polyc-query 2026.9.0

//! The background worker that turns a dirty conversation into published
//! segments, plus the reconcile that re-establishes coverage after the
//! observer loses track of one.
//!
//! The observer marks; this replays. Everything expensive lives here, off the
//! write path: a bounded range replay of whatever a conversation committed
//! since its last watermark, the committed-and-excision-aware projection of
//! that range ([`super::project::project`]), and one atomic publish of the
//! resulting segment.
//!
//! # Committed-only, and what that costs
//!
//! A turn's `turn_start` and its input messages commit before the harness is
//! dialed, so indexing on arrival would make an interrupted turn searchable
//! though the conversation never accepted it. The worker therefore indexes
//! only up to a committed turn boundary the observer saw, and
//! [`polyc_facts::committed_message_facts`] — reached through
//! [`super::project::project`] — does the filtering, so this pipeline uses the
//! same definition of "committed" every other consumer does rather than a
//! second one that could drift.
//!
//! # A partial index is worse than no index
//!
//! Several failure paths converge on the same answer: mark the conversation
//! unavailable and publish nothing. A bounded replay that trips its byte
//! budget has read only part of a turn, and a segment built from that would
//! sit behind a clean-looking watermark claiming coverage it does not have. So
//! would a conversation whose stored rows exceed the store's read cap. Both
//! refuse rather than publish, because a search that answers "not found" from
//! a partially indexed conversation is the one failure this whole design
//! exists to prevent.
//!
//! # Delta in, prefix on disk
//!
//! [`super::store::SearchProjection::append`] takes the DELTA — the messages
//! this pass indexed — never the conversation. The prefix already lives in the
//! earlier segments and a reader unions them, so re-publishing the whole
//! conversation per turn would reintroduce exactly the O(N²) the segment
//! layout exists to remove. That is why [`super::project::project`] folds the
//! replayed range and nothing else: it has no way to carry earlier postings
//! forward, so a forward pass cannot accidentally republish a prefix. Only a
//! rebuild, which replays from zero and replaces every segment, writes a whole
//! conversation at once.
//!
//! The same bound holds on what this pass READS. Everything it needs about the
//! stored prefix — is it still decodable, how many segments, how many rows
//! outside the folded base — comes from Parquet footers
//! ([`super::store::SearchProjection::segment_stats`]) and from what the write
//! itself returns. Nothing on the per-turn path decodes a row. A pass that
//! decoded the conversation to answer those questions would be O(N) per turn
//! however cheap the write was, which is the same quadratic wearing the other
//! half of the pair.
//!
//! # What this does NOT do yet
//!
//! It leaves one gap named rather than hidden: a conversation whose
//! journal vanished without a mutation notification — the only way that
//! happens is a destroy whose mark was dropped by an overflowing dirty set —
//! keeps its segments on disk. [`SearchIndexWorker::reconcile`] refuses that
//! conversation (its coverage cannot be verified against a journal that is not
//! there) but does not delete it, because deleting from the ABSENCE of a
//! partition would tombstone every live conversation on any future day
//! `list_partitions` under-reports. Reclaiming those rows needs an
//! administrative sweep with a stronger signal than absence.

use std::sync::Arc;
use std::time::Duration;

use crate::journal::{JournalError, PartitionJournal};
use tokio_util::sync::CancellationToken;

use super::marks::{DirtySet, MAX_TRACKED_PARTITIONS, Pending, is_conversation_partition};
use super::project;
use super::store::{
    self, CoverageState, ExcisionScan, JournalState, SearchProjection, SegmentStats, StoreError,
};
use super::terms::TermKey;
use crate::metrics;

/// Bytes one conversation's catch-up replay may read before it gives up.
///
/// Generous enough that ordinary catch-up never trips it, and finite so a
/// pathological conversation cannot stall the worker indefinitely. Tripping it
/// is not a truncation: the conversation is marked unavailable, never
/// published half-read.
const REPLAY_BUDGET_BYTES: u64 = 32 * 1024 * 1024;

/// Segments a conversation's directory may hold before the next successful
/// append folds them into one.
///
/// A first guess, not a measurement. Every read opens and parses each
/// segment's footer, so the per-file cost is what this bounds; past roughly a
/// dozen files that overhead starts to dominate the rows themselves. The
/// number is deliberately well under the point where a directory listing gets
/// expensive, because compacting early costs one rewrite and compacting late
/// costs every reader.
///
/// A fold leaves exactly one segment, so this is an EDGE the conversation
/// crosses once per fold rather than a level it sits above forever.
const COMPACT_SEGMENT_THRESHOLD: u32 = 16;

/// Rows a conversation may accumulate OUTSIDE its folded base before the next
/// successful append folds them into one.
///
/// Measured against [`super::store::SegmentStats::unmerged_rows`], never the
/// conversation's total. Compaction merges rows; it never removes them, so a
/// trigger on the total is crossed once and then true on every append
/// afterwards — three O(rows) passes per committed turn, forever, which is the
/// quadratic the segment layout exists to remove. Against the unmerged count it
/// is an edge: a fold returns the number to zero.
///
/// The value is also a first guess. A segment count alone misses the
/// conversation with three enormous segments, which is precisely the shape that
/// hurts most: rows cost about 150 bytes apiece once decoded, so this ceiling
/// is roughly 75 MiB of decoded postings — an order of magnitude under the
/// store's own `MAX_POSTINGS_ROWS` read cap, which is the point at which a
/// conversation stops being readable at all. Folding well before that keeps the
/// cap a backstop rather than a destination.
const COMPACT_ROW_CEILING: u64 = 500_000;

/// Conversations one [`SearchIndexWorker::reconcile`] pass may visit.
///
/// The sweep is O(every partition in the deployment) by construction, so it
/// needs a stop. A pass that hits this bound has NOT re-established coverage
/// everywhere and therefore does not clear `degraded` — it logs the truncation
/// instead, which is the signal that the deployment outgrew both this bound
/// and the dirty set's.
const RECONCILE_PARTITION_LIMIT: usize = 65_536;

/// How long [`SearchIndexWorker::run`] waits between drains.
///
/// A mark lands as soon as the commit feed delivers, so this is purely how
/// long a committed turn waits to become searchable. Short enough that the
/// delay is not user-visible, long enough that an idle deployment is not
/// polling a lock in a tight loop.
const DRAIN_INTERVAL: Duration = Duration::from_secs(2);

/// How long the index may go without proving its own coverage.
///
/// The marks are the steady-state path and this is not: a sweep costs O(every
/// conversation in the deployment), so it runs on a slow clock. What it buys
/// is the one thing marking cannot promise on its own. Marks arrive over a
/// durable feed now (#1565, chunk B6), and a feed can stop delivering without
/// saying so — a subscription that ends quietly, a mutation whose receipt
/// nobody was left to hear. Either would otherwise leave a conversation
/// unindexed forever, with a stale watermark and nothing to disagree with it.
///
/// So the sweep is unconditional rather than degraded-only. A consumer that
/// KNOWS its feed went dark still says so
/// ([`super::marks::CommitMarks::note_coverage_doubt`]) and gets the faster
/// path; this covers the case where nothing knew.
const COVERAGE_SWEEP_INTERVAL: Duration = Duration::from_mins(30);

/// How long [`SearchIndexWorker::run`] waits before sweeping again after a
/// sweep that did not clear the flag.
///
/// Without it a reconcile that cannot finish — a truncated enumeration, a
/// refusal the projection would not accept — is retried every
/// [`DRAIN_INTERVAL`], and each retry is O(every partition in the deployment).
/// Whatever blocks a clear needs a fix from outside this worker, so
/// hammering it starves the drain instead of resolving anything. A caller that
/// wants a sweep NOW calls [`SearchIndexWorker::reconcile`] directly; this
/// paces only the automatic one.
const RECONCILE_RETRY_INTERVAL: Duration = Duration::from_mins(1);

