zeppelin_embed/ingest/

purge.rs

//! Verifiable physical purge tokens, artifact rewriting, and recovery intent.

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

use xxhash_rust::xxh3::xxh3_64;

use crate::format::FormatFamily;
use crate::format::frame::{FormatError, decode_artifact, encode_artifact};
use crate::lifecycle::durability::{DurabilityPolicy, SyncRequirement};
use crate::lifecycle::stats::Accounting;
use crate::lifecycle::{PublishedSnapshot, Store, StoreError, StoreState};
use crate::manifest::Manifest;
use crate::manifest::io::{MANIFEST_FILE, load_manifest};
use crate::meta::{
    AliveSet, Column, ColumnInput, ColumnStore, ColumnStoreBuilder, ColumnValue, TIMESTAMP_COLUMN,
};
use crate::quant::Bit4Factors;
use crate::segment::layout::Int8Factors;
use crate::segment::reader::SegmentReader;
use crate::segment::writer::{
    SegmentBuild, SegmentDocumentVersions, SegmentFactors, SegmentPayloads, SegmentPostings,
    SegmentStoredMetadata, SegmentStoredText, write_segment_with_documents_payloads,
};
use crate::segment::{ClusteringKeyRange, SegmentId, SegmentMeta};
use crate::vfs::Vfs;
use crate::wal::LogSeq;

use super::{ActiveState, DocId, IngestDocument, SealedTombstoneDemand, wal_payload};

pub(crate) const PURGE_INTENT_FILE: &str = "purge.ze";
const PURGE_INTENT_TEMP_FILE: &str = ".purge.ze.tmp";
const WAL_PURGE_TEMP_FILE: &str = ".wal.ze.purge.tmp";

type WalImageRecords = (Vec<(u16, Vec<u8>)>, Vec<bool>);

#[derive(Clone, Copy)]
pub(crate) struct SealedDocumentMatch {
    pub(crate) segment_index: usize,
    pub(crate) row: usize,
    pub(crate) version: super::DocumentVersion,
}

pub(crate) struct PreparedSealedTombstones {
    manifest: Manifest,
    snapshot: PublishedSnapshot,
    replaced_paths: Vec<PathBuf>,
    replacement_ids: Vec<SegmentId>,
}

impl PreparedSealedTombstones {
    pub(crate) fn into_namespace_manifest(self) -> (Manifest, Vec<SegmentId>) {
        (self.manifest, self.replacement_ids)
    }

    pub(crate) fn abort(
        self,
        vfs: &dyn Vfs,
        directory: &Path,
        policy: DurabilityPolicy,
    ) -> Result<(), StoreError> {
        cleanup_replacement_segments(vfs, directory, &self.replacement_ids, policy)
    }

    pub(crate) fn commit(
        mut self,
        publication: &mut super::active::ManifestPublication,
        store: &Store,
        vfs: &dyn Vfs,
        directory: &Path,
        policy: DurabilityPolicy,
    ) -> Result<(PublishedSnapshot, Vec<PathBuf>), StoreError> {
        self.manifest.epochs = store.epoch_registry(&self.manifest.epochs);
        publication
            .commit_manifest(vfs, directory, &self.manifest, policy)
            .map_err(StoreError::Manifest)?;
        Ok((self.snapshot, self.replaced_paths))
    }
}

pub(crate) fn sealed_document_matches(
    snapshot: &PublishedSnapshot,
    ids: &[DocId],
) -> Result<Vec<SealedDocumentMatch>, StoreError> {
    sealed_document_matches_in(snapshot.segments(), ids)
}

fn all_sealed_document_matches(
    snapshot: &PublishedSnapshot,
    ids: &[DocId],
) -> Result<Vec<SealedDocumentMatch>, StoreError> {
    sealed_document_matches_in(snapshot.all_segments(), ids)
}

fn sealed_document_matches_in(
    segments: &[SegmentReader],
    ids: &[DocId],
) -> Result<Vec<SealedDocumentMatch>, StoreError> {
    // Each segment's identity-ordered index answers an id by binary search,
    // so a write costs O(ids x segments x log rows), not a scan of every
    // sealed row. Matches keep the (segment, ascending row) order of a scan.
    let requested = ids.iter().copied().collect::<HashSet<_>>();
    let mut matches = Vec::new();
    let mut rows = Vec::new();
    for (segment_index, segment) in segments.iter().enumerate() {
        rows.clear();
        for doc_id in &requested {
            rows.extend(segment.query_rows_with_doc_id(*doc_id)?);
        }
        rows.sort_unstable();
        for row in rows.iter().map(|row| *row as usize) {
            if let Some(version) = segment.document_version(row).map_err(StoreError::Segment)? {
                matches.push(SealedDocumentMatch {
                    segment_index,
                    row,
                    version,
                });
            }
        }
    }
    Ok(matches)
}

#[allow(clippy::too_many_arguments)]
fn prepare_one_segment_replacement(
    vfs: &dyn Vfs,
    directory: &Path,
    snapshot: &PublishedSnapshot,
    manifest: &mut Manifest,
    segment_index: usize,
    rows: &mut Vec<usize>,
    nonce: u64,
    generation: u64,
    policy: DurabilityPolicy,
    replacement_ids: &mut Vec<SegmentId>,
) -> Result<Option<(SegmentId, PathBuf)>, StoreError> {
    if rows.is_empty() {
        return Ok(None);
    }
    let reader = snapshot
        .segments()
        .get(segment_index)
        .ok_or(StoreError::ActiveRowOverflow)?;
    let alive = reader.alive().map_err(StoreError::Segment)?;
    rows.retain(|row| {
        u32::try_from(*row)
            .ok()
            .is_some_and(|row| alive.is_alive(row))
    });
    if rows.is_empty() {
        return Ok(None);
    }
    let original = manifest
        .segments
        .iter()
        .find(|segment| segment.id == reader.meta().id)
        .cloned()
        .ok_or_else(|| {
            StoreError::Segment(crate::segment::SegmentError::Geometry(format!(
                "manifest lost sealed mutation segment {}",
                reader.meta().id
            )))
        })?;
    let replacement_id = replacement_segment_id(original.id, nonce, generation);
    replacement_ids.push(replacement_id);
    let all_rows = (0..reader.meta().row_count as usize).collect::<Vec<_>>();
    let mut replacement = rewrite_segment(
        vfs,
        directory,
        reader,
        &all_rows,
        rows,
        replacement_id,
        policy,
    )?;
    replacement.epoch_id = original.epoch_id;
    let target = manifest
        .segments
        .iter_mut()
        .find(|segment| segment.id == original.id)
        .ok_or_else(|| {
            StoreError::Segment(crate::segment::SegmentError::Geometry(format!(
                "manifest lost sealed mutation segment {}",
                original.id
            )))
        })?;
    *target = replacement;
    Ok(Some((
        replacement_id,
        directory.join(original.id.file_name()),
    )))
}

#[allow(clippy::too_many_arguments)]
pub(crate) fn prepare_sealed_tombstones(
    vfs: &dyn Vfs,
    directory: &Path,
    snapshot: &PublishedSnapshot,
    matches: &[SealedDocumentMatch],
    ids: &[DocId],
    durable_end: u64,
    generation: u64,
    nonce: u64,
    #[cfg(feature = "graph-cypher")] generation_absorbed_through: Option<u64>,
    policy: DurabilityPolicy,
    accounting: &Arc<Accounting>,
) -> Result<Option<PreparedSealedTombstones>, StoreError> {
    if matches.is_empty() || ids.is_empty() {
        return Ok(None);
    }
    let requested = ids.iter().copied().collect::<HashSet<_>>();
    let mut rows_by_segment = vec![Vec::<usize>::new(); snapshot.segments().len()];
    let mut seen = HashSet::new();
    for matched in matches {
        if requested.contains(&matched.version.doc_id()) {
            let rows = rows_by_segment
                .get_mut(matched.segment_index)
                .ok_or(StoreError::ActiveRowOverflow)?;
            super::lookup::work(1);
            if seen.insert((matched.segment_index, matched.row)) {
                rows.push(matched.row);
            }
        }
    }
    let manifest_path = directory.join(MANIFEST_FILE);
    let mut manifest =
        load_manifest(vfs, &manifest_path, durable_end).map_err(StoreError::Manifest)?;
    let mut replaced_paths = Vec::new();
    let mut replacement_ids = Vec::new();
    let prepared = (|| {
        for (segment_index, rows) in rows_by_segment.iter_mut().enumerate() {
            if let Some((_replacement_id, replaced_path)) = prepare_one_segment_replacement(
                vfs,
                directory,
                snapshot,
                &mut manifest,
                segment_index,
                rows,
                nonce,
                generation,
                policy,
                &mut replacement_ids,
            )? {
                replaced_paths.push(replaced_path);
            }
        }
        if replacement_ids.is_empty() {
            return Ok(None);
        }
        manifest.generation = generation;
        #[cfg(feature = "graph-cypher")]
        if let Some(graph) = &mut manifest.graph {
            // The pending delete's WAL batch is counted in this generation,
            // but neither replay watermark can move past unsealed documents.
            if let Some(boundary) = generation_absorbed_through {
                graph.generation_absorbed_through = Some(boundary);
            }
        }
        let remapped = PublishedSnapshot::from_manifest(vfs, directory, &manifest, accounting)?;
        Ok(Some(PreparedSealedTombstones {
            manifest,
            snapshot: remapped,
            replaced_paths,
            replacement_ids: replacement_ids.clone(),
        }))
    })();
    if prepared.is_err() {
        cleanup_replacement_segments(vfs, directory, &replacement_ids, policy)?;
    }
    prepared
}

/// Caller handle for one scheduled physical-purge guarantee.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PurgeToken {
    id: u64,
    generation: u64,
    unknown_ids: Vec<DocId>,
    no_op: bool,
}

impl PurgeToken {
    /// Returns the stable token identifier.
    #[must_use]
    pub const fn id(&self) -> u64 {
        self.id
    }

    /// Returns the generation observed when the token was issued.
    #[must_use]
    pub const fn generation(&self) -> u64 {
        self.generation
    }

    /// Returns requested identifiers absent from active and sealed state.
    #[must_use]
    pub fn unknown_ids(&self) -> &[DocId] {
        &self.unknown_ids
    }

    /// Returns whether this token represents no physical work.
    #[must_use]
    pub const fn is_no_op(&self) -> bool {
        self.no_op
    }
}

/// Completed physical-purge accounting.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PurgeReport {
    generation: u64,
    segments_rewritten: usize,
    wal_rewritten: bool,
    unknown_ids: Vec<DocId>,
}

impl PurgeReport {
    /// Returns the generation at which the physical guarantee holds.
    #[must_use]
    pub const fn generation(&self) -> u64 {
        self.generation
    }

    /// Returns the number of sealed segments physically rewritten.
    #[must_use]
    pub const fn segments_rewritten(&self) -> usize {
        self.segments_rewritten
    }

    /// Returns whether the on-disk WAL was atomically replaced.
    #[must_use]
    pub const fn wal_rewritten(&self) -> bool {
        self.wal_rewritten
    }

    /// Returns requested identifiers absent from active and sealed state.
    #[must_use]
    pub fn unknown_ids(&self) -> &[DocId] {
        &self.unknown_ids
    }

    /// Returns whether no artifact needed mutation.
    #[must_use]
    pub const fn is_no_op(&self) -> bool {
        self.segments_rewritten == 0 && !self.wal_rewritten
    }
}

