fix(mongo): pipeline document decode in CDC PullRecords

[itsbilal]

Previously, we serially decoded the change event, then the full document that was changed, inside the PullRecords loop for a given batch. This would leave a lot of CPU underutilized as the bulk of work would happen on just one thread.

This change moves the full document decode part to a different goroutine, and the record publish to another goroutine, which should allow for greater CPU utilization by the decoding function.

Existing tests pass; currently looking into benchmarking this with a fast-changing table on Mongo.

[claude]

Code review

Four issues found in the new pipelined PullRecords path.

Note: inline review comments were blocked by tool permissions in this environment, so findings are consolidated here with permalinks.

---

1. wg.Add(1) is called inside the worker goroutines, not before go

peerdb/flow/connectors/mongo/cdc.go

Lines 239 to 242 in 11938ef

	) {
	wg.Add(1)
	defer wg.Done()

Same pattern in decodeLoop; spawn sites are L560 and L607.

sync.WaitGroup requires the Add that raises the counter from zero to happen before Wait. That ordering does not hold here, and no exotic scheduling is needed to break it. decodeChan and sender are both cap 10, so for a batch of at most 10 records nothing ever blocks:

1. Main loop buffers N items into decodeChan and never blocks, so it never synchronizes with decodeLoop.
2. decodeLoop runs, Add raises the counter to 1, then does go c.sendLoop(...) at L283. The child is queued but not yet scheduled, so its Add(1) has not run.
3. decodeLoop drains all N into the sender buffer, sees recv closed, calls close(sender), returns, and Done drops the counter to 0.
4. wg.Wait() in drainWorkers returns nil, PullRecords returns nil, and the deferred req.RecordStream.Close() runs close(r.records).
5. sendLoop is finally scheduled, reads the buffered records, calls AddRecord, and panics with send on closed channel. Those records were already checkpointed but never delivered.

Secondary consequences: in recreateChangeStream, an early Wait() lets a second decodeLoop/sendLoop pair start while the old pair is still live, giving two concurrent writers to the same RecordStream and breaking ordering. Separately, a late Add(1) re-raising the counter while a Wait is in flight is exactly the sync: WaitGroup misuse: Add called concurrently with Wait panic condition.

Fix: hoist wg.Add(1) to the callers, before go c.sendLoop(...) at L283 and before both go c.decodeLoop(...) at L560 and L607, leaving only defer wg.Done() in the goroutine bodies.

---

2. The default: skip for unsupported operation types was removed

peerdb/flow/connectors/mongo/cdc.go

Lines 710 to 722 in 11938ef

	incrementRecordCount()
	items := recordItems{
	documentKey: changeEvent.DocumentKey,
	maybeFullDocument: changeEvent.FullDocument,
	operationType: operationType(changeEvent.OperationType),
	sourceTableName: sourceTableName,
	destinationTableName: destinationTableName,
	commitTimeNanos: commitTimeNanos,
	}
	select {
	case decodeChan <- items:
	case err := <-errChan:

The old switch in PullRecords had a default: branch that logged skipping event with unsupported operation type and did a continue. That branch is gone: the main loop now calls incrementRecordCount() and enqueues every event onto decodeChan without inspecting operationType.

createPipeline only matches on ns.db / ns.coll plus the user-configured excludedOps (which parseOperationType restricts to insert/update/replace/delete). There is no allowlist, so drop and rename events on a replicated collection carry ns.db and ns.coll, pass the filter, and reach decodeLoop.

Those events have no documentKey, so decodeLoop takes the document key is nil branch and fails the whole PullRecords call. Because the batch aborts, the offset is not committed and the workflow retries from LastOffset, replaying the same drop event. That turns a previously harmless warn-and-skip into a permanent retry loop.

Relatedly, the switch at L356 has no default, so record stays nil and is unconditionally sent at case sender <- record:, reaching AddRecord(ctx, nil). That path is currently shadowed by the documentKey check, but it is a live gap if the checks are ever reordered.

Fix: restore the op-type guard on the producer side before incrementRecordCount() (reuse parseOperationType and continue with the warn log), and add a default: to the switch in decodeLoop that reports an error rather than sending nil.

---

3. drainWorkers can silently swallow a worker error

peerdb/flow/connectors/mongo/cdc.go

Lines 571 to 582 in 11938ef

	select {
	case <-wgWaiter:
	case err := <-errChan:
	workerCtxCancel()
	<-wgWaiter
	return err
	case <-time.After(workerDrainTimeout):
	workerCtxCancel()
	<-wgWaiter
	return errors.New("timed out waiting for PullRecords workers to drain")
	}
	return nil

When a worker fails it writes to the buffered errChan (cap 2, so the send never blocks) and then returns, which also drives the WaitGroup to zero. Both <-wgWaiter and <-errChan are then ready, and Go picks a ready case uniformly at random, so the <-wgWaiter branch can win and fall straight through to return nil at L582. There is no non-blocking errChan re-check afterwards, and errChan is recreated per PullRecords call, so the error is discarded permanently.

This is lossy, not just noisy: checkpoint() now advances LatestCheckpointText as soon as the item is handed to the buffered decodeChan, long before sendLoop calls AddRecord. So a decode failure (failed to convert key, failed to convert document, InvalidIdValueError) or an AddRecord failure returns success for a batch whose records never reached the stream, with the offset already committed.

The batch-boundary case hits this routinely: when a bad document appears among the last records before recordCount == MaxBatchSize, decodeChan has no reader once decodeLoop dies, so the error sits in the buffer until drainWorkers races it.

Fix:

		case <-wgWaiter:
			// Workers may have exited after buffering an error; do not lose it.
			select {
			case err := <-errChan:
				return err
			default:
			}

---

4. The decodeChan send has no cancellation case, so parent-context cancellation deadlocks

peerdb/flow/connectors/mongo/cdc.go

Lines 719 to 726 in 11938ef

	}
	select {
	case decodeChan <- items:
	case err := <-errChan:
	workerCtxCancel()
	return err
	}
	checkpoint()

workerCtx is derived from ctx, and on cancellation both workers return via their case <-ctx.Done() branches (decodeLoop L386, sendLoop L256) without writing to errChan. Nothing drains decodeChan afterwards and nothing writes errChan, so both cases are permanently unready.

The reachable path does not need an unlikely interleaving. A main goroutine parked on this send is the designed steady state under backpressure: CDCStream.AddRecord blocks when the destination channel is full, which backs up sender (cap 10), which backs up decodeChan (cap 10), which parks the producer here. From that state, if sendLoop is parked at its outer select (that is, decodeLoop, the CPU-heavy stage, is the bottleneck), neither worker writes errChan and the block is permanent. A goroutine parked here never re-evaluates changeStream.Next(timeoutCtx), so the timeout path cannot rescue it. PullRecords never returns, which means the deferred cancelTimeout(), workerCtxCancel(), wg.Wait(), changeStream.Close() and req.RecordStream.Close() never run: a leaked goroutine, a leaked server-side cursor, and a CDCStream whose records channel is never closed. Parent-ctx cancellation here is routine (Temporal activity cancel on mirror pause/edit, worker shutdown).

Use ctx, not timeoutCtx, so idle timeouts still fall through to the graceful drainWorkers() path.

Fix:

		select {
		case decodeChan <- items:
		case err := <-errChan:
			workerCtxCancel()
			return err
		case <-ctx.Done():
			workerCtxCancel()
			return ctx.Err()
		}

The error-reporting selects in sendLoop / decodeLoop are worth tightening too: selecting between errChan <- err and <-ctx.Done() drops the error roughly half the time when cancellation is what caused the failure. A non-blocking send into the buffered errChan would avoid that.

[pfcoperez]

One is decodeLoop, which takes records from ... has this code comment been cut?

[itsbilal]

ah yes, my bad - forgot to finish up this comment.

[Copilot]

Pull request overview

This PR parallelizes MongoDB CDC record processing in PullRecords by offloading full-document conversion and record publishing to separate goroutines, aiming to improve CPU utilization and avoid blocking on downstream record streaming.

Changes:

• Introduces decodeLoop (BSON-to-QValue conversion + record construction) and sendLoop (publishes to req.RecordStream) worker goroutines.
• Adds worker draining/restart logic to support change stream recreation while attempting to avoid record loss.
• Refactors PullRecords hot path to enqueue lightweight items to the decode worker instead of doing full decode inline.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[pfcoperez]

Aside from the more superficial comments I left in the PR I wanted to confirm this deeper point:

The goal of the PR is to boost event processing performance through the use of parallelism in the so far serial process of:

1. Decoding BSON
2. Sending the decoded record to the record stream (req.RecordStream.AddRecord(ctx, record) ) were this action is, in turn, sends the record to another channel. When this final destination channel is full (back-pressure we already logged waiting on adding record to stream ) messages.

So I am assuming that the sought performance boost is expected to come from the added ability to decode at most 10 BSON documents when (2) back-pressure kicks in.

Q1: Without such back-pressure, both the serial version (prior to the PR) and the one with these changes should take the same time, right? Events are decoded, pushed to the channel and sent as they come. So, in this case, there is no huge performance boost beyond being able to pull more records will decoding is happening, right?

In this case (no back-pressure from AddRecord), the bottleneck could be the decoding part.

Q2: However, with this implementation as I understand it, decoding is still serial (we are not decoding two or more different events at the same time at different go-routines). Am I understanding this correctly?

It's true that it adds a buffer workerBufferedChanSize that might allow for the continued processing of events in the main loop until the decoding queue grows up to 10 (current constant value).

Then we'd allow for at most other 10 events to be sent while decoding is working.

Q3: But back to point (2) above, there we already had a channel consumed by another go-routine. So, would this change actually improve things in this case?

Q4: And, finally, if the bottleneck actually happens at AddRecord point, then following a similar reasoning we wouldn't expect noticeable performance boost through parallelism as decoding needs to wait to be able to send to the destination channel, the pipeline becomes serialized.

Q5: Would it be easy to add https://pkg.go.dev/net/http/pprof and play with some operation in the local dev env?

[pfcoperez]

I wonder if a a buffer of size 10 really makes a difference vs a size 1 if the addition back-pressures on the decoding.

Have you been able to play with different values to notice real differences?

[itsbilal]

10 ended up being useful once I added the dynamically increasing number of decode workers - but I can experiment with just matching the number of workers if you want?

[pfcoperez]

but I can experiment with just matching the number of workers if you want?

I was mostly wondering what drove choosing that number, maybe others can drive to better results? Is 10 just good enough?

If you experimented and observed benefits from 10 that's fine by me. I was curious as to were it's coming from 🙂

[Jeremyyang920]

@pfcoperez Just for some more context around this issue, there was a situation that came up while you were out where it took almost 48 hours to drain 24 hours of source side lag so this is what is driving this change.

[jgao54]

to answer this question first:

Q1: Without such back-pressure, both the serial version (prior to the PR) and the one with these changes should take the same time, right? Events are decoded, pushed to the channel and sent as they come. So, in this case, there is no huge performance boost beyond being able to pull more records will decoding is happening, right?

Back-pressure kicks in only after some time (~24 hours), so even if it doesn't kick in, this optimization will reduce latency if the processing is behind but yet triggered back-pressure. The most expensive/slow part is bson deser/ser into QRecord, so by parallelizing deserialization we can saturate cpu better and get some perf gain.

I'm a bit behind on reviewing this, taking a look now

[jgao54]

After taking an initial look, here's my thoughts:

• pipelining is going to help a bit but not give us quite the perf gain we want.
• what i had in mind was fan out the decoding to multiple workers, so decoding itself does not have to be bottlenecked
• we probably want to put this behind a feature flag to start, and enable it when needed (e.g. back-pressure observed); and later on assess enabling it globally by default.

[itsbilal]

@jgao54 @pfcoperez thanks for the reviews! I expect that the current approach will provide some perf gains by parallelizing the event decode (in PullRecords) with the document decode (in decodeLoop). Based on how some of my in-progress stresshouse testing goes, I plan to update the code with an approach that allows for multiple parallel decodeLoops that still maintain order, and if that yields a greater improvement I will go ahead with that approach.

It's also possible that the bottlenecks are elsewhere and not with full document decode. But I'm looking forward to seeing what my experimentation yields.

Also yes, I'm open to putting this behind a feature flag.

[itsbilal]

I couldn't test this end-to-end using stresshouse but instead I wrote a microbenchmark that I can point at a mongoDB and a clickhouse running locally. I will polish the benchmark up in a separate PR, but in the meantime I noticed a few things:

1. Adding more workers to parallelize decode is very necessary, as that's still the most CPU-intensive part of the whole pipe. I've made this change.
2. Adding more workers isn't sufficient on its own; we also need batching because individual documents are still relatively quick to decode, so the goroutine coordination starts to dominate unless we pass larger batches of documents at once.
3. The time it takes to Next() while we wait for the next 16MB of change events from MongoDB is also significant. Some prefetching could help here; I'll try this out next.

I've updated the code to parallelize decode across multiple workers while still maintaining order of events, and with this I now see a ~40-50% reduction in the amount of time it takes to run PullRecords ( ~0.13 elapsedMinutes logged by PullRecords down to ~0.06-0.07 for my contrived benchmark of writing 2 million rows across multiple parallel writers to mongo and CDCing it over to a ClickHouse instance). Also translates to a ~40% reduction in end-to-end replication time because we were (and continue to be) bound by PullRecords.

This should be ready for a review again. I'll scrutinize the checkpoint logic a bit more as it might now be possible to write an incorrectly-forwarded checkpoint while an inflight batch errors out and breaks the pipeline. But other than that this should be good to go.

[github-actions]

❌ Test Failure

Analysis: Not flaky: the mongo connector test package fails to compile because PR #4722 added RemainingBatchLength to the ChangeStream interface without implementing it on mockChangeStream, failing identically in all three matrix jobs and on retry.
Confidence: 0.98

⚠️ This appears to be a real bug - manual intervention needed

View workflow run

[github-actions]

🔄 Flaky Test Detected

Analysis: Two unrelated e2e tests in different matrix legs both hit the fixed 60-second SetupCDCFlowStatusQuery poll cap while the mirror was still in STATUS_SNAPSHOT (no assertion mismatch, panic, or race), a load-sensitive timeout typical of the 32-way-parallel Tilt e2e suite, and a third matrix leg passed clean.
Confidence: 0.75

✅ Automatically retrying the workflow

View workflow run

[itsbilal]

The time it takes to Next() while we wait for the next 16MB of change events from MongoDB is also significant. Some prefetching could help here; I'll try this out next.

Tried this out, didn't help significantly (< 5% improvement) so I'll keep this change as-is.

Only other TODO is to add more testing around the error / early termination cases as mentioned previously, but otherwise this is ready for a look.

[jgao54]

The time it takes to Next() while we wait for the next 16MB of change events from MongoDB is also significant. Some prefetching could help here; I'll try this out next

this would also depends on network roundtrip between peerdb node and source mongo node, and I'd expect putting them in the same region should shorten this if they are tested far away from each other.

[jgao54]

(i haven't yet looked at the code changes but just commenting based on what you wrote)

I'll scrutinize the checkpoint logic a bit more as it might now be possible to write an incorrectly-forwarded checkpoint while an inflight batch errors out and breaks the pipeline.

yeah this one needs to be very careful as order matters. before we checkpoint offset after every event in-memory, and then persist it to disk at the end of a batch. if we introduce concurrent checkpointing, we have to make sure the goroutines are coordinated in a way such that we never checkpoint until all the worker are past a certain point (kind of similar to cockroachdb's quirk we had to workaround recently, only this we have to manage it ourselves)

if an error is hit, it's less of a concern because we don't persist checkpoint to disk when we encounter an error, so on retry we just resume from previously checkpointed token.

edit: on second thought, maybe the worst case is we checkpoint an older offset than the latest event we send to the stream (when the events are processed out-of-order), so worst case is duplicates, not missing data... (but would still be good to avoid if possible).

[itsbilal]

if we introduce concurrent checkpointing, we have to make sure the goroutines are coordinated in a way such that we never checkpoint until all the worker are past a certain point (kind of similar to cockroachdb's quirk we had to workaround recently, only this we have to manage it ourselves)

This ended up being relatively simple to do; I just moved checkpoint responsibility over to sendLoop and tracked resume tokens alongside batches as they flowed through the goroutines. There's one case where we can have a resumeToken that lags behind a sent record, and that is if RecordStream.AddRecord itself errors out mid-batch. I can track resume tokens at record level if duplicates when resuming in that case are an issue; I had figured (possibly incorrectly) that the normalize step would flatten duplicates after an error.

If duplicates in that specific error case are a non-issue, the current approach is slightly cleaner and marginally more performant.

Also added tests for the error cases, this should be ready for a look, thanks!

[github-actions]

❌ Test Failure

Analysis: Real bug, not flake: the PR's new call to GetLastCheckpoint() at flow/connectors/mongo/cdc.go:588 runs before the CDC stream is closed, panicking with "last checkpoint not set, stream is still active" identically on all three matrix legs, and wedging the entire e2e package into a 20-minute timeout.
Confidence: 0.93

⚠️ This appears to be a real bug - manual intervention needed

View workflow run

[github-actions]

🔄 Flaky Test Detected

Analysis: Not a test failure at all — all three jobs failed in the install lib-geos and pg_dump setup step when apt-get update hit a transient "Hash Sum mismatch" fetching Google's Chrome apt repo index (exit 100), so the test step never ran.
Confidence: 0.98

✅ Automatically retrying the workflow

View workflow run

[github-actions]

🔄 Flaky Test Detected

Analysis: Not a test failure — both matrix jobs died in the "install lib-geos and pg_dump" setup step when sudo apt-get update exited 100 due to a transient Hash Sum mismatch on the runner image's pre-installed Google Chrome apt repo, so no tests ever ran.
Confidence: 0.97

✅ Automatically retrying the workflow

View workflow run

[github-actions]

🔄 Flaky Test Detected

Analysis: Not a test failure at all — all three matrix jobs died in the "install lib-geos and pg_dump" setup step because apt-get update got a "Hash Sum mismatch" from the runner image's Google Chrome apt repo (CDN metadata lag), exiting 100 before any test ran.
Confidence: 0.97

✅ Automatically retrying the workflow

View workflow run

[github-actions]

🔄 Flaky Test Detected

Analysis: Not a test failure at all — all three flow_test matrix jobs aborted during setup because apt-get update exited 100 on a transient "Hash Sum mismatch" from Google's pre-installed chrome-stable apt repo, so no tests ran; a re-run should pass.
Confidence: 0.97

✅ Automatically retrying the workflow

View workflow run

[github-actions]

🔄 Flaky Test Detected

Analysis: TestPullRecordsWorkerPoolErrorFromWorker asserts one specific worker's error token ("token-3") but a different worker's error ("token-1") won the goroutine race, and the same test passed in the pg18 matrix job on the identical commit — a race-dependent assertion in a test newly added by PR #4722.
Confidence: 0.85

✅ Automatically retrying the workflow

View workflow run

[pfcoperez]

@fable run a review on these changes focusing on potential data loss or event order changes regressions.

[claude]

Claude finished @pfcoperez's task in 11m 9s —— View job

---

Review: data loss & event ordering in the pipelined PullRecords

• Gather context (changed files, worker pool, CDC pull path)
• Analyze pkg/common/workers.go for ordering/loss hazards
• Analyze flow/connectors/mongo/cdc.go (checkpointing, stream recreation, error paths)
• Verify checkpoint/commit interplay with flow/activities/flowable_core.go and model.CDCStream
• Review tests for loss/ordering coverage
• Post findings

TL;DR: I found no data-loss or reordering regression in the happy and error paths I traced. The core invariants hold by construction and are well tested. I did find one config-triggered wedge/panic (decode threads ≤ 0), one error-path behavior that delays failure and buffers unboundedly within a batch, and a couple of smaller hardening items.

Why ordering and checkpoint safety hold

Walking the design end to end:

• Ordering: AddItem/Flush run on the single pull goroutine, and Flush enqueues each chunk's unbuffered result channel into p.sender before spawning its worker. The single send loop consumes those result channels strictly in FIFO order and blocks on each one, so chunks reach Send in dispatch order no matter how workers interleave. A failed chunk blocks the send loop on its result channel until the errgroup cancels it, so later chunks can never leapfrog a failed one.
• Checkpoint never runs ahead of delivered records: recordSender calls UpdateLatestCheckpointText only after every record of the chunk was accepted by AddRecord, and the token is the resume token of the chunk's last event. The batch-cut paths (MaxBatchSize, idle timeout, fatal error) all run Flush + Wait before touching checkpoint() or recreating the stream, so the PBRT checkpoint at cdc.go:594 and the recreate at cdc.go:602/cdc.go:611 only happen once every in-flight record is either delivered or the pull has failed. Worst case anywhere is a checkpoint that lags delivered records → duplicates on retry, which is the right at-least-once failure mode.
• Activity-level commit: flowable_core.go:347-360 only reads GetLastCheckpoint() and commits after errGroup.Wait() succeeds, i.e. after a clean pull, so a failed pull doesn't commit an offset (matching the new tests' requireOffsetsCoverOnlyDeliveredRecords invariant — that helper is a genuinely nice addition).
• The classic deferred-decode hazard is safe: changeStream.Current() is only valid until the next Next(), but decodeEvent (bson.Unmarshal) copies documentKey/fullDocument into freshly allocated bson.Raw buffers (the driver's raw-value decoder copies document bytes), so handing them to workers past subsequent Next() calls does not alias the cursor buffer.
• No send-on-closed-stream: defer order guarantees workerPool.Wait runs before req.RecordStream.Close(), and pool re-Init in recreateChangeStream only ever happens after a completed Wait, so there is never more than one live send loop per stream.

Findings

1. PEERDB_MONGODB_NUM_PARALLEL_DECODE_THREADS ≤ 0 wedges or panics the pull (medium)

Concurrency is taken verbatim from the dynamic config at cdc.go:464-469 and used as the semaphore capacity at workers.go:54. If an operator sets it to 0, p.sem <- struct{}{} in Flush can never proceed (no worker ever releases a slot) and the mirror hangs in AddItem until the activity is killed; a negative value panics in make(chan struct{}, n). Since this knob exists precisely to be tuned on live mirrors, clamp it: Concurrency: max(1, int(numParallelDecodeWorkers)) (or validate in Init). Fix this →

2. After a worker/send error, Flush returns nil and the pull keeps buffering until the batch cut (medium)

Once any worker or Send fails, the errgroup cancels workerCtx, and both selects in Flush take case <-p.workerCtx.Done(): return nil (workers.go:113-114, 142-143). From then until MaxBatchSize or the idle timeout, the pull loop keeps reading events and AddItem keeps appending — and because those Flush paths return before resetting inProgressChunk, the slice grows past ChunkSize without bound (each entry holding full raw BSON documents), while up to ~10 already-dispatched chunks get decoded and dropped. No records are lost — the eventual Wait at the batch boundary surfaces the error and nothing is committed — but the failure is detected a full batch late with avoidable memory growth. Suggest Flush return the pool's error instead: case <-p.workerCtx.Done(): return context.Cause(p.workerCtx) (errgroup sets the first error as the cancel cause), which makes the pull loop abort promptly; the same path currently also leaks a semaphore slot when the sender-enqueue select hits workerCtx.Done after acquiring one (harmless today only because the pool is already dead). Fix this →

3. First-batch empty-checkpoint overwrite window (low, pre-existing, cheap to close)

The stream's lastCheckpointText starts as "" and is never seeded with req.LastOffset.Text. If AddRecord fails mid-first-chunk after SignalAsNotEmpty (cdc.go:257), the stream can end up non-empty with checkpoint ""; a destination that races its batch-finish ahead of the pull error propagating could persist Text: "", and the next pull would then start a change stream with no ResumeAfter — skipping everything since the real offset. The window is very narrow (the pull error cancels the sync ctx via the shared errgroup) and the same window existed in the serial code, so this is not a regression — but jgao54's earlier suggestion of seeding req.RecordStream.UpdateLatestCheckpointText(req.LastOffset.Text) at the top of PullRecords is a one-liner that eliminates the class entirely, and I'd take it.

4. Effective decode parallelism is silently capped at ~11 (perf note)

workerBufferedChanSize = 10 bounds dispatched-but-unsent chunks: with Concurrency above ~11, Flush blocks on p.sender <- sendChan regardless of free semaphore slots. If someone cranks the dynamic config to 16+, they won't get it. Either size sender as max(workerBufferedChanSize, Concurrency) or document the cap next to the constant (whose comment still references the removed decodeWorkerBufSize).

5. TestPullRecordsWorkerPoolErrorFromWorker flake (test)

CI flagged this test flaking on run 34385388819 (lastResumeToken = token-1 instead of token-3). I traced the current barrier-based version and could not construct an interleaving that produces token-1 — the barrier forces Send([2,3]) to complete before the failing worker can return — so I believe the flake predates the barrier commit. Worth confirming the flake doesn't recur on 7ec39fc before merging, since this test guards the ordering invariant.

Test coverage suggestion

The one loss/ordering-relevant path without coverage is stream recreation with records already delivered: recreateChangeStream after a resume token not found error mid-batch (the deadline-exceeded recreate is only ever hit with recordCount == 0). That's the path where a future regression in the drain-before-Init ordering would produce two concurrent send loops and interleaved records. An iterationType for a resume-token-not-found error (e.g. insert, insert, rtnfError, insert, idle) asserting gap-free order across the recreation would lock it down.

Nits

• flow/connectors/mongo/cdc.go:240-243: the doc comment on recordSender is stale — it describes the old sendLoop design and contains a duplicated fragment ("to signal timeouts to signal timeouts").
• Two functions named decodeEvent (package-level at cdc.go:223, method at cdc.go:281) — the method decodes a chunk of events; decodeChunk/decodeEvents would disambiguate.
• workers.go:126: sendMsg := sendMsg[D, RT]{...} shadows the type name.
• cdc.go:521: incrementRecordCount reassigns timeoutCtx, cancelTimeout without cancelling the prior hour-long timeout ctx, leaking its timer (pre-existing pattern).

Nothing here blocks the PR on the data-loss/ordering question the review asked about — findings 1 and 2 are the ones I'd address before merge.

---

[pfcoperez]

@itsbilal From the Fable review I'd address this one, specially making sure the defaults don't lead to hanging syncs:

oncurrency is taken verbatim from the dynamic config at cdc.go:464-469 and used as the semaphore capacity at workers.go:54. If an operator sets it to 0, p.sem <- struct{}{} in Flush can never proceed (no worker ever releases a slot) and the mirror hangs in AddItem until the activity is killed; a negative value panics in make(chan struct{}, n). Since this knob exists precisely to be tuned on live mirrors, clamp it: Concurrency: max(1, int(numParallelDecodeWorkers)) (or validate in Init). Fix this →

[itsbilal]

@pfcoperez addressed! thanks

[pfcoperez]

Nice!

[pfcoperez]

This reads like a charm now, thank you!

[github-actions]

❌ Test Failure

Analysis: Not flaky: the new unit test TestPullRecordsWorkerPoolErrorFromWorker (flow/pkg/common/workers_test.go:76) fails deterministically in all three matrix jobs in 0.00s with an unexpected "decode failed" error, indicating a real bug in the PR's new worker-pool decode error handling rather than an environmental or timing issue.
Confidence: 0.93

⚠️ This appears to be a real bug - manual intervention needed

View workflow run

[github-actions]

🔄 Flaky Test Detected

Analysis: A single CDC event was delivered twice in TestCockroachDBSuite/Test_CDC_Exactly_Once_Across_Batches on only the crdb-v26 matrix leg (the same test passed on crdb-v24/v25 in the same run), which matches the timing-sensitive at-least-once replay behavior the CockroachDB changefeed connector's resolved-timestamp buffering guards against, rather than a deterministic bug.
Confidence: 0.72

✅ Automatically retrying the workflow

View workflow run

[jgao54]

one thing i missed earlier during review that may worth a follow-up: currently this hits the db on every PullRecords, which can be as frequent as every second. so applied to all the pipes in all services this can cause some db pressure.

given that this changes infrequently, and requires a pause/resume to update anyways, can be a connector level setting that gets read once upfront (similar to how we handle PeerDBMongoDBExcludedOperationTypes today).