armdb 0.11.2

//! T10 Task 10: the namespace repair pass — see the spec's own "`repair` and
//! adoption bootstrap" section
//! (`docs/superpowers/specs/26-08-10-unique-index-cas-reservation.md`) for
//! the design rationale, and `docs/schema-metadata.md`'s "Unique namespace"
//! section for the user-facing contract (declaration, guarantees,
//! boundaries, findings) this module implements.
//!
//! **A namespace is repaired by its own pass, never the generic one.** T8's
//! [`crate::projection::reconcile_carrier_projections`] builds its expected
//! set from *one* source's scan and deletes everything else in the target —
//! correct for an index fed by one producer, destructive for a namespace fed
//! by several. A namespace is not a [`crate::projection::ProjectableIndex`]
//! (see `unique/index.rs`'s own module doc for why), so `IndexTargets` cannot
//! even hold one and the generic reconcile can never reach it structurally —
//! `carrier_reconcile_never_touches_a_namespace` in
//! `armdb/tests/unique_validate_repair.rs` pins the observable half of that
//! at the call-site level (not just the type level), because the failure mode
//! of a future refactor accidentally reaching a namespace from there would be
//! silent.
//!
//! The expected set is the same union-of-every-declared-source
//! ([`super::expected::build_unique`]) `validate`'s own namespace pass reads,
//! and every key is decided with the very same rule the write path uses
//! ([`crate::unique::resolve::decide`]) — so `validate` and this pass can
//! never disagree about what a namespace should hold, only about what they do
//! once they know it: report, or write.
//!
//! `decide` is always called here with [`crate::txn::participant::ClaimKind::Claim`]
//! — repair only ever fills a *missing* claim, it never releases one — so of
//! its four possible decisions this pass reaches exactly three:
//!
//! - `Write` — the row is put through the crate-private [`crate::unique::ClaimTarget`]
//!   seam (`write_row`), never a public API (`UniqueIndex` has none). Covers
//!   both a fresh claim (adoption's own case: a namespace opens empty, every
//!   claim reads as this) and a reservation this same claimant already holds
//!   being promoted to a permanent claim.
//! - `Conflict` — refused: a winner cannot be chosen without the business
//!   rule, so nothing is written for that key and the occurrence is recorded
//!   — the same shape `Finding::UniqueConflict`/T8's `DbError::IndexKeyCollision`
//!   use (the key plus every owner involved, see [`NamespaceConflict`]),
//!   except a namespace conflict does not abort the whole repair, it only
//!   skips that one key (every key is decided independently, mirroring
//!   `validate`'s own per-key pass).
//! - `NoOp` — already correct; no stats bump.
//! - `Delete` — [`crate::unique::resolve::decide`]'s own doc: only
//!   `ClaimKind::Release` ever decides this. Reaching it here would mean this
//!   module started passing the wrong `ClaimKind`, so it is reported as
//!   `DbError::Internal` rather than silently repaired or panicked on — the
//!   same "must never crash on an unreachable state" discipline
//!   `registry::run_decls` uses for the identical case in `validate`.
//!
//! A **plain orphan** (no declared claimant at all, on a *permanent* row) is
//! deleted only under [`super::SchemaRegistry::complete_namespace`] +
//! [`super::RepairOptions::delete_orphans`] — reusing the exact gate
//! `complete_index`/`IndexOrphan` already uses (same underlying
//! `complete_indices` set), because under partial adoption the other,
//! not-yet-declared sources' live claims are indistinguishable from orphans
//! even with the union rule.
//!
//! **A `Decl::Unique` whose source is not a registered [`super::registry::CollectionEntry`]
//! is never read as "this source claims nothing".** `expected::build_unique`
//! silently skips such a declaration (mirroring `expected::build`/
//! `build_counters` — a `validate`-only report-only tolerance), which on its
//! own would make every one of that source's live, permanent claims in its
//! declared namespace(s) look like a plain orphan against an empty expected
//! set. This pass additionally protects those namespaces the same way
//! `repair.rs`'s counter pass already protects an unregistered counter
//! source's targets — see the `src_entry` guard just above `Decl::Counter`'s
//! repair loop in `repair.rs`.
//!
//! A **reservation** with no declared claimant is never touched, in either
//! mode: an absent owner is its normal state, not drift, and its own bound is
//! the deadline — deleting it early would race a legitimate future claim.
//!
//! A **corrupt row** (fails [`crate::UniqueRow::is_well_formed`]) is excluded
//! from this pass entirely, exactly as `validate`'s own missing/conflict/
//! orphan pass excludes it — its `key_len` cannot be trusted, so it must
//! never reach `decide`. Repair leaves it exactly as `validate` reports it
//! (`Finding::UniqueRowCorrupt`); it is neither deleted nor overwritten, even
//! when a live source row declares a claim on the same key (a corrupt row
//! reads as "absent" to a naive per-key pass, which would otherwise silently
//! overwrite it via `ClaimDecision::Write` — this pass tracks corrupt keys
//! per namespace and removes them from scope entirely, the same technique
//! `registry::run_decls` uses).
//!
//! `UndecodableRows` holds back the destructive half exactly as it does for
//! `repair`'s index/counter passes: a source whose scan skipped rows makes
//! this namespace's expected set short, so its live, correctly-owned rows
//! would look like orphans — deletion is suppressed for the *whole* namespace
//! those rows feed, regardless of `complete_namespace`/`delete_orphans`.
//! Missing claims from *other*, fully-decodable sources into the same
//! namespace are still written — only the destructive half is held back.
//!
//! A namespace scan failure (`ClaimTarget::scan_rows` returning `Err`, an
//! I/O-level failure) propagates via `?` out of [`repair_namespaces`] and
//! therefore out of [`super::repair::repair`] — same as every other hard
//! failure in this module. Because this pass runs last (`repair`'s pass 5,
//! after the index and counter passes), any writes those earlier passes
//! already made have already landed on disk; this is not a new failure shape
//! for `repair` — the counter pass's own `expected::build_counters(reg)?`
//! already aborts the same way, after the index passes' writes.