/// Bytes the excision scan behind [`SearchProjection::verified_coverage`] may
/// read before it gives up and answers [`ExcisionScan::Unknown`].
///
/// Far smaller than [`REPLAY_BUDGET_BYTES`] because the range is far smaller: a
/// pass records the ceiling it replayed, so what this reads is only whatever
/// landed since — an in-flight turn and the marker itself. Overshooting is not
/// a truncation, because `Unknown` refuses exactly as `Pending` does; it only
/// means the conversation waits for the rebuild that will move the frontier.
const EXCISION_SCAN_BUDGET_BYTES: u64 = 4 * 1024 * 1024;

/// How far the journal may run ahead of a pinned open-turn barrier before the
/// pass says so at warning level.
///
/// A barrier itself is ordinary — every conversation with a turn in flight has
/// one for the length of that turn. What is not ordinary is a barrier that
/// stays put while thousands of events pile up behind it, which is what an
/// orphaned dispatch or an indefinitely-held approval looks like from here.
/// The number is a first guess: high enough that a normal turn never trips it,
/// low enough to fire long before the pass reaches [`REPLAY_BUDGET_BYTES`] and
/// refuses the conversation outright.
const BARRIER_LAG_WARN_EVENTS: u64 = 10_000;

/// What one conversation's indexing attempt produced.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum Outcome {
    /// A segment was published covering through the given exclusive position.
    Published {
        /// The exclusive journal position now indexed through.
        indexed_through: u64,
    },
    /// The conversation was left unsearchable and the index refuses for it
    /// until some later pass succeeds.
    Unavailable {
        /// Why, for the log line and the metric.
        reason: UnavailableReason,
    },
    /// The conversation's segments were erased with no trace — the migration
    /// source, whose coverage now resolves from the destination.
    Removed,
    /// The conversation's rows are gone and its tombstone records that it was
    /// destroyed. Terminal: nothing publishes for it again.
    Destroyed,
    /// Nothing to do — the watermark already covers everything this pass could
    /// have indexed.
    AlreadyCurrent,
    /// A turn is still open, so the watermark stayed where it was even though
    /// the journal has moved on.
    ///
    /// Split out of [`Outcome::AlreadyCurrent`] because the two are opposite
    /// states easily conflated: "caught up" is the healthy end of
    /// a pass, and "pinned behind a turn that never closed" is a conversation
    /// silently accumulating unindexable committed turns behind a watermark
    /// that still advertises itself as searchable. Without this variant,
    /// nothing distinguishes them — no log line, no metric, no return value —
    /// so the second would stay unobservable until it eventually tripped the
    /// replay budget.
    ///
    /// This still does not FORCE anything, deliberately. The barrier is what
    /// makes publication committed-only ([`super::project::project`]), so
    /// indexing past it would make an uncommitted turn searchable — the one
    /// failure the barrier exists to prevent, traded for an availability
    /// problem it does not fix. The right repair is upstream: complete or fail
    /// the turn. This makes that visible instead of guessing.
    BarrierHeld {
        /// The `turn_start` position holding the watermark down.
        open_turn_at: u64,
        /// How far the journal has moved past that position.
        journal_lag: u64,
    },
    /// This process is stopping, so nothing was read and nothing is published.
    ///
    /// Distinct from [`Outcome::Unavailable`] on purpose. A read that did not
    /// reach the journal while everything is SHUTTING DOWN says nothing about
    /// this conversation, and marking it unsearchable would take a perfectly
    /// healthy one down until something rebuilt it. Leaving the stored
    /// watermark exactly where it was means the next boot resumes from it.
    ///
    /// Only a real shutdown reaches here. The same
    /// [`JournalError::Unreachable`] also answers a journal that is simply not
    /// answering while this process keeps serving, and that IS a failure worth
    /// publishing — [`PartitionJournal::is_stopping`] is what tells them
    /// apart.
    Deferred,
}

/// Why a conversation was left unsearchable.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum UnavailableReason {
    /// The catch-up replay hit its byte budget before reaching the boundary,
    /// so it read only part of what it needed.
    ReplayBudgetExceeded,
    /// The stored rows exceeded the store's read cap, so the prefix this pass
    /// would have to carry cannot be decoded at all.
    RecordTooLarge,
    /// The event log could not be read, for a reason that may not be there on
    /// the next attempt: a transport failure, or a partition worker that
    /// panicked and will be respawned.
    ReplayFailed,
    /// The event log answered, and what it answered with cannot be replayed at
    /// all: tamper evidence failed, the stored bytes will not decode, or an
    /// event is larger than every replay path's decode cap.
    ///
    /// A property of the conversation's own content, exactly as a budget
    /// overrun is. The identical replay produces the identical failure.
    SourceUnreadable,
    /// The journal holds no events, so there is nothing to publish and no
    /// honest coverage to publish it behind.
    SourceEmpty,
    /// The projection could not be read or written.
    StoreFailed,
}

impl UnavailableReason {
    /// Whether retrying this conversation unchanged could plausibly succeed.
    ///
    /// A budget, size, unreadable-source, or empty-journal failure is a property
    /// of the conversation's own content, so the identical attempt fails
    /// identically. Re-enqueuing one would spin the worker on a conversation it
    /// can never finish, replaying tens of megabytes each pass and starving
    /// every other conversation behind it — a livelock, not a repair. Those wait
    /// for the operational fix (a raised bound, an excision that shrinks the
    /// conversation, or the destroy whose notification went missing) that the
    /// log line exists to prompt.
    ///
    /// [`UnavailableReason::SourceUnreadable`] must never map to "retryable":
    /// a tamper-evidence failure, a corrupt blob, or an over-cap payload are
    /// all deterministic, so mapping them to
    /// [`UnavailableReason::ReplayFailed`] instead would re-queue them into
    /// precisely the two-second 32-MiB loop this split exists to prevent.
    const fn is_transient(self) -> bool {
        match self {
            Self::ReplayFailed | Self::StoreFailed => true,
            Self::ReplayBudgetExceeded
            | Self::RecordTooLarge
            | Self::SourceUnreadable
            | Self::SourceEmpty => false,
        }
    }

    /// A stable, low-cardinality label for this reason's metric series.
    const fn label(self) -> &'static str {
        match self {
            Self::ReplayBudgetExceeded => "replay_budget_exceeded",
            Self::RecordTooLarge => "record_too_large",
            Self::ReplayFailed => "replay_failed",
            Self::SourceUnreadable => "source_unreadable",
            Self::SourceEmpty => "source_empty",
            Self::StoreFailed => "store_failed",
        }
    }
}

/// How a [`JournalError`] that is not a shutdown should be classified.
///
/// Free-standing and total over the enum, so a new journal error class has to
/// be classified here rather than inheriting "retryable" by default — which is
/// how the deterministic cases ended up in a retry loop.
const fn replay_reason(error: &JournalError) -> UnavailableReason {
    match error {
        // The bytes themselves are the problem, and a second replay reads the
        // same bytes. This is the side that cannot livelock: a genuinely
        // transient storage failure then waits for the conversation's next
        // committed turn to re-mark it, or for a reconcile, rather than
        // replaying up to 32 MiB every two seconds forever.
        JournalError::Unreadable(_) => UnavailableReason::SourceUnreadable,
        // The read never reached the bytes at all — the authority did not
        // answer, or this process is stopping. Another attempt is worth making.
        JournalError::Unreachable(_) => UnavailableReason::ReplayFailed,
    }
}

/// Why a forward pass turned itself into a rebuild.
///
/// Carried as a type rather than a string so the retry bound can tell the two
/// apart: only a recovery rebuild is suppressible, because only it is about
/// availability. An excision rebuild is about correctness and is never skipped.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RebuildCause {
    /// The replayed window held a taint-excision marker naming positions the
    /// window cannot reach.
    Excision,
    /// The stored coverage belongs to another durable physical lineage.
    SourceReplacement,
    /// The conversation was already refusing, and only a rebuild clears that.
    UnavailableRecovery,
}

impl RebuildCause {
    /// A stable, low-cardinality label for this cause's metric series.
    const fn label(self) -> &'static str {
        match self {
            Self::Excision => "excision",
            Self::SourceReplacement => "source_replacement",
            Self::UnavailableRecovery => "unavailable_recovery",
        }
    }
}

