liminal-server 0.25.4

//! Durable connection-incarnation transition-input stream.
//!
//! The log stores canonical inputs, never server-computed allocator snapshots.
//! Replay re-executes every transition through `liminal-protocol` from a resume
//! seed that is READ OFF THE STREAM'S OWN KEY, so an independently valid but
//! regressing scalar header is not representable: no event in any schema
//! version carries a header field for one to be written into. The server owns
//! only event framing, append/flush mechanics, and the log grammar: the first
//! event of a stream is a startup, allocations are legal only after a startup,
//! and each allocation records the immutable reference bound which validated
//! that input, so later configuration reductions do not reinterpret valid
//! history. All increment, collision, reset, and exhaustion decisions remain in
//! the shared protocol crate.
//!
//! History is bounded by policy rather than accumulated: each server
//! incarnation owns its own epoch stream and a boot replays only that one, so
//! neither the stream of record nor the boot's read cost grows with everything
//! the estate has ever appended. [`epoch`] holds that rotation and the
//! non-regression argument behind it; the pre-rotation genesis stream still
//! opens unchanged and rotates forward on its first boot.
//!
//! The transition-input format uses the `/v2` stream namespace. The earlier
//! `/v1` checkpoint stored scalar `LPIC` snapshots, so reusing that key for
//! canonical `LPIE` events would make existing durable data ambiguous.
//!
//! Replay advances through bounded pages and stops at the first empty page. It
//! never sleeps, retries, or treats an empty page as a polling signal.

mod codec;
mod epoch;
mod error;
mod replay;

use liminal::durability::settle::{append_settled, flush_settled};
use std::sync::Arc;

use liminal::durability::DurableStore;
use liminal_protocol::{
    lifecycle::{
        ConnectionIncarnationAllocationDecision, ConnectionIncarnationAllocator,
        ConnectionIncarnationAllocatorRestore, ConnectionOrdinalExhaustion,
        DurableIncarnationReferences, ServerIncarnationStartupDecision,
        allocate_connection_incarnation, prepare_server_incarnation_startup,
    },
    outcome::ConnectionIncarnationExhausted,
    wire::ConnectionIncarnation,
};

pub(in crate::server) use super::dispatch::ConnectionFateClass;

use codec::{IncarnationEvent, decode_event, encode_event};
use epoch::{EPOCH_STARTUP_SEQUENCE, EpochSource};
use replay::{
    AllocationGeneration, ReplayAccumulator, ReplayedIncarnationState, UnmatchedConnectionFates,
    retain_active_allocations,
};

pub(in crate::server) use epoch::AnchorSwap;
pub(in crate::server) use error::{
    DurableWriteReach, IncarnationOperationError, IncarnationStreamError,
};
pub(in crate::server) use replay::ConnectionFateIntent;

#[cfg(test)]
pub(in crate::server) use codec::{
    encode_allocate_event_fixture, encode_complete_connection_fate_event_fixture,
    encode_frozen_v1_startup_event_fixture, encode_open_connection_fate_event_fixture,
    encode_startup_event_fixture,
};

const REPLAY_PAGE_SIZE: usize = 256;

/// Cold-start result after applying the protocol-owned server increment.
#[derive(Debug)]
pub(in crate::server) enum IncarnationStartup {
    /// The startup input was appended and flushed before this allocator became usable.
    Started(StartedIncarnationStream),
    /// Unmatched historical Opens must Complete before a new Startup can append.
    RecoveryRequired(ConnectionFateRecovery),
    /// The protocol selected terminal server-incarnation exhaustion.
    Exhausted(ConnectionIncarnationExhausted),
}

/// Persisted connection-ordinal result returned only after its input-event commit.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(in crate::server) enum IncarnationAllocation {
    /// One collision-free pair was durably allocated.
    Allocated {
        /// Exact protocol-emitted pair safe to publish.
        connection_incarnation: ConnectionIncarnation,
        /// Number of exact durable collisions skipped by the protocol.
        skipped_collisions: usize,
    },
    /// The protocol selected stable connection-ordinal exhaustion.
    Exhausted(ConnectionIncarnationExhausted),
}

