liminal-server 0.20.2

//! Durable transition-input log for one participant conversation.
//!
//! Every entry stores the *inputs* of one committed operation together with
//! the canonical shell event bytes minted for it (when the operation is one
//! of the six shell-event lifecycle operations). Replay re-runs the same
//! protocol transitions from the stored inputs and re-mints the same shell
//! event under the aggregate durability barrier, then cross-checks the
//! re-minted canonical bytes against the stored bytes — so the server never
//! grows a second implementation of lifecycle rules and byte drift between
//! live and replayed decisions fails loudly.

use liminal::durability::settle::{append_all_settled, append_settled, cas_settled, flush_settled};
use std::{collections::VecDeque, sync::Arc};

use liminal::durability::{DurabilityError, DurableStore};
use liminal_protocol::wire::{
    AttachAttemptToken, AttachSecret, BindingEpoch, ConnectionIncarnation, CredentialAttachRequest,
    DeliverySeq, DetachAttemptToken, DetachRequest, EnrollmentRequest, EnrollmentToken, Generation,
    LeaveAttemptToken, LeaveRequest, ParticipantAck, ParticipantId, RecordAdmission,
    RecordAdmissionAttemptToken, TransactionOrder,
};
use serde::{Deserialize, Serialize};

use super::facts::{Digest, FactsError};
pub(super) use super::log_error::FencedAttachProofRefusal;
use super::operation_batch::OperationBatch;
use super::routed_stream::{RoutedStreamError, RoutedStreamRefusal};
use super::write_issuance::WriteIssuance;

/// Stream-key prefix for production participant conversation logs.
pub(super) const STREAM_PREFIX: &str = "liminal:participant-production:";
/// Stream-key prefix for a conversation log rewritten by a migration; the
/// key carries the generation, then the conversation id.
pub(super) const GENERATION_STREAM_PREFIX: &str = "liminal:participant-production-generation:";
/// Value-key prefix naming which generation of a conversation's log is
/// authoritative. Absent means generation 0, the original stream.
pub(super) const GENERATION_KEY_PREFIX: &str = "liminal:participant-production-log-generation:";
/// Durable page size used during replay reads.
pub(super) const READ_BATCH_SIZE: usize = 64;
/// Stored-entry schema version written by every new append: v3 grammar whose
/// retained-charge audit is the change a row made, never the complete list.
pub(super) const SCHEMA_VERSION: u8 = 4;
/// Historical schema version whose rows carry the complete retained-charge
/// list; accepted only before the first v4 row.
pub(super) const SCHEMA_VERSION_V3: u8 = 3;
/// Frozen historical schema version accepted only in the contiguous prefix.
pub(super) const SCHEMA_VERSION_V2: u8 = 2;

/// Why a log-generation switch did not complete.
#[derive(Debug, thiserror::Error)]
pub(super) enum GenerationSwitchFailure {
    /// The compare-and-swap was refused; the generation key is unchanged.
    #[error("the generation compare-and-swap was refused: {0}")]
    NotSwapped(DurabilityError),
    /// The swap was made but the flush that makes it durable failed.
    #[error("the generation was swapped but its flush failed: {0}")]
    SwappedUnconfirmed(DurabilityError),
}

