polyc-state-connect 2026.9.0

//! 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::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 {
    async fn submit(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, SubmitRequest>,
    ) -> ServiceResult<SubmitReply> {
        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(),
        })
    }

    async fn get_receipt(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, GetReceiptRequest>,
    ) -> ServiceResult<GetReceiptReply> {
        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(),
        })
    }

    async fn create_snapshot(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, CreateSnapshotRequest>,
    ) -> ServiceResult<CreateSnapshotReply> {
        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(),
        })
    }

    async fn read_page(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ReadPageRequest>,
    ) -> ServiceResult<ReadPageReply> {
        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(),
        })
    }

    async fn read_chunk(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ReadChunkRequest>,
    ) -> ServiceResult<ReadChunkReply> {
        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(),
        })
    }

    async fn describe_stream(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, DescribeStreamRequest>,
    ) -> ServiceResult<DescribeStreamReply> {
        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(),
        })
    }

    async fn observe(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ObserveRequest>,
    ) -> ServiceResult<ObserveReply> {
        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(),
        })
    }

    async fn advance_clock(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, AdvanceClockRequest>,
    ) -> ServiceResult<AdvanceClockReply> {
        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(),
        })
    }
}