polyc-state-connect 2026.10.2

//! The server half of the conformance surface.
//!
//! Every handler is the same four lines: 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 — the admission
//! functions and the module own every decision between them (INV-24) — and no
//! handler ever mints a receipt of its own (INV-22): it carries across exactly
//! the one the module recorded.

use std::{
    future::Future,
    sync::{Arc, atomic::AtomicBool},
};

use connectrpc::{ConnectError, RequestContext, Response, Router, ServiceRequest, ServiceResult};
use polyc_proto::proto::polychrome::state::v1::{
    AdvanceClockReply, AdvanceClockRequest, CreateSnapshotReply, CreateSnapshotRequest,
    DescribeStreamReply, DescribeStreamRequest, GetReceiptReply, GetReceiptRequest, ObserveReply,
    ObserveRequest, ReadChunkReply, ReadChunkRequest, ReadPageReply, ReadPageRequest,
    StateConformanceService, StateConformanceServiceExt, SubmitReply, SubmitRequest,
};
use polyc_state::{
    conformance::{SyntheticCommand, family},
    context::CallContext,
    error::StateError,
    id::{Audience, CommandId, OperationFamily},
    page::PageRequest,
    stream::StreamRequest,
};

use crate::{
    admission::{
        check_audience, check_call_context_version, check_not_draining, check_transport_deadline,
    },
    conformance::backend::ConformanceBackend,
    error::to_connect_error,
    wire::{DeclaredCall, Kernel, declared_call, duration_from_nanos},
};

/// The conformance kit's synthetic family, served over Connect.
///
/// A test surface, never an authority module. A listener mounts it only when a
/// composition hands it a backend; the shipped binary hands it none.
pub struct ConformanceSvc {
    backend: Arc<dyn ConformanceBackend>,
    draining: Arc<AtomicBool>,
}

impl ConformanceSvc {
    /// Serves `backend`, refusing new calls once `draining` is set.
    #[must_use]
    pub const fn new(backend: Arc<dyn ConformanceBackend>, draining: Arc<AtomicBool>) -> Self {
        Self { backend, draining }
    }

