feat(java): incremental workflow caching

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

@@ -68,6 +68,8 @@ struct StepSpec {
     sub_workflow: Option<String>,
     /// Rollback task name — the saga runs it to compensate this node.
     compensate: Option<String>,
+    /// JSON `{ttlMs}` marking a cacheable node whose result the tracker may reuse.
+    cache: Option<String>,
 }
 
 /// Combined run + node snapshot returned by `getWorkflowStatus`. Timestamps are
@@ -233,6 +235,7 @@ fn submit(
                     gate: s.gate.clone(),
                     sub_workflow: s.sub_workflow.clone(),
                     compensate: s.compensate.clone(),
+                    cache: s.cache.clone(),
                     ..Default::default()
                 },
             )
@@ -522,6 +525,7 @@ struct PlanNodeView<'a> {
     gate: Option<&'a str>,
     sub_workflow: Option<&'a str>,
     compensate: Option<&'a str>,
+    cache: Option<&'a str>,
 }
 
 /// The run + node a job belongs to.
@@ -583,6 +587,7 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_getWor
                     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()),
+                    cache: meta.and_then(|m| m.cache.as_deref()),
                 }
             })
             .collect();
@@ -1063,6 +1068,31 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_skipWo
     })
 }
 
+/// `void setWorkflowNodeCacheHit(long, String runId, String nodeName)` — mark a
+/// node as a cache hit (terminal, treated as completed) without running it.
+#[no_mangle]
+pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_setWorkflowNodeCacheHit<
+    'local,
+>(
+    mut env: JNIEnv<'local>,
+    _class: JClass<'local>,
+    handle: jlong,
+    run_id: JString<'local>,
+    node_name: JString<'local>,
+) {
+    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()?.update_workflow_node_status(
+            &run_id,
+            &node_name,
+            WorkflowNodeStatus::CacheHit,
+        )?;
+        Ok(())
+    })
+}
+
 /// `void setWorkflowRunCompensating(long, String runId)` — mark a run as
 /// compensating (the saga is rolling back completed nodes).
 #[no_mangle]

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

