polyc-state-connect 2026.8.3

//! The server half of the durable commit feed.
//!
//! The unary handlers are the same shape the journal's are: admit the call,
//! translate the wire into the kernel's vocabulary, hand it to the module,
//! translate what the module said back. No handler branches on a State question
//! (INV-24) and no handler mints a receipt of its own (INV-22).
//!
//! `SubscribeCommits` is the one that is not unary, and the shape of it is worth
//! stating plainly. The module contract is synchronous and returns one bounded
//! chunk per call; the RPC is a server stream. What bridges them is a pump: a
//! loop on the blocking pool that asks the module for the next chunk, writes it
//! into a bounded channel, and advances its own cursor. The stream the client
//! reads is that channel.
//!
//! Three properties fall out of that arrangement, and each is the declared
//! contract rather than an accident:
//!
//! - **The cursor is the pump's, not the connection's.** Every chunk carries the
//!   cursor that resumes it, so a client that reconnects continues from what it
//!   persisted rather than from where the pump happened to be.
//! - **A slow consumer is shed.** The channel is bounded and the pump never
//!   waits on it. A consumer that stops reading loses the connection and
//!   resumes from its cursor, which is `Backpressure::ShedSlowConsumer` and not
//!   a policy invented here.
//! - **A quiet partition pauses rather than ending.** An idle tail, and a shed
//!   consumer, both get one final empty chunk marked `StreamEnd::More` carrying
//!   the resume cursor — *more may follow, resume from here* — and then the
//!   stream ends. `Exhausted` is reserved for a read that is genuinely over, so
//!   a consumer can tell "nothing right now" from "nothing ever again".
//! - **A drain ends the stream gracefully.** A draining listener gets one final
//!   drained chunk carrying its resume cursor, then end of stream — never an
//!   error, because a drain is not a failure and the consumer has lost nothing.

