polyc-query 2026.8.3

//! What the commit feed tells the search index, and how little it is allowed
//! to do about it.
//!
//! A mark runs on whatever task is draining the feed, and that task's next
//! chunk must not wait on a replay: every replay this index performs therefore
//! happens on a background worker, and marking only records what the worker
//! still owes. The mark is a map lookup and an integer merge, nothing more.
//!
//! # Two inputs, because the feed has one blind spot
//!
//! [`CommitMarks::note_commit`] takes a chunk of the durable commit feed
//! (#1565, chunk B6), which is the steady state. A destroy, an excision, a
//! repair, or a migration changes a partition's durable content without
//! committing, so none of them appears on the feed at all —
//! [`CommitMarks::note_partition_change`] is how they arrive, from whoever
//! issued the command, after that command returned its receipt.
//!
//! # Why the mark must not panic
//!
//! A panic here would take out the task draining the feed, and that
//! partition's mark would be lost — with nothing on the read path ever
//! disagreeing with the stale watermark left behind. So the mark does only
//! infallible work (integer comparison, map insert), and a poisoned lock
//! escalates to `DirtySet::degraded` rather than unwinding: a mark that
//! cannot be recorded makes the whole index refuse, which is loud, instead of
//! silently dropping one conversation's updates, which is not.
//!
//! # A missed mark costs latency, not correctness
//!
//! Feed delivery is at-least-once and a subscription can stop without saying
//! so, so no mark may be the only thing standing between a conversation and
//! its coverage. Two properties close that. Marking is idempotent by
//! construction — `Pending::merge` keeps whichever side demands more work,
//! so a redelivered chunk merges to the same pending state it already had.
//! And [`crate::search_index::worker::SearchIndexWorker::reconcile`] sweeps
//! the whole deployment on its own schedule, re-establishing coverage for
//! every partition whether anything marked it or not; a consumer that knows
//! its own feed went dark says so with [`CommitMarks::note_coverage_doubt`]
//! and brings that sweep forward.
//!
//! # Why the queue is bounded, and what overflow means
//!
//! An unbounded dirty set is a memory leak with a busy fleet behind it. A
//! bounded one that silently forgets is worse: the forgotten partition keeps
//! serving a stale watermark forever. So overflow sets `DirtySet::degraded`,
//! and a degraded index refuses every search until a full reconcile
//! re-establishes coverage.
//!
//! An earlier draft of the design record asked overflow to enqueue the
//! partition for rebuild and apply backpressure, and named fleet-wide refusal
//! as the outcome to avoid. Neither half of that is available here: the set is
//! already full by definition, and an observer runs on the write path where it
//! cannot block. Global refusal is the only fail-closed option left with a
//! hard bound — a search quietly answering "not found" from a partition nobody
//! re-read is a correctness failure, where a loud refusal is an availability
//! one. The record now states this behavior
//! (`docs/proposals/participation-scoped-agent-search.md`, "No silent document
//! eviction").
//!
//! The forgotten partition is recovered by
//! [`crate::search_index::worker::SearchIndexWorker::reconcile`], which
//! enumerates every partition rather than draining this queue — so it does not
//! depend on the entry that was dropped, and it is the only caller of
//! [`DirtySet::clear_degraded`].
//!
//! With that reconcile in place the bound stays where it is. Raising it trades
//! a larger resident map for a later refusal without removing the refusal, and
//! the recovery no longer depends on the bound at all: overflow now costs one
//! full sweep, not a restart.

use std::collections::BTreeMap;
use std::sync::{Mutex, PoisonError};

use polyc_proto::kinds;
use polyc_state::feed::FeedRecord;

use crate::feed::PartitionChange;

/// Prefix every conversation partition carries — `partition_for` in the
/// control plane.
///
/// Duplicated here rather than depended on, exactly as
/// [`crate::dashboard`] duplicates it: a Component cannot depend on the
/// Container that composes it.
const CONVERSATION_PARTITION_PREFIX: &str = "conv-";

