ddp-server: publications leak stream listeners when disconnect races with async subscription setup

Read packages/ddp-server/src/Publication.ts, packages/streamer/src/streamer.module.ts, and the stream-notify-user handler in packages/streamer/src/notifications.module.ts; trace how stop state and cleanup behave when async setup resumes after disconnect. The issue includes reproduction steps and expected listener cleanup. Done means raced subscriptions leave no listeners or closed-socket writes and each onStop callback runs once.

Summary

Publication in @rocket.chat/ddp-server does not honour the subscription lifecycle the streamer code is written against. When a client disconnects while a stream subscription is still awaiting a database lookup, the handler resumes after the publication has already stopped, attaches listeners to the streamer, and registers an onStop cleanup that never runs. On stream-notify-user <uid>/rooms-changed those listeners are never removed: they keep the closed Session reachable and write every later room event to the closed socket.

Since #43281 the monolith serves its streams through this Publication, so this affects develop for every deployment.

The lifecycle mismatch

Meteor's Subscription (which the streamer was written for, ddp-server livedata_server.js):

• onStop(callback) runs callback immediately when the subscription is already deactivated.
• _isDeactivated() is this._deactivated || !this._session || this._session.inQueue === null, so it becomes true when the subscription stops or its session goes away.

packages/ddp-server/src/Publication.ts:

• _isDeactivated() returns this._session === null, but _session is never set to null, so it is always false.
• onStop(fn) is this.once('stop', fn). A callback registered after stop was emitted never runs, because EventEmitter does not replay events.
• stop() sends nosub and emits stop again on every call.

The streamer relies on Streamer.isPublicationActive() (built on _isDeactivated()) after each await, and on onStop for cleanup. With this Publication, the guard always passes and a late onStop is lost.

The race

stream-notify-user <uid>/rooms-changed (notifications.module.ts L434-L501):

1. The client subscribes. Streamer._publish adds the subscription, registers its onStop, sends ready, and emits _afterPublish (not awaited).
2. The _afterPublish listener passes isPublicationActive() and awaits Subscriptions.find({ 'u._id': userId }).toArray().
3. The client disconnects. Session emits close, the Publication emits stop, and every callback registered so far runs.
4. The lookup resolves. The handler calls streamer.on(rid, roomEvent) for each room, streamer.on(userId, userEvent), and then publication.onStop(...), which never fires.
5. From then on, every streamUser.__emit(rid, ...) (room changes from ListenersModule) calls roomEvent, which passes isPublicationActive() and writes to the closed socket. Every inserted subscription event for the user calls Rooms.findOneById once per leaked listener. Nothing calls stop() on that publication again, so the listeners stay until the process restarts.

stream-room-messages has the same race inside allowRead (Rooms.findOneById, the cached user lookup, canReadRoom) before Streamer._publish reaches its guard. The stopped publication is added to the streamer, ready is written to the closed socket, and the _afterPublish user listener is attached. That leak is temporary: on the next message in that room, StreamAdapter.sendToManySubscriptions finds the socket closed and calls subscription.stop(), which emits stop again and finally runs the late callbacks.

Steps to reproduce

1. Wire NotificationsModule to createStreamAdapter(server) as apps/meteor/server/lib/notifications/core/lib/Notifications.ts does, with a ws server creating a Session per connection.
2. Hold Subscriptions.find(...).toArray() on a deferred promise.
3. Connect, log in as u1, subscribe to stream-notify-user with u1/rooms-changed, wait for the lookup to be pending.
4. Close the socket, wait for the server-side close, then resolve the lookup.
5. Read notifications.streamUser.listenerCount('u1') and listenerCount('<rid>').

My reproduction ran the real Server, Session, Publication, ConnectionLifecycle, createStreamAdapter, Streamer and NotificationsModule from develop over real ws sockets on loopback. Only these boundaries were mocked: @rocket.chat/models (each lookup behind a gate the test resolves, no timers), the core-services proxies (Authorization allows, Settings, api.broadcast), the logger, and login (session.userId set on connected). Publication.onStop/emit and Session.send were wrapped to count late registrations, callback runs, and writes while the socket is not open.

Expected behavior

After each disconnect, no listener for that publication remains, every onStop callback has run exactly once, and nothing is written to the closed socket.

Actual behavior