use std::{
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use connectrpc::{
    ConnectError, RequestContext, Response, Router, ServiceRequest, ServiceResult, ServiceStream,
};
use polyc_proto::proto::polychrome::state::v1::{
    AcknowledgeProjectorCursorReply, AcknowledgeProjectorCursorRequest, CompactFeedPrefixReply,
    CompactFeedPrefixRequest, CreateFeedSnapshotReply, CreateFeedSnapshotRequest,
    DescribeFeedStreamReply, DescribeFeedStreamRequest, FeedChunk as PbFeedChunk,
    GetFeedRetentionReply, GetFeedRetentionRequest, ListProjectorsReply, ListProjectorsRequest,
    PageCompleteness, RegisterProjectorReply, RegisterProjectorRequest, StateFeedService,
    StateFeedServiceExt, SubscribeCommitsRequest,
};
use polyc_state::{
    command::CommandMetadata,
    context::CallContext,
    error::{BoundKind, StateError},
    feed::{
        self, AcknowledgeProjectorCursor, CompactFeedPrefix, CreateSnapshot, GetFeedRetention,
        JournalFeed, ListProjectors, ProjectorRegistration, RegisterProjector, SubscribeCommits,
    },
    id::{ConsumerId, OperationFamily},
    page::{Cursor, ReadStart},
    revision::JournalPosition,
    stream::{StreamChunk, StreamEnd},
};

use crate::{
    MAX_FEED_WIRE_MESSAGE_BYTES,
    admission::{
        AudienceBinding, PeerIdentity, check_audience_binding, check_call_context_version,
        check_not_draining, check_transport_deadline, state_audience,
    },
    error::to_connect_error,
    feed::wire::subscribe_request,
    trace::adopt_caller_trace,
    wire::{DeclaredCall, Kernel, declared_call},
};

/// How many chunks a subscription may buffer before its consumer is shed.
///
/// Small on purpose. The buffer exists to absorb a consumer that is briefly
/// busy, not to hide one that cannot keep up: a deep buffer would turn shedding
/// into stalling and hold a partition's commits in memory while it did.
const SUBSCRIBE_STREAM_BOUND: usize = 8;

/// How many times a shed pump tries to leave its resume marker behind.
///
/// A shed consumer's channel is full by definition, so the marker cannot be
/// written immediately. The pump waits briefly for the consumer to read one
/// chunk and make room; if it never does, the pump leaves anyway — the last
/// chunk it did deliver already carried a cursor, so nothing is lost either
/// way.
const SHED_MARKER_ATTEMPTS: u32 = 25;

/// How long the pump waits before asking a caught-up partition again.
///
/// A poll rather than a notification, because the feed is served from durable
/// state and depends on no process-local signal reaching it (INV-11). The
/// interval is the latency a caught-up projector sees.
const TAIL_POLL: Duration = Duration::from_millis(20);

/// How long a caught-up subscription is held open before the stream pauses.
///
/// A State call is a bounded call, and a subscription is not exempt from that
/// just because it is long-lived: a listener that held an idle stream forever
/// could never finish draining, and a pump that never returned would leak the
/// thread it runs on. So an idle tail *pauses* — one final empty chunk marked
/// [`StreamEnd::More`], carrying the cursor, then the transport stream ends —
/// and the client resumes from that cursor. It costs a consumer nothing,
/// because a resume from a cursor is indistinguishable from never having
/// disconnected.
///
/// Comfortably inside the listener's own call deadline, so the stream ends on
/// this side's terms rather than the transport's.
const MAX_TAIL_IDLE: Duration = Duration::from_secs(20);

/// How many subscriptions one listener pumps at once without queueing, summed
/// over every client it serves.
///
/// A THREAD budget and nothing else. Every open subscription holds one
/// blocking-pool thread for as long as it runs — [`FeedSvc::subscribe_commits`]
/// pumps it on [`tokio::task::spawn_blocking`], because every call it makes into
/// the module below is synchronous — so a listener whose runtime has fewer
/// blocking threads than this does not refuse the extra subscriptions: it queues
/// their pumps, and a queued pump is the worst failure shape available. The
/// stream opens, yields nothing, never ends, and reports no error for a client
/// to retry on.
///
/// It is therefore an obligation on the listener: size the runtime's blocking
/// pool to at least this, plus whatever else runs there (`polychrome-state`'s
/// `main` does).
///
/// It is NOT what bounds any one client's fan-out, and reading it that way is
/// how a caller lands on a number the transport cannot honor. A client reaches
/// this listener over HTTP/2, one connection carries
/// [`crate::MAX_CONCURRENT_STREAMS_PER_CONNECTION`] streams, and a subscription
/// is one stream — so for a single client on a single connection the binding
/// limit is that ceiling, well below this budget, and it is the one a client
/// derives its cap from (`polyc_control_plane::commit_feed`'s `MAX_FOLLOWERS`
/// does). What this budget buys is headroom on the side that cannot see how
/// many clients there are: the reconnect churn a paused idle tail produces
/// (`MAX_TAIL_IDLE`), where a client's replacement pump can start while the
/// pump it replaces is still finishing.
pub const MAX_CONCURRENT_SUBSCRIPTIONS: usize = 512;

/// What a listener tunes about the subscriptions it serves.
///
/// Two values, both of which have one right answer in production and need a
/// different one to be *observable* in a test: an idle window a test cannot
/// wait out, and a buffer depth a test cannot realistically overrun. A
/// composition that says nothing gets the production values.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FeedStreamTuning {
    max_tail_idle: Duration,
    stream_bound: usize,
}

impl Default for FeedStreamTuning {
    fn default() -> Self {
        Self::new()
    }
}

impl FeedStreamTuning {
    /// Returns the production tuning.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            max_tail_idle: MAX_TAIL_IDLE,
            stream_bound: SUBSCRIBE_STREAM_BOUND,
        }
    }

    /// Returns the same tuning pausing an idle tail after `idle`.
    #[must_use]
    pub const fn with_max_tail_idle(mut self, idle: Duration) -> Self {
        self.max_tail_idle = idle;
        self
    }

    /// Returns the same tuning buffering `chunks` before shedding a consumer.
    #[must_use]
    pub const fn with_stream_bound(mut self, chunks: usize) -> Self {
        self.stream_bound = chunks;
        self
    }

    /// Returns how long an idle tail is held open.
    #[must_use]
    pub const fn max_tail_idle(self) -> Duration {
        self.max_tail_idle
    }

    /// Returns how many chunks a subscription buffers.
    #[must_use]
    pub const fn stream_bound(self) -> usize {
        self.stream_bound
    }
}

/// State's durable commit feed, served over Connect.
///
/// Mounted with the partition journal and behind the same gate: the feed serves
/// what the journal committed, and a composition that may not serve one has no
/// business serving the other.
///
/// # Blocking
///
/// The module contract is synchronous, so every unary handler below blocks its
/// thread for the duration of the durable write or read rather than yielding
/// while it waits — the same deliberate deviation the journal's handlers carry.
///
/// `SubscribeCommits` does not: its pump runs on the blocking pool, so a
/// long-lived subscription occupies a blocking-pool thread rather than a runtime
/// worker. A listener serving many concurrent subscriptions needs a blocking
/// pool sized for them.
pub struct FeedSvc {
    feed: Arc<dyn JournalFeed>,
    draining: Arc<AtomicBool>,
    binding: Arc<AudienceBinding>,
    tuning: FeedStreamTuning,
}

