icydb-core 0.214.1

//! Module: executor::mutation::commit_window
//! Responsibility: commit-window open/apply orchestration for prepared row ops.
//! Does not own: save/delete logical planning or relation policy decisions.
//! Boundary: shared commit marker and prepared-op apply pipeline for mutations.

use crate::{
    db::journal::{JournalBatch, JournalRecord},
    db::{
        Db,
        commit::{
            CommitApplyGuard, CommitGuard, CommitMarker, CommitRowOp, PreparedIndexMutation,
            PreparedRowCommitOp, begin_commit, database_incarnation_id, finish_commit,
            generate_commit_id, generate_marker_batch_id, prepare_row_commit_with_context,
            rollback_prepared_row_ops_reverse,
        },
        data::{DecodedDataStoreKey, RawDataStoreKey, RawRow},
        direction::Direction,
        index::{
            IndexEntryValue, IndexReadContract, IndexStore, RawIndexStoreKey,
            StructuralIndexEntryReader, StructuralPrimaryRowReader, key_within_envelope,
        },
        key_taxonomy::PrimaryKeyValue,
        registry::{
            StoreCommitParticipation, StoreHandle, StoreRecoveryCapability,
            StoreSchemaMetadataCapability,
        },
        schema::{
            IdentityAdvanceId, IdentityRangeAdvance, apply_live_identity_range_checkpoint,
            preflight_live_identity_range_checkpoint,
        },
    },
    error::InternalError,
    metrics::sink::{MetricsEvent, MutationCommitClass, record},
    traits::CanisterKind,
};
use std::{
    cell::RefCell,
    collections::{BTreeMap, HashMap},
    ops::Bound,
    ptr,
    thread::LocalKey,
};

use super::constraint_scheduler::AcceptedMutationConstraintBatch;

const MUTATION_COMMIT_INITIAL_RESERVE_ROWS: usize = 64;

/// Test-only durable interruption boundaries around Identity publication.
#[cfg(test)]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db) enum IdentityCommitInterruption {
    MarkerPersisted,
    JournalPublished,
    RowsPublished,
    StateMaterialized,
}

#[cfg(test)]
thread_local! {
    static NEXT_IDENTITY_COMMIT_INTERRUPTION: std::cell::Cell<Option<IdentityCommitInterruption>> =
        const { std::cell::Cell::new(None) };
}

#[cfg(test)]
pub(in crate::db) fn interrupt_next_identity_commit_for_tests(
    interruption: IdentityCommitInterruption,
) {
    NEXT_IDENTITY_COMMIT_INTERRUPTION.with(|next| next.set(Some(interruption)));
}

#[cfg(test)]
fn take_identity_commit_interruption(
    interruption: IdentityCommitInterruption,
    has_identity_ranges: bool,
) -> bool {
    has_identity_ranges
        && NEXT_IDENTITY_COMMIT_INTERRUPTION.with(|next| {
            if next.get() == Some(interruption) {
                next.set(None);
                true
            } else {
                false
            }
        })
}

///
/// PreparedRowOpDelta
///
/// Aggregated mutation deltas from preflight-prepared row operations.
/// Used by save/delete executors to emit consistent metrics without duplicating
/// per-field folding logic.
///

pub(in crate::db::executor) struct PreparedRowOpDelta {
    pub(in crate::db::executor) index_inserts: usize,
    pub(in crate::db::executor) index_removes: usize,
    pub(in crate::db::executor) reverse_index_inserts: usize,
    pub(in crate::db::executor) reverse_index_removes: usize,
}

impl PreparedRowOpDelta {
    // Construct one zeroed delta accumulator.
    const fn zero() -> Self {
        Self {
            index_inserts: 0,
            index_removes: 0,
            reverse_index_inserts: 0,
            reverse_index_removes: 0,
        }
    }

    // Project one delete-only metric view that suppresses insert counts.
    const fn delete_only(&self) -> Self {
        Self {
            index_inserts: 0,
            index_removes: self.index_removes,
            reverse_index_inserts: 0,
            reverse_index_removes: self.reverse_index_removes,
        }
    }
}

///
/// OpenCommitWindow
///
/// Commit-window staging bundle shared across save/delete executors.
/// Contains the persisted commit guard, preflight-prepared row ops, and
/// precomputed delta counters.
///

pub(in crate::db::executor) struct OpenCommitWindow {
    pub(in crate::db::executor) commit: CommitGuard,
    pub(in crate::db::executor) prepared_row_ops: Vec<PreparedRowCommitOp>,
    effects: PreparedCommitEffects,
    pub(in crate::db::executor) index_store_guards: Vec<IndexStoreGenerationGuard>,
    pub(in crate::db::executor) delta: PreparedRowOpDelta,
    pub(in crate::db::executor) commit_class: MutationCommitClass,
}

#[derive(Clone)]
pub(in crate::db::executor) struct PreparedJournalAppend {
    journal_store: &'static LocalKey<RefCell<crate::db::journal::JournalTailStore>>,
    batch: JournalBatch,
}

struct CommitWindowPayload {
    marker: CommitMarker,
    effects: PreparedCommitEffects,
}

struct PreparedCommitEffects {
    journal_appends: Vec<PreparedJournalAppend>,
    identity_range_applies: Vec<PreparedIdentityRangeApply>,
}

#[derive(Clone, Copy)]
struct PreparedIdentityRangeApply {
    store_path: &'static str,
    handle: StoreHandle,
    range: IdentityRangeAdvance,
    advance_id: IdentityAdvanceId,
}

///
/// IndexStoreGenerationGuard
///
/// Snapshot of one index store generation captured after preflight.
/// Apply must observe the same generation before it starts mutating state.
///

pub(in crate::db::executor) struct IndexStoreGenerationGuard {
    index_store: &'static LocalKey<RefCell<IndexStore>>,
    expected_generation: u64,
}

impl IndexStoreGenerationGuard {
    // Capture one index store generation at preflight time.
    fn capture(index_store: &'static LocalKey<RefCell<IndexStore>>) -> Self {
        Self {
            index_store,
            expected_generation: index_store.with_borrow(IndexStore::generation),
        }
    }

    // Verify one touched index store still matches its preflight generation.
    fn verify(&self) -> Result<(), InternalError> {
        let observed_generation = self.index_store.with_borrow(IndexStore::generation);
        if observed_generation != self.expected_generation {
            return Err(InternalError::mutation_index_store_generation_changed(
                self.expected_generation,
                observed_generation,
            ));
        }

        Ok(())
    }
}

///
/// PreparedRowOpBatch
///
/// Streaming preflight output for one commit window.
/// The batch keeps prepared row operations, generation guards, and delta
/// metrics together so preflight can produce all apply metadata in one pass.
///

struct PreparedRowOpBatch {
    prepared_row_ops: Vec<PreparedRowCommitOp>,
    index_store_guards: Vec<IndexStoreGenerationGuard>,
    delta: PreparedRowOpDelta,
}

impl PreparedRowOpBatch {
    // Allocate one preflight batch with enough room for the expected row count.
    fn with_row_capacity(row_count: usize) -> Self {
        let reserve_rows = row_count.min(MUTATION_COMMIT_INITIAL_RESERVE_ROWS);

        Self {
            prepared_row_ops: Vec::with_capacity(reserve_rows),
            index_store_guards: Vec::new(),
            delta: PreparedRowOpDelta::zero(),
        }
    }

