polyc-state-connect 2026.8.3

//! The client half of the durable commit feed.
//!
//! Callers speak the kernel's vocabulary in and out: a [`CreateSnapshot`] goes
//! in, a [`FeedSnapshot`] or a [`StateError`] comes back. The generated types
//! never leave this crate, and neither does the transport's status code.
//!
//! # The subscription is the point
//!
//! The unary methods below are ordinary. [`FeedSubscription`] is not: it is the
//! whole reason this feed is a stream rather than a poll, and it owns the one
//! piece of state that makes a stream survivable — the cursor.
//!
//! A transport connection is not a record of progress. It drops, the listener
//! drains, a replica moves, a quiet partition ends an idle stream, and none of
//! that may cost a projector a commit or make it apply one twice without
//! knowing. So the subscription persists a cursor after every chunk and dials
//! from *that*, never from wherever the previous connection happened to stop. A
//! resume is therefore indistinguishable from a fresh subscription that started
//! at the same place, which is exactly what makes it safe.
//!
//! That is also why the resume is unbounded rather than capped at a couple of
//! attempts. A partition can be silent for hours; a subscription that gave up
//! after two closed streams would turn every quiet hour into a dead projector.
//! Every resume is cursor-anchored and delivery is at-least-once, so retrying
//! forever costs at most a redelivered chunk — and the wait between attempts is
//! capped exponential with jitter, so a fleet coming back together does not
//! arrive as one wave.

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use std::time::Duration;

use polyc_state::{
    error::StateError,
    feed::{
        self, AcknowledgeProjectorCursor, CompactFeedPrefix, CreateSnapshot, FeedChunk,
        FeedCompaction, FeedRetention, FeedSnapshot, GetFeedRetention, ListProjectors,
        ProjectorListing, ProjectorStatus, RegisterProjector, SubscribeCommits,
    },
    page::{Cursor, ReadStart},
    receipt::Receipt,
    stream::{StreamContract, StreamEnd},
};

use crate::{
    MAX_FEED_WIRE_MESSAGE_BYTES,
    error::{TransportFallback, from_connect_error},
    feed::wire::subscription,
    trace::{bounded_traced_options, streaming_traced_options},
    wire::{DeclaredCall, Kernel},
};

/// The shortest a subscription waits before resuming a paused stream.
const RESUME_BACKOFF_BASE: Duration = Duration::from_millis(20);

/// The longest a subscription waits before resuming a paused stream.
///
/// A quiet partition is the common case for a projector, so the ceiling is the
/// latency a commit into a long-idle partition may wait before its consumer
/// sees it. Low enough to be unremarkable, high enough that a thousand idle
/// projectors are not a load source.
const RESUME_BACKOFF_CAP: Duration = Duration::from_secs(5);

/// How many times the base delay may double before the cap governs.
const RESUME_BACKOFF_MAX_DOUBLINGS: u32 = 8;

/// Returns the jitter stream one subscription uses.
///
/// Seeded from what makes the subscription distinct rather than from a clock,
/// so two projectors on different partitions decorrelate and one projector's
/// waits stay reproducible.
fn jitter_seed(request: &SubscribeCommits) -> u64 {
    let mut seed: u64 = 0xcbf2_9ce4_8422_2325;
    let mut absorb = |bytes: &[u8]| {
        for byte in bytes {
            seed ^= u64::from(*byte);
            seed = seed.wrapping_mul(0x0100_0000_01b3);
        }
    };
    absorb(request.partition().as_str().as_bytes());
    if let Some(consumer) = request.consumer() {
        absorb(consumer.as_str().as_bytes());
    }
    // Never zero: an xorshift seeded with zero produces only zero.
    seed | 1
}

/// A typed client for State's durable commit feed.
///
/// # Errors
///
/// Every method returns a [`StateError`]. When the listener refused with a
/// typed outcome, that is the exact variant it refused with — including
/// [`StateError::CompactedRange`], which is terminal and means the consumer must
/// rebootstrap rather than retry. When the transport refused before any handler
/// ran, 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 snapshot, registration, acknowledgement, or compaction may already have
/// been recorded; what the drop destroys is only this side's chance to hear the
/// receipt. Retry the identical command identity — never a fresh one — and the
/// replay returns the original outcome. The read methods carry no such risk.
pub struct FeedClient<T> {
    inner: pb::StateFeedServiceClient<T>,
}

