ozzy-journal 0.7.0

//! Generation-scoped accounting for complete journal writes and syncs.
//!
//! This does not perform I/O or decide commit. The adapter reports a write only
//! after a complete sealed group, and a sync only after the required barrier.
//! Tickets identify local actions, not hashes, quorum evidence, or disk proofs.

/// One active journal writer incarnation, supplied by the owning runtime.
///
/// Never reuse a value across journals, reopen, or selected-lineage replacement.
/// It is independent of the persisted segment/manifest generation and VSR view.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct JournalGeneration(pub u128);

/// Contiguous group-operation prefix. Zero denotes the empty journal.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord)]
pub struct OpNumber(pub u64);

/// One admitted, indivisible local write group. Fields cannot be forged externally.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WriteTicket {
    generation: JournalGeneration,
    first: OpNumber,
    through: OpNumber,
}

impl WriteTicket {
    /// Writer incarnation that admitted this group.
    pub const fn generation(self) -> JournalGeneration {
        self.generation
    }
    /// First operation in this group, inclusive.
    pub const fn first(self) -> OpNumber {
        self.first
    }
    /// Last operation in this group, inclusive.
    pub const fn through(self) -> OpNumber {
        self.through
    }
}

/// Prefix captured when a synchronization request is issued.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SyncTicket {
    generation: JournalGeneration,
    through: OpNumber,
}

impl SyncTicket {
    /// Writer incarnation that issued this synchronization.
    pub const fn generation(self) -> JournalGeneration {
        self.generation
    }
    /// Complete written prefix this synchronization may acknowledge.
    pub const fn through(self) -> OpNumber {
        self.through
    }
}

/// Read-only local progress. None of these positions implies quorum commit.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct JournalSnapshot {
    /// Current writer incarnation.
    pub generation: JournalGeneration,
    /// Last admitted operation.
    pub accepted: OpNumber,
    /// Last operation in a completely written group.
    pub written: OpNumber,
    /// Last operation covered by successful sync evidence.
    pub durable: OpNumber,
    /// Uncertain I/O outcome blocks subsequent actions until recovery.
    pub faulted: bool,
}

/// Single-writer local completion state. No allocation, runtime, or I/O.
#[derive(Debug)]
pub struct JournalProgress {
    state: JournalSnapshot,
}

impl JournalProgress {
    /// Start from an independently verified and synchronized recovered prefix.
    ///
    /// The caller must complete storage recovery before supplying this prefix.
    /// Bytes visible only in a surviving OS page cache are not stable evidence.
    /// Reopening requires a fresh generation, including after an I/O error.
    pub const fn recover(generation: JournalGeneration, through: OpNumber) -> Self {
        Self {
            state: JournalSnapshot {
                generation,
                accepted: through,
                written: through,
                durable: through,
                faulted: false,
            },
        }
    }

    /// Inspect progress without obtaining mutation authority.
    pub const fn snapshot(&self) -> JournalSnapshot {
        self.state
    }

    /// Reserve a contiguous group after external payload/capacity validation.
    /// Errors leave all state unchanged. Zero and counter wrap are rejected.
    pub fn admit(&mut self, operations: u64) -> Result<WriteTicket, ProgressError> {
        self.require_active(self.state.generation)?;
        if operations == 0 {
            return Err(ProgressError::EmptyGroup);
        }
        let through = self
            .state
            .accepted
            .0
            .checked_add(operations)
            .ok_or(ProgressError::OpExhausted)?;
        let ticket = WriteTicket {
            generation: self.state.generation,
            first: OpNumber(self.state.accepted.0 + 1),
            through: OpNumber(through),
        };
        self.state.accepted = ticket.through;
        Ok(ticket)
    }

    /// Report a complete sealed write, in journal order. Duplicates are harmless.
    /// Partial writes must not call this method. They remain the adapter's work.
    pub fn complete_write(&mut self, ticket: WriteTicket) -> Result<(), ProgressError> {
        self.require_active(ticket.generation)?;
        if ticket.through > self.state.accepted {
            return Err(ProgressError::BeyondAccepted);
        }
        if ticket.through <= self.state.written {
            return Ok(());
        }
        if Some(ticket.first.0) != self.state.written.0.checked_add(1) {
            return Err(ProgressError::WriteGap);
        }
        self.state.written = ticket.through;
        Ok(())
    }

    /// Capture the complete written prefix before submitting a sync action.
    pub fn begin_sync(&self) -> Result<SyncTicket, ProgressError> {
        self.require_active(self.state.generation)?;
        Ok(SyncTicket {
            generation: self.state.generation,
            through: self.state.written,
        })
    }

    /// Report successful sync for the captured prefix, not newer written data.
    /// Duplicate/older completions cannot lower the durable prefix.
    pub fn complete_sync(&mut self, ticket: SyncTicket) -> Result<(), ProgressError> {
        self.require_active(ticket.generation)?;
        if ticket.through > self.state.written {
            return Err(ProgressError::BeyondWritten);
        }
        self.state.durable = self.state.durable.max(ticket.through);
        Ok(())
    }

    /// Report one complete write whose backend operation also synchronized the
    /// exact group before returning. This supports transactional local adapters
    /// that do not expose separate write and sync callbacks.
    pub fn complete_durable_write(&mut self, ticket: WriteTicket) -> Result<(), ProgressError> {
        self.complete_write(ticket)?;
        self.state.durable = self.state.durable.max(ticket.through);
        Ok(())
    }

    /// An uncertain write/sync faults this generation. Past evidence is retained.
    /// A stale error from a replaced generation cannot fault its replacement.
    pub fn fail(&mut self, generation: JournalGeneration) -> Result<(), ProgressError> {
        if generation != self.state.generation {
            return Err(ProgressError::StaleGeneration);
        }
        self.fault();
        Ok(())
    }

    /// Fault the current generation after its owner found conflicting history.
    pub fn fault(&mut self) {
        self.state.faulted = true;
    }

    fn require_active(&self, generation: JournalGeneration) -> Result<(), ProgressError> {
        if generation != self.state.generation {
            return Err(ProgressError::StaleGeneration);
        }
        if self.state.faulted {
            return Err(ProgressError::Faulted);
        }
        Ok(())
    }
}

/// Invalid or stale journal transition. No error advances any prefix.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum ProgressError {
    /// The writer has an uncertain I/O outcome and needs recovery.
    #[error("journal generation is faulted")]
    Faulted,
    /// Completion belongs to another writer incarnation.
    #[error("stale journal generation")]
    StaleGeneration,
    /// A write group must contain at least one operation.
    #[error("empty journal group")]
    EmptyGroup,
    /// Operation numbering cannot wrap.
    #[error("journal operation number exhausted")]
    OpExhausted,
    /// Completion claims operations that were never admitted.
    #[error("write completion exceeds admitted prefix")]
    BeyondAccepted,
    /// An earlier complete write has not been reported.
    #[error("write completion leaves a journal gap")]
    WriteGap,
    /// Sync claims data not known to have been completely written.
    #[error("sync completion exceeds written prefix")]
    BeyondWritten,
}

#[cfg(kani)]
mod proofs;