polyc-query 2026.9.0

//! The durable projection behind participation-scoped search: append-only
//! Parquet segments per conversation, in the layout `DataFusion` reads.
//!
//! # Why Parquet rather than a hand-rolled store
//!
//! The accepted design already settled this: `ListingTable` reads Parquet over an
//! object store **with zero custom code**, including row-group and page-index
//! statistics pruning and Bloom-filter pruning for high-selectivity equality
//! predicates, and the layout is organized per conversation so a session's
//! catalog registers only the conversations it is already authorized for —
//! the same scoping rule as the live `events` table.
//!
//! An earlier draft built a bespoke key-value store instead, on the stated
//! grounds that "there is no range-queryable durable store in this
//! workspace." That was wrong, and the mechanisms it hand-rolled each have a
//! native counterpart: a membership filter is a Bloom filter, a postings
//! record is columnar rows, a delta-varint codec is Parquet's own encoding.
//!
//! # Why segments rather than one file per conversation
//!
//! A single file rewritten on every committed turn costs O(N) rows per turn
//! and therefore **O(N²)** over a conversation's life — the same quadratic
//! that disqualified `Metadata` for this job, reintroduced one level down.
//!
//! So a publish [`SearchProjection::append`]s a NEW segment covering only the
//! positions it just indexed. Turn cost becomes proportional to the delta.
//! `ListingTable` reads every segment in the directory as one table, so the
//! read side is unchanged; [`SearchProjection::compact`] folds them back to
//! one when the count grows.
//!
//! # What the rows are
//!
//! One row per (committed message, distinct term):
//!
//! | column | meaning |
//! |---|---|
//! | `turn_id` | the turn the message belongs to; ranking deduplicates to one hit per turn |
//! | `position` | the message's durable journal position |
//! | `term_hash` | the KEYED hash of one term the message contains |
//!
//! Storing the keyed hash rather than the plaintext term costs nothing here —
//! a Bloom filter prunes on column-value equality either way, so a query
//! rewrites its own terms through the same key — and it keeps the projection
//! from ever holding readable user words. A term-membership structure is an
//! oracle over user text by construction, and more precise for short
//! conversations, which are most of them.
//!
//! # What prunes today, and what does not
//!
//! This module WRITES the layout that makes pruning possible — Hive-style
//! `conversation_id=` directories, a Bloom filter and declared sort order on
//! `term_hash`. It does not yet READ through them:
//! [`SearchProjection::postings`] decodes each segment whole.
//!
//! That is deliberate and temporary. Pruning happens when the search port
//! registers a `ListingTable` over the authorized segments and pushes a
//! `term_hash IN (...)` predicate down — a later change in this stack.
//! Writing the layout first makes that change a query, not a migration.
//!
//! # Coverage lives in the segment, not beside it
//!
//! Each segment's watermark, source incarnation, availability, term-key
//! identity, and format version are written into its Parquet footer, so they
//! are published in the same object as the rows they describe. Atomic
//! publication is then a property of the write rather than of writer
//! discipline. A conversation's coverage is its NEWEST segment's — segments
//! are ordered by the watermark they reach, so the newest is the one that
//! describes the whole indexed prefix.
//!
//! The footer is also where a state with no rows at all lives: a destroy
//! writes a rows-less tombstone segment rather than deleting the directory,
//! because an absent directory is indistinguishable from a conversation
//! nothing has indexed yet, and those two demand opposite responses. See
//! [`CoverageState`].
//!
//! # Coverage is a claim, not freshness
//!
//! A footer states what the writer indexed. It cannot state that the journal
//! still is what was indexed: a rewrite that commits durably and then dies
//! before its notification leaves a perfectly readable segment describing a
//! journal that no longer exists, behind an apparently current watermark.
//! [`SearchProjection::verified_coverage`] closes that by comparing the
//! segment's exact physical source with the live journal on every read, which
//! is what let the design record drop its startup barrier rather than pay
//! O(every partition) on each boot. The journal side arrives through
//! [`JournalState`], so this module never learns what an event log is.
//!
//! Atomicity here rests on POSIX rename, so it holds for the local filesystem
//! this writes to today. An object store has no atomic rename; moving to one
//! means a single-object PUT instead, atomic for a different reason. The
//! invariant survives; the mechanism does not, so it is named rather than
//! assumed.
//!
//! # Blocking work, async surface
//!
//! Filesystem access, zstd, and Parquet encode/decode all block. Consumers run
//! on the control plane's runtime, so every method here is `async` and wraps
//! its work in `spawn_blocking` — the same shape
//! `polyc_wallet_delegation::secret_store` uses for equivalent work. A caller
//! cannot forget to.

use std::path::{Path, PathBuf};
use std::sync::Arc;

use arrow::array::{RecordBatch, StringArray, UInt32Array, UInt64Array};
use arrow::datatypes::{DataType, Field, Schema};
use parquet::arrow::ArrowWriter;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use parquet::basic::{Compression, ZstdLevel};
use parquet::file::metadata::SortingColumn;
use parquet::file::properties::WriterProperties;

use polyc_eventlog_host::encode_partition;
use polyc_state::revision::PartitionIncarnation;

use super::{IndexedMessage, PostingsRecord};

/// Footer key carrying the exclusive journal position this file is indexed
/// through.
const META_INDEXED_THROUGH: &str = "polychrome.search.indexed_through";

/// Footer key carrying the exact physical journal lineage, lower-hex.
const META_INCARNATION: &str = "polychrome.search.incarnation";

/// Footer key carrying whether this conversation may currently be searched.
const META_AVAILABLE: &str = "polychrome.search.available";

/// Footer key carrying the exclusive journal position this file's writer
/// examined for taint-excision markers.
///
/// The durable half of the excision obligation, and the reason a dropped
/// rebuild is recoverable. An excision marker is an ORDINARY append that never
/// changes the physical source, so
/// [`SearchProjection::verified_coverage`]'s source check structurally cannot
/// see one.
/// The only thing that ever recorded the owed rebuild was an in-memory mark,
/// which a restart discards. This key records how far the index has actually
/// looked, so the outstanding work is the difference between it and the
/// journal — two durable things, neither of which a restart can lose.
const META_EXCISION_SCANNED_THROUGH: &str = "polychrome.search.excision_scanned_through";

/// Footer key marking a rows-less segment as the destroy tombstone.
///
/// Without it a destroyed conversation is an ABSENT directory, which reads
/// exactly like one nothing ever indexed — so a reader cannot tell "there is
/// nothing left to search, ever" from "this has not been indexed yet, ask
/// again". The design record requires the destroyed state to be recorded
/// rather than merely absent, because the two demand opposite responses: the
/// first must never be rebuilt, the second must be.
const META_DESTROYED: &str = "polychrome.search.destroyed";

/// Footer key carrying the identity of the [`super::terms::TermKey`] the
/// hashes were computed under.
///
/// Without it a key rotation is undetectable: every file still decodes, still
/// reports a valid watermark, and still says `available` — and every query
/// returns zero hits for text the file demonstrably holds. That is the same
/// silent-not-found [`StoreError::Corrupt`] exists to prevent, reached by a
/// different door.
const META_KEY_ID: &str = "polychrome.search.key_id";

/// Footer key carrying the projection's format version.
///
/// Covers everything a reader must agree with the writer about beyond the
/// schema: the term hash width, its domain separator, and the truncation rule.
/// Changing any of them without a rebuild produces the same silent zero-hit
/// failure a rotated key does.
const META_FORMAT: &str = "polychrome.search.format";

/// Current projection format. Bump on any change to how a term becomes a
/// `term_hash`.
const FORMAT_VERSION: &str = "2";

/// The column term-hash predicates probe, and the only one carrying a Bloom
/// filter.
const TERM_HASH_COLUMN: &str = "term_hash";

/// Stands in for "this message committed text, but none of it is searchable".
///
/// A real term hash is 24-bit (`super::terms`), so no term can collide with
/// this value — it is outside the space by construction rather than by luck.
const NO_TERMS_SENTINEL: u32 = u32::MAX;