impl<T> FeedClient<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. It applies to every chunk of a subscription too, which is
    /// the practical ceiling on how large a feed chunk can be regardless of the
    /// commit bound the contract declares.
    pub fn new(transport: T, config: ClientConfig) -> Self {
        Self {
            inner: pb::StateFeedServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_FEED_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(
            feed::family(),
            MAX_FEED_WIRE_MESSAGE_BYTES as u64,
            attempted_bytes as u64,
        )
    }

    /// Takes one immutable snapshot of one partition's feed.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the command earned, and
    /// [`StateError::Malformed`] naming `snapshot` when a listener answered
    /// successfully but carried none.
    ///
    /// # Cancellation safety
    ///
    /// Mutating: a dropped call may already have recorded the binding. Retry the
    /// same command identity, which is a replay.
    pub async fn create_snapshot(
        &self,
        declared: &DeclaredCall,
        command: &CreateSnapshot,
    ) -> Result<FeedSnapshot, StateError> {
        let request = pb::CreateFeedSnapshotRequest {
            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
            .create_snapshot_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(Kernel::<FeedSnapshot>::try_from(
            reply
                .snapshot
                .into_option()
                .ok_or_else(|| missing("snapshot", "a recorded snapshot carries its binding"))?,
        )?
        .into_inner())
    }

    /// Registers one projector against one partition's feed.
    ///
    /// # 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 registered. Retry the same
    /// command identity, which is a replay.
    pub async fn register(
        &self,
        declared: &DeclaredCall,
        command: &RegisterProjector,
    ) -> Result<Receipt, StateError> {
        let request = pb::RegisterProjectorRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command: buffa::MessageField::some(Kernel(command).into()),
            registration: buffa::MessageField::some(Kernel(command.registration()).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .register_projector_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        receipt_of(reply.receipt)
    }

    /// Records how far one projector has durably applied.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the command earned, and
    /// [`StateError::Malformed`] naming `receipt` when a listener answered
    /// successfully but carried none.
    ///
    /// # Cancellation safety
    ///
    /// Mutating, but harmlessly so: an acknowledgement is monotone, so a
    /// dropped call is retried under the same identity or simply superseded by
    /// the next one. Nothing a reader sees depends on it either way.
    pub async fn acknowledge(
        &self,
        declared: &DeclaredCall,
        command: &AcknowledgeProjectorCursor,
    ) -> Result<Receipt, StateError> {
        let request = pb::AcknowledgeProjectorCursorRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command: buffa::MessageField::some(Kernel(command).into()),
            consumer: command.consumer().as_str().to_owned(),
            cursor: buffa::MessageField::some(pb::Cursor::from(Kernel(command.cursor()))),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .acknowledge_projector_cursor_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        receipt_of(reply.receipt)
    }

    /// Lists one partition's registered projectors and what each has applied.
    ///
    /// A request naming a consumer ([`ListProjectors::for_consumer`]) asks about
    /// that one and answers with its row or nothing. A request naming none
    /// returns a page, which truncates silently — see
    /// [`JournalFeed::projectors`](polyc_state::feed::JournalFeed::projectors)
    /// for why a caller after one consumer names it rather than searching.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with, including
    /// [`StateError::BoundsExceeded`] past the family's listing bound.
    ///
    /// # Cancellation safety
    ///
    /// Read-only: dropping it loses the listing and nothing else.
    pub async fn projectors(
        &self,
        declared: &DeclaredCall,
        request: &ListProjectors,
    ) -> Result<ProjectorListing, StateError> {
        let request = pb::ListProjectorsRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            partition: request.partition().as_str().to_owned(),
            limit: request.limit(),
            consumer: request
                .consumer()
                .map(|consumer| consumer.as_str().to_owned()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .list_projectors_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        let completeness = crate::wire::completeness("completeness", reply.completeness)?;
        let projectors = reply
            .projectors
            .into_iter()
            .map(|status| Kernel::<ProjectorStatus>::try_from(status).map(Kernel::into_inner))
            .collect::<Result<Vec<_>, _>>()?;
        Ok(ProjectorListing::new(projectors, completeness))
    }

    /// Returns what one partition's feed still retains.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with, and
    /// [`StateError::Malformed`] naming `retention` when a listener answered
    /// successfully but carried none.
    ///
    /// # Cancellation safety
    ///
    /// Read-only: dropping it loses the answer and nothing else.
    pub async fn retention(
        &self,
        declared: &DeclaredCall,
        request: &GetFeedRetention,
    ) -> Result<FeedRetention, StateError> {
        let request = pb::GetFeedRetentionRequest {
            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_feed_retention_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(Kernel::<FeedRetention>::from(
            reply
                .retention
                .into_option()
                .ok_or_else(|| missing("retention", "a successful read carries its retention"))?,
        )
        .into_inner())
    }

    /// Moves one partition's feed floor forward, as far as retention allows.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the command earned, and
    /// [`StateError::Malformed`] naming `compaction` when a listener answered
    /// successfully but carried none.
    ///
    /// # Cancellation safety
    ///
    /// Mutating and irreversible: a dropped call may already have moved the
    /// floor or evicted a projector. Retry the same command identity, which is
    /// a replay.
    pub async fn compact(
        &self,
        declared: &DeclaredCall,
        command: &CompactFeedPrefix,
    ) -> Result<FeedCompaction, StateError> {
        let request = pb::CompactFeedPrefixRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            command: buffa::MessageField::some(Kernel(command).into()),
            through: command.through().get(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .compact_feed_prefix_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        Ok(Kernel::<FeedCompaction>::try_from(
            reply
                .compaction
                .into_option()
                .ok_or_else(|| missing("compaction", "a compaction reports what it did"))?,
        )?
        .into_inner())
    }

    /// Returns the contract this feed's subscriptions declare.
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with, and
    /// [`StateError::Malformed`] naming `contract` when a listener answered
    /// successfully but carried none.
    ///
    /// # Cancellation safety
    ///
    /// Read-only: dropping it loses the answer and nothing else.
    pub async fn contract(&self, declared: &DeclaredCall) -> Result<StreamContract, StateError> {
        let request = pb::DescribeFeedStreamRequest {
            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_feed_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(|| missing("contract", "a stream declares its contract"))?,
        )?
        .into_inner())
    }

    /// Opens a resumable subscription to one partition's commit feed.
    ///
    /// Nothing is dialed here: the first [`FeedSubscription::next_chunk`] opens
    /// the stream, and every reconnect after that opens it again from the
    /// cursor the subscription is holding.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] when `request` starts from a snapshot
    /// identity this contract could not have issued — caught here rather than
    /// after a round trip, because a start this side cannot read is a start it
    /// could never resume from either.
    pub fn subscribe(
        &self,
        declared: DeclaredCall,
        request: SubscribeCommits,
    ) -> Result<FeedSubscription<'_, T>, StateError> {
        let cursor = feed::resume_cursor(request.start())?;
        let jitter = jitter_seed(&request);
        Ok(FeedSubscription {
            client: &self.inner,
            declared,
            request,
            cursor,
            stream: None,
            ended: false,
            last_end: None,
            stalls: 0,
            jitter,
        })
    }
}

/// One resumable subscription to a partition's commit feed.
///
/// Holds the cursor, dials from it, and re-dials from it. A caller drives it
/// with [`FeedSubscription::next_chunk`] and persists
/// [`FeedSubscription::cursor`] alongside whatever it applied, so a restart of
/// the *consumer* resumes exactly like a reconnect of the *connection*.
///
/// # Cancellation safety
///
/// Dropping the subscription abandons the connection and loses nothing else: a
/// feed read commits nothing, and the cursor the caller persisted is the whole
/// record of progress. Dropping a `next_chunk` future mid-await may lose one
/// chunk that was in flight; the cursor did not move for it, so the next
/// subscription redelivers it — which at-least-once delivery already permits.
pub struct FeedSubscription<'a, T: ClientTransport> {
    client: &'a pb::StateFeedServiceClient<T>,
    declared: DeclaredCall,
    request: SubscribeCommits,
    cursor: Cursor,
    stream: Option<
        connectrpc::client::ServerStream<
            T::ResponseBody,
            pb::__buffa::view::FeedChunkView<'static>,
        >,
    >,
    ended: bool,
    /// Why the last stream this subscription held ended, where it said.
    ///
    /// The marker on the final chunk is what turns a closed connection into a
    /// decision: `More` means pause and resume, `Exhausted` means the read is
    /// over, and nothing at all means the transport dropped without the
    /// listener getting a word in.
    last_end: Option<StreamEnd>,
    /// How many times in a row the stream has closed without delivering a
    /// chunk. Reset by any progress, and the exponent the backoff uses.
    stalls: u32,
    /// This subscription's jitter stream.
    jitter: u64,
}

impl<T> FeedSubscription<'_, T>
where
    T: ClientTransport,
    // The stream reads its body in place, so the transport's body must not move
    // under it. Every transport this workspace dials satisfies it; the bound is
    // stated rather than assumed.
    T::ResponseBody: Unpin,
    <T::ResponseBody as connectrpc::http_body::Body>::Error: std::fmt::Display,
{
    /// Returns the cursor the consumer persists alongside what it applied.
    #[must_use]
    pub const fn cursor(&self) -> &Cursor {
        &self.cursor
    }

    /// Reports whether the subscription is finished.
    ///
    /// True after a drained chunk, after a stream that ended on an exhausted
    /// one, and after a terminal refusal. A finished subscription yields
    /// [`None`] forever rather than resuming into a loop. A stream that ended on
    /// a paused chunk is not finished — that is the case this distinction
    /// exists for.
    #[must_use]
    pub const fn is_ended(&self) -> bool {
        self.ended
    }

    /// Returns the next chunk, resuming from the persisted cursor whenever the
    /// stream under it closes.
    ///
    /// What a closed stream means is whatever its last chunk said it meant. A
    /// chunk marked [`StreamEnd::More`] — including the empty one an idle or
    /// shedding listener leaves behind — means *pause*, and the subscription
    /// waits out a capped exponential backoff and dials again from its cursor,
    /// for as long as the caller keeps asking. A chunk marked
    /// [`StreamEnd::Exhausted`] or a drained one means the read is over.
    /// A close with no chunk to speak for it is the transport dropping without
    /// the listener getting a word in, which is treated as a pause for the same
    /// reason: the feed may still be growing, and resuming from a cursor cannot
    /// lose anything.
    ///
    /// [`None`] therefore means the subscription is genuinely over, never that
    /// the partition went quiet. Anything else is a chunk the caller applies
    /// before persisting [`FeedSubscription::cursor`].
    ///
    /// # Errors
    ///
    /// Returns the typed outcome the listener refused with. A terminal one —
    /// [`StateError::CompactedRange`] above all — also ends the subscription,
    /// because retrying the same cursor against it is futile and the remedy is
    /// a fresh snapshot. A retry-safe one leaves the subscription open, and the
    /// next call resumes from the same cursor.
    pub async fn next_chunk(&mut self) -> Result<Option<FeedChunk>, StateError> {
        loop {
            if self.ended {
                return Ok(None);
            }
            if self.stream.is_none() {
                // A stall is a pause, not a failure, so it is waited out rather
                // than reported. Progress resets the wait.
                if self.stalls > 0 {
                    tokio::time::sleep(self.backoff()).await;
                }
                self.dial().await?;
            }
            let received = match self.stream.as_mut() {
                Some(stream) => stream.message::<pb::FeedChunk>().await,
                None => return Ok(None),
            };

            match received {
                Ok(Some(message)) => {
                    // A chunk this client cannot read is refused exactly as a
                    // listener's own refusal is. Leaving the stream open would
                    // hand the next call the message AFTER the malformed one,
                    // whose `cursor_after` moves the cursor past commits that
                    // were never delivered — a subscription that claims to have
                    // applied what it silently skipped.
                    let chunk = match Kernel::<FeedChunk>::try_from(message.to_owned_message()) {
                        Ok(chunk) => chunk.into_inner(),
                        Err(typed) => return Err(self.refuse(typed)),
                    };
                    if let Some(next) = feed::cursor_after(&chunk, Some(&self.cursor)) {
                        self.cursor = next;
                    }
                    self.last_end = Some(chunk.end());
                    self.stalls = 0;
                    // A drain is a graceful end, not an error, and the cursor
                    // it left behind still resumes — which is why the chunk is
                    // handed back rather than swallowed.
                    if chunk.is_drained() {
                        self.ended = true;
                    }
                    return Ok(Some(chunk));
                }
                // The transport stream closed. What that means is whatever the
                // last chunk said it meant.
                Ok(None) => {
                    self.stream = None;
                    match self.last_end.take() {
                        // The read finished. Nothing more was coming.
                        Some(StreamEnd::Exhausted | StreamEnd::Drained) => {
                            self.ended = true;
                            return Ok(None);
                        }
                        // A pause, or a close the listener never annotated —
                        // which is the same thing from here, because both leave
                        // the feed possibly still growing. Resume from the
                        // cursor.
                        Some(StreamEnd::More) | None => {
                            self.stalls = self.stalls.saturating_add(1);
                        }
                    }
                }
                Err(error) => {
                    let typed = from_connect_error(
                        &error,
                        &TransportFallback::new(
                            feed::family(),
                            MAX_FEED_WIRE_MESSAGE_BYTES as u64,
                            0,
                        ),
                    );
                    return Err(self.refuse(typed));
                }
            }
        }
    }

    /// Records `typed` against the subscription and hands it back.
    ///
    /// The stream is dropped either way — whatever arrived on it was not
    /// applied, so resuming means dialing from the cursor rather than reading
    /// on past it. A terminal outcome ends the subscription as well, which is
    /// what makes [`FeedSubscription::next_chunk`]'s contract true: retrying
    /// the same cursor against it is futile, and the remedy is a fresh
    /// snapshot.
    fn refuse(&mut self, typed: StateError) -> StateError {
        self.stream = None;
        if !typed.is_retry_safe() {
            self.ended = true;
        }
        typed
    }

    /// Returns how long to wait before resuming after `stalls` closed streams.
    ///
    /// Capped exponential with jitter, per the design's retry rule. The
    /// subscription owns the retry because it owns the cursor: nothing above it
    /// knows where to resume from, so nothing above it could retry correctly.
    ///
    /// The jitter is derived from the subscription's own identity rather than a
    /// clock, so a fleet of projectors restarting together spreads out
    /// deterministically instead of arriving in one wave.
    fn backoff(&mut self) -> Duration {
        let exponent = (self.stalls - 1).min(RESUME_BACKOFF_MAX_DOUBLINGS);
        let ceiling = RESUME_BACKOFF_BASE
            .saturating_mul(1_u32 << exponent)
            .min(RESUME_BACKOFF_CAP);

        // xorshift64: a few instructions, no dependency, and decorrelated
        // enough for what jitter is for.
        self.jitter ^= self.jitter << 13;
        self.jitter ^= self.jitter >> 7;
        self.jitter ^= self.jitter << 17;

        let half = ceiling / 2;
        let spread = u64::from(half.subsec_nanos())
            .saturating_add(half.as_secs().saturating_mul(1_000_000_000));
        let offset = if spread == 0 { 0 } else { self.jitter % spread };
        half.saturating_add(Duration::from_nanos(offset))
    }

    /// Opens the stream from the cursor this subscription is holding.
    async fn dial(&mut self) -> Result<(), StateError> {
        let resumed = SubscribeCommits::new(
            self.request.partition().clone(),
            ReadStart::Resume(self.cursor.clone()),
            self.request.max_chunk_commits(),
        );
        let resumed = match self.request.consumer() {
            Some(consumer) => resumed.on_behalf_of(consumer.clone()),
            None => resumed,
        };
        let (partition, start, max_chunk_commits, consumer) = subscription(&resumed);

        let request = pb::SubscribeCommitsRequest {
            context: buffa::MessageField::some(Kernel(&self.declared).into()),
            partition,
            start: buffa::MessageField::some(start),
            max_chunk_commits,
            consumer,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let stream = self
            .client
            .subscribe_commits_with_options(request, streaming_traced_options())
            .await
            .map_err(|error| {
                let typed = from_connect_error(
                    &error,
                    &TransportFallback::new(
                        feed::family(),
                        MAX_FEED_WIRE_MESSAGE_BYTES as u64,
                        attempted as u64,
                    ),
                );
                if !typed.is_retry_safe() {
                    self.ended = true;
                }
                typed
            })?;
        self.stream = Some(stream);
        Ok(())
    }
}

/// Reads the receipt a successful command 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 tests {
    #![allow(
        clippy::pedantic,
        clippy::nursery,
        missing_docs,
        reason = "a test module: panics are the failure mode"
    )]

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

    use bytes::Bytes;
    use connectrpc::{
        client::{BoxFuture, ClientBody, ClientConfig},
        envelope::Envelope,
        http_body::{Body, Frame},
    };
    use polyc_state::{
        error::RetryClass, id::PartitionId, revision::JournalPosition, stream::StreamChunk,
    };

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

    /// One canned response body: the frames a case wrote for that dial.
    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>,
            _cx: &mut Context<'_>,
        ) -> Poll<Option<Result<Frame<Bytes>, Self::Error>>> {
            Poll::Ready(self.0.next().map(|bytes| Ok(Frame::data(bytes))))
        }
    }

    /// A transport that answers each dial with the next canned response, and
    /// counts how many dials it saw — which is how a case tells "resumed from
    /// the cursor" from "read on down the same open stream".
    #[derive(Clone)]
    struct CannedTransport {
        dials: Arc<Mutex<std::collections::VecDeque<Vec<Bytes>>>>,
        dialed: Arc<Mutex<usize>>,
        content_type: &'static str,
    }

    impl CannedTransport {
        fn serving(responses: Vec<Vec<Bytes>>) -> Self {
            Self {
                dials: Arc::new(Mutex::new(responses.into())),
                dialed: Arc::new(Mutex::new(0)),
                content_type: "application/connect+proto",
            }
        }

        /// Answers one unary call with `reply`.
        ///
        /// A unary Connect reply is a bare protobuf body under
        /// `application/proto`. It carries no envelope, and the client refuses
        /// the streaming content type on a unary route, so a unary case cannot
        /// reuse [`CannedTransport::serving`].
        fn serving_unary(reply: Bytes) -> Self {
            Self {
                dials: Arc::new(Mutex::new(vec![vec![reply]].into())),
                dialed: Arc::new(Mutex::new(0)),
                content_type: "application/proto",
            }
        }

        fn dials(&self) -> usize {
            *self.dialed.lock().expect("the dial count")
        }
    }

    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>> {
            *self.dialed.lock().expect("the dial count") += 1;
            let frames = self
                .dials
                .lock()
                .expect("the canned responses")
                .pop_front()
                .unwrap_or_default();
            let content_type = self.content_type;
            Box::pin(async move {
                Ok(http::Response::builder()
                    .status(http::StatusCode::OK)
                    .header(http::header::CONTENT_TYPE, content_type)
                    .body(CannedBody(frames.into_iter()))
                    .expect("a canned response"))
            })
        }
    }

    /// One data envelope carrying `chunk`.
    fn framed(chunk: &pb::FeedChunk) -> Bytes {
        Envelope::data(Bytes::from(buffa::Message::encode_to_vec(chunk))).encode()
    }

    /// The envelope that closes a Connect stream cleanly.
    fn end_of_stream() -> Bytes {
        Envelope::end_stream(Bytes::from_static(b"{}")).encode()
    }

    /// A well-formed, record-free chunk that leaves the cursor at `position`.
    fn chunk_at(position: u64, end: StreamEnd) -> pb::FeedChunk {
        let chunk: FeedChunk = StreamChunk::new(
            Vec::new(),
            Some(Cursor::at(JournalPosition::new(position))),
            end,
        );
        pb::FeedChunk::from(Kernel(&chunk))
    }

    fn declared() -> DeclaredCall {
        DeclaredCall::live(state_audience(), Duration::MAX)
    }

    fn subscription_request() -> SubscribeCommits {
        SubscribeCommits::new(
            PartitionId::new("conv-feed-client"),
            ReadStart::Resume(Cursor::at(JournalPosition::ORIGIN)),
            8,
        )
    }

    /// A chunk this client cannot read ends the subscription, and never lets
    /// the cursor step over what it did not deliver.
    ///
    /// The decode failure used to return before any of the bookkeeping the
    /// sibling refusal does: the subscription stayed open and unfinished, so
    /// the next call read the message AFTER the malformed one off the same
    /// stream and moved the cursor to it. The commits in between were never
    /// handed to the consumer, and the cursor said they had been — silent loss
    /// under a contract that promises at-least-once.
    #[tokio::test]
    async fn a_malformed_chunk_ends_the_subscription_without_moving_the_cursor() {
        let mut malformed = chunk_at(7, StreamEnd::More);
        malformed.end = pb::StreamEnd::STREAM_END_UNSPECIFIED.into();

        let transport = CannedTransport::serving(vec![vec![
            framed(&chunk_at(5, StreamEnd::More)),
            framed(&malformed),
            framed(&chunk_at(9, StreamEnd::Exhausted)),
            end_of_stream(),
        ]]);
        let client = FeedClient::new(
            transport.clone(),
            ClientConfig::new("http://feed.invalid".parse().expect("a base url")),
        );
        let mut subscription = client
            .subscribe(declared(), subscription_request())
            .expect("a resumable subscription");

        let first = subscription
            .next_chunk()
            .await
            .expect("the well-formed chunk is delivered")
            .expect("a chunk, not the end of the subscription");
        assert_eq!(first.end(), StreamEnd::More);
        assert_eq!(
            subscription.cursor(),
            &Cursor::at(JournalPosition::new(5)),
            "the delivered chunk moved the cursor"
        );

        let refused = subscription
            .next_chunk()
            .await
            .expect_err("a chunk this build cannot read is a refusal");
        assert!(
            matches!(refused, StateError::Malformed { ref field, .. } if field == "end"),
            "got {refused}"
        );
        assert_eq!(
            refused.retry_class(),
            RetryClass::Terminal,
            "a malformed chunk is terminal, and the contract says terminal ends the subscription"
        );
        assert!(
            subscription.is_ended(),
            "a terminal refusal ends the subscription"
        );
        assert_eq!(
            subscription.cursor(),
            &Cursor::at(JournalPosition::new(5)),
            "the cursor never steps over commits that were not delivered"
        );

        assert!(
            subscription
                .next_chunk()
                .await
                .expect("an ended subscription answers rather than failing")
                .is_none(),
            "an ended subscription yields nothing forever"
        );
        assert_eq!(
            transport.dials(),
            1,
            "the refusal is not a stall, so nothing re-dials"
        );
    }

    /// A listing that does not say whether the limit cut it is refused, not
    /// read as whole.
    ///
    /// This is the decode half of the completeness marker. The encode half is
    /// pinned by the loopback suite. Nothing pinned this call site, so
    /// softening the `?` here to `unwrap_or(PageCompleteness::Complete)` left
    /// the whole suite green: the silent-truncation defect the marker exists
    /// to prevent, moved one layer out into the client.
    ///
    /// An unset marker is what an older state plane sends. Proto3 gives an
    /// absent enum field the value zero.
    #[tokio::test]
    async fn a_listing_that_does_not_say_whether_it_was_cut_is_refused() {
        let reply = pb::ListProjectorsReply {
            projectors: Vec::new(),
            completeness: pb::PageCompleteness::PAGE_COMPLETENESS_UNSPECIFIED.into(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let transport =
            CannedTransport::serving_unary(Bytes::from(buffa::Message::encode_to_vec(&reply)));
        let client = FeedClient::new(
            transport.clone(),
            ClientConfig::new("http://feed.invalid".parse().expect("a base url")),
        );

        let refused = client
            .projectors(
                &declared(),
                &ListProjectors::new(PartitionId::new("conv-feed-client"), 4),
            )
            .await
            .expect_err("a listing that does not report its completeness is refused");

        assert!(
            matches!(&refused, StateError::Malformed { field, .. } if field == "completeness"),
            "the refusal names the field it could not read: {refused:?}"
        );
        assert_eq!(
            refused.retry_class(),
            RetryClass::Terminal,
            "a reply this client cannot read does not become readable on a retry"
        );
        assert_eq!(transport.dials(), 1, "the refusal is not a stall");
    }
}