6f26d03c77
Adds the M9 purge re-materialization feature: a WAL-replay background
engine that rebuilds community cohort aggregates for a (user, community)
pair after retroactive signal purge, restoring ranking correctness without
modifying the immutable WAL.
Key additions:
- cohort::rematerialization module: PurgeJobQueue, RematerializationEngine,
WAL replay, atomic CohortSignalLedger swap, BLAKE3 audit log, metrics counters
- TidalDb::{submit_purge_job, purge_job_status, rematerialization_metrics}
public API (db/rematerialization.rs)
- Engine auto-starts in persistent mode; clean shutdown before WAL teardown
- 6 integration tests in tests/m9_purge_remat.rs covering ephemeral and
persistent modes, job lifecycle, and multi-job independence
- Split oversized files to comply with 600-line limit: db/mod.rs →
db/from_parts.rs, entities/revocation.rs → revocation/{mod,tests}.rs,
schema/validation/builders.rs → builders/{mod,tests}.rs,
signals/warm.rs → warm/{mod,tests,proptests}.rs
- Fix pre-existing bootstrap errors: export AuditKind from session module,
add overrides_rejected to SessionSnapshot deserialization
95 lines
3.9 KiB
Markdown
95 lines
3.9 KiB
Markdown
# Design: Re-materialization after Purge
|
||
|
||
## Module Layout
|
||
|
||
All re-materialization code lives under `tidal/src/cohort/rematerialization/`:
|
||
|
||
```
|
||
cohort/
|
||
rematerialization/
|
||
mod.rs — RematerializationConfig, RematerializationHandle, start(), worker_loop
|
||
job.rs — PurgeJob, PurgeJobQueue, PurgeJobStatus, JobId
|
||
replay.rs — filtered_replay() — WAL scan + scratch CohortSignalLedger build
|
||
swap.rs — atomic_swap() — drain scratch into live ledger
|
||
audit.rs — AuditLogWriter, PurgeAuditEntry, BLAKE3-framed binary format
|
||
metrics.rs — RematerializationMetrics, RematerializationMetricsSnapshot
|
||
```
|
||
|
||
## Data Structures
|
||
|
||
### PurgeJob
|
||
```rust
|
||
pub struct PurgeJob {
|
||
pub job_id: JobId, // [u8; 16] — timestamp+counter hash
|
||
pub user_id: u64,
|
||
pub community_id: String,
|
||
pub signal_type_filter: Option<Vec<String>>,
|
||
pub submitted_at_ns: u64,
|
||
pub status: PurgeJobStatus,
|
||
pub attempt_count: u8,
|
||
pub last_heartbeat_ns: Option<u64>,
|
||
}
|
||
```
|
||
|
||
### PurgeJobQueue
|
||
- `DashMap<JobId, PurgeJob>` for O(1) lookup.
|
||
- `Arc<Mutex<Vec<JobId>>>` for FIFO submission order.
|
||
- Methods: `submit`, `claim_pending`, `update_status`, `requeue_interrupted`, `requeue_failed`, `list_jobs`, `job_status`, `pending_count`.
|
||
|
||
### RematerializationConfig
|
||
Passed to `start()` at open time:
|
||
- `cohort_ledger: Arc<CohortSignalLedger>`
|
||
- `contribution_log: Arc<CohortContributionLog>`
|
||
- `schema: Schema`
|
||
- `wal_dir: PathBuf`
|
||
- `audit_log_path: PathBuf`
|
||
- `job_queue: Arc<Mutex<PurgeJobQueue>>`
|
||
- `metrics: Arc<RematerializationMetrics>`
|
||
|
||
## WAL Replay Algorithm (`filtered_replay`)
|
||
|
||
1. Read `CohortContributionLog::drain_for(user_id, community_id)` to obtain the set of `ContributionRecord`s to exclude.
|
||
2. Build an `ExclusionSet`: `HashSet<(entity_id, signal_type_id, weight_bits, timestamp_ns)>` — exact 4-tuple match.
|
||
3. Scan WAL segments in order (using the existing WAL reader).
|
||
4. For each WAL event matching `community_id`:
|
||
- If the event's 4-tuple is in the `ExclusionSet`, skip it (`events_excluded++`).
|
||
- Otherwise, `record()` the event into a scratch `CohortSignalLedger`.
|
||
5. Return `ReplayResult { ledger: scratch, stats: ReplayStats }`.
|
||
|
||
## Atomic Ledger Swap (`atomic_swap`)
|
||
|
||
After replay, the scratch ledger is swapped into the live ledger for the target community:
|
||
1. Collect all `live` entries for `community_id` (the "before" set).
|
||
2. Call `scratch.drain_community_into(live, community_id)` — uses `DashMap::remove + insert` to take ownership of each entry without cloning.
|
||
3. For entries in the "before" set but absent from scratch (zero retained events), call `live.remove_entry(community_id, entity_id, type_id)`.
|
||
|
||
## Audit Log Format
|
||
|
||
Binary file, one entry per job:
|
||
```
|
||
[magic: 4 bytes "PRGT"]
|
||
[body_len: 4 bytes LE u32]
|
||
[body: JSON of PurgeAuditEntry]
|
||
[checksum: 4 bytes — first 4 bytes of BLAKE3(body)]
|
||
```
|
||
|
||
`PurgeAuditEntry` contains: `job_id`, `user_id`, `community_id`, `filter`, `submitted_at_ns`, `completed_at_ns`, `events_replayed`, `events_excluded`, `entries_updated`, `verification_checksum`, `success`, `error_message`.
|
||
|
||
The `verification_checksum` is a BLAKE3 hash of all `(entity_id, type_id, score_bits)` tuples from the live ledger for `community_id`, sorted for determinism.
|
||
|
||
## TidalDb Integration
|
||
|
||
New fields on `TidalDb`:
|
||
- `purge_job_queue: Arc<Mutex<PurgeJobQueue>>`
|
||
- `rematerialization_metrics: Arc<RematerializationMetrics>`
|
||
- `rematerialization_handle: Mutex<Option<RematerializationHandle>>`
|
||
|
||
Engine started in `from_parts()` when `config.data_dir` is `Some` (persistent mode). Stopped in `shutdown_inner()` before WAL shutdown. Interrupted jobs are requeued before start.
|
||
|
||
## Concurrency
|
||
|
||
- Worker polls queue every 500ms.
|
||
- Exponential backoff on failure: 100ms × 2^attempt, capped at 30s.
|
||
- After `MAX_ATTEMPTS = 5` failures, job is `PermanentlyFailed`.
|
||
- `DashMap` shard locking ensures concurrent ranking queries are not blocked during the atomic swap.
|