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..e7844cf5 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; } @@ -502,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))); 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..19968a98 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java @@ -0,0 +1,248 @@ +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; + } + 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( + node.compensate, + forward, + node.queueOrDefault(), + node.retries(), + node.timeout(), + node.priorityOrDefault())); + } + if (targets.isEmpty()) { + return false; + } + SagaRun run = new SagaRun(reverseTopoWaves(plan, targets.keySet()), targets); + // 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); + dispatchOrFail(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; + } + // 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 { + 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 { + 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()) { + finalizeSaga(runId, run); + return; + } + 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()); + } + } + + 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) { + // 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() { + 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..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 @@ -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"); @@ -209,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); } } 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..26d61bb8 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSagaTest.java @@ -0,0 +1,92 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.nio.file.Path; +import java.time.Duration; +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; +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); + Queue undone = new ConcurrentLinkedQueue<>(); + 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); + // 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)); + } + } + } + + @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); + } + } + } +}