@@ -365,6 +365,7 @@ public WorkflowRun submitWorkflow(Workflow workflow) {
     @Override
     public WorkflowRun submitWorkflow(Workflow workflow, Map<String, Object> suppliedPayloads) {
         List<Step> steps = workflow.steps();
+        rejectCachedFanProducers(steps);
         Set<String> deferred = deferredNodes(steps);
         List<Map<String, Object>> specs = new ArrayList<>(steps.size());
         // Deferred nodes (fan-out/fan-in + their downstream) have no job — and so
@@ -405,6 +406,33 @@ public WorkflowRun submitWorkflow(Workflow workflow, Map<String, Object> supplie
      * downstream of them — are deferred: they carry a node row but no job at
      * submit, and the worker tracker creates/parks/evaluates them at runtime.
      */
+    /**
+     * A cache hit produces no forward result, so a fan-out/fan-in over a cached step
+     * would have nothing to expand or collect (the run would hang). Reject it at submit.
+     */
+    private static void rejectCachedFanProducers(List<Step> steps) {
+        Set<String> cacheable = new HashSet<>();
+        for (Step step : steps) {
+            if (step.cacheTtlMs != null) {
+                cacheable.add(step.name);
+            }
+        }
+        if (cacheable.isEmpty()) {
+            return;
+        }
+        for (Step step : steps) {
+            if (step.fanOut == null && step.fanIn == null) {
+                continue;
+            }
+            for (String predecessor : step.after) {
+                if (cacheable.contains(predecessor)) {
+                    throw new WorkflowException("step '" + step.name + "' fans over cached step '" + predecessor
+                            + "', but a cache hit yields no result to fan — don't cache a fan-out/fan-in producer");
+                }
+            }
+        }
+    }
+
     private static Set<String> deferredNodes(List<Step> steps) {
         Set<String> deferred = new HashSet<>();
         for (Step step : steps) {
@@ -415,6 +443,7 @@ private static Set<String> deferredNodes(List<Step> steps) {
                     || step.fanIn != null
                     || step.gate != null
                     || step.subWorkflow != null
+                    || step.cacheTtlMs != null
                     || runtimeCondition) {
                 deferred.add(step.name);
             }
@@ -478,6 +507,9 @@ private Map<String, Object> stepSpec(Step step) {
         if (step.compensate != null) {
             spec.put("compensate", step.compensate);
         }
+        if (step.cacheTtlMs != null) {
+            spec.put("cache", encode(Map.of("ttlMs", step.cacheTtlMs)));
+        }
         return spec;
     }
 

sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java

@@ -340,6 +340,11 @@ public void skipWorkflowNode(String runId, String nodeName) {
         NativeWorkflows.skipWorkflowNode(handle, runId, nodeName);
     }
 
+    @Override
+    public void setWorkflowNodeCacheHit(String runId, String nodeName) {
+        NativeWorkflows.setWorkflowNodeCacheHit(handle, runId, nodeName);
+    }
+
     @Override
     public void setWorkflowRunCompensating(String runId) {
         NativeWorkflows.setWorkflowRunCompensating(handle, runId);

sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.java

@@ -104,6 +104,9 @@ 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);
 
+    /** Mark a node as a cache hit (terminal, treated as completed) without running it. */
+    public static native void setWorkflowNodeCacheHit(long handle, String runId, String nodeName);
+
     // ── Saga compensation (driven by the worker-side tracker) ──
 
     public static native void setWorkflowRunCompensating(long handle, String runId);

sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java

@@ -239,6 +239,11 @@ default void skipWorkflowNode(String runId, String nodeName) {
         throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
     }
 
+    /** Mark a node as a cache hit (terminal) without running it. */
+    default void setWorkflowNodeCacheHit(String runId, String nodeName) {
+        throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
+    }
+
     // ── Saga compensation ─────────────────────────────────────────
 
     default void setWorkflowRunCompensating(String runId) {

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

@@ -22,6 +22,7 @@ final class PlanNode {
     final String condition;
     final String subWorkflow;
     final String compensate;
+    final String cache;
 
     @JsonCreator
     PlanNode(
@@ -37,7 +38,8 @@ final class PlanNode {
             @JsonProperty("gate") String gate,
             @JsonProperty("condition") String condition,
             @JsonProperty("subWorkflow") String subWorkflow,
-            @JsonProperty("compensate") String compensate) {
+            @JsonProperty("compensate") String compensate,
+            @JsonProperty("cache") String cache) {
         this.name = name;
         this.predecessors = predecessors == null ? Collections.emptyList() : predecessors;
         this.taskName = taskName;
@@ -51,6 +53,7 @@ final class PlanNode {
         this.condition = condition;
         this.subWorkflow = subWorkflow;
         this.compensate = compensate;
+        this.cache = cache;
     }
 
     /** 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/Step.java

@@ -23,6 +23,7 @@ public final class Step {
     public final Condition callableCondition;
     public final Workflow subWorkflow;
     public final String compensate;
+    public final Long cacheTtlMs;
 
     private Step(Builder builder) {
         this.name = builder.name;
@@ -40,6 +41,7 @@ private Step(Builder builder) {
         this.callableCondition = builder.callableCondition;
         this.subWorkflow = builder.subWorkflow;
         this.compensate = builder.compensate;
+        this.cacheTtlMs = builder.cacheTtlMs;
     }
 
     /** Begin a step bound to a typed task. */
@@ -74,6 +76,7 @@ public static final class Builder {
         private Condition callableCondition;
         private Workflow subWorkflow;
         private String compensate;
+        private Long cacheTtlMs;
 
         private Builder(String name, String taskName, Object payload) {
             this.name = name;
@@ -209,7 +212,31 @@ public Builder compensate(Task<?> compensateTask) {
             return compensate(compensateTask.name());
         }
 
+        /**
+         * Cache this step's execution for {@code ttl}: on a later run of the same
+         * workflow, if this step's task + payload are unchanged and within the TTL,
+         * the worker marks it a cache hit and skips re-running it. (Cache state is
+         * per worker process.)
+         *
+         * <p>A cache hit does not produce a forward result, so a cached step's result
+         * is not available downstream: it cannot feed a fan-out/fan-in (rejected at
+         * submit) and a callable {@code condition} won't see its result.
+         */
+        public Builder cache(java.time.Duration ttl) {
+            if (ttl == null || ttl.isNegative() || ttl.isZero()) {
+                throw new IllegalArgumentException("cache ttl must be positive");
+            }
+            this.cacheTtlMs = ttl.toMillis();
+            return this;
+        }
+
         public Step build() {
+            // A cacheable step is a deferred node; deferred roots are never promoted,
+            // so a cacheable step with no predecessor would wedge the run in PENDING.
+            if (cacheTtlMs != null && after.isEmpty()) {
+                throw new IllegalArgumentException("step '" + name
+                        + "' cannot be cacheable without a predecessor; a deferred root is never promoted");
+            }
             if (fanOut != null && fanIn != null) {
                 throw new IllegalArgumentException("step '" + name + "' cannot be both fan-out and fan-in");
             }

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

@@ -5,9 +5,11 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import java.security.MessageDigest;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.HexFormat;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -67,6 +69,8 @@ public final class WorkflowTracker {
     private final ConcurrentMap<String, String[]> childToParent = new ConcurrentHashMap<>();
     /** Rolls back failed runs that declared compensators. */
     private final SagaOrchestrator saga;
+    /** Incremental cache: {@code cacheKey -> expiry millis}, so a re-run of a cacheable step is a cache hit. */
+    private final ConcurrentMap<String, Long> resultCache = new ConcurrentHashMap<>();
 
     public WorkflowTracker(QueueBackend backend, Serializer serializer) {
         this.backend = backend;
@@ -121,6 +125,14 @@ private void onOutcome(String jobId, boolean succeeded, String error) {
         backend.markWorkflowNodeResult(jobId, succeeded, error, true);
         RunPlan plan = plans.computeIfAbsent(ref.runId, this::loadPlan);
 
+        // Cache a cacheable step's success so a later run of the same workflow can skip it.
+        if (succeeded && plan != null) {
+            PlanNode settled = plan.node(ref.nodeName);
+            if (settled != null && settled.cache != null) {
+                cacheResult(ref.runId, settled);
+            }
+        }
+
         if (ref.nodeName.indexOf('[') >= 0) {
             handleFanOutChild(ref.runId, ref.nodeName, plan);
             return;
@@ -273,13 +285,55 @@ private void promoteOrSkip(String runId, PlanNode node, Map<String, NodeSnapshot
                 submitSubWorkflow(runId, node);
                 return;
             }
+            if (node.cache != null && reuseFromCache(runId, node, plan)) {
+                return; // cache hit — skipped re-running this step
+            }
             createDeferredJobFor(runId, node);
         } catch (RuntimeException e) {
             promotedNodes.remove(promotionKey);
             throw e;
         }
     }
 
+    /** If a prior run cached this step's result within its TTL, mark it a cache hit instead of running it. */
+    private boolean reuseFromCache(String runId, PlanNode node, RunPlan plan) {
+        String key = cacheKey(runId, node.name);
+        Long expiry = resultCache.get(key);
+        if (expiry == null) {
+            return false;
+        }
+        if (expiry <= System.currentTimeMillis()) {
+            resultCache.remove(key, expiry); // drop the stale entry rather than let it linger
+            return false;
+        }
+        backend.setWorkflowNodeCacheHit(runId, node.name);
+        promoteSuccessors(runId, node.name, plan); // CacheHit is terminal — advance successors
+        return true;
+    }
+
+    private void cacheResult(String runId, PlanNode node) {
+        CacheMeta meta = decode(node.cache, CacheMeta.class);
+        if (meta != null && meta.ttlMs != null) {
+            long now = System.currentTimeMillis();
+            resultCache.values().removeIf(expiry -> expiry <= now); // sweep expired keys before adding
+            resultCache.put(cacheKey(runId, node.name), now + meta.ttlMs);
+        }
+    }
+
+    /** Stable across runs: workflow name + node + a hash of the node's payload. */
+    private String cacheKey(String runId, String nodeName) {
+        byte[] payload = deferredPayload(runId, nodeName);
+        return workflowName(runId) + ":" + nodeName + ":" + sha256(payload == null ? new byte[0] : payload);
+    }
+
+    private static String sha256(byte[] data) {
+        try {
+            return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(data));
+        } catch (Exception e) {
+            throw new WorkflowException("failed to hash payload for the workflow cache", e);
+        }
+    }
+
     /** Submit a sub-workflow node's child run and mark the node running until the child finalizes. */
     private void submitSubWorkflow(String runId, PlanNode node) {
         ChildSpec child = decode(node.subWorkflow, ChildSpec.class);
@@ -663,6 +717,17 @@ public int hashCode() {
         }
     }
 
+    /** {@code {ttlMs}} parsed from a node's cache metadata. */
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    private static final class CacheMeta {
+        final Long ttlMs;
+
+        @JsonCreator
+        CacheMeta(@JsonProperty("ttlMs") Long ttlMs) {
+            this.ttlMs = ttlMs;
+        }
+    }
+
     /** {@code {timeoutMs, onTimeout, message}} parsed from a node's gate metadata. */
     @JsonIgnoreProperties(ignoreUnknown = true)
     private static final class GateMeta {