> ## Documentation Index
> Fetch the complete documentation index at: https://docs.odds-api.io/llms.txt
> Use this file to discover all available pages before exploring further.

# 2026 09 02 droplog time buckets

# Time-bucketed drop log Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Store the gateway drop log in 10-minute Redis stream buckets with TTLs so no single key can exhaust a Dragonfly shard, while keeping the `<ms>-<seq>` cursor contract intact.

**Architecture:** A new `droplog_buckets.go` owns bucket key math, explicit id assignment, per-bucket TTL stamping, and two page iterators (forward, backward) that walk buckets in time order and read the legacy `ws:droplog` key as a pseudo-bucket during the 25 hour cutover. `droplog.go` and `droplog_cache.go` swap their direct `XRANGE`/`XREVRANGE` calls for those iterators and keep every other line of their logic. The trim job shrinks to legacy cleanup plus a bounded stale-bucket unlink.

**Tech Stack:** Go 1.25, go-redis v9, miniredis v2 (tests only), Dragonfly 1.40 in production.

**Spec:** `docs/superpowers/specs/2026-09-02-droplog-time-buckets-design.md`

## Global Constraints

* Bucket width `dropLogBucketMs = 600_000`; TTL `dropLogBucketTTL = 25 * time.Hour`; retention floor `now - 24h - dropLogBucketMs`.
* Ids are complete `<ms>-<seq>`; every id's `ms` lies inside the bucket that stores it.
* Cursor strings passed by `/internal/drops`, the board and the cache keep their exact meaning.
* Entry encoding (`marshalDropLogEntry` / `decodeDropLogEntry`), dedupe and decode-skip behaviour are untouched.
* Comments 1 to 3 lines, no narration (repo CLAUDE.md).
* All Redis behaviour unit-tested against miniredis via `go-redis`; a `dropLogNow func() time.Time` seam controls time.

***

## File structure

* Create `v3/websocket/droplog_buckets.go`: constants, `dropLogBucketKey(ms)`, `dropLogBucketStart(ms)`, `parseStreamID(id) (ms, seq, ok)`, `formatStreamID(ms, seq)`, `dropLogRetentionFloorMs(now)`, `dropLogIDAllocator`, `stampBucketTTL`, `dropLogCutover` (cutover ms cache), `forwardScan`, `backwardScan`, `staleBucketKeys(now)`.
* Create `v3/websocket/droplog_buckets_test.go`: miniredis harness `newDropLogTestRedis(t)` returning a `*redis.Client` wired into `utils.RedisClient`, plus tests for every item in the spec's coverage list.
* Modify `v3/websocket/droplog.go`: `AppendDropEvents`, `ReadDropEventsSince`, `ReadDropEventsBack`, `sweepMatchedSlice`, `TrimDropLog`; retire `dropLogMaxLenApprox`/`DROPLOG_MAX_LEN` in favour of `DROPLOG_BUCKET_MAX_LEN`.
* Modify `v3/websocket/droplog_cache.go`: `warm`, `syncTail` use the iterators.
* Modify `v3/websocket/droplog_test.go`: replace `TestDropLog_MaxLenApproxIsRunawayBackstopNotRetentionAuthority` with the bucket cap equivalent.
* Modify `v3/websocket/gateway.go:1322-1345` only if `TrimDropLog`'s signature changes (it does not).

***

### Task 1: Bucket math and id helpers

**Files:** Create `droplog_buckets.go`, `droplog_buckets_test.go`.

**Produces:**

```go theme={null}
const dropLogBucketMs int64 = 600_000
const dropLogBucketTTL = 25 * time.Hour
const dropLogRetention = 24 * time.Hour
var dropLogNow = time.Now
func dropLogBucketStart(ms int64) int64
func dropLogBucketKey(bucketStart int64) string        // "ws:droplog:b:<start>"
func parseStreamID(id string) (ms, seq int64, ok bool) // accepts "(" prefix
func formatStreamID(ms, seq int64) string
func dropLogRetentionFloorMs(now time.Time) int64
```

* [ ] Test: boundaries (`ms % bucket == 0` maps to itself, `ms-1` to previous), key format, `parseStreamID` on `"123-4"`, `"(123-4"`, `"123"`, `"-"`, `"+"`, garbage; floor value.
* [ ] Implement; `go test -run 'TestDropLogBucket|TestParseStreamID' ./websocket/`.
* [ ] Commit `feat(ws): drop log bucket key and id helpers`.

### Task 2: Id allocator and TTL stamp

**Produces:**

```go theme={null}
type dropLogIDAllocator struct { mu sync.Mutex; lastMs, lastSeq int64 }
func (a *dropLogIDAllocator) next(now time.Time) (ms, seq int64)      // monotonic
func (a *dropLogIDAllocator) bump(ms, seq int64)                      // after a top-id retry
var dropLogIDs dropLogIDAllocator
type bucketTTLStamper struct { mu sync.Mutex; stamped map[int64]struct{} }
func (s *bucketTTLStamper) ensure(ctx, client redis.Cmdable, bucketStart int64) error // EXPIRE once per process, only recorded on success
```

* [ ] Tests: `next` never goes backwards with a clock that steps back; same ms increments seq; new ms resets seq to 0; `ensure` issues one EXPIRE per bucket (miniredis `TTL` \~25h, second call no-op), failed EXPIRE (closed client) is retried on next call.
* [ ] Implement, test, commit `feat(ws): drop log id allocator and bucket TTL stamp`.

