polyc-state-connect 2026.10.2

//! 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::future::Future;
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, CreateJournalDirectorySnapshotReply,
    CreateJournalDirectorySnapshotRequest, DestroyJournalPartitionReply,
    DestroyJournalPartitionRequest, ExciseJournalRecordsReply, ExciseJournalRecordsRequest,
    GetJournalHeadReply, GetJournalHeadRequest, GetJournalProofReply, GetJournalProofRequest,
    GetJournalReceiptReply, GetJournalReceiptRequest, GetJournalRootReply, GetJournalRootRequest,
    GetJournalSourceReply, GetJournalSourceRequest, GetPartitionLastModifiedReply,
    GetPartitionLastModifiedRequest, ListJournalDirectorySnapshotReply,
    ListJournalDirectorySnapshotRequest, ReadJournalRangeReply, ReadJournalRangeRequest,
    ReleaseJournalDirectorySnapshotReply, ReleaseJournalDirectorySnapshotRequest,
    RepairJournalPartitionReply, RepairJournalPartitionRequest, StageJournalDestructionReply,
    StageJournalDestructionRequest, StateJournalService, StateJournalServiceExt,
    VerifyPartitionReplayReply, VerifyPartitionReplayRequest,
};
use polyc_state::{
    command::CommandMetadata,
    context::CallContext,
    error::StateError,
    id::{CommandId, OperationFamily, PartitionId},
    journal::{
        self, CommitJournalBatch, CreateJournalDirectorySnapshot, DestroyPartition,
        ExcisePartitionRecords, GetJournalHead, GetJournalRoot, GetJournalSource,
        GetPartitionLastModified, JournalDirectorySnapshotId, ReadJournalRange, RecordDecision,
        ReleaseJournalDirectorySnapshot, RepairPartition,
    },
    revision::JournalPosition,
};

use crate::{
    admission::{
        AudienceBinding, PeerIdentity, check_audience_binding, check_call_context_version,
        check_not_draining, check_transport_deadline, slot_wait, state_audience,
    },
    blocking,
    error::to_connect_error,
    journal::{
        verify::{JournalAuthority, VerifyPartitionReplay},
        wire::{directory_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 runs its whole
/// authority call inside one `blocking` closure on Tokio's blocking pool —
/// one call, one pool thread, one journal-gate slot — rather than parking a
/// runtime worker on the journal's mutation mutex. 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, std::time::Duration), 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),
            slot_wait(ctx.time_remaining(), &declared),
        ))
    }
}