/// Failure to encode, decode, append, or read one durable log entry.
#[derive(Debug, thiserror::Error)]
pub enum OperationLogError {
    /// The underlying durable store rejected an operation.
    #[error(transparent)]
    Durability(#[from] DurabilityError),
    /// Routed stream structure or framing was refused.
    #[error("routed stream refused: {0}")]
    RoutedStream(#[source] RoutedStreamRefusal),
    /// The bounded synchronous durability bridge rejected a suspending backend.
    #[error(transparent)]
    Bridge(#[from] liminal::durability::bridge::BridgeError),
    /// A durable entry could not be serialized or deserialized.
    #[error("participant production log serialization failed: {0}")]
    Serialization(#[from] serde_json::Error),
    /// A durable entry uses an unsupported schema version.
    #[error("unsupported participant production log schema version {0}")]
    SchemaVersion(u8),
    /// A durable row's retained-charge audit does not have the form its
    /// schema version requires.
    #[error(
        "participant production log row at sequence {sequence} carries the wrong retained-charge audit form for schema version {schema_version}"
    )]
    RetainedChargeForm { sequence: u64, schema_version: u8 },
    /// A durable stream regressed from a later schema back to an earlier one.
    #[error(
        "participant production log schema regressed at sequence {sequence}: previous {previous}, actual {actual}"
    )]
    SchemaVersionTransition {
        sequence: u64,
        previous: u8,
        actual: u8,
    },
    /// The durable stream was not contiguous at the expected sequence.
    #[error("participant production log expected sequence {expected}, found {actual}")]
    Sequence {
        /// Next sequence required by replay.
        expected: u64,
        /// Sequence read from durable storage.
        actual: u64,
    },
    /// The store assigned a different sequence than the optimistic head.
    #[error("participant production log append expected {expected}, got {actual}")]
    AssignedSequence {
        /// Optimistic-concurrency sequence supplied to the store.
        expected: u64,
        /// Assigned sequence returned by the store.
        actual: u64,
    },
    /// A stored numeric field violated a protocol domain (zero generation).
    #[error("participant production log entry carries zero generation")]
    ZeroGeneration,
    /// A durable row the protocol aggregate itself refuses to replay.
    #[error("durable participant row at sequence {sequence} was refused during restore")]
    CorruptRow {
        /// Durable sequence of the refused row.
        sequence: u64,
    },
    /// A prune receipt the checkpoint would stage does not prove its own cut
    /// (DUR-018): the plan is a defect, named before any row goes.
    #[error("prune receipt {key} refused its proof: {source}")]
    PruneReceipt {
        /// The receipt's ledger key inside the conversation's route.
        key: String,
        #[source]
        source: super::prune::PruneProofError,
    },
    /// Frozen v2 marker-bearing Attached bytes cannot supply a fenced proof.
    #[error("v2 Attached row at sequence {sequence} has no durable fenced proof")]
    V2AttachedFencedProofUnavailable { sequence: u64 },
    /// Frozen v2 Attached option fields disagree with replay prestate.
    #[error("v2 Attached row at sequence {sequence} has contradictory mode evidence")]
    V2AttachedModeMismatch { sequence: u64 },
    /// A schema-v3 Attached row carries non-canonical or contradictory fenced proof facts.
    #[error("v3 fenced Attached proof at sequence {sequence} was refused: {reason}")]
    FencedAttachProof {
        sequence: u64,
        reason: FencedAttachProofRefusal,
    },
    /// A retained-charge list is longer than the stored count domain.
    #[error("participant production retained-charge count exceeds u64")]
    RetainedChargeCount,
    /// Reconstructed fate operation disagreed with the complete durable row.
    #[error("reconstructed fate operation drifted from durable sequence {sequence}")]
    FateReplayDrift { sequence: u64 },
    /// The clock could not stamp a row written on its own.
    #[error("participant production log row timestamp: {0}")]
    Clock(#[from] FactsError),
    /// The conversation registry's snapshot was refused by name (DUR-019);
    /// a refused snapshot stops the load and is never walked around.
    #[error("participant conversation registry snapshot refused: {0}")]
    RegistrySnapshot(#[source] super::registry_snapshot::RegistrySnapshotError),
}

impl From<RoutedStreamError> for OperationLogError {
    fn from(error: RoutedStreamError) -> Self {
        match error {
            RoutedStreamError::Store(original) => Self::Durability(original),
            RoutedStreamError::Refusal(refusal) => Self::RoutedStream(refusal),
        }
    }
}

impl From<super::operation_batch::CheckpointStageError> for OperationLogError {
    fn from(error: super::operation_batch::CheckpointStageError) -> Self {
        use super::operation_batch::CheckpointStageError;
        match error {
            CheckpointStageError::Store(original) => Self::Durability(original),
            CheckpointStageError::Checkpoint(original) => {
                Self::RoutedStream(RoutedStreamRefusal::Checkpoint(Box::new(original)))
            }
            CheckpointStageError::Locator(original) => {
                Self::RoutedStream(RoutedStreamRefusal::CheckpointLocator(original))
            }
            CheckpointStageError::BatchRefused => {
                Self::RoutedStream(RoutedStreamRefusal::BatchRefused)
            }
            CheckpointStageError::RouteMismatch => {
                Self::RoutedStream(RoutedStreamRefusal::CheckpointRouteMismatch)
            }
            CheckpointStageError::PositionRegression => {
                Self::RoutedStream(RoutedStreamRefusal::CheckpointPositionRegression)
            }
            CheckpointStageError::BodyMismatch => {
                Self::RoutedStream(RoutedStreamRefusal::CheckpointBodyMismatch)
            }
        }
    }
}

/// Append-only handle over one conversation's production operation stream.
///
/// The stream is the conversation's authoritative log generation, read from
/// its generation key at each use; a handle pinned to a generation reads and
/// writes that generation's stream whatever the key says.
#[derive(Debug)]
pub(super) struct OperationLog {
    store: Arc<dyn DurableStore>,
    conversation_id: u64,
    pinned_generation: Option<u64>,
    issuance: WriteIssuance,
}

impl OperationLog {
    /// Creates a stateless log handle over one durable conversation stream.
    pub(super) fn new(store: Arc<dyn DurableStore>, conversation_id: u64) -> Self {
        Self {
            store,
            conversation_id,
            pinned_generation: None,
            issuance: WriteIssuance::default(),
        }
    }

    /// Creates a handle over one exact generation of the conversation's log.
    pub(super) fn at_generation(
        store: Arc<dyn DurableStore>,
        conversation_id: u64,
        generation: u64,
    ) -> Self {
        Self {
            store,
            conversation_id,
            pinned_generation: Some(generation),
            issuance: WriteIssuance::default(),
        }
    }

    pub(super) const fn issuance(&self) -> &WriteIssuance {
        &self.issuance
    }

    /// The conversation this log belongs to.
    pub(super) const fn conversation_id(&self) -> u64 {
        self.conversation_id
    }

    fn generation_key(&self) -> String {
        format!("{GENERATION_KEY_PREFIX}{}", self.conversation_id)
    }

    /// The generation this handle reads: its pin, else the authoritative one.
    pub(super) async fn generation(&self) -> Result<u64, OperationLogError> {
        Ok(self.read_generation().await?)
    }

    /// The generation this handle reads, with the store's own failure.
    async fn read_generation(&self) -> Result<u64, DurabilityError> {
        if let Some(generation) = self.pinned_generation {
            return Ok(generation);
        }
        Ok(self
            .store
            .read_value(&self.generation_key())
            .await?
            .unwrap_or(0))
    }

    /// The stream key of the generation this handle reads: the flat stream
    /// and, inside the conversation's route, the routed stream.
    ///
    /// # Errors
    /// Preserves the store's failure to read the authoritative generation.
    pub(super) async fn stream_key(&self) -> Result<String, DurabilityError> {
        let generation = self.read_generation().await?;
        Ok(if generation == 0 {
            format!("{STREAM_PREFIX}{}", self.conversation_id)
        } else {
            format!(
                "{GENERATION_STREAM_PREFIX}{generation}:{}",
                self.conversation_id
            )
        })
    }

    /// Reads the raw stored payloads from `from_sequence`, one page.
    pub(super) async fn read_raw_page(
        &self,
        from_sequence: u64,
    ) -> Result<Vec<liminal::durability::StoredEntry>, OperationLogError> {
        let key = self.stream_key().await?;
        Ok(self
            .store
            .read_from(&key, from_sequence, READ_BATCH_SIZE)
            .await?)
    }

    /// Appends already-encoded payloads at `expected_sequence` in one store
    /// write where the store supports it, then flushes.
    pub(super) async fn append_encoded(
        &self,
        payloads: Vec<Vec<u8>>,
        expected_sequence: u64,
    ) -> Result<u64, OperationLogError> {
        let key = self.stream_key().await?;
        self.issuance.mark_submitted();
        let next =
            append_all_settled(self.store.as_ref(), &key, payloads, expected_sequence).await?;
        flush_settled(self.store.as_ref()).await?;
        Ok(next)
    }

    /// Makes `to` the authoritative generation, only while `from` still is.
    ///
    /// The two failures are told apart: a refused compare-and-swap left the
    /// generation key as it was, while a flush that fails after the swap
    /// leaves it unknown which generation a restart will find.
    pub(super) async fn switch_generation(
        &self,
        from: u64,
        to: u64,
    ) -> Result<(), GenerationSwitchFailure> {
        let key = self.generation_key();
        self.issuance.mark_submitted();
        cas_settled(self.store.as_ref(), &key, from, to)
            .await
            .map_err(GenerationSwitchFailure::NotSwapped)?;
        flush_settled(self.store.as_ref())
            .await
            .map_err(GenerationSwitchFailure::SwappedUnconfirmed)?;
        Ok(())
    }

    /// Clones the shared store handle for bounded point-read associations that
    /// are part of replaying one decoded operation.
    pub(super) fn store(&self) -> Arc<dyn DurableStore> {
        Arc::clone(&self.store)
    }

    /// Reads and decodes exactly one committed operation by physical sequence,
    /// wherever it sits: below the native head it is a flat row, at or after
    /// it a routed row on the conversation's route. Every reader of a
    /// committed row goes through here, so none sees the flat side alone.
    ///
    /// # Errors
    /// Refuses a corrupt head, a sequence mismatch and operation framing.
    pub(super) async fn read_at(
        &self,
        sequence: u64,
    ) -> Result<Option<DecodedOperation>, OperationLogError> {
        self.read_committed_at(
            &super::restore_checkpoint_locator::checkpoint_conversation_route(self.conversation_id),
            sequence,
        )
        .await
    }

    /// Reads and decodes exactly one flat native row by physical sequence:
    /// the rows below the native head, where a store written before pruning
    /// holds every row and a migrated one its prefix.
    async fn read_flat_at(
        &self,
        sequence: u64,
    ) -> Result<Option<DecodedOperation>, OperationLogError> {
        let key = self.stream_key().await?;
        let Some(entry) = self.store.read_at(&key, sequence).await? else {
            return Ok(None);
        };
        if entry.sequence != sequence {
            return Err(OperationLogError::Sequence {
                expected: sequence,
                actual: entry.sequence,
            });
        }
        Self::decode_at(&entry)
    }

    /// Reads a source selected by explicit routed-tail authority.
    ///
    /// # Errors
    /// Refuses invalid heads, absent covered rows and operation framing.
    pub(super) async fn read_routed_at(
        &self,
        route: &str,
        sequence: u64,
    ) -> Result<Option<DecodedOperation>, OperationLogError> {
        let key = self.stream_key().await?;
        let head = super::routed_stream::read_head(self.store.as_ref(), route, &key)
            .await?
            .ok_or_else(|| {
                OperationLogError::RoutedStream(RoutedStreamRefusal::MissingHead {
                    stream: key.as_bytes().to_vec(),
                })
            })?;
        let mut entries =
            super::routed_stream::read_page(self.store.as_ref(), route, &key, &head, sequence, 1)
                .await?;
        entries
            .pop()
            .as_ref()
            .map(Self::decode_at)
            .transpose()
            .map(Option::flatten)
    }

    /// Reads one committed operation by position wherever it sits: below the
    /// native head it is a flat row, at or after it a routed row under
    /// `route`. This is the branch a restore takes
    /// (`restore_checkpoint_rebuild`) and the one [`Self::read_at`] takes on
    /// the conversation's own route; a reader that took the flat side alone
    /// saw none of the rows a live commit stages routed. A sequence at or
    /// after the native head with no routed tail selects no row.
    ///
    /// # Errors
    /// Refuses a corrupt head, a sequence mismatch and operation framing.
    pub(super) async fn read_committed_at(
        &self,
        route: &str,
        sequence: u64,
    ) -> Result<Option<DecodedOperation>, OperationLogError> {
        let native_head = self.read_stream_next_seq().await?.unwrap_or(0);
        if sequence < native_head {
            return self.read_flat_at(sequence).await;
        }
        match self.read_routed_at(route, sequence).await {
            Err(OperationLogError::RoutedStream(RoutedStreamRefusal::MissingHead { .. })) => {
                Ok(None)
            }
            read => read,
        }
    }

    /// Decodes only rows belonging to the current staged operation.
    pub(super) async fn read_staged_page(
        &self,
        batch: &mut OperationBatch,
        sequence: u64,
        phase: OperationSchemaPhase,
    ) -> Result<DecodedOperationPage, OperationLogError> {
        let stream = self.stream_key().await?;
        let rows = batch
            .read_stream_page(self.store.as_ref(), &stream, sequence, READ_BATCH_SIZE)
            .await?
            .ok_or(OperationLogError::RoutedStream(
                super::routed_stream::RoutedStreamRefusal::BatchRefused,
            ))?;
        Self::decode_page(rows, phase)
    }

    /// Reads a referenced native-prefix or staged/routed-tail source by position.
    pub(super) async fn read_staged_at(
        &self,
        batch: &OperationBatch,
        sequence: u64,
    ) -> Result<Option<DecodedOperation>, OperationLogError> {
        if let Some(operation) = self.read_at(sequence).await? {
            return Ok(Some(operation));
        }
        let stream = self.stream_key().await?;
        batch
            .read_source(self.store.as_ref(), &stream, sequence)
            .await?
            .as_ref()
            .map(Self::decode_at)
            .transpose()
            .map(Option::flatten)
    }

    fn decode_at(
        entry: &liminal::durability::StoredEntry,
    ) -> Result<Option<DecodedOperation>, OperationLogError> {
        let sequence = entry.sequence;
        let version: StoredEntryVersion = serde_json::from_slice(&entry.payload)?;
        let operation = match version.schema_version {
            SCHEMA_VERSION_V2 => {
                let stored: StoredEntryV2 = serde_json::from_slice(&entry.payload)?;
                DecodedStoredOperation::V2(stored.operation)
            }
            SCHEMA_VERSION_V3 | SCHEMA_VERSION => {
                let stored: StoredEntryV3 = serde_json::from_slice(&entry.payload)?;
                stored.operation.validate_durable(sequence)?;
                DecodedStoredOperation::V3(stored.operation)
            }
            actual => return Err(OperationLogError::SchemaVersion(actual)),
        };
        operation.validate_retained_charge_form(version.schema_version, sequence)?;
        Ok(Some(DecodedOperation {
            sequence,
            schema_version: version.schema_version,
            operation,
        }))
    }

    /// Reads the exclusive head of this exact log generation.
    pub(super) async fn read_stream_next_seq(&self) -> Result<Option<u64>, OperationLogError> {
        let key = self.stream_key().await?;
        Ok(self.store.read_stream_next_seq(&key).await?)
    }

    /// Reads at most the caller's remaining rows under the carried schema phase.
    pub(super) async fn read_bounded_page(
        &self,
        from_sequence: u64,
        phase: OperationSchemaPhase,
        limit: usize,
    ) -> Result<DecodedOperationPage, OperationLogError> {
        if limit == 0 || limit > READ_BATCH_SIZE {
            return Err(DurabilityError::ConfigError(
                "operation page limit is outside its domain".to_owned(),
            )
            .into());
        }
        let key = self.stream_key().await?;
        let entries = self.store.read_from(&key, from_sequence, limit).await?;
        Self::decode_page(entries, phase)
    }

    /// Reads a bounded validated routed tail without falling back to legacy rows.
    ///
    /// # Errors
    /// Refuses a missing or corrupt head, missing row and schema drift.
    pub(super) async fn read_routed_page(
        &self,
        route: &str,
        from_sequence: u64,
        phase: OperationSchemaPhase,
    ) -> Result<DecodedOperationPage, OperationLogError> {
        let key = self.stream_key().await?;
        let head = super::routed_stream::read_head(self.store.as_ref(), route, &key)
            .await?
            .ok_or_else(|| {
                OperationLogError::RoutedStream(RoutedStreamRefusal::MissingHead {
                    stream: key.as_bytes().to_vec(),
                })
            })?;
        let entries = super::routed_stream::read_page(
            self.store.as_ref(),
            route,
            &key,
            &head,
            from_sequence,
            READ_BATCH_SIZE,
        )
        .await?;
        Self::decode_page(entries, phase)
    }

    /// Reads an existing routed continuation during an explicit full rebuild.
    pub(super) async fn read_rebuild_page(
        &self,
        route: &str,
        from_sequence: u64,
        phase: OperationSchemaPhase,
    ) -> Result<Option<DecodedOperationPage>, OperationLogError> {
        let key = self.stream_key().await?;
        let Some(head) = super::routed_stream::read_head(self.store.as_ref(), route, &key).await?
        else {
            return Ok(None);
        };
        let entries = super::routed_stream::read_page(
            self.store.as_ref(),
            route,
            &key,
            &head,
            from_sequence,
            READ_BATCH_SIZE,
        )
        .await?;
        Self::decode_page(entries, phase).map(Some)
    }

    /// Stages the replay-validated operation head with the first checkpoint.
    ///
    /// # Errors
    /// Propagates generation, storage and conditional-head refusals unchanged.
    pub(super) async fn stage_bootstrap_head(
        &self,
        batch: &mut OperationBatch,
        next_sequence: u64,
    ) -> Result<(), OperationLogError> {
        let result: Result<(), OperationLogError> = async {
            let key = self.stream_key().await?;
            batch
                .stage_bootstrap_stream(self.store.as_ref(), &key, next_sequence)
                .await?;
            Ok(())
        }
        .await;
        if result.is_err() {
            batch.refuse();
        }
        result
    }

    fn decode_page(
        entries: Vec<liminal::durability::StoredEntry>,
        mut phase: OperationSchemaPhase,
    ) -> Result<DecodedOperationPage, OperationLogError> {
        let mut rows = Vec::with_capacity(entries.len());
        let mut canonical_rows = VecDeque::with_capacity(entries.len());
        for entry in entries {
            let version: StoredEntryVersion = serde_json::from_slice(&entry.payload)?;
            let operation = match version.schema_version {
                SCHEMA_VERSION_V2 if phase == OperationSchemaPhase::V2Prefix => {
                    let stored: StoredEntryV2 = serde_json::from_slice(&entry.payload)?;
                    DecodedStoredOperation::V2(stored.operation)
                }
                SCHEMA_VERSION_V2 => {
                    return Err(OperationLogError::SchemaVersionTransition {
                        sequence: entry.sequence,
                        previous: phase.schema_version(),
                        actual: SCHEMA_VERSION_V2,
                    });
                }
                SCHEMA_VERSION_V3 if phase == OperationSchemaPhase::V4Suffix => {
                    return Err(OperationLogError::SchemaVersionTransition {
                        sequence: entry.sequence,
                        previous: SCHEMA_VERSION,
                        actual: SCHEMA_VERSION_V3,
                    });
                }
                SCHEMA_VERSION_V3 | SCHEMA_VERSION => {
                    phase = if version.schema_version == SCHEMA_VERSION {
                        OperationSchemaPhase::V4Suffix
                    } else {
                        OperationSchemaPhase::V3Suffix
                    };
                    let stored: StoredEntryV3 = serde_json::from_slice(&entry.payload)?;
                    stored.operation.validate_durable(entry.sequence)?;
                    DecodedStoredOperation::V3(stored.operation)
                }
                actual => return Err(OperationLogError::SchemaVersion(actual)),
            };
            operation.validate_retained_charge_form(version.schema_version, entry.sequence)?;
            canonical_rows.push_back((entry.sequence, entry.payload));
            rows.push(DecodedOperation {
                sequence: entry.sequence,
                schema_version: version.schema_version,
                operation,
            });
        }
        Ok(DecodedOperationPage {
            rows,
            canonical_rows,
            next_phase: phase,
        })
    }

    /// Reads one replay page starting at `from_sequence`.
    #[cfg(test)]
    pub(super) async fn read_v2_page(
        &self,
        from_sequence: u64,
    ) -> Result<Vec<(u64, StoredOperationV2)>, OperationLogError> {
        let key = self.stream_key().await?;
        let entries = self
            .store
            .read_from(&key, from_sequence, READ_BATCH_SIZE)
            .await?;
        let mut decoded = Vec::with_capacity(entries.len());
        for entry in entries {
            let version: StoredEntryVersion = serde_json::from_slice(&entry.payload)?;
            if version.schema_version != SCHEMA_VERSION_V2 {
                return Err(OperationLogError::SchemaVersion(version.schema_version));
            }
            let stored: StoredEntryV2 = serde_json::from_slice(&entry.payload)?;
            decoded.push((entry.sequence, stored.operation));
        }
        Ok(decoded)
    }

    /// Stages one encoded operation without submitting or marking issuance.
    ///
    /// # Errors
    /// Propagates encoding, generation and conditional-head refusals unchanged.
    pub(super) async fn stage_into(
        &self,
        batch: &mut OperationBatch,
        operation: &StoredOperation,
        expected_sequence: u64,
        timestamp: u64,
    ) -> Result<(), OperationLogError> {
        let result: Result<(), OperationLogError> = async {
            let payload = encode_current(operation, expected_sequence)?;
            let key = self.stream_key().await?;
            batch
                .stage_source_stream(
                    self.store.as_ref(),
                    &key,
                    super::restore_checkpoint_locator::CheckpointSource::Base,
                    liminal::durability::StoredEntry {
                        payload,
                        sequence: expected_sequence,
                        timestamp,
                    },
                )
                .await?;
            Ok(())
        }
        .await;
        if result.is_err() {
            batch.refuse();
        }
        result
    }

    /// Appends one operation at the exact optimistic head as one routed
    /// durable operation: a one-row batch on the conversation's route,
    /// committed through the path the handler commits every operation.
    ///
    /// The handler stages its rows into the operation's own batch, and no
    /// production path writes an operation row on its own; this is the append
    /// of a test-local appender, so it exists only under test. A flat native
    /// row beside routed rows leaves the native head behind the routed one,
    /// and the next append at the routed head is refused by sequence; so a
    /// row of a routed conversation is never written flat.
    ///
    /// # Errors
    /// Propagates encoding, generation, head and store refusals unchanged.
    #[cfg(test)]
    pub(super) async fn append(
        &self,
        operation: &StoredOperation,
        expected_sequence: u64,
    ) -> Result<(), OperationLogError> {
        let mut batch = OperationBatch::new(
            &super::restore_checkpoint_locator::checkpoint_conversation_route(self.conversation_id),
        );
        self.stage_into(
            &mut batch,
            operation,
            expected_sequence,
            super::facts::now_unix_nanos()?,
        )
        .await?;
        batch
            .commit(self.store.as_ref(), || self.issuance.mark_submitted())
            .await?;
        Ok(())
    }

    /// Appends one flat native row at the exact optimistic head, then
    /// flushes: the row shape a store written before pruning holds, kept for
    /// the legacy-rows arm alone.
    ///
    /// The flush is the durability barrier the caller's pending shell commit
    /// waits behind: nothing is published until these bytes are durable.
    pub(super) async fn append_flat(
        &self,
        operation: &StoredOperation,
        expected_sequence: u64,
    ) -> Result<(), OperationLogError> {
        let payload = encode_current(operation, expected_sequence)?;
        let key = self.stream_key().await?;
        self.issuance.mark_submitted();
        let assigned =
            append_settled(self.store.as_ref(), &key, payload, expected_sequence).await?;
        if assigned != expected_sequence {
            return Err(OperationLogError::AssignedSequence {
                expected: expected_sequence,
                actual: assigned,
            });
        }
        flush_settled(self.store.as_ref()).await?;
        Ok(())
    }
}

/// The schema version every new append writes.
#[cfg(not(test))]
const fn append_schema_version() -> u8 {
    SCHEMA_VERSION
}

/// The schema version every new append writes, or schema v3 while a test has
/// asked for the complete-list rows older stores hold.
#[cfg(test)]
fn append_schema_version() -> u8 {
    if legacy_rows::armed() {
        SCHEMA_VERSION_V3
    } else {
        SCHEMA_VERSION
    }
}

/// Encodes one operation at the schema version new appends write, after the
/// same validation every append applies.
pub(super) fn encode_current(
    operation: &StoredOperation,
    sequence: u64,
) -> Result<Vec<u8>, OperationLogError> {
    encode_versioned(operation, sequence, append_schema_version())
}

/// Encodes one operation under an exact v3-grammar schema version after the
/// validation every append applies.
pub(super) fn encode_versioned(
    operation: &StoredOperation,
    sequence: u64,
    schema_version: u8,
) -> Result<Vec<u8>, OperationLogError> {
    operation.validate_durable(sequence)?;
    DecodedStoredOperation::V3(operation.clone())
        .validate_retained_charge_form(schema_version, sequence)?;
    Ok(serde_json::to_vec(&StoredEntryV3 {
        schema_version,
        operation: operation.clone(),
    })?)
}

/// Re-encodes one frozen schema-v2 operation exactly as it is stored.
pub(super) fn encode_v2(operation: &StoredOperationV2) -> Result<Vec<u8>, OperationLogError> {
    Ok(serde_json::to_vec(&StoredEntryV2 {
        schema_version: SCHEMA_VERSION_V2,
        operation: operation.clone(),
    })?)
}

/// The retained-charge audit a new row records for one transition: the
/// change from `prior` to `resulting`, or the complete list while a test has
/// asked for schema-v3 rows.
pub(super) fn record_retained_charges(
    prior: &[StoredRetainedCharge],
    resulting: &[StoredRetainedCharge],
) -> Result<
    (
        Option<Vec<StoredRetainedCharge>>,
        Option<StoredRetainedChargeChange>,
    ),
    OperationLogError,
> {
    if append_schema_version() == SCHEMA_VERSION {
        Ok((
            None,
            Some(StoredRetainedChargeChange::between(prior, resulting)?),
        ))
    } else {
        Ok((Some(resulting.to_vec()), None))
    }
}

/// Test-only switch that makes appends write the schema-v3 rows, carrying
/// the complete retained-charge list, that stores written before schema v4
/// hold.
#[cfg(test)]
pub(super) mod legacy_rows {
    use std::cell::Cell;

    thread_local! {
        static ARMED: Cell<bool> = const { Cell::new(false) };
    }

    /// Holds the switch on until dropped.
    pub(in super::super) struct LegacyRows(());

    impl Drop for LegacyRows {
        fn drop(&mut self) {
            ARMED.with(|armed| armed.set(false));
        }
    }

    /// Makes this thread's appends write schema-v3 complete-list rows.
    pub(in super::super) fn arm() -> LegacyRows {
        ARMED.with(|armed| armed.set(true));
        LegacyRows(())
    }

    pub(in super::super) fn armed() -> bool {
        ARMED.with(Cell::get)
    }
}

/// One-way schema phase carried across every bounded replay page.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) enum OperationSchemaPhase {
    #[default]
    V2Prefix,
    V3Suffix,
    V4Suffix,
}

impl OperationSchemaPhase {
    /// The latest schema version this phase has already read.
    const fn schema_version(self) -> u8 {
        match self {
            Self::V2Prefix => SCHEMA_VERSION_V2,
            Self::V3Suffix => SCHEMA_VERSION_V3,
            Self::V4Suffix => SCHEMA_VERSION,
        }
    }
}

/// One decoded versioned operation row.
#[derive(Clone, Debug)]
pub(super) struct DecodedOperation {
    pub(super) sequence: u64,
    pub(super) schema_version: u8,
    pub(super) operation: DecodedStoredOperation,
}

/// One bounded decoded page plus the phase supplied to the next page.
#[derive(Clone, Debug)]
pub(super) struct DecodedOperationPage {
    pub(super) rows: Vec<DecodedOperation>,
    pub(super) canonical_rows: VecDeque<(u64, Vec<u8>)>,
    pub(super) next_phase: OperationSchemaPhase,
}

/// Version-specific operation payload retained until contextual v2 migration.
#[derive(Clone, Debug)]
pub(super) enum DecodedStoredOperation {
    V2(StoredOperationV2),
    V3(StoredOperationV3),
}

impl DecodedStoredOperation {
    /// Refuses a row whose retained-charge audit form disagrees with its
    /// schema version: v2 and v3 rows carry the complete resulting list, v4
    /// rows carry only the change the row made.
    fn validate_retained_charge_form(
        &self,
        schema_version: u8,
        sequence: u64,
    ) -> Result<(), OperationLogError> {
        let audit = match self {
            Self::V2(StoredOperationV2::MarkerDrained { row })
            | Self::V3(StoredOperationV3::MarkerDrained { row }) => {
                row.retained_charge_audit(sequence)?
            }
            Self::V2(StoredOperationV2::RecordAdmission { row })
            | Self::V3(StoredOperationV3::RecordAdmission { row }) => {
                row.retained_charge_audit(sequence)?
            }
            Self::V2(_) | Self::V3(_) => return Ok(()),
        };
        let expects_change = schema_version == SCHEMA_VERSION;
        if expects_change == matches!(audit, RetainedChargeAudit::Change(_)) {
            Ok(())
        } else {
            Err(OperationLogError::RetainedChargeForm {
                sequence,
                schema_version,
            })
        }
    }
}

#[derive(Clone, Copy, Debug, Deserialize)]
struct StoredEntryVersion {
    schema_version: u8,
}

/// Frozen v2 entry envelope. Its field names and representation never change.
#[derive(Clone, Debug, Deserialize, Serialize)]
struct StoredEntryV2 {
    schema_version: u8,
    operation: StoredOperationV2,
}

/// Canonical v3 entry envelope used for every new append.
#[derive(Clone, Debug, Deserialize, Serialize)]
struct StoredEntryV3 {
    schema_version: u8,
    operation: StoredOperationV3,
}

pub(super) use super::log_v2::*;
pub(super) use super::log_v3::*;

/// Canonical operation type accepted by all production append entry points.
pub(super) type StoredOperation = StoredOperationV3;

/// Scalar resource vector in the canonical v2 row schema.
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Hash, Serialize)]
pub(super) struct StoredResourceVector {
    pub(super) entries: u64,
    pub(super) bytes: u64,
}

