feat(java): incremental workflow caching#337
Conversation
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.
A re-run of a cacheable step is a cache hit, not re-executed.
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 23 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (4)
📝 WalkthroughWalkthroughAdds TTL-based result caching for workflow steps. A new ChangesWorkflow Step Caching
Sequence Diagram(s)sequenceDiagram
participant DefaultTaskito
participant WorkflowTracker
participant resultCache
participant JniQueueBackend
participant RustJNI
DefaultTaskito->>RustJNI: submitWorkflowDef(stepSpec with cache:{ttlMs})
RustJNI-->>DefaultTaskito: stored in StepMetadata.cache
WorkflowTracker->>RustJNI: getWorkflowPlan()
RustJNI-->>WorkflowTracker: PlanNode with cache field
Note over WorkflowTracker: On node COMPLETED
WorkflowTracker->>resultCache: store cacheKey → expiry
Note over WorkflowTracker: On successor promotion
WorkflowTracker->>resultCache: reuseFromCache(cacheKey)?
alt Cache hit
WorkflowTracker->>JniQueueBackend: setWorkflowNodeCacheHit(runId, nodeName)
JniQueueBackend->>RustJNI: NativeWorkflows.setWorkflowNodeCacheHit(handle, runId, nodeName)
RustJNI-->>WorkflowTracker: node → CacheHit
WorkflowTracker->>WorkflowTracker: promoteSuccessors immediately
else Cache miss
WorkflowTracker->>JniQueueBackend: createDeferredJob(...)
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
crates/taskito-java/src/workflows/mod.rs (1)
221-239: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftDefinition reuse now ignores cache-policy changes.
step_metanow storescache, but the reuse check below still keys definition identity only ondag_data. Re-submitting the same workflow name/version with identical edges but a different cache TTL (or adding/removing caching entirely) will silently reuse the old definition, so the new policy never reachesgetWorkflowPlan. Please include relevant step metadata in the reuse comparison or reject same-version submissions when metadata changes.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/taskito-java/src/workflows/mod.rs` around lines 221 - 239, The workflow definition reuse check in the `step_meta`/`getWorkflowPlan` path is still comparing only `dag_data`, so cache-policy changes can be missed. Update the reuse logic in `workflows::mod` to include relevant step metadata from `StepMetadata`—at least `cache`, and any other fields that affect execution identity—when deciding whether a same-name/version workflow can be reused, or explicitly reject submissions when those metadata fields differ.sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java (1)
408-421: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winCache misses can be recreated on the wrong queue.
Adding
step.cacheTtlMs != nullhere makes cacheable steps use the deferred-job path on a miss.getWorkflowPlanstill only carries the per-step queue override, andPlanNode.queueOrDefault()falls back to"default", so a cacheable step without explicit.queue(...)can stall named-queue workflows by being recreated on the wrong lane. Either propagate the run default queue through the plan or reject cacheable deferred steps that do not setqueue. Based on learnings, explicitly call.queue(...)on deferred steps becausegetWorkflowPlanmay omit the run default queue and later break deferred job creation inWorkflowTracker.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java` around lines 408 - 421, The deferred-node selection in DefaultTaskito.deferredNodes is now sending cacheable steps through the deferred-job path, but the plan data used by WorkflowTracker/PlanNode.queueOrDefault() does not reliably preserve the run’s default queue. Update the deferred-step handling so cacheable steps either carry an explicit queue through getWorkflowPlan or are rejected unless Step.queue is set; ensure deferred creation never falls back to "default" for named-queue workflows, and reference the Step.cacheTtlMs, getWorkflowPlan, and PlanNode.queueOrDefault flow when fixing it.Source: Learnings
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java`:
- Around line 215-227: The Step.Builder.cache(...) path currently allows
cacheable steps to be attached without predecessors, which creates root
cacheable steps that DefaultTaskito.deferredNodes(...) will defer forever and
never let WorkflowTracker promote to a terminal state. Update the build-time
validation in Step.Builder so that when cacheTtlMs is set and after is empty the
workflow is rejected immediately with an IllegalArgumentException, using the
existing cache(...) / after state in Step.Builder to catch this invalid
configuration before the workflow is built.
In `@sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java`:
- Around line 72-73: The issue is that WorkflowTracker.resultCache keeps expired
cache keys forever, causing unbounded growth on long-lived workers. Update the
caching flow in WorkflowTracker, especially cacheResult(...) and
reuseFromCache(...), so stale entries are actively removed when they expire
rather than only being ignored on reads. Use the existing resultCache and cache
key logic to either delete expired entries on access or replace the map with a
bounded TTL cache implementation that evicts old keys automatically.
- Around line 128-133: Cache reuse is bypassing the normal success flow in
WorkflowTracker, so cached nodes never run the same post-success hooks as a real
success. Update reuseFromCache(...) to restore the cached step’s serialized
result (or equivalent handle) and then execute the same success path used by
onOutcome(..., true, ...) so expandFanOutIfAny(...) and any callable-condition
context can see the predecessor result. Also ensure cacheResult(...) persists
that result data, not just expiry metadata, so buildContext(...) can reconstruct
it for later runs.
---
Outside diff comments:
In `@crates/taskito-java/src/workflows/mod.rs`:
- Around line 221-239: The workflow definition reuse check in the
`step_meta`/`getWorkflowPlan` path is still comparing only `dag_data`, so
cache-policy changes can be missed. Update the reuse logic in `workflows::mod`
to include relevant step metadata from `StepMetadata`—at least `cache`, and any
other fields that affect execution identity—when deciding whether a
same-name/version workflow can be reused, or explicitly reject submissions when
those metadata fields differ.
In `@sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java`:
- Around line 408-421: The deferred-node selection in
DefaultTaskito.deferredNodes is now sending cacheable steps through the
deferred-job path, but the plan data used by
WorkflowTracker/PlanNode.queueOrDefault() does not reliably preserve the run’s
default queue. Update the deferred-step handling so cacheable steps either carry
an explicit queue through getWorkflowPlan or are rejected unless Step.queue is
set; ensure deferred creation never falls back to "default" for named-queue
workflows, and reference the Step.cacheTtlMs, getWorkflowPlan, and
PlanNode.queueOrDefault flow when fixing it.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 3167cee9-c5b4-4696-92c6-7e8f230b6974
📒 Files selected for processing (9)
crates/taskito-java/src/workflows/mod.rssdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.javasdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.javasdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.javasdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.javasdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.javasdks/java/src/main/java/org/byteveda/taskito/workflows/Step.javasdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.javasdks/java/src/test/java/org/byteveda/taskito/WorkflowCacheTest.java
1 participant
Adds opt-in result caching for workflow steps: a step marked
cache(ttl)whoseresult is still fresh from a prior run is marked a cache hit and skipped instead
of re-running its task.
API
Step.cache(Duration ttl)— make a step cacheable forttl.Mechanics
in-memory
resultCachekeyed byworkflow:node:sha256(payload); within TTL itcalls the new
setWorkflowNodeCacheHitJNI fn (marks the nodeCACHE_HIT,terminal) and advances successors without enqueuing the task. On a miss it
enqueues normally and caches the result on completion.
setWorkflowNodeCacheHit+cachefield onStepSpec/PlanNodeView.Scope (documented limitation)
within a live worker, not across restarts (same posture as the rest of the
in-process tracker). A cacheable node also needs a predecessor (a deferred
root isn't promoted).
Tests
WorkflowCacheTestcovers a cache hit on re-run skipping the task and a TTLexpiry forcing re-execution.
Stacked after #335 (saga). Rebased onto master.
Summary by CodeRabbit
New Features
Bug Fixes
Tests