/// What can go wrong reading or writing the projection.
#[derive(Debug, thiserror::Error)]
pub(crate) enum StoreError {
    /// The underlying filesystem or object store failed.
    #[error("search projection io: {0}")]
    Io(String),

    /// Parquet encoding or decoding failed.
    #[error("search projection parquet: {0}")]
    Parquet(String),

    /// A stored file decoded, but its footer metadata is missing or
    /// unparseable.
    ///
    /// Never repaired in place: a file whose watermark cannot be read is a
    /// file whose coverage is unknown, and treating unknown coverage as
    /// complete is how a search answers "not found" for text it holds.
    #[error("search projection metadata is unreadable and must be rebuilt from the journal")]
    Corrupt,

    /// A conversation's segments exceed [`MAX_POSTINGS_ROWS`] to decode.
    ///
    /// Refuses rather than truncating: a partial read that looked complete is
    /// the silent incompleteness this whole design exists to prevent. The
    /// caller marks the conversation unavailable, the same path a replay
    /// budget overrun takes.
    #[error("search projection exceeds the {MAX_POSTINGS_ROWS}-row read cap")]
    TooLarge,

    /// A publication was attempted for a conversation that carries the destroy
    /// tombstone.
    ///
    /// Destroy is terminal: the journal is gone, so nothing can legitimately
    /// produce rows for it again. Refusing loudly rather than writing them is
    /// what keeps a late in-flight publish — one that read its range before the
    /// destroy landed — from appending a segment NEWER than the tombstone and
    /// resurrecting a conversation the deployment was told to forget.
    #[error("search projection for a destroyed conversation cannot be published to")]
    Destroyed,
}

impl StoreError {
    /// Whether the stored bytes cannot be understood, so the repair is a
    /// rebuild from the journal rather than a retry.
    ///
    /// One definition rather than a `Corrupt | Parquet(_)` pattern repeated at
    /// every call site: the set is a property of this type, and three copies of
    /// it is three chances for one to drift into treating an unreadable segment
    /// as a transient store failure — which re-queues it forever instead of
    /// rebuilding it.
    pub(crate) const fn is_unreadable(&self) -> bool {
        matches!(self, Self::Corrupt | Self::Parquet(_))
    }
}

impl From<std::io::Error> for StoreError {
    fn from(err: std::io::Error) -> Self {
        Self::Io(err.to_string())
    }
}

impl From<parquet::errors::ParquetError> for StoreError {
    fn from(err: parquet::errors::ParquetError) -> Self {
        Self::Parquet(err.to_string())
    }
}

impl From<arrow::error::ArrowError> for StoreError {
    fn from(err: arrow::error::ArrowError) -> Self {
        Self::Parquet(err.to_string())
    }
}

/// Everything a reader needs about a conversation before deciding to scan it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Coverage {
    /// Exclusive journal position this conversation is indexed through.
    ///
    /// Compared against the partition's last COMMITTED turn boundary, never
    /// its journal tail: a turn's `turn_start` and its inputs commit before
    /// the harness is dialed, so a partition with any turn in flight has a
    /// tail ahead of anything indexable.
    pub(crate) indexed_through: u64,
    /// Opaque durable incarnation of the physical journal this file describes.
    /// Exactly thirty-two bytes for every live source.
    pub(crate) source_incarnation: PartitionIncarnation,
    /// False when this conversation cannot currently be searched. A search
    /// over any scope containing an unavailable conversation refuses rather
    /// than returning results that look exhaustive.
    pub(crate) available: bool,
    /// Exclusive journal position the writer examined for taint-excision
    /// markers, stripping whatever it found.
    ///
    /// Always at or above [`Coverage::indexed_through`]: a pass scans the whole
    /// range it replayed, including the part above the open-turn barrier that
    /// the watermark stops short of. A journal holding an excision marker at or
    /// above this position is a journal the index has not accounted for, which
    /// [`SearchProjection::verified_coverage`] reports as
    /// [`CoverageState::Stale`] — see [`META_EXCISION_SCANNED_THROUGH`].
    pub(crate) excision_scanned_through: u64,
}

/// What the projection knows about one conversation.
///
/// Distinct states, not an `Option<Coverage>`, because "absent" was answering
/// two questions with one silence: a conversation nothing has indexed yet must
/// be indexed, and a destroyed one must never be. A caller that has to tell
/// them apart cannot, and one that does not have to still has to say which it
/// is doing — which is the point of putting the distinction in the type rather
/// than in a flag beside it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum CoverageState {
    /// No segment has ever been published for this conversation.
    ///
    /// Uncovered, and the repair is to index it.
    NeverIndexed,

    /// The conversation is indexed, and this is what its newest segment
    /// declares.
    ///
    /// Being indexed is not being searchable: the caller still owes the
    /// availability flag and the watermark comparison against the last
    /// committed turn boundary.
    Indexed(Coverage),

    /// The index describes a journal this conversation no longer has.
    ///
    /// A rewrite, repair, or truncation landed and the rebuild that should
    /// have followed did not — most sharply when the mutation commits durably
    /// and the process dies before its notification, which leaves a perfectly
    /// readable segment behind an apparently current watermark. Uncovered
    /// until a rebuild, and only
    /// [`SearchProjection::verified_coverage`] can report it, because only it
    /// asks the journal.
    Stale,

    /// The conversation was destroyed: its rows are gone and the tombstone
    /// naming that state is what remains.
    ///
    /// Uncovered permanently. There is no journal left to rebuild from, so a
    /// caller that treats this as [`CoverageState::NeverIndexed`] schedules an
    /// index build that can never succeed.
    Destroyed,
}

/// What the live journal says about excision markers a stored segment has not
/// accounted for.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ExcisionScan {
    /// The journal holds no excision marker at or above the footer's scanned
    /// frontier, so every marker it does hold was already applied by the pass
    /// that wrote that footer.
    Clear,
    /// At least one excision marker sits at or above the frontier. A rebuild is
    /// owed, and until it lands the stored segments may still hold text the
    /// deployment was told to remove.
    Pending,
    /// The journal could not be asked — the partition is gone, the replay
    /// failed, or the range needed more bytes than the scan may read.
    ///
    /// Treated exactly as [`ExcisionScan::Pending`] everywhere it matters, and
    /// kept distinct only so a log line can say which of the two it was.
    /// "Cannot prove there is no excision" and "there is an excision" call for
    /// the same refusal.
    Unknown,
}

impl ExcisionScan {
    /// Whether the index may be trusted to have applied every excision the
    /// journal holds.
    const fn is_clear(self) -> bool {
        matches!(self, Self::Clear)
    }
}

/// What the live journal says about a conversation the projection has a stored
/// claim for, supplied by the caller.
///
/// Use a trait rather than caller-supplied answers. The store must own the
/// complete read-side protocol. It reads its footer and the exact source. It then
/// scans the unaccounted taint-marker tail and re-reads the source. A caller that
/// assembles those steps can omit or reorder one. Giving the store a way to ask
/// makes the source bracket part of the serving mechanism.
///
/// It is also what keeps `EventLogHost` out of this module. The implementor is
/// the worker, which already holds the host; this store stays storage-only.
///
/// `Sync` and a `Send` future, because a verification runs inside a search
/// that the runtime may move between worker threads — the same requirement the
/// host this will be implemented over already satisfies.
pub(crate) trait JournalState: Sync {
    /// Durable, non-repeating physical lineage of `partition` as held now.
    ///
    /// `None` when the journal cannot answer — the partition is gone, the
    /// position is past its tail, or the replay failed. Every one of those
    /// reads as a mismatch, which is the direction that refuses rather than
    /// serves.
    fn source_incarnation(
        &self,
        partition: &str,
    ) -> impl std::future::Future<Output = Option<PartitionIncarnation>> + Send;