/// What one attempt at [`SearchIndexWorker::index_pass`] produced.
///
/// The escalation is a separate variant rather than a recursive call, so the
/// "a forward pass has to become a rebuild" decision stays visible at the one
/// place that can bound how often it happens.
enum Pass {
    /// The pass finished; this is the conversation's outcome.
    Done(Outcome),
    /// The pass must be redone from position zero.
    RebuildInstead {
        /// Why, for the log line, the metric, and the retry bound.
        cause: RebuildCause,
    },
}

/// What one [`SearchIndexWorker::reconcile`] pass did.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ReconcileSummary {
    /// Conversations whose coverage this pass re-established or confirmed.
    pub(crate) visited: usize,
    /// Conversations this pass left unsearchable.
    pub(crate) refused: usize,
    /// Whether the pass reached every conversation, and therefore whether it
    /// was allowed to clear [`DirtySet::degraded`].
    pub(crate) complete: bool,
}

/// What the live journal says about a conversation.
///
/// The [`JournalState`] side of [`SearchProjection::verified_coverage`],
/// implemented here rather than in the store because this is the layer that
/// holds a journal handle: the store stays storage-only and never learns what
/// a partition journal is.
#[derive(Clone)]
pub(crate) struct LiveJournal {
    journal: Arc<dyn PartitionJournal>,
}

impl LiveJournal {
    /// Answer incarnation questions from `journal`.
    pub(crate) const fn new(journal: Arc<dyn PartitionJournal>) -> Self {
        Self { journal }
    }
}

impl JournalState for LiveJournal {
    async fn source_incarnation(
        &self,
        partition: &str,
    ) -> Option<polyc_state::revision::PartitionIncarnation> {
        self.journal
            .partition_incarnation(partition.to_owned())
            .await
            .ok()
            .flatten()
    }

    /// Look for a taint-excision marker the stored footer never accounted for.
    ///
    /// Matches on the event KIND alone rather than verifying each marker's
    /// signature and conversation id the way [`project::project`] does.
    /// Over-detecting is nearly free — it schedules one rebuild, which then
    /// advances the frontier past the marker and answers [`ExcisionScan::Clear`]
    /// forever after — where under-detecting serves removed text. Verification
    /// belongs on the side that decides what to strip, not on the side that
    /// decides whether to look.
    async fn excision_since(&self, partition: &str, scanned_through: u64) -> ExcisionScan {
        let Ok(tail) = self
            .journal
            .partition_event_count(partition.to_owned())
            .await
        else {
            return ExcisionScan::Unknown;
        };
        if scanned_through >= tail {
            return ExcisionScan::Clear;
        }

        let Ok(replay) = self
            .journal
            .replay_range_with_positions_bounded(
                partition.to_owned(),
                scanned_through,
                tail,
                EXCISION_SCAN_BUDGET_BYTES,
            )
            .await
        else {
            return ExcisionScan::Unknown;
        };
        if replay.events.iter().any(|(_, event)| {
            polyc_proto::kinds::base(&event.kind) == polyc_proto::kinds::TAINT_EXCISION
        }) {
            return ExcisionScan::Pending;
        }
        // A truncated scan proves nothing about the part it did not reach, so
        // it must not answer `Clear` off the prefix it managed.
        if replay.budget_exceeded {
            ExcisionScan::Unknown
        } else {
            ExcisionScan::Clear
        }
    }
}

/// Replays dirty conversations and publishes their segments.
///
/// Owns the projection because it is the only writer: the single-replica
/// invariant means one process holds this, and every mutation to the index
/// goes through [`SearchIndexWorker::index_partition`].
pub(crate) struct SearchIndexWorker {
    projection: SearchProjection,
    journal: Arc<dyn PartitionJournal>,
    dirty: Arc<DirtySet>,
    term_key: TermKey,
    /// Derived once from [`SearchIndexWorker::term_key`] rather than per call:
    /// it is a keyed hash, and every read and write passes it.
    key_id: String,
    /// Bytes a catch-up replay may read. [`REPLAY_BUDGET_BYTES`] in
    /// production; a test lowers it to reach the refusal path without writing
    /// tens of megabytes.
    replay_budget: u64,
    /// Unmerged rows a conversation may hold before a fold.
    /// [`COMPACT_ROW_CEILING`] in production; a test lowers it so the trigger
    /// can be driven end to end without writing half a million rows.
    row_ceiling: u64,
    /// Set when [`SearchProjection::mark_unavailable`] itself failed during the
    /// current reconcile pass, and reset at the start of the next one.
    ///
    /// This is what stops a reconcile from clearing `degraded` off the back of
    /// a conversation whose stale segment still advertises `available: true`.
    /// A refusal the store accepted leaves coverage KNOWN (known-refused); a
    /// refusal it rejected leaves it unknown, which is exactly what `degraded`
    /// means.
    ///
    /// Scoped to the sweep on purpose. It covers only conversations the sweep
    /// itself re-examines, which is every partition `list_partitions` still
    /// reports. Work a sweep CANNOT retry — a destroy or a removal, whose
    /// partition is gone from that listing — is carried by
    /// [`DirtySet::has_pending_removal`] instead.
    unrecorded_refusal: bool,
    /// When [`SearchIndexWorker::run`] last swept, so a sweep that cannot clear
    /// the flag is retried at [`RECONCILE_RETRY_INTERVAL`] rather than on every
    /// drain.
    last_reconcile: Option<std::time::Instant>,
    /// When any sweep last ran, so the unconditional coverage sweep fires at
    /// [`COVERAGE_SWEEP_INTERVAL`].
    ///
    /// Initialized to the worker's own construction rather than `None`: there
    /// is deliberately no sweep at startup — the design record dropped its
    /// startup barrier because it costs O(every partition) on every boot — so
    /// the first periodic sweep lands one interval in.
    last_sweep: std::time::Instant,
    /// Conversations whose recovery rebuild already failed for a reason the
    /// identical rebuild would hit again.
    ///
    /// The bound on the design record's "an append while unavailable enqueues a
    /// rebuild". Executing that row means a committed turn landing on a refused
    /// conversation converts itself into a rebuild
    /// ([`SearchIndexWorker::index_partition`]) — which is the only recovery an
    /// otherwise quiet conversation has, since nothing else ever revisits it.
    /// Unbounded, though, it is a full replay per committed turn forever on a
    /// conversation that is deterministically broken.
    ///
    /// So a recovery rebuild that fails non-transiently records the
    /// conversation here and no later append tries again. Any successful
    /// publish, destroy, or removal for it clears the entry, and a reconcile
    /// clears the whole set — the sweep is the moment a deployment gets a fresh
    /// attempt at everything.
    ///
    /// Deliberately in memory and deliberately NOT a correctness record: losing
    /// it to a restart costs one extra rebuild attempt, which is the safe
    /// direction. Bounded by [`MAX_TRACKED_PARTITIONS`], and a full set stops
    /// suppressing rather than stops recovering.
    recovery_blocked: std::collections::BTreeSet<String>,
}

impl SearchIndexWorker {
    /// Build a worker over an already-open projection.
    ///
    /// The Container supplies every environment-derived input — where the
    /// projection lives, which term key it is written under, and the host it
    /// replays through — because this crate is a Component and reads no
    /// configuration of its own.
    pub(crate) fn new(
        projection: SearchProjection,
        journal: Arc<dyn PartitionJournal>,
        dirty: Arc<DirtySet>,
        term_key: TermKey,
    ) -> Self {
        let key_id = term_key.key_id();
        Self {
            projection,
            journal,
            dirty,
            term_key,
            key_id,
            replay_budget: REPLAY_BUDGET_BYTES,
            row_ceiling: COMPACT_ROW_CEILING,
            unrecorded_refusal: false,
            last_reconcile: None,
            last_sweep: std::time::Instant::now(),
            recovery_blocked: std::collections::BTreeSet::new(),
        }
    }

