pipecircuit 0.0.23

use super::super::observed::{ObservedTable, SourceId};
use super::super::{WarpTable, WarpTableGetOutcome, WarpTableRemoveOutcome, WarpTableRow};
use super::{changes, *};
use crate::{StateMachine, StepInputOf};
use alloc::{collections::VecDeque, vec::Vec};
use core::num::NonZeroU64;

#[derive(WarpTableRow)]
#[index(primary = SessionId)]
struct Session {
    sequence: u64,
}

#[derive(WarpTableRow)]
#[index(primary = OperationId)]
struct Retained<Row: PersistentRow> {
    record: Record<Row, Row::Id>,
}

pub struct Authority<Row: PersistentRow> {
    history: HistoryId,
    epoch: u64,
    clock: RevisionClock,
    row_clock: RevisionClock,
    next_session: u64,
    next_attempt: u64,
    limits: Limits,
    replication: Replication,
    accepted: ObservedTable<Row>,
    staging: WarpTable<Row>,
    local: ObservedTable<Row>,
    remote: Option<ObservedTable<Row>>,
    accepted_revision: Revision,
    local_revision: Revision,
    remote_revision: Revision,
    restored_revision: Revision,
    remote_ready: bool,
    records: WarpTable<Retained<Row>>,
    order: VecDeque<OperationId>,
    sessions: WarpTable<Session>,
    undo: Vec<(Row::Id, Row)>,
    incarnation_scratch: Vec<(Row::Id, u64)>,
    local_attempt: Option<Attempt>,
    remote_attempt: Option<Attempt>,
    poisoned: bool,
    observation_incarnation_valid: bool,
}
impl<Row: PersistentRow> StateMachine for Authority<Row> {}

