hawdb-storage 0.10.0

// Copyright 2026 Nowledge
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use super::{
    codec, durability, relational_row_delta_manifest_generation_file,
    relational_row_delta_run_file, RelationalRowDeltaBaseBinding, RelationalRowDeltaConfig,
    RelationalRowDeltaError, RelationalRowDeltaGeneration, RelationalRowDeltaManifest,
    RelationalRowDeltaPublicationPhase, RelationalRowDeltaReadReport, RelationalRowDeltaReport,
    RelationalRowDeltaTableMetadata, RowDeltaKey, RowDeltaRunContext, RowDeltaRunDescriptor,
    RowDeltaValue, COMPLETE_PUBLICATION_TRACE, RELATIONAL_ROW_DELTA_MANIFEST_FILE,
    RELATIONAL_ROW_DELTA_PUBLICATION_LOCK_FILE,
};
use crate::durability::{durable_replace_file, sync_directory};
use crate::file_io::{self as fs, File, OpenOptions};
use crate::relational::row_page::{RelationalRowPageRecoveredValue, RelationalRowPageRootReader};
use crate::relational::{
    ordered_key::{encode_ordered_relational_key, CheckpointKeyEncoding},
    RelationalKey, RelationalOverflowRootReader, RelationalRecoverySourceIdentity,
    RelationalRowChange, RelationalRowChangeCapture, RelationalRowChangeCaptureLimits,
    RelationalRowPagePublicationConfig, RelationalState,
};
use std::io::Write;
use std::num::NonZeroU32;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

static NEXT_ROW_DELTA_GENERATION: AtomicU64 = AtomicU64::new(1);

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RowRootSelection {
    IndependentLatest,
    CanonicalCheckpoint,
}

#[derive(Debug, Clone, Copy)]
enum DeltaSelection {
    Latest,
    PrivateGeneration,
    PrivatePrefix,
}

#[derive(Debug, Clone)]
pub struct RelationalRowDeltaBuilder {
    directory: PathBuf,
    base: RelationalRowDeltaBaseBinding,
    row_page_config: RelationalRowPagePublicationConfig,
    delta_generation: u64,
    expected_previous: Option<RelationalRowDeltaGeneration>,
    config: RelationalRowDeltaConfig,
    tables: Vec<RelationalRowDeltaTableMetadata>,
    schema_set_digest: hawdb_integrity::Sha256Digest,
    visible_commit_epoch: u64,
    replayed_batches: u64,
    dirty: super::DirtyEntries,
    dirty_bytes: usize,
    runs: Vec<RowDeltaRunDescriptor>,
    run_bytes: u64,
    peak_dirty_entries: usize,
    peak_dirty_bytes: usize,
    row_root_selection: RowRootSelection,
    invalidated: Option<String>,
    // A durable, unselected prefix survives a later reader-admission denial.
    private_prefix: Option<(RelationalRecoverySourceIdentity, RelationalRowDeltaReport)>,
    _runs_memory: crate::background::CheckpointAllocationOwner,
    _bounds_memory: crate::background::CheckpointAllocationOwner,
}

impl RelationalRowDeltaBuilder {
    pub fn validate_base_state(
        base: &RelationalRowPageRootReader,
        state: &RelationalState,
        config: RelationalRowDeltaConfig,
    ) -> Result<(), RelationalRowDeltaError> {
        let tables = table_metadata_for_state(state)?;
        codec::validate_tables_against_base(base, tables, config)?;
        Ok(())
    }

    pub fn new_for_recovery(
        directory: &Path,
        base: &RelationalRowPageRootReader,
        expected_previous: Option<RelationalRowDeltaGeneration>,
        state: &RelationalState,
        config: RelationalRowDeltaConfig,
    ) -> Result<Self, RelationalRowDeltaError> {
        let delta_generation = next_recovery_delta_generation(base, expected_previous)?;
        Self::new_for_state(
            directory,
            base,
            delta_generation,
            expected_previous,
            state,
            config,
        )
    }

    /// Creates a recovery builder whose immutable base is selected by an
    /// enclosing canonical checkpoint manifest rather than the row-page
    /// subsystem's independent latest selector.
    pub fn new_for_checkpoint_recovery(
        directory: &Path,
        base: &RelationalRowPageRootReader,
        expected_previous: Option<RelationalRowDeltaGeneration>,
        state: &RelationalState,
        config: RelationalRowDeltaConfig,
    ) -> Result<Self, RelationalRowDeltaError> {
        let delta_generation = next_recovery_delta_generation(base, expected_previous)?;
        let tables = table_metadata_for_state(state)?;
        Self::new_inner(
            directory,
            base,
            delta_generation,
            expected_previous,
            tables,
            config,
            RowRootSelection::CanonicalCheckpoint,
        )
    }

    pub fn new_for_state(
        directory: &Path,
        base: &RelationalRowPageRootReader,
        delta_generation: u64,
        expected_previous: Option<RelationalRowDeltaGeneration>,
        state: &RelationalState,
        config: RelationalRowDeltaConfig,
    ) -> Result<Self, RelationalRowDeltaError> {
        let tables = table_metadata_for_state(state)?;
        Self::new(
            directory,
            base,
            delta_generation,
            expected_previous,
            tables,
            config,
        )
    }

    pub fn new(
        directory: &Path,
        base: &RelationalRowPageRootReader,
        delta_generation: u64,
        expected_previous: Option<RelationalRowDeltaGeneration>,
        tables: Vec<RelationalRowDeltaTableMetadata>,
        config: RelationalRowDeltaConfig,
    ) -> Result<Self, RelationalRowDeltaError> {
        Self::new_inner(
            directory,
            base,
            delta_generation,
            expected_previous,
            tables,
            config,
            RowRootSelection::IndependentLatest,
        )
    }