Output against develop at 45d9273ed3aea44361d8c9ba5e86617eb6097004. userEvt = streamUser.listenerCount('u1'), r1..r3 = streamUser.listenerCount(rid), nuSubs/rmSubs = streamer subscriptions for u1/rooms-changed / room-messages r1, rmUserEvt = streamRoomMessage.listenerCount('u1'), lateOnStop = onStop calls made after stop, onStopRan = callbacks run / registered, closedWrites = Session.send calls on a socket that is not open.

MODE: UPSTREAM develop (unmodified)
  baseline                     userEvt=0 r1=0 r2=0 r3=0 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=0 onStopRan=0/0 closedWrites=0

A/D/H. notify-user rooms-changed, subscribe completes, then disconnect
  reply: {"msg":"ready","subs":["s1"]}
  after ready                  userEvt=1 r1=1 r2=0 r3=0 nuSubs=1 rmSubs=0 rmUserEvt=0 openSessions=1 lateOnStop=0 onStopRan=0/2 closedWrites=0
  after disconnect             userEvt=0 r1=0 r2=0 r3=0 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=0 onStopRan=2/2 closedWrites=0

H2. notify-user, subscribe completes, client unsubscribes (no disconnect)
  after ready                  userEvt=1 r1=1 r2=0 r3=0 nuSubs=1 rmSubs=0 rmUserEvt=0 openSessions=1 lateOnStop=0 onStopRan=2/4 closedWrites=0
  after unsub                  userEvt=0 r1=0 r2=0 r3=0 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=1 lateOnStop=0 onStopRan=4/4 closedWrites=0
  after disconnect             userEvt=0 r1=0 r2=0 r3=0 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=0 onStopRan=4/4 closedWrites=0

B/C. notify-user: disconnect while Subscriptions.find (last await before onStop) is pending
  reply: {"msg":"ready","subs":["s3"]}  (ready is sent before _afterPublish runs)
  lookup pending               userEvt=0 r1=0 r2=0 r3=0 nuSubs=1 rmSubs=0 rmUserEvt=0 openSessions=1 lateOnStop=0 onStopRan=4/5 closedWrites=0
  after disconnect             userEvt=0 r1=0 r2=0 r3=0 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=0 onStopRan=5/5 closedWrites=0
  after lookup resolves        userEvt=1 r1=1 r2=0 r3=0 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=1 onStopRan=5/6 closedWrites=0

E. same race, 3 more reconnect cycles (same user, same room)
  after cycle 1                userEvt=2 r1=2 r2=0 r3=0 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=2 onStopRan=6/8 closedWrites=0
  after cycle 2                userEvt=3 r1=3 r2=0 r3=0 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=3 onStopRan=7/10 closedWrites=0
  after cycle 3                userEvt=4 r1=4 r2=0 r3=0 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=4 onStopRan=8/12 closedWrites=0

F. room/user events after all those clients disconnected
  room event r1 -> writes to closed sessions: 4
  user event u1 removed r1 -> writes to closed sessions: 0
  after events                 userEvt=4 r1=0 r2=0 r3=0 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=4 onStopRan=8/12 closedWrites=4
  user event u1 inserted r9 -> Rooms.findOneById queries: 4, closed-session writes: 4
  after insert event           userEvt=4 r1=0 r2=0 r3=0 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=4 onStopRan=8/12 closedWrites=8
  sample closed-session writes: ["{\"msg\":\"changed\",\"collection\":\"stream-notify-user\",\"id\":\"id\",\"fields\":{\"eventName\":\"u1/roo","{\"msg\":\"changed\",\"collection\":\"stream-notify-user\",\"id\":\"id\",\"fields\":{\"eventName\":\"u1/roo"]

G. multiple rooms (r1,r2,r3) under the same user, race
  after race (3 rooms)         userEvt=5 r1=1 r2=1 r3=1 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=5 onStopRan=9/14 closedWrites=8

B2. room-messages r1: disconnect while Rooms.findOneById (inside allowRead) is pending
  lookup pending               userEvt=5 r1=1 r2=1 r3=1 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=1 lateOnStop=5 onStopRan=9/14 closedWrites=8
  after disconnect             userEvt=5 r1=1 r2=1 r3=1 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=5 onStopRan=9/14 closedWrites=8
  after lookup resolves        userEvt=5 r1=1 r2=1 r3=1 nuSubs=0 rmSubs=1 rmUserEvt=1 openSessions=0 lateOnStop=7 onStopRan=9/16 closedWrites=9
  -- a live client now subscribes to r1 and a message is emitted on room-messages r1
  live client ready            userEvt=5 r1=1 r2=1 r3=1 nuSubs=0 rmSubs=2 rmUserEvt=2 openSessions=1 lateOnStop=7 onStopRan=9/18 closedWrites=9
  live client got message; writes to closed sessions during emit: 1
  after first emit             userEvt=5 r1=1 r2=1 r3=1 nuSubs=0 rmSubs=1 rmUserEvt=1 openSessions=1 lateOnStop=7 onStopRan=11/18 closedWrites=10
  after live client disconnect userEvt=5 r1=1 r2=1 r3=1 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=7 onStopRan=13/18 closedWrites=10

