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
74 lines
3.9 KiB
Markdown
74 lines
3.9 KiB
Markdown
# Tasks: Re-materialization after Purge
|
|
|
|
## T1: PurgeJob and PurgeJobQueue
|
|
**File**: `tidal/src/cohort/rematerialization/job.rs`
|
|
- `JobId = [u8; 16]`, `new_job_id()` (timestamp+counter), `job_id_to_string()`
|
|
- `PurgeJobStatus` enum: Pending, Running, Succeeded, Failed, PermanentlyFailed
|
|
- `PurgeJob` struct with all lifecycle fields
|
|
- `PurgeJobQueue`: DashMap + FIFO order Vec; `submit`, `claim_pending`, `update_status`, `requeue_interrupted`, `requeue_failed`, `list_jobs`, `job_status`, `pending_count`
|
|
- Status: COMPLETE
|
|
|
|
## T2: RematerializationMetrics
|
|
**File**: `tidal/src/cohort/rematerialization/metrics.rs`
|
|
- `RematerializationMetrics` (AtomicU64 fields): `jobs_pending`, `jobs_succeeded`, `jobs_failed`, `last_job_duration_ms`, `last_job_entries_updated`
|
|
- `RematerializationMetricsSnapshot` (plain u64/usize)
|
|
- `RematerializationMetrics::new() -> Arc<Self>`, `snapshot()`
|
|
- Status: COMPLETE
|
|
|
|
## T3: Filtered WAL Replay
|
|
**File**: `tidal/src/cohort/rematerialization/replay.rs`
|
|
- `ExclusionKey { entity_id, signal_type_id, weight_bits, timestamp_nanos }` 4-tuple
|
|
- `filtered_replay(wal_dir, schema, community_id, contributions, signal_type_filter) -> Result<ReplayResult>`
|
|
- Scans WAL segments; builds scratch `CohortSignalLedger`; excludes matched events
|
|
- `ReplayStats { events_replayed, events_excluded }`
|
|
- Status: COMPLETE
|
|
|
|
## T4: Atomic Ledger Swap
|
|
**File**: `tidal/src/cohort/rematerialization/swap.rs`
|
|
- `atomic_swap(live, scratch, community_id) -> u64`
|
|
- Uses `drain_community_into` (no Clone needed); removes zero-event entries from live
|
|
- Status: COMPLETE
|
|
|
|
## T5: Audit Log
|
|
**File**: `tidal/src/cohort/rematerialization/audit.rs`
|
|
- `AuditLogWriter::open(path)`, `append(&entry) -> Result<()>`
|
|
- BLAKE3-framed binary: `[magic:4][body_len:4 LE][body:JSON][checksum:4]`
|
|
- `PurgeAuditEntry`, `make_success_entry`, `make_failure_entry`, `read_audit_log`
|
|
- `compute_verification_checksum_direct` — BLAKE3 over sorted live ledger entries
|
|
- Status: COMPLETE
|
|
|
|
## T6: CohortSignalLedger Extensions
|
|
**File**: `tidal/src/cohort/ledger.rs`
|
|
- `remove_entry(cohort, entity_id, type_id)` — removes one DashMap entry
|
|
- `drain_community_into(live, community_id) -> Vec<(EntityId, SignalTypeId)>` — ownership transfer via `DashMap::remove + insert`
|
|
- Status: COMPLETE
|
|
|
|
## T7: RematerializationEngine Worker
|
|
**File**: `tidal/src/cohort/rematerialization/mod.rs`
|
|
- `RematerializationConfig` struct
|
|
- `RematerializationHandle` with `shutdown()` and `Drop`
|
|
- `start(config) -> RematerializationHandle` — spawns `tidal-remat` thread
|
|
- `worker_loop`: polls queue every 500ms, executes jobs, exponential backoff on failure
|
|
- Status: COMPLETE
|
|
|
|
## T8: TidalDb Integration
|
|
**Files**: `tidal/src/db/mod.rs`, `tidal/src/db/rematerialization.rs`, `tidal/src/db/lifecycle.rs`
|
|
- Added `purge_job_queue`, `rematerialization_metrics`, `rematerialization_handle` fields
|
|
- Engine started in `from_parts()` when `config.data_dir` is Some
|
|
- Engine stopped in `shutdown_inner()` before WAL shutdown
|
|
- Interrupted jobs requeued on startup
|
|
- Public methods: `submit_purge_job`, `purge_job_status`, `rematerialization_metrics`
|
|
- Status: COMPLETE
|
|
|
|
## T9: Integration Tests
|
|
**File**: `tidal/tests/m9_purge_remat.rs`
|
|
- 6 tests: `submit_purge_job_returns_job_id`, `purge_job_status_pending_after_submit`, `unknown_job_id_returns_none`, `rematerialization_metrics_snapshot_is_valid`, `job_succeeds_in_persistent_mode`, `multiple_jobs_processed_independently`
|
|
- Status: COMPLETE
|
|
|
|
## Bugfixes Applied
|
|
- Fixed `purge_retracts_from_ledger` test in `cohort/purge.rs` (ancient timestamp caused full signal decay)
|
|
- Fixed `query/retrieve/types/tests.rs` direct `Retrieve` struct initializers missing `community` field
|
|
- Fixed `entities/community.rs` temporary drop while borrowed (DashMap entry)
|
|
- Fixed `ranking/executor/mod.rs` `if let Some` on `Result` (bootstrapped code used wrong arm)
|
|
- Replaced `uuid::Uuid` in `entities/revocation.rs` with `[u8; 16]` backed by timestamp+AtomicU64
|