Fix three P0 billing-safety bugs from external review

[pbudzik]

Summary

External code review surfaced three P0 paths to silently-wrong billing answers. This PR fixes all three and adds regression tests shaped so they would fail on the pre-fix code.

P0 #1 — Rollup watermark advanced past unflushed data

Bug. RollupWorker.tick advanced the watermark to floor((now - safety_lag) / 1h) * 1h regardless of whether the events for that hour had actually reached raw segments. With low traffic, the 64 MB memtable threshold isn't hit for hours; meanwhile the rollup worker seals hours and writes empty/partial rollups. The query path's raw fallback only kicks in for timestamps above the watermark, so events stuck in the memtable were silently dropped from rollup queries.

Fix. Three new safety bounds in tick:

1. Skip if a flush is in flight. wal.active_id > manifest.last_sealed_wal_id + 1 means a sealed WAL file hasn't been committed to a raw segment yet. Don't advance the watermark — let the flusher catch up.
2. Cap by oldest memtable event. min_event_timestamp_ms floor-of-hour is the upper bound on target_hour. The watermark never crosses unflushed data.
3. Force-drain stale memtables. If the memtable's oldest event has been pending longer than memtable_max_age_ms (new config, default 60s), the worker drains it + rotates the WAL + queues a flush, then skips this tick. The next tick sees the flushed state. This prevents the watermark from stalling forever in low-traffic deployments.

Memtable gains oldest_insert_at_ms and min_event_timestamp_ms.

P0 #2 — Dedupe lost across restart after WAL was sealed

Bug. Recovery rebuilt dedupe only from un-sealed WAL files. After one full flush cycle the WAL is deleted, so a restart left the cache empty — and the compaction worker's cold dedupe only triggers on >16 segments per bucket and silently keeps "first wins" without payload comparison. So a retry of an ack'd event across restart was accepted as new and re-billed.

Fix. After WAL replay, recovery now scans raw segments whose max_timestamp_ms is within the dedupe TTL window (DEFAULT_TTL_MS, 7 days) and re-registers each event in the rebuilt cache. Older segments are skipped — TTL says the upstream pipeline shouldn't be retrying events that stale.

P0 #3 — Rollups dropped source and unit

Bug. HourlyRollupKey and HourlyRollupRecord carried account / sub / product / meter / model / hour / dims but not source or unit. The query executor's synthetic rollup event filled both with empty strings, so any RollupHourly query that grouped or filtered on either returned wrong answers silently.

Fix. Added source: SourceId and unit: Unit to both the key and record. #[serde(default)] on the record fields means old rollup segments deserialize with empty strings — new segments populate them correctly. SourceId and Unit gain Default for the serde default.

Tests

tests/billing_safety.rs — 7 new tests, each shaped to fail on the pre-fix path:

• rollup_watermark_bounded_by_oldest_memtable_event — fails on pre-fix (watermark would jump past h10 with event still in memtable)
• rollup_tick_skips_when_flush_is_in_flight — fails on pre-fix (would advance watermark while sealed WAL not yet committed)
• rollup_force_drains_stale_memtable — verifies the force-flush path also delivers a FlushMessage to the channel
• dedupe_rebuilds_from_recent_segments_on_recovery — fails on pre-fix (retry after restart accepted as new)
• dedupe_rebuild_skips_segments_older_than_ttl — confirms the 7-day window bound
• rollups_preserve_source_and_unit — fails on pre-fix (filter by source ignored)
• rollups_group_by_source — fails on pre-fix (all rows collapse into one empty-source group)

Test plan

• cargo build --all-targets clean with -D warnings
• cargo test --all-targets — 44 tests pass (was 37; +7 in tests/billing_safety.rs)
• CI green

Existing tests

The four existing tests/rollups.rs tests pass i64::MAX as memtable_max_age_ms, which preserves the prior "never force-drain" behavior they were written against.

Still on the backlog from the same review (P1s)

• Flusher failure makes acknowledged events invisible until restart
• Corrupt manifest silently falls back to Manifest::default()
• SQL parser silently maps SUM(anything) → SUM(quantity), < ≡ <=, etc.
• Segment pruning ignores bucket/product/meter/model metadata
• Time range semantics: inclusive vs exclusive at the to boundary
• No time-based flush trigger from the ingest path itself; no shutdown flush

🤖 Generated with Claude Code