Repository navigation
Change data capture: change streams on the log service, cluster gate, reference consumer - #506
Merged
Merged
Conversation
cdc-change-stream.md specifies v1 of change data capture for EloqDoc: a pull-model change stream served by the runtime, sourced from the log service and consumed by an external process that writes into a destination MongoDB. Section 2 records the facts about the tx service and log service the design rests on, each cited to the code it was verified against, so a later change to any of them can be checked against what depends on it. cdc-implementation-plan.md sequences that design into six workstreams, four milestones and fourteen PRs, and assigns every item of the design's "items to confirm during implementation" list to the task that closes it. Two scope notes for readers of this branch: Both documents describe the whole v1 effort, not just this branch. Their status annotations therefore cover work that lands in later PRs -- in particular the postBatchResumeToken plumbing and the end-of-pipeline resume token integrity check (plan tasks C9 and C14, PR 6), which the plan marks as implemented. That code is not part of this branch. The plan is a draft and is expected to move as the work lands. Where an implementation diverges from the design, the design doc is to be fixed in the same change, per its own rule; several sections here already carry corrections made that way while writing the codecs in this branch, including the resume token byte order in section 5.6 and the decoder notes in section 5.4 step 3. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The exact-integer branch for a decimal carrying a continuation was chosen by
comparing the int64 integer part against kMaxIntForDouble, which is typed
double. That converts the integer first, and 2^53 + 1 converts to 2^53, so
the branch that exists precisely for values a double cannot hold was skipped.
The decimal was then rebuilt from a binary fraction that cannot represent the
integer part, and came back one unit low:
NumberDecimal("9007199254740993.5") -> 9007199254740992.5
Only integer parts that round to 2^53 are affected, so exactly 2^53 + 1;
2^53 and 2^53 + 2 are representable and already compared correctly. The
encoding was never lossy -- the same bytes decode exactly once the bound is
compared as an integer.
Every caller of KeyString::toBson inherited this, EloqIndexCursor::_curr
among them when a plan requests kWantKey. Decode-only change; no storage
format moves and nothing on disk is affected.
storage_key_string_test passes (38 tests, including its randomized
round-trip fuzzing).
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
First code for the change-stream design in docs/cdc-change-stream.md
(implementation plan PR 2, tasks C3 and C8a). Pure codecs, no wiring:
nothing calls them yet.
CdcKeyDecoder decodes KeyString V1 bytes without type bits, which is all a
delete event has. No record follows a Delete op in the redo log, so an event's
documentKey._id has to come from the logged key alone, and KeyString::toBson
cannot be used for that: with empty type bits it reads every numeric as kInt,
narrowing an int64 _id above 32 bits, and it uasserts on a key with a
fractional part or a decimal.
Byte consumption is identical to the stock decoder, because in V1 the layout
of every branch is fixed by the CType byte and by markers inside the key --
the fraction bit, the decimal-continuation markers, the small-magnitude
selector -- while type bits only choose the BSON type of the output. Where
they would have chosen, this decoder picks the value-preserving
representation: int32 when it fits and int64 otherwise, int32 0 for every
zero, double NaN for every NaN, String for a Symbol. MongoDB compares numbers
across int, long, double and Decimal128 by value and treats all zeros and all
NaNs as equal, so the result always compares equal to the original and
deleteOne({_id}) matches. Consumers needing byte-stable routing use the
event's eloq.keyBytes field instead.
CdcResumeToken encodes the token layout of design section 5.6 and the
{_data: <hex>} shape drivers expect, big-endian throughout. Two hardening
rules on the parse path, which takes bytes straight from a client: duplicate
log-group ids are tracked in a hash set rather than by rescanning the entries
so far, since the count reaches 65535 where a pairwise scan is roughly 2.1
billion comparisons for a token of about 1 MB; and a watermark at the maximum
timestamp is rejected, because its successor would wrap to 0 and restart the
stream at the beginning of history rather than past its end.
Both live in a new dependency-light cdc_codec library, so the change-stream
stages can link them without pulling in the storage engine and the tests stay
cheap to build.
55 unit tests. Each key-decoder case encodes its value with the real KeyString
encoder followed by a sentinel field, which is what proves exact byte
consumption: a decoder off by a byte corrupts the sentinel. Equality with the
original and agreement with the stock decoder are both asserted
unconditionally, with no allowance for the two disagreeing.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Implementation plan task C2. CdcScan returns each node group's data-log blob exactly as stored, so the change-stream source needs a reader for the format TransactionExecution::FillDataLogRequest writes: a sequence of per-table sections, each a name, engine and table type followed by a length-prefixed run of entries, and each entry a length-prefixed KeyString, a one-byte op and, for every op but Delete, a serialized record. The layout was checked against the writer and against the reader recovery uses (ReplayLogCc's section walk and its entry loop in template_cc_map.h), including the contract that a record follows every non-Delete op: nothing else delimits an entry. CdcBlobDecoder walks one blob and yields each entry in a stream's scope as views into the blob: namespace, op, the logged _id KeyString (which an event carries verbatim as eloq.keyBytes and CdcKeyDecoder turns into documentKey), and the record, from which documentBson() extracts the document. Sections the stream does not want -- other engines' tables, secondary indexes, namespaces outside the scope, and local/admin/config for every scope -- are skipped by length without being parsed, so EloqKV object-command sections pass through. Positions count what was skipped. A resume token names an entry by the ordinal of its section in the blob and of the entry within that section, and that must mean the same thing to a collection stream and a deployment stream over the same record, so filtering changes which entries are produced but never how they are numbered. The bytes come from the log, so any malformed or truncated input is treated as corruption and throws rather than being skipped. Tests (23, in cdc_codec_test): each op and mixed ops kept in order, records carrying unpack info, arbitrary bytes in keys and documents, the three scopes and the sections every scope skips, positions that stay the same under every scope, and empty blobs, sections, keys and documents; then rejection of a blob truncated at every prefix, of an unknown op byte, and of each length field that points past its section or the blob, with the entry-level cases checked to fail inside the entry rather than against the section as a whole. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Implementation plan task C9. A change stream reports progress to its client through postBatchResumeToken, and this tree predates 4.0.7, so the field does not exist at any layer: CursorResponseBuilder emits only the top-level $_internalLatestOplogTimestamp, PlanExecutor offers only getLatestOplogTimestamp(), ClientCursor holds no token, and there is no ChangeStreamProxyStage. It is introduced here at every layer the existing timestamp already passes through, plus two the timestamp never touches. A pipeline's first stage supplies the token by implementing the new PostBatchResumeTokenProvider. PipelineProxyStage resolves that stage once, at construction, through PipelineD::getPostBatchResumeTokenProvider, since the first stage never changes once the pipeline runs and a per-document dynamic_cast would tax every aggregation; it and PlanExecutor then pass the token up the same chain getLatestOplogTimestamp uses, and both also report whether the pipeline tracks tokens at all. ClientCursor holds the token last reported, so an aggregate with batchSize 0, which pulls nothing, still returns one; and CursorResponseBuilder emits it inside the cursor sub-object, where drivers look for it, rather than at the top level where the internal timestamp goes. The opening token is taken when the cursor is registered, and for a pipeline that tracks tokens an empty one there fails the command with the new ChangeStreamFatalError instead of being stored. Otherwise a source that could not name its start position in time would leave a cursor that looks like an ordinary one, and every reply would silently omit postBatchResumeToken. Both batch loops apply one rule. aggregate builds firstBatch in handleCursorCommand and getMore builds nextBatch in generateBatch, so both change identically: start from the cursor's token, take each appended document's _id, take the executor's watermark at IS_EOF, record nothing for a stashed document, and keep the current value when CloseChangeStream ends the pipeline after an invalidate. Recording only at append time is what keeps the reported token from running ahead of what the client received. Cursors whose pipeline provides no token are untouched, so no non-change-stream reply changes shape. The token travels inside the document as its _id, because the Document to BSON conversion ahead of the batch builder discards metadata. That makes it rewritable by a user stage, and rejecting a $project that drops _id does not cover it: $addFields, a computed $project and $replaceRoot can all replace or reshape _id, and no open-time inspection rules them out. So the guarantee is an end-of-pipeline check instead. Every stage permitted after $changeStream is streaming and yields at most one output per input, so a document leaving the pipeline always derives from the last one the internal stages emitted; PipelineProxyStage::getNextBson compares the two _ids byte for byte and fails the command with ChangeStreamFatalError. Every document leaves through that one point, including documents the batch loops later stash. Tests: six cases in cursor_response_test for the response layer, and a new resume_token_integrity_test with fourteen cases covering _id removed, replaced, swapped for another valid token, and modified or extended in a nested field, against $project, $addFields and $replaceRoot, plus the pass-through cases. Every case pulls through a real PipelineProxyStage, the one place the check lives, so the test cannot drift from it. The stage that provides the token is task C5 and is not in this commit, so every cursor currently carries an empty token and this code is inert. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Implementation plan task C14, a prerequisite for every eloq_cdc jstest that reads a token: the shell has no postBatchResumeToken handling at all, so a jstest cannot observe what the server now reports. DBCommandCursor records the token from the opening reply and from every getMore reply, and getResumeToken() returns the position the caller has actually received. The two are not the same thing, and conflating them hands out a token that skips unread events: hasNext() issues a getMore to refill an empty batch, which replaces the batch token, but nothing from that batch has been returned yet. So the consumed position is tracked separately, advanced by next() and snapshotted before a getMore replaces the batch token. The batch token is reported only once its batch is empty, an unread opening batch reports nothing rather than a fabricated value, and a cursor the server sends no token for never reports one. The watch() helpers on DBCollection, DB and Mongo also forward startAfter, which they did not previously pass through: into the $changeStream stage, and not down as an aggregate option, where the server would reject or ignore it and silently open a fresh stream instead of resuming. Tests in tests/jstests/eloq_cdc/shell_resume_token.js drive the real DBCommandCursor against canned replies through a stubbed db, so they need no server: the prefetch case above, empty getMores advancing to their watermark, an unread opening batch, an ordinary cursor, and the batch that delivers invalidate. tests/jstests/eloq_cdc/shell_watch_helpers.js covers the three watch() helpers the same way, each with its route to aggregate stubbed: startAfter lands in the stage and not in the options, other options and the user's stages still pass through, resumeAfter is unchanged, both resume options together are forwarded for the server to reject, and Mongo.watch opens a deployment-level stream on admin. Wiring the suite into the two CI jstest entry points is task F2. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
While an index is being built, EloqRecordStore writes each document's keys into it on every insert, update and delete. Three bugs there: - Updates left stale keys. updateRecord gets only the new bytes, so it wrote the new keys and never removed the old ones; every update that changed an indexed field left one stale key in the index being built. - A document without the indexed field, written during a sparse index build, failed invariant(keys.size() > 0) in _insertRecords and aborted the server; updateRecord dereferenced the first key unchecked. - Only the first key was written, so array documents written during a multikey build were partly unindexed. The record store now takes the document the caller already read: RecordStore gains deleteRecordWithPreImage and updateRecordWithPreImage (defaulting to deleteRecord/updateRecord), and CollectionImpl passes the document it holds for deleteDocument and updateDocument. The update is read-for-write, so that document is the current version. One helper, maintainCreatingIndexes, handles insert, update and delete for every index under construction: it removes the old keys the new document no longer has, writes every new key (none for a sparse miss), lays out a unique index's key without the record id as EloqUniqueIndex does, and still refuses a write that would add multikey paths the index has not recorded. Without a pre-image (other callers), the record store reads the document once, only while the table has indexes under construction. Verified with a Debug build: inserts, updates and deletes during index builds leave no stale and no missing keys; multikey updates and deletes during a build keep the index complete. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
The tx service builds an index from the existing documents without checking them for duplicate keys: the generator emits unique keys unchecked and the upload keeps the first entry per key, so a "unique" index built over duplicates silently drops documents from the index. Duplicate detection during the build is deferred to a later project. Until then, MultiIndexBlockImpl::init refuses createIndexes with unique: true on a secondary index of a non-empty collection, with CannotCreateIndex. Unique indexes on empty collections are unaffected: DML on a built unique index is checked, and inserts and updates are refused while one is being built. A document inserted between this check and the build's start is not covered. Three jstests build a unique index on a non-empty collection and are skipped: apply_ops_dups.js and covered_index_simple_2.js in eloq_core and core_eloq, and covered_index_simple_3.js in eloq_core (core_eloq already skipped it, for a multikey reason). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
WITH_FAULT_INJECT is a new CMake option of the eloq module: it compiles
the data substrate's fault points in (ACTION_FAULT_INJECTOR,
CODE_FAULT_INJECTOR). It is on by default for Debug builds, as in
EloqKV, and applied as a compile option, not a definition, because the
top-level CMakeLists passes data_substrate's COMPILE_DEFINITIONS to every
Mongo target and the fault macros are used only inside the data
substrate. Until now EloqDoc had no fault points at all: the top-level
CMakeLists defines FAULT_INJECTOR for Debug builds, which nothing checks.
eloqFaultInject (src/eloq_test_commands.cpp, a test command registered
with enableTestCommands) arms, disarms or queries a fault point through
FaultInjectTxRequest, as EloqKV's FAULT_INJECT command does:
{eloqFaultInject: <name>, params, nodes} arms it, params "remove"
disarms it, and {eloqFaultInject: <name>, query: true} replies whether
it is armed on this node. On a build without the fault points every form
fails with CommandNotSupported (FaultInject::CompiledIn), so a test
cannot pass without exercising its fault.
New jstest suite eloq_fault (tests/jstests/eloq_fault), for a server
built with the fault points; each test checks that its fault fired:
- add_index_failure_paths.js: the kv table creation and the prepare log
fail createIndex with their own errors, leaving no index and a
writable collection, and a retry builds a complete index; a failed kv
catalog write after the commit log is retried (the one-shot fault
reads disarmed afterwards) and the build succeeds.
- range_split_upsert_retry.js: a range split whose first UpsertRanges
fails still finds every document once the retry has run (350,000
documents of about 1 KB, past the 256 MB range limit).
CI's CMake server build turns WITH_FAULT_INJECT on, and
cmake_ci_runtime.sh adds eloq_fault to the jstest suites whenever the
build has it; test_server_fixture.py checks both. docs/how-to-compile.md
documents the option. SCons builds the command too; its CI job, whose
build has no fault points, does not run eloq_fault.
Needs data substrate commit bb69dfa (FaultInject::CompiledIn and the
add-index fault points) and b4bbf5d (the range-split fault point); the
submodule pointer moves with PR 5.
Verified with a Debug build, running the jstests with a 4.0 shell: both
pass and the server log shows each fault fired; on a build without
WITH_FAULT_INJECT, eloqFaultInject fails with CommandNotSupported and
the add-index test fails at its first arm. Configure-only runs: the
option is OFF for RelWithDebInfo, ON with -DWITH_FAULT_INJECT=ON and for
Debug, and reaches only data substrate objects. The CI fixture tests
pass.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Design doc (docs/cdc-change-stream.md): - F7/F9 and §4.4(b): an index op's PrepareCommit installs its timestamp D as the catalog entry's commit ts, so every DML that sees the dirty schema commits above D; replay of the prepare log does the same. - F8, §4.3 and §4.4(b): the index's kv table is created before the prepare log, under the prepare-stage write lock, and dropped if the creation or the prepare log fails; the new schema reaches the kv catalog once, after the commit log, under the commit-stage write lock, which orders it before any later DDL's catalog write. So a build constraint violation is the only add-index rollback after the prepare log. The cost (DML and the checkpoint wait on the creation, tried once) is stated as B3 states create's. Implementation plan (docs/cdc-implementation-plan.md): - PR 4 row: the B1 lock registry, the kv table before the prepare log, the catalog write after the commit log, and the catalog entry commit ts. - B8 row: the only add-index failure after the prepare log is a build constraint violation; test it through PACK_SK_ERR. - §11: unique index builds on non-empty collections are deferred to a later project and refused until then (the residual window, the three skipped jstests). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…l_ts, txn) Decisions for A3 (the logger's DDL store), settled while reviewing plan PR 5. Design doc (docs/cdc-change-stream.md): - §4.3: the DDL history is kept in memory by the log state, saved in the raft snapshot as tx_catalog_ops_ already is, and trimmed at the CDC retention floor like data records, not an hour after the checkpoint. No RocksDB column family. - §4.3: one record per schema op. An index op that adds or drops several indexes, as a createIndexes with several specs does (the tx service builds them in one scan), is one record and one event. - §4.1, §4.2, §4.3: a record's position is (ddl_ts, txn), encoded be64 ddl_ts | be64 txn wherever the continuation and tokens record a key; there is no separate storage key. DDLs on one table are serialized by its catalog write lock, so only DDLs on different tables, hence of different transactions, share a timestamp; the two records of a rolled-back add index share a transaction but not a timestamp. The order within a timestamp carries no meaning. The runtime rebuilds the position from ddl_ts and txn_number. - §4.3 "What is recorded": index ops arrive as a table_op with op_type AddIndex or DropIndex and their index names in alter_table_info_blob; SchemaOpMessage.index_op is never set. A rolled-back add index's dropIndexes covers every index the op added. - §5.4, §5.4a: createIndexes and dropIndexes list one entry per index the op added or dropped; one DDL record still yields exactly one event. - §5.6: index DDL identity is the indexes one op added or dropped; a token's DDL key is the record's position; resuming at a collection stream's own drop re-reads the record at an entry whose ddl_ts equals the token's timestamp and checks for a DropTable of the watched collection. - §6: the consumer drops each index a dropIndexes event lists. Implementation plan (docs/cdc-implementation-plan.md), A3 row: the in-memory history and its position, the table_op filter, one record per schema op (tested with an index op that adds two indexes), updating the DDL wording of log.proto and cdc_scan.h, and CdcSourceRecord::ddlKey() on the EloqDoc side for C8b. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…tamp
Plan PR 5 (docs/cdc-change-stream.md §4.2, §5.1, §5.3 steps 1-2, §5.8;
plan F1, C4, C5): the record source a change-stream cursor pulls from.
Nothing wires it into a pipeline yet; PR 6 does.
New cdc_scan library (src/mongo/db/modules/eloq/src/cdc, eloq
SConscript):
- CdcWaitObject: where scan completions meet the waiting coroutine. An
RPC callback or the deadline timer only calls notify(), which resumes
a registered coroutine exactly once through its resume functor, or
records the wake-up for the next wait; no callback holds a coroutine,
which may have returned on its deadline. A coroutine yields with a
bthread timer bounding the wait, and the timer holds its own reference
to the object; a caller without coroutine functors blocks on a
condition variable.
- CdcScanTransport, and TxLogCdcScanTransport over the tx service's
TxLog (the new TxLog::CdcScan and LogNodeByAddress, and the existing
UpdateLeaderCache and RequestRefreshLeader).
- CdcScanClient (C4): one call in flight per group, its state shared
with its callback, so a response that lands after the getMore returned
waits for the next one; the destructor cancels calls and never waits.
A failed scan is retried from the same position: a NotLeader redirect
naming another node ID is followed at once, unless the previous
redirect was followed within one backoff period, so two nodes naming
each other during an election are not bounced between; an empty
redirect or one naming the node just asked backs off; a transport
error or an unknown address asks for a leader refresh and backs off.
Redirects are compared by node ID, since the log service may name a
node by hostname or by IP address.
- CdcScanSource (C5): per group the continuation, the buffer and
scanned_ts (last_ts - 1 for a cut page, else arrival_complete_ts);
emit_bound and drained. It releases the lowest buffered record once
every other group's scanned_ts is at least its timestamp minus one,
and fetches a group whose buffer is empty or that lags, under the
buffer cap except for the lagging group, with wait_ms clamped to the
deadline. An empty exhausted page that leaves a group's bound where it
was pauses that group (doubling from 100 ms to 1 s) unless the logger
held the scan for at least half its wait; a wait under 20 ms never
counts as held, and after a call's first pass no scan goes out with
less than 20 ms left. fixStart() fixes a token-less stream's start
S = min arrival_complete_ts, probing each group at once and
re-probing a group that reports 0 with a long poll. next() returns
none at its deadline instead of throwing; its header states the
deadline contract PR 6 must follow (now for the aggregate's first
batch, the maxTimeMS deadline for a getMore).
- A position below a group's retention floor raises the new error code
ChangeStreamHistoryLost (286).
F1, the scripted log service (cdc_fake_log_service.{h,cpp}), follows the
logger's paging contract (merged prefixes, cuts inside a timestamp, the
two-key continuation) and scripts NotLeader redirects, transport errors,
node IDs known by several addresses, and inline, threaded or held
completion. It is compiled only into cdc_scan_test.
cdc_scan_test, 30 cases: the wait object on both paths (a fake
coroutine whose yield blocks until a resume from another thread: a
notify wins, the timer wins, a notify between registering and yielding,
a stale timer waking only the next wait, the timer's reference released
exactly once); merge order and exactly-once delivery across three
groups, inline, threaded and under a coroutine; cut pages; the emission
rule and a lagging group; idle advance of emit_bound; the idle scan
rate, back-to-back short waits and the last 20 ms of a wait; the buffer
cap; fixStart; every retry case, including a redirect naming the node
just asked by another address and two nodes naming each other; history
loss; and calls outliving their waiter or their source.
Docs: design §5.3 step 1 states the redirect-following cap and when a
page counts as held; the plan marks PR 3 and PR 4 as bumped here and
PR 5 as done.
Bumps the data_substrate submodule to 6080145, past PR 1 (log.proto),
PR 3 (the CdcScan log service), PR 4 (the checkpoint bound and add
index), the range-split fix and the TxLog scan calls.
cdc_scan_test 30/30 over repeated runs; data_substrate unit suite
89/89; eloqdoc builds and links.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
A KeyString encodes numbers by value, so NumberInt(5), NumberLong(5) and 5.0 share key bytes; the original type lives in the type bits, which an index entry keeps in its unpack info. EloqIndexCursor::_idRead reset the cursor's type bits after a hit, so a key read back from seekExact was decoded as if every number were an int32: NumberLong(5000000000) came back narrowed, and a fractional double or decimal failed to decode (50810). The lookup itself, the RecordId and the document were right. Only findSingle's debug check reads that key, so a Debug server aborted on every find, update or remove by such an _id; Release was unaffected. _idRead now reads the type bits from the entry's unpack info, as the scan path (_updateIdAndTypeBits) already does; an entry written without unpack info gives all-zero type bits, which is what its writer meant. eloq_basic/id_point_lookup_types.js finds, updates and removes by int64 values beyond int32, doubles, a decimal, an int and a string, and looks documents up by numerically equal values of another type. With the reset put back, a Debug server aborts on its first find. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Plan PR 6, the M1 exit (docs/cdc-change-stream.md §5.1-§5.6, §5.8; plan C1, C2, C6 decode, C7 DML, C8a, C8b, C9, C11, C13, C14, F2, F3): $changeStream on EloqDoc now reads the log service instead of the oplog, delivers insert, replace and delete events, and resumes exactly from any token it issued. Pipeline (§5.2): - DocumentSourceChangeStream gains a stage factory. eloq_init installs one, and $changeStream then expands to DocumentSourceEloqChangeStream followed by CloseCursor; run_aggregate skips the oplog namespace switch and the majority read-concern upgrade when a factory is installed. startAfter is parsed (upstream code is not copied). - DocumentSourceEloqChangeStream drives two plain components as one stage, since every event's token describes every log group's cursor: CdcScanSource (PR 5) and the new CdcEventStream. The stage answers both of the batch loops' token questions; a getMore waits until maxTimeMS only while its batch is empty. - CdcEventStream decodes one record at a time in storage order, skips sections outside the stream's scope, suppresses forward-write duplicates with a per-group per-timestamp dedup set, and formats each change: _id (token), operationType, clusterTime, fullDocument (or fullDocumentOmitted past 16 MB), ns, documentKey and eloq.keyBytes. Opening and resuming (§5.1, §5.6): - A fresh open fixes S = min arrival_complete_ts, waiting inside the command up to maxTimeMS or cdc_open_wait_ms, and answers with the watermark for S; past the wait it fails with the new ChangeStreamStartNotReady (5000, in an EloqDoc range upstream never allocates), whose message gives the wait and the checkpoint interval. - resumeAfter/startAfter resume from watermarks and event tokens, form-0 and form-1 entries, data before DDL at the token's timestamp. A token leaves out a group whose cursor its own timestamp covers, and resume treats such an entry as absent, so a resume never scans below the token into purged history. - Tokens that name no change fail with FailedToParse: an event or invalidate token at timestamp 0 (it aborted the server), and a watermark at the largest timestamp. startAfter an invalidate token fails with ChangeStreamFatalError until C10. - Rejected: startAtOperationTime and $_resumeAfterClusterTime. - cdc_open_wait_ms and cdc_scan_max_bytes (scan page size, up to the log service's 32 MB) are runtime server parameters, read at open. Tests: - cdc_stream_test (15): every emitted field, scopes, dedup and its overflow, oversized documents, resuming from every event token of three groups with cut pages, watermarks above and below their entries, covered cursors under a retention floor, and a mutation test that opens and drains a stream at every decodable byte mutation of real tokens. - cdc_scan_test (32) and cdc_codec_test (79), extended. - eloq_cdc suite (9 server tests, both CI suite lists): DML events, scopes, resume, postBatchResumeToken rows 1, 2, 5, 6, 9 and the getMore stash (row 4), _id integrity, open errors, the idle tail, and streams on missing collections and databases. - change_stream_drivers.py runs the C13 scenarios (open before the first checkpoint round, resume before the first event, start not ready) through PyMongo, Node.js 6.16.0 and Java 5.4.0, each driver on fresh servers; CI sets up Node and Java. Each is the last release accepting a 4.0 server; later ones refuse EloqDoc's wire version 7. All pass locally against a Debug server with the RocksDB log state. A Debug build reparses every aggregate, so a fresh open's stream runs there as the openAt(S) reopening; CI's RelWithDebInfo job runs the openFresh stream itself. C9 row 3, and row 4 in the aggregate loop, cannot occur on a real server: an aggregate's first batch only issues its scans. Also: CI's data substrate configs spelled checkpointer_interval as checkpoint_interval, which was ignored. Declaration comments in the touched CDC files are Doxygen blocks now. Design and plan updated, including dropping the incarnation filter: the merged timestamp order places every write in its incarnation, and the §10 fault tests assert the drop position rule instead. Bumps the data substrate to 3e7a1e4 (checkpointer_interval rules and Checkpointer::CheckpointInterval()). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…actly Plan PR 7 (docs/cdc-change-stream.md §4.4(b), §10; plan B2-B5, B7, B8 and the B6 tests). The data substrate side, bumped here to 1ad15aa: CREATE, DROP and TRUNCATE commit at a timestamp P above every DML their write lock drained; a create writes its kv tables before its commit log and its catalog row after it; a rolled-back add index commits at P with a rolled-back mark; UpsertTableOp runs as a coroutine; commit-stage recovery restores the coordinator's write lock. Table versions (MongoTableSchema). EloqDoc builds a DDL's image before the DDL has a timestamp, so a table the DDL creates is recorded without one. RecordKeySchemasTs writes every base and index table's resolved version into the image, and the tx service keeps that image (SchemaOp::ResolveSchemaImages), so a schema loaded at a later record version or on another node gives each table its own version. A missing version now reads as 0, the data substrate's unknown timestamp (was 1). Start-up race. ddl_recovery.py exposed it: after an unclean shutdown KVStorageEngine::loadCatalog dropped any collection that getAllCollections listed but getAllIdents did not, and reconcileCatalogAndIdents fasserted (40593) on the same difference. Both lists come from the data store's catalog table, only the second filtered through the tx service, and a recovered create or drop writes that table between the two reads, so a committed create was dropped and a node restarting after a drop could abort. A collection's table and catalog entry are one record in EloqDoc, so start-up now opens every listed collection the tx service's catalog still has, skipping one it no longer has, and the reconcile asks for nothing. Repair mode keeps no orphan handling, as the header says. Tests. - cmake/tests/ddl_recovery.py (target eloqdoc-ddl-recovery, a CI step): seven crash-and-restart scenarios. A create killed after its prepare log, and after its commit log; a drop killed after its commit log; a rolled-back add index that must survive a restart, and the same rollback killed after its commit log; a create that gave up, killed after its failing commit log; an ALTER (collMod) killed after its commit log, finished under the restored lock. - eloq_basic/add_index_rollback.js: a key past the size limit fails the build; writes, reads and later index builds work at the new version. - eloq_fault/ddl_kv_retry.js: a create whose kv tables or sequence table update fail gives up and releases the name; a create's catalog row and a drop's sequence and kv steps fail once and are retried. - eloq_fault/ddl_prepare_log_fail.js: a create and a drop whose prepare log fails roll back; the drop leaves the collection intact. - The CI fixtures and server_smoke_config set ddl_kv_retry_interval_seconds=1, so a retried step takes a second. Docs: the design doc and the plan record PR 7 as done but not committed (B2-B5, B7, B8, D3's gate site in RunUntilPrepareLog, the retry rule, the timestamp rules for catalog readers and lock grants), the EloqSQL changes PR 7 requires once EloqSQL moves to this data substrate (plan §4), and the deferred range-split recovery work (plan §11). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
The data substrate commit is 1ad15aa, bumped by 608720b. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Plan PR 8 (docs/cdc-change-stream.md §4.2, §4.3, §4.5, §4.6, §5.8; plan A3, A5, A8). The data substrate side, bumped here to e51b721: - the log state keeps a DDL history, one record per EloqDoc create, drop and index op (two for an add index rolled back after its build), at the timestamp of the log the op wrote under its catalog write lock, saved in the raft snapshot and trimmed by the data log's purges; - CdcScan's DDL stream reads that history; - a scan with nothing to return is held until the checkpoint advances, its wait (capped at 60 s on the server) elapses, the client cancels, or the node steps down; - a schema op recovered from its commit log restores a write intent, not a write lock, on its coordinator's catalog entry, so the table's data log replays instead of being dropped. Test commands (eloq_test_commands.cpp, test commands only). eloqCatalogVersion replies with a collection's committed schema version from the local catalog. eloqCdcScan sends one CdcScan to a log group and replies with the page, its DDL records decoded; it rejects a group id outside the cluster's log groups, and it waits on the call's own closure, which the log agent runs even when it fails without starting a call (a brpc::Join would hang there). cmake/tests/ddl_recovery.py. Every scenario now reads the DDL history: the op left one record at the position it committed at; a recovered create's record is at the version the catalog holds after recovery; a rolled-back add index has a record at its prepare log's timestamp and a second one, marked rolled back, at its commit log's; a create that gave up and an ALTER leave none. Scenarios (v) and (vii) check that the documents written before the crash survive recovery and a checkpoint, which caught the dropped replay. The test runs on the default configuration, with no CDC retention window, as the CI fixtures do. New --scenario option runs single scenarios. Docs. The design (§4.3, §4.5: the DDL history is trimmed by the purges, not at every checkpoint; the failure table: recovery restores a write intent) and the plan (A3, A5 and A8 done; A8's stale items replaced; the B5 row; A5's forced step-down test open) are updated. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Plan PR 9 (docs/cdc-change-stream.md §5.1, §5.2, §5.4, §5.4a, §5.6; plan
C7, C8b, C10, B6, B7, F2), and a fix to EloqDoc's write-conflict backoff.
DDL events (C7). decodeCdcDdl (src/cdc/cdc_ddl_decoder.*) turns the schema
op of a log-service DDL record into create, drop, createIndexes or
dropIndexes. It reads only the catalog entry BSON at the head of each schema
image, bounded by the image: create reports md.options without the UUID and
EloqDoc's catalog version, as listCollections does; an index op's indexes
are those whose names one image has and the other lacks, reported as their
specs without ns (createIndexes) or {v, key, name} (dropIndexes); the second
record of an add index rolled back after its build reports its indexes
dropped. An undecodable image fails the stream with ChangeStreamFatalError;
a record no event reports (another engine or op type, a rolled-back create
or drop, the feature document) is skipped. CdcEventStream emits one event
per in-scope DDL record, with _id, operationType, clusterTime, ns and, but
for drop, operationDescription; its group's token entry is form 0.
Invalidate (C10, C8b). A collection-level stream returns invalidate on the
pull after its collection's drop, with the drop's position as a kind-2
token, and nothing after it; the $changeStream expansion now appends
MongoDB's DocumentSourceCloseCursor, which closes the cursor on the next
pull, so the reply that carries the invalidate has cursor id 0. A stream
resumed from the drop's own token returns that invalidate alone: the group
named by a form-0 entry whose DDL key is at the token's timestamp restarts
there and re-reads the record, and the stream invalidates when the record is
the watched collection's drop. openAt accepts invalidate tokens (startAfter
resumes after the drop); resumeAfter refuses them with InvalidResumeToken,
as MongoDB does (was BadValue). When the stream closes after its
invalidate, both batch loops report the stream's invalidate token as the
closing batch's postBatchResumeToken, even if a user stage filtered the
invalidate out; the drop's token, the last one delivered then, would
restart the stream into the same invalidate.
Tests. cdc_stream_test: CdcDdlDecoder (7 cases) and the stream's DDL,
scope, invalidate, own-drop and startAfter cases; the three-group resume
script now carries DDL records, so every DDL token is resumed from.
eloq_cdc: ddl_events.js (every DDL event shape, a rolled-back index build,
resume from each DDL event with resumeAfter and startAfter) and
invalidate.js (C9 rows 7 and 8, row 8 also with the invalidate filtered
out). A stream opened without a token starts at the checkpoint bound,
which can lag a DDL that has returned, so it now reports a test's setup
DDL: EloqCdcUtil.settleDdl waits until a fresh
stream starts above it, missing_namespace.js expects the implicit create,
open_errors.js expects startAfter on an invalidate token to open, and the
driver scenario (ii) of PyMongo, Node and Java watches inserts only.
Drop order (B6 (ii)). eloq_fault/ddl_drop_order.js holds a drop between
its prepare log and its commit-stage write locks with the data
substrate's new upsert_table_hold_before_commit_lock fault point
(submodule bumped to fab5f18), inserts on the same node while it is held (MongoDB's locks exclude nothing
in EloqDoc: EloqLockerNoop), and checks in a stream that the inserts come
before the drop, below its commit-stage timestamp, and the next create
after it.
Write-conflict backoff. A drop issued during a foreground index build on
the same collection stalled both for minutes: the drop's catalog write
intent conflicts with the build's, writeConflictRetry retries it, and the
backoff slept the thread. In EloqDoc that thread is a brpc worker that
also drives a core's transaction processor, so retries sleeping back to
back starved the build the drop was waiting for, about 200 times over.
WriteConflictException::logAndBackoffYielding backs off through
coro::sleepUntil, which on a service executor coroutine arms a bthread
timer and yields. Every write-conflict retry loop uses it, after its catch
handler, since it may yield: writeConflictRetry, the plan executor's two,
cloneCollectionAsCapped and the MongoDB-side index insert loop; the latter
two now abandon their snapshot before the pause, as writeConflictRetry
does. The thread-sleeping logAndBackoff is removed, so that no caller can
bring it back unnoticed. OperationContext::sleepFor and sleepUntil yield on a coroutine too, in
100 ms slices that check for interruption, which covers the catalog record
store's retry sleeps. eloq_basic/drop_during_index_build.js.
Docs. docs/cdc-operations.md, the operator runbook, with its first entry:
a schema change whose data-store step keeps failing after its commit log
holds its write lock and stalls every stream (B7). The design (§5.4's DDL
decode, §5.6's own-drop check, §8, §10 items 3 and 4, the status line and
Key files) and the plan (C7, C8b, C10, C9 rows 7 and 8, F2, B7, PR 9) are
updated, B6 (ii) and §10 item 6 included.
The CDC comments of src/base/cdc_blob_decoder.h, touched here for
CdcStreamScope::isCollection, and of
src/mongo/db/pipeline/post_batch_resume_token_provider.h, touched for the
closing token, are converted to Doxygen.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Bumps data_substrate to the commit that carries the cluster gate in the cluster config record and cc map, refuses new schema ops while it is paused (TxErrorCode::CLUSTER_GATED) and keeps the gate's slot in log group 0 (snapshotted, replayed, never purged, found by RecoverTx); plan PR 10: D1, D2, D3's data substrate part, D5. D7, the drain poll, is dropped: every schema op holds the cluster config read lock through its last post write, so the pause's write lock is the drain, and the pause's commit timestamp (pause_ts), raised above its acquire-all results, bounds every earlier DDL; bootstrap picks T0 from it (design §6.2). The submodule also adds Now() to the cluster config map's acquire-all, which §6.2's bound now cites. No EloqDoc code changes: nothing can pause the gate before the gate op (PR 11) and the admin command (PR 12). The plan moves D3's EloqDoc mapping of CLUSTER_GATED to PR 12, since an index build's cleanup retries write conflicts forever and treats other exceptions as fatal (fassert 18644) and createIndexes retries write conflicts in writeConflictRetry; D3's behavioral test to PR 11, which brings a gate that can close; and requires D4 and D6 to land together, since RecoverTx keeps a logged gate tx's locks for D6's replay. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Bumps data_substrate to the commit that adds the gate op and its replay
(plan PR 11: D4, D6). EloqDoc side:
- eloqClusterGate {op: "pause" | "resume", mode: "ddl"}, a test command
for now (D8's syntax; PR 12 makes it the admin command with its
privileges, gauge and alert). Each call is its own transaction,
initialized for log group 0; a held gate fails with
ConflictingOperationInProgress.
- cmake/tests/cluster_gate.py (target eloqdoc-cluster-gate, in CI): on one
node, pause and resume, a pause waiting for an admitted DDL, a pause
surviving restarts, a gate op killed after its log or in its install,
a create recovered beside the gate's slot, concurrent pauses, and a
pause retried after a forced write intent conflict.
Design §6.2 and the plan record how the op locks (a read for write, then
OCC write intents everywhere, then write locks, retrying inside the op
after a conflict, so gate ops and cluster scale cannot deadlock), that
recovery takes no lock (re-taking it deadlocks replay against schema-op
recovery), that the gate log carries the granting node groups' terms, and
what one node cannot test (a second
node group failing over, leader changes, scale under the gate). A refused
drop is retried as a write conflict until it gives up; D3's mapping in
PR 12 decides what it reports.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
EloqDoc keeps a Database, and so a view catalog, per worker thread (DatabaseHolderImpl's _dbMapVector). A view's create, collMod or drop invalidated only the catalog of the thread that wrote system.views, so listCollections and queries on connections served by other threads used older view definitions, and another node's catalogs were refreshed by nothing. Collections avoid this with catalogVersion; views had no equivalent. A view is now data, read like a document. When the storage engine reports record versions (StorageEngine::supportsRecordVersions(), true for Eloq): - ViewCatalog::lookup() and each link of resolveView() read the view's system.views document in the operation's transaction (DurableViewCatalog::lookup) and reuse a cached definition only if it was built from the version read: the commit timestamp of the record version, which grows with every write of the record. A read of the transaction's own uncommitted write reports 0 and is never cached; an older version never replaces a newer one, so an operation on an older snapshot does not undo what a newer one cached. A namespace with no document is not a view. - The cache forgets dropped views: a committed drop removes its entry, every full read of system.views removes the entries of views it no longer holds, and a lookup that finds no view removes the namespace's. A thread that only looks up views dropped elsewhere is bounded by a cap of 1024 entries, past which the cache is cleared. - iterate() (listCollections, dbStats) and createView, modifyView and dropView read all of system.views first, so listings and the name, cycle and collation checks see every view as of the snapshot. - Every read and write of system.views happens with the catalog's mutex released. The coroutines of a worker thread share it, and a read can yield the coroutine; holding it across one could block the thread. RecordStore::findRecordVersion() reads a record and its version (the default reports 0); EloqRecordStore implements it, and EloqRecoveryUnit::getKV can return the version the tx service reports. Engines that report no versions keep the stock behavior. Test: eloq_basic/views_across_connections.js (eight connections each see a create, a collMod, a view on a view, a drop, a collection taking a dropped view's name and a dropDatabase made on another). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…umer Plan PR 12 (D3's EloqDoc mapping, D8, E1, E2, E5); EloqDoc only, no data_substrate bump. - eloqClusterGate is an admin command (eloq_cluster_gate_command.cpp), registered whether or not test commands are enabled. It needs the new eloqClusterGate action on the cluster resource, which clusterManager holds, and so clusterAdmin and root. Syntax and replies are PR 11's. - A schema change refused by a paused gate (CLUSTER_GATED) fails at once with the new ClusterGatePaused (5001), which clients retry after the resume: a create (an insert's implicit one too), drop, dropDatabase, index build or drop, and collMod. The recovery unit returns it as a status, never as a write conflict, and the drop loop in EloqCatalogRecordStore::deleteRecord raises it instead of retrying until it gave up (70001). - A refused index build's cleanup writes back the catalog entry it started from, which differs from the stored one only in its catalogVersion. It used to issue a DDL for that write, which the gate refused too, and ~MultiIndexBlockImpl treats any failure but a write conflict as fatal (fassert 18644), so a createIndexes during a pause killed the server. EloqRecoveryUnit::updateTable now skips a write that changes nothing (sameCatalogEntry). - View changes are held still by the gate too. A view's create, collMod and drop are writes to system.views, not schema ops; EloqRecordStore now admits every write there with EloqRecoveryUnit::checkClusterGate, a local read of the cluster config record whose read lock lasts until the transaction ends, as a schema op's does: a pause waits for an admitted view change, and one that reads a paused gate is refused with ClusterGatePaused. Otherwise a view changed during the bootstrap copy could stop replay for good: the copy holds later definitions, and an earlier change replayed beside them can form a view cycle the sink refuses on every retry. - tools/cdc_consumer/eloq_cdc_consumer.py: the reference consumer (design §6). The apply loop spreads DML over slots by crc32(ns, eloq.keyBytes) and applies each slot's share as ordered bulk writes; a DDL waits for every slot; the token goes to admin.__eloq_cdc_progress after each DDL and each batch. Bootstrap marks the checkpoint bootstrapping, drops the in-scope databases on the sink, pauses the gate, picks T0 from a fresh stream, copies definitions (without unique, expireAfterSeconds or capped) and documents, resumes, and publishes T0. A held gate halts for the operator; a sink collection or index that conflicts with the copy halts as a bypassed gate. Events in databases outside the consumer's scope (--databases, never admin, local or config) are skipped, as the bootstrap skips those databases. An event that arrives without its document (fullDocumentOmitted) halts the consumer until plan E3 adds the point read. Every sink write, DDL included, carries the sink client's write concern, and a write concern error fails it (Database.command() adds neither), so the checkpoint never gets ahead of durable writes; an unacknowledged write concern is refused. One namespace rule serves the copy and the stream: collections and system.js are replicated as documents; views as definitions, read from system.views by the copy, and replayed from the source's system.views inserts, replaces and deletes as create, collMod and drop on the sink; every other system collection is skipped. A bootstrap that halts on an error after pausing the gate resumes it on its way out. Tests: - tools/cdc_consumer/test_eloq_cdc_consumer.py: 20 unit tests of the apply loop, the sink and the tokens. - cmake/tests/cdc_consumer_e2e.py (target eloqdoc-cdc-consumer, with the unit tests, in CI), one fresh node replicated into another: convergence under concurrent writes, with a database outside the consumer's scope whose writes, create and drop never reach the sink, every DDL sent to the sink carrying its client's write concern, and views and system.js matching the source; an index created, dropped and recreated during the copy, refused with 5001 and converging with the gate, halting without it; views changed during the copy (a on b, a back on the collection, b on a), converging with the gate and halting on a view cycle without it; a run halted in its copy, which leaves the gate open, and a run stopped before publishing T0, after which the next run bootstraps again. - eloq_cdc/bootstrap.js: the command's arguments and admin-only rule, every kind of DDL, views' included, refused at once during a pause, T0 picked from a stream, and a stream resumed from T0 holding exactly the later DML and DDL. - cluster_gate.py now drives the admin command, expects 5001 from every kind of DDL under a pause, view changes and direct writes to system.views included, and checks the privileges (x). Found on the way, and fixed in the commit before this one (not CDC work): EloqDoc kept a view catalog per worker thread, refreshed only by that thread's own view changes; the e2e test's view checks rely on it. The cluster_gate_paused gauge moves to PR 13 with the data substrate's other metrics: it reads the cluster config cc map, which only the data substrate sees. Design §6.2 and the plan record the mapping, the privileges and what is done. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Bumps data_substrate to the commit that exports cluster_gate_paused,
cdc_arrival_complete_ts and cdc_scan_parked_requests (plan PR 13: A9,
and D8's gauge, moved there from PR 12).
cmake/tests/cluster_gate.py (xi) turns metrics on and scrapes the
server's /eloq_metrics endpoint: cluster_gate_paused{ng_id="0"} reads 0,
1 after a pause, 1 again after a restart's replay reinstalls it, and 0
after the resume; cdc_arrival_complete_ts advances with the checkpoints
and trails the wall clock by less than two minutes; a change stream's
getMore waiting on an idle database shows in cdc_scan_parked_requests,
which returns to 0 afterwards.
Design §6.2 (the gauge's labels and reset) and §8 (the metrics that
exist and those still to add) and the plan's A9, D8, D9 and PR 13 rows
are updated.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Plan PR 14 (C11, C12, C7's end-to-end check, E3, E4, E6, F4, F5, F6),
the M4 exit. Bumps data_substrate to the commit with docs 03 and 07.
- C11: cdc_cursor_buffer_bytes (default 64 MB, up to 4 GB),
cdc_dedup_set_bytes (default 4 MB, 0 to 1 GB; 0 records nothing) and
cdc_scan_max_wait_ms (default 30 s, up to 10 minutes) are runtime
server parameters, read when a stream opens, documented in
configuration-description.md and bounds-checked in open_errors.js.
- C12: serverStatus' eloqChangeStreams section lists the open streams:
each one's scope, namespace, lag (the emit bound it last saw minus
what it has produced through), positions, dedup-set overflows and the
time since its last pull, and the totals. Each CdcEventStream keeps
these in atomics after every pull; its stage registers it while open
(cdc_stream_registry.{h,cpp}). A per-cursor series cannot go through
the data substrate's meter, whose label values are registered up front.
- The stage factory no longer adds a DocumentSourceCloseCursor:
createFromBson(), its only caller, adds one already.
- E3: the reference consumer applies an insert or replace that arrives
without its document (fullDocumentOmitted) from a read of the source,
in the event's slot, and skips one gone since, whose delete follows.
- E4: ChangeStreamHistoryLost, when the stream opens or while it is
read, makes the consumer bootstrap again (HistoryLost), at most three
times before its stream has caught up with the source once
(--max-rebootstraps); past that it halts for the operator, since each
bootstrap drops the sink's data and a history kept shorter than a
bootstrap takes would lose every one. Caught up takes a watermark
above where the stream started and within a minute of the consumer's
clock: a poll that timed out returns the unchanged opening watermark,
which proves nothing.
- E6, F4: cdc-operations.md §2-§4 (a pause never resumed, a pause
refused, retention, lag and the alerts to set) and
configuration-description.md on txlog_cdc_retention_seconds; in this
version a position older than the retained history goes unreported,
so retention is the only protection.
- F5: the design's facts F7, F8 and F12 re-verified against the code
and corrected (create and drop stamp their schema at the commit-stage
P; a rolled-back add index commits at P; an ALTER can roll back after
its prepare log; schema ops hold the cluster config read lock), §10's
items marked with the PR that settled them or what remains open, and
§5.7 gives the MVCC checkpoint delay. The plan's C7 row loses its
stale planned text.
Tests: cdc_stream_test (StatsFollowThePulls); eloq_cdc/server_status.js;
consumer unit tests (an omitted document read from the source or skipped,
a lost history at open, the run loop bootstrapping again, and its cap,
which only a verified catch-up restarts, through the real stream_from);
cdc_consumer_e2e.py (v), documents just under 16 MB whose events arrive
without them.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
The data substrate's CDC branch now sits on its main, which since the branch began has consolidated the module docs, added cache and checkpoint metrics and per-partition checkpoint publication, and moved its EloqStore and brpc submodules (main's tx_service.h needs brpc's eloq::ModuleType). - Docs: main removed the checkpoint_stall_rounds gauge in favor of node-level checkpoint metrics. The runbook (sections 1 and 4) and the design doc (4.4(b), 5.7, section 8's failure table and metrics list) now name the checkpoint stall warning in the log and cdc_arrival_complete_ts standing still for a stall in progress, and checkpoint_advance_interval_seconds for its length afterwards. - The plan's PR table records what was committed where without commit hashes, which a rebase rewrites. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
The data substrate on the CDC branch, rebased onto its main, pins the brpc whose
eloq::EloqModule::Type() is pure virtual: the module registry places each module
by its type. MongoModule did not implement it, so EloqDoc no longer compiled
("cannot declare variable 'mongoModule' to be of abstract type"). It returns
eloq::ModuleType::kRuntime, the type brpc defines for the mongo service executor,
which registers in the registry's runtime range.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…startup DatabaseImpl::clearTmpCollections, which repairDatabasesAndCheckVersion runs for every database at startup, took the collection names from the data store's table list (EloqKVEngine::listCollections) and each entry from the tx service's catalog, and dereferenced the entry unchecked. A drop that has committed but not yet deleted its catalog row from the store is still listed but has no entry, and DDL recovery rolls such a drop forward after the node has started, while the sweep runs: the restarted node then crashed with a null pointer (the DDL recovery test's drop killed after its commit log, 10 of 12 RelWithDebInfo runs). The sweep now skips such a name; the other walks of the listing already did. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
SCons, which CI uses to build the legacy JavaScript test shell, stopped before compiling anything: - Its error code check found assertion code 40674 at two sites: MongoDB's check of a resume token combined with a start time, and the change-stream stage's check of resumeAfter combined with startAfter. Both now go through one helper holding the single 40674 site, so every combination of resume options still fails with the same code and message. - The same check counted "uassert(50810)" in a comment of cdc_key_decoder.h as an assertion site; the comment is reworded. - The Eloq SConscript had a non-ASCII section sign in four comments, which SCons's Python 2 rejects without an encoding declaration. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
… first Fixes found running CI's CMake job on a 32-core VM: - A test's servers listened on ports picked in the kernel's ephemeral range, and only the tx port was checked: the data substrate also listens on tx port +1, +2, +3 and +7, which client sockets left in TIME_WAIT by earlier servers held, and a server killed or crashed by a fault point left its forked host manager holding its listening sockets for a moment. Restarts and fresh servers then failed with "server exited during start-up" (DDL recovery, the cluster gate, the CDC consumer and the change-stream drivers). server_smoke_config.py now hands out ports below the ephemeral range, the tx port's whole block free, and reaps a server's process group and waits for its ports before the next start; ddl_recovery.Node and the drivers' Server use it. - The smoke test built a unique index on a populated collection, which this branch refuses (unique indexes only on an empty collection): it now builds the index first. - server_smoke_config supports EloqStore with the RocksDB log state: the store stays in S3, WAL on, without the log service's cloud settings, which such a build does not define. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
A future feature. EloqDoc has temporary collections ($out's tmp.agg_out.<n>, and create with temp: true), and MongoDB's restart sweep drops every one of them in the shared catalog at any node's start, including those live operations on other nodes are using. The document records how EloqSQL ties a temporary table to the node that created it and drops it at that node's next start, and the design to adopt: the creating node's id in the collection's options, a sweep that drops only this node's temporary collections, rename clearing both, and a reaper for nodes that never return. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
The data store service now hosts the data shards of every node group a node leads on one EloqStore (eloqdata/eloqstore PR 502), attaching and detaching each as leadership comes and goes. The voter takeover test expects the voter to detach node group 1's data shard when the group comes back, where it expected it to stop a second EloqStore instance. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Scenario (vi) armed the one-shot upsert_table_create_kv_table_fail and raced the server, whose feature document create took the fault first on the VM; the abandoned create then committed and was rolled forward. The fault now takes a table_name parameter (data substrate), and the scenario passes its collection. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Each scenario arms its crash fault right after creating a collection, while the server's feature document create is still in flight; an unaimed fault could kill the node on that create instead (2 of 5 runs on the VM). crash_during takes the DDL's table and passes it to the fault. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…ed passes The shard cleaner releases parked requests as fast as it reclaims memory (shard_clean_release_kb, a new [local] key, default 256 KB) instead of waiting for a share of the heap, and a pass that leaves requests parked retries on a timer with backoff instead of waiting for the next flush. Two data substrate tests that no longer compiled on this branch build again. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
batchCheckDuplicateKey read its keys with a point read on each miss, so a run of sequential inserts went to the data store once per key. It now asks batchGetKV to load the whole slice on a miss, and the checks of the following keys find it in the cache. batchGetKV takes the choice as a parameter that defaults to the point read, so its other callers are unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
server_smoke.py took its node's base port from a bind to port 0, which hands out a port from the ephemeral range, and checked only that one port, while the node listens on eight consecutive ports from it. With other tests running, a client connection could already hold one of the seven: the host manager then failed to bind and the server exited with code 100 at start. It now uses free_node_port with the node's port block, as the other runtime tests do: below the ephemeral range, every port of the block free. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The readers of a node share the node's client and ran their aggregate and
getMores without a session, so each command took an implicit session from
the client's shared pool. When all readers woke at once, two threads could
swap sessions, and the server refused a getMore from a session other than
its cursor's ("Cannot run getMore on cursor ..., which was created in
session ..."). Each reader now runs its aggregate, getMores and killCursors
in an explicit session of its own.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The repository's format check runs clang-format-18 over the eloq module; seven files the CDC branch added did not pass it. Whitespace and the order of one include block only. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The design still read "implemented through milestone M2", with the cluster gate and the reference consumer to come, and the plan called itself a draft; every PR of the plan's sequence is done. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
Important Review skippedToo many files! This PR contains 161 files, which is 11 over the limit of 150. To get a review, reduce the PR to 150 files or fewer by splitting it into smaller PRs or changing its base branch. Upgrade to a paid plan to raise the limit. This review couldn't start because sufficient usage credits or metered capacity aren't available. Add credits or update usage-based reviews in the billing tab, then retry.
 |
5 tasks done
|
|
The SCons build runs the IDL compiler under Python 2, which writes its
generated source through the ASCII codec: the section sign in a
document_sources.idl description failed it ("UnicodeEncodeError ... u'\xa7'"),
so CI's SCons jobs stopped at document_sources_gen.cpp. The CMake build runs
the compiler under Python 3 and was unaffected. The references now read
"section 5.2" and so on, in error_codes.err too, which its generator
accepted but which keeps both files ASCII like the rest of main.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The data substrate now closes and checks the CDC DDL history and cluster gate snapshot files like the other log snapshot files, and starts brpc's bthread workers before its storage handler so EloqStore's shard attach can run during EloqKV's bootstrap. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
EloqDoc creates a collection's table in its own schema transaction, which
commits before the collection object is built. The collection constructor
rejected some options only then: a validator in the local, admin or config
database or on a system collection, and an unknown validationLevel or
validationAction. The create command failed but the table stayed:
listCollections failed in that database from then on, and the server
refused to start ("exception in initAndListen: InvalidOptions"). This
happens on main too.
userCreateNS now applies the same checks before the create, through a new
Collection::checkValidatorAllowed that parseValidator shares, and through
the existing parseValidationLevel and parseValidationAction.
core/doc_validation_invalid_validators.js left such tables in all three
internal databases. CI now runs eloq_cdc (and eloq_fault) after eloq_core
against the same server, and their ValidateCollections hook failed on the
first test. The new eloq_basic test create_refused_options.js covers each
refused option.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The jstest suites declare enableTestCommands for their server, and the CMake job's fixture sets it, but the SCons jobs run the suites against an external server whose config left it out. eloq_cdc's post_batch_resume_token_first_batch.js arms a fail point through configureFailPoint, which only exists with test commands enabled, so it failed in both SCons jobs while passing in the CMake job. The same config serves the SCons TPC-C phase, as test commands do in the CMake job's. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
liangjchen
added a commit
to eloqdata/tx_service
that referenced
this pull request
Oct 29, 2026
…lti-node fixes (#577) Everything EloqDoc's change data capture needs from the data substrate (design in eloqdata/eloqdoc docs/cdc-change-stream.md; EloqDoc side in eloqdata/eloqdoc#506), the fixes its multi-node tests found, one EloqStore per process, and reworked shard memory eviction. Log service - New CdcScan RPC: a log group's committed data records and DDL history merged in timestamp order, paged with an exact continuation, served by the leader only, long-polling when idle (cdc_scan_max_parked, cdc_scan_max_park_ms). - The log state keeps EloqDoc's DDL history and saves it, with the cluster gate state, in its snapshots. - txlog_cdc_retention_seconds (default 0) holds back SST purges for CDC readers; at 0 purging is unchanged. - New log content kind 6, the cluster gate entry, always in log group 0. Schema operations (all products) - CREATE, DROP and TRUNCATE TABLE commit at a timestamp chosen under the commit-stage write lock, above every DML that lock drained. - Lock grants report max(last read ts, local time). - A create writes its kv tables after its prepare log and is abandoned if that fails. Store failures after the point of no return retry every ddl_kv_retry_interval_seconds (default 30). - Add index creates its kv table before its prepare log. - The checkpoint stays below any write lock on a data table or catalog entry. A missing table version now reads as 0 (was 1). Cluster gate - A cluster-wide pause that refuses new schema ops with CLUSTER_GATED, drains the admitted ones, and returns a timestamp above all of them. Log replay reinstates it. Recovery and routing - Log replay is requested once per log group at leader start, in bounded attempts with a watchdog. TxLog::ReplayLog takes one log group. - Data loss fixed: writes to a splitting range's new range were logged to the new owner only for the first such key. - Leader caches update by term. A retried hash-partitioned read no longer asserts. Snapshot transactions start at max(ts base, wall clock). A failed range-split UpsertRanges retries with the full range list instead of an empty one. EloqStore - One EloqStore per process hosts all of a node's data shards (AttachShard/DetachShard, eloqdata/eloqstore PR 502), pinned to the merged commit. - In brpc module mode the bthread workers start before the storage handler, so EloqStore's shard attach can run during EloqKV's bootstrap. Shard memory eviction - One cleaner per shard releases parked requests in batches of 20. After a request hits the limit again, the next release waits until shard_clean_release_kb (default 256) more is reclaimed; at the LRU tail any reclaimed amount goes out. - A pass that ends with requests still parked retries on the timer thread after 100 ms. While passes reclaim less than that threshold, each asks for a checkpoint and the delay doubles up to 1.6 s. The backoff clears when the cleaner goes idle. Smaller - A negative checkpointer_interval refuses to start; 0 to 4 becomes 5 s. - Fault points for the DDL, gate and replay paths. - At shutdown the timer thread no longer walks a freed shard. Compatibility: proto changes are additive, and defaults behave as before except eviction pacing. Logs with a cluster gate entry and snapshots with DDL history are new formats; older binaries were not tested on them. EloqSQL must adapt when it bumps to this data substrate, its ReplayLog forwarder included (EloqDoc CDC plan, section 4). Verified against this head: the unit job (201 tests), EloqDoc's CI and multi-node suites, and EloqKV's CI matrix (both build types, all three stores).
eloqdata/tx_service#577 was squash-merged onto tx_service main. The merged commit's tree is identical to the pull request head this branch pointed at, so no file changes; the pointer now names a commit on main instead of one only on the pull request's branch. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Context
Change data capture for EloqDoc:
$changeStreamserved from the log service instead of an oplog, a cluster gate that holds DDL still while a consumer copies the catalog, and a reference consumer that copies a cluster into another MongoDB and then follows its stream. Design:docs/cdc-change-stream.md. Plan and per-PR history:docs/cdc-implementation-plan.md(all 14 plan PRs done). Operator runbook:docs/cdc-operations.md.The data substrate side, eloqdata/tx_service#577, is merged; this branch's submodule points at its squashed commit on tx_service
main, whose tree is identical to the PR head the tests below ran on. The branch is 58 commits on top of currentmain(no rebase needed); each commit message stands on its own.Behavior before and after
Change streams (new)
aggregatewith$changeStreamat collection, database or deployment level, thengetMore: insert, replace and delete events with full documents; create, drop and index DDL events; collection-levelinvalidate. Updates are reported asreplace(noupdateDescription).postBatchResumeTokenon every reply, the opening reply included;resumeAfterandstartAfterresume exactly across pages, log groups and leader changes. The legacy shell's cursor tracks tokens.watch()work unchanged (tested).cdc_open_wait_ms(default 30 s) andcdc_scan_max_bytes.ChangeStreamStartNotReady(5000, the open could not fix a start position in time; retry),ChangeStreamHistoryLost(286, the position is below the log's retention),ChangeStreamFatalError(280).Cluster gate (new)
{eloqClusterGate: {op: "pause" | "resume", mode: "ddl"}}(mode optional), on the admin database, needing the neweloqClusterGateaction, whichclusterManagerholds (and soclusterAdminandroot). While paused, every schema change fails withClusterGatePaused(5001): creates (implicit too), drops, index builds and drops,collMod, and view changes.Reference consumer (new)
tools/cdc_consumer/: bootstraps an empty sink under the gate, then follows the deployment stream, keeping its position in the sink'sadmin.__eloq_cdc_progress; manual in its README.Fixes outside CDC that the work found
KeyStringdecoded a decimal with an integer part above 2^53 one unit low (key_string.cpp)._idpoint lookups decoded numbers as int32 (lost the stored type bits).CannotCreateIndex): the tx service's build does not detect duplicates. Unique indexes on empty collections are unaffected.system.viewson every lookup; a view change made on one worker thread or node was invisible to others.MongoModuleimplementsEloqModule::Type(), which the pinned brpc requires.local,adminorconfigor on a system collection, and an unknownvalidationLevelorvalidationAction. The command failed, butlistCollectionsfailed in that database from then on and the server refused to start.userCreateNSnow applies the same checks before the create (Collection::checkValidatorAllowed, shared withparseValidator). This was a bug onmaintoo.Tests and CI
eloq_cdcandeloq_fault(the latter only on builds withWITH_FAULT_INJECT, a new CMake option, on by default in Debug), wired into the CMake CI job with the new runtime targetseloqdoc-change-stream-drivers,eloqdoc-ddl-recovery,eloqdoc-cluster-gateandeloqdoc-cdc-consumer.eloqCdcScan,eloqCatalogVersionandeloqFaultInject(test commands only).cmake/tests/multinode/: three node groups as separate processes, EloqStore local and cloud mode. Not run by CI; see the test plan.enableTestCommands, as the suites declare and the CMake fixture already does;eloq_cdc/post_batch_resume_token_first_batch.jsneedsconfigureFailPoint.Implementation
src/mongo/db/modules/eloq/src/cdc/:document_source_eloq_change_stream.cpp(the single stage and its opening protocol),cdc_scan_client.cpp(one in-flightCdcScanper log group,NotLeaderredirects compared by node ID),cdc_scan_source.cpp(the merge across log groups and the long poll),cdc_event_stream.cpp(events, invalidate, DDL),cdc_stream_registry.h.src/mongo/db/modules/eloq/src/base/: the redo-log blob decoder, the canonical delete-key decoder and the resume-token codec, each with unit tests.src/mongo/db/pipeline/,query/plan_executor,commands/run_aggregate.cpp,commands/getmore_cmd.cpp,clientcursor.h,cursor_response:postBatchResumeTokenplumbing, which this 4.0 tree did not have.src/mongo/db/modules/eloq/src/eloq_cluster_gate_command.cpp,eloq_recovery_unit.cpp: the gate command and the mapping of the data substrate'sCLUSTER_GATEDtoClusterGatePaused.tools/cdc_consumer/eloq_cdc_consumer.py: the reference consumer.cmake/tests/: runtime test scripts (smoke, DDL recovery, cluster gate, drivers, consumer end to end) andmultinode/.Design decisions and alternatives
cdc_arrival_complete_ts).replace: EloqDoc logs whole documents, so the event carries the full document and no diff.Test plan
Commands and results:
Known flakes, not caused by this branch: in
voter_takeover, a restarted node whose node group is led elsewhere can fail its first transaction before it follows the group (the data substrate's tx service init does not wait for its native group), and EloqStore's cloud mode occasionally reports a corrupted manifest after akill -9. Reruns pass.Risk assessment
change_streams/jstests do not apply and are not run.eloq_cdcis the suite for this implementation.CannotCreateIndex.system.viewsmore often (a cached definition is reused while its commit timestamp is unchanged).txlog_cdc_retention_secondsat its default 0, a consumer behind the checkpoint when a purge runs getsChangeStreamHistoryLostand must re-bootstrap; the runbook covers sizing.Rollback plan
Revert this PR together with the data substrate bump. The data substrate PR's rollback note applies: a cluster that has logged a cluster gate entry should not go back to an older binary without checking its replay. Change streams have no persistent server-side state; consumers keep their own tokens.
Reviewer guide
docs/cdc-change-stream.md§5.1 (opening protocol), §5.6 (tokens) and §6.2 (gate): the contracts the code implements.src/mongo/db/modules/eloq/src/cdc/document_source_eloq_change_stream.cppandcdc_scan_source.cpp: the stage, the merge and the long poll.src/mongo/db/modules/eloq/src/base/cdc_resume_token.cpp: the token layout, frozen once clients hold tokens.src/mongo/db/catalog/index_create_impl.cppandeloq_record_store.cpp: the index-build fixes and the unique-build refusal.src/mongo/db/views/view_catalog.cpp: views read fromsystem.views.tools/cdc_consumer/eloq_cdc_consumer.py: bootstrap under the gate, then follow.Follow-up work
updateDescription, pre-images, transaction atomicity on the sink, unique indexes on the sink, rename, capped collections, a Kafka connector.🤖 Generated with Claude Code