polyc-state-connect 2026.8.3

//! The client half of the partition-journal surface.
//!
//! Callers speak the kernel's vocabulary in and out: a [`CommitJournalBatch`]
//! 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.
//!
//! Every request carries the active span's trace context, so a commit and the
//! decision that caused it land in one trace rather than two.

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    error::StateError,
    id::{CommandId, PartitionId},
    journal::{
        self, CommitJournalBatch, DestroyPartition, ExcisePartitionRecords, GetJournalHead,
        GetJournalProof, GetJournalRoot, GetPartitionLastModified, JournalAttestation,
        JournalProof, JournalRange, ListJournalPartitions, PartitionLastModified, PartitionPage,
        QuarantinedRecord, ReadJournalRange, RepairOutcome, RepairPartition,
    },
    receipt::Receipt,
    revision::JournalHead,
};

use crate::{
    MAX_JOURNAL_WIRE_MESSAGE_BYTES,
    error::{TransportFallback, from_connect_error},
    journal::verify::{PartitionVerification, VerifyPartitionReplay},
    trace::{bounded_traced_options, bounded_traced_options_from_headers},
    wire::{DeclaredCall, Kernel},
};

/// A typed client for State's partition journal.
///
/// # 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 a mutating future mid-await abandons the call, it does not undo it.
/// The batch may already have committed; what the drop destroys is only this
/// side's chance to hear the receipt. That is the same shape a lost
/// acknowledgement has and it has the same resolution: retry the identical
/// command identity and digest — never a fresh identity — and
/// [`JournalClient::receipt`] settles which it was. The read methods carry no
/// such risk: a dropped read loses an answer and no more.
pub struct JournalClient<T> {
    inner: pb::StateJournalServiceClient<T>,
}