/// Unstarted handle for the server-wide append-only incarnation history.
#[derive(Debug)]
pub(in crate::server) struct IncarnationStream {
    store: Arc<dyn DurableStore>,
    maximum_references: usize,
    #[cfg(test)]
    replay_seed: Option<ConnectionIncarnationAllocatorRestore>,
}

impl IncarnationStream {
    /// Binds the server-wide event stream and complete-reference bound to storage.
    #[must_use]
    pub(in crate::server) const fn new(
        store: Arc<dyn DurableStore>,
        maximum_references: usize,
    ) -> Self {
        Self {
            store,
            maximum_references,
            #[cfg(test)]
            replay_seed: None,
        }
    }

    /// Returns the stable namespaced genesis stream key.
    #[must_use]
    #[cfg(test)]
    pub(in crate::server) const fn stream_key() -> &'static str {
        epoch::GENESIS_STREAM_KEY
    }

    /// Returns the stream key holding one server incarnation's own epoch.
    #[must_use]
    #[cfg(test)]
    pub(in crate::server) fn epoch_stream_key(server_incarnation: u64) -> String {
        epoch::epoch_stream_key(server_incarnation)
    }

    /// Returns the compare-and-swap key holding the epoch anchor hint.
    #[must_use]
    #[cfg(test)]
    pub(in crate::server) const fn epoch_anchor_key() -> &'static str {
        epoch::EPOCH_ANCHOR_KEY
    }

    /// The stream key a cold boot would replay, as the durable store says now.
    ///
    /// Replay's own choice of source, asked as a question instead of taken as a
    /// consequence, so a measurement of what boot costs can name the stream it
    /// will actually read rather than assuming one.
    ///
    /// # Errors
    ///
    /// Returns [`IncarnationStreamError`] when the store cannot be read.
    #[cfg(test)]
    pub(in crate::server) async fn stream_key_of_record(
        store: &Arc<dyn DurableStore>,
    ) -> Result<String, IncarnationStreamError> {
        Ok(EpochSource::from_anchor(epoch::read_anchor(store).await?).stream_key())
    }

    /// Builds a test-only stream whose replay begins at a validated high state.
    ///
    /// Production has no raw-header restoration path. This seam lets exhaustion
    /// tests reach `u64::MAX` without appending billions of canonical inputs; all
    /// events after the seed still replay through public protocol transitions.
    #[cfg(test)]
    pub(in crate::server) fn seeded_for_test(
        store: Arc<dyn DurableStore>,
        maximum_references: usize,
        replay_seed: ConnectionIncarnationAllocatorRestore,
    ) -> Result<Self, IncarnationStreamError> {
        let _validated = ConnectionIncarnationAllocator::try_restore(replay_seed)?;
        Ok(Self {
            store,
            maximum_references,
            replay_seed: Some(replay_seed),
        })
    }

    /// Replays the stream of record and rotates this process into its own epoch.
    ///
    /// The replayed state is the previous incarnation's; the Startup this
    /// appends is the first row of a NEW stream, named by the incarnation the
    /// protocol just produced. Nothing is deleted and nothing is rewritten —
    /// the epoch below simply stops being read.
    ///
    /// An absent production history begins at the fixed zero genesis validated
    /// by [`ConnectionIncarnationAllocator::try_restore`]. Append or flush
    /// ambiguity consumes this handle and is never retried here.
    ///
    /// # Errors
    ///
    /// Returns [`IncarnationStreamError`] for malformed events, an illegal event
    /// chain, invalid protocol state, append conflict, or failed flush.
    pub(in crate::server) async fn startup(
        self,
    ) -> Result<IncarnationStartup, IncarnationStreamError> {
        let (source, replayed) = self.replay_of_record().await?;
        if !replayed.unmatched.by_sequence.is_empty() {
            return Ok(IncarnationStartup::RecoveryRequired(
                ConnectionFateRecovery {
                    store: self.store,
                    stream_key: source.stream_key(),
                    source,
                    maximum_references: self.maximum_references,
                    allocator: replayed.allocator,
                    next_sequence: replayed.next_sequence,
                    unmatched: replayed.unmatched,
                },
            ));
        }
        rotate_into_successor_epoch(
            &self.store,
            self.maximum_references,
            source,
            replayed.allocator,
        )
        .await
    }

    /// Locates the durable stream of record and replays it.
    ///
    /// The anchor is a hint. This walks it FORWARD over every successor epoch
    /// the store actually holds before reading anything, so a boot after a
    /// crash between an epoch's fsync and its anchor swap adopts that epoch
    /// rather than re-minting its incarnation. The genesis source is the one
    /// whose incarnation its key does not name, so its successor question can
    /// only be asked after its replay — which is why the loop exists.
    ///
    /// An epoch that replays to an incarnation other than the one its key names
    /// is refused: the key IS the resume seed, so history that disagrees with
    /// it is history the seed does not describe.
    async fn replay_of_record(
        &self,
    ) -> Result<(EpochSource, ReplayedIncarnationState), IncarnationStreamError> {
        let mut anchor = epoch::read_anchor(&self.store).await?;
        loop {
            anchor = epoch::roll_anchor_forward(&self.store, anchor).await?;
            let source = EpochSource::from_anchor(anchor);
            let replayed = self.replay(source).await?;
            let replayed_server_incarnation = replayed.allocator.as_restore().server_incarnation;
            if let EpochSource::Epoch(named) = source {
                if replayed_server_incarnation == named.get() {
                    return Ok((source, replayed));
                }
                return Err(IncarnationStreamError::EpochHeaderMismatch {
                    epoch_server_incarnation: named.get(),
                    replayed_server_incarnation,
                });
            }
            let Some(successor) = replayed_server_incarnation.checked_add(1) else {
                return Ok((source, replayed));
            };
            if !epoch::epoch_exists(&self.store, successor).await? {
                return Ok((source, replayed));
            }
            epoch::advance_anchor(&self.store, anchor, successor).await?;
            anchor = successor;
        }
    }

    async fn replay(
        &self,
        source: EpochSource,
    ) -> Result<ReplayedIncarnationState, IncarnationStreamError> {
        let started = std::time::Instant::now();
        let stream_key = source.stream_key();
        tracing::info!(
            step = "incarnation-stream recovery",
            stream_key = %stream_key,
            "incarnation-stream replay started"
        );
        #[cfg(test)]
        let replay_seed = self.replay_seed;
        #[cfg(not(test))]
        let replay_seed = None;
        let (allocator, has_started) = Self::initial_replay_state(replay_seed, source)?;
        let mut next_sequence = 0_u64;
        let generation = has_started.then(|| {
            AllocationGeneration::new(next_sequence, allocator.as_restore().server_incarnation)
        });
        let mut replay = ReplayAccumulator {
            allocator,
            generation,
            unmatched: UnmatchedConnectionFates::default(),
            has_started,
        };
        loop {
            let entries = self
                .store
                .read_from(&stream_key, next_sequence, REPLAY_PAGE_SIZE)
                .await?;
            if entries.is_empty() {
                break;
            }
            for entry in entries {
                if entry.sequence != next_sequence {
                    return Err(IncarnationStreamError::EventSequence {
                        expected: next_sequence,
                        actual: entry.sequence,
                    });
                }
                let event = decode_event(&entry.payload)?;
                replay = replay.apply(event, entry.sequence)?;
                next_sequence = next_sequence
                    .checked_add(1)
                    .ok_or(IncarnationStreamError::StreamSequenceExhausted)?;
            }
        }
        tracing::info!(
            step = "incarnation-stream recovery",
            rows = next_sequence,
            unmatched_opens = replay.unmatched.by_sequence.len(),
            elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
            "incarnation-stream replay finished"
        );
        Ok(ReplayedIncarnationState {
            allocator: replay.allocator,
            next_sequence,
            generation: replay.generation,
            unmatched: replay.unmatched,
            has_started: replay.has_started,
        })
    }

    /// Re-establishes ground truth by READING, after an ambiguous durable write.
    ///
    /// An ambiguous append or flush leaves exactly one thing unknown — whether
    /// the bytes landed — and that question has an answer sitting in the store.
    /// This replays this process's OWN epoch and rebuilds a
    /// [`StartedIncarnationStream`] from what is actually there, so a process
    /// that lost track of its own sequence finds it again instead of demanding
    /// an operator restart it.
    ///
    /// It appends NOTHING, and in particular it does not rotate. This process's
    /// Startup is already durable in its own epoch; a second one would burn a
    /// server incarnation and re-frame every live allocation as historical.
    ///
    /// The replayed unmatched-Open set is carried through rather than discarded:
    /// connections that were mid-teardown when the write went ambiguous still
    /// have their Complete to append, and a resumed handle that had forgotten
    /// their Opens would refuse those Completes as absent. An Open whose own
    /// append was the ambiguous one stays unmatched here and is completed by the
    /// ordinary `RecoveryRequired` path at the next process startup.
    ///
    /// `expected_server_incarnation` is the incarnation this process durably
    /// started under. Under rotation that question is answered without reading
    /// this process's history at all: if the anchor has passed it, or if the
    /// successor epoch exists, another process has started over this estate and
    /// resuming would allocate under a foreign incarnation. That is refused
    /// rather than papered over.
    ///
    /// # Errors
    ///
    /// Returns [`IncarnationStreamError`] when the store cannot be read, when the
    /// replayed history is malformed, when it carries no Startup at all, or when
    /// the server incarnation has moved.
    pub(in crate::server) async fn resume_after_ambiguous_write(
        self,
        expected_server_incarnation: u64,
    ) -> Result<StartedIncarnationStream, IncarnationStreamError> {
        let anchor = epoch::read_anchor(&self.store).await?;
        if anchor > expected_server_incarnation {
            return Err(IncarnationStreamError::ResumeServerIncarnationMoved {
                expected: expected_server_incarnation,
                actual: anchor,
            });
        }
        if let Some(successor) = expected_server_incarnation.checked_add(1)
            && epoch::epoch_exists(&self.store, successor).await?
        {
            return Err(IncarnationStreamError::ResumeServerIncarnationMoved {
                expected: expected_server_incarnation,
                actual: successor,
            });
        }
        let source = EpochSource::from_anchor(expected_server_incarnation);
        let replayed = self.replay(source).await?;
        if !replayed.has_started {
            return Err(IncarnationStreamError::MissingStartup);
        }
        let header = replayed.allocator.as_restore();
        if header.server_incarnation != expected_server_incarnation {
            return Err(IncarnationStreamError::ResumeServerIncarnationMoved {
                expected: expected_server_incarnation,
                actual: header.server_incarnation,
            });
        }
        let generation = replayed
            .generation
            .ok_or(IncarnationStreamError::MissingStartup)?;
        Ok(StartedIncarnationStream {
            store: self.store,
            stream_key: source.stream_key(),
            maximum_references: self.maximum_references,
            header,
            next_sequence: replayed.next_sequence,
            generation,
            unmatched: replayed.unmatched,
            anchor_swap: None,
        })
    }

    fn initial_replay_state(
        replay_seed: Option<ConnectionIncarnationAllocatorRestore>,
        source: EpochSource,
    ) -> Result<(ConnectionIncarnationAllocator, bool), IncarnationStreamError> {
        if let Some(seed) = replay_seed {
            return Ok((ConnectionIncarnationAllocator::try_restore(seed)?, true));
        }
        Ok((
            ConnectionIncarnationAllocator::try_restore(source.resume_seed())?,
            false,
        ))
    }

    #[cfg(test)]
    pub(in crate::server) async fn resume_started_for_test(
        self,
    ) -> Result<StartedIncarnationStream, IncarnationStreamError> {
        let (source, replayed) = self.replay_of_record().await?;
        if !replayed.has_started {
            return Err(IncarnationStreamError::MissingStartup);
        }
        if !replayed.unmatched.by_sequence.is_empty() {
            return Err(IncarnationStreamError::UnmatchedConnectionFates {
                count: replayed.unmatched.by_sequence.len(),
            });
        }
        let header = replayed.allocator.as_restore();
        Ok(StartedIncarnationStream {
            store: self.store,
            stream_key: source.stream_key(),
            maximum_references: self.maximum_references,
            header,
            next_sequence: replayed.next_sequence,
            // A replay that carried no generation record falls back to
            // generation 0, and the two cases that produce it are
            // indistinguishable here: a legitimate cold start, which has none
            // yet, and a durable stream truncated before its generation record,
            // whose real generation is unknown and higher. Nothing at this point
            // can tell them apart — the evidence that would separate them is the
            // very record that is missing — so the truncated stream restarts
            // numbering from 0 without a word.
            generation: replayed
                .generation
                .unwrap_or_else(|| AllocationGeneration::new(0, header.server_incarnation)),
            unmatched: UnmatchedConnectionFates::default(),
            anchor_swap: None,
        })
    }

    /// Read-only WAL census: unlike startup, this does not append a new server
    /// incarnation or repair anything. Tests use it to prove a named Open gets
    /// its durable Complete rather than merely disappearing from a live index.
    #[cfg(test)]
    pub(in crate::server) async fn unmatched_intents_for_test(
        &self,
    ) -> Result<Vec<ConnectionFateIntent>, IncarnationStreamError> {
        let (_, replayed) = self.replay_of_record().await?;
        Ok(replayed.unmatched.by_sequence.into_values().collect())
    }
}

