Race Condition: Initial Fetch vs Change Stream Events

Problem Summary

There is a race condition window between when the initial MongoDB find() query returns results and when the Change Stream subscription begins. Events that occur during this window can be missed, leading to inconsistent client state.

Current Implementation Flow

┌─────────────────────────────────────────────────────────────────┐
│ Cursor::start()                                                 │
├─────────────────────────────────────────────────────────────────┤
│ 1. fetch() - runs find() query                                  │
│    └─> Returns snapshot of documents at time T1                 │
│                                                                 │
│ ⚠️ GAP: Events between T1 and T2 are NOT captured ⚠️           │
│                                                                 │
│ 2. watch() - starts Change Stream                               │
│    └─> Begins capturing events from time T2                     │
└─────────────────────────────────────────────────────────────────┘

Relevant code: src/cursor/mod.rs lines 52-82

// 1. Run initial query (blocking)
self.fetcher
    .write()
    .await
    .fetch(&mergeboxes)
    .await
    .context("Cursor::start")?;

// 2. Start background task (Change Stream starts AFTER fetch completes)
let task = async move {
    let receiver_or_interval = fetcher.read().await.watch().await;
    // ...
};

Affected Scenarios

Event Type	Impact
Insert	Document is missed entirely until next refetch (polling)
Update	Client receives stale data until Change Stream delivers a subsequent update
Delete	Client receives a "ghost" document that no longer exists in the database

Current Mitigations

1. MergeBox Idempotency

The MergeBox handles duplicate operations gracefully:

• Duplicate inserts become updates (only changed fields sent)
• No message sent if nothing actually changed

However: This only helps with duplicates, not missed events.

2. Polling Fallback

When Change Streams are unavailable, the system polls every 10 seconds (configurable via polling_interval_ms). This provides eventual consistency but with a delay.

How Meteor (JS) Handles This

The original Meteor DDP server solves this elegantly through a phased approach with queuing:

Phased State Machine

1. QUERYING Phase: During initial load:

   • The initial query runs against the database
   • Oplog entries that arrive are NOT immediately applied
   • Instead, document IDs are stored in _needToFetch for later processing

   // From oplog_observe_driver.js
   _handleOplogEntryQuerying: function (op) {
     var self = this;
     Meteor._noYieldsAllowed(function () {
       self._needToFetch.set(idForOp(op), op);
     });
   },
   

2. Multiplexer Queue: The ObserveMultiplexer uses a _SynchronousQueue ensuring callbacks are processed in order. It also enforces that only added events occur during initial load:

   // From observe_multiplex.js
   if (!self._ready() &&
       (callbackName !== 'added' && callbackName !== 'addedBefore')) {
     throw new Error("Got " + callbackName + " during initial adds");
   }
   

3. Transition to STEADY: After the initial query completes:

   • _multiplexer.ready() is called
   • _doneQuerying() processes all pending oplog entries stored in _needToFetch
   • The driver enters PHASE.STEADY where changes are applied immediately

Result

Documents inserted/updated/deleted during the initial query are:

• Tracked in _needToFetch if oplog-driven
• Processed after the initial snapshot is sent and ready() is called
• Merged with the initial results to provide a consistent view

This ensures the client gets a consistent initial snapshot followed by all changes that happened during the load, in proper order. No changes are lost or duplicated.

Proposed Solutions for ddp-router

Option A: Phased Approach (matches Meteor's design)

Implement a similar QUERYING/STEADY phase model:

┌─────────────────────────────────────────────────────────────────┐
│ 1. watch() - Start Change Stream immediately                    │
│    └─> Buffer events in a queue (QUERYING phase)                │
│                                                                 │
│ 2. fetch() - Run find() query                                   │
│    └─> Send initial documents to client                         │
│                                                                 │
│ 3. Transition to STEADY                                         │
│    └─> Process buffered events                                  │
│    └─> Apply future events immediately                          │
└─────────────────────────────────────────────────────────────────┘

Pros:

• No events missed - matches proven Meteor behavior
• MergeBox already handles duplicate inserts/updates (idempotent)
• Conceptually aligned with original Meteor implementation

Cons:

• May send duplicate data to client (added → changed with same data)
• Memory usage for event buffer during slow initial fetches
• Requires adding phase state to CursorFetcher

Option B: Use Change Stream Resume Token

1. Get current resume token before fetch
2. Perform fetch
3. Start Change Stream from saved resume token

Pros:

• No duplicates
• No event buffer needed

Cons:

• Resume tokens expire (default ~1 hour on Atlas)
• More complex implementation
• Requires storing resume token state

Option C: Use Read Concern + Timestamp

1. Get current cluster time
2. Perform fetch with readConcern: { level: "majority" }
3. Start Change Stream with startAtOperationTime

Pros:

• Precise synchronization
• No duplicates

Cons:

• Requires replica set (standalone MongoDB does not support Change Streams anyway)
• Read concern "majority" may have minor performance implications

MongoDB Version Considerations

Opportunity: Simplify by Requiring MongoDB 7.0+

All MongoDB versions 6.0 and older are now End of Life (EOL):

Version	EOL Date	Status
6.0	July 2025	❌ EOL
7.0	August 2027	✅ Current LTS
8.0	October 2029	✅ Current Release

If we require MongoDB 7.0+, all the features we need are well-established and battle-tested:

• Change Streams: Available since 3.6
• startAtOperationTime: Available since 4.0
• readConcern: "majority": Available since 3.2
• FullDocumentType::Required: Available since 6.0

This would allow us to:

1. Use Option C (startAtOperationTime) without version compatibility concerns
2. Use FullDocumentType::Required for the update race condition fix
3. Simplify documentation and testing (single target platform)

Question for team: Should we require MongoDB 7.0+ given that all older versions are EOL?

Impact Assessment

• Severity: Medium
• Frequency: Rare (only during subscription setup + concurrent writes)
• User Impact: Temporary inconsistent state; self-corrects on next update or refetch

Questions for Team Discussion

1. Which solution approach should we pursue?
2. Is the polling fallback interval (10s) acceptable for the gap window?
3. Should we add a "forced refetch" after Change Stream starts as a simple fix?

Related Issues

FullDocumentType::Required (MongoDB 6.0+)

There's a related race condition in src/watcher.rs (lines 75-77):

// TODO: Ideally we would use `Required` here, but it has to be
// enabled on the database level. It should be configurable.
.full_document(Some(FullDocumentType::UpdateLookup))

Current behavior (UpdateLookup): On update events, MongoDB performs a separate lookup query to fetch the full document. This introduces a race condition where the document might change between the update event and the lookup.

Better option (Required): Available since MongoDB 6.0, this mode guarantees the full document is included directly in the change event - no race condition. Requires enabling changeStreamPreAndPostImages on the collection.

If we decide to require MongoDB 7.0+ (see MongoDB Version Considerations above), we could:

1. Make FullDocumentType configurable (as the TODO suggests)
2. Default to Required when changeStreamPreAndPostImages is enabled
3. Document the collection-level setting requirement

Related Files

• src/cursor/mod.rs - Cursor start/stop logic
• src/cursor/fetcher.rs - Initial fetch and event processing
• src/watcher.rs - Change Stream management (includes FullDocumentType TODO)
• src/mergebox.rs - Document merging and deduplication

References

• MongoDB Change Streams Documentation
• Meteor Oplog Tailing Implementation
• Change Stream Resume Tokens

Start with src/cursor/mod.rs lines 52-82, then read src/cursor/fetcher.rs, src/watcher.rs, and src/mergebox.rs to trace initial fetching and Change Stream processing. The issue presents three possible approaches but does not select one or define acceptance criteria, so the intended behavior and done state need team clarification first.