polyc-state-connect 2026.8.3

//! The client half of the conformance surface.
//!
//! Callers speak the kernel's vocabulary in and out: a [`SyntheticCommand`]
//! goes in, a [`Receipt`] or a [`StateError`] comes back. The generated types
//! never leave this crate, and neither does the transport's status code — a
//! caller reads the same variant the module refused with, whether it arrived
//! as an error detail or as a bare code the transport itself produced.

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

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::conformance::family;

use crate::{
    MAX_WIRE_MESSAGE_BYTES,
    error::{TransportFallback, from_connect_error},
    trace::bounded_traced_options,
    wire::{DeclaredCall, Kernel, nanos_from_duration},
};

/// What one `Observe` call reports: the honesty check and the module's clock.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Observation {
    /// How many times the family's effect has actually been applied.
    pub applied_effects: u64,
    /// Where the module's clock stands.
    pub now: MonotonicInstant,
}

/// A typed client for the conformance surface.
///
/// # Errors
///
/// Every method returns a [`StateError`]. When the listener refused with a
/// typed outcome, that is the exact variant it refused with. When the
/// transport refused before any handler ran — an oversized message, a spent
/// deadline, a draining or unreachable listener — the outcome is derived from
/// the transport code and keeps the retry class the design assigns it.
///
/// # Cancellation safety
///
/// Dropping any of these futures mid-await abandons the call, it does not undo
/// it. The request may already have reached the module, and the module may
/// already have committed; what the drop destroys is only this side's chance to
/// hear the answer. That is the same shape a lost response has, and the design
/// gives it the same resolution: retry the identical command identity and
/// digest — never a fresh identity — and the durable receipt settles which it
/// was. [`ConformanceClient::receipt`] is how a caller asks.
///
/// The read methods carry no such risk, because there is nothing to undo: a
/// dropped read loses an answer and no more, and reissuing it is free. Each
/// method below names which it is.
pub struct ConformanceClient<T> {
    inner: pb::StateConformanceServiceClient<T>,
}