/// Whether `partition` is a conversation this index has any business tracking.
///
/// Marks arrive for every partition in the deployment — `persona-<id>-mem`,
/// `admin-audit`, `skill-share-ledger`, the enrollment and workqueue
/// partitions. Without this filter a memory erasure on `persona-abc-mem` would
/// insert a removal for a conversation the index has never held, consume the
/// tracking budget that overflow turns into a fleet-wide refusal, and hand the
/// worker a partition it cannot act on.
///
/// The sibling projection in this crate guards its own input the same way
/// (`crate::dashboard::conversation_id_from_partition`).
///
/// Shared with [`crate::search_index::worker`] rather than restated there: the
/// reconcile enumerates every partition in the deployment and must admit
/// exactly the set this observer marks, or a conversation is either swept by
/// one and not the other — a partition the reconcile establishes coverage for
/// but no append ever re-marks, or one marked dirty that no reconcile can
/// repair.
pub(crate) fn is_conversation_partition(partition: &str) -> bool {
    partition
        .strip_prefix(CONVERSATION_PARTITION_PREFIX)
        .is_some_and(|id| !id.is_empty())
}

/// The most partitions the dirty set tracks before it declares itself
/// degraded.
///
/// Sized well above any plausible burst of concurrently active conversations,
/// because crossing it is a correctness event rather than a load-shedding one:
/// past this point the index cannot promise it knows what changed.
pub(crate) const MAX_TRACKED_PARTITIONS: usize = 16_384;

/// What the worker still owes a partition.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Pending {
    /// Committed turns landed up to this EXCLUSIVE position; the worker can
    /// index forward from wherever its watermark currently sits.
    IndexThrough(u64),
    /// The partition's journal changed in a way no forward replay can absorb —
    /// a rewrite, a repair, or a migration destination. Everything already
    /// indexed for it is suspect and must be recomputed from scratch.
    Rebuild,
    /// The partition's journal moved elsewhere. Its records must leave no
    /// trace: the conversation is alive under a new id, and coverage for it
    /// resolves from the destination.
    Remove,
    /// The partition's journal was destroyed. Its records must go and the
    /// destroyed state must be recorded, because there is nothing left to
    /// rebuild from and nothing may ever publish for it again.
    Destroy,
}

impl Pending {
    /// Combine two pending states for the same partition, keeping whichever
    /// demands more work.
    ///
    /// Ordering is deliberate and total: [`Pending::Destroy`] beats everything
    /// (it is terminal — a destroyed conversation may never be published to
    /// again), [`Pending::Remove`] beats any indexing work (a partition whose
    /// journal moved cannot be indexed from where it no longer is),
    /// [`Pending::Rebuild`] beats any forward index (a compaction invalidates
    /// the watermark a forward index would resume from), and two forward
    /// positions merge to the higher. Any other rule risks resuming a forward
    /// index across a compaction, which is exactly how an index keeps serving
    /// text the journal no longer holds.
    ///
    /// Destroy over remove matters in one direction only: a conversation id
    /// whose journal migrated away and was then re-created and destroyed must
    /// end at the tombstone. The reverse pairing cannot arise — nothing
    /// migrates a partition that no longer exists — and if it somehow did, the
    /// tombstone refuses, which is the safe direction.
    fn merge(self, other: Self) -> Self {
        match (self, other) {
            (Self::Destroy, _) | (_, Self::Destroy) => Self::Destroy,
            (Self::Remove, _) | (_, Self::Remove) => Self::Remove,
            (Self::Rebuild, _) | (_, Self::Rebuild) => Self::Rebuild,
            (Self::IndexThrough(a), Self::IndexThrough(b)) => Self::IndexThrough(a.max(b)),
        }
    }
}

/// The set of partitions the index owes work on.
///
/// Shared between [`CommitMarks`] (which only ever adds) and the worker (which
/// drains). Deliberately holds no handle to the store, the event log, or
/// anything else that could block: the write path reaches this and nothing
/// beyond it.
#[derive(Debug, Default)]
pub(crate) struct DirtySet {
    inner: Mutex<DirtyInner>,
}

#[derive(Debug, Default)]
struct DirtyInner {
    pending: BTreeMap<String, Pending>,
    degraded: bool,
    /// How many times this set has degraded, ever.
    ///
    /// Monotonic, and the reason a reconcile can tell "still degraded from the
    /// event I started sweeping after" from "degraded again while I swept". The
    /// flag alone cannot: marks land on the task draining the feed, so a mark
    /// dropped by overflow AFTER the sweep visited that partition would be
    /// cleared by a sweep that never accounted for it.
    degrades: u64,
}