/// Opens this process's own epoch with the Startup row the protocol produced.
///
/// The ORDER is the crash contract, and it is one-way on purpose. The epoch's
/// Startup is appended and fsynced FIRST, and only then does the anchor swap to
/// name it. A crash between them leaves an epoch that exists but is unanchored,
/// which the next boot's forward walk adopts — the incarnation it published is
/// never handed out twice. The reverse order would leave the anchor naming a
/// stream with nothing in it, and a replay of nothing is a header of zero: that
/// is precisely the regressing header this design refuses to make representable.
///
/// The swap is returned unrun inside the started stream (SRV-048 R1), so the
/// start publishes on the flushed Startup row alone and the swap is settled
/// after admission opens; the order between the two is unchanged.
async fn rotate_into_successor_epoch(
    store: &Arc<dyn DurableStore>,
    maximum_references: usize,
    source: EpochSource,
    allocator: ConnectionIncarnationAllocator,
) -> Result<IncarnationStartup, IncarnationStreamError> {
    match prepare_server_incarnation_startup(allocator) {
        ServerIncarnationStartupDecision::Fsync(intent) => {
            let server_incarnation = intent.server_incarnation();
            let stream_key = epoch::epoch_stream_key(server_incarnation);
            let payload = encode_event(&IncarnationEvent::Startup)?;
            let appending = std::time::Instant::now();
            let next_sequence =
                append_and_flush(store, &stream_key, EPOCH_STARTUP_SEQUENCE, payload).await?;
            // The stretch between the replay's line and the repair's: one
            // flushed append, named so a slow start says so (the start_stages
            // gap row). The anchor's swap is not in it: it is carried unrun
            // and settled after admission opens (SRV-048 R1), on a line of
            // its own.
            tracing::info!(
                server_incarnation,
                append_ms = u64::try_from(appending.elapsed().as_millis()).unwrap_or(u64::MAX),
                "incarnation-stream epoch rotated"
            );
            let header = intent.complete_after_fsync().as_restore();
            Ok(IncarnationStartup::Started(StartedIncarnationStream {
                store: Arc::clone(store),
                stream_key,
                maximum_references,
                header,
                next_sequence,
                generation: AllocationGeneration::new(
                    EPOCH_STARTUP_SEQUENCE,
                    header.server_incarnation,
                ),
                unmatched: UnmatchedConnectionFates::default(),
                anchor_swap: Some(AnchorSwap::new(
                    Arc::clone(store),
                    source.anchor_value(),
                    server_incarnation,
                )),
            }))
        }
        ServerIncarnationStartupDecision::Exhausted(exhausted) => {
            Ok(IncarnationStartup::Exhausted(exhausted.outcome()))
        }
    }
}

