fix: stream-router contract — SequenceNumber batch failures and REMOVE record guard

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

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")
 

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

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)