/// Typed physical-purge rejection.
#[derive(Debug)]
pub enum PurgeError {
    /// Store lifecycle, format, filesystem, or synchronization failure.
    Store(StoreError),
    /// A graph delete rule refused removal of a document node.
    #[cfg(feature = "graph-cypher")]
    Delete(super::IngestError),
    /// Rewriting one immutable segment would exceed available temporary disk.
    InsufficientTempSpace {
        /// Exact immutable segment bytes that would be duplicated temporarily.
        segment_bytes: u64,
        /// Available filesystem bytes observed before any purge write.
        available_bytes: u64,
        /// Integer floor of the strict 120-percent threshold.
        required_bytes: u64,
    },
    /// One store already owns a scheduled purge that has not resolved.
    PurgeInProgress,
    /// A token was not issued by this store's current pending operation.
    UnknownToken {
        /// Rejected caller token identifier.
        token_id: u64,
    },
    /// The durable purge-intent frame failed checksum or header validation.
    IntentFormat(FormatError),
    /// The durable purge-intent payload violated its fixed-width contract.
    IntentDecode(String),
    /// Re-encoding the surviving active state violated the WAL payload contract.
    WalPayload(wal_payload::PayloadError),
    /// Rewriting the WAL would discard an acknowledged, unabsorbed mutation.
    WalRewriteWouldDropAcked {
        /// First retained sequence not covered by the surviving active state.
        seq: LogSeq,
    },
}

impl std::fmt::Display for PurgeError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Store(error) => error.fmt(formatter),
            #[cfg(feature = "graph-cypher")]
            Self::Delete(error) => error.fmt(formatter),
            Self::InsufficientTempSpace {
                segment_bytes,
                available_bytes,
                required_bytes,
            } => write!(
                formatter,
                "purge segment of {segment_bytes} bytes needs more than {required_bytes} temporary bytes, only {available_bytes} available"
            ),
            Self::PurgeInProgress => formatter.write_str("a physical purge is already pending"),
            Self::UnknownToken { token_id } => {
                write!(formatter, "purge token {token_id} is unknown")
            }
            Self::IntentFormat(error) => write!(formatter, "purge intent: {error}"),
            Self::IntentDecode(detail) => write!(formatter, "purge intent decode failed: {detail}"),
            Self::WalPayload(error) => write!(formatter, "purge WAL payload: {error}"),
            Self::WalRewriteWouldDropAcked { seq } => write!(
                formatter,
                "purge WAL rewrite would drop acknowledged WAL sequence {}",
                seq.get()
            ),
        }
    }
}

impl std::error::Error for PurgeError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Store(error) => Some(error),
            #[cfg(feature = "graph-cypher")]
            Self::Delete(error) => Some(error),
            Self::IntentFormat(error) => Some(error),
            Self::WalPayload(error) => Some(error),
            Self::InsufficientTempSpace { .. }
            | Self::PurgeInProgress
            | Self::UnknownToken { .. }
            | Self::IntentDecode(_)
            | Self::WalRewriteWouldDropAcked { .. } => None,
        }
    }
}

impl From<StoreError> for PurgeError {
    fn from(error: StoreError) -> Self {
        Self::Store(error)
    }
}

pub(crate) struct PurgeIntent {
    token_id: u64,
    ids: Vec<DocId>,
}

fn surviving_sealed_matches(
    snapshot: &PublishedSnapshot,
    demands: &[SealedTombstoneDemand],
) -> Result<Vec<SealedDocumentMatch>, StoreError> {
    let requested = demands
        .iter()
        .map(|demand| demand.doc_id())
        .collect::<Vec<_>>();
    let sealed = sealed_document_matches(snapshot, &requested)?;
    let mut surviving = Vec::new();
    for matched in sealed {
        if !demands.iter().any(|demand| demand.matches(matched.version)) {
            continue;
        }
        let segment = snapshot
            .segments()
            .get(matched.segment_index)
            .ok_or(StoreError::ActiveRowOverflow)?;
        let row = u32::try_from(matched.row).map_err(|_| StoreError::ActiveRowOverflow)?;
        if segment.alive().map_err(StoreError::Segment)?.is_alive(row) {
            surviving.push(matched);
        }
    }
    Ok(surviving)
}

fn partition_known_ids(
    requested: Vec<DocId>,
    active: &super::ActiveSegment,
    sealed: &[SealedDocumentMatch],
) -> (Vec<DocId>, Vec<DocId>) {
    let mut known = Vec::new();
    let mut unknown = Vec::new();
    let mut present = super::lookup::set(active.doc_ids());
    for matched in sealed {
        super::lookup::work(1);
        present.insert(matched.version.doc_id());
    }
    for id in requested {
        let in_active = super::lookup::contains(&present, &id);
        if in_active {
            known.push(id);
        } else {
            unknown.push(id);
        }
    }
    (known, unknown)
}

fn enforce_temp_space_for_targets(
    snapshot: &PublishedSnapshot,
    sealed: &[SealedDocumentMatch],
    known: &[DocId],
    available: u64,
) -> Result<(), PurgeError> {
    let known = super::lookup::set(known);
    let targets: HashSet<_> = sealed
        .iter()
        .filter(|matched| super::lookup::contains(&known, &matched.version.doc_id()))
        .map(|matched| matched.segment_index)
        .collect();
    for (segment_index, segment) in snapshot.all_segments().iter().enumerate() {
        if super::lookup::contains(&targets, &segment_index) {
            enforce_temp_space(segment.meta().file_size, available)?;
        }
    }
    Ok(())
}

