Query layer: reject the inputs that used to return a confident wrong answer

CHANGELOG.md

@@ -262,6 +262,61 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 - `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.
 
+- **Query layer: `Get` / `GetMany` on a windowed pipeline could never return data.** Both RPCs address bucket 0, which is simultaneously the all-time sentinel and the epoch bucket, so on a windowed pipeline they reported `present: false` no matter how much the pipeline had counted. Four shipped runbooks pointed operators at them. They now return `FAILED_PRECONDITION` naming `GetWindow` / `GetWindowMany`; `fresh_read` does not bypass the check. A `Granularity` of zero still legitimately writes bucket 0 and stays allowed.
+- **Query layer: degenerate time bounds returned a fabricated `present: true` zero.** `start_unix > end_unix`, bounds left at the proto3 zero, a non-positive `duration_seconds`, a `duration_seconds` large enough to overflow the nanosecond `time.Duration`, `end_unix = 253402300799`, and the zero `time.Time` that `pkg/query/typed` sends as `-62135596800` — the last two wrap `UnixNano` and landed on arbitrary negative buckets. All now return `INVALID_ARGUMENT`, mirroring the check `pkg/admin` already had.
+- **Query layer: retention was advisory on the read path.** A window longer than `Retention` read TTL-evicted buckets, folded the holes in as the monoid identity, and returned the result labelled as a full window. Windowed reads now reject a duration that reaches past what retention keeps.
+- **Query layer: bucket fan-out was unbounded.** With `windowed.Minute(24h)`, `GetRange(entity, Unix(0,0), now)` built 29,797,201 keys (~715MB) of which all but 1,440 addressed buckets TTL had already evicted. `windowed.Config` gains `MaxBuckets`, defaulting to `ceil(Retention/Granularity)`, enforced on every path that fans a read across buckets (`GetWindow`/`GetRange` and their `Many` forms, `LambdaQuery`, `WarmupWindowed`).
+- **Query layer: the singleflight coalesce key was ambiguous.** Entities were sorted and joined with `|`, so `["a|b"]` shared a group with `["a","b"]` and `["a|b","c"]` with `["a","b|c"]` — and the shipped codegen `key_template` puts a literal `|` inside entity keys. Sorting also merged `["a","b"]` with `["b","a"]`, and since responses are positional the second caller received the first caller's values against the wrong entities. The key is now length-prefixed and order-preserving; permutation coalescing is deliberately given up.
+- **Query layer: one client disconnect failed every coalesced peer.** The shared store call ran on the context of whichever caller happened to lead the singleflight group, so a hang-up cancelled the read for everyone with `CodeInternal` — under exactly the concurrent load coalescing exists to serve. The shared work now runs detached under a server-side `Config.CoalesceTimeout` (default 10s) with each waiter selecting on its own context.
+- **DynamoDB stores: a duplicate key in a batched read failed the whole RPC.** `BatchGetItem` rejects a repeated key (`Provided list of item keys contains duplicates`), and whether it reproduced depended on the two copies landing in the same 100-key chunk. `Int64SumStore.GetMany` and `BytesStore.GetMany` now chunk a de-duplicated key set and scatter results back through the `(entity, bucket)` map they already keep; `Int64MaxStore` inherits the fix. `query.WarmupWindowed` / `WarmupNonWindowed` collapse repeated entities before fetching (their reported "warmed" count is now distinct entities × buckets).
+- **`pkg/query/typed`: `SumClient` wrote past its output slice.** It sized the slice from the entity list but indexed it by the server's value count, so a server returning more values than entities caused an index-out-of-range panic inside the calling application. The sketch clients' `if i >= len(out) { break }` guard turned the same bug into silent truncation. Every batched typed client now rejects a value/entity count mismatch — the wire contract is positional, so a mismatch means the values cannot be attributed at all. `examples/search-rerank` was fixed the same way.
+
+### Changed (breaking, pre-1.0)
+
+- `QueryService.Get` / `GetMany` now fail with `FAILED_PRECONDITION` on windowed pipelines instead of reporting `present: false`.
+- Malformed windows and ranges now return `INVALID_ARGUMENT` instead of a monoid-identity value; `duration_seconds` must be positive, and absolute ranges must set at least one bound.
+- `pkg/query`'s `GetWindow` / `GetRange` / `GetWindowMany` / `GetRangeMany` / `LambdaQuery` / `WarmupWindowed` return errors matching the new `query.ErrInvalidQuery` for these shapes.
+- `windowed.Config` gains `MaxBuckets`; `grpc.Config` gains `CoalesceTimeout`. Both default from existing fields, so existing configs keep working.
+- `pkg/query/typed`'s batched clients (`GetMany`, `GetWindowMany` on all four clients) now return an error when the response's value count does not match the requested entity count.
+
+### Fixed
+
+- `pkg/query`: a `GetRange` / `GetRangeMany` over exactly the retention
+  window is accepted again. `windowed.Config.MaxBucketSpan` derived
+  `ceil(Retention/Granularity)`, but bucket ranges are inclusive at both
+  ends, so a range of duration `Retention` touches
+  `Retention/Granularity + 1` buckets and was rejected as
+  `InvalidArgument`.
+- `pkg/query`: absolute ranges are now held to `Retention` as trailing
+  windows already were. Setting `windowed.Config.MaxBuckets` higher than
+  `Retention` let `GetRange`, `GetRangeMany` and `LambdaQuery.GetRange`
+  read past TTL — a year against a 7-day `Retention` fanned out over 366
+  buckets, folded 359 evicted ones in as `Identity`, and returned the
+  week's total labelled as a year. The bound is on a range's width, not
+  its age; `checkRetention` documents why.
+- `pkg/query/grpc`: coalesced store work keeps the leading caller's
+  deadline instead of only a flat `CoalesceTimeout`. Detaching the shared
+  call from the caller's context dropped the deadline with the
+  cancellation, so a burst of abandoned requests each held up to
+  `CoalesceTimeout` of fan-out open with nobody left to read it; before
+  coalescing existed, a client hangup shed that work immediately. The
+  shared context's deadline is now derived from the leader's, capped by
+  `CoalesceTimeout`.
+
+### Changed
+
+- `pkg/monoid/windowed`: **breaking.** `Config.MaxBucketSpan()` counts
+  both ends of an inclusive range, so the default derived from
+  `Retention` is now `ceil(Retention/Granularity) + 1`. An explicit
+  `MaxBuckets` is unchanged and still used verbatim.
+- `pkg/monoid/windowed`: **breaking.** A `Config` with a `Granularity`
+  but neither `MaxBuckets` nor `Retention` now caps reads at the new
+  `DefaultMaxBucketSpan` (100,000 buckets) instead of being unbounded.
+  `Config{Granularity: time.Minute}` previously let
+  `GetRange(epoch, now)` build ~30 million keys (~715MB of `state.Key`)
+  before the first store call. A `Config` with no `Granularity` still
+  reports 0 — it assigns everything to bucket 0 and cannot fan out.
+
 ### Fixed — graceful shutdown silently lost every in-flight record
 
 `streaming.Run` treated a cancelled context as a poison record. The comment at

STABILITY.md

@@ -13,7 +13,7 @@ edges callers should plan around.
 | `pkg/monoid/core` | mostly stable | `Min` / `Max` use `Bounded[V]` for a proper Identity; lift inputs via `core.NewBounded(v)`. `Monotonic[V](identity)` is the raw-V counterpart that pairs with conditional-update stores like `pkg/state/dynamodb.Int64MaxStore` for the SetCountIfGreater pattern (out-of-order absolute-value safety) |
 | `pkg/monoid/sketch/{hll,topk,bloom}` | experimental | On a decode error `Combine` returns the operand that decoded and discards the other — unavoidable while `Monoid.Combine` has no error return, but now reportable via `WithDecodeErrorHandler`; wire it to a `metrics.Recorder` or the loss stays invisible. For `bloom` the same hook now reports **shape problems**: two filters whose (m, k) differ cannot be OR'd (`Combine` keeps the left operand), and operands that agree with each other but not with the (n, p) passed to `NewWithCapacity` are merged but reported — without that report those parameters were decorative, since every filter carries its own shape on the wire. `bloom.Bloom` / `bloom.NewWithCapacity` Identity is now the empty slice, so it is an identity for an operand of ANY shape. **`topk` counts are Misra-Gries lower bounds and the wire format does not record the stream size**, so a summary that retained 29 of 45,932 events is indistinguishable from an exact answer over 29; carry `n` out of band to size the `n/(K+1)` error bar. Making that visible in the header is a staged wire-format change that has not landed. Cross-runtime encoding portability not yet proven |
 | `pkg/monoid/compose` | mostly stable | `MapMerge` / `Tuple2` / `DecayedSum`; FP-associativity caveats apply to `DecayedSum`. `DecayedSum` / `DecayedSumBytes` take one experimental option, `WithDecodeErrorHandler` (bytes only — a non-17-byte operand is now an error, not a value assembled from someone else's blob), and `DecodeDecayed` returns `(Decayed, error)`. A non-positive half-life means "no decay" in `Combine` as well as `EvaluateAt`, and `EvaluateAt` before the reference time returns the stored value rather than scaling it up. `Combine` is a pure function of its operands — it reads no clock, because `BytesStore.MergeUpdate` recomputes it on every CAS retry. A future-dated observation therefore still freezes a key, and the guard is `compose.ClampFuture(t, now, compose.DefaultSkewBound(halfLife))` applied at the **lift**, which is the last point where a clock reading is honest; `murmur.Trending` stamps at its own clock and is not exposed |
-| `pkg/monoid/windowed` | mostly stable | bucket math is solid; minute-granularity has high read-amplification on long ranges |
+| `pkg/monoid/windowed` | mostly stable | bucket math is solid; minute-granularity has high read-amplification on long ranges. `MaxBucketSpan()` counts both ends of an inclusive range, so the default derived from `Retention` is `ceil(Retention/Granularity) + 1` — a read over exactly the retention window fits. A `Config` with a `Granularity` but neither `MaxBuckets` nor `Retention` falls back to `DefaultMaxBucketSpan` (100,000) rather than going uncapped |
 | `pkg/state` (interfaces) | mostly stable | `Store` / `Cache` interfaces unlikely to change before v1. `state.NewInstrumented` / `state.NewInstrumentedCache` decorate any store/cache with metrics.Recorder hooks (store_get / store_get_many / store_merge_update / cache_get / cache_repopulate latencies + errors) |
 | `pkg/state/dynamodb` | mostly stable | `BatchGetItem` retries `UnprocessedKeys` with chunking + jittered backoff; CAS path retries CCF with the same backoff policy, tunable per store via `WithCASRetries` / `WithCASBackoff` and counted under `<pipeline>:cas_conflict` when `WithCASMetrics` is wired. `BytesStore` pre-flights DynamoDB's 400KB item limit and returns `ErrItemTooLarge` / `*ItemTooLargeError` — sketch size tracks key length, not just K, and an oversized row otherwise fails every write while `Get` keeps serving the last value that fit. `Deduper` claim keys are `"<pipeline>#<EventID>"` (`NewDeduper` takes the pipeline name; `ForPipeline` derives a sibling scope) and carry a per-call `claimant` token so a claim whose response was lost is not mistaken for a peer's. `Int64MaxStore` ships the SetCountIfGreater pattern via DDB `UpdateItem` with conditional expression — out-of-order events with lower values are silently dropped |
 | `pkg/state/valkey` | mostly stable | `Int64Cache` (atomic INCRBY) + `BytesCache` (RMW with caller-supplied byte-monoid; works with HLL/TopK/Bloom/DecayedSumBytes) + `HLLCache` (Valkey-native PFADD/PFCOUNT/PFMERGE accelerator) + `BloomCache` (Valkey-native BF.ADD/BF.MADD/BF.EXISTS/BF.MEXISTS accelerator; requires the valkey-bloom or RedisBloom module loaded into the server). The sketch accelerators run side-by-side with the BytesStore-authoritative sketches — independent estimators, both within the monoid's error bound. No portable axiomhq↔HYLL or bits-and-blooms↔valkey-bloom byte conversion: on Valkey loss the accelerators can only be repopulated by re-feeding events |
@@ -34,9 +34,9 @@ edges callers should plan around.
 | `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. A failed record with an empty SequenceNumber gets no entry at all (an empty ItemIdentifier makes Lambda redeliver the whole batch) and is surfaced via metrics.RecordError plus a `<name>:unreportable_failure` event |
 | `pkg/exec/lambda/sqs` | experimental | SQS Lambda handler; same shape as kinesis/dynamodbstreams. Default EventID is "<arn>/<MessageId>"; override via WithEventID for FIFO content-dedup or upstream-key dedup. Uses SQS SentTimestamp for windowed-bucket assignment so delayed deliveries land in the correct bucket |