    // Add one prepared row op and update all derived apply metadata immediately.
    fn push(&mut self, row_op: PreparedRowCommitOp) {
        for index_op in &row_op.index_ops {
            record_prepared_index_delta(&mut self.delta, index_op);
            record_index_store_generation_guard(&mut self.index_store_guards, index_op.index_store);
        }

        self.prepared_row_ops.push(row_op);
    }
}

///
/// SingleRowIndexStoreGuards
///
/// One-row index-store generation snapshot optimized for the hot 0/1-store
/// case. This preserves the same preflight/apply invariant as the batch guard
/// path without forcing heap allocation when a row touches no indexes or only
/// one unique index store.
///

enum SingleRowIndexStoreGuards {
    Empty,
    One(IndexStoreGenerationGuard),
    Many(Vec<IndexStoreGenerationGuard>),
}

impl SingleRowIndexStoreGuards {
    // Record one unique touched index store under the single-row guard shape.
    fn record(&mut self, index_store: &'static LocalKey<RefCell<IndexStore>>) {
        match self {
            Self::Empty => {
                *self = Self::One(IndexStoreGenerationGuard::capture(index_store));
            }
            Self::One(existing) => {
                if ptr::eq(existing.index_store, index_store) {
                    return;
                }

                let first = IndexStoreGenerationGuard::capture(existing.index_store);
                let second = IndexStoreGenerationGuard::capture(index_store);
                *self = Self::Many(vec![first, second]);
            }
            Self::Many(guards) => {
                if guards
                    .iter()
                    .any(|existing| ptr::eq(existing.index_store, index_store))
                {
                    return;
                }

                guards.push(IndexStoreGenerationGuard::capture(index_store));
            }
        }
    }

    // Verify every captured index store still matches its preflight generation.
    fn verify(&self) -> Result<(), InternalError> {
        match self {
            Self::Empty => Ok(()),
            Self::One(guard) => guard.verify(),
            Self::Many(guards) => {
                for guard in guards {
                    guard.verify()?;
                }

                Ok(())
            }
        }
    }
}

///
/// SingleRowApplyPrep
///
/// Single-row preapply bundle derived from one prepared row operation.
/// This keeps delta aggregation and index-generation capture in one scan so the
/// hot one-row save/delete lanes do not rewalk `index_ops` separately.
///

struct SingleRowApplyPrep {
    guards: SingleRowIndexStoreGuards,
    delta: PreparedRowOpDelta,
}

///
/// PreflightStoreOverlay
///
/// In-memory simulation overlay for commit-window preflight.
/// Data reads first consult the complete final-row batch overlay. Index reads
/// consult staged earlier row operations before falling back to committed
/// stores.
///

struct PreflightStoreOverlay<'a, C: CanisterKind> {
    db: &'a Db<C>,
    data_overrides: HashMap<RawDataStoreKey, Option<RawRow>>,
    index_overrides: HashMap<usize, HashMap<RawIndexStoreKey, Option<IndexEntryValue>>>,
}

impl<'a, C: CanisterKind> PreflightStoreOverlay<'a, C> {
    /// Construct one preflight overlay with every submitted final data image
    /// visible before storage-backed constraint scheduling begins.
    fn from_row_ops(db: &'a Db<C>, row_ops: &[CommitRowOp]) -> Result<Self, InternalError> {
        let reserve_rows = row_ops.len().min(MUTATION_COMMIT_INITIAL_RESERVE_ROWS);
        let mut overlay = Self {
            db,
            data_overrides: HashMap::with_capacity(reserve_rows),
            index_overrides: HashMap::with_capacity(reserve_rows),
        };
        for row_op in row_ops {
            let after = row_op
                .after
                .as_ref()
                .map(|bytes| RawRow::from_untrusted_bytes(bytes.clone()))
                .transpose()?;
            if overlay
                .data_overrides
                .insert(row_op.key.clone(), after)
                .is_some()
            {
                return Err(InternalError::query_executor_invariant());
            }
        }
        Ok(overlay)
    }

    // Stage one prepared row-op into overlay data/index maps.
    fn stage_prepared_row_op(&mut self, row_op: &PreparedRowCommitOp) {
        for index_op in &row_op.index_ops {
            let store_id = index_store_id(index_op.index_store);
            self.index_overrides
                .entry(store_id)
                .or_default()
                .insert(index_op.key.clone(), index_op.value.clone());
        }
        self.data_overrides.insert(
            row_op.data_key.clone(),
            row_op
                .data_value
                .as_ref()
                .map(|row| row.as_raw_row().clone()),
        );
    }
}

impl<C: CanisterKind> StructuralPrimaryRowReader for PreflightStoreOverlay<'_, C> {
    fn read_primary_row(&self, key: &DecodedDataStoreKey) -> Result<Option<RawRow>, InternalError> {
        let raw_key = key.to_raw()?;
        if let Some(override_row) = self.data_overrides.get(&raw_key) {
            return Ok(override_row.clone());
        }

        let runtime_entity = self.db.accepted_runtime_entity_for_tag(key.entity_tag())?;
        let store = self.db.recovered_store(runtime_entity.store_path())?;

        Ok(store.with_data(|data_store| data_store.get(&raw_key)))
    }

    fn has_primary_row_override(&self, key: &DecodedDataStoreKey) -> Result<bool, InternalError> {
        Ok(self.data_overrides.contains_key(&key.to_raw()?))
    }
}