impl<T> ConformanceClient<T>
where
    T: ClientTransport,
    <T::ResponseBody as connectrpc::http_body::Body>::Error: std::fmt::Display,
{
    /// Builds a client over `transport`, dialing whatever `config` names.
    ///
    /// The response bound matches the listener's own message bound, so a reply
    /// this build could not have accepted fails here rather than after an
    /// unbounded read.
    pub fn new(transport: T, config: ClientConfig) -> Self {
        Self {
            inner: pb::StateConformanceServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_WIRE_MESSAGE_BYTES),
            ),
        }
    }

    /// Returns how a bare transport code should be read for a request of
    /// `attempted_bytes`.
    fn fallback(attempted_bytes: usize) -> TransportFallback {
        TransportFallback::new(
            OperationFamily::new(family::FAMILY),
            MAX_WIRE_MESSAGE_BYTES as u64,
            attempted_bytes as u64,
        )
    }

    /// Submits one synthetic command.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the command earned; see the type-level note.
    ///
    /// # Cancellation safety
    ///
    /// Mutating: a dropped call may already have committed. Retry the same
    /// command identity, or retrieve its receipt.
    pub async fn submit(
        &self,
        declared: &DeclaredCall,
        command: &SyntheticCommand,
    ) -> Result<Receipt, StateError> {
        self.submit_with_payload(declared, command, Vec::new())
            .await
    }

    /// Submits one synthetic command carrying `payload` opaque wire bytes.
    ///
    /// The module ignores the bytes. They exist so a caller can present a
    /// genuinely oversized *message* and prove the listener's own wire bound
    /// refuses it before any handler runs — the bound the synthetic family's
    /// declared payload size cannot exercise, since it declares a size rather
    /// than carrying one.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::BoundsExceeded`] naming the payload-bytes bound
    /// when the encoded request exceeds the wire bound, and otherwise the
    /// typed outcome the command earned.
    ///
    /// # Cancellation safety
    ///
    /// Mutating, exactly as [`ConformanceClient::submit`]: a dropped call may
    /// already have committed.
    pub async fn submit_with_payload(
        &self,
        declared: &DeclaredCall,
        command: &SyntheticCommand,
        payload: Vec<u8>,
    ) -> Result<Receipt, StateError> {
        let request = pb::SubmitRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command: buffa::MessageField::some(Kernel(command).into()),
            payload,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .submit_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(
            Kernel::<Receipt>::try_from(reply.receipt.into_option().ok_or_else(|| {
                StateError::Malformed {
                    field: "receipt".to_owned(),
                    reason: "a successful submission carries its receipt".to_owned(),
                }
            })?)?
            .into_inner(),
        )
    }

    /// Retrieves the durable receipt recorded under `command_id`.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with. An identity that
    /// was never submitted is [`None`], not an error.
    ///
    /// # Cancellation safety
    ///
    /// Read-only, and the resolution for every ambiguous call above: dropping
    /// it changes nothing, and reissuing it is free.
    pub async fn receipt(
        &self,
        declared: &DeclaredCall,
        command_id: &CommandId,
    ) -> Result<Option<Receipt>, StateError> {
        let request = pb::GetReceiptRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command_id: command_id.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_receipt_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        reply
            .receipt
            .into_option()
            .map(|receipt| Kernel::<Receipt>::try_from(receipt).map(Kernel::into_inner))
            .transpose()
    }

    /// Takes an immutable snapshot a bounded read or stream can start from.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with.
    ///
    /// # Cancellation safety
    ///
    /// Mutating: a dropped call may leave a snapshot this side never learned
    /// the identity of. It holds records visible and nothing more, so the cost
    /// is retained storage rather than a wrong answer.
    pub async fn create_snapshot(&self, declared: &DeclaredCall) -> Result<SnapshotId, StateError> {
        let request = pb::CreateSnapshotRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .create_snapshot_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(SnapshotId::new(reply.snapshot))
    }

    /// Reads one bounded page.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with, including
    /// [`StateError::BoundsExceeded`] past the family's page bound.
    ///
    /// # Cancellation safety
    ///
    /// Read-only: dropping it loses the page and nothing else.
    pub async fn read_page(
        &self,
        declared: &DeclaredCall,
        page: &PageRequest,
    ) -> Result<Page<SyntheticRecord>, StateError> {
        let request = pb::ReadPageRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            page: buffa::MessageField::some(Kernel(page).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .read_page_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(
            Kernel::<Page<SyntheticRecord>>::try_from(reply.page.into_option().ok_or_else(
                || StateError::Malformed {
                    field: "page".to_owned(),
                    reason: "a successful read carries its page".to_owned(),
                },
            )?)?
            .into_inner(),
        )
    }

    /// Reads one bounded chunk of the family's durable stream.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with, including
    /// [`StateError::BoundsExceeded`] past the family's chunk bound.
    /// # Cancellation safety
    ///
    /// Read-only, and the one method where a declared withdrawal is not a
    /// failure at all: a caller whose [`DeclaredCall`] says it has withdrawn
    /// gets a drained chunk carrying the cursor that resumes it, rather than a
    /// [`StateError::Cancelled`]. Dropping the future instead loses the chunk,
    /// and the cursor the caller already holds still resumes.
    pub async fn read_chunk(
        &self,
        declared: &DeclaredCall,
        chunk: &StreamRequest,
    ) -> Result<StreamChunk<SyntheticRecord>, StateError> {
        let request = pb::ReadChunkRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            chunk: buffa::MessageField::some(Kernel(chunk).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .read_chunk_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(Kernel::<StreamChunk<SyntheticRecord>>::try_from(
            reply
                .chunk
                .into_option()
                .ok_or_else(|| StateError::Malformed {
                    field: "chunk".to_owned(),
                    reason: "a successful read carries its chunk".to_owned(),
                })?,
        )?
        .into_inner())
    }

    /// Returns the contract the family's stream declares.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with.
    ///
    /// # Cancellation safety
    ///
    /// Read-only: dropping it loses the declaration and nothing else.
    pub async fn stream_contract(
        &self,
        declared: &DeclaredCall,
    ) -> Result<StreamContract, StateError> {
        let request = pb::DescribeStreamRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .describe_stream_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(
            Kernel::<StreamContract>::try_from(reply.contract.into_option().ok_or_else(|| {
                StateError::Malformed {
                    field: "contract".to_owned(),
                    reason: "a stream declares its contract".to_owned(),
                }
            })?)?
            .into_inner(),
        )
    }

    /// Returns the applied-effect count and where the module's clock stands.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with.
    ///
    /// # Cancellation safety
    ///
    /// Read-only: dropping it loses the observation and nothing else.
    pub async fn observe(&self, declared: &DeclaredCall) -> Result<Observation, StateError> {
        let request = pb::ObserveRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .observe_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(Observation {
            applied_effects: reply.applied_effects,
            now: MonotonicInstant::from_nanos(reply.now_nanos),
        })
    }

    /// Advances the module's clock by `budget` and returns where it now sits.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with.
    ///
    /// # Cancellation safety
    ///
    /// Mutating: a dropped call may already have moved the module's clock, and
    /// the clock never moves back. [`ConformanceClient::observe`] reports where
    /// it actually stands.
    pub async fn advance_clock(
        &self,
        declared: &DeclaredCall,
        budget: std::time::Duration,
    ) -> Result<MonotonicInstant, StateError> {
        let request = pb::AdvanceClockRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            budget_nanos: nanos_from_duration(budget),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .advance_clock_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(MonotonicInstant::from_nanos(reply.now_nanos))
    }
}