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]

❌ Test Failure

Analysis: A real bug: the PR's new mongo test TestPullRecordsOffsetNeverRunsAheadOfDeliveredRecords/cut_by_max_batch_size panics deterministically in 0.00s ("last checkpoint not set, stream is still active" from model/cdc_stream.go:62 via mongo/cdc.go:587) identically across all three matrix jobs, and the 853 e2e failures are collateral from the resulting 20m package timeout.
Confidence: 0.95

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

View workflow run

[github-actions]

🔄 Flaky Test Detected

Analysis: Five unrelated PG_CH e2e subtests all hit the hardcoded 60s "wait for mirror to reach RUNNING" deadline (UNEXPECTED STATUS TIMEOUT STATUS_SNAPSHOT/SETUP) in a single matrix leg while the same commit passed on pg17 and pg18 with healthy ClickHouse containers, indicating a transient startup stall under parallel load rather than a code defect.
Confidence: 0.8

✅ Automatically retrying the workflow

View workflow run

[github-actions]

🔄 Flaky Test Detected

Analysis: All failures are mirror status/WaitFor timeouts (STATUS_SNAPSHOT/STATUS_SETUP) scattered across unrelated Mongo, PG, MySQL and API suites in two matrix legs while the pg18 leg passed all 4212 tests on the same commit, pointing to runner/service contention rather than a code defect.
Confidence: 0.82

✅ Automatically retrying the workflow

View workflow run

[jgao54]

for parallelism, it should be a dynamic config, default to 1, and can be increased on demand.

given there can be multiple pipes per worker, using GOMAXPROCS to determine pipe-level parallelism can lead to OOM given the buffering introduced in this PR

[itsbilal]

The semaphore achieves this now, right? We'll start with 1 and go up until min(6, GOMAXPROCS/2)

[jgao54]

ah, what i meant is that runtime.GOMAXPROCS(0)/2 is not a super meaningful cap. if there is only one pipe, we could probably get a better perf at higher value than runtime.GOMAXPROCS(0)/2; if there are many pipes, runtime.GOMAXPROCS(0)/2 as ceiling is still too high since the decodeWorkerSem controls per pipe concurrency, not service-level concurrency. So i recommend making it a configurable value here to give more control.

[jgao54]

Or are you thinking of modifying GOMAXPROCS env var to control pipe-level concurrency. downside with that is it would require a deploy; vs. dynamic conf can be done on-the-fly (and it would be a service-level env-var and not something we can customize on the pipe-level)

[itsbilal]

Ah yes that makes sense. We don't have a per-process control or cap in a sense, which is why I went with a fairly low maxNumDecodeWorkers value in this case. I'll replace this cap with one defined in flow/internal/dynamicconf.go which is what I believe you're referring to, and get rid of the runtime.GOMAXPROCS(0)/2 part.

[itsbilal]

Did this part!

[jgao54]

the high-level design: (1) synchronous record pull (2) batch and parallelize for decoding (3) process each batch in order makes sense to me. i think (2) with the worker pool could be simplified with a semaphore-based approach, e.g. in finishBatch (and maybe rename to dispatchBatch since finish is a bit misleading), something like the following as a rough idea:

sender := make(chan chan sendBatch, decodeWorkerBufSize*maxNumDecodeWorkers)
result := make(chan sendBatch)
decodeSem := make(chan struct{}, maxNumDecodeWorkers)


// enqueue result channel before decoding goroutine starts
select {
case sender <- result:
case ... // err handling
}

// manages concurrency and backpressure
select {
case decodeSem <- struct{}{}:
case ... // err handling
}

// spun goroutine for decoding
errGroup.Go()(func() error{
	defer func() { <-decodeSem }()
	records, err := c.decodeItems(ctx, batch)
	if err != nil {
		return err
	}
	result <- sendBatch{records: records, resumeToken: batch.resumeToken}
	return nil
})

The semaphore makes sure there are no more than N parallel decoding goroutines; which achieves what you had originally but with less worker pool orchestration.

[jgao54]

on the topic of framework, given the nature of this logic being fairly generic, would it be feasible to move it to to a separate shared module and import it here? it will also make the logic here easier to reason about as the PullRecords code is getting a bit long.

Given this PR introduces a relatively big change to mongo cdc processing, it would be ideal if we can keep the original logic for the general synchronous case, and having this logic in a separate package will make this a bit cleaner to introduce parallel-processing as a separate mode.

[itsbilal]

Implemented this approach - it significantly simplified things, thanks!

It does make sense to have a more general framework to be able to swap in parallel processing for any generic PullRecords implementation that wants to implement parallel workers while still maintaining order of records in and out. I'll do this as a follow-up, but at least with the last refactor the amount of extra code in this change has gone down significantly.

[github-actions]

❌ Test Failure

Analysis: Not flaky: flow/connectors/mongo fails to compile because the PR's new cdc_batch_test.go references an undefined identifier pullRecordsItemsBatchSize, reproducing identically across all three matrix jobs and on retry.
Confidence: 0.97

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

View workflow run

[github-actions]

❌ Test Failure

Analysis: The new pipelined-decode code in PR #4722 deterministically breaks Mongo CDC record ordering and offset tracking — its own mock-only unit tests in flow/connectors/mongo/cdc_batch_test.go fail identically in 0.25s across all three matrix jobs with out-of-order/dropped records and offsets advancing past undelivered records.
Confidence: 0.95

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

View workflow run

[pfcoperez]

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.

Wow @itsbilal, impressive! I am going through your changes again now 🙇

