Lambda partial-batch identifiers, dead config, and docs that taught the bug

CHANGELOG.md

@@ -212,6 +212,56 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 - `pkg/monoid/compose.ClampFuture` and `pkg/monoid/compose.DefaultSkewBound` (experimental).
 - `pkg/monoid/sketch/topk`: saturation tests documenting that a `K=32` summary drops from 32 counters covering 45,932 events to 29 counters covering 29 when a 33rd entity appears, and that the counts are Misra-Gries lower bounds with an `n/(K+1)` error bar that callers must size from an `n` the sketch does not record.
 
+- **DynamoDB Streams Lambda: partial-batch failures named an identifier
+  AWS cannot resolve.** The handler reported each failed record's
+  `eventID` as the `BatchItemFailures` ItemIdentifier, where Lambda
+  matches against the shard's **sequence numbers**. Outcome was one of:
+  the whole batch redelivered (duplicate merges — dedup is off by
+  default), a stalled iterator, or the failure silently discarded.
+  `pkg/exec/lambda/dynamodbstreams` and `examples/search-projector` now
+  report `Change.SequenceNumber`; the `eventID` keeps its real job as the
+  dedup key. `pkg/exec/lambda/kinesis` was already correct.
+- **`replay.WithDedup` promised more than it delivers.** Re-running an
+  archive folds idempotently only inside the deduper's TTL horizon:
+  claims expire, and an archive replayed after they do is
+  indistinguishable from new data and merges a second time. Documented on
+  `replay.WithDedup`, the `replay` package, and `dynamodb.NewDeduper`, and
+  pinned by tests (identical re-run → 100; re-run past a 1h TTL → 200).
+- **`recently-interacted-topk` built a K the query server could not
+  read.** `Config{K: 0}` resolved to `topk.DefaultK` (10) while the
+  Config doc, both writers' `TOPK_K` defaults, and the query server all
+  said 32. Mismatched-K Misra-Gries sketches refuse to merge, so the
+  symptom is an empty or stale Top-N rather than an error. K now resolves
+  once through `Config.ResolveK()` (default `example.DefaultK` = 32) for
+  every binary, and `cmd/query` reads `TOPK_K` instead of hard-coding 32.
+  `topk.DefaultK` is unchanged at 10.
+- **Docs that contradicted the code.** The `Trending` decay clock is
+  processing-time (`Clock` is a `func() time.Time` and never sees the
+  event), not per-record `EventTime` — so a replayed archive scores as
+  fresh. And `WithBatchWindow`'s crash story is not "dedup catches the
+  redelivery": the un-flushed records never reached the store, so the
+  redelivery is their first apply.
+
+### Removed
+
+- **BREAKING:** `windowed.Config.EventTimeField`. It was documented as
+  honored by backends and read by nothing — event time has only ever come
+  from `source.Record.EventTime`. Callers setting it were configuring
+  nothing; delete the field from your `Config` literal. To bucket by a
+  timestamp inside the payload, use the source's own `EventTime`
+  extractor (S3 / JSONL / Parquet snapshot readers take one).
+
+### Fixed
+
+- **DDB Streams Lambda docs no longer teach the `eventID` bug.** `doc/design.md` §6.2 claimed the `BatchItemFailures` `ItemIdentifier` is the `eventID` for DynamoDB Streams; it is the stream record's `SequenceNumber`, which is what Lambda resolves against the shard's checkpoint. Replaced with a per-source table and an explanation of why the two identifiers are not interchangeable (`eventID` feeds the `Deduper`; the sequence number is the cursor). The copy-pasteable Lambda handler in `doc/search-integration.md` had the same defect in code form and now reports `rec.Change.SequenceNumber`.
+- **Empty `SequenceNumber` no longer produces an empty `ItemIdentifier`.** Lambda treats a null or empty `itemIdentifier` as a malformed response and redelivers the entire batch, re-merging every record that had already succeeded. `pkg/exec/lambda/dynamodbstreams` and `examples/search-projector` now drop the unreportable entry and surface it — via `metrics.RecordError` plus a `<name>:unreportable_failure` event in the handler, and via a new `Stats.Unreportable` counter in the example.
+- **`WithBatchWindow` crash-safety documentation corrected.** The `pkg/exec/streaming` option doc, `doc/design.md` §5.4, §14.1, §14.4 and the failure-mode diagram all asserted crash safety on the premise that the dedup claim is taken at flush time. It is taken in `aggregator.accept`, when the record enters the accumulator. The consequence — with `WithDedup`, a crash before the flush loses the accumulated batch, because the surviving claim suppresses the very redelivery that would restore it; without a `Deduper`, the redelivery is the records' first apply and nothing is lost — is now stated plainly, along with the contrast to `processor.MergeOne`'s release-on-failed-merge.
+
+### Changed
+
+- `examples/recently-interacted-topk/multisource_test.go` drives the example's real `Build()` instead of a hand-rolled copy that had drifted to `K=10` against the deployment's `K=32`, and asserts the built sketch's K against `Config.ResolveK()`.
+- The replay dedup-TTL contract moved from a unit test asserting against its own hand-rolled expiring fake to `test/e2e/replay_dedup_ttl_test.go`, which exercises the real `pkg/state/dynamodb.Deduper` behind the `DDB_LOCAL_ENDPOINT` gate.
+
 ### Fixed — graceful shutdown silently lost every in-flight record
 
 `streaming.Run` treated a cancelled context as a poison record. The comment at