    /// Whether `partition` holds a taint-excision marker at or above
    /// `scanned_through`, the frontier the stored footer claims to have applied
    /// every excision below.
    ///
    /// The question the source identity cannot answer. A content rewrite
    /// rotates the physical source, but its later durable taint/completion
    /// marker can append without another rotation. A source-consistent
    /// publication may therefore still lag that marker tail. This asks the one
    /// thing that distinguishes that state. It asks the journal. The answer
    /// therefore survives a restart that discards an in-memory rebuild mark.
    ///
    /// The range is small in the steady state. A pass records the ceiling it
    /// replayed. The unscanned tail contains only records that landed later.
    fn excision_since(
        &self,
        partition: &str,
        scanned_through: u64,
    ) -> impl std::future::Future<Output = ExcisionScan> + Send;
}

/// Whether a segment being written is an ordinary publication or the tombstone
/// that records a destroy.
///
/// A parameter rather than two writers: the tombstone is a segment like any
/// other — same directory, same ordering, same atomic rename — and the whole
/// reason it can carry an authoritative state at all is that the footer is
/// published in the same object as the rows.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Marker {
    /// An ordinary segment: these rows, this coverage.
    Live,
    /// The rows-less tombstone: this conversation was destroyed.
    Destroyed,
}

/// The Arrow schema every projection file carries.
///
/// `term_hash` is `UInt32` so Parquet's Bloom filter and min/max statistics
/// both apply to it directly — the pruning this whole layout exists for.
fn projection_schema() -> Arc<Schema> {
    Arc::new(Schema::new(vec![
        Field::new("turn_id", DataType::Utf8, false),
        Field::new("position", DataType::UInt64, false),
        Field::new("term_hash", DataType::UInt32, false),
    ]))
}

/// Rows one conversation's segments may decode before a read refuses.
///
/// The design's work bounds cap partitions READ, never the size of one, so
/// without this a single pathological conversation is unbounded memory inside
/// the sealed funnel. Exceeding it marks the conversation unavailable and
/// refuses, exactly as the indexer's replay budget does — never a truncated
/// result that looks complete.
///
/// Counted in rows rather than bytes because each row costs a fixed ~150 bytes
/// once decoded (map slot, `String`, `IndexedMessage`) regardless of how well
/// it compressed. At this cap that is roughly 1.5 GiB worst case.
const MAX_POSTINGS_ROWS: u64 = 10_000_000;

/// One segment on disk, with the state its footer declares.
#[derive(Debug, Clone)]
struct Segment {
    path: PathBuf,
    state: CoverageState,
    /// Decoded row count, read off the footer this pass already parses.
    rows: u64,
}

/// How much one conversation's segment directory holds.
///
/// Read off the directory listing and the footers a write already performs, so
/// a caller that needs these numbers pays no pass over the rows. Nothing here
/// is a remembered counter: a restarted process, or one whose bookkeeping table
/// filled, reads the same values back off disk.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct SegmentStats {
    /// Segments in the directory.
    pub(crate) segments: u32,
    /// Rows across every segment.
    pub(crate) rows: u64,
    /// Rows in every segment but the OLDEST.
    ///
    /// The oldest segment is the folded base — [`SearchProjection::rebuild`]
    /// and [`SearchProjection::compact`] both leave exactly one segment
    /// behind — so this is the quantity a fold actually removes, and it returns
    /// to zero the moment one runs. [`SegmentStats::rows`] does not: a fold
    /// merges rows, it never drops them, so a threshold on the total is crossed
    /// once and then true forever.
    pub(crate) unmerged_rows: u64,
}

/// What one [`SearchProjection::append`] left behind.
///
/// The stats alone were not enough. The design record's invalidation table asks
/// an append landing on an unavailable conversation to "remain unavailable" AND
/// to make a rebuild happen, and the store can only do the first half — it has
/// no queue and no journal. Reporting the clamp is what lets the caller do the
/// second half instead of publishing a segment that quietly stays refused with
/// nothing scheduled to repair it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct Appended {
    /// What the conversation's directory holds afterwards.
    pub(crate) stats: SegmentStats,
    /// Whether this append asked for `available: true` over a conversation the
    /// newest segment already said was unavailable, and was forced back down.
    pub(crate) clamped_unavailable: bool,
}

/// Fold a directory's per-segment row counts, oldest first, into its stats.
fn stats_from_rows(rows: &[u64]) -> SegmentStats {
    let total = rows.iter().copied().fold(0u64, u64::saturating_add);
    let base = rows.first().copied().unwrap_or(0);
    SegmentStats {
        segments: u32::try_from(rows.len()).unwrap_or(u32::MAX),
        rows: total,
        unmerged_rows: total.saturating_sub(base),
    }
}

/// The per-conversation projection.
///
/// Single-writer: exactly one process publishes, matching the deployment's
/// single-replica invariant.
#[derive(Clone)]
pub(crate) struct SearchProjection {
    root: Arc<PathBuf>,
    #[cfg(test)]
    fail_next_remove: Arc<std::sync::atomic::AtomicBool>,
}

impl SearchProjection {
    /// Open (creating if absent) the projection rooted at `root`.
    ///
    /// # Errors
    ///
    /// [`StoreError::Io`] if the root cannot be created.
    pub(crate) fn open(root: PathBuf) -> Result<Self, StoreError> {
        std::fs::create_dir_all(&root)?;
        Ok(Self {
            root: Arc::new(root),
            #[cfg(test)]
            fail_next_remove: Arc::new(std::sync::atomic::AtomicBool::new(false)),
        })
    }

    /// Make the next removal fail before touching disk.
    #[cfg(test)]
    pub(crate) fn fail_next_remove(&self) {
        self.fail_next_remove
            .store(true, std::sync::atomic::Ordering::SeqCst);
    }

    /// The directory holding one conversation's segments.
    ///
    /// Named with a Hive-style `conversation_id=<name>` component, which is
    /// what a `ListingTable` needs to treat it as a partition column.
    ///
    /// Encoding is ENFORCED here, and it happens ONCE. Every method takes a
    /// `&str`, so a caller that forgets would otherwise write a raw namespaced
    /// id (`web:<uuid>` is the actual on-the-wire form) into a path — where the
    /// `:` is percent-encoded downstream and the partition value stops
    /// equalling the journal's. [`SearchProjection::remove`] and
    /// [`SearchProjection::destroy`] both do `remove_dir_all` on this path, so
    /// getting it wrong is not merely a lookup miss.
    ///
    /// This is the only place the codec runs on this path, because the codec is
    /// NOT idempotent: `:` encodes to `_3a`, and encoding that again gives
    /// `__3a`. A caller that normalized before calling in would address a
    /// directory nothing ever wrote.
    ///
    /// The column therefore carries the PHYSICAL partition key. A reader that
    /// wants the conversation id decodes it; it is not the logical id.
    ///
    /// # Panics
    ///
    /// Panics if `partition` cannot be encoded. Every caller here holds a name
    /// the journal already accepted, so an unencodable one means this
    /// projection and the journal disagree about what a partition is.
    fn dir_for(root: &Path, partition: &str) -> PathBuf {
        let encoded = encode_partition(partition)
            .expect("the journal accepted this partition name, so it encodes");
        root.join(format!("conversation_id={encoded}"))
    }

    /// Every segment path for one conversation, ordered oldest first, without
    /// decoding a single footer.
    ///
    /// The filename already encodes `(watermark, sequence)` zero-padded, so a
    /// path sort is the publication order — and a caller that only needs to
    /// enumerate or delete must not be blocked by a footer it cannot read.
    fn segment_paths(root: &Path, partition: &str) -> Result<Vec<PathBuf>, StoreError> {
        let dir = Self::dir_for(root, partition);
        if !dir.exists() {
            return Ok(Vec::new());
        }
        let mut paths = Vec::new();
        for entry in std::fs::read_dir(&dir)? {
            let path = entry?.path();
            if path.extension().and_then(std::ffi::OsStr::to_str) == Some("parquet") {
                paths.push(path);
            }
        }
        paths.sort();
        Ok(paths)
    }