impl<Row: PersistentRow> Authority<Row> {
    /// Validate a proposed group without publishing it. This is used before
    /// allocating external record locations; all touched rows are restored.
    pub fn validate_changes(&mut self, proposed: &[Change<Row, Row::Id>]) -> Result<(), Error> {
        if self.poisoned {
            return Err(Error::Corrupt);
        }
        if proposed.len() > self.limits.changes {
            return Err(Error::InvalidBounds);
        }
        let mut inverse = Vec::new();
        inverse
            .try_reserve_exact(proposed.len())
            .map_err(|_| Error::Allocation)?;
        let previous = |id| match self.accepted.table().get(id) {
            WarpTableGetOutcome::Present { row } => Ok(Row::own(row)),
            WarpTableGetOutcome::Absent { .. } => Err(Error::Absent),
        };
        for change in proposed {
            inverse.push(match change {
                Change::Insert { id, .. } => Change::Remove { id: *id },
                Change::Replace { id, .. } => Change::Replace {
                    id: *id,
                    row: previous(*id)?,
                },
                Change::Remove { id } => Change::Insert {
                    id: *id,
                    row: previous(*id)?,
                },
            });
        }
        if let Err(error) = changes::apply(&mut self.staging, proposed, &mut self.undo) {
            if error == Error::Corrupt {
                self.poisoned = true;
            }
            return Err(error);
        }
        if changes::apply(&mut self.staging, &inverse, &mut self.undo).is_err() {
            self.poisoned = true;
            return Err(Error::Corrupt);
        }
        Ok(())
    }
    /// The backend must durably reserve `reserved_through` before construction.
    /// Recovery and every physical completion belong to the trusted wiring side.
    pub fn recover(
        recovery: Recovery<Row, Row::Id>,
        reserved_through: u64,
        limits: Limits,
        replication: Replication,
    ) -> Result<Self, Error> {
        if limits.changes == 0 || limits.row_bytes == 0 {
            return Err(Error::InvalidBounds);
        }
        if recovery.revision.0 > recovery.reserved_through
            || (reserved_through <= recovery.reserved_through && reserved_through != u64::MAX)
        {
            return Err(Error::InvalidRecovery);
        }
        // MAX reopens read-only: the exhausted clock refuses every new session.
        // It is not a reusable operation epoch.
        let epoch = recovery.reserved_through.saturating_add(1);
        let create = || WarpTable::<Row>::try_new(limits.rows).map_err(|_| Error::Allocation);
        let mut accepted = create()?;
        let mut staging = create()?;
        let mut local = create()?;
        let mut remote = match replication {
            Replication::Disabled => None,
            Replication::Required => Some(create()?),
        };
        for (id, row) in recovery.rows {
            if row.retained_bytes() > limits.row_bytes {
                return Err(Error::InvalidBounds);
            }
            changes::insert(&mut accepted, id, row.clone())?;
            changes::insert(&mut staging, id, row.clone())?;
            changes::insert(&mut local, id, row.clone())?;
            if let Some(remote) = &mut remote {
                changes::insert(remote, id, row)?;
            }
        }
        let mut records = WarpTable::<Retained<Row>>::try_new(limits.operations)
            .map_err(|_| Error::Allocation)?;
        let mut order = VecDeque::new();
        order
            .try_reserve_exact(limits.operations.value().get() as usize)
            .map_err(|_| Error::Allocation)?;
        let mut previous = Revision(0);
        for record in recovery.outcomes {
            super::replay::validate(&record, limits)?;
            if record.revision <= previous
                || record.revision > recovery.revision
                || record.operation.history != recovery.history
                || (record.operation.epoch >= epoch && recovery.reserved_through != u64::MAX)
            {
                return Err(Error::InvalidRecovery);
            }
            previous = record.revision;
            let operation = record.operation;
            match records.insert_with_id(operation, Retained { record }) {
                super::super::WarpTableInsertWithIdOutcome::Inserted { .. } => {
                    order.push_back(operation)
                }
                super::super::WarpTableInsertWithIdOutcome::AlreadyPresent { .. }
                | super::super::WarpTableInsertWithIdOutcome::Full { .. }
                | super::super::WarpTableInsertWithIdOutcome::UniqueConflict { .. } => {
                    return Err(Error::InvalidRecovery);
                }
            }
        }
        let mut undo = Vec::new();
        undo.try_reserve_exact(limits.changes)
            .map_err(|_| Error::Allocation)?;
        let source = SourceId::persistent(
            recovery.history.0,
            NonZeroU64::new(epoch).ok_or(Error::InvalidRecovery)?,
        );
        let observe = |table: WarpTable<Row>| table.observe(source).map_err(|_| Error::Allocation);
        let accepted = observe(accepted)?;
        let local = observe(local)?;
        let remote = remote.map(observe).transpose()?;
        let mut incarnation_scratch = Vec::new();
        incarnation_scratch
            .try_reserve_exact(limits.changes)
            .map_err(|_| Error::Allocation)?;
        Ok(Self {
            history: recovery.history,
            epoch,
            clock: RevisionClock::restart(recovery.reserved_through, reserved_through)?,
            row_clock: RevisionClock::restart(recovery.reserved_through, reserved_through)?,
            next_session: 0,
            next_attempt: 0,
            limits,
            replication,
            accepted,
            staging,
            local,
            remote,
            accepted_revision: recovery.revision,
            local_revision: recovery.revision,
            // The restored image is private until the replica confirms this baseline.
            remote_revision: Revision(0),
            restored_revision: recovery.revision,
            remote_ready: false,
            records,
            order,
            sessions: WarpTable::try_new(limits.sessions).map_err(|_| Error::Allocation)?,
            undo,
            incarnation_scratch,
            local_attempt: None,
            remote_attempt: None,
            poisoned: false,
            observation_incarnation_valid: recovery.reserved_through != u64::MAX,
        })
    }
    pub const fn history(&self) -> HistoryId {
        self.history
    }
    pub const fn revision(&self) -> Revision {
        self.accepted_revision
    }
    pub const fn local_revision(&self) -> Revision {
        self.local_revision
    }
    pub const fn remote_revision(&self) -> Revision {
        self.remote_revision
    }
    pub const fn reserved_through(&self) -> u64 {
        self.clock.reserved_through()
    }
    pub const fn limits(&self) -> Limits {
        self.limits
    }
    pub fn reservation_required(&self) -> bool {
        self.clock.available() == Err(Error::ReservationRequired)
            || self.row_clock.available() == Err(Error::ReservationRequired)
    }
    pub const fn epoch(&self) -> u64 {
        self.epoch
    }

