feat(java): native logs, replay, job DAG, worker resources

crates/taskito-java/src/convert.rs

@@ -7,8 +7,8 @@ use serde::{Deserialize, Serialize};
 use taskito_core::job::{now_millis, Job, NewJob};
 use taskito_core::resilience::circuit_breaker::CircuitState;
 use taskito_core::storage::models::{
-    CircuitBreakerRow, JobErrorRow, LockInfoRow, PeriodicTaskRow, TaskLogRow, TaskMetricRow,
-    WorkerRow,
+    CircuitBreakerRow, JobErrorRow, LockInfoRow, PeriodicTaskRow, ReplayHistoryRow, TaskLogRow,
+    TaskMetricRow, WorkerRow,
 };
 use taskito_core::storage::{DeadJob, QueueStats};
 
@@ -379,6 +379,10 @@ pub struct WorkerView<'a> {
     pub pool_type: Option<&'a str>,
     pub threads: i32,
     pub tags: Option<&'a str>,
+    /// JSON array of resource names the worker advertised at registration.
+    pub resources: Option<&'a str>,
+    /// JSON object of per-resource health, written by the worker's heartbeat.
+    pub resource_health: Option<&'a str>,
 }
 
 impl<'a> From<&'a WorkerRow> for WorkerView<'a> {
@@ -394,6 +398,33 @@ impl<'a> From<&'a WorkerRow> for WorkerView<'a> {
             pool_type: w.pool_type.as_deref(),
             threads: w.threads,
             tags: w.tags.as_deref(),
+            resources: w.resources.as_deref(),
+            resource_health: w.resource_health.as_deref(),
+        }
+    }
+}
+
+/// One replay of a job (result blobs are omitted; ids + errors suffice).
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ReplayEntryView<'a> {
+    pub id: &'a str,
+    pub original_job_id: &'a str,
+    pub replay_job_id: &'a str,
+    pub replayed_at: i64,
+    pub original_error: Option<&'a str>,
+    pub replay_error: Option<&'a str>,
+}
+
+impl<'a> From<&'a ReplayHistoryRow> for ReplayEntryView<'a> {
+    fn from(r: &'a ReplayHistoryRow) -> Self {
+        Self {
+            id: &r.id,
+            original_job_id: &r.original_job_id,
+            replay_job_id: &r.replay_job_id,
+            replayed_at: r.replayed_at,
+            original_error: r.original_error.as_deref(),
+            replay_error: r.replay_error.as_deref(),
         }
     }
 }

crates/taskito-java/src/queue/admin.rs

@@ -3,10 +3,12 @@
 use jni::objects::{JClass, JString};
 use jni::sys::{jboolean, jlong, jstring, JNI_FALSE};
 use jni::JNIEnv;
+use taskito_core::job::{now_millis, NewJob};
 use taskito_core::Storage;
 
 use super::borrow_queue;
-use crate::convert::{to_json, DeadJobView};
+use crate::convert::{to_json, DeadJobView, ReplayEntryView};
+use crate::error::BindingError;
 use crate::ffi::{guard, new_string, read_string};
 
 /// `String listDead(long handle, long limit, long offset)` — dead-letter entries.
@@ -229,3 +231,67 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listSettin
         new_string(env, to_json(&queue.storage.list_settings()?)?)
     })
 }
+
+/// `String replayJob(long handle, String jobId)` — re-enqueue a copy of an
+/// existing job and record it in the replay history. Returns the new job id.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_replayJob<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    job_id: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let id = read_string(env, &job_id)?;
+        let original = queue
+            .storage
+            .get_job(&id)?
+            .ok_or_else(|| BindingError::new(format!("Job not found: {id}")))?;
+        let new_job = NewJob {
+            queue: original.queue.clone(),
+            task_name: original.task_name.clone(),
+            payload: original.payload.clone(),
+            priority: original.priority,
+            scheduled_at: now_millis(),
+            max_retries: original.max_retries,
+            timeout_ms: original.timeout_ms,
+            unique_key: None,
+            metadata: Some(format!("{{\"replayed_from\":\"{id}\"}}")),
+            notes: original.notes.clone(),
+            depends_on: Vec::new(),
+            expires_at: None,
+            result_ttl_ms: original.result_ttl_ms,
+            namespace: original.namespace.clone(),
+        };
+        let job = queue.storage.enqueue(new_job)?;
+        // Best-effort audit row — a history write must not fail the replay.
+        let _ = queue.storage.record_replay(
+            &id,
+            &job.id,
+            original.result.as_deref(),
+            None,
+            original.error.as_deref(),
+            None,
+        );
+        new_string(env, job.id)
+    })
+}
+
+/// `String getReplayHistory(long handle, String jobId)` — replays recorded for
+/// a job, as a JSON array.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_getReplayHistory<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    job_id: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let id = read_string(env, &job_id)?;
+        let rows = queue.storage.get_replay_history(&id)?;
+        let views: Vec<ReplayEntryView> = rows.iter().map(ReplayEntryView::from).collect();
+        new_string(env, to_json(&views)?)
+    })
+}

crates/taskito-java/src/queue/inspect.rs

@@ -1,10 +1,12 @@
 //! Read-only inspection entry points for `NativeQueue`.
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use jni::objects::{JClass, JString};
 use jni::sys::{jlong, jstring};
 use jni::JNIEnv;
+use serde::Serialize;
+use taskito_core::job::Job;
 use taskito_core::Storage;
 
 use super::borrow_queue;
@@ -15,6 +17,22 @@ use crate::convert::{
 use crate::error::BindingError;
 use crate::ffi::{guard, new_string, read_optional_string, read_string};
 
+/// One `dependency -> dependent` edge in a job DAG.
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct DagEdgeView {
+    from: String,
+    to: String,
+}
+
+/// The dependency DAG reachable from a job: full job rows plus edges.
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct JobDagView<'a> {
+    nodes: Vec<JobView<'a>>,
+    edges: Vec<DagEdgeView>,
+}
+
 const DEFAULT_LIMIT: i64 = 50;
 
 /// `String stats(long handle)` — job counts by status across all queues.
@@ -165,3 +183,56 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listCircui
         new_string(env, to_json(&views)?)
     })
 }
+
+/// `String jobDag(long handle, String jobId)` — the dependency DAG reachable
+/// from a job as a JSON `{nodes, edges}` object. Walks dependencies and
+/// dependents in both directions, deduping nodes and edges.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_jobDag<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    job_id: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let start = read_string(env, &job_id)?;
+
+        let mut visited: HashSet<String> = HashSet::new();
+        // Both endpoints of an edge report it; dedupe by (from, to).
+        let mut seen_edges: HashSet<(String, String)> = HashSet::new();
+        // Keep the jobs alive so the borrowing `JobView`s can be built at the end.
+        let mut jobs: Vec<Job> = Vec::new();
+        let mut edges: Vec<DagEdgeView> = Vec::new();
+        let mut pending = vec![start];
+        while let Some(current) = pending.pop() {
+            if !visited.insert(current.clone()) {
+                continue;
+            }
+            let Some(job) = queue.storage.get_job(&current)? else {
+                continue;
+            };
+            jobs.push(job);
+            for dep_id in queue.storage.get_dependencies(&current)? {
+                if seen_edges.insert((dep_id.clone(), current.clone())) {
+                    edges.push(DagEdgeView {
+                        from: dep_id.clone(),
+                        to: current.clone(),
+                    });
+                }
+                pending.push(dep_id);
+            }
+            for dep_id in queue.storage.get_dependents(&current)? {
+                if seen_edges.insert((current.clone(), dep_id.clone())) {
+                    edges.push(DagEdgeView {
+                        from: current.clone(),
+                        to: dep_id.clone(),
+                    });
+                }
+                pending.push(dep_id);
+            }
+        }
+        let nodes: Vec<JobView> = jobs.iter().map(JobView::from).collect();
+        new_string(env, to_json(&JobDagView { nodes, edges })?)
+    })
+}

crates/taskito-java/src/queue/logs.rs

@@ -51,3 +51,51 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_getTaskLog
         new_string(env, to_json(&views)?)
     })
 }
+
+/// `String getTaskLogsAfter(long handle, String jobId, String afterIdOrNull)` —
+/// a JSON array of log lines with id strictly after `afterId` (`null` = all).
+/// UUIDv7 ids are time-ordered, so the last id doubles as a stream cursor.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_getTaskLogsAfter<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    job_id: JString<'local>,
+    after_id: JString<'local>,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let id = read_string(env, &job_id)?;
+        let after = read_optional_string(env, &after_id)?;
+        let logs = queue.storage.get_task_logs_after(&id, after.as_deref())?;
+        let views: Vec<LogView> = logs.iter().map(LogView::from).collect();
+        new_string(env, to_json(&views)?)
+    })
+}
+
+/// `String queryTaskLogs(long handle, String taskOrNull, String levelOrNull, long sinceMs, long limit)`
+/// — logs across jobs filtered by task/level, at or after `sinceMs`, newest capped at `limit`.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_queryTaskLogs<'local>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    task_name: JString<'local>,
+    level: JString<'local>,
+    since_ms: jlong,
+    limit: jlong,
+) -> jstring {
+    guard(&mut env, std::ptr::null_mut(), |env| {
+        let queue = unsafe { borrow_queue(handle) };
+        let task = read_optional_string(env, &task_name)?;
+        let level = read_optional_string(env, &level)?;
+        let logs = queue.storage.query_task_logs(
+            task.as_deref(),
+            level.as_deref(),
+            since_ms,
+            limit.max(0),
+        )?;
+        let views: Vec<LogView> = logs.iter().map(LogView::from).collect();
+        new_string(env, to_json(&views)?)
+    })
+}

sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java

@@ -32,10 +32,12 @@
 import org.byteveda.taskito.model.CircuitBreakerState;
 import org.byteveda.taskito.model.DeadJob;
 import org.byteveda.taskito.model.Job;
+import org.byteveda.taskito.model.JobDag;
 import org.byteveda.taskito.model.JobError;
 import org.byteveda.taskito.model.JobFilter;
 import org.byteveda.taskito.model.PeriodicInfo;
 import org.byteveda.taskito.model.QueueStats;
+import org.byteveda.taskito.model.ReplayEntry;
 import org.byteveda.taskito.model.TaskLog;
 import org.byteveda.taskito.model.TaskMetric;
 import org.byteveda.taskito.model.WorkerInfo;
@@ -506,6 +508,21 @@ public String retry(String deadId) {
         return backend.retryDead(deadId);
     }
 
+    @Override
+    public String replayJob(String jobId) {
+        return backend.replayJob(jobId);
+    }
+
+    @Override
+    public List<ReplayEntry> getReplayHistory(String jobId) {
+        return decodeList(backend.getReplayHistoryJson(jobId), ReplayEntry.class);
+    }
+
+    @Override
+    public JobDag jobDag(String jobId) {
+        return decode(backend.jobDagJson(jobId), JobDag.class);
+    }
+
     @Override
     public boolean deleteDead(String deadId) {
         return backend.deleteDead(deadId);
@@ -573,6 +590,16 @@ public List<TaskLog> getTaskLogs(String jobId) {
         return decodeList(backend.getTaskLogsJson(jobId), TaskLog.class);
     }
 
+    @Override
+    public List<TaskLog> getTaskLogsAfter(String jobId, String afterId) {
+        return decodeList(backend.getTaskLogsAfterJson(jobId, afterId), TaskLog.class);
+    }
+
+    @Override
+    public List<TaskLog> queryTaskLogs(String taskName, String level, long sinceMs, long limit) {
+        return decodeList(backend.queryTaskLogsJson(taskName, level, sinceMs, limit), TaskLog.class);
+    }
+
     @Override
     public Lock lock(String name, long ttlMs) {
         return new Lock(backend, name, ttlMs);

sdks/java/src/main/java/org/byteveda/taskito/Taskito.java

@@ -24,10 +24,12 @@
 import org.byteveda.taskito.model.CircuitBreakerState;
 import org.byteveda.taskito.model.DeadJob;
 import org.byteveda.taskito.model.Job;
+import org.byteveda.taskito.model.JobDag;
 import org.byteveda.taskito.model.JobError;
 import org.byteveda.taskito.model.JobFilter;
 import org.byteveda.taskito.model.PeriodicInfo;
 import org.byteveda.taskito.model.QueueStats;
+import org.byteveda.taskito.model.ReplayEntry;
 import org.byteveda.taskito.model.TaskLog;
 import org.byteveda.taskito.model.TaskMetric;
 import org.byteveda.taskito.model.WorkerInfo;
@@ -198,6 +200,15 @@ static Builder builder() {
     /** Alias of {@link #retryDead(String)} in the guide's vocabulary. */
     String retry(String deadId);
 
+    /** Re-enqueue a copy of a job (recording it in the replay history); returns the new job id. */
+    String replayJob(String jobId);
+
+    /** A job's replay history. */
+    List<ReplayEntry> getReplayHistory(String jobId);
+
+    /** The dependency DAG reachable from a job (nodes plus {@code from → to} edges). */
+    JobDag jobDag(String jobId);
+
     boolean deleteDead(String deadId);
 
     long purgeDead(long olderThanMs);
@@ -223,6 +234,12 @@ static Builder builder() {
 
     List<TaskLog> getTaskLogs(String jobId);
 
+    /** Logs for a job with id after {@code afterId} (UUIDv7-ordered cursor); null = all. */
+    List<TaskLog> getTaskLogsAfter(String jobId, String afterId);
+
+    /** Logs across jobs filtered by task/level, at or after {@code sinceMs}, capped at {@code limit}. */
+    List<TaskLog> queryTaskLogs(String taskName, String level, long sinceMs, long limit);
+
     // ── Locks ───────────────────────────────────────────────────────
 
     /** A distributed lock {@code name} with the given TTL; call {@link Lock#acquire()}. */