/// Exclusive startup owner while historical Opens are completed under their persisted bound.
///
/// The Completes it appends go to the stream that carries their Opens — the
/// epoch of record as it was found — because an Open's identity IS its sequence
/// in that stream. Rotation happens afterwards, in [`Self::finish_startup`],
/// once the unmatched set is empty and nothing is left to carry across.
#[derive(Debug)]
pub struct ConnectionFateRecovery {
    store: Arc<dyn DurableStore>,
    stream_key: String,
    source: EpochSource,
    maximum_references: usize,
    allocator: ConnectionIncarnationAllocator,
    next_sequence: u64,
    unmatched: UnmatchedConnectionFates,
}

impl ConnectionFateRecovery {
    #[must_use]
    pub fn intents(&self) -> Vec<ConnectionFateIntent> {
        self.unmatched.by_sequence.values().cloned().collect()
    }

    pub async fn complete(&mut self, open_sequence: u64) -> Result<(), IncarnationStreamError> {
        if !self.unmatched.by_sequence.contains_key(&open_sequence) {
            return Err(IncarnationStreamError::RecoveryCompleteForAbsentOpen { open_sequence });
        }
        let payload = encode_event(&IncarnationEvent::CompleteConnectionFate {
            open_event_sequence: open_sequence,
        })?;
        let complete_sequence = self.next_sequence;
        self.next_sequence =
            append_and_flush(&self.store, &self.stream_key, complete_sequence, payload).await?;
        self.unmatched.complete(complete_sequence, open_sequence)?;
        Ok(())
    }