/// One keyed retained-row charge in the canonical v2 row schema.
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Hash, Serialize)]
pub(super) struct StoredRetainedCharge {
    pub(super) delivery_seq: DeliverySeq,
    pub(super) transaction_order: TransactionOrder,
    pub(super) candidate_phase: u8,
    pub(super) participant_id: ParticipantId,
    pub(super) charge: StoredResourceVector,
}

/// The retained charges one row's transition removed and added, by position.
///
/// A row records only what it changed, so its size is bounded by the change
/// and never by how much history the conversation has retained. The change is
/// exact: applied to the list the transition started from it rebuilds the
/// resulting list element for element, in order, so replay audits the whole
/// poststate exactly as the complete list did.
#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredRetainedChargeChange {
    /// Positions in the prior list, ascending, of the charges removed.
    pub(super) removed: Vec<u64>,
    /// The charges added, each at its position in the resulting list,
    /// ascending.
    pub(super) added: Vec<(u64, StoredRetainedCharge)>,
    /// Length of the resulting list.
    pub(super) resulting_count: u64,
}

impl StoredRetainedChargeChange {
    /// The change from `prior` to `resulting`.
    ///
    /// One walk over both lists: equal heads are kept, and otherwise the
    /// lower delivery sequence goes (from `prior`) or comes (into
    /// `resulting`). The walk consumes both lists whole, so the change it
    /// records rebuilds `resulting` exactly whatever order either list holds.
    pub(super) fn between(
        prior: &[StoredRetainedCharge],
        resulting: &[StoredRetainedCharge],
    ) -> Result<Self, OperationLogError> {
        let position =
            |index: usize| u64::try_from(index).map_err(|_| OperationLogError::RetainedChargeCount);
        let mut removed = Vec::new();
        let mut added = Vec::new();
        let (mut prior_index, mut resulting_index) = (0_usize, 0_usize);
        loop {
            match (prior.get(prior_index), resulting.get(resulting_index)) {
                (None, None) => break,
                (Some(before), Some(after)) if before == after => {
                    prior_index += 1;
                    resulting_index += 1;
                }
                (Some(before), Some(after)) if before.delivery_seq <= after.delivery_seq => {
                    removed.push(position(prior_index)?);
                    prior_index += 1;
                }
                (Some(_), None) => {
                    removed.push(position(prior_index)?);
                    prior_index += 1;
                }
                (_, Some(after)) => {
                    added.push((position(resulting_index)?, *after));
                    resulting_index += 1;
                }
            }
        }
        Ok(Self {
            removed,
            added,
            resulting_count: position(resulting.len())?,
        })
    }