impl DirtyInner {
    /// Record that this set lost track of something.
    const fn degrade(&mut self) {
        self.degraded = true;
        self.degrades = self.degrades.saturating_add(1);
    }
}

impl DirtySet {
    /// Record that `partition` has committed turns through `boundary`
    /// (exclusive).
    ///
    /// Infallible and allocation-bounded: the only work under the lock is a
    /// map lookup and an integer merge.
    pub(crate) fn mark(&self, partition: &str, pending: Pending) {
        let mut inner = self.lock();

        if let Some(existing) = inner.pending.get_mut(partition) {
            *existing = existing.merge(pending);
            return;
        }

        if inner.pending.len() >= MAX_TRACKED_PARTITIONS {
            // Forgetting this partition would leave it serving a stale
            // watermark with nothing to disagree with it. Refusing everything
            // is worse for availability and better for correctness.
            inner.degrade();
            return;
        }

        inner.pending.insert(partition.to_owned(), pending);
    }

    /// Take everything currently outstanding, leaving the set empty.
    ///
    /// The degraded flag is NOT cleared here: only a completed full reconcile
    /// may clear it, via [`DirtySet::clear_degraded`]. Draining the queue
    /// proves the worker caught up with what it still knew about, never with
    /// what overflow already discarded.
    pub(crate) fn drain(&self) -> BTreeMap<String, Pending> {
        std::mem::take(&mut self.lock().pending)
    }

    /// Whether `partition` is outstanding, and what it owes, without taking it.
    ///
    /// A read, not a drain: the worker's own progress runs through
    /// [`DirtySet::drain`], and observing what is owed must not consume it.
    pub(crate) fn pending_for(&self, partition: &str) -> Option<Pending> {
        self.lock().pending.get(partition).copied()
    }

    /// Whether the index has lost track of what changed and must refuse.
    pub(crate) fn degraded(&self) -> bool {
        self.lock().degraded
    }

    /// How many partitions still owe work, for the drain-lag metric.
    pub(crate) fn pending_len(&self) -> usize {
        self.lock().pending.len()
    }

    /// Declare the index degraded for a reason that did not come from this set
    /// overflowing.
    ///
    /// The only caller outside this module is the Container's wiring, and it
    /// has exactly one reason to use it: the deployment's term key was MINTED
    /// rather than read back. Every stored segment then carries a key identity
    /// nothing on disk matches, so every conversation's coverage reads
    /// unreadable and its participants' search refuses — permanently, because
    /// nothing sweeps on a key change and a conversation that never commits
    /// another turn is never revisited. Degrading turns that into one reconcile
    /// that rebuilds the fleet, which is what the mint's own rationale always
    /// claimed happened and never did.
    pub(crate) fn degrade(&self) {
        self.lock().degrade();
    }

    /// How many times this set has degraded, ever.
    ///
    /// A reconcile snapshots this BEFORE its sweep and hands it back to
    /// [`DirtySet::clear_degraded`], which is what makes the clear safe against
    /// a degrade that lands mid-sweep. The flag alone cannot express that: it
    /// reads the same whether the sweep is about to repair the event that set
    /// it or a newer one it never saw.
    pub(crate) fn degrade_count(&self) -> u64 {
        self.lock().degrades
    }

    /// Whether the set still owes a destroy or a removal.
    ///
    /// The one class of work a reconcile cannot retry for itself: the sweep
    /// enumerates partitions that still EXIST, and a destroyed or migrated-away
    /// one is gone from that listing. So a failed removal lives on as its
    /// re-queued mark, and this is how the sweep sees it — a deployment told to
    /// forget user text must not report its index whole while the removal is
    /// still outstanding.
    pub(crate) fn has_pending_removal(&self) -> bool {
        self.lock()
            .pending
            .values()
            .any(|pending| matches!(pending, Pending::Destroy | Pending::Remove))
    }

    /// Clear the degraded flag after a full reconcile has re-established
    /// coverage for every partition, unless the set degraded again meanwhile.
    ///
    /// `observed` is [`DirtySet::degrade_count`] as the caller read it before
    /// its sweep. The comparison happens under the same lock as the clear, so
    /// there is no window between checking and clearing — a degrade recorded on
    /// the write path anywhere between the snapshot and this call blocks it.
    ///
    /// Returns whether the flag was cleared.
    pub(crate) fn clear_degraded(&self, observed: u64) -> bool {
        let mut inner = self.lock();
        if inner.degrades != observed {
            return false;
        }
        inner.degraded = false;
        true
    }