    pub async fn finish_startup(self) -> Result<IncarnationStartup, IncarnationStreamError> {
        if !self.unmatched.by_sequence.is_empty() {
            return Err(IncarnationStreamError::RecoveryIncomplete {
                count: self.unmatched.by_sequence.len(),
            });
        }
        rotate_into_successor_epoch(
            &self.store,
            self.maximum_references,
            self.source,
            self.allocator,
        )
        .await
    }
}

/// Started allocator whose latest server-startup input is already durable.
#[derive(Debug)]
pub(in crate::server) struct StartedIncarnationStream {
    store: Arc<dyn DurableStore>,
    /// The epoch this process owns. Every append below goes here and nowhere
    /// else, which is what keeps one incarnation's history in one stream.
    stream_key: String,
    maximum_references: usize,
    header: ConnectionIncarnationAllocatorRestore,
    next_sequence: u64,
    generation: AllocationGeneration,
    unmatched: UnmatchedConnectionFates,
    /// The anchor swap the rotation left unrun; none on a resumed stream or
    /// once taken (SRV-048 R1).
    anchor_swap: Option<AnchorSwap>,
}

impl StartedIncarnationStream {
    /// Current server incarnation durably fsynced by this stream's Startup row.
    #[must_use]
    pub(in crate::server) const fn server_incarnation(&self) -> u64 {
        self.header.server_incarnation
    }