    fn new_inner(
        directory: &Path,
        base: &RelationalRowPageRootReader,
        delta_generation: u64,
        expected_previous: Option<RelationalRowDeltaGeneration>,
        tables: Vec<RelationalRowDeltaTableMetadata>,
        config: RelationalRowDeltaConfig,
        row_root_selection: RowRootSelection,
    ) -> Result<Self, RelationalRowDeltaError> {
        config.validate_run_policy()?;
        if delta_generation == 0 {
            return Err(RelationalRowDeltaError::Admission(
                "row delta generation must be non-zero".to_string(),
            ));
        }
        if config.max_manifest_bytes.get() < codec::MANIFEST_HEADER_BYTES {
            return Err(RelationalRowDeltaError::Admission(format!(
                "row delta manifest requires {} bytes, exceeding limit {}",
                codec::MANIFEST_HEADER_BYTES,
                config.max_manifest_bytes
            )));
        }
        let tables = codec::validate_tables_against_base(base, tables, config)?;
        let schema_set_digest = codec::schema_set_digest(&tables)?;
        let row_page_config = base.publication_config();
        let base = RelationalRowDeltaBaseBinding::from_reader(base);
        if expected_previous.is_some_and(|previous| {
            previous.base_generation == base.generation
                && previous.delta_generation >= delta_generation
        }) {
            return Err(RelationalRowDeltaError::Admission(format!(
                "row delta generation {delta_generation} must exceed the previous generation"
            )));
        }
        let generation_manifest = directory.join(relational_row_delta_manifest_generation_file(
            base.generation,
            delta_generation,
        ));
        if generation_manifest.exists() {
            return Err(RelationalRowDeltaError::Admission(format!(
                "immutable row delta generation manifest {} already exists",
                generation_manifest.display()
            )));
        }
        fs::create_dir_all(directory).map_err(durability("create row delta directory"))?;
        Ok(Self {
            directory: directory.to_path_buf(),
            base,
            row_page_config,
            delta_generation,
            expected_previous,
            config,
            tables,
            schema_set_digest,
            visible_commit_epoch: base.source_commit_epoch,
            replayed_batches: 0,
            dirty: super::DirtyEntries::default(),
            dirty_bytes: 0,
            runs: Vec::new(),
            run_bytes: 0,
            peak_dirty_entries: 0,
            peak_dirty_bytes: 0,
            row_root_selection,
            invalidated: None,
            private_prefix: None,
            _runs_memory: crate::background::CheckpointAllocationOwner::default(),
            _bounds_memory: crate::background::CheckpointAllocationOwner::default(),
        })
    }

    pub const fn capture_limits(&self) -> RelationalRowChangeCaptureLimits {
        self.config.capture_limits()
    }

    pub const fn base_commit_epoch(&self) -> u64 {
        self.base.source_commit_epoch
    }

    pub const fn visible_commit_epoch(&self) -> u64 {
        self.visible_commit_epoch
    }

    /// Looks up the newest value already staged by this unpublished builder.
    ///
    /// The bounded dirty map takes precedence over immutable runs. Run reads
    /// use the same checksummed decoder as a published delta reader, allowing
    /// WAL recovery to hydrate one later transaction without retaining all
    /// earlier recovered rows in memory.
    pub fn lookup_staged(
        &self,
        table: &str,
        primary_key: &RelationalKey,
    ) -> Result<
        (
            Option<RelationalRowPageRecoveredValue>,
            RelationalRowDeltaReadReport,
        ),
        RelationalRowDeltaError,
    > {
        self.lookup_staged_with_allocations(table, primary_key, None)
    }

    pub(crate) fn lookup_staged_with_allocations(
        &self,
        table: &str,
        primary_key: &RelationalKey,
        allocations: Option<&crate::background::CheckpointDecodeContext>,
    ) -> Result<
        (
            Option<RelationalRowPageRecoveredValue>,
            RelationalRowDeltaReadReport,
        ),
        RelationalRowDeltaError,
    > {
        self.require_available()?;
        let Ok(table_ordinal) = self
            .tables
            .binary_search_by(|candidate| candidate.table.as_str().cmp(table))
        else {
            return Ok((None, RelationalRowDeltaReadReport::default()));
        };
        let table_ordinal = u32::try_from(table_ordinal).map_err(|_| {
            RelationalRowDeltaError::Corrupt("row delta table ordinal does not fit u32".to_string())
        })?;
        // Query-key scratch dies with this lookup, independently of the
        // caller's inventory for returned row payloads.
        let key_allocations =
            allocations.map(|context| crate::background::CheckpointDecodeContext {
                work: context.work.clone(),
                memory: Default::default(),
            });
        let encoded_primary_key = match key_allocations.as_ref() {
            Some(context) => {
                let encoding = CheckpointKeyEncoding::new(primary_key, &[], context)
                    .map_err(super::reader::lookup_key_encoding_error)?;
                if encoding.encoded_len() > self.config.row_limits.max_key_bytes.get() {
                    return Err(RelationalRowDeltaError::Admission(format!(
                        "row delta lookup key contains {} bytes, exceeding limit {}",
                        encoding.encoded_len(),
                        self.config.row_limits.max_key_bytes
                    )));
                }
                encoding
                    .encode_owned(context)
                    .map_err(super::reader::lookup_key_encoding_error)?
            }
            None => encode_ordered_relational_key(primary_key).map_err(|error| {
                RelationalRowDeltaError::Admission(format!(
                    "row delta lookup key cannot be encoded: {error}"
                ))
            })?,
        };
        if encoded_primary_key.len() > self.config.row_limits.max_key_bytes.get() {
            return Err(RelationalRowDeltaError::Admission(format!(
                "row delta lookup key contains {} bytes, exceeding limit {}",
                encoded_primary_key.len(),
                self.config.row_limits.max_key_bytes
            )));
        }
        if let Some(context) = key_allocations.as_ref() {
            context
                .reserve(std::mem::size_of::<Vec<u8>>() + 2 * std::mem::size_of::<usize>())
                .map_err(|error| RelationalRowDeltaError::Admission(error.to_string()))?;
        }
        let key = RowDeltaKey {
            table_ordinal,
            encoded_primary_key: encoded_primary_key.into(),
        };
        if let Some(value) = self.dirty.get(&key) {
            let value = match allocations {
                Some(context) => super::reader::decode_staged_value_with_allocations(
                    value,
                    &self.tables[table_ordinal as usize],
                    self.config,
                    context,
                )?,
                None => super::reader::decode_staged_value(
                    value,
                    &self.tables[table_ordinal as usize],
                    self.config,
                )?,
            };
            return Ok((
                Some(value),
                RelationalRowDeltaReadReport {
                    entries_visited: 1,
                    ..RelationalRowDeltaReadReport::default()
                },
            ));
        }
        // Free the temporary dirty-map key before encoding the immutable-run
        // lookup key. Physical backing always dies before its permit owner.
        drop(key);
        drop(key_allocations);
        super::reader::lookup_runs_with_allocations(
            RowDeltaRunContext {
                directory: &self.directory,
                base: self.base,
                delta_generation: self.delta_generation,
                schema_set_digest: self.schema_set_digest,
                tables: &self.tables,
                config: self.config,
            },
            &self.runs,
            table,
            primary_key,
            allocations,
        )
    }