    /// Drain and apply on a fixed interval until `shutdown` is cancelled.
    ///
    /// Reached from the Container through [`super::super::SearchIndex::run`],
    /// which puts it in the supervised task set. It is a plain future rather
    /// than a detached spawn on purpose: a perpetual loop nobody joins is a
    /// loop nobody notices has stopped.
    ///
    /// A reconcile runs when the dirty set reports itself degraded, and
    /// otherwise every [`COVERAGE_SWEEP_INTERVAL`] regardless — see that
    /// constant for why a periodic sweep is what keeps a feed that stopped
    /// delivering from becoming a permanently unindexed conversation. There is
    /// deliberately no sweep at startup: the design record dropped its startup
    /// barrier because it costs O(every partition in the deployment) on every
    /// boot and still only catches what was already broken at boot, which
    /// [`SearchProjection::verified_coverage`] catches per read instead.
    ///
    /// # Cancellation safety
    ///
    /// Cancellation is observed between conversations, never inside one: a
    /// partly applied conversation would either lose its dirty mark or leave a
    /// half-written segment, and neither is recoverable from the queue. Work
    /// the drain did not reach is put back before this returns.
    // polychrome-work-family: search-projection
    pub(crate) async fn run(mut self, shutdown: CancellationToken) {
        loop {
            // `timeout` rather than `select!`: `cancelled()` is cancel-safe,
            // so racing it against a sleep needs no extra tokio feature and
            // cannot drop a wakeup.
            if tokio::time::timeout(DRAIN_INTERVAL, shutdown.cancelled())
                .await
                .is_ok()
            {
                break;
            }

            if self.sweep_is_due() {
                let degraded = self.dirty.degraded();
                match self.reconcile(&shutdown).await {
                    Ok(summary) => tracing::info!(
                        visited = summary.visited,
                        refused = summary.refused,
                        complete = summary.complete,
                        degraded,
                        "search index swept every conversation to re-establish coverage"
                    ),
                    Err(error) => tracing::error!(
                        %error,
                        "search index could not enumerate partitions to reconcile; coverage \
                         stays whatever the last sweep left"
                    ),
                }
                self.last_sweep = std::time::Instant::now();
                // Pace only a sweep that did NOT clear the flag. A fresh
                // overflow after a successful one is a new event and deserves
                // its own immediate sweep, not the tail of the previous one's
                // cooldown.
                self.last_reconcile = self.dirty.degraded().then(std::time::Instant::now);
            }

            for (partition, outcome) in self.drain_once(&shutdown).await {
                match outcome {
                    Outcome::Unavailable { reason } => tracing::warn!(
                        partition,
                        ?reason,
                        "search index left a conversation unsearchable"
                    ),
                    // Loud only once the lag says this is not an ordinary turn
                    // in flight. The state is otherwise routine, and a warning
                    // per pass per active conversation would bury the case that
                    // matters.
                    Outcome::BarrierHeld {
                        open_turn_at,
                        journal_lag,
                    } if journal_lag >= BARRIER_LAG_WARN_EVENTS => tracing::warn!(
                        partition,
                        open_turn_at,
                        journal_lag,
                        "search index cannot advance a conversation's watermark: a turn opened \
                         and never completed, so every committed turn behind it stays \
                         unindexable until that turn completes or fails"
                    ),
                    Outcome::BarrierHeld { .. }
                    | Outcome::Published { .. }
                    | Outcome::Removed
                    | Outcome::Destroyed
                    | Outcome::AlreadyCurrent
                    | Outcome::Deferred => {}
                }
            }
        }
    }

    /// Whether a sweep should run on this pass.
    ///
    /// Two independent reasons, and the degraded one is the urgent one: the
    /// index has said it lost track of something, so it sweeps as soon as its
    /// cooldown allows. Absent that, the sweep is the slow coverage proof —
    /// see [`COVERAGE_SWEEP_INTERVAL`] — which is what corrects a conversation
    /// no mark ever reached because nothing was left to mark it.
    fn sweep_is_due(&self) -> bool {
        if self.dirty.degraded() {
            return self.reconcile_is_due();
        }
        self.last_sweep.elapsed() >= COVERAGE_SWEEP_INTERVAL
    }

    /// Whether the degraded-driven sweep may run again.
    ///
    /// The first one is always due, and so is the first after a sweep that
    /// cleared the flag. Only a sweep that left the index degraded starts the
    /// [`RECONCILE_RETRY_INTERVAL`] cooldown, because whatever blocked it is
    /// resolved from outside this worker rather than by another O(every
    /// partition) pass.
    fn reconcile_is_due(&self) -> bool {
        self.last_reconcile
            .is_none_or(|last| last.elapsed() >= RECONCILE_RETRY_INTERVAL)
    }

    /// Drain everything currently outstanding and apply it.
    ///
    /// Returns each conversation's outcome, in partition order. A failure on
    /// one conversation never abandons the rest: an unavailable conversation
    /// is a coverage statement about that conversation alone, and stopping the
    /// drain would spread it to every conversation behind it in the queue.
    ///
    /// # Cancellation safety
    ///
    /// Whatever the drain took but did not apply is marked dirty again before
    /// returning, so a cancelled pass loses no work — [`DirtySet::mark`] merges
    /// rather than overwrites, so a mark that arrived meanwhile still wins on
    /// the side that demands more work.
    pub(crate) async fn drain_once(
        &mut self,
        shutdown: &CancellationToken,
    ) -> Vec<(String, Outcome)> {
        let pending = self.dirty.drain();
        let mut outcomes = Vec::with_capacity(pending.len());
        let mut remaining = pending.into_iter();

        for (partition, pending) in remaining.by_ref() {
            if shutdown.is_cancelled() {
                self.dirty.mark(&partition, pending);
                break;
            }
            let outcome = self.apply(&partition, pending).await;
            // A deferred pass read nothing and published nothing, so the work
            // is still owed. Putting the mark back is what keeps a panicked
            // partition worker — which answers the same `Closed` a shutdown
            // does — from freezing one conversation behind a watermark that
            // still advertises `available: true` with nothing queued to move
            // it. On a real shutdown the mark dies with the process either
            // way, so this costs nothing there.
            if outcome == Outcome::Deferred {
                self.dirty.mark(&partition, pending);
            }
            outcomes.push((partition, outcome));
        }

        for (partition, pending) in remaining {
            self.dirty.mark(&partition, pending);
        }

        metrics::record_search_index_queue(self.dirty.pending_len(), self.dirty.degraded());
        outcomes
    }