    /// Rebuilds the resulting list from the list the transition started from.
    #[cfg(test)]
    pub(super) fn apply(&self, prior: &[StoredRetainedCharge]) -> Vec<StoredRetainedCharge> {
        let mut kept = prior
            .iter()
            .enumerate()
            .filter(|(index, _)| {
                !self
                    .removed
                    .iter()
                    .any(|removed| u64::try_from(*index).is_ok_and(|index| index == *removed))
            })
            .map(|(_, charge)| *charge);
        let mut added = self.added.iter().peekable();
        let mut rebuilt = Vec::new();
        loop {
            let at = u64::try_from(rebuilt.len()).unwrap_or(u64::MAX);
            match added.peek() {
                Some((position, charge)) if *position == at => {
                    rebuilt.push(*charge);
                    added.next();
                }
                _ => match kept.next() {
                    Some(charge) => rebuilt.push(charge),
                    None => match added.next() {
                        Some((_, charge)) => rebuilt.push(*charge),
                        None => return rebuilt,
                    },
                },
            }
        }
    }
}

/// How one row records the retained charges its transition left behind.
#[derive(Clone, Copy, Debug)]
pub(super) enum RetainedChargeAudit<'row> {
    /// Schema v2 and v3: the complete resulting list.
    Complete(&'row [StoredRetainedCharge]),
    /// Schema v4: only the change the row made.
    Change(&'row StoredRetainedChargeChange),
}

impl RetainedChargeAudit<'_> {
    /// Whether the transition from `prior` to `resulting` is exactly the one
    /// this row recorded.
    pub(super) fn matches(
        self,
        prior: &[StoredRetainedCharge],
        resulting: &[StoredRetainedCharge],
    ) -> Result<bool, OperationLogError> {
        Ok(match self {
            Self::Complete(stored) => stored == resulting,
            Self::Change(stored) => {
                *stored == StoredRetainedChargeChange::between(prior, resulting)?
            }
        })
    }
}

