feat: pubsub worker (builtin migration) - #394
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 WalkthroughWalkthroughAdds a new Rust "pubsub" worker crate implementing topic-based publish/subscribe messaging with pluggable local and Redis adapters, hub-based subscription tracking with hot-swap support, trigger/function bridging, configuration integration, CLI entrypoint, documentation, and e2e/unit tests. Updates CI workflows to recognize the new worker. ChangesPubsub worker implementation
Estimated code review effort: 4 (Complex) | ~75 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant Engine
participant PubsubBoot as Pubsub Worker
participant Hub
participant Adapter as Local/Redis Adapter
participant Subscriber as Subscriber Function
Client->>Engine: trigger("subscribe", topic, function_id)
Engine->>PubsubBoot: register_trigger
PubsubBoot->>Hub: subscribe(id, topic, function_id)
Hub->>Adapter: subscribe(topic, id, function_id)
Client->>Engine: trigger("publish", {topic, data})
Engine->>PubsubBoot: publish function call
PubsubBoot->>Hub: publish(topic, data)
Hub->>Adapter: publish(topic, data)
Adapter->>Subscriber: invoker.call(function_id, data)
sequenceDiagram
participant ConfigWorker as Configuration Worker
participant PubsubMain as Pubsub Main
participant ConfigHandler as on-config-change
participant Hub
participant NewAdapter
PubsubMain->>ConfigWorker: register_config(schema, seed)
ConfigWorker-->>ConfigHandler: configuration:updated
ConfigHandler->>ConfigWorker: fetch_config
ConfigHandler->>ConfigHandler: swap_needed check
ConfigHandler->>NewAdapter: build_adapter(new config)
ConfigHandler->>Hub: swap_adapter(new adapter)
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
…ype, publish function
…e stamps caller id)
77670ee to
890255d
Compare
skill-check — worker0 verified, 33 skipped (no docs/).
Four for four. Nicely done. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
pubsub/src/configuration.rs (1)
80-93: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winPermanent
NOT_FOUNDerrors are retried like transient failures.
try_get_config_valueroutes everyconfiguration::getcall throughtrigger_with_retry, which retries all errors (including a legitimate "not found" for a brand-new install) up to 3 times with backoff. On a fresh boot this cost is paid twice (once viashould_seed_initial_value, once viafetch_config), adding ~1.5s of pure retry-backoff latency for a case that isn't transient at all.Separately, the
NOT_FOUNDdetection at Line 90 does substring matching on an error message that has already been reformatted bytrigger_with_retry(Line 219-221:"{function_id} failed after {CONFIG_RETRIES} attempts: {last_err}"). This works only because the original error text happens to nest inside the wrapped one — fragile if the bus's error format changes.Consider a fast-path that treats
NOT_FOUNDas non-retryable insidetrigger_with_retry(or a dedicated helper) rather than exhausting all attempts.Also applies to: 190-222
🤖 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 `@pubsub/src/configuration.rs` around lines 80 - 93, `try_get_config_value` is treating a legitimate `NOT_FOUND` from `configuration::get` as a retriable failure because all calls go through `trigger_with_retry`; update the retry flow so `NOT_FOUND` is classified as non-retryable immediately (either inside `trigger_with_retry` or via a dedicated helper) and returns `Ok(None)` without exhausting retries. Also make the `NOT_FOUND` handling in `try_get_config_value` rely on the original error signal rather than substring matching against the wrapped `function_id`/attempts message, so the logic in `should_seed_initial_value` and `fetch_config` stays fast and robust.pubsub/src/adapters/local.rs (1)
37-55: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winFire-and-forget delivery has no timeout/bound on
invoker.call.Each publish spawns one detached task per subscriber (documented as parity with the builtin). If a subscriber's handler hangs (e.g. blocked on a lock or an external call), that task never completes — repeated publishes to a stuck subscriber will keep spawning new tasks indefinitely, growing memory/task count with no backpressure or timeout. The same pattern recurs in the Redis listener's per-message spawn (
pubsub/src/adapters/redis.rs:118-122).Consider wrapping
invoker.callintokio::time::timeoutso a hung handler doesn't accumulate unbounded background tasks.♻️ Suggested guard
tokio::spawn(async move { - if let Err(e) = invoker.call(&function_id, data).await { - tracing::debug!(function_id = %function_id, error = %e, "pubsub delivery failed"); - } + match tokio::time::timeout(std::time::Duration::from_secs(30), invoker.call(&function_id, data)).await { + Ok(Err(e)) => tracing::debug!(function_id = %function_id, error = %e, "pubsub delivery failed"), + Err(_) => tracing::warn!(function_id = %function_id, "pubsub delivery timed out"), + Ok(Ok(_)) => {} + } });This depends on whether
SdkInvoker(inboot.rs, not in this review batch) already enforces a timeout oncall— if so this is redundant.🤖 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 `@pubsub/src/adapters/local.rs` around lines 37 - 55, The fire-and-forget publish path in local::publish can spawn unbounded detached tasks if invoker.call hangs, so add a timeout guard around the call before logging delivery failures. Update the async task in publish so it waits on tokio::time::timeout with a reasonable bound, and apply the same protection in the Redis listener’s per-message spawn path for parity. Keep the existing tracing::debug failure logging, but make sure timeout errors are handled as delivery failures.pubsub/src/adapters/redis.rs (1)
27-31: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win
ConnectionManagerdoesn't need an externalMutex.
redis::aio::ConnectionManageris explicitly designed to be cloned and used concurrently on the same underlying connection. Wrapping it inArc<Mutex<ConnectionManager>>serializes everypublishcall unnecessarily.♻️ Suggested refactor
pub struct RedisAdapter { - publisher: Arc<Mutex<ConnectionManager>>, + publisher: ConnectionManager, subscriber: Arc<Client>, ... }Then in
publish, useself.publisher.clone()(cheap) instead of locking, per the redis-rs docs: "this manager can be cloned, allowing requests to be sent concurrently on the same underlying connection."Also applies to: 39-57, 62-74
🤖 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 `@pubsub/src/adapters/redis.rs` around lines 27 - 31, The Redis adapter stores `ConnectionManager` behind an unnecessary external mutex, which serializes concurrent publishes. Update `RedisPubSubAdapter` to keep `publisher` as a clonable `ConnectionManager` handle instead of `Arc<Mutex<ConnectionManager>>`, and adjust `publish` to use a cloned manager directly rather than locking. Make the same change anywhere else in `RedisPubSubAdapter` that assumes exclusive access to `publisher`, while leaving `subscriber`, `subscriptions`, and `invoker` unchanged.
🤖 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 `@pubsub/src/adapters/redis.rs`:
- Around line 76-134: In subscribe on RedisAdapter, fix the race by making the
topic check and subscription insert atomic under the same subscriptions write
lock, using the existing subscriptions map and SubscriptionInfo to enforce the
single-listener invariant. Also avoid inserting into self.subscriptions until
the spawned listener has actually connected and subscribed successfully; if
get_async_pubsub or pubsub.subscribe fails inside the tokio::spawn block, do not
record the topic as subscribed so future subscribe calls can retry instead of
being blocked by a ghost entry.
In `@pubsub/src/boot.rs`:
- Around line 51-69: SdkInvoker::call currently forwards TriggerRequest with
timeout_ms set to None, which leaves subscriber fan-out work unbounded. Update
the SdkInvoker implementation to accept or derive a finite timeout and pass it
into TriggerRequest, and make sure the pubsub adapter paths that invoke
SdkInvoker still compile and propagate this timeout consistently so hung
consumers cannot stall spawned tasks indefinitely.
---
Nitpick comments:
In `@pubsub/src/adapters/local.rs`:
- Around line 37-55: The fire-and-forget publish path in local::publish can
spawn unbounded detached tasks if invoker.call hangs, so add a timeout guard
around the call before logging delivery failures. Update the async task in
publish so it waits on tokio::time::timeout with a reasonable bound, and apply
the same protection in the Redis listener’s per-message spawn path for parity.
Keep the existing tracing::debug failure logging, but make sure timeout errors
are handled as delivery failures.
In `@pubsub/src/adapters/redis.rs`:
- Around line 27-31: The Redis adapter stores `ConnectionManager` behind an
unnecessary external mutex, which serializes concurrent publishes. Update
`RedisPubSubAdapter` to keep `publisher` as a clonable `ConnectionManager`
handle instead of `Arc<Mutex<ConnectionManager>>`, and adjust `publish` to use a
cloned manager directly rather than locking. Make the same change anywhere else
in `RedisPubSubAdapter` that assumes exclusive access to `publisher`, while
leaving `subscriber`, `subscriptions`, and `invoker` unchanged.
In `@pubsub/src/configuration.rs`:
- Around line 80-93: `try_get_config_value` is treating a legitimate `NOT_FOUND`
from `configuration::get` as a retriable failure because all calls go through
`trigger_with_retry`; update the retry flow so `NOT_FOUND` is classified as
non-retryable immediately (either inside `trigger_with_retry` or via a dedicated
helper) and returns `Ok(None)` without exhausting retries. Also make the
`NOT_FOUND` handling in `try_get_config_value` rely on the original error signal
rather than substring matching against the wrapped `function_id`/attempts
message, so the logic in `should_seed_initial_value` and `fetch_config` stays
fast and robust.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: c9bce905-a733-4840-b1f9-88e65d4b532c
⛔ Files ignored due to path filters (1)
pubsub/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (21)
.github/workflows/create-tag.yml.github/workflows/release.ymlpubsub/Cargo.tomlpubsub/README.mdpubsub/build.rspubsub/iii.worker.yamlpubsub/skills/SKILL.mdpubsub/src/adapters/local.rspubsub/src/adapters/mod.rspubsub/src/adapters/redis.rspubsub/src/boot.rspubsub/src/config.rspubsub/src/configuration.rspubsub/src/hub.rspubsub/src/lib.rspubsub/src/main.rspubsub/src/manifest.rspubsub/src/trigger.rspubsub/tests/common/engine.rspubsub/tests/common/mod.rspubsub/tests/e2e_pubsub.rs
1 participant
Summary
pubsubregistry worker, replacing the built-iniii-pubsub.subscribetrigger type and the barepublishfunction; boot-time guard refuses to start if the built-iniii-pubsubis still connected.Invokerabstraction:local(in-process, default) andredis(Redis Pub/Sub, cross-instance); selected via thepubsubconfiguration entry with a gated, gap-free adapter hot-swap (re-subscribe before swap).create-tag.yml+release.yml), README parity documentation, the registry skill (pubsub/skills/SKILL.md), and connect-or-skip e2e coverage.Behavior
subscribe; the public function id stays the barepublish(stream bridge compatibility).data(no envelope), fire-and-forget; publish with empty topic fails withtopic_not_set.iii-pubsubworker is still connected.unsubscriberemoves only the given id instead of dropping the whole topic entry (the builtin bug killed co-subscribers). Documented in the README parity table.Validation
cd pubsub && cargo test— 27 unit tests greencd pubsub && cargo fmt --checkcd pubsub && cargo clippy --all-targets -- -D warningscd pubsub && cargo build && ./target/debug/pubsub --manifest | head -5python3 .github/scripts/validate_worker.py --worker pubsub --base-ref main --source-changed '["pubsub"]'— exit 0python3 .github/scripts/build_skills_payload.py --worker pubsub --version 0.1.0— collected 1 skill fileIII_E2E_REQUIRE=1):publishover the bus fans rawdataout to everysubscribetrigger; null result; empty topic →topic_not_settest_python_pubsub/README walkthrough → ALL PASS (fan-out to both subscribers, other-topic isolation, empty-topic error, unregister stops delivery)Notes
feat/http-worker; rebase ontomainonce http lands.pubsub/v0.1.0is post-merge only, via the Create Tag workflow.test_python_pubsub/and is intentionally untracked.Summary by CodeRabbit
New Features
pubsubworker option and release trigger coverage forpubsub/v*tags.Bug Fixes