polyc-state-connect 2026.9.0

//! 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, FeedCursor, FeedReadStart, FeedRetention, FeedSnapshot, GetFeedRetention,
        GetProjectorStatus, ListProjectors, ProjectorListing, ProjectorStatus, RegisterProjector,
        SubscribeCommits,
    },
    receipt::Receipt,
    stream::{StreamContract, StreamEnd},
};

use crate::{
    MAX_FEED_WIRE_MESSAGE_BYTES,
    error::{TransportFallback, from_connect_error},
    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::FeedCursor::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 exact source's registered projectors and what each applied.
    ///
    /// The page truncates, and says so: the reply carries its own
    /// completeness, and this client refuses one that does not. A caller
    /// looking for one consumer asks
    /// [`Self::projector_status`] instead, which answers by exact
    /// `(source, consumer)` key and cannot be cut by a limit.
    ///
    /// # 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 asked = request;
        let request = pb::ListProjectorsRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            source: buffa::MessageField::some(pb::JournalSource::from(Kernel(request.source()))),
            limit: request.limit(),
            __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<_>, _>>()?;
        let listing = ProjectorListing::new(projectors, completeness);
        if !listing.answers(asked) {
            // Named fields only. A row from another source carries that
            // source's identity and cursor, and neither belongs in a refusal
            // handed back to a caller that asked about its own.
            return Err(StateError::Malformed {
                field: "projectors".to_owned(),
                reason: "the listing does not answer the exact source and bound this read named"
                    .to_owned(),
            });
        }
        Ok(listing)
    }

    /// Returns one exact projector status, if that consumer is registered.
    ///
    /// # Errors
    ///
    /// - Returns [`StateError::Malformed`] when the request or reply is invalid.
    /// - Returns [`StateError::PartitionHeld`] when destruction holds the
    ///   partition.
    /// - Returns [`StateError::Denied`] when the caller lacks feed access.
    /// - Returns [`StateError::DeadlineExpired`] when the call spends its budget.
    /// - Returns [`StateError::Cancelled`] when the caller withdraws interest.
    /// - Returns [`StateError::Unavailable`] when the state plane cannot answer.
    ///
    /// # Cancellation safety
    ///
    /// Read-only: dropping it loses the answer and nothing else.
    pub async fn projector_status(
        &self,
        declared: &DeclaredCall,
        request: &GetProjectorStatus,
    ) -> Result<Option<ProjectorStatus>, StateError> {
        let asked = request;
        let request = pb::GetProjectorStatusRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            source: buffa::MessageField::some(pb::JournalSource::from(Kernel(request.source()))),
            consumer: request.consumer().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_projector_status_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|e| from_connect_error(&e, &Self::fallback(attempted)))?
            .into_owned();
        let Some(status) = reply.status.into_option() else {
            return Ok(None);
        };
        let status = Kernel::<ProjectorStatus>::try_from(status)?.into_inner();
        if !status.answers(asked) {
            // Named fields only. The mismatched source, consumer, and cursor
            // are another projector's progress, and this refusal travels to a
            // caller that asked about its own.
            return Err(StateError::Malformed {
                field: "status".to_owned(),
                reason: "the status does not answer the exact projector this read named".to_owned(),
            });
        }
        Ok(Some(status))
    }

    /// 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()),
            source: buffa::MessageField::some(pb::JournalSource::from(Kernel(request.source()))),
            __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>::try_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: FeedCursor,
    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) -> &FeedCursor {
        &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 !feed::chunk_is_honest(&chunk, Some(&self.cursor)) {
                        return Err(self.refuse(StateError::Malformed {
                            field: "chunk".to_owned(),
                            reason: "a feed chunk preserves source, continuity, bounds, and its exact cursor"
                                .to_owned(),
                        }));
                    }
                    self.cursor = feed::cursor_after(&chunk, Some(&self.cursor));
                    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.source().clone(),
            FeedReadStart::Resume(Box::new(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 request = pb::SubscribeCommitsRequest::from(Kernel((&self.declared, &resumed)));
        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::{
        digest::ContentDigest,
        error::RetryClass,
        feed::{
            ATTESTATION_SIGNATURE_BYTES, ATTESTATION_SIGNER_BYTES, CommitEnvelope, FeedRecord,
            SourceCheckpoint,
        },
        id::{CommandId, PartitionId},
        journal::JournalAttestation,
        revision::{CommitRoot, JournalPosition, JournalSource, PartitionIncarnation},
    };

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

    /// One canned response body: the frames a case wrote for that dial.
    pub(super) 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)]
    pub(super) 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`].
        pub(super) 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()
    }

    fn source() -> JournalSource {
        JournalSource::new(
            PartitionId::new("conv-feed-client"),
            PartitionIncarnation::from_bytes([7; PartitionIncarnation::LEN]),
        )
    }

    fn checkpoint(position: u64) -> SourceCheckpoint {
        SourceCheckpoint::try_new(
            source(),
            JournalPosition::new(position),
            JournalPosition::new(position),
            position,
            JournalAttestation::new(
                CommitRoot::from_bytes([position as u8; CommitRoot::LEN]),
                position + 1,
                vec![3; ATTESTATION_SIGNATURE_BYTES],
                vec![4; ATTESTATION_SIGNER_BYTES],
            ),
        )
        .expect("a complete test checkpoint")
    }

    fn cursor(position: u64) -> FeedCursor {
        if position == 0 {
            FeedCursor::origin(source())
        } else {
            FeedCursor::at(checkpoint(position))
        }
    }

    /// A well-formed chunk carrying every commit after `from` through `to`.
    fn chunk(from: u64, to: u64, end: StreamEnd) -> pb::FeedChunk {
        let records = ((from + 1)..=to)
            .map(|position| {
                FeedRecord::new(
                    JournalPosition::new(position),
                    CommitEnvelope::new(
                        checkpoint(position),
                        CommandId::new(format!("commit-{position}")),
                        ContentDigest::from_bytes([position as u8; ContentDigest::LEN]),
                        JournalPosition::new(position - 1),
                        JournalPosition::new(position),
                        0,
                    ),
                    Vec::new(),
                )
            })
            .collect();
        let chunk = FeedChunk::new(records, cursor(to), end);
        pb::FeedChunk::from(Kernel(&chunk))
    }

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

    fn subscription_request() -> SubscribeCommits {
        SubscribeCommits::new(source(), FeedReadStart::Resume(Box::new(cursor(0))), 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(5, 7, StreamEnd::More);
        malformed.end = pb::StreamEnd::STREAM_END_UNSPECIFIED.into();

        let transport = CannedTransport::serving(vec![vec![
            framed(&chunk(0, 5, StreamEnd::More)),
            framed(&malformed),
            framed(&chunk(7, 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(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(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(
                    JournalSource::new(
                        PartitionId::new("conv-feed-client"),
                        PartitionIncarnation::from_bytes([7; PartitionIncarnation::LEN]),
                    ),
                    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");
    }
}

#[cfg(test)]
mod projector_binding_tests {
    #![allow(
        clippy::pedantic,
        clippy::nursery,
        missing_docs,
        reason = "a test module: panics are the failure mode"
    )]

    use super::tests::*;
    use super::*;
    use bytes::Bytes;
    use polyc_state::feed::{
        ConsumerPolicy, FeedCursor, GetProjectorStatus, ListProjectors, ProjectorRegistration,
        ProjectorStatus,
    };
    use polyc_state::id::{ConsumerId, PartitionId};
    use polyc_state::revision::{JournalSource, PartitionIncarnation};

    fn source_of(byte: u8) -> JournalSource {
        JournalSource::new(
            PartitionId::new("conv-bind"),
            PartitionIncarnation::from_bytes([byte; PartitionIncarnation::LEN]),
        )
    }

    fn status_of(
        registration: JournalSource,
        cursor: JournalSource,
        consumer: &str,
    ) -> pb::ProjectorStatus {
        let status = ProjectorStatus::new(
            ProjectorRegistration::new(
                ConsumerId::new(consumer),
                registration,
                ConsumerPolicy::Optional,
                0,
            ),
            FeedCursor::origin(cursor),
        );
        pb::ProjectorStatus::from(Kernel(&status))
    }

    fn client_over(reply: &impl buffa::Message) -> FeedClient<CannedTransport> {
        FeedClient::new(
            CannedTransport::serving_unary(Bytes::from(buffa::Message::encode_to_vec(reply))),
            ClientConfig::new("http://feed.invalid".parse().expect("a base url")),
        )
    }

    fn status_reply(status: pb::ProjectorStatus) -> pb::GetProjectorStatusReply {
        pb::GetProjectorStatusReply {
            status: buffa::MessageField::some(status),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    fn listing_reply(
        projectors: Vec<pb::ProjectorStatus>,
        completeness: polyc_state::page::PageCompleteness,
    ) -> pb::ListProjectorsReply {
        pb::ListProjectorsReply {
            projectors,
            completeness: pb::PageCompleteness::from(Kernel(completeness)).into(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    /// An exact lookup refuses a status that belongs to another source.
    ///
    /// A projector resumes from the cursor this read returns, so a status from
    /// another incarnation would let a restart continue a lineage that is no
    /// longer current.
    #[tokio::test]
    async fn an_exact_status_from_another_source_is_refused() {
        let client = client_over(&status_reply(status_of(
            source_of(9),
            source_of(9),
            "search",
        )));
        let refused = client
            .projector_status(
                &declared(),
                &GetProjectorStatus::new(source_of(7), ConsumerId::new("search")),
            )
            .await
            .expect_err("a status from another source is refused");
        assert!(
            matches!(&refused, StateError::Malformed { field, .. } if field == "status"),
            "got: {refused:?}"
        );
        assert!(
            !format!("{refused}").contains("09090909"),
            "the refusal must not carry the mismatched source back"
        );
    }

    /// An exact lookup refuses a status for another consumer.
    #[tokio::test]
    async fn an_exact_status_for_another_consumer_is_refused() {
        let client = client_over(&status_reply(status_of(
            source_of(7),
            source_of(7),
            "someone-else",
        )));
        let refused = client
            .projector_status(
                &declared(),
                &GetProjectorStatus::new(source_of(7), ConsumerId::new("search")),
            )
            .await
            .expect_err("a status for another consumer is refused");
        assert!(
            matches!(&refused, StateError::Malformed { field, .. } if field == "status"),
            "got: {refused:?}"
        );
    }

    /// A status whose acknowledged cursor names another source is refused.
    ///
    /// The registration can match while the cursor does not, and the cursor is
    /// the half a restart actually resumes from. The wire decode refuses this
    /// shape first, under the `acknowledged` field, because a status whose two
    /// halves disagree is not a status at all. The request binding behind it
    /// stays as defense in depth for a status that is internally coherent and
    /// still answers another read.
    #[tokio::test]
    async fn an_acknowledged_cursor_from_another_source_is_refused() {
        let client = client_over(&status_reply(status_of(
            source_of(7),
            source_of(9),
            "search",
        )));
        let refused = client
            .projector_status(
                &declared(),
                &GetProjectorStatus::new(source_of(7), ConsumerId::new("search")),
            )
            .await
            .expect_err("a cursor from another source is refused");
        assert!(
            matches!(&refused, StateError::Malformed { field, .. } if field == "acknowledged"),
            "got: {refused:?}"
        );
    }

    /// A matching status is returned, so the refusals above are not vacuous.
    #[tokio::test]
    async fn a_matching_exact_status_is_returned() {
        let client = client_over(&status_reply(status_of(
            source_of(7),
            source_of(7),
            "search",
        )));
        let status = client
            .projector_status(
                &declared(),
                &GetProjectorStatus::new(source_of(7), ConsumerId::new("search")),
            )
            .await
            .expect("a matching status is returned")
            .expect("the consumer is registered");
        assert_eq!(status.registration().consumer().as_str(), "search");
    }

    /// A listing over its own limit is refused.
    #[tokio::test]
    async fn a_listing_over_the_requested_limit_is_refused() {
        let client = client_over(&listing_reply(
            vec![
                status_of(source_of(7), source_of(7), "aa"),
                status_of(source_of(7), source_of(7), "bb"),
                status_of(source_of(7), source_of(7), "cc"),
            ],
            polyc_state::page::PageCompleteness::Complete,
        ));
        let refused = client
            .projectors(&declared(), &ListProjectors::new(source_of(7), 2))
            .await
            .expect_err("more rows than asked for is refused");
        assert!(
            matches!(&refused, StateError::Malformed { field, .. } if field == "projectors"),
            "got: {refused:?}"
        );
    }

    /// A listing carrying a row from another source is refused.
    #[tokio::test]
    async fn a_listing_row_from_another_source_is_refused() {
        let client = client_over(&listing_reply(
            vec![
                status_of(source_of(7), source_of(7), "aa"),
                status_of(source_of(9), source_of(9), "bb"),
            ],
            polyc_state::page::PageCompleteness::Complete,
        ));
        let refused = client
            .projectors(&declared(), &ListProjectors::new(source_of(7), 8))
            .await
            .expect_err("a row from another source is refused");
        assert!(
            matches!(&refused, StateError::Malformed { field, .. } if field == "projectors"),
            "got: {refused:?}"
        );
    }

    /// Duplicate and unordered consumers are both refused.
    #[tokio::test]
    async fn duplicate_or_unordered_consumers_are_refused() {
        for rows in [
            vec![
                status_of(source_of(7), source_of(7), "aa"),
                status_of(source_of(7), source_of(7), "aa"),
            ],
            vec![
                status_of(source_of(7), source_of(7), "bb"),
                status_of(source_of(7), source_of(7), "aa"),
            ],
        ] {
            let client = client_over(&listing_reply(
                rows,
                polyc_state::page::PageCompleteness::Complete,
            ));
            let refused = client
                .projectors(&declared(), &ListProjectors::new(source_of(7), 8))
                .await
                .expect_err("a listing must be strictly ordered and unique");
            assert!(
                matches!(&refused, StateError::Malformed { field, .. } if field == "projectors"),
                "got: {refused:?}"
            );
        }
    }

    /// A listing that stopped short of its limit cannot claim truncation.
    #[tokio::test]
    async fn a_short_listing_claiming_truncation_is_refused() {
        let client = client_over(&listing_reply(
            vec![status_of(source_of(7), source_of(7), "aa")],
            polyc_state::page::PageCompleteness::Truncated,
        ));
        let refused = client
            .projectors(&declared(), &ListProjectors::new(source_of(7), 8))
            .await
            .expect_err("a short page cannot have been cut by its limit");
        assert!(
            matches!(&refused, StateError::Malformed { field, .. } if field == "projectors"),
            "got: {refused:?}"
        );
    }
}