feat(java): idempotent, dependsOn, notes, circuit breakers#383
Merged
Conversation
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 7 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 (9)
📝 WalkthroughWalkthroughThis PR adds circuit-breaker support, structured job notes, dependency gating, and idempotency handling across the Rust JNI layer and Java SDK, including new public configuration/state types, enqueue behavior, and inspection APIs. It also relocates existing Java tests into feature-specific packages. ChangesCircuit breaker, notes, dependsOn, idempotency feature
Java test package restructuring
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant JavaCaller
participant NativeQueue
participant inspect_rs
participant Storage
JavaCaller->>NativeQueue: listCircuitBreakers(handle)
NativeQueue->>inspect_rs: Java_..._listCircuitBreakers
inspect_rs->>Storage: list_circuit_breakers()
Storage-->>inspect_rs: CircuitBreakerRow list
inspect_rs-->>JavaCaller: JSON string of CircuitBreakerView
sequenceDiagram
participant Caller
participant DefaultTaskito
participant Serializer
participant IdempotencyKeys
participant Backend
Caller->>DefaultTaskito: enqueue(task, payload, options)
DefaultTaskito->>Serializer: serialize(payload)
Serializer-->>DefaultTaskito: payloadBytes
DefaultTaskito->>DefaultTaskito: resolveUniqueKey(options, task.idempotent())
DefaultTaskito->>IdempotencyKeys: autoKey(taskName, payloadBytes)
IdempotencyKeys-->>DefaultTaskito: dedup key
DefaultTaskito->>Backend: submit(finalOptions with uniqueKey)
Possibly related PRs
Suggested labels: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 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: 1
🧹 Nitpick comments (4)
sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java (1)
428-462: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider extracting policy/breaker encoding into helper methods.
The loop body mixes two independent concerns (retry-policy fields, breaker fields) in one block, which is why this segment was flagged as high complexity. Splitting into
encodePolicyFields(RetryPolicy, Map)andencodeBreakerFields(CircuitBreakerConfig, Map)would keepencodeTaskConfigs()a simple orchestrator.♻️ Proposed refactor
- RetryPolicy policy = taskPolicies.get(name); - if (policy != null) { - if (policy.baseDelay() != null) { - config.put("baseDelayMs", policy.baseDelay().toMillis()); - } - if (policy.maxDelay() != null) { - config.put("maxDelayMs", policy.maxDelay().toMillis()); - } - if (!policy.customDelays().isEmpty()) { - List<Long> delaysMs = new ArrayList<>(policy.customDelays().size()); - policy.customDelays().forEach(delay -> delaysMs.add(delay.toMillis())); - config.put("customDelaysMs", delaysMs); - } - } - CircuitBreakerConfig breaker = taskCircuitBreakers.get(name); - if (breaker != null) { - config.put("circuitBreakerThreshold", breaker.threshold()); - config.put("circuitBreakerWindowMs", breaker.window().toMillis()); - config.put("circuitBreakerCooldownMs", breaker.cooldown().toMillis()); - config.put("circuitBreakerHalfOpenProbes", breaker.halfOpenProbes()); - config.put("circuitBreakerHalfOpenSuccessRate", breaker.halfOpenSuccessRate()); - } + encodePolicy(config, taskPolicies.get(name)); + encodeBreaker(config, taskCircuitBreakers.get(name));🤖 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/worker/Worker.java` around lines 428 - 462, Refactor the high-complexity loop in Worker.encodeTaskConfigs() by moving the retry-policy field population and circuit-breaker field population into separate helpers, such as encodePolicyFields(RetryPolicy, Map<String, Object>) and encodeBreakerFields(CircuitBreakerConfig, Map<String, Object>). Keep encodeTaskConfigs() focused on collecting task names, creating the per-task map, and delegating to those helpers so the two concerns are encoded independently.crates/taskito-java/src/convert.rs (1)
222-234: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCircuit-breaker defaults duplicated with
worker.rs.The comment notes these
Options "mirror the core's" defaults, but the actual default values (60_000/300_000/5/0.8) are hardcoded again inworker.rs'sregister_task_policiesrather than sourced fromtaskito_core's ownCircuitBreakerConfig::default(). If the core's defaults ever change, these two locations can silently drift.🤖 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/convert.rs` around lines 222 - 234, The circuit-breaker default values are duplicated between TaskRetryConfig handling and register_task_policies, which can drift from taskito_core::CircuitBreakerConfig::default(). Update register_task_policies to source the defaults from the core config instead of hardcoding the 60_000/300_000/5/0.8 values, and keep TaskRetryConfig aligned with that single source of truth so the breaker settings stay consistent.sdks/java/src/main/java/org/byteveda/taskito/task/CircuitBreakerConfig.java (1)
13-56: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider adding
toString()for debuggability.Config objects like this are often logged (e.g., during worker startup/policy registration). A
toString()override would make debug logs and test failures easier to diagnose.🤖 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/task/CircuitBreakerConfig.java` around lines 13 - 56, Add a toString() override to CircuitBreakerConfig so the config is readable in logs and test failures. Implement it in the CircuitBreakerConfig class using the existing fields exposed by threshold(), window(), cooldown(), halfOpenProbes(), and halfOpenSuccessRate(), so startup and policy-registration debug output is more informative.sdks/java/src/main/java/org/byteveda/taskito/task/EnqueueOptions.java (1)
37-49: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMissing
notes()getter for symmetry with the other new fields.
dependsOn,idempotent,idempotencyKey, anduniqueKeyall get public getters, butnotesdoesn't. Serialization to the native layer still works (Jackson picks up the field-level@JsonPropertyvia reflection), but there's no way to read the encoded notes back off anEnqueueOptionsinstance — e.g. for tests or middleware that want to introspect it.♻️ Proposed fix
/** An explicit idempotency key (used as the {`@code` uniqueKey} when set), or {`@code` null}. */ public String idempotencyKey() { return idempotencyKey; } + + /** Canonical JSON encoding of the structured notes, or {`@code` null} when none are set. */ + public String notes() { + return notes; + }Also applies to: 91-113
🤖 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/task/EnqueueOptions.java` around lines 37 - 49, Add a public notes() getter on EnqueueOptions to match the other new fields and allow callers/tests to read the encoded notes value from the object. Keep it consistent with the existing accessor pattern used for dependsOn, idempotent, idempotencyKey, and uniqueKey, and ensure it simply returns the notes field without changing the Jackson serialization behavior.
🤖 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 259-291: The batched enqueue path still skips idempotent
auto-derivation, so `enqueueMany`/`enqueueAll` can drop dedup behavior even when
`idempotent(true)` is set. Update the bulk enqueue flow in `DefaultTaskito` so
each payload resolves its effective key via `resolveUniqueKey(...)` before
creating `perJob`, mirroring the single-enqueue path and preserving explicit
`uniqueKey`/`idempotencyKey` precedence.
---
Nitpick comments:
In `@crates/taskito-java/src/convert.rs`:
- Around line 222-234: The circuit-breaker default values are duplicated between
TaskRetryConfig handling and register_task_policies, which can drift from
taskito_core::CircuitBreakerConfig::default(). Update register_task_policies to
source the defaults from the core config instead of hardcoding the
60_000/300_000/5/0.8 values, and keep TaskRetryConfig aligned with that single
source of truth so the breaker settings stay consistent.
In `@sdks/java/src/main/java/org/byteveda/taskito/task/CircuitBreakerConfig.java`:
- Around line 13-56: Add a toString() override to CircuitBreakerConfig so the
config is readable in logs and test failures. Implement it in the
CircuitBreakerConfig class using the existing fields exposed by threshold(),
window(), cooldown(), halfOpenProbes(), and halfOpenSuccessRate(), so startup
and policy-registration debug output is more informative.
In `@sdks/java/src/main/java/org/byteveda/taskito/task/EnqueueOptions.java`:
- Around line 37-49: Add a public notes() getter on EnqueueOptions to match the
other new fields and allow callers/tests to read the encoded notes value from
the object. Keep it consistent with the existing accessor pattern used for
dependsOn, idempotent, idempotencyKey, and uniqueKey, and ensure it simply
returns the notes field without changing the Jackson serialization behavior.
In `@sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java`:
- Around line 428-462: Refactor the high-complexity loop in
Worker.encodeTaskConfigs() by moving the retry-policy field population and
circuit-breaker field population into separate helpers, such as
encodePolicyFields(RetryPolicy, Map<String, Object>) and
encodeBreakerFields(CircuitBreakerConfig, Map<String, Object>). Keep
encodeTaskConfigs() focused on collecting task names, creating the per-task map,
and delegating to those helpers so the two concerns are encoded independently.
🪄 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: bef5e141-4382-499c-8ce2-d77aad25a66b
📒 Files selected for processing (64)
crates/taskito-java/src/convert.rscrates/taskito-java/src/queue/inspect.rscrates/taskito-java/src/worker.rssdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.javasdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.javasdks/java/src/main/java/org/byteveda/taskito/Taskito.javasdks/java/src/main/java/org/byteveda/taskito/annotation/TaskHandler.javasdks/java/src/main/java/org/byteveda/taskito/dashboard/Contract.javasdks/java/src/main/java/org/byteveda/taskito/errors/NotesValidationException.javasdks/java/src/main/java/org/byteveda/taskito/internal/IdempotencyKeys.javasdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.javasdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.javasdks/java/src/main/java/org/byteveda/taskito/model/CircuitBreakerState.javasdks/java/src/main/java/org/byteveda/taskito/model/Job.javasdks/java/src/main/java/org/byteveda/taskito/serialization/Notes.javasdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.javasdks/java/src/main/java/org/byteveda/taskito/task/CircuitBreakerConfig.javasdks/java/src/main/java/org/byteveda/taskito/task/EnqueueOptions.javasdks/java/src/main/java/org/byteveda/taskito/task/Task.javasdks/java/src/main/java/org/byteveda/taskito/worker/Worker.javasdks/java/src/test/java/org/byteveda/taskito/autoscale/ScalerTest.javasdks/java/src/test/java/org/byteveda/taskito/codegen/AnnotatedPolicies.javasdks/java/src/test/java/org/byteveda/taskito/codegen/AnnotatedPolicyTest.javasdks/java/src/test/java/org/byteveda/taskito/codegen/AnnotationProcessorTest.javasdks/java/src/test/java/org/byteveda/taskito/codegen/Greeter.javasdks/java/src/test/java/org/byteveda/taskito/codegen/ResourceGreeter.javasdks/java/src/test/java/org/byteveda/taskito/contrib/TaskitoLoggerTest.javasdks/java/src/test/java/org/byteveda/taskito/contrib/WebhookTest.javasdks/java/src/test/java/org/byteveda/taskito/core/DeadLetterByTaskTest.javasdks/java/src/test/java/org/byteveda/taskito/core/DependsOnTest.javasdks/java/src/test/java/org/byteveda/taskito/core/EnqueueDecisionTest.javasdks/java/src/test/java/org/byteveda/taskito/core/ErgonomicsTest.javasdks/java/src/test/java/org/byteveda/taskito/core/ExceptionHierarchyTest.javasdks/java/src/test/java/org/byteveda/taskito/core/IdempotencyTest.javasdks/java/src/test/java/org/byteveda/taskito/core/InterceptionTest.javasdks/java/src/test/java/org/byteveda/taskito/core/NotesTest.javasdks/java/src/test/java/org/byteveda/taskito/core/PredicateTest.javasdks/java/src/test/java/org/byteveda/taskito/core/QueueTest.javasdks/java/src/test/java/org/byteveda/taskito/core/RecipesTest.javasdks/java/src/test/java/org/byteveda/taskito/dashboard/DashboardTest.javasdks/java/src/test/java/org/byteveda/taskito/internal/FfmRoundTripTest.javasdks/java/src/test/java/org/byteveda/taskito/resources/ProxyTest.javasdks/java/src/test/java/org/byteveda/taskito/resources/ResourceTest.javasdks/java/src/test/java/org/byteveda/taskito/serialization/EncryptedGreeter.javasdks/java/src/test/java/org/byteveda/taskito/serialization/MsgpackSerializerTest.javasdks/java/src/test/java/org/byteveda/taskito/serialization/PayloadCodecTest.javasdks/java/src/test/java/org/byteveda/taskito/serialization/PerTaskCodecTest.javasdks/java/src/test/java/org/byteveda/taskito/worker/BatcherTest.javasdks/java/src/test/java/org/byteveda/taskito/worker/CircuitBreakerTest.javasdks/java/src/test/java/org/byteveda/taskito/worker/LockTest.javasdks/java/src/test/java/org/byteveda/taskito/worker/MeshWorkerTest.javasdks/java/src/test/java/org/byteveda/taskito/worker/PeriodicTest.javasdks/java/src/test/java/org/byteveda/taskito/worker/RetryPolicyTest.javasdks/java/src/test/java/org/byteveda/taskito/worker/WorkerTest.javasdks/java/src/test/java/org/byteveda/taskito/workflows/CanvasTest.javasdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowCacheTest.javasdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowConditionTest.javasdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowFanOutTest.javasdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowGateTest.javasdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowGraphTest.javasdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowSagaTest.javasdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowSubWorkflowTest.javasdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowSubmitMapTest.javasdks/java/src/test/java/org/byteveda/taskito/workflows/WorkflowTest.java
Batch enqueue skipped resolveUniqueKey, so idempotent(true) lost dedup semantics in bulk.
pratyush618
changed the title
This was referenced Jul 7, 2026
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.
Summary
Closes four features the Java SDK was missing. Three are core-backed — the Rust core already
implements them and the JNI binding was hardcoding defaults at the seam — so the change is mostly
threading an existing field through; one is a pure SDK-side convenience. All are opt-in and
non-breaking. Also regroups the flat Java test suite into feature-area packages.
Features
@TaskHandler(idempotent = true)orEnqueueOptions.idempotent(true)derive a
uniqueKeyfrom the payload so a duplicate enqueue is a no-op while the first job ispending/running. Key =
auto:+sha256(name ‖ 0x00 ‖ pre-codec payload), truncated to 32 hexchars — matching the cross-SDK idempotency-key contract (a golden-vector test pins it). Precedence:
explicit
uniqueKey→idempotencyKey→ per-call opt-out → task/enqueue flag → auto-derive.dependsOnon enqueue —EnqueueOptions.dependsOn(...)gates a job on other job ids. The corealready validates edges, gates dequeue, and cascade-cancels on a failed dependency; the binding now
threads the field instead of discarding it.
EnqueueOptions.notes(Map)attaches a bounded, validated annotation map(≤15 fields, ≤64-char keys, ≤500-char string leaves, ≤3 nesting, ≤4 KiB), stored as canonical JSON
and surfaced via
Job.notesMap()and the dashboard contract.Task.circuitBreaker(...)/@TaskHandler(circuitBreaker*)registerthe core's per-task breaker (state machine + enforcement already live in the core; the binding only
supplies config). A new
listCircuitBreakers()inspection call surfaces breaker state.Tests
dependency gating + cascade cancel, notes limit boundaries + storage round-trip, breaker
open/half-open behaviour, and annotation-processor codegen for the new per-task attributes.
clippyclean; workspace tests pass.core,worker,workflows,serialization,codegen,resources,contrib,dashboard,cli,autoscale,internal), matching the layout the rest of the repo already uses.Out of scope (follow-up)
Summary by CodeRabbit
dependsOnso jobs can wait for other jobs to finish.