​

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.

​

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

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

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

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

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

• 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

• 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

• 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

• 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

• 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.