/// 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 {
    fn commit_journal_batch(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, CommitJournalBatchRequest>,
    ) -> impl Future<Output = ServiceResult<CommitJournalBatchReply>> {
        let message = request.to_owned_message();
        async move {
            let (context, span, wait) = self.admit(&ctx, message.context, "CommitJournalBatch")?;

            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 journal = Arc::clone(&self.journal);
            let receipt = blocking::mutation(span, Self::family(), wait, move || {
                journal.commit_batch(batch, &context)
            })
            .await?;
            Response::ok(CommitJournalBatchReply {
                receipt: buffa::MessageField::some(Kernel(&receipt).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

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

            let partition = PartitionId::new(message.partition);
            let command_id = CommandId::new(message.command_id);
            let journal = Arc::clone(&self.journal);
            let recorded = blocking::read(span, Self::family(), wait, move || {
                journal.committed_receipt(&partition, &command_id)
            })
            .await?;
            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(),
            })
        }
    }

    fn read_journal_range(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ReadJournalRangeRequest>,
    ) -> impl Future<Output = ServiceResult<ReadJournalRangeReply>> {
        let message = request.to_owned_message();
        async move {
            let (context, span, wait) = self.admit(&ctx, message.context, "ReadJournalRange")?;

            let range =
                Kernel::<ReadJournalRange>::try_from(message.range.into_option().ok_or_else(
                    || required_field("range", "a bounded read carries the range it asks for"),
                )?)
                .map_err(|e| to_connect_error(&e))?
                .into_inner();

            let journal = Arc::clone(&self.journal);
            let read = blocking::read(span, Self::family(), wait, move || {
                journal.read_range(range, &context)
            })
            .await?;
            Response::ok(ReadJournalRangeReply {
                range: buffa::MessageField::some(Kernel(&read).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

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

            let journal = Arc::clone(&self.journal);
            let head = blocking::read(span, Self::family(), wait, move || {
                journal.head(
                    GetJournalHead::new(PartitionId::new(message.partition)),
                    &context,
                )
            })
            .await?;
            Response::ok(GetJournalHeadReply {
                head: buffa::MessageField::some(Kernel(head).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

    fn get_journal_source(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, GetJournalSourceRequest>,
    ) -> impl Future<Output = ServiceResult<GetJournalSourceReply>> {
        let message = request.to_owned_message();
        async move {
            let (context, span, wait) = self.admit(&ctx, message.context, "GetJournalSource")?;

            let journal = Arc::clone(&self.journal);
            let head = blocking::read(span, Self::family(), wait, move || {
                journal.source_head(
                    GetJournalSource::new(PartitionId::new(message.partition)),
                    &context,
                )
            })
            .await?;
            Response::ok(GetJournalSourceReply {
                head: head
                    .as_ref()
                    .map_or_else(buffa::MessageField::default, |head| {
                        buffa::MessageField::some(Kernel(head).into())
                    }),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

    fn create_journal_directory_snapshot(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, CreateJournalDirectorySnapshotRequest>,
    ) -> impl Future<Output = ServiceResult<CreateJournalDirectorySnapshotReply>> {
        let message = request.to_owned_message();
        async move {
            let (context, span, wait) =
                self.admit(&ctx, message.context, "CreateJournalDirectorySnapshot")?;

            let journal = Arc::clone(&self.journal);
            let snapshot = blocking::read(span, Self::family(), wait, move || {
                journal.create_directory_snapshot(CreateJournalDirectorySnapshot, &context)
            })
            .await?;
            Response::ok(CreateJournalDirectorySnapshotReply {
                snapshot: buffa::MessageField::some(Kernel(&snapshot).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

    fn list_journal_directory_snapshot(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ListJournalDirectorySnapshotRequest>,
    ) -> impl Future<Output = ServiceResult<ListJournalDirectorySnapshotReply>> {
        let message = request.to_owned_message();
        async move {
            let (context, span, wait) =
                self.admit(&ctx, message.context, "ListJournalDirectorySnapshot")?;

            let journal = Arc::clone(&self.journal);
            let page = blocking::read(span, Self::family(), wait, move || {
                journal.directory_page(
                    directory_listing(message.snapshot, message.start_after, message.limit),
                    &context,
                )
            })
            .await?;
            Response::ok(ListJournalDirectorySnapshotReply {
                page: buffa::MessageField::some(Kernel(&page).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

    fn release_journal_directory_snapshot(
        &self,
        ctx: RequestContext,
        request: ServiceRequest<'_, ReleaseJournalDirectorySnapshotRequest>,
    ) -> impl Future<Output = ServiceResult<ReleaseJournalDirectorySnapshotReply>> {
        let message = request.to_owned_message();
        async move {
            let (context, span, wait) =
                self.admit(&ctx, message.context, "ReleaseJournalDirectorySnapshot")?;

            let journal = Arc::clone(&self.journal);
            blocking::read(span, Self::family(), wait, move || {
                journal.release_directory_snapshot(
                    ReleaseJournalDirectorySnapshot::new(JournalDirectorySnapshotId::new(
                        message.snapshot,
                    )),
                    &context,
                )
            })
            .await?;
            Response::ok(ReleaseJournalDirectorySnapshotReply {
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

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

            let journal = Arc::clone(&self.journal);
            // A mutation, not a read: `verify_replay` finishes pending
            // excisions under the journal's mutation lock, so a lost result
            // may have applied durable bookkeeping. It queues on the replay
            // share, not the gate's — the expensive pass runs off the
            // mutation lock, so a long replay must not hold a commit's slot.
            let verdict = blocking::mutation(span, journal::replay_family(), wait, move || {
                journal.verify_replay(
                    VerifyPartitionReplay::new(PartitionId::new(message.partition)),
                    &context,
                )
            })
            .await?;
            Response::ok(Kernel(&verdict).into())
        }
    }

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

            let journal = Arc::clone(&self.journal);
            let reported = blocking::read(span, Self::family(), wait, move || {
                journal.last_modified(
                    GetPartitionLastModified::new(PartitionId::new(message.partition)),
                    &context,
                )
            })
            .await?;
            Response::ok(GetPartitionLastModifiedReply {
                at_ms: reported.at_ms(),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

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

            let journal = Arc::clone(&self.journal);
            let attestation = blocking::read(span, Self::family(), wait, move || {
                journal.root(
                    GetJournalRoot::new(PartitionId::new(message.partition)),
                    &context,
                )
            })
            .await?;
            Response::ok(GetJournalRootReply {
                attestation: buffa::MessageField::some(Kernel(&attestation).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

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

            let journal = Arc::clone(&self.journal);
            let proof = blocking::read(span, Self::family(), wait, move || {
                journal.proof(proof_request(message.partition, message.position), &context)
            })
            .await?;
            Response::ok(GetJournalProofReply {
                proof: buffa::MessageField::some(Kernel(&proof).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

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

            let command = destruction_command(message.command)?;
            let journal = Arc::clone(&self.journal);
            let receipt = blocking::mutation(span, Self::family(), wait, move || {
                journal.destroy(command, &context)
            })
            .await?;
            Response::ok(DestroyJournalPartitionReply {
                receipt: buffa::MessageField::some(Kernel(&receipt).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

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

            let command = destruction_command(message.command)?;
            let journal = Arc::clone(&self.journal);
            blocking::mutation(span, Self::family(), wait, move || {
                journal.stage_destroy(command, &context)
            })
            .await?;
            Response::ok(StageJournalDestructionReply::default())
        }
    }

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

            let command = destruction_command(message.command)?;
            let journal = Arc::clone(&self.journal);
            let receipt = blocking::mutation(span, Self::family(), wait, move || {
                journal.complete_destroy(command, &context)
            })
            .await?;
            Response::ok(CompleteJournalDestructionReply {
                receipt: buffa::MessageField::some(Kernel(&receipt).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

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

            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 journal = Arc::clone(&self.journal);
            let receipt = blocking::mutation(span, Self::family(), wait, move || {
                journal.excise(ExcisePartitionRecords::new(metadata, decisions), &context)
            })
            .await?;
            Response::ok(ExciseJournalRecordsReply {
                receipt: buffa::MessageField::some(Kernel(&receipt).into()),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
        }
    }

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

            let metadata = mutation_metadata(message.command)?;
            let journal = Arc::clone(&self.journal);
            let outcome = blocking::mutation(span, Self::family(), wait, move || {
                journal.repair(RepairPartition::new(metadata), &context)
            })
            .await?;
            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)
}