GC. does a raced Session stay reachable after a forced GC?
  raced session retained: true   clean session retained: false

FINAL userEvt=6 r1=2 r2=2 r3=2 nuSubs=0 rmSubs=0 rmUserEvt=0 openSessions=0 lateOnStop=8 onStopRan=16/22 closedWrites=10

What it shows:

• A normal disconnect (A) and a normal unsubscribe (H2) clean up: 0 listeners, every callback run.
• A disconnect while Subscriptions.find is pending (B/C) leaves the user listener and the room listener behind with no open session, and the late onStop never runs (5/6).
• Three more racy reconnects (E) grow the listeners 1 → 2 → 3 → 4.
• After every client has disconnected (F), one room event produces 4 writes to closed sockets, and one inserted event produces 4 Rooms.findOneById queries and 4 more closed-socket writes. A removed event drops the room listeners but the user listeners stay.
• With three rooms (G), one race adds one listener per room plus one user listener.
• On stream-room-messages (B2), the stopped publication is added and ready is written to the closed socket; the next message in the room writes nosub to the closed socket and then clears it.
• With node --expose-gc, the raced Session is still reachable after gc(), while a cleanly closed one is collected.

ws drops writes to a closed socket without throwing, so nothing is logged; the cost is the retained session per race, the work done per event for each dead listener, and the extra queries.

Scope

Deployment	Status
Monolith on develop (since #43281, 2026-10-30)	Reproduced as above.
Monolith, 9.0.0-rc.4 and earlier	Not exposed: Notifications.ts still publishes through Meteor.publish, whose onStop/_isDeactivated behave as described above.
Microservices (ee/apps/ddp-streamer), 9.0.0-rc.4	Same Publication (ee/apps/ddp-streamer/src/ddp/Publication.ts) and the same notify-user handler. Identified by code inspection only, not reproduced on the tag.

Related code paths with the same register-after-await shape, identified by inspection only: StreamPresence._publish (onStop after await client.refreshHiddenUsers()), and trackHistoryCutoff in the open #43238.

The window is the duration of the lookup, so it is small per connection, but reconnect storms (deploys, network drops, mobile clients backgrounding) make it more likely.

Suggested fix

Fix the contract once, in Publication, instead of at each call site:

• On the first stop, whether from the client closing or from stop(), mark the publication stopped and set _session to null, so _isDeactivated() and isPublicationActive() report it.
• Make stop() idempotent, and do not send nosub once the socket is closed.
• In onStop(fn), run fn immediately when the publication has already stopped.
• Run each callback exactly once.

The existing callers already handle a null _session, since #40429 and #41302 adapted them to Meteor 3.5.

Regression tests

• Publication: _isDeactivated() is true after the client closes and after stop(); onStop registered after the stop runs immediately, once; stop() twice sends one nosub and runs each callback once.
• NotificationsModule with createStreamAdapter and a deferred Subscriptions.find: a disconnect during the lookup leaves 0 user and room listeners; three such races do not accumulate; three rooms leave 0; a room event afterwards writes nothing.
• stream-room-messages with a deferred Rooms.findOneById: no subscription is added and ready is not sent after the stop.
• Normal subscribe, unsubscribe and disconnect unchanged.

Related

• #43281 switched the monolith's streams to @rocket.chat/ddp-server.
• #43217 moved Publication from ddp-streamer to @rocket.chat/ddp-server unchanged.
• #40429 added _isDeactivated() to the ddp-streamer Publication as _session === null; #41302 made isPublicationActive() also require a socket.
• #42031 / #42174 fixed other publication cleanup paths (the client close listener, and handlers that fail); neither covers a disconnect during a handler that succeeds.
• #43238 (open) adds trackHistoryCutoff, which registers onStop after an await.

Server Setup Information

• Version of Rocket.Chat Server: develop at 45d9273ed3aea44361d8c9ba5e86617eb6097004
• Deployment: monolith
• NodeJS Version: v26.0.0 (reproduction)
• MongoDB: not used (models replaced by the deferred gates described above)