impl<C: CanisterKind> StructuralIndexEntryReader for PreflightStoreOverlay<'_, C> {
    fn read_index_entry(
        &self,
        index_store: &'static LocalKey<RefCell<IndexStore>>,
        key: &RawIndexStoreKey,
    ) -> Result<Option<IndexEntryValue>, InternalError> {
        let store_id = index_store_id(index_store);
        if let Some(store_overrides) = self.index_overrides.get(&store_id)
            && let Some(override_entry) = store_overrides.get(key)
        {
            return Ok(override_entry.clone());
        }

        Ok(index_store.with_borrow(|store| store.get(key)))
    }

    fn read_index_keys_in_raw_range(
        &self,
        entity_path: &str,
        _entity_tag: crate::types::EntityTag,
        index_store: &'static LocalKey<RefCell<IndexStore>>,
        index: IndexReadContract<'_>,
        bounds: (&Bound<RawIndexStoreKey>, &Bound<RawIndexStoreKey>),
        limit: usize,
    ) -> Result<Vec<PrimaryKeyValue>, InternalError> {
        // Phase 1: untouched stores can use the canonical index-store range
        // reader directly instead of materializing one merged entry map first.
        let store_id = index_store_id(index_store);
        let Some(store_overrides) = self.index_overrides.get(&store_id) else {
            let mut out = Vec::with_capacity(limit.min(32));
            index_store.with_borrow(|store| {
                store.visit_raw_entries_in_range(bounds, Direction::Asc, |raw_key, raw_entry| {
                    push_index_entry_primary_key_values(
                        index,
                        raw_key,
                        raw_entry,
                        &mut out,
                        limit,
                        entity_path,
                    )
                })
            })?;

            return Ok(out);
        };

        // Phase 2: staged stores stream a sorted merge of committed entries and
        // staged overrides. Only the small override set is sorted here; the
        // committed store still streams through its canonical range visitor.
        let mut out = Vec::new();
        let bounded_overrides = store_overrides
            .iter()
            .filter(|(raw_key, _)| key_within_bounds(raw_key, bounds))
            .collect::<BTreeMap<&RawIndexStoreKey, &Option<IndexEntryValue>>>();
        let mut overrides = bounded_overrides.into_iter().peekable();
        let mut limit_reached = false;

        index_store.with_borrow(|index_store| {
            index_store.visit_raw_entries_in_range(bounds, Direction::Asc, |raw_key, raw_entry| {
                while let Some((override_key, _)) = overrides.peek() {
                    match (*override_key).cmp(raw_key) {
                        std::cmp::Ordering::Less => {
                            let override_key = (*override_key).clone();
                            let Some((_, override_entry)) = overrides.next() else {
                                return Err(InternalError::query_executor_invariant());
                            };
                            if push_optional_index_entry_primary_key_values(
                                index,
                                &override_key,
                                override_entry.as_ref(),
                                &mut out,
                                limit,
                                entity_path,
                            )? {
                                limit_reached = true;
                                return Ok(true);
                            }
                        }
                        std::cmp::Ordering::Equal => {
                            let override_key = (*override_key).clone();
                            let Some((_, override_entry)) = overrides.next() else {
                                return Err(InternalError::query_executor_invariant());
                            };
                            if push_optional_index_entry_primary_key_values(
                                index,
                                &override_key,
                                override_entry.as_ref(),
                                &mut out,
                                limit,
                                entity_path,
                            )? {
                                limit_reached = true;
                                return Ok(true);
                            }

                            return Ok(false);
                        }
                        std::cmp::Ordering::Greater => break,
                    }
                }

                if push_index_entry_primary_key_values(
                    index,
                    raw_key,
                    raw_entry,
                    &mut out,
                    limit,
                    entity_path,
                )? {
                    limit_reached = true;
                    return Ok(true);
                }

                Ok(false)
            })
        })?;

        if !limit_reached {
            for (override_key, override_entry) in overrides {
                if push_optional_index_entry_primary_key_values(
                    index,
                    override_key,
                    override_entry.as_ref(),
                    &mut out,
                    limit,
                    entity_path,
                )? {
                    break;
                }
            }
        }

        Ok(out)
    }
}

fn push_optional_index_entry_primary_key_values(
    index: IndexReadContract<'_>,
    raw_key: &RawIndexStoreKey,
    raw_entry: Option<&IndexEntryValue>,
    out: &mut Vec<PrimaryKeyValue>,
    limit: usize,
    entity_path: &str,
) -> Result<bool, InternalError> {
    let Some(raw_entry) = raw_entry else {
        return Ok(false);
    };

    push_index_entry_primary_key_values(index, raw_key, raw_entry, out, limit, entity_path)
}

// Decode one raw index entry into structural primary-key values under the
// preflight overlay's corruption mapping.
fn push_index_entry_primary_key_values(
    _index: IndexReadContract<'_>,
    raw_key: &RawIndexStoreKey,
    raw_entry: &IndexEntryValue,
    out: &mut Vec<PrimaryKeyValue>,
    limit: usize,
    _entity_path: &str,
) -> Result<bool, InternalError> {
    raw_entry.push_row_identity_primary_key_values_limited(raw_key, out, limit, |_err| {
        InternalError::index_plan_index_corruption()
    })
}

// Fold one prepared index mutation into saturated commit-window counters.
const fn record_prepared_index_delta(
    summary: &mut PreparedRowOpDelta,
    index_op: &PreparedIndexMutation,
) {
    let (index_inserts, index_removes, reverse_index_inserts, reverse_index_removes) =
        index_op.counter_increments();

    summary.index_inserts = summary.index_inserts.saturating_add(index_inserts);
    summary.index_removes = summary.index_removes.saturating_add(index_removes);
    summary.reverse_index_inserts = summary
        .reverse_index_inserts
        .saturating_add(reverse_index_inserts);
    summary.reverse_index_removes = summary
        .reverse_index_removes
        .saturating_add(reverse_index_removes);
}

// Capture one unique index-store guard while preflight streams prepared row
// operations. This replaces the old post-preflight guard collection pass.
fn record_index_store_generation_guard(
    guards: &mut Vec<IndexStoreGenerationGuard>,
    index_store: &'static LocalKey<RefCell<IndexStore>>,
) {
    if guards
        .iter()
        .any(|existing| ptr::eq(existing.index_store, index_store))
    {
        return;
    }

    guards.push(IndexStoreGenerationGuard::capture(index_store));
}

// Structural preflight variant used by nongeneric delete paths. It shares the
// same streaming batch accumulator as the typed save/delete path.
fn preflight_prepare_row_op_batch_structural<C: CanisterKind>(
    db: &Db<C>,
    row_ops: &[CommitRowOp],
) -> Result<PreparedRowOpBatch, InternalError> {
    let Some(first_row_op) = row_ops.first() else {
        return Ok(PreparedRowOpBatch::with_row_capacity(0));
    };
    let runtime_entity = db.accepted_runtime_entity_for_path(first_row_op.entity_path.as_ref())?;
    let context = runtime_entity.prepare_commit_context(
        db,
        first_row_op.schema_fingerprint,
        crate::db::commit::CommitPrepareMode::NormalWrite,
    )?;

    // The structural accepted-entity path can also bypass overlay simulation for
    // one-row commits because there is no staged cross-row state to read.
    if let [row_op] = row_ops {
        let store = db.store_handle(runtime_entity.store_path())?;
        return prepare_row_commit_with_context(db, row_op, &context, db, &store).map(|prepared| {
            let mut batch = PreparedRowOpBatch::with_row_capacity(1);
            batch.push(prepared);
            batch
        });
    }

    preflight_prepare_row_ops_with_overlay(db, row_ops, |overlay, row_op| {
        prepare_row_commit_with_context(db, row_op, &context, overlay, overlay)
    })
}

// Run the shared overlay-driven preflight sequence for multi-row commit
// windows, leaving only per-row preparation policy to the caller.
fn preflight_prepare_row_ops_with_overlay<C: CanisterKind>(
    db: &Db<C>,
    row_ops: &[CommitRowOp],
    mut prepare_one: impl FnMut(
        &PreflightStoreOverlay<'_, C>,
        &CommitRowOp,
    ) -> Result<PreparedRowCommitOp, InternalError>,
) -> Result<PreparedRowOpBatch, InternalError> {
    let mut batch = PreparedRowOpBatch::with_row_capacity(row_ops.len());
    let mut overlay = PreflightStoreOverlay::<C>::from_row_ops(db, row_ops)?;

    for row_op in row_ops {
        let row = prepare_one(&overlay, row_op)?;
        overlay.stage_prepared_row_op(&row);
        batch.push(row);
    }

    Ok(batch)
}

