fix(publisher, bun)!: keep Redis Pub/Sub delivery in stream order - #2073
Conversation
More templates
@orpc/ai-sdk
@orpc/arktype
@orpc/bun
@orpc/client
@orpc/cloudflare
@orpc/contract
@orpc/experimental-effect
@orpc/evlog
@orpc/hibernation
@orpc/json-schema
@orpc/experimental-lock
@orpc/experimental-msw
@orpc/nest
@orpc/next
@orpc/node
@orpc/openapi
@orpc/opentelemetry
@orpc/pinia-colada
@orpc/pino
@orpc/publisher
@orpc/ratelimit
@orpc/server
@orpc/shared
@orpc/swr
@orpc/tanstack-query
@orpc/trpc
@orpc/valibot
@orpc/zod
commit: |
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
With resume enabled, publish() ran XADD and PUBLISH as two round trips, so concurrent publishers could deliver events out of stream order and a subscriber resuming from its last received ID could skip or repeat events. The Redis adapters now add and publish in one Lua script. BREAKING CHANGE: BaseRedisPublisher subclasses implement evalScript() instead of addStreamEntry(), and RedisStreamTrimOptions is removed.
8584fe9 to
8b41094
Compare
dinwwwh
changed the title
![pullfrog[bot]](https://avatars.githubusercontent.com/in/1768019?s=60&v=4){kind=small}
There was a problem hiding this comment.
Important
The atomic-publish script publishes to KEYS[1], which node-redis prefixes, while Pub/Sub channels are never prefixed — so a client-level keyPrefix silently breaks live delivery on the resume path. Details inline.
Reviewed changes
- Atomic resume publish —
publish()now runs one Lua script (XADD+ optionalXTRIM/EXPIRE+PUBLISH) through a newevalScript()hook, so live Pub/Sub delivery order matches stream order under concurrent publishers. - Adapter port — node-redis, Upstash, and Bun implement
evalScript()instead ofaddStreamEntry();RedisStreamTrimOptionsis removed (breaking). - Cluster support (node-redis) — new structural
RedisPublisherClientaccepts standalone and cluster clients;connectIfNeededdedupes in-flight connects;unsubscribeswallowsClientClosedError. - Tests — concurrent-publish ordering tests added to the Redis, Upstash, and Bun suites; docs note cluster support.
ℹ️ Cluster-only paths have no automated coverage
The new cluster support rests on three things no CI job exercises (CI runs standalone redis:7-alpine): the in-flight connect() dedup in connectIfNeeded, the ClientClosedError swallow in unsubscribe, and the claim that createCluster() clients satisfy RedisPublisherClient. Manual testing against a real cluster (as the PR describes) is reasonable, but these paths can regress unnoticed. Two cheap additions lock in the parts that do not need a live cluster: a *.test-d.ts asserting createCluster() is assignable to RedisPublisherClient (the lock/ratelimit adapters encode cluster support as an explicit RedisClientType | RedisClusterType union rather than a structural interface), and a unit test using a stub client whose isOpen stays false while connect() is pending, asserting concurrent callers await a single connect.
Technical details
# Lock the type-level and dedup guarantees without a live cluster
## Affected sites
- `packages/publisher/src/adapters/redis.ts:11-20` — `RedisPublisherClient`; no type test asserts cluster assignability.
- `packages/publisher/src/adapters/redis.ts:92-102` — `pendingConnects` dedup, only reachable with a cluster-like client whose `isOpen` is `true` before topology is ready.
- `packages/publisher/src/adapters/redis.ts:61-72` — `ClientClosedError` swallow.
## Required outcome
- A failing change to the interface (e.g. requiring an extra member cluster clients lack) or to the dedup logic should be caught by CI.
## Suggested approach
- `*.test-d.ts` with `const client: RedisPublisherClient = createCluster({...})`.
- A unit test passing a stub implementing `RedisPublisherClient` where `connect()` resolves after a tick while `isOpen` remains `false`, then firing concurrent `publish()` calls and asserting `connect()` ran once.DeepSeek Flash (free via Pullfrog for OSS) ｜ 𝕏
There was a problem hiding this comment.
ℹ️ No new issues in this delta — the node-redis cluster work was reverted, narrowing the PR to the atomic-publish fix. The
keyPrefixfinding from the prior review is unaffected by that revert; it is not repeated here.
Reviewed changes
Reviewed the delta since the prior pullfrog review (8584fe9 → 8b41094): the cluster-support changes are gone, leaving only the atomic-publish fix.
- Cluster support reverted —
RedisPublisherClient, the in-flightconnect()dedup (pendingConnects), theClientClosedErrorswallow inunsubscribe, and thexReadcast are all removed;redis.tsis back toRedisClientTypewith the originalconnectIfNeeded/unsubscribe. Thepublisher.mdxcluster comment is removed too. - Atomic publish retained —
PUBLISH_SCRIPT+evalScript()inbase-redis.ts, the three adapter ports, and the concurrent-publish tests are unchanged from the prior revision. - No new issues from the revert — the restored
connectIfNeeded/unsubscribematch the base behavior, which is correct for standalone clients (a standaloneconnect()flipsisOpensynchronously, so concurrent callers do not double-connect).
DeepSeek Flash (free via Pullfrog for OSS) ｜ 𝕏
…fixes keys node-redis applies a client keyPrefix to EVAL keys but not to Pub/Sub channels, so publishing to KEYS[1] sent events to a channel no subscriber listened on. The script now takes the channel as an argument.
There was a problem hiding this comment.
✅ No new issues found.
Reviewed changes
Reviewed the delta since the prior pullfrog review (8b41094 → da88065): the new commit resolves the keyPrefix regression flagged in review 5301495850, with no other behavioral changes.
keyPrefixfix —PUBLISH_SCRIPTnow takes the raw channel asARGV[1]and publishes to it, whileKEYS[1]remains the (client-prefixed) stream key forXADD/XTRIM/EXPIRE. The argument list shifted by one soXTRIM MINIDstill receivesexactnessthenminId.- Regression test — a new node-redis test publishes and resumes against a client created with
keyPrefix, asserting both live delivery and resume-on-the-unprefixed-channel; it fails against the previous revision.
The concern raised in the prior review is resolved: the stream stays under the prefixed key (so readStreamEntries keeps matching) and PUBLISH targets the unprefixed channel subscribers listen on.
DeepSeek Flash (free via Pullfrog for OSS) ｜ 𝕏
Trim times are inserted in time order, so publish() no longer walks every channel it has used to find expired entries. Also shortens the publish script comment and closes the keyPrefix test's subscriber on failure.
There was a problem hiding this comment.
✅ No new issues found.
Reviewed changes
Reviewed the delta since the prior pullfrog review (da88065 → 81ef9b5), a performance follow-up on top of the atomic-publish fix.
- Trim-time pruning short-circuits —
publish()now breaks out of thelastTrimTimessweep at the first still-live entry instead of scanning every channel each publish. keyPrefixregression test cleanup — the test now passes an explicitprefixedRedis.duplicate()subscriber and destroys it viaonTestFinished, replacing the(publisher as any).subscriber.destroy()poke.PUBLISH_SCRIPTJSDoc — condensed to the script's argument layout; the old two-round-trip description is gone.
The early break is sound: lastTrimTimes only ever sets a key when it is absent and deletes expired entries from the front, so Map iteration order stays ascending by first-trim time — the first live entry implies every later one is live too. The prior keyPrefix thread remains resolved and no other threads are open.
DeepSeek Flash (free via Pullfrog for OSS) ｜ 𝕏
The publish script now uses KEYS[1] as the channel too, dropping the extra argument. RedisPublisher adds the client's keyPrefix to channels so they still match, which also isolates channels for keyPrefix users. The prefix option docs now say it covers Pub/Sub channels as well as keys.
There was a problem hiding this comment.
✅ No new issues found.
Reviewed changes
Reviewed the delta since the prior pullfrog review (81ef9b5 → 4b0069e): a deliberate redesign of node-redis keyPrefix handling that replaces the previous raw-channel publish with channel prefixing that matches the stream key.
- Channels now carry the client
keyPrefix—PUBLISH_SCRIPTpublishes onKEYS[1](prefixed by node-redis) instead of a rawARGVchannel, andRedisPublisherreadsredis.options.keyPrefixinto achannelPrefixapplied to bothpublishMessage()andsubscribeChannel(), so publish, subscribe,XADD, andXREADall resolve tokeyPrefix + prefix + event. - Script args shift —
ARGV[1]is now the payload;ARGV[2..4]stay exactness/minId/expireSeconds. keyPrefixtest extended — adds a non-resume publisher (covering thepublishMessage()path) alongside the resume path, with an explicit duplicated subscriber.- Docs/JSDoc — the
prefixoption now states it namespaces Pub/Sub channels as well as keys.
The design is internally consistent: verified against the installed @redis/client@6.2.1 that EVAL prefixes keys while PUBLISH/SUBSCRIBE do not, so the explicit channelPrefix is both necessary and correctly applied, and redis.options.keyPrefix is populated from createClient({ keyPrefix }). tsc -b packages/publisher and eslint pass.
ℹ️ Rolling deploy loses events in both directions
The breaking change (channels carry the client keyPrefix) means a mixed-version fleet loses events in both directions, not just the one the PR body calls out. This only affects users who set a node-redis client keyPrefix.
Technical details
# Channel name changes under a client keyPrefix
## Affected sites
- `packages/publisher/src/adapters/redis.ts:38` — `channelPrefix` sourced from `redis.options.keyPrefix`.
- `packages/publisher/src/adapters/redis.ts:43`, `:47` — publish/subscribe on `keyPrefix + prefix + event`.
- `packages/publisher/src/adapters/base-redis.ts:15` — script `PUBLISH KEYS[1]` (prefixed).
## Required outcome
- None for the code; the change is intentional and matches the documented breaking note.
## Open questions for the human
- Confirm the release/migration note covers both directions: a new subscriber misses an old publisher's events, and an old subscriber misses a new publisher's events. The PR body mentions only the latter.DeepSeek Flash (free via Pullfrog for OSS) ｜ 𝕏
Channels get the client's keyPrefix as text, so a Buffer prefix with non-UTF-8 bytes would no longer match the stream key and live delivery would silently stop. The constructor now throws a TypeError instead.
There was a problem hiding this comment.
ℹ️ No critical issues — one minor suggestion inline.
Reviewed changes
Reviewed the delta since the prior pullfrog review (4b0069e → 0480e48): a single constructor guard in RedisPublisher.
- Reject non-string
keyPrefix—RedisPublishernow throws aTypeErrorat construction whenredis.options.keyPrefixis defined but not a string, checked beforechannelPrefixis derived and beforeredis.duplicate()runs. - Regression test — a unit test asserts the constructor throws for a
BufferkeyPrefix.
DeepSeek Flash (free via Pullfrog for OSS) ｜ 𝕏
|
|
||
| const keyPrefix = redis.options?.keyPrefix | ||
|
|
||
| if (keyPrefix !== undefined && typeof keyPrefix !== 'string') { |
There was a problem hiding this comment.
node-redis types keyPrefix as RedisArgument (string | Buffer) and its parser prefixes keys from a Buffer via Buffer.concat, so a Buffer keyPrefix is type-legal and previously worked: the prior revision derived the channel name with String(keyPrefix), which matches the prefixed key bytes for UTF-8. This guard turns that into a construction-time TypeError, an additional runtime restriction that is not listed in the PR body's breaking-changes section.
Technical details
# Non-string `keyPrefix` now throws at construction
## Affected sites
- `packages/publisher/src/adapters/redis.ts:39` — guard rejects any defined non-string `keyPrefix`.
- `packages/publisher/src/adapters/redis.ts:44` — `channelPrefix = keyPrefix ?? ''`.
## Evidence
- `@redis/client@6.2.1`: `RedisClientOptions.keyPrefix?: RedisArgument`, `RedisArgument = string | Buffer` (`dist/lib/RESP/types.d.ts:74`).
- `prefixKey()` (`dist/lib/client/parser.js:11`) uses `Buffer.concat` when either side is a `Buffer`, so a `Buffer` prefix is a supported runtime path.
## Required outcome
- A `Buffer` keyPrefix either keeps working (UTF-8) or is called out as now throwing in the release notes / breaking-changes list.
## Open questions for the human
- Is rejecting `Buffer` intentional? If so, add it to the PR body's breaking-changes list.
1 participant
With resume enabled,
publish()added the stream entry and published it in two separate round trips, so concurrent publishers could deliver events out of stream order. A subscriber resuming from its last received ID could then skip an event or receive one twice. The Redis adapters now add and publish in a single Lua script, so live delivery always matches stream order.Fixes
Performance
publish()no longer scans every channel it has used to prune trim times. Measured per publish: 118 µs → 0.06 µs with 10,000 channels, 1.2 ms → 0.04 µs with 100,000.Breaking changes
BaseRedisPublishersubclasses implementevalScript()instead ofaddStreamEntry(), andRedisStreamTrimOptionsis removed.keyPrefix, Pub/Sub channels now carry the prefix too, so they are isolated like keys. During a rolling deploy, subscribers on older versions miss events from newer publishers.RedisPublisherthrows aTypeErrorwhen the client'skeyPrefixis not a string. ABufferprefix could hold bytes that are not valid text, so channels would stop matching the stream key.Docs
prefixoption now states it namespaces Pub/Sub channels as well as keys.Testing
keyPrefixtests cover live delivery with and without resume, resume itself, and rejecting aBufferprefix.