    pub fn open_session(&mut self) -> Result<SessionId, Error> {
        if self.poisoned {
            return Err(Error::Corrupt);
        }
        if self.clock.available() == Err(Error::Exhausted) {
            return Err(Error::Exhausted);
        }
        let next = self.next_session.checked_add(1).ok_or(Error::Exhausted)?;
        let id = SessionId(next);
        match self.sessions.insert_with_id(id, Session { sequence: 0 }) {
            super::super::WarpTableInsertWithIdOutcome::Inserted { .. } => {
                self.next_session = next;
                Ok(id)
            }
            super::super::WarpTableInsertWithIdOutcome::Full { .. } => Err(Error::Full),
            super::super::WarpTableInsertWithIdOutcome::AlreadyPresent { .. }
            | super::super::WarpTableInsertWithIdOutcome::UniqueConflict { .. } => {
                Err(Error::Corrupt)
            }
        }
    }
    pub fn operation(&self, session: SessionId, sequence: u64) -> OperationId {
        OperationId {
            history: self.history,
            epoch: self.epoch,
            session,
            sequence,
        }
    }
    pub fn next_operation(&self, session: SessionId) -> Result<OperationId, Error> {
        match self.sessions.get(session) {
            WarpTableGetOutcome::Present { row } => Ok(self.operation(
                session,
                row.sequence.checked_add(1).ok_or(Error::Exhausted)?,
            )),
            WarpTableGetOutcome::Absent { .. } => Err(Error::UnknownSession),
        }
    }
    pub fn close_session(&mut self, session: SessionId) -> Result<(), Error> {
        match self.sessions.remove(session) {
            WarpTableRemoveOutcome::Removed { .. } => Ok(()),
            WarpTableRemoveOutcome::Absent { .. } => Err(Error::UnknownSession),
        }
    }
    /// Call only after the backend has durably extended the reservation.
    pub fn reserved(&mut self, through: u64) -> Result<(), Error> {
        self.clock.reserve(through)?;
        self.row_clock.reserve(through)?;
        Ok(())
    }
    pub fn record(&self, operation: OperationId) -> Option<&Record<Row, Row::Id>> {
        match self.records.get(operation) {
            WarpTableGetOutcome::Present { row } => Some(row.record),
            WarpTableGetOutcome::Absent { .. } => None,
        }
    }
    pub fn inquire(&self, operation: OperationId, required: Guarantee) -> Confirmation {
        if self.poisoned {
            return Confirmation::Unavailable(Error::Corrupt);
        }
        if required == Guarantee::ReplicatedDurable && self.replication == Replication::Disabled {
            return Confirmation::Unsupported;
        }
        let Some(record) = self.record(operation) else {
            return Confirmation::Unknown;
        };
        let established = if self.remote_revision >= record.revision
            && self.replication == Replication::Required
        {
            Guarantee::ReplicatedDurable
        } else if self.local_revision >= record.revision {
            Guarantee::LocallyDurable
        } else {
            Guarantee::Accepted
        };
        let met = match required {
            Guarantee::Accepted => true,
            Guarantee::LocallyDurable => established != Guarantee::Accepted,
            Guarantee::ReplicatedDurable => established == Guarantee::ReplicatedDurable,
        };
        if met {
            Confirmation::Confirmed {
                revision: record.revision,
                established,
            }
        } else {
            Confirmation::Pending {
                revision: record.revision,
                established,
            }
        }
    }
    /// Explicit requester retirement; the request can never be admitted again.
    pub fn retire(&mut self, operation: OperationId) -> Result<Record<Row, Row::Id>, Error> {
        let required = match self.replication {
            Replication::Disabled => Guarantee::LocallyDurable,
            Replication::Required => Guarantee::ReplicatedDurable,
        };
        if !matches!(
            self.inquire(operation, required),
            Confirmation::Confirmed { .. }
        ) {
            return Err(Error::Busy);
        }
        self.retire_retained(operation)
    }