    /// Lock the inner state, recovering from a poisoned lock rather than
    /// unwinding.
    ///
    /// A poisoned lock means some other holder panicked mid-update, so the
    /// map may be missing a mark it was in the middle of applying. Recovering
    /// the data and flagging the index degraded keeps this callable from the
    /// feed-draining task — where an unwind would take the drain down — while
    /// still refusing to answer from state that may have lost an update.
    ///
    /// The escalation goes through [`DirtyInner::degrade`], so a poisoning that
    /// lands during a reconcile blocks that pass's clear exactly as an overflow
    /// does — including the poisoning this very call discovers.
    fn lock(&self) -> std::sync::MutexGuard<'_, DirtyInner> {
        match self.inner.lock() {
            Ok(guard) => guard,
            Err(poisoned) => {
                let mut guard = PoisonError::into_inner(poisoned);
                guard.degrade();
                guard
            }
        }
    }
}

/// Marks partitions dirty as commits and mutations reach this index.
///
/// # Only ever driven alongside a running worker
///
/// Driving it by itself is safe from a crash standpoint — every method here is
/// infallible — but the set it fills would never be emptied. It grows by one
/// entry per newly-dirtied conversation until it crosses
/// `MAX_TRACKED_PARTITIONS`, at which point `DirtySet::degraded` flips and
/// stays flipped until something re-establishes coverage for every partition.
///
/// [`crate::search_index::SearchIndex`] is what makes the correct wiring the
/// easy one: it hands out this handle and owns
/// `SearchIndexWorker` over the SAME
/// `DirtySet`, so a Container cannot accidentally build a second set that
/// nothing drains, and `marks()` borrows where `run()` consumes, so the handle
/// cannot outlive its own loop by construction.
///
/// It does not make the mistake impossible. `open(...).marks()` followed by
/// dropping the [`crate::search_index::SearchIndex`] compiles, and driving that
/// handle would fill a set nothing empties. Closing that by construction would
/// mean the handle could not exist before the worker is supervised, which no
/// ownership arrangement here expresses — so it is stated rather than claimed
/// away, and the Container's own wiring (`crates/control-plane`) is where the
/// pairing is checked.
///
/// Holds nothing but the shared `DirtySet`.
pub struct CommitMarks {
    dirty: std::sync::Arc<DirtySet>,
}

impl CommitMarks {
    /// Build a handle feeding `dirty`.
    pub(crate) const fn new(dirty: std::sync::Arc<DirtySet>) -> Self {
        Self { dirty }
    }

    /// Note the highest committed turn boundary one chunk of `partition`'s
    /// commit feed established.
    ///
    /// Only `turn_complete` matters. A turn's `turn_start` and its input
    /// messages commit BEFORE the harness is dialed, so treating any commit as
    /// indexable would index text from a turn that may never commit — and an
    /// interrupted turn would become searchable though the conversation never
    /// accepted it. The boundary is therefore the position AFTER a
    /// `turn_complete`, which is the first position not yet known to be
    /// committed.
    ///
    /// A chunk carrying no `turn_complete` marks nothing: it moves the journal
    /// tail but not the committed tail, and coverage is defined against the
    /// latter.
    ///
    /// Safe to call with a chunk this index already saw. Delivery is
    /// at-least-once, so a redelivery is expected: the boundary a chunk
    /// establishes is a function of the chunk's own positions, and
    /// `Pending::merge` keeps the higher of two boundaries, so re-marking
    /// leaves the pending state exactly where it was.
    pub fn note_commit(&self, partition: &str, commits: &[FeedRecord]) {
        if !is_conversation_partition(partition) {
            return;
        }

        let mut boundary: Option<u64> = None;
        let mut excised = false;

        for (position, event) in crate::feed::commit_events(commits) {
            let (base, _turn_id) = kinds::parse(&event.kind);
            if base == kinds::TURN_COMPLETE {
                // Exclusive: everything strictly below this is committed.
                let candidate = position.saturating_add(1);
                boundary = Some(boundary.map_or(candidate, |seen: u64| seen.max(candidate)));
            } else if base == kinds::TAINT_EXCISION {
                excised = true;
            }
        }

        // A verified excision marker is an ORDINARY commit: it arrives on the
        // feed like any other and never changes a partition's identity, so
        // nothing else in this pipeline would notice it. It forces a REBUILD
        // rather than a forward index, because the positions it names sit below
        // the current watermark — a forward window would never contain them,
        // and the postings carry-forward would restore the excised text
        // verbatim. Serving excised content is the worst failure this feature
        // can have.
        if excised {
            self.dirty.mark(partition, Pending::Rebuild);
            return;
        }

        if let Some(boundary) = boundary {
            self.dirty.mark(partition, Pending::IndexThrough(boundary));
        }
    }