    /// The durable store this stream is bound to.
    ///
    /// Handed out so an owner that has to discard this handle after an ambiguous
    /// write still knows where to go and read ground truth back from. Nothing
    /// else about the handle survives that discard, which is the point.
    #[must_use]
    pub(in crate::server) fn store(&self) -> Arc<dyn DurableStore> {
        Arc::clone(&self.store)
    }

    /// The signed complete-reference bound this stream was constructed with.
    #[must_use]
    pub(in crate::server) const fn maximum_references(&self) -> usize {
        self.maximum_references
    }

    /// Returns the current protocol-derived scalar state for assertions only.
    #[must_use]
    #[cfg(test)]
    pub(super) const fn header(&self) -> ConnectionIncarnationAllocatorRestore {
        self.header
    }

    /// Allocates through the protocol and durably records its complete input.
    ///
    /// The complete reference set is bounded before allocation, encoding, or
    /// append. An already-exhausted decision emits no event. Every successful
    /// allocation and first transition to exhaustion appends and flushes the
    /// exact input before the protocol result becomes live. No failure is retried.
    ///
    /// # Errors
    ///
    /// Returns an [`IncarnationOperationError`] for an over-bound reference set,
    /// invalid current state, event-length overflow, append conflict, inconsistent
    /// assigned sequence, or failed flush. Its [`DurableWriteReach`] says whether
    /// the caller may keep using this handle: only
    /// [`DurableWriteReach::Ambiguous`] consumes it.
    pub(in crate::server) async fn allocate(
        &mut self,
        referenced_incarnations: &[ConnectionIncarnation],
    ) -> Result<IncarnationAllocation, IncarnationOperationError> {
        self.generation
            .establish_bound(self.next_sequence, self.maximum_references)?;
        let references = DurableIncarnationReferences::try_new(
            referenced_incarnations,
            self.maximum_references,
        )?;
        let event = IncarnationEvent::Allocate {
            declared_reference_bound: self.maximum_references,
            referenced_incarnations: referenced_incarnations.to_vec(),
        };
        let payload = encode_event(&event)?;
        let allocator = ConnectionIncarnationAllocator::try_restore(self.header)?;
        match allocate_connection_incarnation(allocator, references) {
            ConnectionIncarnationAllocationDecision::Allocated(allocation) => {
                let connection_incarnation = allocation.connection_incarnation();
                let skipped_collisions = allocation.skipped_collisions();
                let next_sequence =
                    append_and_flush(&self.store, &self.stream_key, self.next_sequence, payload)
                        .await?;
                let resulting = allocation.into_resulting();
                self.header = resulting.as_restore();
                retain_active_allocations(
                    &mut self.generation,
                    referenced_incarnations,
                    Some((connection_incarnation, self.next_sequence)),
                );
                self.next_sequence = next_sequence;
                Ok(IncarnationAllocation::Allocated {
                    connection_incarnation,
                    skipped_collisions,
                })
            }
            ConnectionIncarnationAllocationDecision::Exhausted(exhaustion) => {
                let outcome = exhaustion.outcome();
                let must_append =
                    matches!(&exhaustion, ConnectionOrdinalExhaustion::MarkExhausted(_));
                let next_sequence = if must_append {
                    append_and_flush(&self.store, &self.stream_key, self.next_sequence, payload)
                        .await?
                } else {
                    self.next_sequence
                };
                let resulting = exhaustion.into_resulting();
                self.header = resulting.as_restore();
                if must_append {
                    retain_active_allocations(&mut self.generation, referenced_incarnations, None);
                }
                self.next_sequence = next_sequence;
                Ok(IncarnationAllocation::Exhausted(outcome))
            }
        }
    }