[github-actions]

❌ Test Failure

Analysis: Not flaky: a deterministic Go compile error in the PR's own code (connectors/mongo/cdc.go:411 "declared and not used: fullDocumentColumnName") broke the peer-flow image build in both matrix jobs, so the test step never ran.
Confidence: 0.97

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

View workflow run

[pfcoperez]

Adding AI generated diagram that helped me with the review:

flowchart TB
    CS[("MongoDB change stream")]

    subgraph main["PullRecords main goroutine (1 per batch)"]
        direction TB
        NEXT["changeStream.Next + decodeEvent<br/>(envelope only, full document kept as raw BSON)"]
        CHUNK["existingChunk []recordItems<br/>buffers up to 256 items (pullRecordsItemsChunkSize)"]
        DISPATCH["dispatchChunk()<br/>1. acquire decodeWorkerSem slot<br/>2. resultChan := make(chan sendChunk)<br/>3. sendChan ← resultChan<br/>4. workerEg.Go(decodeWorker)"]
        DRAIN["drainWorkers()<br/>dispatch partial chunk, close(sendChan), workerEg.Wait()"]
    end

    SEM{{"decodeWorkerSem<br/>chan struct{}<br/>cap = max(1, PEERDB_MONGODB_NUM_PARALLEL_DECODE_THREADS)<br/>default 1"}}
    SENDCHAN{{"sendChan<br/>chan chan sendChunk<br/>cap = 10 (workerBufferedChanSize)<br/>FIFO of per-chunk result channels"}}

    subgraph workers["decodeWorker goroutines (at most N in flight, one per chunk)"]
        direction TB
        W1["decodeWorker for chunk 1<br/>BSON → QValue, build model.Record slice"]
        W2["decodeWorker for chunk 2"]
        WK["decodeWorker for chunk k"]
    end

    RC1{{"resultChan 1<br/>chan sendChunk, unbuffered"}}
    RC2{{"resultChan 2"}}
    RCK{{"resultChan k"}}

    subgraph send["sendLoop goroutine (1 per change stream)"]
        SL["for resultChan := range sendChan<br/>chunk := ← resultChan (blocks until that chunk is decoded)<br/>RecordStream.AddRecord for each record<br/>RecordStream.UpdateLatestCheckpointText(chunk.resumeToken)"]
    end

    RS{{"req.RecordStream<br/>CDCStream records channel"}}
    SYNC["sync side of the activity<br/>(destination connector)"]

    CS -->|change events| NEXT
    NEXT -->|append| CHUNK
    CHUNK -->|"len == 256"| DISPATCH
    CHUNK -->|"batch cut: idle timeout / MaxBatchSize / stream error"| DRAIN
    DRAIN --> DISPATCH

    DISPATCH -->|"acquire: sem ← struct{}{} (blocks while N busy)"| SEM
    DISPATCH -->|"sendChan ← resultChan (blocks when 10 queued)"| SENDCHAN
    DISPATCH -.->|go| W1
    DISPATCH -.->|go| W2
    DISPATCH -.->|go| WK

    W1 -->|"resultChan ← sendChunk"| RC1
    W2 -->|"resultChan ← sendChunk"| RC2
    WK -->|"resultChan ← sendChunk"| RCK
    W1 -.->|"defer release: ← sem"| SEM
    W2 -.->|"defer release: ← sem"| SEM
    WK -.->|"defer release: ← sem"| SEM

    SENDCHAN -->|"next resultChan, in dispatch order"| SL
    RC1 -->|sendChunk| SL
    RC2 -->|sendChunk| SL
    RCK -->|sendChunk| SL
    DRAIN -.->|"close(sendChan) → sendLoop exits"| SENDCHAN

    SL -->|AddRecord| RS
    RS --> SYNC

Loading

[pfcoperez]

Did we choose to implement the semaphore as a bounded channel instead of https://pkg.go.dev/golang.org/x/sync/semaphore because of performance?

[pfcoperez]

Suggested change

	case decodeWorkerSem <- struct{}{}:
	// Acquire worker slot in semaphore
	case decodeWorkerSem <- struct{}{}:

Would it make sense to use a mini interface around decodeWorkerSem with methods .AcquireOrBlock() and .Free() to improve code readability?

[pfcoperez]

I would rename this struct to something that conveys its content better. As I understand it, this struct contains the information associated with a MongoDB event.

If the event contains a document (maybeFullDocument != nil), then we'll generate a record which itself contains the RecordItems as the map of columns and their QValues

[pfcoperez]

Each element of this collection actually represents a document record which, prior to this change, was populated with:

peerdb/flow/connectors/mongo/cdc.go

Line 381 in 3780357

addRecordItems := func(documentKey bson.Raw, maybeFullDocument *bson.Raw, items *model.RecordItems, tableName string) error {

function.

The world items is a bit overloaded and I think we should use it for the record items:

type RecordItems struct {
	ColToVal map[string]types.QValue
}

Our private recordItems, representing the source record envelop, clashes with RecordItems representing decoded, processed and ready to be sent records.

[pfcoperez]

This has the potential of further increasing parallelism, with N NewDirectBsonConverter, given that we are decoding positional elements in the same batch, this parsing could be done in N threads, one per converter.

This makes me wonder. Could we get parallelism changes simplified by exploiting it at this level and just at this level?

Instead of having M workers processing and sending batches, we could just accumulate one batch (e.g: 256 elements) launch the decoding ops in parallel, order is given by the position. Once the whole batch is one we move to the next. No sempahores, no workers management, ....