From d03adc437e4a0dd7d535e35bbf4507ddab042467 Mon Sep 17 00:00:00 2001 From: Kyle Galloway Date: Fri, 28 Aug 2026 10:51:56 -0300 Subject: [PATCH 1/4] Report the sequence number Lambda checkpoints on, not the eventID MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A DDB Streams BatchItemFailure is resolved against the shard's sequence numbers. Both handlers reported the record's eventID there, which names nothing Lambda can find: the batch is redelivered whole (duplicate merges, and dedup defaults to off), left on a stalled iterator, or the failure is discarded. kinesis.go already had this right. The eventID keeps its real job — it is what feeds the Deduper. The fixtures now carry an eventID and a realistic numeric SequenceNumber that share no characters, so an assertion on the wrong identifier cannot pass by coincidence. Alongside, four claims the code never honored: - windowed.Config.EventTimeField was read by nothing. Event time comes from source.Record.EventTime and always has. Deleted (breaking). - replay.WithDedup had no coverage at any level and promised idempotent re-runs without qualification. Two tests pin the contract: identical archives fold to one, and the same archive past the deduper's TTL merges twice — 200, not 100 — because an evicted claim is indistinguishable from a record never seen. - The recently-interacted example documented K=32 while Build() resolved a zero K to topk.DefaultK (10), under a query server hard-coded to 32. Mismatched-K sketches refuse to merge, so that lands as an empty Top-N rather than an error. K now resolves once, through Config.ResolveK(), for every binary; the example's default is the 32 its deployment has always run. - WithBatchWindow's doc said a crash loses at most a window of records because "dedup catches the redelivery". It does not: those records never reached the store, so the redelivery is their first apply, and a claim taken before the flush would suppress it. Prose only. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01X2YMxeLgyRc9i5XPyEV75S --- STABILITY.md | 2 +- doc/design.md | 37 +++-- doc/use-cases.md | 28 +++- examples/recently-interacted-topk/README.md | 6 + .../cmd/lambda/main.go | 2 +- .../cmd/query/main.go | 24 ++- .../cmd/worker/main.go | 2 +- examples/recently-interacted-topk/pipeline.go | 24 ++- .../recently-interacted-topk/pipeline_test.go | 74 +++++++++ examples/search-projector/projector.go | 7 +- examples/search-projector/projector_test.go | 35 +++- .../lambda/dynamodbstreams/dynamodbstreams.go | 20 ++- .../dynamodbstreams/dynamodbstreams_test.go | 60 ++++--- pkg/exec/replay/runtime.go | 11 +- pkg/exec/replay/runtime_test.go | 151 ++++++++++++++++++ pkg/exec/streaming/runtime.go | 10 +- pkg/monoid/windowed/windowed.go | 10 +- pkg/murmur/lambda_test.go | 64 ++++++++ pkg/murmur/preset.go | 11 +- pkg/state/dynamodb/dedup.go | 6 + 20 files changed, 511 insertions(+), 73 deletions(-) create mode 100644 examples/recently-interacted-topk/pipeline_test.go diff --git a/STABILITY.md b/STABILITY.md index 158c347..498d4bc 100644 --- a/STABILITY.md +++ b/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 | | `pkg/exec/lambda/sqs` | experimental | SQS Lambda handler; same shape as kinesis/dynamodbstreams. Default EventID is "/"; 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) | diff --git a/doc/design.md b/doc/design.md index 8539d94..39d5c69 100644 --- a/doc/design.md +++ b/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, @@ -1101,9 +1101,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; @@ -2596,12 +2599,16 @@ A worker crash mid-batch: 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. + Ack'd), and aggregated for the first time on restart. Dedup is not + the mechanism here — those records never reached the store, so the + redelivery is their apply. This holds only while the dedup claim is + taken at flush time: a claim taken when the record enters the + accumulator outlives the crash, suppresses the redelivery, and the + whole in-flight batch is silently lost. 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). +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. ### 14.2 DDB throttles or is unavailable diff --git a/doc/use-cases.md b/doc/use-cases.md index 19f9412..83b0982 100644 --- a/doc/use-cases.md +++ b/doc/use-cases.md @@ -220,9 +220,15 @@ an event's contribution drops by half. - Decay associativity is approximate (FP-rounding-bounded); see `STABILITY.md`'s row on `pkg/monoid/compose`. -- The "now" used for decay comes from `EventTime` on each record, - not processing-time. Late-arriving events still get their original - decay applied — they don't artificially count as "fresh." +- The "now" stamped on each observation is processing-time: the + builder calls its `Clock` (default `time.Now`) when the value + extractor runs, and that hook is a `func() time.Time` — it gets no + access to the event, so it cannot read `EventTime`. A late-arriving + event therefore decays from when it was *processed*, not when it + happened. Replaying a week-old archive through a Trending pipeline + scores every one of those events as fresh; use `Clock` to pin the + replay's timestamp, or aggregate with a windowed Sum and apply decay + at query time, when you can decay from the bucket's own time. --- @@ -246,10 +252,18 @@ murmur.MustKinesisHandler(buildPipeline(clickDecoder)) murmur.RunStreamingWorker(ctx, buildPipeline(purchaseDecoder)) ``` -Both call `buildPipeline()` with the same monoid (`topk.New(10)`), -same key extractor (`func(e Event) string { return e.CategoryID }`), -same DDB table. The store's `MergeUpdate` semantics ensure neither -worker's contribution is lost. +Both call `buildPipeline()` with the same monoid (`topk.New(k)` for +one shared `k` — the example resolves it once via `Config.ResolveK()`, +default 32), same key extractor +(`func(e Event) string { return e.CategoryID }`), same DDB table. The +store's `MergeUpdate` semantics ensure neither worker's contribution is +lost. + +The K has to be identical in every binary that touches the row, query +server included: sketches sized for different K refuse to merge, and +that shows up as an empty Top-N rather than an error. Note the example's +32 is its own choice — `topk.DefaultK`, what `topk.TopK()` and +`topk.Single()` use, is 10. **Runnable example.** [`examples/recently-interacted-topk/`](../examples/recently-interacted-topk/). diff --git a/examples/recently-interacted-topk/README.md b/examples/recently-interacted-topk/README.md index c9c5d5b..504fde7 100644 --- a/examples/recently-interacted-topk/README.md +++ b/examples/recently-interacted-topk/README.md @@ -48,6 +48,12 @@ pipeline.NewPipeline[Interaction, []byte]("recently_interacted"). The Lambda binary leaves `Source` unset (Lambda owns polling). The Kafka worker calls `AttachKafkaSource(pipe, cfg)` to attach a franz-go source. +`K` is one number for the whole deployment: `Config.ResolveK()`, which is +`TOPK_K` in each binary's environment and `DefaultK` (32) when unset. Both +writers and the query server go through it. Sketches sized for different K +refuse to merge, and the symptom is an empty or stale Top-N at query time +rather than an error at startup — so never hard-code K anywhere. + ## Run locally Stand up dependencies: diff --git a/examples/recently-interacted-topk/cmd/lambda/main.go b/examples/recently-interacted-topk/cmd/lambda/main.go index 580f1dc..b734c07 100644 --- a/examples/recently-interacted-topk/cmd/lambda/main.go +++ b/examples/recently-interacted-topk/cmd/lambda/main.go @@ -67,7 +67,7 @@ func run() int { DDBRegion: envOr("AWS_REGION", "us-east-1"), DDBDedupTable: os.Getenv("DDB_DEDUP_TABLE"), // recommended in production DedupTTL: 24 * time.Hour, - K: envU32("TOPK_K", 32), + K: envU32("TOPK_K", example.DefaultK), WindowRetention: 30 * 24 * time.Hour, } diff --git a/examples/recently-interacted-topk/cmd/query/main.go b/examples/recently-interacted-topk/cmd/query/main.go index a212a87..3be4322 100644 --- a/examples/recently-interacted-topk/cmd/query/main.go +++ b/examples/recently-interacted-topk/cmd/query/main.go @@ -28,6 +28,7 @@ import ( "net/http" "os" "os/signal" + "strconv" "syscall" "time" @@ -46,6 +47,7 @@ func run() int { DDBEndpoint: os.Getenv("DDB_ENDPOINT"), DDBTable: envOr("DDB_TABLE", "recently_interacted"), DDBRegion: envOr("AWS_REGION", "us-east-1"), + K: envU32("TOPK_K", example.DefaultK), WindowRetention: 30 * 24 * time.Hour, } addr := envOr("GRPC_ADDR", ":50051") @@ -65,8 +67,11 @@ func run() int { window := windowed.Daily(cfg.WindowRetention) srv := mgrpc.NewServer(mgrpc.Config[[]byte]{ - Store: store, - Monoid: topk.New(32), // K must match the Build() default; pin via env in production + Store: store, + // Same Config, same ResolveK: a query server whose K differs from the + // writers' reads a sketch that refuses to merge, and the Top-N comes + // back empty instead of erroring. + Monoid: topk.New(cfg.ResolveK()), Window: &window, Encode: mgrpc.BytesIdentity(), }) @@ -121,3 +126,18 @@ func envOr(key, fallback string) string { } return fallback } + +// envU32 mirrors the writers' TOPK_K parsing so one env var pins K across +// every binary in the deployment. +func envU32(key string, fallback uint32) uint32 { + v := os.Getenv(key) + if v == "" { + return fallback + } + n, err := strconv.ParseUint(v, 10, 32) + if err != nil || n == 0 { + slog.Warn("invalid TopK size, using default", "key", key, "value", v, "default", fallback, "err", err) + return fallback + } + return uint32(n) +} diff --git a/examples/recently-interacted-topk/cmd/worker/main.go b/examples/recently-interacted-topk/cmd/worker/main.go index d3eae5e..ae1d408 100644 --- a/examples/recently-interacted-topk/cmd/worker/main.go +++ b/examples/recently-interacted-topk/cmd/worker/main.go @@ -57,7 +57,7 @@ func run() int { DDBRegion: envOr("AWS_REGION", "us-east-1"), DDBDedupTable: os.Getenv("DDB_DEDUP_TABLE"), DedupTTL: 24 * time.Hour, - K: envU32("TOPK_K", 32), + K: envU32("TOPK_K", example.DefaultK), WindowRetention: 30 * 24 * time.Hour, } diff --git a/examples/recently-interacted-topk/pipeline.go b/examples/recently-interacted-topk/pipeline.go index 4398945..2d00b72 100644 --- a/examples/recently-interacted-topk/pipeline.go +++ b/examples/recently-interacted-topk/pipeline.go @@ -84,7 +84,7 @@ type Config struct { ConsumerGroup string // TopK parameters - K uint32 // sketch size (memory ~K; default 32 if zero) + K uint32 // sketch size (memory ~K); zero means DefaultK WindowRetention time.Duration // daily-bucket retention (default 30d) // Metrics, when set, is handed to the byte store so CAS contention lands @@ -95,6 +95,23 @@ type Config struct { Metrics metrics.Recorder } +// DefaultK is the K the example builds with when Config.K is left zero. It is +// deliberately larger than topk.DefaultK: the deployed writers (TOPK_K) and +// the query server have always used 32, and a library caller who omitted K +// used to get a K=10 store read through a K=32 query server. +const DefaultK uint32 = 32 + +// ResolveK returns the K this Config actually builds with. Every binary in the +// example goes through it — writers and the query server alike — because +// sketches with mismatched K refuse to merge, and the failure shows up at +// query time as an empty or stale Top-N rather than as an error at startup. +func (c Config) ResolveK() uint32 { + if c.K == 0 { + return DefaultK + } + return c.K +} + // PipelineName is the canonical pipeline identifier — surfaces in metrics, // the admin UI, and the auto-generated query service. const PipelineName = "recently_interacted" @@ -110,10 +127,7 @@ func Build(ctx context.Context, cfg Config) (*pipeline.Pipeline[Interaction, []b if cfg.DDBTable == "" { return nil, nil, nil, errors.New("recently-interacted: DDBTable is required") } - k := cfg.K - if k == 0 { - k = topk.DefaultK - } + k := cfg.ResolveK() retention := cfg.WindowRetention if retention == 0 { retention = 30 * 24 * time.Hour diff --git a/examples/recently-interacted-topk/pipeline_test.go b/examples/recently-interacted-topk/pipeline_test.go new file mode 100644 index 0000000..fd6da83 --- /dev/null +++ b/examples/recently-interacted-topk/pipeline_test.go @@ -0,0 +1,74 @@ +package recentlyinteracted_test + +import ( + "context" + "encoding/binary" + "testing" + + example "github.com/gallowaysoftware/murmur/examples/recently-interacted-topk" +) + +// sketchK reads the K a marshaled Misra-Gries sketch was sized for. The wire +// format opens with a little-endian uint32 K (pkg/monoid/sketch/topk). +func sketchK(t *testing.T, b []byte) uint32 { + t.Helper() + if len(b) < 4 { + t.Fatalf("sketch too short to carry a K header: %d bytes", len(b)) + } + return binary.LittleEndian.Uint32(b[:4]) +} + +// TestBuild_KDefaultsToDocumentedSize pins the K a caller gets when Config.K +// is left zero. Sketches with mismatched K refuse to merge, so a documented +// default that disagrees with the built one is not cosmetic: it puts a K=10 +// store under the K=32 query server this example ships, and the Top-N comes +// back empty rather than erroring. +func TestBuild_KDefaultsToDocumentedSize(t *testing.T) { + // 32 is the number the README, the Config doc, the TOPK_K default in both + // writer binaries, and the query server all name. + const documentedK uint32 = 32 + + if example.DefaultK != documentedK { + t.Fatalf("example.DefaultK = %d, want the documented %d", example.DefaultK, documentedK) + } + + cases := []struct { + name string + cfg uint32 + want uint32 + }{ + {"zero takes the documented default", 0, documentedK}, + {"explicit K is honored", 7, 7}, + {"explicit K equal to the default", documentedK, documentedK}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + cfg := example.Config{ + DDBTable: "recently_interacted_test", + DDBRegion: "us-east-1", + DDBEndpoint: "http://127.0.0.1:8000", // no call is made; keeps creds static + K: tc.cfg, + } + pipe, store, _, err := example.Build(context.Background(), cfg) + if err != nil { + t.Fatalf("Build: %v", err) + } + t.Cleanup(func() { _ = store.Close() }) + + // The per-event delta and the aggregating monoid must agree with + // each other AND with what the query server would construct from + // the same Config. + delta := sketchK(t, pipe.ValueFn()(example.Interaction{EntityID: "entity-1"})) + if delta != tc.want { + t.Errorf("value extractor built a K=%d sketch, want K=%d", delta, tc.want) + } + if got := sketchK(t, pipe.Monoid().Identity()); got != tc.want { + t.Errorf("aggregate monoid is K=%d, want K=%d", got, tc.want) + } + if got := cfg.ResolveK(); got != tc.want { + t.Errorf("ResolveK (what the query server builds with) = %d, want %d", got, tc.want) + } + }) + } +} diff --git a/examples/search-projector/projector.go b/examples/search-projector/projector.go index 9129e36..61fd428 100644 --- a/examples/search-projector/projector.go +++ b/examples/search-projector/projector.go @@ -173,12 +173,17 @@ func (p *Projector) Handle(ctx context.Context, rec *events.DynamoDBEventRecord) // HandleEvent is the convenience handler for a full SQS-style batch. // Returns the BatchItemFailures slice for Lambda's response shape. +// +// The ItemIdentifier is the record's SequenceNumber, which is what Lambda +// resolves against the shard's checkpoint. Reporting the eventID instead +// names nothing Lambda can find, and the failed record is either redelivered +// as part of the whole batch or dropped outright. func (p *Projector) HandleEvent(ctx context.Context, evt events.DynamoDBEvent) []events.DynamoDBBatchItemFailure { var failures []events.DynamoDBBatchItemFailure for i := range evt.Records { if err := p.Handle(ctx, &evt.Records[i]); err != nil { failures = append(failures, events.DynamoDBBatchItemFailure{ - ItemIdentifier: evt.Records[i].EventID, + ItemIdentifier: evt.Records[i].Change.SequenceNumber, }) } } diff --git a/examples/search-projector/projector_test.go b/examples/search-projector/projector_test.go index 66bb078..8e17fbc 100644 --- a/examples/search-projector/projector_test.go +++ b/examples/search-projector/projector_test.go @@ -50,10 +50,15 @@ func (f *fakeIndex) snapshot() []indexedDoc { // makeRecord constructs a DDB Streams record carrying a counter // transition for entity `pk` going from `oldV` to `newV`. +// +// EventID and SequenceNumber deliberately look nothing alike: Lambda +// checkpoints on the sequence number, so a fixture that let the two collide +// would let a handler report the useless one and still pass. func makeRecord(pk string, oldV, newV int64, hasOld bool) events.DynamoDBEventRecord { rec := events.DynamoDBEventRecord{ EventID: "ev-" + pk + "-" + strconv.FormatInt(newV, 10), Change: events.DynamoDBStreamRecord{ + SequenceNumber: seqFor(pk, newV), Keys: map[string]events.DynamoDBAttributeValue{ "pk": events.NewStringAttribute(pk), }, @@ -70,6 +75,17 @@ func makeRecord(pk string, oldV, newV int64, hasOld bool) events.DynamoDBEventRe return rec } +// seqFor returns a realistic DynamoDB Streams sequence number: a long numeric +// string that shares no characters with the fixtures' eventIDs. +func seqFor(pk string, newV int64) string { + var sum int64 + for _, c := range pk { + sum += int64(c) + } + return "4959033827149025660855969253361571095921" + + strconv.FormatInt(1_000_000+sum*1000+newV, 10) +} + func TestProjector_BucketTransitionTriggersIndex(t *testing.T) { idx := &fakeIndex{} p := projector.New(projector.Config{Index: "posts"}, idx) @@ -208,18 +224,27 @@ func TestProjector_OpenSearchFailureReportsToBatchItemFailures(t *testing.T) { idx := &fakeIndex{err: errors.New("opensearch 503")} p := projector.New(projector.Config{Index: "posts"}, idx) - evt := events.DynamoDBEvent{Records: []events.DynamoDBEventRecord{ + records := []events.DynamoDBEventRecord{ makeRecord("post-A", 999, 1000, true), // would index → fails makeRecord("post-B", 100, 500, true), // skipped (same bucket) - }} - failures := p.HandleEvent(context.Background(), evt) + } + failures := p.HandleEvent(context.Background(), events.DynamoDBEvent{Records: records}) // Only the failing record should appear in BatchItemFailures. if len(failures) != 1 { t.Fatalf("BatchItemFailures: got %d, want 1", len(failures)) } - if failures[0].ItemIdentifier != "ev-post-A-1000" { - t.Errorf("ItemIdentifier: got %q", failures[0].ItemIdentifier) + // The identifier must be the sequence number Lambda checkpoints on. An + // eventID resolves to nothing on the shard, which costs a whole-batch + // redelivery or a dropped failure. + got := failures[0].ItemIdentifier + if want := records[0].Change.SequenceNumber; got != want { + t.Errorf("ItemIdentifier: got %q, want the record's SequenceNumber %q", got, want) + } + for _, rec := range records { + if got == rec.EventID { + t.Errorf("ItemIdentifier is eventID %q; Lambda checkpoints on sequence numbers", rec.EventID) + } } } diff --git a/pkg/exec/lambda/dynamodbstreams/dynamodbstreams.go b/pkg/exec/lambda/dynamodbstreams/dynamodbstreams.go index 5a52ccf..57f46ba 100644 --- a/pkg/exec/lambda/dynamodbstreams/dynamodbstreams.go +++ b/pkg/exec/lambda/dynamodbstreams/dynamodbstreams.go @@ -47,10 +47,17 @@ // # Partial-batch failure handling // // Records that exhaust their retry budget are reported via BatchItemFailures -// with the DDB Streams `eventID` as ItemIdentifier. Configure your -// event-source mapping with `FunctionResponseTypes=["ReportBatchItemFailures"]` -// so Lambda only redelivers the failures (or, in shard-order replay mode, -// all records from the earliest failure forward). +// with the record's `SequenceNumber` as ItemIdentifier — NOT its `eventID`. +// Lambda resolves an ItemIdentifier against the shard's sequence numbers, so +// an eventID there names nothing it can find: depending on the event-source +// mapping that degrades to whole-batch redelivery, a stalled iterator, or a +// silently discarded failure. The eventID is still what feeds the Deduper — +// it is the stream-unique record identity, just not the checkpoint cursor. +// +// Configure your event-source mapping with +// `FunctionResponseTypes=["ReportBatchItemFailures"]` so Lambda only +// redelivers the failures (or, in shard-order replay mode, all records from +// the earliest failure forward). package dynamodbstreams import ( @@ -230,8 +237,11 @@ func NewHandler[T any, V any]( if err := processor.MergeMany(ctx, &cfg.Config, name, rec.EventID, eventTime, keysFn(value), valueFn(value), store, cacheStore, window); err != nil { + // SequenceNumber, not EventID: Lambda checkpoints the shard by + // sequence number, and an identifier it can't resolve is either + // ignored or takes the whole batch down with it. resp.BatchItemFailures = append(resp.BatchItemFailures, events.DynamoDBBatchItemFailure{ - ItemIdentifier: rec.EventID, + ItemIdentifier: rec.Change.SequenceNumber, }) } } diff --git a/pkg/exec/lambda/dynamodbstreams/dynamodbstreams_test.go b/pkg/exec/lambda/dynamodbstreams/dynamodbstreams_test.go index 6646946..7e9babd 100644 --- a/pkg/exec/lambda/dynamodbstreams/dynamodbstreams_test.go +++ b/pkg/exec/lambda/dynamodbstreams/dynamodbstreams_test.go @@ -138,12 +138,16 @@ func decodeOrder(rec *events.DynamoDBEventRecord) (order, error) { return order{customerID: cid.String(), amount: n}, nil } -func mustChange(t *testing.T, eventID, eventName, customerID string, amount int64) events.DynamoDBEventRecord { +// mustChange builds a DDB Streams record. eventID and seq are deliberately +// separate arguments: the two identifiers serve different masters (eventID +// feeds the Deduper, seq is what Lambda checkpoints on) and a fixture that +// conflates them cannot catch a handler that reports the wrong one. +func mustChange(t *testing.T, eventID, seq, eventName, customerID string, amount int64) events.DynamoDBEventRecord { t.Helper() rec := events.DynamoDBEventRecord{ EventID: eventID, EventName: eventName, - Change: events.DynamoDBStreamRecord{}, + Change: events.DynamoDBStreamRecord{SequenceNumber: seq}, } if eventName != "REMOVE" { rec.Change.NewImage = map[string]events.DynamoDBAttributeValue{ @@ -154,6 +158,13 @@ func mustChange(t *testing.T, eventID, eventName, customerID string, amount int6 return rec } +// seqNum returns a realistic DynamoDB Streams sequence number — a long +// numeric string, nothing like an eventID — so an assertion on the wrong +// identifier cannot accidentally pass. +func seqNum(n int64) string { + return "495903382714902566085596925383615710959215759891" + itoa(1000+n) +} + // itoa is strconv.Itoa for int64 without pulling in strconv in test code. func itoa(n int64) string { if n == 0 { @@ -187,9 +198,9 @@ func TestHandler_HappyPath(t *testing.T) { t.Fatalf("NewHandler: %v", err) } evt := events.DynamoDBEvent{Records: []events.DynamoDBEventRecord{ - mustChange(t, "ev-1", "INSERT", "cust-A", 100), - mustChange(t, "ev-2", "INSERT", "cust-A", 50), - mustChange(t, "ev-3", "MODIFY", "cust-B", 200), + mustChange(t, "ev-1", seqNum(1), "INSERT", "cust-A", 100), + mustChange(t, "ev-2", seqNum(2), "INSERT", "cust-A", 50), + mustChange(t, "ev-3", seqNum(3), "MODIFY", "cust-B", 200), }} resp, err := h(context.Background(), evt) if err != nil { @@ -221,9 +232,9 @@ func TestHandler_SkipRecordSentinel(t *testing.T) { // Mix INSERT (counted) + REMOVE (skipped) — the REMOVE must NOT decrement // or otherwise affect state, and must NOT be reported as a failure. evt := events.DynamoDBEvent{Records: []events.DynamoDBEventRecord{ - mustChange(t, "ev-1", "INSERT", "cust-A", 100), - mustChange(t, "ev-2", "REMOVE", "cust-A", 0), - mustChange(t, "ev-3", "INSERT", "cust-A", 50), + mustChange(t, "ev-1", seqNum(1), "INSERT", "cust-A", 100), + mustChange(t, "ev-2", seqNum(2), "REMOVE", "cust-A", 0), + mustChange(t, "ev-3", seqNum(3), "INSERT", "cust-A", 50), }} resp, err := h(context.Background(), evt) if err != nil { @@ -251,7 +262,7 @@ func TestHandler_RetriesAndRecovers(t *testing.T) { t.Fatalf("NewHandler: %v", err) } evt := events.DynamoDBEvent{Records: []events.DynamoDBEventRecord{ - mustChange(t, "ev-1", "INSERT", "cust-A", 1), + mustChange(t, "ev-1", seqNum(1), "INSERT", "cust-A", 1), }} resp, err := h(context.Background(), evt) if err != nil { @@ -265,6 +276,11 @@ func TestHandler_RetriesAndRecovers(t *testing.T) { } } +// TestHandler_ReportsExhaustedRetries pins the identifier Lambda can actually +// resolve. AWS matches BatchItemFailures against the shard's sequence numbers; +// an eventID there names nothing, and the batch is then redelivered whole +// (duplicate merges — dedup is off by default), left on a stalled iterator, or +// dropped outright. func TestHandler_ReportsExhaustedRetries(t *testing.T) { store := newFlakyStore(10) h, err := dynamodbstreams.NewHandler(newPipe(store), decodeOrder, @@ -274,19 +290,27 @@ func TestHandler_ReportsExhaustedRetries(t *testing.T) { if err != nil { t.Fatalf("NewHandler: %v", err) } - evt := events.DynamoDBEvent{Records: []events.DynamoDBEventRecord{ - mustChange(t, "ev-doom", "INSERT", "cust-A", 1), - }} - resp, err := h(context.Background(), evt) + // eventID and SequenceNumber share no characters, so neither assertion + // below can pass by coincidence. + const doomSeq = "49590338271490256608559692538361571095921575989136588898" + records := []events.DynamoDBEventRecord{ + mustChange(t, "ev-doom", doomSeq, "INSERT", "cust-A", 1), + } + resp, err := h(context.Background(), events.DynamoDBEvent{Records: records}) if err != nil { t.Fatalf("handler: %v", err) } if got := len(resp.BatchItemFailures); got != 1 { t.Fatalf("BatchItemFailures = %d, want 1; got %+v", got, resp.BatchItemFailures) } - if resp.BatchItemFailures[0].ItemIdentifier != "ev-doom" { - t.Errorf("ItemIdentifier: got %q, want %q", - resp.BatchItemFailures[0].ItemIdentifier, "ev-doom") + got := resp.BatchItemFailures[0].ItemIdentifier + if got != doomSeq { + t.Errorf("ItemIdentifier: got %q, want the record's SequenceNumber %q", got, doomSeq) + } + for _, rec := range records { + if got == rec.EventID { + t.Errorf("ItemIdentifier is eventID %q; Lambda checkpoints on sequence numbers", rec.EventID) + } } } @@ -302,7 +326,7 @@ func TestHandler_DedupSkipsSecondInvocation(t *testing.T) { t.Fatalf("NewHandler: %v", err) } evt := events.DynamoDBEvent{Records: []events.DynamoDBEventRecord{ - mustChange(t, "ev-dup", "INSERT", "cust-A", 100), + mustChange(t, "ev-dup", seqNum(7), "INSERT", "cust-A", 100), }} for i := 0; i < 3; i++ { resp, err := h(context.Background(), evt) @@ -340,7 +364,7 @@ func TestHandler_DecodeErrorCallback(t *testing.T) { "amount": events.NewNumberAttribute("100"), }, }}, - mustChange(t, "good", "INSERT", "cust-A", 50), + mustChange(t, "good", seqNum(9), "INSERT", "cust-A", 50), }} resp, err := h(context.Background(), evt) if err != nil { diff --git a/pkg/exec/replay/runtime.go b/pkg/exec/replay/runtime.go index 42c7272..353bb2a 100644 --- a/pkg/exec/replay/runtime.go +++ b/pkg/exec/replay/runtime.go @@ -9,7 +9,9 @@ // table swap as a deployment concern; the runtime here just drives records through // the pipeline. // -// At-least-once dedup at the state-store level handles re-runs safely. +// Re-runs are safe only when a Deduper is wired (see WithDedup) and only +// within its TTL horizon: the claims that suppress a repeated merge expire, +// and an archive replayed after they do is indistinguishable from new data. package replay import ( @@ -78,7 +80,12 @@ func WithRetryBackoff(base, max time.Duration) RunOption { // WithDedup installs a state.Deduper. The replay driver typically emits // stable per-event IDs (S3 archive line position, Kafka offset), so a -// re-run of the same archive folds idempotently when a Deduper is wired. +// re-run of the same archive folds idempotently — but only within the +// deduper's TTL horizon. Claims are TTL'd (DDB evicts them natively), and +// once they expire the same archive looks like new data and merges again: +// re-running a backfill a day later behind a 1h dedup TTL double-counts +// every record. Size the TTL to the longest re-run window you actually +// want protected, and remember it is a state-table cost, not free. func WithDedup(d state.Deduper) RunOption { return func(c *runConfig) { if d != nil { diff --git a/pkg/exec/replay/runtime_test.go b/pkg/exec/replay/runtime_test.go index bad239a..62ee660 100644 --- a/pkg/exec/replay/runtime_test.go +++ b/pkg/exec/replay/runtime_test.go @@ -131,6 +131,88 @@ func (d *slowDriver) Replay(ctx context.Context, out chan<- source.Record[int]) func (*slowDriver) Name() string { return "slow-driver" } func (*slowDriver) Close() error { return nil } +// fakeClock is a hand-wound clock. Replay idempotency is bounded by the +// deduper's TTL, and a test that waited out a real one would never run. +type fakeClock struct { + mu sync.Mutex + t time.Time +} + +func newFakeClock() *fakeClock { + return &fakeClock{t: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)} +} + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *fakeClock) Advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.t = c.t.Add(d) +} + +// ttlDeduper models dynamodb.Deduper: one claim per EventID, which DDB's +// native TTL drops once it expires, after which the same EventID is claimable +// again. That expiry is exactly what bounds a replay's idempotency. +type ttlDeduper struct { + mu sync.Mutex + clock *fakeClock + ttl time.Duration + expires map[string]time.Time +} + +func newTTLDeduper(clock *fakeClock, ttl time.Duration) *ttlDeduper { + return &ttlDeduper{clock: clock, ttl: ttl, expires: map[string]time.Time{}} +} + +func (d *ttlDeduper) MarkSeen(_ context.Context, id string) (bool, error) { + if id == "" { + return true, nil + } + d.mu.Lock() + defer d.mu.Unlock() + now := d.clock.Now() + if exp, ok := d.expires[id]; ok && now.Before(exp) { + return false, nil + } + d.expires[id] = now.Add(d.ttl) + return true, nil +} + +func (d *ttlDeduper) Release(_ context.Context, id string) error { + d.mu.Lock() + defer d.mu.Unlock() + delete(d.expires, id) + return nil +} + +func (*ttlDeduper) Close() error { return nil } + +// newCountingPipe sums one unit per record into a single entity, so a re-run +// that double-counts shows up as a doubled total rather than a per-key puzzle. +func newCountingPipe(store state.Store[int64]) *pipeline.Pipeline[int, int64] { + return pipeline.NewPipeline[int, int64]("replay-dedup"). + Key(func(int) string { return "all" }). + Value(func(int) int64 { return 1 }). + Aggregate(core.Sum[int64]()). + StoreIn(store) +} + +// archive returns n records of replay input. fakeDriver assigns each record a +// positional EventID, which is what a real S3-archive or Kafka-offset driver +// does: stable across re-runs of the same archive, which is the whole basis +// for dedup catching a re-run. +func archive(n int) *fakeDriver { + vals := make([]int, n) + for i := range vals { + vals[i] = i + } + return &fakeDriver{values: vals} +} + func newPipe(store state.Store[int64]) *pipeline.Pipeline[int, int64] { return pipeline.NewPipeline[int, int64]("replay-test"). Key(func(i int) string { return strconv.Itoa(i % 3) }). @@ -155,6 +237,75 @@ func TestReplay_HappyPath(t *testing.T) { } } +// TestReplay_RerunWithDedupIsIdempotent is the replay analogue of the +// bootstrap re-run test: an operator who re-runs the same archive (a retry +// after a partial failure, a second pass over a shadow table) must not double +// the totals. Sum is non-idempotent, so WithDedup is the only thing standing +// between a re-run and a corrupted backfill. +func TestReplay_RerunWithDedupIsIdempotent(t *testing.T) { + const records = 100 + + store := newFakeStore() + clock := newFakeClock() + dedup := newTTLDeduper(clock, time.Hour) + rec := metrics.NewInMemory() + + for run := 1; run <= 2; run++ { + if err := replay.Run(context.Background(), newCountingPipe(store), archive(records), + replay.WithDedup(dedup), + replay.WithMetrics(rec), + ); err != nil { + t.Fatalf("run %d: %v", run, err) + } + } + + if got := store.m[state.Key{Entity: "all"}]; got != records { + t.Errorf("total after two identical replays: got %d, want %d", got, records) + } + if got := rec.SnapshotOne("replay-dedup:dedup_skip").EventsProcessed; got != records { + t.Errorf("dedup_skip events on the second replay: got %d, want %d", got, records) + } +} + +// TestReplay_RerunAfterClaimExpiryDoubleCounts pins the horizon on that +// idempotency: the Deduper's claims are TTL'd, and once they expire the same +// archive merges a second time. 200, not 100, is the intended contract — an +// operator re-running a backfill a day later with a 1h dedup TTL is not +// protected, and nothing in the runtime can tell that re-run from new data. +func TestReplay_RerunAfterClaimExpiryDoubleCounts(t *testing.T) { + const ( + records = 100 + ttl = time.Hour + ) + + store := newFakeStore() + clock := newFakeClock() + dedup := newTTLDeduper(clock, ttl) + + if err := replay.Run(context.Background(), newCountingPipe(store), archive(records), + replay.WithDedup(dedup), + ); err != nil { + t.Fatalf("first replay: %v", err) + } + if got := store.m[state.Key{Entity: "all"}]; got != records { + t.Fatalf("after first replay: got %d, want %d", got, records) + } + + // Past the TTL horizon: DDB has evicted every claim, so the identical + // archive looks brand new. + clock.Advance(ttl + time.Minute) + + if err := replay.Run(context.Background(), newCountingPipe(store), archive(records), + replay.WithDedup(dedup), + ); err != nil { + t.Fatalf("second replay: %v", err) + } + if got := store.m[state.Key{Entity: "all"}]; got != 2*records { + t.Errorf("re-run past the dedup TTL: got %d, want %d (claims expire; the merge repeats)", + got, 2*records) + } +} + func TestReplay_RetriesOnTransientStoreFailure(t *testing.T) { store := newFlakyStore(2) drv := &fakeDriver{values: []int{1, 2, 3}} diff --git a/pkg/exec/streaming/runtime.go b/pkg/exec/streaming/runtime.go index 750d004..2358cfe 100644 --- a/pkg/exec/streaming/runtime.go +++ b/pkg/exec/streaming/runtime.go @@ -138,9 +138,13 @@ func WithDeadLetter(fn func(eventID string, err error)) RunOption { // immediate-merge semantics; common production values are 500ms–5s // depending on read-staleness tolerance. // - Durability under crash: records are Ack'd to the source AFTER the -// batch flushes, so a worker crash loses at most `window`-worth of -// in-flight records (the source replays them on restart, dedup -// catches the redelivery). +// batch flushes, so a worker crash drops at most `window`-worth of +// accumulated records and the source redelivers them on restart. +// Those records never reached the store, so the redelivery is their +// first apply, not a double-apply — dedup is not what saves you here. +// It only holds while the deduper's claim is taken at flush time: a +// claim taken on accept would suppress exactly the redelivery this +// depends on, and the batch's contribution would be lost for good. // - Memory: at most `maxBatch` records per (entity, bucket) before // forced flush. Default 1024 if unset. The number of concurrent keys // in flight is unbounded — for high-cardinality pipelines (per-user diff --git a/pkg/monoid/windowed/windowed.go b/pkg/monoid/windowed/windowed.go index 1b3a821..f7dfd89 100644 --- a/pkg/monoid/windowed/windowed.go +++ b/pkg/monoid/windowed/windowed.go @@ -11,6 +11,11 @@ import ( ) // Config describes the bucket layout for a windowed aggregation. +// +// The timestamp a record is bucketed by is source.Record.EventTime — every +// runtime passes it to BucketID, falling back to the wall clock only when a +// record carries no event time. There is no per-field extractor here: sources +// own event-time, not the window config. type Config struct { // Granularity is the size of each tumbling bucket (e.g. 24h for daily, 1h for hourly). // Smaller granularity means finer query precision and higher state cost. @@ -19,11 +24,6 @@ type Config struct { // Retention is how long buckets are kept before TTL eviction. Sliding-window queries // can ask for any range up to Retention. Retention time.Duration - - // EventTimeField, if non-empty, names the field on the source record from which to - // extract event-time. If empty, processing-time is used. Backends honor this when - // computing BucketID. - EventTimeField string } // Daily returns a Config with 24h granularity and the given retention. The most common diff --git a/pkg/murmur/lambda_test.go b/pkg/murmur/lambda_test.go index 17e471a..c49874f 100644 --- a/pkg/murmur/lambda_test.go +++ b/pkg/murmur/lambda_test.go @@ -3,6 +3,7 @@ package murmur_test import ( "context" "encoding/json" + "errors" "sync" "testing" "time" @@ -128,6 +129,69 @@ func TestDynamoDBStreamsHandler_BuildsAndProcesses(t *testing.T) { } } +// failingLambdaStore fails every MergeUpdate, so every record exhausts its +// retry budget and lands in BatchItemFailures. +type failingLambdaStore struct{} + +func (failingLambdaStore) Get(context.Context, state.Key) (int64, bool, error) { + return 0, false, nil +} +func (failingLambdaStore) GetMany(context.Context, []state.Key) ([]int64, []bool, error) { + return nil, nil, nil +} +func (failingLambdaStore) MergeUpdate(context.Context, state.Key, int64, time.Duration) error { + return errors.New("store down") +} +func (failingLambdaStore) Close() error { return nil } + +// TestDynamoDBStreamsHandler_ReportsSequenceNumberOnFailure guards the facade +// against the same mistake the underlying handler had: Lambda resolves a +// BatchItemFailures ItemIdentifier against the shard's sequence numbers, so an +// eventID there costs a whole-batch redelivery, a stalled iterator, or a +// silently dropped failure. +func TestDynamoDBStreamsHandler_ReportsSequenceNumberOnFailure(t *testing.T) { + handler, err := murmur.DynamoDBStreamsHandler(newLambdaPipe(failingLambdaStore{}), + func(rec *events.DynamoDBEventRecord) (lambdaEvent, error) { + pk := rec.Change.Keys["pk"] + return lambdaEvent{K: pk.String()}, nil + }, + murmur.LambdaConfig{}, + ) + if err != nil { + t.Fatalf("DynamoDBStreamsHandler: %v", err) + } + // eventID and SequenceNumber share no characters on purpose. + const doomSeq = "49590338271490256608559692538361571095921575989136588898" + records := []events.DynamoDBEventRecord{ + { + EventID: "ev-doom", + EventName: "INSERT", + Change: events.DynamoDBStreamRecord{ + SequenceNumber: doomSeq, + Keys: map[string]events.DynamoDBAttributeValue{ + "pk": events.NewStringAttribute("a"), + }, + }, + }, + } + resp, err := handler(context.Background(), events.DynamoDBEvent{Records: records}) + if err != nil { + t.Fatalf("handler: %v", err) + } + if len(resp.BatchItemFailures) != 1 { + t.Fatalf("BatchItemFailures: got %d, want 1", len(resp.BatchItemFailures)) + } + got := resp.BatchItemFailures[0].ItemIdentifier + if got != doomSeq { + t.Errorf("ItemIdentifier: got %q, want the record's SequenceNumber %q", got, doomSeq) + } + for _, rec := range records { + if got == rec.EventID { + t.Errorf("ItemIdentifier is eventID %q; Lambda checkpoints on sequence numbers", rec.EventID) + } + } +} + func TestSQSHandler_BuildsAndProcesses(t *testing.T) { store := newLambdaStore() pipe := newLambdaPipe(store) diff --git a/pkg/murmur/preset.go b/pkg/murmur/preset.go index aff4796..8876f5e 100644 --- a/pkg/murmur/preset.go +++ b/pkg/murmur/preset.go @@ -371,8 +371,15 @@ func (b *TrendingBuilder[T]) Hourly(retention time.Duration) *TrendingBuilder[T] return b } -// Clock overrides time.Now for the per-event timestamp. Useful for tests -// with deterministic clocks; production code should leave this unset. +// Clock overrides time.Now for the per-event decay timestamp. It takes a +// func() time.Time, not a func(T) time.Time: an observation is stamped when +// it is PROCESSED, so a late-arriving event decays from arrival rather than +// from when it happened, and a replay of last week's archive scores every +// record as fresh. Pin the clock for such a replay, or aggregate a windowed +// Sum and decay from the bucket's time at query instead. +// +// Useful for tests with deterministic clocks; production code should leave +// this unset. func (b *TrendingBuilder[T]) Clock(now func() time.Time) *TrendingBuilder[T] { if now != nil { b.now = now diff --git a/pkg/state/dynamodb/dedup.go b/pkg/state/dynamodb/dedup.go index a5e0c6a..63a2a67 100644 --- a/pkg/state/dynamodb/dedup.go +++ b/pkg/state/dynamodb/dedup.go @@ -56,6 +56,12 @@ type Deduper struct { // ttl is how long each claim is retained before DDB's TTL feature evicts it; // pick a value > the source's max delivery latency. 24h is a reasonable default // for Kafka with bounded retention; longer for Kinesis with extended retention. +// +// The ttl is also the window in which a re-run is idempotent: a replay or +// bootstrap of the same input after the claims expire merges everything a +// second time, because an evicted claim is indistinguishable from a record +// never seen. Operators who re-run backfills days later need a ttl that +// spans that gap. func NewDeduper(client *dynamodb.Client, table, pipeline string, ttl time.Duration) *Deduper { return &Deduper{client: client, table: table, pipeline: pipeline, ttl: ttl} } From 880527fb7eb47e68e0088c6cc7b89f0202f7c0fa Mon Sep 17 00:00:00 2001 From: Kyle Galloway Date: Fri, 28 Aug 2026 11:38:55 -0300 Subject: [PATCH 2/4] Stop the docs teaching the eventID bug this branch fixed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The code fix landed but every prose copy of it survived, so anyone reading the design doc or copy-pasting the search-integration handler reproduced the defect verbatim. - doc/design.md 6.2: replaced "DDB Streams uses EventID" with a per-source table of what Lambda actually checkpoints on, and spelled out why eventID is the wrong field even though it is the record's unique name. The dedup-EventID bullet now says explicitly that it is the only place EventID is used. - doc/search-integration.md: the Pattern B handler sample reported rec.EventID. It reports rec.Change.SequenceNumber now, and skips the entry when there isn't one. Guard the empty SequenceNumber. EventID was always non-empty, so swapping in Change.SequenceNumber introduced a shape the handler couldn't produce before: a hand-constructed or synthetic record with no cursor. Lambda treats a null/empty itemIdentifier as a malformed response and redelivers the WHOLE batch, re-merging every record that already succeeded — strictly worse than losing the one entry. Both the handler and the projector example now drop the unreportable entry and make the drop loud (metrics.RecordError plus a ":unreportable_failure" event; Stats.Unreportable in the example). Correct the WithBatchWindow crash-safety prose. It asserted safety on a precondition that is false in this tree: aggregator.accept claims the EventID when the record ENTERS the accumulator, not at flush. So with WithDedup the claim outlives a crash, the redelivery is dedup-skipped, and the in-flight batch is lost — the opposite of the unbatched path, where MergeOne releases the claim on a failed merge. Fixed the three places that said otherwise (runtime.go's WithBatchWindow note, design 5.4, design 14.1 + its failure diagram) and left coalesce.go / aggregator.go untouched. Two tests that could not fail: - examples/recently-interacted-topk/multisource_test.go carried `const k uint32 = 10` and a hand-rolled buildPipeline claiming to mirror Build, while every deployed binary uses K=32. It calls the real example.Build now and substitutes only the Store, with assertions tying the built sketch's K to Config.ResolveK(). - TestReplay_RerunAfterClaimExpiryDoubleCounts mostly tested its own fake: ttlDeduper hand-rolled an expiry nothing in murmur implements (DynamoDB's TTL does). Deleted it, along with the fake clock, and reasserted the claim against the real dynamodb.Deduper in test/e2e/replay_dedup_ttl_test.go behind the DDB-local gate — evicting the claim rows the way TTL does, through the raw client rather than the type under test. The surviving unit test keeps a plain claim-once deduper and covers only the runtime's half of the contract. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01X2YMxeLgyRc9i5XPyEV75S --- STABILITY.md | 2 +- doc/design.md | 110 +++++++--- doc/search-integration.md | 17 +- .../multisource_test.go | 128 +++++++---- examples/search-projector/projector.go | 34 ++- examples/search-projector/projector_test.go | 41 ++++ .../lambda/dynamodbstreams/dynamodbstreams.go | 33 ++- .../dynamodbstreams/dynamodbstreams_test.go | 58 +++++ pkg/exec/replay/runtime_test.go | 119 +++------- pkg/exec/streaming/runtime.go | 24 +- test/e2e/replay_dedup_ttl_test.go | 205 ++++++++++++++++++ 11 files changed, 598 insertions(+), 173 deletions(-) create mode 100644 test/e2e/replay_dedup_ttl_test.go diff --git a/STABILITY.md b/STABILITY.md index 498d4bc..060b392 100644 --- a/STABILITY.md +++ b/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. BatchItemFailures report the record's SequenceNumber (what Lambda checkpoints on); the eventID feeds dedup only | +| `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 `:unreportable_failure` event | | `pkg/exec/lambda/sqs` | experimental | SQS Lambda handler; same shape as kinesis/dynamodbstreams. Default EventID is "/"; 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) | diff --git a/doc/design.md b/doc/design.md index 39d5c69..c6597f1 100644 --- a/doc/design.md +++ b/doc/design.md @@ -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 @@ -1319,17 +1324,41 @@ 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 +`: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`: @@ -1337,7 +1366,9 @@ EventID format suitable for `Deduper.MarkSeen`: - Kinesis: `//` — 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: `/` by default, override-able. The dedup contract is the same across all three: pass @@ -2577,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
the duplicate re-apply"] + Replay --> Lost["WithBatchWindow + WithDedup:
claim taken on accept suppresses
the replay — batch lost (14.1)"] end subgraph Storage["Storage failures"] DDBT["DDB throttle / unavailable"] @@ -2598,17 +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 aggregated for the first time on restart. Dedup is not - the mechanism here — those records never reached the store, so the - redelivery is their apply. This holds only while the dedup claim is - taken at flush time: a claim taken when the record enters the - accumulator outlives the crash, suppresses the redelivery, and the - whole in-flight batch is silently lost. - -The result: 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. + 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 @@ -2656,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 diff --git a/doc/search-integration.md b/doc/search-integration.md index b0c0cac..2178d35 100644 --- a/doc/search-integration.md +++ b/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 diff --git a/examples/recently-interacted-topk/multisource_test.go b/examples/recently-interacted-topk/multisource_test.go index d2f9a86..455e444 100644 --- a/examples/recently-interacted-topk/multisource_test.go +++ b/examples/recently-interacted-topk/multisource_test.go @@ -11,34 +11,28 @@ import ( "github.com/aws/aws-lambda-go/events" + example "github.com/gallowaysoftware/murmur/examples/recently-interacted-topk" mkinesis "github.com/gallowaysoftware/murmur/pkg/exec/lambda/kinesis" "github.com/gallowaysoftware/murmur/pkg/exec/streaming" "github.com/gallowaysoftware/murmur/pkg/monoid" "github.com/gallowaysoftware/murmur/pkg/monoid/sketch/topk" - "github.com/gallowaysoftware/murmur/pkg/monoid/windowed" "github.com/gallowaysoftware/murmur/pkg/pipeline" "github.com/gallowaysoftware/murmur/pkg/source" "github.com/gallowaysoftware/murmur/pkg/state" ) -// This test mirrors the recently-interacted-topk example pipeline against an -// in-memory state store driven by BOTH the Lambda Kinesis handler and the -// streaming runtime simultaneously. The point is to prove that the +// This test drives the recently-interacted-topk example pipeline from BOTH +// the Lambda Kinesis handler and the streaming runtime simultaneously, +// against an in-memory state store. The point is to prove that the // multi-source claim — one pipeline definition, two ingest paths, merged // state — actually holds end-to-end. // -// We import the example's package indirectly: rather than depend on the -// Build() function (which spins up a real DDB client), we re-state the same -// pipeline definition here with a synthetic Store. This keeps the test -// hermetic; the example's pipeline.go still owns the canonical definition. - -type interaction struct { - EntityID string `json:"entity_id"` - UserID string `json:"user_id"` - Source string `json:"source,omitempty"` -} - -const k uint32 = 10 +// It calls the example's real Build(). An earlier version re-stated the +// pipeline inline "to keep it hermetic", and promptly drifted: the copy +// aggregated at K=10 while every deployed binary uses K=32, so the test +// asserted a Top-N shape nothing in production would produce. Build() is +// hermetic enough — it constructs a DDB client but issues no request — so +// the only thing this test substitutes is the Store. // memBytesStore is an in-memory state.Store[[]byte] backed by the supplied // monoid's Combine. The streaming runtime and the Lambda handler both @@ -84,12 +78,12 @@ func (s *memBytesStore) Close() error { return nil } // kafkaLikeSource satisfies source.Source — used to drive streaming.Run with // a fixed batch of synthetic interactions, then close. type kafkaLikeSource struct { - events []interaction + events []example.Interaction } -func (s *kafkaLikeSource) Read(_ context.Context, out chan<- source.Record[interaction]) error { +func (s *kafkaLikeSource) Read(_ context.Context, out chan<- source.Record[example.Interaction]) error { for i, e := range s.events { - out <- source.Record[interaction]{ + out <- source.Record[example.Interaction]{ EventID: fmt.Sprintf("kafka-%d", i), EventTime: time.Now(), Value: e, @@ -101,28 +95,77 @@ func (s *kafkaLikeSource) Read(_ context.Context, out chan<- source.Record[inter func (*kafkaLikeSource) Name() string { return "test-kafka" } func (*kafkaLikeSource) Close() error { return nil } -// buildPipeline mirrors examples/recently-interacted-topk.Build but with the -// store and source plugged in by the test. The aggregation shape (TopK -// monoid, daily windowing, "global" key, SingleN value lift) is identical to -// the example. -func buildPipeline(store state.Store[[]byte], src source.Source[interaction]) *pipeline.Pipeline[interaction, []byte] { - w := windowed.Daily(30 * 24 * time.Hour) - pipe := pipeline.NewPipeline[interaction, []byte]("recently_interacted"). - Key(func(interaction) string { return "global" }). - Value(func(e interaction) []byte { return topk.SingleN(k, e.EntityID, 1) }). - Aggregate(topk.New(k), w). - StoreIn(store) +// exampleConfig is the Config every binary in the example builds from. The +// DDB endpoint is a loopback address that is never dialled — Build creates +// the client, but the test swaps the Store out before a single request is +// issued — and setting it keeps the AWS credential chain from reaching for +// real credentials. +func exampleConfig() example.Config { + return example.Config{ + DDBTable: "recently_interacted_test", + DDBRegion: "us-east-1", + DDBEndpoint: "http://127.0.0.1:8000", + WindowRetention: 30 * 24 * time.Hour, + } +} + +// buildExamplePipeline returns the example's own pipeline with its DynamoDB +// store swapped for the test's in-memory one and, optionally, a source +// attached. Everything else — the "global" key, the SingleN value lift, the +// TopK monoid and its K, the daily windowing — comes from Build, so a change +// to the example is a change to what this test exercises. +// +// Pass store == nil to build the first pipeline and read back the monoid the +// caller needs in order to construct the shared store. +func buildExamplePipeline( + t *testing.T, + store state.Store[[]byte], + src source.Source[example.Interaction], +) (*pipeline.Pipeline[example.Interaction, []byte], monoid.Monoid[[]byte]) { + t.Helper() + pipe, ddbStore, _, err := example.Build(context.Background(), exampleConfig()) + if err != nil { + t.Fatalf("example.Build: %v", err) + } + // The real store is never written to; close it so the test leaks nothing. + t.Cleanup(func() { _ = ddbStore.Close() }) + + if got := pipe.Name(); got != example.PipelineName { + t.Fatalf("pipeline name: got %q, want %q", got, example.PipelineName) + } + // Guard the K contradiction this test used to carry in a local const: the + // sketch the pipeline actually builds must be the one every binary in the + // example resolves to. Sketches sized for different K refuse to merge, + // and the symptom is an empty Top-N rather than an error. + wantK := exampleConfig().ResolveK() + if got := sketchK(t, pipe.ValueFn()(example.Interaction{EntityID: "probe"})); got != wantK { + t.Fatalf("example builds K=%d sketches, but ResolveK says %d", got, wantK) + } + if got := sketchK(t, pipe.Monoid().Identity()); got != wantK { + t.Fatalf("example aggregates at K=%d, but ResolveK says %d", got, wantK) + } + if pipe.Window() == nil { + t.Fatal("example.Build returned an unwindowed pipeline; the test asserts on a daily bucket") + } + + mon := pipe.Monoid() + if store != nil { + pipe = pipe.StoreIn(store) + } if src != nil { pipe = pipe.From(src) } - return pipe + return pipe, mon } func TestMultiSource_KinesisLambdaPlusKafkaWorker_ShareState(t *testing.T) { - store := newMemBytesStore(topk.New(k)) + // The monoid comes from the example's own Build, so the shared store + // combines exactly the way the deployed DDB BytesStore does. + _, mon := buildExamplePipeline(t, nil, nil) + store := newMemBytesStore(mon) // --- Drive Kafka side via streaming.Run --- - kafkaEvents := []interaction{ + kafkaEvents := []example.Interaction{ {EntityID: "ent-A", Source: "kafka"}, {EntityID: "ent-A", Source: "kafka"}, {EntityID: "ent-A", Source: "kafka"}, @@ -131,7 +174,7 @@ func TestMultiSource_KinesisLambdaPlusKafkaWorker_ShareState(t *testing.T) { {EntityID: "ent-C", Source: "kafka"}, } src := &kafkaLikeSource{events: kafkaEvents} - kafkaPipe := buildPipeline(store, src) + kafkaPipe, _ := buildExamplePipeline(t, store, src) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() @@ -141,13 +184,13 @@ func TestMultiSource_KinesisLambdaPlusKafkaWorker_ShareState(t *testing.T) { // --- Drive Kinesis side via NewKinesisHandler --- // Same pipeline definition, no Source attached. - lambdaPipe := buildPipeline(store, nil) - handler, err := mkinesis.NewHandler(lambdaPipe, mkinesis.JSONDecoder[interaction]()) + lambdaPipe, _ := buildExamplePipeline(t, store, nil) + handler, err := mkinesis.NewHandler(lambdaPipe, mkinesis.JSONDecoder[example.Interaction]()) if err != nil { t.Fatalf("NewHandler: %v", err) } - kinesisEvents := []interaction{ + kinesisEvents := []example.Interaction{ {EntityID: "ent-A", Source: "kinesis"}, {EntityID: "ent-A", Source: "kinesis"}, {EntityID: "ent-B", Source: "kinesis"}, @@ -182,9 +225,9 @@ func TestMultiSource_KinesisLambdaPlusKafkaWorker_ShareState(t *testing.T) { // --- Verify merged Top-N --- // Bucket today gets all 10 events: ent-A=5 (3 kafka + 2 kinesis), ent-B=3 (2k+1k), - // ent-C=1 (kafka), ent-D=1 (kinesis). - w := windowed.Daily(30 * 24 * time.Hour) - bucket := w.BucketID(time.Now()) + // ent-C=1 (kafka), ent-D=1 (kinesis). The bucket comes from the example's + // own windowing config, not a restatement of it. + bucket := lambdaPipe.Window().BucketID(time.Now()) raw, ok, err := store.Get(ctx, state.Key{Entity: "global", Bucket: bucket}) if err != nil { t.Fatalf("store Get: %v", err) @@ -197,8 +240,9 @@ func TestMultiSource_KinesisLambdaPlusKafkaWorker_ShareState(t *testing.T) { t.Fatalf("topk.Items decode: %v", err) } - // Misra-Gries with K=10 retains every distinct key when the unique-count - // fits, so we can assert exact counts. + // Misra-Gries retains every distinct key while the unique-count fits + // under K (4 distinct entities against the example's K), so the counts + // below are exact rather than approximate. got := map[string]uint64{} for _, it := range items { got[it.Key] = it.Count diff --git a/examples/search-projector/projector.go b/examples/search-projector/projector.go index 61fd428..7ab4483 100644 --- a/examples/search-projector/projector.go +++ b/examples/search-projector/projector.go @@ -72,6 +72,13 @@ type Stats struct { Indexed atomic.Int64 // bucket changed → reindex emitted IndexErrors atomic.Int64 // OpenSearch update returned a non-2xx DecodeErrs atomic.Int64 // missing pk / bad attribute / etc. + + // Unreportable counts records that failed but carry no SequenceNumber, + // so no BatchItemFailures entry can name them. Lambda rejects an empty + // itemIdentifier and redelivers the whole batch, which is worse than + // dropping the entry — but the drop must not be invisible. A nonzero + // value here means records are failing and Lambda is not being told. + Unreportable atomic.Int64 } // IndexClient abstracts the OpenSearch UpdateDoc surface so the projector @@ -178,12 +185,24 @@ func (p *Projector) Handle(ctx context.Context, rec *events.DynamoDBEventRecord) // resolves against the shard's checkpoint. Reporting the eventID instead // names nothing Lambda can find, and the failed record is either redelivered // as part of the whole batch or dropped outright. +// +// A record with an EMPTY SequenceNumber gets no entry at all: Lambda treats a +// null or empty itemIdentifier as a malformed response and redelivers the +// WHOLE batch, so one unreportable record would cost every other record in +// the batch a redundant reindex. Those drops are counted in +// Stats.Unreportable rather than being silent. Real DDB Streams records +// always carry a sequence number; an empty one means a synthetic event. func (p *Projector) HandleEvent(ctx context.Context, evt events.DynamoDBEvent) []events.DynamoDBBatchItemFailure { var failures []events.DynamoDBBatchItemFailure for i := range evt.Records { if err := p.Handle(ctx, &evt.Records[i]); err != nil { + seq := evt.Records[i].Change.SequenceNumber + if seq == "" { + p.stats.Unreportable.Add(1) + continue + } failures = append(failures, events.DynamoDBBatchItemFailure{ - ItemIdentifier: evt.Records[i].Change.SequenceNumber, + ItemIdentifier: seq, }) } } @@ -213,12 +232,15 @@ func (s *Stats) MarshalJSON() ([]byte, error) { Indexed int64 `json:"indexed"` IndexErrors int64 `json:"index_errors"` DecodeErrs int64 `json:"decode_errors"` + + Unreportable int64 `json:"unreportable_failures"` }{ - Decoded: s.Decoded.Load(), - Skipped: s.Skipped.Load(), - Indexed: s.Indexed.Load(), - IndexErrors: s.IndexErrors.Load(), - DecodeErrs: s.DecodeErrs.Load(), + Decoded: s.Decoded.Load(), + Skipped: s.Skipped.Load(), + Indexed: s.Indexed.Load(), + IndexErrors: s.IndexErrors.Load(), + DecodeErrs: s.DecodeErrs.Load(), + Unreportable: s.Unreportable.Load(), }) } diff --git a/examples/search-projector/projector_test.go b/examples/search-projector/projector_test.go index 8e17fbc..f520874 100644 --- a/examples/search-projector/projector_test.go +++ b/examples/search-projector/projector_test.go @@ -220,6 +220,47 @@ func TestProjector_TombstoneProjectsToZero(t *testing.T) { } } +// TestProjector_FailureWithoutSequenceNumberIsNotReported covers the record +// shape with no checkpoint cursor. Lambda rejects a response whose +// itemIdentifier is null or empty as malformed and redelivers the ENTIRE +// batch — one unreportable record would then cost every other record in the +// batch a redundant reindex. The projector must drop the entry and count it +// rather than emit an empty identifier or fall back to the eventID. +func TestProjector_FailureWithoutSequenceNumberIsNotReported(t *testing.T) { + idx := &fakeIndex{err: errors.New("opensearch 503")} + p := projector.New(projector.Config{Index: "posts"}, idx) + + noSeq := makeRecord("post-A", 999, 1000, true) // would index → fails + noSeq.Change.SequenceNumber = "" // hand-constructed: no cursor + withSeq := makeRecord("post-B", 999, 1000, true) + + records := []events.DynamoDBEventRecord{noSeq, withSeq} + failures := p.HandleEvent(context.Background(), events.DynamoDBEvent{Records: records}) + + if len(failures) != 1 { + t.Fatalf("BatchItemFailures: got %d, want 1 (only the record with a SequenceNumber): %+v", + len(failures), failures) + } + if got, want := failures[0].ItemIdentifier, withSeq.Change.SequenceNumber; got != want { + t.Errorf("ItemIdentifier: got %q, want %q", got, want) + } + for _, f := range failures { + if f.ItemIdentifier == "" { + t.Error("empty ItemIdentifier reported; Lambda treats that response as malformed " + + "and redelivers the whole batch") + } + if f.ItemIdentifier == noSeq.EventID { + t.Errorf("ItemIdentifier fell back to the eventID %q; Lambda cannot resolve one", noSeq.EventID) + } + } + + // The drop must be observable — a silent one hides records that failed + // and were never handed back to Lambda. + if got := p.Stats().Unreportable.Load(); got != 1 { + t.Errorf("Stats.Unreportable: got %d, want 1", got) + } +} + func TestProjector_OpenSearchFailureReportsToBatchItemFailures(t *testing.T) { idx := &fakeIndex{err: errors.New("opensearch 503")} p := projector.New(projector.Config{Index: "posts"}, idx) diff --git a/pkg/exec/lambda/dynamodbstreams/dynamodbstreams.go b/pkg/exec/lambda/dynamodbstreams/dynamodbstreams.go index 57f46ba..309e40d 100644 --- a/pkg/exec/lambda/dynamodbstreams/dynamodbstreams.go +++ b/pkg/exec/lambda/dynamodbstreams/dynamodbstreams.go @@ -54,6 +54,14 @@ // silently discarded failure. The eventID is still what feeds the Deduper — // it is the stream-unique record identity, just not the checkpoint cursor. // +// A record that fails with an EMPTY `SequenceNumber` gets no +// BatchItemFailures entry at all. An empty or null ItemIdentifier makes +// Lambda treat the response as malformed and redeliver the WHOLE batch — +// worse than losing the one entry — so the handler drops it, counts a +// `:unreportable_failure` event, and reports it through +// metrics.RecordError. Real DDB Streams records always carry a sequence +// number; an empty one means a synthetic or hand-constructed event. +// // Configure your event-source mapping with // `FunctionResponseTypes=["ReportBatchItemFailures"]` so Lambda only // redelivers the failures (or, in shard-order replay mode, all records from @@ -104,8 +112,10 @@ type handlerConfig struct { // WithMetrics installs a metrics.Recorder. Defaults to metrics.Noop{}. // The handler records events under the pipeline's Name; retries under // ":retry"; dedup skips under ":dedup_skip"; dead letters under -// ":dead_letter"; and skipped records (ErrSkipRecord) under -// ":skip" — same conventions as the streaming runtime. +// ":dead_letter"; skipped records (ErrSkipRecord) under ":skip" +// — same conventions as the streaming runtime — and failures that cannot be +// reported to Lambda because the record carries no SequenceNumber under +// ":unreportable_failure". func WithMetrics(r metrics.Recorder) HandlerOption { return func(c *handlerConfig) { if r != nil { @@ -240,8 +250,25 @@ func NewHandler[T any, V any]( // SequenceNumber, not EventID: Lambda checkpoints the shard by // sequence number, and an identifier it can't resolve is either // ignored or takes the whole batch down with it. + seq := rec.Change.SequenceNumber + if seq == "" { + // An empty ItemIdentifier is worse than no entry at all: + // Lambda rejects the whole response as malformed and + // redelivers the ENTIRE batch, re-merging every record + // that already succeeded (dedup is off by default). Drop + // the unreportable entry and make the drop loud instead. + // Real DDB Streams records always carry a sequence + // number; an empty one means a synthetic event. + cfg.Recorder.RecordEvent(name + ":unreportable_failure") + cfg.Recorder.RecordError(name, fmt.Errorf( + "record %q exhausted retries but carries no Change.SequenceNumber; "+ + "dropping its BatchItemFailures entry rather than sending an empty "+ + "ItemIdentifier (which would redeliver the whole batch): %w", + rec.EventID, err)) + continue + } resp.BatchItemFailures = append(resp.BatchItemFailures, events.DynamoDBBatchItemFailure{ - ItemIdentifier: rec.Change.SequenceNumber, + ItemIdentifier: seq, }) } } diff --git a/pkg/exec/lambda/dynamodbstreams/dynamodbstreams_test.go b/pkg/exec/lambda/dynamodbstreams/dynamodbstreams_test.go index 7e9babd..2a47004 100644 --- a/pkg/exec/lambda/dynamodbstreams/dynamodbstreams_test.go +++ b/pkg/exec/lambda/dynamodbstreams/dynamodbstreams_test.go @@ -314,6 +314,64 @@ func TestHandler_ReportsExhaustedRetries(t *testing.T) { } } +// TestHandler_EmptySequenceNumberIsNotReported covers the record shape that +// has no checkpoint cursor at all. Lambda rejects a response whose +// itemIdentifier is null or empty as MALFORMED and redelivers the entire +// batch — so an empty entry does not cost one record, it costs every record +// in the batch a second merge (dedup is off by default). The handler must +// drop the unreportable entry and surface it through the metrics recorder +// instead of handing Lambda a batch-wide redelivery. +func TestHandler_EmptySequenceNumberIsNotReported(t *testing.T) { + store := newFlakyStore(10) // never succeeds within the retry budget + rec := metrics.NewInMemory() + h, err := dynamodbstreams.NewHandler(newPipe(store), decodeOrder, + dynamodbstreams.WithMaxAttempts(2), + dynamodbstreams.WithRetryBackoff(time.Millisecond, 2*time.Millisecond), + dynamodbstreams.WithMetrics(rec), + ) + if err != nil { + t.Fatalf("NewHandler: %v", err) + } + + // A hand-constructed record: an eventID but no SequenceNumber. Real DDB + // Streams always fills the sequence number; synthetic events and + // non-AWS drivers do not. + records := []events.DynamoDBEventRecord{ + mustChange(t, "ev-no-seq", "", "INSERT", "cust-A", 1), + mustChange(t, "ev-ok", seqNum(2), "INSERT", "cust-B", 1), + } + resp, err := h(context.Background(), events.DynamoDBEvent{Records: records}) + if err != nil { + t.Fatalf("handler: %v", err) + } + + // Only the record that HAS a sequence number is reportable. + if got := len(resp.BatchItemFailures); got != 1 { + t.Fatalf("BatchItemFailures = %d, want 1; got %+v", got, resp.BatchItemFailures) + } + if got, want := resp.BatchItemFailures[0].ItemIdentifier, seqNum(2); got != want { + t.Errorf("ItemIdentifier: got %q, want %q", got, want) + } + for _, f := range resp.BatchItemFailures { + if f.ItemIdentifier == "" { + t.Errorf("empty ItemIdentifier reported; Lambda treats that response as " + + "malformed and redelivers the whole batch") + } + if f.ItemIdentifier == "ev-no-seq" { + t.Errorf("ItemIdentifier fell back to the eventID %q; Lambda cannot resolve one", f.ItemIdentifier) + } + } + + // The dropped failure must not be silent — an operator has to be able to + // see that a record failed and could not be handed back to Lambda. + if got := rec.SnapshotOne("orders:unreportable_failure").EventsProcessed; got != 1 { + t.Errorf("orders:unreportable_failure events: got %d, want 1", got) + } + if got := rec.SnapshotOne("orders").Errors; got == 0 { + t.Error("no error recorded for the dropped BatchItemFailures entry") + } +} + func TestHandler_DedupSkipsSecondInvocation(t *testing.T) { store := newFakeStore() dedup := newMemDeduper() diff --git a/pkg/exec/replay/runtime_test.go b/pkg/exec/replay/runtime_test.go index 62ee660..50bcd2a 100644 --- a/pkg/exec/replay/runtime_test.go +++ b/pkg/exec/replay/runtime_test.go @@ -131,65 +131,48 @@ func (d *slowDriver) Replay(ctx context.Context, out chan<- source.Record[int]) func (*slowDriver) Name() string { return "slow-driver" } func (*slowDriver) Close() error { return nil } -// fakeClock is a hand-wound clock. Replay idempotency is bounded by the -// deduper's TTL, and a test that waited out a real one would never run. -type fakeClock struct { - mu sync.Mutex - t time.Time -} - -func newFakeClock() *fakeClock { - return &fakeClock{t: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)} -} - -func (c *fakeClock) Now() time.Time { - c.mu.Lock() - defer c.mu.Unlock() - return c.t -} - -func (c *fakeClock) Advance(d time.Duration) { - c.mu.Lock() - defer c.mu.Unlock() - c.t = c.t.Add(d) -} - -// ttlDeduper models dynamodb.Deduper: one claim per EventID, which DDB's -// native TTL drops once it expires, after which the same EventID is claimable -// again. That expiry is exactly what bounds a replay's idempotency. -type ttlDeduper struct { - mu sync.Mutex - clock *fakeClock - ttl time.Duration - expires map[string]time.Time -} - -func newTTLDeduper(clock *fakeClock, ttl time.Duration) *ttlDeduper { - return &ttlDeduper{clock: clock, ttl: ttl, expires: map[string]time.Time{}} -} - -func (d *ttlDeduper) MarkSeen(_ context.Context, id string) (bool, error) { +// claimOnceDeduper is a minimal state.Deduper: one claim per EventID, held +// for the life of the test. It stands in for any Deduper implementation +// while the runtime's own behaviour is under test — whether a replay re-run +// re-merges depends only on what MarkSeen answers, so nothing here needs to +// model DynamoDB. +// +// It deliberately does NOT model claim expiry. Nothing in murmur implements +// a dedup TTL — DynamoDB's native TTL sweeper does, in +// pkg/state/dynamodb.Deduper's table — so a hand-rolled expiry here would +// only be this file asserting against itself. The horizon that expiry puts +// on replay idempotency is pinned against the real Deduper in +// test/e2e/replay_dedup_ttl_test.go, behind the DDB-local gate. +type claimOnceDeduper struct { + mu sync.Mutex + claim map[string]bool +} + +func newClaimOnceDeduper() *claimOnceDeduper { + return &claimOnceDeduper{claim: map[string]bool{}} +} + +func (d *claimOnceDeduper) MarkSeen(_ context.Context, id string) (bool, error) { if id == "" { return true, nil } d.mu.Lock() defer d.mu.Unlock() - now := d.clock.Now() - if exp, ok := d.expires[id]; ok && now.Before(exp) { + if d.claim[id] { return false, nil } - d.expires[id] = now.Add(d.ttl) + d.claim[id] = true return true, nil } -func (d *ttlDeduper) Release(_ context.Context, id string) error { +func (d *claimOnceDeduper) Release(_ context.Context, id string) error { d.mu.Lock() defer d.mu.Unlock() - delete(d.expires, id) + delete(d.claim, id) return nil } -func (*ttlDeduper) Close() error { return nil } +func (*claimOnceDeduper) Close() error { return nil } // newCountingPipe sums one unit per record into a single entity, so a re-run // that double-counts shows up as a doubled total rather than a per-key puzzle. @@ -242,12 +225,17 @@ func TestReplay_HappyPath(t *testing.T) { // after a partial failure, a second pass over a shadow table) must not double // the totals. Sum is non-idempotent, so WithDedup is the only thing standing // between a re-run and a corrupted backfill. +// +// This covers the runtime's half of the contract — that it consults the +// Deduper for every record and skips the merge on a claimed one. The other +// half, that the protection lapses once the claims expire, belongs to the +// real DynamoDB-backed Deduper and is asserted in +// test/e2e/replay_dedup_ttl_test.go. func TestReplay_RerunWithDedupIsIdempotent(t *testing.T) { const records = 100 store := newFakeStore() - clock := newFakeClock() - dedup := newTTLDeduper(clock, time.Hour) + dedup := newClaimOnceDeduper() rec := metrics.NewInMemory() for run := 1; run <= 2; run++ { @@ -267,45 +255,6 @@ func TestReplay_RerunWithDedupIsIdempotent(t *testing.T) { } } -// TestReplay_RerunAfterClaimExpiryDoubleCounts pins the horizon on that -// idempotency: the Deduper's claims are TTL'd, and once they expire the same -// archive merges a second time. 200, not 100, is the intended contract — an -// operator re-running a backfill a day later with a 1h dedup TTL is not -// protected, and nothing in the runtime can tell that re-run from new data. -func TestReplay_RerunAfterClaimExpiryDoubleCounts(t *testing.T) { - const ( - records = 100 - ttl = time.Hour - ) - - store := newFakeStore() - clock := newFakeClock() - dedup := newTTLDeduper(clock, ttl) - - if err := replay.Run(context.Background(), newCountingPipe(store), archive(records), - replay.WithDedup(dedup), - ); err != nil { - t.Fatalf("first replay: %v", err) - } - if got := store.m[state.Key{Entity: "all"}]; got != records { - t.Fatalf("after first replay: got %d, want %d", got, records) - } - - // Past the TTL horizon: DDB has evicted every claim, so the identical - // archive looks brand new. - clock.Advance(ttl + time.Minute) - - if err := replay.Run(context.Background(), newCountingPipe(store), archive(records), - replay.WithDedup(dedup), - ); err != nil { - t.Fatalf("second replay: %v", err) - } - if got := store.m[state.Key{Entity: "all"}]; got != 2*records { - t.Errorf("re-run past the dedup TTL: got %d, want %d (claims expire; the merge repeats)", - got, 2*records) - } -} - func TestReplay_RetriesOnTransientStoreFailure(t *testing.T) { store := newFlakyStore(2) drv := &fakeDriver{values: []int{1, 2, 3}} diff --git a/pkg/exec/streaming/runtime.go b/pkg/exec/streaming/runtime.go index 2358cfe..f69b265 100644 --- a/pkg/exec/streaming/runtime.go +++ b/pkg/exec/streaming/runtime.go @@ -140,11 +140,19 @@ func WithDeadLetter(fn func(eventID string, err error)) RunOption { // - Durability under crash: records are Ack'd to the source AFTER the // batch flushes, so a worker crash drops at most `window`-worth of // accumulated records and the source redelivers them on restart. -// Those records never reached the store, so the redelivery is their -// first apply, not a double-apply — dedup is not what saves you here. -// It only holds while the deduper's claim is taken at flush time: a -// claim taken on accept would suppress exactly the redelivery this -// depends on, and the batch's contribution would be lost for good. +// What the redelivery is worth depends on whether a Deduper is wired: +// without WithDedup, those records never reached the store, so the +// redelivery is their first apply and nothing is lost. WithDedup, the +// claim is taken when the record ENTERS the accumulator, not at +// flush — so the claim survives the crash, the redelivery is +// dedup-skipped, and the in-flight batch's contribution is lost for +// good. Pairing WithBatchWindow with WithDedup trades a possible +// double-count for a possible silent loss of up to one flush window +// (or maxBatch records per key) per crash; pick the failure you can +// live with. There is no release-on-crash: flushOne dead-letters and +// Acks a batch whose merge exhausted its retries but leaves the +// claims standing, so replaying those EventIDs by hand is suppressed +// until the dedup TTL expires. // - Memory: at most `maxBatch` records per (entity, bucket) before // forced flush. Default 1024 if unset. The number of concurrent keys // in flight is unbounded — for high-cardinality pipelines (per-user @@ -240,6 +248,12 @@ func WithBatchTick(d time.Duration) RunOption { // // The Deduper itself is typically backed by a small DDB table with TTL — // see pkg/state/dynamodb.NewDeduper. +// +// Note the interaction with WithBatchWindow. On the unbatched path the +// claim is released when the merge fails, so a redelivery is re-applied. +// The aggregator claims on accept and never releases, so a crash before +// the flush loses the accumulated records rather than double-counting +// them — see WithBatchWindow's "Durability under crash" note. func WithDedup(d state.Deduper) RunOption { return func(c *runConfig) { if d != nil { diff --git a/test/e2e/replay_dedup_ttl_test.go b/test/e2e/replay_dedup_ttl_test.go new file mode 100644 index 0000000..326c9e9 --- /dev/null +++ b/test/e2e/replay_dedup_ttl_test.go @@ -0,0 +1,205 @@ +// End-to-end test for the HORIZON on replay idempotency: +// +// in-memory archive → replay.Run(WithDedup(real dynamodb.Deduper)) → +// DynamoDB Int64SumStore +// +// pkg/exec/replay's own tests prove the runtime consults the Deduper and +// skips claimed records. They cannot prove what happens when the claims go +// away, because nothing in murmur expires them — DynamoDB's native TTL does, +// against the real dedup table. A unit test with a hand-rolled expiring fake +// would only assert against its own fake, so the claim lives here, against +// pkg/state/dynamodb.Deduper. +// +// DynamoDB Local runs no TTL sweeper on a schedule a test can wait for (real +// DynamoDB takes up to 48h), so the eviction is performed the way TTL +// performs it: DeleteItem on the claim rows, issued through the raw client +// rather than through the Deduper's own Release, so the assertion does not +// lean on the type under test to set itself up. +// +// Skipped unless DDB_LOCAL_ENDPOINT is set. +package e2e_test + +import ( + "context" + "fmt" + "os" + "strconv" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + awsddb "github.com/aws/aws-sdk-go-v2/service/dynamodb" + ddbtypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + + "github.com/gallowaysoftware/murmur/pkg/exec/replay" + "github.com/gallowaysoftware/murmur/pkg/monoid/core" + "github.com/gallowaysoftware/murmur/pkg/pipeline" + "github.com/gallowaysoftware/murmur/pkg/source" + "github.com/gallowaysoftware/murmur/pkg/state" + mddb "github.com/gallowaysoftware/murmur/pkg/state/dynamodb" +) + +// archiveDriver replays a fixed number of records with stable, positional +// EventIDs — what a real S3-archive or Kafka-offset driver produces, and the +// whole basis for a Deduper catching a re-run of the same archive. +type archiveDriver struct{ n int } + +func (d *archiveDriver) Replay(_ context.Context, out chan<- source.Record[int]) error { + for i := 0; i < d.n; i++ { + out <- source.Record[int]{ + EventID: "archive-line-" + strconv.Itoa(i), + Value: 1, + Ack: func() error { return nil }, + } + } + return nil +} +func (*archiveDriver) Name() string { return "archive-driver" } +func (*archiveDriver) Close() error { return nil } + +// TestE2E_ReplayDedupExpiryDoubleCounts pins the horizon on replay +// idempotency against the REAL DynamoDB-backed Deduper: the protection is +// exactly as durable as the claim rows, and once DynamoDB's TTL sweeper has +// taken them, the identical archive is indistinguishable from new data and +// merges a second time. +// +// 2N, not N, is the intended contract. An operator re-running a backfill a +// day later behind a 1h dedup TTL is not protected, and neither the runtime +// nor the Deduper can tell that re-run from fresh input. Sizing the TTL to +// span the longest re-run window is the only defence. +func TestE2E_ReplayDedupExpiryDoubleCounts(t *testing.T) { + endpoint := os.Getenv("DDB_LOCAL_ENDPOINT") + if endpoint == "" { + t.Skip("DDB_LOCAL_ENDPOINT must be set") + } + + const records = 100 + + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + + awsCfg, err := awsconfig.LoadDefaultConfig(ctx, + awsconfig.WithRegion("us-east-1"), + awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")), + ) + if err != nil { + t.Fatalf("aws config: %v", err) + } + ddbClient := awsddb.NewFromConfig(awsCfg, func(o *awsddb.Options) { + o.BaseEndpoint = aws.String(endpoint) + }) + + suffix := time.Now().UnixNano() + stateTable := fmt.Sprintf("murmur_e2e_replayttl_state_%d", suffix) + dedupTable := fmt.Sprintf("murmur_e2e_replayttl_dedup_%d", suffix) + + if err := mddb.CreateInt64Table(ctx, ddbClient, stateTable); err != nil { + t.Fatalf("create state table: %v", err) + } + t.Cleanup(func() { + _, _ = ddbClient.DeleteTable(context.Background(), &awsddb.DeleteTableInput{TableName: &stateTable}) + }) + if err := mddb.CreateDedupTable(ctx, ddbClient, dedupTable); err != nil { + t.Fatalf("create dedup table: %v", err) + } + t.Cleanup(func() { + _, _ = ddbClient.DeleteTable(context.Background(), &awsddb.DeleteTableInput{TableName: &dedupTable}) + }) + + store := mddb.NewInt64SumStore(ddbClient, stateTable) + t.Cleanup(func() { _ = store.Close() }) + + // A generous TTL: this test never waits it out, it evicts the rows the + // way DynamoDB's sweeper eventually would. + deduper := mddb.NewDeduper(ddbClient, dedupTable, time.Hour) + + newPipe := func() *pipeline.Pipeline[int, int64] { + return pipeline.NewPipeline[int, int64]("replay_dedup_ttl"). + Key(func(int) string { return "all" }). + Value(func(int) int64 { return 1 }). + Aggregate(core.Sum[int64]()). + StoreIn(store) + } + total := func() int64 { + t.Helper() + v, ok, err := store.Get(ctx, state.Key{Entity: "all"}) + if err != nil { + t.Fatalf("store Get: %v", err) + } + if !ok { + return 0 + } + return v + } + + // --- Pass 1: the backfill itself. --- + if err := replay.Run(ctx, newPipe(), &archiveDriver{n: records}, replay.WithDedup(deduper)); err != nil { + t.Fatalf("first replay: %v", err) + } + if got := total(); got != records { + t.Fatalf("after first replay: got %d, want %d", got, records) + } + + // --- Pass 2: an immediate re-run, claims still live. --- + if err := replay.Run(ctx, newPipe(), &archiveDriver{n: records}, replay.WithDedup(deduper)); err != nil { + t.Fatalf("second replay: %v", err) + } + if got := total(); got != records { + t.Fatalf("re-run inside the dedup TTL: got %d, want %d (the claims should suppress every merge)", + got, records) + } + + // --- Evict the claims, exactly as DynamoDB's TTL sweeper does. --- + if evicted := deleteAllDedupClaims(ctx, t, ddbClient, dedupTable); evicted != records { + t.Fatalf("evicted %d dedup claims, want %d — the Deduper did not write one row per record", + evicted, records) + } + + // --- Pass 3: same archive, no claims. It looks brand new. --- + if err := replay.Run(ctx, newPipe(), &archiveDriver{n: records}, replay.WithDedup(deduper)); err != nil { + t.Fatalf("third replay: %v", err) + } + if got := total(); got != 2*records { + t.Errorf("re-run past the dedup TTL: got %d, want %d (claims expire; the merge repeats)", + got, 2*records) + } +} + +// deleteAllDedupClaims scans the dedup table and deletes every claim row, +// standing in for DynamoDB's TTL sweeper. It goes through the raw client +// rather than Deduper.Release so the eviction does not depend on the +// behaviour of the type under test. Returns the number of rows removed. +func deleteAllDedupClaims(ctx context.Context, t *testing.T, client *awsddb.Client, table string) int { + t.Helper() + var deleted int + var start map[string]ddbtypes.AttributeValue + for { + out, err := client.Scan(ctx, &awsddb.ScanInput{ + TableName: &table, + ExclusiveStartKey: start, + }) + if err != nil { + t.Fatalf("scan dedup table: %v", err) + } + for _, item := range out.Items { + pk, ok := item["pk"] + if !ok { + t.Fatalf("dedup row has no pk attribute: %+v", item) + } + if _, err := client.DeleteItem(ctx, &awsddb.DeleteItemInput{ + TableName: &table, + Key: map[string]ddbtypes.AttributeValue{"pk": pk}, + }); err != nil { + t.Fatalf("delete dedup row: %v", err) + } + deleted++ + } + if len(out.LastEvaluatedKey) == 0 { + break + } + start = out.LastEvaluatedKey + } + return deleted +} From 577be923e42887ae4773a8129aaeb3c560cfde81 Mon Sep 17 00:00:00 2001 From: Kyle Galloway Date: Fri, 28 Aug 2026 11:46:17 -0300 Subject: [PATCH 3/4] CHANGELOG for this branch Kept out of the implementing commits so parallel branches did not all conflict on the [Unreleased] heading. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01X2YMxeLgyRc9i5XPyEV75S --- CHANGELOG.md | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5be9153..b9c9fc1 100644 --- a/CHANGELOG.md +++ b/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 `: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 From 0c6804ec96f4f92d61fc32f6f21bc68086735a80 Mon Sep 17 00:00:00 2001 From: Kyle Galloway Date: Fri, 28 Aug 2026 11:56:26 -0300 Subject: [PATCH 4/4] Scope the replay e2e deduper to its pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NewDeduper gained a pipeline-name argument in #88 (claim keys are now "#"). This test file was written on a parallel branch against the old three-argument signature. go build does not compile _test.go files, so only golangci-lint's typecheck caught it — worth noting for the next cross-branch rebase. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01X2YMxeLgyRc9i5XPyEV75S --- test/e2e/replay_dedup_ttl_test.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/test/e2e/replay_dedup_ttl_test.go b/test/e2e/replay_dedup_ttl_test.go index 326c9e9..e716ee9 100644 --- a/test/e2e/replay_dedup_ttl_test.go +++ b/test/e2e/replay_dedup_ttl_test.go @@ -113,7 +113,10 @@ func TestE2E_ReplayDedupExpiryDoubleCounts(t *testing.T) { // A generous TTL: this test never waits it out, it evicts the rows the // way DynamoDB's sweeper eventually would. - deduper := mddb.NewDeduper(ddbClient, dedupTable, time.Hour) + // The pipeline name scopes the claim key ("#"), so it + // must match the pipeline built below or the claims land in a namespace + // nothing reads. + deduper := mddb.NewDeduper(ddbClient, dedupTable, "replay_dedup_ttl", time.Hour) newPipe := func() *pipeline.Pipeline[int, int64] { return pipeline.NewPipeline[int, int64]("replay_dedup_ttl").