impl Store {
    pub(crate) fn recover_sealed_tombstones(
        &self,
        demands: &[SealedTombstoneDemand],
    ) -> Result<(), StoreError> {
        if demands.is_empty() {
            return Ok(());
        }
        let snapshot = self
            .snapshot
            .read()
            .map_err(|_| StoreError::Synchronization {
                component: "published snapshot",
            })?
            .as_ref()
            .cloned()
            .ok_or(StoreError::Closed)?;
        let surviving = surviving_sealed_matches(&snapshot, demands)?;
        if surviving.is_empty() {
            return Ok(());
        }
        let state = self
            .state
            .lock()
            .map_err(|_| StoreError::Synchronization { component: "state" })?;
        match *state {
            StoreState::Open => {}
            StoreState::Closing => return Err(StoreError::Closing),
            StoreState::Closed => return Err(StoreError::Closed),
        }
        let writer_lock = self
            .writer_lock
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "writer lock",
            })?;
        if writer_lock.is_none() {
            return Err(StoreError::SealedTombstoneRecoveryRequired);
        }
        let mut wal = self
            .wal_writer
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "WAL writer",
            })?;
        let writer = wal
            .as_mut()
            .ok_or(StoreError::SealedTombstoneRecoveryRequired)?;
        let mut active = self
            .active
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "active segment",
            })?;
        let current = active.as_ref().ok_or(StoreError::Closed)?;
        // Graph-free recovery retains main's repair publication and segment ID.
        // Graph stores publish the already replayed delete generation exactly.
        #[cfg(feature = "graph-cypher")]
        let graph_enabled = snapshot.graph_manifest.is_some();
        #[cfg(not(feature = "graph-cypher"))]
        let graph_enabled = false;
        let generation = if graph_enabled {
            current.generation
        } else {
            current
                .generation
                .checked_add(1)
                .ok_or(StoreError::GenerationOverflow)?
        };
        let active_segment = Arc::clone(&current.segment);
        let mut ids = surviving
            .iter()
            .map(|matched| matched.version.doc_id())
            .collect::<Vec<_>>();
        ids.sort_unstable();
        ids.dedup();
        let prepared = prepare_sealed_tombstones(
            self.vfs.as_ref(),
            &self.directory,
            &snapshot,
            &surviving,
            &ids,
            writer.durable_end(),
            generation,
            writer.durable_end().saturating_add(1),
            #[cfg(feature = "graph-cypher")]
            Some(writer.durable_end()),
            self.durability_policy,
            &self.accounting,
        )?;
        let Some(prepared) = prepared else {
            return Ok(());
        };
        let mut publication = writer.manifest_publication()?;
        let (remapped, replaced_paths) = prepared.commit(
            &mut publication,
            self,
            self.vfs.as_ref(),
            &self.directory,
            self.durability_policy,
        )?;
        let mut published = self
            .snapshot
            .write()
            .map_err(|_| StoreError::Synchronization {
                component: "published snapshot",
            })?;
        let previous = published.replace(Arc::new(remapped));
        *active = Some(ActiveState {
            generation,
            segment: active_segment,
        });
        publication.complete();
        drop(published);
        drop(previous);
        unlink_replaced_segments(
            self,
            self.vfs.as_ref(),
            &self.directory,
            &replaced_paths,
            self.durability_policy,
        );
        drop(active);
        drop(wal);
        drop(writer_lock);
        drop(state);
        Ok(())
    }

    /// Schedules a physical purge and returns before artifact rewriting begins.
    pub fn purge(&self, ids: &[DocId]) -> Result<PurgeToken, PurgeError> {
        let available = self.available_purge_bytes()?;
        self.purge_inner(ids, available, self.vfs.as_ref())
    }

    /// Test seam that supplies a deterministic free-space observation.
    #[doc(hidden)]
    pub fn purge_with_available_space(
        &self,
        ids: &[DocId],
        available_bytes: u64,
    ) -> Result<PurgeToken, PurgeError> {
        self.purge_inner(ids, available_bytes, self.vfs.as_ref())
    }

    fn purge_inner(
        &self,
        ids: &[DocId],
        available_bytes: u64,
        vfs: &dyn Vfs,
    ) -> Result<PurgeToken, PurgeError> {
        #[cfg(feature = "graph-cypher")]
        {
            self.with_native_document_recovery(
                || self.purge_attempt(ids, available_bytes, vfs),
                |error| matches!(error, PurgeError::Delete(error) if error.is_native_reclaim_debt()),
                PurgeError::Delete,
            )
        }
        #[cfg(not(feature = "graph-cypher"))]
        self.purge_attempt(ids, available_bytes, vfs)
    }

    fn purge_attempt(
        &self,
        ids: &[DocId],
        available_bytes: u64,
        vfs: &dyn Vfs,
    ) -> Result<PurgeToken, PurgeError> {
        let _maintenance = self
            .maintenance
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "physical purge",
            })?;
        #[cfg(feature = "graph-cypher")]
        let mut native_writer = self.native_document_writer()?;
        #[cfg(feature = "graph-cypher")]
        if let Some(writer) = native_writer.as_mut().and_then(|slot| slot.as_mut())
            && self
                .documents_have_native_nodes(ids)
                .map_err(PurgeError::Delete)?
        {
            // Validate only; schedule_purge_locked writes the durable intent.
            self.prepare_purge_intent(ids, available_bytes, vfs)?;
            self.delete_native_documents_locked(&super::DeleteBatch::new(ids.to_vec()), writer)
                .map_err(PurgeError::Delete)?;
        }
        let state = self
            .state
            .lock()
            .map_err(|_| StoreError::Synchronization { component: "state" })?;
        match *state {
            StoreState::Open => {}
            StoreState::Closing => return Err(StoreError::Closing.into()),
            StoreState::Closed => return Err(StoreError::Closed.into()),
        }
        let writer_lock = self
            .writer_lock
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "writer lock",
            })?;
        if writer_lock.is_none() {
            return Err(StoreError::ReadOnly.into());
        }
        let wal = self
            .wal_writer
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "WAL writer",
            })?;
        let token = self.schedule_purge_locked(
            ids,
            available_bytes,
            vfs,
            wal.as_ref().ok_or(StoreError::ReadOnly)?,
        )?;
        drop(writer_lock);
        drop(state);
        Ok(token)
    }

    /// Current free bytes on the store's volume, as `purge` observes them.
    pub(crate) fn available_purge_bytes(&self) -> Result<u64, PurgeError> {
        available_disk_bytes(&self.directory).map_err(|source| {
            PurgeError::Store(StoreError::Io {
                path: self.directory.clone(),
                source,
            })
        })
    }

    /// Durably records a purge intent for `ids`. The caller holds the
    /// maintenance, state and writer locks and has checked the store is open
    /// and writable; this takes the active and snapshot locks itself.
    pub(crate) fn schedule_purge_locked(
        &self,
        ids: &[DocId],
        available_bytes: u64,
        vfs: &dyn Vfs,
        writer: &super::StoreWal,
    ) -> Result<PurgeToken, PurgeError> {
        let publication = writer.manifest_publication()?;
        let (token, intent) = self.prepare_purge_intent(ids, available_bytes, vfs)?;
        if token.no_op {
            return Ok(token);
        }
        let token_id = intent.token_id;
        let known = intent.ids;
        #[cfg(any(test, feature = "test-seams"))]
        let crash_target_ids = known.iter().map(|id| id.get()).collect::<Vec<_>>();
        let publication = publication.arm();
        write_intent(
            vfs,
            &self.directory,
            &PurgeIntent {
                token_id,
                ids: known,
            },
            self.durability_policy,
        )?;
        #[cfg(any(test, feature = "test-seams"))]
        if let Some(controller) = self.ingest_retention_fault_controller.as_ref() {
            let plan = controller.take_purge_crash_boundary_plan().map_err(|_| {
                StoreError::Synchronization {
                    component: "ingest-retention controller",
                }
            })?;
            if let Some((invocation_id, receipt_sink)) = plan {
                let receipt = super::IngestRetentionFaultReceiptV1::purge_crash_boundary(
                    invocation_id,
                    crash_target_ids,
                    token_id,
                );
                receipt
                    .write_purge_crash_test_evidence(vfs, &receipt_sink)
                    .map_err(|source| StoreError::Io {
                        path: receipt_sink,
                        source,
                    })?;
                std::process::abort();
            }
        }
        publication.complete();
        Ok(token)
    }

    /// Performs every definite purge refusal before a document/node deletion.
    pub(crate) fn prepare_purge_intent(
        &self,
        ids: &[DocId],
        available_bytes: u64,
        vfs: &dyn Vfs,
    ) -> Result<(PurgeToken, PurgeIntent), PurgeError> {
        self.require_no_snapshot_views()?;
        let intent_path = self.directory.join(PURGE_INTENT_FILE);
        match vfs.open(&intent_path) {
            Ok(_) => return Err(PurgeError::PurgeInProgress),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(source) => {
                return Err(StoreError::Io {
                    path: intent_path,
                    source,
                }
                .into());
            }
        }
        let mut requested = ids.to_vec();
        requested.sort_unstable();
        requested.dedup();
        let active = self
            .active
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "active segment",
            })?;
        let active_state = active.as_ref().ok_or(StoreError::Closed)?;
        let snapshot = self
            .snapshot
            .read()
            .map_err(|_| StoreError::Synchronization {
                component: "published snapshot",
            })?
            .as_ref()
            .cloned()
            .ok_or(StoreError::Closed)?;
        let sealed = all_sealed_document_matches(&snapshot, &requested)?;
        let (known, unknown) = partition_known_ids(requested, &active_state.segment, &sealed);
        enforce_temp_space_for_targets(&snapshot, &sealed, &known, available_bytes)?;
        let token_id = purge_token_id(active_state.generation, &known);
        let no_op = known.is_empty();
        let token = PurgeToken {
            id: token_id,
            generation: active_state.generation,
            unknown_ids: unknown,
            no_op,
        };
        Ok((
            token,
            PurgeIntent {
                token_id,
                ids: known,
            },
        ))
    }

    /// Removes the intent written by [`Self::schedule_purge_locked`] when the
    /// mutation it was scheduled for did not commit. Same locks held.
    pub(crate) fn abandon_scheduled_purge_locked(
        &self,
        vfs: &dyn Vfs,
        writer: &super::StoreWal,
    ) -> Result<(), PurgeError> {
        let publication = writer.manifest_publication()?.arm();
        remove_intent(vfs, &self.directory, self.durability_policy)?;
        publication.complete();
        Ok(())
    }

    #[allow(clippy::too_many_arguments)]
    fn replace_one_purged_segment(
        &self,
        vfs: &dyn Vfs,
        manifest: &mut Manifest,
        active_state: &mut ActiveState,
        writer: &super::StoreWal,
        original: &SegmentMeta,
        ids: &super::lookup::LookupSet<DocId>,
        token_id: u64,
        policy: DurabilityPolicy,
    ) -> Result<bool, PurgeError> {
        let _durable_end = writer.durable_end();
        let path = self.directory.join(original.id.file_name());
        let reader = SegmentReader::open(vfs, &path, original.id).map_err(StoreError::Segment)?;
        let survivors = survivor_rows(&reader, ids)?;
        if survivors.len() == original.row_count as usize {
            return Ok(false);
        }
        let generation = active_state
            .generation
            .max(manifest.generation)
            .checked_add(1)
            .ok_or(StoreError::GenerationOverflow)?;
        let replacement_id = replacement_segment_id(original.id, token_id, generation);
        let mut replacement = rewrite_segment(
            vfs,
            &self.directory,
            &reader,
            &survivors,
            &[],
            replacement_id,
            policy,
        )?;
        replacement.epoch_id = original.epoch_id;
        drop(reader);
        let position = manifest
            .segments
            .iter()
            .position(|segment| segment.id == original.id)
            .ok_or_else(|| {
                PurgeError::IntentDecode(format!("manifest lost purge segment {}", original.id))
            })?;
        let target = manifest.segments.get_mut(position).ok_or_else(|| {
            PurgeError::IntentDecode("purge segment position is invalid".to_owned())
        })?;
        *target = replacement;
        manifest.generation = generation;
        #[cfg(feature = "graph-cypher")]
        manifest
            .record_generation_bump(_durable_end)
            .map_err(StoreError::Manifest)?;
        manifest.epochs = self.epoch_registry(&manifest.epochs);
        let remapped =
            PublishedSnapshot::from_manifest(vfs, &self.directory, manifest, &self.accounting)?;
        let mut publication = writer.manifest_publication()?;
        publication
            .commit_manifest(vfs, &self.directory, manifest, policy)
            .map_err(StoreError::Manifest)?;
        let mut published = self
            .snapshot
            .write()
            .map_err(|_| StoreError::Synchronization {
                component: "published snapshot",
            })?;
        let previous = published.replace(Arc::new(remapped));
        active_state.generation = generation;
        publication.complete();
        drop(published);
        drop(previous);
        if let Err(source) = vfs.delete(&path) {
            #[cfg(any(test, feature = "test-seams"))]
            if source.kind() == std::io::ErrorKind::Other
                && let Some(controller) = self.ingest_retention_fault_controller.as_ref()
            {
                let invocation_id = controller
                    .purge_unlink_error_plan(*original.id.as_bytes())
                    .map_err(|_| StoreError::Synchronization {
                        component: "ingest-retention controller",
                    })?;
                if let Some(invocation_id) = invocation_id {
                    let intent_present = vfs.open(&self.directory.join(PURGE_INTENT_FILE)).is_ok();
                    let old_path_linked = vfs.open(&path).is_ok();
                    controller
                        .push_receipt(super::IngestRetentionFaultReceiptV1::purge_unlink_error(
                            invocation_id,
                            *original.id.as_bytes(),
                            *replacement_id.as_bytes(),
                            original.id.file_name(),
                            true,
                            intent_present,
                            old_path_linked,
                        ))
                        .map_err(|_| StoreError::Synchronization {
                            component: "ingest-retention controller",
                        })?;
                }
            }
            return Err(StoreError::Io {
                path: path.clone(),
                source,
            }
            .into());
        }
        sync_directory(vfs, &self.directory, policy)?;
        Ok(true)
    }

    fn commit_bumped_manifest(
        &self,
        vfs: &dyn Vfs,
        manifest: &mut Manifest,
        generation: u64,
        writer: &super::StoreWal,
        _generation_bump: bool,
        policy: DurabilityPolicy,
    ) -> Result<(), PurgeError> {
        // The caller already advanced active generation or rewrote the WAL.
        // Even a pre-rename failure requires recovery before another write.
        let mut publication = writer.manifest_publication()?.arm();
        let _durable_end = writer.durable_end();
        manifest.generation = generation;
        #[cfg(feature = "graph-cypher")]
        if _generation_bump {
            manifest
                .record_generation_bump(_durable_end)
                .map_err(StoreError::Manifest)?;
        } else if let Some(graph) = &mut manifest.graph {
            // WAL replay, not a fresh mutation, supplied this generation.
            graph.generation_absorbed_through = Some(_durable_end);
        }
        manifest.epochs = self.epoch_registry(&manifest.epochs);
        let remapped =
            PublishedSnapshot::from_manifest(vfs, &self.directory, manifest, &self.accounting)?;
        publication
            .commit_manifest(vfs, &self.directory, manifest, policy)
            .map_err(StoreError::Manifest)?;
        let mut published = self
            .snapshot
            .write()
            .map_err(|_| StoreError::Synchronization {
                component: "published snapshot",
            })?;
        let previous = published.replace(Arc::new(remapped));
        publication.complete();
        drop(published);
        drop(previous);
        Ok(())
    }

    #[allow(clippy::too_many_arguments)]
    fn rewrite_wal_for_purge<'a>(
        &self,
        vfs: &dyn Vfs,
        writer: &'a mut super::StoreWal,
        manifest: &Manifest,
        intent: &PurgeIntent,
        next_active: &mut super::ActiveSegment,
        records: &[(u16, Vec<u8>)],
        tombstoned: &[bool],
        policy: DurabilityPolicy,
    ) -> Result<super::active::WalReplacement<'a>, PurgeError> {
        let retained_first_seq = LogSeq::new(
            manifest
                .log_seq
                .checked_add(1)
                .ok_or(StoreError::GenerationOverflow)?,
        );
        let durable_end = writer.durable_end();
        if durable_end >= retained_first_seq.get() {
            let snapshot = self
                .snapshot
                .read()
                .map_err(|_| StoreError::Synchronization {
                    component: "published snapshot",
                })?;
            let snapshot = snapshot.as_ref().ok_or(StoreError::Closed)?;
            ensure_wal_rewrite_covers_retained(
                vfs,
                &self.directory,
                manifest,
                &intent.ids,
                next_active,
                snapshot.all_segments(),
                self.private_preparation.as_ref(),
            )?;
        }
        let rewrite_first_seq = LogSeq::new(durable_end.checked_add(1).ok_or(
            StoreError::WalWrite(crate::wal::WalWriteError::SequenceExhausted),
        )?);
        #[cfg(feature = "graph-cypher")]
        let framed;
        #[cfg(feature = "graph-cypher")]
        let records = if manifest.graph.is_some() && records.len() > 1 {
            // Replay counts batches, so frame the replacement image as one batch
            // covered by the purge cutoff, including any trailing tombstones.
            let count = u32::try_from(records.len())
                .map_err(|_| PurgeError::WalPayload(wal_payload::PayloadError::LengthOverflow))?;
            framed = records
                .iter()
                .zip(0_u32..)
                .map(|((op, payload), index)| {
                    wal_payload::encode_mixed_batch_member(index, count, *op, payload)
                        .map(|payload| (wal_payload::MIXED_BATCH_MEMBER_V1, payload))
                        .map_err(PurgeError::WalPayload)
                })
                .collect::<Result<Vec<_>, _>>()?;
            framed.as_slice()
        } else {
            records
        };
        let publication =
            writer.rewrite(vfs, &self.directory, policy, rewrite_first_seq, records)?;
        assign_rewritten_sequences(next_active, rewrite_first_seq, tombstoned)?;
        Ok(publication)
    }

    #[cfg(feature = "graph-cypher")]
    pub(crate) fn graph_checkpoint_through(
        &self,
        vfs: &dyn Vfs,
    ) -> Result<Option<u64>, StoreError> {
        let snapshot = self
            .snapshot
            .read()
            .map_err(|_| StoreError::Synchronization {
                component: "published snapshot",
            })?;
        let snapshot = snapshot.as_ref().ok_or(StoreError::Closed)?;
        if !snapshot.graph_enabled {
            return Ok(None);
        }
        let reader = crate::wal::WalReader::open(vfs, &self.directory.join("wal.ze"))
            .map_err(StoreError::Wal)?;
        let clean = reader.into_clean().map_err(StoreError::WalRecovery)?;
        let watermark = snapshot
            .absorbed_through()
            .min(snapshot.graph_absorbed_through);
        let decisions = crate::lifecycle::namespace_batch::transaction_decisions(
            vfs,
            &self.directory,
            clean.records(),
            watermark,
            self.private_preparation.as_ref(),
        )?;
        let mut through = None;
        for (seq, _, mutation) in super::atomic_batch::committed_mutations_with_decisions(
            clean.records(),
            watermark,
            |binding| decisions.get(&binding.transaction).copied(),
        )? {
            if seq.get() > snapshot.graph_absorbed_through
                && matches!(mutation, wal_payload::MutationPayload::GraphCommit(_))
            {
                through = Some(seq.get());
            }
        }
        Ok(through)
    }

    /// Resolves only after rewritten artifacts are committed and old paths are unlinked.
    pub fn await_physical_purge(&self, token: PurgeToken) -> Result<PurgeReport, PurgeError> {
        self.await_physical_purge_on_vfs(token, self.vfs.as_ref())
    }

    /// Test seam that runs the purge protocol through an observable filesystem.
    #[doc(hidden)]
    pub fn await_physical_purge_on_vfs(
        &self,
        token: PurgeToken,
        vfs: &dyn Vfs,
    ) -> Result<PurgeReport, PurgeError> {
        if token.no_op {
            return Ok(PurgeReport {
                generation: token.generation,
                segments_rewritten: 0,
                wal_rewritten: false,
                unknown_ids: token.unknown_ids,
            });
        }
        let _maintenance = self
            .maintenance
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "physical purge",
            })?;
        let state = self
            .state
            .lock()
            .map_err(|_| StoreError::Synchronization { component: "state" })?;
        match *state {
            StoreState::Open => {}
            StoreState::Closing => return Err(StoreError::Closing.into()),
            StoreState::Closed => return Err(StoreError::Closed.into()),
        }
        let writer_lock = self
            .writer_lock
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "writer lock",
            })?;
        if writer_lock.is_none() {
            return Err(StoreError::ReadOnly.into());
        }
        self.require_no_snapshot_views()?;
        let mut wal = self
            .wal_writer
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "WAL writer",
            })?;
        let writer = wal.as_mut().ok_or(StoreError::ReadOnly)?;
        let report = self.complete_physical_purge_locked(vfs, writer, token)?;
        drop(wal);
        drop(writer_lock);
        drop(state);
        Ok(report)
    }

    /// Runs the scheduled purge protocol to completion. The caller holds the
    /// maintenance, state, writer and WAL locks; this takes the active lock.
    pub(crate) fn complete_physical_purge_locked(
        &self,
        vfs: &dyn Vfs,
        writer: &mut super::StoreWal,
        token: PurgeToken,
    ) -> Result<PurgeReport, PurgeError> {
        let intent = read_intent(vfs, &self.directory)?;
        let policy = if crate::lifecycle::namespace_batch::owns_purge_obligation(
            vfs,
            &self.directory,
            intent.token_id,
            self.private_preparation.as_ref(),
        )? {
            DurabilityPolicy::new(
                crate::lifecycle::durability::DurabilityMode::Durable,
                crate::lifecycle::durability::CommitTier::Durable,
            )
            .map_err(StoreError::Durability)?
        } else {
            self.durability_policy
        };
        if intent.token_id != token.id {
            return Err(PurgeError::UnknownToken { token_id: token.id });
        }
        #[cfg(feature = "graph-cypher")]
        let graph_checkpoint = self.graph_checkpoint_through(vfs)?;
        let mut active = self
            .active
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "active segment",
            })?;
        let active_state = active.as_mut().ok_or(StoreError::Closed)?;
        #[cfg(feature = "graph-cypher")]
        if let Some(through) = graph_checkpoint {
            self.checkpoint_native_graph_locked(writer, active_state, through, vfs)
                .map_err(|error| StoreError::PurgeRecovery {
                    detail: format!("purge graph checkpoint: {error}"),
                })?;
        }
        let manifest_path = self.directory.join(MANIFEST_FILE);
        let mut manifest = match vfs.open(&manifest_path) {
            Ok(_) => load_manifest(vfs, &manifest_path, writer.durable_end())
                .map_err(StoreError::Manifest)?,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Manifest {
                #[cfg(feature = "graph-cypher")]
                graph: None,
                generation: active_state.generation,
                log_seq: 0,
                segments: Vec::new(),
                epochs: Vec::new(),
                epoch_alias: None,
                schema: crate::meta::Schema::new(Vec::new()).map_err(|source| {
                    StoreError::Segment(crate::segment::SegmentError::Columns(source.to_string()))
                })?,
            },
            Err(source) => {
                return Err(StoreError::Io {
                    path: manifest_path,
                    source,
                }
                .into());
            }
        };
        let mut rewritten = 0_usize;
        let purge_ids = super::lookup::set(&intent.ids);
        let original_segments = manifest.segments.clone();
        for original in original_segments {
            if self.replace_one_purged_segment(
                vfs,
                &mut manifest,
                active_state,
                writer,
                &original,
                &purge_ids,
                token.id,
                policy,
            )? {
                rewritten = rewritten.saturating_add(1);
            }
        }
        sweep_purge_orphans(vfs, &self.directory, &manifest, policy)?;

        let active_has_target = active_state
            .segment
            .doc_ids()
            .iter()
            .any(|id| super::lookup::contains(&purge_ids, id));
        let mut next_active = if active_has_target {
            let (purged, removed) =
                active_state
                    .segment
                    .purge(&intent.ids, &self.accounting, &self.tokenizer)?;
            if removed == 0 {
                return Err(PurgeError::IntentDecode(
                    "active purge target disappeared during admission".to_owned(),
                ));
            }
            active_state.generation = active_state
                .generation
                .checked_add(1)
                .ok_or(StoreError::GenerationOverflow)?;
            purged
        } else {
            active_state
                .segment
                .purge(&[], &self.accounting, &self.tokenizer)?
                .0
        };
        let (records, tombstoned) = active_wal_records(&next_active)?;
        if manifest.generation < active_state.generation {
            self.commit_bumped_manifest(
                vfs,
                &mut manifest,
                active_state.generation,
                writer,
                active_has_target,
                policy,
            )?;
        }
        #[cfg(feature = "graph-cypher")]
        let retired_through = writer.durable_end();
        let publication = self.rewrite_wal_for_purge(
            vfs,
            writer,
            &manifest,
            &intent,
            &mut next_active,
            &records,
            &tombstoned,
            policy,
        )?;
        #[cfg(feature = "graph-cypher")]
        let writer = &*publication.wal;
        #[cfg(feature = "graph-cypher")]
        let history_folded = if let Some(graph) = &mut manifest.graph
            && !graph.generation_bumps.is_empty()
        {
            // The replacement WAL retires every old record. No retained graph
            // envelope needs the old ordering; restore these increments once,
            // before the replacement image. Keep the v2 boundary row that
            // proves a cutoff immediately before the new WAL header.
            let count = graph
                .generation_bumps
                .iter()
                .try_fold(0_u64, |total, (_, count)| {
                    total
                        .checked_add(*count)
                        .ok_or(StoreError::GenerationOverflow)
                })?;
            let folded = (retired_through, count);
            let changed = graph.generation_bumps.as_slice() != [folded];
            graph.generation_bumps.clear();
            graph.generation_bumps.push(folded);
            changed
        } else {
            false
        };
        #[cfg(feature = "graph-cypher")]
        if history_folded
            || manifest.graph.as_ref().is_some_and(|graph| {
                graph.generation_absorbed_through != Some(writer.durable_end())
            })
        {
            // Rewritten document batches were already counted in this
            // generation. Publish their new cutoff before purge returns or
            // another graph write can be acknowledged.
            self.commit_bumped_manifest(
                vfs,
                &mut manifest,
                active_state.generation,
                writer,
                false,
                policy,
            )?;
        }
        active_state.segment = Arc::new(next_active);
        #[cfg(feature = "graph-cypher")]
        self.publish_native_documents(active_state)?;
        remove_intent(vfs, &self.directory, policy)?;
        publication.complete();
        let generation = active_state.generation;
        drop(active);
        Ok(PurgeReport {
            generation,
            segments_rewritten: rewritten,
            wal_rewritten: true,
            unknown_ids: token.unknown_ids,
        })
    }

    pub(crate) fn recover_pending_physical_purge(&self) -> Result<(), PurgeError> {
        let path = self.directory.join(PURGE_INTENT_FILE);
        match self.vfs.open(&path) {
            Ok(_) => {
                let intent = read_intent(self.vfs.as_ref(), &self.directory)?;
                // Accepted namespace deletes already have a committed logical
                // state. Ordinary purge intents still require writable recovery.
                if self
                    .writer_lock
                    .lock()
                    .map_err(|_| StoreError::Synchronization {
                        component: "writer lock",
                    })?
                    .is_none()
                {
                    return if crate::lifecycle::namespace_batch::owns_purge_obligation(
                        self.vfs.as_ref(),
                        &self.directory,
                        intent.token_id,
                        self.private_preparation.as_ref(),
                    )? {
                        Ok(())
                    } else {
                        Err(StoreError::ReadOnly.into())
                    };
                }
                let generation = self
                    .active
                    .lock()
                    .map_err(|_| StoreError::Synchronization {
                        component: "active segment",
                    })?
                    .as_ref()
                    .ok_or(StoreError::Closed)?
                    .generation;
                let token = PurgeToken {
                    id: intent.token_id,
                    generation,
                    unknown_ids: Vec::new(),
                    no_op: false,
                };
                self.await_physical_purge(token).map(|_| ())
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(source) => Err(StoreError::Io { path, source }.into()),
        }
    }
}

// Prepared delete records are the transaction-gated obligation. Adoption
// materializes the ordinary intent before recording local acceptance.
pub(crate) fn namespace_purge_intent(
    vfs: &dyn Vfs,
    directory: &Path,
    transaction: u128,
    ids: Vec<DocId>,
) -> Result<(), PurgeError> {
    if ids.is_empty() {
        return Ok(());
    }
    write_intent(
        vfs,
        directory,
        &PurgeIntent {
            token_id: xxh3_64(&transaction.to_le_bytes()),
            ids,
        },
        DurabilityPolicy::new(
            crate::lifecycle::durability::DurabilityMode::Durable,
            crate::lifecycle::durability::CommitTier::Durable,
        )
        .map_err(StoreError::Durability)?,
    )
}

impl Store {
    pub(crate) fn complete_namespace_purge_locked(
        &self,
        writer: &mut super::StoreWal,
    ) -> Result<u64, PurgeError> {
        let intent = read_intent(self.vfs.as_ref(), &self.directory)?;
        let token = PurgeToken {
            id: intent.token_id,
            generation: 0,
            unknown_ids: Vec::new(),
            no_op: false,
        };
        self.complete_physical_purge_locked(self.vfs.as_ref(), writer, token)
            .map(|report| report.generation())
    }
}

fn enforce_temp_space(segment_bytes: u64, available_bytes: u64) -> Result<(), PurgeError> {
    let required_bytes = segment_bytes
        .checked_mul(6)
        .map(|value| value / 5)
        .unwrap_or(u64::MAX);
    if available_bytes > required_bytes {
        Ok(())
    } else {
        Err(PurgeError::InsufficientTempSpace {
            segment_bytes,
            available_bytes,
            required_bytes,
        })
    }
}

fn purge_token_id(generation: u64, ids: &[DocId]) -> u64 {
    let mut bytes = Vec::with_capacity(8_usize.saturating_add(ids.len().saturating_mul(16)));
    bytes.extend_from_slice(&generation.to_le_bytes());
    for id in ids {
        bytes.extend_from_slice(&id.get().to_le_bytes());
    }
    xxh3_64(&bytes)
}

fn write_intent(
    vfs: &dyn Vfs,
    directory: &Path,
    intent: &PurgeIntent,
    policy: DurabilityPolicy,
) -> Result<(), PurgeError> {
    let count = u32::try_from(intent.ids.len())
        .map_err(|_| PurgeError::IntentDecode("purge id count exceeds u32".to_owned()))?;
    let mut payload =
        Vec::with_capacity(16_usize.saturating_add(intent.ids.len().saturating_mul(16)));
    payload.extend_from_slice(&intent.token_id.to_le_bytes());
    payload.extend_from_slice(&count.to_le_bytes());
    payload.extend_from_slice(&0_u32.to_le_bytes());
    for id in &intent.ids {
        payload.extend_from_slice(&id.get().to_le_bytes());
    }
    let bytes = encode_artifact(FormatFamily::PurgeIntent, 0, &payload);
    let temporary = directory.join(PURGE_INTENT_TEMP_FILE);
    let committed = directory.join(PURGE_INTENT_FILE);
    vfs.write(&temporary, &bytes)
        .map_err(|source| StoreError::Io {
            path: temporary.clone(),
            source,
        })?;
    if let SyncRequirement::Sync(kind) = policy.data_file_sync() {
        vfs.sync(&temporary, kind)
            .map_err(|source| StoreError::Io {
                path: temporary.clone(),
                source,
            })?;
    }
    vfs.rename(&temporary, &committed)
        .map_err(|source| StoreError::Io {
            path: committed,
            source,
        })?;
    sync_directory(vfs, directory, policy).map_err(PurgeError::from)
}

