feat: per-message ack for log topics#463
Merged
Merged
Conversation
Opt-in per-message consumption: lease a message with a visibility timeout, then ack or nack it individually, so one poison message no longer blocks the cursor. Redelivery on nack or lease timeout.
|
Caution Review failedAn error occurred during the review process. Please try again later. 📝 WalkthroughWalkthroughAdds opt-in per-message leasing, acknowledgement, negative acknowledgement, timeout redelivery, and delivery-state compaction for log topics across Diesel, Redis, Java, Node, and Python APIs. ChangesPer-message topic delivery
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Application
participant SDK
participant StorageBackend
Application->>SDK: lease topic messages
SDK->>StorageBackend: lease_topic_messages
StorageBackend-->>SDK: leased TopicMessage values
SDK-->>Application: decoded messages
Application->>SDK: ackMessage or nackMessage
SDK->>StorageBackend: update delivery state
StorageBackend-->>SDK: boolean result
Possibly related PRs
🚥 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: 10
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
crates/taskito-core/src/storage/diesel_common/pubsub.rs (1)
371-457: 🚀 Performance & Scalability | 🟠 Major | ⚡ Quick winPer-message compaction can be blocked forever by a since-unsubscribed subscriber's orphaned delivery rows.
pm_subsis built from every distinct(topic, subscription_name)pair ever recorded intopic_deliveries, not from currently-registered subscriptions.unsubscribe()never deletes the unsubscribed consumer'stopic_deliveriesrows, so a subscriber that unsubscribes with unacked leases permanently preventssub_countfrom being satisfied for those (and later) messages on that topic — unbounded row growth with noexpires_atsafety net once a log subscriber exists.The cleanest fix is at
unsubscribe()(unchanged in this diff, so shown here rather than as an in-range diff):pub fn unsubscribe(&self, topic: &str, subscription_name: &str) -> Result<bool> { let mut conn = self.conn()?; diesel::delete( topic_deliveries::table .filter(topic_deliveries::topic.eq(topic)) .filter(topic_deliveries::subscription_name.eq(subscription_name)), ) .execute(&mut conn)?; let affected = diesel::delete(topic_subscriptions::table.find((topic, subscription_name))) .execute(&mut conn)?; Ok(affected > 0) }Alternatively, filter
pm_subshere againstlist_subscriptions_for_topic/an active-subscriptions query before using it as the compaction denominator.🤖 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/diesel_common/pubsub.rs` around lines 371 - 457, Update unsubscribe() to delete all topic_deliveries rows for the specified topic and subscription_name before deleting the topic_subscriptions record, while preserving its existing boolean result based on subscription deletion. This prevents stale delivery entries from being included in the per-message compaction denominator.
🤖 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 `@crates/taskito-core/src/storage/diesel_common/pubsub.rs`:
- Around line 485-518: Update lease_topic_messages to perform the same
subscription registration and mode validation used by read_topic_messages and
ack_topic_cursor before querying or creating delivery rows. Require
subscription_name to exist and have mode exactly "log"; reject unregistered or
non-log subscriptions through the existing error path, while preserving the
current leasing behavior for valid log subscriptions.
- Around line 485-549: The lease_topic_messages transaction must atomically
claim messages before returning them to prevent concurrent callers from
receiving the same row. Update the topic_messages selection to use row-level
locking with SKIP LOCKED where supported, or an equivalent atomic claim
mechanism, ensuring locks/claims cover the availability check through delivery
upsert while preserving ordering and limit behavior.
In `@crates/taskito-core/src/storage/redis_backend/pubsub.rs`:
- Around line 838-844: Update purge_topic_messages to remove each purged
message’s field from every matching pmdeliv topic/subscription hash, enumerating
the topic’s subscribers before or during XTRIM/XDEL cleanup. Ensure fields for
both trimmed and explicitly deleted stream entries are removed while preserving
existing stream purge behavior.
- Around line 849-863: The lease_topic_messages method must enforce the same
subscription-mode restriction as read_topic_messages and ack_topic_cursor. Load
the subscription’s SubEntry and return the established non-log rejection result
unless entry.mode equals SUBSCRIPTION_MODE_LOG, before leasing any messages;
preserve the existing behavior for valid log subscriptions.
- Around line 865-870: Update lease_topic_messages so its XRANGE scan is
COUNT-bounded and paginated rather than issuing an unbounded “-” to “+” query.
Loop through pages with a fixed per-call cap, advancing the range cursor while
skipping in-flight entries, and stop once enough leaseable messages are
collected or the cap/stream end is reached.
- Around line 849-892: The lease_topic_messages method must make availability
checks and lease writes atomic across concurrent callers. Replace the snapshot
HGETALL plus per-entry HSET flow with a Redis transaction or Lua script using
WATCH/MULTI/EXEC, ensuring each message is rechecked against the current
delivery state before writing its lease and retried on conflicts; preserve the
limit, expiry, acknowledged-message, and result-ordering behavior.
In `@crates/taskito-core/tests/rust/storage_tests.rs`:
- Around line 1488-1492: Update the Redis implementation of purge_topic_messages
to compact per-message entries acknowledged by every relevant subscriber, rather
than relying only on log cursors. Remove each fully acknowledged message from
the Redis message storage and delete its associated per-message delivery-state
entries, while preserving unacknowledged messages such as m1 and returning the
number removed.
In `@docs/content/docs/java/api-reference/queue/pubsub.mdx`:
- Around line 23-24: The Java documentation at
docs/content/docs/java/api-reference/queue/pubsub.mdx lines 23-24 and Python
documentation at docs/content/docs/python/api-reference/queue/pubsub.mdx lines
176-201 must consistently describe TopicMessage as returned by both read and
lease APIs. Clarify that cursor-read message IDs use ackTopic/ack_topic, while
leased message IDs use ackMessage/nackMessage or ack_message/nack_message,
including the relevant read_topic, lease_topic, and acknowledgment API
relationships.
In `@docs/content/docs/shared/guides/core/pubsub.mdx`:
- Around line 612-616: Update the lease visibility wording in the pubsub guide
to scope an unexpired lease to the subscription: other lease reads on that same
subscription skip the message until the visibility window expires, while
separate subscriptions may still read it. Preserve the existing ack, nack, and
lease-timeout behavior descriptions.
- Around line 661-665: Update the retention guidance in the subscription cleanup
section to state that Diesel ack-based compaction applies only when a topic is
consumed exclusively per-message; when cursor and per-message subscribers are
mixed, document expiration-based cleanup instead. Preserve the existing Redis
expiration/stream-trim guidance and the warning against mixing consumption
styles.
---
Outside diff comments:
In `@crates/taskito-core/src/storage/diesel_common/pubsub.rs`:
- Around line 371-457: Update unsubscribe() to delete all topic_deliveries rows
for the specified topic and subscription_name before deleting the
topic_subscriptions record, while preserving its existing boolean result based
on subscription deletion. This prevents stale delivery entries from being
included in the per-message compaction denominator.
🪄 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: 98bd84ec-9c87-40fb-b2a3-bdac3223b0b2
📒 Files selected for processing (29)
crates/taskito-core/BINDING_CONTRACT.mdcrates/taskito-core/migrations/m0007_topic_deliveries.rscrates/taskito-core/src/storage/diesel_common/pubsub.rscrates/taskito-core/src/storage/mod.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/traits.rscrates/taskito-core/tests/rust/storage_tests.rscrates/taskito-java/src/queue/pubsub.rscrates/taskito-node/src/queue/pubsub.rscrates/taskito-python/src/py_queue/pubsub.rsdocs/content/docs/java/api-reference/queue/pubsub.mdxdocs/content/docs/node/api-reference/queue/pubsub.mdxdocs/content/docs/python/api-reference/queue/pubsub.mdxdocs/content/docs/shared/guides/core/pubsub.mdxsdks/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/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/PerMessageAckTest.javasdks/node/src/queue.tssdks/node/test/core/per-message-ack.test.tssdks/python/taskito/_taskito.pyisdks/python/taskito/mixins/pubsub.pysdks/python/tests/core/test_pubsub_log.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.
Per-message ack / nack / redelivery for log topics
Third and final S28 pub/sub follow-up. Adds an opt-in per-message consumption
path for log topics alongside the existing high-water-mark cursor, so a poison
message no longer blocks its siblings.
What it does
lease_topic_messages(topic, sub, limit, visibility_ms, now))instead of reading against the cursor. Each leased message is tracked per
(subscription, message)with a visibility timeout.ack_messageends a delivery (never redelivered);nack_messagemakes itimmediately available again. An un-acked lease redelivers once its visibility
window expires.
attemptscounts (re)deliveries.subscription can be read either way; per-message just tracks delivery state.
Storage
topic_deliveriestable (migrationm0007), one boundedanti-join to find available messages + a per-message lease upsert in one txn.
Retention additionally compacts messages every per-message subscriber has acked
(mixed cursor+per-message topics fall back to
expires_at).pmdeliv:<topic>:<sub>hash mirrors the delivery state over theexisting stream; ack/nack are guarded no-ops on unknown/already-acked entries.
Cross-SDK contract
Full parity —
leaseTopic/ackMessage/nackMessage(+ Pythonlease_topic)across all shells, wired through the same binding-wrapper + facade pattern. The
BINDING_CONTRACT.mdTopic pub/subsection documents the semantics and the onedocumented backend difference (Redis reclaims via
expires_at/stream-trim only).Tests
test_per_message_ack,test_per_message_purge): lease →visibility-timeout redelivery, ack removes, nack redelivers now,
attemptsbumps, retention only after all-acked. SQLite locally; PG/Redis via CI services.
Depends on nothing in the registry PR; layered on current
master.Summary by CodeRabbit
New Features
Documentation
Tests