impl FeedSvc {
    /// Serves `feed` for [`STATE_AUDIENCE`](crate::STATE_AUDIENCE), admitting
    /// only the workloads `binding` names, and refusing new calls once
    /// `draining` is set.
    ///
    /// The audience is not a parameter, for the reason
    /// [`JournalSvc::new`](crate::journal::JournalSvc::new) gives: the feed
    /// serves what the journal committed, on the same plane, under the same
    /// one name.
    #[must_use]
    pub const fn new(
        feed: Arc<dyn JournalFeed>,
        draining: Arc<AtomicBool>,
        binding: Arc<AudienceBinding>,
    ) -> Self {
        Self {
            feed,
            draining,
            binding,
            tuning: FeedStreamTuning::new(),
        }
    }

    /// Returns the same service serving subscriptions under `tuning`.
    #[must_use]
    pub const fn with_tuning(mut self, tuning: FeedStreamTuning) -> Self {
        self.tuning = tuning;
        self
    }

    /// Registers this service on `router`.
    #[must_use]
    pub fn register_on(self, router: Router) -> Router {
        Arc::new(self).register(router)
    }

    /// Returns the family every call to this surface belongs to.
    fn family() -> OperationFamily {
        feed::family()
    }

    /// Returns the identity the connection proved, or anonymity when it proved
    /// none.
    fn peer(ctx: &RequestContext) -> PeerIdentity {
        PeerIdentity::from_verified_leaf(
            ctx.peer_certs().and_then(<[_]>::first).map(|leaf| &**leaf),
        )
    }

    /// Admits one call, or refuses it with the typed outcome it earned.
    ///
    /// The same order the journal's admission uses, and for the same reasons:
    /// lifecycle first, then version, then authorization, then budget.
    fn admit(
        &self,
        ctx: &RequestContext,
        context: impl Into<Option<polyc_proto::proto::polychrome::state::v1::CallContext>>,
        method: &'static str,
    ) -> Result<(DeclaredCall, tracing::Span), ConnectError> {
        check_not_draining(self.draining.load(Ordering::Relaxed))?;
        let declared: DeclaredCall = declared_call(context).map_err(|e| to_connect_error(&e))?;
        let family = Self::family();
        check_call_context_version(declared.version).map_err(|e| to_connect_error(&e))?;
        check_audience_binding(
            &Self::peer(ctx),
            &declared.audience,
            &state_audience(),
            &self.binding,
            &family,
        )
        .map_err(|e| to_connect_error(&e))?;
        check_transport_deadline(ctx.time_remaining(), &family)
            .map_err(|e| to_connect_error(&e))?;
        Ok((declared, adopt_caller_trace(ctx.headers(), method)))
    }
}

/// One subscription's chunk, as it travels from the pump to the transport.
type PumpedChunk = Result<PbFeedChunk, ConnectError>;

/// Builds the bounded channel one subscription's chunks travel through.
fn new_chunk_channel(
    bound: usize,
) -> (
    futures::channel::mpsc::Sender<PumpedChunk>,
    futures::channel::mpsc::Receiver<PumpedChunk>,
) {
    futures::channel::mpsc::channel(bound)
}

/// Builds the refusal for a request field the caller was required to set.
fn required_field(field: &str, reason: &str) -> ConnectError {
    to_connect_error(&StateError::Malformed {
        field: field.to_owned(),
        reason: reason.to_owned(),
    })
}

/// Reads the identity, addressing, and preconditions a feed command carries.
fn command_metadata(
    command: impl Into<Option<polyc_proto::proto::polychrome::state::v1::FeedCommand>>,
) -> Result<CommandMetadata, ConnectError> {
    Ok(Kernel::<CommandMetadata>::try_from(
        command.into().ok_or_else(|| {
            required_field("command", "a feed command carries its command identity")
        })?,
    )
    .map_err(|e| to_connect_error(&e))?
    .into_inner())
}

/// What one subscription's pump needs, once it is off the request.
struct Pump {
    feed: Arc<dyn JournalFeed>,
    draining: Arc<AtomicBool>,
    declared: DeclaredCall,
    request: SubscribeCommits,
    cursor: Cursor,
    tuning: FeedStreamTuning,
}