/// Preflight row ops, build marker, and persist the nongeneric delete commit window.
pub(in crate::db::executor) fn open_commit_window_structural<C: CanisterKind>(
    db: &Db<C>,
    row_ops: Vec<CommitRowOp>,
    identity_ranges: Vec<IdentityRangeAdvance>,
) -> Result<OpenCommitWindow, InternalError> {
    let PreparedRowOpBatch {
        prepared_row_ops,
        index_store_guards,
        delta,
    } = preflight_prepare_row_op_batch_structural(db, &row_ops)?;
    let affected_store_handles = affected_store_handles_for_prepared_row_ops(db, &prepared_row_ops);
    let commit_class = classify_mutation_commit_plan(affected_store_handles.as_slice());
    let CommitWindowPayload { marker, effects } = commit_window_payload_for_prepared_row_ops(
        db,
        &row_ops,
        &prepared_row_ops,
        identity_ranges.as_slice(),
    )?;
    preflight_identity_range_applies(effects.identity_range_applies.as_slice())?;
    let commit = begin_commit_window_payload(marker)?;

    Ok(OpenCommitWindow {
        commit,
        prepared_row_ops,
        effects,
        index_store_guards,
        delta,
        commit_class,
    })
}

/// Apply prepared row ops under the shared commit-window guard.
fn apply_prepared_row_ops(
    commit: CommitGuard,
    apply_phase: &'static str,
    prepared_row_ops: Vec<PreparedRowCommitOp>,
    effects: PreparedCommitEffects,
    index_store_guards: Vec<IndexStoreGenerationGuard>,
    on_index_applied: impl FnOnce(),
    on_data_applied: impl FnOnce(),
) -> Result<(), InternalError> {
    finish_commit(commit, |guard| {
        let mut apply_guard = CommitApplyGuard::new(apply_phase);
        let _ = guard;
        #[cfg(test)]
        let has_identity_ranges = !effects.identity_range_applies.is_empty();

        // Enforce that index stores are unchanged between preflight and apply.
        for index_store_guard in &index_store_guards {
            index_store_guard.verify()?;
        }
        #[cfg(test)]
        if take_identity_commit_interruption(
            IdentityCommitInterruption::MarkerPersisted,
            has_identity_ranges,
        ) {
            std::mem::forget(apply_guard);
            return Err(InternalError::executor_invariant());
        }
        append_prepared_journal_batches(&effects.journal_appends)?;
        #[cfg(test)]
        if take_identity_commit_interruption(
            IdentityCommitInterruption::JournalPublished,
            has_identity_ranges,
        ) {
            std::mem::forget(apply_guard);
            return Err(InternalError::executor_invariant());
        }

        // Single-row writes dominate the hot write lanes, so avoid the extra
        // rollback vector and reverse-apply scaffolding when only one prepared
        // row op remains.
        if prepared_row_ops.len() == 1 {
            let mut prepared_iter = prepared_row_ops.into_iter();
            let Some(row_op) = prepared_iter.next() else {
                return Err(InternalError::query_executor_invariant());
            };
            apply_guard.record_single_row_rollback(row_op.snapshot_rollback());

            row_op.apply();
            #[cfg(test)]
            if take_identity_commit_interruption(
                IdentityCommitInterruption::RowsPublished,
                has_identity_ranges,
            ) {
                std::mem::forget(apply_guard);
                return Err(InternalError::executor_invariant());
            }
            apply_identity_range_applies(effects.identity_range_applies.as_slice())?;
            #[cfg(test)]
            if take_identity_commit_interruption(
                IdentityCommitInterruption::StateMaterialized,
                has_identity_ranges,
            ) {
                std::mem::forget(apply_guard);
                return Err(InternalError::executor_invariant());
            }
            on_index_applied();
            on_data_applied();
            apply_guard.finish()?;

            return Ok(());
        }

        let mut rollback = Vec::with_capacity(prepared_row_ops.len());
        for row_op in &prepared_row_ops {
            rollback.push(row_op.snapshot_rollback());
        }
        apply_guard.record_rollback(move || rollback_prepared_row_ops_reverse(rollback));

        for row_op in prepared_row_ops {
            row_op.apply();
        }
        #[cfg(test)]
        if take_identity_commit_interruption(
            IdentityCommitInterruption::RowsPublished,
            has_identity_ranges,
        ) {
            std::mem::forget(apply_guard);
            return Err(InternalError::executor_invariant());
        }
        apply_identity_range_applies(effects.identity_range_applies.as_slice())?;
        #[cfg(test)]
        if take_identity_commit_interruption(
            IdentityCommitInterruption::StateMaterialized,
            has_identity_ranges,
        ) {
            std::mem::forget(apply_guard);
            return Err(InternalError::executor_invariant());
        }
        on_index_applied();
        on_data_applied();
        apply_guard.finish()?;

        Ok(())
    })
}

// Apply one prepared row op under the shared commit-window guard without
// routing through the multi-row vector machinery.
fn apply_prepared_single_row_op(
    commit: CommitGuard,
    apply_phase: &'static str,
    prepared_row_op: PreparedRowCommitOp,
    effects: PreparedCommitEffects,
    index_store_guards: SingleRowIndexStoreGuards,
    on_index_applied: impl FnOnce(),
    on_data_applied: impl FnOnce(),
) -> Result<(), InternalError> {
    finish_commit(commit, |guard| {
        let mut apply_guard = CommitApplyGuard::new(apply_phase);
        let _ = guard;
        #[cfg(test)]
        let has_identity_ranges = !effects.identity_range_applies.is_empty();

        // Enforce that index stores are unchanged between preflight and apply.
        index_store_guards.verify()?;
        #[cfg(test)]
        if take_identity_commit_interruption(
            IdentityCommitInterruption::MarkerPersisted,
            has_identity_ranges,
        ) {
            std::mem::forget(apply_guard);
            return Err(InternalError::executor_invariant());
        }
        append_prepared_journal_batches(&effects.journal_appends)?;
        #[cfg(test)]
        if take_identity_commit_interruption(
            IdentityCommitInterruption::JournalPublished,
            has_identity_ranges,
        ) {
            std::mem::forget(apply_guard);
            return Err(InternalError::executor_invariant());
        }

        apply_guard.record_single_row_rollback(prepared_row_op.snapshot_rollback());

        prepared_row_op.apply();
        #[cfg(test)]
        if take_identity_commit_interruption(
            IdentityCommitInterruption::RowsPublished,
            has_identity_ranges,
        ) {
            std::mem::forget(apply_guard);
            return Err(InternalError::executor_invariant());
        }
        apply_identity_range_applies(effects.identity_range_applies.as_slice())?;
        #[cfg(test)]
        if take_identity_commit_interruption(
            IdentityCommitInterruption::StateMaterialized,
            has_identity_ranges,
        ) {
            std::mem::forget(apply_guard);
            return Err(InternalError::executor_invariant());
        }
        on_index_applied();
        on_data_applied();
        apply_guard.finish()?;

        Ok(())
    })
}