    fn retire_retained(&mut self, operation: OperationId) -> Result<Record<Row, Row::Id>, Error> {
        let (_, removed, _) = super::super::remove::require_removal(
            self.records.remove(operation),
            Error::ExpiredOperation,
        )?;
        self.order.retain(|id| *id != operation);
        Ok(removed.record)
    }
    pub fn prepare(&mut self, guarantee: Guarantee) -> Result<Option<Attempt>, Error> {
        if self.poisoned {
            return Err(Error::Corrupt);
        }
        let (current, through, pending) = match guarantee {
            Guarantee::Accepted => return Err(Error::Unsupported),
            Guarantee::LocallyDurable => (
                self.local_revision,
                self.accepted_revision,
                &mut self.local_attempt,
            ),
            Guarantee::ReplicatedDurable => {
                if self.replication == Replication::Disabled {
                    return Err(Error::Unsupported);
                }
                (
                    self.remote_revision,
                    self.local_revision,
                    &mut self.remote_attempt,
                )
            }
        };
        if pending.is_some() {
            return Err(Error::Busy);
        }
        if current == through && (guarantee != Guarantee::ReplicatedDurable || self.remote_ready) {
            return Ok(None);
        }
        let sequence = self.next_attempt.checked_add(1).ok_or(Error::Exhausted)?;
        let attempt = Attempt {
            history: self.history,
            epoch: self.epoch,
            sequence,
            guarantee,
            through,
        };
        self.next_attempt = sequence;
        *pending = Some(attempt);
        Ok(Some(attempt))
    }
    pub fn batch(
        &self,
        attempt: Attempt,
    ) -> Result<impl Iterator<Item = &Record<Row, Row::Id>>, Error> {
        let (pending, after) = match attempt.guarantee {
            Guarantee::LocallyDurable => (self.local_attempt, self.local_revision),
            Guarantee::ReplicatedDurable => (self.remote_attempt, self.remote_revision),
            Guarantee::Accepted => return Err(Error::Unsupported),
        };
        if pending != Some(attempt) {
            return Err(Error::StaleCompletion);
        }
        Ok(self
            .order
            .iter()
            .filter_map(|id| self.record(*id))
            .filter(move |record| record.revision > after && record.revision <= attempt.through))
    }
    /// Allocating physical checkpoint construction, outside the semantic hot path.
    pub fn checkpoint(&self) -> Result<Recovery<Row, Row::Id>, Error> {
        if self.poisoned {
            return Err(Error::Corrupt);
        }
        let mut rows = Vec::new();
        rows.try_reserve_exact(self.local.table().len())
            .map_err(|_| Error::Allocation)?;
        for id in self.local.table().ids() {
            match self.local.table().get(*id) {
                WarpTableGetOutcome::Present { row } => rows.push((*id, Row::own(row))),
                WarpTableGetOutcome::Absent { .. } => return Err(Error::Corrupt),
            }
        }
        let mut outcomes = Vec::new();
        outcomes
            .try_reserve_exact(self.records.len())
            .map_err(|_| Error::Allocation)?;
        for id in &self.order {
            let record = self.record(*id).ok_or(Error::Corrupt)?;
            if record.revision <= self.local_revision {
                outcomes.push(record.clone());
            }
        }
        Ok(Recovery {
            history: self.history,
            revision: self.local_revision,
            reserved_through: self.clock.reserved_through(),
            rows,
            outcomes,
        })
    }
}

