From 3995fade6a9df9982cbdb1d2fd9c7d51108425fc Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sun, 28 Jun 2026 21:39:42 +0530 Subject: [PATCH 1/4] feat(java): incremental workflow caching Step.cache(ttl) reuses a step's result across runs of the same workflow: when its task + payload are unchanged and within the TTL, the tracker marks it a cache hit and skips re-running it. Adds a setWorkflowNodeCacheHit JNI fn + cache metadata; cache state is per worker process. --- crates/taskito-java/src/workflows/mod.rs | 30 ++++++++++ .../org/byteveda/taskito/DefaultTaskito.java | 4 ++ .../taskito/internal/JniQueueBackend.java | 5 ++ .../taskito/internal/NativeWorkflows.java | 3 + .../byteveda/taskito/spi/QueueBackend.java | 5 ++ .../byteveda/taskito/workflows/PlanNode.java | 5 +- .../org/byteveda/taskito/workflows/Step.java | 17 ++++++ .../taskito/workflows/WorkflowTracker.java | 58 +++++++++++++++++++ 8 files changed, 126 insertions(+), 1 deletion(-) diff --git a/crates/taskito-java/src/workflows/mod.rs b/crates/taskito-java/src/workflows/mod.rs index f9f7d4f1..51bdaff6 100644 --- a/crates/taskito-java/src/workflows/mod.rs +++ b/crates/taskito-java/src/workflows/mod.rs @@ -68,6 +68,8 @@ struct StepSpec { sub_workflow: Option, /// Rollback task name — the saga runs it to compensate this node. compensate: Option, + /// JSON `{ttlMs}` marking a cacheable node whose result the tracker may reuse. + cache: Option, } /// 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] 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 e7844cf5..ca6a2bcc 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -415,6 +415,7 @@ private static Set deferredNodes(List steps) { || step.fanIn != null || step.gate != null || step.subWorkflow != null + || step.cacheTtlMs != null || runtimeCondition) { deferred.add(step.name); } @@ -478,6 +479,9 @@ private Map 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; } 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 c7e42083..d97ffacc 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,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); 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 f79f4869..e28dfc00 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 @@ -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); 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 b6b1ef8f..1313827e 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,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) { 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 5206b622..270831ea 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 @@ -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). */ 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 fdcc4f5a..730985cb 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 @@ -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,6 +212,20 @@ 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.) + */ + 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() { 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 637adea2..6867c852 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 @@ -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 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 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,6 +285,9 @@ private void promoteOrSkip(String runId, PlanNode node, Map Date: Sun, 28 Jun 2026 21:39:43 +0530 Subject: [PATCH 2/4] test(java): cover incremental workflow caching A re-run of a cacheable step is a cache hit, not re-executed. --- .../byteveda/taskito/WorkflowCacheTest.java | 62 +++++++++++++++++++ 1 file changed, 62 insertions(+) create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/WorkflowCacheTest.java diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowCacheTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowCacheTest.java new file mode 100644 index 00000000..aecd6899 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowCacheTest.java @@ -0,0 +1,62 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicInteger; +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 WorkflowCacheTest { + + private static final Task SEED = Task.of("wc.seed", Integer.class); + private static final Task COMPUTE = Task.of("wc.compute", Integer.class); + private static final Task AFTER = Task.of("wc.after", Integer.class); + + @Test + @Timeout(40) + void cachedStepIsReusedOnRerun(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("wc.db").toString()).open()) { + // compute is cacheable; it has a predecessor (a deferred root is never promoted). + Workflow wf = Workflow.named("cached") + .step("seed", SEED, 0) + .step(Step.of("compute", COMPUTE, 7) + .cache(Duration.ofMinutes(5)) + .after("seed") + .build()) + .step("after", AFTER, 1, "compute"); + + AtomicInteger computeRuns = new AtomicInteger(); + try (Worker worker = queue.worker() + .handle(SEED, p -> p) + .handle(COMPUTE, p -> computeRuns.incrementAndGet()) + .handle(AFTER, p -> p) + .trackWorkflows(wf) + .start()) { + // First run executes compute. + WorkflowStatus first = queue.submitWorkflow(wf).await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.COMPLETED, first.state); + assertEquals(NodeStatus.COMPLETED, first.node("compute").orElseThrow().status); + + // Second run reuses it: compute is a cache hit, not re-executed. + WorkflowRun run2 = queue.submitWorkflow(wf); + WorkflowStatus second = run2.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.COMPLETED, second.state); + assertEquals(NodeStatus.CACHE_HIT, second.node("compute").orElseThrow().status); + assertEquals(NodeStatus.COMPLETED, second.node("after").orElseThrow().status); + assertEquals(1, computeRuns.get()); + } + } + } +} From f8337622232b51d9993696c7fdcea3ae6f074741 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 30 Jun 2026 12:01:23 +0530 Subject: [PATCH 3/4] fix(java): evict expired workflow cache entries --- .../byteveda/taskito/workflows/WorkflowTracker.java | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) 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 6867c852..e819dbc3 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 @@ -297,8 +297,13 @@ private void promoteOrSkip(String runId, PlanNode node, Map expiry <= now); // sweep expired keys before adding + resultCache.put(cacheKey(runId, node.name), now + meta.ttlMs); } } From b28eae03d0752ffd6653dca910d63a1940a47476 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Tue, 30 Jun 2026 12:01:23 +0530 Subject: [PATCH 4/4] fix(java): reject cacheable steps that would never run --- .../org/byteveda/taskito/DefaultTaskito.java | 28 +++++++++++++++++ .../org/byteveda/taskito/workflows/Step.java | 10 +++++++ .../byteveda/taskito/WorkflowCacheTest.java | 30 +++++++++++++++++++ 3 files changed, 68 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 ca6a2bcc..b459e73f 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/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 suppliedPayloads) { List steps = workflow.steps(); + rejectCachedFanProducers(steps); Set deferred = deferredNodes(steps); List> 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 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 steps) { + Set 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 deferredNodes(List steps) { Set deferred = new HashSet<>(); for (Step step : steps) { 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 730985cb..8c6b9e68 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 @@ -217,6 +217,10 @@ public Builder compensate(Task compensateTask) { * 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.) + * + *

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()) { @@ -227,6 +231,12 @@ public Builder cache(java.time.Duration ttl) { } 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"); } diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowCacheTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowCacheTest.java index aecd6899..49a42ca5 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowCacheTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowCacheTest.java @@ -1,12 +1,15 @@ package org.byteveda.taskito; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import java.nio.file.Path; import java.time.Duration; import java.util.concurrent.atomic.AtomicInteger; +import org.byteveda.taskito.errors.WorkflowException; import org.byteveda.taskito.task.Task; import org.byteveda.taskito.worker.Worker; +import org.byteveda.taskito.workflows.FanMode; import org.byteveda.taskito.workflows.NodeStatus; import org.byteveda.taskito.workflows.Step; import org.byteveda.taskito.workflows.Workflow; @@ -59,4 +62,31 @@ void cachedStepIsReusedOnRerun(@TempDir Path dir) throws Exception { } } } + + @Test + void cacheableRootStepIsRejected() { + // A deferred root is never promoted, so a cacheable step with no predecessor would hang. + assertThrows( + IllegalArgumentException.class, + () -> Step.of("root", COMPUTE, 1).cache(Duration.ofMinutes(1)).build()); + } + + @Test + void cachedFanProducerIsRejected(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("wc2.db").toString()).open()) { + // A cache hit yields no result list, so a fan-out over a cached step can't expand. + Workflow wf = Workflow.named("badcache") + .step("seed", SEED, 0) + .step(Step.of("producer", COMPUTE, 1) + .cache(Duration.ofMinutes(5)) + .after("seed") + .build()) + .step(Step.of("fan", AFTER) + .fanOut(FanMode.EACH) + .after("producer") + .build()); + assertThrows(WorkflowException.class, () -> queue.submitWorkflow(wf)); + } + } }