chio_kernel/
admission_operation.rs

1use chio_core::canonical::canonical_json_bytes;
2use chio_core::crypto::sha256_hex;
3pub use chio_core_types::provider_attempt::ProviderAttemptBindingV1;
4pub use chio_core_types::StoreMutationFence;
5use serde::{Deserialize, Deserializer, Serialize};
6
7mod capture;
8mod identity;
9mod projection;
10mod remote_projection;
11mod sequencer;
12mod state;
13mod store;
14
15use state::*;
16
17pub use capture::*;
18pub use identity::*;
19pub use projection::*;
20pub use remote_projection::*;
21pub use sequencer::*;
22pub use store::*;
23
24pub const ADMISSION_OPERATION_SCHEMA: &str = "chio.admission-operation.v1";
25pub const ADMISSION_OPERATION_DOMAIN: &[u8] = b"chio.admission-operation.v1\0";
26pub const ADMISSION_REQUEST_NAMESPACE_SCHEMA: &str = "chio.admission-request-namespace.v1";
27pub const ADMISSION_REQUEST_NAMESPACE_DOMAIN: &[u8] = b"chio.admission-request-namespace.v1\0";
28pub const ADMISSION_REQUEST_BINDING_DOMAIN: &[u8] = b"chio.admission-request-binding.v1\0";
29pub const ADMISSION_RECEIPT_METADATA_KEY: &str = "admission_operation";
30pub const ADMISSION_RECEIPT_SCHEMA: &str = "chio.admission-receipt.v1";
31pub const MAX_ADMISSION_IDENTIFIER_BYTES: usize = 512;
32pub const MAX_ADMISSION_TENANT_BYTES: usize = 256;
33pub const MAX_ADMISSION_ERROR_BYTES: usize = 2_048;
34pub const MAX_AUTHORIZATION_ARTIFACT_DIGESTS: usize = 8;
35pub const LOCAL_SYSTEM_TENANT_ID: &str = "local-system";
36pub(super) const I_JSON_MAX_SAFE_INTEGER: u64 = (1_u64 << 53) - 1;
37
38pub type AdmissionIdentifier = BoundedAdmissionText<MAX_ADMISSION_IDENTIFIER_BYTES>;
39pub type AdmissionErrorDetail = BoundedAdmissionText<MAX_ADMISSION_ERROR_BYTES>;
40
41pub fn expected_dispatch_committed_version(
42    kind: AdmissionOperationKind,
43    requirements: AdmissionParticipantRequirements,
44    prepared_version: u64,
45) -> Result<u64, AdmissionOperationError> {
46    requirements.validate_for_kind(kind)?;
47    state::dispatch_committed_version_from_prepared(kind, requirements, prepared_version)
48}
49
50#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
51pub enum AdmissionOperationError {
52    #[error("{field} must not be empty")]
53    Empty { field: &'static str },
54    #[error("{field} exceeds its {maximum}-byte limit")]
55    TooLong { field: &'static str, maximum: usize },
56    #[error("{field} contains a control character")]
57    ControlCharacter { field: &'static str },
58    #[error("{field} must not have leading or trailing whitespace")]
59    Padded { field: &'static str },
60    #[error("{field} must be lowercase SHA-256 hex")]
61    InvalidDigest { field: &'static str },
62    #[error("canonical JSON encoding failed: {0}")]
63    CanonicalJson(String),
64    #[error("operation version and coordinator lease epoch must be nonzero")]
65    ZeroVersionOrEpoch,
66    #[error("{field} must be an I-JSON safe positive integer")]
67    UnsafeInteger { field: &'static str },
68    #[error("operation version overflow")]
69    VersionOverflow,
70    #[error(
71        "durable admission may be disabled only for unsafe development with ephemeral receipts"
72    )]
73    UnsafeDurableAdmissionOff,
74    #[error("operation id does not match its immutable binding")]
75    OperationIdMismatch,
76    #[error("request namespace digest does not match its authenticated tenant binding")]
77    RequestNamespaceMismatch,
78    #[error("authenticated tenant id is reserved for the local-system namespace")]
79    ReservedLocalSystemTenant,
80    #[error("state {state:?} is invalid for operation kind {kind:?}")]
81    StateKindMismatch {
82        kind: AdmissionOperationKind,
83        state: AdmissionOperationState,
84    },
85    #[error("stored dispatch state does not match operation state")]
86    DispatchStateMismatch,
87    #[error("terminal replay reference does not match operation state")]
88    TerminalReplayMismatch,
89    #[error("command targets a different operation")]
90    WrongOperation,
91    #[error("recovery lease was not issued for this command version")]
92    LeaseVersionMismatch,
93    #[error("recovery lease is fenced by coordinator epoch")]
94    CoordinatorFenced,
95    #[error("recovery lease expired")]
96    LeaseExpired,
97    #[error("store mutation fence is invalid")]
98    InvalidStoreFence,
99    #[error("stale operation version: expected {expected}, actual {actual}")]
100    StaleVersion { expected: u64, actual: u64 },
101    #[error("illegal operation transition from {from:?} to {to:?}")]
102    IllegalTransition {
103        from: AdmissionOperationState,
104        to: AdmissionOperationState,
105    },
106    #[error("attachment {field} is already bound to a different value")]
107    AttachmentConflict { field: &'static str },
108    #[error("attachment {field} is not legal in state {state:?}")]
109    AttachmentPhase {
110        field: &'static str,
111        state: AdmissionOperationState,
112    },
113    #[error("attachment {field} is not allowed by the immutable operation binding")]
114    ForbiddenAttachment { field: &'static str },
115    #[error("stored threshold proposal does not match its bound digest")]
116    ThresholdProposalMismatch,
117    #[error("provider attempt does not match the immutable admission operation")]
118    ProviderAttemptBindingMismatch,
119    #[error("authorization artifact digests must be bounded, sorted, and unique")]
120    InvalidAuthorizationArtifactDigests,
121    #[error("capture decision does not match its record disposition")]
122    CaptureDispositionMismatch,
123    #[error("combined capture requires the exact CapturePending operation snapshot")]
124    CapturePreconditionMismatch,
125    #[error("combined capture participant binding does not match the operation")]
126    CaptureBindingMismatch,
127    #[error("combined capture authority does not match its fenced store")]
128    CaptureAuthorityMismatch,
129    #[error("combined capture global authority commit is invalid")]
130    CaptureCommitMismatch,
131    #[error("combined capture response digest is invalid")]
132    CaptureResponseDigestMismatch,
133    #[error("combined capture quota mutation is invalid")]
134    CaptureQuotaMismatch,
135    #[error("combined capture authorization expiry disposition is invalid")]
136    CaptureExpiryMismatch,
137    #[error("combined capture authority time is ahead of its qualified clock")]
138    CaptureAuthorityTimeMismatch,
139    #[error("admission operation command contains no mutation")]
140    EmptyCommand,
141    #[error("admission operation command attaches {field} more than once")]
142    DuplicateAttachment { field: &'static str },
143    #[error("verified economic mutation result binding is invalid")]
144    InvalidEconomicMutationBinding,
145    #[error("state transition requires attachment {field}")]
146    MissingParticipantAttachment { field: &'static str },
147    #[error("admission participant requirements are inconsistent")]
148    InvalidParticipantRequirements,
149    #[error("request binding hash does not cover its immutable request and requirements")]
150    RequestBindingMismatch,
151    #[error("terminal operation states require the atomic receipt-side projection")]
152    TerminalProjectionRequired,
153    #[error("terminal projection does not match the operation binding")]
154    TerminalProjectionBindingMismatch,
155    #[error("terminal projection store lacks required capability {capability}")]
156    MissingProjectionCapability { capability: &'static str },
157    #[error("durable admission mutation sequencer is poisoned")]
158    MutationSequencerPoisoned,
159}
160
161#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
162#[serde(rename_all = "snake_case")]
163pub enum AdmissionOperationState {
164    Prepared,
165    BrokerAttemptRegistered,
166    ApprovalRequired,
167    BudgetAuthorized,
168    ApprovalReserved,
169    ReadyToDispatch,
170    CapturePending,
171    DispatchCommitted,
172    Finalizing,
173    Completed,
174    CompensatedBeforeDispatch,
175    NotAcceptedAfterDispatchCommit,
176    OutcomeUnknownAfterDispatch,
177    MutationReady,
178    MutationSubmitted,
179    EconomicMutationApplied,
180    EconomicMutationNotApplied,
181}
182
183impl AdmissionOperationState {
184    pub const ALL: [Self; 17] = [
185        Self::Prepared,
186        Self::BrokerAttemptRegistered,
187        Self::ApprovalRequired,
188        Self::BudgetAuthorized,
189        Self::ApprovalReserved,
190        Self::ReadyToDispatch,
191        Self::CapturePending,
192        Self::DispatchCommitted,
193        Self::Finalizing,
194        Self::Completed,
195        Self::CompensatedBeforeDispatch,
196        Self::NotAcceptedAfterDispatchCommit,
197        Self::OutcomeUnknownAfterDispatch,
198        Self::MutationReady,
199        Self::MutationSubmitted,
200        Self::EconomicMutationApplied,
201        Self::EconomicMutationNotApplied,
202    ];
203
204    #[must_use]
205    pub fn is_terminal(self) -> bool {
206        matches!(
207            self,
208            Self::Completed
209                | Self::CompensatedBeforeDispatch
210                | Self::NotAcceptedAfterDispatchCommit
211                | Self::OutcomeUnknownAfterDispatch
212                | Self::EconomicMutationApplied
213                | Self::EconomicMutationNotApplied
214        )
215    }
216
217    fn is_pre_dispatch(self) -> bool {
218        matches!(
219            self,
220            Self::Prepared
221                | Self::BrokerAttemptRegistered
222                | Self::ApprovalRequired
223                | Self::BudgetAuthorized
224                | Self::ApprovalReserved
225                | Self::ReadyToDispatch
226                | Self::CapturePending
227        )
228    }
229}
230
231#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
232#[serde(rename_all = "snake_case")]
233pub enum AdmissionDispatchState {
234    NotCommitted,
235    CapturePending,
236    Committed,
237    Finalizing,
238    Terminal,
239    NotApplicable,
240}
241
242#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
243pub struct AdmissionDispatchCommitBindingV1 {
244    pub committed_version: u64,
245    pub coordinator_lease_id: AdmissionIdentifier,
246    pub coordinator_lease_epoch: u64,
247    pub store_fence: StoreMutationFence,
248    pub provider_attempt: Option<ProviderAttemptBindingV1>,
249}
250
251impl<'de> Deserialize<'de> for AdmissionDispatchCommitBindingV1 {
252    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
253    where
254        D: Deserializer<'de>,
255    {
256        #[derive(Deserialize)]
257        #[serde(deny_unknown_fields)]
258        struct Persisted {
259            committed_version: u64,
260            coordinator_lease_id: AdmissionIdentifier,
261            coordinator_lease_epoch: u64,
262            store_fence: StoreMutationFence,
263            provider_attempt: Option<ProviderAttemptBindingV1>,
264        }
265        let value = Persisted::deserialize(deserializer)?;
266        let binding = Self {
267            committed_version: value.committed_version,
268            coordinator_lease_id: value.coordinator_lease_id,
269            coordinator_lease_epoch: value.coordinator_lease_epoch,
270            store_fence: value.store_fence,
271            provider_attempt: value.provider_attempt,
272        };
273        binding.validate().map_err(serde::de::Error::custom)?;
274        Ok(binding)
275    }
276}
277
278impl AdmissionDispatchCommitBindingV1 {
279    fn validate(&self) -> Result<(), AdmissionOperationError> {
280        validate_positive_ijson("committed_version", self.committed_version)?;
281        validate_positive_ijson("coordinator_lease_epoch", self.coordinator_lease_epoch)?;
282        validate_store_fence(&self.store_fence)?;
283        self.provider_attempt
284            .as_ref()
285            .map_or(Ok(()), ProviderAttemptBindingV1::validate)
286            .map_err(|_| AdmissionOperationError::ProviderAttemptBindingMismatch)
287    }
288}
289
290#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
291pub enum AdmissionAttachment {
292    ThresholdProposalHash(AdmissionDigest),
293    ThresholdProposal(Box<chio_core::capability::governance::ThresholdApprovalProposal>),
294    SupplementalAuthorizationDigest(AdmissionDigest),
295    BrokerAttempt(ProviderAttemptBindingV1),
296    BudgetHoldId(AdmissionIdentifier),
297    ApprovalSetHash(AdmissionDigest),
298    ExecutionNonceId(AdmissionIdentifier),
299    OutcomeEligibilityDigest(AdmissionDigest),
300    PaymentParticipantId(AdmissionIdentifier),
301    ToolOutcomeId(AdmissionDigest),
302    ChannelReservationProposalDigest(AdmissionDigest),
303    ChannelReservationDigest(AdmissionDigest),
304    CreditExposureReservationDigest(AdmissionDigest),
305}
306
307#[derive(Debug, Clone, Copy, PartialEq, Eq)]
308pub(crate) enum AdmissionAttachmentKind {
309    ThresholdProposal,
310    ThresholdProposalBody,
311    SupplementalAuthorization,
312    BrokerAttempt,
313    BudgetHold,
314    ApprovalSet,
315    ExecutionNonce,
316    OutcomeEligibility,
317    PaymentParticipant,
318    ToolOutcome,
319    ChannelReservationProposal,
320    ChannelReservation,
321    CreditExposureReservation,
322}
323
324impl AdmissionAttachment {
325    fn kind(&self) -> AdmissionAttachmentKind {
326        match self {
327            Self::ThresholdProposalHash(_) => AdmissionAttachmentKind::ThresholdProposal,
328            Self::ThresholdProposal(_) => AdmissionAttachmentKind::ThresholdProposalBody,
329            Self::SupplementalAuthorizationDigest(_) => {
330                AdmissionAttachmentKind::SupplementalAuthorization
331            }
332            Self::BrokerAttempt(_) => AdmissionAttachmentKind::BrokerAttempt,
333            Self::BudgetHoldId(_) => AdmissionAttachmentKind::BudgetHold,
334            Self::ApprovalSetHash(_) => AdmissionAttachmentKind::ApprovalSet,
335            Self::ExecutionNonceId(_) => AdmissionAttachmentKind::ExecutionNonce,
336            Self::OutcomeEligibilityDigest(_) => AdmissionAttachmentKind::OutcomeEligibility,
337            Self::PaymentParticipantId(_) => AdmissionAttachmentKind::PaymentParticipant,
338            Self::ToolOutcomeId(_) => AdmissionAttachmentKind::ToolOutcome,
339            Self::ChannelReservationProposalDigest(_) => {
340                AdmissionAttachmentKind::ChannelReservationProposal
341            }
342            Self::ChannelReservationDigest(_) => AdmissionAttachmentKind::ChannelReservation,
343            Self::CreditExposureReservationDigest(_) => {
344                AdmissionAttachmentKind::CreditExposureReservation
345            }
346        }
347    }
348
349    fn slot(&self) -> u8 {
350        self.kind().slot()
351    }
352
353    fn field_name(&self) -> &'static str {
354        match self {
355            Self::ThresholdProposalHash(_) => "threshold_proposal_hash",
356            Self::ThresholdProposal(_) => "threshold_proposal",
357            Self::SupplementalAuthorizationDigest(_) => "supplemental_authorization_digest",
358            Self::BrokerAttempt(_) => "broker_attempt",
359            Self::BudgetHoldId(_) => "budget_hold_id",
360            Self::ApprovalSetHash(_) => "approval_set_hash",
361            Self::ExecutionNonceId(_) => "execution_nonce_id",
362            Self::OutcomeEligibilityDigest(_) => "outcome_eligibility_digest",
363            Self::PaymentParticipantId(_) => "payment_participant_id",
364            Self::ToolOutcomeId(_) => "tool_outcome_id",
365            Self::ChannelReservationProposalDigest(_) => "channel_reservation_proposal_digest",
366            Self::ChannelReservationDigest(_) => "channel_reservation_digest",
367            Self::CreditExposureReservationDigest(_) => "credit_exposure_reservation_digest",
368        }
369    }
370}
371
372impl AdmissionAttachmentKind {
373    fn slot(self) -> u8 {
374        match self {
375            Self::ThresholdProposal => 0,
376            Self::SupplementalAuthorization => 1,
377            Self::BrokerAttempt => 2,
378            Self::BudgetHold => 3,
379            Self::ApprovalSet => 4,
380            Self::ExecutionNonce => 5,
381            Self::OutcomeEligibility => 6,
382            Self::PaymentParticipant => 7,
383            Self::ToolOutcome => 8,
384            Self::ChannelReservationProposal => 9,
385            Self::ChannelReservation => 10,
386            Self::CreditExposureReservation => 11,
387            Self::ThresholdProposalBody => 12,
388        }
389    }
390}
391
392#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
393#[serde(transparent)]
394pub struct AdmissionOperationAttachmentsV1(Vec<AdmissionAttachment>);
395
396impl AdmissionOperationAttachmentsV1 {
397    fn has_slot(&self, slot: u8) -> bool {
398        self.0.iter().any(|attachment| attachment.slot() == slot)
399    }
400
401    fn existing_matches(&self, attachment: &AdmissionAttachment) -> Option<bool> {
402        self.0
403            .iter()
404            .find(|existing| existing.slot() == attachment.slot())
405            .map(|existing| existing == attachment)
406    }
407
408    fn tool_outcome_id(&self) -> Option<&AdmissionDigest> {
409        self.0.iter().find_map(|attachment| match attachment {
410            AdmissionAttachment::ToolOutcomeId(id) => Some(id),
411            _ => None,
412        })
413    }
414
415    fn attach(&mut self, attachment: AdmissionAttachment) {
416        let index = self
417            .0
418            .partition_point(|existing| existing.slot() < attachment.slot());
419        self.0.insert(index, attachment);
420    }
421
422    fn validate(&self) -> Result<(), AdmissionOperationError> {
423        if self.0.len() > 13
424            || self
425                .0
426                .windows(2)
427                .any(|pair| pair[0].slot() >= pair[1].slot())
428        {
429            return Err(AdmissionOperationError::DuplicateAttachment {
430                field: "persisted_attachment_set",
431            });
432        }
433        Ok(())
434    }
435}
436
437impl<'de> Deserialize<'de> for AdmissionOperationAttachmentsV1 {
438    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
439    where
440        D: Deserializer<'de>,
441    {
442        let attachments = Self(Vec::deserialize(deserializer)?);
443        attachments.validate().map_err(serde::de::Error::custom)?;
444        Ok(attachments)
445    }
446}
447
448#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
449#[serde(rename_all = "snake_case")]
450pub enum AdmissionTerminalReplay {
451    Receipt {
452        receipt_id: AdmissionIdentifier,
453        projection_digest: AdmissionDigest,
454    },
455    Incident {
456        incident_id: AdmissionIdentifier,
457        projection_digest: AdmissionDigest,
458    },
459    EconomicMutation {
460        result_id: AdmissionIdentifier,
461        result_digest: AdmissionDigest,
462        projection_digest: AdmissionDigest,
463    },
464}
465
466impl AdmissionTerminalReplay {
467    #[must_use]
468    pub const fn projection_digest(&self) -> &AdmissionDigest {
469        match self {
470            Self::Receipt {
471                projection_digest, ..
472            }
473            | Self::Incident {
474                projection_digest, ..
475            }
476            | Self::EconomicMutation {
477                projection_digest, ..
478            } => projection_digest,
479        }
480    }
481}
482
483#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
484pub enum AdmissionOperationSchema {
485    #[serde(rename = "chio.admission-operation.v1")]
486    V1,
487}
488
489#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
490#[serde(deny_unknown_fields)]
491pub struct PersistedAdmissionOperationV1 {
492    pub schema: AdmissionOperationSchema,
493    pub binding: PersistedAdmissionOperationBindingV1,
494    pub attachments: AdmissionOperationAttachmentsV1,
495    pub state: AdmissionOperationState,
496    pub dispatch_state: AdmissionDispatchState,
497    pub dispatch_commit: Option<AdmissionDispatchCommitBindingV1>,
498    pub coordinator_lease_epoch: u64,
499    pub version: u64,
500    pub last_error: Option<AdmissionErrorDetail>,
501    pub terminal_replay: Option<AdmissionTerminalReplay>,
502}
503
504#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
505pub struct AdmissionOperationV1 {
506    binding: AdmissionOperationBindingV1,
507    attachments: AdmissionOperationAttachmentsV1,
508    state: AdmissionOperationState,
509    dispatch_state: AdmissionDispatchState,
510    dispatch_commit: Option<AdmissionDispatchCommitBindingV1>,
511    coordinator_lease_epoch: u64,
512    version: u64,
513    last_error: Option<AdmissionErrorDetail>,
514    terminal_replay: Option<AdmissionTerminalReplay>,
515}
516
517impl AdmissionOperationV1 {
518    pub fn prepare(
519        binding: AdmissionOperationBindingV1,
520        coordinator_lease_epoch: u64,
521    ) -> Result<Self, AdmissionOperationError> {
522        validate_positive_ijson("coordinator_lease_epoch", coordinator_lease_epoch)?;
523        binding.validate()?;
524        let dispatch_state = dispatch_state_for(binding.kind, AdmissionOperationState::Prepared)?;
525        Ok(Self {
526            binding,
527            attachments: AdmissionOperationAttachmentsV1::default(),
528            state: AdmissionOperationState::Prepared,
529            dispatch_state,
530            dispatch_commit: None,
531            coordinator_lease_epoch,
532            version: 1,
533            last_error: None,
534            terminal_replay: None,
535        })
536    }
537
538    pub fn with_initial_channel_reservation_proposal_digest(
539        mut self,
540        proposal_digest: AdmissionDigest,
541    ) -> Result<Self, AdmissionOperationError> {
542        self.validate()?;
543        if self.state != AdmissionOperationState::Prepared
544            || self.version != 1
545            || !self.binding.participant_requirements().channel
546            || !self.attachments.0.is_empty()
547        {
548            return Err(AdmissionOperationError::ForbiddenAttachment {
549                field: "channel_reservation_proposal_digest",
550            });
551        }
552        self.attachments
553            .0
554            .push(AdmissionAttachment::ChannelReservationProposalDigest(
555                proposal_digest,
556            ));
557        self.validate()?;
558        Ok(self)
559    }
560
561    /// Decode and validate a persisted operation value.
562    ///
563    /// This checks the operation's internal invariants but does not authenticate
564    /// its storage provenance. Authority-sensitive callers must obtain the value
565    /// from a fenced [`AdmissionOperationStore`] operation.
566    pub fn from_persisted(
567        persisted: PersistedAdmissionOperationV1,
568    ) -> Result<Self, AdmissionOperationError> {
569        match persisted.schema {
570            AdmissionOperationSchema::V1 => {}
571        }
572        let operation = Self {
573            binding: AdmissionOperationBindingV1::from_persisted(persisted.binding)?,
574            attachments: persisted.attachments,
575            state: persisted.state,
576            dispatch_state: persisted.dispatch_state,
577            dispatch_commit: persisted.dispatch_commit,
578            coordinator_lease_epoch: persisted.coordinator_lease_epoch,
579            version: persisted.version,
580            last_error: persisted.last_error,
581            terminal_replay: persisted.terminal_replay,
582        };
583        operation.validate()?;
584        Ok(operation)
585    }
586
587    pub fn validate(&self) -> Result<(), AdmissionOperationError> {
588        validate_positive_ijson("operation_version", self.version)?;
589        validate_positive_ijson("coordinator_lease_epoch", self.coordinator_lease_epoch)?;
590        self.binding.validate()?;
591        self.attachments.validate()?;
592        if let Some(attempt) = self.provider_attempt() {
593            attempt
594                .validate()
595                .map_err(|_| AdmissionOperationError::ProviderAttemptBindingMismatch)?;
596            if attempt.operation_id != self.binding.operation_id.as_str() {
597                return Err(AdmissionOperationError::ProviderAttemptBindingMismatch);
598            }
599        }
600        validate_state_requirements(
601            self.binding.kind,
602            self.binding.participant_requirements(),
603            self.state,
604        )?;
605        validate_state_attachments(
606            self.binding.kind,
607            self.binding.participant_requirements(),
608            self.state,
609            &self.attachments,
610        )?;
611        let expected_dispatch = dispatch_state_for(self.binding.kind, self.state)?;
612        if expected_dispatch != self.dispatch_state {
613            return Err(AdmissionOperationError::DispatchStateMismatch);
614        }
615        validate_dispatch_commit(self)?;
616        validate_terminal_replay(self.binding.kind, self.state, self.terminal_replay.as_ref())
617    }
618
619    #[must_use]
620    pub fn binding(&self) -> &AdmissionOperationBindingV1 {
621        &self.binding
622    }
623
624    #[must_use]
625    pub fn state(&self) -> AdmissionOperationState {
626        self.state
627    }
628
629    #[must_use]
630    pub fn dispatch_state(&self) -> AdmissionDispatchState {
631        self.dispatch_state
632    }
633
634    #[must_use]
635    pub fn dispatch_commit(&self) -> Option<&AdmissionDispatchCommitBindingV1> {
636        self.dispatch_commit.as_ref()
637    }
638
639    #[must_use]
640    pub fn coordinator_lease_epoch(&self) -> u64 {
641        self.coordinator_lease_epoch
642    }
643
644    #[must_use]
645    pub fn version(&self) -> u64 {
646        self.version
647    }
648
649    #[must_use]
650    pub fn terminal_replay(&self) -> Option<&AdmissionTerminalReplay> {
651        self.terminal_replay.as_ref()
652    }
653
654    #[must_use]
655    pub fn tool_outcome_id(&self) -> Option<&AdmissionDigest> {
656        self.attachments.tool_outcome_id()
657    }
658
659    #[must_use]
660    pub fn supplemental_authorization_digest(&self) -> Option<&AdmissionDigest> {
661        match self.attachment(AdmissionAttachmentKind::SupplementalAuthorization) {
662            Some(AdmissionAttachment::SupplementalAuthorizationDigest(digest)) => Some(digest),
663            _ => None,
664        }
665    }
666
667    #[must_use]
668    pub fn threshold_proposal_hash(&self) -> Option<&AdmissionDigest> {
669        match self.attachment(AdmissionAttachmentKind::ThresholdProposal) {
670            Some(AdmissionAttachment::ThresholdProposalHash(digest)) => Some(digest),
671            _ => None,
672        }
673    }
674
675    #[must_use]
676    pub fn threshold_proposal(
677        &self,
678    ) -> Option<&chio_core::capability::governance::ThresholdApprovalProposal> {
679        match self.attachment(AdmissionAttachmentKind::ThresholdProposalBody) {
680            Some(AdmissionAttachment::ThresholdProposal(proposal)) => Some(proposal.as_ref()),
681            _ => None,
682        }
683    }
684
685    #[must_use]
686    pub fn approval_set_hash(&self) -> Option<&AdmissionDigest> {
687        match self.attachment(AdmissionAttachmentKind::ApprovalSet) {
688            Some(AdmissionAttachment::ApprovalSetHash(digest)) => Some(digest),
689            _ => None,
690        }
691    }
692
693    #[allow(dead_code)]
694    pub(crate) fn has_attachment(&self, kind: AdmissionAttachmentKind) -> bool {
695        self.attachments.has_slot(kind.slot())
696    }
697
698    pub(crate) fn attachment(&self, kind: AdmissionAttachmentKind) -> Option<&AdmissionAttachment> {
699        self.attachments
700            .0
701            .iter()
702            .find(|attachment| attachment.kind() == kind)
703    }
704
705    pub(crate) fn provider_attempt(&self) -> Option<&ProviderAttemptBindingV1> {
706        match self.attachment(AdmissionAttachmentKind::BrokerAttempt) {
707            Some(AdmissionAttachment::BrokerAttempt(attempt)) => Some(attempt),
708            _ => None,
709        }
710    }
711
712    #[must_use]
713    pub fn budget_hold_id(&self) -> Option<&AdmissionIdentifier> {
714        match self.attachment(AdmissionAttachmentKind::BudgetHold) {
715            Some(AdmissionAttachment::BudgetHoldId(hold_id)) => Some(hold_id),
716            _ => None,
717        }
718    }
719
720    #[must_use]
721    pub fn channel_reservation_proposal_digest(&self) -> Option<&AdmissionDigest> {
722        match self.attachment(AdmissionAttachmentKind::ChannelReservationProposal) {
723            Some(AdmissionAttachment::ChannelReservationProposalDigest(digest)) => Some(digest),
724            _ => None,
725        }
726    }
727
728    #[must_use]
729    pub fn channel_reservation_digest(&self) -> Option<&AdmissionDigest> {
730        match self.attachment(AdmissionAttachmentKind::ChannelReservation) {
731            Some(AdmissionAttachment::ChannelReservationDigest(digest)) => Some(digest),
732            _ => None,
733        }
734    }
735
736    #[must_use]
737    pub fn credit_exposure_reservation_digest(&self) -> Option<&AdmissionDigest> {
738        match self.attachment(AdmissionAttachmentKind::CreditExposureReservation) {
739            Some(AdmissionAttachment::CreditExposureReservationDigest(digest)) => Some(digest),
740            _ => None,
741        }
742    }
743
744    #[must_use]
745    pub fn replay_key(&self) -> AdmissionReplayKey {
746        self.binding.replay_key()
747    }
748
749    #[must_use]
750    pub fn to_persisted(&self) -> PersistedAdmissionOperationV1 {
751        self.into()
752    }
753
754    #[must_use]
755    pub fn classify_replay(&self, candidate: &Self) -> AdmissionReplayClassification {
756        if self.replay_key() == candidate.replay_key()
757            && self.binding.operation_id == candidate.binding.operation_id
758            && self.binding == candidate.binding
759        {
760            AdmissionReplayClassification::Exact {
761                terminal_replay: self.terminal_replay.clone(),
762            }
763        } else {
764            AdmissionReplayClassification::Conflict
765        }
766    }
767
768    pub fn apply_command(
769        &self,
770        command: &AdmissionOperationCommand,
771        trusted_now_unix_ms: u64,
772    ) -> Result<AdmissionCommandResult, AdmissionOperationError> {
773        command
774            .recovery_lease
775            .validate_for(self, command, trusted_now_unix_ms)?;
776        let mut updated = self.clone();
777        let mut changed = false;
778        for attachment in &command.attachments {
779            match self.attachments.existing_matches(attachment) {
780                Some(true) => continue,
781                Some(false) => {
782                    return Err(AdmissionOperationError::AttachmentConflict {
783                        field: attachment.field_name(),
784                    });
785                }
786                None if attachment_allowed(
787                    self.binding.kind,
788                    self.binding.participant_requirements(),
789                    self.state,
790                    attachment,
791                ) =>
792                {
793                    updated.attachments.attach(attachment.clone());
794                    changed = true;
795                }
796                None => {
797                    return Err(
798                        if !attachment_supported(
799                            self.binding.kind,
800                            self.binding.participant_requirements(),
801                            attachment,
802                        ) {
803                            AdmissionOperationError::ForbiddenAttachment {
804                                field: attachment.field_name(),
805                            }
806                        } else {
807                            AdmissionOperationError::AttachmentPhase {
808                                field: attachment.field_name(),
809                                state: self.state,
810                            }
811                        },
812                    );
813                }
814            }
815        }
816        if let Some(next_state) = command.next_state {
817            if next_state.is_terminal() {
818                return Err(AdmissionOperationError::TerminalProjectionRequired);
819            }
820            if self.state == next_state {
821                if self.terminal_replay != command.terminal_replay
822                    || self.last_error != command.last_error
823                {
824                    return Err(AdmissionOperationError::TerminalReplayMismatch);
825                }
826            } else {
827                if !is_legal_transition(
828                    self.binding.kind,
829                    self.binding.participant_requirements(),
830                    self.state,
831                    next_state,
832                ) {
833                    return Err(AdmissionOperationError::IllegalTransition {
834                        from: self.state,
835                        to: next_state,
836                    });
837                }
838                validate_terminal_replay(
839                    self.binding.kind,
840                    next_state,
841                    command.terminal_replay.as_ref(),
842                )?;
843                updated.state = next_state;
844                updated.dispatch_state = dispatch_state_for(self.binding.kind, next_state)?;
845                if next_state == AdmissionOperationState::DispatchCommitted {
846                    let provider_attempt =
847                        if self.binding.kind == AdmissionOperationKind::ToolDispatch {
848                            Some(updated.provider_attempt().cloned().ok_or(
849                                AdmissionOperationError::MissingParticipantAttachment {
850                                    field: "broker_attempt",
851                                },
852                            )?)
853                        } else {
854                            None
855                        };
856                    updated.dispatch_commit = Some(AdmissionDispatchCommitBindingV1 {
857                        committed_version: next_version(self.version)?,
858                        coordinator_lease_id: command.recovery_lease.coordinator_lease_id().clone(),
859                        coordinator_lease_epoch: command.recovery_lease.coordinator_lease_epoch(),
860                        store_fence: command.recovery_lease.store_fence().clone(),
861                        provider_attempt,
862                    });
863                }
864                updated.terminal_replay.clone_from(&command.terminal_replay);
865                updated.last_error.clone_from(&command.last_error);
866                validate_state_attachments(
867                    self.binding.kind,
868                    self.binding.participant_requirements(),
869                    next_state,
870                    &updated.attachments,
871                )?;
872                changed = true;
873            }
874        }
875        if !changed {
876            return Ok(AdmissionCommandResult::Idempotent(updated));
877        }
878        self.require_version(command.expected_version)?;
879        updated.version = next_version(self.version)?;
880        updated.validate()?;
881        Ok(AdmissionCommandResult::Applied(updated))
882    }
883
884    fn require_version(&self, expected_version: u64) -> Result<(), AdmissionOperationError> {
885        if self.version != expected_version {
886            return Err(AdmissionOperationError::StaleVersion {
887                expected: expected_version,
888                actual: self.version,
889            });
890        }
891        Ok(())
892    }
893}
894
895impl From<&AdmissionOperationV1> for PersistedAdmissionOperationV1 {
896    fn from(operation: &AdmissionOperationV1) -> Self {
897        Self {
898            schema: AdmissionOperationSchema::V1,
899            binding: PersistedAdmissionOperationBindingV1::from(&operation.binding),
900            attachments: operation.attachments.clone(),
901            state: operation.state,
902            dispatch_state: operation.dispatch_state,
903            dispatch_commit: operation.dispatch_commit.clone(),
904            coordinator_lease_epoch: operation.coordinator_lease_epoch,
905            version: operation.version,
906            last_error: operation.last_error.clone(),
907            terminal_replay: operation.terminal_replay.clone(),
908        }
909    }
910}
911
912#[derive(Debug, Clone, PartialEq, Eq)]
913pub enum AdmissionReplayClassification {
914    Exact {
915        terminal_replay: Option<AdmissionTerminalReplay>,
916    },
917    Conflict,
918}
919
920#[derive(Debug, Clone, PartialEq, Eq)]
921pub struct AdmissionOperationCommand {
922    operation_id: AdmissionOperationId,
923    expected_version: u64,
924    recovery_lease: AdmissionRecoveryLease,
925    attachments: Vec<AdmissionAttachment>,
926    next_state: Option<AdmissionOperationState>,
927    terminal_replay: Option<AdmissionTerminalReplay>,
928    last_error: Option<AdmissionErrorDetail>,
929}
930
931impl AdmissionOperationCommand {
932    #[allow(clippy::too_many_arguments)]
933    pub fn new(
934        operation_id: AdmissionOperationId,
935        expected_version: u64,
936        recovery_lease: AdmissionRecoveryLease,
937        attachments: Vec<AdmissionAttachment>,
938        next_state: Option<AdmissionOperationState>,
939        terminal_replay: Option<AdmissionTerminalReplay>,
940        last_error: Option<AdmissionErrorDetail>,
941    ) -> Result<Self, AdmissionOperationError> {
942        validate_positive_ijson("expected_operation_version", expected_version)?;
943        if attachments.is_empty() && next_state.is_none() {
944            return Err(AdmissionOperationError::EmptyCommand);
945        }
946        if next_state.is_none() && (terminal_replay.is_some() || last_error.is_some()) {
947            return Err(AdmissionOperationError::TerminalReplayMismatch);
948        }
949        if next_state.is_some_and(AdmissionOperationState::is_terminal) {
950            return Err(AdmissionOperationError::TerminalProjectionRequired);
951        }
952        for (index, attachment) in attachments.iter().enumerate() {
953            if attachments[..index]
954                .iter()
955                .any(|prior| prior.field_name() == attachment.field_name())
956            {
957                return Err(AdmissionOperationError::DuplicateAttachment {
958                    field: attachment.field_name(),
959                });
960            }
961        }
962        Ok(Self {
963            operation_id,
964            expected_version,
965            recovery_lease,
966            attachments,
967            next_state,
968            terminal_replay,
969            last_error,
970        })
971    }
972
973    #[must_use]
974    pub fn operation_id(&self) -> &AdmissionOperationId {
975        &self.operation_id
976    }
977
978    #[must_use]
979    pub fn expected_version(&self) -> u64 {
980        self.expected_version
981    }
982
983    #[must_use]
984    pub fn recovery_lease(&self) -> &AdmissionRecoveryLease {
985        &self.recovery_lease
986    }
987
988    #[must_use]
989    pub fn attachments(&self) -> &[AdmissionAttachment] {
990        &self.attachments
991    }
992
993    #[must_use]
994    pub fn next_state(&self) -> Option<AdmissionOperationState> {
995        self.next_state
996    }
997
998    #[must_use]
999    pub fn terminal_replay(&self) -> Option<&AdmissionTerminalReplay> {
1000        self.terminal_replay.as_ref()
1001    }
1002
1003    #[must_use]
1004    pub fn last_error(&self) -> Option<&AdmissionErrorDetail> {
1005        self.last_error.as_ref()
1006    }
1007}
1008
1009#[derive(Debug, Clone, PartialEq, Eq)]
1010pub enum AdmissionCommandResult {
1011    Applied(AdmissionOperationV1),
1012    Idempotent(AdmissionOperationV1),
1013}
1014
1015impl AdmissionCommandResult {
1016    #[must_use]
1017    pub fn into_operation(self) -> AdmissionOperationV1 {
1018        match self {
1019            Self::Applied(operation) | Self::Idempotent(operation) => operation,
1020        }
1021    }
1022}
1023
1024/// Structurally checked claim returned by an admission-operation store.
1025///
1026/// This value is not authority and cannot be used to build a command until a
1027/// qualified store rechecks it against durable operation and lease state.
1028#[derive(Debug, Clone, PartialEq, Eq)]
1029pub struct UntrustedAdmissionRecoveryClaim {
1030    operation_id: AdmissionOperationId,
1031    claimant_id: AdmissionIdentifier,
1032    coordinator_lease_id: AdmissionIdentifier,
1033    coordinator_lease_epoch: u64,
1034    claimed_version: u64,
1035    expires_at_unix_ms: u64,
1036    store_fence: StoreMutationFence,
1037}
1038
1039impl UntrustedAdmissionRecoveryClaim {
1040    pub fn new(
1041        operation_id: AdmissionOperationId,
1042        claimant_id: AdmissionIdentifier,
1043        coordinator_lease_id: AdmissionIdentifier,
1044        coordinator_lease_epoch: u64,
1045        claimed_version: u64,
1046        expires_at_unix_ms: u64,
1047        store_fence: StoreMutationFence,
1048    ) -> Result<Self, AdmissionOperationError> {
1049        validate_positive_ijson("coordinator_lease_epoch", coordinator_lease_epoch)?;
1050        validate_positive_ijson("claimed_version", claimed_version)?;
1051        validate_positive_ijson("expires_at_unix_ms", expires_at_unix_ms)?;
1052        validate_store_fence(&store_fence)?;
1053        Ok(Self {
1054            operation_id,
1055            claimant_id,
1056            coordinator_lease_id,
1057            coordinator_lease_epoch,
1058            claimed_version,
1059            expires_at_unix_ms,
1060            store_fence,
1061        })
1062    }
1063
1064    fn validate_for_qualification(
1065        &self,
1066        operation: &AdmissionOperationV1,
1067        expected_version: u64,
1068        claimant_id: &AdmissionIdentifier,
1069        trusted_now_unix_ms: u64,
1070        expires_at_unix_ms: u64,
1071        current_store_fence: &StoreMutationFence,
1072    ) -> Result<(), AdmissionOperationError> {
1073        operation.validate()?;
1074        validate_store_fence(&self.store_fence)?;
1075        validate_store_fence(current_store_fence)?;
1076        if self.operation_id != operation.binding.operation_id {
1077            return Err(AdmissionOperationError::WrongOperation);
1078        }
1079        if self.claimed_version != expected_version {
1080            return Err(AdmissionOperationError::LeaseVersionMismatch);
1081        }
1082        if operation.version != expected_version {
1083            return Err(AdmissionOperationError::StaleVersion {
1084                expected: expected_version,
1085                actual: operation.version,
1086            });
1087        }
1088        if self.claimant_id != *claimant_id
1089            || self.expires_at_unix_ms > expires_at_unix_ms
1090            || self.store_fence != *current_store_fence
1091            || self.coordinator_lease_epoch != operation.coordinator_lease_epoch
1092        {
1093            return Err(AdmissionOperationError::CoordinatorFenced);
1094        }
1095        if trusted_now_unix_ms >= self.expires_at_unix_ms {
1096            return Err(AdmissionOperationError::LeaseExpired);
1097        }
1098        Ok(())
1099    }
1100
1101    #[must_use]
1102    pub fn operation_id(&self) -> &AdmissionOperationId {
1103        &self.operation_id
1104    }
1105
1106    #[must_use]
1107    pub fn claimant_id(&self) -> &AdmissionIdentifier {
1108        &self.claimant_id
1109    }
1110
1111    #[must_use]
1112    pub fn coordinator_lease_id(&self) -> &AdmissionIdentifier {
1113        &self.coordinator_lease_id
1114    }
1115
1116    #[must_use]
1117    pub fn coordinator_lease_epoch(&self) -> u64 {
1118        self.coordinator_lease_epoch
1119    }
1120
1121    #[must_use]
1122    pub fn claimed_version(&self) -> u64 {
1123        self.claimed_version
1124    }
1125
1126    #[must_use]
1127    pub fn expires_at_unix_ms(&self) -> u64 {
1128        self.expires_at_unix_ms
1129    }
1130
1131    #[must_use]
1132    pub fn store_fence(&self) -> &StoreMutationFence {
1133        &self.store_fence
1134    }
1135}
1136
1137/// Store-qualified recovery authority required by an operation command.
1138#[derive(Debug, Clone, PartialEq, Eq)]
1139pub struct AdmissionRecoveryLease(UntrustedAdmissionRecoveryClaim);
1140
1141impl AdmissionRecoveryLease {
1142    fn from_qualified(claim: UntrustedAdmissionRecoveryClaim) -> Self {
1143        Self(claim)
1144    }
1145
1146    #[must_use]
1147    pub fn operation_id(&self) -> &AdmissionOperationId {
1148        self.0.operation_id()
1149    }
1150
1151    #[must_use]
1152    pub fn claimant_id(&self) -> &AdmissionIdentifier {
1153        self.0.claimant_id()
1154    }
1155
1156    #[must_use]
1157    pub fn coordinator_lease_id(&self) -> &AdmissionIdentifier {
1158        self.0.coordinator_lease_id()
1159    }
1160
1161    #[must_use]
1162    pub fn coordinator_lease_epoch(&self) -> u64 {
1163        self.0.coordinator_lease_epoch()
1164    }
1165
1166    #[must_use]
1167    pub fn claimed_version(&self) -> u64 {
1168        self.0.claimed_version()
1169    }
1170
1171    #[must_use]
1172    pub fn expires_at_unix_ms(&self) -> u64 {
1173        self.0.expires_at_unix_ms()
1174    }
1175
1176    #[must_use]
1177    pub fn store_fence(&self) -> &StoreMutationFence {
1178        self.0.store_fence()
1179    }
1180
1181    #[must_use]
1182    pub fn untrusted_claim(&self) -> &UntrustedAdmissionRecoveryClaim {
1183        &self.0
1184    }
1185
1186    fn validate_for(
1187        &self,
1188        operation: &AdmissionOperationV1,
1189        command: &AdmissionOperationCommand,
1190        trusted_now_unix_ms: u64,
1191    ) -> Result<(), AdmissionOperationError> {
1192        validate_store_fence(self.store_fence())?;
1193        if self.operation_id() != &operation.binding.operation_id
1194            || command.operation_id != operation.binding.operation_id
1195        {
1196            return Err(AdmissionOperationError::WrongOperation);
1197        }
1198        if self.claimed_version() != command.expected_version {
1199            return Err(AdmissionOperationError::LeaseVersionMismatch);
1200        }
1201        operation.require_version(command.expected_version)?;
1202        if self.coordinator_lease_epoch() != operation.coordinator_lease_epoch {
1203            return Err(AdmissionOperationError::CoordinatorFenced);
1204        }
1205        if trusted_now_unix_ms >= self.expires_at_unix_ms() {
1206            return Err(AdmissionOperationError::LeaseExpired);
1207        }
1208        Ok(())
1209    }
1210}
1211
1212#[cfg(test)]
1213pub(crate) fn qualify_recovery_claim_for_test(
1214    operation: &AdmissionOperationV1,
1215    claim: UntrustedAdmissionRecoveryClaim,
1216    trusted_now_unix_ms: u64,
1217    current_store_fence: &StoreMutationFence,
1218) -> Result<AdmissionRecoveryLease, AdmissionOperationError> {
1219    claim.validate_for_qualification(
1220        operation,
1221        claim.claimed_version,
1222        &claim.claimant_id,
1223        trusted_now_unix_ms,
1224        claim.expires_at_unix_ms,
1225        current_store_fence,
1226    )?;
1227    Ok(AdmissionRecoveryLease::from_qualified(claim))
1228}
1229
1230#[cfg(test)]
1231#[path = "admission_operation_tests.rs"]
1232mod tests;
chio_kernel/admission_operation.rs

chio_kernel/
admission_operation.rs