    /// Every segment for one conversation, ordered oldest first.
    ///
    /// Ordering is by the watermark each segment reaches, then by its sequence
    /// — so a republish at the same watermark (an availability flip) sorts
    /// after the content it supersedes.
    fn segments(root: &Path, partition: &str, key_id: &str) -> Result<Vec<Segment>, StoreError> {
        let dir = Self::dir_for(root, partition);
        if !dir.exists() {
            return Ok(Vec::new());
        }

        let mut segments = Vec::new();
        for entry in std::fs::read_dir(&dir)? {
            let path = entry?.path();
            if path.extension().and_then(std::ffi::OsStr::to_str) != Some("parquet") {
                continue;
            }
            let file = std::fs::File::open(&path)?;
            let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
            let state = state_from(
                builder.metadata().file_metadata().key_value_metadata(),
                key_id,
            )?;
            let rows = u64::try_from(builder.metadata().file_metadata().num_rows())
                .map_err(|_| StoreError::Corrupt)?;
            segments.push(Segment { path, state, rows });
        }

        // The filename encodes (watermark, sequence) zero-padded, so a plain
        // path sort is the publication order.
        segments.sort_by(|a, b| a.path.cmp(&b.path));
        Ok(segments)
    }

    /// Append a segment covering the messages just indexed.
    ///
    /// `messages` is the DELTA, not the conversation. Writing the whole
    /// conversation each time is what made this quadratic.
    ///
    /// Returns what the directory holds afterwards, so a caller deciding
    /// whether to compact reads the true segment count and row totals rather
    /// than a counter it maintained beside the store. The listing is one the
    /// write already performs, so the numbers are free — and unlike a counter
    /// they cannot be lost to a restart or to a bookkeeping table that filled.
    ///
    /// # Errors
    ///
    /// [`StoreError::Io`] or [`StoreError::Parquet`] on a write failure.
    /// Existing segments are untouched, so a failed append never degrades what
    /// is already readable. [`StoreError::Destroyed`] when the conversation
    /// carries the destroy tombstone.
    pub(crate) async fn append(
        &self,
        partition: &str,
        messages: &[IndexedMessage],
        coverage: &Coverage,
        key_id: &str,
    ) -> Result<Appended, StoreError> {
        let (root, partition, messages, coverage, key_id) = (
            Arc::clone(&self.root),
            partition.to_owned(),
            messages.to_vec(),
            coverage.clone(),
            key_id.to_owned(),
        );
        blocking(move || {
            // Read ONE footer, the newest, and answer both questions off it.
            //
            // A destroy is terminal, so an append after one is refused rather
            // than written: the tombstone carries watermark zero, so any later
            // segment sorts after it, and coverage would read the resurrected
            // conversation as ordinary.
            //
            // And an append must never restore availability. The design is
            // explicit -- "append while unavailable: remain unavailable; the
            // append becomes the rebuild rather than waiting for one" -- and
            // without this a routine turn landing after an excision marker
            // silently made the conversation searchable again, excised terms
            // included. Only a rebuild clears it. The store owns the first half
            // of that row and cannot own the second: it has no queue and no
            // journal, so it REPORTS the clamp (`Appended::clamped_unavailable`)
            // and the worker turns the append into the rebuild.
            //
            // An unreadable footer stays swallowed: a stale term key or format
            // is what a rebuild repairs, and blocking the append would not make
            // that rebuild happen any sooner.
            let state = Self::newest_state(&root, &partition, &key_id);
            if matches!(state, Ok(CoverageState::Destroyed)) {
                return Err(StoreError::Destroyed);
            }
            let newest = match state {
                Ok(CoverageState::Indexed(newest)) => Some(newest),
                _ => None,
            };
            let clamped_unavailable =
                coverage.available && newest.as_ref().is_some_and(|newest| !newest.available);
            // The excision frontier is a MONOTONIC property of the whole
            // directory, so it takes the higher of what this pass scanned and
            // what the prefix already recorded. A pass that read a shorter range
            // than its predecessor must not retract a scan that already
            // happened — and it must not advance one that did not, which is why
            // this is a max rather than an overwrite in either direction.
            let coverage = Coverage {
                indexed_through: coverage.indexed_through,
                source_incarnation: coverage.source_incarnation,
                available: coverage.available && !clamped_unavailable,
                excision_scanned_through: coverage
                    .excision_scanned_through
                    .max(newest.map_or(0, |newest| newest.excision_scanned_through)),
            };
            Self::write_segment(
                &root,
                &partition,
                &messages,
                &coverage,
                &key_id,
                Marker::Live,
            )?;
            Ok(Appended {
                stats: Self::stats_of(&Self::dir_for(&root, &partition))?,
                clamped_unavailable,
            })
        })
        .await
    }

    /// Replace every segment with one covering `messages`.
    ///
    /// For a rebuild: a rewrite or repair compacts journal positions, so the
    /// existing segments describe a prefix that no longer exists and must go
    /// rather than be appended to.
    ///
    /// Returns the resulting stats, as [`SearchProjection::append`] does. They
    /// are one segment and no unmerged rows by construction, which is what
    /// makes a rebuild reset every compaction trigger without the caller
    /// special-casing it.
    ///
    /// # Errors
    ///
    /// As [`SearchProjection::append`], including [`StoreError::Destroyed`] —
    /// a destroyed conversation has no journal to rebuild FROM, so a rebuild
    /// against one is a caller bug rather than a repair. The new segment is
    /// written before the old ones are removed, so a failure mid-way leaves a
    /// readable conversation rather than an empty directory.
    pub(crate) async fn rebuild(
        &self,
        partition: &str,
        messages: &[IndexedMessage],
        coverage: &Coverage,
        key_id: &str,
    ) -> Result<SegmentStats, StoreError> {
        let (root, partition, messages, coverage, key_id) = (
            Arc::clone(&self.root),
            partition.to_owned(),
            messages.to_vec(),
            coverage.clone(),
            key_id.to_owned(),
        );
        blocking(move || {
            // Listed by PATH, not by decoding footers: `segments()` refuses the
            // whole conversation when any footer carries a stale key or format,
            // and a rebuild is exactly what must clear that. Swallowing the
            // error left the stale segments in place forever and made the
            // rebuild a permanent no-op against the one state it exists to
            // repair.
            if matches!(
                Self::newest_state(&root, &partition, &key_id),
                Ok(CoverageState::Destroyed)
            ) {
                return Err(StoreError::Destroyed);
            }
            let existing = Self::segment_paths(&root, &partition)?;
            Self::write_segment(
                &root,
                &partition,
                &messages,
                &coverage,
                &key_id,
                Marker::Live,
            )?;
            for path in existing {
                // Loudly: a surviving old segment with a HIGHER watermark sorts
                // after the rebuilt one, so coverage reports the stale
                // watermark and the union resurrects the excised terms. A
                // rebuild that reports success while leaving that behind is the
                // worst outcome available here.
                std::fs::remove_file(&path)?;
            }
            // AFTER the removals, so the stats describe the one segment a
            // rebuild leaves rather than the prefix it just replaced.
            Self::stats_of(&Self::dir_for(&root, &partition))
        })
        .await
    }

    /// Fold every segment for one conversation into a single one.
    ///
    /// Bounds read cost after a run of appends. Coverage is unchanged — this
    /// rewrites how the same prefix is stored, never what it covers.
    ///
    /// # Errors
    ///
    /// As [`SearchProjection::append`], plus [`StoreError::TooLarge`] when the
    /// segments exceed the [`MAX_POSTINGS_ROWS`] read cap to decode.
    pub(crate) async fn compact(&self, partition: &str, key_id: &str) -> Result<(), StoreError> {
        let (root, partition, key_id) = (
            Arc::clone(&self.root),
            partition.to_owned(),
            key_id.to_owned(),
        );
        blocking(move || {
            let segments = Self::segments(&root, &partition, &key_id)?;
            if segments.len() < 2 {
                return Ok(());
            }
            // A destroyed conversation has exactly one segment, so the count
            // guard above already returned; matching rather than assuming keeps
            // that a property of the code instead of of the reader's memory.
            let Some(CoverageState::Indexed(coverage)) = segments.last().map(|s| s.state.clone())
            else {
                return Ok(());
            };
            let merged = Self::read_segments(&segments)?;
            Self::write_segment(&root, &partition, &merged, &coverage, &key_id, Marker::Live)?;
            for segment in segments {
                std::fs::remove_file(&segment.path)?;
            }
            Ok(())
        })
        .await
    }