    /// 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 {
        OperationFamily::new(family::FAMILY)
    }

    /// Returns the audience this surface serves — the family's own.
    fn served_audience() -> Audience {
        Audience::new(family::AUDIENCE)
    }

    /// Admits one call, or refuses it with the typed outcome it earned.
    ///
    /// Order matters, and it is the order the design names: lifecycle first
    /// (a draining listener decides nothing at all), then version, then
    /// authorization, then budget. Checking authorization before version would
    /// let a peer probe an audience with a shape this build cannot even read.
    fn admit(
        &self,
        ctx: &RequestContext,
        context: impl Into<Option<polyc_proto::proto::polychrome::state::v1::CallContext>>,
    ) -> Result<CallContext, ConnectError> {
        check_not_draining(self.draining.load(std::sync::atomic::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(&declared.audience, &Self::served_audience(), &family)
            .map_err(|e| to_connect_error(&e))?;
        check_transport_deadline(ctx.time_remaining(), &family)
            .map_err(|e| to_connect_error(&e))?;
        Ok(declared
            .origin_relative_context()
            .in_frame(self.backend.now()))
    }
}

// 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 control plane's handlers carry.
#[allow(refining_impl_trait)]
impl StateConformanceService for ConformanceSvc {
    fn submit(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, SubmitRequest>,
    ) -> impl Future<Output = ServiceResult<SubmitReply>> {
        std::future::ready((|| {
            let message = request.to_owned_message();
            let context = self.admit(&ctx, message.context)?;
            let command = Kernel::<SyntheticCommand>::try_from(
                message.command.into_option().ok_or_else(|| {
                    to_connect_error(&StateError::Malformed {
                        field: "command".to_owned(),
                        reason: "a submission carries the command it submits".to_owned(),
                    })
                })?,
            )
            .map_err(|e| to_connect_error(&e))?
            .into_inner();

            let receipt = self
                .backend
                .submit(command, &context)
                .map_err(|e| to_connect_error(&e))?;
            Response::ok(SubmitReply {
                receipt: buffa::MessageField::some(Kernel(&receipt).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        })())
    }

    fn get_receipt(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, GetReceiptRequest>,
    ) -> impl Future<Output = ServiceResult<GetReceiptReply>> {
        std::future::ready((|| {
            let message = request.to_owned_message();
            let _context = self.admit(&ctx, message.context)?;
            let recorded = self
                .backend
                .receipt(&CommandId::new(message.command_id))
                .map_err(|e| to_connect_error(&e))?;
            Response::ok(GetReceiptReply {
                receipt: recorded
                    .as_ref()
                    .map_or_else(buffa::MessageField::default, |receipt| {
                        buffa::MessageField::some(Kernel(receipt).into())
                    }),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        })())
    }

    fn create_snapshot(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, CreateSnapshotRequest>,
    ) -> impl Future<Output = ServiceResult<CreateSnapshotReply>> {
        std::future::ready((|| {
            let message = request.to_owned_message();
            let context = self.admit(&ctx, message.context)?;
            let snapshot = self
                .backend
                .create_snapshot(&context)
                .map_err(|e| to_connect_error(&e))?;
            Response::ok(CreateSnapshotReply {
                snapshot: snapshot.as_str().to_owned(),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        })())
    }

    fn read_page(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ReadPageRequest>,
    ) -> impl Future<Output = ServiceResult<ReadPageReply>> {
        std::future::ready((|| {
            let message = request.to_owned_message();
            let context = self.admit(&ctx, message.context)?;
            let page_request =
                Kernel::<PageRequest>::try_from(message.page.into_option().ok_or_else(|| {
                    to_connect_error(&StateError::Malformed {
                        field: "page".to_owned(),
                        reason: "a bounded read carries the page it asks for".to_owned(),
                    })
                })?)
                .map_err(|e| to_connect_error(&e))?
                .into_inner();

            let page = self
                .backend
                .read_page(page_request, &context)
                .map_err(|e| to_connect_error(&e))?;
            Response::ok(ReadPageReply {
                page: buffa::MessageField::some(Kernel(&page).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        })())
    }

    fn read_chunk(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ReadChunkRequest>,
    ) -> impl Future<Output = ServiceResult<ReadChunkReply>> {
        std::future::ready((|| {
            let message = request.to_owned_message();
            let context = self.admit(&ctx, message.context)?;
            let chunk_request = Kernel::<StreamRequest>::try_from(
                message.chunk.into_option().ok_or_else(|| {
                    to_connect_error(&StateError::Malformed {
                        field: "chunk".to_owned(),
                        reason: "a bounded stream read carries the chunk it asks for".to_owned(),
                    })
                })?,
            )
            .map_err(|e| to_connect_error(&e))?
            .into_inner();

            let chunk = self
                .backend
                .read_chunk(chunk_request, &context)
                .map_err(|e| to_connect_error(&e))?;
            Response::ok(ReadChunkReply {
                chunk: buffa::MessageField::some(Kernel(&chunk).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        })())
    }

    fn describe_stream(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, DescribeStreamRequest>,
    ) -> impl Future<Output = ServiceResult<DescribeStreamReply>> {
        std::future::ready((|| {
            let message = request.to_owned_message();
            let _context = self.admit(&ctx, message.context)?;
            Response::ok(DescribeStreamReply {
                contract: buffa::MessageField::some(Kernel(self.backend.stream_contract()).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        })())
    }

    fn observe(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ObserveRequest>,
    ) -> impl Future<Output = ServiceResult<ObserveReply>> {
        std::future::ready((|| {
            let message = request.to_owned_message();
            let _context = self.admit(&ctx, message.context)?;
            Response::ok(ObserveReply {
                applied_effects: self.backend.applied_effects(),
                now_nanos: self.backend.now().as_nanos(),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        })())
    }

    fn advance_clock(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, AdvanceClockRequest>,
    ) -> impl Future<Output = ServiceResult<AdvanceClockReply>> {
        std::future::ready((|| {
            let message = request.to_owned_message();
            let _context = self.admit(&ctx, message.context)?;
            self.backend
                .advance_clock(duration_from_nanos(message.budget_nanos));
            Response::ok(AdvanceClockReply {
                now_nanos: self.backend.now().as_nanos(),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        })())
    }
}