use std::collections::{HashMap, HashSet};

use crate::error::{DbError, DbResult};
use crate::txn::participant::ClaimKind;
use crate::unique::resolve::{ClaimDecision, decide};

use super::expected::{self, SkippedSource};
use super::registry::{BuiltRegistry, Decl, unique_owner_pair};
use super::repair::RepairOptions;
use super::validate::hex;

fn retain_owner_source_dirty(
    reg: &BuiltRegistry,
    repaired_sources: &mut HashSet<&'static str>,
    owner: &crate::unique::UniqueOwner,
) {
    if let Some(entry) = reg
        .reg
        .entries
        .iter()
        .find(|entry| crate::unique::collection_tag(entry.name) == owner.collection_id())
    {
        repaired_sources.remove(entry.name);
    }
}

// One-shot fault injection for the namespace repair pass. It is deliberately
// checked after the namespace union has been assembled, so tests exercise the
// fail-closed retirement boundary rather than registry construction.
#[cfg(any(test, feature = "test-faults"))]
std::thread_local! {
    static FAIL_NEXT_NAMESPACE_REPAIR: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}

#[cfg(any(test, feature = "test-faults"))]
pub(crate) fn fail_next_namespace_repair() {
    FAIL_NEXT_NAMESPACE_REPAIR.set(true);
}

#[cfg(any(test, feature = "test-faults"))]
fn take_fail_next_namespace_repair() -> bool {
    FAIL_NEXT_NAMESPACE_REPAIR.replace(false)
}

/// Per-namespace repair counters (T10 Task 10) — see the module doc for what
/// drives each field. Mirrors [`super::IndexRepairStats`] in shape.
#[derive(Debug, Clone, Default)]
pub struct NamespaceRepairStats {
    pub namespace: &'static str,
    /// Missing claims written, and reservations promoted to a permanent
    /// claim by their own owner — both are `ClaimDecision::Write`.
    pub written: u64,
    /// Orphan rows deleted — only under `complete_namespace` +
    /// `RepairOptions::delete_orphans`.
    pub deleted: u64,
    /// Orphan rows found but left in place (gate not open, or the namespace
    /// is protected by an `UndecodableRows` source).
    pub declined_orphans: u64,
    /// Keys where two or more live claimants disagree, or a single claimant
    /// disagrees with the row's actual (foreign) owner — refused, nothing
    /// written for that key. `.len()` is the count; each entry carries the
    /// key and every owner involved, the same detail
    /// `Finding::UniqueConflict` reports, so an operator can correlate a
    /// `repair` run against a `validate` run without a second pass.
    pub conflicts: Vec<NamespaceConflict>,
}

/// One namespace key `repair_namespaces` refused to write to, because more
/// than one live claimant disagreed about who owns it — either two or more
/// declared claimants, or one declared claimant against the row's actual
/// (foreign) owner. Mirrors `Finding::UniqueConflict`'s shape exactly.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NamespaceConflict {
    pub key: String,
    pub owners: Vec<(String, String)>,
}

