redis: implement psubscribe/punsubscribe with listener routing - #35521
redis: implement psubscribe/punsubscribe with listener routing#35521robobun wants to merge 8 commits into
Conversation
WalkthroughChangesPattern subscriptions
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
Comment |
|
Mock-server tests in All automated review feedback addressed. CI build 80153: the 80 test lanes that ran all pass; no valkey-related failures. The red is build infrastructure unrelated to this diff:
Ready for review. |
|
Updated 2:19 AM PT - Jul 25th, 2026
❌ @robobun, your commit b59249e has 1 failures in
🧪 To try this PR locally: bunx bun-pr 35521That installs a local version of the PR into your bun-35521 --bun |
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
There was a problem hiding this comment.
Beyond the two inline nits, I also checked: (1) the is_reply_kind change dropping the p prefix is safe — psubscribe/punsubscribe/pmessage now match via from_bytes directly, and ssubscribe/sunsubscribe still route through is_reply_kind, so promise-pair consumption in handle_response is unchanged; (2) the new patternCallbackMap slot is declared in values in valkey.classes.ts (so it's GC-visited) and cleared alongside subscriptionCallbackMap in SubscriptionCtx::close.
Extended reasoning...
This PR adds user-facing API (psubscribe/punsubscribe with listener routing), a second GC-rooted JSMap on the RedisClient wrapper, and refactors the existing subscribe/unsubscribe host functions into shared helpers parametrized by SubscriptionKind. The two findings are cosmetic nits (docs typo, two dead predicate helpers). I separately verified that the is_reply_kind narrowing preserves push-frame routing semantics for both pattern and sharded pub/sub kinds, and that the new cached-value slot is wired into codegen's values list and the context teardown path. Deferring because this is a non-trivial feature/API change, not because of correctness concerns.
There was a problem hiding this comment.
Beyond the inline nit, I also checked: (1) the is_reply_kind narrowing from p|s → s only — the p* push kinds now hit from_bytes first so promise-pair routing in handle_response still consumes the pair for psubscribe/punsubscribe acks; (2) the new patternCallbackMap slot is declared in valkey.classes.ts values so it's GC-visited, and it's cleared alongside subscriptionCallbackMap in SubscriptionCtx::close().
Extended reasoning...
This run's only new finding is a cosmetic nit (hard-coded command name in two rare send-failure error strings). The two nits from the previous run — the docs UNPSUBSCRIBE typo and the dead is_subscribe/is_unsubscribe helpers — were addressed in 460fd14. The PR itself is a substantial feature (new user-facing API surface, second GC-rooted JSMap, refactor of existing subscribe/unsubscribe into shared helpers, protocol push-frame routing changes), so it should get a human look regardless; the note above records the two non-obvious correctness concerns I traced through and ruled out so a human reviewer doesn't have to re-derive them.
psubscribe() was a first-class prototype method but a silent no-op: it took no listener, resolved with the raw RESP3 push frame as its value, and every pmessage the server delivered afterward was dropped with no callback and no error. This gives psubscribe(pattern, listener) and punsubscribe() the same treatment as subscribe/unsubscribe: - SubscriptionPushMessage now recognizes pmessage/psubscribe/punsubscribe - SubscriptionCtx keeps a second JSMap keyed by pattern so pattern and literal-channel listeners cannot collide and punsubscribe() only clears pattern listeners - pmessage pushes are routed to the pattern listener with (message, channel) arguments, matching node-redis v4 pSubscribe - subscribe/unsubscribe are refactored into shared helpers parametrized by SubscriptionKind so psubscribe/punsubscribe share the same validation, error messages and rollback path - fixed UNPSUBSCRIBE typo in the auto-pipelining disallow list
…ibe/punsubscribe name the right command
b323c67 to
730c58c
Compare
…dx subscriber-mode note
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
![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
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/runtime/valkey_jsc/js_valkey_functions.rs (1)
1716-1725: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winRollback on subscribe failure wipes unrelated, already-active subscriptions.
On
send()failure,clear_all_receive_handlers(kind, global)callsmap.clear()— this empties every channel/pattern currently registered for that kind, not just the ones added by this call. If the client already has other working subscriptions of the same kind, a later failedsubscribe()/psubscribe()(e.g. OOM inEntry::create) silently drops all of them client-side, even though Redis was never told to unsubscribe. This now affects bothsubscribeandpsubscribesince the rollback is shared.🛠️ Proposed fix: track and roll back only the channels/patterns added in this call
fn do_subscribe( this: &Self, global: &JSGlobalObject, frame: &CallFrame, kind: SubscriptionKind, fn_name: &'static str, arg_name: &'static str, redis_command: &'static [u8], err_msg: &'static str, ) -> JsResult<JSValue> { let _guard = this.ref_scope(); let [channel_or_many, handler_callback] = frame.arguments_as_array::<2>(); let mut redis_channels: Vec<JSArgument> = Vec::with_capacity(1); + let mut registered_channels: Vec<JSValue> = Vec::with_capacity(1); ... redis_channels.push(channel); this._subscription_ctx.get().upsert_receive_handler( kind, global, channel_arg, handler_callback, )?; + registered_channels.push(channel_arg); ... redis_channels.push(channel); this._subscription_ctx.get().upsert_receive_handler( kind, global, channel_or_many, handler_callback, )?; + registered_channels.push(channel_or_many); ... let promise = match this.send(global, frame.this(), &command) { Ok(p) => p, Err(err) => { - // If we catch an error, we need to clean up any handlers we may have added and fall out of subscription mode - this._subscription_ctx - .get() - .clear_all_receive_handlers(kind, global)?; + // Only remove the handlers we just added in this call — other + // active subscriptions for this `kind` must be left intact. + for ch in ®istered_channels { + let _ = this._subscription_ctx.get().remove_receive_handler( + kind, + global, + *ch, + handler_callback, + ); + } return send_err_to_js(global, err_msg, &err); } };🤖 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 `@src/runtime/valkey_jsc/js_valkey_functions.rs` around lines 1716 - 1725, Update the shared subscribe/psubscribe failure rollback around send and the subscription context handlers so it removes only channels or patterns registered by the current call, rather than invoking clear_all_receive_handlers, which clears the entire kind map. Track the entries added during this call and roll back those entries on send failure while preserving existing active subscriptions.
🤖 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 `@test/js/valkey/valkey.test.ts`:
- Around line 7137-7151: Remove the : any annotations from all three RedisClient
declarations in the psubscribe tests, including the sites surrounding
“psubscribe without a listener throws” and “psubscribe delivers pmessage pushes
to its listener,” so they use inferred RedisClient types and validate the new
psubscribe/punsubscribe declarations.
---
Outside diff comments:
In `@src/runtime/valkey_jsc/js_valkey_functions.rs`:
- Around line 1716-1725: Update the shared subscribe/psubscribe failure rollback
around send and the subscription context handlers so it removes only channels or
patterns registered by the current call, rather than invoking
clear_all_receive_handlers, which clears the entire kind map. Track the entries
added during this call and roll back those entries on send failure while
preserving existing active subscriptions.
🪄 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: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: d28b423a-fc41-4ad0-be1e-7e7d450456b5
📒 Files selected for processing (11)
docs/runtime/redis.mdxpackages/bun-types/redis.d.tssrc/runtime/valkey_jsc/ValkeyCommand.rssrc/runtime/valkey_jsc/js_valkey.rssrc/runtime/valkey_jsc/js_valkey_functions.rssrc/runtime/valkey_jsc/valkey.classes.tssrc/runtime/valkey_jsc/valkey.rssrc/valkey/valkey_protocol.rstest/js/valkey/reliability/resp-nesting-depth.test.tstest/js/valkey/test-utils.tstest/js/valkey/valkey.test.ts
There was a problem hiding this comment.
No issues found, but this adds new public API surface (psubscribe(pattern, listener) / punsubscribe() with the node-redis-v4 (message, channel) callback shape) and changes psubscribe's arity/behavior, so worth a human sign-off on the API.
What was reviewed:
is_reply_kindnarrowing tos-prefix only is safe —p*kinds now hitfrom_bytesfirst, so theNonearm inhandle_responsenever sees them.patternCallbackMapis added to thevalues:array invalkey.classes.ts, so it's GC-visited alongsidesubscriptionCallbackMap;SubscriptionCtx::closeclears both.channels_subscribed_to_countsums both maps, soremove_subscription/has_subscriptionscorrectly keep subscriber mode when only pattern subs remain (covered by the mixed subscribe+psubscribe test).- The send-failure rollback now removes only this call's listeners via
remove_receive_handler, notclear_all_receive_handlers.
Extended reasoning...
Overview
This PR promotes psubscribe / punsubscribe from arity-0 cmd_strings_varargs! no-ops to first-class subscriber-mode methods with listener routing, mirroring the existing subscribe / unsubscribe implementation. It touches 11 files: the RESP protocol enum (valkey_protocol.rs) gains Pmessage/Psubscribe/Punsubscribe variants and an is_message() predicate; SubscriptionCtx in js_valkey.rs grows a second JSMap keyed by pattern with a SubscriptionKind enum threaded through every accessor; js_valkey_functions.rs refactors subscribe/unsubscribe into shared do_subscribe/do_unsubscribe helpers parametrized by kind/fn_name/arg_name/redis_command/err_msg; valkey.rs routes pmessage pushes and pairs Psubscribe/Punsubscribe with the existing subscribe/unsubscribe arms; valkey.classes.ts adds patternCallbackMap to the visited values and sets length: 2 on both subscribe methods. Docs, .d.ts, and ~280 lines of tests (mock-server + real-Redis) round it out. Also fixes the UNPSUBSCRIBE typo in AUTO_PIPELINE_DISALLOWED_COMMANDS.
Security risks
None identified. The change is confined to pub/sub listener bookkeeping and RESP push-frame routing on an already-authenticated connection. No new parsing of untrusted lengths (the pmessage frame's element indices are bounds-checked via value.len() <= message_idx before indexing). The pattern string is passed through to Redis verbatim as it already was.
Level of scrutiny
Medium-high. This is user-facing API design (callback signature, arity, error messages, return value of the promise) plus a behavioral change to an existing prototype method — psubscribe() now throws ERR_INVALID_ARG_TYPE without a listener where it previously returned the raw push frame. The refactor of subscribe/unsubscribe into shared helpers is behavior-preserving as far as I can trace (all error strings parametrized, rollback narrowed to per-call listeners which is strictly better than the old clear_all_receive_handlers), but it's ~170 lines of moved/reshaped code in a path with re-entrant JS callbacks and refcounted lifetime management. The second cached JSMap is correctly declared in values: so codegen emits the visitor, and SubscriptionCtx::init/close handle both slots symmetrically.
Other factors
Four rounds of prior review feedback (docs typo mirror, dead is_subscribe/is_unsubscribe helpers, hard-coded error strings in shared helpers, unhandled-rejection in the mock punsubscribe test, stale subscribe() JSDoc, comment-cop flags, : any in tests) have all been addressed and the threads resolved. Test coverage is solid: 4 mock-server tests exercise the listener path without docker (so they run in every CI lane), plus 5 real-Redis integration tests covering single/array patterns, targeted punsubscribe(pattern) and punsubscribe(pattern, listener), and mixed channel+pattern subscriptions on one client. The resp-nesting-depth.test.ts update confirms the arity change is intentional.
Deferring because new public API shape and a breaking arity change on an existing method are decisions a maintainer should confirm, not because anything looks wrong.
…s tables (#39271) ### Problem - `class RedisClient` in `packages/bun-types/redis.d.ts` is a hand-written copy of the `proto` table in `src/runtime/valkey_jsc/valkey.classes.ts`, and nothing compares the two. A command added to the table and not to the d.ts works at runtime and fails to type-check (`Property 'pubsub' does not exist on type 'RedisClient'`). - On main, 5 of the 167 table entries are undeclared: `psubscribe`, `pubsub`, `punsubscribe`, `script`, `select`. All five have been registered since the commit that introduced `Bun.redis` (ec87a27, #18812) and stayed undeclared through the 14 redis.d.ts commits since, including the 614-line batch in #23116, so the gap is systematic rather than a one-off. - Each of the five is already being declared by an open PR: `pubsub` and `select` by #39208, `script` by #29339, `psubscribe`/`punsubscribe` by #35521 (which also adds the listener routing they are missing; today `psubscribe()` drops every pmessage). This PR adds the check, not the declarations. ### Fix - `test/internal/source-lints/redis-client-types.test.ts` imports `valkey.classes.ts` (the module the codegen reads; `class-definitions.ts` has no dependencies), collects the members the codegen installs (`proto` entries by name, `klass` entries as `static name`, `constructor` when `construct` is set, skipping `internal`/`privateSymbol`/`publicSymbol` entries, which the codegen does not install under an identifier), parses the member names declared in the `class RedisClient` body of `redis.d.ts` (`[Symbol.x]` maps onto the table's `@@x` spelling), and requires the two sets to be equal. Each direction is its own test, so a table entry without a declaration and a declaration without a table entry are both reported by name. - The five names missing today sit in a `pendingDeclarations` table keyed by the PR that declares each. A fourth test fails as soon as a listed name is declared or unregistered, so the entry gets deleted when its PR lands (whichever of this PR and #39208/#29339/#35521 lands second trips it on rebase, and the message says which entry to delete). A name that is neither declared nor listed fails the lint outright. - Declaration lines the parser does not recognize are reported as failures rather than skipped, so a new d.ts shape cannot silently hide a member from the comparison. - `.github/workflows/source-lints.yml` gains `src/**/*.classes.ts`, `src/codegen/class-definitions.ts` and `packages/bun-types/redis.d.ts` as triggers: the workflow is path-filtered and those are the files this lint reads, so without them the edits it guards would not run it. The directory README now states that rule; a comment on the `proto` table points at the lint. - Scope is RedisClient only. A checker for every `*.classes.ts` needs a class-to-declaration mapping and inheritance handling and is a separate project. - Verified: - `bun test test/internal/source-lints/redis-client-types.test.ts` passes on this branch (4 tests); the whole directory is 170 green. - With `pendingDeclarations` emptied, the lint fails against main's files naming exactly `psubscribe`, `pubsub`, `punsubscribe`, `script`, `select` (output below). - Simulated the eight drift shapes in the details block (new table entry, pending name declared, phantom declaration, pending name unregistered, unparseable member, `@@asyncDispose` + a `klass` entry with and without declarations, `internal`/`privateSymbol` entries); each fails the intended test or passes as intended. - No runtime code changes: the `valkey.classes.ts` hunk is a comment and produces identical codegen output. This PR has no src/packages diff for a fail-before run to strip; the fail-before evidence is the emptied-pending-table run above. ### Background - `*.classes.ts` files are the input of `src/codegen/generate-classes.ts`. `define({ proto, klass, construct })` describes a native class: `proto` entries become properties of the prototype (`fn` methods, `getter`/`setter` accessors), `klass` entries become statics, and `construct: true` makes it newable. Entries with `internal`, `privateSymbol` or `publicSymbol` are installed under private names or `Symbol.for()` symbols (or not at all), and keys spelled `@@x` are installed under the well-known symbol `Symbol.x`. - `packages/bun-types` is the published `@types/bun` surface; it is hand-written, not generated from the class definitions, which is why it can drift. - `test/internal/source-lints/` holds tests that only read the source tree; `.buildkite/ci.mjs` excludes the directory from the binary lanes and `source-lints.yml` runs it against a released bun on a bare checkout, so tests there may only import built-ins and relative paths. <details> <summary>Lint output against main's files with the pending table emptied, and the simulated drift shapes</summary> ``` (pass) every member of class RedisClient in packages/bun-types/redis.d.ts has a shape this lint can read (fail) packages/bun-types/redis.d.ts declares every RedisClient member src/runtime/valkey_jsc/valkey.classes.ts installs - [] + [ + "psubscribe", + "pubsub", + "punsubscribe", + "script", + "select", + ] (pass) packages/bun-types/redis.d.ts declares no RedisClient member src/runtime/valkey_jsc/valkey.classes.ts does not install (pass) pendingDeclarations lists only members that are still registered and still undeclared ``` ``` ### proto gains waitaof, d.ts untouched + "waitaof", (fail) packages/bun-types/redis.d.ts declares every RedisClient member src/runtime/valkey_jsc/valkey.classes.ts installs ### d.ts declares select while it is still pending + "select (#39208) is declared in packages/bun-types/redis.d.ts now; delete its entry", (fail) pendingDeclarations lists only members that are still registered and still undeclared ### d.ts declares flushall, which is not registered + "flushall", (fail) packages/bun-types/redis.d.ts declares no RedisClient member src/runtime/valkey_jsc/valkey.classes.ts does not install ### proto drops script while it is still pending + "script (#29339) is no longer registered in src/runtime/valkey_jsc/valkey.classes.ts", (fail) pendingDeclarations lists only members that are still registered and still undeclared ### d.ts gains `private brand: never;` + "private brand: never;", (fail) every member of class RedisClient in packages/bun-types/redis.d.ts has a shape this lint can read ### proto gains "@@asyncDispose" and a klass entry, d.ts declares [Symbol.asyncDispose]() and the static 4 pass ### same, with nothing declared + "@@asyncDispose", + "static parseURL", (fail) packages/bun-types/redis.d.ts declares every RedisClient member src/runtime/valkey_jsc/valkey.classes.ts installs ### proto gains an `internal: true` entry and a `privateSymbol` entry, d.ts untouched 4 pass ``` </details>
2 participants
Problem
RedisClient.prototype.psubscribe()was a first-class prototype method but a silent no-op: it took no listener (arity 0), resolved with the raw RESP3>psubscribepush frame as its return value, and everypmessagethe server delivered afterward was dropped with no callback and no error. The client stayed connected, so pattern subscriptions appeared to work but lost 100% of their messages.send("PSUBSCRIBE", ...)on a client that already had a realsubscribe()listener would additionally never settle and zombie the connection, becausepmessage/psubscribepush kinds were unknown to the subscriber handler.Fix
Give
psubscribe(pattern, listener)andpunsubscribe()the same listener routing thatsubscribe/unsubscribealready have:SubscriptionPushMessagenow recognizespmessage/psubscribe/punsubscribe.SubscriptionCtxkeeps a secondJSMapkeyed by pattern so pattern and literal-channel listeners cannot collide, andpunsubscribe()with no arguments clears only pattern listeners.pmessagepushes are routed to the pattern listener with(message, channel)arguments, matching node-redis v4pSubscribe.subscribe/unsubscribeare refactored into shared helpers parametrized bySubscriptionKind, sopsubscribe/punsubscribeshare the same validation, error messages and rollback path.UNPSUBSCRIBEtypo in the auto-pipelining disallow list.psubscribenow has arity 2 and throwsERR_INVALID_ARG_TYPEwhen called without a listener.Verification
RedisClient PSUBSCRIBE (mock server)block invalkey.test.ts(4 tests, self-contained RESP3 stub, no docker required): fails on current main, passes with this change.PUB/SUBblock covering single/array patterns,punsubscribe(pattern),punsubscribe(pattern, listener), and mixedsubscribe+psubscribeon one client.resp-nesting-depth.test.tsto pass a listener now that one is required.[review] gate passed · iteration 2 · 11 files touched
fails on main (without fix)
passes on PR (with fix)
diff hotspot
gate history · 3 passed · 2 rejected · iteration 2
evidence per changed file