tidaldb/.sdlc/features/m9-retroactive-purge/design.md

Design: Retroactive Signal Purge

Module Layout

tidal/src/
├── cohort/
│   ├── contribution.rs        # CohortContributionLog (new)
│   ├── purge.rs               # PurgeCoordinator, PurgeManifest, PurgeId (new)
│   ├── ledger.rs              # +retract(), +lambdas_for(), +remove_entry()
│   └── mod.rs                 # re-exports
├── signals/
│   ├── hot.rs                 # +subtract_contribution()
│   └── warm.rs                # +subtract_bucket()
├── storage/
│   └── keys.rs                # +Tag::PurgeManifest = 0x10
└── db/
    ├── purge.rs               # TidalDb::request_community_purge(), list_purge_manifests() (new)
    ├── signals.rs             # try_cohort_attribution() +contribution_log.push()
    └── mod.rs                 # +contribution_log, +purge_coordinator fields

Component Design

CohortContributionLog (cohort/contribution.rs)

Bounded FIFO ring-buffer of ContributionRecord entries.

CohortContributionLog {
    entries: Mutex<VecDeque<ContributionRecord>>,
    cap: usize,                      // default: 5_000_000
    eviction_count: AtomicU64,
}

• push(record) — O(1) amortised. Evicts front when at capacity.
• drain_for(user_id, cohort) -> Vec<ContributionRecord> — O(n) drain.
• eviction_count() -> u64 — monotonic counter for manifest.

PurgeCoordinator (cohort/purge.rs)

Orchestrates the in-memory retraction. Has no I/O responsibilities.

PurgeCoordinator {
    contribution_log: Arc<CohortContributionLog>,
    cohort_ledger: Arc<CohortSignalLedger>,
}

request_purge(user_id, cohort, now_ns, evicted) -> (PurgeId, PurgeManifest):

1. Drain contribution log for (user_id, cohort).
2. For each entry, call cohort_ledger.retract(...).
3. Build PurgeManifest with drained entries.
4. Return (rand::random::<u128>(), manifest).

Ledger Retraction (cohort/ledger.rs)

retract(cohort, entity_id, type_id, weight, timestamp_ns):

• Looks up entry; silent no-op if absent.
• Calls entry.hot.subtract_contribution(weight, timestamp_ns, lambdas).
• Calls entry.warm.subtract_bucket(timestamp_ns).

Hot Tier Subtraction (signals/hot.rs)

subtract_contribution(weight, contribution_ts_ns, lambdas):

• Computes dt = last_update_ns - contribution_ts_ns (clamped to 0).
• For each lambda: decayed = weight * exp(-lambda * dt).
• CAS loop: new_score = max(0.0, old_score - decayed).

Warm Tier Subtraction (signals/warm.rs)

subtract_bucket(contribution_ts_ns):

• Decrements all_time_count (CAS, floor 0).
• If timestamp within current minute: decrements minute bucket.
• If timestamp within current hour: decrements hour bucket.

Storage Key Format

Tag::PurgeManifest = 0x10
Key: encode_key(EntityId::new(user_id), Tag::PurgeManifest, &purge_id.to_be_bytes())
     = [user_id: 8B BE][0x00][0x10][purge_id: 16B BE]
Value: JSON-serialized PurgeManifest

Public API (db/purge.rs)

request_community_purge(user_id, cohort_name):

1. require_writeable().
2. Get evicted = contribution_log.eviction_count().
3. purge_coordinator.request_purge(...) → (purge_id, manifest).
4. If storage: storage.items_engine().put(key, manifest.to_json()).
5. Return (purge_id, manifest).

list_purge_manifests(user_id):

• Scan prefix entity_tag_prefix(user_id, PurgeManifest).
• Deserialize each value via PurgeManifest::from_json().

Data Flow

signal_with_context(signal, entity, weight, ts, user_id)
  └─ try_cohort_attribution(...)
       ├─ cohort_ledger.record(cohort, entity, type_id, weight, ts_ns)
       └─ contribution_log.push(ContributionRecord { user_id, cohort, entity, type_id, weight, ts_ns })

request_community_purge(user_id, cohort_name)
  └─ purge_coordinator.request_purge(user_id, cohort_name, now_ns, eviction_count)
       ├─ contribution_log.drain_for(user_id, cohort_name) → entries
       ├─ for each entry: cohort_ledger.retract(...)
       │    ├─ hot.subtract_contribution(weight, ts_ns, lambdas)  [CAS loop]
       │    └─ warm.subtract_bucket(ts_ns)                        [CAS loop]
       └─ return (purge_id, PurgeManifest { entries })
  └─ storage.put(encode_key(user_id, PurgeManifest, purge_id), manifest.to_json())

Design Decisions

1. No WAL event for purge — The purge manifest in storage is the durable record. WAL replay is the responsibility of m9-purge-rematerialization. Adding a WAL event would create a dependency on the WAL format that this feature intentionally avoids.

2. f32 weight in contribution log — Lossy compression (f64→f32) reduces log memory by 4B/entry. Subtraction is approximate; full correctness is deferred to WAL replay by m9-purge-rematerialization.

3. Score floors at 0.0 — CAS loops in subtract_contribution and subtract_bucket clamp to 0 to prevent negative scores from floating-point imprecision on the hot path.

4. PurgeJobQueue for async re-materialization — db/purge.rs also exposes submit_purge_job() so callers can enqueue a full WAL-replay re-materialization after the in-memory retraction.