pub struct Submit<Row: PersistentRow> {
    pub receipt: Option<CommandReceipt>,
    pub operation: OperationId,
    pub expected: Revision,
    pub guarantee: Guarantee,
    pub changes: Vec<Change<Row, Row::Id>>,
}
impl<Row: PersistentRow> StepInputOf<Authority<Row>> for Submit<Row> {
    type Outcome = Result<Confirmation, (Self, Error)>;
    fn step(self, machine: &mut Authority<Row>) -> Self::Outcome {
        match machine.submit(&self) {
            Ok(Some(confirmation)) => Ok(confirmation),
            Err(error) => Err((self, error)),
            Ok(None) => {
                let revision = machine.accepted_revision;
                let operation = self.operation;
                let guarantee = self.guarantee;
                let record = Record {
                    receipt: self.receipt,
                    operation,
                    revision,
                    expected: self.expected,
                    changes: self.changes,
                };
                if let Err(rejected) = retention_result(
                    machine
                        .records
                        .insert_with_id(operation, Retained { record }),
                    guarantee,
                ) {
                    machine.poisoned = true;
                    return Err((rejected, Error::Corrupt));
                }
                machine.order.push_back(operation);
                machine.poisoned = false;
                Ok(machine.inquire(operation, guarantee))
            }
        }
    }
}

#[allow(
    clippy::result_large_err,
    reason = "retention refusal returns the owned submission without allocating"
)]
fn retention_result<Row: PersistentRow>(
    outcome: super::super::WarpTableInsertWithIdOutcome<OperationId, Retained<Row>>,
    guarantee: Guarantee,
) -> Result<(), Submit<Row>> {
    use super::super::WarpTableInsertWithIdOutcome;
    let record = match outcome {
        WarpTableInsertWithIdOutcome::Inserted { .. } => return Ok(()),
        WarpTableInsertWithIdOutcome::AlreadyPresent { rejected, .. }
        | WarpTableInsertWithIdOutcome::Full { rejected, .. } => rejected.record,
        WarpTableInsertWithIdOutcome::UniqueConflict { conflict, .. } => match conflict {},
    };
    Err(Submit {
        receipt: record.receipt,
        operation: record.operation,
        expected: record.expected,
        guarantee,
        changes: record.changes,
    })
}