    /// Validates, appends, and flushes one bounded connection-fate Open.
    ///
    /// The caller supplies the signed semantic-conversation limit, not the
    /// observed set length. The Open is staged through the same generation and
    /// unmatched-set validation used by cold replay before any bytes append.
    ///
    /// # Errors
    ///
    /// Returns a typed validation, encoding, append, or flush failure carrying
    /// its [`DurableWriteReach`]. Only [`DurableWriteReach::Ambiguous`] consumes
    /// this handle; a validation refusal leaves it exact and reusable.
    pub(in crate::server) async fn open_connection_fate(
        &mut self,
        connection_incarnation: ConnectionIncarnation,
        class: ConnectionFateClass,
        declared_conversation_bound: usize,
        conversations: &[u64],
    ) -> Result<ConnectionFateIntent, IncarnationOperationError> {
        let event = IncarnationEvent::OpenConnectionFate {
            connection_incarnation,
            class,
            declared_conversation_bound,
            conversations: conversations.to_vec(),
        };
        let payload = encode_event(&event)?;
        let open_sequence = self.next_sequence;
        let mut staged_unmatched = self.unmatched.clone();
        let intent = staged_unmatched.open(
            open_sequence,
            &self.generation,
            connection_incarnation,
            class,
            declared_conversation_bound,
            conversations.to_vec(),
        )?;
        self.next_sequence =
            append_and_flush(&self.store, &self.stream_key, open_sequence, payload).await?;
        self.unmatched = staged_unmatched;
        Ok(intent)
    }

    /// Appends and flushes Complete for one live Open after all targets finish.
    ///
    /// # Errors
    ///
    /// Returns a typed absent-Open, encoding, append, or flush failure carrying
    /// its [`DurableWriteReach`]. As with Open, only an ambiguous durable result
    /// consumes the outer authority; an absent-Open refusal does not.
    pub(in crate::server) async fn complete_connection_fate(
        &mut self,
        open_sequence: u64,
    ) -> Result<(), IncarnationOperationError> {
        let payload = encode_event(&IncarnationEvent::CompleteConnectionFate {
            open_event_sequence: open_sequence,
        })?;
        let complete_sequence = self.next_sequence;
        let mut staged_unmatched = self.unmatched.clone();
        staged_unmatched.complete(complete_sequence, open_sequence)?;
        self.next_sequence =
            append_and_flush(&self.store, &self.stream_key, complete_sequence, payload).await?;
        self.unmatched = staged_unmatched;
        Ok(())
    }
}

/// Appends one canonical event and fsyncs it, reporting how far it got.
///
/// This function IS the durable-write boundary, which is why it is the only
/// place [`DurableWriteReach`] is decided. Everything before `store.append` is
/// arithmetic on this process's own numbers; everything from `store.append`
/// onward may have put bytes on disk that this process cannot see the fate of.
/// The `AssignedSequence` mismatch is deliberately `Ambiguous` even though it is
/// a logic failure rather than an I/O one: the append RETURNED, so it landed
/// somewhere, and where it landed is exactly what is no longer known.
async fn append_and_flush(
    store: &Arc<dyn DurableStore>,
    stream_key: &str,
    expected_sequence: u64,
    payload: Vec<u8>,
) -> Result<u64, IncarnationOperationError> {
    let next_sequence = expected_sequence.checked_add(1).ok_or_else(|| {
        IncarnationOperationError::not_attempted(IncarnationStreamError::StreamSequenceExhausted)
    })?;
    let assigned = append_settled(store.as_ref(), stream_key, payload, expected_sequence)
        .await
        .map_err(|error| IncarnationOperationError::ambiguous(error.into()))?;
    if assigned != expected_sequence {
        return Err(IncarnationOperationError::ambiguous(
            IncarnationStreamError::AssignedSequence {
                expected: expected_sequence,
                actual: assigned,
            },
        ));
    }
    flush_settled(store.as_ref())
        .await
        .map_err(|error| IncarnationOperationError::ambiguous(error.into()))?;
    Ok(next_sequence)
}