/// Commit delete-mode row operations through one nongeneric commit window.
pub(in crate::db) fn commit_delete_row_ops_with_window_for_path<C: CanisterKind>(
    db: &Db<C>,
    entity_path: &str,
    batch: AcceptedMutationConstraintBatch,
    apply_phase: &'static str,
) -> Result<(), InternalError> {
    let row_ops = batch.into_delete_rows()?;
    if row_ops.len() == 1 {
        let Some(row_op) = row_ops.into_iter().next() else {
            return Err(InternalError::query_executor_invariant());
        };

        return commit_single_delete_row_op_with_window_for_path(
            db,
            entity_path,
            row_op,
            apply_phase,
        );
    }

    let OpenCommitWindow {
        commit,
        prepared_row_ops,
        effects,
        index_store_guards,
        delta,
        commit_class,
    } = open_commit_window_structural(db, row_ops, Vec::new())?;
    record_mutation_commit_plan(entity_path, commit_class);
    let synchronized_store_handles =
        synchronized_store_handles_for_prepared_row_ops(db, prepared_row_ops.as_slice());

    apply_prepared_row_ops(
        commit,
        apply_phase,
        prepared_row_ops,
        effects,
        index_store_guards,
        || emit_delete_index_delta_metrics_for_path(entity_path, &delta),
        || {},
    )?;
    mark_store_handles_index_ready(synchronized_store_handles.as_slice());
    Ok(())
}

/// Commit save-mode row operations through the nongeneric structural window.
pub(in crate::db) fn commit_structural_save_row_ops_with_window_for_path<C: CanisterKind>(
    db: &Db<C>,
    entity_path: &str,
    batch: AcceptedMutationConstraintBatch,
    identity_ranges: Vec<IdentityRangeAdvance>,
    apply_phase: &'static str,
) -> Result<(), InternalError> {
    let row_ops = batch.into_save_rows()?;
    let OpenCommitWindow {
        commit,
        prepared_row_ops,
        effects,
        index_store_guards,
        delta,
        commit_class,
    } = open_commit_window_structural(db, row_ops, identity_ranges)?;
    record_mutation_commit_plan(entity_path, commit_class);
    let synchronized_store_handles =
        synchronized_store_handles_for_prepared_row_ops(db, prepared_row_ops.as_slice());

    apply_prepared_row_ops(
        commit,
        apply_phase,
        prepared_row_ops,
        effects,
        index_store_guards,
        || emit_index_delta_metrics_for_path(entity_path, &delta),
        || {},
    )?;
    mark_store_handles_index_ready(synchronized_store_handles.as_slice());
    Ok(())
}
// Commit one already-prepared row op through the shared single-row fast path.
fn commit_prepared_single_row_op_with_window(
    db: &Db<impl CanisterKind>,
    row_op: CommitRowOp,
    prepared_row_op: PreparedRowCommitOp,
    synchronized_store_handles: Vec<StoreHandle>,
    apply_phase: &'static str,
    on_index_applied: impl FnOnce(&PreparedRowOpDelta),
    on_data_applied: impl FnOnce(),
) -> Result<(), InternalError> {
    let SingleRowApplyPrep {
        guards: index_store_guards,
        delta,
    } = prepare_single_row_apply(&prepared_row_op);
    let CommitWindowPayload { marker, effects } = commit_window_payload_for_prepared_row_ops(
        db,
        &[row_op],
        std::slice::from_ref(&prepared_row_op),
        &[],
    )?;
    preflight_identity_range_applies(effects.identity_range_applies.as_slice())?;
    let commit = begin_commit_window_payload(marker)?;

    apply_prepared_single_row_op(
        commit,
        apply_phase,
        prepared_row_op,
        effects,
        index_store_guards,
        || on_index_applied(&delta),
        on_data_applied,
    )?;
    mark_store_handles_index_ready(synchronized_store_handles.as_slice());

    Ok(())
}

// Commit one delete-mode row operation through the runtime-hook single-row
// structural commit-window fast path.
fn commit_single_delete_row_op_with_window_for_path<C: CanisterKind>(
    db: &Db<C>,
    entity_path: &str,
    row_op: CommitRowOp,
    apply_phase: &'static str,
) -> Result<(), InternalError> {
    let prepared_row_op = db.prepare_row_commit_op(&row_op)?;
    let synchronized_store_handles =
        synchronized_store_handles_for_prepared_row_ops(db, std::slice::from_ref(&prepared_row_op));
    let affected_store_handles =
        affected_store_handles_for_prepared_row_ops(db, std::slice::from_ref(&prepared_row_op));
    record_mutation_commit_plan(
        entity_path,
        classify_mutation_commit_plan(affected_store_handles.as_slice()),
    );
    commit_prepared_single_row_op_with_window(
        db,
        row_op,
        prepared_row_op,
        synchronized_store_handles,
        apply_phase,
        |delta| emit_delete_index_delta_metrics_for_path(entity_path, delta),
        || {},
    )
}

// Derive single-row delta metrics and index-store generation guards in one
// scan so the hot write lane does not rewalk the same `index_ops` slice.
fn prepare_single_row_apply(prepared_row_op: &PreparedRowCommitOp) -> SingleRowApplyPrep {
    let mut delta = PreparedRowOpDelta::zero();
    let mut guards = SingleRowIndexStoreGuards::Empty;

    for index_op in &prepared_row_op.index_ops {
        record_prepared_index_delta(&mut delta, index_op);
        guards.record(index_op.index_store);
    }

    SingleRowApplyPrep { guards, delta }
}

/// Resolve the exact registered store pairs that one prepared-op batch
/// synchronized through authoritative row + paired index mutation.
#[must_use]
pub(in crate::db::executor) fn synchronized_store_handles_for_prepared_row_ops<C: CanisterKind>(
    db: &Db<C>,
    prepared_row_ops: &[PreparedRowCommitOp],
) -> Vec<StoreHandle> {
    let registered_handles = db.with_store_registry(|registry| {
        registry
            .iter()
            .map(|(_, handle)| handle)
            .collect::<Vec<StoreHandle>>()
    });

    registered_handles
        .into_iter()
        .filter(|handle| {
            prepared_row_ops.iter().any(|row_op| {
                ptr::eq(handle.data_store(), row_op.data_store)
                    && row_op
                        .index_ops
                        .iter()
                        .any(|index_op| ptr::eq(handle.index_store(), index_op.index_store))
            })
        })
        .collect()
}

// Resolve every registered store touched by one prepared row-op batch. Unlike
// the synchronized-index helper, this includes data-only rows and cross-store
// reverse-index mutations so durable/live commit classification can describe
// the full write footprint.
pub(in crate::db::executor) fn affected_store_handles_for_prepared_row_ops<C: CanisterKind>(
    db: &Db<C>,
    prepared_row_ops: &[PreparedRowCommitOp],
) -> Vec<StoreHandle> {
    let registered_handles = db.with_store_registry(|registry| {
        registry
            .iter()
            .map(|(_, handle)| handle)
            .collect::<Vec<StoreHandle>>()
    });

    registered_handles
        .into_iter()
        .filter(|handle| {
            prepared_row_ops.iter().any(|row_op| {
                ptr::eq(handle.data_store(), row_op.data_store)
                    || row_op
                        .index_ops
                        .iter()
                        .any(|index_op| ptr::eq(handle.index_store(), index_op.index_store))
            })
        })
        .collect()
}