impl<Row: PersistentRow> Authority<Row> {
    /// Check identity and admission before a command handler is evaluated.
    pub fn admit_command(&self, operation: OperationId, guarantee: Guarantee) -> Result<(), Error> {
        if self.poisoned {
            return Err(Error::Corrupt);
        }
        if guarantee == Guarantee::ReplicatedDurable && self.replication == Replication::Disabled {
            return Err(Error::Unsupported);
        }
        if operation.history != self.history {
            return Err(Error::InvalidRecovery);
        }
        if operation.epoch != self.epoch {
            return Err(Error::ExpiredOperation);
        }
        let sequence = match self.sessions.get(operation.session) {
            WarpTableGetOutcome::Present { row } => *row.sequence,
            WarpTableGetOutcome::Absent { .. } => return Err(Error::UnknownSession),
        };
        if operation.sequence <= sequence {
            return Err(Error::ExpiredOperation);
        }
        if self.records.len() == self.limits.operations.value().get() as usize {
            return Err(Error::Full);
        }
        self.clock.available()?;
        Ok(())
    }
    fn submit(&mut self, request: &Submit<Row>) -> Result<Option<Confirmation>, Error> {
        if self.poisoned {
            return Err(Error::Corrupt);
        }
        if request.guarantee == Guarantee::ReplicatedDurable
            && self.replication == Replication::Disabled
        {
            return Err(Error::Unsupported);
        }
        if let Some(record) = self.record(request.operation) {
            if record.expected != request.expected
                || record.changes != request.changes
                || record.receipt != request.receipt
            {
                return Err(Error::ConflictingRetry);
            }
            return Ok(Some(self.inquire(request.operation, request.guarantee)));
        }
        self.admit_command(request.operation, request.guarantee)?;
        if request.expected != self.accepted_revision {
            return Err(Error::StaleRevision);
        }
        if let Some(receipt) = &request.receipt
            && receipt.retained_bytes().ok_or(Error::InvalidBounds)? > self.limits.receipt_bytes
        {
            return Err(Error::InvalidBounds);
        }
        if (request.changes.is_empty() && request.receipt.is_none())
            || request.changes.capacity() > self.limits.changes
        {
            return Err(Error::InvalidBounds);
        }
        for change in &request.changes {
            match change {
                Change::Insert { row, .. } | Change::Replace { row, .. }
                    if row.retained_bytes() > self.limits.row_bytes =>
                {
                    return Err(Error::InvalidBounds);
                }
                Change::Insert { .. } | Change::Replace { .. } | Change::Remove { .. } => {}
            }
        }
        let mut next_clock = self.clock;
        let revision = next_clock.issue()?;
        self.accepted
            .sequence
            .checked_add(u64::try_from(request.changes.len()).map_err(|_| Error::Exhausted)?)
            .ok_or(Error::Exhausted)?;
        self.poisoned = true;
        if let Err(error) = changes::apply(&mut self.staging, &request.changes, &mut self.undo) {
            if error != Error::Corrupt {
                self.poisoned = false;
            }
            return Err(error);
        }
        // Only touched rows are copied. Both private tables are valid before
        // publication; callers cannot run another step inside this transition.
        if changes::apply_observed(
            &mut self.accepted,
            &request.changes,
            &mut self.incarnation_scratch,
        )
        .is_err()
        {
            self.poisoned = true;
            return Err(Error::Corrupt);
        }
        self.finish_submission(request.operation, revision, next_clock)
    }

    fn finish_submission(
        &mut self,
        operation: OperationId,
        revision: u64,
        next_clock: RevisionClock,
    ) -> Result<Option<Confirmation>, Error> {
        let (_id, _row, ()) = super::super::update::require_update(
            self.sessions.update(operation.session, |mut row| {
                *row.sequence_mut() = operation.sequence
            }),
            Error::Corrupt,
        )?;
        self.accepted_revision = Revision(revision);
        self.clock = next_clock;
        Ok(None)
    }
}

pub enum ReadOutcome<'a, Row: PersistentRow> {
    Ready {
        history: HistoryId,
        revision: Revision,
        guarantee: Guarantee,
        table: &'a WarpTable<Row>,
    },
    Pending,
    Refused(Error),
}

impl<'a, Row: PersistentRow> ReadOutcome<'a, Row> {
    pub(super) fn into_table(self) -> Result<&'a WarpTable<Row>, Error> {
        match self {
            Self::Ready { table, .. } => Ok(table),
            Self::Pending => Err(Error::Busy),
            Self::Refused(error) => Err(error),
        }
    }
}

