polyc-state-connect 2026.8.3

//! What the conformance surface serves.
//!
//! The kernel's [`ConformanceAdapter`] takes `&mut self`, because the suite
//! drives one adapter from one place. A served surface cannot: a handler holds
//! a shared reference and calls may overlap. So the seam a listener composes
//! is this shared-reference mirror of the same operations, and
//! [`MemoryBackend`] is the one implementation B0 ships — the kernel's
//! deterministic in-memory reference behind a mutex.
//!
//! Nothing here decides anything. Every method hands its arguments to the
//! module and returns what the module said.

use std::{
    sync::{Mutex, PoisonError},
    time::Duration,
};

use polyc_state::{
    conformance::{ConformanceAdapter, SyntheticCommand, SyntheticRecord},
    context::CallContext,
    deadline::MonotonicInstant,
    error::StateError,
    id::{CommandId, SnapshotId},
    memory::MemoryState,
    page::{Page, PageRequest},
    receipt::Receipt,
    stream::{StreamChunk, StreamContract, StreamRequest},
};

/// The conformance kit's synthetic family, served behind a shared reference.
///
/// # Cancellation safety
///
/// The same guarantees the kernel's adapter documents carry over unchanged: a
/// cancelled [`ConformanceBackend::submit`] that already committed keeps its
/// receipt, and a later call under the same command identity retrieves it;
/// [`ConformanceBackend::read_chunk`] never fails on cancellation but drains
/// and returns a resumable cursor.
pub trait ConformanceBackend: Send + Sync + 'static {
    /// Submits one synthetic command.
    ///
    /// # Errors
    ///
    /// Returns whatever typed outcome the module gave the command — a
    /// [`StateError::DigestConflict`], [`StateError::RevisionConflict`],
    /// [`StateError::StaleFence`], [`StateError::BoundsExceeded`],
    /// [`StateError::DeadlineExpired`], [`StateError::Cancelled`], or
    /// [`StateError::AmbiguousOutcome`].
    fn submit(
        &self,
        command: SyntheticCommand,
        context: &CallContext,
    ) -> Result<Receipt, StateError>;

    /// Retrieves the durable receipt recorded under `command_id`.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] for an identity the module could
    /// never have recorded and [`StateError::Denied`] when the caller may not
    /// read it. An identity that was never submitted is [`None`], not an
    /// error.
    fn receipt(&self, command_id: &CommandId) -> Result<Option<Receipt>, StateError>;

    /// Takes an immutable snapshot a bounded read or stream can start from.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::DeadlineExpired`] or [`StateError::Cancelled`]
    /// before any snapshot is taken.
    fn create_snapshot(&self, context: &CallContext) -> Result<SnapshotId, StateError>;

    /// Reads one bounded page.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::BoundsExceeded`] past the family's page bound and
    /// [`StateError::Malformed`] when the start cannot be resolved.
    fn read_page(
        &self,
        request: PageRequest,
        context: &CallContext,
    ) -> Result<Page<SyntheticRecord>, StateError>;

    /// Reads one bounded chunk of the family's durable stream.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::BoundsExceeded`] past the family's chunk bound
    /// and [`StateError::Malformed`] when the start cannot be resolved.
    fn read_chunk(
        &self,
        request: StreamRequest,
        context: &CallContext,
    ) -> Result<StreamChunk<SyntheticRecord>, StateError>;

    /// Returns the contract the family's stream declares.
    fn stream_contract(&self) -> StreamContract;

    /// Returns how many times the family's effect has actually been applied.
    fn applied_effects(&self) -> u64;

    /// Returns the module's current monotonic instant.
    fn now(&self) -> MonotonicInstant;

    /// Advances the module's clock by `budget`.
    fn advance_clock(&self, budget: Duration);
}

/// The kernel's deterministic in-memory reference, served behind a mutex.
///
/// The mutex is what turns `&mut self` into `&self`; it adds no behavior. A
/// poisoned lock is recovered rather than propagated, because a panic in one
/// call must not silently convert every later call's typed outcome into a
/// transport failure.
#[derive(Debug, Default)]
pub struct MemoryBackend {
    state: Mutex<MemoryState>,
}

