Repository navigation
🐛 Destroy every stream an Agent holds when it closes - #751
alecgibson wants to merge 1 commit into
Conversation
There was a problem hiding this comment.
🟡 Changes recommended
In-flight presence broadcasts can survive cleanup, and synthetic disconnect messages include an invalid sequence value.
2 open findings
What changed in this PR
Tracks Agent-owned streams for reliable cleanup and broadcasts null presence on unsubscribe or disconnect.
Changes:
- Adds centralized stream tracking and cleanup.
- Moves presence-disconnect broadcasts into explicit cleanup paths.
- Adds tests and documents null presence behavior.
| File | Description |
|---|---|
lib/agent.js |
Implements stream tracking and presence cleanup. |
test/agent.js |
Tests cleanup of all stream types. |
test/client/presence/presence.js |
Tests null presence broadcasts. |
docs/api/presence.md |
Documents unsubscribe behavior. |
docs/middleware/actions.md |
Documents null middleware presence. |
🧠 Review effort: Balanced
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
84cb60e to
c5c9a12
Compare
c5c9a12 to
7c1a081
Compare
Migration noteNot breaking. This can ship as a 7.x 🐛 fix. On
On backend.use('receivePresence', function(context, next) {
if (context.presence.p === null) return next();
// ...
});Two other changes that middleware with side effects (metering, auditing, "last known presence") may notice:
|
dawidreedsy
left a comment
There was a problem hiding this comment.
I don't have comments, but my review panel has few comments :P
| return callback(error); | ||
| } | ||
| backend.trigger(backend.MIDDLEWARE_ACTIONS.receivePresence, agent, context, function(error) { | ||
| agent._removePendingPresenceBroadcast(pendingBroadcast); |
There was a problem hiding this comment.
[🤖 Operability + Intent + Correctness · should] The pending-broadcast guard ends here, but the transform (getOps) and publish still run after it. A presence in that window that isn't the stored requests[id] can still publish live after the leave.
- The client sends pv1 and pv2 for the same id without waiting for acks (
local-presence.js:30-54). - pv1 finishes middleware second, through async middleware or two concurrent
backend.subscribes in_subscribeDocVersion. Line 863 doesn't store it, because pv2 is newer. - The client disconnects while pv1 is in
transformPresenceToLatestVersion._retractPresencesnulls only the stored pv2. - pv1 publishes its real
pafter the null. On receivers, the null already ran_removeRemotePresence(client/presence/presence.js:136-137), which dropped the pv high-water mark. pv1 lands on a freshRemotePresencewithpresenceVersion0, is accepted, and leaves a ghost.
The stored-request case is only safe because the transform shares the object and mutates it in place. #748 lists the transform as one of its windows, but none of the new tests holds a broadcast inside it.
Suggestion: keep the pending entry until just before pubsub.publish, removing it in a single exit wrapper around callback. Re-check agent.closed || pendingBroadcast.retracted at that point, then publish p: null or drop the stale one. To test it: make the presence stale (as in doc-presence.js:253), hold agent1's op middleware (fired via getOps → _sanitizeOp), unsubscribe inside it, release it, and assert the last receive is null.
Dissent (Operability): base had the same hole, and a wider one, so this narrows #748 rather than regressing it. It could be a follow-up.
There was a problem hiding this comment.
Fixed, but with a publish-time check rather than keeping the pending entry: once the transform finishes, a presence that's no longer the stored presenceRequests[ch][id] (because a newer one from the same client, or its leave, has replaced it) is acknowledged but not published. That needs no extra bookkeeping, and it also covers #763, so the commit now closes that too. Leaves are exempt, because _cleanup() deletes the channel's map while one may still be in its transform.
A presence re-created with the same id starts again at pv 0, so any presence now replaces a stored leave whatever its pv. Otherwise the check above would drop it. That would also let an older presence still in flight replace its own client's later leave, which happens without any middleware, because a doc presence's first submit waits on a DB subscribe and its leave doesn't. So a leave from the client now flags its older pending broadcasts with the same id, and they're dropped as they leave middleware, releasing the doc version they subscribed to after the leave had already released it. _pendingPresenceBroadcasts is already in arrival order, so that needs no new bookkeeping. Async receive middleware can still reorder a presence and its own leave before either reaches this point. That's no worse than 7.0.0, and I've filed #767 for it.
Tests:
- "does not broadcast an older presence that finishes transforming after its client unsubscribes" (holds agent1's
opmiddleware, as you suggested) - "broadcasts a null presence still being transformed when the connection closes"
- "does not broadcast presence still looking up its doc version when it is destroyed"
- "does not broadcast presence that gets through middleware after its own leave"
- the two "re-created with the same id" tests
| var requests = presenceRequests[presence.ch] || (presenceRequests[presence.ch] = Object.create(null)); | ||
| if ((agent.closed || pendingBroadcast.retracted) && !leaving) { | ||
| presence.p = null; | ||
| return agent._broadcastPresence(presence, callback); |
There was a problem hiding this comment.
[🤖 Operability · should] When the agent is closed, errors from this rewritten p: null pass are dropped without a trace. That covers middleware rejecting the null, the doc-version unsubscribe, the transform and the publish.
callback goes to _reply and then send, which returns early once this.closed (lines 287-289). The other synthetic-leave path, _retractPresences, logs with logger.error (line 201). This path exists to prevent ghosts, so a silent failure here leaves a ghost with nothing logged.
Suggestion:
return agent._broadcastPresence(presence, function(error, result) {
if (error) logger.error('Error broadcasting disconnect presence', presence.ch, presence.id, error);
callback(error, result);
});There was a problem hiding this comment.
Done. Both synthetic leaves now go through _broadcastLeave(), which logs failures with the channel, presence id and client src. Test: "logs an error when the leave for presence in flight fails".
| if ((agent.closed || pendingBroadcast.retracted) && !leaving) { | ||
| presence.p = null; | ||
| return agent._broadcastPresence(presence, callback); | ||
| } |
There was a problem hiding this comment.
[🤖 Readability · should] This condition carries the PR's key new rule (an in-flight presence whose client has left goes out as a leave), but the rule has no name, and the recursion's stopping condition is hidden.
To follow it, a reader has to know three things:
retractedis set far away, in_retractPresences.agent.closedis needed too, because_cleanuponly retracts channels that already have apresenceRequestsentry.- The recursion stops only because
leavingis recomputed astrueon the second pass.
Suggestion:
var clientHasLeft = agent.closed || pendingBroadcast.retracted;
if (clientHasLeft && !leaving) {
// Re-broadcast once as a leave: `leaving` is true on that pass, so it can't loop
presence.p = null;
return agent._broadcastPresence(presence, callback);
}Alternatively, flag every pending broadcast in _cleanup, so this becomes pendingBroadcast.retracted && !leaving. Correctness checked that it behaves the same, because only leaving broadcasts can start once the agent is closed.
There was a problem hiding this comment.
Named it clientHasLeft, and moved the rewrite into _broadcastLeave(). The !leaving beside it is what stops it recursing, the same way it does on the replaced check further down.
| Agent.prototype._trackStream = function(stream) { | ||
| var streams = this._streams; | ||
| var key = this._nextStreamKey++; | ||
| streams[key] = stream; | ||
| stream.once('close', function() { | ||
| delete streams[key]; | ||
| }); | ||
| }; |
There was a problem hiding this comment.
[🤖 Tests & conventions · should] No test checks that _trackStream drops a stream from _streams when the stream closes. A long-lived connection could build up a map of dead streams and CI would stay green.
Mutation test: deleting the stream.once('close', … delete streams[key]) listener still gives 902 passing, 0 failing. The new cleanup tests assert backend.pubsub.streamsCount, which drops on destroy whether or not the agent's map is pruned. No test reads agent._streams. A script that subscribes and unsubscribes shows it does drain today; the tests just don't pin that.
Suggestion: in the cleanup describe block, subscribe and unsubscribe a doc and a presence channel, and submit and then destroy a doc presence (doc-version stream). Then assert expect(connection.agent._streams).to.be.empty before closing. Existing tests already read agent internals such as presenceRequests.
There was a problem hiding this comment.
Added "stops tracking streams once they close". It covers doc subscribe/unsubscribe, presence subscribe/unsubscribe, and a doc presence submit then destroy, and expects agent._streams to be empty. Deleting the 'close' listener now fails it.
|
|
||
| > The presence object that was received. Its shape depends on its [type]({{ site.baseurl }}{% link types/index.md %}) | ||
| > | ||
| > `presence.p` is `null` when the presence leaves the channel: when it is destroyed, or when its client unsubscribes or disconnects. If its client leaves while a presence is still in this middleware, the presence passes through it a second time with `p: null`. |
There was a problem hiding this comment.
[🤖 Intent + Compatibility + Readability · should] This sentence understates when the second pass happens and what it receives. The author's migration note says more than these docs do.
- When: the pending broadcast is registered at
lib/agent.js:838, before the doc-version lookup at line 845. A leave during that lookup also triggers the second pass, not only a leave "while a presence is still in this middleware". - What: it's the same
presenceobject, already changed by the first pass, withpset tonull. Middleware with side effects (metering, auditing, counting) will see one submitted presence twice.
This is a documentation caveat, not a semver break: base already ran p: null leaves through receivePresence.
Suggestion: "If its client leaves while a presence is still being processed (its doc version lookup or this middleware), the same presence object passes through this middleware a second time, with p set to null and any changes from the first pass kept. Middleware with side effects should expect this." Add the same line to the migration note.
There was a problem hiding this comment.
Reworded. Any leave from an unsubscribe or disconnect reuses the client's latest presence object, not only one still in flight, so the docs now say that. They also say that a presence which passes this middleware may still not be broadcast if a newer one replaces it first. Both are in the migration note too.
| var requests = this.presenceRequests[channel]; | ||
| for (var id in requests) { | ||
| var request = requests[id]; | ||
| if (request.p === null) continue; |
There was a problem hiding this comment.
[🤖 Readability · nit] This skip has no explanation. The old code re-broadcast every request, so a reader can't tell why null requests are now skipped: the presence has already left, and its null has already gone out.
Suggestion:
var alreadyLeft = request.p === null;
if (alreadyLeft) continue;There was a problem hiding this comment.
Done, as you suggested.
| this._streams = Object.create(null); | ||
| this._nextStreamKey = 1; |
There was a problem hiding this comment.
[🤖 Readability · nit] _streams and _nextStreamKey are the only new fields with no comment. _streams is also easy to confuse with this.stream (the client connection) and with the per-path maps above it, and a reader can't tell that it leaves out query emitters.
Suggestion:
// Every pub/sub stream this agent holds (doc, doc version and presence
// subscriptions). Streams remove themselves on 'close', and _cleanup()
// destroys the rest, even ones no longer in their own map
this._streams = Object.create(null);There was a problem hiding this comment.
Done. Every other field in that constructor documents its map, so these do too.
| for (var channel in this.presenceRequests) { | ||
| this._retractPresences(channel); | ||
| } | ||
| this.subscribedDocs = Object.create(null); | ||
|
|
||
| for (var channel in this.subscribedPresences) { | ||
| this.subscribedPresences[channel].destroy(); | ||
| for (var key in this._streams) { | ||
| this._streams[key].destroy(); | ||
| } |
There was a problem hiding this comment.
[🤖 Readability · nit] The PR removed the // Clean up doc subscription streams comment, and the two new loops have none, while the query loop below keeps its comment. So _cleanup is now half documented, and a reader comparing it with base will wonder whether the map resets were dropped on purpose.
Suggestion:
// Tell remote subscribers this client's presence has left
for (var channel in this.presenceRequests) { … }
// Destroy doc, doc version and presence streams. Their maps empty
// out through each stream's 'end' handler
for (var key in this._streams) { … }There was a problem hiding this comment.
I'd rather leave these out. They'd restate _retractPresences() and destroy(), and the resets are back, so there's nothing left for a reader comparing it with base to wonder about.
| for (var id in streams) streams[id].destroy(); | ||
| } | ||
| this.latestDocVersionStreams = Object.create(null); | ||
| Agent.prototype._trackStream = function(stream) { |
There was a problem hiding this comment.
[🤖 Performance + Readability · praise] _trackStream follows the same counter-key plus once('close') self-removal pattern as PubSub#_createStream, so it reads as familiar. It also holds up under scrutiny:
- keys can't collide;
- destroying twice does nothing, because of
OpStream'sopenguard; - deleting inside
_cleanup's for-in loop is defined behaviour; - streams that arrive after close are destroyed before they are tracked.
The cost is about 150-200 B per stream, which is roughly 2 MB for a client holding 10k doc subscriptions.
| agent._removePendingPresenceBroadcast(pendingBroadcast); | ||
| if (error) return callback(error); | ||
| var requests = presenceRequests[presence.ch] || (presenceRequests[presence.ch] = Object.create(null)); | ||
| if ((agent.closed || pendingBroadcast.retracted) && !leaving) { |
There was a problem hiding this comment.
[🤖 Intent · praise] This fixes #748's stale-pv problem for the doc-version lookup and middleware windows. The newer in-flight presence is never published live. It goes out as p: null with its own newer pv, so whichever null arrives last, a leave has the last word. leaving stops a second rewrite, and the "doesn't rewrite a null presence again" test pins that.
|
Heads-up from #752, which now also carries #749 (reject presence ahead of the doc) and is due to land first. This PR reworks the same #749 caches a presence before its transform and rolls it back if the transform rejects it. If the client disconnects while that transform is still running, the leave never reaches subscribers:
The leave replay nulls the in-flight cached request in place, and #749 rejects that request as ahead of the doc. The rollback then writes the previous request into a map the Repro, as a drop-in for it('broadcasts null presence on disconnect while presence ahead of the doc is in its transform', function(done) {
var localPresence1 = presence1.create('presence-1');
var releaseGetOps;
async.series([
presence1.subscribe.bind(presence1),
presence2.subscribe.bind(presence2),
function(next) {
localPresence1.submit({index: 1}, errorHandler(done));
presence2.once('receive', function() {
next();
});
},
function(next) {
var getOps = backend.db.getOps;
sinon.stub(backend.db, 'getOps').callsFake(function() {
var args = arguments;
releaseGetOps = function() {
getOps.apply(backend.db, args);
};
backend.db.getOps.restore();
next();
});
var message = localPresence1._message();
message.v = doc1.version + 1;
message.t = presenceTestType.type.uri;
connection1.send(message);
},
function(next) {
presence2.once('receive', function(id, presence) {
expect(id).to.equal('presence-1');
expect(presence).to.be.null;
next();
});
connection1.close();
setTimeout(releaseGetOps, 20);
}
], done);
}); |
7c1a081 to
528f9cc
Compare
528f9cc to
8ce846f
Compare
Fixes #740 Fixes #748 Fixes #763 At the moment, `Agent._cleanup()` only destroys streams that are still in each subscription path's own map, so any bookkeeping slip in one of those paths leaves a stream subscribed after the client disconnects. This change tracks every stream handed to the agent in one map, and destroys them all on cleanup. So that presence streams can be replaced like doc streams, the `p: null` "client left" broadcast moves off the presence stream's `'end'` and into `_unsubscribePresence()` and `_cleanup()`. A broadcast still in middleware at that point goes out as `p: null` instead, and one that a newer presence or that leave has replaced by the time it would be published isn't sent at all. Any presence replaces a stored leave whatever its `pv`, because a presence re-created with the same id starts again at `pv` 0. That also fixes presence outliving its client when it was never subscribed, or still in flight, which `receivePresence` middleware now sees as `p: null`: 1. Client A submits presence on a channel it hasn't subscribed to 2. Client B, subscribed to the channel, receives it 3. Client A disconnects, and B keeps A's presence forever 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
8ce846f to
af9c279
Compare
There was a problem hiding this comment.
🔵 Needs a closer look
The interactions between asynchronous middleware, presence transformations, and disconnect cleanup warrant final human validation despite the added regression coverage.
No open issues 🎉
2 resolved since last review
Review effort: Balanced
|
A heads-up for rebasing this over #752, which a Copilot fix has since changed (c3be55e). When a newer presence for the same id is still in flight, or gets rejected, #752 publishes an older presence once its transform succeeds, and caches it so the disconnect leave goes out at its
To keep what |



Fixes #740
Fixes #748
At the moment,
Agent._cleanup()only destroys streams that are still in each subscription path's own map, so any bookkeeping slip in one of those paths leaves a stream subscribed after the client disconnects.This change tracks every stream handed to the agent in one map, and destroys them all on cleanup. So that presence streams can be replaced like doc streams, the
p: null"client left" broadcast moves off the presence stream's'end'and into_unsubscribePresence()and_cleanup(). A broadcast still in middleware at that point goes out asp: nullinstead.That also fixes presence outliving its client when it was never subscribed, or still in flight, which
receivePresencemiddleware now sees asp: null:🤖 Generated with Claude Code
Co-Authored-By: Claude noreply@anthropic.com