/// Repair every declared unique namespace — the T10 twin of
/// [`super::repair::repair`]'s index/counter passes. Returns per-namespace
/// stats plus the sources whose scan skipped undecodable rows, which the
/// caller merges into [`super::repair::RepairReport::undecodable_sources`]
/// exactly as its own counter pass merges into the index pass's list (same
/// source may feed both an index/counter and a namespace).
pub(super) fn repair_namespaces(
    reg: &BuiltRegistry,
    opts: &RepairOptions,
) -> DbResult<(
    Vec<NamespaceRepairStats>,
    Vec<SkippedSource>,
    Vec<&'static str>,
)> {
    let (unique_expected, skipped) = expected::build_unique(reg)?;
    let mut protected: HashSet<&'static str> = skipped
        .iter()
        .flat_map(|s| s.targets.iter().copied())
        .collect();

    // An unregistered `Decl::Unique` source is not read as "confirmed empty"
    // any more than an undecodable one is — see the module doc. Without
    // this, every permanent claim the source's declared namespace(s)
    // genuinely hold would look like a plain orphan against an empty
    // expected set, and `delete_orphans: true` would delete them all.
    for d in &reg.reg.decls {
        if let Decl::Unique {
            source, targets, ..
        } = d
            && reg.reg.entries.iter().all(|e| e.name != *source)
        {
            for t in targets.all() {
                protected.insert(t.collection_name());
            }
        }
    }

    let handles = expected::declared_unique_handles(reg);

    // Start with every registered collection: an explicit format migration
    // cannot know whether a later opener will install a claim hook, so it
    // leaves the same conservative marker even for a registered collection
    // with no unique declaration. Such a source has a successful empty pass.
    // A declared source is eligible only when every namespace it names is
    // registered as a unique index and has a live handle in this union.
    let mut repaired_sources: HashSet<&'static str> =
        reg.reg.entries.iter().map(|entry| entry.name).collect();
    for d in &reg.reg.decls {
        let Decl::Unique {
            source, targets, ..
        } = d
        else {
            continue;
        };
        let complete = targets.all().all(|target| {
            let name = target.collection_name();
            handles.contains_key(name)
                && reg.reg.external.iter().any(|(registered, kind)| {
                    *registered == name && *kind == armour_core::CollectionKind::UniqueIndex
                })
        });
        if !complete {
            repaired_sources.remove(source);
        }
    }

    // Pass 1: scan every declared namespace once. A corrupt row is excluded
    // from `stored` *and* tracked in `corrupt_by_ns`, so its key can be
    // removed from `all_pairs` below — never overwritten, never deleted,
    // same as `validate`'s own guard (T10 Task 9 review, C1): `is_well_formed`
    // exists precisely so a row like this never reaches `decide`. Excluding
    // it only from `stored` (and not from the key space) would make it read
    // as "absent" to the per-key pass below, which is indistinguishable from
    // a genuinely missing claim and would let a live source's claim silently
    // overwrite it.
    let mut stored_by_ns: HashMap<&'static str, HashMap<Vec<u8>, crate::UniqueRow>> =
        HashMap::new();
    let mut corrupt_by_ns: HashMap<&'static str, HashSet<Vec<u8>>> = HashMap::new();
    for (ns_name, handle) in &handles {
        let stored = stored_by_ns.entry(ns_name).or_default();
        let corrupt = corrupt_by_ns.entry(ns_name).or_default();
        handle.scan_rows(&mut |key_bytes, row| {
            if row.is_well_formed() {
                stored.insert(key_bytes.to_vec(), row);
            } else {
                corrupt.insert(key_bytes.to_vec());
            }
            Ok(())
        })?;
    }

    #[cfg(any(test, feature = "test-faults"))]
    if take_fail_next_namespace_repair() {
        return Err(DbError::Internal("injected namespace repair failure"));
    }

    // Pass 2: union every expected `(namespace, key)` with every physically
    // stored one — the same union `validate`'s missing/conflict/orphan pass
    // walks, so the two can never disagree about which keys are in scope —
    // then drop every corrupt key from scope entirely.
    let mut all_pairs: HashSet<(&'static str, Vec<u8>)> = unique_expected.keys().cloned().collect();
    for (ns, rows) in &stored_by_ns {
        for key in rows.keys() {
            all_pairs.insert((*ns, key.clone()));
        }
    }
    for (ns, keys) in &corrupt_by_ns {
        for key in keys {
            // A corrupt stored row blocks comparison for this key, but an
            // expected claimant still has unresolved namespace drift. Keep
            // every such source dirty rather than letting the exclusion below
            // turn "cannot inspect" into "successfully repaired".
            if let Some(claims) = unique_expected.get(&(*ns, key.clone())) {
                for claim in claims {
                    repaired_sources.remove(claim.source);
                }
            }
            all_pairs.remove(&(*ns, key.clone()));
        }
    }

    let mut stats_by_ns: HashMap<&'static str, NamespaceRepairStats> = HashMap::new();
    for (ns, key) in all_pairs {
        let Some(handle) = handles.get(ns) else {
            continue;
        };
        let claims = unique_expected.get(&(ns, key.clone()));
        let row = stored_by_ns.get(ns).and_then(|m| m.get(&key)).copied();

        match claims {
            None => {
                // No declared claimant at all. A reservation's absent owner
                // is its normal state — never touched, in either mode. A
                // permanent row with nobody claiming it is a plain orphan.
                let Some(row) = row else { continue };
                if row.deadline().is_some() {
                    continue;
                }
                let stats = stats_by_ns
                    .entry(ns)
                    .or_insert_with(|| NamespaceRepairStats {
                        namespace: ns,
                        ..Default::default()
                    });
                let complete = reg.complete_indices.contains(ns) && !protected.contains(ns);
                if complete && opts.delete_orphans {
                    handle.delete_row(&key)?;
                    stats.deleted += 1;
                } else {
                    retain_owner_source_dirty(reg, &mut repaired_sources, &row.owner());
                    stats.declined_orphans += 1;
                }
            }
            Some(claims) if claims.len() == 1 => {
                let claim = &claims[0];
                let stats = stats_by_ns
                    .entry(ns)
                    .or_insert_with(|| NamespaceRepairStats {
                        namespace: ns,
                        ..Default::default()
                    });
                match decide(row, &claim.owner, ClaimKind::Claim) {
                    ClaimDecision::NoOp => {}
                    ClaimDecision::Write(new_row) => {
                        handle.write_row(&key, new_row)?;
                        stats.written += 1;
                    }
                    ClaimDecision::Conflict(cur) => {
                        repaired_sources.remove(claim.source);
                        retain_owner_source_dirty(reg, &mut repaired_sources, &cur);
                        let (cur_source, cur_key) = unique_owner_pair(reg, &cur);
                        stats.conflicts.push(NamespaceConflict {
                            key: hex(&key),
                            owners: vec![
                                (claim.source.to_string(), claim.source_key.clone()),
                                (cur_source, cur_key),
                            ],
                        });
                    }
                    ClaimDecision::Delete => {
                        return Err(DbError::Internal(
                            "unique repair: decide(Claim) returned Delete — unreachable",
                        ));
                    }
                }
            }
            Some(claims) => {
                // Two or more declared claimants for one key: a conflict
                // regardless of what is stored (mirrors `validate`'s own
                // rule — see `expected::UniqueExpected`'s doc). No winner
                // without the business rule, so nothing is written.
                let stats = stats_by_ns
                    .entry(ns)
                    .or_insert_with(|| NamespaceRepairStats {
                        namespace: ns,
                        ..Default::default()
                    });
                let mut owners: Vec<(String, String)> = claims
                    .iter()
                    .map(|c| (c.source.to_string(), c.source_key.clone()))
                    .collect();
                for claim in claims {
                    repaired_sources.remove(claim.source);
                }
                if let Some(row) = row {
                    retain_owner_source_dirty(reg, &mut repaired_sources, &row.owner());
                }
                owners.sort();
                stats.conflicts.push(NamespaceConflict {
                    key: hex(&key),
                    owners,
                });
            }
        }
    }

    let mut per_namespace: Vec<NamespaceRepairStats> = stats_by_ns.into_values().collect();
    per_namespace.sort_by_key(|s| s.namespace);
    for s in &mut per_namespace {
        s.conflicts.sort_by(|a, b| a.key.cmp(&b.key));
    }

    // A skipped source has an incomplete expected set. Its namespace may be
    // repaired for other producers, but this source's marker must survive.
    for source in &skipped {
        repaired_sources.remove(source.name);
    }

    let mut repaired_sources: Vec<_> = repaired_sources.into_iter().collect();
    repaired_sources.sort_unstable();

    Ok((per_namespace, skipped, repaired_sources))
}

#[cfg(test)]
mod tests {
    use super::{fail_next_namespace_repair, take_fail_next_namespace_repair};

    #[test]
    fn namespace_repair_failpoint_is_test_thread_isolated() {
        fail_next_namespace_repair();
        let other_thread_consumed = std::thread::spawn(take_fail_next_namespace_repair)
            .join()
            .expect("join failpoint observer");

        assert!(
            !other_thread_consumed,
            "another test thread consumed the failpoint"
        );
        assert!(
            take_fail_next_namespace_repair(),
            "the arming thread lost its failpoint"
        );
    }
}