/// Selects the one retained-charge audit a row carries; a row with both or
/// neither is corrupt.
fn retained_charge_audit<'row>(
    complete: Option<&'row Vec<StoredRetainedCharge>>,
    change: Option<&'row StoredRetainedChargeChange>,
    sequence: u64,
) -> Result<RetainedChargeAudit<'row>, OperationLogError> {
    match (complete, change) {
        (Some(complete), None) => Ok(RetainedChargeAudit::Complete(complete)),
        (None, Some(change)) => Ok(RetainedChargeAudit::Change(change)),
        (Some(_), Some(_)) | (None, None) => Err(OperationLogError::CorruptRow { sequence }),
    }
}

/// Durable marker drain written before the same-lock `RecordAdmission` retry.
#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredMarkerDrain {
    pub(super) marker: Vec<u8>,
    pub(super) retained_charge: StoredRetainedCharge,
    /// Schema v2 and v3 only.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) resulting_retained_charges: Option<Vec<StoredRetainedCharge>>,
    /// Schema v4 only.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) retained_charge_change: Option<StoredRetainedChargeChange>,
    pub(super) successor: Vec<u8>,
}

impl StoredMarkerDrain {
    /// The one retained-charge audit this row carries.
    pub(super) fn retained_charge_audit(
        &self,
        sequence: u64,
    ) -> Result<RetainedChargeAudit<'_>, OperationLogError> {
        retained_charge_audit(
            self.resulting_retained_charges.as_ref(),
            self.retained_charge_change.as_ref(),
            sequence,
        )
    }
}

