P0: fix flusher shutdown deadlock; use notify_one for all workers

[pbudzik]

Summary

Fixes the P0 shutdown deadlock flagged in the latest external review.

The FlusherWorker holds an Arc<AppState> clone, and AppState owns a flush_sender. The flusher's run loop was a plain while let Some(msg) = recv.await { ... } that exits only when all senders drop — but the flusher's own state clone keeps one sender alive forever. drop(state); flusher_handle.await in main.rs therefore hangs on graceful shutdown.

Changes

• FlusherWorker::run now takes an Arc<Notify> shutdown signal and uses tokio::select!, matching the rollup/compaction pattern. On signal it drains any already-queued messages (the shutdown-drain payload and any last-tick flushes) before exiting.
• main.rs shutdown is reordered: stop rollup + compaction (producers) and await them, then signal the flusher and await it, then drop state.
• All three workers' shutdown call sites switched from notify_waiters() to notify_one(). notify_waiters is lossy if no waiter is currently registered — if a worker is mid-handle_message/mid-tick when shutdown fires, the notification disappears and the deadlock returns. notify_one stores a permit that the next notified().await consumes.

Test plan

• New tests/p0_shutdown.rs::flusher_shuts_down_even_with_state_clones_outstanding reproduces the original deadlock (outstanding AppState clone present at shutdown) and confirms exit within 5s.
• New tests/p0_shutdown.rs::flusher_drains_queued_messages_on_shutdown confirms drain-on-shutdown still processes queued messages (the shutdown-drain payload depends on this).
• Full cargo test passes — 143 tests across the suite.

Notes

This is the first of several P0 fixes from the external review. Follow-ups (in order of priority):

1. ManifestStore::commit(|m| …) helper — fix mutate-then-save atomicity in flusher / rollup / compaction / period-close
2. Watermark race-safe commit (validate last_sealed_wal_id and raw-segment generation are unchanged before saving)
3. Open → Closing → Closed period lifecycle so close-period can't miss racing ingests
4. Dedupe TTL on ingested_at_ms, not event timestamp_ms

🤖 Generated with Claude Code