/// Encodes one semantic feed chunk and, where Buffa framing pushes it over the
/// transport ceiling, shortens it at a commit boundary until the actual wire
/// message fits.
///
/// The kernel deliberately counts semantic content rather than knowing about a
/// transport encoding. That is sufficient for allocation control but not for
/// liveness: thousands of small records can add more framing than a fixed
/// headroom predicts. The adapter owns that last bound because it is the first
/// layer that can measure the real encoded message. A commit is never split.
fn bounded_wire_chunk(
    chunk: &polyc_state::feed::FeedChunk,
    previous: &Cursor,
) -> Result<(PbFeedChunk, Option<Cursor>, bool), StateError> {
    let original_len = chunk.len();
    let mut wire = PbFeedChunk::from(Kernel(chunk));
    while usize::try_from(buffa::Message::encoded_len(&wire)).unwrap_or(usize::MAX)
        > MAX_FEED_WIRE_MESSAGE_BYTES
    {
        if wire.records.len() <= 1 {
            return Err(StateError::BoundsExceeded {
                bound: BoundKind::PayloadBytes,
                limit: u64::try_from(MAX_FEED_WIRE_MESSAGE_BYTES).unwrap_or(u64::MAX),
                requested: u64::from(buffa::Message::encoded_len(&wire)),
            });
        }
        wire.records.pop();
    }

    let shortened = wire.records.len() < original_len;
    if shortened {
        let delivered = wire
            .records
            .last()
            .expect("a shortened non-empty feed chunk retains one commit")
            .position;
        let cursor = previous.snapshot().map_or_else(
            || Cursor::at(JournalPosition::new(delivered)),
            |snapshot| Cursor::in_snapshot(snapshot.clone(), JournalPosition::new(delivered)),
        );
        wire.next = buffa::MessageField::some(Kernel(&cursor).into());
        wire.end =
            polyc_proto::proto::polychrome::state::v1::StreamEnd::from(Kernel(StreamEnd::More))
                .into();
        Ok((wire, Some(cursor), false))
    } else {
        Ok((
            wire,
            feed::cursor_after(chunk, Some(previous)),
            chunk.is_drained(),
        ))
    }
}

impl Pump {
    /// Returns the subscription resuming from the cursor the pump holds.
    fn resumed(&self) -> SubscribeCommits {
        let resumed = SubscribeCommits::new(
            self.request.partition().clone(),
            ReadStart::Resume(self.cursor.clone()),
            self.request.max_chunk_commits(),
        );
        match self.request.consumer() {
            Some(consumer) => resumed.on_behalf_of(consumer.clone()),
            None => resumed,
        }
    }

    /// Returns the chunk a draining listener ends the stream with.
    fn drained(&self) -> PbFeedChunk {
        PbFeedChunk::from(Kernel(&feed::drained_chunk(self.cursor.clone())))
    }

    /// Returns the chunk a pausing pump ends the stream with.
    ///
    /// Empty, carrying the resume cursor, and marked [`StreamEnd::More`] —
    /// which is the whole message: *more may follow, resume from here*. It is
    /// deliberately not [`StreamEnd::Exhausted`], because a quiet partition is
    /// not a finished one, and a consumer that could not tell those apart would
    /// treat every silent minute as the end of the feed.
    fn paused(&self) -> PbFeedChunk {
        PbFeedChunk::from(Kernel(&StreamChunk::new(
            Vec::new(),
            Some(self.cursor.clone()),
            StreamEnd::More,
        )))
    }

    /// Leaves the resume marker behind for a consumer that fell behind.
    ///
    /// Its channel is full by definition, so there is no room for the marker
    /// yet. The pump waits briefly for the consumer to read one chunk; if it
    /// never does, the pump leaves without the marker, and the last chunk it
    /// did deliver already carried a cursor that resumes just as well.
    fn shed(&self, sender: &mut futures::channel::mpsc::Sender<PumpedChunk>) {
        for _ in 0..SHED_MARKER_ATTEMPTS {
            if sender.is_closed() {
                return;
            }
            if sender.try_send(Ok(self.paused())).is_ok() {
                return;
            }
            std::thread::sleep(TAIL_POLL);
        }
    }