    /// What the store alone knows about one conversation: its newest segment's
    /// statement.
    ///
    /// Costs one footer read rather than a scan, which is what keeps a refusal
    /// cheap across a large participation set.
    ///
    /// This asks no journal, so it can report that a conversation is indexed
    /// but never that the index still describes the journal it was built from
    /// — it never returns [`CoverageState::Stale`], and a search must go
    /// through [`SearchProjection::verified_coverage`] instead. What survives
    /// here is the administrative and observability read.
    ///
    /// # Errors
    ///
    /// [`StoreError::Corrupt`] when a footer is absent, unparseable, or
    /// written under a different term key or format version;
    /// [`StoreError::Io`]/[`StoreError::Parquet`] on a read failure.
    pub(crate) async fn coverage(
        &self,
        partition: &str,
        key_id: &str,
    ) -> Result<CoverageState, StoreError> {
        let (root, partition, key_id) = (
            Arc::clone(&self.root),
            partition.to_owned(),
            key_id.to_owned(),
        );
        blocking(move || Self::newest_state(&root, &partition, &key_id)).await
    }

    /// One conversation's coverage, checked against the journal it claims to
    /// describe.
    ///
    /// The stored incarnation names the exact physical source; `journal`
    /// reads the live source and the two must agree. A content-changing
    /// physical rewrite rotates that source before stale rows may be served.
    ///
    /// This is what let the design record drop its startup barrier. A barrier
    /// costs O(every partition in the deployment) on each boot and still only
    /// catches what was already broken at boot; verifying per read costs
    /// O(partitions in scope) — bounded by
    /// [`super::SearchIndexConfig::max_partitions_read`] — one footer read, two
    /// exact-source reads, and one bounded tail scan apiece. It catches the
    /// stale index at the moment it would be trusted.
    ///
    /// A conversation that is [`CoverageState::NeverIndexed`] or
    /// [`CoverageState::Destroyed`] is returned unchanged and the journal is
    /// never asked: neither claims a prefix, so neither has one to check.
    ///
    /// # Errors
    ///
    /// As [`SearchProjection::coverage`]. A journal that cannot answer is not
    /// an error — it is a mismatch, and a mismatch reads as
    /// [`CoverageState::Stale`] rather than as coverage.
    pub(crate) async fn verified_coverage(
        &self,
        partition: &str,
        key_id: &str,
        journal: &impl JournalState,
    ) -> Result<CoverageState, StoreError> {
        // ONE footer read for both the state and the watermark the journal is
        // asked about. Reading it again for the position would be the same
        // two-call gap the trait exists to close, moved inside the store.
        let coverage = match self.coverage(partition, key_id).await? {
            CoverageState::Indexed(coverage) => coverage,
            other => return Ok(other),
        };

        let live = journal.source_incarnation(partition).await;
        if live != Some(coverage.source_incarnation) {
            return Ok(CoverageState::Stale);
        }

        // The source check just passed, which says the indexed physical source
        // still exists — and says nothing at all about an excision appended to
        // it. Excision is a pure append: the source remains unchanged whether
        // or not a marker sits at the tail naming positions inside the prefix.
        // Asking the journal is the only way to tell, and it is what makes the
        // obligation survive a restart: before this, the sole record that a
        // rebuild was owed was the in-memory dirty mark, which a graceful
        // shutdown discards.
        let scan = journal
            .excision_since(partition, coverage.excision_scanned_through)
            .await;
        let still_live = journal.source_incarnation(partition).await;
        if scan.is_clear() && still_live == Some(coverage.source_incarnation) {
            Ok(CoverageState::Indexed(coverage))
        } else {
            tracing::warn!(
                partition,
                ?scan,
                scanned_through = coverage.excision_scanned_through,
                "search index holds a conversation whose journal has an excision it never \
                 applied; refusing until a rebuild strips it"
            );
            Ok(CoverageState::Stale)
        }
    }

    /// The state the NEWEST segment declares, without decoding the others.
    ///
    /// Decoding every segment's footer would be O(segments) on a call whose
    /// whole job is deciding cheaply whether to read at all — and with nothing
    /// compacting yet that is one open-and-parse per committed turn, across up
    /// to `max_partitions_read` conversations per search. The filename already
    /// encodes the ordering, so the newest is a path max.
    ///
    /// Reading only the newest also means one stale-key segment elsewhere in
    /// the directory no longer makes the whole conversation unreadable, which
    /// is what made a rebuild unable to repair it.
    fn newest_state(
        root: &Path,
        partition: &str,
        key_id: &str,
    ) -> Result<CoverageState, StoreError> {
        let Some(newest) = Self::segment_paths(root, partition)?.pop() else {
            return Ok(CoverageState::NeverIndexed);
        };
        let file = std::fs::File::open(&newest)?;
        let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
        state_from(
            builder.metadata().file_metadata().key_value_metadata(),
            key_id,
        )
    }

    /// One conversation's messages, unioned across every segment.
    ///
    /// A later segment supersedes an earlier one at the same position, which is
    /// what makes an at-least-once replay idempotent.
    ///
    /// # Errors
    ///
    /// [`StoreError::TooLarge`] when the segments exceed
    /// [`MAX_POSTINGS_ROWS`]; otherwise as
    /// [`SearchProjection::coverage`].
    pub(crate) async fn postings(
        &self,
        partition: &str,
        key_id: &str,
    ) -> Result<Option<PostingsRecord>, StoreError> {
        let (root, partition, key_id) = (
            Arc::clone(&self.root),
            partition.to_owned(),
            key_id.to_owned(),
        );
        blocking(move || {
            let segments = Self::segments(&root, &partition, &key_id)?;
            // A destroyed conversation has rows-less segments only, so decoding
            // them would return an EMPTY postings record -- "indexed, nothing
            // in it" -- for a conversation that holds nothing and never will.
            // `None` is the same answer this gives for one nothing indexed,
            // which is what a caller reaching postings without consulting
            // coverage deserves to see.
            if segments.is_empty()
                || matches!(
                    segments.last().map(|s| &s.state),
                    Some(&CoverageState::Destroyed)
                )
            {
                return Ok(None);
            }
            Ok(Some(PostingsRecord {
                messages: Self::read_segments(&segments)?,
            }))
        })
        .await
    }

    /// One conversation's segment stats, with every footer checked.
    ///
    /// The checking listing, and the cheap way to ask whether a stored prefix
    /// is still readable. It opens EVERY segment's footer rather than only the
    /// newest, so a file that no longer decodes — a torn write, bit rot, a
    /// segment left under a rotated term key — surfaces as
    /// [`StoreError::Corrupt`] or [`StoreError::Parquet`] instead of hiding
    /// behind a healthy newest footer. Footers only: it costs one open per
    /// segment and reads no rows, where decoding the conversation to learn the
    /// same thing costs O(rows) on every call.
    ///
    /// A conversation with no segments reports zeroes rather than an error —
    /// "nothing here" is a state, and [`SearchProjection::coverage`] is what
    /// says which state.
    ///
    /// # Errors
    ///
    /// As [`SearchProjection::coverage`], plus [`StoreError::TooLarge`] when
    /// the footers already sum past [`MAX_POSTINGS_ROWS`] — the read cap,
    /// answered without decoding a single row.
    pub(crate) async fn segment_stats(
        &self,
        partition: &str,
        key_id: &str,
    ) -> Result<SegmentStats, StoreError> {
        let (root, partition, key_id) = (
            Arc::clone(&self.root),
            partition.to_owned(),
            key_id.to_owned(),
        );
        blocking(move || {
            let segments = Self::segments(&root, &partition, &key_id)?;
            let rows: Vec<u64> = segments.iter().map(|segment| segment.rows).collect();
            let stats = stats_from_rows(&rows);
            if stats.rows > MAX_POSTINGS_ROWS {
                return Err(StoreError::TooLarge);
            }
            Ok(stats)
        })
        .await
    }