STABILITY.md

@@ -32,7 +32,7 @@ edges callers should plan around.
 | `pkg/exec/replay` | mostly stable | Shares the `pkg/exec/processor` core. Same retry / dead-letter / `KeyByMany` semantics as bootstrap. metrics.Recorder fully wired; the historical "metrics integration not yet wired" note is fixed |
 | `pkg/exec/batch/sparkconnect` | experimental | own Go submodule (separate `go.mod`) so root `github.com/gallowaysoftware/murmur` doesn't pull `apache/spark-connect-go`. Consumers who DO depend on this submodule must mirror its `replace` line for the `pequalsnp/spark-connect-go` fork in their own `go.mod` |
 | `pkg/exec/lambda/kinesis` | experimental | `NewHandler` returns the Lambda Kinesis handler signature; partial-batch failures via BatchItemFailures; pair with `WithDedup` so adjacent-redelivered records fold idempotently |
-| `pkg/exec/lambda/dynamodbstreams` | experimental | DDB Streams Lambda handler; same retry/dedup/BatchItemFailures shape as the Kinesis variant. Decoder takes the whole change record so callers can branch on EventName / inspect OldImage |
+| `pkg/exec/lambda/dynamodbstreams` | experimental | DDB Streams Lambda handler; same retry/dedup/BatchItemFailures shape as the Kinesis variant. Decoder takes the whole change record so callers can branch on EventName / inspect OldImage. BatchItemFailures report the record's SequenceNumber (what Lambda checkpoints on); the eventID feeds dedup only. A failed record with an empty SequenceNumber gets no entry at all (an empty ItemIdentifier makes Lambda redeliver the whole batch) and is surfaced via metrics.RecordError plus a `<name>:unreportable_failure` event |
 | `pkg/exec/lambda/sqs` | experimental | SQS Lambda handler; same shape as kinesis/dynamodbstreams. Default EventID is "<arn>/<MessageId>"; override via WithEventID for FIFO content-dedup or upstream-key dedup. Uses SQS SentTimestamp for windowed-bucket assignment so delayed deliveries land in the correct bucket |
 | `pkg/query` | mostly stable | `Get` / `GetWindow` / `GetRange` / `LambdaQuery` are likely v1 surface |
 | `pkg/query/grpc` | mostly stable | generic byte-encoded responses; `cmd/murmur-codegen-typed` emits per-service typed `.proto` + Go server stubs (sum / hll / topk / bloom; get_all_time / get_window / get_window_many / get_many / get_range) over `pkg/query/typed` clients. `HealthHandler` serves `grpc.health.v1.Health` and `HealthzHandler` serves `/healthz` (liveness, always 200) + `/readyz` (readiness, store round-trip, cached so probe traffic is not billed reads) |

doc/design.md

@@ -718,24 +718,24 @@ layer. The DSL split is invisible at the boundary.
 
 ### 4.6 Windowing: bucket math and TTL integration
 
-`pkg/monoid/windowed.Config` is two fields and a granularity:
+`pkg/monoid/windowed.Config` is two fields:
 
 ```go
 type Config struct {
-    Granularity    time.Duration
-    Retention      time.Duration
-    EventTimeField string
+    Granularity time.Duration
+    Retention   time.Duration
 }
 ```
 
 `Granularity` is the bucket size — 24h for daily, 1h for hourly, 1m for
 per-minute. `Retention` is how long buckets persist before DDB TTL
-evicts them. `EventTimeField`, when set, names a struct field on the
-record from which the runtime extracts a timestamp; when empty,
-processing-time is used.
+evicts them. The timestamp a record is bucketed by is not configured
+here: it is `source.Record.EventTime`, which every runtime hands to
+`BucketID`, falling back to the wall clock only for a record that
+carries no event time.
 
 Bucket assignment is `BucketID(t) = t.UnixNano() / Granularity.Nanoseconds()`
-(`pkg/monoid/windowed/windowed.go:54-60`). Buckets are tumbling and
+(`pkg/monoid/windowed/windowed.go:51-56`). Buckets are tumbling and
 aligned to the Unix epoch. The implication: bucket 0 is "the first
 bucket since 1970" for any granularity, not "the bucket containing
 midnight today." This matters for queries — `GetWindow(now,
@@ -1040,7 +1040,7 @@ isn't in the box."
 
 The single optional concession to a non-trivial runtime is
 `streaming.WithBatchWindow(window, maxBatch)`
-(`pkg/exec/streaming/runtime.go:78`). It enables a per-(entity, bucket)
+(`pkg/exec/streaming/runtime.go`). It enables a per-(entity, bucket)
 delta accumulator: instead of issuing one MergeUpdate per record, the
 runtime accumulates deltas in memory for `window` time, then flushes a
 single MergeUpdate per key.
@@ -1061,8 +1061,13 @@ The trade is real:
   side for the read-your-writes case.
 - **Crash durability.** Records are Ack'd to the source AFTER the
   batch flushes. A worker crash loses up to `window`-worth of in-flight
-  records, which the source replays on restart. Dedup catches the
-  redelivery.
+  records, which the source replays on restart. Whether that replay
+  restores them turns on `WithDedup`, and not in the direction the
+  word "dedup" suggests: the aggregator claims each EventID when the
+  record ENTERS the accumulator, so the claim survives the crash, the
+  replay is dedup-skipped, and the batch's contribution is gone. Run
+  without a Deduper and the replay is the records' first apply, so
+  nothing is lost. Section 14.1 walks both cases.
 - **Memory.** At most `maxBatch` records per (entity, bucket) before
   a forced flush, but the *number of concurrent keys* is unbounded.
   For high-cardinality pipelines (per-user keys, with long-tail
@@ -1101,9 +1106,12 @@ canceled or the source is exhausted. A `Record[T]` carries:
 - `Value T` — the decoded record body.
 - `EventTime time.Time` — used for windowed bucket assignment.
   Sources fill this from their native timestamp (Kafka record
-  timestamp, Kinesis ApproximateArrivalTimestamp, SQS SentTimestamp);
-  the user's value extractor can override via the `EventTimeField`
-  windowing config.
+  timestamp, Kinesis ApproximateArrivalTimestamp, SQS SentTimestamp).
+  This is the ONLY event-time input to bucketing; a record that arrives
+  with a zero EventTime is bucketed by the runtime's clock instead. To
+  bucket by a timestamp carried inside the payload, use the source's
+  `EventTime` extractor (the S3 / JSONL / Parquet snapshot readers all
+  take one) — the windowing config has no say in it.
 - `Ack func() error` — called when the record has been successfully
   processed (or duplicate-skipped). For Kafka, this marks the offset
   for commit; for Kinesis, this advances the per-shard checkpoint;
@@ -1316,25 +1324,51 @@ poison-pill semantics are uniform across all three Lambda variants.
 `FunctionResponseTypes=["ReportBatchItemFailures"]` on the
 event-source mapping), Lambda redelivers only the failed records, not
 the whole batch. The Murmur Lambda handlers populate the
-`BatchItemFailures` slice with the EventIDs of records that exhausted
-their retry budget; Lambda redelivers those on the next invocation.
-This is the single most operationally important detail of the Lambda
-runtime — without it, one bad record retries the entire batch
-indefinitely. With it, one bad record is dead-lettered and the batch
-proceeds.
-
-The `ItemIdentifier` for `BatchItemFailures` differs by source:
-Kinesis uses the record `SequenceNumber`, DDB Streams uses
-`EventID`, SQS uses `MessageId`. The handlers fill the right shape
-for each.
+`BatchItemFailures` slice with the *checkpoint identifier* of records
+that exhausted their retry budget; Lambda redelivers those on the next
+invocation. This is the single most operationally important detail of
+the Lambda runtime — without it, one bad record retries the entire
+batch indefinitely. With it, one bad record is dead-lettered and the
+batch proceeds.
+
+The `ItemIdentifier` is whatever the event source checkpoints on, which
+is not the same thing as the identity Murmur dedups on:
+
+| Source | `ItemIdentifier` | Go field |
+| --- | --- | --- |
+| Kinesis | record sequence number | `rec.Kinesis.SequenceNumber` |
+| DDB Streams | stream-record sequence number | `rec.Change.SequenceNumber` |
+| SQS | message ID | `msg.MessageId` |
+
+The DDB Streams row is the one that reads wrong at a glance. A change
+record also carries an `eventID`, and it is the record's unique name —
+but it is not a cursor. Lambda resolves an `ItemIdentifier` against the
+shard's sequence numbers, so an `eventID` there names nothing it can
+find, and the mapping degrades to whole-batch redelivery, a stalled
+iterator, or a failure discarded outright. The `eventID` is still what
+feeds the `Deduper` (point 3 below); record identity and checkpoint
+cursor are two different jobs and the handlers fill both.
+
+A record whose checkpoint identifier is empty cannot be reported at
+all: Lambda treats a null or empty `itemIdentifier` in the response as
+malformed and redelivers the WHOLE batch, which for a non-idempotent
+monoid without dedup means re-merging every record that had already
+succeeded. Rather than hand Lambda that, the DDB Streams handler drops
+the entry and surfaces the record through `metrics.RecordError` plus a
+`<name>:unreportable_failure` event — the same "never hand Lambda a
+redelivery loop" policy as the poison-pill path above. Real DDB
+Streams records always carry a sequence number; an empty one means a
+synthetic or hand-constructed event.
 
 **3. Dedup-friendly EventID shapes.** Each variant produces an
 EventID format suitable for `Deduper.MarkSeen`:
 
 - Kinesis: `<stream>/<shard>/<sequenceNumber>` — globally unique
   across the stream's lifetime.
 - DDB Streams: the stream record's `EventID` field — unique per
-  stream record, ordering preserved within a shard.
+  stream record, ordering preserved within a shard. This is the only
+  place `EventID` is used; the `BatchItemFailures` identifier above is
+  the sequence number.
 - SQS: `<arn>/<MessageId>` by default, override-able.
 
 The dedup contract is the same across all three: pass
@@ -2574,7 +2608,8 @@ flowchart TB
       WC["Worker crashes"]
       WC --> NoAck["In-flight records not Ack'd"]
       NoAck --> Replay["Source replays on restart"]
-      Replay --> Dedup2["Dedup catches duplicates"]
+      Replay --> Dedup2["Unbatched: dedup catches<br/>the duplicate re-apply"]
+      Replay --> Lost["WithBatchWindow + WithDedup:<br/>claim taken on accept suppresses<br/>the replay — batch lost (14.1)"]
     end
     subgraph Storage["Storage failures"]
       DDBT["DDB throttle / unavailable"]
