feat(java): saga compensation for workflows#335
Merged
Conversation
A failed run with compensable steps rolls back: Step.compensate registers a rollback task, and on failure the tracker compensates completed steps in reverse-dependency waves (parallel within a wave). Adds saga JNI fns + compensate metadata + a SagaOrchestrator; the run ends compensated (or compensation_failed). The terminal state is set without a transient Failed.
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 8 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 (1)
📝 WalkthroughWalkthroughAdds saga compensation support for workflow steps. Steps can declare compensators, plan data now carries that field, failed top-level runs trigger reverse-order compensation waves, and JNI-backed storage records run and node compensation states. Integration tests cover compensated and non-compensated failure paths. ChangesSaga Compensation Feature
Sequence Diagram(s)sequenceDiagram
participant WorkflowTracker
participant SagaOrchestrator
participant QueueBackend
WorkflowTracker->>SagaOrchestrator: startCompensation(runId, plan, statuses)
SagaOrchestrator->>QueueBackend: setWorkflowRunCompensating(runId)
SagaOrchestrator->>QueueBackend: enqueue compensation wave
QueueBackend-->>WorkflowTracker: onOutcome(compensation job)
WorkflowTracker->>SagaOrchestrator: onCompensationCompleted(...)
SagaOrchestrator->>QueueBackend: setWorkflowNodeCompensated / setWorkflowNodeCompensationFailed
alt more waves remain
SagaOrchestrator->>QueueBackend: enqueue next wave
else saga ends
SagaOrchestrator->>QueueBackend: setWorkflowRunCompensated / setWorkflowRunCompensationFailed
SagaOrchestrator->>WorkflowTracker: forget(runId)
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested labels
🚥 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: 9
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
crates/taskito-java/src/workflows/mod.rs (1)
524-585: 🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy liftPreserve the effective queue in
getWorkflowPlan().Saga rollback now re-enqueues compensators from this plan JSON, but this view still returns
queue = nullwhen a step inherited the run default. The Java side then falls back toPlanNode.queueOrDefault(), so the undo job can land on literal"default"instead of the forward lane and never be picked up by the worker that handled the original step. Based on learnings,getWorkflowPlanmay omit the run default queue, which later breaks deferred job creation inWorkflowTracker; the long-term fix is a native serialization change sogetWorkflowPlanincludes the run default queue.🤖 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 524 - 585, `Java_org_byteveda_taskito_internal_NativeWorkflows_getWorkflowPlan` is serializing `PlanNodeView.queue` as null when a step inherits the run default, which later causes rollback/retry path logic to fall back to `"default"` instead of the original lane. Update the native plan serialization in `getWorkflowPlan` so it emits the effective queue for each node, using the step metadata when present and otherwise the workflow run’s default queue, and keep `PlanNodeView` as the place where this value is populated for the JSON returned to Java.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/DefaultTaskito.java`:
- Around line 478-480: The child-workflow path is now serializing step-level
compensators even though nested runs cannot be rolled back by
WorkflowTracker.finalizeIfTerminal(), so child failures may leave uncompensated
side effects. Update encodeChild()/stepSpec() in DefaultTaskito to either reject
or omit compensate for child workflows until child-run compensation is
supported, and keep the top-level saga behavior unchanged.
In
`@sdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.java`:
- Around line 69-72: Make saga startup in SagaOrchestrator.startCompensation
idempotent and failure-safe by preventing duplicate initialization for the same
runId: use runs.putIfAbsent or a per-run lock instead of overwriting the
existing SagaRun. Ensure dispatchNextWave is only invoked once the compensating
run is safely recorded, and if dispatching the first wave fails, transition the
backend state from COMPENSATING to compensation_failed so the run is not left
half-started.
- Around line 41-43: The compensation routing in SagaOrchestrator is relying too
much on the in-memory compensationJobs map, which can miss fast completions and
is lost on restart. Update isCompensationJob and the enqueue/registration flow
so compensation jobs are identified from persisted compensation metadata or
durable compensation job IDs, not only by compensationJobs.containsKey(jobId).
Ensure the backend.enqueue path cannot expose a job before routing information
is available, and restore the same routing after tracker restart.
- Around line 178-180: Update SagaOrchestrator.isCompensable so that
compensation only applies to steps that actually executed in the current run;
remove CACHE_HIT from the compensable statuses and keep only
NodeStatus.COMPLETED (or otherwise gate compensation on execution state). Use
the isCompensable helper and the compensation flow in SagaOrchestrator to ensure
cached results do not trigger compensators by default.
- Around line 53-55: The compensation path in SagaOrchestrator is silently
falling back to an empty payload when backend.getResult(snap.jobId) returns
nothing, which violates the rollback contract. Update the forward-result
retrieval logic in the compensation flow to fail fast if the result is missing
instead of using new byte[0], and ensure the rollback invocation only proceeds
with a real forward payload obtained from backend.getResult.
- Around line 88-100: Serialize per-run wave advancement in SagaOrchestrator:
the compensation completion path can let multiple outcomes see run.inflight
empty at the same time and both advance the saga. Update the logic around
run.inflight.remove(node) and the in-flight empty check so only one thread can
transition a run from one wave to the next, and guard the
finalizeSaga/run.dispatchNextWave decision with synchronization on the per-run
state (or another single-run lock) to prevent duplicate dispatches.
In `@sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java`:
- Around line 197-211: The compensation API on Step.Builder currently allows
registering rollback tasks for step kinds that never produce a forward result,
which later causes SagaOrchestrator.startCompensation() to enqueue an undo task
with an empty payload. Update Step.Builder.build() (or the validation path
around compensate(String)/compensate(Task<?>)) to reject compensation for node
types that cannot emit a forward job result, using the existing Step/Builder and
node-kind checks to fail fast before the API can be used incorrectly.
In `@sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java`:
- Around line 458-466: The compensation gate in WorkflowTracker should not
trigger on any failed node alone, because runs with on_failure or always
recovery may still finalize as completed_with_failures. Update the saga start
check to use the same terminal-state classification as
backend.finalizeRunIfTerminal(...) or a shared non-mutating classifier before
calling saga.startCompensation, and keep the childToParent/runId top-level check
intact. Ensure the decision is based on whether the run would actually finalize
as failed, not just whether NodeStatus.FAILED appears in statuses.
In `@sdks/java/src/test/java/org/byteveda/taskito/WorkflowSagaTest.java`:
- Around line 40-63: Update WorkflowSagaTest so the compensation assertions
verify both execution order and single invocation, not just unique payloads in a
Set. Replace the current undone collection in the test with an order-preserving
structure plus per-compensator call tracking, then assert that UNDO_B runs
before UNDO_A and that each compensator is invoked exactly once after
trackWorkflows() completes and run.await(...) returns.
---
Outside diff comments:
In `@crates/taskito-java/src/workflows/mod.rs`:
- Around line 524-585:
`Java_org_byteveda_taskito_internal_NativeWorkflows_getWorkflowPlan` is
serializing `PlanNodeView.queue` as null when a step inherits the run default,
which later causes rollback/retry path logic to fall back to `"default"` instead
of the original lane. Update the native plan serialization in `getWorkflowPlan`
so it emits the effective queue for each node, using the step metadata when
present and otherwise the workflow run’s default queue, and keep `PlanNodeView`
as the place where this value is populated for the JSON returned to Java.
🪄 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: a558422e-644e-4c07-b708-183b2fe05266
📒 Files selected for processing (11)
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/RunPlan.javasdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.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/WorkflowSagaTest.java
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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/SagaOrchestrator.java`:
- Around line 114-121: `SagaOrchestrator.onCompensationCompleted` advances to
`dispatchNextWave(...)` without the same safety net used in
`startCompensation()`, so an enqueue/dispatch exception can leave the run stuck
in COMPENSATING. Wrap the `dispatchNextWave(runId, run)` call in a try/catch
that handles the `WorkflowException` path, logs the failure, and calls
`finalizeSaga(runId, run)` just like the start-path guard; use the existing
`finalizeSaga`, `dispatchNextWave`, and `onCompensationCompleted` flow to keep
the run from remaining in `runs`.
🪄 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: e7cb0bf8-c579-4b91-8929-3e497dc9480b
📒 Files selected for processing (4)
sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.javasdks/java/src/main/java/org/byteveda/taskito/workflows/SagaOrchestrator.javasdks/java/src/main/java/org/byteveda/taskito/workflows/Step.javasdks/java/src/test/java/org/byteveda/taskito/WorkflowSagaTest.java
🚧 Files skipped from review as they are similar to previous changes (1)
- sdks/java/src/test/java/org/byteveda/taskito/WorkflowSagaTest.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds saga-style rollback to Java workflows. When a run with compensable steps
fails, completed steps are compensated in reverse-dependency order — each wave
of independent rollbacks runs in parallel, the next wave only after the previous
drains. The run ends
compensated(orcompensation_failedif a rollbackitself fails), and observers never see a transient
failedfirst.API
Step.compensate(Task<?>)/compensate(String)— register a rollback task.The compensation job runs with the forward step's result as its payload.
Workflowsurface; compensation is driven entirely by the workertracker (
trackWorkflows()).Mechanics
SagaOrchestrator(worker-side): on a failed run with compensable completednodes, sets the run
compensating, dispatches reverse-topo waves ofcompensation jobs (idempotency-keyed
compensation:{run}:{node}, taggedcompensation:truemetadata so they never advance the forward run), thenfinalizes
compensated/compensation_failed.WorkflowTracker.finalizeIfTerminalchecks for compensable failure BEFORE thecore marks the run failed; compensation-job outcomes route to the saga via
isCompensationJob.taskito-workflowssaga primitives):setWorkflowRunCompensating/Compensated/CompensationFailed,setWorkflowNodeCompensationJob/Compensated/CompensationFailed,setWorkflowRunCompletedWithFailures. No new core storage.Tests
WorkflowSagaTest: a 3-step run whose last step fails compensates the twocompleted steps (correct results as payloads, reverse order, run
compensated);a run with no compensators still fails normally.
Stacked on #334 (needs
WorkflowException). Rebased onto master.Summary by CodeRabbit
compensatetask, which is reflected in workflow plans.