// Project durable recovery payloads into one marker-bound commit payload.
// Journaled stores embed logical journal records in the marker so journal
// publication can happen before live projections are updated. Heap stores
// remain live-only and absent from durable recovery payloads.
#[expect(
    clippy::too_many_lines,
    reason = "one builder must bind row records, range ordinals, journal sequence, and marker identity before publication"
)]
fn commit_window_payload_for_prepared_row_ops<C: CanisterKind>(
    db: &Db<C>,
    row_ops: &[CommitRowOp],
    prepared_row_ops: &[PreparedRowCommitOp],
    identity_ranges: &[IdentityRangeAdvance],
) -> Result<CommitWindowPayload, InternalError> {
    if row_ops.len() != prepared_row_ops.len() {
        return Err(InternalError::executor_invariant());
    }

    let marker_id = generate_commit_id()?;
    let registered_stores = db.with_store_registry(|registry| registry.iter().collect::<Vec<_>>());
    let incarnation = database_incarnation_id()?;
    let mut range_routes = Vec::with_capacity(identity_ranges.len());
    for (ordinal, range) in identity_ranges.iter().copied().enumerate() {
        if range.owner().database_incarnation_id() != incarnation
            || identity_ranges[..ordinal]
                .iter()
                .any(|existing| existing.owner() == range.owner())
        {
            return Err(InternalError::identity_state_corruption());
        }
        let runtime_entity = db.accepted_runtime_entity_for_tag(range.owner().entity_tag())?;
        let (store_path, handle) = registered_stores
            .iter()
            .copied()
            .find(|(store_path, _)| *store_path == runtime_entity.store_path())
            .ok_or_else(InternalError::executor_invariant)?;
        range_routes.push((range, store_path, handle));
    }
    let mut journal_records = Vec::<(StoreHandle, Vec<JournalRecord>)>::new();

    for (row_op, prepared_row_op) in row_ops.iter().zip(prepared_row_ops) {
        let handle = registered_stores
            .iter()
            .map(|(_, handle)| handle)
            .find(|handle| ptr::eq(handle.data_store(), prepared_row_op.data_store))
            .ok_or_else(InternalError::executor_invariant)?;

        let range_binds_store = range_routes
            .iter()
            .any(|(_, _, route_handle)| ptr::eq(route_handle.data_store(), handle.data_store()));
        if handle.storage_capabilities().recovery()
            == StoreRecoveryCapability::StableBasePlusJournalReplay
            || range_binds_store
        {
            let record = journal_record_for_row_op(row_op)?;
            push_journal_record(&mut journal_records, *handle, record);
        }
    }
    for (range, _, handle) in &range_routes {
        push_journal_record(
            &mut journal_records,
            *handle,
            JournalRecord::identity_range_advance(*range)?,
        );
    }

    let mut journal_appends = Vec::with_capacity(journal_records.len());
    let mut marker_batches = Vec::with_capacity(journal_records.len());
    let mut identity_range_applies = Vec::with_capacity(identity_ranges.len());
    for (ordinal, (handle, records)) in journal_records.into_iter().enumerate() {
        let journal_store = handle.journal_tail_store();
        let sequence =
            journal_store.map_or(Ok(crate::db::journal::JournalSequence::new(0)), |store| {
                store.with_borrow(
                    crate::db::journal::JournalTailStore::next_mutation_append_sequence,
                )
            })?;
        // Preserve the established single-store bytes while giving every
        // additional tail its own marker-local batch identity.
        let batch_id = generate_marker_batch_id(marker_id, ordinal)?;
        let batch = JournalBatch::new(batch_id, marker_id, sequence, records)?;
        let store_path = registered_stores
            .iter()
            .find_map(|(store_path, registered_handle)| {
                ptr::eq(registered_handle.data_store(), handle.data_store()).then_some(*store_path)
            })
            .ok_or_else(InternalError::executor_invariant)?;
        for (record_ordinal, record) in batch.records().iter().enumerate() {
            let JournalRecord::IdentityRangeAdvance { range } = record else {
                continue;
            };
            let record_ordinal =
                u32::try_from(record_ordinal).map_err(|_| InternalError::store_invariant())?;
            let advance_id = IdentityAdvanceId::try_new(
                batch.commit_marker_id(),
                batch.batch_id(),
                batch.journal_sequence().get(),
                record_ordinal,
            )?;
            identity_range_applies.push(PreparedIdentityRangeApply {
                store_path,
                handle,
                range: *range,
                advance_id,
            });
        }
        marker_batches.push(batch.clone());
        if let Some(journal_store) = journal_store {
            journal_appends.push(PreparedJournalAppend {
                journal_store,
                batch,
            });
        }
    }
    if identity_range_applies.len() != identity_ranges.len() {
        return Err(InternalError::identity_state_corruption());
    }

    let marker = CommitMarker::from_parts(marker_id, marker_batches)?;

    Ok(CommitWindowPayload {
        marker,
        effects: PreparedCommitEffects {
            journal_appends,
            identity_range_applies,
        },
    })
}

fn preflight_identity_range_applies(
    identity_ranges: &[PreparedIdentityRangeApply],
) -> Result<(), InternalError> {
    for prepared in identity_ranges {
        prepared
            .handle
            .with_schema(|store| store.preflight_identity_range_advance(prepared.range))?;
        if prepared.handle.storage_capabilities().schema_metadata()
            == StoreSchemaMetadataCapability::LiveRebuiltMetadata
        {
            preflight_live_identity_range_checkpoint(prepared.store_path, prepared.range)?;
        }
    }
    Ok(())
}

fn apply_identity_range_applies(
    identity_ranges: &[PreparedIdentityRangeApply],
) -> Result<(), InternalError> {
    for prepared in identity_ranges {
        if prepared.handle.storage_capabilities().schema_metadata()
            == StoreSchemaMetadataCapability::LiveRebuiltMetadata
        {
            apply_live_identity_range_checkpoint(
                prepared.store_path,
                prepared.range,
                prepared.advance_id,
            )?;
        }
        prepared.handle.with_schema_mut(|store| {
            store.apply_identity_range_advance(prepared.range, prepared.advance_id)
        })?;
    }
    Ok(())
}

fn begin_commit_window_payload(marker: CommitMarker) -> Result<CommitGuard, InternalError> {
    begin_commit(marker)
}

fn journal_record_for_row_op(row_op: &CommitRowOp) -> Result<JournalRecord, InternalError> {
    match row_op.after.as_ref() {
        Some(after) => JournalRecord::row_put(
            row_op.entity_path.as_ref(),
            row_op.key.clone(),
            after.clone(),
            row_op.schema_fingerprint,
        ),
        None => JournalRecord::row_delete(
            row_op.entity_path.as_ref(),
            row_op.key.clone(),
            row_op.schema_fingerprint,
        ),
    }
}

fn push_journal_record(
    journal_records: &mut Vec<(StoreHandle, Vec<JournalRecord>)>,
    handle: StoreHandle,
    record: JournalRecord,
) {
    if let Some((_, records)) = journal_records
        .iter_mut()
        .find(|(existing, _)| ptr::eq(existing.data_store(), handle.data_store()))
    {
        records.push(record);
        return;
    }

    journal_records.push((handle, vec![record]));
}

