polyc-state-connect 2026.9.0

//! 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, CreateJournalDirectorySnapshot, DestroyPartition,
        ExcisePartitionRecords, GetJournalHead, GetJournalProof, GetJournalRoot, GetJournalSource,
        GetPartitionLastModified, JournalAttestation, JournalDirectoryPage,
        JournalDirectorySnapshot, JournalProof, JournalRange, JournalSourceHead,
        ListJournalDirectorySnapshot, LogicalJournalHead, PartitionLastModified, QuarantinedRecord,
        ReadJournalRange, ReleaseJournalDirectorySnapshot, RepairOutcome, RepairPartition,
    },
    receipt::Receipt,
};

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. Reads risk no durable
/// mutation. Snapshot creation may retain one bounded ephemeral entry until
/// release or expiry.
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, and checks the reply against what was asked.
    ///
    /// Every reply passes [`JournalRange::answers`] before it reaches a
    /// caller, so a listener that named another partition, moved the anchor,
    /// invented an absence for an anchored request, returned records outside
    /// the window or out of order, or contradicted its own completeness is
    /// refused here rather than believed.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with, including
    /// [`StateError::BoundsExceeded`] past the family's read bound, and
    /// [`StateError::Malformed`] naming `range` for a reply that does not
    /// answer this request.
    ///
    /// # 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();
        let decoded = Kernel::<JournalRange>::try_from(
            reply
                .range
                .into_option()
                .ok_or_else(|| missing("range", "a successful read carries its range"))?,
        )?
        .into_inner();
        if !decoded.answers(range) {
            return Err(StateError::Malformed {
                field: "range".to_owned(),
                reason: "the reply does not answer the bounded read that was sent".to_owned(),
            });
        }
        Ok(decoded)
    }

    /// 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<LogicalJournalHead, 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::<LogicalJournalHead>::try_from(
            reply
                .head
                .into_option()
                .ok_or_else(|| missing("head", "a successful read carries its head"))?,
        )?
        .into_inner())
    }

    /// Returns one partition's exact current physical source and logical head.
    ///
    /// # Errors
    ///
    /// Returns the typed State outcome when the request is refused, times out,
    /// or the successful reply is malformed.
    pub async fn source_head(
        &self,
        declared: &DeclaredCall,
        request: &GetJournalSource,
    ) -> Result<Option<JournalSourceHead>, StateError> {
        let expected = request.partition().clone();
        let request = pb::GetJournalSourceRequest {
            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_source_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        let head = reply
            .head
            .into_option()
            .map(|head| Kernel::<JournalSourceHead>::try_from(head).map(Kernel::into_inner))
            .transpose()?;
        if head
            .as_ref()
            .is_some_and(|head| head.source().partition() != &expected)
        {
            return Err(missing(
                "source_head",
                "a successful source response binds the requested partition",
            ));
        }
        Ok(head)
    }

    /// Retains one immutable snapshot of logical partition names.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome State refused with, including a capacity
    /// refusal when it cannot admit another snapshot.
    ///
    /// # Cancellation safety
    ///
    /// A lost successful response can orphan one bounded snapshot until its
    /// expiry. It changes no durable fact.
    pub async fn create_directory_snapshot(
        &self,
        declared: &DeclaredCall,
        _request: &CreateJournalDirectorySnapshot,
    ) -> Result<JournalDirectorySnapshot, StateError> {
        let request = pb::CreateJournalDirectorySnapshotRequest {
            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_journal_directory_snapshot_with_options(
                request,
                bounded_traced_options(declared),
            )
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Kernel::<JournalDirectorySnapshot>::try_from(
            reply
                .snapshot
                .into_option()
                .ok_or_else(|| missing("snapshot", "a successful creation carries its snapshot"))?,
        )
        .map(Kernel::into_inner)
    }

    /// Reads one bounded page from an exact directory snapshot.
    ///
    /// # Errors
    ///
    /// Returns the typed lifecycle, bound, or malformed outcome from State.
    /// A malformed successful response also fails closed.
    ///
    /// # Cancellation safety
    ///
    /// Dropping the future loses the page. A completed server page may have
    /// refreshed the bounded snapshot's idle lifetime.
    pub async fn directory_page(
        &self,
        declared: &DeclaredCall,
        request: &ListJournalDirectorySnapshot,
    ) -> Result<JournalDirectoryPage, StateError> {
        let wire_request = pb::ListJournalDirectorySnapshotRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            snapshot: request.snapshot().as_str().to_owned(),
            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(&wire_request) as usize;
        let reply = self
            .inner
            .list_journal_directory_snapshot_with_options(
                wire_request,
                bounded_traced_options(declared),
            )
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let page = Kernel::<JournalDirectoryPage>::try_from(
            reply
                .page
                .into_option()
                .ok_or_else(|| missing("page", "a successful listing carries its page"))?,
        )?
        .into_inner();
        page.validate(request)?;
        Ok(page)
    }

    /// Releases one retained directory snapshot idempotently.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome State refused with.
    ///
    /// # Cancellation safety
    ///
    /// A lost response is safe to retry with the same snapshot identity.
    pub async fn release_directory_snapshot(
        &self,
        declared: &DeclaredCall,
        request: &ReleaseJournalDirectorySnapshot,
    ) -> Result<(), StateError> {
        let request = pb::ReleaseJournalDirectorySnapshotRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            snapshot: request.snapshot().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        self.inner
            .release_journal_directory_snapshot_with_options(
                request,
                bounded_traced_options(declared),
            )
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?;
        Ok(())
    }

    /// 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(),
    }
}

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

    use std::pin::Pin;
    use std::sync::Arc;
    use std::task::{Context, Poll};
    use std::time::Duration;

    use bytes::Bytes;
    use connectrpc::client::{BoxFuture, ClientBody};
    use connectrpc::http_body::{Body, Frame};
    use polyc_state::revision::{JournalPosition, JournalSource, PartitionIncarnation};

    use super::*;
    use crate::state_audience;

    struct CannedBody(std::vec::IntoIter<Bytes>);

    impl Body for CannedBody {
        type Data = Bytes;
        type Error = std::io::Error;

        fn poll_frame(
            mut self: Pin<&mut Self>,
            _context: &mut Context<'_>,
        ) -> Poll<Option<Result<Frame<Bytes>, Self::Error>>> {
            Poll::Ready(self.0.next().map(|bytes| Ok(Frame::data(bytes))))
        }
    }

    #[derive(Clone)]
    struct CannedTransport {
        frames: Arc<Vec<Bytes>>,
    }

    impl ClientTransport for CannedTransport {
        type ResponseBody = CannedBody;
        type Error = std::io::Error;

        fn send(
            &self,
            _request: http::Request<ClientBody>,
        ) -> BoxFuture<'static, Result<http::Response<Self::ResponseBody>, Self::Error>> {
            let frames = self.frames.as_ref().clone();
            Box::pin(async move {
                Ok(http::Response::builder()
                    .status(http::StatusCode::OK)
                    .header(http::header::CONTENT_TYPE, "application/proto")
                    .body(CannedBody(frames.into_iter()))
                    .unwrap())
            })
        }
    }

    #[tokio::test]
    async fn source_head_refuses_a_crossed_partition_reply() {
        let crossed = JournalSourceHead::new(
            JournalSource::new(
                PartitionId::new("conv-b"),
                PartitionIncarnation::from_bytes([7; PartitionIncarnation::LEN]),
            ),
            JournalPosition::new(9),
        );
        let reply = pb::GetJournalSourceReply {
            head: buffa::MessageField::some(Kernel(&crossed).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let frames = vec![Bytes::from(buffa::Message::encode_to_vec(&reply))];
        let client = JournalClient::new(
            CannedTransport {
                frames: Arc::new(frames),
            },
            ClientConfig::new("http://journal.invalid".parse().unwrap()),
        );
        let error = client
            .source_head(
                &DeclaredCall::live(state_audience(), Duration::MAX),
                &GetJournalSource::new(PartitionId::new("conv-a")),
            )
            .await
            .expect_err("the successful reply crossed the requested partition");

        assert!(
            matches!(error, StateError::Malformed { ref field, .. } if field == "source_head"),
            "got {error:?}"
        );
    }
}