/// Durable payload-bearing `RecordAdmission` request.
#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredRecordAdmissionRequest {
    pub(super) conversation_id: u64,
    pub(super) participant_id: ParticipantId,
    pub(super) capability_generation: u64,
    pub(super) token: [u8; 16],
    pub(super) payload: Vec<u8>,
}

impl From<&RecordAdmission> for StoredRecordAdmissionRequest {
    fn from(request: &RecordAdmission) -> Self {
        Self {
            conversation_id: request.conversation_id,
            participant_id: request.participant_id,
            capability_generation: request.capability_generation.get(),
            token: request.record_admission_attempt_token.into_bytes(),
            payload: request.payload.clone(),
        }
    }
}

impl StoredRecordAdmissionRequest {
    pub(super) fn into_request(self) -> Result<RecordAdmission, OperationLogError> {
        Ok(RecordAdmission {
            conversation_id: self.conversation_id,
            participant_id: self.participant_id,
            capability_generation: stored_generation(self.capability_generation)?,
            record_admission_attempt_token: RecordAdmissionAttemptToken::new(self.token),
            payload: self.payload,
        })
    }
}

/// Atomic ordinary-record poststate persisted in one append/flush transaction.
#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredRecordAdmission {
    pub(super) request: StoredRecordAdmissionRequest,
    pub(super) receiving_epoch: StoredBindingEpoch,
    pub(super) transaction_order: TransactionOrder,
    pub(super) delivery_seq: DeliverySeq,
    pub(super) encoded_record_charge: StoredResourceVector,
    pub(super) resulting_connection_count: u64,
    pub(super) newly_tracked: bool,
    /// Schema v2 and v3 only.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) resulting_retained_charges: Option<Vec<StoredRetainedCharge>>,
    pub(super) resulting_closure_accounting: Vec<u8>,
    /// Schema v4 only.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) retained_charge_change: Option<StoredRetainedChargeChange>,
}

