mongodb: add AWS IAM authentication (MONGODB-AWS) for Atlas - #4690
Conversation
Add an aws config block to the mongodb input, output, processor and cache, and to the mongodb_cdc input, enabling the driver-native MONGODB-AWS SASL mechanism for MongoDB Atlas. Supports the ambient AWS credential chain (env vars, EC2 instance profile, EKS pod role — refreshed automatically by the driver), explicit static keys, and STS assume-role chaining with a configurable session_duration. Credential resolution is two-phase: the aws block is parsed and validated at component construction, while credentials are resolved at connect time, so role-derived session credentials are re-resolved whenever the input, output or CDC input reconnects. The processor and cache connect once at creation under a bounded context. The AWS SDK dependency is isolated behind an injection point populated by the new internal/impl/mongodb/aws package, imported via components/aws, so binaries without AWS components fail with a clear error instead of linking the SDK. Validation rejects username/password or URL-embedded credentials combined with aws.enabled, unpaired static keys, combining role with roles, and session durations below the STS minimum. CON-527
ae207e6 to
8421ec1
Compare
…and cache The aws sub-field names in AWSIAMAuthField() were raw string literals that the aws subpackage re-typed as separate literals to read, so a typo or rename would silently resolve to "". Promote them to shared constants and use them on both sides. The mongodb processor and cache connect once at construction with no reconnect lifecycle, so role-derived STS session credentials (which expire in at most an hour under role chaining) can never be refreshed. Reject aws.role/aws.roles for those two components at startup instead of letting them fail silently later; the ambient credential chain and static keys are unaffected.
…t timeout MongoDB never re-authenticates an established pool connection, so a successful ping on Connect can succeed over an old socket even though the STS credential baked into the client for any new socket is already expired. Role-assuming input/output/CDC clients now always drop and rebuild on Connect instead of reusing a pinged client, so the credential builder re-runs and the "re-resolved whenever it reconnects" guarantee actually holds. Non-role configs keep the ping-reuse fast path. Also promotes the literal time.Minute construction bound in the processor and cache constructors to a named clientConstructTimeout constant.
The `mongodb` output returned bulk-write failures as generic errors, which the framework nacks forever against the same client. A broken connection pool - notably an expired MONGODB-AWS session credential failing the handshake for every new socket - therefore never triggered a reconnect and never re-resolved credentials. Classify driver errors that indicate the pool is unusable and return `service.ErrNotConnected` so `Connect` rebuilds the client and the batch is retried against it. The `mongodb` input ignored `cursor.Err()` on the read loop's exhaustion path, so a cursor that died mid-read looked like a clean end of input: the pipeline shut down gracefully having silently truncated the query results. Surface the error and tear the client down so the next connect re-runs the query. Also log the stream error drained on a `mongodb_cdc` reconnect, which was previously discarded, and document that a `mongodb_cdc` snapshot must complete within one STS session duration since snapshot progress is not checkpointed.
The client built in Connect() bakes in STS session credentials when role assumption is used. A long initial snapshot can consume most or all of the session (default 1h), so the streaming phase started on a client whose credentials were expired or nearly so, causing pooled connections to fail auth and forcing a reconnect that re-runs the snapshot from scratch, since snapshot progress is not checkpointed. Rebuild the client after the snapshot completes and before the change stream opens, only when aws.role/aws.roles is configured. The change stream's resume position is computed earlier from server-side state (resume token / operation timestamp) and is unaffected by the client swap.
Store the pre-snapshot stream position as soon as the snapshot completes (CONTRIBUTING 5.4.1) instead of waiting for the first streamed message, so a reconnect in that window resumes the change stream rather than replaying the whole snapshot. The store is gated on every snapshot batch having been acknowledged: persisting the position any earlier would let a restart skip a snapshot that was never fully delivered. Only a resume token is ever written to the checkpoint cache, so the pre-snapshot position is now captured as a token on replica sets too rather than only on sharded clusters; when no token is available the checkpoint is skipped and behaviour stays as it was. Also gate the post-snapshot credential refresh on a snapshot actually having run, and stop reporting an ordinary shutdown of the `mongodb` input as a cursor failure.
A snapshot batch blocked on the read channel while the connection is shutting down used to resolve its checkpoint slot without ever being delivered, which let the pending count drain and the post-snapshot checkpoint be written for a snapshot that lost its tail. Fail the snapshot in that case, and refuse to store the checkpoint when the context is already cancelled, since a drained count proves nothing once the connection is going away. Only capture the pre-snapshot resume token when a snapshot will actually run, or when the deployment reports no oplog timestamp to start from, so non-snapshot fresh starts keep their previous start position. Report no token if the position-capturing stream ever returns an event, rather than adopting a token that would skip it. Also stop the `mongodb` input from reporting a driver operation timeout as the end of its result set: only a cancelled read context, meaning shutdown, ends the input cleanly now.
…the last ack The shutdown guard discarded a completed, fully-acked snapshot's checkpoint whenever cancellation arrived during the ack-poll sleep, so a pipeline stopped shortly after its snapshot re-ran it on the next start. Since every resolve-without-delivery path now fails the snapshot errgroup, a drained pending count is trustworthy even after cancellation: store the checkpoint whenever all acks landed, using a detached context for the cache write, and abort only while acks are genuinely outstanding.
…e cases Four integration tests covering edge cases in the connection-pool recovery and checkpoint paths that the AWS IAM work depends on. `TestIntegrationConnPoolErrorShapes` feeds `isConnPoolError` errors produced by the live driver against a real server, rather than the hand-copied v2.5.0 strings the unit test pins, so a driver bump that rewords a handshake or server-selection message fails CI instead of silently defeating the classifier. `TestIntegrationOutputReconnectsAfterOutage` proves the write-error -> ErrNotConnected -> Connect rebuild loop end to end by stopping and restarting the server under a running output. The container is pinned to a fixed host port because docker re-allocates an ephemeral published port on restart, which would otherwise move the server out from under the output's url. `TestIntegrationMongoCDCUnresumableCheckpointToken` pins current behaviour when the checkpoint holds a token the server rejects: the snapshot is skipped, no documents are delivered, and the input cycles through reconnects without clearing the checkpoint. This is a known limitation - when recovery lands, the test should assert the re-snapshot instead. `TestIntegrationMongoCDCSnapshotRestartChaos` restarts the input repeatedly mid-snapshot and requires at-least-once delivery across all runs, then requires that a completed snapshot is not re-run after a further restart. Also extracts the cdc container boot out of `setup` into `startMongoContainer`, so tests needing control of the checkpoint cache directory and logger can build their own stream against the same container setup.
…stop The stream goroutine registered TriggerHasStopped last, so it fired before the deferred checkpointFlusher.Stop() ran. Connect's reconnect path waits only on HasStoppedChan and then calls Start() on the same unsynchronized Periodic, racing the in-flight Stop(); a lost race no-ops the Start and leaves the flusher dead for the new connection, silently halting periodic checkpoint writes. Register TriggerHasStopped first so it fires only after the flusher has fully stopped.
|
Two additions from local edge-case testing against real MongoDB containers: 99a3485 — four integration tests converting the review-era analysis into executable coverage: an unresumable checkpoint token pins the documented hard-fail behavior (no silent re-snapshot, no skip — and the test's first 'bogus' token turned out to be one mongo:7 happily resumes from, hence the deliberately undecodable one); real driver error shapes (handshake failure, duplicate key, server-selection timeout) act as a canary so a driver bump that changes error text fails CI instead of silently breaking the output's reconnect classification; a container stop/start drill proves the output's ErrNotConnected → rebuild loop end-to-end; and a snapshot restart-chaos test (throttled so interruptions actually land mid-snapshot) verifies at-least-once delivery plus no re-snapshot after completion. 7618d39 — a real bug the testing surfaced: the CDC stream goroutine signalled HasStopped before its deferred checkpointFlusher.Stop() ran; a reconnect's Start() could race the in-flight Stop() on the unsynchronized Periodic and lose, leaving periodic checkpoint writes silently dead for the new connection. Fixed by registering TriggerHasStopped first so it fires only after the flusher fully stops. Full mongodb integration suite: 28/28 locally. |
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| // NOTE: the fast reconnect loop this test produces also exposes a pre-existing | ||
| // race between checkpointFlusher.Start() in a new Connect and the previous | ||
| // stream goroutine's deferred Stop() (input.go's goroutine runs | ||
| // TriggerHasStopped before that deferred Stop, so the waiting Connect resumes | ||
| // first, and asyncroutine.Periodic guards neither). It is unrelated to what this | ||
| // test asserts, and only shows up under -race; integration tests are not run | ||
| // with -race. Do not "fix" it by weakening this test. |
There was a problem hiding this comment.
This NOTE is stale as of the last commit in this PR (mongodb_cdc: stop the checkpoint flusher before signalling goroutine stop). It claims "input.go's goroutine runs TriggerHasStopped before that deferred Stop", but input.go#L464-L477 now registers defer shutsig.TriggerHasStopped() first, so it fires last — after defer m.checkpointFlusher.Stop(). The race described here no longer exists.
Leaving it in place documents a fixed race as still-present and instructs future maintainers not to touch it ("Do not "fix" it by weakening this test"), which is actively misleading. Suggest deleting the NOTE paragraph, or rewording it to say the reconnect loop this test drives is what motivated the defer reordering in input.go.
| require.Eventually(t, func() bool { return len(missing(seenIDs(t))) == 0 }, | ||
| 60*time.Second, 250*time.Millisecond, "documents never delivered: %v", missing(seenIDs(t))) |
There was a problem hiding this comment.
The condition closure calls seenIDs(t), which uses require.True(t, ok, "unexpected message shape: %T", msg) and require.NoError(t, err) (L1094-L1107). require.Eventually runs the condition in a spawned goroutine, and require calls t.FailNow() → runtime.Goexit, which is invalid off the test goroutine: the tick goroutine dies without sending on testify's channel, so instead of a clear failure the test silently spins until the 60s timeout and then reports the generic "condition never satisfied" message.
This is the documented rule in the project test patterns ("Do not use require inside assert.Eventually … Use assert or return bool"). Note the pre-existing require.Eventually calls in this file only go through output.Messages(t), whose require.NoError is effectively infallible; the new _id shape/parse assertions in seenIDs can genuinely fire. Suggest switching those two to assert (or returning a bool/error from the helper) so a shape mismatch surfaces as a real failure.
| require.Eventually(t, func() bool { return len(missing(seenIDs(t))) == 0 }, | ||
| 60*time.Second, 250*time.Millisecond, "documents never delivered: %v", missing(seenIDs(t))) |
There was a problem hiding this comment.
require assertions run inside the require.Eventually condition.
seenIDs(t) calls require.True / require.NoError (L1094-L1108) and output.Messages(t) calls require.NoError (L205-L218). testify runs the Eventually condition in its own goroutine (go func() { ch <- condition() }()), so a failing require there calls FailNow() → runtime.Goexit() off the test goroutine: the condition never sends on the channel, the test does not stop, and the real cause (an unexpected message or _id shape) is replaced by the generic "documents never delivered" timeout 60s later.
This is the exact case called out in the project test patterns: "Do not use require inside assert.Eventually. require calls FailNow() which panics when called from a non-test goroutine. Use assert or return bool." (.claude/agents/tester.md, Polling).
Suggested fix: make the condition read output.Messages/decode without require — return false (or collect into a variable checked after Eventually) on an unexpected shape — and keep the require assertions on the test goroutine.
Separately, the missing(seenIDs(t)) in the failure-message args is evaluated eagerly before Eventually polls, so the reported list is always the pre-wait state.
There was a problem hiding this comment.
Fixed across 7aa9fe6 and 9bb6570 — the shape checks moved to an error-returning parser in the first commit, and the second adds a non-failing messages() variant on the output helper so the Eventually condition never reaches require.NoError through output.Messages either. The eagerly-evaluated missing(seenIDs(t)) message arg is gone; the post-wait require.Empty(missing(...)) on the test goroutine reports the real gap on failure.
| // keeps being retried - more than one failure proves the input is cycling | ||
| // through reconnects rather than having quietly settled into an idle state. | ||
| failures := logs.matching("error watching MongoDB change stream") | ||
| require.NotEmpty(t, failures, "expected the change stream open to fail, captured logs: %v", logs.records) |
There was a problem hiding this comment.
Data race: logs.records is read here without holding logs.mu, while the pipeline is still running and logCapture.Handle is appending to that same slice from the stream's goroutines (L995-L1002). stream.StopWithin is only called ~15 lines later, so the input is definitely still logging.
msgAndArgs are evaluated eagerly at the call site regardless of whether the assertion passes, so this races on every run, not just on failure. Line L1052 has the same issue via failures (that one is already a safe snapshot from matching).
Suggested fix: add a mutex-guarded snapshot accessor (or reuse logs.matching("")) and pass that instead of touching logs.records directly — the mu on logCapture is clearly intended to cover exactly this.
There was a problem hiding this comment.
Fixed in 9bb6570 — the assertion's message arg now snapshots via the existing mutex-guarded logs.matching("") instead of touching logs.records while the stream is running, with a comment noting msgAndArgs evaluate eagerly even on success.
| // NOTE: the fast reconnect loop this test produces also exposes a pre-existing | ||
| // race between checkpointFlusher.Start() in a new Connect and the previous | ||
| // stream goroutine's deferred Stop() (input.go's goroutine runs | ||
| // TriggerHasStopped before that deferred Stop, so the waiting Connect resumes | ||
| // first, and asyncroutine.Periodic guards neither). It is unrelated to what this | ||
| // test asserts, and only shows up under -race; integration tests are not run |
There was a problem hiding this comment.
This NOTE is stale as of the final state of the PR and now contradicts the code it describes.
It asserts that "input.go's goroutine runs TriggerHasStopped before that deferred Stop", but commit 7618d39 in this same PR reordered those defers precisely so that no longer happens — defer shutsig.TriggerHasStopped() is now registered first and therefore runs last, after defer m.checkpointFlusher.Stop() (input.go L464-L477).
As written, a future maintainer reading this test will believe the Start()/Stop() race is still live and may either re-"fix" it or conclude the defer ordering in input.go is arbitrary and safe to change back. Please drop the NOTE (or rewrite it to say the ordering in input.go is what keeps this fast reconnect loop safe).
There was a problem hiding this comment.
Fixed in 7aa9fe6 (same fix as the earlier thread on this NOTE — the commit hadn't been pushed yet when this re-raise ran).
The unresumable-token test's NOTE documented the flusher Start/Stop race as still present after the previous commit fixed it; reword it as the regression context. The chaos test's Eventually condition called require helpers, which FailNow on testify's tick goroutine and silently spin out the timeout instead of failing; the ID parser now returns an error checked on the test goroutine.
4.105.0 was released from main without this feature (the branch-update merge dropped the changelog entry during conflict resolution); restore the entry under 4.106.0 and bump the aws field's version tag to match.
A stored resume position can become permanently unresumable, most realistically by ageing out of the oplog window during a long snapshot, a long downtime, or slow acking downstream. The input then had no way forward: the checkpoint's presence skipped the snapshot, the dead token stopped the change stream from opening, and every reconnect retried the same position, delivering nothing, forever. Detect that case and recover from it. An unresumable position is identified only by the server error codes that prove it dead - ChangeStreamHistoryLost, InvalidResumeToken and the keystring decode failure - never by message text or error category, because clearing a live checkpoint would cost a duplicate-producing re-snapshot while transient failures must keep retrying as before. On a match the in-memory token is cleared before the terminal save can write it back and the checkpoint is deleted, so the next connection re-runs the snapshot and captures a fresh position. That is at-least-once: a re-snapshot can duplicate, it cannot lose. The delete runs on a detached, bounded context, since the stream goroutine's context is already cancelled on shutdown paths and a shutdown racing the failure would otherwise leave the dead checkpoint in place for one more start. TestIntegrationMongoCDCUnresumableCheckpointToken pinned the wedge as a known limitation and now asserts the recovery instead.
…ecovery Clearing an unresumable checkpoint was not enough on its own. Acks outlive the goroutine that produced them, so a batch resolved after the clear would set the resume token again - and with checkpoint_interval: 0 write it straight to the cache, bypassing every other guard. The restored position is no better than the one just discarded: for ChangeStreamHistoryLost the oplog is a capped FIFO, so every tracked token sits at or before the lost position and is equally gone. Version the checkpoint state instead. Each connection opens a token epoch, every write carries the epoch it was created under, and a write whose epoch has been superseded is dropped. Both the reconnect path and the unresumable-token clear open a new epoch, so a late ack can no longer resurrect a position the input has moved off. This also fixes two pre-existing problems: the cross-connection race where an ack from a previous connection could overwrite the checkpoint the current one just loaded, and the resume-token monotonicity hazard that allowed a stale token to move the stored position backwards. Connect's load of the checkpoint now takes the same mutex as every other access to it, rather than writing the field unguarded. Tighten which failures count as proof that the cached position is dead. ChangeStreamHistoryLost still clears in any phase, since history loss is a property of the oplog rather than of one token. InvalidResumeToken and the keystring decode failure now only clear when they came from opening the change stream, which is the only phase where the rejected token is provably the cached one - mid-stream they implicate the driver's in-memory token, and clearing on them would cost a needless re-snapshot. The keystring code is a generic Location code, so it is matched together with its message. Recover from a corrupt checkpoint too: bytes that fail to decode as a resume token can never decode on a retry, so failing Connect on them wedged the input just as an unresumable token used to. It is now cleared and the input starts over. Finally, describe the stream_snapshot: false case honestly. Behaviour is unchanged - a dead position leaves no better option than restarting from the current oplog position - but the warning now says the changes since the lost position are skipped, and the documented description says so too, instead of promising a re-run snapshot that cannot happen without a snapshot configured.
| // chain them (ie, from local role, privileged then cross-account). The | ||
| // resulting credentials provider is lazy: the STS calls happen when the | ||
| // credentials are first retrieved. | ||
| func assumeRoleChain(awsCfg aws.Config, roles []roleConfig, sessionDuration time.Duration, log *service.Logger) aws.Config { |
There was a problem hiding this comment.
Test coverage: the STS role-assumption path is never exercised.
assumeRoleChain is the only non-trivial logic in this file, and nothing in aws_test.go reaches it. Every test either configures no roles (TestAmbientChainCredential, TestStaticKeysCredential, TestSessionDurationIgnoredWithoutRoles) or asserts a validation error before the builder is returned (TestRolesRequireARN, TestRoleAndRolesMutuallyExclusive, TestTooShortSessionDurationRejected). TestParseRoleConfigsOrdering/TestParseRoleConfigsSingleRole call parseRoleConfigs directly, so they cover parsing but never invoke the returned builder with roleConfigs non-empty. Nothing verifies that chaining order is applied (each provider built from the previous config), that opts.Duration/opts.ExternalID reach stscreds.AssumeRoleProvider, or that Retrieve errors are wrapped.
That is the headline capability of this PR — the whole two-phase "resolve at connect, re-resolve on reconnect" design in ClientConfig, plus the processor/cache rejection and the mongodb_cdc post-snapshot refresh, exist only for role-derived credentials. Per CONTRIBUTING.md §1.3.2 tests should "prove that the connector works across supported configurations", and §1.3.3 asks for integration tests covering core workflows runnable in CI.
Suggested fix: exercise the builder with roles configured against a stubbed STS endpoint — either an httptest server plus awsconfig.WithBaseEndpoint, or the localstack testcontainers module already used elsewhere in this repo — asserting the resolved credentials come from the final role in the chain and that external ID/duration are sent.
There was a problem hiding this comment.
Fixed in bec3caf — TestRoleChainAgainstStubbedSTS exercises the full builder path against an httptest STS stub (via AWS_ENDPOINT_URL_STS, no production seams): it asserts the two AssumeRole calls happen in order, that hop 2 is SigV4-signed with the credentials hop 1 minted (the actual proof of chaining), that the external ID reaches only the hop that configured it, that session_duration propagates as DurationSeconds, and that a rejected hop fails the builder with the existing error context. Mutate-checked: reversing the chain broke the order/signing assertions; dropping external-ID forwarding broke exactly that assertion.
|
|
||
| func TestStoreSnapshotCheckpointWaitsForSnapshotAcks(t *testing.T) { | ||
| cp := checkpoint.NewCapped[bson.Raw](10) | ||
| resolve, err := cp.Track(context.Background(), nil, 5) |
There was a problem hiding this comment.
Use t.Context() rather than context.Background() in tests.
The project test patterns state: "Use t.Context() for test contexts. Exception: in t.Cleanup() functions, use context.Background() because t.Context() is already canceled during cleanup." None of the uses here are in cleanup functions.
Affected spots in this new file: cp.Track(context.Background(), …) and the storeSnapshotCheckpoint(context.Background(), …) call in TestStoreSnapshotCheckpointWaitsForSnapshotAcks, plus cp.Track / context.WithCancel(context.Background()) in TestStoreSnapshotCheckpointStopsOnShutdown and TestStoreSnapshotCheckpointStoredDespiteCancelledContext, context.WithCancel(context.Background()) in TestStoreSnapshotCheckpointStoredWhenAckWinsShutdownRace, and ctx := context.Background() in TestCommitResumeTokenDropsSupersededEpoch, TestCommitResumeTokenDefersToFlusher and TestCheckpointCacheRoundTripAndRecoverableFailures.
Each of these works with t.Context() (including the cancelled-context cases, via context.WithCancel(t.Context())), and it ensures the goroutine started in TestStoreSnapshotCheckpointWaitsForSnapshotAcks is torn down with the test rather than being left blocked on the poll loop if the test fails early.
There was a problem hiding this comment.
Fixed in bec3caf — all listed spots (plus two more of the same pattern) now use t.Context() / context.WithCancel(t.Context()); t.Cleanup uses keep context.Background() per the documented exception.
| // produced by the live driver against a real server instead, so the same bump | ||
| // fails CI. The literal substring assertions are deliberate: they name which | ||
| // message shape moved. | ||
| func TestIntegrationConnPoolErrorShapes(t *testing.T) { |
There was a problem hiding this comment.
Integration test placed in a non-integration test file.
The project test file conventions state: "Unit tests: internal/impl/category/thing_test.go next to the code they test. Integration tests: integration_test.go or {feature}_integration_test.go."
TestIntegrationConnPoolErrorShapes boots a testcontainer and guards with integration.CheckSkip(t), but lives in common_test.go alongside pure unit tests, so it isn't picked up by the naming convention the rest of the repo relies on to separate the two suites.
It can't simply move into the existing internal/impl/mongodb/integration_test.go, since that file is package mongodb_test and this test needs the unexported isConnPoolError. Splitting it into a new internal/impl/mongodb/conn_pool_integration_test.go in package mongodb keeps both the access and the convention.
There was a problem hiding this comment.
Fixed in 6efd782 — moved to internal/impl/mongodb/conn_pool_integration_test.go per the {feature}_integration_test.go convention, staying in package mongodb (with a file comment explaining why) since the canary probes the unexported isConnPoolError.
TestIntegrationConnPoolErrorShapes boots a container and is integration-
gated, so it belongs in a {feature}_integration_test.go per the test file
conventions. It stays in package mongodb (not mongodb_test) because it
probes the unexported isConnPoolError.
…eadiness The role-assumption path had no coverage: nothing invoked the credential builder with roles configured, so the sequential AssumeRole calls, the external IDs and the session duration were all unverified. A stubbed STS server, reached via AWS_ENDPOINT_URL_STS, now drives the whole chain hermetically and asserts the shape the SDK produces: one AssumeRole per configured role in config order, each hop signed with the previous hop's minted access key, the external ID only on the hop that configured it, and the final hop's credentials being the ones handed to the driver. A denied hop covers the error path. Also sweeps the CDC unit tests from context.Background() to t.Context(), so a failing test cannot leak goroutines past its own lifetime, and makes the CDC integration harness wait for a writable primary rather than just a successful ping. A ping is answered before the single-node replica set elects itself, so a fast test start raced the election and hit `(NotWritablePrimary) not primary` on its first CreateCollection - a flake that has already failed CI.
| if err := m.cursor.Err(); err != nil && ctx.Err() == nil { | ||
| // The cursor died mid-read (e.g. an expired credential broke the | ||
| // connection pool, or the client-side operation timeout elapsed waiting | ||
| // on a getMore). Tear down so the next Connect rebuilds the client | ||
| // — re-resolving IAM credentials when roles are used — and re-runs | ||
| // the query from the start (at-least-once delivery). A detached | ||
| // context lets the driver finish its endSessions handshake even when | ||
| // the read context is already gone. The condition tests our own | ||
| // context rather than the error value: an ordinary shutdown cancels it | ||
| // and must fall through to ErrEndOfInput, while a driver deadline with | ||
| // a live read context is a real failure that must not be mistaken for | ||
| // the end of the result set. | ||
| tearCtx := context.WithoutCancel(ctx) | ||
| _ = m.cursor.Close(tearCtx) | ||
| m.cursor = nil | ||
| _ = m.client.Disconnect(tearCtx) | ||
| m.client = nil | ||
| return nil, nil, fmt.Errorf("mongodb cursor failure: %w", err) | ||
| } |
There was a problem hiding this comment.
This new branch changes the mongodb input's delivery semantics — a cursor that dies mid-read used to look like a clean end of input, and now it tears down the client/cursor and returns an error, which makes the framework re-run Connect and re-issue the query from the start (re-delivering everything already emitted). That is a meaningful behavioural change and it has no test coverage: nothing in internal/impl/mongodb/common_test.go, integration_test.go or conn_pool_integration_test.go exercises the mongodb cursor failure path, the ctx.Err() == nil guard that distinguishes it from an ordinary shutdown, or the re-query on the following Connect.
By contrast the sibling recovery paths added in this PR are all covered (isConnPoolError has both a unit table and a live-driver canary, the output's ErrNotConnected → rebuild loop has TestIntegrationOutputReconnectsAfterOutage, and the CDC checkpoint recovery has unit plus integration tests), so this is the one recovery path landing untested.
Suggested fix: add coverage for it — e.g. an integration test that starts a find with a small batch_size, kills the cursor server-side (killCursors) or stops the container mid-read, then asserts ReadBatch returns an error rather than ErrEndOfInput, that m.cursor/m.client are dropped, and that the following Connect re-runs the query; plus a case proving a cancelled read context still yields service.ErrEndOfInput.
Per CONTRIBUTING.md §1.3.2/§1.3.3 ("Tests should cover end-to-end functionality and prove that the connector works across supported configurations" / "Integration tests verify core workflows and are runnable in CI"). Code in question: internal/impl/mongodb/input.go#L269-L292.
There was a problem hiding this comment.
Covered in 088bc21. TestIntegrationMongoInputCursorFailureRequery kills the cursor mid-read with a failCommand failpoint on getMore (errorCode 43 CursorNotFound — non-retryable, so the driver surfaces it rather than retrying) and asserts the full teardown → ErrNotConnected → re-query path delivers all documents, duplicates included; mutate-checked — short-circuiting the cursor.Err() branch drops 4 of 6 documents. TestIntegrationMongoInputShutdownIsNotCursorFailure pins the ctx.Err() == nil guard: a cancelled read context mid-cursor yields clean ErrEndOfInput, no error. That one drives ReadBatch directly by design — a stream-level version is provably blind to the guard mutation, since benthos stops on the same soft-stop signal that cancels the read context and filters context.Canceled from its logs (rationale in the test comment).
…ion tests ReadBatch's cursor error branch had no coverage on either side. Both are now pinned against a real mongod. The recovery side is staged with the failCommand failpoint on getMore (errorCode 43 CursorNotFound, scoped by appName), which fails exactly one getMore at a deterministic point: with batch_size 2 the find serves the first batch and every later batch needs a getMore, so the cursor always dies after the first batch is delivered. getMore is not a retryable read and 43 carries no retryable label, so the driver surfaces it instead of hiding it behind a retry. The test then requires all six documents to arrive - with the duplicate first batch that proves the query really was re-run - and the stream to terminate on its own. The shutdown side drives ReadBatch directly rather than stopping a running stream, because the distinction is invisible at the stream level: the reader cancels the read context via the same soft-stop signal it checks immediately after ReadBatch returns, so it stops on either answer, and it does not log errors wrapping context.Canceled. A stream-level test would pass with the ctx.Err() guard deleted; this one fails with "mongodb cursor failure: context canceled" instead of end-of-input. auto_replay_nacks is disabled so the auto-retry wrapper does not answer the cancelled context from its own queue before the value under test is produced.
| require.Eventually(t, func() bool { return len(output.Messages(t)) > 0 }, 30*time.Second, 10*time.Millisecond) | ||
| stream.StopWithin(t, 30*time.Second) | ||
| wait() | ||
| require.JSONEq(t, `[{"_id":{"$numberInt":"1"}, "data":"hello"}]`, output.MessagesJSON(t)) | ||
|
|
||
| // The ack of the first event wrote its position through to the cache, so this | ||
| // run resumes after it rather than replaying it. | ||
| wait = stream.RunAsync(t) | ||
| time.Sleep(time.Second) | ||
| db.InsertOne(t, "foo", bson.M{"_id": 2, "data": "world"}) | ||
| require.Eventually(t, func() bool { return len(output.Messages(t)) > 1 }, 30*time.Second, 10*time.Millisecond) |
There was a problem hiding this comment.
output.Messages(t) is called from inside a require.Eventually condition, which testify runs on its own tick goroutine. Messages calls require.NoError (via t.Helper() + require), and require's FailNow/runtime.Goexit from a non-test goroutine silently kills the tick rather than failing the test — the wait then just spins out to the 30s timeout with a misleading message.
This PR added outputHelper.messages() specifically for this case and documents it as "the non-failing variant of Messages for use inside Eventually conditions" (see the helper at integration_test.go#L207-L212), and the other three new tests in this file use it. This new test should use messages() in both conditions too:
require.Eventually(t, func() bool {
msgs, err := output.messages()
return err == nil && len(msgs) > 0
}, ...)
Per the project test patterns: "Do not use require inside assert.Eventually. require calls FailNow() which panics when called from a non-test goroutine."
There was a problem hiding this comment.
Fixed in 12c45c6 — both Eventually conditions in the flusherless resume test now go through output.messages() with the error folded into the condition; the test predated the helper's adoption in its own waits.
…est's waits The interval-0 resume test predated the messages() helper's adoption in its Eventually conditions; require inside those conditions FailNows on testify's tick goroutine and silently kills it.
Jeffail
left a comment
Jeffail
left a comment
There was a problem hiding this comment.
thanks for this — there's a lot here that reads really well. the token-epoch guard is a proper fix rather than a patch (late acks genuinely can resurrect an abandoned position, and a generation counter kills them all at once), the post-snapshot ack gate gets the delivery guarantee right (position persists after downstream acks, not at read time), and the SDK injection matches the postgres/mysql prior art exactly.
the main thing i want to flag is a composition issue: the post-snapshot checkpoint and the unresumable-position recovery each look right alone, but together they can livelock — detail in the inline comment. the rest are smaller and inline too.
two process bits:
- CHANGELOG.md — my read of
.github/workflows/release-notes.ymlis that the release job splices a fresh generated## <next> - <date>section in after the preamble and re-derives this PR's entry from its commits, so the hand-written## 4.106.0 - TBDsection would end up duplicated (second header + second entry) at release time. i think the wording wants to live in the PR title/body instead — happy to be corrected if there's a step i'm not seeing. - PR body drift — the body still says the processor and cache "resolve once at creation", but at HEAD they reject
role/rolesoutright, and well over half the diff is now the CDC checkpoint/recovery/reconnect work that the body doesn't mention. since release notes are generated from body + commits, it's probably worth either growing the body or splitting that work out — either seems fine to me.
things we checked and cleared, so they don't get re-raised: the apparent snapshot/stream gap from SetResumeAfter(initialResumeToken) (the snapshot's Find has no atClusterTime, so it covers the window — duplicates only), the epoch machinery (the un-guarded terminal save and the flusher are both safe under resumeTokenMu), the defer reordering in the stream goroutine, the m.client write from the stream goroutine (reads happen only after HasStoppedChan), and the missing slices.Clone on stream.ResumeToken() (the driver allocates per read, so the aliasing looks benign).
verified locally at 12c45c6: go build ./... clean, go test -race ./internal/impl/mongodb/... green on all three packages. integration tests left to CI via the label.
no rush on any of this — happy to dig in further on any of the inline threads.
| } | ||
| return | ||
| } | ||
| // CONTRIBUTING §5.4.1: checkpoint as soon as the snapshot completes |
There was a problem hiding this comment.
this is the big one, i think: the post-snapshot checkpoint and the unresumable-position recovery compose into a potential livelock.
the token persisted here is initialResumeToken, captured before the snapshot ran. if the snapshot outlives the oplog window, that position is already dead at the moment it's written. from what i can tell the sequence is then:
- snapshot completes, dead pre-snapshot token persisted
- stream open fails 286
ChangeStreamHistoryLost clearUnresumableCheckpointclears it- next
Connectfinds no checkpoint and re-runs the whole snapshot - → step 1, indefinitely — re-emitting the entire collection downstream each pass
before the recovery landed this wedged, which was at least loud and diagnosable; now it churns quietly behind a Warnf. nothing counts the clears or gives up. and it's exactly the case the IAM work makes likelier — role sessions cap at 1h, and the session_duration docs already concede the snapshot has to fit inside one session.
a few options, roughly in order of how much i like them:
- checkpoint snapshot progress so the snapshot converges —
readSnapshotRangealready works in_idkey ranges, so per-range completion looks like a natural checkpoint unit, and it'd fix credential-expiry-mid-snapshot at the same time - refuse to persist a position that's already provably outside the oplog window, so this fails loudly instead of churning
- give up after N consecutive unresumable clears — cheapest, least satisfying
totally fine if (1) is a follow-up PR — but i'd want at least (3) here so the failure mode isn't silent.
There was a problem hiding this comment.
Implemented your option (3) in 3a82f58, refined in 48810e8: maxConsecutiveUnresumableRecoveries = 3 — two automatic clear-and-re-snapshot recoveries, then the third consecutive failure refuses to clear, keeps the checkpoint, and fails on every reconnect with an error naming the likely cause (snapshot outliving the oplog window), the remediations (oplog size, snapshot scope/parallelism/batching, session_duration for roles), and the mandatory final step: restart the pipeline (the counter is per process) or delete the checkpoint entry. The counter resets only on proven stream progress — a live-epoch commitResumeToken, which is reachable solely from a successfully opened stream — deliberately not on snapshot completion, since every pass of the livelock completes a snapshot; and the epoch opens atomically with the counter so no in-flight ack can reset the breaker mid-decision.
Your option (1) — checkpointing snapshot progress per _id range so the snapshot converges — agreed as the durable fix and queued as the follow-up; it would also solve credential-expiry-mid-snapshot, which is the other pathology sharing this shape.
| // an ordinary shutdown. | ||
| func (m *mongoCDC) clearUnresumableCheckpoint(ctx context.Context, cause error) { | ||
| if m.snapshotParallelism == 0 { | ||
| m.logger.Warnf("Change stream position is no longer resumable and stream_snapshot is disabled; clearing the checkpoint and restarting from the current oplog position — changes since the lost position will be skipped: %v", cause) |
There was a problem hiding this comment.
with stream_snapshot: false this branch skips every change between the lost position and now — data loss in a CDC connector, surfaced only as a log warning.
the doc comment's reasoning is fair as far as it goes (the stored position is unreadable, so nothing in that window was recoverable by any means), but my read is that the choice between a silent gap and a loud stop belongs to the operator rather than the connector — some pipelines would much rather page a human than quietly resume with a hole.
maybe worth a config knob — something like on_unresumable_position: fail | reset, defaulting to fail? happy to discuss if you think the unconditional reset is the right default for CDC.
There was a problem hiding this comment.
Implemented in 3a82f58 with your knob and your default: on_unresumable_position: fail | reset, default fail — the input keeps the checkpoint for inspection and fails loudly on every reconnect; reset opts into the skip with the honest gap warning.
One scoping decision to check against your intent: the knob only governs the stream_snapshot: false case. With the snapshot enabled, recovery re-runs it automatically (bounded by the breaker from the other thread), on the reasoning that re-snapshotting is loss-free — duplicates within the at-least-once contract — so the operator-pages-me preference applies where recovery would create a gap, not where it merely repeats work. If you'd rather the knob also gate the loss-free path (some operators may care more about surprise re-snapshot cost than gaps), it's a two-line change to widen.
| // rejection of the aggregate that opens the stream (a mongo.CommandError, which | ||
| // implements ServerError) and a mid-stream stream.Err(). errors.As and errors.Is | ||
| // walk wrapped errors, so callers may classify before or after adding context. | ||
| func isUnresumableTokenError(err error) bool { |
There was a problem hiding this comment.
the code selection here might be narrower than the server's own taxonomy. mongod defines a NonResumableChangeStreamError category in error_codes.yml — literally "codes that prevent change stream resumption" — containing three codes:
- 286
ChangeStreamHistoryLost✔ (covered) - 280
ChangeStreamFatalError✘ - 464
ShardRemovedError✘
280 is what the invalidate family surfaces as — i.e. a watched collection being dropped or renamed, which is exactly the case the pre-existing TODO: Handle the resume token becoming invalid due to collection rename/drop below names. as written that path still wedges forever, one code away from the recovery you've already built.
the same open-vs-mid-stream reasoning you applied to 260 probably deserves a think for 280 (i haven't traced every way a mid-stream 280 can arise), but matching the category's codes rather than hand-picking seems closer to the server's intent. worth a look?
There was a problem hiding this comment.
Partially implemented in 99180d4, with one empirical surprise worth your eyes. The category point stands: 280/464 added as named constants (verified NonResumableChangeStreamError is exactly {280, 286, 464}), clearing in any phase.
But the drop/rename premise didn't reproduce: this input's database-level watch with the ns.coll $match is not invalidated by a collection drop or rename — both arrive as drop/rename events (the rename's ns is the source, so it passes the filter) that the default: arm skips, the stream carries on, and resuming across them succeeds. A database drop closes the cursor with a nil error because the invalidate event itself is filtered out by the $match. So 280 appears unreachable through this input's stream shape — TestIntegrationMongoCDCCollectionDropAndRename now pins the measured behavior (no recovery fires, no duplicates, exact delivery) where the stale rename/drop TODO used to sit.
One judgment call to flag: 280's any-phase clearing rests on taxonomy alone (the category is the server telling drivers not to auto-resume, not strictly that every historical position is dead — unlike 286/464 which are structurally fatal). Since we couldn't reproduce a 280 at all, we kept the category-faithful classification, with the trade documented in the classifier comment and the downside bounded by the recovery breaker. Happy to phase-gate 280 like 260 if you'd rather.
| token bson.Raw, | ||
| store func(context.Context, bson.Raw) error, | ||
| ) bool { | ||
| if token == nil { |
There was a problem hiding this comment.
this early return skips the cp.Pending() > 0 wait as well as the store — so on the no-token path, streaming starts while snapshot batches are still tracked in the same cp, which is the interleaving the gate exists to prevent.
the mechanism (checked against Jeffail/checkpoint@v1.1.0): the resolve closure does newNode.prev.payload = newNode.payload, so when a stream batch (payload tok1) resolves before an earlier snapshot batch (payload nil), the snapshot node inherits tok1. when that slot finally resolves it's the head, resolve() returns &tok1, and the snapshot ackFn trips its own unexpected resume token for snapshot batch guard.
reachable whenever stream_snapshot is on and no token could be captured (replica sets < 4.0.7, or the "stream returned an event" path in getCurrentResumeToken), with out-of-order acks — which are normal with pipeline.threads > 1 or a batching output. impact is contained — a spurious ack error plus one missed commit, later stream acks still advance the position — but it's a hole in an invariant this PR built specifically to hold.
fix looks small: run the wait whenever anything was tracked, and only skip the store when there's no token.
There was a problem hiding this comment.
Fixed in 99180d4, exactly as you described — the gate now runs whenever anything was tracked, and only the store is skipped without a token. Your inheritance mechanism verified precisely (uncapped.go:44-54: newNode.prev.payload = newNode.payload), and it's worse than a log line: the snapshot ackFn's guard turns the inherited token into a failed ack, so the batch nacks and a commit is lost. TestSnapshotSlotInheritsStreamTokenWithoutGate demonstrates the corruption against the real library, and the nil-token wait has its own test.
| // Probe the URL once so that malformed connection strings are rejected at | ||
| // startup, and so we can tell whether it already carries credentials. | ||
| probe := options.Client().ApplyURI(c.url) | ||
| if err := probe.Validate(); err != nil { |
There was a problem hiding this comment.
heads-up that this probe does a blocking SRV lookup at construction for mongodb+srv:// URLs — i reproduced it against driver v2.5.0:
Validate() err = error parsing uri: lookup _mongodb._tcp.<host> on <resolver>: no such host
so every Atlas cluster (the target of the feature) resolves DNS at startup, where a transient failure is fatal to the pipeline rather than retried with backoff by the connect lifecycle this PR just introduced. to be fair, the old getClient had the same behaviour, so it's not a regression — but it does cut against the lazy-connect rationale, and it now costs an extra SRV+TXT round trip per component since Connect re-applies the URI later.
maybe the userinfo check could ride on a plain URL parse instead of ApplyURI, deferring DNS to Connect where it's retryable? fine as-is if you'd rather keep the eager validation — just flagging the trade-off.
There was a problem hiding this comment.
Fixed in 3e25eaf, mostly your way: the userinfo rejection now rides on a textual check (per the connection-string spec's userinfo rule — the driver rejects raw '@' in the host section, so presence reduces to the authority segment containing '@'), and for mongodb+srv:// URLs the probe is skipped entirely — parse and DNS errors surface at Connect, where the lifecycle retries them. Kept the eager Validate() for plain mongodb:// URLs only, where it's DNS-free typo-catching for free. TestAWSAuthSRVURLConstructsWithoutDNS pins the headline: an +srv URL on a .invalid TLD now constructs offline (it failed with the SRV lookup error before the change), and the double SRV+TXT round trip is gone.
Trade-off accepted along the way: an +srv typo now shows up as a retrying Connect loop instead of a startup failure — the warn-level connect errors name the lookup failure, which seemed like the right side of the trade for transient-DNS resilience on exactly the Atlas URLs this feature targets.
| // to read a position, the streaming phase opens its own. | ||
| _ = stream.Close(ctx) | ||
| }() | ||
| if stream.TryNext(ctx) { |
There was a problem hiding this comment.
the comment says SetBatchSize(0) makes this unreachable, but i don't think that holds: batchSize 0 only empties the first batch, and TryNext on an empty batch issues one getMore (next(ctx, true) → loopNext in driver v2.5.0), which on a busy watched collection can return an event straight away.
when it fires, the two deployments diverge:
- mongos (
!hasTS): the caller hard-failsConnect. the framework retries with backoff so it recovers, but startup can flap under sustained write load, where the old code just resumed after the event. - replica sets: the
defaultbranch logs at Debug and silently disables the post-snapshot checkpoint — the headline behaviour of this PR skipped with no operator-visible signal.
using the event's token (as the old mongos path did) seems safe on the snapshot path, since the snapshot re-reads current data and duplicates are within the at-least-once contract. if you'd rather keep the conservative error, i think the replica-set fallback at least deserves a Warn rather than a Debug.
There was a problem hiding this comment.
You're right — I checked driver v2.5.0 and the comment was wrong: TryNext → next(ctx, true) → loopNext issues one getMore (change_stream.go:727-731), and a batchSize of 0 is omitted from the getMore command entirely (batch_cursor.go:412-414), so a busy collection can absolutely hand back an event there.
Fixed in 3e25eaf with your first option: the event's token is adopted as the start position (cloned, same as the empty-poll path), which restores the old mongos behavior and removes both divergent failure modes. The correctness argument is now in the comment: adopting the token makes that event pre-snapshot data by definition, and the snapshot scan — which starts after this call — reads the collection's current state, so nothing is lost within the at-least-once contract. ResumeToken() after a successful TryNext is the delivered event's _id or that batch's PBRT (change_stream.go:382-403), both at-or-after the event, so the resume position is consistent either way.
| return | ||
| } | ||
| if cc.AssumesRole() { | ||
| return nil, errors.New("aws.role and aws.roles cannot be used with the mongodb processor: role-derived session credentials expire and this component has no reconnect lifecycle to refresh them; use the ambient credential chain or static keys instead") |
There was a problem hiding this comment.
this guard (and its twin in cache.go) rejects role/roles because session credentials expire with no refresh path — but then points users at "static keys", and aws.token is documented as "required when using short term credentials". a processor or cache configured with id/secret/token from STS gets the same expiring snapshot with the same absence of a refresh path, is accepted silently, and starts failing auth mid-run once the session lapses.
so from what i can tell the error message currently recommends the failure mode it just rejected. two ways out: reject aws.token for these two components as well, or reword the guidance to steer at the ambient chain only. i'd lean towards rejecting, for symmetry with the reasoning already given here.
There was a problem hiding this comment.
Fixed in 3e25eaf — went with rejection for the symmetry you describe: the processor and cache now reject aws.token alongside role/roles, and the error text steers at the ambient credential chain or long-lived access keys (the docs sentence matches). Tests cover both rejections plus acceptance of long-lived id+secret.
One residual your comment surfaced: aws.token remains accepted on the input, output and mongodb_cdc, where a reconnect re-reads the same static token — so expiry there is a visible auth-failure/reconnect loop rather than the silent mid-run death these two components had, but it's still a config that can't outlive its session. Happy to tighten that too (or add a docs caveat) if you'd rather not leave it.
…up events, defer SRV lookups The processor and cache build their client once and have no reconnect lifecycle, so they already rejected aws.role/aws.roles. A static aws.token is the same expiring snapshot with no refresh path, so reject it there too and reword both errors (and the aws field docs) to name long-lived access keys as the alternative. getCurrentResumeToken treated an event from its TryNext poll as unreachable and returned an error, which hard-fails mongos startup under load and silently skips the post-snapshot checkpoint on replica sets. A batchSize of 0 only empties the first batch: TryNext still issues one getMore, which the server answers with its default batch size, so a busy collection does deliver an event. Adopt that event's resume token instead - it makes the event pre-snapshot data by definition, and the snapshot scan reads the collection's current state after this call returns, so the event's effect is captured. The construction-time options.Client().ApplyURI probe resolves SRV and TXT records for every mongodb+srv URL, which makes transient DNS fatal in a constructor that is never retried. Detect URL credentials textually with uriHasUserInfo instead and keep the eager probe only for plain mongodb:// URLs, leaving parse and DNS errors for SRV URLs to Connect, which is retried.
| // satisfied by *checkpoint.Capped. | ||
| type pendingTracker interface { | ||
| Pending() int64 | ||
| } | ||
|
|
There was a problem hiding this comment.
Hardcoded durations — the project Go patterns require every time-related value to be YAML-configurable: "Configurable Time Parameters: Every time-related value (timeouts, backoffs, intervals, retry delays) must be exposed as a YAML-configurable field. Do not hardcode durations." (.claude/agents/godev.md)
Three new durations land in this PR without a config field:
checkpointWriteTimeout = 10 * time.Second— bounds the detached post-snapshot store and the unresumable-checkpoint delete. A slow checkpoint cache (e.g. a remoteredis/dynamodbcache under load) silently loses the post-snapshot checkpoint after 10s, which costs a full re-snapshot on the next start — exactly the outcome the new checkpointing is meant to avoid, with no way for an operator to raise the bound.snapshotAckPollInterval = 100 * time.MillisecondclientConstructTimeout = time.Minuteincommon.go, which bounds processor/cache client construction.
Suggested fix: expose at least checkpointWriteTimeout (and ideally the construct timeout) as advanced duration fields with these values as defaults; the ack poll interval is the weakest case for a knob if you want to argue an exception for it.
There was a problem hiding this comment.
Partially implemented in 90ce4ab: checkpoint_write_timeout is now an advanced duration field (default 10s, rejects non-positive values since a detached context treats zero as already-expired), covering both detached writes — the post-snapshot store and the recovery clear — which is where an operator with a slow remote cache genuinely needs the knob.
Arguing the named-constant exception for the other two: snapshotAckPollInterval is internal poll mechanics with no operator-observable trade-off (the wait's duration is governed by the acks, not the interval), and clientConstructTimeout only bounds driver handshake work already capped by the client's own hardcoded 10s connect / 30s server-selection timeouts, which predate this PR — if those become configurable in a broader timeout pass, it should join them rather than gaining a lone knob here.
| _ = m.client.Disconnect(ctx) | ||
| client, db, err := m.cc.Connect(ctx, mongoClientBSONOptions) | ||
| if err != nil { | ||
| select { | ||
| case m.errorChan <- fmt.Errorf("error refreshing MongoDB credentials after snapshot: %w", err): | ||
| default: | ||
| } | ||
| return | ||
| } | ||
| m.client, m.db = client, db | ||
| } | ||
| } | ||
| if err := m.readFromStream(ctx, cp, opts); err != nil { | ||
| if err := m.readFromStream(ctx, tokenEpoch, cp, opts); err != nil { | ||
| if isUnresumableTokenError(err) { | ||
| m.clearUnresumableCheckpoint(ctx, err) | ||
| } | ||
| select { | ||
| case m.errorChan <- fmt.Errorf("error watching MongoDB change stream: %w", err): |
There was a problem hiding this comment.
Test coverage: the credential-refresh-on-reconnect guarantee is untested.
Coverage elsewhere in this PR is genuinely thorough (STS chain against a stub, pool-error classification against a live driver, checkpoint edge cases, cursor failure/shutdown paths). The one mechanism with no test is the one the feature's central promise rests on — the field docs state "Role-derived session credentials are resolved when the component connects and are re-resolved whenever it reconnects" and "credentials are freshly resolved after the initial snapshot completes":
- This post-snapshot rebuild (
snapshotParallelism > 0 && cc.AssumesRole()). - The
AssumesRole()drop-and-rebuild branches that replace the ping-reuse fast path incdc/input.goConnect,output.goConnect andinput.goConnect.
TestAWSAuthInputAcceptsRoleAssumption only proves construction succeeds; nothing asserts the builder is invoked again on a second Connect, so a regression that restored the ping-reuse path for role configs (the exact bug commit cdd8cff fixed) would leave every test green.
A cheap unit-level version: stub AWSOptFn with a builder that counts invocations, then assert the count increments across two Connect calls (and across the post-snapshot rebuild) — per the project test patterns this is the kind of lifecycle assertion Connect/ReadBatch/Close tests are for, and stubAWSOptFn in common_test.go already gives you the seam.
There was a problem hiding this comment.
Covered in fe58a40 — the counting-builder tests pin the guarantee at the exact regression you named: the input and output tests call Connect twice against a live container and assert the builder ran again for role configs (ping-reuse would leave the count at 1) AND that the ambient fast path did not rebuild (require.Same on the client pointer, so an over-correction fails too); the CDC test asserts ≥2 builds across the initial connect and the post-snapshot refresh plus continued streaming on the rebuilt client. The stub builder returns (nil, nil) — verified legal against ClientConfig.Connect and pinned by its own test with a real insert — so the counting works against a no-auth container.
… set The post-snapshot ack gate returned early when no resume token could be captured (replica sets before 4.0.7), which skipped the wait as well as the store. The wait is not only about the write: the streaming phase tracks its batches in the same checkpointer, so starting it while snapshot slots are unresolved lets a stream batch resolve ahead of an earlier snapshot batch. The checkpointer's resolve copies the resolving node's payload onto its predecessor, so the snapshot slot inherits the stream token, and its ack then trips its own "unexpected resume token for snapshot batch" guard - a spurious error, and the commit that token should have produced is lost. Hoist the wait above the token check so only the store is skipped, and keep the return contract: false still means "ctx died with acks outstanding, stop". Also classify ChangeStreamFatalError (280) and ShardRemovedError (464) as position-fatal in any phase, which makes the set exactly the server's own NonResumableChangeStreamError category alongside ChangeStreamHistoryLost. The SetResumeAfter TODO about drops and renames invalidating the stored token is replaced by what a mongo:7 replica set actually does, now pinned by TestIntegrationMongoCDCCollectionDropAndRename: this input watches at the database level, and a database-level stream survives both operations as ordinary skipped events, with the stored position still resumable afterwards.
Clearing an unresumable checkpoint and starting over is the right default, but two cases turn it into the wrong answer, and both used to happen silently. A snapshot that outlives the oplog window checkpoints a position that has already aged out, so the stream fails to open on it, recovery clears it, the snapshot re-runs and produces another dead position - forever, re-delivering the collection on every pass behind a warning. Count consecutive recoveries and stop after three, keeping the checkpoint and reporting what to do about it (grow the oplog window, shorten the snapshot, or cover it with a longer STS session). The counter resets only on proven stream progress, meaning a live-generation token accepted by commitResumeToken, which is reachable only from the streaming phase: resetting on snapshot completion would disable the breaker outright, since every pass of the loop completes a snapshot. Checkpointing progress within the snapshot is the durable fix, left as follow-up. With stream_snapshot disabled there is no snapshot to re-run, so clearing restarts from the current oplog position and skips every change since the lost one. That is now opt-in via on_unresumable_position, which defaults to fail: the checkpoint is preserved for inspection and the input stops with an error naming the opt-in. The knob has no effect when stream_snapshot is enabled, where recovery loses nothing and the breaker governs instead.
The two checkpoint writes that run on a context detached from the read loop - storing the position a completed snapshot reached, and clearing a position that can no longer be resumed from - were bounded by a hard-coded ten seconds. That is too short for slow remote caches, where the cost of losing the post-snapshot write is a full re-snapshot on the next start, so make it configurable and keep the constant as the default. A non-positive value is rejected at construction: because the context is detached, zero would not mean "unbounded" but "already expired", which would silently drop both writes.
MongoDB never re-authenticates an established pool connection, so a successful ping proves the old socket is alive but says nothing about whether the credential baked into the client's options is still valid for a new one. Role assumption therefore has to rebuild the client on every reconnect, and the ping-reuse fast path has to stay for credentials that cannot expire that way. Nothing held either half of that in place; a regression restored the ping reuse for roles once already. Cover both against a live server, plus the cdc input's post-snapshot refresh, by counting credential-builder invocations through the exported AWSOptFn seam. The counting builder returns a (nil, nil) credential, which ClientConfig.Connect treats as "apply no auth", so the containers run without authentication - a MONGODB-AWS credential could not authenticate against a test container. That contract is asserted directly too, since the tests depend on it. The cdc container helper grows a no-auth variant for the same reason.
… atomic The refuse-churn error now names the mandatory final step (restart the pipeline or delete the checkpoint entry - the refusal counter is per process), the breaker prose consistently describes two clears with refusal on the third consecutive failure, and the epoch now opens inside the same critical section as the recovery counter so no in-flight ack can reset the breaker between the two. Docs and changelog wording aligned.
| m.unresumableRecoveries++ | ||
| if m.unresumableRecoveries >= maxConsecutiveUnresumableRecoveries { | ||
| // Pinned at the limit so a wedged input's counter cannot grow without bound | ||
| // across reconnects. Only proven stream progress clears it, in | ||
| // commitResumeToken. | ||
| m.unresumableRecoveries = maxConsecutiveUnresumableRecoveries | ||
| return unresumableRecoveryRefuseChurn, m.unresumableRecoveries | ||
| } | ||
| // Open the new epoch inside the same critical section as the counter | ||
| // increment: a live-epoch ack landing between the two would reset the | ||
| // breaker without genuine progress. After the bump every in-flight ack is | ||
| // stale and is dropped instead. |
There was a problem hiding this comment.
The churn breaker also governs the no-snapshot path, which contradicts what on_unresumable_position: reset promises and produces a message that cannot apply.
The gap short-circuit only fires when the mode is fail. With stream_snapshot: false and on_unresumable_position: reset, execution falls through to m.unresumableRecoveries++, so unresumableRecoveryRefuseChurn is reachable for a configuration that has no snapshot at all. Two consequences:
- The operator explicitly opted into "clear the checkpoint and restart streaming from the current oplog position" (the field docs on
fieldOnUnresumablePosition), but on the third consecutive unresumable position in one process the input stops permanently instead. - The refusal text names snapshot-only remediations —
replSetResizeOplog, "shorten the snapshot by narrowingcollections, raisingsnapshot_parallelism" — none of which exist for this configuration, and asserts "the snapshot is very likely taking longer than the oplog window" when no snapshot ran.
Suggested fix: gate the counter increment (and therefore the churn plan) on m.snapshotParallelism > 0, so that with reset the recovery always clears, and the breaker stays scoped to the snapshot-churn livelock it was designed for. That also matches the prose already in the field description — "When stream_snapshot is enabled this field has no effect: recovery re-runs the snapshot ... the breaker governs instead" — which reads as the two mechanisms being mutually exclusive.
There was a problem hiding this comment.
Fixed in 3be6fee per your suggestion — the counter (and therefore the churn plan) is now inside the snapshotParallelism > 0 scope: with reset and no snapshot, recovery always clears, matching the field's promise, and the snapshot-remediation message can no longer fire for a configuration with no snapshot. The two mechanisms are now genuinely mutually exclusive as the field docs describe.
| // state rather than an as-of-`ts` view - therefore captures its effect. | ||
| // Anything later than the token is streamed as usual, so at-least-once | ||
| // still holds. This is the behaviour sharded clusters already had, where | ||
| // a token is the only available start position. | ||
| m.logger.Debugf("change stream returned an event while capturing the start position, adopting its resume token as the pre-snapshot position") | ||
| } | ||
| if rt := stream.ResumeToken(); rt != nil { | ||
| return rt, nil | ||
| // The driver hands back a token aliasing the cursor's batch buffer, so | ||
| // copy it before the stream is closed. | ||
| return bson.Raw(slices.Clone([]byte(rt))), nil | ||
| } | ||
| return nil, errors.New("unable to determine start position prior to snapshot phase") | ||
| } |
There was a problem hiding this comment.
The "at-least-once still holds" justification depends on a snapshot running, but this call is now also reached when no snapshot will run.
After TryNext returns true, the driver's cached resume token is the _id of the delivered event (the per-document token, not the post-batch one), so SetResumeAfter(token) at L753 resumes after that event — it is never delivered downstream, since this short-lived stream is closed and discarded.
The comment argues the snapshot scan covers it. That is sound on the m.snapshotParallelism > 0 branch of the new gate at L673, but the gate has a second arm, !hasTS, which is taken on a sharded cluster (mongos reports no lastWrite.majorityOpTime) regardless of stream_snapshot. On mongodb_cdc against a sharded deployment with stream_snapshot: false and no existing checkpoint, there is no snapshot to capture the event's effect, so a change on a busy cluster is silently skipped on first start.
Either return no token when m.snapshotParallelism == 0 and an event was consumed (falling back to the existing error/behaviour), or scope the comment's claim to the snapshot case and record the gap as a documented limitation per §1.2.3.
There was a problem hiding this comment.
Fixed in 3be6fee with your first option: when snapshotParallelism == 0 and the capture stream delivers an event, no token is returned — the sharded no-snapshot caller turns that into a retried Connect failure (loud and recoverable) rather than a silent skip, and the error text points at enabling stream_snapshot for robust first starts on busy clusters. The comment's at-least-once claim is now scoped to the snapshot arm.
| _, err = db.Collection("testcoll").InsertOne(t.Context(), bson.M{"_id": 1}) | ||
| require.NoError(t, err) | ||
| if err != nil && strings.Contains(err.Error(), "auth") { | ||
| t.Fatalf("unexpected authentication involvement: %v", err) | ||
| } |
There was a problem hiding this comment.
This block is unreachable: require.NoError(t, err) on the preceding line calls FailNow when err != nil, so err is always nil here and the strings.Contains check never runs. It's also the only use of the strings import in the file.
The assertion the comment is reaching for ("no authentication was involved") is already covered by require.NoError — an auth failure would fail there. Dropping the dead if (and the now-unused import) keeps the test honest about what it actually verifies.
There was a problem hiding this comment.
Fixed in 3be6fee — dead branch and the strings import dropped; the comment on the insert now states that require.NoError is the auth assertion.
…shot runs The churn breaker counts snapshot re-runs, so it must not govern the no-snapshot reset path where the operator opted into always clearing; and an event consumed while capturing the start position is only covered when a snapshot follows, so the sharded no-snapshot case errors loudly instead of silently skipping the event. Also drops an unreachable assertion from the credential-refresh test.
| if derr := m.checkpoint.Delete(ctx); derr != nil { | ||
| return fmt.Errorf("unable to clear the corrupt checkpoint: %w", derr) | ||
| } | ||
| resumeToken = nil | ||
| } | ||
| // Open this connection's checkpoint generation, abandoning any token still in | ||
| // flight from a previous one: those belong to a position this connection is | ||
| // not resuming from, and the loaded checkpoint is the only start position that | ||
| // counts now. resumeToken is used locally from here on, since the field can be | ||
| // advanced concurrently by acks once the stream is running. | ||
| tokenEpoch := m.beginTokenEpoch(resumeToken) |
There was a problem hiding this comment.
Corrupt-checkpoint recovery bypasses on_unresumable_position and the churn breaker, so it can silently skip the gap this PR made opt-in.
Every other dead-position path in this PR goes through recoverFromUnresumablePosition/planUnresumableRecovery, which refuses to clear when there is no snapshot to re-run unless the operator opted in (on_unresumable_position, default fail), and counts consecutive clears towards the breaker. This branch clears unconditionally.
Failure scenario: stream_snapshot: false with the default on_unresumable_position: fail, and a checkpoint entry whose bytes are no longer decodable (partial cache write, cache backend migration, hand-edited entry). checkpoint.Load returns errCorruptCheckpoint, this branch deletes the entry, resumeToken is nil and initialResumeToken stays nil on a replica set (hasTS is true, snapshotParallelism == 0), so the stream opens at nextTimestamp(ts) and every change between the stored position and now is dropped behind one warning — the exact outcome the input otherwise "refuses to do silently". m.unresumableRecoveries is not incremented either, so an entry that keeps getting corrupted never trips the breaker.
Suggested fix: treat a corrupt entry as an unresumable position — consult onUnresumablePosition (and the recovery counter) before deleting, so the no-snapshot case fails loudly by default. If the asymmetry is deliberate, say so in the on_unresumable_position description, since operators will otherwise read the default as covering all unusable stored positions.
There was a problem hiding this comment.
Fixed in 222eb5b — a corrupt entry now goes through the same planUnresumableRecovery policy as every other dead position: with no snapshot and the default fail, Connect refuses, keeps the entry for inspection, and fails on every reconnect with an error naming the reset opt-in and the repair/delete remediation; with a snapshot (or reset), it clears as before, and snapshot-path clears now count toward the churn breaker so a repeatedly-corrupted entry can no longer churn silently. TestIntegrationMongoCDCCorruptCheckpointWithoutSnapshot pins the refusal: no messages delivered, corrupt bytes byte-identical after shutdown.
A corrupt checkpoint entry is an unresumable position by another route, so clearing it now consults on_unresumable_position and the churn breaker like every other dead-position path: with no snapshot to re-run and the default fail mode, the input keeps the entry and fails loudly instead of silently skipping every change since the stored position.
Jeffail
left a comment
Jeffail
left a comment
There was a problem hiding this comment.
thanks for the thorough follow-through here — every thread from the review got a real fix rather than a patch, and the tests pinning the checkpoint gate corruption and the drop/rename behaviour are genuinely nice to have. from what I can tell all the judgment calls you flagged (280's any-phase clearing, the knob scoping, the hardcoded-duration exceptions) land on the right side, so happy to leave them as they are. the residual aws.token acceptance on the reconnecting components seems fine to leave too — worth a docs caveat or a follow-up someday, but no rush. lovely work 👍
2 participants
Adds AWS IAM authentication (the driver-native
MONGODB-AWSSASL mechanism) to themongodbinput, output, processor and cache, and to themongodb_cdcinput, primarily for MongoDB Atlas database users backed by AWS IAM.Ticket: CON-527
Configuration
A new
awsblock, shaped after thepostgres_cdc/mysql_cdcprior art:Three credential paths:
id/secret/token): passed through to the driver.roleorroleschaining, optional external IDs): resolved via STS with a configurablesession_duration(default 1h).Design notes
awsblock is parsed and validated at component construction (misconfiguration fails at startup), but credentials are resolved when the component connects — so the input, output and CDC input re-resolve role-derived session credentials whenever they reconnect. The processor and cache have no connect lifecycle and resolve once at creation (documented; the ambient chain is recommended for long-running pipelines).internal/impl/mongodb/awspackage, imported throughpublic/components/aws(same injection pattern as postgres/mysql). Binaries without AWS components rejectaws.enabledwith a clear error.username/password(explicit or URL-embedded) cannot combine withaws.enabled; static keys must be paired (id+secret,tokenrequires both);roleandrolesare mutually exclusive;session_durationhas a 15-minute STS floor. Credential-less URLs carryingauthMechanism=MONGODB-AWSremain accepted.Testing
17 hermetic unit tests (no network/AWS/Docker): validation rejections, credential mapping (static keys →
Username/Password/AWS_SESSION_TOKEN, role parsing/ordering), the not-imported stub error, and a lint-based test proving theawsfield is registered in themongodb_cdcspec. The MONGODB-AWS SASL handshake itself is driver-delegated and cannot run against vanillamongod, matching the coverage stance of the postgres/mysql IAM implementations.The Atlas database user must be created with the AWS IAM authentication type; TLS is required (always on for Atlas).