feat: persist per-subscription delivery settings#415
Conversation
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 33 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 (30)
📝 WalkthroughWalkthroughChangesThe pub/sub system now persists delivery settings on subscription rows. Publish-time task defaults are removed, and fan-out resolves each field from explicit publish overrides, persisted subscription settings, or queue defaults. Java, Node, and Python bindings expose the updated registration APIs. Subscription delivery settings
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Subscriber
participant SubscriptionStore
participant Producer
participant PubSubCore
Subscriber->>SubscriptionStore: register subscription delivery settings
Producer->>PubSubCore: publish message with queue defaults and overrides
PubSubCore->>SubscriptionStore: load persisted subscription row
PubSubCore-->>Producer: create delivery job using resolved settings
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
Store priority/max_retries/timeout_ms on the subscription row at registration so publish applies each subscriber's own settings even from a producer process that never loaded the task. Resolution: publish override > row > queue default. Drops the in-process task_defaults map — the row is a strict superset.
1da1e47 to
cbc5ba3
Compare
![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 (3)
crates/taskito-core/src/storage/redis_backend/pubsub.rs (1)
14-31: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider a regression test for legacy-blob deserialization.
The
#[serde(default)]fallback for old blobs lackingpriority/max_retries/timeout_msis exactly the kind of backward-compat behavior that's easy to silently break in a future refactor. A small test deserializing aSubEntryJSON blob without these fields (asserting they come backNone) would lock this in.🧪 Example test
#[test] fn sub_entry_deserializes_legacy_blob_without_delivery_settings() { let legacy = r#"{"topic":"orders","subscription_name":"email","task_name":"send_email","queue":"default","active":true,"durable":true,"owner_worker_id":null,"created_at":0}"#; let entry: SubEntry = serde_json::from_str(legacy).unwrap(); assert_eq!(entry.priority, None); assert_eq!(entry.max_retries, None); assert_eq!(entry.timeout_ms, None); }🤖 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-core/src/storage/redis_backend/pubsub.rs` around lines 14 - 31, Add a focused regression test for SubEntry deserialization using a legacy JSON blob that omits priority, max_retries, and timeout_ms; deserialize it with serde_json and assert all three fields are None while retaining the existing legacy fields and values.crates/taskito-core/tests/rust/storage_tests.rs (1)
860-876: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winExtend the shared CRUD test to assert persisted delivery-setting values round-trip.
This test runs across Postgres, SQLite, and Redis (per the storage trait tests), but
sub(...)always passesNoneforpriority/max_retries/timeout_ms, so no backend is asserted to actually persist and return aSome(x)value for these new columns. A mapping bug specific to one backend'sregister_subscription/list mapping wouldn't be caught here.🧪 Suggested addition
// After the existing upsert-idempotency assertions: s.register_subscription(&NewSubscriptionRow { topic: "ts-orders", subscription_name: "priced", task_name: "priced_task", queue: "default", active: true, durable: true, owner_worker_id: None, created_at: now, priority: Some(5), max_retries: Some(1), timeout_ms: Some(42_000), }) .unwrap(); let priced = s .list_subscriptions_for_topic("ts-orders") .unwrap() .into_iter() .find(|r| r.subscription_name == "priced") .unwrap(); assert_eq!(priced.priority, Some(5)); assert_eq!(priced.max_retries, Some(1)); assert_eq!(priced.timeout_ms, Some(42_000));🤖 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-core/tests/rust/storage_tests.rs` around lines 860 - 876, Extend the shared CRUD test after the existing upsert-idempotency assertions to register a subscription with non-None priority, max_retries, and timeout_ms values, then list it by topic and assert those fields round-trip unchanged. Keep the test backend-agnostic and use the existing register_subscription and list_subscriptions_for_topic flow.sdks/node/src/queue.ts (1)
540-554: 🎯 Functional Correctness | 🔵 Trivial | 🏗️ Heavy liftAdd
priorityto the Node subscriber optionsTaskOptions/SubscriberOptionsdon’t expose it, so this path always persistsundefinedand Node subscriptions can only inherit the queue default. If parity with Java/Python is intended, threadprioritythroughregisterSubscription().🤖 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/node/src/queue.ts` around lines 540 - 554, Expose priority on the Node TaskOptions/SubscriberOptions types, then update the subscription registration flow around registerSubscription to read the subscriber task’s priority and pass it through instead of undefined. Preserve queue-default fallback when no priority is configured.
🤖 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/spi/QueueBackend.java`:
- Around line 170-206: Update the 9-argument registerSubscription overload in
QueueBackend so its default implementation delegates to the existing 6-argument
registerSubscription overload, discarding the optional priority, maxRetries, and
timeoutMs values. Preserve the 6-argument overload as the compatibility path so
existing backends that override it continue handling subscriptions.
---
Nitpick comments:
In `@crates/taskito-core/src/storage/redis_backend/pubsub.rs`:
- Around line 14-31: Add a focused regression test for SubEntry deserialization
using a legacy JSON blob that omits priority, max_retries, and timeout_ms;
deserialize it with serde_json and assert all three fields are None while
retaining the existing legacy fields and values.
In `@crates/taskito-core/tests/rust/storage_tests.rs`:
- Around line 860-876: Extend the shared CRUD test after the existing
upsert-idempotency assertions to register a subscription with non-None priority,
max_retries, and timeout_ms values, then list it by topic and assert those
fields round-trip unchanged. Keep the test backend-agnostic and use the existing
register_subscription and list_subscriptions_for_topic flow.
In `@sdks/node/src/queue.ts`:
- Around line 540-554: Expose priority on the Node TaskOptions/SubscriberOptions
types, then update the subscription registration flow around
registerSubscription to read the subscriber task’s priority and pass it through
instead of undefined. Preserve queue-default fallback when no priority is
configured.
🪄 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: 6ec1b577-ab21-4f76-a8a6-60bd68721279
📒 Files selected for processing (30)
crates/taskito-core/src/pubsub.rscrates/taskito-core/src/storage/diesel_common/migrations.rscrates/taskito-core/src/storage/models.rscrates/taskito-core/src/storage/postgres/pubsub.rscrates/taskito-core/src/storage/redis_backend/pubsub.rscrates/taskito-core/src/storage/schema.rscrates/taskito-core/src/storage/sqlite/pubsub.rscrates/taskito-core/src/storage/sqlite/tests.rscrates/taskito-core/tests/rust/storage_tests.rscrates/taskito-java/src/convert.rscrates/taskito-java/src/queue/pubsub.rscrates/taskito-java/src/worker.rscrates/taskito-node/src/config.rscrates/taskito-node/src/queue/pubsub.rscrates/taskito-python/src/py_queue/pubsub.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/NativeQueue.javasdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.javasdks/java/src/main/java/org/byteveda/taskito/worker/Worker.javasdks/java/src/test/java/org/byteveda/taskito/core/PubSubTest.javasdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.javasdks/java/test-support/src/test/java/org/byteveda/taskito/test/InMemoryPubSubTest.javasdks/node/src/native.tssdks/node/src/queue.tssdks/node/src/types.tssdks/node/test/core/pubsub.test.tssdks/python/taskito/_taskito.pyisdks/python/taskito/mixins/pubsub.pysdks/python/tests/core/test_pubsub.py
💤 Files with no reviewable changes (1)
- sdks/node/src/native.ts
An external backend that overrides only the settings-less overload broke once subscribe() called the 9-arg form (its default threw). The 6-arg default is now the terminal base, the 9-arg default delegates to it, and both built-in backends override the 6-arg to route through their own 9-arg impl.
1 participant
Summary
Changes
topic_subscriptions, populated at registration;publish_to_topicresolves delivery settings from the row. Drops the in-processtask_defaultsmap — the row is a strict superset (works cross-process and in-process)Test plan
Summary by CodeRabbit
New Features
Bug Fixes