feat(java): saga compensation for workflows

crates/taskito-java/src/workflows/mod.rs

@@ -66,6 +66,8 @@ struct StepSpec {
     gate: Option<String>,
     /// Serialized child-workflow spec marking a sub-workflow node.
     sub_workflow: Option<String>,
+    /// Rollback task name — the saga runs it to compensate this node.
+    compensate: Option<String>,
 }
 
 /// 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<String>) -> WorkflowNode {
     WorkflowNode {

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

@@ -475,6 +475,9 @@ private Map<String, Object> 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)));

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));

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);
 }

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);

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). */

sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java

@@ -33,6 +33,16 @@ static RunPlan from(List<PlanNode> 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<PlanNode> allNodes() {
+        return byName.values();
+    }
+
     /** Every successor of {@code node} whose plan entry matches {@code match}. */
     List<PlanNode> successorsMatching(String node, Predicate<PlanNode> match) {
         List<PlanNode> matches = new ArrayList<>();