impl MemoryBackend {
    /// Builds an empty backend whose clock sits at its origin.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            state: Mutex::new(MemoryState::new()),
        }
    }

    /// Runs `call` against the module.
    fn with<R>(&self, call: impl FnOnce(&mut MemoryState) -> R) -> R {
        let mut guard = self.state.lock().unwrap_or_else(PoisonError::into_inner);
        call(&mut guard)
    }
}

impl ConformanceBackend for MemoryBackend {
    fn submit(
        &self,
        command: SyntheticCommand,
        context: &CallContext,
    ) -> Result<Receipt, StateError> {
        self.with(|state| state.submit(command, context))
    }

    fn receipt(&self, command_id: &CommandId) -> Result<Option<Receipt>, StateError> {
        self.with(|state| state.receipt(command_id))
    }

    fn create_snapshot(&self, context: &CallContext) -> Result<SnapshotId, StateError> {
        self.with(|state| state.create_snapshot(context))
    }

    fn read_page(
        &self,
        request: PageRequest,
        context: &CallContext,
    ) -> Result<Page<SyntheticRecord>, StateError> {
        self.with(|state| state.read_page(request, context))
    }

    fn read_chunk(
        &self,
        request: StreamRequest,
        context: &CallContext,
    ) -> Result<StreamChunk<SyntheticRecord>, StateError> {
        self.with(|state| state.read_chunk(request, context))
    }

    fn stream_contract(&self) -> StreamContract {
        self.with(|state| state.stream_contract())
    }

    fn applied_effects(&self) -> u64 {
        self.with(|state| state.applied_effects())
    }

    fn now(&self) -> MonotonicInstant {
        self.with(|state| state.now())
    }

    fn advance_clock(&self, budget: Duration) {
        self.with(|state| state.advance_clock(budget));
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use super::*;
    use polyc_state::{cancel::CancellationToken, conformance::SyntheticOp, deadline::Deadline};

    fn context(backend: &MemoryBackend) -> CallContext {
        CallContext::new(
            Deadline::after(backend.now(), Duration::from_secs(30)),
            CancellationToken::new(),
        )
    }

    /// The mutex is plumbing, not behavior: what the module decides is what
    /// the backend returns.
    #[test]
    fn the_backend_returns_what_the_module_decided() {
        let backend = MemoryBackend::new();
        let ctx = context(&backend);
        let receipt = backend
            .submit(
                SyntheticCommand::new("cmd-1", SyntheticOp::Append { amount: 4 }),
                &ctx,
            )
            .unwrap();
        assert!(receipt.is_new());
        assert_eq!(backend.applied_effects(), 1);

        let replay = backend
            .submit(
                SyntheticCommand::new("cmd-1", SyntheticOp::Append { amount: 4 }),
                &ctx,
            )
            .unwrap();
        assert!(replay.is_deduplicated());
        assert_eq!(backend.applied_effects(), 1);

        let conflict = backend
            .submit(
                SyntheticCommand::new("cmd-1", SyntheticOp::Append { amount: 5 }),
                &ctx,
            )
            .unwrap_err();
        assert!(matches!(conflict, StateError::DigestConflict { .. }));
    }

    #[test]
    fn the_clock_moves_only_when_a_caller_moves_it() {
        let backend = MemoryBackend::new();
        assert_eq!(backend.now(), MonotonicInstant::ORIGIN);
        backend.advance_clock(Duration::from_secs(3));
        assert_eq!(backend.now(), MonotonicInstant::from_nanos(3_000_000_000));
    }

    #[test]
    fn a_poisoned_lock_still_answers() {
        let backend = std::sync::Arc::new(MemoryBackend::new());
        let poisoner = std::sync::Arc::clone(&backend);
        let _ = std::thread::spawn(move || {
            poisoner.with(|_| panic!("poison the lock"));
        })
        .join();
        assert_eq!(backend.applied_effects(), 0);
    }
}