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) ｜ 𝕏
| * concurrent publishers, so a subscriber resuming from its last received ID skips nothing. | ||
| * Kept on one line because `EVAL` sends it with every call. | ||
| */ | ||
| const PUBLISH_SCRIPT = `local id=redis.call('XADD',KEYS[1],'*','data',ARGV[1]) if ARGV[2] then redis.call('XTRIM',KEYS[1],'MINID',ARGV[2],ARGV[3]) redis.call('EXPIRE',KEYS[1],ARGV[4]) end redis.call('PUBLISH',KEYS[1],'{"data":'..ARGV[1]..',"id":"'..id..'"}')` |
There was a problem hiding this comment.
With a node-redis client-level keyPrefix, this publishes to the wrong channel. node-redis prefixes the EVAL keys (KEYS[1]) but never Pub/Sub channels (the keyPrefix option docs state "Pub/Sub channels are NOT prefixed"), while subscribe() listens on the unprefixed channel — so live delivery on the resume path silently stops. Before this PR publishMessage() sent the unprefixed channel, so it worked.
Technical details
# Publish to the same channel subscribers use
## Affected sites
- `packages/publisher/src/adapters/base-redis.ts:18` — `PUBLISH` targets `KEYS[1]`.
- `packages/publisher/src/adapters/redis.ts:76` — `redis.eval(script, { keys: [channel], ... })` applies the client `keyPrefix` to that key.
## Evidence
- `@redis/client@6.2.1` parser: `EVAL` uses `pushKeys` (prefix applied), `XADD`/`XREAD`/`XTRIM` use `pushKey` (prefix applied), `PUBLISH` uses a plain `parser.push` (no prefix). The `keyPrefix` JSDoc: "Pub/Sub channels are NOT prefixed".
- Reproduced at the parser level with `keyPrefix: 'app:'`: EVAL args are `["EVAL","S","1","app:chan","d"]` while PUBLISH args are `["PUBLISH","chan","msg"]`.
- Old code: `addStreamEntry` used the prefixed `XADD` key, `publishMessage` used the unprefixed channel — the stream and the channel were independently correct.
## Required outcome
- With `keyPrefix` set, a live subscriber on `channel` must receive events, while the stream stays under the prefixed key so `readStreamEntries` keeps matching.
## Suggested approach
- Pass the raw (unprefixed) channel to the script as an `ARGV` and `PUBLISH` to that, keeping `KEYS[1]` for `XADD`/`XTRIM`/`EXPIRE` (needed for stream reads and cluster slot routing). Bun and Upstash have no key prefix, so the two values are identical there today.
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) ｜ 𝕏
This branch has not been deployed
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
Breaking changes
BaseRedisPublishersubclasses implementevalScript()instead ofaddStreamEntry(), andRedisStreamTrimOptionsis removed.Testing