    /// Re-establish coverage for every conversation in the deployment, then
    /// clear the degraded flag.
    ///
    /// The only caller of [`DirtySet::clear_degraded`], and the reason
    /// overflow is not a one-way door. It enumerates partitions rather than
    /// draining the queue precisely because the queue is what lost an entry:
    /// the conversation it forgot is recoverable only from the storage
    /// directory.
    ///
    /// A conversation whose coverage verifies against its journal is caught up
    /// forward to the journal's current ceiling — the open-turn barrier in
    /// [`super::project::project`] decides what within that ceiling is
    /// actually committed, so a ceiling is safe where a guessed watermark
    /// would not be. Everything else is rebuilt from position zero: a stale
    /// index describes positions that no longer exist, and an unavailable one
    /// can only be cleared by a rebuild.
    ///
    /// The flag is cleared ONLY when the pass reached every conversation, every
    /// refusal it made was durably recorded, no removal is still outstanding,
    /// and nothing degraded the dirty set while the pass was running. A
    /// truncated or cancelled pass proves nothing about the conversations it
    /// never looked at, and a refusal the store rejected leaves a segment still
    /// advertising itself as searchable — which is the state `degraded` exists
    /// to describe.
    ///
    /// The last of those four is not hypothetical. The observer runs on the
    /// event-log write thread, and this worker's own
    /// [`SearchIndexWorker::fail_unavailable`] re-marks every transiently
    /// failed partition DURING the sweep — so a fleet-wide store failure
    /// overflows the set after the sweep has already visited those partitions,
    /// and a clear that only looked at where the sweep had been would declare
    /// the index whole over marks it never saw. [`DirtySet::degrade_count`] is
    /// snapshotted before the sweep and compared under the same lock that
    /// clears, so a degrade landing anywhere in between blocks it.
    ///
    /// # Errors
    ///
    /// Returns the [`JournalError`] class the partition list failed with. The
    /// flag stays set.
    ///
    /// # Cancellation safety
    ///
    /// Cancellation is observed between conversations. Each conversation's
    /// publish is atomic on its own, so an interrupted pass leaves the
    /// projection consistent and the flag untouched.
    pub(crate) async fn reconcile(
        &mut self,
        shutdown: &CancellationToken,
    ) -> Result<ReconcileSummary, JournalError> {
        // Snapshotted FIRST, so every degrade from here on — including one the
        // observer records on the write thread while this pass is halfway
        // through the deployment — is one this pass cannot clear.
        let degrade_mark = self.dirty.degrade_count();
        let partitions = self.journal.list_partitions().await?;
        let journal = LiveJournal::new(Arc::clone(&self.journal));

        // This pass re-examines every partition `list_partitions` reports, so a
        // coverage write that failed before it began is about to be retried.
        // Only a failure DURING the pass may block the clear. Destroys and
        // removals are NOT in that set — their partition is gone from the
        // listing, so no sweep can retry them — and they are carried by the
        // dirty set instead, which the clear below consults.
        self.unrecorded_refusal = false;
        // A sweep is the deployment-wide "try everything again" event, so it is
        // also where a conversation whose recovery rebuild was suppressed earns
        // another attempt.
        self.recovery_blocked.clear();

        let mut visited = 0usize;
        let mut refused = 0usize;
        let mut complete = true;

        for partition in partitions
            .iter()
            .filter(|partition| is_conversation_partition(partition))
        {
            if shutdown.is_cancelled() {
                complete = false;
                break;
            }
            if visited >= RECONCILE_PARTITION_LIMIT {
                complete = false;
                tracing::error!(
                    limit = RECONCILE_PARTITION_LIMIT,
                    "search index reconcile stopped at its partition bound; coverage is \
                     re-established for a prefix of the deployment only, so the index stays \
                     degraded"
                );
                break;
            }
            visited += 1;

            match self.reestablish(partition, &journal).await {
                Outcome::Unavailable { .. } => refused += 1,
                // The HOST is gone, so every conversation after this one would
                // fail the same way and none of them was actually examined.
                // Stopping — and refusing to call the pass complete — is what
                // keeps a shutdown from clearing `degraded` off a sweep that
                // proved nothing. A single panicked partition worker never
                // reaches here: `replay_failed` checks the host itself and
                // routes that to a per-conversation refusal, so one bad
                // partition can no longer abort the sweep and wedge the whole
                // fleet in `degraded` forever.
                Outcome::Deferred => {
                    complete = false;
                    break;
                }
                Outcome::Published { .. }
                | Outcome::Removed
                | Outcome::Destroyed
                | Outcome::BarrierHeld { .. }
                | Outcome::AlreadyCurrent => {}
            }
        }

        let cleared = complete
            && !self.unrecorded_refusal
            && !self.dirty.has_pending_removal()
            && self.dirty.clear_degraded(degrade_mark);
        metrics::record_search_index_reconcile(cleared, refused);
        metrics::record_search_index_queue(self.dirty.pending_len(), self.dirty.degraded());

        Ok(ReconcileSummary {
            visited,
            refused,
            complete,
        })
    }

    /// Bring one conversation's coverage back to something this index can
    /// stand behind.
    ///
    /// The forward catch-up below is safe only because
    /// [`SearchProjection::verified_coverage`] checks exact source and the
    /// durable excision frontier. Exact source catches a physical rewrite.
    /// The frontier catches a later excision that keeps the same incarnation.
    /// Either mismatch reports [`CoverageState::Stale`] and selects a rebuild
    /// before `degraded` can be cleared.
    async fn reestablish(&mut self, partition: &str, journal: &LiveJournal) -> Outcome {
        let state = match self
            .projection
            .verified_coverage(partition, &self.key_id, journal)
            .await
        {
            Ok(state) => state,
            // Unreadable is repaired by a rebuild, never in place — see
            // `StoreError::is_unreadable`.
            Err(err) if err.is_unreadable() => CoverageState::NeverIndexed,
            Err(err) => return self.store_failed(partition, &err).await,
        };

        match state {
            // Terminal. Rebuilding a destroyed conversation is the one thing
            // its tombstone exists to forbid.
            CoverageState::Destroyed => Outcome::Destroyed,
            CoverageState::Indexed(coverage) if coverage.available => {
                // Verified and searchable, but the mark that would have told
                // this worker how far the journal has moved is exactly what
                // overflow discarded. Catch up to the journal's ceiling rather
                // than trusting the watermark to be current.
                match self
                    .journal
                    .partition_event_count(partition.to_owned())
                    .await
                {
                    Ok(end) => self.index_partition(partition, Some(end)).await,
                    Err(error) => {
                        self.replay_failed(
                            partition,
                            &error,
                            "search index reconcile could not read a partition's length",
                        )
                        .await
                    }
                }
            }
            // Stale, never indexed, or indexed-but-refused. Only a rebuild
            // clears any of the three.
            CoverageState::Indexed(_) | CoverageState::NeverIndexed | CoverageState::Stale => {
                self.index_partition(partition, None).await
            }
        }
    }

    /// Apply one conversation's pending work.
    async fn apply(&mut self, partition: &str, pending: Pending) -> Outcome {
        match pending {
            Pending::Destroy => self.destroy(partition).await,
            Pending::Replace => self.replace(partition).await,
            Pending::Remove => self.remove(partition).await,
            Pending::Rebuild => self.index_partition(partition, None).await,
            Pending::IndexThrough(boundary) => {
                self.index_partition(partition, Some(boundary)).await
            }
        }
    }

    /// Record that a conversation was destroyed: its rows go, its tombstone
    /// stays.
    async fn destroy(&self, partition: &str) -> Outcome {
        match self.projection.destroy(partition, &self.key_id).await {
            Ok(()) => Outcome::Destroyed,
            Err(err) => {
                tracing::error!(
                    partition,
                    error = %err,
                    "search index could not record a destroyed conversation; its rows may still \
                     be on disk"
                );
                // A destroy that did not land leaves user text the deployment
                // was told to forget, so it is re-queued unconditionally. The
                // re-queue is also what keeps a reconcile from declaring the
                // index whole while the removal is outstanding: the mark stays
                // in the dirty set until a drain lands it, and
                // `DirtySet::has_pending_removal` is what the sweep consults.
                // A flag reset at the top of every sweep could not do that job —
                // the sweep enumerates EXISTING partitions, and a destroyed one
                // is gone from that listing, so no sweep can retry it.
                self.dirty.mark(partition, Pending::Destroy);
                Outcome::Unavailable {
                    reason: UnavailableReason::StoreFailed,
                }
            }
        }
    }

    /// Erase a migrated-away conversation's segments, leaving no trace.
    async fn remove(&self, partition: &str) -> Outcome {
        match self.projection.remove(partition).await {
            Ok(()) => Outcome::Removed,
            Err(err) => {
                tracing::error!(
                    partition,
                    error = %err,
                    "search index could not remove a migrated conversation's segments"
                );
                // As `destroy`: the re-queue is what carries this, because the
                // migrated-away partition is gone from what a sweep enumerates.
                self.dirty.mark(partition, Pending::Remove);
                Outcome::Unavailable {
                    reason: UnavailableReason::StoreFailed,
                }
            }
        }
    }

    /// Remove one stale physical source, then rebuild the replacement which
    /// now owns the same logical partition name.
    async fn replace(&mut self, partition: &str) -> Outcome {
        // Revoke read authority before the destructive step. Removing a
        // directory can fail while every old Parquet file remains readable. Without
        // this persisted unavailable footer, a same-length replacement could leave
        // erased earlier text servable when it has the same final event.
        if let Err(err) = self
            .projection
            .mark_unavailable(partition, &self.key_id)
            .await
        {
            tracing::error!(
                partition,
                error = %err,
                "search index could not revoke stale-source read authority before replacement"
            );
            self.unrecorded_refusal = true;
            self.dirty.mark(partition, Pending::Replace);
            return Outcome::Unavailable {
                reason: UnavailableReason::StoreFailed,
            };
        }
        if let Err(err) = self.projection.remove(partition).await {
            tracing::error!(
                partition,
                error = %err,
                "search index could not remove a stale source before replacement rebuild"
            );
            self.dirty.mark(partition, Pending::Replace);
            return Outcome::Unavailable {
                reason: UnavailableReason::StoreFailed,
            };
        }
        self.index_partition(partition, None).await
    }