impl StoredRecordAdmission {
    /// The one retained-charge audit this row carries.
    pub(super) fn retained_charge_audit(
        &self,
        sequence: u64,
    ) -> Result<RetainedChargeAudit<'_>, OperationLogError> {
        retained_charge_audit(
            self.resulting_retained_charges.as_ref(),
            self.retained_charge_change.as_ref(),
            sequence,
        )
    }
}

/// Durable payload-bearing Leave request.
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredLeaveRequest {
    pub(super) conversation_id: u64,
    pub(super) participant_id: ParticipantId,
    pub(super) capability_generation: u64,
    pub(super) attach_secret: [u8; 32],
    pub(super) token: [u8; 16],
}

impl From<&LeaveRequest> for StoredLeaveRequest {
    fn from(request: &LeaveRequest) -> Self {
        Self {
            conversation_id: request.conversation_id,
            participant_id: request.participant_id,
            capability_generation: request.capability_generation.get(),
            attach_secret: request.attach_secret.into_bytes(),
            token: request.leave_attempt_token.into_bytes(),
        }
    }
}

impl StoredLeaveRequest {
    pub(super) fn into_request(self) -> Result<LeaveRequest, OperationLogError> {
        Ok(LeaveRequest {
            conversation_id: self.conversation_id,
            participant_id: self.participant_id,
            capability_generation: stored_generation(self.capability_generation)?,
            attach_secret: AttachSecret::new(self.attach_secret),
            leave_attempt_token: LeaveAttemptToken::new(self.token),
        })
    }
}