    /// Mark a conversation unsearchable, leaving its rows in place.
    ///
    /// A coverage statement, not a deletion: appends an empty segment at the
    /// current watermark carrying `available: false`, so the newest segment —
    /// the one coverage reads — says unavailable while every row survives for
    /// a rebuild to reuse.
    ///
    /// A conversation with no segments is already unsearchable, so this is a
    /// no-op for one — never a fabricated segment claiming a watermark nothing
    /// established. A destroyed one is unsearchable by a stronger statement
    /// already, and writing a segment over its tombstone would weaken that to
    /// "merely unavailable", which invites the rebuild the tombstone exists to
    /// forbid.
    ///
    /// # Errors
    ///
    /// As [`SearchProjection::append`].
    pub(crate) async fn mark_unavailable(
        &self,
        partition: &str,
        key_id: &str,
    ) -> Result<(), StoreError> {
        let CoverageState::Indexed(coverage) = self.coverage(partition, key_id).await? else {
            return Ok(());
        };
        if !coverage.available {
            return Ok(());
        }
        self.append(
            partition,
            &[],
            &Coverage {
                available: false,
                ..coverage
            },
            key_id,
        )
        .await
        .map(|_| ())
    }

    /// Remove a conversation's rows and record that it was destroyed.
    ///
    /// The rows go — there is no journal left to rebuild them from, so leaving
    /// them merely unavailable is an indefinite hold on text the deployment was
    /// told to forget. The STATE stays, as a rows-less tombstone segment: the
    /// design record's invalidation table asks destroy to "record the
    /// authoritative destroyed state", and an absent directory records nothing.
    /// A reader that cannot tell destroyed from never-indexed either schedules
    /// an index build that can never succeed, or — worse, once the worker
    /// treats an unindexed conversation as merely behind — reports the
    /// conversation as one it is about to catch up on.
    ///
    /// Terminal: [`SearchProjection::append`] and
    /// [`SearchProjection::rebuild`] refuse afterwards with
    /// [`StoreError::Destroyed`].
    ///
    /// Rows first, tombstone second. The reverse order would leave the
    /// tombstone — which carries watermark zero, so it sorts FIRST — behind
    /// segments that outrank it, and coverage would keep reading the old
    /// conversation as live. In this order a crash between the two leaves
    /// [`CoverageState::NeverIndexed`], which is uncovered and rebuildable
    /// rather than searchable and wrong.
    ///
    /// # Errors
    ///
    /// [`StoreError::Io`] on a removal failure, or as
    /// [`SearchProjection::append`] on the tombstone write. A conversation with
    /// no segments still gets a tombstone: destroy is a statement about the
    /// conversation, not about what happened to be indexed for it.
    pub(crate) async fn destroy(&self, partition: &str, key_id: &str) -> Result<(), StoreError> {
        let (root, partition, key_id) = (
            Arc::clone(&self.root),
            partition.to_owned(),
            key_id.to_owned(),
        );
        blocking(move || {
            let dir = Self::dir_for(&root, &partition);
            if dir.exists() {
                std::fs::remove_dir_all(&dir)?;
            }
            Self::write_segment(
                &root,
                &partition,
                &[],
                &Coverage {
                    // Tombstones return before coverage fields are parsed; the
                    // fixed-width zero value is framing only and names no live
                    // source.
                    indexed_through: 0,
                    source_incarnation: PartitionIncarnation::from_bytes([0; 32]),
                    available: false,
                    // Nothing left to scan and nothing that may ever be
                    // published again, so the frontier stays where a
                    // never-indexed conversation's would be.
                    excision_scanned_through: 0,
                },
                &key_id,
                Marker::Destroyed,
            )
        })
        .await
    }

    /// Segment count and row totals for a conversation directory, off footers
    /// alone.
    ///
    /// Deliberately checks neither the term key nor the format version: this
    /// answers "how much is on disk", which is a question about files rather
    /// than about whether they still decode under the current key. A write that
    /// succeeded must not be reported as failed because a sibling segment
    /// carries a rotated key — that is what
    /// [`SearchProjection::segment_stats`] is for, and it runs before the
    /// write rather than after it.
    fn stats_of(dir: &Path) -> Result<SegmentStats, StoreError> {
        let mut paths = Vec::new();
        if dir.exists() {
            for entry in std::fs::read_dir(dir)? {
                let path = entry?.path();
                if path.extension().and_then(std::ffi::OsStr::to_str) == Some("parquet") {
                    paths.push(path);
                }
            }
        }
        // The filename encodes (watermark, sequence) zero-padded, so a plain
        // path sort puts the folded base first.
        paths.sort();

        let mut rows = Vec::with_capacity(paths.len());
        for path in &paths {
            let file = std::fs::File::open(path)?;
            let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
            rows.push(
                u64::try_from(builder.metadata().file_metadata().num_rows())
                    .map_err(|_| StoreError::Corrupt)?,
            );
        }
        Ok(stats_from_rows(&rows))
    }

    /// Remove a conversation's segments entirely, leaving no trace.
    ///
    /// The migration SOURCE, and only it. The design record resolves a
    /// migrated conversation's coverage "from the destination", so the source
    /// must read as though it were never indexed — a tombstone there would
    /// report the conversation as destroyed while it is alive under its new
    /// id, which is the same silent not-found
    /// [`StoreError::Corrupt`] exists to prevent, reached by a third door.
    /// Destroy takes [`SearchProjection::destroy`] instead.
    ///
    /// # Errors
    ///
    /// [`StoreError::Io`] on a removal failure. A conversation with no
    /// segments is not an error.
    pub(crate) async fn remove(&self, partition: &str) -> Result<(), StoreError> {
        #[cfg(test)]
        if self
            .fail_next_remove
            .swap(false, std::sync::atomic::Ordering::SeqCst)
        {
            return Err(std::io::Error::other("injected remove failure").into());
        }
        let (root, partition) = (Arc::clone(&self.root), partition.to_owned());
        blocking(move || {
            let dir = Self::dir_for(&root, &partition);
            if dir.exists() {
                std::fs::remove_dir_all(&dir)?;
            }
            Ok(())
        })
        .await
    }

    /// The Parquet paths for exactly the conversations `authorized` names.
    ///
    /// This is what a `ListingTable` is registered over — never
    /// [`SearchProjection::root`]. The accepted architecture makes catalog
    /// scoping the authorization boundary ("a session's catalog registers only
    /// the authorized conversations' Parquet paths"), and registering the root
    /// would demote that to a `WHERE conversation_id = ...` predicate, where
    /// one predicate-construction bug leaks one persona's term membership to
    /// another.
    ///
    /// A destroyed conversation still has a directory, holding its tombstone.
    /// The tombstone carries no rows, so registering it contributes nothing to
    /// a scan — and a caller reaches this only for conversations whose
    /// [`SearchProjection::coverage`] it already accepted, which a destroyed
    /// one never passes.
    ///
    /// # Errors
    ///
    /// [`StoreError::Io`] when a conversation's directory cannot be listed.
    pub(crate) async fn authorized_paths(
        &self,
        authorized: Vec<String>,
    ) -> Result<Vec<PathBuf>, StoreError> {
        let root = Arc::clone(&self.root);
        blocking(move || {
            let mut paths = Vec::new();
            for partition in &authorized {
                let dir = Self::dir_for(&root, partition);
                if !dir.exists() {
                    continue;
                }
                for entry in std::fs::read_dir(&dir)? {
                    let path = entry?.path();
                    if path.extension().and_then(std::ffi::OsStr::to_str) == Some("parquet") {
                        paths.push(path);
                    }
                }
            }
            paths.sort();
            Ok(paths)
        })
        .await
    }

    /// The projection root.
    ///
    /// Administrative and test use only: it names every conversation, so it is
    /// never what a scoped session's catalog is registered over — see
    /// [`SearchProjection::authorized_paths`].
    pub(crate) fn root(&self) -> &Path {
        &self.root
    }

