From 4158ffcfb041a8ad16fd6e5af75b68c893cd84f7 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Mon, 10 Aug 2026 14:20:31 -0400 Subject: [PATCH 01/11] aws_dynamodb_cdc: add auto_replay_nacks The input had no nack redelivery: a rejected batch pinned its shard's checkpoint frontier (correctly, no loss) but the records were only ever redelivered by a restart. Wrap with the standard AutoRetryNacksBatchedToggled so transient downstream failures replay in-process by default, consistent with the other CDC inputs; disabling the toggle keeps the pin-until-restart behavior. --- internal/impl/aws/dynamodb/input_cdc.go | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/internal/impl/aws/dynamodb/input_cdc.go b/internal/impl/aws/dynamodb/input_cdc.go index 6d455a25ef..de8ea0c7b7 100644 --- a/internal/impl/aws/dynamodb/input_cdc.go +++ b/internal/impl/aws/dynamodb/input_cdc.go @@ -245,6 +245,7 @@ When `+"`global_table`"+` is enabled the principal additionally needs `+"`dynamo Description("Maximum number of shards to track simultaneously. Prevents memory issues with extremely large tables."). Default(10000). Advanced(), + service.NewAutoRetryNacksToggleField(), service.NewDurationField(dciFieldThrottleBackoff). Description("Time to wait when applying backpressure due to too many in-flight messages."). Default(defaultDynamoDBThrottleBackoff). @@ -360,7 +361,14 @@ func init() { err := service.RegisterBatchInput( "aws_dynamodb_cdc", dynamoDBCDCInputConfig(), func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchInput, error) { - return newDynamoDBCDCInputFromConfig(conf, mgr) + in, err := newDynamoDBCDCInputFromConfig(conf, mgr) + if err != nil { + return nil, err + } + // With the toggle on (default) transient downstream failures are + // replayed in-process; with it off a nack pins the shard's + // checkpoint frontier and redelivery happens on restart. + return service.AutoRetryNacksBatchedToggled(conf, in) }) if err != nil { panic(err) From 943c926a7f059fc98566acf6d449b1d1f8246de8 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Mon, 10 Aug 2026 14:29:02 -0400 Subject: [PATCH 02/11] aws_dynamodb_cdc: apply start_from only to genuinely fresh pipelines start_from: latest was applied to every checkpoint-less shard, including stream-rotation children discovered by the periodic refresh (~every 4h on active streams) and children created while the pipeline was down. Starting those at LATEST silently skipped their backlog in steady state, no crash required. start_from is now honored only on the first shard discovery of a pipeline whose checkpoint store holds no prior state (new Checkpointer.HasAnyState probe); shards discovered on later refresh cycles or on restart with existing state always start at TRIM_HORIZON. Exact-checkpoint and global-table failover resume paths are unchanged. --- internal/impl/aws/dynamodb/checkpoint.go | 26 ++++++++ internal/impl/aws/dynamodb/checkpoint_test.go | 43 +++++++++++++ internal/impl/aws/dynamodb/input_cdc.go | 60 +++++++++++++++++-- 3 files changed, 125 insertions(+), 4 deletions(-) diff --git a/internal/impl/aws/dynamodb/checkpoint.go b/internal/impl/aws/dynamodb/checkpoint.go index be59f7bfb0..cb045117ae 100644 --- a/internal/impl/aws/dynamodb/checkpoint.go +++ b/internal/impl/aws/dynamodb/checkpoint.go @@ -332,6 +332,32 @@ func (c *Checkpointer) CheckpointLimit() int { return c.checkpointLimit } +// HasAnyState reports whether any checkpoint state (shard checkpoints or +// snapshot progress) exists under this pipeline's namespace and stream/table +// key. Used to scope start_from to genuinely fresh pipelines: shards that +// appear once state exists are stream-rotation children whose backlog must +// not be skipped. +func (c *Checkpointer) HasAnyState(ctx context.Context) (bool, error) { + result, err := c.svc.Query(ctx, &dynamodb.QueryInput{ + TableName: aws.String(c.tableName), + KeyConditionExpression: aws.String("#hk = :hv"), + ExpressionAttributeNames: map[string]string{ + "#hk": c.hashAttrName(), + }, + ExpressionAttributeValues: map[string]types.AttributeValue{ + ":hv": &types.AttributeValueMemberS{Value: c.hashKeyValue()}, + }, + Limit: aws.Int32(1), + }) + if err != nil { + if _, ok := errors.AsType[*types.ResourceNotFoundException](err); ok { + return false, nil + } + return false, fmt.Errorf("probing checkpoint state for table=%s key=%s: %w", c.tableName, c.hashKeyValue(), err) + } + return len(result.Items) > 0, nil +} + type resumeMode int const ( diff --git a/internal/impl/aws/dynamodb/checkpoint_test.go b/internal/impl/aws/dynamodb/checkpoint_test.go index 5831efc508..85ed15e0f0 100644 --- a/internal/impl/aws/dynamodb/checkpoint_test.go +++ b/internal/impl/aws/dynamodb/checkpoint_test.go @@ -277,6 +277,49 @@ func shardRow(streamArn, shardID, seq, ts string) map[string]types.AttributeValu return row } +func TestHasAnyState(t *testing.T) { + t.Run("false on an empty partition", func(t *testing.T) { + c := globalCheckpointerWithPartition(t, "arn:A", nil) + has, err := c.HasAnyState(context.Background()) + require.NoError(t, err) + require.False(t, has) + }) + + t.Run("true when any row exists", func(t *testing.T) { + c := globalCheckpointerWithPartition(t, "arn:A", []map[string]types.AttributeValue{ + shardRow("arn:A", "shard-1", "seq-9", ""), + }) + has, err := c.HasAnyState(context.Background()) + require.NoError(t, err) + require.True(t, has) + }) + + t.Run("queries the namespaced hash key", func(t *testing.T) { + var queryIn *dynamodb.QueryInput + api := &fakeCheckpointAPI{ + describeTable: func(context.Context, *dynamodb.DescribeTableInput, ...func(*dynamodb.Options)) (*dynamodb.DescribeTableOutput, error) { + return &dynamodb.DescribeTableOutput{Table: &types.TableDescription{TableStatus: types.TableStatusActive}}, nil + }, + query: func(_ context.Context, in *dynamodb.QueryInput, _ ...func(*dynamodb.Options)) (*dynamodb.QueryOutput, error) { + queryIn = in + return &dynamodb.QueryOutput{}, nil + }, + } + c, err := NewCheckpointer(context.Background(), api, CheckpointerConfig{ + TableName: "cps", SourceTable: "t", StreamArn: "arn:A", + Namespace: "dev", CheckpointLimit: 1, Region: "us-east-1", + }, checkpointTestLogger()) + require.NoError(t, err) + + _, err = c.HasAnyState(context.Background()) + require.NoError(t, err) + require.NotNil(t, queryIn) + hv, ok := queryIn.ExpressionAttributeValues[":hv"].(*types.AttributeValueMemberS) + require.True(t, ok) + require.Equal(t, "dev#arn:A", hv.Value) + }) +} + func TestCDCCheckpointProbeNeeded(t *testing.T) { // Only an exact, same-region resume can be a stale checkpoint: its sequence // number belongs to this stream, so a failed iterator means trimmed data. diff --git a/internal/impl/aws/dynamodb/input_cdc.go b/internal/impl/aws/dynamodb/input_cdc.go index de8ea0c7b7..270291ae17 100644 --- a/internal/impl/aws/dynamodb/input_cdc.go +++ b/internal/impl/aws/dynamodb/input_cdc.go @@ -414,6 +414,13 @@ type tableStream struct { shardReaders map[string]*dynamoDBShardReader snapshot *snapshotState shardRefreshCh chan struct{} // Signal coordinator to refresh shards immediately + + // honorStartFrom is true only until the first successful shard discovery + // of a pipeline with no pre-existing checkpoint state. start_from applies + // exclusively to that case: shards that appear later (stream rotation + // children) or after a restart with state must start at TRIM_HORIZON, or + // their backlog would be silently skipped under start_from: latest. + honorStartFrom atomic.Bool } // dynamoDBCDCInput is the main input struct for DynamoDB CDC. @@ -446,6 +453,14 @@ type dynamoDBCDCInput struct { pendingAcks sync.WaitGroup backgroundWorkers sync.WaitGroup // Tracks background goroutines for proper cleanup closed atomic.Bool + + // honorStartFrom (single-table path; see tableStream.honorStartFrom for + // multi-table) is true only until the first successful shard discovery of + // a pipeline with no pre-existing checkpoint state. start_from applies + // exclusively to that case: shards that appear later (stream rotation + // children) or after a restart with state must start at TRIM_HORIZON, or + // their backlog would be silently skipped under start_from: latest. + honorStartFrom atomic.Bool } type dynamoDBCDCMetrics struct { @@ -1038,6 +1053,15 @@ func (d *dynamoDBCDCInput) connectSingleTable(ctx context.Context, tableName str // Initialize record batcher d.recordBatcher = NewRecordBatcher(d.conf.maxTrackedShards, d.conf.checkpointLimit, d.log) + // start_from only applies to a genuinely fresh pipeline: if any checkpoint + // state already exists, shards without checkpoints are rotation children + // created while we were down and must be read from TRIM_HORIZON. + hasState, err := d.checkpointer.HasAnyState(ctx) + if err != nil { + return fmt.Errorf("probing checkpoint state: %w", err) + } + d.honorStartFrom.Store(!hasState) + d.log.Infof("Connected to DynamoDB stream: %s", *d.streamArn) // Handle snapshot mode @@ -1151,6 +1175,15 @@ func (d *dynamoDBCDCInput) initializeTableStream(ctx context.Context, tableName // Initialize record batcher for this table recordBatcher := NewRecordBatcher(d.conf.maxTrackedShards, d.conf.checkpointLimit, d.log) + // start_from only applies to a genuinely fresh pipeline: if any checkpoint + // state already exists for this table, shards without checkpoints are + // rotation children created while we were down and must be read from + // TRIM_HORIZON. + hasState, err := checkpointer.HasAnyState(ctx) + if err != nil { + return false, fmt.Errorf("probing checkpoint state for table %s: %w", tableName, err) + } + // Re-check under write lock before inserting (another goroutine may have // initialized this table concurrently during periodic discovery). d.mu.Lock() @@ -1172,6 +1205,7 @@ func (d *dynamoDBCDCInput) initializeTableStream(ctx context.Context, tableName shardReaders: make(map[string]*dynamoDBShardReader), shardRefreshCh: make(chan struct{}, 1), } + ts.honorStartFrom.Store(!hasState) d.tableStreams[tableName] = ts d.log.Infof("Initialized table stream for %s (stream ARN: %s)", tableName, streamArn) @@ -1565,12 +1599,17 @@ func (d *dynamoDBCDCInput) refreshShards(ctx context.Context) error { cutoff = decision.Cutoff d.log.Infof("Failover resume for shard %s from trim horizon, skipping records at/before %s", shardID, cutoff.Format(time.RFC3339)) default: - if d.conf.startFrom == "latest" { + // start_from: latest is honored only on the first discovery of a + // fresh pipeline. Checkpoint-less shards found on later refresh + // cycles (or after a restart with existing state) are stream + // rotation children: starting them at LATEST would silently skip + // their backlog. + if d.conf.startFrom == "latest" && d.honorStartFrom.Load() { iteratorType = types.ShardIteratorTypeLatest } else { iteratorType = types.ShardIteratorTypeTrimHorizon } - d.log.Infof("Starting shard %s from %s", shardID, d.conf.startFrom) + d.log.Infof("Starting shard %s from %s", shardID, iteratorType) } // Get shard iterator (I/O operation - do not hold lock) @@ -1612,6 +1651,10 @@ func (d *dynamoDBCDCInput) refreshShards(ctx context.Context) error { d.metrics.shardsTracked.Set(int64(totalShards)) } + // The first successful discovery has positioned the fresh pipeline's + // initial shards; anything discovered from here on is a rotation child. + d.honorStartFrom.Store(false) + return nil } @@ -1866,12 +1909,17 @@ func (d *dynamoDBCDCInput) refreshTableShards(ctx context.Context, tableName str cutoff = decision.Cutoff d.log.Infof("Failover resume for shard %s (table %s) from trim horizon, skipping records at/before %s", shardID, tableName, cutoff.Format(time.RFC3339)) default: - if d.conf.startFrom == "latest" { + // start_from: latest is honored only on the first discovery of a + // fresh pipeline. Checkpoint-less shards found on later refresh + // cycles (or after a restart with existing state) are stream + // rotation children: starting them at LATEST would silently skip + // their backlog. + if d.conf.startFrom == "latest" && ts.honorStartFrom.Load() { iteratorType = types.ShardIteratorTypeLatest } else { iteratorType = types.ShardIteratorTypeTrimHorizon } - d.log.Infof("Starting shard %s (table %s) from %s", shardID, tableName, d.conf.startFrom) + d.log.Infof("Starting shard %s (table %s) from %s", shardID, tableName, iteratorType) } // Get shard iterator @@ -1912,6 +1960,10 @@ func (d *dynamoDBCDCInput) refreshTableShards(ctx context.Context, tableName str d.updateTotalShardsMetric() } + // The first successful discovery has positioned the fresh pipeline's + // initial shards; anything discovered from here on is a rotation child. + ts.honorStartFrom.Store(false) + return nil } From 3ede5136f414e04e3c9d600fb27d42a35872aab2 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Mon, 10 Aug 2026 14:35:55 -0400 Subject: [PATCH 03/11] aws_dynamodb_cdc: gate snapshot checkpoints and completion on downstream acks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Snapshot segment progress (UpdateSnapshotProgress) was persisted at read time inside the scanner, and MarkSnapshotComplete fired as soon as the scan finished, while the snapshot batches' ack function was a no-op — so a crash (or a terminal nack with auto_replay_nacks disabled) after persistence but before delivery skipped the un-acked items on restart. The scanner no longer persists anything: each batch carries its scan resume position to the input, a per-segment ordered tracker (mirroring the CDC path's RecordBatcher) persists only the highest contiguous acknowledged position, segments are sealed behind their in-flight batches so Complete=true persists only after they all ack, and the snapshot is only marked complete once every emitted batch has been acknowledged (a nack fails the gate, with per-attempt reset so it cannot livelock reconnects). --- internal/impl/aws/dynamodb/input_cdc.go | 139 ++++++++++++++--- internal/impl/aws/dynamodb/snapshot.go | 116 ++++++-------- internal/impl/aws/dynamodb/snapshot_ack.go | 145 +++++++++++++++++ .../impl/aws/dynamodb/snapshot_ack_test.go | 147 ++++++++++++++++++ 4 files changed, 459 insertions(+), 88 deletions(-) create mode 100644 internal/impl/aws/dynamodb/snapshot_ack.go create mode 100644 internal/impl/aws/dynamodb/snapshot_ack_test.go diff --git a/internal/impl/aws/dynamodb/input_cdc.go b/internal/impl/aws/dynamodb/input_cdc.go index 270291ae17..f4c42dd7f5 100644 --- a/internal/impl/aws/dynamodb/input_cdc.go +++ b/internal/impl/aws/dynamodb/input_cdc.go @@ -499,6 +499,62 @@ type snapshotState struct { scanner *SnapshotScanner recordsRead atomic.Int64 segmentsTotal int + + // ackTracker gates snapshot progress persistence on downstream acks. + ackTracker *snapshotAckTracker + // ackWG counts emitted snapshot batches that have not yet been acked or + // nacked; the snapshot is only marked complete once it drains cleanly. + ackWG sync.WaitGroup + // nackErr records the first snapshot batch nack. auto_replay_nacks is + // user-toggleable, so a nack can be terminal: the completion gate must + // fail rather than mark the snapshot complete over undelivered items. + // Cleared per connection attempt (the state outlives reconnects). + nackMu sync.Mutex + nackErr error +} + +func (s *snapshotState) recordNack(err error) { + s.nackMu.Lock() + defer s.nackMu.Unlock() + if s.nackErr == nil { + s.nackErr = err + } +} + +func (s *snapshotState) nackError() error { + s.nackMu.Lock() + defer s.nackMu.Unlock() + return s.nackErr +} + +// resetAckGate clears any nack recorded by a previous snapshot attempt so the +// completion gate judges only the current run. The WaitGroup is deliberately +// left untouched — batches from a previous attempt that are still in flight +// can yet be acked or nacked, and both must keep counting. +func (s *snapshotState) resetAckGate() { + s.nackMu.Lock() + defer s.nackMu.Unlock() + s.nackErr = nil +} + +// waitAcks blocks until every emitted snapshot batch has been acked or +// nacked, or ctx is cancelled; a recorded nack fails the gate. +func (s *snapshotState) waitAcks(ctx context.Context) error { + drained := make(chan struct{}) + go func() { + // May outlive this call if ctx fires first; bounded by process lifetime. + s.ackWG.Wait() + close(drained) + }() + select { + case <-drained: + if err := s.nackError(); err != nil { + return fmt.Errorf("snapshot batch was rejected downstream: %w", err) + } + return nil + case <-ctx.Done(): + return ctx.Err() + } } // snapshotSequenceBuffer tracks sequence numbers seen during snapshot for deduplication. @@ -1332,21 +1388,24 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st d.snapshot.state.Store(snapshotStateInProgress) d.metrics.snapshotState.Set(int64(snapshotStateInProgress)) - // Initialize snapshot scanner + // Initialize snapshot scanner. Progress persistence is ack-gated: the + // tracker below persists a segment's position only once every batch at or + // below it has been acknowledged downstream, and the completion gate is + // reset per connection attempt. + d.snapshot.ackTracker = newSnapshotAckTracker(d.checkpointer, 10 /* persist every 10 acked batches */, d.log) + d.snapshot.resetAckGate() d.snapshot.scanner = NewSnapshotScanner(SnapshotScannerConfig{ - Client: d.dynamoClient, - Table: tableName, - Segments: d.conf.snapshot.segments, - BatchSize: d.conf.snapshot.batchSize, - Throttle: d.conf.snapshot.throttle, - Checkpointer: d.checkpointer, - CheckpointInterval: 10, // Checkpoint every 10 batches (10x cost reduction) - Logger: d.log, + Client: d.dynamoClient, + Table: tableName, + Segments: d.conf.snapshot.segments, + BatchSize: d.conf.snapshot.batchSize, + Throttle: d.conf.snapshot.throttle, + Logger: d.log, }) // Set batch callback to send snapshot records to msgChan - d.snapshot.scanner.SetBatchCallback(func(ctx context.Context, items []map[string]dynamodbtypes.AttributeValue, segment int) error { - return d.handleSnapshotBatch(ctx, items, segment, tableName) + d.snapshot.scanner.SetBatchCallback(func(ctx context.Context, items []map[string]dynamodbtypes.AttributeValue, segment int, lastKey map[string]dynamodbtypes.AttributeValue) error { + return d.handleSnapshotBatch(ctx, items, segment, tableName, lastKey) }) // Set progress callback to update metrics @@ -1354,9 +1413,14 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st d.metrics.snapshotSegmentsActive.Set(int64(d.snapshot.scanner.ActiveSegments())) }) - // Set checkpoint failure callback to track failures - d.snapshot.scanner.SetCheckpointFailedCallback(func(_ int, _ error) { - d.metrics.checkpointFailures.Incr(1) + // Seal each segment behind its in-flight batches so Complete=true only + // persists after they are all acknowledged. + d.snapshot.scanner.SetSegmentSealedCallback(func(ctx context.Context, segment int) error { + if err := d.snapshot.ackTracker.SealSegment(ctx, segment); err != nil { + d.metrics.checkpointFailures.Incr(1) + return err + } + return nil }) // Set segment completion callback to track scan duration @@ -1394,6 +1458,21 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st } } + // The scan has read everything, but the snapshot is only complete once + // every emitted batch is acknowledged downstream: marking it complete + // any earlier would let a crash (or a terminal nack) skip un-acked + // items on restart. Blocks until acks drain or soft-stop. + if err := d.snapshot.waitAcks(scanCtx); err != nil { + wrappedErr := fmt.Errorf("snapshot completion gate for table %s: %w", tableName, err) + d.log.Errorf("%v (the snapshot will resume from acknowledged progress on restart)", wrappedErr) + d.snapshot.errOnce.Do(func() { + d.snapshot.err = wrappedErr + }) + d.snapshot.state.Store(snapshotStateFailed) + d.metrics.snapshotState.Set(int64(snapshotStateFailed)) + return + } + // Snapshot complete d.snapshot.endTime = time.Now() d.snapshot.state.Store(snapshotStateComplete) @@ -2705,8 +2784,10 @@ func (d *dynamoDBCDCInput) startShardReader(ctx context.Context, shardID string) } } -// handleSnapshotBatch processes a batch of items from the snapshot scan -func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[string]dynamodbtypes.AttributeValue, segment int, tableName string) error { +// handleSnapshotBatch processes a batch of items from the snapshot scan. +// lastKey is the scan position after this batch; it is registered with the +// segment's ordered ack tracker and persisted only once acknowledged. +func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[string]dynamodbtypes.AttributeValue, segment int, tableName string, lastKey map[string]dynamodbtypes.AttributeValue) error { if len(items) == 0 { return nil } @@ -2762,12 +2843,19 @@ func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[ d.log.Warn("Snapshot deduplication buffer overflowed - duplicates may occur during CDC overlap") } - // Track pending ack + // Register with the segment's ordered ack tracker: the scan position is + // only persisted once this batch (and everything before it) is acked. + snapshot := d.snapshot + resolve := snapshot.ackTracker.TrackBatch(segment, lastKey, len(batch)) + tracker := snapshot.ackTracker + + // Track pending acks: the global gauge and the snapshot completion gate. d.pendingAcks.Add(1) + snapshot.ackWG.Add(1) - // Create simple ack function for snapshot records - ackFunc := func(_ context.Context, err error) error { + ackFunc := func(ackCtx context.Context, err error) error { defer d.pendingAcks.Done() + defer snapshot.ackWG.Done() if d.closed.Load() { d.log.Debug("Received snapshot ack after close, dropping") @@ -2775,10 +2863,20 @@ func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[ } if err != nil { - d.log.Warnf("Snapshot batch nacked from segment %d: %v", segment, err) + // auto_replay_nacks is user-toggleable, so a nack can be terminal. + // Never resolve: the segment's persisted position stays pinned + // before this batch, and the completion gate fails so the snapshot + // is not marked complete over undelivered items. + snapshot.recordNack(err) + d.log.Errorf("Snapshot batch rejected downstream (segment %d): the segment's checkpoint is pinned before this batch and the snapshot will not be marked complete, unless the batch is redelivered (auto_replay_nacks) or the pipeline restarts: %v", segment, err) return err } + if ackErr := tracker.Ack(ackCtx, segment, len(batch), resolve); ackErr != nil { + d.metrics.checkpointFailures.Incr(1) + d.log.Errorf("Failed to checkpoint snapshot segment %d after ack: %v", segment, ackErr) + return ackErr + } return nil } @@ -2786,6 +2884,7 @@ func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[ select { case <-ctx.Done(): d.pendingAcks.Done() // Undo the Add(1) above + snapshot.ackWG.Done() return ctx.Err() case d.msgChan <- asyncMessage{msg: batch, ackFn: ackFunc}: d.log.Debugf("Sent snapshot batch of %d records from segment %d", len(batch), segment) diff --git a/internal/impl/aws/dynamodb/snapshot.go b/internal/impl/aws/dynamodb/snapshot.go index 3f02e00eda..8b0f88fd21 100644 --- a/internal/impl/aws/dynamodb/snapshot.go +++ b/internal/impl/aws/dynamodb/snapshot.go @@ -32,15 +32,13 @@ type DynamoItems = []map[string]dynamodbtypes.AttributeValue // SnapshotScannerConfig holds configuration for snapshot scanning. type SnapshotScannerConfig struct { - Client *dynamodb.Client - Table string - Segments int - BatchSize int - Throttle time.Duration - MaxBackoff time.Duration // Maximum backoff on throttling errors (0 = no limit). - Checkpointer *Checkpointer - CheckpointInterval int // Checkpoint every N batches (default: 10). - Logger *service.Logger + Client *dynamodb.Client + Table string + Segments int + BatchSize int + Throttle time.Duration + MaxBackoff time.Duration // Maximum backoff on throttling errors (0 = no limit). + Logger *service.Logger } // SnapshotScanner performs a parallel scan of a DynamoDB table using the @@ -48,21 +46,21 @@ type SnapshotScannerConfig struct { // resumable checkpointing, adaptive backoff on throttling, and reports // progress through user-supplied callbacks. type SnapshotScanner struct { - client *dynamodb.Client - table string - segments int - batchSize int - throttle time.Duration - maxBackoff time.Duration - checkpointer *Checkpointer - checkpointInterval int // Checkpoint every N batches (0 = every batch) - log *service.Logger - - // Callbacks - onBatch func(ctx context.Context, items DynamoItems, segment int) error - onProgress func(segment, totalSegments int, recordsRead int64) - onCheckpointFailed func(segment int, err error) - onSegmentComplete func(segment int, duration time.Duration, recordsRead int64) + client *dynamodb.Client + table string + segments int + batchSize int + throttle time.Duration + maxBackoff time.Duration + log *service.Logger + + // Callbacks. Progress persistence is NOT the scanner's job: batches carry + // their scan resume position (lastKey) to the batch callback, and the + // consumer persists positions only after downstream acknowledgement. + onBatch func(ctx context.Context, items DynamoItems, segment int, lastKey map[string]dynamodbtypes.AttributeValue) error + onProgress func(segment, totalSegments int, recordsRead int64) + onSegmentSealed func(ctx context.Context, segment int) error + onSegmentComplete func(segment int, duration time.Duration, recordsRead int64) // State tracking activeSegments atomic.Int32 @@ -70,26 +68,21 @@ type SnapshotScanner struct { // NewSnapshotScanner creates a new snapshot scanner. func NewSnapshotScanner(conf SnapshotScannerConfig) *SnapshotScanner { - checkpointInterval := conf.CheckpointInterval - if checkpointInterval == 0 { - checkpointInterval = 10 // Default: checkpoint every 10 batches. - } - return &SnapshotScanner{ - client: conf.Client, - table: conf.Table, - segments: conf.Segments, - batchSize: conf.BatchSize, - throttle: conf.Throttle, - maxBackoff: conf.MaxBackoff, - checkpointer: conf.Checkpointer, - checkpointInterval: checkpointInterval, - log: conf.Logger, + client: conf.Client, + table: conf.Table, + segments: conf.Segments, + batchSize: conf.BatchSize, + throttle: conf.Throttle, + maxBackoff: conf.MaxBackoff, + log: conf.Logger, } } -// SetBatchCallback sets the callback for processing batches of items. -func (s *SnapshotScanner) SetBatchCallback(fn func(ctx context.Context, items DynamoItems, segment int) error) { +// SetBatchCallback sets the callback for processing batches of items. lastKey +// is the scan position after the batch (nil when the batch ends the segment); +// it is the only position safe to persist once the batch is acknowledged. +func (s *SnapshotScanner) SetBatchCallback(fn func(ctx context.Context, items DynamoItems, segment int, lastKey map[string]dynamodbtypes.AttributeValue) error) { s.onBatch = fn } @@ -98,9 +91,11 @@ func (s *SnapshotScanner) SetProgressCallback(fn func(segment, totalSegments int s.onProgress = fn } -// SetCheckpointFailedCallback sets the callback for checkpoint failures. -func (s *SnapshotScanner) SetCheckpointFailedCallback(fn func(segment int, err error)) { - s.onCheckpointFailed = fn +// SetSegmentSealedCallback sets the callback fired when a segment's scan has +// emitted its last batch, so the consumer can register the segment-complete +// marker behind all of the segment's in-flight batches. +func (s *SnapshotScanner) SetSegmentSealedCallback(fn func(ctx context.Context, segment int) error) { + s.onSegmentSealed = fn } // SetSegmentCompleteCallback sets the callback for segment completion with duration tracking. @@ -160,7 +155,6 @@ func (s *SnapshotScanner) scanSegment(ctx context.Context, segment int, startKey var ( lastEvaluatedKey = startKey recordsRead int64 - batchCount int throttleTicker = time.NewTicker(s.throttle) firstRequest = true ) @@ -215,27 +209,15 @@ func (s *SnapshotScanner) scanSegment(ctx context.Context, segment int, startKey if len(result.Items) == 0 { lastEvaluatedKey = result.LastEvaluatedKey if lastEvaluatedKey == nil { - return s.completeSegment(segment, startTime, recordsRead) + return s.completeSegment(ctx, segment, startTime, recordsRead) } continue } - if err := s.onBatch(ctx, result.Items, segment); err != nil { + if err := s.onBatch(ctx, result.Items, segment, result.LastEvaluatedKey); err != nil { return fmt.Errorf("processing batch for segment %d: %w", segment, err) } recordsRead += int64(len(result.Items)) - batchCount++ - - if s.shouldCheckpoint(batchCount, result.LastEvaluatedKey) { - if err := s.checkpointer.UpdateSnapshotProgress(ctx, segment, result.LastEvaluatedKey, recordsRead); err != nil { - s.log.Warnf("Failed to update checkpoint for segment %d: %v", segment, err) - if s.onCheckpointFailed != nil { - s.onCheckpointFailed(segment, err) - } - } else { - s.log.Debugf("Checkpointed segment %d at %d records (%d batches)", segment, recordsRead, batchCount) - } - } if s.onProgress != nil { s.onProgress(segment, s.segments, recordsRead) @@ -243,21 +225,19 @@ func (s *SnapshotScanner) scanSegment(ctx context.Context, segment int, startKey lastEvaluatedKey = result.LastEvaluatedKey if lastEvaluatedKey == nil { - return s.completeSegment(segment, startTime, recordsRead) + return s.completeSegment(ctx, segment, startTime, recordsRead) } } } -// shouldCheckpoint returns true when a checkpoint should be written. -func (s *SnapshotScanner) shouldCheckpoint(batchCount int, lastKey map[string]dynamodbtypes.AttributeValue) bool { - if s.checkpointer == nil || batchCount == 0 { - return false +// completeSegment seals the segment (registering its ack-gated completion +// marker), logs completion, and fires the metrics callback. +func (s *SnapshotScanner) completeSegment(ctx context.Context, segment int, startTime time.Time, recordsRead int64) error { + if s.onSegmentSealed != nil { + if err := s.onSegmentSealed(ctx, segment); err != nil { + return fmt.Errorf("sealing segment %d: %w", segment, err) + } } - return batchCount%s.checkpointInterval == 0 || lastKey == nil -} - -// completeSegment logs segment completion and fires the callback. -func (s *SnapshotScanner) completeSegment(segment int, startTime time.Time, recordsRead int64) error { duration := time.Since(startTime) s.log.Infof("Segment %d completed: %d records read in %v", segment, recordsRead, duration) if s.onSegmentComplete != nil { diff --git a/internal/impl/aws/dynamodb/snapshot_ack.go b/internal/impl/aws/dynamodb/snapshot_ack.go new file mode 100644 index 0000000000..b861b50f21 --- /dev/null +++ b/internal/impl/aws/dynamodb/snapshot_ack.go @@ -0,0 +1,145 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md + +package dynamodb + +import ( + "context" + "sync" + + "github.com/Jeffail/checkpoint" + dynamodbtypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + + "github.com/redpanda-data/benthos/v4/public/service" +) + +// snapshotProgressStore is the persistence surface used by the snapshot ack +// tracker; satisfied by *Checkpointer. +type snapshotProgressStore interface { + UpdateSnapshotProgress(ctx context.Context, segment int, lastKey map[string]dynamodbtypes.AttributeValue, recordsRead int64) error +} + +// segmentCheckpoint is the ordered-tracker payload for a snapshot segment: +// either a scan resume position (lastKey) or the segment-complete marker. +type segmentCheckpoint struct { + lastKey map[string]dynamodbtypes.AttributeValue + complete bool +} + +type segmentAckState struct { + tracker *checkpoint.Uncapped[segmentCheckpoint] + // frontier is the highest contiguous acked checkpoint. + frontier segmentCheckpoint + hasFrontier bool + // ackedRecords/ackedBatches accumulate acknowledged progress; persistence + // is throttled to every interval batches (plus completion). + ackedRecords int64 + ackedBatches int + persistedBatches int + persistedComplete bool +} + +// snapshotAckTracker gates snapshot progress persistence on downstream acks. +// Batches are tracked per segment in scan order; UpdateSnapshotProgress is +// only ever called with the highest *contiguous* acknowledged position, so a +// crash never skips unacked snapshot items on resume, and a segment is only +// marked complete once every one of its batches has been acknowledged. +type snapshotAckTracker struct { + store snapshotProgressStore + interval int + log *service.Logger + + mu sync.Mutex + segments map[int]*segmentAckState +} + +func newSnapshotAckTracker(store snapshotProgressStore, interval int, log *service.Logger) *snapshotAckTracker { + if interval <= 0 { + interval = 10 + } + return &snapshotAckTracker{ + store: store, + interval: interval, + log: log, + segments: make(map[int]*segmentAckState), + } +} + +func (t *snapshotAckTracker) segmentLocked(segment int) *segmentAckState { + st, ok := t.segments[segment] + if !ok { + st = &segmentAckState{tracker: checkpoint.NewUncapped[segmentCheckpoint]()} + t.segments[segment] = st + } + return st +} + +// TrackBatch registers a snapshot batch with its segment's ordered tracker and +// returns the resolve function for its ack. Must be called in scan order per +// segment (the segment's single scan goroutine). +func (t *snapshotAckTracker) TrackBatch(segment int, lastKey map[string]dynamodbtypes.AttributeValue, n int) func() *segmentCheckpoint { + t.mu.Lock() + defer t.mu.Unlock() + return t.segmentLocked(segment).tracker.Track(segmentCheckpoint{lastKey: lastKey}, int64(n)) +} + +// Ack marks a tracked batch as acknowledged and persists the segment's +// contiguous frontier once enough batches have been acked since the last +// persist (or immediately when the frontier reaches the completion marker). +func (t *snapshotAckTracker) Ack(ctx context.Context, segment, n int, resolve func() *segmentCheckpoint) error { + t.mu.Lock() + st, ok := t.segments[segment] + if !ok { + t.mu.Unlock() + return nil + } + if fr := resolve(); fr != nil { + st.frontier = *fr + st.hasFrontier = true + } + st.ackedRecords += int64(n) + st.ackedBatches++ + + var ( + toPersist *segmentCheckpoint + records int64 + ) + if st.hasFrontier { + completeDue := st.frontier.complete && !st.persistedComplete + intervalDue := st.frontier.lastKey != nil && st.ackedBatches-st.persistedBatches >= t.interval + if completeDue || intervalDue { + cp := st.frontier + toPersist = &cp + records = st.ackedRecords + st.persistedBatches = st.ackedBatches + if cp.complete { + st.persistedComplete = true + } + } + } + t.mu.Unlock() + + if toPersist == nil { + return nil + } + if toPersist.complete { + return t.store.UpdateSnapshotProgress(ctx, segment, nil, records) + } + return t.store.UpdateSnapshotProgress(ctx, segment, toPersist.lastKey, records) +} + +// SealSegment registers the segment-complete marker. Segments can end on an +// empty scan page, so completion cannot ride on a final batch; the marker +// resolves immediately and Complete=true persists as soon as every batch +// before it has been acknowledged (possibly inside a later Ack call). +func (t *snapshotAckTracker) SealSegment(ctx context.Context, segment int) error { + t.mu.Lock() + resolve := t.segmentLocked(segment).tracker.Track(segmentCheckpoint{complete: true}, 0) + t.mu.Unlock() + return t.Ack(ctx, segment, 0, resolve) +} diff --git a/internal/impl/aws/dynamodb/snapshot_ack_test.go b/internal/impl/aws/dynamodb/snapshot_ack_test.go new file mode 100644 index 0000000000..f494b98086 --- /dev/null +++ b/internal/impl/aws/dynamodb/snapshot_ack_test.go @@ -0,0 +1,147 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md + +package dynamodb + +import ( + "context" + "log/slog" + "sync" + "testing" + + dynamodbtypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + "github.com/stretchr/testify/require" + + "github.com/redpanda-data/benthos/v4/public/service" +) + +type recordedProgress struct { + segment int + lastKey map[string]dynamodbtypes.AttributeValue + recordsRead int64 +} + +type fakeProgressStore struct { + mu sync.Mutex + updates []recordedProgress +} + +func (f *fakeProgressStore) UpdateSnapshotProgress(_ context.Context, segment int, lastKey map[string]dynamodbtypes.AttributeValue, recordsRead int64) error { + f.mu.Lock() + defer f.mu.Unlock() + f.updates = append(f.updates, recordedProgress{segment: segment, lastKey: lastKey, recordsRead: recordsRead}) + return nil +} + +func (f *fakeProgressStore) recorded() []recordedProgress { + f.mu.Lock() + defer f.mu.Unlock() + return append([]recordedProgress(nil), f.updates...) +} + +func scanKey(v string) map[string]dynamodbtypes.AttributeValue { + return map[string]dynamodbtypes.AttributeValue{ + "pk": &dynamodbtypes.AttributeValueMemberS{Value: v}, + } +} + +func keyVal(k map[string]dynamodbtypes.AttributeValue) string { + if s, ok := k["pk"].(*dynamodbtypes.AttributeValueMemberS); ok { + return s.Value + } + return "" +} + +func newTestSnapshotAckTracker(interval int) (*snapshotAckTracker, *fakeProgressStore) { + store := &fakeProgressStore{} + return newSnapshotAckTracker(store, interval, service.NewLoggerFromSlog(slog.Default())), store +} + +func TestSnapshotAckTracker(t *testing.T) { + ctx := context.Background() + + t.Run("persists only acknowledged positions", func(t *testing.T) { + tracker, store := newTestSnapshotAckTracker(1) + + r1 := tracker.TrackBatch(0, scanKey("k1"), 5) + require.Empty(t, store.recorded(), "nothing may persist at read time") + + require.NoError(t, tracker.Ack(ctx, 0, 5, r1)) + got := store.recorded() + require.Len(t, got, 1) + require.Equal(t, "k1", keyVal(got[0].lastKey)) + require.Equal(t, int64(5), got[0].recordsRead) + }) + + t.Run("out-of-order acks never persist past an unacked batch", func(t *testing.T) { + tracker, store := newTestSnapshotAckTracker(1) + + r1 := tracker.TrackBatch(0, scanKey("k1"), 5) + r2 := tracker.TrackBatch(0, scanKey("k2"), 5) + + // Acking only the LATER batch must not persist k2: the items before + // it are not durable. (The pre-fix code persisted at read time - the + // data-loss window.) + require.NoError(t, tracker.Ack(ctx, 0, 5, r2)) + require.Empty(t, store.recorded(), "the frontier must not advance past a still-unacked earlier batch") + + require.NoError(t, tracker.Ack(ctx, 0, 5, r1)) + got := store.recorded() + require.Len(t, got, 1) + require.Equal(t, "k2", keyVal(got[0].lastKey), "acking the gap resolves the full prefix") + }) + + t.Run("interval throttles persistence", func(t *testing.T) { + tracker, store := newTestSnapshotAckTracker(3) + + var resolves []func() *segmentCheckpoint + for i := range 3 { + resolves = append(resolves, tracker.TrackBatch(0, scanKey(string(rune('a'+i))), 1)) + } + require.NoError(t, tracker.Ack(ctx, 0, 1, resolves[0])) + require.NoError(t, tracker.Ack(ctx, 0, 1, resolves[1])) + require.Empty(t, store.recorded(), "persistence is throttled to every interval batches") + require.NoError(t, tracker.Ack(ctx, 0, 1, resolves[2])) + require.Len(t, store.recorded(), 1) + }) + + t.Run("seal persists completion only after all batches ack", func(t *testing.T) { + tracker, store := newTestSnapshotAckTracker(1000) // interval never triggers + + r1 := tracker.TrackBatch(1, scanKey("k1"), 5) + require.NoError(t, tracker.SealSegment(ctx, 1)) + require.Empty(t, store.recorded(), "completion must wait for the segment's in-flight batches") + + require.NoError(t, tracker.Ack(ctx, 1, 5, r1)) + got := store.recorded() + require.Len(t, got, 1) + require.Nil(t, got[0].lastKey, "a nil lastKey marks the segment complete") + require.Equal(t, 1, got[0].segment) + require.Equal(t, int64(5), got[0].recordsRead) + }) + + t.Run("seal on an empty segment persists completion immediately", func(t *testing.T) { + tracker, store := newTestSnapshotAckTracker(1000) + + require.NoError(t, tracker.SealSegment(ctx, 2)) + got := store.recorded() + require.Len(t, got, 1) + require.Nil(t, got[0].lastKey) + }) + + t.Run("a never-acked batch pins the segment forever", func(t *testing.T) { + tracker, store := newTestSnapshotAckTracker(1) + + tracker.TrackBatch(3, scanKey("k1"), 5) // nacked: resolve never called + r2 := tracker.TrackBatch(3, scanKey("k2"), 5) + require.NoError(t, tracker.Ack(ctx, 3, 5, r2)) + require.NoError(t, tracker.SealSegment(ctx, 3)) + + require.Empty(t, store.recorded(), "neither progress nor completion may pass a nacked batch") + }) +} From 120371b3ae9ac599fcdd23e8ae02dce048209d11 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Mon, 10 Aug 2026 14:43:28 -0400 Subject: [PATCH 04/11] aws_dynamodb_cdc: adversarial crash test for the snapshot ack gate Receive the whole snapshot without acking, assert no checkpoint state is persisted (the pre-fix code wrote both segment progress and the completion marker at read time), then restart and assert every item is re-delivered. Also commits the regenerated docs for the new auto_replay_nacks field. --- .../pages/inputs/aws_dynamodb_cdc.adoc | 11 ++ .../dynamodb/input_cdc_integration_test.go | 120 ++++++++++++++++++ 2 files changed, 131 insertions(+) diff --git a/docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc b/docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc index 05e689dd36..361a8bc4d2 100644 --- a/docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc +++ b/docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc @@ -43,6 +43,7 @@ input: checkpoint_table: redpanda_dynamodb_checkpoints checkpoint_namespace: "" start_from: trim_horizon + auto_replay_nacks: true snapshot_mode: none ``` @@ -69,6 +70,7 @@ input: start_from: trim_horizon checkpoint_limit: 1000 max_tracked_shards: 10000 + auto_replay_nacks: true throttle_backoff: 100ms snapshot_mode: none snapshot_segments: 1 @@ -412,6 +414,15 @@ Maximum number of shards to track simultaneously. Prevents memory issues with ex *Default*: `10000` +=== `auto_replay_nacks` + +Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to `false` these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. + + +*Type*: `bool` + +*Default*: `true` + === `throttle_backoff` Time to wait when applying backpressure due to too many in-flight messages. diff --git a/internal/impl/aws/dynamodb/input_cdc_integration_test.go b/internal/impl/aws/dynamodb/input_cdc_integration_test.go index d7327d561b..76f95d003f 100644 --- a/internal/impl/aws/dynamodb/input_cdc_integration_test.go +++ b/internal/impl/aws/dynamodb/input_cdc_integration_test.go @@ -534,6 +534,126 @@ credentials: } // TestIntegrationDynamoDBSnapshot tests snapshot functionality. +// TestIntegrationDynamoDBSnapshotAckGate verifies that snapshot progress and +// completion are gated on downstream acks: a crash after the scan has read +// (and emitted) everything but before acknowledgement must leave no snapshot +// checkpoint state, so a restart re-delivers every item. See CON-504. +func TestIntegrationDynamoDBSnapshotAckGate(t *testing.T) { + integration.CheckSkip(t) + + ctx := context.Background() + + ctr, err := testcontainers.Run(ctx, + "amazon/dynamodb-local:latest", + testcontainers.WithExposedPorts("8000/tcp"), + testcontainers.WithWaitStrategy(wait.ForListeningPort("8000/tcp")), + ) + require.NoError(t, err) + t.Cleanup(func() { + if err := ctr.Terminate(context.Background()); err != nil { + t.Logf("failed to terminate dynamodb container: %v", err) + } + }) + + mappedPort, err := ctr.MappedPort(ctx, "8000/tcp") + require.NoError(t, err) + port := mappedPort.Port() + + var client *dynamodb.Client + tableName := "test-snapshot-ack-gate-table" + checkpointTable := "test-snapshot-ack-gate-checkpoint" + + require.Eventually(t, func() bool { + var cerr error + client, cerr = createTableWithStreams(ctx, t, port, tableName) + return cerr == nil + }, 60*time.Second, 500*time.Millisecond) + + const itemCount = 5 + for i := range itemCount { + require.NoError(t, putTestItem(ctx, client, tableName, fmt.Sprintf("gate-%d", i), fmt.Sprintf("value-%d", i))) + } + + confStr := fmt.Sprintf(` +tables: [%s] +checkpoint_table: %s +endpoint: http://localhost:%s +region: us-east-1 +snapshot_mode: snapshot_only +snapshot_segments: 1 +snapshot_batch_size: 10 +credentials: + id: xxxxx + secret: xxxxx + token: xxxxx +`, tableName, checkpointTable, port) + + countCheckpointRows := func() int { + out, err := client.Scan(ctx, &dynamodb.ScanInput{TableName: &checkpointTable}) + if err != nil { + return -1 // table may not exist yet + } + return len(out.Items) + } + + // Run 1: receive the snapshot batch but never acknowledge it, then + // simulate a crash. Nothing may be persisted. + { + spec := dynamoDBCDCInputConfig() + parsed, err := spec.ParseYAML(confStr, nil) + require.NoError(t, err) + input, err := newDynamoDBCDCInputFromConfig(parsed, service.MockResources()) + require.NoError(t, err) + require.NoError(t, input.Connect(ctx)) + + readCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + batch, _, err := input.ReadBatch(readCtx) + cancel() + require.NoError(t, err) + require.Len(t, batch, itemCount, "expected the whole snapshot in one batch") + + // Give the input time to (wrongly) persist progress or the completion + // marker - the pre-fix code did both at read time. + time.Sleep(3 * time.Second) + require.Zero(t, countCheckpointRows(), + "no snapshot checkpoint state may be persisted before the batch is acknowledged") + + // Simulated crash: abandon without acking. + _ = input.Close(ctx) + } + + // Run 2: restart against the same checkpoint table. Since nothing was + // acked, the snapshot must re-run and deliver every item again. + { + spec := dynamoDBCDCInputConfig() + parsed, err := spec.ParseYAML(confStr, nil) + require.NoError(t, err) + input, err := newDynamoDBCDCInputFromConfig(parsed, service.MockResources()) + require.NoError(t, err) + require.NoError(t, input.Connect(ctx)) + t.Cleanup(func() { _ = input.Close(ctx) }) + + seen := 0 + readCtx, cancel := context.WithTimeout(ctx, 60*time.Second) + defer cancel() + for seen < itemCount { + batch, ackFn, err := input.ReadBatch(readCtx) + if errors.Is(err, service.ErrEndOfInput) { + break + } + require.NoError(t, err) + seen += len(batch) + require.NoError(t, ackFn(ctx, nil)) + } + require.Equal(t, itemCount, seen, "the snapshot should have re-run and re-delivered every item after the crash") + + // With everything acked, completion must now persist. + require.Eventually(t, func() bool { + return countCheckpointRows() > 0 + }, 30*time.Second, 500*time.Millisecond, "snapshot completion was never persisted after a fully-acked run") + } +} + func TestIntegrationDynamoDBSnapshot(t *testing.T) { integration.CheckSkip(t) From 76385acd578e399024654836ea1fc16a8538e66f Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Mon, 10 Aug 2026 16:16:10 -0400 Subject: [PATCH 05/11] aws_dynamodb_cdc: address review - retryable snapshot persists, graceful-stop gate, expired-iterator positioning - Snapshot checkpoint bookkeeping was committed before the store write and never rolled back: a transient DynamoDB error on the completion write left the segment's durable row stale forever while the completion gate proceeded. Writes now happen first and bookkeeping commits only on success, so the next ack retries a failed write. - The completion gate treated graceful-shutdown cancellation as a snapshot failure (ERROR log + failed state surfaced to the framework on every mid-snapshot stop). Cancellation is now a debug-level normal shutdown. - resolveResumeIterator could re-acquire LATEST for a shard whose expired iterator had positioned it at TRIM_HORIZON (or at an older LATEST), silently skipping everything published since - the exact hole the start_from scoping closed at discovery time. Expired-iterator recovery never uses LATEST now. - The discovery-time decision is extracted into initialIteratorType with a table test locking in the contract, and the snapshot persist interval is a named constant. --- internal/impl/aws/dynamodb/input_cdc.go | 56 ++++++++++--------- .../input_cdc_expired_iterator_test.go | 52 ++++++++++++----- internal/impl/aws/dynamodb/snapshot_ack.go | 31 +++++++--- .../impl/aws/dynamodb/snapshot_ack_test.go | 35 +++++++++++- 4 files changed, 125 insertions(+), 49 deletions(-) diff --git a/internal/impl/aws/dynamodb/input_cdc.go b/internal/impl/aws/dynamodb/input_cdc.go index f4c42dd7f5..6947242975 100644 --- a/internal/impl/aws/dynamodb/input_cdc.go +++ b/internal/impl/aws/dynamodb/input_cdc.go @@ -1392,7 +1392,7 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st // tracker below persists a segment's position only once every batch at or // below it has been acknowledged downstream, and the completion gate is // reset per connection attempt. - d.snapshot.ackTracker = newSnapshotAckTracker(d.checkpointer, 10 /* persist every 10 acked batches */, d.log) + d.snapshot.ackTracker = newSnapshotAckTracker(d.checkpointer, defaultSnapshotCheckpointBatchInterval, d.log) d.snapshot.resetAckGate() d.snapshot.scanner = NewSnapshotScanner(SnapshotScannerConfig{ Client: d.dynamoClient, @@ -1463,6 +1463,13 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st // any earlier would let a crash (or a terminal nack) skip un-acked // items on restart. Blocks until acks drain or soft-stop. if err := d.snapshot.waitAcks(scanCtx); err != nil { + if errors.Is(err, context.Canceled) { + // Graceful shutdown mid-snapshot: nothing is wrong, the + // un-acked items simply resume from acknowledged progress on + // the next run. + d.log.Debug("Snapshot completion gate interrupted by shutdown") + return + } wrappedErr := fmt.Errorf("snapshot completion gate for table %s: %w", tableName, err) d.log.Errorf("%v (the snapshot will resume from acknowledged progress on restart)", wrappedErr) d.snapshot.errOnce.Do(func() { @@ -1678,16 +1685,7 @@ func (d *dynamoDBCDCInput) refreshShards(ctx context.Context) error { cutoff = decision.Cutoff d.log.Infof("Failover resume for shard %s from trim horizon, skipping records at/before %s", shardID, cutoff.Format(time.RFC3339)) default: - // start_from: latest is honored only on the first discovery of a - // fresh pipeline. Checkpoint-less shards found on later refresh - // cycles (or after a restart with existing state) are stream - // rotation children: starting them at LATEST would silently skip - // their backlog. - if d.conf.startFrom == "latest" && d.honorStartFrom.Load() { - iteratorType = types.ShardIteratorTypeLatest - } else { - iteratorType = types.ShardIteratorTypeTrimHorizon - } + iteratorType = initialIteratorType(d.conf.startFrom, d.honorStartFrom.Load()) d.log.Infof("Starting shard %s from %s", shardID, iteratorType) } @@ -1988,16 +1986,7 @@ func (d *dynamoDBCDCInput) refreshTableShards(ctx context.Context, tableName str cutoff = decision.Cutoff d.log.Infof("Failover resume for shard %s (table %s) from trim horizon, skipping records at/before %s", shardID, tableName, cutoff.Format(time.RFC3339)) default: - // start_from: latest is honored only on the first discovery of a - // fresh pipeline. Checkpoint-less shards found on later refresh - // cycles (or after a restart with existing state) are stream - // rotation children: starting them at LATEST would silently skip - // their backlog. - if d.conf.startFrom == "latest" && ts.honorStartFrom.Load() { - iteratorType = types.ShardIteratorTypeLatest - } else { - iteratorType = types.ShardIteratorTypeTrimHorizon - } + iteratorType = initialIteratorType(d.conf.startFrom, ts.honorStartFrom.Load()) d.log.Infof("Starting shard %s (table %s) from %s", shardID, tableName, iteratorType) } @@ -2082,20 +2071,35 @@ func lastRecordSequenceNumber(records []types.Record) string { return "" } +// initialIteratorType decides where a shard with no checkpoint state starts. +// start_from is honored only while honorStartFrom is true — the first shard +// discovery of a pipeline with no prior checkpoint state. Shards discovered +// on later refresh cycles (or after a restart with existing state) are stream +// rotation children: starting them at LATEST would silently skip their +// backlog. +func initialIteratorType(startFrom string, honorStartFrom bool) types.ShardIteratorType { + if startFrom == "latest" && honorStartFrom { + return types.ShardIteratorTypeLatest + } + return types.ShardIteratorTypeTrimHorizon +} + // resolveResumeIterator decides which iterator type and sequence number to use // when re-acquiring an iterator after the previous one expired. It prefers the // last sequence number actually read from the shard (resuming exactly where a // healthy iterator would be, so no records are skipped or re-read beyond the // pipeline's normal at-least-once guarantee), then the persisted checkpoint, -// and finally the configured start position when nothing has been read yet. -func resolveResumeIterator(lastSeq, checkpoint, startFrom string) (types.ShardIteratorType, *string) { +// and otherwise the trim horizon. LATEST is never used here: the shard was +// already positioned when its previous iterator was acquired, so re-acquiring +// LATEST would silently skip everything published since — including the whole +// backlog of a TRIM_HORIZON-positioned rotation child that expired before its +// first read. +func resolveResumeIterator(lastSeq, checkpoint string) (types.ShardIteratorType, *string) { switch { case lastSeq != "": return types.ShardIteratorTypeAfterSequenceNumber, &lastSeq case checkpoint != "": return types.ShardIteratorTypeAfterSequenceNumber, &checkpoint - case startFrom == "latest": - return types.ShardIteratorTypeLatest, nil default: return types.ShardIteratorTypeTrimHorizon, nil } @@ -2117,7 +2121,7 @@ func (d *dynamoDBCDCInput) refreshExpiredIterator(ctx context.Context, cp *Check } } - iteratorType, sequenceNumber := resolveResumeIterator(lastSeq, checkpoint, d.conf.startFrom) + iteratorType, sequenceNumber := resolveResumeIterator(lastSeq, checkpoint) iter, err := d.streamsClient.GetShardIterator(ctx, &dynamodbstreams.GetShardIteratorInput{ StreamArn: &streamArn, diff --git a/internal/impl/aws/dynamodb/input_cdc_expired_iterator_test.go b/internal/impl/aws/dynamodb/input_cdc_expired_iterator_test.go index 92919bbc1a..b9f6e83e88 100644 --- a/internal/impl/aws/dynamodb/input_cdc_expired_iterator_test.go +++ b/internal/impl/aws/dynamodb/input_cdc_expired_iterator_test.go @@ -29,7 +29,6 @@ func TestResolveResumeIterator(t *testing.T) { name string lastSeq string checkpoint string - startFrom string wantType types.ShardIteratorType wantSeq *string }{ @@ -37,7 +36,6 @@ func TestResolveResumeIterator(t *testing.T) { name: "prefers last read sequence over checkpoint", lastSeq: "100", checkpoint: "50", - startFrom: "latest", wantType: types.ShardIteratorTypeAfterSequenceNumber, wantSeq: aws.String("100"), }, @@ -45,29 +43,30 @@ func TestResolveResumeIterator(t *testing.T) { name: "falls back to checkpoint when nothing read", lastSeq: "", checkpoint: "50", - startFrom: "latest", wantType: types.ShardIteratorTypeAfterSequenceNumber, wantSeq: aws.String("50"), }, { - name: "falls back to latest when no sequence available", - lastSeq: "", - startFrom: "latest", - wantType: types.ShardIteratorTypeLatest, - wantSeq: nil, + name: "falls back to trim horizon when no sequence available", + lastSeq: "", + wantType: types.ShardIteratorTypeTrimHorizon, + wantSeq: nil, }, { - name: "falls back to trim horizon when no sequence available", - lastSeq: "", - startFrom: "trim_horizon", - wantType: types.ShardIteratorTypeTrimHorizon, - wantSeq: nil, + // LATEST must never be re-acquired: the shard was already + // positioned when the expired iterator was obtained, so LATEST + // would silently skip everything published since. + name: "never re-acquires latest", + lastSeq: "", + checkpoint: "", + wantType: types.ShardIteratorTypeTrimHorizon, + wantSeq: nil, }, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - gotType, gotSeq := resolveResumeIterator(tc.lastSeq, tc.checkpoint, tc.startFrom) + gotType, gotSeq := resolveResumeIterator(tc.lastSeq, tc.checkpoint) assert.Equal(t, tc.wantType, gotType) if tc.wantSeq == nil { assert.Nil(t, gotSeq) @@ -77,3 +76,28 @@ func TestResolveResumeIterator(t *testing.T) { }) } } + +// TestInitialIteratorType locks in the start_from contract: latest applies +// only to the first discovery of a genuinely fresh pipeline; every other +// checkpoint-less shard (rotation children found on refresh cycles, or any +// shard after a restart with existing state) starts at TRIM_HORIZON so its +// backlog is never silently skipped. +func TestInitialIteratorType(t *testing.T) { + cases := []struct { + name string + startFrom string + honor bool + want types.ShardIteratorType + }{ + {"fresh pipeline honors latest", "latest", true, types.ShardIteratorTypeLatest}, + {"fresh pipeline honors trim_horizon", "trim_horizon", true, types.ShardIteratorTypeTrimHorizon}, + {"rotation child ignores latest", "latest", false, types.ShardIteratorTypeTrimHorizon}, + {"restart with state ignores latest", "latest", false, types.ShardIteratorTypeTrimHorizon}, + {"trim_horizon unaffected by honor flag", "trim_horizon", false, types.ShardIteratorTypeTrimHorizon}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, initialIteratorType(tc.startFrom, tc.honor)) + }) + } +} diff --git a/internal/impl/aws/dynamodb/snapshot_ack.go b/internal/impl/aws/dynamodb/snapshot_ack.go index b861b50f21..cb36f5c0f9 100644 --- a/internal/impl/aws/dynamodb/snapshot_ack.go +++ b/internal/impl/aws/dynamodb/snapshot_ack.go @@ -18,6 +18,11 @@ import ( "github.com/redpanda-data/benthos/v4/public/service" ) +// defaultSnapshotCheckpointBatchInterval is how many acknowledged batches a +// segment accumulates between checkpoint-store writes (bounding write volume +// the same way the pre-ack-gating scanner checkpointed every 10 batches). +const defaultSnapshotCheckpointBatchInterval = 10 + // snapshotProgressStore is the persistence surface used by the snapshot ack // tracker; satisfied by *Checkpointer. type snapshotProgressStore interface { @@ -60,7 +65,7 @@ type snapshotAckTracker struct { func newSnapshotAckTracker(store snapshotProgressStore, interval int, log *service.Logger) *snapshotAckTracker { if interval <= 0 { - interval = 10 + interval = defaultSnapshotCheckpointBatchInterval } return &snapshotAckTracker{ store: store, @@ -116,10 +121,6 @@ func (t *snapshotAckTracker) Ack(ctx context.Context, segment, n int, resolve fu cp := st.frontier toPersist = &cp records = st.ackedRecords - st.persistedBatches = st.ackedBatches - if cp.complete { - st.persistedComplete = true - } } } t.mu.Unlock() @@ -127,10 +128,26 @@ func (t *snapshotAckTracker) Ack(ctx context.Context, segment, n int, resolve fu if toPersist == nil { return nil } + var persistErr error if toPersist.complete { - return t.store.UpdateSnapshotProgress(ctx, segment, nil, records) + persistErr = t.store.UpdateSnapshotProgress(ctx, segment, nil, records) + } else { + persistErr = t.store.UpdateSnapshotProgress(ctx, segment, toPersist.lastKey, records) + } + if persistErr != nil { + // Bookkeeping is deliberately untouched: the next ack (or seal) for + // this segment retries the write instead of silently treating the + // failed position as durable. + return persistErr } - return t.store.UpdateSnapshotProgress(ctx, segment, toPersist.lastKey, records) + + t.mu.Lock() + st.persistedBatches = st.ackedBatches + if toPersist.complete { + st.persistedComplete = true + } + t.mu.Unlock() + return nil } // SealSegment registers the segment-complete marker. Segments can end on an diff --git a/internal/impl/aws/dynamodb/snapshot_ack_test.go b/internal/impl/aws/dynamodb/snapshot_ack_test.go index f494b98086..527b840570 100644 --- a/internal/impl/aws/dynamodb/snapshot_ack_test.go +++ b/internal/impl/aws/dynamodb/snapshot_ack_test.go @@ -10,6 +10,7 @@ package dynamodb import ( "context" + "errors" "log/slog" "sync" "testing" @@ -27,13 +28,19 @@ type recordedProgress struct { } type fakeProgressStore struct { - mu sync.Mutex - updates []recordedProgress + mu sync.Mutex + updates []recordedProgress + failNext error } func (f *fakeProgressStore) UpdateSnapshotProgress(_ context.Context, segment int, lastKey map[string]dynamodbtypes.AttributeValue, recordsRead int64) error { f.mu.Lock() defer f.mu.Unlock() + if f.failNext != nil { + err := f.failNext + f.failNext = nil + return err + } f.updates = append(f.updates, recordedProgress{segment: segment, lastKey: lastKey, recordsRead: recordsRead}) return nil } @@ -44,6 +51,8 @@ func (f *fakeProgressStore) recorded() []recordedProgress { return append([]recordedProgress(nil), f.updates...) } +var errStoreDown = errors.New("store down") + func scanKey(v string) map[string]dynamodbtypes.AttributeValue { return map[string]dynamodbtypes.AttributeValue{ "pk": &dynamodbtypes.AttributeValueMemberS{Value: v}, @@ -134,6 +143,28 @@ func TestSnapshotAckTracker(t *testing.T) { require.Nil(t, got[0].lastKey) }) + t.Run("a failed store write is retried by the next ack", func(t *testing.T) { + tracker, store := newTestSnapshotAckTracker(1) + + r1 := tracker.TrackBatch(4, scanKey("k1"), 5) + r2 := tracker.TrackBatch(4, scanKey("k2"), 5) + + store.mu.Lock() + store.failNext = errStoreDown + store.mu.Unlock() + + // The first ack's write fails; bookkeeping must not treat the failed + // position as durable. + require.Error(t, tracker.Ack(context.Background(), 4, 5, r1)) + require.Empty(t, store.recorded()) + + // The next ack retries and persists. + require.NoError(t, tracker.Ack(context.Background(), 4, 5, r2)) + got := store.recorded() + require.Len(t, got, 1) + require.Equal(t, "k2", keyVal(got[0].lastKey)) + }) + t.Run("a never-acked batch pins the segment forever", func(t *testing.T) { tracker, store := newTestSnapshotAckTracker(1) From 5765a61a39df8794f1af04e2a4da738075ff7683 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Tue, 11 Aug 2026 10:13:27 -0400 Subject: [PATCH 06/11] aws_dynamodb_cdc: nacks resolve snapshot checkpoints (auto_replay_nacks off is an opt-in drop) Unwinds the nack-recording completion-gate changes from the review rounds. Per the framework's documented contract for auto_replay_nacks ("If set to false these messages will instead be deleted"), disabling replay is an explicit opt-in to drop rejected messages, so a nacked snapshot batch now resolves its tracker slot like an ack and the segment checkpoint advances past it; failing the completion gate on nack produced permanent backpressure and blocked the snapshot from ever completing. The crash window is still guarded: the completion gate waits for every in-flight batch to settle before the snapshot is marked complete, and segment positions persist only behind ordered acks (write-then-commit). Full integration suite re-verified green. --- internal/impl/aws/dynamodb/input_cdc.go | 63 ++++--------------- .../impl/aws/dynamodb/snapshot_ack_test.go | 4 +- 2 files changed, 13 insertions(+), 54 deletions(-) diff --git a/internal/impl/aws/dynamodb/input_cdc.go b/internal/impl/aws/dynamodb/input_cdc.go index 6947242975..63984d1a4d 100644 --- a/internal/impl/aws/dynamodb/input_cdc.go +++ b/internal/impl/aws/dynamodb/input_cdc.go @@ -503,42 +503,12 @@ type snapshotState struct { // ackTracker gates snapshot progress persistence on downstream acks. ackTracker *snapshotAckTracker // ackWG counts emitted snapshot batches that have not yet been acked or - // nacked; the snapshot is only marked complete once it drains cleanly. + // nacked; the snapshot is only marked complete once it drains. ackWG sync.WaitGroup - // nackErr records the first snapshot batch nack. auto_replay_nacks is - // user-toggleable, so a nack can be terminal: the completion gate must - // fail rather than mark the snapshot complete over undelivered items. - // Cleared per connection attempt (the state outlives reconnects). - nackMu sync.Mutex - nackErr error -} - -func (s *snapshotState) recordNack(err error) { - s.nackMu.Lock() - defer s.nackMu.Unlock() - if s.nackErr == nil { - s.nackErr = err - } -} - -func (s *snapshotState) nackError() error { - s.nackMu.Lock() - defer s.nackMu.Unlock() - return s.nackErr -} - -// resetAckGate clears any nack recorded by a previous snapshot attempt so the -// completion gate judges only the current run. The WaitGroup is deliberately -// left untouched — batches from a previous attempt that are still in flight -// can yet be acked or nacked, and both must keep counting. -func (s *snapshotState) resetAckGate() { - s.nackMu.Lock() - defer s.nackMu.Unlock() - s.nackErr = nil } // waitAcks blocks until every emitted snapshot batch has been acked or -// nacked, or ctx is cancelled; a recorded nack fails the gate. +// nacked, or ctx is cancelled. func (s *snapshotState) waitAcks(ctx context.Context) error { drained := make(chan struct{}) go func() { @@ -548,9 +518,6 @@ func (s *snapshotState) waitAcks(ctx context.Context) error { }() select { case <-drained: - if err := s.nackError(); err != nil { - return fmt.Errorf("snapshot batch was rejected downstream: %w", err) - } return nil case <-ctx.Done(): return ctx.Err() @@ -1390,10 +1357,8 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st // Initialize snapshot scanner. Progress persistence is ack-gated: the // tracker below persists a segment's position only once every batch at or - // below it has been acknowledged downstream, and the completion gate is - // reset per connection attempt. + // below it has been acknowledged downstream. d.snapshot.ackTracker = newSnapshotAckTracker(d.checkpointer, defaultSnapshotCheckpointBatchInterval, d.log) - d.snapshot.resetAckGate() d.snapshot.scanner = NewSnapshotScanner(SnapshotScannerConfig{ Client: d.dynamoClient, Table: tableName, @@ -1459,9 +1424,9 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st } // The scan has read everything, but the snapshot is only complete once - // every emitted batch is acknowledged downstream: marking it complete - // any earlier would let a crash (or a terminal nack) skip un-acked - // items on restart. Blocks until acks drain or soft-stop. + // every emitted batch has settled downstream: marking it complete any + // earlier would let a crash skip un-acked items on restart. Blocks + // until acks drain or soft-stop. if err := d.snapshot.waitAcks(scanCtx); err != nil { if errors.Is(err, context.Canceled) { // Graceful shutdown mid-snapshot: nothing is wrong, the @@ -2857,7 +2822,7 @@ func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[ d.pendingAcks.Add(1) snapshot.ackWG.Add(1) - ackFunc := func(ackCtx context.Context, err error) error { + ackFunc := func(ackCtx context.Context, _ error) error { defer d.pendingAcks.Done() defer snapshot.ackWG.Done() @@ -2866,16 +2831,10 @@ func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[ return nil } - if err != nil { - // auto_replay_nacks is user-toggleable, so a nack can be terminal. - // Never resolve: the segment's persisted position stays pinned - // before this batch, and the completion gate fails so the snapshot - // is not marked complete over undelivered items. - snapshot.recordNack(err) - d.log.Errorf("Snapshot batch rejected downstream (segment %d): the segment's checkpoint is pinned before this batch and the snapshot will not be marked complete, unless the batch is redelivered (auto_replay_nacks) or the pipeline restarts: %v", segment, err) - return err - } - + // The ack error is deliberately ignored: nacks are replayed by + // auto_replay_nacks (the default), and disabling that is a documented + // opt-in to DROP rejected messages, so the segment's checkpoint must + // advance past them rather than pin the tracker. if ackErr := tracker.Ack(ackCtx, segment, len(batch), resolve); ackErr != nil { d.metrics.checkpointFailures.Incr(1) d.log.Errorf("Failed to checkpoint snapshot segment %d after ack: %v", segment, ackErr) diff --git a/internal/impl/aws/dynamodb/snapshot_ack_test.go b/internal/impl/aws/dynamodb/snapshot_ack_test.go index 527b840570..8c1b66fbc4 100644 --- a/internal/impl/aws/dynamodb/snapshot_ack_test.go +++ b/internal/impl/aws/dynamodb/snapshot_ack_test.go @@ -168,11 +168,11 @@ func TestSnapshotAckTracker(t *testing.T) { t.Run("a never-acked batch pins the segment forever", func(t *testing.T) { tracker, store := newTestSnapshotAckTracker(1) - tracker.TrackBatch(3, scanKey("k1"), 5) // nacked: resolve never called + tracker.TrackBatch(3, scanKey("k1"), 5) // still in flight: resolve never called r2 := tracker.TrackBatch(3, scanKey("k2"), 5) require.NoError(t, tracker.Ack(ctx, 3, 5, r2)) require.NoError(t, tracker.SealSegment(ctx, 3)) - require.Empty(t, store.recorded(), "neither progress nor completion may pass a nacked batch") + require.Empty(t, store.recorded(), "neither progress nor completion may pass an in-flight batch") }) } From e7881f2af9d53a782ab875bee06e3e4caacb2101 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Fri, 14 Aug 2026 11:30:06 -0400 Subject: [PATCH 07/11] aws_dynamodb_cdc: serialize segment persists and scope the completion gate per attempt Two review findings on the snapshot ack path: - snapshotAckTracker.Ack computed the persist under the tracker mutex but issued the store write outside it, so concurrent acks for one segment could land PutItems out of order - a stale write could regress the durable position or overwrite Complete=true. Each segment now serializes resolve+compute+write+commit behind a per-segment mutex (matching the CDC RecordBatcher's discipline), which also stops the commit from crediting acks the write did not cover. New concurrency test proven red against the unfixed code. - The snapshot completion gate was a WaitGroup on the input-lifetime snapshotState while Connect rebuilds msgChan per attempt: batches orphaned in a replaced channel never ran their ack fn, so a re-run's completion gate could wait forever. The gate is now allocated per connection attempt and captured by the scanner callbacks, so it only ever counts batches its own attempt emitted. Also reattaches TestIntegrationDynamoDBSnapshot's doc comment to its function. Full integration suite re-verified green. --- internal/impl/aws/dynamodb/input_cdc.go | 29 +++++++------- .../dynamodb/input_cdc_integration_test.go | 2 +- internal/impl/aws/dynamodb/snapshot_ack.go | 21 ++++++++-- .../impl/aws/dynamodb/snapshot_ack_test.go | 39 +++++++++++++++++++ 4 files changed, 74 insertions(+), 17 deletions(-) diff --git a/internal/impl/aws/dynamodb/input_cdc.go b/internal/impl/aws/dynamodb/input_cdc.go index 63984d1a4d..fee6bb8f9d 100644 --- a/internal/impl/aws/dynamodb/input_cdc.go +++ b/internal/impl/aws/dynamodb/input_cdc.go @@ -502,18 +502,17 @@ type snapshotState struct { // ackTracker gates snapshot progress persistence on downstream acks. ackTracker *snapshotAckTracker - // ackWG counts emitted snapshot batches that have not yet been acked or - // nacked; the snapshot is only marked complete once it drains. - ackWG sync.WaitGroup } -// waitAcks blocks until every emitted snapshot batch has been acked or -// nacked, or ctx is cancelled. -func (s *snapshotState) waitAcks(ctx context.Context) error { +// waitAckGate blocks until every batch counted on gate has been acked or +// nacked, or ctx is cancelled. The gate is per connection attempt: batches a +// previous attempt left buffered in a replaced msgChan settle (or leak) on +// their own attempt's gate and can never wedge the current one. +func waitAckGate(ctx context.Context, gate *sync.WaitGroup) error { drained := make(chan struct{}) go func() { // May outlive this call if ctx fires first; bounded by process lifetime. - s.ackWG.Wait() + gate.Wait() close(drained) }() select { @@ -1359,6 +1358,10 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st // tracker below persists a segment's position only once every batch at or // below it has been acknowledged downstream. d.snapshot.ackTracker = newSnapshotAckTracker(d.checkpointer, defaultSnapshotCheckpointBatchInterval, d.log) + // The completion gate is scoped to this connection attempt: it counts only + // batches this attempt emits, so batches orphaned in a previous attempt's + // msgChan cannot leave it permanently un-drainable. + ackGate := &sync.WaitGroup{} d.snapshot.scanner = NewSnapshotScanner(SnapshotScannerConfig{ Client: d.dynamoClient, Table: tableName, @@ -1370,7 +1373,7 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st // Set batch callback to send snapshot records to msgChan d.snapshot.scanner.SetBatchCallback(func(ctx context.Context, items []map[string]dynamodbtypes.AttributeValue, segment int, lastKey map[string]dynamodbtypes.AttributeValue) error { - return d.handleSnapshotBatch(ctx, items, segment, tableName, lastKey) + return d.handleSnapshotBatch(ctx, items, segment, tableName, lastKey, ackGate) }) // Set progress callback to update metrics @@ -1427,7 +1430,7 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st // every emitted batch has settled downstream: marking it complete any // earlier would let a crash skip un-acked items on restart. Blocks // until acks drain or soft-stop. - if err := d.snapshot.waitAcks(scanCtx); err != nil { + if err := waitAckGate(scanCtx, ackGate); err != nil { if errors.Is(err, context.Canceled) { // Graceful shutdown mid-snapshot: nothing is wrong, the // un-acked items simply resume from acknowledged progress on @@ -2756,7 +2759,7 @@ func (d *dynamoDBCDCInput) startShardReader(ctx context.Context, shardID string) // handleSnapshotBatch processes a batch of items from the snapshot scan. // lastKey is the scan position after this batch; it is registered with the // segment's ordered ack tracker and persisted only once acknowledged. -func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[string]dynamodbtypes.AttributeValue, segment int, tableName string, lastKey map[string]dynamodbtypes.AttributeValue) error { +func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[string]dynamodbtypes.AttributeValue, segment int, tableName string, lastKey map[string]dynamodbtypes.AttributeValue, ackGate *sync.WaitGroup) error { if len(items) == 0 { return nil } @@ -2820,11 +2823,11 @@ func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[ // Track pending acks: the global gauge and the snapshot completion gate. d.pendingAcks.Add(1) - snapshot.ackWG.Add(1) + ackGate.Add(1) ackFunc := func(ackCtx context.Context, _ error) error { defer d.pendingAcks.Done() - defer snapshot.ackWG.Done() + defer ackGate.Done() if d.closed.Load() { d.log.Debug("Received snapshot ack after close, dropping") @@ -2847,7 +2850,7 @@ func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[ select { case <-ctx.Done(): d.pendingAcks.Done() // Undo the Add(1) above - snapshot.ackWG.Done() + ackGate.Done() return ctx.Err() case d.msgChan <- asyncMessage{msg: batch, ackFn: ackFunc}: d.log.Debugf("Sent snapshot batch of %d records from segment %d", len(batch), segment) diff --git a/internal/impl/aws/dynamodb/input_cdc_integration_test.go b/internal/impl/aws/dynamodb/input_cdc_integration_test.go index 76f95d003f..b019b6304e 100644 --- a/internal/impl/aws/dynamodb/input_cdc_integration_test.go +++ b/internal/impl/aws/dynamodb/input_cdc_integration_test.go @@ -533,7 +533,6 @@ credentials: assert.NotEmpty(t, batch2, "Should read new events after resumption") } -// TestIntegrationDynamoDBSnapshot tests snapshot functionality. // TestIntegrationDynamoDBSnapshotAckGate verifies that snapshot progress and // completion are gated on downstream acks: a crash after the scan has read // (and emitted) everything but before acknowledgement must leave no snapshot @@ -654,6 +653,7 @@ credentials: } } +// TestIntegrationDynamoDBSnapshot tests snapshot functionality. func TestIntegrationDynamoDBSnapshot(t *testing.T) { integration.CheckSkip(t) diff --git a/internal/impl/aws/dynamodb/snapshot_ack.go b/internal/impl/aws/dynamodb/snapshot_ack.go index cb36f5c0f9..cdcd556e3a 100644 --- a/internal/impl/aws/dynamodb/snapshot_ack.go +++ b/internal/impl/aws/dynamodb/snapshot_ack.go @@ -37,7 +37,13 @@ type segmentCheckpoint struct { } type segmentAckState struct { - tracker *checkpoint.Uncapped[segmentCheckpoint] + // persistMu serializes the whole resolve+compute+write+commit sequence + // for one segment. Acks for a segment's batches arrive on concurrent + // pipeline goroutines: without this, two acks can both trip the persist + // check and race their store writes, landing an older position (or a + // Complete=false row) over a newer one. Always acquired before t.mu. + persistMu sync.Mutex + tracker *checkpoint.Uncapped[segmentCheckpoint] // frontier is the highest contiguous acked checkpoint. frontier segmentCheckpoint hasFrontier bool @@ -99,10 +105,18 @@ func (t *snapshotAckTracker) TrackBatch(segment int, lastKey map[string]dynamodb func (t *snapshotAckTracker) Ack(ctx context.Context, segment, n int, resolve func() *segmentCheckpoint) error { t.mu.Lock() st, ok := t.segments[segment] + t.mu.Unlock() if !ok { - t.mu.Unlock() return nil } + + // One ack at a time per segment, held across the store write: each + // persist is computed after the previous write finished, so the durable + // row can only move forward, and the interval check cannot double-fire. + st.persistMu.Lock() + defer st.persistMu.Unlock() + + t.mu.Lock() if fr := resolve(); fr != nil { st.frontier = *fr st.hasFrontier = true @@ -123,6 +137,7 @@ func (t *snapshotAckTracker) Ack(ctx context.Context, segment, n int, resolve fu records = st.ackedRecords } } + ackedAtCompute := st.ackedBatches t.mu.Unlock() if toPersist == nil { @@ -142,7 +157,7 @@ func (t *snapshotAckTracker) Ack(ctx context.Context, segment, n int, resolve fu } t.mu.Lock() - st.persistedBatches = st.ackedBatches + st.persistedBatches = ackedAtCompute if toPersist.complete { st.persistedComplete = true } diff --git a/internal/impl/aws/dynamodb/snapshot_ack_test.go b/internal/impl/aws/dynamodb/snapshot_ack_test.go index 8c1b66fbc4..931559f6a5 100644 --- a/internal/impl/aws/dynamodb/snapshot_ack_test.go +++ b/internal/impl/aws/dynamodb/snapshot_ack_test.go @@ -11,6 +11,7 @@ package dynamodb import ( "context" "errors" + "fmt" "log/slog" "sync" "testing" @@ -165,6 +166,44 @@ func TestSnapshotAckTracker(t *testing.T) { require.Equal(t, "k2", keyVal(got[0].lastKey)) }) + t.Run("concurrent acks never persist positions out of order", func(t *testing.T) { + // Acks arrive on concurrent pipeline goroutines. Each persist must be + // computed after the previous store write finished, otherwise a stale + // in-flight write can land over a newer position (or overwrite the + // Complete=true row). interval 1 makes every ack a persist candidate. + tracker, store := newTestSnapshotAckTracker(1) + + const batches = 200 + resolves := make([]func() *segmentCheckpoint, batches) + keys := make([]string, batches) + for i := range batches { + keys[i] = fmt.Sprintf("k%06d", i) + resolves[i] = tracker.TrackBatch(7, scanKey(keys[i]), 1) + } + + var wg sync.WaitGroup + for i := range batches { + wg.Go(func() { + require.NoError(t, tracker.Ack(ctx, 7, 1, resolves[i])) + }) + } + wg.Wait() + require.NoError(t, tracker.SealSegment(ctx, 7)) + + got := store.recorded() + require.NotEmpty(t, got) + prev := "" + for i, u := range got { + if u.lastKey == nil { + require.Equal(t, len(got)-1, i, "Complete=true must be the final write, nothing may land after it") + continue + } + k := keyVal(u.lastKey) + require.GreaterOrEqual(t, k, prev, "persisted position regressed at write %d: %q after %q", i, k, prev) + prev = k + } + }) + t.Run("a never-acked batch pins the segment forever", func(t *testing.T) { tracker, store := newTestSnapshotAckTracker(1) From bea7052606f59298796c54fe8cfc1f1ed215abf1 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Sat, 15 Aug 2026 20:50:04 -0400 Subject: [PATCH 08/11] aws_dynamodb_cdc: scope the snapshot tracker per attempt and probe state with a consistent read Follow-ups to the previous round's fixes: - The completion gate was scoped per connection attempt but the ack tracker stayed on the input-lifetime snapshotState and was read live in the scanner callbacks: a previous attempt's still-running scanner could interleave a second scan cursor into the new attempt's ordered tracker (racing the field write, breaking TrackBatch's scan-order contract, and potentially persisting positions or Complete=true past un-acked items). The tracker is now a local allocated next to the gate and captured by the batch and seal callbacks; the now write-only field is removed. - HasAnyState, the sole gate for honoring start_from: latest, used an eventually consistent Query: a fast crash-restart could miss checkpoint rows written moments earlier and reposition shards at LATEST, skipping their backlog. The probe now uses ConsistentRead (Limit 1 keeps the cost negligible). Also collects ack errors outside the test goroutines (require.FailNow is unsupported off the test goroutine) and uses new(sync.WaitGroup) per the project style rule. Full integration suite re-verified green. --- internal/impl/aws/dynamodb/checkpoint.go | 6 +++++ internal/impl/aws/dynamodb/input_cdc.go | 27 +++++++++---------- .../impl/aws/dynamodb/snapshot_ack_test.go | 9 ++++++- 3 files changed, 27 insertions(+), 15 deletions(-) diff --git a/internal/impl/aws/dynamodb/checkpoint.go b/internal/impl/aws/dynamodb/checkpoint.go index cb045117ae..23b2abf52a 100644 --- a/internal/impl/aws/dynamodb/checkpoint.go +++ b/internal/impl/aws/dynamodb/checkpoint.go @@ -348,6 +348,12 @@ func (c *Checkpointer) HasAnyState(ctx context.Context) (bool, error) { ":hv": &types.AttributeValueMemberS{Value: c.hashKeyValue()}, }, Limit: aws.Int32(1), + // This probe is the sole gate for honoring start_from: latest. An + // eventually consistent read could miss checkpoint rows written + // moments before a crash-restart and reposition shards at LATEST, + // silently skipping their backlog; Limit 1 makes the strong read + // nearly free. + ConsistentRead: aws.Bool(true), }) if err != nil { if _, ok := errors.AsType[*types.ResourceNotFoundException](err); ok { diff --git a/internal/impl/aws/dynamodb/input_cdc.go b/internal/impl/aws/dynamodb/input_cdc.go index fee6bb8f9d..7500191807 100644 --- a/internal/impl/aws/dynamodb/input_cdc.go +++ b/internal/impl/aws/dynamodb/input_cdc.go @@ -499,9 +499,6 @@ type snapshotState struct { scanner *SnapshotScanner recordsRead atomic.Int64 segmentsTotal int - - // ackTracker gates snapshot progress persistence on downstream acks. - ackTracker *snapshotAckTracker } // waitAckGate blocks until every batch counted on gate has been acked or @@ -1357,11 +1354,15 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st // Initialize snapshot scanner. Progress persistence is ack-gated: the // tracker below persists a segment's position only once every batch at or // below it has been acknowledged downstream. - d.snapshot.ackTracker = newSnapshotAckTracker(d.checkpointer, defaultSnapshotCheckpointBatchInterval, d.log) - // The completion gate is scoped to this connection attempt: it counts only - // batches this attempt emits, so batches orphaned in a previous attempt's - // msgChan cannot leave it permanently un-drainable. - ackGate := &sync.WaitGroup{} + // The tracker and completion gate are scoped to this connection attempt + // and captured by the scanner callbacks below: a previous attempt's + // still-live scanner keeps its own pair, so it can neither interleave a + // second scan cursor into this attempt's ordered tracker (which would + // break TrackBatch's scan-order contract and could persist positions or + // Complete=true past un-acked items) nor leave this attempt's gate + // permanently un-drainable via batches orphaned in a replaced msgChan. + ackTracker := newSnapshotAckTracker(d.checkpointer, defaultSnapshotCheckpointBatchInterval, d.log) + ackGate := new(sync.WaitGroup) d.snapshot.scanner = NewSnapshotScanner(SnapshotScannerConfig{ Client: d.dynamoClient, Table: tableName, @@ -1373,7 +1374,7 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st // Set batch callback to send snapshot records to msgChan d.snapshot.scanner.SetBatchCallback(func(ctx context.Context, items []map[string]dynamodbtypes.AttributeValue, segment int, lastKey map[string]dynamodbtypes.AttributeValue) error { - return d.handleSnapshotBatch(ctx, items, segment, tableName, lastKey, ackGate) + return d.handleSnapshotBatch(ctx, items, segment, tableName, lastKey, ackTracker, ackGate) }) // Set progress callback to update metrics @@ -1384,7 +1385,7 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st // Seal each segment behind its in-flight batches so Complete=true only // persists after they are all acknowledged. d.snapshot.scanner.SetSegmentSealedCallback(func(ctx context.Context, segment int) error { - if err := d.snapshot.ackTracker.SealSegment(ctx, segment); err != nil { + if err := ackTracker.SealSegment(ctx, segment); err != nil { d.metrics.checkpointFailures.Incr(1) return err } @@ -2759,7 +2760,7 @@ func (d *dynamoDBCDCInput) startShardReader(ctx context.Context, shardID string) // handleSnapshotBatch processes a batch of items from the snapshot scan. // lastKey is the scan position after this batch; it is registered with the // segment's ordered ack tracker and persisted only once acknowledged. -func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[string]dynamodbtypes.AttributeValue, segment int, tableName string, lastKey map[string]dynamodbtypes.AttributeValue, ackGate *sync.WaitGroup) error { +func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[string]dynamodbtypes.AttributeValue, segment int, tableName string, lastKey map[string]dynamodbtypes.AttributeValue, tracker *snapshotAckTracker, ackGate *sync.WaitGroup) error { if len(items) == 0 { return nil } @@ -2817,9 +2818,7 @@ func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[ // Register with the segment's ordered ack tracker: the scan position is // only persisted once this batch (and everything before it) is acked. - snapshot := d.snapshot - resolve := snapshot.ackTracker.TrackBatch(segment, lastKey, len(batch)) - tracker := snapshot.ackTracker + resolve := tracker.TrackBatch(segment, lastKey, len(batch)) // Track pending acks: the global gauge and the snapshot completion gate. d.pendingAcks.Add(1) diff --git a/internal/impl/aws/dynamodb/snapshot_ack_test.go b/internal/impl/aws/dynamodb/snapshot_ack_test.go index 931559f6a5..d1d1d6f118 100644 --- a/internal/impl/aws/dynamodb/snapshot_ack_test.go +++ b/internal/impl/aws/dynamodb/snapshot_ack_test.go @@ -181,13 +181,20 @@ func TestSnapshotAckTracker(t *testing.T) { resolves[i] = tracker.TrackBatch(7, scanKey(keys[i]), 1) } + // Ack errors are collected and asserted after Wait: require inside a + // non-test goroutine panics instead of failing the test. + ackErrs := make(chan error, batches) var wg sync.WaitGroup for i := range batches { wg.Go(func() { - require.NoError(t, tracker.Ack(ctx, 7, 1, resolves[i])) + ackErrs <- tracker.Ack(ctx, 7, 1, resolves[i]) }) } wg.Wait() + close(ackErrs) + for err := range ackErrs { + require.NoError(t, err) + } require.NoError(t, tracker.SealSegment(ctx, 7)) got := store.recorded() From 3008635e33dd8f6a8023114192c64aa754789581 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Mon, 17 Aug 2026 10:17:51 -0400 Subject: [PATCH 09/11] aws_dynamodb_cdc: document the narrowed start_from semantics Review finding: the start_from description still described the old behaviour ('latest starts from new records' whenever no checkpoint exists), but since the honorStartFrom change latest applies only to the first shard discovery of a genuinely fresh pipeline - rotation children and checkpoint-less shards after a restart always start at trim_horizon so their backlog is never skipped. The field description now states this, including the restart-under-latest replay implication. Docs regenerated. --- docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc | 4 +++- internal/impl/aws/dynamodb/input_cdc.go | 4 ++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc b/docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc index 361a8bc4d2..2ca9a1519f 100644 --- a/docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc +++ b/docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc @@ -384,7 +384,9 @@ Time to wait between polling attempts when no records are available. === `start_from` -Where to start reading when no checkpoint exists. `trim_horizon` starts from the oldest available record, `latest` starts from new records. +Where to start reading on a genuinely fresh pipeline (no checkpoint state exists yet under this `checkpoint_namespace` for the stream). `trim_horizon` starts from the oldest available record, `latest` starts from new records. + +`latest` is honoured only on that first discovery: once any checkpoint state exists, shards discovered later - rotation children found by the periodic refresh, and any checkpoint-less shard after a restart - always start at `trim_horizon` so their backlog is never skipped. In practice a restart under `latest` therefore replays from each shard's oldest retained record rather than only new records; at-least-once delivery takes precedence over the configured start position. *Type*: `string` diff --git a/internal/impl/aws/dynamodb/input_cdc.go b/internal/impl/aws/dynamodb/input_cdc.go index 7500191807..514d7a8cb6 100644 --- a/internal/impl/aws/dynamodb/input_cdc.go +++ b/internal/impl/aws/dynamodb/input_cdc.go @@ -233,8 +233,8 @@ When `+"`global_table`"+` is enabled the principal additionally needs `+"`dynamo Default(defaultDynamoDBPollInterval). Advanced(), service.NewStringEnumField(dciFieldStartFrom, "trim_horizon", "latest"). - Description("Where to start reading when no checkpoint exists. `trim_horizon` starts from the oldest available record, `latest` starts from new records."). - ShortDescription("Where to start when no checkpoint exists: trim_horizon for the oldest record, or latest."). + Description("Where to start reading on a genuinely fresh pipeline (no checkpoint state exists yet under this `checkpoint_namespace` for the stream). `trim_horizon` starts from the oldest available record, `latest` starts from new records.\n\n`latest` is honoured only on that first discovery: once any checkpoint state exists, shards discovered later - rotation children found by the periodic refresh, and any checkpoint-less shard after a restart - always start at `trim_horizon` so their backlog is never skipped. In practice a restart under `latest` therefore replays from each shard's oldest retained record rather than only new records; at-least-once delivery takes precedence over the configured start position."). + ShortDescription("Where a fresh pipeline starts: trim_horizon for the oldest record, or latest. After any checkpoint state exists, new shards always start at trim_horizon."). Default("trim_horizon"), service.NewIntField(dciFieldCheckpointLimit). Description("Maximum number of unacknowledged messages before forcing a checkpoint update. Lower values provide better recovery guarantees but increase write overhead."). From 35c0e8df1ff01c52e9708731cc5c074cef431210 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Tue, 18 Aug 2026 09:20:40 -0400 Subject: [PATCH 10/11] aws_dynamodb_cdc: log interrupted snapshot gate at info with resume context Review suggestion was warn (matching Close's unclean-shutdown warnings), but this branch fires on a clean cancellation - normal operation, which per the earlier review round should not warn. Info splits the difference: visible at default levels with the table and the resume-on-restart consequence spelled out, and consistent with how oracledb/mssqlserver log the same interrupted-handoff case. --- internal/impl/aws/dynamodb/input_cdc.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/internal/impl/aws/dynamodb/input_cdc.go b/internal/impl/aws/dynamodb/input_cdc.go index 514d7a8cb6..1043bbb25d 100644 --- a/internal/impl/aws/dynamodb/input_cdc.go +++ b/internal/impl/aws/dynamodb/input_cdc.go @@ -1435,8 +1435,11 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st if errors.Is(err, context.Canceled) { // Graceful shutdown mid-snapshot: nothing is wrong, the // un-acked items simply resume from acknowledged progress on - // the next run. - d.log.Debug("Snapshot completion gate interrupted by shutdown") + // the next run. Info rather than warn - a clean stop is normal + // operation - but visible, since the snapshot will resume and + // redeliver on the next run (matching oracledb/mssqlserver's + // interrupted-handoff logging). + d.log.Infof("Snapshot for table %s interrupted by shutdown; it will resume from acknowledged progress on the next run", tableName) return } wrappedErr := fmt.Errorf("snapshot completion gate for table %s: %w", tableName, err) From d8f164dae6d80bc2246a2aae2edf9e848d74edbc Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Tue, 18 Aug 2026 09:39:06 -0400 Subject: [PATCH 11/11] aws_dynamodb_cdc: survive transient seal-write failures and re-drive lost completions Two review findings on the segment-completion path: - A throttled UpdateSnapshotProgress inside SealSegment propagated through the scanner's errgroup and aborted the entire snapshot, cancelling every sibling segment - where the pre-ack-gating code treated checkpoint-write failures as non-fatal. The seal callback now logs and counts the failure and lets the scan finish; the completion marker is registered in the tracker either way. - The 'next ack retries the failed write' guarantee was unreachable for the completion marker: the seal is a segment's last settle event, so a failed Complete=true write had no retry path and the next run re-scanned the segment's tail from a stale position. FlushCompleted now re-drives any acked-but-unpersisted completion once the snapshot ack gate drains, before MarkSnapshotComplete; its own failure is non-fatal (the global complete marker is the durable fact once written, and a restart before it means duplicates, never loss). The Ack retry comment now describes the real scope of each mechanism. Covered by a seal-failure-then-re-drive unit test (including idempotency). --- internal/impl/aws/dynamodb/input_cdc.go | 22 +++++++-- internal/impl/aws/dynamodb/snapshot_ack.go | 48 +++++++++++++++++-- .../impl/aws/dynamodb/snapshot_ack_test.go | 29 +++++++++++ 3 files changed, 93 insertions(+), 6 deletions(-) diff --git a/internal/impl/aws/dynamodb/input_cdc.go b/internal/impl/aws/dynamodb/input_cdc.go index 1043bbb25d..c1748632f9 100644 --- a/internal/impl/aws/dynamodb/input_cdc.go +++ b/internal/impl/aws/dynamodb/input_cdc.go @@ -1383,11 +1383,15 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st }) // Seal each segment behind its in-flight batches so Complete=true only - // persists after they are all acknowledged. + // persists after they are all acknowledged. A failed completion WRITE is + // non-fatal: aborting the scan over a retryable store error would cancel + // every sibling segment and discard their un-persisted acked progress. + // The marker is already registered in the tracker either way, and the + // write is re-driven by FlushCompleted once the ack gate drains. d.snapshot.scanner.SetSegmentSealedCallback(func(ctx context.Context, segment int) error { if err := ackTracker.SealSegment(ctx, segment); err != nil { d.metrics.checkpointFailures.Incr(1) - return err + d.log.Warnf("Failed to persist completion for snapshot segment %d (will retry once all acks drain): %v", segment, err) } return nil }) @@ -1431,7 +1435,19 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st // every emitted batch has settled downstream: marking it complete any // earlier would let a crash skip un-acked items on restart. Blocks // until acks drain or soft-stop. - if err := waitAckGate(scanCtx, ackGate); err != nil { + if err := waitAckGate(scanCtx, ackGate); err == nil { + // Every batch has settled; re-drive any completion write that + // failed transiently at seal time (nothing else can retry it - + // the seal is each segment's last settle event). Failure here is + // non-fatal: the global complete marker below is the durable + // fact once written, and a restart before it merely re-scans + // from the last durable per-segment position (duplicates, never + // loss). + if flushErr := ackTracker.FlushCompleted(scanCtx); flushErr != nil { + d.metrics.checkpointFailures.Incr(1) + d.log.Warnf("Failed to re-drive snapshot completion writes for table %s: %v", tableName, flushErr) + } + } else { if errors.Is(err, context.Canceled) { // Graceful shutdown mid-snapshot: nothing is wrong, the // un-acked items simply resume from acknowledged progress on diff --git a/internal/impl/aws/dynamodb/snapshot_ack.go b/internal/impl/aws/dynamodb/snapshot_ack.go index cdcd556e3a..b27f920fc4 100644 --- a/internal/impl/aws/dynamodb/snapshot_ack.go +++ b/internal/impl/aws/dynamodb/snapshot_ack.go @@ -10,6 +10,8 @@ package dynamodb import ( "context" + "errors" + "fmt" "sync" "github.com/Jeffail/checkpoint" @@ -150,9 +152,11 @@ func (t *snapshotAckTracker) Ack(ctx context.Context, segment, n int, resolve fu persistErr = t.store.UpdateSnapshotProgress(ctx, segment, toPersist.lastKey, records) } if persistErr != nil { - // Bookkeeping is deliberately untouched: the next ack (or seal) for - // this segment retries the write instead of silently treating the - // failed position as durable. + // Bookkeeping is deliberately untouched so the failed position is + // never treated as durable. Interval persists are retried by the + // segment's next ack; a failed COMPLETION write has no later ack to + // retry it (the seal is the segment's last settle event), so it is + // re-driven by FlushCompleted once the snapshot's ack gate drains. return persistErr } @@ -165,6 +169,44 @@ func (t *snapshotAckTracker) Ack(ctx context.Context, segment, n int, resolve fu return nil } +// FlushCompleted re-drives the Complete=true write for every segment whose +// frontier has fully resolved to its seal marker but whose completion was +// never durably persisted - a throttled completion PutItem otherwise stays +// lost forever, and the next run re-scans the segment's tail from a stale +// position. Called after the snapshot ack gate has drained, so no acks are +// in flight; per-segment persistMu is still taken for consistency. +func (t *snapshotAckTracker) FlushCompleted(ctx context.Context) error { + t.mu.Lock() + var pending []int + for seg, st := range t.segments { + if st.hasFrontier && st.frontier.complete && !st.persistedComplete { + pending = append(pending, seg) + } + } + t.mu.Unlock() + + var errs []error + for _, seg := range pending { + t.mu.Lock() + st := t.segments[seg] + records := st.ackedRecords + t.mu.Unlock() + + st.persistMu.Lock() + err := t.store.UpdateSnapshotProgress(ctx, seg, nil, records) + if err == nil { + t.mu.Lock() + st.persistedComplete = true + t.mu.Unlock() + } + st.persistMu.Unlock() + if err != nil { + errs = append(errs, fmt.Errorf("re-driving completion for segment %d: %w", seg, err)) + } + } + return errors.Join(errs...) +} + // SealSegment registers the segment-complete marker. Segments can end on an // empty scan page, so completion cannot ride on a final batch; the marker // resolves immediately and Complete=true persists as soon as every batch diff --git a/internal/impl/aws/dynamodb/snapshot_ack_test.go b/internal/impl/aws/dynamodb/snapshot_ack_test.go index d1d1d6f118..6c09bcf04c 100644 --- a/internal/impl/aws/dynamodb/snapshot_ack_test.go +++ b/internal/impl/aws/dynamodb/snapshot_ack_test.go @@ -211,6 +211,35 @@ func TestSnapshotAckTracker(t *testing.T) { } }) + t.Run("FlushCompleted re-drives a completion write that failed at seal time", func(t *testing.T) { + // The seal is a segment's last settle event: when its Complete=true + // write fails transiently, no later ack exists to retry it. Without + // the post-gate re-drive the durable row stays Complete=false and the + // next run re-scans the segment's tail. + tracker, store := newTestSnapshotAckTracker(1) + + r1 := tracker.TrackBatch(5, scanKey("k1"), 3) + require.NoError(t, tracker.Ack(ctx, 5, 3, r1)) + + store.mu.Lock() + store.failNext = errStoreDown + store.mu.Unlock() + require.Error(t, tracker.SealSegment(ctx, 5), "the throttled completion write surfaces its error") + + before := store.recorded() + require.NoError(t, tracker.FlushCompleted(ctx)) + got := store.recorded() + require.Len(t, got, len(before)+1, "the re-drive must issue exactly the missing completion write") + last := got[len(got)-1] + require.Equal(t, 5, last.segment) + require.Nil(t, last.lastKey, "the re-driven write must be the Complete=true marker") + require.Equal(t, int64(3), last.recordsRead) + + // Idempotent: nothing left to re-drive. + require.NoError(t, tracker.FlushCompleted(ctx)) + require.Len(t, store.recorded(), len(got)) + }) + t.Run("a never-acked batch pins the segment forever", func(t *testing.T) { tracker, store := newTestSnapshotAckTracker(1)