/// Exact Leave tombstone inputs and causal allocation.
#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredLeave {
    pub(super) request: StoredLeaveRequest,
    pub(super) request_verifier: Digest,
    pub(super) receiving_epoch: StoredBindingEpoch,
    pub(super) left_transaction_order: TransactionOrder,
    pub(super) left_delivery_seq: DeliverySeq,
    pub(super) ended_binding_epoch: Option<StoredBindingEpoch>,
    pub(super) prior_terminal_delivery_seq: Option<DeliverySeq>,
}

/// Stored enrollment request fields.
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredEnrollmentRequest {
    pub(super) conversation_id: u64,
    pub(super) token: [u8; 16],
}

impl From<&EnrollmentRequest> for StoredEnrollmentRequest {
    fn from(request: &EnrollmentRequest) -> Self {
        Self {
            conversation_id: request.conversation_id,
            token: request.enrollment_token.into_bytes(),
        }
    }
}

impl StoredEnrollmentRequest {
    pub(super) const fn to_request(self) -> EnrollmentRequest {
        EnrollmentRequest {
            conversation_id: self.conversation_id,
            enrollment_token: EnrollmentToken::new(self.token),
        }
    }
}

/// Server allocations committed with one enrollment transition.
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredEnrollmentAllocation {
    pub(super) participant_id: ParticipantId,
    pub(super) identity_limit: u64,
    pub(super) attach_secret: [u8; 32],
    pub(super) origin_epoch: StoredBindingEpoch,
    pub(super) attached_order: TransactionOrder,
    pub(super) attached_seq: DeliverySeq,
    pub(super) receipt_expires_at: StoredU128,
    pub(super) provenance_expires_at: StoredU128,
    pub(super) enrollment_fingerprint: Digest,
}

/// Stored credential-attach request fields.
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredAttachRequest {
    pub(super) conversation_id: u64,
    pub(super) participant_id: ParticipantId,
    pub(super) capability_generation: u64,
    pub(super) attach_secret: [u8; 32],
    pub(super) token: [u8; 16],
    pub(super) accept_marker_delivery_seq: Option<DeliverySeq>,
}

impl From<&CredentialAttachRequest> for StoredAttachRequest {
    fn from(request: &CredentialAttachRequest) -> Self {
        Self {
            conversation_id: request.conversation_id,
            participant_id: request.participant_id,
            capability_generation: request.capability_generation.get(),
            attach_secret: request.attach_secret.into_bytes(),
            token: request.attach_attempt_token.into_bytes(),
            accept_marker_delivery_seq: request.accept_marker_delivery_seq,
        }
    }
}

impl StoredAttachRequest {
    pub(super) fn to_request(self) -> Result<CredentialAttachRequest, OperationLogError> {
        Ok(CredentialAttachRequest {
            conversation_id: self.conversation_id,
            participant_id: self.participant_id,
            capability_generation: stored_generation(self.capability_generation)?,
            attach_secret: AttachSecret::new(self.attach_secret),
            attach_attempt_token: AttachAttemptToken::new(self.token),
            accept_marker_delivery_seq: self.accept_marker_delivery_seq,
        })
    }
}

/// Stored detach request fields.
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredDetachRequest {
    pub(super) conversation_id: u64,
    pub(super) participant_id: ParticipantId,
    pub(super) capability_generation: u64,
    pub(super) token: [u8; 16],
}

impl From<&DetachRequest> for StoredDetachRequest {
    fn from(request: &DetachRequest) -> Self {
        Self {
            conversation_id: request.conversation_id,
            participant_id: request.participant_id,
            capability_generation: request.capability_generation.get(),
            token: request.detach_attempt_token.into_bytes(),
        }
    }
}

impl StoredDetachRequest {
    pub(super) fn to_request(self) -> Result<DetachRequest, OperationLogError> {
        Ok(DetachRequest {
            conversation_id: self.conversation_id,
            participant_id: self.participant_id,
            capability_generation: stored_generation(self.capability_generation)?,
            detach_attempt_token: DetachAttemptToken::new(self.token),
        })
    }
}

/// Stored cumulative-ack request fields.
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredAck {
    pub(super) conversation_id: u64,
    pub(super) participant_id: ParticipantId,
    pub(super) capability_generation: u64,
    pub(super) through_seq: DeliverySeq,
}

impl From<&ParticipantAck> for StoredAck {
    fn from(request: &ParticipantAck) -> Self {
        Self {
            conversation_id: request.conversation_id,
            participant_id: request.participant_id,
            capability_generation: request.capability_generation.get(),
            through_seq: request.through_seq,
        }
    }
}

impl StoredAck {
    pub(super) fn to_request(self) -> Result<ParticipantAck, OperationLogError> {
        Ok(ParticipantAck {
            conversation_id: self.conversation_id,
            participant_id: self.participant_id,
            capability_generation: stored_generation(self.capability_generation)?,
            through_seq: self.through_seq,
        })
    }
}

/// Stored binding-epoch fields.
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredBindingEpoch {
    pub(super) server_incarnation: u64,
    pub(super) connection_ordinal: u64,
    pub(super) capability_generation: u64,
}

impl From<BindingEpoch> for StoredBindingEpoch {
    fn from(epoch: BindingEpoch) -> Self {
        Self {
            server_incarnation: epoch.connection_incarnation.server_incarnation,
            connection_ordinal: epoch.connection_incarnation.connection_ordinal,
            capability_generation: epoch.capability_generation.get(),
        }
    }
}

impl StoredBindingEpoch {
    pub(super) fn to_epoch(self) -> Result<BindingEpoch, OperationLogError> {
        Ok(BindingEpoch::new(
            ConnectionIncarnation::new(self.server_incarnation, self.connection_ordinal),
            stored_generation(self.capability_generation)?,
        ))
    }
}

/// Big-endian byte capsule for `u128` deadlines.
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(super) struct StoredU128(pub(super) [u8; 16]);

impl StoredU128 {
    pub(super) const fn get(self) -> u128 {
        u128::from_be_bytes(self.0)
    }
}

impl From<u128> for StoredU128 {
    fn from(value: u128) -> Self {
        Self(value.to_be_bytes())
    }
}

fn stored_generation(value: u64) -> Result<Generation, OperationLogError> {
    Generation::new(value).ok_or(OperationLogError::ZeroGeneration)
}