fn append_prepared_journal_batches(appends: &[PreparedJournalAppend]) -> Result<(), InternalError> {
    for append in appends {
        append
            .journal_store
            .with_borrow_mut(|store| store.append_batch(&append.batch))?;
    }

    Ok(())
}

/// Classify the durable/live commit footprint represented by affected stores.
#[must_use]
pub(in crate::db::executor) fn classify_mutation_commit_plan(
    handles: &[StoreHandle],
) -> MutationCommitClass {
    let mut touches_durable = false;
    let mut touches_live = false;

    for handle in handles {
        match handle.storage_capabilities().commit_participation() {
            StoreCommitParticipation::Durable => touches_durable = true,
            StoreCommitParticipation::LiveOnly => touches_live = true,
        }
    }

    match (touches_durable, touches_live) {
        (true, true) => MutationCommitClass::MixedDurableAndLive,
        (false, true) => MutationCommitClass::LiveOnly,
        _ => MutationCommitClass::DurableOnly,
    }
}

pub(in crate::db::executor) fn record_mutation_commit_plan(
    entity_path: &str,
    class: MutationCommitClass,
) {
    record(MetricsEvent::MutationCommitPlan {
        entity_path: entity_path.into(),
        class,
    });
}

// Mark one batch of synchronized index stores as `Ready` after commit apply
// succeeds and the commit marker is already closed.
fn mark_store_handles_index_ready(handles: &[StoreHandle]) {
    for handle in handles {
        handle.mark_index_ready();
    }
}

fn index_store_id(index_store: &'static LocalKey<RefCell<IndexStore>>) -> usize {
    std::ptr::from_ref::<LocalKey<RefCell<IndexStore>>>(index_store) as usize
}

fn emit_index_delta_metrics_for_path(entity_path: &str, delta: &PreparedRowOpDelta) {
    record(MetricsEvent::IndexDelta {
        entity_path: entity_path.into(),
        inserts: u64::try_from(delta.index_inserts).unwrap_or(u64::MAX),
        removes: u64::try_from(delta.index_removes).unwrap_or(u64::MAX),
    });

    record(MetricsEvent::ReverseIndexDelta {
        entity_path: entity_path.into(),
        inserts: u64::try_from(delta.reverse_index_inserts).unwrap_or(u64::MAX),
        removes: u64::try_from(delta.reverse_index_removes).unwrap_or(u64::MAX),
    });
}

// Emit delete-only index metrics through the shared path-shaped sink contract.
fn emit_delete_index_delta_metrics_for_path(entity_path: &str, delta: &PreparedRowOpDelta) {
    emit_index_delta_metrics_for_path(entity_path, &delta.delete_only());
}