impl<T> JournalClient<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 for THIS
    /// family — [`MAX_JOURNAL_WIRE_MESSAGE_BYTES`], not the tight bound the
    /// other families carry — so a reply this build could not have accepted
    /// fails here rather than after an unbounded read, and a batch the kernel
    /// accepts is never refused by the caller's own transport first.
    pub fn new(transport: T, config: ClientConfig) -> Self {
        Self {
            inner: pb::StateJournalServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_JOURNAL_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(
            journal::family(),
            MAX_JOURNAL_WIRE_MESSAGE_BYTES as u64,
            attempted_bytes as u64,
        )
    }

    /// Commits one atomic batch.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the batch earned; see the type-level note.
    /// Also [`StateError::BoundsExceeded`] naming the payload-bytes bound when
    /// the encoded request or its reply exceeds this protocol's wire bound —
    /// the transport refuses those before any handler runs, and the fallback
    /// reports them in the same vocabulary the module would have — and
    /// [`StateError::Malformed`] naming `receipt` when a listener answered
    /// successfully but carried none.
    ///
    /// # Cancellation safety
    ///
    /// Mutating: a dropped call may already have committed. Retry the same
    /// command identity, or retrieve its receipt.
    pub async fn commit(
        &self,
        declared: &DeclaredCall,
        batch: &CommitJournalBatch,
    ) -> Result<Receipt, StateError> {
        let (request, attempted) = Self::commit_request(declared, batch);
        let reply = self
            .inner
            .commit_journal_batch_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        receipt_of(reply.receipt)
    }

    /// Commits one batch with W3C headers captured by the caller before an
    /// isolated runtime boundary.
    ///
    /// Only the Control journal bridge needs this shape. It preserves the same
    /// method, State deadline, and body bounds as [`Self::commit`]; it merely
    /// avoids deriving the trace carrier from the bridge runtime's ambient span.
    ///
    /// The options stay spelled out at the dial rather than passed in, because
    /// the deadline guard reads this seam as source: it must see that this
    /// shape derives its transport budget from `declared` too. Only the request
    /// encoding is shared with [`Self::commit`]
    /// (`scripts/check_state_call_deadlines.py`).
    ///
    /// # Errors
    ///
    /// Returns the State transport or protocol error from the underlying commit.
    pub async fn commit_with_trace_headers(
        &self,
        declared: &DeclaredCall,
        batch: &CommitJournalBatch,
        headers: http::HeaderMap,
    ) -> Result<Receipt, StateError> {
        let (request, attempted) = Self::commit_request(declared, batch);
        let reply = self
            .inner
            .commit_journal_batch_with_options(
                request,
                bounded_traced_options_from_headers(declared, headers),
            )
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        receipt_of(reply.receipt)
    }

    /// Encodes one commit request and the attempted size its fallback reports.
    fn commit_request(
        declared: &DeclaredCall,
        batch: &CommitJournalBatch,
    ) -> (pb::CommitJournalBatchRequest, usize) {
        let request = pb::CommitJournalBatchRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            batch: buffa::MessageField::some(Kernel(batch).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        (request, attempted)
    }

    /// Retrieves the durable receipt `partition` recorded under `command_id`.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with. An identity that
    /// never committed to `partition` is [`None`], not an error.
    ///
    /// # Cancellation safety
    ///
    /// Read-only, and the resolution for every ambiguous commit above: dropping
    /// it changes nothing, and reissuing it is free.
    pub async fn receipt(
        &self,
        declared: &DeclaredCall,
        partition: &PartitionId,
        command_id: &CommandId,
    ) -> Result<Option<Receipt>, StateError> {
        let request = pb::GetJournalReceiptRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command_id: command_id.as_str().to_owned(),
            partition: partition.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_journal_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()
    }

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

    /// Returns one partition's current head.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with, and
    /// [`StateError::Malformed`] naming `head` when a listener answered
    /// successfully but carried no head.
    ///
    /// # Cancellation safety
    ///
    /// Read-only: dropping it loses the answer and nothing else.
    pub async fn head(
        &self,
        declared: &DeclaredCall,
        request: &GetJournalHead,
    ) -> Result<JournalHead, StateError> {
        let request = pb::GetJournalHeadRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            partition: String::from(Kernel(request)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_journal_head_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(Kernel::<JournalHead>::try_from(
            reply
                .head
                .into_option()
                .ok_or_else(|| missing("head", "a successful read carries its head"))?,
        )?
        .into_inner())
    }

    /// Lists one bounded page of partitions.
    ///
    /// # 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 list_partitions(
        &self,
        declared: &DeclaredCall,
        request: &ListJournalPartitions,
    ) -> Result<PartitionPage, StateError> {
        let request = pb::ListJournalPartitionsRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            start_after: request
                .start_after()
                .map(|partition| partition.as_str().to_owned()),
            limit: request.limit(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .list_journal_partitions_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(Kernel::<PartitionPage>::try_from(
            reply
                .page
                .into_option()
                .ok_or_else(|| missing("page", "a successful listing carries its page"))?,
        )?
        .into_inner())
    }

    /// Replays one partition and reports whether its tamper-evidence holds.
    ///
    /// A partition that does not verify is a
    /// [`PartitionVerification::Violation`], not an error: the listener
    /// answered, and the answer is bad news. An error here means the listener
    /// could not answer at all, which a caller must never report as a healthy
    /// log.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with, and
    /// [`StateError::Malformed`] naming `verdict` when a listener answered
    /// successfully but reached no verdict.
    ///
    /// # Cancellation safety
    ///
    /// Read-only: dropping it loses the verdict and nothing else.
    pub async fn verify_replay(
        &self,
        declared: &DeclaredCall,
        request: &VerifyPartitionReplay,
    ) -> Result<PartitionVerification, StateError> {
        let request = pb::VerifyPartitionReplayRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            partition: request.partition().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .verify_partition_replay_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(Kernel::<PartitionVerification>::try_from(reply)?.into_inner())
    }

    /// Returns when one partition's storage last changed.
    ///
    /// The operational read a retention policy's idle axis is decided on. A
    /// partition the module holds nothing for reports
    /// [`PartitionLastModified::at_ms`] of [`None`], which a caller reads as
    /// "nothing to say" rather than as a very old time.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with, including
    /// [`StateError::Malformed`] naming `partition` for a name the module could
    /// not address.
    ///
    /// # Cancellation safety
    ///
    /// Read-only: dropping it loses the answer and nothing else.
    pub async fn last_modified(
        &self,
        declared: &DeclaredCall,
        request: &GetPartitionLastModified,
    ) -> Result<PartitionLastModified, StateError> {
        let request = pb::GetPartitionLastModifiedRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            partition: request.partition().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_partition_last_modified_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(PartitionLastModified::new(reply.at_ms))
    }

    /// Returns one partition's current signed root.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with, and
    /// [`StateError::Malformed`] naming `attestation` when a listener answered
    /// successfully but carried no attestation.
    ///
    /// # Cancellation safety
    ///
    /// Read-only: dropping it loses the attestation and nothing else.
    pub async fn root(
        &self,
        declared: &DeclaredCall,
        request: &GetJournalRoot,
    ) -> Result<JournalAttestation, StateError> {
        let request = pb::GetJournalRootRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            partition: String::from(Kernel(request)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_journal_root_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(Kernel::<JournalAttestation>::try_from(
            reply.attestation.into_option().ok_or_else(|| {
                missing("attestation", "a successful read carries its attestation")
            })?,
        )?
        .into_inner())
    }

    /// Returns one record's inclusion proof.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with, and
    /// [`StateError::Malformed`] naming `proof` when a listener answered
    /// successfully but carried no proof.
    ///
    /// # Cancellation safety
    ///
    /// Read-only: dropping it loses the proof and nothing else.
    pub async fn proof(
        &self,
        declared: &DeclaredCall,
        request: &GetJournalProof,
    ) -> Result<JournalProof, StateError> {
        let request = pb::GetJournalProofRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            partition: request.partition().as_str().to_owned(),
            position: request.position().get(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_journal_proof_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(Kernel::<JournalProof>::try_from(
            reply
                .proof
                .into_option()
                .ok_or_else(|| missing("proof", "a successful read carries its proof"))?,
        )?
        .into_inner())
    }

    /// Destroys one partition.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the command earned, and
    /// [`StateError::Malformed`] naming `receipt` when a listener answered
    /// successfully but carried none.
    ///
    /// # Cancellation safety
    ///
    /// Mutating and irreversible: a dropped call may already have destroyed the
    /// partition. Retry the same command identity, which is a replay.
    pub async fn destroy(
        &self,
        declared: &DeclaredCall,
        command: &DestroyPartition,
    ) -> Result<Receipt, StateError> {
        let request = pb::DestroyJournalPartitionRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command: buffa::MessageField::some(Kernel(command).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .destroy_journal_partition_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        receipt_of(reply.receipt)
    }

    /// Physically destroys one exact incarnation and leaves its durable hold
    /// active for a cross-family erasure.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the command earned. A dropped or lost reply
    /// may already have staged the destruction; retry the identical command.
    pub async fn stage_destroy(
        &self,
        declared: &DeclaredCall,
        command: &DestroyPartition,
    ) -> Result<(), StateError> {
        let request = pb::StageJournalDestructionRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command: buffa::MessageField::some(Kernel(command).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        self.inner
            .stage_journal_destruction_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?;
        Ok(())
    }

    /// Records a staged destruction and releases its durable hold.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the command earned, and
    /// [`StateError::Malformed`] when the stage is absent or a successful
    /// reply carries no receipt.
    pub async fn complete_destroy(
        &self,
        declared: &DeclaredCall,
        command: &DestroyPartition,
    ) -> Result<Receipt, StateError> {
        let request = pb::CompleteJournalDestructionRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command: buffa::MessageField::some(Kernel(command).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .complete_journal_destruction_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        receipt_of(reply.receipt)
    }

    /// Rewrites the named records of one partition in place.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the command earned, and
    /// [`StateError::Malformed`] naming `receipt` when a listener answered
    /// successfully but carried none.
    ///
    /// # Cancellation safety
    ///
    /// Mutating and irreversible: a dropped call may already have rewritten the
    /// partition. Retry the same command identity, which is a replay.
    pub async fn excise(
        &self,
        declared: &DeclaredCall,
        command: &ExcisePartitionRecords,
    ) -> Result<Receipt, StateError> {
        let request = pb::ExciseJournalRecordsRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command: buffa::MessageField::some(Kernel(command).into()),
            decisions: command
                .decisions()
                .iter()
                .map(|decision| pb::RecordDecision::from(Kernel(decision)))
                .collect(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .excise_journal_records_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        receipt_of(reply.receipt)
    }

    /// Repairs one partition, quarantining every record it cannot read.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the command earned, and
    /// [`StateError::Malformed`] naming `receipt` when a listener answered
    /// successfully but carried none.
    ///
    /// # Cancellation safety
    ///
    /// Mutating: a dropped call may already have repaired the partition. Retry
    /// the same command identity, which is a replay.
    pub async fn repair(
        &self,
        declared: &DeclaredCall,
        command: &RepairPartition,
    ) -> Result<RepairOutcome, StateError> {
        let request = pb::RepairJournalPartitionRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command: buffa::MessageField::some(Kernel(command).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .repair_journal_partition_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        let quarantined = reply
            .quarantined
            .into_iter()
            .map(|record| Kernel::<QuarantinedRecord>::from(record).into_inner())
            .collect();
        Ok(RepairOutcome::new(receipt_of(reply.receipt)?, quarantined))
    }
}

/// Reads the receipt a successful mutation carries.
fn receipt_of(receipt: impl Into<Option<pb::Receipt>>) -> Result<Receipt, StateError> {
    Ok(Kernel::<Receipt>::try_from(
        receipt
            .into()
            .ok_or_else(|| missing("receipt", "a committed command carries its receipt"))?,
    )?
    .into_inner())
}

/// Builds the outcome for a reply field the listener was required to set.
fn missing(field: &str, reason: &str) -> StateError {
    StateError::Malformed {
        field: field.to_owned(),
        reason: reason.to_owned(),
    }
}