From 5f9e88dadfe213be6f665820745887100ff8d00a Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Sat, 12 Sep 2026 09:20:21 +0700 Subject: [PATCH 1/9] fix: report stream batch failures by SequenceNumber AWS matches ReportBatchItemFailures identifiers against the DynamoDB stream record SequenceNumber. HandleStreamEvent returned the EventID, so Lambda did not recognise the identifier and re-drove the whole batch -- with bisect_batch_on_function_error and maximum_retry_attempts = 3 that replayed already-successful records whose WriteJobEvent and PublishEvent side effects are not idempotent. Adds the first test file to internal/lambda/stream. --- internal/lambda/stream/handler.go | 6 ++- internal/lambda/stream/handler_test.go | 72 ++++++++++++++++++++++++++ 2 files changed, 77 insertions(+), 1 deletion(-) create mode 100644 internal/lambda/stream/handler_test.go diff --git a/internal/lambda/stream/handler.go b/internal/lambda/stream/handler.go index 8e736e5..67292a8 100644 --- a/internal/lambda/stream/handler.go +++ b/internal/lambda/stream/handler.go @@ -27,9 +27,13 @@ func HandleStreamEvent(ctx context.Context, d *lambda.Deps, event lambda.StreamE d.Logger.Error("stream record error", "error", err, "eventID", event.Records[i].EventID, + "sequenceNumber", event.Records[i].Change.SequenceNumber, ) + // AWS matches ReportBatchItemFailures identifiers against the + // stream record SequenceNumber. Returning the EventID makes the + // identifier unrecognisable and re-drives the entire batch. resp.BatchItemFailures = append(resp.BatchItemFailures, events.DynamoDBBatchItemFailure{ - ItemIdentifier: event.Records[i].EventID, + ItemIdentifier: event.Records[i].Change.SequenceNumber, }) } } diff --git a/internal/lambda/stream/handler_test.go b/internal/lambda/stream/handler_test.go new file mode 100644 index 0000000..d62e005 --- /dev/null +++ b/internal/lambda/stream/handler_test.go @@ -0,0 +1,72 @@ +package stream + +import ( + "context" + "io" + "log/slog" + "testing" + "time" + + "github.com/aws/aws-lambda-go/events" + + lambda "github.com/dwsmith1983/interlock/internal/lambda" + "github.com/dwsmith1983/interlock/internal/store" + "github.com/dwsmith1983/interlock/internal/store/storetest" + "github.com/dwsmith1983/interlock/pkg/types" +) + +// discardLogger keeps test output readable; every routing branch logs. +func discardLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// TestHandleStreamEvent_BatchFailureUsesSequenceNumber pins the AWS +// ReportBatchItemFailures contract: Lambda matches each returned +// ItemIdentifier against the stream record's SequenceNumber. An EventID is not +// a recognised identifier, so Lambda treats the whole batch as failed and +// re-drives records whose side effects (WriteJobEvent, PublishEvent) are not +// idempotent. +func TestHandleStreamEvent_BatchFailureUsesSequenceNumber(t *testing.T) { + s := storetest.NewStore(&storetest.FakeDynamo{}) + d := &lambda.Deps{ + Store: s, + ConfigCache: store.NewConfigCache(s, 5*time.Minute), + Logger: discardLogger(), + } + + // No Keys at all -> handleRecord returns "record missing PK or SK". + bad := events.DynamoDBEventRecord{ + EventID: "evt-1", + EventName: "INSERT", + Change: events.DynamoDBStreamRecord{ + SequenceNumber: "111", + Keys: map[string]events.DynamoDBAttributeValue{}, + }, + } + // An unrouted SK prefix hits the default branch and returns nil without + // touching the store. + good := events.DynamoDBEventRecord{ + EventID: "evt-2", + EventName: "INSERT", + Change: events.DynamoDBStreamRecord{ + SequenceNumber: "222", + Keys: map[string]events.DynamoDBAttributeValue{ + "PK": events.NewStringAttribute(types.PipelinePK("p")), + "SK": events.NewStringAttribute("UNROUTED#1"), + }, + }, + } + + resp, err := HandleStreamEvent(context.Background(), d, lambda.StreamEvent{ + Records: []events.DynamoDBEventRecord{bad, good}, + }) + if err != nil { + t.Fatalf("HandleStreamEvent returned error: %v", err) + } + if len(resp.BatchItemFailures) != 1 { + t.Fatalf("BatchItemFailures = %+v, want exactly 1 (only the keyless record)", resp.BatchItemFailures) + } + if got := resp.BatchItemFailures[0].ItemIdentifier; got != "111" { + t.Errorf("ItemIdentifier = %q, want %q (the record SequenceNumber, not the EventID)", got, "111") + } +} From af6721b7ddb4d24e52436aac238ae83ff984c973 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Sat, 12 Sep 2026 09:25:05 +0700 Subject: [PATCH 2/9] fix: skip REMOVE records in the stream router handleRecord ignored EventName, so a deleted row reached the write handlers with a nil NewImage. Deleting a RERUN_REQUEST# row (those rows have no TTL and are never cleaned up) started a rerun with reason "manual"; deleting a SENSOR# row -- including the postrun-baseline# delete the rerun path itself issues -- could publish POST_RUN_PASSED or POST_RUN_FAILED; 30-day JOB# TTL expiries invoked the Lambda for nothing. REMOVE records are now skipped and logged at info with the pk, sk, event id and whether the delete came from TTL expiry. SK = CONFIG is the one exception: it still invalidates the config cache. --- internal/lambda/stream/handler.go | 36 ++++ internal/lambda/stream/handler_test.go | 236 +++++++++++++++++++++++++ internal/store/storetest/fake.go | 4 +- 3 files changed, 274 insertions(+), 2 deletions(-) diff --git a/internal/lambda/stream/handler.go b/internal/lambda/stream/handler.go index 67292a8..8e2a0a5 100644 --- a/internal/lambda/stream/handler.go +++ b/internal/lambda/stream/handler.go @@ -15,6 +15,21 @@ import ( "github.com/dwsmith1983/interlock/pkg/types" ) +const ( + // removeEventName is the DynamoDB stream EventName for a deleted item. + removeEventName = "REMOVE" + // ttlServiceIdentity is the UserIdentity.Type DynamoDB sets on records it + // deletes itself via the TTL process (PrincipalID "dynamodb.amazonaws.com"). + ttlServiceIdentity = "Service" +) + +// isTTLExpiry reports whether a stream record was produced by DynamoDB's TTL +// deleter rather than by an application or an operator. UserIdentity is a +// pointer and is nil on every non-TTL record. +func isTTLExpiry(record events.DynamoDBEventRecord) bool { + return record.UserIdentity != nil && record.UserIdentity.Type == ttlServiceIdentity +} + // HandleStreamEvent processes a DynamoDB stream event, routing each record // to the appropriate handler based on the SK prefix. Per-record errors are // collected as BatchItemFailures so the Lambda runtime can use DynamoDB's @@ -47,6 +62,27 @@ func handleRecord(ctx context.Context, d *lambda.Deps, record events.DynamoDBEve return fmt.Errorf("record missing PK or SK") } + // A deleted row carries no NewImage, so routing it makes the downstream + // handlers act on absent data: a deleted RERUN_REQUEST# row would start a + // rerun with reason "manual", and a deleted SENSOR# row would publish a + // POST_RUN_* verdict. The one actionable delete is CONFIG: a pipeline + // whose config disappeared must drop out of the cache. An empty EventName + // (synthetic records) is treated as a write, as before. + if record.EventName == removeEventName { + if sk == types.ConfigSK { + d.Logger.Info("config deleted, invalidating cache", "pk", pk) + d.ConfigCache.Invalidate() + return nil + } + d.Logger.Info("skipping REMOVE stream record", + "pk", pk, + "sk", sk, + "eventID", record.EventID, + "ttlExpiry", isTTLExpiry(record), + ) + return nil + } + switch { case strings.HasPrefix(sk, "SENSOR#"): return handleSensorEvent(ctx, d, pk, sk, record) diff --git a/internal/lambda/stream/handler_test.go b/internal/lambda/stream/handler_test.go index d62e005..7b7dd8c 100644 --- a/internal/lambda/stream/handler_test.go +++ b/internal/lambda/stream/handler_test.go @@ -2,12 +2,18 @@ package stream import ( "context" + "encoding/json" "io" "log/slog" + "strings" "testing" "time" "github.com/aws/aws-lambda-go/events" + "github.com/aws/aws-sdk-go-v2/service/dynamodb" + ddbtypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + "github.com/aws/aws-sdk-go-v2/service/eventbridge" + "github.com/aws/aws-sdk-go-v2/service/sfn" lambda "github.com/dwsmith1983/interlock/internal/lambda" "github.com/dwsmith1983/interlock/internal/store" @@ -70,3 +76,233 @@ func TestHandleStreamEvent_BatchFailureUsesSequenceNumber(t *testing.T) { t.Errorf("ItemIdentifier = %q, want %q (the record SequenceNumber, not the EventID)", got, "111") } } + +// testDate is the execution date every REMOVE-guard case uses. It matches +// testNow so ResolveExecutionDate derives the same date from an absent +// NewImage (REMOVE records carry no NewImage). +const testDate = "2026-03-07" + +func testNow() time.Time { + return time.Date(2026, 3, 7, 12, 0, 0, 0, time.UTC) +} + +// countingSFN is a lambda.SFNAPI double that records StartExecution calls, so +// a test can prove no run was started. +type countingSFN struct{ calls int } + +func (c *countingSFN) StartExecution(context.Context, *sfn.StartExecutionInput, ...func(*sfn.Options)) (*sfn.StartExecutionOutput, error) { + c.calls++ + return &sfn.StartExecutionOutput{}, nil +} + +// countingEventBridge is a lambda.EventBridgeAPI double that records PutEvents +// calls, so a test can prove no verdict was published. +type countingEventBridge struct{ calls int } + +func (c *countingEventBridge) PutEvents(context.Context, *eventbridge.PutEventsInput, ...func(*eventbridge.Options)) (*eventbridge.PutEventsOutput, error) { + c.calls++ + return &eventbridge.PutEventsOutput{}, nil +} + +// removeGuardConfig is a pipeline with both a stream trigger and post-run +// rules, so a REMOVE on any routed SK prefix reaches a handler that would +// write or publish if the guard were missing. +func removeGuardConfig() types.PipelineConfig { + return types.PipelineConfig{ + Pipeline: types.PipelineIdentity{ID: "p"}, + Schedule: types.ScheduleConfig{ + Trigger: &types.TriggerCondition{Key: "upstream", Check: types.CheckExists}, + }, + PostRun: &types.PostRunConfig{ + Rules: []types.ValidationRule{{Key: "row-count", Check: types.CheckExists}}, + }, + } +} + +// configScanItem builds the CONFIG row that store.ScanConfigs (and therefore +// ConfigCache) expects for pipeline "p". +func configScanItem(t *testing.T) map[string]ddbtypes.AttributeValue { + t.Helper() + data, err := json.Marshal(removeGuardConfig()) + if err != nil { + t.Fatalf("marshal config: %v", err) + } + return map[string]ddbtypes.AttributeValue{ + "PK": &ddbtypes.AttributeValueMemberS{Value: types.PipelinePK("p")}, + "SK": &ddbtypes.AttributeValueMemberS{Value: types.ConfigSK}, + "config": &ddbtypes.AttributeValueMemberS{Value: string(data)}, + } +} + +// completedTriggerItem is a COMPLETED TRIGGER# row, the state that sends a +// SENSOR# record down the post-run verdict path. +func completedTriggerItem() map[string]ddbtypes.AttributeValue { + return map[string]ddbtypes.AttributeValue{ + "PK": &ddbtypes.AttributeValueMemberS{Value: types.PipelinePK("p")}, + "SK": &ddbtypes.AttributeValueMemberS{Value: types.TriggerSK("stream", testDate)}, + "status": &ddbtypes.AttributeValueMemberS{Value: types.TriggerStatusCompleted}, + } +} + +// newNoWriteFake serves the CONFIG row on Scan and a COMPLETED TRIGGER row on +// GetItem, and fails the test if the router issues any write. +func newNoWriteFake(t *testing.T) *storetest.FakeDynamo { + t.Helper() + return &storetest.FakeDynamo{ + ScanFn: func(context.Context, *dynamodb.ScanInput) (*dynamodb.ScanOutput, error) { + return &dynamodb.ScanOutput{ + Items: []map[string]ddbtypes.AttributeValue{configScanItem(t)}, + }, nil + }, + GetItemFn: func(_ context.Context, in *dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error) { + if skAttr, ok := in.Key["SK"].(*ddbtypes.AttributeValueMemberS); ok && strings.HasPrefix(skAttr.Value, "TRIGGER#") { + return &dynamodb.GetItemOutput{Item: completedTriggerItem()}, nil + } + return &dynamodb.GetItemOutput{}, nil + }, + PutItemFn: func(_ context.Context, in *dynamodb.PutItemInput) (*dynamodb.PutItemOutput, error) { + t.Errorf("PutItem called for a REMOVE record: %v", in.Item) + return &dynamodb.PutItemOutput{}, nil + }, + UpdateItemFn: func(_ context.Context, in *dynamodb.UpdateItemInput) (*dynamodb.UpdateItemOutput, error) { + t.Errorf("UpdateItem called for a REMOVE record: %v", in.Key) + return &dynamodb.UpdateItemOutput{}, nil + }, + DeleteItemFn: func(_ context.Context, in *dynamodb.DeleteItemInput) (*dynamodb.DeleteItemOutput, error) { + t.Errorf("DeleteItem called for a REMOVE record: %v", in.Key) + return &dynamodb.DeleteItemOutput{}, nil + }, + } +} + +// TestHandleRecord_SkipsRemoveRecords pins H11: a deleted row carries no +// NewImage, so routing it makes downstream handlers act on absent data. A +// deleted RERUN_REQUEST# row starts a rerun with reason "manual"; a deleted +// SENSOR# row publishes a POST_RUN_* verdict. TTL expiries (30-day JOB# rows, +// trigger-lock rows) arrive as REMOVE with a Service user identity. +func TestHandleRecord_SkipsRemoveRecords(t *testing.T) { + ttlIdentity := &events.DynamoDBUserIdentity{ + Type: "Service", + PrincipalID: "dynamodb.amazonaws.com", + } + + tests := []struct { + name string + sk string + userIdentity *events.DynamoDBUserIdentity + }{ + {name: "operator deletes a rerun request", sk: types.RerunRequestSK("stream", testDate)}, + {name: "sensor row deleted", sk: types.SensorSK("row-count")}, + {name: "job row deleted", sk: types.JobSK("stream", testDate, "1772000000000")}, + {name: "job row expired by ttl", sk: types.JobSK("stream", testDate, "1772000000000"), userIdentity: ttlIdentity}, + {name: "trigger row expired by ttl", sk: types.TriggerSK("stream", testDate), userIdentity: ttlIdentity}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := storetest.NewStore(newNoWriteFake(t)) + sfnFake := &countingSFN{} + ebFake := &countingEventBridge{} + d := &lambda.Deps{ + Store: s, + ConfigCache: store.NewConfigCache(s, 5*time.Minute), + SFNClient: sfnFake, + EventBridge: ebFake, + EventBusName: "test-bus", + NowFunc: testNow, + Logger: discardLogger(), + } + + record := events.DynamoDBEventRecord{ + EventID: "evt-remove", + EventName: "REMOVE", + UserIdentity: tt.userIdentity, + Change: events.DynamoDBStreamRecord{ + SequenceNumber: "900", + Keys: map[string]events.DynamoDBAttributeValue{ + "PK": events.NewStringAttribute(types.PipelinePK("p")), + "SK": events.NewStringAttribute(tt.sk), + }, + OldImage: map[string]events.DynamoDBAttributeValue{ + "reason": events.NewStringAttribute("data-drift"), + }, + }, + } + + if err := handleRecord(context.Background(), d, record); err != nil { + t.Fatalf("handleRecord returned error: %v", err) + } + if sfnFake.calls != 0 { + t.Errorf("StartExecution called %d times, want 0 -- a REMOVE must never start a run", sfnFake.calls) + } + if ebFake.calls != 0 { + t.Errorf("PutEvents called %d times, want 0 -- a REMOVE must never publish a verdict", ebFake.calls) + } + }) + } +} + +// TestHandleRecord_ConfigChangesInvalidateCache pins the one delete the router +// must still act on: a CONFIG row that disappears has to drop out of the +// cache, otherwise the router keeps triggering a pipeline whose config no +// longer exists for up to the cache TTL. INSERT and MODIFY must keep working. +// ConfigCache exposes no state, so invalidation is observed through the extra +// ScanConfigs call a stale cache would have skipped. +func TestHandleRecord_ConfigChangesInvalidateCache(t *testing.T) { + tests := []struct { + name string + eventName string + }{ + {name: "config inserted", eventName: "INSERT"}, + {name: "config modified", eventName: "MODIFY"}, + {name: "config deleted", eventName: "REMOVE"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + scans := 0 + fake := &storetest.FakeDynamo{ + ScanFn: func(context.Context, *dynamodb.ScanInput) (*dynamodb.ScanOutput, error) { + scans++ + return &dynamodb.ScanOutput{ + Items: []map[string]ddbtypes.AttributeValue{configScanItem(t)}, + }, nil + }, + } + s := storetest.NewStore(fake) + cache := store.NewConfigCache(s, 5*time.Minute) + d := &lambda.Deps{Store: s, ConfigCache: cache, Logger: discardLogger()} + + // Prime the cache: a second GetAll is served from memory unless + // the stream record invalidates it. + if _, err := cache.GetAll(context.Background()); err != nil { + t.Fatalf("prime cache: %v", err) + } + if scans != 1 { + t.Fatalf("scans after priming = %d, want 1", scans) + } + + record := events.DynamoDBEventRecord{ + EventID: "evt-config", + EventName: tt.eventName, + Change: events.DynamoDBStreamRecord{ + SequenceNumber: "800", + Keys: map[string]events.DynamoDBAttributeValue{ + "PK": events.NewStringAttribute(types.PipelinePK("p")), + "SK": events.NewStringAttribute(types.ConfigSK), + }, + }, + } + if err := handleRecord(context.Background(), d, record); err != nil { + t.Fatalf("handleRecord returned error: %v", err) + } + + if _, err := cache.GetAll(context.Background()); err != nil { + t.Fatalf("second GetAll: %v", err) + } + if scans != 2 { + t.Errorf("scans after a %s CONFIG record = %d, want 2 -- the cache must be invalidated", tt.eventName, scans) + } + }) + } +} diff --git a/internal/store/storetest/fake.go b/internal/store/storetest/fake.go index b987672..54bffe0 100644 --- a/internal/store/storetest/fake.go +++ b/internal/store/storetest/fake.go @@ -1,6 +1,6 @@ // Package storetest provides test doubles for the store package. It is used -// only from _test.go files (internal/lambda/orchestrator, deploy); no -// production code imports it. +// only from _test.go files (internal/lambda/orchestrator, internal/lambda/sla, +// internal/lambda/stream, deploy); no production code imports it. package storetest import ( From 1c177f32b68708e4ffead373073ec1979235fed1 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Sat, 12 Sep 2026 09:27:31 +0700 Subject: [PATCH 3/9] fix: filter REMOVE records at the DynamoDB event source mappings Defense in depth behind the handleRecord REMOVE guard. Both stream mappings now carry filter_criteria; the OR-ed filters pass INSERT and MODIFY on both tables, plus REMOVE of the CONFIG row on the control table so the config cache is still invalidated when a pipeline config is deleted. The joblog table holds no CONFIG rows, so it gets the writes filter only. Filtered records advance the stream iterator without invoking or billing the function, which also removes the 30-day JOB# TTL-expiry invocations. Requires terraform apply to take effect on existing deployments. --- deploy/terraform/lambda.tf | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/deploy/terraform/lambda.tf b/deploy/terraform/lambda.tf index 780e4de..fd7f03c 100644 --- a/deploy/terraform/lambda.tf +++ b/deploy/terraform/lambda.tf @@ -554,6 +554,21 @@ resource "aws_lambda_event_source_mapping" "control_stream" { maximum_retry_attempts = 3 function_response_types = ["ReportBatchItemFailures"] + # Defense in depth. The primary guard is in handleRecord + # (internal/lambda/stream/handler.go), which skips REMOVE records; these + # filters stop them being delivered or billed at all. Multiple filter blocks + # are OR-ed: writes always pass, and deletes pass only for the CONFIG row so + # the handler can still invalidate the config cache. Operator deletes and TTL + # expiries of SENSOR#, JOB#, RERUN_REQUEST# and TRIGGER# rows never arrive. + filter_criteria { + filter { + pattern = jsonencode({ eventName = ["INSERT", "MODIFY"] }) + } + filter { + pattern = jsonencode({ eventName = ["REMOVE"], dynamodb = { Keys = { SK = { S = ["CONFIG"] } } } }) + } + } + destination_config { on_failure { destination_arn = aws_sqs_queue.stream_router_control_dlq.arn @@ -570,6 +585,16 @@ resource "aws_lambda_event_source_mapping" "joblog_stream" { maximum_retry_attempts = 3 function_response_types = ["ReportBatchItemFailures"] + # Only writes. Unlike the control table, the joblog table holds no CONFIG + # rows (store.ScanConfigs scans the control table only), so there is no + # CONFIG-delete escape hatch to add here. This filter also stops the 30-day + # JOB# TTL expiries from invoking the function. + filter_criteria { + filter { + pattern = jsonencode({ eventName = ["INSERT", "MODIFY"] }) + } + } + destination_config { on_failure { destination_arn = aws_sqs_queue.stream_router_joblog_dlq.arn From 8e7e07c7da7d25aeaf125f35ab8dfb4bf40e6c0e Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Sat, 12 Sep 2026 09:29:34 +0700 Subject: [PATCH 4/9] fix: apply the stream mapping filters in the LocalStack deploy Mirrors the filter_criteria blocks added to deploy/terraform/lambda.tf so local E2E runs exercise the same delivery contract as production: writes always pass, and REMOVE passes only for the control table's CONFIG row. Existing mappings are left alone (the function is still idempotent), so a running stack needs a teardown and redeploy to pick the filters up. --- deploy/localstack/deploy.py | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/deploy/localstack/deploy.py b/deploy/localstack/deploy.py index 53e15d6..3023c49 100644 --- a/deploy/localstack/deploy.py +++ b/deploy/localstack/deploy.py @@ -601,18 +601,35 @@ def _find_mapping(lam, source_arn: str, function_name: str) -> str | None: return items[0]["UUID"] if items else None -def ensure_stream_mapping(stream_arn: str, function_name: str) -> None: +# Mirrors the filter_criteria blocks on the aws_lambda_event_source_mapping +# resources in deploy/terraform/lambda.tf. Filters are OR-ed: writes always +# pass, and deletes pass only for the CONFIG row so the stream router can +# still invalidate its config cache. +_FILTER_WRITES = {"eventName": ["INSERT", "MODIFY"]} +_FILTER_CONFIG_DELETE = { + "eventName": ["REMOVE"], + "dynamodb": {"Keys": {"SK": {"S": ["CONFIG"]}}}, +} + + +def ensure_stream_mapping( + stream_arn: str, function_name: str, *, allow_config_delete: bool = False +) -> None: lam = _client("lambda") existing = _find_mapping(lam, stream_arn, function_name) if existing: print(f" [esm] mapping {stream_arn} -> {function_name} already exists") return + patterns: list[dict[str, Any]] = [_FILTER_WRITES] + if allow_config_delete: + patterns.append(_FILTER_CONFIG_DELETE) lam.create_event_source_mapping( EventSourceArn=stream_arn, FunctionName=function_name, StartingPosition="LATEST", BatchSize=10, FunctionResponseTypes=["ReportBatchItemFailures"], + FilterCriteria={"Filters": [{"Pattern": json.dumps(p)} for p in patterns]}, ) print(f" [esm] created mapping {stream_arn} -> {function_name}") @@ -769,7 +786,9 @@ def cmd_deploy() -> None: ensure_watchdog_schedule(lambda_arns["watchdog"]) print("--> Event source mappings") - ensure_stream_mapping(stream_arns[TABLE_CONTROL], f"{PREFIX}-stream-router") + ensure_stream_mapping( + stream_arns[TABLE_CONTROL], f"{PREFIX}-stream-router", allow_config_delete=True + ) ensure_stream_mapping(stream_arns[TABLE_JOBLOG], f"{PREFIX}-stream-router") ensure_sqs_mapping(alert_queue_arn, f"{PREFIX}-alert-dispatcher") From ae5c424b2987ec92d78b1982aef29b8b66c98cb3 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Sat, 12 Sep 2026 09:31:30 +0700 Subject: [PATCH 5/9] docs: changelog entries for the stream-router contract fixes Documents the SequenceNumber batch-failure fix and the REMOVE guard, including the event source mapping filters and the terraform apply that operators need to run to pick them up. --- CHANGELOG.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 662cdd1..77ac09e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- **Stream router reported the wrong batch-item identifier (C3)** — `stream.HandleStreamEvent` returned `ItemIdentifier: EventID` in `BatchItemFailures`. AWS matches `ReportBatchItemFailures` identifiers against the DynamoDB stream record `SequenceNumber`, so an unrecognised identifier made Lambda treat the entire batch as failed. Combined with `bisect_batch_on_function_error = true` and `maximum_retry_attempts = 3`, that replayed already-successful records whose `WriteJobEvent` and `PublishEvent` side effects are not idempotent. The handler now returns `record.Change.SequenceNumber` and logs it alongside the event ID (the event ID remains the log correlation ID). +- **Stream router acted on REMOVE records (H11)** — `handleRecord` never checked `EventName`, so a deleted row reached the write handlers with a nil `NewImage`. Deleting a `RERUN_REQUEST#` row (those rows have no TTL and are never cleaned up) started a rerun with `reason="manual"`; deleting a `SENSOR#` row — including the `postrun-baseline#` delete the rerun path issues itself — could publish `POST_RUN_PASSED`/`POST_RUN_FAILED` verdicts from absent data; and 30-day `JOB#` TTL expiries invoked the function for nothing. REMOVE records are now skipped and logged at info with the key, event ID and whether the delete came from DynamoDB's TTL process, with the single exception of `SK = CONFIG`, which still invalidates the config cache so a deleted pipeline config drops out immediately. Defense in depth: both DynamoDB event source mappings in `deploy/terraform/lambda.tf` now carry `filter_criteria` (writes always pass; REMOVE passes only for the control table's `CONFIG` row), so filtered records are neither delivered nor billed. **Operators must run `terraform apply` to pick up the new event source mapping filters**; the handler guard protects deployments in the meantime. `deploy/localstack/deploy.py` applies the equivalent `FilterCriteria`, but only when a mapping is created — tear down and redeploy an existing LocalStack stack to pick them up. + ## [0.10.0] - 2026-09-12 ### Added From 7e63898c1db1f3aaf44bb15238adfa1763acfe58 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Sat, 12 Sep 2026 09:43:46 +0700 Subject: [PATCH 6/9] docs: state the residual partial-batch replay for the SequenceNumber fix --- CHANGELOG.md | 2 +- internal/lambda/stream/handler.go | 8 +++++--- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 77ac09e..0eaf831 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- **Stream router reported the wrong batch-item identifier (C3)** — `stream.HandleStreamEvent` returned `ItemIdentifier: EventID` in `BatchItemFailures`. AWS matches `ReportBatchItemFailures` identifiers against the DynamoDB stream record `SequenceNumber`, so an unrecognised identifier made Lambda treat the entire batch as failed. Combined with `bisect_batch_on_function_error = true` and `maximum_retry_attempts = 3`, that replayed already-successful records whose `WriteJobEvent` and `PublishEvent` side effects are not idempotent. The handler now returns `record.Change.SequenceNumber` and logs it alongside the event ID (the event ID remains the log correlation ID). +- **Stream router reported the wrong batch-item identifier (C3)** — `stream.HandleStreamEvent` returned `ItemIdentifier: EventID` in `BatchItemFailures`. AWS matches `ReportBatchItemFailures` identifiers against the DynamoDB stream record `SequenceNumber`, so an unrecognised identifier made Lambda treat the entire batch as failed and re-drive every record in it. The handler now returns `record.Change.SequenceNumber` and logs it alongside the event ID (the event ID remains the log correlation ID) — but partial-batch reporting only checkpoints at the *first* failed record and still re-drives that record and every later one in the batch, so this narrows the replay of non-idempotent `WriteJobEvent`/`PublishEvent` side effects rather than eliminating it; making those handlers idempotent remains a separate follow-up. - **Stream router acted on REMOVE records (H11)** — `handleRecord` never checked `EventName`, so a deleted row reached the write handlers with a nil `NewImage`. Deleting a `RERUN_REQUEST#` row (those rows have no TTL and are never cleaned up) started a rerun with `reason="manual"`; deleting a `SENSOR#` row — including the `postrun-baseline#` delete the rerun path issues itself — could publish `POST_RUN_PASSED`/`POST_RUN_FAILED` verdicts from absent data; and 30-day `JOB#` TTL expiries invoked the function for nothing. REMOVE records are now skipped and logged at info with the key, event ID and whether the delete came from DynamoDB's TTL process, with the single exception of `SK = CONFIG`, which still invalidates the config cache so a deleted pipeline config drops out immediately. Defense in depth: both DynamoDB event source mappings in `deploy/terraform/lambda.tf` now carry `filter_criteria` (writes always pass; REMOVE passes only for the control table's `CONFIG` row), so filtered records are neither delivered nor billed. **Operators must run `terraform apply` to pick up the new event source mapping filters**; the handler guard protects deployments in the meantime. `deploy/localstack/deploy.py` applies the equivalent `FilterCriteria`, but only when a mapping is created — tear down and redeploy an existing LocalStack stack to pick them up. ## [0.10.0] - 2026-09-12 diff --git a/internal/lambda/stream/handler.go b/internal/lambda/stream/handler.go index 8e2a0a5..651895a 100644 --- a/internal/lambda/stream/handler.go +++ b/internal/lambda/stream/handler.go @@ -44,9 +44,11 @@ func HandleStreamEvent(ctx context.Context, d *lambda.Deps, event lambda.StreamE "eventID", event.Records[i].EventID, "sequenceNumber", event.Records[i].Change.SequenceNumber, ) - // AWS matches ReportBatchItemFailures identifiers against the - // stream record SequenceNumber. Returning the EventID makes the - // identifier unrecognisable and re-drives the entire batch. + // AWS matches each BatchItemFailures identifier against the record + // SequenceNumber and checkpoints at the lowest one returned, + // re-driving that record and every later record in the batch. An + // unrecognised identifier (such as the EventID) instead falls back + // to re-driving the entire batch. resp.BatchItemFailures = append(resp.BatchItemFailures, events.DynamoDBBatchItemFailure{ ItemIdentifier: event.Records[i].Change.SequenceNumber, }) From e8cb765f6ca8425ca5934d7c441c4dcb000a3170 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Sat, 12 Sep 2026 09:43:48 +0700 Subject: [PATCH 7/9] test: assert isTTLExpiry and REMOVE key-extraction ordering --- internal/lambda/stream/handler_test.go | 60 ++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/internal/lambda/stream/handler_test.go b/internal/lambda/stream/handler_test.go index 7b7dd8c..8339e3b 100644 --- a/internal/lambda/stream/handler_test.go +++ b/internal/lambda/stream/handler_test.go @@ -242,6 +242,66 @@ func TestHandleRecord_SkipsRemoveRecords(t *testing.T) { } } +// TestIsTTLExpiry pins the UserIdentity shape DynamoDB's TTL deleter sets on +// records it removes itself, versus an operator- or application-driven +// delete (nil UserIdentity, or a non-Service identity such as an assumed +// role). +func TestIsTTLExpiry(t *testing.T) { + tests := []struct { + name string + userIdentity *events.DynamoDBUserIdentity + want bool + }{ + {name: "nil user identity", userIdentity: nil, want: false}, + { + name: "ttl service identity", + userIdentity: &events.DynamoDBUserIdentity{ + Type: "Service", + PrincipalID: "dynamodb.amazonaws.com", + }, + want: true, + }, + { + name: "assumed role identity", + userIdentity: &events.DynamoDBUserIdentity{Type: "AssumedRole"}, + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + record := events.DynamoDBEventRecord{UserIdentity: tt.userIdentity} + if got := isTTLExpiry(record); got != tt.want { + t.Errorf("isTTLExpiry() = %v, want %v", got, tt.want) + } + }) + } +} + +// TestHandleRecord_RemoveWithoutKeysStillErrors pins the order of operations +// in handleRecord: key extraction happens before the REMOVE guard, so a +// REMOVE record with no PK/SK still returns the "missing PK or SK" error +// instead of being silently skipped. +func TestHandleRecord_RemoveWithoutKeysStillErrors(t *testing.T) { + d := &lambda.Deps{Logger: discardLogger()} + record := events.DynamoDBEventRecord{ + EventID: "evt-remove-no-keys", + EventName: "REMOVE", + Change: events.DynamoDBStreamRecord{ + SequenceNumber: "1", + Keys: map[string]events.DynamoDBAttributeValue{}, + }, + } + + err := handleRecord(context.Background(), d, record) + if err == nil { + t.Fatal("handleRecord returned nil error, want an error for missing PK or SK") + } + if !strings.Contains(err.Error(), "missing PK or SK") { + t.Errorf("handleRecord error = %q, want it to contain %q", err.Error(), "missing PK or SK") + } +} + // TestHandleRecord_ConfigChangesInvalidateCache pins the one delete the router // must still act on: a CONFIG row that disappears has to drop out of the // cache, otherwise the router keeps triggering a pipeline whose config no From 92f3fcd9bc07538a00455d194b4a5ce4a3193a44 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Sat, 12 Sep 2026 09:43:51 +0700 Subject: [PATCH 8/9] docs: flag the known bugs carried by the deprecated stream router copy --- internal/lambda/stream_router.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/internal/lambda/stream_router.go b/internal/lambda/stream_router.go index 22b92d9..f80ccfe 100644 --- a/internal/lambda/stream_router.go +++ b/internal/lambda/stream_router.go @@ -12,6 +12,11 @@ import ( ) // Deprecated: Use stream.HandleStreamEvent instead. Retained for test compatibility. +// This copy still carries known bugs C3 (ItemIdentifier uses EventID instead +// of the record SequenceNumber) and H11 (no EventName check, so REMOVE +// records reach the write handlers). The ItemIdentifier assertions in +// stream_router_test.go encode the C3 bug rather than the correct contract; +// the live implementation is internal/lambda/stream. func HandleStreamEvent(ctx context.Context, d *Deps, event StreamEvent) (events.DynamoDBEventResponse, error) { var resp events.DynamoDBEventResponse for i := range event.Records { From 6b37c9d11a4c2cf81bb66e675eb8a26b2442b2c0 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Sat, 12 Sep 2026 09:53:57 +0700 Subject: [PATCH 9/9] docs: drop internal audit labels from changelog and deprecation note --- CHANGELOG.md | 4 ++-- internal/lambda/stream_router.go | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0eaf831..20a3981 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,8 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- **Stream router reported the wrong batch-item identifier (C3)** — `stream.HandleStreamEvent` returned `ItemIdentifier: EventID` in `BatchItemFailures`. AWS matches `ReportBatchItemFailures` identifiers against the DynamoDB stream record `SequenceNumber`, so an unrecognised identifier made Lambda treat the entire batch as failed and re-drive every record in it. The handler now returns `record.Change.SequenceNumber` and logs it alongside the event ID (the event ID remains the log correlation ID) — but partial-batch reporting only checkpoints at the *first* failed record and still re-drives that record and every later one in the batch, so this narrows the replay of non-idempotent `WriteJobEvent`/`PublishEvent` side effects rather than eliminating it; making those handlers idempotent remains a separate follow-up. -- **Stream router acted on REMOVE records (H11)** — `handleRecord` never checked `EventName`, so a deleted row reached the write handlers with a nil `NewImage`. Deleting a `RERUN_REQUEST#` row (those rows have no TTL and are never cleaned up) started a rerun with `reason="manual"`; deleting a `SENSOR#` row — including the `postrun-baseline#` delete the rerun path issues itself — could publish `POST_RUN_PASSED`/`POST_RUN_FAILED` verdicts from absent data; and 30-day `JOB#` TTL expiries invoked the function for nothing. REMOVE records are now skipped and logged at info with the key, event ID and whether the delete came from DynamoDB's TTL process, with the single exception of `SK = CONFIG`, which still invalidates the config cache so a deleted pipeline config drops out immediately. Defense in depth: both DynamoDB event source mappings in `deploy/terraform/lambda.tf` now carry `filter_criteria` (writes always pass; REMOVE passes only for the control table's `CONFIG` row), so filtered records are neither delivered nor billed. **Operators must run `terraform apply` to pick up the new event source mapping filters**; the handler guard protects deployments in the meantime. `deploy/localstack/deploy.py` applies the equivalent `FilterCriteria`, but only when a mapping is created — tear down and redeploy an existing LocalStack stack to pick them up. +- **Stream router reported the wrong batch-item identifier** — `stream.HandleStreamEvent` returned `ItemIdentifier: EventID` in `BatchItemFailures`. AWS matches `ReportBatchItemFailures` identifiers against the DynamoDB stream record `SequenceNumber`, so an unrecognised identifier made Lambda treat the entire batch as failed and re-drive every record in it. The handler now returns `record.Change.SequenceNumber` and logs it alongside the event ID (the event ID remains the log correlation ID) — but partial-batch reporting only checkpoints at the *first* failed record and still re-drives that record and every later one in the batch, so this narrows the replay of non-idempotent `WriteJobEvent`/`PublishEvent` side effects rather than eliminating it; making those handlers idempotent remains a separate follow-up. +- **Stream router acted on REMOVE records** — `handleRecord` never checked `EventName`, so a deleted row reached the write handlers with a nil `NewImage`. Deleting a `RERUN_REQUEST#` row (those rows have no TTL and are never cleaned up) started a rerun with `reason="manual"`; deleting a `SENSOR#` row — including the `postrun-baseline#` delete the rerun path issues itself — could publish `POST_RUN_PASSED`/`POST_RUN_FAILED` verdicts from absent data; and 30-day `JOB#` TTL expiries invoked the function for nothing. REMOVE records are now skipped and logged at info with the key, event ID and whether the delete came from DynamoDB's TTL process, with the single exception of `SK = CONFIG`, which still invalidates the config cache so a deleted pipeline config drops out immediately. Defense in depth: both DynamoDB event source mappings in `deploy/terraform/lambda.tf` now carry `filter_criteria` (writes always pass; REMOVE passes only for the control table's `CONFIG` row), so filtered records are neither delivered nor billed. **Operators must run `terraform apply` to pick up the new event source mapping filters**; the handler guard protects deployments in the meantime. `deploy/localstack/deploy.py` applies the equivalent `FilterCriteria`, but only when a mapping is created — tear down and redeploy an existing LocalStack stack to pick them up. ## [0.10.0] - 2026-09-12 diff --git a/internal/lambda/stream_router.go b/internal/lambda/stream_router.go index f80ccfe..61a3f02 100644 --- a/internal/lambda/stream_router.go +++ b/internal/lambda/stream_router.go @@ -12,10 +12,10 @@ import ( ) // Deprecated: Use stream.HandleStreamEvent instead. Retained for test compatibility. -// This copy still carries known bugs C3 (ItemIdentifier uses EventID instead -// of the record SequenceNumber) and H11 (no EventName check, so REMOVE +// This copy still carries known bugs (ItemIdentifier uses EventID instead +// of the record SequenceNumber; no EventName check, so REMOVE // records reach the write handlers). The ItemIdentifier assertions in -// stream_router_test.go encode the C3 bug rather than the correct contract; +// stream_router_test.go encode the EventID identifier bug rather than the correct contract; // the live implementation is internal/lambda/stream. func HandleStreamEvent(ctx context.Context, d *Deps, event StreamEvent) (events.DynamoDBEventResponse, error) { var resp events.DynamoDBEventResponse