diff --git a/CHANGELOG.md b/CHANGELOG.md index 662cdd1..20a3981 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** — `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 ### Added 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") 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 diff --git a/internal/lambda/stream/handler.go b/internal/lambda/stream/handler.go index 8e736e5..651895a 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 @@ -27,9 +42,15 @@ 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 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].EventID, + ItemIdentifier: event.Records[i].Change.SequenceNumber, }) } } @@ -43,6 +64,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 new file mode 100644 index 0000000..8339e3b --- /dev/null +++ b/internal/lambda/stream/handler_test.go @@ -0,0 +1,368 @@ +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" + "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") + } +} + +// 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) + } + }) + } +} + +// 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 +// 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/lambda/stream_router.go b/internal/lambda/stream_router.go index 22b92d9..61a3f02 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 (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 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 for i := range event.Records { 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 (