Skip to content

GO-7373: reconnect the mongo change stream instead of dying silently#103

Open
requilence wants to merge 2 commits into
mainfrom
go-7373-changestream-listener-reconnect
Open

GO-7373: reconnect the mongo change stream instead of dying silently#103
requilence wants to merge 2 commits into
mainfrom
go-7373-changestream-listener-reconnect

Conversation

@requilence

Copy link
Copy Markdown
Contributor

Problem

db.streamListener is a bare for stream.Next(ctx) {} loop. The driver resumes a change stream only once per failure, and returns false with a nil error when the server closes the cursor:

if !cs.isResumableError() { return }                                  // non-resumable -> false
if cs.err = cs.executeOperation(ctx, true); cs.err != nil { return }  // one resume attempt; fails -> false

So a mongo restart (resume hits server selection timeout), a non-resumable error (ChangeStreamHistoryLost, CappedPositionLost) or an invalidate ends the loop and the goroutine exits without a trace. The node stays up and keeps serving.

That listener is the only thing that refreshes the log cache, and the cache cannot heal on its own: object.AddStream serves the cached snapshot and never re-reads the db, while object.TryClose pins the object as long as any stream is attached — and a tree node keeps a permanent Watch on every space it holds.

Net effect: a dead listener freezes the cache for good. Every new subscriber gets a stale snapshot, so the coordinator's acl head stays behind the db and it rejects the next correctly chained record forever — surfacing to users as incorrect prev id of a record when sharing a space.

Evidence

Caught on a live consensus node:

  • pprof: 57 goroutines, healthy mongo driver pool, 3 LogWatch streams served — zero streamListener goroutines, zero ChangeStream frames
  • mongo $currentOp: zero change-stream cursors on consensus.log
  • the coordinator wrote a record to consensus itself, and an hour later — after a cold acl cache reload — still validated against the previous head
  • reproduced against a real replica-set mongod: stop mongod for >30s, restart it, and the listener never delivers another change

Fix

  • log the interruption and reopen the change stream with a backoff (capped at 30s) instead of exiting
  • add db.ResetReceiver, fired on reconnect: updates made while the stream was down are never replayed, so the consumer has to resync
  • stream.Service.resync reloads every cached log from the db and sends the missed records to the subscribed streams, healing both the cache and the live subscribers
  • open the change stream on the long-lived stream context rather than the Run one, and stop the listener before disconnecting the client in Close

Tests

Both fail without the fix:

  • stream: a new subscriber gets a stale head after a missed update
  • db: forces an invalidate, asserts the listener reconnects and reports the reset

Follow-ups (not here)

  • object.AddStream could reconcile with the db instead of serving a pinned snapshot
  • object.AddRecords diffs by slice length, so it cannot detect a gap in the chain
  • any-sync/acl.AddRecord relies entirely on the consensus broadcast to learn about its own accepted record

streamListener was a bare `for stream.Next(ctx) {}` loop. The driver
resumes a change stream only once per failure, and returns false with a
nil error when the server closes the cursor, so a mongo restart, a
non-resumable error or an invalidate event ended the loop and the
goroutine exited without a trace.

The listener is the only thing that refreshes the log cache, and the
cache never reloads from the db while a stream is attached to it, so a
dead listener froze the cache for good: every new subscriber was served
a stale snapshot and the coordinator kept rejecting valid acl records
with "incorrect prev id of a record".

Log the interruption and reopen the stream with a backoff. The updates
made while the stream was down are not replayed, so notify the reset
receiver on reconnect and resync the cached logs with the db, sending
the missed records to the subscribed streams.
- resync retries a log it failed to fetch: it runs right after mongo came
  back, so a fetch may well fail, and giving up would leave the log stale
  forever, since it is pinned in the cache by its streams and never reloaded
- ResetReceiver takes a ctx, so a resync can't outlive the service: it runs
  on the listener goroutine, and Close waits for it before disconnecting
- keep the backoff between the reconnects of a stream that dies right after
  being opened, otherwise the listener reopens it and resyncs in a tight loop
- fail loudly if the listener ever stops while the service is running: without
  it the cache silently serves stale records

Do not resume the stream from the last token: the pipeline filters out the
invalidate event, so the token never moves past it and the resumed stream
dies on it again and again.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant