feat: fix fetch vs watch race condition using Change Stream Resume Tokens by cbnsndwch · Pull Request #5 · meteor/ddp-router

cbnsndwch · 2026-03-08T05:48:13Z

Resolves #4

Description

This PR fixes the race condition window that existed between the initial MongoDB find() query and the start of the Change Stream subscription (watch()). Previously, any events that occurred during this gap were missed, leading to temporarily inconsistent client state.

To resolve this, I've implemented Option B from the issue proposal (Change Stream Resume Tokens):

Captures the current resume token prior to executing the initial fetch() query.
Performs the fetch() to seed the MergeBox.
Starts the Change Stream watch() using the saved resume token, ensuring no operations in the intermediary window are missed.

Additional Changes & Reproduction

As requested in the issue thread, I have included a repro script that demonstrates the problem without the fix, and empirically verifies it operates correctly with the fix.

Examples: Added a reproduction script (kill-streams.js) to the meteor-repro example and bumped the mongodb dependency. This generates concurrent events that precisely hit the gap window between the initial fetch and watch.

Testing

Rust unit and integration tests (cargo test) pass successfully.
Verified against the new kill-streams.js reproduction script, confirming that intermediary events are captured immediately during initialization via the resume token without being dropped or waiting on polling fallbacks.

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

# Conflicts: # .gitignore

…tion

…find config.toml

Copilot

Pull request overview

This PR addresses the MongoDB find() vs Change Stream watch() startup race by starting the watch first and then performing a refetch + buffered replay so writes occurring during initialization aren’t missed. It also adds a Meteor-based reproduction harness and expands documentation to describe the project and configuration.

Changes:

Refactors the Change Stream watcher to support checkpointing (operation time + resume/startAfter tokens), reconnection recovery, and resync signaling.
Updates cursor startup to watch first, then refetch and replay buffered events, with lagged-receiver recovery via resync/refetch.
Adds a replica-set startup test plus a Meteor repro app/scripts and updates docs/config to support full_document.

Reviewed changes

Copilot reviewed 34 out of 36 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
`src/watcher.rs`	Implements per-collection shared change stream with checkpoints, recovery, and `Resync` events.
`src/cursor/mod.rs`	Changes cursor startup flow to watch-first and refetch/replay; improves lagged receiver handling.
`src/cursor/fetcher.rs`	Adds buffering phases and `refetch_and_replay`, plus insert de-dupe handling.
`src/main.rs`	Adds `mongo.full_document` parsing and passes it into `Watcher`.
`src/settings.rs`	Adds `mongo.full_document` config field.
`src/lib.rs`	Exposes crate modules for integration tests / library usage.
`tests/change_stream_startup_replica.rs`	Adds ignored replica-set integration test for startup replay correctness.
`examples/mongo_probe.rs`	Adds probe utility for Mongo connectivity/session/watch behavior.
`examples/meteor-repro/kill-streams.js`	Adds repro script to terminate active change streams.
`examples/meteor-repro/package.json`	Adds Meteor repro app dependencies/scripts (incl. mongodb bump).
`examples/meteor-repro/package-lock.json`	Locks Meteor repro npm dependencies.
`examples/meteor-repro/README.md`	Documents how to run the Meteor repro + router.
`examples/meteor-repro/run.ps1`	PowerShell runner configuring client env vars for repro.
`examples/meteor-repro/run-router.ps1`	PowerShell runner for starting the Rust router against repro Mongo.
`examples/meteor-repro/server/main.js`	Patches `Meteor.publish` to return cursor descriptions; adds mock insert method.
`examples/meteor-repro/tests/main.js`	Adds minimal Meteor app test scaffold.
`examples/meteor-repro/imports/api/links.js`	Defines Links collection used by the repro UI and publish.
`examples/meteor-repro/imports/ui/App.jsx`	Repro UI root component.
`examples/meteor-repro/imports/ui/Hello.jsx`	Repro UI sample counter component.
`examples/meteor-repro/imports/ui/Info.jsx`	Repro UI subscription + list rendering for Links.
`examples/meteor-repro/client/main.jsx`	Mounts React UI.
`examples/meteor-repro/client/main.html`	Adds mount point HTML.
`examples/meteor-repro/client/main.css`	Adds basic styling.
`examples/meteor-repro/.meteor/release`	Pins Meteor release for repro app.
`examples/meteor-repro/.meteor/packages`	Declares Meteor packages used by repro app.
`examples/meteor-repro/.meteor/versions`	Captures Meteor package versions for repro app.
`examples/meteor-repro/.meteor/platforms`	Declares supported Meteor platforms.
`examples/meteor-repro/.meteor/.id`	Meteor project id file for repro app.
`examples/meteor-repro/.meteor/.gitignore`	Ignores Meteor local state.
`examples/meteor-repro/.meteor/.finished-upgraders`	Meteor upgrader state for repro app.
`examples/meteor-repro/.gitignore`	Ignores `node_modules` in repro app.
`README.md`	Major documentation expansion (architecture, config, quickstart, limitations).
`DEVELOPING.md`	Adds Windows/MSVC build guidance and toolchain pinning notes.
`Cargo.toml`	Tweaks test profile (`codegen-units = 1`).
`.gitignore`	Expands ignored paths.
`test_rwlock.rs`	Adds a standalone RwLock demo source file at repo root.

Files not reviewed (1)

examples/meteor-repro/package-lock.json: Language not supported

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-03-08T05:53:25Z

+      </button>
+      <ul>{links.map(
+        link => <li key={link._id}>
+          <a href={link.url} target="_blank">{link.title}</a>


The link uses target="_blank" without also setting rel="noopener noreferrer", which allows the opened page to access window.opener (tabnabbing risk). Add the appropriate rel attribute when opening untrusted URLs in a new tab.

Suggested change

<a href={link.url} target="_blank">{link.title}</a>

<a href={link.url} target="_blank" rel="noopener noreferrer">{link.title}</a>

Copilot · 2026-03-08T05:53:26Z

+#[tokio::main]
+async fn main() {
+    let lock = tokio::sync::RwLock::new(5);
+    match *lock.read().await {
+        x => {
+            println!("Got {}, trying to write...", x);
+            *lock.write().await = 6;
+            println!("Done.");
+        }
+    }
+}


This looks like an ad-hoc executable used for debugging (#[tokio::main] fn main()), but it lives at the repository root where Cargo won’t treat it as an example or bin target. Consider removing it from the PR, or moving it under examples/ (or src/bin/) with a clear name/purpose so it’s discoverable and doesn’t look like stray source.

Copilot · 2026-03-08T05:53:26Z

+            let watch_result = self.fetcher.read().await.watch().await;
+            let receiver_or_interval = match watch_result {
+                Ok(mut subscription) => {
+                    self.fetcher
+                        .write()
+                        .await
+                        .refetch_and_replay(
+                            &mut subscription.receiver,
+                            &mergeboxes,
+                            subscription.start_at_operation_time,
+                        )


PR description says it implements the “resume token captured before fetch, then watch(resumeAfter)” approach (Option B), but this startup path passes start_at_operation_time (a Timestamp) into refetch_and_replay rather than a resume token. The implementation appears closer to an operation-time boundary + buffered replay approach. Please update the PR description to match the actual mechanism, or adjust the code to truly capture and use a resume token if that’s the intended design.

Copilot · 2026-03-08T05:53:26Z

+    let full_document = match settings.mongo.full_document.as_deref() {
+        Some("default") => None,
+        Some("updateLookup") => Some(FullDocumentType::UpdateLookup),
+        Some("whenAvailable") => Some(FullDocumentType::WhenAvailable),
+        Some("required") => Some(FullDocumentType::Required),
+        None => Some(FullDocumentType::UpdateLookup),


Config value full_document = "default" maps to None, but watcher::classify_change_event only emits Event::Update/Replace when full_document is present. With None, update/replace events will be classified as Ignore, so subscribers can silently miss updates. Either remove/rename the "default" option, force UpdateLookup when change streams are enabled, or implement handling based on updateDescription when full_document isn’t requested.

Copilot · 2026-03-08T05:53:26Z

+    #[test]
+    async fn fetch_state_tracks_query_replay_steady_transitions() {
+        let mut fetch_state = FetchState::default();
+        assert_eq!(fetch_state.phase, FetchPhase::Steady);


These tests are declared as #[test] async fn ..., but Rust’s standard test harness doesn’t support async tests under #[test], so this will fail to compile. Use #[tokio::test] (and update the simulate! macro similarly) or make the functions non-async when they don’t need .await.

Copilot · 2026-03-08T05:53:26Z

+export DISABLE_SOCKJS=true
+
+# Point clients to DDP Router
+export DDP_DEFAULT_CONNECTION_URL=127.0.0.1:4000


DDP_DEFAULT_CONNECTION_URL in Meteor expects a URL (including scheme). Other docs/scripts in this repo use http://127.0.0.1:4000; using 127.0.0.1:4000 without a scheme may not work consistently. Update the example to include the scheme (and path if required).

Suggested change

export DDP_DEFAULT_CONNECTION_URL=127.0.0.1:4000

export DDP_DEFAULT_CONNECTION_URL=http://127.0.0.1:4000

Grubba27 · 2026-03-12T18:26:50Z

I would prefer not to have the example in this repo...

Could you make a repo with your example and remove it from the PR?

Also, what is src/lib.rs for?

Grubba27 · 2026-03-12T18:29:14Z

Also, I see the kill streams script – but it still makes me wonder how you're killing those operations? Could you mind sharing a bit more about your example? How/when does this happen?

cbnsndwch and others added 20 commits January 29, 2026 02:05

docs: add extended architecture documentation to README.md

eae6dd8

docs: add Windows development setup guide

3b886e9

Update README.md

268ddb3

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Update README.md

8ba546f

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

chore: keep ./.local directory out of git index

f66a7ba

chore: keep vcode and github files out of the git index

fe65dbc

chore: keep vcode and github files out of the git index

1ae63be

Merge branch 'master' into feat/change-streams-resume-tokens

808b875

# Conflicts: # .gitignore

feat(watcher): configure full document lookup

24d8346

fix(cursor): subscribe before initial fetch

f6ac6e2

build(cargo): set test profile codegen-units to 1

05769ed

feat(cursor): implement change stream resume tokens

c350b1f

refactor(core): extract lib module to allow external testing

1b30123

test(cursor): add change stream resume and rwlock tests

79f7669

feat(examples): add mongo probe tool

5a8cbcc

docs(examples): add meteor repro app demonstrating ddp-router integra…

b56d9d8

…tion

fix(examples): ensure run-router.ps1 executes from workspace root to …

3214d70

…find config.toml

feat(examples): add mock data generation button to test UI reactivity

662ed0f

build(examples): add mongodb dependency to meteor-repro

969cfd2

feat(examples): add kill-streams script to meteor-repro

bd7744b

Copilot AI review requested due to automatic review settings March 8, 2026 05:48

Copilot started reviewing on behalf of cbnsndwch March 8, 2026 05:48 View session

Copilot AI reviewed Mar 8, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat: fix fetch vs watch race condition using Change Stream Resume Tokens#5

feat: fix fetch vs watch race condition using Change Stream Resume Tokens#5
cbnsndwch wants to merge 20 commits into
meteor:masterfrom
cbnsndwch:feat/change-streams-resume-tokens

cbnsndwch commented Mar 8, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Mar 8, 2026

Uh oh!

Copilot AI Mar 8, 2026

Uh oh!

Copilot AI Mar 8, 2026

Uh oh!

Copilot AI Mar 8, 2026

Uh oh!

Copilot AI Mar 8, 2026

Uh oh!

Copilot AI Mar 8, 2026

Uh oh!

Grubba27 commented Mar 12, 2026

Uh oh!

Grubba27 commented Mar 12, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

	<a href={link.url} target="_blank">{link.title}</a>
	<a href={link.url} target="_blank" rel="noopener noreferrer">{link.title}</a>

	export DDP_DEFAULT_CONNECTION_URL=127.0.0.1:4000
	export DDP_DEFAULT_CONNECTION_URL=http://127.0.0.1:4000

Uh oh!

Conversation

cbnsndwch commented Mar 8, 2026

Description

Additional Changes & Reproduction

Testing

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Mar 8, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 8, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 8, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 8, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 8, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 8, 2026

Choose a reason for hiding this comment

Uh oh!

Grubba27 commented Mar 12, 2026

Uh oh!

Grubba27 commented Mar 12, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants