polyc-state-connect 2026.8.3

//! The server half of the partition-journal surface.
//!
//! Every handler is the same shape: 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.
//!
//! This is the first authority surface the State plane serves, so admission is
//! stricter than the conformance surface's: the caller's proven mutual-TLS
//! identity is checked against an [`AudienceBinding`] rather than only against
//! the audience it addressed.

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

use connectrpc::{ConnectError, RequestContext, Response, Router, ServiceRequest, ServiceResult};
use polyc_proto::proto::polychrome::state::v1::{
    CommitJournalBatchReply, CommitJournalBatchRequest, CompleteJournalDestructionReply,
    CompleteJournalDestructionRequest, DestroyJournalPartitionReply,
    DestroyJournalPartitionRequest, ExciseJournalRecordsReply, ExciseJournalRecordsRequest,
    GetJournalHeadReply, GetJournalHeadRequest, GetJournalProofReply, GetJournalProofRequest,
    GetJournalReceiptReply, GetJournalReceiptRequest, GetJournalRootReply, GetJournalRootRequest,
    GetPartitionLastModifiedReply, GetPartitionLastModifiedRequest, ListJournalPartitionsReply,
    ListJournalPartitionsRequest, ReadJournalRangeReply, ReadJournalRangeRequest,
    RepairJournalPartitionReply, RepairJournalPartitionRequest, StageJournalDestructionReply,
    StageJournalDestructionRequest, StateJournalService, StateJournalServiceExt,
    VerifyPartitionReplayReply, VerifyPartitionReplayRequest,
};
use polyc_state::{
    command::CommandMetadata,
    context::CallContext,
    error::StateError,
    id::{CommandId, OperationFamily, PartitionId},
    journal::{
        self, CommitJournalBatch, DestroyPartition, ExcisePartitionRecords, GetJournalHead,
        GetJournalRoot, GetPartitionLastModified, ReadJournalRange, RecordDecision,
        RepairPartition,
    },
    revision::JournalPosition,
};

use crate::{
    admission::{
        AudienceBinding, PeerIdentity, check_audience_binding, check_call_context_version,
        check_not_draining, check_transport_deadline, state_audience,
    },
    error::to_connect_error,
    journal::{
        verify::{JournalAuthority, VerifyPartitionReplay},
        wire::{listing, proof_request},
    },
    trace::adopt_caller_trace,
    wire::{DeclaredCall, Kernel, declared_call},
};

/// State's partition journal, served over Connect.
///
/// A listener mounts this only once its module has passed the conformance kit;
/// the composition that mounts it is where that gate lives.
///
/// # Blocking
///
/// The module contract is synchronous, so every handler below blocks its
/// thread for the duration of the durable write or read rather than yielding
/// while it waits. A listener serving this needs enough threads for the calls
/// it admits. The deviation is deliberate and documented on the module that
/// carries it.
pub struct JournalSvc {
    journal: Arc<dyn JournalAuthority>,
    draining: Arc<AtomicBool>,
    binding: Arc<AudienceBinding>,
}

impl JournalSvc {
    /// Serves `journal` 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. It is the one name this plane serves,
    /// and a composition that could pass its own would be a composition that
    /// could disagree with its callers — silently, until the first call was
    /// denied.
    #[must_use]
    pub const fn new(
        journal: Arc<dyn JournalAuthority>,
        draining: Arc<AtomicBool>,
        binding: Arc<AudienceBinding>,
    ) -> Self {
        Self {
            journal,
            draining,
            binding,
        }
    }

    /// 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 {
        journal::family()
    }

    /// Returns the identity the connection proved, or anonymity when it proved
    /// none.
    ///
    /// The leaf is whatever rustls verified; nothing here re-verifies it,
    /// because a certificate that reached this point is one the handshake
    /// already accepted against this listener's authority.
    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.
    ///
    /// 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.
    ///
    /// The returned span is the one the handler runs in, already re-parented on
    /// the caller's trace.
    fn admit(
        &self,
        ctx: &RequestContext,
        context: impl Into<Option<polyc_proto::proto::polychrome::state::v1::CallContext>>,
        method: &'static str,
    ) -> Result<(CallContext, 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.origin_relative_context(),
            adopt_caller_trace(ctx.headers(), method),
        ))
    }
}