    /// Drives the subscription until the consumer, the listener, or the module
    /// ends it.
    ///
    /// Runs on the blocking pool: every call into the module below is
    /// synchronous, and the channel writes are not `async` either, so nothing
    /// here needs a runtime and nothing here occupies a runtime worker.
    fn run(mut self, mut sender: futures::channel::mpsc::Sender<PumpedChunk>) {
        let mut idle = Duration::ZERO;
        loop {
            // The consumer is gone: its connection dropped, or the listener tore
            // the call down. Nothing left to serve, and continuing would leak
            // this thread for as long as the process lived.
            if sender.is_closed() {
                return;
            }
            // A tail that has been caught up for long enough pauses rather than
            // being held open indefinitely. The marker says "more may follow",
            // so the client resumes from its cursor instead of concluding the
            // feed ended.
            if idle >= self.tuning.max_tail_idle() {
                let _ = sender.try_send(Ok(self.paused()));
                return;
            }
            // A draining listener ends the stream gracefully. The consumer keeps
            // its cursor and reconnects wherever it is served next.
            if self.draining.load(Ordering::Relaxed) {
                let _ = sender.try_send(Ok(self.drained()));
                return;
            }

            let context: CallContext = self.declared.origin_relative_context();
            let chunk = match self.feed.commits(self.resumed(), &context) {
                Ok(chunk) => chunk,
                Err(error) => {
                    // The refusal is the last thing on the stream, and it
                    // carries the typed outcome — a compacted range above all,
                    // which tells the consumer to rebootstrap rather than
                    // reconnect.
                    let _ = sender.try_send(Err(to_connect_error(&error)));
                    return;
                }
            };
            // A caught-up partition produces nothing to send. Waiting is the
            // whole of tailing; sending an empty chunk every interval would be
            // a busy loop with a wire attached.
            if chunk.is_empty() && !chunk.is_drained() {
                std::thread::sleep(TAIL_POLL);
                idle = idle.saturating_add(TAIL_POLL);
                continue;
            }
            idle = Duration::ZERO;

            // The semantic module bounds content before cloning it; this
            // adapter additionally measures Buffa's actual encoded message and
            // shortens only at commit boundaries. The cursor therefore names
            // exactly what this wire item carries, never commits trimmed from
            // its tail.
            let (wire, advanced, drained) = match bounded_wire_chunk(&chunk, &self.cursor) {
                Ok(bounded) => bounded,
                Err(error) => {
                    let _ = sender.try_send(Err(to_connect_error(&error)));
                    return;
                }
            };

            // Never `send`, always `try_send`: the declared backpressure is to
            // shed a consumer that cannot keep up, and waiting here is what
            // "block the producer" would mean instead. The chunk that did not
            // fit is dropped rather than queued behind the consumer — it is
            // redelivered when the consumer resumes, which at-least-once
            // delivery already permits.
            if sender.try_send(Ok(wire)).is_err() {
                // The cursor is deliberately still where the last delivered
                // chunk left it, so the marker resumes from what the consumer
                // has, not from what the pump read.
                self.shed(&mut sender);
                return;
            }
            if let Some(next) = advanced {
                self.cursor = next;
            }
            if drained {
                return;
            }
        }
    }
}