    /// Write one segment with an explicit format for legacy-format refusal tests.
    #[cfg(test)]
    pub(crate) async fn write_format_for_test(
        &self,
        partition: &str,
        messages: &[IndexedMessage],
        coverage: &Coverage,
        key_id: &str,
        format: &str,
    ) -> Result<(), StoreError> {
        let (root, partition, messages, coverage, key_id, format) = (
            Arc::clone(&self.root),
            partition.to_owned(),
            messages.to_vec(),
            coverage.clone(),
            key_id.to_owned(),
            format.to_owned(),
        );
        blocking(move || {
            Self::write_segment_with_format(
                &root,
                &partition,
                &messages,
                &coverage,
                &key_id,
                Marker::Live,
                &format,
            )
        })
        .await
    }

    /// Write one segment, atomically.
    fn write_segment(
        root: &Path,
        partition: &str,
        messages: &[IndexedMessage],
        coverage: &Coverage,
        key_id: &str,
        marker: Marker,
    ) -> Result<(), StoreError> {
        Self::write_segment_with_format(
            root,
            partition,
            messages,
            coverage,
            key_id,
            marker,
            FORMAT_VERSION,
        )
    }

    /// Write one segment, atomically, with the supplied format tag.
    fn write_segment_with_format(
        root: &Path,
        partition: &str,
        messages: &[IndexedMessage],
        coverage: &Coverage,
        key_id: &str,
        marker: Marker,
        format: &str,
    ) -> Result<(), StoreError> {
        let dir = Self::dir_for(root, partition);
        std::fs::create_dir_all(&dir)?;

        // A process-unique `O_EXCL` temp, matching this repo's own atomic-write
        // reference (`polyc_wallet_delegation::secret_store`). A fixed name
        // opened truncating would let two publishes for one conversation
        // interleave into a single file and rename the corruption into place —
        // "single writer" is a process-level claim, not a serialization of two
        // tasks inside that process.
        let temp = dir.join(format!(
            "seg.{}.{}.tmp",
            std::process::id(),
            TEMP_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
        ));

        // The sequence is DURABLE state, derived from what is already on disk —
        // never a process-local counter. A counter restarts at zero while the
        // directory keeps its old names, and `rename` silently REPLACES a
        // colliding destination, so `rebuild` would delete the very file it
        // just wrote and leave the conversation indistinguishable from never
        // indexed. Reading the max off the directory costs one listing and
        // cannot rewind.
        let sequence = next_sequence(&dir)?;
        // Zero-padded so a plain path sort is the publication order.
        let final_path = dir.join(format!(
            "part-{:020}-{:010}.parquet",
            coverage.indexed_through, sequence
        ));
        if final_path.exists() {
            // Belt and braces: never let a rename replace a live segment. The
            // temp file does not exist yet — `open_owner_only` runs below — so
            // there is nothing to clean up on this path.
            return Err(StoreError::Io(format!(
                "search projection segment {} already exists",
                final_path.display()
            )));
        }

        let props = WriterProperties::builder()
            .set_compression(Compression::ZSTD(ZstdLevel::default()))
            .set_column_bloom_filter_enabled(TERM_HASH_COLUMN.into(), true)
            .set_sorting_columns(Some(vec![SortingColumn {
                column_idx: 2,
                descending: false,
                nulls_first: false,
            }]))
            .set_key_value_metadata(Some(vec![
                kv(META_INDEXED_THROUGH, &coverage.indexed_through.to_string()),
                kv(
                    META_INCARNATION,
                    &hex(coverage.source_incarnation.as_bytes()),
                ),
                kv(META_AVAILABLE, &coverage.available.to_string()),
                kv(
                    META_EXCISION_SCANNED_THROUGH,
                    &coverage.excision_scanned_through.to_string(),
                ),
                kv(META_KEY_ID, key_id),
                kv(META_FORMAT, format),
                kv(
                    META_DESTROYED,
                    &matches!(marker, Marker::Destroyed).to_string(),
                ),
            ]))
            .build();

        let written = (|| -> Result<(), StoreError> {
            let file = open_owner_only(&temp)?;
            let schema = projection_schema();
            let mut writer =
                ArrowWriter::try_new(file.try_clone()?, Arc::clone(&schema), Some(props))?;
            if let Some(batch) = batch_for(&schema, messages)? {
                writer.write(&batch)?;
            }
            writer.close()?;
            // `close` flushes to the `Write` impl, which for a file is a write
            // into page cache, not durability. Rename is atomic against
            // concurrent READERS; it says nothing about a crash.
            file.sync_all()?;
            Ok(())
        })();

        if let Err(err) = written {
            let _ = std::fs::remove_file(&temp);
            return Err(err);
        }

        std::fs::rename(&temp, &final_path)?;
        Ok(())
    }

    /// Decode every segment, later positions winning, under the byte cap.
    fn read_segments(segments: &[Segment]) -> Result<Vec<IndexedMessage>, StoreError> {
        // Bound ROWS, not compressed bytes. On-disk size is a terrible proxy
        // for decoded memory here: a conversation of reactions and non-Latin
        // text is all sentinel rows, so the hash column compresses to almost
        // nothing while every row still costs its map slot, its `String`, and
        // its `IndexedMessage`. Measured overshoot against a byte cap ran from
        // 1.6x to 91x. The row count is already in the footers being parsed.
        let rows: u64 = segments.iter().map(|segment| segment.rows).sum();
        if rows > MAX_POSTINGS_ROWS {
            return Err(StoreError::TooLarge);
        }

        // Keyed on `position` alone: a position identifies exactly one message
        // by the type's own invariant, so pairing it with the turn id would
        // allocate a `String` per row only to discard all but one per group.
        let mut by_position: std::collections::BTreeMap<u64, (String, Vec<u32>)> =
            std::collections::BTreeMap::new();

        for segment in segments {
            // Accumulate WITHIN a segment, then replace across segments. A
            // position re-indexed with fewer terms -- a redaction, or text
            // edited until every term is non-ASCII -- must lose the terms it no
            // longer has. Unioning across segments would keep them forever,
            // which is a silent failure to forget.
            let mut this_segment: std::collections::BTreeMap<u64, (String, Vec<u32>)> =
                std::collections::BTreeMap::new();

            let file = std::fs::File::open(&segment.path)?;
            let reader = ParquetRecordBatchReaderBuilder::try_new(file)?.build()?;
            for batch in reader {
                let batch = batch?;
                let turns = column::<StringArray>(&batch, 0)?;
                let positions = column::<UInt64Array>(&batch, 1)?;
                let hashes = column::<UInt32Array>(&batch, 2)?;
                for row in 0..batch.num_rows() {
                    let entry = this_segment
                        .entry(positions.value(row))
                        .or_insert_with(|| (turns.value(row).to_owned(), Vec::new()));
                    let hash = hashes.value(row);
                    // The sentinel records that a message exists with no
                    // searchable term; it is not itself a term.
                    if hash != NO_TERMS_SENTINEL {
                        entry.1.push(hash);
                    }
                }
            }

            // Segments are iterated oldest first, so this is last-writer-wins.
            by_position.extend(this_segment);
        }

        Ok(by_position
            .into_iter()
            .map(|(position, (turn_id, mut term_hashes))| {
                term_hashes.sort_unstable();
                term_hashes.dedup();
                IndexedMessage {
                    position,
                    turn_id,
                    term_hashes,
                }
            })
            .collect())
    }
}

/// Run blocking filesystem and Parquet work off the runtime's worker threads.
async fn blocking<T, F>(work: F) -> Result<T, StoreError>
where
    F: FnOnce() -> Result<T, StoreError> + Send + 'static,
    T: Send + 'static,
{
    tokio::task::spawn_blocking(work)
        .await
        .map_err(|err| StoreError::Io(format!("search projection task: {err}")))?
}

