diff --git a/docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc b/docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc index 05e689dd36..2ca9a1519f 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 @@ -382,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` @@ -412,6 +416,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/checkpoint.go b/internal/impl/aws/dynamodb/checkpoint.go index be59f7bfb0..23b2abf52a 100644 --- a/internal/impl/aws/dynamodb/checkpoint.go +++ b/internal/impl/aws/dynamodb/checkpoint.go @@ -332,6 +332,38 @@ 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), + // 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 { + 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 6d455a25ef..c1748632f9 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."). @@ -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) @@ -406,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. @@ -438,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 { @@ -478,6 +501,25 @@ type snapshotState struct { segmentsTotal int } +// 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. + gate.Wait() + close(drained) + }() + select { + case <-drained: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + // snapshotSequenceBuffer tracks sequence numbers seen during snapshot for deduplication. // // Architecture: Lock-free sharded hash table design @@ -1030,6 +1072,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 @@ -1143,6 +1194,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() @@ -1164,6 +1224,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) @@ -1290,21 +1351,30 @@ 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. + // 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, - 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, ackTracker, ackGate) }) // Set progress callback to update metrics @@ -1312,9 +1382,18 @@ 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. 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) + d.log.Warnf("Failed to persist completion for snapshot segment %d (will retry once all acks drain): %v", segment, err) + } + return nil }) // Set segment completion callback to track scan duration @@ -1352,6 +1431,43 @@ func (d *dynamoDBCDCInput) connectWithSnapshot(ctx context.Context, tableName st } } + // The scan has read everything, but the snapshot is only complete once + // 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 { + // 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 + // 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) + 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) @@ -1557,12 +1673,8 @@ 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" { - iteratorType = types.ShardIteratorTypeLatest - } else { - iteratorType = types.ShardIteratorTypeTrimHorizon - } - d.log.Infof("Starting shard %s from %s", shardID, d.conf.startFrom) + iteratorType = initialIteratorType(d.conf.startFrom, d.honorStartFrom.Load()) + d.log.Infof("Starting shard %s from %s", shardID, iteratorType) } // Get shard iterator (I/O operation - do not hold lock) @@ -1604,6 +1716,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 } @@ -1858,12 +1974,8 @@ 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" { - iteratorType = types.ShardIteratorTypeLatest - } else { - iteratorType = types.ShardIteratorTypeTrimHorizon - } - d.log.Infof("Starting shard %s (table %s) from %s", shardID, tableName, d.conf.startFrom) + iteratorType = initialIteratorType(d.conf.startFrom, ts.honorStartFrom.Load()) + d.log.Infof("Starting shard %s (table %s) from %s", shardID, tableName, iteratorType) } // Get shard iterator @@ -1904,6 +2016,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 } @@ -1943,20 +2059,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 } @@ -1978,7 +2109,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, @@ -2645,8 +2776,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, tracker *snapshotAckTracker, ackGate *sync.WaitGroup) error { if len(items) == 0 { return nil } @@ -2702,23 +2835,32 @@ 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. + resolve := tracker.TrackBatch(segment, lastKey, len(batch)) + + // Track pending acks: the global gauge and the snapshot completion gate. d.pendingAcks.Add(1) + ackGate.Add(1) - // Create simple ack function for snapshot records - ackFunc := func(_ context.Context, err error) error { + ackFunc := func(ackCtx context.Context, _ error) error { defer d.pendingAcks.Done() + defer ackGate.Done() if d.closed.Load() { d.log.Debug("Received snapshot ack after close, dropping") return nil } - if err != nil { - d.log.Warnf("Snapshot batch nacked from segment %d: %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) + return ackErr } - return nil } @@ -2726,6 +2868,7 @@ func (d *dynamoDBCDCInput) handleSnapshotBatch(ctx context.Context, items []map[ select { case <-ctx.Done(): d.pendingAcks.Done() // Undo the Add(1) above + 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_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/input_cdc_integration_test.go b/internal/impl/aws/dynamodb/input_cdc_integration_test.go index d7327d561b..b019b6304e 100644 --- a/internal/impl/aws/dynamodb/input_cdc_integration_test.go +++ b/internal/impl/aws/dynamodb/input_cdc_integration_test.go @@ -533,6 +533,126 @@ credentials: assert.NotEmpty(t, batch2, "Should read new events after resumption") } +// 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") + } +} + // TestIntegrationDynamoDBSnapshot tests snapshot functionality. func TestIntegrationDynamoDBSnapshot(t *testing.T) { integration.CheckSkip(t) 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..b27f920fc4 --- /dev/null +++ b/internal/impl/aws/dynamodb/snapshot_ack.go @@ -0,0 +1,219 @@ +// 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" + "errors" + "fmt" + "sync" + + "github.com/Jeffail/checkpoint" + dynamodbtypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + + "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 { + 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 { + // 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 + // 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 = defaultSnapshotCheckpointBatchInterval + } + 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] + t.mu.Unlock() + if !ok { + 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 + } + 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 + } + } + ackedAtCompute := st.ackedBatches + t.mu.Unlock() + + if toPersist == nil { + return nil + } + var persistErr error + if toPersist.complete { + 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 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 + } + + t.mu.Lock() + st.persistedBatches = ackedAtCompute + if toPersist.complete { + st.persistedComplete = true + } + t.mu.Unlock() + 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 +// 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..6c09bcf04c --- /dev/null +++ b/internal/impl/aws/dynamodb/snapshot_ack_test.go @@ -0,0 +1,253 @@ +// 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" + "errors" + "fmt" + "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 + 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 +} + +func (f *fakeProgressStore) recorded() []recordedProgress { + f.mu.Lock() + defer f.mu.Unlock() + 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}, + } +} + +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 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("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) + } + + // 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() { + 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() + 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("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) + + 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 an in-flight batch") + }) +}