From d6e9748e1f9c17a9eb087f3069c8c1fffc0b77bf Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 7 Jul 2026 21:21:32 +0530 Subject: [PATCH 1/2] feat(java): workflow dashboard reads + saga node fields New NativeWorkflows read bindings (listWorkflowRuns, getWorkflowRun, getWorkflowChildren, getWorkflowDag) and the four saga compensation fields on the node view/NodeSnapshot. Dashboard gains /api/workflows/runs, /runs/{id}, /runs/{id}/dag (edge-enriched), /runs/{id}/children. New WorkflowRunInfo model. --- crates/taskito-java/src/workflows/mod.rs | 145 ++++++++++++++++++ .../org/byteveda/taskito/DefaultTaskito.java | 21 +++ .../java/org/byteveda/taskito/Taskito.java | 13 ++ .../taskito/dashboard/DashboardServer.java | 8 + .../taskito/dashboard/api/Contract.java | 34 ++++ .../dashboard/api/WorkflowsHandlers.java | 134 ++++++++++++++++ .../taskito/internal/JniQueueBackend.java | 21 +++ .../taskito/internal/NativeWorkflows.java | 9 ++ .../taskito/model/WorkflowRunInfo.java | 49 ++++++ .../byteveda/taskito/spi/QueueBackend.java | 16 ++ .../taskito/workflows/NodeSnapshot.java | 16 +- .../dashboard/DashboardWorkflowTest.java | 61 ++++++++ 12 files changed, 526 insertions(+), 1 deletion(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/WorkflowsHandlers.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/model/WorkflowRunInfo.java create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardWorkflowTest.java diff --git a/crates/taskito-java/src/workflows/mod.rs b/crates/taskito-java/src/workflows/mod.rs index 51bdaff6..9fab3710 100644 --- a/crates/taskito-java/src/workflows/mod.rs +++ b/crates/taskito-java/src/workflows/mod.rs @@ -97,6 +97,10 @@ struct NodeView<'a> { started_at: Option, completed_at: Option, error: Option<&'a str>, + compensation_job_id: Option<&'a str>, + compensation_started_at: Option, + compensation_completed_at: Option, + compensation_error: Option<&'a str>, } impl<'a> From<&'a WorkflowNode> for NodeView<'a> { @@ -110,6 +114,43 @@ impl<'a> From<&'a WorkflowNode> for NodeView<'a> { started_at: n.started_at, completed_at: n.completed_at, error: n.error.as_deref(), + compensation_job_id: n.compensation_job_id.as_deref(), + compensation_started_at: n.compensation_started_at, + compensation_completed_at: n.compensation_completed_at, + compensation_error: n.compensation_error.as_deref(), + } + } +} + +/// A workflow run summary (no node detail — `getWorkflowStatus` provides that). +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct WorkflowRunView<'a> { + id: &'a str, + definition_id: &'a str, + state: &'static str, + params: Option<&'a str>, + error: Option<&'a str>, + started_at: Option, + completed_at: Option, + created_at: i64, + parent_run_id: Option<&'a str>, + parent_node_name: Option<&'a str>, +} + +impl<'a> From<&'a WorkflowRun> for WorkflowRunView<'a> { + fn from(r: &'a WorkflowRun) -> Self { + Self { + id: &r.id, + definition_id: &r.definition_id, + state: r.state.as_str(), + params: r.params.as_deref(), + error: r.error.as_deref(), + started_at: r.started_at, + completed_at: r.completed_at, + created_at: r.created_at, + parent_run_id: r.parent_run_id.as_deref(), + parent_node_name: r.parent_node_name.as_deref(), } } } @@ -1348,3 +1389,107 @@ fn cascade_skip_pending( } } } + +/// `String listWorkflowRuns(long, String definitionNameOrNull, String stateOrNull, +/// long limit, long offset)` — run summaries as a JSON array (dashboard read). +#[no_mangle] +#[allow(clippy::too_many_arguments)] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_listWorkflowRuns< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + definition_name: JString<'local>, + state: JString<'local>, + limit: jlong, + offset: jlong, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let definition_name = read_optional_string(env, &definition_name)?; + let state_raw = read_optional_string(env, &state)?; + let state = match state_raw.as_deref() { + Some(s) => Some( + WorkflowState::from_str_val(s) + .ok_or_else(|| BindingError::new(format!("unknown workflow state '{s}'")))?, + ), + None => None, + }; + let wf = queue.workflow_store()?; + let runs = wf.list_workflow_runs( + definition_name.as_deref(), + state, + limit.max(0), + offset.max(0), + )?; + let views: Vec = runs.iter().map(WorkflowRunView::from).collect(); + new_string(env, to_json(&views)?) + }) +} + +/// `String getWorkflowRun(long, String runId)` — a single run summary, or +/// `null` if the run no longer exists. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_getWorkflowRun<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let wf = queue.workflow_store()?; + match wf.get_workflow_run(&run_id)? { + Some(run) => new_string(env, to_json(&WorkflowRunView::from(&run))?), + None => Ok(std::ptr::null_mut()), + } + }) +} + +/// `String getWorkflowChildren(long, String runId)` — sub-workflow runs spawned +/// by a run, as a JSON array of run summaries. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_getWorkflowChildren< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let wf = queue.workflow_store()?; + let runs = wf.get_child_workflow_runs(&run_id)?; + let views: Vec = runs.iter().map(WorkflowRunView::from).collect(); + new_string(env, to_json(&views)?) + }) +} + +/// `String getWorkflowDag(long, String runId)` — the serialized DAG JSON backing +/// a run, or `null` if the run or its definition no longer exists. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_getWorkflowDag<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let wf = queue.workflow_store()?; + let Some(run) = wf.get_workflow_run(&run_id)? else { + return Ok(std::ptr::null_mut()); + }; + let Some(def) = wf.get_workflow_definition_by_id(&run.definition_id)? else { + return Ok(std::ptr::null_mut()); + }; + let dag = String::from_utf8(def.dag_data) + .map_err(|e| BindingError::new(format!("invalid DAG utf8: {e}")))?; + new_string(env, dag) + }) +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java index ba86ef93..1d2dd0a8 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -41,6 +41,7 @@ import org.byteveda.taskito.model.TaskLog; import org.byteveda.taskito.model.TaskMetric; import org.byteveda.taskito.model.WorkerInfo; +import org.byteveda.taskito.model.WorkflowRunInfo; import org.byteveda.taskito.predicates.EnqueueDecision; import org.byteveda.taskito.predicates.EnqueueGate; import org.byteveda.taskito.predicates.Predicate; @@ -774,6 +775,26 @@ public void cancelWorkflow(String runId) { backend.cancelWorkflowRun(runId); } + @Override + public List listWorkflowRuns(String definitionName, String state, long limit, long offset) { + return decodeList(backend.listWorkflowRunsJson(definitionName, state, limit, offset), WorkflowRunInfo.class); + } + + @Override + public Optional getWorkflowRun(String runId) { + return backend.getWorkflowRunJson(runId).map(json -> decode(json, WorkflowRunInfo.class)); + } + + @Override + public List getWorkflowChildren(String runId) { + return decodeList(backend.getWorkflowChildrenJson(runId), WorkflowRunInfo.class); + } + + @Override + public Optional getWorkflowDag(String runId) { + return backend.getWorkflowDagJson(runId); + } + /** Encode a step's structure + per-step overrides for the native submit call. */ private Map stepSpec(Step step) { Map spec = new LinkedHashMap<>(); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java index f0929421..28374f97 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java @@ -33,6 +33,7 @@ import org.byteveda.taskito.model.TaskLog; import org.byteveda.taskito.model.TaskMetric; import org.byteveda.taskito.model.WorkerInfo; +import org.byteveda.taskito.model.WorkflowRunInfo; import org.byteveda.taskito.predicates.EnqueueGate; import org.byteveda.taskito.predicates.Predicate; import org.byteveda.taskito.resources.PoolConfig; @@ -292,6 +293,18 @@ static Builder builder() { /** Cancel a workflow run: skip its pending nodes and mark it cancelled. */ void cancelWorkflow(String runId); + /** Workflow run summaries, filtered by definition name and/or state, paged. Nulls mean no filter. */ + List listWorkflowRuns(String definitionName, String state, long limit, long offset); + + /** A single workflow run summary, or empty if the run no longer exists. */ + Optional getWorkflowRun(String runId); + + /** Sub-workflow runs spawned by a run. */ + List getWorkflowChildren(String runId); + + /** The serialized DAG JSON backing a run, or empty if the run/definition is gone. */ + Optional getWorkflowDag(String runId); + // ── Worker ────────────────────────────────────────────────────── /** Begin building a worker over this client. */ diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java index 27f7621b..74c221d9 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java @@ -19,6 +19,7 @@ import org.byteveda.taskito.dashboard.api.OpsHandlers; import org.byteveda.taskito.dashboard.api.OverridesHandlers; import org.byteveda.taskito.dashboard.api.SettingsHandlers; +import org.byteveda.taskito.dashboard.api.WorkflowsHandlers; import org.byteveda.taskito.dashboard.auth.AuthHandlers; import org.byteveda.taskito.dashboard.auth.AuthStore; import org.byteveda.taskito.dashboard.auth.Cookies; @@ -202,6 +203,7 @@ private Router buildRouter() { SettingsHandlers settingsApi = new SettingsHandlers(settings); OverridesHandlers overrides = new OverridesHandlers(queue, new OverridesStore(settings)); MetricsHandlers metrics = new MetricsHandlers(queue); + WorkflowsHandlers workflows = new WorkflowsHandlers(queue); long ttl = AuthStore.DEFAULT_SESSION_TTL_SECONDS; Router r = new Router(); @@ -243,6 +245,12 @@ private Router buildRouter() { r.get("/api/scaler", req -> ops.scaler(req.query())); r.get("/api/resources", req -> ops.resources()); + // Workflows (read) + r.get("/api/workflows/runs", req -> workflows.runs(req.query())); + r.get("/api/workflows/runs/([^/]+)/dag", req -> workflows.dag(req.param(0))); + r.get("/api/workflows/runs/([^/]+)/children", req -> workflows.children(req.param(0))); + r.get("/api/workflows/runs/([^/]+)", req -> workflows.run(req.param(0))); + // Settings KV r.get("/api/settings", req -> settingsApi.list()); r.get("/api/settings/(.+)", req -> settingsApi.get(req.param(0))); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java index d5e0c36d..84c1956d 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/Contract.java @@ -12,6 +12,8 @@ import org.byteveda.taskito.model.ReplayEntry; import org.byteveda.taskito.model.TaskLog; import org.byteveda.taskito.model.WorkerInfo; +import org.byteveda.taskito.model.WorkflowRunInfo; +import org.byteveda.taskito.workflows.NodeSnapshot; /** * Maps the SDK's camelCase types to the snake_case JSON contract the React SPA @@ -139,4 +141,36 @@ private static Map dagEdge(DagEdge e) { m.put("to", e.to); return m; } + + static Map workflowRun(WorkflowRunInfo r) { + Map m = new LinkedHashMap<>(); + m.put("id", r.id); + m.put("definition_id", r.definitionId); + m.put("state", r.state); + m.put("params", r.params); + m.put("started_at", r.startedAt); + m.put("completed_at", r.completedAt); + m.put("error", r.error); + m.put("parent_run_id", r.parentRunId); + m.put("parent_node_name", r.parentNodeName); + m.put("created_at", r.createdAt); + return m; + } + + static Map workflowNode(NodeSnapshot n) { + Map m = new LinkedHashMap<>(); + m.put("node_name", n.nodeName); + m.put("status", n.status); + m.put("job_id", n.jobId); + m.put("result_hash", n.resultHash); + m.put("fan_out_count", n.fanOutCount); + m.put("started_at", n.startedAt); + m.put("completed_at", n.completedAt); + m.put("error", n.error); + m.put("compensation_job_id", n.compensationJobId); + m.put("compensation_started_at", n.compensationStartedAt); + m.put("compensation_completed_at", n.compensationCompletedAt); + m.put("compensation_error", n.compensationError); + return m; + } } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/WorkflowsHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/WorkflowsHandlers.java new file mode 100644 index 00000000..4b840dbd --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/WorkflowsHandlers.java @@ -0,0 +1,134 @@ +package org.byteveda.taskito.dashboard.api; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.dashboard.support.Json; +import org.byteveda.taskito.workflows.NodeSnapshot; + +/** + * Workflow read endpoints: run listing, run detail (run + nodes), children, and + * the DAG. The DAG is enriched — edges rebuilt from each node's {@code deps} and + * live status/job-id folded in — so the SPA's graph view renders real links. + */ +public final class WorkflowsHandlers { + private static final long DEFAULT_LIMIT = 50; + + private final Taskito queue; + + public WorkflowsHandlers(Taskito queue) { + this.queue = queue; + } + + public Object runs(Map query) { + long limit = longParam(query, "limit", DEFAULT_LIMIT); + long offset = longParam(query, "offset", 0); + List runs = + queue.listWorkflowRuns(query.get("definition_name"), query.get("state"), limit, offset).stream() + .map(Contract::workflowRun) + .collect(Collectors.toList()); + Map out = new LinkedHashMap<>(); + out.put("runs", runs); + out.put("limit", limit); + out.put("offset", offset); + return out; + } + + public Object run(String id) { + var run = queue.getWorkflowRun(id).orElse(null); + if (run == null) { + return null; + } + List nodes = nodesOf(id).stream().map(Contract::workflowNode).collect(Collectors.toList()); + Map out = new LinkedHashMap<>(); + out.put("run", Contract.workflowRun(run)); + out.put("nodes", nodes); + return out; + } + + public Object children(String id) { + Map out = new LinkedHashMap<>(); + out.put( + "children", + queue.getWorkflowChildren(id).stream() + .map(Contract::workflowRun) + .collect(Collectors.toList())); + return out; + } + + public Object dag(String id) { + String dag = queue.getWorkflowDag(id).orElse(null); + if (dag == null) { + return null; + } + Map out = new LinkedHashMap<>(); + out.put("dag", enrichDag(dag, nodesOf(id))); + return out; + } + + private List nodesOf(String id) { + return queue.workflowStatus(id).map(status -> status.nodes).orElse(List.of()); + } + + /** + * Rewrite the raw {@code SerializableGraph} into the SPA's DAG shape: edges + * come from each node's incoming edges, and live status/job-id are folded in. + * Returns a JSON string (the SPA parses it). + */ + private static String enrichDag(String dagJson, List nodes) { + Map graph = Json.parseMap(dagJson); + if (graph == null) { + return dagJson; // not our JSON — pass through + } + Map byName = new HashMap<>(); + for (NodeSnapshot node : nodes) { + byName.put(node.nodeName, node); + } + List> edges = asMapList(graph.get("edges")); + List> enriched = new ArrayList<>(); + for (Map raw : asMapList(graph.get("nodes"))) { + String name = raw.get("name") == null ? "" : String.valueOf(raw.get("name")); + NodeSnapshot node = byName.get(name); + List deps = new ArrayList<>(); + for (Map edge : edges) { + if (name.equals(edge.get("to"))) { + deps.add(edge.get("from")); + } + } + Map entry = new LinkedHashMap<>(); + entry.put("name", name); + entry.put("node_name", name); + entry.put("status", node != null ? node.status : "pending"); + entry.put("id", node != null && node.jobId != null ? node.jobId : name); + entry.put("deps", deps); + enriched.add(entry); + } + Map result = new LinkedHashMap<>(); + result.put("nodes", enriched); + result.put("edges", edges); + return Json.toString(result); + } + + @SuppressWarnings("unchecked") + private static List> asMapList(Object value) { + if (!(value instanceof List)) { + return List.of(); + } + List> out = new ArrayList<>(); + for (Object item : (List) value) { + if (item instanceof Map) { + out.add((Map) item); + } + } + return out; + } + + private static long longParam(Map query, String key, long fallback) { + String value = query.get(key); + return value == null ? fallback : Long.parseLong(value); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java index 07cf43de..730d01dd 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java @@ -323,6 +323,27 @@ public Optional getWorkflowStatusJson(String runId) { return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.getWorkflowStatus(handle, runId))); } + @Override + public String listWorkflowRunsJson(String definitionNameOrNull, String stateOrNull, long limit, long offset) { + return withOpenHandle( + () -> NativeWorkflows.listWorkflowRuns(handle, definitionNameOrNull, stateOrNull, limit, offset)); + } + + @Override + public Optional getWorkflowRunJson(String runId) { + return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.getWorkflowRun(handle, runId))); + } + + @Override + public String getWorkflowChildrenJson(String runId) { + return withOpenHandle(() -> NativeWorkflows.getWorkflowChildren(handle, runId)); + } + + @Override + public Optional getWorkflowDagJson(String runId) { + return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.getWorkflowDag(handle, runId))); + } + @Override public void cancelWorkflowRun(String runId) { withOpenHandle(() -> { diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.java index e28dfc00..20c24bd9 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.java @@ -40,6 +40,15 @@ public static native String markWorkflowNodeResult( /** Returns a JSON run + node snapshot, or {@code null} if the run is absent. */ public static native String getWorkflowStatus(long handle, String runId); + public static native String listWorkflowRuns( + long handle, String definitionNameOrNull, String stateOrNull, long limit, long offset); + + public static native String getWorkflowRun(long handle, String runId); + + public static native String getWorkflowChildren(long handle, String runId); + + public static native String getWorkflowDag(long handle, String runId); + public static native void cancelWorkflowRun(long handle, String runId); // ── Fan-out / fan-in orchestration (driven by the worker-side tracker) ── diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/WorkflowRunInfo.java b/sdks/java/src/main/java/org/byteveda/taskito/model/WorkflowRunInfo.java new file mode 100644 index 00000000..8c6259f0 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/WorkflowRunInfo.java @@ -0,0 +1,49 @@ +package org.byteveda.taskito.model; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import org.byteveda.taskito.workflows.WorkflowState; + +/** + * A workflow run summary (no node detail). {@code createdAt} is Unix + * milliseconds; start/complete are nullable. Named {@code WorkflowRunInfo} to + * avoid clashing with the live {@code workflows.WorkflowRun} handle. + */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class WorkflowRunInfo { + public final String id; + public final String definitionId; + public final WorkflowState state; + public final String params; + public final String error; + public final Long startedAt; + public final Long completedAt; + public final long createdAt; + public final String parentRunId; + public final String parentNodeName; + + @JsonCreator + public WorkflowRunInfo( + @JsonProperty("id") String id, + @JsonProperty("definitionId") String definitionId, + @JsonProperty("state") WorkflowState state, + @JsonProperty("params") String params, + @JsonProperty("error") String error, + @JsonProperty("startedAt") Long startedAt, + @JsonProperty("completedAt") Long completedAt, + @JsonProperty("createdAt") long createdAt, + @JsonProperty("parentRunId") String parentRunId, + @JsonProperty("parentNodeName") String parentNodeName) { + this.id = id; + this.definitionId = definitionId; + this.state = state; + this.params = params; + this.error = error; + this.startedAt = startedAt; + this.completedAt = completedAt; + this.createdAt = createdAt; + this.parentRunId = parentRunId; + this.parentNodeName = parentNodeName; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java index 63c55a9a..30e2f7ad 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java @@ -190,6 +190,22 @@ default Optional getWorkflowStatusJson(String runId) { throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); } + default String listWorkflowRunsJson(String definitionNameOrNull, String stateOrNull, long limit, long offset) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + default Optional getWorkflowRunJson(String runId) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + default String getWorkflowChildrenJson(String runId) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + default Optional getWorkflowDagJson(String runId) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + default void cancelWorkflowRun(String runId) { throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/NodeSnapshot.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/NodeSnapshot.java index d5f5415c..197e353f 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/NodeSnapshot.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/NodeSnapshot.java @@ -15,6 +15,12 @@ public final class NodeSnapshot { public final Long startedAt; public final Long completedAt; public final String error; + /** Saga rollback job for this node; null outside a compensation flow. */ + public final String compensationJobId; + + public final Long compensationStartedAt; + public final Long compensationCompletedAt; + public final String compensationError; @JsonCreator public NodeSnapshot( @@ -25,7 +31,11 @@ public NodeSnapshot( @JsonProperty("fanOutCount") Integer fanOutCount, @JsonProperty("startedAt") Long startedAt, @JsonProperty("completedAt") Long completedAt, - @JsonProperty("error") String error) { + @JsonProperty("error") String error, + @JsonProperty("compensationJobId") String compensationJobId, + @JsonProperty("compensationStartedAt") Long compensationStartedAt, + @JsonProperty("compensationCompletedAt") Long compensationCompletedAt, + @JsonProperty("compensationError") String compensationError) { this.nodeName = nodeName; this.status = status; this.jobId = jobId; @@ -34,5 +44,9 @@ public NodeSnapshot( this.startedAt = startedAt; this.completedAt = completedAt; this.error = error; + this.compensationJobId = compensationJobId; + this.compensationStartedAt = compensationStartedAt; + this.compensationCompletedAt = compensationCompletedAt; + this.compensationError = compensationError; } } diff --git a/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardWorkflowTest.java b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardWorkflowTest.java new file mode 100644 index 00000000..530b1e29 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardWorkflowTest.java @@ -0,0 +1,61 @@ +package org.byteveda.taskito.dashboard; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.workflows.Workflow; +import org.byteveda.taskito.workflows.WorkflowRun; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** Coverage of the workflow read endpoints: runs, run detail, DAG, children. */ +@Timeout(30) +class DashboardWorkflowTest { + + private static final Task STEP = Task.of("wf.step", Integer.class); + + private static Workflow demo() { + return Workflow.named("demo").step("a", STEP, 1).step("b", STEP, 2, "a"); + } + + @Test + void listsRunsAndReturnsDetail(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().sqlite(dir.resolve("t.db").toString()).open(); + DashboardServer server = DashboardServer.start(queue, 0)) { + WorkflowRun run = queue.submitWorkflow(demo()); + String runId = run.id(); + DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue)); + + String runs = client.get("/api/workflows/runs").body(); + assertTrue(runs.contains("\"runs\"") && runs.contains(runId)); + + String detail = client.get("/api/workflows/runs/" + runId).body(); + assertTrue(detail.contains("\"run\"") && detail.contains("\"nodes\"")); + assertTrue(detail.contains("\"node_name\":\"a\"")); + + assertTrue( + client.get("/api/workflows/runs/" + runId + "/dag").body().contains("\"dag\"")); + assertTrue(client.get("/api/workflows/runs/" + runId + "/children") + .body() + .contains("\"children\"")); + } + } + + @Test + void missingRunIs404(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().sqlite(dir.resolve("t.db").toString()).open(); + DashboardServer server = DashboardServer.start(queue, 0)) { + // Touch the workflow store so it exists, then query a missing run. + queue.submitWorkflow(demo()); + DashboardClient client = new DashboardClient(server.port()).as(DashboardClient.seedAdmin(queue)); + assertEquals(404, client.get("/api/workflows/runs/nope").statusCode()); + assertEquals(404, client.get("/api/workflows/runs/nope/dag").statusCode()); + } + } +} From d4874490b9d7463d5f615e41f342e3779a0ba15a Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Wed, 8 Jul 2026 09:19:51 +0530 Subject: [PATCH 2/2] fix(java): guard workflow list params via shared parser Addresses PR review: WorkflowsHandlers.runs used a local longParam that 500'd on malformed limit/offset; route through Http.longParam so bad input returns 400. --- .../taskito/dashboard/api/WorkflowsHandlers.java | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/WorkflowsHandlers.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/WorkflowsHandlers.java index 4b840dbd..f8f1f381 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/WorkflowsHandlers.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/api/WorkflowsHandlers.java @@ -7,6 +7,7 @@ import java.util.Map; import java.util.stream.Collectors; import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.dashboard.support.Http; import org.byteveda.taskito.dashboard.support.Json; import org.byteveda.taskito.workflows.NodeSnapshot; @@ -25,8 +26,8 @@ public WorkflowsHandlers(Taskito queue) { } public Object runs(Map query) { - long limit = longParam(query, "limit", DEFAULT_LIMIT); - long offset = longParam(query, "offset", 0); + long limit = Http.longParam(query, "limit", DEFAULT_LIMIT); + long offset = Http.longParam(query, "offset", 0); List runs = queue.listWorkflowRuns(query.get("definition_name"), query.get("state"), limit, offset).stream() .map(Contract::workflowRun) @@ -126,9 +127,4 @@ private static List> asMapList(Object value) { } return out; } - - private static long longParam(Map query, String key, long fallback) { - String value = query.get(key); - return value == null ? fallback : Long.parseLong(value); - } }