/// 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(),
    })
}

// 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 StateJournalService for JournalSvc {
    async fn commit_journal_batch(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, CommitJournalBatchRequest>,
    ) -> ServiceResult<CommitJournalBatchReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "CommitJournalBatch")?;
        // 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 journal call below is synchronous and this handler
        // never awaits while the guard is alive. An edit that introduces an
        // await inside this scope must replace the guard with `Instrument`.
        let _entered = span.enter();

        let batch = Kernel::<CommitJournalBatch>::try_from(
            message
                .batch
                .into_option()
                .ok_or_else(|| required_field("batch", "a commit carries the batch it commits"))?,
        )
        .map_err(|e| to_connect_error(&e))?
        .into_inner();

        let receipt = self
            .journal
            .commit_batch(batch, &context)
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(CommitJournalBatchReply {
            receipt: buffa::MessageField::some(Kernel(&receipt).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn get_journal_receipt(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, GetJournalReceiptRequest>,
    ) -> ServiceResult<GetJournalReceiptReply> {
        let message = request.to_owned_message();
        let (_context, span) = self.admit(&ctx, message.context, "GetJournalReceipt")?;
        let _entered = span.enter();

        let recorded = self
            .journal
            .committed_receipt(
                &PartitionId::new(message.partition),
                &CommandId::new(message.command_id),
            )
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(GetJournalReceiptReply {
            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 read_journal_range(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ReadJournalRangeRequest>,
    ) -> ServiceResult<ReadJournalRangeReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "ReadJournalRange")?;
        let _entered = span.enter();

        let range =
            Kernel::<ReadJournalRange>::from(message.range.into_option().ok_or_else(|| {
                required_field("range", "a bounded read carries the range it asks for")
            })?)
            .into_inner();

        let read = self
            .journal
            .read_range(range, &context)
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(ReadJournalRangeReply {
            range: buffa::MessageField::some(Kernel(&read).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn get_journal_head(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, GetJournalHeadRequest>,
    ) -> ServiceResult<GetJournalHeadReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "GetJournalHead")?;
        let _entered = span.enter();

        let head = self
            .journal
            .head(
                GetJournalHead::new(PartitionId::new(message.partition)),
                &context,
            )
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(GetJournalHeadReply {
            head: buffa::MessageField::some(Kernel(head).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn list_journal_partitions(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ListJournalPartitionsRequest>,
    ) -> ServiceResult<ListJournalPartitionsReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "ListJournalPartitions")?;
        let _entered = span.enter();

        let page = self
            .journal
            .list_partitions(listing(message.start_after, message.limit), &context)
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(ListJournalPartitionsReply {
            page: buffa::MessageField::some(Kernel(&page).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn verify_partition_replay(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, VerifyPartitionReplayRequest>,
    ) -> ServiceResult<VerifyPartitionReplayReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "VerifyPartitionReplay")?;
        let _entered = span.enter();

        let verdict = self
            .journal
            .verify_replay(
                VerifyPartitionReplay::new(PartitionId::new(message.partition)),
                &context,
            )
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(Kernel(&verdict).into())
    }

    async fn get_partition_last_modified(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, GetPartitionLastModifiedRequest>,
    ) -> ServiceResult<GetPartitionLastModifiedReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "GetPartitionLastModified")?;
        let _entered = span.enter();

        let reported = self
            .journal
            .last_modified(
                GetPartitionLastModified::new(PartitionId::new(message.partition)),
                &context,
            )
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(GetPartitionLastModifiedReply {
            at_ms: reported.at_ms(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn get_journal_root(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, GetJournalRootRequest>,
    ) -> ServiceResult<GetJournalRootReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "GetJournalRoot")?;
        let _entered = span.enter();

        let attestation = self
            .journal
            .root(
                GetJournalRoot::new(PartitionId::new(message.partition)),
                &context,
            )
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(GetJournalRootReply {
            attestation: buffa::MessageField::some(Kernel(&attestation).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn get_journal_proof(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, GetJournalProofRequest>,
    ) -> ServiceResult<GetJournalProofReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "GetJournalProof")?;
        let _entered = span.enter();

        let proof = self
            .journal
            .proof(proof_request(message.partition, message.position), &context)
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(GetJournalProofReply {
            proof: buffa::MessageField::some(Kernel(&proof).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn destroy_journal_partition(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, DestroyJournalPartitionRequest>,
    ) -> ServiceResult<DestroyJournalPartitionReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "DestroyJournalPartition")?;
        let _entered = span.enter();

        let command = destruction_command(message.command)?;
        let receipt = self
            .journal
            .destroy(command, &context)
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(DestroyJournalPartitionReply {
            receipt: buffa::MessageField::some(Kernel(&receipt).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn stage_journal_destruction(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, StageJournalDestructionRequest>,
    ) -> ServiceResult<StageJournalDestructionReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "StageJournalDestruction")?;
        let _entered = span.enter();

        let command = destruction_command(message.command)?;
        self.journal
            .stage_destroy(command, &context)
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(StageJournalDestructionReply::default())
    }

    async fn complete_journal_destruction(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, CompleteJournalDestructionRequest>,
    ) -> ServiceResult<CompleteJournalDestructionReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "CompleteJournalDestruction")?;
        let _entered = span.enter();

        let command = destruction_command(message.command)?;
        let receipt = self
            .journal
            .complete_destroy(command, &context)
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(CompleteJournalDestructionReply {
            receipt: buffa::MessageField::some(Kernel(&receipt).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn excise_journal_records(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ExciseJournalRecordsRequest>,
    ) -> ServiceResult<ExciseJournalRecordsReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "ExciseJournalRecords")?;
        let _entered = span.enter();

        let metadata = mutation_metadata(message.command)?;
        let decisions = message
            .decisions
            .into_iter()
            .map(|decision| Kernel::<RecordDecision>::try_from(decision).map(Kernel::into_inner))
            .collect::<Result<Vec<_>, _>>()
            .map_err(|e| to_connect_error(&e))?;

        let receipt = self
            .journal
            .excise(ExcisePartitionRecords::new(metadata, decisions), &context)
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(ExciseJournalRecordsReply {
            receipt: buffa::MessageField::some(Kernel(&receipt).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }

    async fn repair_journal_partition(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, RepairJournalPartitionRequest>,
    ) -> ServiceResult<RepairJournalPartitionReply> {
        let message = request.to_owned_message();
        let (context, span) = self.admit(&ctx, message.context, "RepairJournalPartition")?;
        let _entered = span.enter();

        let metadata = mutation_metadata(message.command)?;
        let outcome = self
            .journal
            .repair(RepairPartition::new(metadata), &context)
            .map_err(|e| to_connect_error(&e))?;
        Response::ok(RepairJournalPartitionReply {
            receipt: buffa::MessageField::some(Kernel(outcome.receipt()).into()),
            quarantined: outcome
                .quarantined()
                .iter()
                .map(|record| Kernel(record).into())
                .collect(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        })
    }
}

/// Reads the identity, addressing, and preconditions a mutation carries.
fn mutation_metadata(
    command: impl Into<Option<polyc_proto::proto::polychrome::state::v1::JournalMutationCommand>>,
) -> Result<CommandMetadata, ConnectError> {
    let command = command.into().ok_or_else(|| {
        required_field("command", "a journal mutation carries its command identity")
    })?;
    if command.expected_root.is_some() {
        return Err(required_field(
            "expected_root",
            "only a partition destruction can bind an expected root",
        ));
    }
    Ok(Kernel::<CommandMetadata>::try_from(command)
        .map_err(|e| to_connect_error(&e))?
        .into_inner())
}

fn destruction_command(
    command: impl Into<Option<polyc_proto::proto::polychrome::state::v1::JournalMutationCommand>>,
) -> Result<DestroyPartition, ConnectError> {
    Ok(
        Kernel::<DestroyPartition>::try_from(command.into().ok_or_else(|| {
            required_field(
                "command",
                "a journal destruction carries its command identity",
            )
        })?)
        .map_err(|e| to_connect_error(&e))?
        .into_inner(),
    )
}

/// The position a caller named, for a module that reports it back.
///
/// Exposed so a composition can build the same value a handler does when it
/// needs to name a position outside a request — a repair report, say — without
/// reaching for the generated types.
#[must_use]
pub const fn position(raw: u64) -> JournalPosition {
    JournalPosition::new(raw)
}