    /// Index `partition` forward to `boundary`, or rebuild it from scratch
    /// when `boundary` is `None`.
    ///
    /// A rebuild starts at position zero and ignores the stored watermark: a
    /// rewrite or repair compacts positions, so resuming from the old
    /// watermark would index across a discontinuity and keep serving text the
    /// journal no longer holds.
    ///
    /// # A forward pass may decide it has to be a rebuild
    ///
    /// Twice, for two different reasons, and both are decisions this layer must
    /// make because they depend on what the pass READ rather than on what
    /// marked it:
    ///
    /// - The window holds a taint-excision marker. Excision names positions
    ///   below the watermark, so a forward append cannot apply it — and worse,
    ///   the segment it would write claims an excision frontier past a marker it
    ///   only half-applied, which is the value the read path trusts. The
    ///   observer normally marks these `Rebuild`, but that mark lives in memory
    ///   and a restart discards it, so this is where a dropped one is recovered.
    /// - The conversation is already unavailable. The design record's
    ///   invalidation table says an append landing on one must "remain
    ///   unavailable" AND make a rebuild happen; only a rebuild can clear the
    ///   flag, and nothing else will ever revisit a quiet conversation. So the
    ///   append becomes the rebuild rather than publishing another refused
    ///   segment behind a promise nobody kept.
    ///
    /// Escalation happens in place rather than through the queue, so it cannot
    /// livelock: it is driven by an external committed turn, never by this
    /// worker's own failure, and it runs at most once per pass. The repeated
    /// case — a conversation whose rebuild is deterministically doomed — is
    /// bounded by [`SearchIndexWorker::recovery_blocked`].
    pub(crate) async fn index_partition(
        &mut self,
        partition: &str,
        boundary: Option<u64>,
    ) -> Outcome {
        // At most two passes: the second always has `boundary == None`, and a
        // rebuild never escalates. A loop rather than recursion because an
        // `async fn` calling itself needs a boxed future for a fixed-size one.
        let mut boundary = boundary;
        let mut attempted_recovery = false;
        loop {
            match self.index_pass(partition, boundary).await {
                Pass::Done(outcome) => {
                    self.note_recovery(partition, attempted_recovery, &outcome);
                    return outcome;
                }
                Pass::RebuildInstead { cause } => {
                    tracing::info!(
                        partition,
                        cause = cause.label(),
                        "search index is rebuilding a conversation a forward pass cannot repair"
                    );
                    metrics::record_search_index_rebuild(cause.label());
                    attempted_recovery |= cause == RebuildCause::UnavailableRecovery;
                    boundary = None;
                }
            }
        }
    }

    /// Keep [`SearchIndexWorker::recovery_blocked`] honest about what the pass
    /// just proved.
    ///
    /// A conversation that published is working, so nothing about it is
    /// suppressed any more — which is what keeps the set from outliving the
    /// condition it describes. A recovery rebuild that failed for a reason the
    /// identical rebuild would hit again is what earns an entry: without it,
    /// every committed turn on a deterministically broken conversation buys
    /// another full replay of it, forever.
    fn note_recovery(&mut self, partition: &str, attempted_recovery: bool, outcome: &Outcome) {
        match outcome {
            Outcome::Published { .. } | Outcome::Destroyed | Outcome::Removed => {
                self.recovery_blocked.remove(partition);
            }
            Outcome::Unavailable { reason } if attempted_recovery && !reason.is_transient() => {
                // A full set stops SUPPRESSING rather than stops recovering:
                // the cost of forgetting an entry is a repeated rebuild, and
                // the cost of an unbounded set is the memory leak the dirty
                // set's own bound exists to prevent.
                if self.recovery_blocked.len() < MAX_TRACKED_PARTITIONS {
                    self.recovery_blocked.insert(partition.to_owned());
                }
            }
            Outcome::Unavailable { .. }
            | Outcome::BarrierHeld { .. }
            | Outcome::AlreadyCurrent
            | Outcome::Deferred => {}
        }
    }

    /// One indexing attempt, which may report that it has to be redone as a
    /// rebuild.
    async fn index_pass(&mut self, partition: &str, boundary: Option<u64>) -> Pass {
        let partition = partition.to_owned();
        let state = match self.projection.coverage(&partition, &self.key_id).await {
            Ok(state) => state,
            // A segment that will not decode is rebuilt from the journal, not
            // repaired in place: an unreadable footer is unknown coverage, and
            // `rebuild` lists by path so it can replace what it cannot read.
            Err(err) if err.is_unreadable() => CoverageState::NeverIndexed,
            Err(err) => return Pass::Done(self.store_failed(&partition, &err).await),
        };

        if matches!(state, CoverageState::Destroyed) {
            // Destroy is terminal, and the store would refuse the publish
            // anyway. Returning here keeps that refusal from being read as a
            // store failure and re-queued forever.
            return Pass::Done(Outcome::Destroyed);
        }

        let source_incarnation = match self.journal.partition_incarnation(partition.clone()).await {
            Ok(Some(incarnation)) => incarnation,
            Ok(None) => {
                return Pass::Done(
                    self.fail_unavailable(&partition, UnavailableReason::SourceEmpty)
                        .await,
                );
            }
            Err(error) => {
                return Pass::Done(
                    self.replay_failed(
                        &partition,
                        &error,
                        "search index could not resolve the partition's exact source",
                    )
                    .await,
                );
            }
        };
        if let Some(cause) = self.rebuild_instead(&partition, boundary, &state, source_incarnation)
        {
            return Pass::RebuildInstead { cause };
        }

        let start = match self.resolve_start(&partition, boundary, &state).await {
            Ok(start) => start,
            Err(outcome) => return Pass::Done(outcome),
        };

        let end = match self.resolve_end(&partition, boundary).await {
            Ok(end) => end,
            Err(outcome) => return Pass::Done(outcome),
        };

        if boundary.is_some() && end <= start {
            return Pass::Done(Outcome::AlreadyCurrent);
        }

        // `partition_event_count` reports 0 for a partition that no longer
        // exists, so a rebuild that reached here with nothing to read must not
        // publish `{ indexed_through: 0, available: true }` — that claims
        // complete coverage of a conversation nothing has read. A journal that
        // is merely uncommitted is a different state and is handled below, by
        // the barrier, with events actually in hand.
        if end == 0 {
            tracing::warn!(
                partition,
                "search index found no events to rebuild a conversation from; leaving it \
                 unsearchable"
            );
            return Pass::Done(
                self.fail_unavailable(&partition, UnavailableReason::SourceEmpty)
                    .await,
            );
        }

        let replay = match self
            .journal
            .replay_range_with_positions_bounded(partition.clone(), start, end, self.replay_budget)
            .await
        {
            Ok(replay) => replay,
            Err(error) => {
                return Pass::Done(
                    self.replay_failed(
                        &partition,
                        &error,
                        "search index replay failed; leaving the conversation unsearchable",
                    )
                    .await,
                );
            }
        };

        if replay.budget_exceeded {
            // The replay stopped short of `end`, so the range it read covers
            // only part of what the watermark would claim. Publishing it would
            // put a partial index behind a clean watermark.
            tracing::warn!(
                partition,
                bytes_read = replay.bytes_read,
                "search index replay exceeded its byte budget; leaving the conversation \
                 unsearchable"
            );
            return Pass::Done(
                self.fail_unavailable(&partition, UnavailableReason::ReplayBudgetExceeded)
                    .await,
            );
        }

        // The projection folds the replayed range alone, so what this publishes
        // is the DELTA and the earlier segments keep the prefix. `start` still
        // matters here — it bounds the replay above and picks rebuild against
        // append below — but it no longer reaches the fold.
        let built = project::project(
            &self.term_key,
            &partition,
            source_incarnation,
            &replay.events,
            end,
        );

        self.settle_pass(&partition, source_incarnation, boundary, start, end, &built)
            .await
    }

