Hourly rollup scheduler + atomic watermark + rollup-aware query path

[pbudzik]

Summary

Closes review item #9. Wires the orphaned rollup builder up to a real background scheduler. The headline spec use case — "monthly account usage from rollups" — now actually goes through rollups instead of falling back to a raw scan.

What works now

Scheduler. A tokio background task runs every rollup_tick_interval_secs (30s default). Each tick:

1. Reads the manifest, computes target = floor((now - safety_lag) / 1h) * 1h
2. For each hour in [current_watermark, target), scans raw segments that overlap the hour, aggregates events whose timestamp falls inside it (grouped by bucket and dimension canonical JSON), writes one rollup_<uuid>.rseg segment per bucket touched
3. Atomically appends the new SegmentMeta entries to manifest.rollup_segments and advances manifest.watermarks.hourly_rollup_ms in a single save() — a crash mid-tick leaves the previous watermark in place and the next tick redoes the work, so the operation is idempotent (covered by test)

Query path. RollupHourly queries now scan rollup segments for hours ≤ watermark, then fall back to raw segments + memtable for the open-period tail (timestamps > watermark). SUM(quantity) matches across RawEvents and RollupHourly regardless of how the data is split (covered by test).

Open-period correctness. A separate test sets up two hours of data, ticks the worker with a now that places the watermark between them, and verifies that the rollup-source query sees both: the older hour from rollups, the newer hour from raw fallback. Bug found and fixed during this: collect_raw_events was returning all events in a segment regardless of the caller-supplied time range — fine for RawEvents (the aggregator filters later) but it caused double-counting on the fallback path. Now it filters per-event.

Notable design choices

• HourlyRollupKey / HourlyRollupRecord drop the per-builder u32 dimensions_key (meaningless across segments) in favor of canonical JSON. Records are now self-describing and cross-segment comparable.
• RollupSegmentWriter::finish() returns (rows, checksum). Rollup segment metadata carries a blake3 checksum like raw segments, verified on every open.
• RollupHourly re-derives the bucket per event from account_id rather than trusting the source segment's bucket label, so future compaction that mixes buckets can't poison the rollup.
• Config gains rollup_tick_interval_secs (default 30) and rollup_safety_lag_ms (default 60_000). The safety lag bounds the open-period duration the query path has to merge across.

COUNT caveat

Each rollup row counts as 1, not as the number of underlying events. This is documented in the executor doc-comment and in the README — callers needing exact event counts should use the RawEvents source. SUM(quantity) is correct in both.

Test plan

• cargo build --all-targets clean with -D warnings
• cargo test --all-targets — 27 tests pass (was 23; +4 in tests/rollups.rs)
• CI green

Still on the backlog

• Compaction worker scheduler
• Event validation / rejected counts (Phase D operability: explain + verify endpoints #14)
• Buffered WAL (Parquet export for raw events (Phase D operability) #17), durability mode (Admin CLI: check / rebuild-rollups / inspect-segment / verify-period / export-parquet #18), endpoint path alignment (Polish bundle: file locking, sort-on-flush, RLE for kind #19)
• Hot-path event clones (Frozen-snapshot semantics for closed periods #23)
• Rollup segment columnar format (small; can come later)

🤖 Generated with Claude Code