From 97b290cf9bbb5612341ee1a3c6ddc719f8f403df Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 28 Jun 2026 19:57:57 +0530 Subject: [PATCH 1/7] feat(java): saga compensation for workflows A failed run with compensable steps rolls back: Step.compensate registers a rollback task, and on failure the tracker compensates completed steps in reverse-dependency waves (parallel within a wave). Adds saga JNI fns + compensate metadata + a SagaOrchestrator; the run ends compensated (or compensation_failed). The terminal state is set without a transient Failed. --- crates/taskito-java/src/workflows/mod.rs | 178 ++++++++++++++ .../org/byteveda/taskito/DefaultTaskito.java | 3 + .../taskito/internal/JniQueueBackend.java | 36 +++ .../taskito/internal/NativeWorkflows.java | 19 ++ .../byteveda/taskito/spi/QueueBackend.java | 31 +++ .../byteveda/taskito/workflows/PlanNode.java | 5 +- .../byteveda/taskito/workflows/RunPlan.java | 10 + .../taskito/workflows/SagaOrchestrator.java | 218 ++++++++++++++++++ .../org/byteveda/taskito/workflows/Step.java | 18 ++ .../taskito/workflows/WorkflowTracker.java | 33 ++- .../byteveda/taskito/WorkflowSagaTest.java | 90 ++++++++ 11 files changed, 639 insertions(+), 2 deletions(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/WorkflowSagaTest.java diff --git a/crates/taskito-java/src/workflows/mod.rs b/crates/taskito-java/src/workflows/mod.rs index 6a111a1c..f9f7d4f1 100644 --- a/crates/taskito-java/src/workflows/mod.rs +++ b/crates/taskito-java/src/workflows/mod.rs @@ -66,6 +66,8 @@ struct StepSpec { gate: Option, /// Serialized child-workflow spec marking a sub-workflow node. sub_workflow: Option, + /// Rollback task name — the saga runs it to compensate this node. + compensate: Option, } /// Combined run + node snapshot returned by `getWorkflowStatus`. Timestamps are @@ -230,6 +232,7 @@ fn submit( condition: s.condition.clone(), gate: s.gate.clone(), sub_workflow: s.sub_workflow.clone(), + compensate: s.compensate.clone(), ..Default::default() }, ) @@ -518,6 +521,7 @@ struct PlanNodeView<'a> { condition: Option<&'a str>, gate: Option<&'a str>, sub_workflow: Option<&'a str>, + compensate: Option<&'a str>, } /// The run + node a job belongs to. @@ -578,6 +582,7 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_getWor condition: meta.and_then(|m| m.condition.as_deref()), gate: meta.and_then(|m| m.gate.as_deref()), sub_workflow: meta.and_then(|m| m.sub_workflow.as_deref()), + compensate: meta.and_then(|m| m.compensate.as_deref()), } }) .collect(); @@ -1058,6 +1063,179 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_skipWo }) } +/// `void setWorkflowRunCompensating(long, String runId)` — mark a run as +/// compensating (the saga is rolling back completed nodes). +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_setWorkflowRunCompensating< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + queue.workflow_store()?.update_workflow_run_state( + &run_id, + WorkflowState::Compensating, + None, + )?; + Ok(()) + }) +} + +/// `void setWorkflowRunCompensated(long, String runId, long completedAt)` — the +/// saga rolled everything back successfully. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_setWorkflowRunCompensated< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + completed_at: jlong, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let wf = queue.workflow_store()?; + wf.update_workflow_run_state(&run_id, WorkflowState::Compensated, None)?; + wf.set_workflow_run_completed(&run_id, completed_at)?; + Ok(()) + }) +} + +/// `void setWorkflowRunCompensationFailed(long, String runId, long completedAt, +/// String error)` — a compensation job itself failed; rollback is incomplete. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_setWorkflowRunCompensationFailed< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + completed_at: jlong, + error: JString<'local>, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let error = read_optional_string(env, &error)?; + let wf = queue.workflow_store()?; + wf.update_workflow_run_state(&run_id, WorkflowState::CompensationFailed, error.as_deref())?; + wf.set_workflow_run_completed(&run_id, completed_at)?; + Ok(()) + }) +} + +/// `void setWorkflowRunCompletedWithFailures(long, String runId, long completedAt)` +/// — the run finished with some failed nodes (on-failure=continue). +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_setWorkflowRunCompletedWithFailures< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + completed_at: jlong, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let wf = queue.workflow_store()?; + wf.update_workflow_run_state(&run_id, WorkflowState::CompletedWithFailures, None)?; + wf.set_workflow_run_completed(&run_id, completed_at)?; + Ok(()) + }) +} + +/// `void setWorkflowNodeCompensationJob(long, String runId, String nodeName, +/// String compensationJobId, long startedAt)` — bind a node's rollback job. +#[no_mangle] +#[allow(clippy::too_many_arguments)] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_setWorkflowNodeCompensationJob< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + node_name: JString<'local>, + compensation_job_id: JString<'local>, + started_at: jlong, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let node_name = read_string(env, &node_name)?; + let comp_job_id = read_string(env, &compensation_job_id)?; + queue.workflow_store()?.set_workflow_node_compensation_job( + &run_id, + &node_name, + &comp_job_id, + started_at, + )?; + Ok(()) + }) +} + +/// `void setWorkflowNodeCompensated(long, String runId, String nodeName, +/// long completedAt)` — a node's rollback succeeded. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_setWorkflowNodeCompensated< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + node_name: JString<'local>, + completed_at: jlong, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let node_name = read_string(env, &node_name)?; + queue + .workflow_store()? + .set_workflow_node_compensated(&run_id, &node_name, completed_at)?; + Ok(()) + }) +} + +/// `void setWorkflowNodeCompensationFailed(long, String runId, String nodeName, +/// String error, long completedAt)` — a node's rollback failed. +#[no_mangle] +#[allow(clippy::too_many_arguments)] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_setWorkflowNodeCompensationFailed< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + node_name: JString<'local>, + error: JString<'local>, + completed_at: jlong, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let node_name = read_string(env, &node_name)?; + let error = + read_optional_string(env, &error)?.unwrap_or_else(|| "compensation failed".to_string()); + queue + .workflow_store()? + .set_workflow_node_compensation_failed(&run_id, &node_name, &error, completed_at)?; + Ok(()) + }) +} + /// A fresh `Pending` node. `job_id` is `None` for deferred nodes. fn new_node(run_id: &str, node_name: &str, job_id: Option) -> WorkflowNode { WorkflowNode { 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 40da2b1e..82299187 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -475,6 +475,9 @@ private Map stepSpec(Step step) { if (step.subWorkflow != null) { spec.put("subWorkflow", encodeChild(step.subWorkflow)); } + if (step.compensate != null) { + spec.put("compensate", step.compensate); + } return spec; } 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 795379e1..c7e42083 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 @@ -340,6 +340,42 @@ public void skipWorkflowNode(String runId, String nodeName) { NativeWorkflows.skipWorkflowNode(handle, runId, nodeName); } + @Override + public void setWorkflowRunCompensating(String runId) { + NativeWorkflows.setWorkflowRunCompensating(handle, runId); + } + + @Override + public void setWorkflowRunCompensated(String runId, long completedAt) { + NativeWorkflows.setWorkflowRunCompensated(handle, runId, completedAt); + } + + @Override + public void setWorkflowRunCompensationFailed(String runId, long completedAt, String error) { + NativeWorkflows.setWorkflowRunCompensationFailed(handle, runId, completedAt, error); + } + + @Override + public void setWorkflowRunCompletedWithFailures(String runId, long completedAt) { + NativeWorkflows.setWorkflowRunCompletedWithFailures(handle, runId, completedAt); + } + + @Override + public void setWorkflowNodeCompensationJob( + String runId, String nodeName, String compensationJobId, long startedAt) { + NativeWorkflows.setWorkflowNodeCompensationJob(handle, runId, nodeName, compensationJobId, startedAt); + } + + @Override + public void setWorkflowNodeCompensated(String runId, String nodeName, long completedAt) { + NativeWorkflows.setWorkflowNodeCompensated(handle, runId, nodeName, completedAt); + } + + @Override + public void setWorkflowNodeCompensationFailed(String runId, String nodeName, String error, long completedAt) { + NativeWorkflows.setWorkflowNodeCompensationFailed(handle, runId, nodeName, error, completedAt); + } + @Override public WorkerControl startWorker(WorkerBridge bridge, String optionsJson) { return new JniWorkerControl(NativeQueue.runWorker(handle, bridge, optionsJson)); 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 05272562..f79f4869 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 @@ -103,4 +103,23 @@ public static native void resolveWorkflowGate( /** Mark a node skipped (its condition evaluated false) and cancel any bound job. */ public static native void skipWorkflowNode(long handle, String runId, String nodeName); + + // ── Saga compensation (driven by the worker-side tracker) ── + + public static native void setWorkflowRunCompensating(long handle, String runId); + + public static native void setWorkflowRunCompensated(long handle, String runId, long completedAt); + + public static native void setWorkflowRunCompensationFailed( + long handle, String runId, long completedAt, String error); + + public static native void setWorkflowRunCompletedWithFailures(long handle, String runId, long completedAt); + + public static native void setWorkflowNodeCompensationJob( + long handle, String runId, String nodeName, String compensationJobId, long startedAt); + + public static native void setWorkflowNodeCompensated(long handle, String runId, String nodeName, long completedAt); + + public static native void setWorkflowNodeCompensationFailed( + long handle, String runId, String nodeName, String error, long completedAt); } 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 18f9e577..b6b1ef8f 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 @@ -239,6 +239,37 @@ default void skipWorkflowNode(String runId, String nodeName) { throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); } + // ── Saga compensation ───────────────────────────────────────── + + default void setWorkflowRunCompensating(String runId) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + default void setWorkflowRunCompensated(String runId, long completedAt) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + default void setWorkflowRunCompensationFailed(String runId, long completedAt, String error) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + default void setWorkflowRunCompletedWithFailures(String runId, long completedAt) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + default void setWorkflowNodeCompensationJob( + String runId, String nodeName, String compensationJobId, long startedAt) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + default void setWorkflowNodeCompensated(String runId, String nodeName, long completedAt) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + default void setWorkflowNodeCompensationFailed(String runId, String nodeName, String error, long completedAt) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + // ── Worker ────────────────────────────────────────────────────── /** Start a worker that dispatches jobs to {@code bridge}; returns its control. */ WorkerControl startWorker(WorkerBridge bridge, String optionsJson); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java index a0e57a5d..5206b622 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java @@ -21,6 +21,7 @@ final class PlanNode { final String gate; final String condition; final String subWorkflow; + final String compensate; @JsonCreator PlanNode( @@ -35,7 +36,8 @@ final class PlanNode { @JsonProperty("fanIn") String fanIn, @JsonProperty("gate") String gate, @JsonProperty("condition") String condition, - @JsonProperty("subWorkflow") String subWorkflow) { + @JsonProperty("subWorkflow") String subWorkflow, + @JsonProperty("compensate") String compensate) { this.name = name; this.predecessors = predecessors == null ? Collections.emptyList() : predecessors; this.taskName = taskName; @@ -48,6 +50,7 @@ final class PlanNode { this.gate = gate; this.condition = condition; this.subWorkflow = subWorkflow; + this.compensate = compensate; } /** Whether this node is a fan-out/fan-in node (its job is created by the fan-out machinery). */ diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java index b1d7fbc6..291e2d4b 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java @@ -33,6 +33,16 @@ static RunPlan from(List nodes) { return new RunPlan(byName, successors); } + /** The plan node by name, or {@code null} if absent. */ + PlanNode node(String name) { + return byName.get(name); + } + + /** Every plan node. */ + java.util.Collection allNodes() { + return byName.values(); + } + /** Every successor of {@code node} whose plan entry matches {@code match}. */ List successorsMatching(String node, Predicate match) { List matches = new ArrayList<>(); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java new file mode 100644 index 00000000..a5025068 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java @@ -0,0 +1,218 @@ +package org.byteveda.taskito.workflows; + +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import org.byteveda.taskito.errors.WorkflowException; +import org.byteveda.taskito.spi.QueueBackend; +import org.byteveda.taskito.task.EnqueueOptions; + +/** + * Rolls back a failed workflow run. When a run with compensable steps fails, the + * tracker hands it here: completed steps are compensated in reverse-dependency + * order — each wave of independent rollbacks runs in parallel, the next wave only + * after the previous drains. A compensation job is enqueued with the step's + * forward result as its payload, an idempotency key (so a tracker restart is a + * no-op), and {@code compensation:true} metadata (so it never advances the + * forward run). The run ends {@code compensated}, or {@code compensation_failed} + * if a rollback itself fails. + */ +final class SagaOrchestrator { + private final QueueBackend backend; + private final WorkflowTracker tracker; + private final ObjectMapper json; + private final ConcurrentMap runs = new ConcurrentHashMap<>(); + private final ConcurrentMap compensationJobs = new ConcurrentHashMap<>(); + + SagaOrchestrator(QueueBackend backend, WorkflowTracker tracker, ObjectMapper json) { + this.backend = backend; + this.tracker = tracker; + this.json = json; + } + + boolean isCompensationJob(String jobId) { + return compensationJobs.containsKey(jobId); + } + + /** Begin compensating a failed run; returns false (caller finalizes normally) if nothing to roll back. */ + boolean startCompensation(String runId, RunPlan plan, Map statuses) { + Map targets = new HashMap<>(); + for (PlanNode node : plan.allNodes()) { + NodeSnapshot snap = statuses.get(node.name); + if (node.compensate == null || snap == null || !isCompensable(snap.status)) { + continue; + } + byte[] forward = snap.jobId == null + ? new byte[0] + : backend.getResult(snap.jobId).orElseGet(() -> new byte[0]); + targets.put( + node.name, + new CompTarget( + node.compensate, + forward, + node.queueOrDefault(), + node.retries(), + node.timeout(), + node.priorityOrDefault())); + } + if (targets.isEmpty()) { + return false; + } + backend.setWorkflowRunCompensating(runId); + SagaRun run = new SagaRun(reverseTopoWaves(plan, targets.keySet()), targets); + runs.put(runId, run); + dispatchNextWave(runId, run); + return true; + } + + /** Record a compensation job's outcome and advance (or finalize) the saga. */ + void onCompensationCompleted(String jobId, boolean succeeded, String error) { + String[] key = compensationJobs.remove(jobId); + if (key == null) { + return; + } + String runId = key[0]; + String node = key[1]; + SagaRun run = runs.get(runId); + if (run == null) { + return; + } + if (succeeded) { + backend.setWorkflowNodeCompensated(runId, node, now()); + } else { + backend.setWorkflowNodeCompensationFailed(runId, node, error, now()); + run.anyFailed = true; + } + run.inflight.remove(node); + if (run.inflight.isEmpty()) { + if (run.anyFailed) { + finalizeSaga(runId, run); // first failure stops further rollback + } else { + dispatchNextWave(runId, run); + } + } + } + + private void dispatchNextWave(String runId, SagaRun run) { + if (run.waveIndex >= run.waves.size()) { + finalizeSaga(runId, run); + return; + } + List wave = run.waves.get(run.waveIndex++); + run.inflight.addAll(wave); + for (String node : wave) { + String compJobId = enqueueCompensation(runId, node, run.targets.get(node)); + compensationJobs.put(compJobId, new String[] {runId, node}); + backend.setWorkflowNodeCompensationJob(runId, node, compJobId, now()); + } + } + + private String enqueueCompensation(String runId, String node, CompTarget target) { + Map metadata = new LinkedHashMap<>(); + metadata.put("compensation", true); + metadata.put("workflow_run_id", runId); + metadata.put("workflow_node_name", node); + try { + EnqueueOptions options = EnqueueOptions.builder() + .jobId("compensation:" + runId + ":" + node) + .metadata(json.writeValueAsString(metadata)) + .queue(target.queue) + .maxRetries(target.retries) + .timeoutMs(target.timeout) + .priority(target.priority) + .build(); + return backend.enqueue(target.task, target.payload, json.writeValueAsString(options)); + } catch (Exception e) { + throw new WorkflowException("failed to enqueue compensation for node '" + node + "'", e); + } + } + + private void finalizeSaga(String runId, SagaRun run) { + runs.remove(runId); + compensationJobs.values().removeIf(key -> key[0].equals(runId)); + if (run.anyFailed) { + backend.setWorkflowRunCompensationFailed(runId, now(), "saga compensation failed"); + } else { + backend.setWorkflowRunCompensated(runId, now()); + } + tracker.forget(runId); + } + + /** Compensable nodes grouped by topo depth, deepest first (roll back dependents before dependencies). */ + private List> reverseTopoWaves(RunPlan plan, Set compensable) { + Map depth = new HashMap<>(); + for (String node : compensable) { + computeDepth(plan, node, depth); + } + Map> byDepth = new TreeMap<>(Collections.reverseOrder()); + for (String node : compensable) { + byDepth.computeIfAbsent(depth.get(node), key -> new ArrayList<>()).add(node); + } + return new ArrayList<>(byDepth.values()); + } + + private int computeDepth(RunPlan plan, String node, Map memo) { + Integer cached = memo.get(node); + if (cached != null) { + return cached; + } + PlanNode planNode = plan.node(node); + int depth = 0; + if (planNode != null) { + for (String predecessor : planNode.predecessors) { + depth = Math.max(depth, computeDepth(plan, predecessor, memo) + 1); + } + } + memo.put(node, depth); + return depth; + } + + private static boolean isCompensable(NodeStatus status) { + return status == NodeStatus.COMPLETED || status == NodeStatus.CACHE_HIT; + } + + private static long now() { + return System.currentTimeMillis(); + } + + /** What to enqueue to roll back one node. */ + private static final class CompTarget { + final String task; + final byte[] payload; + final String queue; + final int retries; + final long timeout; + final int priority; + + CompTarget(String task, byte[] payload, String queue, int retries, long timeout, int priority) { + this.task = task; + this.payload = payload; + this.queue = queue; + this.retries = retries; + this.timeout = timeout; + this.priority = priority; + } + } + + /** Per-run compensation progress. */ + private static final class SagaRun { + final List> waves; + final Map targets; + final Set inflight = ConcurrentHashMap.newKeySet(); + int waveIndex; + volatile boolean anyFailed; + + SagaRun(List> waves, Map targets) { + this.waves = waves; + this.targets = targets; + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java index d86e09b4..67b9f325 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java @@ -22,6 +22,7 @@ public final class Step { public final String condition; public final Condition callableCondition; public final Workflow subWorkflow; + public final String compensate; private Step(Builder builder) { this.name = builder.name; @@ -38,6 +39,7 @@ private Step(Builder builder) { this.condition = builder.condition; this.callableCondition = builder.callableCondition; this.subWorkflow = builder.subWorkflow; + this.compensate = builder.compensate; } /** Begin a step bound to a typed task. */ @@ -71,6 +73,7 @@ public static final class Builder { private String condition; private Condition callableCondition; private Workflow subWorkflow; + private String compensate; private Builder(String name, String taskName, Object payload) { this.name = name; @@ -191,6 +194,21 @@ public Builder subWorkflow(Workflow child) { return this; } + /** + * Register a rollback task for this step. If the run later fails, the saga + * runs {@code compensateTask} (with this step's result as its payload) to + * compensate it, rolling back completed steps in reverse order. + */ + public Builder compensate(String compensateTask) { + this.compensate = compensateTask; + return this; + } + + /** Register a typed rollback task; see {@link #compensate(String)}. */ + public Builder compensate(Task compensateTask) { + return compensate(compensateTask.name()); + } + public Step build() { if (fanOut != null && fanIn != null) { throw new IllegalArgumentException("step '" + name + "' cannot be both fan-out and fan-in"); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java index eb730a00..637adea2 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java @@ -65,10 +65,13 @@ public final class WorkflowTracker { private final Set promotedNodes = ConcurrentHashMap.newKeySet(); /** Sub-workflow child run → its parent {@code [runId, nodeName]}, so a finalized child resolves its parent node. */ private final ConcurrentMap childToParent = new ConcurrentHashMap<>(); + /** Rolls back failed runs that declared compensators. */ + private final SagaOrchestrator saga; public WorkflowTracker(QueueBackend backend, Serializer serializer) { this.backend = backend; this.serializer = serializer; + this.saga = new SagaOrchestrator(backend, this, json); } /** @@ -102,6 +105,12 @@ public void onDead(OutcomeEvent event) { } private void onOutcome(String jobId, boolean succeeded, String error) { + // Compensation jobs carry compensation:true metadata, so they never map to + // a forward node — route them to the saga instead. + if (saga.isCompensationJob(jobId)) { + saga.onCompensationCompleted(jobId, succeeded, error); + return; + } NodeRef ref = backend.workflowNodeForJobJson(jobId) .map(raw -> decode(raw, NodeRef.class)) .orElse(null); @@ -442,6 +451,19 @@ private String workflowName(String runId) { } private void finalizeIfTerminal(String runId) { + Map statuses = statusMap(runId); + if (statuses.isEmpty() || !allNodesTerminal(statuses)) { + return; // not every node has settled yet + } + // A failed top-level run with compensators rolls back BEFORE the core marks + // it failed, so observers never see a transient Failed before Compensating. + boolean anyFailed = statuses.values().stream().anyMatch(node -> node.status == NodeStatus.FAILED); + if (anyFailed && !childToParent.containsKey(runId)) { + RunPlan plan = plans.computeIfAbsent(runId, this::loadPlan); + if (plan != null && saga.startCompensation(runId, plan, statuses)) { + return; // the saga owns the run and will finalize it + } + } Optional finalState = backend.finalizeRunIfTerminal(runId); if (finalState.isEmpty()) { return; @@ -469,7 +491,7 @@ private void resolveSubWorkflowParent(String parentRun, String parentNode, boole } /** Drop all per-run state once a run is terminal, cancelling any lingering timers. */ - private void forget(String runId) { + void forget(String runId) { plans.remove(runId); runNames.remove(runId); gateTimers.keySet().removeIf(key -> { @@ -560,6 +582,15 @@ private static boolean anyFailed(List predecessors, Map statuses) { + for (NodeSnapshot node : statuses.values()) { + if (!isTerminal(node.status)) { + return false; + } + } + return true; + } + private static boolean isTerminal(NodeStatus status) { switch (status) { case COMPLETED: diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSagaTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSagaTest.java new file mode 100644 index 00000000..5ae0aa90 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSagaTest.java @@ -0,0 +1,90 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.byteveda.taskito.workflows.NodeStatus; +import org.byteveda.taskito.workflows.Step; +import org.byteveda.taskito.workflows.Workflow; +import org.byteveda.taskito.workflows.WorkflowRun; +import org.byteveda.taskito.workflows.WorkflowState; +import org.byteveda.taskito.workflows.WorkflowStatus; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +class WorkflowSagaTest { + + private static final Task A = Task.of("sg.a", Integer.class); + private static final Task B = Task.of("sg.b", Integer.class); + private static final Task C = Task.of("sg.c", Integer.class); + private static final Task UNDO_A = Task.of("sg.undoA", Integer.class); + private static final Task UNDO_B = Task.of("sg.undoB", Integer.class); + + @Test + @Timeout(30) + void compensatesCompletedStepsOnFailure(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("sg.db").toString()).open()) { + Workflow wf = Workflow.named("saga") + .step(Step.of("a", A, 1).compensate(UNDO_A).build()) + .step(Step.of("b", B, 2).after("a").compensate(UNDO_B).build()) + .step(Step.of("c", C, 3).after("b").maxRetries(0).build()); + + WorkflowRun run = queue.submitWorkflow(wf); + Set undone = ConcurrentHashMap.newKeySet(); + try (Worker worker = queue.worker() + .handle(A, p -> p) + .handle(B, p -> p) + .handle(C, p -> { + throw new IllegalStateException("boom"); + }) + .handle(UNDO_A, p -> { + undone.add(100 + p); + return p; + }) + .handle(UNDO_B, p -> { + undone.add(200 + p); + return p; + }) + .trackWorkflows() + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.COMPENSATED, status.state); + assertEquals(NodeStatus.COMPENSATED, status.node("a").orElseThrow().status); + assertEquals(NodeStatus.COMPENSATED, status.node("b").orElseThrow().status); + assertEquals(NodeStatus.FAILED, status.node("c").orElseThrow().status); + // undoA saw a's result (1) → 101; undoB saw b's result (2) → 202. + assertEquals(Set.of(101, 202), undone); + } + } + } + + @Test + @Timeout(30) + void noCompensatorsStillFailsNormally(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("sg2.db").toString()).open()) { + Workflow wf = Workflow.named("plain") + .step("a", A, 1) + .step(Step.of("b", B, 2).after("a").maxRetries(0).build()); + + WorkflowRun run = queue.submitWorkflow(wf); + try (Worker worker = queue.worker() + .handle(A, p -> p) + .handle(B, p -> { + throw new IllegalStateException("boom"); + }) + .trackWorkflows() + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.FAILED, status.state); + } + } + } +} From cc9a9876a53aab491756b81616571037e4a5a443 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 30 Jun 2026 10:10:17 +0530 Subject: [PATCH 2/7] fix(java): reject compensators on non-task workflow steps --- .../src/main/java/org/byteveda/taskito/workflows/Step.java | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java index 67b9f325..fdcc4f5a 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java @@ -227,6 +227,13 @@ public Step build() { throw new IllegalArgumentException( "step '" + name + "' cannot be both a sub-workflow and a gate/fan-out/fan-in"); } + // Compensation replays the step's forward result as the rollback payload. + // Gate, sub-workflow, and fan-out nodes never run a task of their own + // (so they have no forward result) — reject a compensator on them. + if (compensate != null && (gate != null || subWorkflow != null || fanOut != null)) { + throw new IllegalArgumentException("step '" + name + + "' cannot be compensated: a gate/sub-workflow/fan-out node has no forward result"); + } return new Step(this); } } From f894c51f6057e347e934118c0a9089d7b6830461 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 30 Jun 2026 10:10:17 +0530 Subject: [PATCH 3/7] fix(java): reject compensators in child workflows --- .../src/main/java/org/byteveda/taskito/DefaultTaskito.java | 6 ++++++ 1 file changed, 6 insertions(+) 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 82299187..e7844cf5 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -505,6 +505,12 @@ private String encodeChild(Workflow child) { throw new WorkflowException("child workflow step '" + step.name + "' has no payload; a sub-workflow cannot supply child step payloads at submit"); } + // The saga only compensates top-level runs; a child run resolves its parent + // node without rolling back, so a child compensator would never fire. + if (step.compensate != null) { + throw new WorkflowException("child workflow step '" + step.name + + "' declares a compensator; sub-workflow runs cannot compensate yet"); + } specs.add(stepSpec(step)); if (step.payload != null) { payloads.put(step.name, Base64.getEncoder().encodeToString(serializer.serialize(step.payload))); From a811be9be3020149e51fbbdd0361ab893858a51d Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 30 Jun 2026 10:10:17 +0530 Subject: [PATCH 4/7] test(java): assert saga rollback order and single run --- .../java/org/byteveda/taskito/WorkflowSagaTest.java | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSagaTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSagaTest.java index 5ae0aa90..26d61bb8 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSagaTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSagaTest.java @@ -4,8 +4,9 @@ import java.nio.file.Path; import java.time.Duration; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; import org.byteveda.taskito.task.Task; import org.byteveda.taskito.worker.Worker; import org.byteveda.taskito.workflows.NodeStatus; @@ -37,7 +38,7 @@ void compensatesCompletedStepsOnFailure(@TempDir Path dir) throws Exception { .step(Step.of("c", C, 3).after("b").maxRetries(0).build()); WorkflowRun run = queue.submitWorkflow(wf); - Set undone = ConcurrentHashMap.newKeySet(); + Queue undone = new ConcurrentLinkedQueue<>(); try (Worker worker = queue.worker() .handle(A, p -> p) .handle(B, p -> p) @@ -59,8 +60,9 @@ void compensatesCompletedStepsOnFailure(@TempDir Path dir) throws Exception { assertEquals(NodeStatus.COMPENSATED, status.node("a").orElseThrow().status); assertEquals(NodeStatus.COMPENSATED, status.node("b").orElseThrow().status); assertEquals(NodeStatus.FAILED, status.node("c").orElseThrow().status); - // undoA saw a's result (1) → 101; undoB saw b's result (2) → 202. - assertEquals(Set.of(101, 202), undone); + // Reverse-dependency order, each exactly once: b rolls back before a + // (undoB saw b's result 2 → 202; undoA saw a's result 1 → 101). + assertEquals(List.of(202, 101), List.copyOf(undone)); } } } From 66e6d68305a5188d6feb6ed759f127f70e7dbc6f Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 30 Jun 2026 10:11:21 +0530 Subject: [PATCH 5/7] fix(java): compensate only executed nodes with a result --- .../taskito/workflows/SagaOrchestrator.java | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java index a5025068..d5df596e 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java @@ -50,9 +50,13 @@ boolean startCompensation(String runId, RunPlan plan, Map if (node.compensate == null || snap == null || !isCompensable(snap.status)) { continue; } - byte[] forward = snap.jobId == null - ? new byte[0] - : backend.getResult(snap.jobId).orElseGet(() -> new byte[0]); + if (snap.jobId == null) { + throw new WorkflowException( + "completed workflow node '" + node.name + "' has no forward job to compensate from"); + } + byte[] forward = backend.getResult(snap.jobId) + .orElseThrow(() -> + new WorkflowException("missing forward result for workflow node '" + node.name + "'")); targets.put( node.name, new CompTarget( @@ -176,7 +180,9 @@ private int computeDepth(RunPlan plan, String node, Map memo) { } private static boolean isCompensable(NodeStatus status) { - return status == NodeStatus.COMPLETED || status == NodeStatus.CACHE_HIT; + // Only nodes that actually executed their forward task this run. A CACHE_HIT + // skipped the task, so its side effects were never performed here. + return status == NodeStatus.COMPLETED; } private static long now() { From c310bfe73d70547ef3f783ffa218ad02af8e791a Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 30 Jun 2026 10:12:11 +0530 Subject: [PATCH 6/7] fix(java): make saga state transitions concurrency-safe --- .../taskito/workflows/SagaOrchestrator.java | 47 +++++++++++++------ 1 file changed, 33 insertions(+), 14 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java index d5df596e..523107e4 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java @@ -70,10 +70,22 @@ boolean startCompensation(String runId, RunPlan plan, Map if (targets.isEmpty()) { return false; } - backend.setWorkflowRunCompensating(runId); SagaRun run = new SagaRun(reverseTopoWaves(plan, targets.keySet()), targets); - runs.put(runId, run); - dispatchNextWave(runId, run); + // Concurrent terminal outcomes can both reach here for the same run; only the + // first owns it, so the run is marked compensating and dispatched exactly once. + if (runs.putIfAbsent(runId, run) != null) { + return true; + } + synchronized (run) { + backend.setWorkflowRunCompensating(runId); + try { + dispatchNextWave(runId, run); + } catch (RuntimeException e) { + run.anyFailed = true; + finalizeSaga(runId, run); // never leave the run stuck in COMPENSATING + throw e; + } + } return true; } @@ -89,22 +101,27 @@ void onCompensationCompleted(String jobId, boolean succeeded, String error) { if (run == null) { return; } - if (succeeded) { - backend.setWorkflowNodeCompensated(runId, node, now()); - } else { - backend.setWorkflowNodeCompensationFailed(runId, node, error, now()); - run.anyFailed = true; - } - run.inflight.remove(node); - if (run.inflight.isEmpty()) { - if (run.anyFailed) { - finalizeSaga(runId, run); // first failure stops further rollback + // Serialize wave advancement: two outcomes in one wave must not both observe an + // empty in-flight set and double-dispatch (or skip) the next wave. + synchronized (run) { + if (succeeded) { + backend.setWorkflowNodeCompensated(runId, node, now()); } else { - dispatchNextWave(runId, run); + backend.setWorkflowNodeCompensationFailed(runId, node, error, now()); + run.anyFailed = true; + } + run.inflight.remove(node); + if (run.inflight.isEmpty()) { + if (run.anyFailed) { + finalizeSaga(runId, run); // first failure stops further rollback + } else { + dispatchNextWave(runId, run); + } } } } + /** Enqueue the next wave's compensation jobs. The caller must hold {@code run}'s monitor. */ private void dispatchNextWave(String runId, SagaRun run) { if (run.waveIndex >= run.waves.size()) { finalizeSaga(runId, run); @@ -113,6 +130,8 @@ private void dispatchNextWave(String runId, SagaRun run) { List wave = run.waves.get(run.waveIndex++); run.inflight.addAll(wave); for (String node : wave) { + // The runtime job id (the saga's routing key) is the value enqueue returns, + // not the idempotency key we pass in — so register routing once we have it. String compJobId = enqueueCompensation(runId, node, run.targets.get(node)); compensationJobs.put(compJobId, new String[] {runId, node}); backend.setWorkflowNodeCompensationJob(runId, node, compJobId, now()); From 71d0e1b7c9466059f31d10c71aa07b0c549fa820 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 30 Jun 2026 10:27:09 +0530 Subject: [PATCH 7/7] fix(java): finalize saga if a later wave fails to dispatch --- .../taskito/workflows/SagaOrchestrator.java | 21 ++++++++++++------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java index 523107e4..19968a98 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java @@ -78,13 +78,7 @@ boolean startCompensation(String runId, RunPlan plan, Map } synchronized (run) { backend.setWorkflowRunCompensating(runId); - try { - dispatchNextWave(runId, run); - } catch (RuntimeException e) { - run.anyFailed = true; - finalizeSaga(runId, run); // never leave the run stuck in COMPENSATING - throw e; - } + dispatchOrFail(runId, run); } return true; } @@ -115,12 +109,23 @@ void onCompensationCompleted(String jobId, boolean succeeded, String error) { if (run.anyFailed) { finalizeSaga(runId, run); // first failure stops further rollback } else { - dispatchNextWave(runId, run); + dispatchOrFail(runId, run); } } } } + /** Dispatch the next wave, finalizing the run instead of stranding it in COMPENSATING if enqueue fails. */ + private void dispatchOrFail(String runId, SagaRun run) { + try { + dispatchNextWave(runId, run); + } catch (RuntimeException e) { + run.anyFailed = true; + finalizeSaga(runId, run); + throw e; + } + } + /** Enqueue the next wave's compensation jobs. The caller must hold {@code run}'s monitor. */ private void dispatchNextWave(String runId, SagaRun run) { if (run.waveIndex >= run.waves.size()) {