pub(crate) fn read_intent(vfs: &dyn Vfs, directory: &Path) -> Result<PurgeIntent, PurgeError> {
    let path = directory.join(PURGE_INTENT_FILE);
    let bytes = vfs.read(&path).map_err(|source| StoreError::Io {
        path: path.clone(),
        source,
    })?;
    let decoded = decode_artifact(
        &path.display().to_string(),
        FormatFamily::PurgeIntent,
        &bytes,
    )
    .map_err(PurgeError::IntentFormat)?;
    let token_id = read_u64(decoded.payload, 0)?;
    let count = read_u32(decoded.payload, 8)? as usize;
    if read_u32(decoded.payload, 12)? != 0 {
        return Err(PurgeError::IntentDecode(
            "purge intent reserved field is non-zero".to_owned(),
        ));
    }
    let expected = 16_usize
        .checked_add(count.saturating_mul(16))
        .ok_or_else(|| PurgeError::IntentDecode("purge intent length overflow".to_owned()))?;
    if decoded.payload.len() != expected {
        return Err(PurgeError::IntentDecode(format!(
            "purge intent bytes {}, expected {expected}",
            decoded.payload.len()
        )));
    }
    let mut ids = Vec::with_capacity(count);
    for index in 0..count {
        let offset = 16_usize
            .checked_add(index.saturating_mul(16))
            .ok_or_else(|| PurgeError::IntentDecode("purge id offset overflow".to_owned()))?;
        let end = offset
            .checked_add(16)
            .ok_or_else(|| PurgeError::IntentDecode("purge id end overflow".to_owned()))?;
        let value = decoded
            .payload
            .get(offset..end)
            .and_then(|bytes| bytes.try_into().ok())
            .map(u128::from_le_bytes)
            .ok_or_else(|| PurgeError::IntentDecode(format!("purge id {index} is truncated")))?;
        ids.push(DocId::new(value));
    }
    Ok(PurgeIntent { token_id, ids })
}

fn read_u32(bytes: &[u8], offset: usize) -> Result<u32, PurgeError> {
    bytes
        .get(offset..offset.saturating_add(4))
        .and_then(|value| value.try_into().ok())
        .map(u32::from_le_bytes)
        .ok_or_else(|| PurgeError::IntentDecode(format!("u32 at {offset} is truncated")))
}

fn read_u64(bytes: &[u8], offset: usize) -> Result<u64, PurgeError> {
    bytes
        .get(offset..offset.saturating_add(8))
        .and_then(|value| value.try_into().ok())
        .map(u64::from_le_bytes)
        .ok_or_else(|| PurgeError::IntentDecode(format!("u64 at {offset} is truncated")))
}

fn remove_intent(
    vfs: &dyn Vfs,
    directory: &Path,
    policy: DurabilityPolicy,
) -> Result<(), PurgeError> {
    let path = directory.join(PURGE_INTENT_FILE);
    vfs.delete(&path)
        .map_err(|source| StoreError::Io { path, source })?;
    sync_directory(vfs, directory, policy).map_err(PurgeError::from)
}

fn sync_directory(
    vfs: &dyn Vfs,
    directory: &Path,
    policy: DurabilityPolicy,
) -> Result<(), StoreError> {
    if let SyncRequirement::Sync(kind) = policy.directory_sync() {
        vfs.sync(directory, kind).map_err(|source| StoreError::Io {
            path: directory.to_path_buf(),
            source,
        })?;
    }
    Ok(())
}

pub(crate) fn unlink_replaced_segments(
    store: &Store,
    vfs: &dyn Vfs,
    directory: &Path,
    replaced_paths: &[PathBuf],
    policy: DurabilityPolicy,
) {
    if store.has_snapshot_views() {
        return;
    }
    for path in replaced_paths {
        // The replacement manifest and snapshot are already published. A
        // failed unlink leaves an unreachable orphan for open-time cleanup;
        // it cannot roll back or deny the acknowledged mutation.
        let _ = vfs.delete(path);
    }
    if !replaced_paths.is_empty() {
        let _ = sync_directory(vfs, directory, policy);
    }
}

fn cleanup_replacement_segments(
    vfs: &dyn Vfs,
    directory: &Path,
    replacement_ids: &[SegmentId],
    policy: DurabilityPolicy,
) -> Result<(), StoreError> {
    let mut deleted = false;
    for id in replacement_ids {
        for path in [
            directory.join(id.file_name()),
            directory.join(format!(".{}.tmp", id.file_name())),
        ] {
            match vfs.open(&path) {
                Ok(_) => {
                    vfs.delete(&path)
                        .map_err(|source| StoreError::Io { path, source })?;
                    deleted = true;
                }
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(source) => return Err(StoreError::Io { path, source }),
            }
        }
    }
    if deleted {
        sync_directory(vfs, directory, policy)?;
    }
    Ok(())
}

fn survivor_rows(
    reader: &SegmentReader,
    ids: &super::lookup::LookupSet<DocId>,
) -> Result<Vec<usize>, PurgeError> {
    let mut survivors = Vec::with_capacity(reader.meta().row_count as usize);
    for row in 0..reader.meta().row_count as usize {
        let version = reader.document_version(row).map_err(StoreError::Segment)?;
        if version.is_none_or(|version| !super::lookup::contains(ids, &version.doc_id())) {
            survivors.push(row);
        }
    }
    Ok(survivors)
}

pub(crate) enum OwnedFactors {
    Bit4(Vec<Bit4Factors>),
    Int8(Vec<Int8Factors>),
}

impl OwnedFactors {
    fn borrowed(&self) -> SegmentFactors<'_> {
        match self {
            Self::Bit4(values) => SegmentFactors::Bit4(values),
            Self::Int8(values) => SegmentFactors::Int8(values),
        }
    }
}

pub(crate) fn gather_survivor_codes(
    reader: &SegmentReader,
    survivors: &[usize],
    dims: usize,
) -> Result<(Vec<u8>, OwnedFactors), StoreError> {
    match reader.meta().scheme {
        4 => {
            let stride = dims.div_ceil(2);
            let source_codes = reader.bit4_codes().map_err(StoreError::Segment)?;
            let source_factors = reader.bit4_factors().map_err(StoreError::Segment)?;
            let mut codes = Vec::with_capacity(survivors.len().saturating_mul(stride));
            let mut factors = Vec::with_capacity(survivors.len());
            for row in survivors {
                let start = row
                    .checked_mul(stride)
                    .ok_or(StoreError::ActiveRowOverflow)?;
                let end = start
                    .checked_add(stride)
                    .ok_or(StoreError::ActiveRowOverflow)?;
                codes.extend_from_slice(
                    source_codes
                        .get(start..end)
                        .ok_or(StoreError::ActiveRowOverflow)?,
                );
                factors.push(
                    source_factors
                        .get(*row)
                        .copied()
                        .ok_or(StoreError::ActiveRowOverflow)?,
                );
            }
            Ok((codes, OwnedFactors::Bit4(factors)))
        }
        2 => {
            let stride = dims;
            let source_codes = reader.int8_codes().map_err(StoreError::Segment)?;
            let source_factors = reader.int8_factors().map_err(StoreError::Segment)?;
            let mut codes = Vec::with_capacity(survivors.len().saturating_mul(stride));
            let mut factors = Vec::with_capacity(survivors.len());
            for row in survivors {
                let start = row
                    .checked_mul(stride)
                    .ok_or(StoreError::ActiveRowOverflow)?;
                let end = start
                    .checked_add(stride)
                    .ok_or(StoreError::ActiveRowOverflow)?;
                for value in source_codes
                    .get(start..end)
                    .ok_or(StoreError::ActiveRowOverflow)?
                {
                    codes.push(value.to_ne_bytes()[0]);
                }
                factors.push(
                    source_factors
                        .get(*row)
                        .copied()
                        .ok_or(StoreError::ActiveRowOverflow)?,
                );
            }
            Ok((codes, OwnedFactors::Int8(factors)))
        }
        scheme => Err(StoreError::Segment(crate::segment::SegmentError::Geometry(
            format!("segment rewrite cannot encode scheme {scheme}"),
        ))),
    }
}

pub(crate) fn gather_survivor_rescore(
    reader: &SegmentReader,
    survivors: &[usize],
    dims: usize,
) -> Result<Vec<f32>, StoreError> {
    let source_rescore = reader.rescore_f32().map_err(StoreError::Segment)?;
    let mut rescore = Vec::with_capacity(survivors.len().saturating_mul(dims));
    for row in survivors {
        let start = row.checked_mul(dims).ok_or(StoreError::ActiveRowOverflow)?;
        let end = start
            .checked_add(dims)
            .ok_or(StoreError::ActiveRowOverflow)?;
        rescore.extend_from_slice(
            source_rescore
                .get(start..end)
                .ok_or(StoreError::ActiveRowOverflow)?,
        );
    }
    Ok(rescore)
}

fn gather_survivor_alive(
    reader: &SegmentReader,
    survivors: &[usize],
    additional_tombstones: &[usize],
) -> Result<AliveSet, StoreError> {
    let additional_tombstones = super::lookup::set(additional_tombstones);
    let source_alive = reader.alive().map_err(StoreError::Segment)?;
    let row_count = u32::try_from(survivors.len()).map_err(|_| StoreError::ActiveRowOverflow)?;
    let mut alive = AliveSet::new(row_count);
    for (new_row, old_row) in survivors.iter().enumerate() {
        let old_row_u32 = u32::try_from(*old_row).map_err(|_| StoreError::ActiveRowOverflow)?;
        if !source_alive.is_alive(old_row_u32)
            || super::lookup::contains(&additional_tombstones, old_row)
        {
            alive
                .tombstone(u32::try_from(new_row).map_err(|_| StoreError::ActiveRowOverflow)?)
                .map_err(|_| StoreError::ActiveRowOverflow)?;
        }
    }
    Ok(alive)
}

pub(crate) fn gather_survivor_documents(
    reader: &SegmentReader,
    survivors: &[usize],
) -> Result<(Vec<DocId>, Vec<super::Revision>), StoreError> {
    let mut doc_ids = Vec::with_capacity(survivors.len());
    let mut revisions = Vec::with_capacity(survivors.len());
    for row in survivors {
        let version = reader
            .document_version(*row)
            .map_err(StoreError::Segment)?
            .ok_or_else(|| {
                StoreError::Segment(crate::segment::SegmentError::Geometry(
                    "purge target segment has no document-version row".to_owned(),
                ))
            })?;
        doc_ids.push(version.doc_id());
        revisions.push(version.revision());
    }
    Ok((doc_ids, revisions))
}

type StoredMetadataGather = (Vec<u64>, Vec<u8>);

fn gather_survivor_metadata(
    reader: &SegmentReader,
    survivors: &[usize],
) -> Result<Option<StoredMetadataGather>, StoreError> {
    let source_metadata = reader.stored_metadata().map_err(StoreError::Segment)?;
    let mut metadata_offsets = Vec::with_capacity(survivors.len());
    let mut metadata_bytes = Vec::new();
    if let Some(source) = source_metadata {
        for row in survivors {
            metadata_bytes.extend_from_slice(source.row(*row).ok_or_else(|| {
                StoreError::Segment(crate::segment::SegmentError::Geometry(format!(
                    "stored metadata row {row} is missing"
                )))
            })?);
            metadata_offsets.push(
                u64::try_from(metadata_bytes.len()).map_err(|_| StoreError::ActiveRowOverflow)?,
            );
        }
    }
    Ok(source_metadata.map(|_| (metadata_offsets, metadata_bytes)))
}