### Task 3: Cutover marker

**Produces:**

```go theme={null}
const dropLogCutoverKey = "ws:droplog:cutover_ms"
type dropLogCutover struct { mu sync.Mutex; ms int64; checkedAt time.Time }
func (c *dropLogCutover) record(ctx, client, ms int64) error   // SET NX, caches result
func (c *dropLogCutover) legacyActive(ctx, client, now time.Time) (cutoverMs int64, active bool)
   // active = cutoverMs > 0 && now < cutover + dropLogBucketTTL; re-reads Redis at most once a minute while unknown
var dropLogCut dropLogCutover
```

* [ ] Tests: unknown then recorded; `SET NX` keeps the first value; `legacyActive` false once past the window; cache avoids Redis calls (count via miniredis command log is unavailable, so assert via changing the key after the first read and observing the cached value).
* [ ] Implement, test, commit.

### Task 4: Iterators

**Produces:**

```go theme={null}
type streamPage struct { key string; msgs []redis.XMessage }
type visitFn func(page streamPage) (cont bool)
func forwardScan(ctx, client redis.Cmdable, startExclusive, endInclusive string, page int64, now time.Time, visit visitFn) error
func backwardScan(ctx, client redis.Cmdable, end string /* "" | "<id>" inclusive | "(<id>" exclusive */, minInclusive string, page int64, now time.Time, visit visitFn) error
```

Behaviour: enumerate bucket starts between the floor and `bucketStart(now)+bucketMs`, clipped by the bounds' ms; per bucket loop pages until fewer than `page` messages return or `visit` says stop; the first bucket uses the caller's bound, later buckets `-`/`+`; when the legacy window is active and the range reaches below cutover, the legacy key is visited last (backward) or first (forward) with the same bounds.

* [ ] Tests (seed via `XADD` with explicit ids across three buckets plus an empty bucket in the middle and a legacy key below cutover):
  * forward from `""` returns everything in ascending order including legacy first;
  * forward from a cursor exactly on a bucket boundary starts at the next id;
  * backward from `""` returns descending order, legacy last;
  * backward with exclusive `(id` end and a `min` in an older bucket clips both ends;
  * visit returning false stops early;
  * legacy skipped when past window.
* [ ] Implement, test, commit `feat(ws): bucket-aware drop log scans`.

### Task 5: Append through buckets

**Files:** Modify `droplog.go` `AppendDropEvents`.

Per event: `ms, seq := dropLogIDs.next(dropLogNow())`; `bucket := dropLogBucketStart(ms)`; `XADD key ID formatStreamID(ms,seq) MAXLEN ~ dropLogBucketMaxLen`; on error containing `equal or smaller`: `XREVRANGE key + - 1`, parse top, `dropLogIDs.bump(top.ms, top.seq+1)`, retry once with that id; after success `dropLogTTLs.ensure(bucket)` and `dropLogCut.record(ms)` (first append only, cached).

* [ ] Tests: two appends straddling a boundary land in two keys; TTL set on both; returned ids equal the written ids; retry path (pre-seed a higher top id in the bucket) yields `top.seq+1`; partial failure returns the persisted prefix (close the client mid-batch is impractical, so inject an error via `dropLogAppendHook` test seam? No: use a miniredis `SetError` on the server, which makes the second XADD fail, and assert `len(ids) == 1`).
* [ ] Implement, test, commit.

### Task 6: Reads through the iterators

**Files:** Modify `ReadDropEventsSince`, `ReadDropEventsBack`, `sweepMatchedSlice`.

* [ ] Tests on a three-bucket fixture with a duplicate DropID pair straddling a boundary and one undecodable entry: newest window of 5; cursor paging with `limit 2` walks the whole log with no gaps or repeats; `ReadDropEventsBack` `head`, `cursor`, `hasMore` across buckets; `ReadMatchedDropsBack` with `min` two buckets back matches across buckets and dedupes.
* [ ] Implement: each function builds its result inside a `visit` closure, retaining existing dedupe/skip/log code.
* [ ] Run the whole `websocket` package; commit.

### Task 7: Memory cache

**Files:** Modify `droplog_cache.go` `warm`, `syncTail`.

* [ ] Tests: warm across 7 buckets stores oldest-first and respects `memCacheRetainMs`; `syncTail` after new appends in a new bucket appends them.
* [ ] Implement, test, commit.

### Task 8: Trim job and cap

**Files:** Modify `TrimDropLog`; `droplog_test.go`.

`TrimDropLog`: `XTRIM ws:droplog MINID <24h ago>`; if `XLEN == 0` then `UNLINK ws:droplog`; `UNLINK` all `staleBucketKeys(now)` (floor-6h .. floor); per-bucket near-cap warning on the current bucket only. `DROPLOG_BUCKET_MAX_LEN` (default 60\_000) replaces `DROPLOG_MAX_LEN`.

* [ ] Tests: legacy trimmed then unlinked; stale bucket keys removed, live ones kept; `staleBucketKeys` returns exactly 36 keys.
* [ ] Implement, update `droplog_test.go`, commit.

### Task 9: Verification and rollout

* [ ] `go build ./... && go vet ./... && go test ./websocket/ -race -count=1`.
* [ ] `/code-review` at high effort; fix findings.
* [ ] PR with the spec linked; after deploy remove `DROPLOG_MAX_LEN` from the v3 Railway service and run the spec's rollout checks.