@@ -2595,13 +2630,36 @@ A worker crash mid-batch:
 - Records in the source's in-flight buffer: redelivered on restart,
   caught by dedup if configured.
 - Records in the `WithBatchWindow` accumulator: lost from the
-  accumulator, but redelivered by the source (since they weren't
-  Ack'd), and re-aggregated on restart. Dedup catches the first
-  redelivery; the second is the real apply.
-
-The result: at-least-once with no data loss, modulo the edge case
-where dedup is disabled for a non-idempotent monoid (then crashes
-double-count by the in-flight buffer's worth of records).
+  accumulator and redelivered by the source (they were never Ack'd —
+  the aggregator defers each record's Ack until its batch flushes).
+  Whether the redelivery restores them depends on whether a `Deduper`
+  is wired, and today the two cases differ:
+  - **No `WithDedup`:** the records never reached the store, so the
+    redelivery is their first and only apply. Nothing is lost and
+    nothing is double-counted.
+  - **With `WithDedup`:** the claim is taken in `aggregator.accept`,
+    when the record enters the accumulator — not at flush. The claim
+    outlives the crash — with the recommended
+    `pkg/state/dynamodb.Deduper` it is a durable table row — so the
+    redelivery is dedup-skipped and the in-flight batch's contribution
+    is lost for good. `WithBatchWindow` + `WithDedup` therefore has a data-loss
+    window of up to one flush interval (or `maxBatch` records per key,
+    whichever comes first) per crash.
+
+  This is the opposite trade from the unbatched path, where
+  `processor.MergeOne` releases the claim on a detached context when a
+  merge fails, so the redelivery is re-applied rather than skipped. The
+  aggregator has no equivalent release: `flushOne` dead-letters and
+  Acks a batch whose merge exhausted its retries but leaves the claims
+  standing, so a manual replay of the dead-lettered EventIDs is also
+  suppressed until the dedup TTL expires.
+
+The result for the unbatched path: at-least-once with no data loss,
+modulo the edge case where dedup is disabled for a non-idempotent
+monoid — a crash between a flush and the Acks it releases re-applies
+that batch on restart. Under `WithBatchWindow` the guarantee is
+weaker in exactly the way above, and choosing between a lost window
+and a double-counted one is currently the operator's call.
 
 ### 14.2 DDB throttles or is unavailable
 
@@ -2649,7 +2707,10 @@ recovery for known-popular keys.
 - SQS (Lambda): same as DDB Streams.
 
 In all cases, dedup catches re-deliveries that arrive during the
-reconnect window.
+reconnect window — with the `WithBatchWindow` caveat from 14.1: there
+the claim is taken on accept, so dedup suppresses the re-delivery
+rather than absorbing a duplicate, and whatever the accumulator was
+holding is lost.
 
 ### 14.5 The handoff token is lost or corrupted
 

doc/search-integration.md

@@ -826,9 +826,20 @@ func main() {
                 Body:       strings.NewReader(body),
             }
             if _, err := osClient.Update(ctx, updateReq); err != nil {
-                resp.BatchItemFailures = append(resp.BatchItemFailures, events.DynamoDBBatchItemFailure{
-                    ItemIdentifier: rec.EventID,
-                })
+                // ItemIdentifier is the SequenceNumber, NOT the eventID.
+                // Lambda resolves this against the shard's sequence
+                // numbers; an eventID names nothing it can find, and the
+                // failure is either dropped or escalated to a whole-batch
+                // redelivery. An empty identifier is worse still — Lambda
+                // rejects the response as malformed and redelivers the
+                // whole batch — so skip the entry rather than emit one.
+                if seq := rec.Change.SequenceNumber; seq != "" {
+                    resp.BatchItemFailures = append(resp.BatchItemFailures, events.DynamoDBBatchItemFailure{
+                        ItemIdentifier: seq,
+                    })
+                } else {
+                    log.Printf("unreportable failure: %v (eventID=%s has no SequenceNumber)", err, rec.EventID)
+                }
             }
         }
         return resp, nil