type StoredTextGather = (Vec<u8>, Vec<u64>, Vec<u8>);

fn gather_survivor_text(
    reader: &SegmentReader,
    survivors: &[usize],
) -> Result<Option<StoredTextGather>, StoreError> {
    let source_text = reader.stored_text().map_err(StoreError::Segment)?;
    let mut text_present = Vec::with_capacity(survivors.len());
    let mut text_end_offsets = Vec::with_capacity(survivors.len());
    let mut text_bytes = Vec::new();
    if let Some(source) = source_text {
        for row in survivors {
            match source.row(*row).ok_or_else(|| {
                StoreError::Segment(crate::segment::SegmentError::Geometry(format!(
                    "stored text row {row} is missing"
                )))
            })? {
                Some(text) => {
                    text_present.push(1);
                    text_bytes.extend_from_slice(text.as_bytes());
                }
                None => text_present.push(0),
            }
            text_end_offsets
                .push(u64::try_from(text_bytes.len()).map_err(|_| StoreError::ActiveRowOverflow)?);
        }
    }
    Ok(source_text.map(|_| (text_present, text_end_offsets, text_bytes)))
}

fn rewrite_segment(
    vfs: &dyn Vfs,
    directory: &Path,
    reader: &SegmentReader,
    survivors: &[usize],
    additional_tombstones: &[usize],
    replacement_id: SegmentId,
    policy: DurabilityPolicy,
) -> Result<SegmentMeta, StoreError> {
    let dims = reader.meta().dims as usize;
    let (codes, factors) = gather_survivor_codes(reader, survivors, dims)?;
    let rescore = gather_survivor_rescore(reader, survivors, dims)?;
    let source_columns = reader.columns().map_err(StoreError::Segment)?;
    let columns = rewrite_columns(&source_columns, survivors)?;
    let alive = gather_survivor_alive(reader, survivors, additional_tombstones)?;
    let (doc_ids, revisions) = gather_survivor_documents(reader, survivors)?;
    let metadata = gather_survivor_metadata(reader, survivors)?;
    let text = gather_survivor_text(reader, survivors)?;
    let build = SegmentBuild {
        id: replacement_id,
        scheme: reader.meta().scheme,
        dims: reader.meta().dims,
        codes: &codes,
        factors: factors.borrowed(),
        rescore: &rescore,
        columns: &columns,
        alive: &alive,
    };
    let documents = SegmentDocumentVersions {
        doc_ids: &doc_ids,
        revisions: &revisions,
    };
    let postings = reader
        .postings()
        .map_err(StoreError::Segment)?
        .map(|postings| postings.retain_rows(survivors))
        .transpose()
        .map_err(crate::segment::SegmentError::from)
        .map_err(StoreError::Segment)?
        .map(|postings| postings.encode_region())
        .transpose()
        .map_err(crate::segment::SegmentError::from)
        .map_err(StoreError::Segment)?;
    let metadata = metadata
        .as_ref()
        .map(|(end_offsets, bytes)| SegmentStoredMetadata { end_offsets, bytes });
    let text = text
        .as_ref()
        .map(|(present, end_offsets, bytes)| SegmentStoredText {
            present,
            end_offsets,
            bytes,
        });
    let postings = postings.as_deref().map(|bytes| SegmentPostings { bytes });
    let written = write_segment_with_documents_payloads(
        vfs,
        directory,
        build,
        SegmentPayloads {
            documents,
            metadata,
            text,
            postings,
        },
        policy,
    )
    .map_err(StoreError::Segment)?;
    let mut written = written;
    written.clustering_key_range = clustering_range(&columns, &alive)?;
    Ok(written)
}

fn rewrite_columns(source: &ColumnStore, survivors: &[usize]) -> Result<ColumnStore, StoreError> {
    let mut builder = ColumnStoreBuilder::new(source.schema().clone());
    append_survivor_columns(&mut builder, source, survivors)?;
    builder.finish().map_err(|source| {
        StoreError::Segment(crate::segment::SegmentError::Columns(source.to_string()))
    })
}

pub(crate) fn append_survivor_columns(
    builder: &mut ColumnStoreBuilder,
    source: &ColumnStore,
    survivors: &[usize],
) -> Result<(), StoreError> {
    for row in survivors {
        let row_u32 = u32::try_from(*row).map_err(|_| StoreError::ActiveRowOverflow)?;
        let timestamp = source.timestamp(row_u32).ok_or_else(|| {
            StoreError::Segment(crate::segment::SegmentError::Columns(format!(
                "timestamp row {row} is missing"
            )))
        })?;
        let mut inputs = Vec::with_capacity(source.schema().user_column_count());
        for definition in source
            .schema()
            .columns()
            .iter()
            .filter(|definition| definition.id() != TIMESTAMP_COLUMN)
        {
            let column = source.column(definition.id()).ok_or_else(|| {
                StoreError::Segment(crate::segment::SegmentError::Columns(format!(
                    "column {} is missing",
                    definition.id().get()
                )))
            })?;
            let value = match column {
                Column::Id128(values) => values.get(row_u32).map(ColumnValue::Id128),
                Column::U64(values) => values.get(row_u32).map(ColumnValue::U64),
                Column::I64(values) => values.get(row_u32).map(ColumnValue::I64),
                Column::F64(values) => values.get(row_u32).map(ColumnValue::F64),
                Column::Bool(values) => values.get(row_u32).map(ColumnValue::Bool),
                Column::DictionaryString(values) => values.get(row_u32).map(ColumnValue::String),
                Column::RawString(values) => values.get(row_u32).map(ColumnValue::String),
            };
            if let Some(value) = value {
                inputs.push(ColumnInput {
                    column: definition.id(),
                    value,
                });
            }
        }
        builder.push_row(timestamp, &inputs).map_err(|source| {
            StoreError::Segment(crate::segment::SegmentError::Columns(source.to_string()))
        })?;
    }
    Ok(())
}

pub(crate) fn clustering_range(
    columns: &ColumnStore,
    alive: &AliveSet,
) -> Result<ClusteringKeyRange, StoreError> {
    let mut bounds: Option<(i64, i64)> = None;
    for row in alive.iter_alive() {
        let timestamp = columns
            .timestamp(row)
            .ok_or(StoreError::ActiveRowOverflow)?;
        bounds = Some(match bounds {
            None => (timestamp, timestamp),
            Some((minimum, maximum)) => (minimum.min(timestamp), maximum.max(timestamp)),
        });
    }
    Ok(match bounds {
        Some((min_ts, max_ts)) => ClusteringKeyRange::Bounded { min_ts, max_ts },
        None => ClusteringKeyRange::Empty,
    })
}

fn replacement_segment_id(original: SegmentId, token_id: u64, generation: u64) -> SegmentId {
    // XOR with the previous ID can cancel earlier generations/nonces and
    // overwrite another live segment. Keep the publication generation intact,
    // and distinguish sources within that generation with a seeded hash.
    let mut bytes = [0_u8; 16];
    for (target, value) in bytes.iter_mut().take(8).zip(generation.to_be_bytes()) {
        *target = value;
    }
    let source = xxhash_rust::xxh3::xxh3_64_with_seed(original.as_bytes(), token_id);
    for (target, value) in bytes.iter_mut().skip(8).zip(source.to_be_bytes()) {
        *target = value;
    }
    SegmentId::from_bytes(bytes)
}

fn sweep_purge_orphans(
    vfs: &dyn Vfs,
    directory: &Path,
    manifest: &Manifest,
    policy: DurabilityPolicy,
) -> Result<(), PurgeError> {
    let reachable = manifest
        .segments
        .iter()
        .map(|segment| directory.join(segment.id.file_name()))
        .collect::<HashSet<PathBuf>>();
    let mut deleted = false;
    for path in vfs.list(directory).map_err(|source| StoreError::Io {
        path: directory.to_path_buf(),
        source,
    })? {
        let is_segment = path
            .file_name()
            .and_then(|name| name.to_str())
            .is_some_and(|name| name.starts_with("segment-") && name.ends_with(".zseg"));
        let is_segment_temp = path
            .file_name()
            .and_then(|name| name.to_str())
            .is_some_and(|name| name.starts_with(".segment-") && name.ends_with(".zseg.tmp"));
        let is_wal_temp = path
            .file_name()
            .is_some_and(|name| name == WAL_PURGE_TEMP_FILE);
        if (is_segment && !reachable.contains(&path)) || is_segment_temp || is_wal_temp {
            vfs.delete(&path)
                .map_err(|source| StoreError::Io { path, source })?;
            deleted = true;
        }
    }
    if deleted {
        sync_directory(vfs, directory, policy)?;
    }
    Ok(())
}

fn active_wal_records(segment: &super::ActiveSegment) -> Result<WalImageRecords, PurgeError> {
    let dims = segment.dims().unwrap_or(0);
    let mut records = Vec::with_capacity(segment.row_count().saturating_add(1));
    let mut tombstoned = Vec::with_capacity(segment.row_count());
    let mut deleted = Vec::new();
    let alive = segment.alive()?;
    for row in 0..segment.row_count() {
        let start = row.checked_mul(dims).ok_or(StoreError::ActiveRowOverflow)?;
        let end = start
            .checked_add(dims)
            .ok_or(StoreError::ActiveRowOverflow)?;
        let version = segment.document(row).ok_or(StoreError::ActiveRowOverflow)?;
        let mut document = IngestDocument::new(
            version,
            segment
                .vectors()
                .get(start..end)
                .ok_or(StoreError::ActiveRowOverflow)?
                .to_vec(),
        )
        .with_timestamp(
            segment
                .timestamps()
                .get(row)
                .copied()
                .ok_or(StoreError::ActiveRowOverflow)?,
        )
        .with_metadata(
            segment
                .metadata(row)
                .ok_or(StoreError::ActiveRowOverflow)?
                .to_vec(),
        );
        if let Some(text) = segment.text(row)? {
            document = document.with_text(text);
        }
        document = document.with_columns(segment.column_values(row)?);
        records.push(super::encode_persisted_upsert(&document).map_err(purge_ingest_error)?);
        let row_u32 = u32::try_from(row).map_err(|_| StoreError::ActiveRowOverflow)?;
        let is_tombstoned = !alive.is_alive(row_u32);
        tombstoned.push(is_tombstoned);
        if is_tombstoned {
            deleted.push(version.doc_id());
        }
    }
    if !deleted.is_empty() {
        records.push((
            wal_payload::DELETE_V1,
            wal_payload::encode_delete(&deleted).map_err(PurgeError::WalPayload)?,
        ));
    }
    Ok((records, tombstoned))
}