// The generated trait returns `impl Encodable<Reply>`; every handler here
// returns the concrete reply type, which refines that bound rather than
// matching it. Same allow the journal's handlers carry.
#[allow(refining_impl_trait)]
impl StateFeedService for FeedSvc {
    async fn create_snapshot(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, CreateFeedSnapshotRequest>,
    ) -> ServiceResult<CreateFeedSnapshotReply> {
        let message = request.to_owned_message();
        let (declared, span) = self.admit(&ctx, message.context, "CreateSnapshot")?;
        // A span guard held across an `await` is illegal — it would leak the
        // span onto whatever task the thread picks up next. It is sound here
        // only because every feed call below is synchronous and this handler
        // never awaits while the guard is alive.
        let _entered = span.enter();

        let metadata = command_metadata(message.command)?;
        let snapshot = self
            .feed
            .create_snapshot(
                CreateSnapshot::new(metadata),
                &declared.origin_relative_context(),
            )
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(CreateFeedSnapshotReply {
            snapshot: buffa::MessageField::some(Kernel(&snapshot).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn subscribe_commits(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, SubscribeCommitsRequest>,
    ) -> ServiceResult<ServiceStream<PbFeedChunk>> {
        let message = request.to_owned_message();
        let (declared, span) = self.admit(&ctx, message.context, "SubscribeCommits")?;
        let _entered = span.enter();

        let subscription = subscribe_request(
            message.partition,
            message.start,
            message.max_chunk_commits,
            message.consumer,
        )
        .map_err(|e| to_connect_error(&e))?;

        // Resolved once, here, so a start this listener could never serve is
        // refused before the stream opens rather than as its first item.
        let cursor = feed::resume_cursor(subscription.start()).map_err(|e| to_connect_error(&e))?;

        let (sender, receiver) = new_chunk_channel(self.tuning.stream_bound());
        let pump = Pump {
            feed: Arc::clone(&self.feed),
            draining: Arc::clone(&self.draining),
            declared,
            request: subscription,
            cursor,
            tuning: self.tuning,
        };
        tokio::task::spawn_blocking(move || pump.run(sender));
        Response::stream_ok(receiver)
    }

    async fn register_projector(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, RegisterProjectorRequest>,
    ) -> ServiceResult<RegisterProjectorReply> {
        let message = request.to_owned_message();
        let (declared, span) = self.admit(&ctx, message.context, "RegisterProjector")?;
        let _entered = span.enter();

        let metadata = command_metadata(message.command)?;
        let registration = Kernel::<ProjectorRegistration>::try_from(
            message.registration.into_option().ok_or_else(|| {
                required_field("registration", "a registration carries what it declares")
            })?,
        )
        .map_err(|e| to_connect_error(&e))?
        .into_inner();

        let receipt = self
            .feed
            .register_projector(
                RegisterProjector::new(metadata, registration),
                &declared.origin_relative_context(),
            )
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(RegisterProjectorReply {
            receipt: buffa::MessageField::some(Kernel(&receipt).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn acknowledge_projector_cursor(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, AcknowledgeProjectorCursorRequest>,
    ) -> ServiceResult<AcknowledgeProjectorCursorReply> {
        let message = request.to_owned_message();
        let (declared, span) = self.admit(&ctx, message.context, "AcknowledgeProjectorCursor")?;
        let _entered = span.enter();

        let metadata = command_metadata(message.command)?;
        let cursor = Kernel::<Cursor>::from(message.cursor.into_option().ok_or_else(|| {
            required_field("cursor", "an acknowledgement names how far it applied")
        })?)
        .into_inner();

        let receipt = self
            .feed
            .acknowledge(
                AcknowledgeProjectorCursor::new(
                    metadata,
                    ConsumerId::new(message.consumer),
                    cursor,
                ),
                &declared.origin_relative_context(),
            )
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(AcknowledgeProjectorCursorReply {
            receipt: buffa::MessageField::some(Kernel(&receipt).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn list_projectors(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ListProjectorsRequest>,
    ) -> ServiceResult<ListProjectorsReply> {
        let message = request.to_owned_message();
        let (declared, span) = self.admit(&ctx, message.context, "ListProjectors")?;
        let _entered = span.enter();

        let listing = ListProjectors::new(
            polyc_state::id::PartitionId::new(message.partition),
            message.limit,
        );
        let listing = match message.consumer {
            Some(consumer) => listing.for_consumer(ConsumerId::new(consumer)),
            None => listing,
        };
        let listed = self
            .feed
            .projectors(listing, &declared.origin_relative_context())
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(ListProjectorsReply {
            projectors: listed
                .projectors()
                .iter()
                .map(|status| Kernel(status).into())
                .collect(),
            completeness: PageCompleteness::from(Kernel(listed.completeness())).into(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn get_feed_retention(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, GetFeedRetentionRequest>,
    ) -> ServiceResult<GetFeedRetentionReply> {
        let message = request.to_owned_message();
        let (declared, span) = self.admit(&ctx, message.context, "GetFeedRetention")?;
        let _entered = span.enter();

        let retention = self
            .feed
            .retention(
                GetFeedRetention::new(polyc_state::id::PartitionId::new(message.partition)),
                &declared.origin_relative_context(),
            )
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(GetFeedRetentionReply {
            retention: buffa::MessageField::some(Kernel(&retention).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn compact_feed_prefix(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, CompactFeedPrefixRequest>,
    ) -> ServiceResult<CompactFeedPrefixReply> {
        let message = request.to_owned_message();
        let (declared, span) = self.admit(&ctx, message.context, "CompactFeedPrefix")?;
        let _entered = span.enter();

        let metadata = command_metadata(message.command)?;
        let compaction = self
            .feed
            .compact(
                CompactFeedPrefix::new(metadata, JournalPosition::new(message.through)),
                &declared.origin_relative_context(),
            )
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(CompactFeedPrefixReply {
            compaction: buffa::MessageField::some(Kernel(&compaction).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn describe_feed_stream(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, DescribeFeedStreamRequest>,
    ) -> ServiceResult<DescribeFeedStreamReply> {
        let message = request.to_owned_message();
        let (_declared, span) = self.admit(&ctx, message.context, "DescribeFeedStream")?;
        let _entered = span.enter();

        Response::ok(DescribeFeedStreamReply {
            contract: buffa::MessageField::some(Kernel(self.feed.contract()).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }
}

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

    use std::time::Duration;

    use futures::StreamExt as _;
    use polyc_state::{
        deadline::{Deadline, MonotonicInstant},
        digest::ContentDigest,
        feed::{CommitEnvelope, FeedRecord},
        id::{CommandId, PartitionId},
        journal::{JournalRecord, JournalRecordDraft, RecordKind, RecordTrust},
        memory::MemoryFeed,
    };

    use super::*;

    fn partition() -> PartitionId {
        PartitionId::new("conv-1")
    }

    fn feed_of(commits: u64) -> Arc<MemoryFeed> {
        let feed = Arc::new(MemoryFeed::new());
        for n in 1..=commits {
            feed.append_commit(
                &partition(),
                &CommandId::new(format!("commit-{n}")),
                vec![JournalRecordDraft::new(
                    RecordKind::new("proof.commit"),
                    n.to_be_bytes().to_vec(),
                )],
            );
        }
        feed
    }

    fn pump_over(feed: Arc<dyn JournalFeed>, tuning: FeedStreamTuning) -> Pump {
        Pump {
            feed,
            draining: Arc::new(AtomicBool::new(false)),
            declared: DeclaredCall::live(state_audience(), Duration::MAX),
            request: SubscribeCommits::new(
                partition(),
                ReadStart::Resume(Cursor::at(JournalPosition::ORIGIN)),
                1,
            ),
            cursor: Cursor::at(JournalPosition::ORIGIN),
            tuning,
        }
    }

    /// Reads what a chunk carries, in the kernel's vocabulary.
    fn decode(chunk: PbFeedChunk) -> polyc_state::feed::FeedChunk {
        Kernel::<polyc_state::feed::FeedChunk>::try_from(chunk)
            .unwrap()
            .into_inner()
    }

    /// Builds the legal shape that invalidates a fixed Buffa-headroom
    /// assumption: all 32 commits and all 8,192 records fit the semantic byte
    /// budget, but their per-record framing does not fit in another 64 KiB.
    fn maximally_fragmented_chunk() -> polyc_state::feed::FeedChunk {
        let records = (1..=feed::MAX_CHUNK_COMMITS)
            .map(|commit| {
                let items = (0..polyc_state::journal::MAX_RECORDS_PER_BATCH)
                    .map(|index| {
                        JournalRecord::new(
                            JournalPosition::new(
                                u64::from(commit - 1)
                                    * u64::from(polyc_state::journal::MAX_RECORDS_PER_BATCH)
                                    + u64::from(index)
                                    + 1,
                            ),
                            RecordKind::new("k"),
                            RecordTrust::QuarantinedContent,
                            vec![b'x'; 766],
                        )
                    })
                    .collect::<Vec<_>>();
                FeedRecord::new(
                    JournalPosition::new(u64::from(commit)),
                    CommitEnvelope::new(
                        PartitionId::new("p"),
                        CommandId::new("c"),
                        ContentDigest::from_bytes([7; ContentDigest::LEN]),
                        JournalPosition::new(u64::from(commit - 1)),
                        JournalPosition::new(u64::from(commit)),
                        u64::from(polyc_state::journal::MAX_RECORDS_PER_BATCH),
                    ),
                    items,
                )
            })
            .collect();
        StreamChunk::new(
            records,
            Some(Cursor::at(JournalPosition::new(u64::from(
                feed::MAX_CHUNK_COMMITS,
            )))),
            StreamEnd::More,
        )
    }

    #[test]
    fn buffa_overhead_shortens_a_feed_only_at_commit_boundaries() {
        let semantic = maximally_fragmented_chunk();
        assert!(feed::chunk_is_honest(&semantic, None));
        let unbounded = PbFeedChunk::from(Kernel(&semantic));
        assert!(
            usize::try_from(buffa::Message::encoded_len(&unbounded)).unwrap()
                > MAX_FEED_WIRE_MESSAGE_BYTES,
            "the fixture must reproduce framing amplification beyond the old fixed headroom"
        );

        let (wire, advanced, drained) =
            bounded_wire_chunk(&semantic, &Cursor::at(JournalPosition::ORIGIN)).unwrap();
        assert!(
            usize::try_from(buffa::Message::encoded_len(&wire)).unwrap()
                <= MAX_FEED_WIRE_MESSAGE_BYTES
        );
        assert!(
            !wire.records.is_empty(),
            "at least one whole commit advances"
        );
        assert!(
            wire.records.len() < semantic.len(),
            "the oversized tail is deferred"
        );
        assert!(!drained, "a shortened chunk has more work by definition");

        let delivered = wire.records.last().unwrap().position;
        assert_eq!(
            advanced.unwrap().position(),
            JournalPosition::new(delivered)
        );
        let decoded = decode(wire);
        assert_eq!(decoded.end(), StreamEnd::More);
        assert!(feed::chunk_is_honest(&decoded, None));
        assert!(decoded.resumes_without_gap(Some(&Cursor::at(JournalPosition::ORIGIN))));
    }

    /// A consumer that cannot keep up is shed, and the marker left behind
    /// resumes from what it actually received.
    ///
    /// The shed is forced at the layer that owns it. Over a real connection the
    /// transport's own buffers absorb a handful of small chunks long before the
    /// pump's channel fills, so a wire test cannot reach this path reliably —
    /// and this is the path where getting the cursor wrong would silently cost
    /// a commit.
    #[test]
    fn a_shed_consumer_gets_a_resume_marker_for_what_it_actually_received() {
        let pump = pump_over(
            feed_of(8) as Arc<dyn JournalFeed>,
            FeedStreamTuning::new()
                .with_stream_bound(1)
                .with_max_tail_idle(Duration::from_millis(50)),
        );
        let (sender, mut receiver) = new_chunk_channel(1);

        // The consumer stalls long enough for the pump to fill the channel and
        // shed, then starts reading.
        let reader = std::thread::spawn(move || {
            std::thread::sleep(Duration::from_millis(200));
            futures::executor::block_on(async move {
                let mut received = Vec::new();
                while let Some(item) = receiver.next().await {
                    received.push(decode(item.unwrap()));
                }
                received
            })
        });
        pump.run(sender);
        let received = reader.join().unwrap();

        assert!(
            received.len() >= 2,
            "the consumer received what fit before it fell behind: {}",
            received.len()
        );
        let marker = received.last().unwrap();
        assert!(marker.is_empty(), "the marker delivers nothing new");
        assert_eq!(
            marker.end(),
            StreamEnd::More,
            "a shed says more may follow, so the consumer resumes rather than stopping"
        );

        // The whole point: the marker resumes from the last chunk the consumer
        // actually got, never past a chunk the pump dropped.
        let delivered = received
            .iter()
            .rev()
            .find(|chunk| !chunk.is_empty())
            .expect("at least one chunk carried commits");
        let last_delivered = feed::cursor_after(delivered, None).unwrap();
        assert_eq!(
            marker.next_cursor(),
            Some(&last_delivered),
            "a marker past a dropped chunk would manufacture the gap it exists to avoid"
        );

        // And resuming there really does hand back everything after it.
        let resumed = feed_of(8)
            .commits(
                SubscribeCommits::new(partition(), ReadStart::Resume(last_delivered.clone()), 8),
                &CallContext::new(
                    Deadline::at(MonotonicInstant::from_nanos(u64::MAX)),
                    polyc_state::cancel::CancellationToken::new(),
                ),
            )
            .unwrap();
        assert!(!feed::chunk_skips_a_commit(&resumed, Some(&last_delivered)));
        assert_eq!(
            resumed
                .records()
                .first()
                .map(polyc_state::page::Positioned::position),
            Some(last_delivered.position().next()),
            "the resume continues immediately after what was delivered"
        );
    }

    /// An idle tail pauses rather than ending, and the pause carries the cursor.
    #[test]
    fn an_idle_tail_pauses_with_the_cursor_it_reached() {
        let pump = pump_over(
            feed_of(2) as Arc<dyn JournalFeed>,
            FeedStreamTuning::new()
                .with_stream_bound(8)
                .with_max_tail_idle(Duration::from_millis(50)),
        );
        let (sender, mut receiver) = new_chunk_channel(8);
        pump.run(sender);

        let received: Vec<_> = futures::executor::block_on(async move {
            let mut received = Vec::new();
            while let Some(item) = receiver.next().await {
                received.push(decode(item.unwrap()));
            }
            received
        });

        let marker = received.last().unwrap();
        assert!(marker.is_empty());
        assert_eq!(marker.end(), StreamEnd::More);
        assert_eq!(
            marker.next_cursor().map(Cursor::position),
            Some(JournalPosition::new(2)),
            "the pause resumes after everything the tail delivered"
        );
    }
}