impl<Row: PersistentRow> Authority<Row> {
    pub fn read(&self, request: ReadRequest) -> ReadOutcome<'_, Row> {
        if self.poisoned {
            return ReadOutcome::Refused(Error::Corrupt);
        }
        let (revision, table) = match request.guarantee {
            Guarantee::Accepted => (self.accepted_revision, self.accepted.table()),
            Guarantee::LocallyDurable => (self.local_revision, self.local.table()),
            Guarantee::ReplicatedDurable => {
                if self.replication == Replication::Disabled {
                    return ReadOutcome::Refused(Error::Unsupported);
                }
                if !self.remote_ready {
                    return ReadOutcome::Pending;
                }
                let Some(remote) = &self.remote else {
                    return ReadOutcome::Refused(Error::Corrupt);
                };
                (self.remote_revision, remote.table())
            }
        };
        ReadOutcome::Ready {
            history: self.history,
            revision,
            guarantee: request.guarantee,
            table,
        }
    }

    /// Read-only observation at the same guarantee boundary as ordinary reads.
    /// Successful commits record identities internally, including whole groups.
    pub fn observation(&self, guarantee: Guarantee) -> Result<&ObservedTable<Row>, Error> {
        // An exhausted store remains readable, but cannot issue a fresh stream
        // identity after restart. Never reuse an incarnation for a new stream.
        if !self.observation_incarnation_valid {
            return Err(Error::Exhausted);
        }
        self.read(ReadRequest { guarantee }).into_table()?;
        match guarantee {
            Guarantee::Accepted => Ok(&self.accepted),
            Guarantee::LocallyDurable => Ok(&self.local),
            Guarantee::ReplicatedDurable => self.remote.as_ref().ok_or(Error::Unsupported),
        }
    }
}

enum DurableDestination {
    Local,
    Remote,
}

pub struct Complete {
    pub attempt: Attempt,
    pub result: PhysicalResult,
}
impl<Row: PersistentRow> StepInputOf<Authority<Row>> for Complete {
    type Outcome = Result<(), Error>;
    fn step(self, machine: &mut Authority<Row>) -> Self::Outcome {
        if machine.poisoned {
            return Err(Error::Corrupt);
        }
        let attempt = self.attempt;
        let (pending, after, destination) = match attempt.guarantee {
            Guarantee::Accepted => return Err(Error::Unsupported),
            Guarantee::LocallyDurable => (
                &mut machine.local_attempt,
                machine.local_revision,
                DurableDestination::Local,
            ),
            Guarantee::ReplicatedDurable => (
                &mut machine.remote_attempt,
                machine.remote_revision,
                DurableDestination::Remote,
            ),
        };
        if *pending != Some(attempt) {
            return Err(Error::StaleCompletion);
        }
        *pending = None;
        if self.result == PhysicalResult::Unresolved {
            return Ok(());
        }
        machine.poisoned = true;
        let table = match destination {
            DurableDestination::Local => &mut machine.local,
            DurableDestination::Remote => machine.remote.as_mut().ok_or(Error::Unsupported)?,
        };
        for id in &machine.order {
            let record = match machine.records.get(*id) {
                WarpTableGetOutcome::Present { row } => row.record,
                WarpTableGetOutcome::Absent { .. } => {
                    machine.poisoned = true;
                    return Err(Error::Corrupt);
                }
            };
            // Every materialization already contains the recovered checkpoint.
            // Its retained outcomes answer inquiries; they are never replayed
            // onto that image, including at the exhausted u64 revision boundary.
            if record.revision > after
                && record.revision <= attempt.through
                && record.revision > machine.restored_revision
                && changes::apply_observed(table, &record.changes, &mut machine.incarnation_scratch)
                    .is_err()
            {
                machine.poisoned = true;
                return Err(Error::Corrupt);
            }
        }
        match destination {
            DurableDestination::Local => machine.local_revision = attempt.through,
            DurableDestination::Remote => {
                machine.remote_revision = attempt.through;
                machine.remote_ready = true;
            }
        }
        machine.poisoned = false;
        Ok(())
    }
}

impl<Row: PersistentRow<Id = super::super::WarpId<Row>>> Authority<Row> {
    /// Generated identities use the same durable reservation boundary as revisions,
    /// with an independent cursor. Abandoned allocations are never reused.
    pub fn allocate_id(&mut self) -> Result<super::super::WarpId<Row>, Error> {
        if self.poisoned {
            return Err(Error::Corrupt);
        }
        let value = self.row_clock.issue()?;
        let value = core::num::NonZeroU64::new(value).ok_or(Error::Corrupt)?;
        Ok(super::super::WarpId::new(value))
    }
}

#[cfg(test)]
mod tests;