fn ensure_wal_rewrite_covers_retained(
    vfs: &dyn Vfs,
    directory: &Path,
    manifest: &Manifest,
    purged_ids: &[DocId],
    next_active: &super::ActiveSegment,
    sealed: &[SegmentReader],
    preparation: Option<&crate::lifecycle::namespace_batch::PrivatePreparation>,
) -> Result<(), PurgeError> {
    let purged_ids = super::lookup::set(purged_ids);
    let alive = next_active.alive()?;
    let mut active_versions = std::collections::HashMap::new();
    for (row, id) in next_active.doc_ids().iter().enumerate() {
        super::lookup::work(1);
        let version = next_active
            .document(row)
            .ok_or(StoreError::ActiveRowOverflow)?;
        let row_u32 = u32::try_from(row).map_err(|_| StoreError::ActiveRowOverflow)?;
        active_versions
            .entry(*id)
            .or_insert((version, !alive.is_alive(row_u32)));
    }
    #[cfg(feature = "graph-cypher")]
    let graph_enabled = manifest.graph.is_some();
    #[cfg(not(feature = "graph-cypher"))]
    let graph_enabled = false;
    let wal_path = directory.join("wal.ze");
    let reader = crate::wal::WalReader::open(vfs, &wal_path).map_err(StoreError::Wal)?;
    let clean = reader.into_clean().map_err(StoreError::WalRecovery)?;
    let absorbed_through = manifest.log_seq;
    let decisions = crate::lifecycle::namespace_batch::transaction_decisions(
        vfs,
        directory,
        clean.records(),
        absorbed_through,
        preparation,
    )?;
    let mutations = if decisions.is_empty() {
        super::atomic_batch::committed_mutations(clean.records(), absorbed_through)?
    } else {
        super::atomic_batch::committed_mutations_with_decisions(
            clean.records(),
            absorbed_through,
            |binding| decisions.get(&binding.transaction).copied(),
        )?
    };
    for (seq, _, mutation) in mutations {
        let covered = match mutation {
            #[cfg(feature = "graph-cypher")]
            super::wal_payload::MutationPayload::GraphCommit(_) => manifest
                .graph
                .as_ref()
                .is_some_and(|graph| seq.get() <= graph.graph_absorbed_through),
            super::wal_payload::MutationPayload::Upsert(document) => {
                let version = document.version();
                super::lookup::contains(&purged_ids, &version.doc_id())
                    || active_versions
                        .get(&version.doc_id())
                        .is_some_and(|(existing, _)| existing.revision() >= version.revision())
            }
            super::wal_payload::MutationPayload::Delete(ids) => {
                let mut covered = true;
                for id in ids {
                    if !super::lookup::contains(&purged_ids, &id)
                        && !active_versions
                            .get(&id)
                            .is_some_and(|(_, tombstoned)| *tombstoned)
                        && !delete_is_covered_by_manifest(
                            sealed,
                            id,
                            graph_enabled && !active_versions.contains_key(&id),
                        )?
                    {
                        covered = false;
                        break;
                    }
                }
                covered
            }
            super::wal_payload::MutationPayload::MetadataEdit(_)
            | super::wal_payload::MutationPayload::MixedBatchMember { .. }
            | super::wal_payload::MutationPayload::BatchMember { .. } => false,
        };
        if !covered {
            return Err(PurgeError::WalRewriteWouldDropAcked { seq });
        }
    }
    Ok(())
}

// The manifest-selected segments and replacement active image jointly cover
// a delete: no sealed live copy, and either a persisted tombstone or, on graph
// stores only, no active copy. Graph-free stores require main's stored tombstone.
// Retention may have removed the segment that held the tombstone. Every
// retained upsert must still independently be covered by the rewrite check.
fn delete_is_covered_by_manifest(
    sealed: &[SegmentReader],
    id: DocId,
    active_absent: bool,
) -> Result<bool, StoreError> {
    let mut found = false;
    for segment in sealed {
        let rows = segment.query_rows_with_doc_id(id)?;
        if !rows.is_empty() {
            let alive = segment.alive().map_err(StoreError::Segment)?;
            if rows.iter().any(|row| alive.is_alive(*row)) {
                return Ok(false);
            }
            found = true;
        }
    }
    Ok(found || active_absent)
}

fn purge_ingest_error(error: super::IngestError) -> PurgeError {
    match error {
        #[cfg(feature = "graph-cypher")]
        super::IngestError::Graph(error) => PurgeError::IntentDecode(format!(
            "active WAL rewrite rejected a graph mutation: {error}"
        )),
        super::IngestError::Store(error) => PurgeError::Store(error),
        super::IngestError::EpochMismatch(error) => {
            PurgeError::Store(StoreError::EpochMismatch(error))
        }
        super::IngestError::EpochUndeclared => PurgeError::Store(StoreError::EpochUndeclared),
        super::IngestError::EpochUnstamped => PurgeError::Store(StoreError::EpochUnstamped),
        super::IngestError::Payload(error) => PurgeError::WalPayload(error),
        super::IngestError::EmptyBatch => {
            PurgeError::IntentDecode("active WAL rewrite produced an empty batch error".to_owned())
        }
        super::IngestError::StaleRevision { .. } => PurgeError::IntentDecode(
            "active WAL rewrite produced a stale revision error".to_owned(),
        ),
        super::IngestError::RevisionConflict { .. } => PurgeError::IntentDecode(
            "active WAL rewrite produced a revision condition error".to_owned(),
        ),
        super::IngestError::Vector(error) => PurgeError::IntentDecode(format!(
            "active WAL rewrite rejected a persisted vector: {error}"
        )),
        super::IngestError::Lexical(error) => PurgeError::IntentDecode(format!(
            "active WAL rewrite rejected persisted text: {error}"
        )),
        super::IngestError::Tokenizer(error) => PurgeError::IntentDecode(format!(
            "active WAL rewrite rejected the frozen tokenizer: {error}"
        )),
        super::IngestError::Columns(error) => PurgeError::IntentDecode(format!(
            "active WAL rewrite rejected persisted columns: {error}"
        )),
    }
}

fn assign_rewritten_sequences(
    segment: &mut super::ActiveSegment,
    first_seq: LogSeq,
    tombstoned: &[bool],
) -> Result<(), PurgeError> {
    let delete_seq = (!tombstoned.is_empty() && tombstoned.iter().any(|value| *value))
        .then(|| LogSeq::new(first_seq.get().saturating_add(segment.row_count() as u64)));
    for row in 0..segment.row_count() {
        let row_seq = LogSeq::new(first_seq.get().saturating_add(row as u64));
        let sequence = if tombstoned.get(row).copied().unwrap_or(false) {
            delete_seq.unwrap_or(row_seq)
        } else {
            row_seq
        };
        segment.set_sequence(row, sequence)?;
    }
    Ok(())
}

#[cfg(unix)]
fn available_disk_bytes(path: &Path) -> std::io::Result<u64> {
    use std::ffi::CString;
    use std::os::unix::ffi::OsStrExt as _;

    let encoded = CString::new(path.as_os_str().as_bytes()).map_err(|_| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            "store path contains a NUL byte",
        )
    })?;
    let mut output = std::mem::MaybeUninit::<libc::statvfs>::uninit();
    // SAFETY: `encoded` is NUL-terminated and `output` points to writable storage
    // for exactly one `statvfs` result initialized by a successful call.
    let result = unsafe { libc::statvfs(encoded.as_ptr(), output.as_mut_ptr()) };
    if result != 0 {
        return Err(std::io::Error::last_os_error());
    }
    // SAFETY: the successful `statvfs` call initialized every field in `output`.
    let output = unsafe { output.assume_init() };
    let blocks = u128::from(output.f_bavail);
    let fragment = u128::from(output.f_frsize);
    u64::try_from(blocks.saturating_mul(fragment)).map_err(|_| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "available filesystem bytes exceed u64",
        )
    })
}

/// Bytes available to this caller on the volume that actually holds `path`.
///
/// `GetDiskFreeSpaceExW` is quota-aware and resolves the volume from the store
/// path itself, so a store on a mount point or junction is measured against the
/// volume it really lives on rather than the one its path prefix suggests.
#[cfg(windows)]
fn available_disk_bytes(path: &Path) -> std::io::Result<u64> {
    crate::sys::windows::available_disk_bytes(path)
}

#[cfg(not(any(unix, windows)))]
fn available_disk_bytes(_path: &Path) -> std::io::Result<u64> {
    Err(std::io::Error::new(
        std::io::ErrorKind::Unsupported,
        "physical purge free-space probing requires statvfs",
    ))
}

#[cfg(test)]
#[allow(clippy::expect_used)]
mod tests {
    use super::{
        PURGE_INTENT_FILE, PurgeError, PurgeIntent, PurgeReport, PurgeToken, enforce_temp_space,
        purge_token_id, read_intent, read_u32, read_u64, remove_intent, write_intent,
    };
    use crate::format::FormatFamily;
    use crate::format::frame::{decode_artifact, encode_artifact};
    use crate::ingest::{DocId, wal_payload};
    use crate::lifecycle::StoreError;
    use crate::lifecycle::durability::{CommitTier, DurabilityMode, DurabilityPolicy};
    use crate::vfs::StdVfs;
    use crate::wal::LogSeq;

    #[test]
    fn a_graph_free_purge_after_retention_matches_main() {
        use crate::ingest::{DeleteBatch, DocumentVersion, IngestBatch, IngestDocument, Revision};
        use crate::lifecycle::{OpenOptions, Store};
        let directory = tempfile::tempdir().expect("directory");
        let store = Store::open(directory.path(), OpenOptions::default()).expect("open");
        store
            .ingest(IngestBatch::new(vec![
                IngestDocument::new(
                    DocumentVersion::new(DocId::new(91), Revision::new(1)),
                    vec![1.0],
                )
                .with_timestamp(91),
            ]))
            .expect("A");
        store.seal().expect("seal A");
        store
            .ingest(IngestBatch::new(
                vec![92, 93]
                    .into_iter()
                    .map(|id| {
                        IngestDocument::new(
                            DocumentVersion::new(DocId::new(id), Revision::new(1)),
                            vec![1.0],
                        )
                    })
                    .collect(),
            ))
            .expect("B C");
        store
            .delete(DeleteBatch::new(vec![DocId::new(91)]))
            .expect("delete A");
        store.drop_partition(91..92).expect("drop A");
        let token = store.purge(&[DocId::new(92)]).expect("purge B");
        let before = std::fs::read(directory.path().join("wal.ze")).expect("WAL before");
        let result = store.await_physical_purge(token);
        assert!(
            matches!(result, Err(PurgeError::WalRewriteWouldDropAcked { seq }) if seq == LogSeq::new(4)),
            "{result:?}"
        );
        assert_eq!(
            std::fs::read(directory.path().join("wal.ze")).expect("WAL after"),
            before
        );
        // Measured on main 9d217196 in /private/tmp/w346-p1p2-main, a separate
        // temporary worktree: same test, graph-cypher, rejection at sequence 4.
        // Pin every durable byte, including the pending intent and unchanged WAL.
        let main = [
            (".ze-readers.lock", ""),
            ("manifest.ze", "5a4550454d4245440a000200000000002000000000000000700000000000000038000000000000000600000000000000010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000f6dd00afe67b79862c016c188e60eca8"),
            ("purge.ze", "5a4550454d4245440f00010000000000200000000000000058000000000000002000000000000000bf09f9ca40c7b1e501000000000000005c0000000000000000000000000000004c66d4b4b29ca4158fc5f105cfc7c200"),
            ("wal.ze", "5a4550454d4245440b000100000000002800000000000000000000000000000002000000000000002c000000020000000000000008000000000002000000010000005c0000000000000000000000000000000100000000000000010000000000803f14975d8581be8d602c000000030000000000000008000100000002000000010000005d0000000000000000000000000000000100000000000000010000000000803f6c6214603baa8205180000000400000000000000020001000000010000005b000000000000000000000000000000c95f15ae22a86feb"),
            ("writer.lock", ""),
        ].into_iter().map(|(name, hex)| {
            let bytes = hex.as_bytes().chunks_exact(2).map(|pair|
                u8::from_str_radix(std::str::from_utf8(pair).expect("hex pair"), 16).expect("hex byte")
            ).collect::<Vec<_>>();
            (name.to_owned(), bytes)
        }).collect::<std::collections::BTreeMap<_, _>>();
        let files = std::fs::read_dir(directory.path())
            .expect("files")
            .map(|entry| {
                let entry = entry.expect("entry");
                (
                    entry.file_name().to_string_lossy().into_owned(),
                    std::fs::read(entry.path()).expect("bytes"),
                )
            })
            .collect::<std::collections::BTreeMap<_, _>>();
        assert_eq!(files, main);
    }