    /// Reports why an append pass must become a rebuild instead, if it must.
    ///
    /// Both reasons are append-only concerns, so a pass with no boundary — a
    /// rebuild already — answers `None`.
    fn rebuild_instead(
        &self,
        partition: &str,
        boundary: Option<u64>,
        state: &CoverageState,
        source_incarnation: polyc_state::revision::PartitionIncarnation,
    ) -> Option<RebuildCause> {
        boundary?;
        if matches!(
            state,
            CoverageState::Indexed(coverage)
                if coverage.source_incarnation != source_incarnation
        ) {
            return Some(RebuildCause::SourceReplacement);
        }
        // Invalidation row 6, the half that was quoted and never executed. An
        // append onto a refused conversation stays refused — the store enforces
        // that — but "remain unavailable" without a rebuild is a permanent
        // participation-wide outage for everyone in the conversation, because
        // only a rebuild clears the flag and nothing else ever comes back to a
        // conversation that has stopped being marked. So the append becomes the
        // rebuild, unless this conversation has already shown that its rebuild
        // fails the same way every time.
        if matches!(state, CoverageState::Indexed(coverage) if !coverage.available)
            && !self.recovery_blocked.contains(partition)
        {
            return Some(RebuildCause::UnavailableRecovery);
        }
        None
    }

    /// Decides what one built pass publishes, once the replay has folded.
    ///
    /// Split from [`Self::index_pass`] because it is a separate phase with its
    /// own refusals: a lineage that moved under the replay, an excision inside
    /// the window, a barrier that left nothing to publish, and only then the
    /// publication itself.
    async fn settle_pass(
        &mut self,
        partition: &str,
        source_incarnation: polyc_state::revision::PartitionIncarnation,
        boundary: Option<u64>,
        start: u64,
        end: u64,
        built: &project::Projected,
    ) -> Pass {
        let partition = partition.to_owned();
        let current_incarnation = self.journal.partition_incarnation(partition.clone()).await;
        if !matches!(current_incarnation, Ok(Some(current)) if current == source_incarnation) {
            tracing::warn!(
                partition,
                "search index source changed while replaying; refusing the stale publication"
            );
            let outcome = self
                .fail_unavailable(&partition, UnavailableReason::ReplayFailed)
                .await;
            self.dirty.mark(&partition, Pending::Replace);
            return Pass::Done(outcome);
        }

        // Before ANY early return below, and before any publish: an excision in
        // this window means the removed positions sit in segments this pass is
        // not rewriting. Appending here would strip only the window and then
        // record an excision frontier past the marker — the exact value the read
        // path trusts to decide the conversation is clean — so the append would
        // not merely fail to remove the text, it would certify it as removed.
        if boundary.is_some() && built.excision_in_range {
            return Pass::RebuildInstead {
                cause: RebuildCause::Excision,
            };
        }

        // The barrier held the watermark at or below where this pass started,
        // so there is nothing new to publish — a conversation whose only open
        // turn is still running reaches this on every committed turn of every
        // OTHER turn in flight. Publishing anyway would add a no-progress
        // segment and could hide whether a later pass advanced the watermark.
        if boundary.is_some() && built.coverage.indexed_through <= start {
            // Which of the two this is matters. "Caught up" is the healthy end of a pass;
            // "pinned behind a turn that opened and never closed" is a
            // conversation whose every later committed turn is silently
            // unindexable while its coverage still reads searchable.
            return Pass::Done(built.open_turn_at.map_or(Outcome::AlreadyCurrent, |at| {
                metrics::record_search_index_barrier_held();
                Outcome::BarrierHeld {
                    open_turn_at: at,
                    journal_lag: end.saturating_sub(at),
                }
            }));
        }

        let outcome = self.publish(&partition, start, built).await;
        if let (Some(open_turn_at), Outcome::Published { .. }) = (built.open_turn_at, &outcome) {
            metrics::record_search_index_barrier_held();
            Pass::Done(Outcome::BarrierHeld {
                open_turn_at,
                journal_lag: end.saturating_sub(open_turn_at),
            })
        } else {
            Pass::Done(outcome)
        }
    }

    /// Write one pass's result, as a rebuild when it starts at position zero
    /// and as an appended segment otherwise.
    async fn publish(
        &mut self,
        partition: &str,
        start: u64,
        built: &project::Projected,
    ) -> Outcome {
        let published = if start == 0 {
            self.projection
                .rebuild(partition, &built.messages, &built.coverage, &self.key_id)
                .await
                // A rebuild is the one write that may raise availability, so it
                // never clamps.
                .map(|stats| store::Appended {
                    stats,
                    clamped_unavailable: false,
                })
        } else {
            self.projection
                .append(partition, &built.messages, &built.coverage, &self.key_id)
                .await
        };

        match published {
            Ok(appended) => {
                if appended.clamped_unavailable {
                    // Reachable only when the recovery rebuild this append
                    // would otherwise have become is suppressed, so it is a
                    // statement about that suppression rather than a surprise.
                    tracing::warn!(
                        partition,
                        "search index indexed a conversation that stays refused: its recovery \
                         rebuild already failed the same way, so it waits for the next sweep or \
                         for a fix outside this worker"
                    );
                }
                // The write answered with what the directory now holds, so a
                // rebuild needs no special case: it leaves one segment and no
                // unmerged rows, which is below both triggers by construction.
                self.compact_if_due(partition, appended.stats).await;
                Outcome::Published {
                    indexed_through: built.coverage.indexed_through,
                }
            }
            // The conversation was destroyed between this pass's coverage read
            // and its publish. The tombstone is the newer statement and this
            // pass's rows are the stale one.
            Err(StoreError::Destroyed) => Outcome::Destroyed,
            Err(StoreError::TooLarge) => {
                self.fail_unavailable(partition, UnavailableReason::RecordTooLarge)
                    .await
            }
            Err(err) => self.store_failed(partition, &err).await,
        }
    }

    /// Fold a conversation's segments once either compaction trigger fires.
    ///
    /// Two independent triggers, because either alone misses a real shape: a
    /// segment count misses the conversation with three enormous segments, and
    /// a row ceiling misses the conversation with forty tiny ones. Both are
    /// first guesses ([`COMPACT_SEGMENT_THRESHOLD`],
    /// [`COMPACT_ROW_CEILING`]), and both read off the stats the write returned
    /// rather than off a counter kept beside the store — so a restart, or a
    /// conversation the worker has never seen before, gets the same answer the
    /// directory would give.
    ///
    /// A failed compaction is logged and nothing more. It rewrites how a
    /// prefix is stored, never what it covers, and the store writes the merged
    /// segment before removing the ones it replaces — so a failure leaves
    /// duplicated rows that a reader unions back to the same answer, not a
    /// coverage claim to withdraw.
    async fn compact_if_due(&self, partition: &str, stats: SegmentStats) {
        if !should_compact(stats, self.row_ceiling) {
            return;
        }
        if let Err(err) = self.projection.compact(partition, &self.key_id).await {
            tracing::warn!(
                partition,
                error = %err,
                "search index could not compact a conversation's segments; its rows are intact \
                 and reads stay correct, only slower"
            );
        }
    }

