feat: add opt-in log+cursor pub/sub#459
Merged
Merged
Conversation
S28: a log topic stores one topic_messages row per publish (vs one job per subscriber) and each log subscription tracks a cursor. Adds the topic_messages table (m0005), mode/cursor columns, four Storage methods (publish_message/read_topic_messages/ack_topic_cursor/topic_log_stats) plus purge_topic_messages across SQLite, Postgres, and Redis Streams.
publish_to_topic writes one topic_messages row when a topic has a log subscriber and fans out jobs only to fan-out subscribers. The retention sweep now compacts log messages every subscriber has acked past.
subscribe_log registers a durable cursor; read_topic pulls decoded messages after it; ack_topic advances it. Adds the TopicMessage dataclass and topic_log_stats.
subscribeLog registers a durable cursor; readTopic pulls decoded messages after it; ackTopic advances it. Adds TopicMessage/TopicLogStat types and topicLogStats.
subscribeLog registers a durable cursor; readTopic pulls messages after it (payload as bytes); ackTopic advances it. Adds TopicMessage/TopicLogStat models and topicLogStats.
Add the Topic pub/sub section to BINDING_CONTRACT (fanout vs log mode, cursor/ack/at-least-once semantics, salting test vector) and log-topic docs across the shared guide and Python/Node/Java API references.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (6)
🚧 Files skipped from review as they are similar to previous changes (5)
📝 WalkthroughWalkthroughChangesTopic log pub/sub
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant SDK
participant NativeBinding
participant Storage
participant TopicLog
SDK->>NativeBinding: subscribeLog and readTopic
NativeBinding->>Storage: register or read subscription
Storage->>TopicLog: fetch messages after cursor
TopicLog-->>Storage: ordered topic messages
Storage-->>NativeBinding: messages and log stats
NativeBinding-->>SDK: decoded message or statistics
SDK->>NativeBinding: ackTopic
NativeBinding->>Storage: advance cursor
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 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: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
crates/taskito-core/src/storage/sqlite/pubsub.rs (1)
32-52: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winReject unknown subscription modes before persisting them.
sub.modeis written verbatim, whilepublish_to_topictreats only"log"specially and silently treats every other value as fan-out. A typo such as"logs"can produce fan-out jobs with the empty task name used by log subscribers instead of storing messages. Validate against the two shared constants before registration across all backends.🤖 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/sqlite/pubsub.rs` around lines 32 - 52, Validate sub.mode in the subscription registration flow before constructing or persisting the row, accepting only the two shared mode constants used by publish_to_topic and rejecting any other value. Apply the same validation across every backend’s registration implementation, including the SQLite path around topic_subscriptions, without changing valid log or fan-out behavior.crates/taskito-core/src/pubsub.rs (1)
75-105: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftMake mixed-mode publishing atomic.
publish_messagesucceeds beforeenqueue_batch/enqueue_unique_batchruns. If the enqueue fails, callers receive an error although the log entry is already durable; a retry appends another entry and can leave log and fan-out deliveries inconsistent. Add one backend-level publish operation that atomically appends the log entry and enqueues fan-out jobs (a transaction for Diesel and an atomic Redis operation).🤖 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/pubsub.rs` around lines 75 - 105, Make the publishing flow around publish_message and enqueue_batch/enqueue_unique_batch atomic by introducing a backend-level operation that performs the log append and all fan-out enqueues as one unit. Implement the operation transactionally for Diesel and as a single atomic Redis operation, then update the caller to use it while preserving log-only and fan-out-only behavior.
🧹 Nitpick comments (3)
sdks/python/tests/core/test_pubsub_log.py (1)
12-21: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCover the metadata and notes mapping.
read_topic()decodes both fields, but this test only verifies payload decoding. Assert them to protect the publicTopicMessagecontract.Proposed test update
- assert queue.publish("events", 1, kind="a") == [] + assert queue.publish( + "events", + 1, + kind="a", + metadata={"tenant": "acme"}, + notes={"source": "test"}, + ) == [] assert queue.publish("events", 2, kind="b") == [] msgs = queue.read_topic("events", "analytics") assert [(m.args, m.kwargs) for m in msgs] == [((1,), {"kind": "a"}), ((2,), {"kind": "b"})] + assert msgs[0].metadata == {"tenant": "acme"} + assert msgs[0].notes == {"source": "test"} assert all(isinstance(m, TopicMessage) for m in msgs)🤖 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/python/tests/core/test_pubsub_log.py` around lines 12 - 21, Update test_publish_stores_one_message_per_call to also assert each decoded TopicMessage’s metadata and notes fields returned by read_topic(). Verify the expected values for both published messages, preserving the existing payload and TopicMessage type assertions.crates/taskito-core/tests/rust/storage_tests.rs (1)
1356-1385: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd coverage for the
expires_atretention path.Both log tests only exercise cursor-based purge; nothing publishes with
expires_ator asserts TTL-based deletion. That leaves theTopicMessage.expires_at"safety net" untested — and it's exactly the path that behaves differently across backends (Diesel deletes expired regardless of ack; Redis ignores it). A test settingexpires_atin the past and asserting purge behavior would document/guard that contract.Want me to draft this test case?
🤖 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 1356 - 1385, Extend the topic purge tests around test_topic_log_purge to publish a message with TopicMessage.expires_at set in the past, then invoke purge_topic_messages and assert the documented TTL-based deletion behavior. Ensure the test covers an expired message independently of cursor acknowledgements and makes the backend contract explicit, including the expected behavior for expired messages across implementations.crates/taskito-core/src/storage/redis_backend/pubsub.rs (1)
648-681: 🚀 Performance & Scalability | 🔵 Trivial | ⚖️ Poor tradeoff
topic_log_statsstill scans the full stream suffix (crates/taskito-core/src/storage/redis_backend/pubsub.rs:650-671).XRANGEreturns every matching entry here, so large backlogs pay O(unacked) network/memory cost just to computelagand the oldest age. If this path matters, move the count server-side or keep a separate lag index.🤖 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 648 - 681, Update topic_log_stats to avoid fetching the full unacknowledged stream suffix through XRANGE. Use a server-side count/range operation or an existing lag index to compute lag and oldest_unacked_age_ms with bounded response size, while preserving cursor handling and the existing TopicLogStats output.
🤖 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/BINDING_CONTRACT.md`:
- Around line 148-162: Update read_topic_messages and ack_topic_cursor to
require mode = "log" and validate the cursor belongs to the requested topic
before reading or acknowledging. For acknowledgements, atomically verify cursor
membership before advancing the stored cursor, preserving no-op behavior for
older or equal cursors; apply these checks consistently in both Diesel and Redis
implementations.
In `@crates/taskito-core/src/storage/redis_backend/pubsub.rs`:
- Around line 683-724: Update purge_topic_messages to perform an expiry-based
cleanup using TopicMessage.expires_at in addition to the existing minimum-cursor
XTRIM pass. Ensure expired Redis log messages are removed independently of
subscription cursor state, including NULL or stalled cursors, while preserving
the current cursor-based compaction behavior and removal count reporting.
In `@docs/content/docs/shared/guides/core/pubsub.mdx`:
- Around line 518-521: Scope the operational notes to fan-out subscriptions: in
docs/content/docs/shared/guides/core/pubsub.mdx lines 518-521, replace the
generic durable-subscriber wording with fan-out-specific wording or explicitly
distinguish log cursor and stored-message behavior; in lines 527-531, restrict
the write-amplification statement to fan-out subscribers while preserving the
log publishing behavior.
In `@sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java`:
- Around line 238-268: Update the default mode-aware registerSubscription method
to preserve delegation only for mode "fanout"; when mode is "log", reject the
request explicitly instead of calling the legacy fan-out overload. Keep existing
fan-out compatibility and ensure unsupported or invalid modes fail clearly
without silently changing subscription semantics.
---
Outside diff comments:
In `@crates/taskito-core/src/pubsub.rs`:
- Around line 75-105: Make the publishing flow around publish_message and
enqueue_batch/enqueue_unique_batch atomic by introducing a backend-level
operation that performs the log append and all fan-out enqueues as one unit.
Implement the operation transactionally for Diesel and as a single atomic Redis
operation, then update the caller to use it while preserving log-only and
fan-out-only behavior.
In `@crates/taskito-core/src/storage/sqlite/pubsub.rs`:
- Around line 32-52: Validate sub.mode in the subscription registration flow
before constructing or persisting the row, accepting only the two shared mode
constants used by publish_to_topic and rejecting any other value. Apply the same
validation across every backend’s registration implementation, including the
SQLite path around topic_subscriptions, without changing valid log or fan-out
behavior.
---
Nitpick comments:
In `@crates/taskito-core/src/storage/redis_backend/pubsub.rs`:
- Around line 648-681: Update topic_log_stats to avoid fetching the full
unacknowledged stream suffix through XRANGE. Use a server-side count/range
operation or an existing lag index to compute lag and oldest_unacked_age_ms with
bounded response size, while preserving cursor handling and the existing
TopicLogStats output.
In `@crates/taskito-core/tests/rust/storage_tests.rs`:
- Around line 1356-1385: Extend the topic purge tests around
test_topic_log_purge to publish a message with TopicMessage.expires_at set in
the past, then invoke purge_topic_messages and assert the documented TTL-based
deletion behavior. Ensure the test covers an expired message independently of
cursor acknowledgements and makes the backend contract explicit, including the
expected behavior for expired messages across implementations.
In `@sdks/python/tests/core/test_pubsub_log.py`:
- Around line 12-21: Update test_publish_stores_one_message_per_call to also
assert each decoded TopicMessage’s metadata and notes fields returned by
read_topic(). Verify the expected values for both published messages, preserving
the existing payload and TopicMessage type assertions.
🪄 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: f16dcd80-cc66-4fc0-90fb-baa40cd852e9
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (44)
Cargo.tomlcrates/taskito-core/BINDING_CONTRACT.mdcrates/taskito-core/migrations/m0005_topic_messages.rscrates/taskito-core/src/pubsub.rscrates/taskito-core/src/scheduler/maintenance.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/records.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/src/storage/traits.rscrates/taskito-core/tests/rust/storage_tests.rscrates/taskito-java/Cargo.tomlcrates/taskito-java/src/convert.rscrates/taskito-java/src/queue/pubsub.rscrates/taskito-java/src/worker.rscrates/taskito-node/src/convert/mod.rscrates/taskito-node/src/convert/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/model/TopicLogStat.javasdks/java/src/main/java/org/byteveda/taskito/model/TopicMessage.javasdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.javasdks/java/src/test/java/org/byteveda/taskito/core/PubSubLogTest.javasdks/node/src/native.tssdks/node/src/queue.tssdks/node/src/types.tssdks/node/test/core/pubsub-log.test.tssdks/python/taskito/__init__.pysdks/python/taskito/_taskito.pyisdks/python/taskito/mixins/pubsub.pysdks/python/tests/core/test_pubsub_log.py
read_topic_messages/ack_topic_cursor now filter mode='log', so a fan-out subscription can't read a mixed topic's log or advance a cursor. Redis purge_topic_messages gains an expires_at XDEL pass, matching the Diesel TTL safety net so a stalled cursor can't block reclamation.
The mode-aware registerSubscription default silently downgraded 'log' to fan-out; now it throws UnsupportedOperationException for any non-fanout mode, keeping the compatibility fallback for fan-out only.
Backlog accumulation and per-subscriber write amplification are fan-out properties; a log topic writes one row per publish regardless of subscriber count.
This was referenced Jul 19, 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
S28 — the last deferred Tier-4 item. Taskito pub/sub is fan-out-on-write:
publish_to_topicwrites onejobsrow per active subscriber (O(subscribers)/publish). This adds an opt-in log topic mode: one durabletopic_messagesrow per publish (O(1)), consumed by pulling via a per-subscription cursor. Fan-out stays the default; nothing changes unless a subscription opts into log mode. A topic may mix both — a publish stores the log message once and fans out jobs to fan-out subscribers.Consumer contract is at-least-once:
read → process → ack. The cursor is an opaque monotonic high-water mark (acking id X acks everything ≤ X); a consumer that dies before acking re-reads. Retention is min-cursor compaction (a message drops once every log subscriber has acked past it), reaper-gated in the retention sweep.Backends (full parity)
topic_messagestable (migrationm0005) +mode/cursorcolumns ontopic_subscriptions; message id = UUIDv7 (time-ordered cursor key).XADD/XRANGE (cursor +/XTRIM MINID); stream id is the cursor). Enabled therediscrate'sstreamsfeature.SDKs (full parity)
subscribe_log/read_topic/ack_topic/topic_log_stats+TopicMessagedataclass.subscribeLog/readTopic/ackTopic/topicLogStats+TopicMessagetype.subscribeLog/readTopic/ackTopic/topicLogStats+TopicMessage/TopicLogStatmodels (payload asbyte[]— statically-typed decode is the caller's).New
## Topic pub/subsection inBINDING_CONTRACT.md+ log-topic docs across the shared guide and all three API references.Testing
cargo test --workspacegreen (all feature combos); new cross-backend contract teststest_topic_log_messages/test_topic_log_purge.clippy --all-targets --all-features+doc -D warningsclean. The Redis Streams path runs against the CI Redis service (no local hosted URL).test_pubsub_log.py(8) + existing pub/sub (23); ruff + mypy clean.pubsub-log.test.ts(6) + existing (18); tsc + biome clean.PubSubLogTest(7) + existingPubSubTest(14).Deferred (noted in the plan)
Managed consumer loop (auto-invoke + auto-ack), a first-class topics registry (retain messages with zero subscribers at publish time), and per-message ack/redelivery.
Summary by CodeRabbit