-| `pkg/query` | mostly stable | `Get` / `GetWindow` / `GetRange` / `LambdaQuery` are likely v1 surface |
-| `pkg/query/grpc` | mostly stable | generic byte-encoded responses; `cmd/murmur-codegen-typed` emits per-service typed `.proto` + Go server stubs (sum / hll / topk / bloom; get_all_time / get_window / get_window_many / get_many / get_range) over `pkg/query/typed` clients. `HealthHandler` serves `grpc.health.v1.Health` and `HealthzHandler` serves `/healthz` (liveness, always 200) + `/readyz` (readiness, store round-trip, cached so probe traffic is not billed reads) |
-| `pkg/query/typed` | mostly stable | typed-client wrappers over the generic QueryService — `SumClient`, `HLLClient`, `TopKClient`, `BloomClient`. All four expose `Get` / `GetMany` / `GetWindow` / `GetWindowMany` / `GetRange`. `GetMany` returns parallel value + present arrays so callers can distinguish "absent" from "present-and-empty"; `GetWindowMany` can't (the generic RPC merges before returning). The decoders + typed shape behind application-service typed-wrapper RPCs (see `examples/typed-wrapper`). Building block under `cmd/murmur-codegen-typed` |
+| `pkg/query` | mostly stable | `Get` / `GetWindow` / `GetRange` / `LambdaQuery` are likely v1 surface. Windowed reads validate the request before fanning out: non-positive durations, swapped / unset / unrepresentable range bounds, windows longer than `Retention`, and bucket spans past `windowed.Config.MaxBuckets` all return an error matching `query.ErrInvalidQuery` instead of a monoid Identity that looks like a real zero. Absolute ranges (`GetRange` / `GetRangeMany` / `LambdaQuery.GetRange`) are held to `Retention` too, so raising `MaxBuckets` above it does not buy reads past TTL — but the bound is on a range's WIDTH, not its age: a within-retention-width range sitting entirely in the past is still answered from whatever buckets survive |
+| `pkg/query/grpc` | mostly stable | `Get` / `GetMany` return `FAILED_PRECONDITION` on a windowed pipeline (they read bucket 0, which windowed writes never populate) — route to `GetWindow` / `GetWindowMany`; `fresh_read` does not bypass it. Malformed windows and ranges return `INVALID_ARGUMENT`. Singleflight coalescing keys on a length-prefixed, order-preserving entity encoding. The shared call is detached from the leading caller's CANCELLATION but keeps its DEADLINE, capped by `Config.CoalesceTimeout` — one client hanging up no longer fails its coalesced peers, and abandoned work still expires when the caller's own deadline would have. 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) |
+| `pkg/query/typed` | mostly stable | typed-client wrappers over the generic QueryService — `SumClient`, `HLLClient`, `TopKClient`, `BloomClient`. All four expose `Get` / `GetMany` / `GetWindow` / `GetWindowMany` / `GetRange`. `GetMany` returns parallel value + present arrays so callers can distinguish "absent" from "present-and-empty"; `GetWindowMany` can't (the generic RPC merges before returning). Every batched client errors when the response's value count doesn't match the requested entity count — the wire contract is positional, and the old handling either panicked in the caller or silently truncated. The decoders + typed shape behind application-service typed-wrapper RPCs (see `examples/typed-wrapper`). Building block under `cmd/murmur-codegen-typed` |
 | `pkg/admin` | mostly stable | CORS is closed by default; opt in via `WithAllowedOrigins`. Bearer-token (`WithAuthToken`, constant-time, multi-token for rotation) and JWT (`WithJWTVerifier`, BYO verifier) auth via a single middleware; auth is off by default — same-origin / network-isolated deploys keep the historical behavior. The `cmd/murmur-ui` binary exposes `--auth-token` + `MURMUR_ADMIN_TOKEN` env fallback |
 | `pkg/swap` | mostly stable | small surface; integrated into `deploy/terraform/modules/pipeline-counter` via opt-in `swap_enabled` (control table + IAM + seed + `SWAP_CONTROL_TABLE` / `SWAP_ALIAS` env vars in every task definition) |
 | `pkg/metrics/emf` | experimental | CloudWatch Embedded Metric Format `Recorder`. Aggregates in memory and emits one document per flush interval as counters plus EMF StatisticSets, so the hot path stays cheap and CloudWatch Logs ingestion does not scale with throughput. Needs no CloudWatch API permission or SDK client — Lambda and the ECS awslogs driver already ship stdout to Logs. Splits `pipeline:sub_event` names so `dedup_skip` / `dedup_release` become their own metrics rather than fragmenting the Pipeline dimension. Lambda callers should `Flush()` per invocation; the environment freezes on return and a background ticker may never fire |