    /// The position this pass should resume from.
    ///
    /// A rebuild deliberately discards the watermark and starts at zero; a
    /// forward index resumes from it.
    ///
    /// A forward index REQUIRES the prefix it is about to advance the watermark
    /// over to be readable. If the newest segment's coverage decoded but an
    /// older segment did not — a torn file, bit rot, a segment left under a
    /// rotated key — appending one more segment would leave a watermark
    /// claiming a prefix no reader can decode, and the conversation would refuse
    /// every search until something rebuilt it. Falling back to a rebuild is
    /// what repairs that.
    ///
    /// The probe is [`SearchProjection::segment_stats`], which opens every
    /// segment's FOOTER and reads no rows. That is what makes it affordable on
    /// every committed turn: decoding the conversation instead would make each
    /// turn O(rows indexed so far), which is the O(N²) the segment layout exists
    /// to remove — writes O(delta) and reads O(N) is still O(N) per turn. A
    /// footer read catches every failure this probe is for, because all of them
    /// are whole-file failures: a truncated or overwritten file fails footer
    /// parse (the footer is the last thing in a Parquet file), and a rotated
    /// key or bumped format is a footer value. A corrupt page inside an
    /// otherwise valid file is what it no longer catches, and the read path
    /// already answers that one — [`StoreError::is_unreadable`] routes it to a
    /// rebuild wherever it surfaces.
    ///
    /// # Errors
    ///
    /// Returns the terminal [`Outcome`] when the stored rows already exceed
    /// what the store will decode, so the caller can return it unchanged.
    async fn resolve_start(
        &mut self,
        partition: &str,
        boundary: Option<u64>,
        state: &CoverageState,
    ) -> Result<u64, Outcome> {
        let (CoverageState::Indexed(coverage), Some(_)) = (state, boundary) else {
            return Ok(0);
        };
        if coverage.indexed_through == 0 {
            return Ok(0);
        }

        match self.projection.segment_stats(partition, &self.key_id).await {
            Ok(on_disk) if on_disk.segments > 0 => Ok(coverage.indexed_through),
            // Coverage without segments behind it: an interrupted publish, or a
            // directory emptied underneath. Rebuild rather than advance a
            // watermark over a prefix that is not there.
            Ok(_) => Ok(0),
            // Same answer, different door: a segment that will not decode is
            // replaced by `rebuild`, which lists by path.
            Err(err) if err.is_unreadable() => Ok(0),
            // A rebuild would not help here: the row count is a property of the
            // conversation, so the rebuilt segments would be just as
            // undecodable — behind coverage claiming otherwise.
            Err(StoreError::TooLarge) => {
                tracing::warn!(
                    partition,
                    "search index cannot decode a conversation's own rows; leaving it unsearchable"
                );
                Err(self
                    .fail_unavailable(partition, UnavailableReason::RecordTooLarge)
                    .await)
            }
            Err(err) => Err(self.store_failed(partition, &err).await),
        }
    }

    /// The exclusive position this pass should read up to.
    ///
    /// A forward index uses the boundary the observer supplied. A rebuild has
    /// none, so it asks the log how far the conversation currently goes and
    /// lets the committed projection decide what within that range is
    /// indexable — the journal tail is a CEILING here, never the watermark
    /// that gets published.
    ///
    /// # Errors
    ///
    /// Returns the terminal [`Outcome`] when the partition's length cannot be
    /// read, so the caller can return it unchanged.
    async fn resolve_end(
        &mut self,
        partition: &str,
        boundary: Option<u64>,
    ) -> Result<u64, Outcome> {
        match boundary {
            Some(boundary) => Ok(boundary),
            None => match self
                .journal
                .partition_event_count(partition.to_owned())
                .await
            {
                Ok(count) => Ok(count),
                Err(error) => Err(self
                    .replay_failed(
                        partition,
                        &error,
                        "search index could not read a partition's length; leaving the \
                         conversation unsearchable",
                    )
                    .await),
            },
        }
    }

    /// A journal read failed: mark the conversation unsearchable, unless this
    /// process is on its way out.
    ///
    /// A [`JournalError::Unreachable`] while everything is SHUTTING DOWN is the
    /// one failure that is not about this conversation. A shutdown that lands
    /// mid-drain would otherwise publish `available: false` for whichever
    /// healthy conversation the pass happened to be holding. Under
    /// all-or-nothing coverage that one segment takes participation-wide search
    /// down for everyone in that conversation until something rebuilds it, and
    /// the mark that would have scheduled the rebuild dies with the process. So
    /// this publishes nothing and leaves the stored watermark exactly where it
    /// was: the next boot resumes from it, and the conversation's next
    /// committed turn re-marks it.
    ///
    /// The same class also answers a journal that is unreachable while this
    /// process keeps serving, and reading THAT as a shutdown would be wrong in
    /// both directions — the pass would drop the mark with nothing queued,
    /// freezing that conversation behind an `available: true` watermark no
    /// later read disagrees with; and one such partition would abort a whole
    /// reconcile pass, so `degraded` never cleared and the entire fleet refused
    /// forever. Asking whether this process is stopping costs an atomic load.
    async fn replay_failed(
        &mut self,
        partition: &str,
        error: &JournalError,
        message: &'static str,
    ) -> Outcome {
        if matches!(error, JournalError::Unreachable(_)) && self.journal.is_stopping() {
            tracing::debug!(
                partition,
                "search index stopped mid-conversation because the event log has shut down; \
                 nothing published and the watermark is untouched"
            );
            return Outcome::Deferred;
        }
        tracing::warn!(partition, %error, "{message}");
        self.fail_unavailable(partition, replay_reason(error)).await
    }

    /// A projection read or write failed: mark the conversation unsearchable
    /// and schedule its repair, exactly as every other failure path does.
    ///
    /// An earlier version of this worker logged and returned without doing
    /// either, which left the old segment advertising `available: true` and
    /// nothing queued to fix it — the failure mode the re-enqueue exists to
    /// prevent.
    async fn store_failed(&mut self, partition: &str, err: &StoreError) -> Outcome {
        tracing::error!(
            partition,
            error = %err,
            "search index projection operation failed; leaving the conversation unsearchable"
        );
        self.fail_unavailable(partition, UnavailableReason::StoreFailed)
            .await
    }

    /// Mark a conversation unsearchable and report why.
    ///
    /// A failure to record the failure is loud and blocks the next reconcile
    /// from declaring the index whole: the stale segment still says
    /// `available: true`, so this conversation's coverage is unknown rather
    /// than known-refused, which is what the degraded flag means.
    async fn fail_unavailable(&mut self, partition: &str, reason: UnavailableReason) -> Outcome {
        if let Err(err) = self
            .projection
            .mark_unavailable(partition, &self.key_id)
            .await
        {
            tracing::error!(
                partition,
                error = %err,
                "search index could not even record that a conversation is unsearchable"
            );
            self.unrecorded_refusal = true;
        }
        // Schedule the repair rather than waiting for the conversation's next
        // turn to arrive. Under all-or-nothing coverage one unavailable
        // conversation takes participation-wide search down for everyone in
        // it, and a quiet conversation may never produce another
        // `turn_complete` — so "unavailable" without a re-enqueue is a
        // permanent outage wearing a transient name.
        //
        // Only for a cause that could resolve on its own, though: re-enqueuing
        // a deterministic failure spins the worker forever on one
        // conversation.
        if reason.is_transient() {
            self.dirty.mark(partition, Pending::Rebuild);
        }
        metrics::record_search_index_unavailable(reason.label());
        Outcome::Unavailable { reason }
    }

    /// Lower the replay budget so a test can reach the refusal path without
    /// writing tens of megabytes into a journal.
    #[cfg(test)]
    pub(crate) const fn with_replay_budget(mut self, bytes: u64) -> Self {
        self.replay_budget = bytes;
        self
    }

    /// Lower the row ceiling so a test can drive the row trigger end to end
    /// without writing half a million rows into a projection.
    #[cfg(test)]
    pub(crate) const fn with_row_ceiling(mut self, rows: u64) -> Self {
        self.row_ceiling = rows;
        self
    }
}

/// Whether a conversation's segments should be folded now.
///
/// Both inputs are quantities a fold RESETS — the segment count to one, the
/// unmerged rows to zero — so each trigger is an edge the conversation crosses
/// once per fold. A trigger on [`SegmentStats::rows`] would not be: compaction
/// merges rows rather than dropping them, so it would fire on every append
/// forever once crossed.
///
/// A free function so both triggers are testable without a journal, a
/// projection, or a runtime behind them. `row_ceiling` is
/// [`COMPACT_ROW_CEILING`] outside tests.
const fn should_compact(stats: SegmentStats, row_ceiling: u64) -> bool {
    stats.segments >= COMPACT_SEGMENT_THRESHOLD || stats.unmerged_rows >= row_ceiling
}

#[cfg(test)]
mod tests;