fn key_within_bounds(
    key: &RawIndexStoreKey,
    bounds: (&Bound<RawIndexStoreKey>, &Bound<RawIndexStoreKey>),
) -> bool {
    key_within_envelope(key, bounds.0, bounds.1)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        db::{
            data::{DataStore, DecodedDataStoreKey},
            index::IndexStore,
            integrity::DatabaseIncarnationId,
            journal::JournalTailStore,
            key_taxonomy::{PrimaryKeyComponent, PrimaryKeyValue},
            registry::{
                StoreAllocationIdentities, StoreAllocationIdentity, StoreRegistry,
                StoreRuntimeStorageCapabilities,
            },
            schema::{
                AcceptedFieldKind, AcceptedSchemaRevision, FieldId, FieldInsertGeneration,
                FieldStorageDecode, PersistedFieldSnapshot, PersistedSchemaSnapshot,
                SchemaFieldSlot, SchemaFieldWritePolicy, SchemaInsertDefault, SchemaRowLayout,
                SchemaStore, SchemaVersion, accepted_schema_candidate_for_tests,
            },
        },
        error::{ErrorClass, ErrorOrigin},
        testing::test_memory,
        traits::Path,
        types::EntityTag,
    };
    use std::collections::BTreeMap;

    struct SchedulerOverlayTestCanister;

    impl Path for SchedulerOverlayTestCanister {
        const PATH: &'static str = "executor::mutation::tests::SchedulerOverlayTestCanister";
    }

    impl CanisterKind for SchedulerOverlayTestCanister {
        const COMMIT_MEMORY_ID: u8 = 1;
        const COMMIT_STABLE_KEY: &'static str = "icydb.scheduler_overlay.commit.v1";
        const INTEGRITY_PROGRESS_MEMORY_ID: u8 = 2;
        const INTEGRITY_PROGRESS_STABLE_KEY: &'static str =
            "icydb.scheduler_overlay.integrity.progress.v1";
    }

    thread_local! {
        static TEST_REGISTRY: StoreRegistry = StoreRegistry::new();
        static FIRST_IDENTITY_DATA: RefCell<DataStore> =
            RefCell::new(DataStore::init_journaled(test_memory(240)));
        static FIRST_IDENTITY_INDEX: RefCell<IndexStore> =
            RefCell::new(IndexStore::init_journaled(test_memory(241)));
        static FIRST_IDENTITY_SCHEMA: RefCell<SchemaStore> =
            RefCell::new(SchemaStore::init_journaled(test_memory(242)));
        static FIRST_IDENTITY_JOURNAL: RefCell<JournalTailStore> =
            RefCell::new(JournalTailStore::init(test_memory(243)));
        static SECOND_IDENTITY_DATA: RefCell<DataStore> =
            RefCell::new(DataStore::init_journaled(test_memory(244)));
        static SECOND_IDENTITY_INDEX: RefCell<IndexStore> =
            RefCell::new(IndexStore::init_journaled(test_memory(245)));
        static SECOND_IDENTITY_SCHEMA: RefCell<SchemaStore> =
            RefCell::new(SchemaStore::init_journaled(test_memory(246)));
        static SECOND_IDENTITY_JOURNAL: RefCell<JournalTailStore> =
            RefCell::new(JournalTailStore::init(test_memory(247)));
    }

    const IDENTITY_INCARNATION: DatabaseIncarnationId = DatabaseIncarnationId::for_tests(0x71);

    fn identity_candidate(
        store_path: &str,
        entity_tag: EntityTag,
    ) -> crate::db::schema::CandidateSchemaRevision {
        let field_id = FieldId::new(1);
        let kind = AcceptedFieldKind::Nat64;
        let leaf_codec = kind.leaf_codec_for_storage(FieldStorageDecode::ByKind);
        let snapshot = PersistedSchemaSnapshot::new(
            SchemaVersion::initial(),
            format!("tests::Identity{}", entity_tag.value()),
            format!("Identity{}", entity_tag.value()),
            field_id,
            SchemaRowLayout::initial(vec![(field_id, SchemaFieldSlot::new(0))]),
            vec![PersistedFieldSnapshot::new_initial_with_write_policy(
                field_id,
                "id".to_string(),
                SchemaFieldSlot::new(0),
                kind,
                Vec::new(),
                false,
                SchemaInsertDefault::None,
                SchemaFieldWritePolicy::from_model_policies(
                    Some(FieldInsertGeneration::Identity),
                    None,
                ),
                FieldStorageDecode::ByKind,
                leaf_codec,
            )],
        );
        accepted_schema_candidate_for_tests(
            store_path,
            AcceptedSchemaRevision::INITIAL,
            BTreeMap::from([(entity_tag, snapshot)]),
        )
    }

    fn identity_store_handles() -> (StoreHandle, StoreHandle) {
        (
            StoreHandle::new_journaled(
                &FIRST_IDENTITY_DATA,
                &FIRST_IDENTITY_INDEX,
                &FIRST_IDENTITY_SCHEMA,
                &FIRST_IDENTITY_JOURNAL,
                StoreAllocationIdentities::new_journaled(
                    StoreAllocationIdentity::new(240, "icydb.test.identity-set.first.data.v1"),
                    StoreAllocationIdentity::new(241, "icydb.test.identity-set.first.index.v1"),
                    StoreAllocationIdentity::new(242, "icydb.test.identity-set.first.schema.v1"),
                    StoreAllocationIdentity::new(243, "icydb.test.identity-set.first.journal.v1"),
                ),
                StoreRuntimeStorageCapabilities::journaled(),
            ),
            StoreHandle::new_journaled(
                &SECOND_IDENTITY_DATA,
                &SECOND_IDENTITY_INDEX,
                &SECOND_IDENTITY_SCHEMA,
                &SECOND_IDENTITY_JOURNAL,
                StoreAllocationIdentities::new_journaled(
                    StoreAllocationIdentity::new(244, "icydb.test.identity-set.second.data.v1"),
                    StoreAllocationIdentity::new(245, "icydb.test.identity-set.second.index.v1"),
                    StoreAllocationIdentity::new(246, "icydb.test.identity-set.second.schema.v1"),
                    StoreAllocationIdentity::new(247, "icydb.test.identity-set.second.journal.v1"),
                ),
                StoreRuntimeStorageCapabilities::journaled(),
            ),
        )
    }

    fn test_key(value: u64) -> DecodedDataStoreKey {
        DecodedDataStoreKey::new(
            EntityTag::new(41),
            &PrimaryKeyValue::Scalar(PrimaryKeyComponent::Nat64(value)),
        )
    }

    fn test_row_op(key: &DecodedDataStoreKey, after: Option<Vec<u8>>) -> CommitRowOp {
        CommitRowOp::new(
            "tests::SelfRelation",
            key.to_raw().expect("test key should encode"),
            None,
            after,
            [7; 16],
        )
    }

    #[test]
    fn scheduler_overlay_seeds_later_final_after_images_before_preflight() {
        let db: Db<SchedulerOverlayTestCanister> = Db::new(&TEST_REGISTRY);
        let first = test_key(1);
        let later = test_key(2);
        let row_ops = vec![
            test_row_op(&first, Some(vec![1])),
            test_row_op(&later, Some(vec![2])),
        ];

        let overlay = PreflightStoreOverlay::from_row_ops(&db, row_ops.as_slice())
            .expect("complete batch overlay should build");
        let visible = overlay
            .read_primary_row(&later)
            .expect("later batch target lookup should succeed")
            .expect("later batch target must be visible before row-order preflight");

        assert_eq!(visible.as_bytes(), &[2]);
    }

    #[test]
    fn scheduler_overlay_seeds_delete_absence_before_preflight() {
        let db: Db<SchedulerOverlayTestCanister> = Db::new(&TEST_REGISTRY);
        let deleted = test_key(3);
        let row_ops = vec![test_row_op(&deleted, None)];

        let overlay = PreflightStoreOverlay::from_row_ops(&db, row_ops.as_slice())
            .expect("delete overlay should build");

        assert!(
            overlay
                .read_primary_row(&deleted)
                .expect("delete target lookup should succeed")
                .is_none(),
            "the complete batch must mask deleted rows before storage-backed proofs",
        );
    }

    #[test]
    fn stale_multi_owner_preflight_is_read_only_for_the_complete_owner_set() {
        let (first, second) = identity_store_handles();
        let first_entity = EntityTag::new(51);
        let second_entity = EntityTag::new(52);
        for (store_path, handle, entity_tag) in [
            ("tests::FirstIdentityStore", first, first_entity),
            ("tests::SecondIdentityStore", second, second_entity),
        ] {
            handle
                .with_schema_mut(|store| {
                    *store =
                        SchemaStore::init_journaled(test_memory(if entity_tag == first_entity {
                            242
                        } else {
                            246
                        }));
                    store.publish_accepted_schema_candidate(
                        IDENTITY_INCARNATION,
                        AcceptedSchemaRevision::NONE,
                        &identity_candidate(store_path, entity_tag),
                    )
                })
                .expect("the Identity owner should publish with explicit zero state");
        }

        let first_owner = crate::db::schema::IdentityStateOwner::try_new(
            IDENTITY_INCARNATION,
            first_entity,
            FieldId::new(1),
        )
        .expect("the first owner should admit");
        let second_owner = crate::db::schema::IdentityStateOwner::try_new(
            IDENTITY_INCARNATION,
            second_entity,
            FieldId::new(1),
        )
        .expect("the second owner should admit");
        let second_committed = IdentityRangeAdvance::try_new(second_owner, 0, 1, 1)
            .expect("the committed second-owner range should admit");
        second
            .with_schema_mut(|store| {
                store.apply_identity_range_advance(
                    second_committed,
                    IdentityAdvanceId::try_new([0x11; 16], [0x21; 16], 1, 0)
                        .expect("the committed advance identity should admit"),
                )
            })
            .expect("the competing second-owner range should materialize");

        let prepared = [
            PreparedIdentityRangeApply {
                store_path: "tests::FirstIdentityStore",
                handle: first,
                range: IdentityRangeAdvance::try_new(first_owner, 0, 1, 1)
                    .expect("the first pending range should admit"),
                advance_id: IdentityAdvanceId::try_new([0x12; 16], [0x22; 16], 2, 0)
                    .expect("the first pending advance identity should admit"),
            },
            PreparedIdentityRangeApply {
                store_path: "tests::SecondIdentityStore",
                handle: second,
                range: IdentityRangeAdvance::try_new(second_owner, 0, 1, 1)
                    .expect("the stale second-owner range should remain structurally valid"),
                advance_id: IdentityAdvanceId::try_new([0x12; 16], [0x23; 16], 3, 0)
                    .expect("the second pending advance identity should admit"),
            },
        ];

        let error = preflight_identity_range_applies(prepared.as_slice())
            .expect_err("one stale owner must reject the complete set before publication");
        assert_eq!(error.class(), ErrorClass::Conflict);
        assert_eq!(error.origin(), ErrorOrigin::Identity);
        for (handle, entity_tag, expected_high_water) in
            [(first, first_entity, 0), (second, second_entity, 1)]
        {
            let cursor = handle
                .with_schema(|store| {
                    store.identity_statement_cursor(
                        IDENTITY_INCARNATION,
                        entity_tag,
                        FieldId::new(1),
                        &AcceptedFieldKind::Nat64,
                    )
                })
                .expect("set preflight must leave every owner unchanged");
            assert_eq!(cursor.expected_high_water(), expected_high_water);
        }
    }
}