/// Build one `RecordBatch` from a conversation's messages, or `None` when it
/// has no searchable terms.
fn batch_for(
    schema: &Arc<Schema>,
    messages: &[IndexedMessage],
) -> Result<Option<RecordBatch>, StoreError> {
    let mut turns: Vec<&str> = Vec::new();
    let mut positions: Vec<u64> = Vec::new();
    let mut hashes: Vec<u32> = Vec::new();

    for message in messages {
        if message.term_hashes.is_empty() {
            // A message whose text yields no searchable term still has to
            // survive: the tokenizer drops punctuation-only text and every
            // non-ASCII term, so a CJK-only message or a bare reaction lands
            // here. Emitting no row would make `publish` then `postings` lose
            // it, and a conversation held in a non-Latin script would round
            // trip to nothing. The sentinel is a hash no real term can take.
            turns.push(&message.turn_id);
            positions.push(message.position);
            hashes.push(NO_TERMS_SENTINEL);
            continue;
        }
        for &hash in &message.term_hashes {
            turns.push(&message.turn_id);
            positions.push(message.position);
            hashes.push(hash);
        }
    }

    if turns.is_empty() {
        return Ok(None);
    }

    // Sorting by `term_hash` is what makes the page statistics selective; the
    // remaining keys keep the order total so a rewrite of unchanged content
    // produces an identical file.
    let mut rows: Vec<(usize, &&str)> = turns.iter().enumerate().collect();
    rows.sort_by_key(|&(i, _)| (hashes[i], positions[i], turns[i]));
    let order: Vec<usize> = rows.into_iter().map(|(i, _)| i).collect();
    let turns: Vec<&str> = order.iter().map(|&i| turns[i]).collect();
    let positions: Vec<u64> = order.iter().map(|&i| positions[i]).collect();
    let hashes: Vec<u32> = order.iter().map(|&i| hashes[i]).collect();

    RecordBatch::try_new(
        Arc::clone(schema),
        vec![
            Arc::new(StringArray::from(turns)),
            Arc::new(UInt64Array::from(positions)),
            Arc::new(UInt32Array::from(hashes)),
        ],
    )
    .map(Some)
    .map_err(Into::into)
}

/// Downcast one column, mapping a schema surprise to [`StoreError::Corrupt`].
fn column<T: 'static>(batch: &RecordBatch, index: usize) -> Result<&T, StoreError> {
    batch
        .column(index)
        .as_any()
        .downcast_ref::<T>()
        .ok_or(StoreError::Corrupt)
}

/// Read one segment's state out of a Parquet footer's key-value metadata.
///
/// Never [`CoverageState::NeverIndexed`]: a footer only exists where a segment
/// does, so absence is the caller's to report.
fn state_from(
    metadata: Option<&Vec<parquet::file::metadata::KeyValue>>,
    expected_key_id: &str,
) -> Result<CoverageState, StoreError> {
    let entries = metadata.ok_or(StoreError::Corrupt)?;
    let get = |key: &str| {
        entries
            .iter()
            .find(|kv| kv.key == key)
            .and_then(|kv| kv.value.as_deref())
    };

    // The tombstone answers BEFORE the term key and format are checked, and
    // before any coverage value is parsed. It holds no rows, so there is
    // nothing for a key rotation or a format bump to make unreadable — and a
    // destroyed conversation that reported `Corrupt` after a rotation would be
    // routed to the rebuild its whole purpose is to forbid.
    if get(META_DESTROYED) == Some("true") {
        return Ok(CoverageState::Destroyed);
    }

    let indexed_through = get(META_INDEXED_THROUGH)
        .and_then(|v| v.parse::<u64>().ok())
        .ok_or(StoreError::Corrupt)?;
    let source_incarnation =
        PartitionIncarnation::parse_hex(get(META_INCARNATION).ok_or(StoreError::Corrupt)?)
            .map_err(|_| StoreError::Corrupt)?;
    let available = get(META_AVAILABLE)
        .and_then(|v| v.parse::<bool>().ok())
        .ok_or(StoreError::Corrupt)?;
    // Required, with no default. A missing frontier would have to be read as
    // either zero (every conversation permanently unverifiable) or as the
    // watermark (every conversation silently vouching for an excision scan
    // nobody performed) — and the second is the failure this key exists to
    // close. A segment without it is a segment from a writer that did not have
    // this rule, so it is unreadable and gets rebuilt, which is what this
    // project does with every other legacy shape.
    let excision_scanned_through = get(META_EXCISION_SCANNED_THROUGH)
        .and_then(|v| v.parse::<u64>().ok())
        .ok_or(StoreError::Corrupt)?;

    // A file written under a different term key, or a different hashing
    // format, decodes perfectly and answers every query with zero hits. Refuse
    // it as unreadable so it is rebuilt, rather than serving a confident
    // "not found" for text it holds.
    if get(META_FORMAT) != Some(FORMAT_VERSION) || get(META_KEY_ID) != Some(expected_key_id) {
        return Err(StoreError::Corrupt);
    }

    Ok(CoverageState::Indexed(Coverage {
        indexed_through,
        source_incarnation,
        available,
        excision_scanned_through,
    }))
}

/// The next durable sequence for a conversation directory: one past the
/// highest any existing segment carries.
///
/// Derived from disk rather than memory because the value ends up in a
/// filename that outlives the process. Global across watermarks, so ordering
/// stays total even when two segments share one.
fn next_sequence(dir: &Path) -> Result<u64, StoreError> {
    let mut highest: Option<u64> = None;
    for entry in std::fs::read_dir(dir)? {
        let path = entry?.path();
        if path.extension().and_then(std::ffi::OsStr::to_str) != Some("parquet") {
            continue;
        }
        let Some(sequence) = path
            .file_stem()
            .and_then(std::ffi::OsStr::to_str)
            .and_then(|stem| stem.rsplit_once('-'))
            .and_then(|(_, seq)| seq.parse::<u64>().ok())
        else {
            continue;
        };
        highest = Some(highest.map_or(sequence, |seen: u64| seen.max(sequence)));
    }
    Ok(highest.map_or(0, |seen| seen.saturating_add(1)))
}

/// Monotonic suffix making each in-flight temp file unique within a process.
static TEMP_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// Create `path` exclusively, owner-readable only.
///
/// `O_EXCL` so two concurrent publishes cannot share a temp file, and mode
/// 0600 because these files are a membership oracle over user text — see
/// [`super::terms`] — which makes world-readable the wrong default even inside
/// a container.
fn open_owner_only(path: &Path) -> Result<std::fs::File, StoreError> {
    let mut options = std::fs::OpenOptions::new();
    options.write(true).create_new(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt as _;
        options.mode(0o600);
    }
    Ok(options.open(path)?)
}

/// Test hook: decode an incarnation value through the same helper the footer
/// read uses, so a decoder test pins the decoder rather than the file format.
#[cfg(test)]
pub(crate) fn unhex_for_test(incarnation: &str) -> Result<Vec<u8>, StoreError> {
    unhex(incarnation)
}

/// One footer key-value entry.
fn kv(key: &str, value: &str) -> parquet::file::metadata::KeyValue {
    parquet::file::metadata::KeyValue::new(key.to_owned(), value.to_owned())
}

/// Lower-hex encoding for the incarnation, which the footer carries as text.
fn hex(bytes: &[u8]) -> String {
    bytes.iter().fold(String::new(), |mut out, byte| {
        use std::fmt::Write as _;
        let _ = write!(out, "{byte:02x}");
        out
    })
}

/// Inverse of [`hex`], rejecting anything that is not an even run of hex
/// digits.
///
/// Chunks BYTES rather than slicing the `str`. The footer is untrusted content
/// — a hand-edited, bit-rotted, or foreign-writer value can hold any byte —
/// and `&text[i..i + 2]` panics when the index lands mid-codepoint, which an
/// even *byte* length does not prevent. A corrupt footer must return
/// [`StoreError::Corrupt`] so the conversation is rebuilt; aborting the
/// reading thread is not that.
fn unhex(text: &str) -> Result<Vec<u8>, StoreError> {
    let bytes = text.as_bytes();
    if !bytes.len().is_multiple_of(2) {
        return Err(StoreError::Corrupt);
    }
    bytes
        .chunks_exact(2)
        .map(|pair| {
            let digits = std::str::from_utf8(pair).map_err(|_| StoreError::Corrupt)?;
            u8::from_str_radix(digits, 16).map_err(|_| StoreError::Corrupt)
        })
        .collect()
}