    pub fn record(
        &mut self,
        epoch: u64,
        capture: RelationalRowChangeCapture,
    ) -> Result<(), RelationalRowDeltaError> {
        self.record_with_work_context(epoch, capture, &Default::default())
    }

    pub(crate) fn record_with_work_context(
        &mut self,
        epoch: u64,
        capture: RelationalRowChangeCapture,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<(), RelationalRowDeltaError> {
        work.checkpoint()
            .map_err(super::super::checkpoint::work_error)?;
        self.require_available()?;
        self.private_prefix = None;
        let result = self.record_inner(epoch, capture, work);
        if let Err(error) = &result {
            self.invalidated = Some(error.to_string());
        }
        result
    }

    pub fn advance_empty(&mut self, epoch: u64) -> Result<(), RelationalRowDeltaError> {
        self.record(
            epoch,
            RelationalRowChangeCapture::Captured {
                changes: Vec::new(),
                encoded_bytes: 0,
            },
        )
    }

    #[cfg(test)]
    pub fn finish(
        self,
        expected_visible_commit_epoch: u64,
        overflow_root: Option<&RelationalOverflowRootReader>,
    ) -> Result<RelationalRowDeltaReport, RelationalRowDeltaError> {
        let recovery_source = RelationalRecoverySourceIdentity::for_test(
            self.base.source_commit_epoch,
            expected_visible_commit_epoch,
        );
        self.finish_inner(
            expected_visible_commit_epoch,
            recovery_source,
            overflow_root,
            None,
        )
    }

    /// Publishes a recovery delta with exact final row counts bound to the
    /// same relational state that produced its captured row changes.
    pub fn finish_with_state(
        self,
        expected_visible_commit_epoch: u64,
        recovery_source: RelationalRecoverySourceIdentity,
        overflow_root: Option<&RelationalOverflowRootReader>,
        final_state: &RelationalState,
    ) -> Result<RelationalRowDeltaReport, RelationalRowDeltaError> {
        self.finish_with_state_inner(
            expected_visible_commit_epoch,
            recovery_source,
            overflow_root,
            final_state,
            DeltaSelection::Latest,
        )
    }

    pub(crate) fn seal_private_with_state(
        &mut self,
        expected_visible_commit_epoch: u64,
        recovery_source: RelationalRecoverySourceIdentity,
        overflow_root: Option<&RelationalOverflowRootReader>,
        final_state: &RelationalState,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<RelationalRowDeltaReport, RelationalRowDeltaError> {
        self.require_available()?;
        if let Some((identity, report)) = &self.private_prefix
            && *identity == recovery_source
            && report.visible_commit_epoch == expected_visible_commit_epoch
            && self.visible_commit_epoch == expected_visible_commit_epoch
        {
            return Ok(report.clone());
        }
        // Flush the live builder first. Cloning dirty entries before flushing
        // would reuse run ordinals and overwrite an already pinned prefix.
        self.flush_with_work_context(work)?;
        // Admit every backing allocation copied by Clone before creating the
        // snapshot. The live builder remains reusable if admission fails.
        let mut memory = crate::background::CheckpointAllocationOwner::default();
        let mut reserve = |bytes: usize| {
            memory
                .reserve(bytes, work)
                .map(|_| ())
                .map_err(|error| RelationalRowDeltaError::Admission(error.to_string()))
        };
        let checked_bytes = |count: usize, size: usize| {
            count.checked_mul(size).ok_or_else(|| {
                RelationalRowDeltaError::Admission("private row builder clone size overflow".into())
            })
        };
        reserve(self.directory.as_os_str().len())?;
        reserve(checked_bytes(
            self.tables.len(),
            std::mem::size_of::<RelationalRowDeltaTableMetadata>(),
        )?)?;
        for table in &self.tables {
            reserve(table.table.len())?;
        }
        reserve(checked_bytes(
            self.runs.len(),
            std::mem::size_of::<RowDeltaRunDescriptor>(),
        )?)?;
        for run in &self.runs {
            reserve(run.lower_bound.encoded_primary_key.len())?;
            reserve(run.upper_bound.encoded_primary_key.len())?;
        }
        if let Some(reason) = &self.invalidated {
            reserve(reason.len())?;
        }
        reserve(codec::manifest_encoding_workspace_bytes(
            &self.tables,
            &self.runs,
        )?)?;
        let result = (|| {
            let mut sealed = self.clone();
            sealed.refresh_private_table_metadata(final_state, work)?;
            sealed.finish_selection_inner(
                expected_visible_commit_epoch,
                recovery_source,
                overflow_root,
                None,
                DeltaSelection::PrivatePrefix,
            )
        })();
        // The cloned builder and its manifest are destroyed before refunding.
        drop(memory);
        if let Ok(report) = &result {
            self.private_prefix = Some((recovery_source, report.clone()));
        }
        result
    }

    // A private prefix preserves the base schema. Validate borrowed schemas
    // before updating counts in the admitted clone instead of cloning a second
    // vector of table names just to replace the first one.
    fn refresh_private_table_metadata(
        &mut self,
        state: &RelationalState,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<(), RelationalRowDeltaError> {
        let mut changed = state.table_schemas().count() != self.tables.len();
        for (schema, table) in state.table_schemas().zip(&self.tables) {
            let columns = u32::try_from(schema.columns.len()).map_err(|_| {
                RelationalRowDeltaError::Admission(format!(
                    "row delta table {} column count does not fit u32",
                    schema.name
                ))
            })?;
            let columns = NonZeroU32::new(columns).ok_or_else(|| {
                RelationalRowDeltaError::Corrupt(format!(
                    "row delta table {} has no columns",
                    schema.name
                ))
            })?;
            let digest =
                crate::relational::index_shadow::relational_schema_digest_with_work_context(
                    schema, work,
                )
                .map_err(|error| RelationalRowDeltaError::Corrupt(error.to_string()))?;
            changed |= table.table != schema.name
                || table.schema_digest != digest
                || table.column_count != columns;
        }
        if changed {
            return Err(RelationalRowDeltaError::RequiresCheckpoint {
                tables: state
                    .table_schemas()
                    .map(|schema| schema.name.clone())
                    .collect(),
            });
        }
        for table in &mut self.tables {
            table.row_count = state.row_count(&table.table) as u64;
        }
        Ok(())
    }

    fn finish_with_state_inner(
        mut self,
        expected_visible_commit_epoch: u64,
        recovery_source: RelationalRecoverySourceIdentity,
        overflow_root: Option<&RelationalOverflowRootReader>,
        final_state: &RelationalState,
        selection: DeltaSelection,
    ) -> Result<RelationalRowDeltaReport, RelationalRowDeltaError> {
        let final_tables = table_metadata_for_state(final_state)?;
        if final_tables.len() != self.tables.len()
            || final_tables
                .iter()
                .zip(&self.tables)
                .any(|(final_table, base_table)| {
                    final_table.table != base_table.table
                        || final_table.schema_digest != base_table.schema_digest
                        || final_table.column_count != base_table.column_count
                })
        {
            return Err(RelationalRowDeltaError::RequiresCheckpoint {
                tables: final_tables
                    .iter()
                    .map(|table| table.table.clone())
                    .collect(),
            });
        }
        self.tables = final_tables;
        if matches!(selection, DeltaSelection::Latest) {
            self.finish_inner(
                expected_visible_commit_epoch,
                recovery_source,
                overflow_root,
                None,
            )
        } else {
            self.finish_selection_inner(
                expected_visible_commit_epoch,
                recovery_source,
                overflow_root,
                None,
                selection,
            )
        }
    }

    pub(super) fn finish_inner(
        self,
        expected_visible_commit_epoch: u64,
        recovery_source: RelationalRecoverySourceIdentity,
        overflow_root: Option<&RelationalOverflowRootReader>,
        stop_after: Option<RelationalRowDeltaPublicationPhase>,
    ) -> Result<RelationalRowDeltaReport, RelationalRowDeltaError> {
        self.finish_selected_inner(
            expected_visible_commit_epoch,
            recovery_source,
            overflow_root,
            stop_after,
            true,
        )
    }

    pub(super) fn finish_selected_inner(
        self,
        expected_visible_commit_epoch: u64,
        recovery_source: RelationalRecoverySourceIdentity,
        overflow_root: Option<&RelationalOverflowRootReader>,
        stop_after: Option<RelationalRowDeltaPublicationPhase>,
        select_latest: bool,
    ) -> Result<RelationalRowDeltaReport, RelationalRowDeltaError> {
        self.finish_selection_inner(
            expected_visible_commit_epoch,
            recovery_source,
            overflow_root,
            stop_after,
            if select_latest {
                DeltaSelection::Latest
            } else {
                DeltaSelection::PrivateGeneration
            },
        )
    }

    fn finish_selection_inner(
        mut self,
        expected_visible_commit_epoch: u64,
        recovery_source: RelationalRecoverySourceIdentity,
        overflow_root: Option<&RelationalOverflowRootReader>,
        stop_after: Option<RelationalRowDeltaPublicationPhase>,
        selection: DeltaSelection,
    ) -> Result<RelationalRowDeltaReport, RelationalRowDeltaError> {
        let select_latest = matches!(selection, DeltaSelection::Latest);
        self.require_available()?;
        if self.visible_commit_epoch != expected_visible_commit_epoch {
            return Err(RelationalRowDeltaError::Corrupt(format!(
                "row delta consumed through epoch {}, expected {expected_visible_commit_epoch}",
                self.visible_commit_epoch
            )));
        }
        self.flush()?;
        maybe_stop(
            stop_after,
            RelationalRowDeltaPublicationPhase::CandidateRunsDurable,
        )?;
        let overflow_binding = overflow_root.map(|reader| reader.manifest().binding());
        if overflow_binding
            .is_some_and(|binding| binding.source_commit_epoch != expected_visible_commit_epoch)
        {
            return Err(RelationalRowDeltaError::Admission(format!(
                "row delta overflow root epoch does not match visible epoch {expected_visible_commit_epoch}"
            )));
        }
        let total_entries = self.runs.iter().try_fold(0u64, |entries, run| {
            entries.checked_add(run.entry_count as u64).ok_or_else(|| {
                RelationalRowDeltaError::Admission(
                    "row delta total entry count overflow".to_string(),
                )
            })
        })?;
        let manifest = RelationalRowDeltaManifest {
            base: self.base,
            delta_generation: self.delta_generation,
            visible_commit_epoch: expected_visible_commit_epoch,
            recovery_source,
            schema_set_digest: self.schema_set_digest,
            run_set_digest: codec::run_set_digest(&self.runs)?,
            overflow_root: overflow_binding,
            tables: self.tables,
            runs: self.runs,
            total_entries,
        };
        super::reader::validate_candidate_overflow_closure(
            &self.directory,
            &manifest,
            self.config,
            overflow_root,
        )?;
        let encoded_manifest = codec::encode_manifest(&manifest, self.config)?;
        let filename = if matches!(selection, DeltaSelection::PrivatePrefix) {
            super::relational_row_delta_prefix_file(
                manifest.base.generation,
                manifest.delta_generation,
                manifest.visible_commit_epoch,
            )
        } else {
            relational_row_delta_manifest_generation_file(
                manifest.base.generation,
                manifest.delta_generation,
            )
        };
        let generation_manifest = self.directory.join(filename);
        if select_latest {
            let generation_tmp = generation_manifest.with_extension("hawdb.tmp");
            remove_if_exists(&generation_tmp)?;
            write_synced(&generation_tmp, &encoded_manifest)?;
            durable_publish_immutable(&generation_tmp, &generation_manifest)?;
        } else {
            // This generation is unselected until its enclosing checkpoint
            // publishes. Exclusive creation protects both existing immutable
            // evidence and unrelated temporary files in this directory.
            let mut file = OpenOptions::new()
                .write(true)
                .create_new(true)
                .open(&generation_manifest)
                .map_err(durability("create private row recovery manifest"))?;
            file.write_all(&encoded_manifest)
                .map_err(durability("write private row recovery manifest"))?;
            file.sync_all()
                .map_err(durability("sync private row recovery manifest"))?;
            sync_directory(&self.directory)
                .map_err(durability("sync private row recovery directory"))?;
        }
        maybe_stop(
            stop_after,
            RelationalRowDeltaPublicationPhase::CandidateManifestDurable,
        )?;

        let _row_page_lock = super::super::publication::acquire_publication_lock(&self.directory)?;
        let actual_base = match self.row_root_selection {
            RowRootSelection::IndependentLatest => {
                RelationalRowPageRootReader::open_latest(&self.directory, self.row_page_config)?
                    .as_ref()
                    .map(RelationalRowDeltaBaseBinding::from_reader)
            }
            RowRootSelection::CanonicalCheckpoint => {
                Some(RelationalRowDeltaBaseBinding::from_reader(
                    &RelationalRowPageRootReader::open_generation(
                        &self.directory,
                        self.base.generation,
                        self.row_page_config,
                    )?,
                ))
            }
        };
        if actual_base != Some(self.base) {
            return Err(RelationalRowDeltaError::StaleBase {
                expected: self.base,
                actual: actual_base,
            });
        }
        let _delta_lock = acquire_publication_lock(&self.directory)?;
        let latest = self.directory.join(RELATIONAL_ROW_DELTA_MANIFEST_FILE);
        let actual_previous = codec::read_manifest_if_exists(&latest, self.config)?
            .map(|manifest| manifest.generation());
        if actual_previous != self.expected_previous {
            return Err(RelationalRowDeltaError::StaleGeneration {
                expected_previous: self.expected_previous,
                actual_previous,
            });
        }
        maybe_stop(
            stop_after,
            RelationalRowDeltaPublicationPhase::BaseRevalidated,
        )?;
        let mut events = COMPLETE_PUBLICATION_TRACE;
        if select_latest {
            let latest_tmp = latest.with_extension("hawdb.tmp");
            remove_if_exists(&latest_tmp)?;
            write_synced(&latest_tmp, &encoded_manifest)?;
            durable_replace_file(&latest_tmp, &latest)
                .map_err(durability("publish latest row delta manifest"))?;
        } else {
            events[3] = RelationalRowDeltaPublicationPhase::PrivateManifestPrepared;
        }

        Ok(RelationalRowDeltaReport {
            generation: manifest.generation(),
            base_commit_epoch: manifest.base.source_commit_epoch,
            visible_commit_epoch: manifest.visible_commit_epoch,
            runs: manifest.runs.len(),
            entries: manifest.total_entries,
            run_bytes: self.run_bytes,
            manifest_bytes: encoded_manifest.len() as u64,
            replayed_batches: self.replayed_batches,
            peak_dirty_entries: self.peak_dirty_entries,
            peak_dirty_bytes: self.peak_dirty_bytes,
            events,
        })
    }

    fn record_inner(
        &mut self,
        epoch: u64,
        capture: RelationalRowChangeCapture,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<(), RelationalRowDeltaError> {
        self.require_current_or_next_epoch(epoch)?;
        let changes = match capture {
            RelationalRowChangeCapture::Captured { changes, .. } => changes,
            RelationalRowChangeCapture::RequiresCheckpoint { tables } => {
                return Err(RelationalRowDeltaError::RequiresCheckpoint { tables });
            }
            RelationalRowChangeCapture::Invalidated { reason } => {
                return Err(RelationalRowDeltaError::Invalidated(reason));
            }
        };
        if epoch == self.base.source_commit_epoch && !changes.is_empty() {
            return Err(RelationalRowDeltaError::Corrupt(
                "row delta cannot apply changes at the immutable base epoch".to_string(),
            ));
        }
        if changes.len() > self.config.max_dirty_entries.get() {
            return Err(RelationalRowDeltaError::Admission(format!(
                "row delta capture contains {} changes, exceeding limit {}",
                changes.len(),
                self.config.max_dirty_entries
            )));
        }
        let advances_epoch = epoch != self.visible_commit_epoch;
        for change in changes {
            work.checkpoint()
                .map_err(super::super::checkpoint::work_error)?;
            let (key, value) = self.encode_change(epoch, change, work)?;
            self.insert_change(key, value, work)?;
        }
        if advances_epoch {
            self.replayed_batches = self.replayed_batches.checked_add(1).ok_or_else(|| {
                RelationalRowDeltaError::Admission(
                    "row delta replayed batch count overflow".to_string(),
                )
            })?;
        }
        self.visible_commit_epoch = epoch;
        Ok(())
    }

    fn encode_change(
        &self,
        epoch: u64,
        change: RelationalRowChange,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<(RowDeltaKey, RowDeltaValue), RelationalRowDeltaError> {
        let table_ordinal = self
            .tables
            .binary_search_by(|table| table.table.cmp(&change.table))
            .map_err(|_| {
                RelationalRowDeltaError::Corrupt(format!(
                    "row delta change references unknown table {}",
                    change.table
                ))
            })?;
        let table = &self.tables[table_ordinal];
        let key_allocations = crate::background::CheckpointDecodeContext {
            work: work.clone(),
            memory: Default::default(),
        };
        let map_key_error = |error| match error {
            crate::relational::ordered_key::CheckpointKeyEncodeError::Work(error) => {
                RelationalRowDeltaError::Admission(error.to_string())
            }
            crate::relational::ordered_key::CheckpointKeyEncodeError::Key(
                crate::relational::RelationalError::Corruption(message),
            ) => RelationalRowDeltaError::Corrupt(format!(
                "row delta primary key cannot be encoded: {message}"
            )),
            crate::relational::ordered_key::CheckpointKeyEncodeError::Key(error) => {
                RelationalRowDeltaError::Admission(error.to_string())
            }
        };
        let encoding =
            CheckpointKeyEncoding::new(&change.primary_key, &[], work).map_err(map_key_error)?;
        if encoding.encoded_len() > self.config.row_limits.max_key_bytes.get() {
            return Err(RelationalRowDeltaError::Admission(format!(
                "row delta key contains {} bytes, exceeding limit {}",
                encoding.encoded_len(),
                self.config.row_limits.max_key_bytes
            )));
        }
        let encoded_primary_key = encoding
            .encode_owned(&key_allocations)
            .map_err(map_key_error)?;
        key_allocations
            .reserve(std::mem::size_of::<Vec<u8>>() + 2 * std::mem::size_of::<usize>())
            .map_err(|error| RelationalRowDeltaError::Admission(error.to_string()))?;
        let encoded_primary_key = std::sync::Arc::new(encoded_primary_key);
        if encoded_primary_key.len() > self.config.row_limits.max_key_bytes.get() {
            return Err(RelationalRowDeltaError::Admission(format!(
                "row delta key contains {} bytes, exceeding limit {}",
                encoded_primary_key.len(),
                self.config.row_limits.max_key_bytes
            )));
        }
        let row_allocations = crate::background::CheckpointDecodeContext {
            work: work.clone(),
            memory: Default::default(),
        };
        let (is_present, encoded_row) = match change.row {
            Some(row) => (
                true,
                super::super::value::encode_row_with_allocations(
                    &row,
                    table.column_count.get() as usize,
                    self.config.row_limits,
                    Some(&row_allocations),
                )?,
            ),
            None => (false, Vec::new()),
        };
        let encoded_row = super::RowDeltaBytes::from_encoded(encoded_row, &row_allocations)?;
        let charged_bytes =
            codec::charged_entry_bytes(encoded_primary_key.len(), encoded_row.len())?;
        Ok((
            RowDeltaKey {
                table_ordinal: table_ordinal as u32,
                encoded_primary_key,
            },
            RowDeltaValue {
                is_present,
                encoded_row,
                last_modified_epoch: epoch,
                charged_bytes,
                _key_memory: key_allocations.memory.into_inner(),
            },
        ))
    }

    fn insert_change(
        &mut self,
        key: RowDeltaKey,
        value: RowDeltaValue,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<(), RelationalRowDeltaError> {
        let inserted_bytes = value.charged_bytes;
        if value.charged_bytes > self.config.max_dirty_bytes.get() {
            return Err(RelationalRowDeltaError::Admission(format!(
                "one row delta entry needs {} bytes, exceeding dirty limit {}",
                value.charged_bytes, self.config.max_dirty_bytes
            )));
        }
        let existing_bytes = self
            .dirty
            .get(&key)
            .map_or(0, |existing| existing.charged_bytes);
        let next_entries = self
            .dirty
            .len()
            .checked_add(usize::from(existing_bytes == 0))
            .ok_or_else(|| {
                RelationalRowDeltaError::Admission(
                    "row delta dirty entry accounting overflow".to_string(),
                )
            })?;
        let next_bytes = self
            .dirty_bytes
            .checked_sub(existing_bytes)
            .and_then(|bytes| bytes.checked_add(value.charged_bytes))
            .ok_or_else(|| {
                RelationalRowDeltaError::Admission(
                    "row delta dirty byte accounting overflow".to_string(),
                )
            })?;
        if !self.dirty.is_empty()
            && (next_entries > self.config.max_dirty_entries.get()
                || next_bytes > self.config.max_dirty_bytes.get())
        {
            self.flush_with_work_context(work)?;
        }
        let replaced = self.dirty.insert(key, value, work)?;
        if let Some(replaced) = replaced {
            self.dirty_bytes = self
                .dirty_bytes
                .checked_sub(replaced.charged_bytes)
                .ok_or_else(|| {
                    RelationalRowDeltaError::Corrupt(
                        "row delta replacement byte accounting underflow".to_string(),
                    )
                })?;
        }
        self.dirty_bytes = self
            .dirty_bytes
            .checked_add(inserted_bytes)
            .ok_or_else(|| {
                RelationalRowDeltaError::Admission(
                    "row delta dirty byte accounting overflow".to_string(),
                )
            })?;
        self.peak_dirty_entries = self.peak_dirty_entries.max(self.dirty.len());
        self.peak_dirty_bytes = self.peak_dirty_bytes.max(self.dirty_bytes);
        Ok(())
    }

    fn flush(&mut self) -> Result<(), RelationalRowDeltaError> {
        self.flush_with_work_context(&crate::background::CheckpointWorkContext::default())
    }

    fn flush_with_work_context(
        &mut self,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<(), RelationalRowDeltaError> {
        if self.dirty.is_empty() {
            return Ok(());
        }
        if self.runs.len() >= self.config.max_runs.get() {
            return Err(RelationalRowDeltaError::Admission(format!(
                "row delta needs more than {} immutable runs",
                self.config.max_runs
            )));
        }
        let ordinal = u32::try_from(self.runs.len()).map_err(|_| {
            RelationalRowDeltaError::Admission("row delta run ordinal does not fit u32".to_string())
        })?;
        let final_path = self.directory.join(relational_row_delta_run_file(
            self.base.generation,
            self.delta_generation,
            ordinal,
        ));
        if final_path.exists() {
            return Err(RelationalRowDeltaError::Admission(format!(
                "immutable row delta run {} already exists",
                final_path.display()
            )));
        }
        let tmp_path = final_path.with_extension("hawdb.tmp");
        remove_if_exists(&tmp_path)?;
        let estimated_run_bytes = codec::estimated_run_encoded_len(&self.dirty)?;
        let next_run_bytes = self
            .run_bytes
            .checked_add(estimated_run_bytes)
            .ok_or_else(|| {
                RelationalRowDeltaError::Admission(
                    "row delta run byte accounting overflow".to_string(),
                )
            })?;
        if next_run_bytes > self.config.max_run_bytes.get() {
            return Err(RelationalRowDeltaError::Admission(format!(
                "row delta runs require {next_run_bytes} bytes, exceeding limit {}",
                self.config.max_run_bytes
            )));
        }
        let (lower_key, _) = self
            .dirty
            .first_key_value()
            .expect("non-empty row delta run");
        let (upper_key, _) = self
            .dirty
            .last_key_value()
            .expect("non-empty row delta run");
        codec::ensure_next_run_manifest_capacity(
            &self.tables,
            &self.runs,
            lower_key.encoded_primary_key.len(),
            upper_key.encoded_primary_key.len(),
            self.config,
        )?;
        // All fallible admission precedes immutable publication. Reserve both
        // vector backings during replacement, freeing the old one before refund.
        if self.runs.len() == self.runs.capacity() {
            let capacity = self
                .runs
                .len()
                .checked_add(1)
                .ok_or_else(|| {
                    RelationalRowDeltaError::Admission("row run capacity overflow".into())
                })?
                .max(self.runs.capacity().saturating_mul(2))
                .min(self.config.max_runs.get());
            let bytes = capacity
                .checked_mul(std::mem::size_of::<RowDeltaRunDescriptor>())
                .ok_or_else(|| {
                    RelationalRowDeltaError::Admission("row run allocation size overflow".into())
                })?;
            let mut next_memory = crate::background::CheckpointAllocationOwner::default();
            next_memory
                .reserve(bytes, work)
                .map_err(|error| RelationalRowDeltaError::Admission(error.to_string()))?;
            let mut replacement = Vec::new();
            replacement.try_reserve_exact(capacity).map_err(|error| {
                let failure =
                    work.record_failure(crate::background::CheckpointWorkError::Allocation {
                        bytes: bytes as u64,
                        reason: error.to_string(),
                    });
                RelationalRowDeltaError::Admission(failure.to_string())
            })?;
            if replacement.capacity() != capacity {
                return Err(RelationalRowDeltaError::Admission(
                    "row run allocation exceeded admitted capacity".into(),
                ));
            }
            replacement.append(&mut self.runs);
            let old = std::mem::replace(&mut self.runs, replacement);
            drop(old);
            self._runs_memory = next_memory;
        }
        let mut bounds_memory = crate::background::CheckpointAllocationOwner::default();
        bounds_memory
            .reserve(lower_key.encoded_primary_key.len(), work)
            .map_err(|error| RelationalRowDeltaError::Admission(error.to_string()))?;
        bounds_memory
            .reserve(upper_key.encoded_primary_key.len(), work)
            .map_err(|error| RelationalRowDeltaError::Admission(error.to_string()))?;
        self._bounds_memory
            .append(&mut bounds_memory, work)
            .map_err(|error| RelationalRowDeltaError::Admission(error.to_string()))?;
        let lower_bound = super::RowDeltaBound {
            table_ordinal: lower_key.table_ordinal,
            encoded_primary_key: lower_key.encoded_primary_key.to_vec(),
        };
        let upper_bound = super::RowDeltaBound {
            table_ordinal: upper_key.table_ordinal,
            encoded_primary_key: upper_key.encoded_primary_key.to_vec(),
        };
        let descriptor = codec::write_run(
            &tmp_path,
            codec::RunWrite {
                base: self.base,
                delta_generation: self.delta_generation,
                schema_set_digest: self.schema_set_digest,
                ordinal,
                entries: &self.dirty,
                lower_bound,
                upper_bound,
            },
            self.config,
        )?;
        debug_assert_eq!(descriptor.encoded_len, estimated_run_bytes);
        durable_publish_immutable(&tmp_path, &final_path)?;
        self.runs.push(descriptor);
        self.run_bytes = next_run_bytes;
        self.dirty.clear();
        self.dirty_bytes = 0;
        Ok(())
    }

    fn require_available(&self) -> Result<(), RelationalRowDeltaError> {
        if let Some(reason) = &self.invalidated {
            return Err(RelationalRowDeltaError::Invalidated(reason.clone()));
        }
        Ok(())
    }

    fn require_current_or_next_epoch(&self, epoch: u64) -> Result<(), RelationalRowDeltaError> {
        if epoch == self.visible_commit_epoch {
            return Ok(());
        }
        let expected = self.visible_commit_epoch.checked_add(1).ok_or_else(|| {
            RelationalRowDeltaError::Corrupt("row delta visible commit epoch overflow".to_string())
        })?;
        if epoch != expected {
            return Err(RelationalRowDeltaError::Corrupt(format!(
                "row delta epoch gap: expected {expected}, found {epoch}"
            )));
        }
        Ok(())
    }
}

fn acquire_publication_lock(directory: &Path) -> Result<File, RelationalRowDeltaError> {
    let lock = OpenOptions::new()
        .create(true)
        .truncate(false)
        .read(true)
        .write(true)
        .open(directory.join(RELATIONAL_ROW_DELTA_PUBLICATION_LOCK_FILE))
        .map_err(durability("open row delta publication lock"))?;
    lock.lock()
        .map_err(durability("lock row delta publication"))?;
    Ok(lock)
}

fn write_synced(path: &Path, encoded: &[u8]) -> Result<(), RelationalRowDeltaError> {
    let mut file = File::create(path).map_err(durability("create row delta manifest candidate"))?;
    file.write_all(encoded)
        .map_err(durability("write row delta manifest candidate"))?;
    file.sync_all()
        .map_err(durability("sync row delta manifest candidate"))
}

fn durable_publish_immutable(
    source: &Path,
    destination: &Path,
) -> Result<(), RelationalRowDeltaError> {
    if destination.exists() {
        return Err(RelationalRowDeltaError::Admission(format!(
            "immutable row delta artifact {} already exists",
            destination.display()
        )));
    }
    durable_replace_file(source, destination)
        .map_err(durability("publish immutable row delta artifact"))
}

fn remove_if_exists(path: &Path) -> Result<(), RelationalRowDeltaError> {
    match fs::remove_file(path) {
        Ok(()) => sync_directory(path.parent().expect("row delta path has a parent")).map_err(
            durability("sync row delta directory after candidate cleanup"),
        ),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(durability("remove stale row delta candidate")(error)),
    }
}

fn maybe_stop(
    stop_after: Option<RelationalRowDeltaPublicationPhase>,
    phase: RelationalRowDeltaPublicationPhase,
) -> Result<(), RelationalRowDeltaError> {
    if stop_after == Some(phase) {
        return Err(RelationalRowDeltaError::Durability(format!(
            "injected stop after {phase:?}"
        )));
    }
    Ok(())
}

fn next_recovery_delta_generation(
    base: &RelationalRowPageRootReader,
    expected_previous: Option<RelationalRowDeltaGeneration>,
) -> Result<u64, RelationalRowDeltaError> {
    let minimum = expected_previous
        .filter(|previous| previous.base_generation == base.manifest().generation)
        .map_or(Ok(1), |previous| {
            previous.delta_generation.checked_add(1).ok_or_else(|| {
                RelationalRowDeltaError::Admission(
                    "row delta generation space is exhausted".to_string(),
                )
            })
        })?;
    Ok(next_row_delta_generation().max(minimum))
}

fn next_row_delta_generation() -> u64 {
    let clock = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos() as u64;
    let sequence = NEXT_ROW_DELTA_GENERATION.fetch_add(1, Ordering::Relaxed);
    let generation = clock.rotate_left(17) ^ sequence ^ ((std::process::id() as u64) << 32);
    generation.max(1)
}

fn table_metadata_for_state(
    state: &RelationalState,
) -> Result<Vec<RelationalRowDeltaTableMetadata>, RelationalRowDeltaError> {
    state
        .table_schemas()
        .map(|schema| {
            let column_count = u32::try_from(schema.columns.len()).map_err(|_| {
                RelationalRowDeltaError::Admission(format!(
                    "row delta table {} column count does not fit u32",
                    schema.name
                ))
            })?;
            let column_count = NonZeroU32::new(column_count).ok_or_else(|| {
                RelationalRowDeltaError::Corrupt(format!(
                    "row delta table {} has no columns",
                    schema.name
                ))
            })?;
            let schema_digest = state
                .table_schema_digest(&schema.name)
                .map_err(|error| RelationalRowDeltaError::Corrupt(error.to_string()))?
                .ok_or_else(|| {
                    RelationalRowDeltaError::Corrupt(format!(
                        "row delta table {} disappeared while deriving its schema fence",
                        schema.name
                    ))
                })?;
            Ok(RelationalRowDeltaTableMetadata {
                table: schema.name.clone(),
                schema_digest,
                column_count,
                row_count: u64::try_from(state.row_count(&schema.name)).map_err(|_| {
                    RelationalRowDeltaError::Admission(format!(
                        "row delta table {} row count does not fit u64",
                        schema.name
                    ))
                })?,
            })
        })
        .collect()
}

#[cfg(test)]
mod lock_tests {
    use super::*;

    #[test]
    fn publication_lock_contract() {
        crate::file_lock_tests::assert_contract(
            RELATIONAL_ROW_DELTA_PUBLICATION_LOCK_FILE,
            acquire_publication_lock,
            "open row delta publication lock",
        );
    }

    #[test]
    #[ignore = "deterministic local publication lock campaign"]
    fn publication_lock_state_machine_campaign() {
        crate::file_lock_tests::assert_state_machine(
            RELATIONAL_ROW_DELTA_PUBLICATION_LOCK_FILE,
            acquire_publication_lock,
        );
    }
}