feat: add opt-in max_pending admission cap#449
Merged
Conversation
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 22 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 ChangesPending count storage and bindings
Java admission caps
Node admission caps
Python admission caps
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Queue
participant Storage
participant JobsTable
Queue->>Storage: count_pending_by_queue(queue)
Storage->>JobsTable: count Pending rows for queue
JobsTable-->>Storage: pending count
Storage-->>Queue: pending count
Queue->>Queue: compare count with maxPending
Queue-->>Queue: enqueue or raise QueueFullError
🚥 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: 7
🤖 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 297-298: Extend the admission-cap enforcement from dispatchEnqueue
to enqueueMany: before calling backend.enqueueMany, validate the target queue’s
capacity against the current pending count plus the full batch size. Update or
reuse rejectIfQueueFull and the enqueueMany flow so the entire batch is rejected
before serialization or insertion when it would exceed the configured cap.
- Around line 151-173: Update maxPending(String queue, int cap) to validate that
cap is non-negative before storing it in maxPending. Reject negative values
immediately and leave the existing configuration unchanged; preserve the fluent
return behavior and allow zero as a valid cap.
In `@sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java`:
- Around line 45-47: Update QueueBackend.countPendingByQueue to be a default
optional capability that throws the established unsupported-operation exception,
matching other optional methods and preserving existing implementations. Ensure
callers invoke it only when maxPending is configured, while retaining the
pending-count behavior when that setting is enabled.
In `@sdks/node/src/queue.ts`:
- Around line 473-476: Update enqueueMany in sdks/node/src/queue.ts at lines
473-476 to count jobs by target queue and reject each queue when its current
pending count plus the batch count exceeds its cap, while preserving
all-or-nothing behavior. Add the requested admission test in
sdks/node/test/core/admission.test.ts at lines 51-60 covering pending=1, cap=3,
and a five-job batch that throws.
In `@sdks/python/taskito/app.py`:
- Around line 927-930: Update the batch admission precheck around
_reject_if_queue_full in sdks/python/taskito/app.py: aggregate requested rows by
queue and reject when each queue’s pending count plus its requested rows exceeds
max_pending, before inserting any rows. Add a regression test in
sdks/python/tests/core/test_admission.py covering a batch larger than the
remaining capacity and assert that no rows are inserted.
- Around line 695-696: Ensure producer-side batched tasks enforce the configured
queue cap before dispatching accumulated items: update the batching flow around
`_dispatch_batched_payload()` so it invokes `_reject_if_queue_full()` for the
target queue before calling `_inner.enqueue()`. Preserve the existing
single-task check and apply the same default-queue fallback when no queue is
specified.
In `@sdks/python/tests/core/test_admission.py`:
- Around line 87-102: Strengthen test_cap_frees_after_drain by using a small
pending cap and enough queued work to fill it, asserting an additional enqueue
raises QueueFullError. Wait for the admitted jobs to complete, then enqueue one
more job and verify it succeeds, proving completed jobs release admission
capacity.
🪄 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: 2cd5ef1e-56ff-43f5-b63d-f21c15a61afe
📒 Files selected for processing (28)
crates/taskito-core/src/storage/diesel_common/jobs.rscrates/taskito-core/src/storage/mod.rscrates/taskito-core/src/storage/redis_backend/jobs/query.rscrates/taskito-core/src/storage/sqlite/tests.rscrates/taskito-core/src/storage/traits.rscrates/taskito-core/tests/rust/storage_tests.rscrates/taskito-java/src/queue/inspect.rscrates/taskito-node/src/queue/mod.rscrates/taskito-python/src/py_queue/mod.rssdks/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/errors/QueueFullException.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/spi/QueueBackend.javasdks/java/src/test/java/org/byteveda/taskito/core/AdmissionTest.javasdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.javasdks/node/src/errors.tssdks/node/src/index.tssdks/node/src/queue.tssdks/node/src/types.tssdks/node/test/core/admission.test.tssdks/python/taskito/__init__.pysdks/python/taskito/_taskito.pyisdks/python/taskito/app.pysdks/python/taskito/exceptions.pysdks/python/taskito/mixins/runtime_config.pysdks/python/tests/core/test_admission.py
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/python/tests/core/test_admission.py`:
- Line 150: Update the worker cleanup in the test around worker.join to assert
that worker.is_alive() is false after the timed join, ensuring the test fails
when the worker remains stuck.
🪄 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: 98c96943-b3a9-4c29-8a48-3007b3d0aef7
📒 Files selected for processing (8)
sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.javasdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.javasdks/java/src/test/java/org/byteveda/taskito/core/AdmissionTest.javasdks/node/src/queue.tssdks/node/test/core/admission.test.tssdks/python/taskito/app.pysdks/python/taskito/mixins/runtime_config.pysdks/python/tests/core/test_admission.py
🚧 Files skipped from review as they are similar to previous changes (4)
- sdks/python/taskito/mixins/runtime_config.py
- sdks/node/test/core/admission.test.ts
- sdks/node/src/queue.ts
- sdks/python/taskito/app.py
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
Opt-in per-queue admission cap (S26, Tier 4). Once a queue's pending backlog reaches
max_pending,enqueue/enqueue_manyraise instead of letting the backlog grow unbounded — the inflow companion to retention's backlog bound. Off by default; a queue absent from the config is uncapped with zero overhead.Design
QueueConfig.count_pending_by_queue(queue)— a single-status sibling ofcount_running_by_task, on all three backends (SQLite/Postgres via the Diesel macro, RedisSINTERCARDonjobs:by_queue).Cross-SDK surface
Queue(max_pending={...})ctor kwarg +set_queue_max_pending; raisesQueueFullError(QueueError).configureQueue(q, { maxPending }); throwsQueueFullError extends QueueError; publiccountPendingByQueue.Taskito.maxPending(q, n);QueueFullException;countPendingByQueueon theTaskitointerface +InMemoryQueueBackend.Tests
count_pending_by_queueunit test + cross-backend contract suite.Summary by CodeRabbit
countPendingByQueue/count_pending_by_queueprimitives for per-queue pending counts across storage backends and Java, Node.js, and Python SDKs.maxPending/max_pending) for Java, Node.js, and Python, rejecting enqueues via new queue-full errors/exceptions.enqueueMany/enqueue_many).QueueFullException(Java),QueueFullError(Node/Python), and runtime configuration support for Python caps.