    #[test]
    fn a_graph_free_tombstone_repair_matches_main_bytes() {
        use crate::ingest::{DeleteBatch, DocumentVersion, IngestBatch, IngestDocument, Revision};
        use crate::lifecycle::{OpenOptions, Store};
        let directory = tempfile::tempdir().expect("directory");
        let store = Store::open(directory.path(), OpenOptions::default()).expect("open");
        assert_eq!(
            store
                .ingest(IngestBatch::new(vec![IngestDocument::new(
                    DocumentVersion::new(DocId::new(91), Revision::new(1)),
                    vec![1.0]
                )]))
                .expect("ingest")
                .generation(),
            1
        );
        assert_eq!(store.seal().expect("seal"), 2);
        let manifest_path = directory.path().join("manifest.ze");
        let old_manifest = std::fs::read(&manifest_path).expect("old manifest");
        let segment_path = directory
            .path()
            .join("segment-00000000000000020000000000000001.zseg");
        let old_segment = std::fs::read(&segment_path).expect("old segment");
        assert_eq!(
            store
                .delete(DeleteBatch::new(vec![DocId::new(91)]))
                .expect("delete")
                .generation(),
            3
        );
        // Lose only the delete publication: retain the synced WAL and old sealed bytes.
        drop(store);
        std::fs::write(&manifest_path, old_manifest).expect("restore manifest");
        std::fs::write(&segment_path, old_segment).expect("restore segment");
        let recovered = Store::open(directory.path(), OpenOptions::default()).expect("recover");
        let bytes = std::fs::read(&manifest_path).expect("repaired manifest");
        let manifest = crate::manifest::decode_manifest("repaired", &bytes).expect("decode");
        // Measured twice on main 9d217196 in a separate temporary worktree,
        // without stashing: review_main_bytes::record_main_repair_bytes.
        assert_eq!(manifest.generation, 4);
        assert_eq!(
            manifest
                .segments
                .first()
                .expect("replacement")
                .id
                .to_string(),
            "00000000000000049e7321e873816107"
        );
        let main_hex = concat!(
            "5a4550454d4245440a000200000000002000000000000000c4000000000000008c00000000000000",
            "04000000000000000100000000000000010000000000000000000000000000000000000000000000",
            "0000000000000000000000000000000000000000000000049e7321e8738161070100000004000000",
            "0100000070c001000000000000000000000000000000000000000000545352310100000001000000",
            "00000000000000000000000000000000000000000026433d9a49bbe058e92629d0d9c543",
        );
        let expected = main_hex
            .as_bytes()
            .chunks_exact(2)
            .map(|pair| {
                u8::from_str_radix(std::str::from_utf8(pair).expect("hex pair"), 16)
                    .expect("hex byte")
            })
            .collect::<Vec<_>>();
        assert_eq!(
            bytes, expected,
            "graph-free recovery must preserve main's complete manifest bytes"
        );
        assert_eq!(recovered.snapshot().expect("snapshot").generation(), 4);
        assert_eq!(
            recovered.count_documents(None, None).expect("count").count,
            0
        );
    }

    fn graph_free_pending_purge_crash(ids: &[u128]) -> tempfile::TempDir {
        use crate::ingest::{DeleteBatch, DocumentVersion, IngestBatch, IngestDocument, Revision};
        use crate::lifecycle::{OpenOptions, Store};
        let directory = tempfile::tempdir().expect("directory");
        let store = Store::open(directory.path(), OpenOptions::default()).expect("open");
        store
            .ingest(IngestBatch::new(
                ids.iter()
                    .map(|id| {
                        IngestDocument::new(
                            DocumentVersion::new(DocId::new(*id), Revision::new(1)),
                            vec![1.0],
                        )
                    })
                    .collect(),
            ))
            .expect("ingest");
        store.seal().expect("seal");
        store
            .purge(&[DocId::new(if ids.len() == 1 { 91 } else { 92 })])
            .expect("schedule purge");
        let path = directory.path().join("manifest.ze");
        let before = std::fs::read(&path).expect("manifest before delete");
        let manifest = crate::manifest::decode_manifest("before", &before).expect("decode");
        let segment_path = directory.path().join(
            manifest
                .segments
                .first()
                .expect("sealed segment")
                .id
                .file_name(),
        );
        let segment = std::fs::read(&segment_path).expect("sealed bytes");
        assert_eq!(
            store
                .delete(DeleteBatch::new(vec![DocId::new(91)]))
                .expect("delete")
                .generation(),
            3
        );
        drop(store);
        // Keep the synced delete WAL; lose its manifest and segment publication.
        std::fs::write(&path, before).expect("restore manifest");
        std::fs::write(&segment_path, segment).expect("restore segment");
        directory
    }

    #[test]
    fn a_graph_free_pending_purge_with_an_unmanifested_delete_matches_main() {
        use crate::lifecycle::{OpenOptions, Store};
        let directory = graph_free_pending_purge_crash(&[91]);
        let recovered = Store::open(directory.path(), OpenOptions::default()).expect("recover");
        let bytes = std::fs::read(directory.path().join("manifest.ze")).expect("manifest");
        let manifest = crate::manifest::decode_manifest("recovered", &bytes).expect("decode");
        // Measured on main 9d217196 in the separate temporary worktree
        // /private/tmp/w346-main-byte-reference: review_pending_purge::record_main_pending_purge.
        assert_eq!(manifest.generation, 4);
        assert_eq!(
            manifest
                .segments
                .first()
                .expect("replacement")
                .id
                .to_string(),
            "0000000000000004e326c259b011dc7d"
        );
        let main_hex = "5a4550454d4245440a000200000000002000000000000000c4000000000000008c0000000000000004000000000000000100000000000000010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000004e326c259b011dc7d00000000040000000100000060800100000000000000000000000000000000000000000054535231010000000100000000000000000000000000000000000000000000002a37ad4743e41889dc2754017bb05dc9";
        let expected = main_hex
            .as_bytes()
            .chunks_exact(2)
            .map(|pair| {
                u8::from_str_radix(std::str::from_utf8(pair).expect("hex pair"), 16)
                    .expect("hex byte")
            })
            .collect::<Vec<_>>();
        assert_eq!(
            bytes, expected,
            "complete graph-free manifest must match main"
        );
        assert_eq!(recovered.snapshot().expect("snapshot").generation(), 4);
        assert_eq!(
            recovered.count_documents(None, None).expect("count").count,
            0
        );
    }

    #[test]
    fn a_graph_free_pending_purge_of_another_document_preserves_main_refusal() {
        use crate::lifecycle::{OpenOptions, Store};
        let directory = graph_free_pending_purge_crash(&[91, 92, 93]);
        // The same independent main worktree reports this exact refusal.
        // Preserve it even though repairing first could make this state open.
        let error = Store::open(directory.path(), OpenOptions::default())
            .err()
            .expect("main refuses this crash state");
        assert!(
            matches!(error, StoreError::PurgeRecovery { ref detail } if detail == "purge WAL rewrite would drop acknowledged WAL sequence 4"),
            "{error:?}"
        );
    }

    #[test]
    fn sealed_purge_lookup_work_is_linear() {
        use crate::ingest::{DocumentVersion, IngestBatch, IngestDocument, Revision};
        use crate::lifecycle::{OpenOptions, Store};
        for n in [128, 512, 2048] {
            let directory = tempfile::tempdir().expect("directory");
            let store = Store::open(directory.path(), OpenOptions::default()).expect("open");
            let documents = (0..n)
                .map(|id| {
                    IngestDocument::new(
                        DocumentVersion::new(DocId::new(id as u128), Revision::new(1)),
                        vec![1.0],
                    )
                })
                .collect();
            store.ingest(IngestBatch::new(documents)).expect("ingest");
            store.seal().expect("seal");
            let snapshot = store.snapshot().expect("snapshot");
            let reader = snapshot.segments().first().expect("segment");
            let mut ids: Vec<_> = (0..n)
                .step_by(2)
                .rev()
                .map(|id| DocId::new(id as u128))
                .collect();
            ids.extend([DocId::new(0), DocId::new(u128::MAX)]);
            super::super::lookup::take_work();
            let requested = super::super::lookup::set(&ids);
            let survivors = super::survivor_rows(reader, &requested).expect("survivors");
            let work = super::super::lookup::take_work();
            assert_eq!(
                survivors,
                (0..n).filter(|id| id % 2 != 0).collect::<Vec<_>>()
            );
            assert!(work <= 2 * (n + ids.len()), "{n} rows: {work} lookup work");
        }
    }

    #[test]
    fn purge_tokens_reports_and_errors_preserve_their_contract_values() {
        let unknown = DocId::new(19);
        let token = PurgeToken {
            id: 41,
            generation: 42,
            unknown_ids: vec![unknown],
            no_op: false,
        };
        assert_eq!(token.id(), 41);
        assert_eq!(token.generation(), 42);
        assert_eq!(token.unknown_ids(), &[unknown]);
        assert!(!token.is_no_op());

        let report = PurgeReport {
            generation: 43,
            segments_rewritten: 2,
            wal_rewritten: true,
            unknown_ids: vec![unknown],
        };
        assert_eq!(report.generation(), 43);
        assert_eq!(report.segments_rewritten(), 2);
        assert!(report.wal_rewritten());
        assert_eq!(report.unknown_ids(), &[unknown]);
        assert!(!report.is_no_op());

        let format = decode_artifact("purge", FormatFamily::PurgeIntent, &[])
            .expect_err("empty intent is invalid");
        let errors = [
            PurgeError::Store(StoreError::ReadOnly),
            PurgeError::InsufficientTempSpace {
                segment_bytes: 100,
                available_bytes: 120,
                required_bytes: 120,
            },
            PurgeError::PurgeInProgress,
            PurgeError::UnknownToken { token_id: 44 },
            PurgeError::IntentFormat(format),
            PurgeError::IntentDecode("bad payload".to_owned()),
            PurgeError::WalPayload(wal_payload::PayloadError::Truncated),
            PurgeError::WalRewriteWouldDropAcked {
                seq: LogSeq::new(45),
            },
        ];
        for error in errors {
            assert!(!error.to_string().is_empty());
            let sourced = std::error::Error::source(&error).is_some();
            assert_eq!(
                sourced,
                matches!(
                    error,
                    PurgeError::Store(_) | PurgeError::IntentFormat(_) | PurgeError::WalPayload(_)
                )
            );
        }
        let converted = PurgeError::from(StoreError::Closed);
        assert!(matches!(converted, PurgeError::Store(StoreError::Closed)));

        let ids = [DocId::new(3), DocId::new(5)];
        assert_eq!(purge_token_id(7, &ids), purge_token_id(7, &ids));
        assert_ne!(purge_token_id(7, &ids), purge_token_id(8, &ids));
        assert!(enforce_temp_space(100, 121).is_ok());
        assert!(matches!(
            enforce_temp_space(100, 120),
            Err(PurgeError::InsufficientTempSpace {
                required_bytes: 120,
                ..
            })
        ));
    }

    #[test]
    fn purge_intent_round_trip_rejects_reserved_and_length_corruption() {
        let directory = tempfile::tempdir().expect("purge intent test directory");
        let policy = DurabilityPolicy::new(DurabilityMode::Derived, CommitTier::None)
            .expect("derived durability");
        let intent = PurgeIntent {
            token_id: 0x0102_0304_0506_0708,
            ids: vec![DocId::new(11), DocId::new(12)],
        };
        write_intent(&StdVfs, directory.path(), &intent, policy).expect("write valid intent");
        let decoded = read_intent(&StdVfs, directory.path()).expect("read valid intent");
        assert_eq!(decoded.token_id, intent.token_id);
        assert_eq!(decoded.ids, intent.ids);

        let path = directory.path().join(PURGE_INTENT_FILE);
        let bytes = std::fs::read(&path).expect("read framed intent");
        let payload = decode_artifact("purge", FormatFamily::PurgeIntent, &bytes)
            .expect("decode framed intent")
            .payload
            .to_vec();

        let mut reserved = payload.clone();
        let field = reserved.get_mut(12).expect("reserved byte exists");
        *field = 1;
        std::fs::write(
            &path,
            encode_artifact(FormatFamily::PurgeIntent, 0, &reserved),
        )
        .expect("write reserved corruption");
        assert!(matches!(
            read_intent(&StdVfs, directory.path()),
            Err(PurgeError::IntentDecode(detail)) if detail.contains("reserved")
        ));

        let truncated = payload.get(..payload.len() - 1).expect("truncate payload");
        std::fs::write(
            &path,
            encode_artifact(FormatFamily::PurgeIntent, 0, truncated),
        )
        .expect("write length corruption");
        assert!(matches!(
            read_intent(&StdVfs, directory.path()),
            Err(PurgeError::IntentDecode(detail)) if detail.contains("expected")
        ));
        assert!(read_u32(&[], 0).is_err());
        assert!(read_u64(&[], 0).is_err());

        std::fs::write(&path, bytes).expect("restore valid intent");
        remove_intent(&StdVfs, directory.path(), policy).expect("remove valid intent");
        assert!(!path.exists());
    }
}