    /// Note that a partition's journal changed underneath the index.
    ///
    /// Reported by whoever issued the mutating command, once that command
    /// returned its receipt — the feed carries none of these. Transport
    /// success is not authority (INV-22): a command that has not earned a
    /// receipt may never commit, and tombstoning a conversation on the
    /// strength of one would erase an index nothing asked to erase.
    ///
    /// A destroyed partition and the source side of a migration both have no
    /// journal left to rebuild from, but they are NOT the same state and the
    /// projection treats them oppositely: a destroy records an authoritative
    /// tombstone that forbids every later publish, while a migration source
    /// must read as though it were never indexed, because the conversation is
    /// alive under its new id and its coverage resolves from the destination.
    /// Collapsing the two would either tombstone a live conversation or leave
    /// a destroyed one looking merely unindexed, so the distinction is carried
    /// here rather than guessed at by the worker.
    ///
    /// A rewrite compacts or renumbers positions, which invalidates the
    /// watermark a forward index would resume from — so it forces a full
    /// recompute rather than a resume.
    pub fn note_partition_change(&self, partition: &str, change: PartitionChange) {
        if !is_conversation_partition(partition) {
            return;
        }

        let pending = match change {
            PartitionChange::Destroyed => Pending::Destroy,
            PartitionChange::MigratedAway => Pending::Remove,
            PartitionChange::Rewritten => Pending::Rebuild,
        };

        self.dirty.mark(partition, pending);
    }

    /// Note that a partition is being followed for the first time and nothing
    /// delivered so far describes its existing contents.
    ///
    /// A subscription that registers against a partition starts from a
    /// snapshot taken at the current head, so commits that predate the
    /// registration never arrive on the feed. [`Self::note_commit`] therefore
    /// never sees them, and a partition whose only commits predate its
    /// follower would stay unindexed until the coverage sweep noticed.
    ///
    /// Marks a full recompute rather than a forward index, for the same
    /// reason an excision does: the positions involved sit below any
    /// watermark a forward window would start from.
    pub fn note_bootstrap(&self, partition: &str) {
        if !is_conversation_partition(partition) {
            return;
        }
        self.dirty.mark(partition, Pending::Rebuild);
    }

    /// Reports whether `partition` is waiting on a full recompute.
    ///
    /// The one thing a Container driving both projections can check about this
    /// one: whether the index still owes this partition a rebuild. Answering it
    /// takes nothing off the queue, so asking never costs the worker work it
    /// would otherwise have done.
    #[must_use]
    pub fn awaits_rebuild(&self, partition: &str) -> bool {
        self.dirty.pending_for(partition) == Some(Pending::Rebuild)
    }

    /// Declare that this index can no longer promise it knows what changed.
    ///
    /// The one call a Container makes when its own feed went dark — a
    /// subscription that ended, refused terminally, or could not be resumed.
    /// Nothing about the marks already recorded is wrong; what is gone is the
    /// promise that they are the whole story, and the honest response is the
    /// same one an overflow gets: refuse until a full reconcile re-establishes
    /// coverage for every partition. A search quietly answering "not found"
    /// from a conversation nobody re-read is a correctness failure; a loud
    /// refusal is an availability one.
    pub fn note_coverage_doubt(&self) {
        self.dirty.degrade();
    }

    /// Whether this index is currently refusing because it cannot promise it
    /// knows what changed.
    ///
    /// Read by a Container that wants to report the state honestly — a search
    /// surface answering nothing right now is answering nothing for a reason,
    /// and that reason is here rather than inferred from an empty result.
    /// Cleared only by a completed sweep.
    #[must_use]
    pub fn coverage_is_doubted(&self) -> bool {
        self.dirty.degraded()
    }
}

#[cfg(test)]
mod tests;