liminal_server/server/participant/production/
marker_source.rs

1//! Bounded durable marker-source association for fenced Attached rows.
2
3use std::sync::Arc;
4
5use liminal::durability::DurableStore;
6use liminal_protocol::lifecycle::{
7    FencedMarkerSourceExpectation, LiveFrontierOwner, MarkerSequenceOwner,
8    RetainedFencedMarkerSource,
9};
10
11use super::log::{
12    DecodedStoredOperation, OperationLog, StoredMarkerDrain, StoredOperationV2, StoredOperationV3,
13};
14
15/// Typed reason an exact marker source cannot authorize later mint inputs.
16#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
17pub enum MarkerSourceRefusalReason {
18    /// The point read or durable row decode failed.
19    #[error("durable marker source could not be read or decoded")]
20    DurableRead,
21    /// No durable row exists at the selected sequence.
22    #[error("marker source sequence does not exist")]
23    Missing,
24    /// The selected durable row is not a marker source.
25    #[error("marker source sequence does not select MarkerDrained")]
26    WrongOperation,
27    /// Canonical marker identity bytes disagree with restored frontier truth.
28    #[error("durable marker body differs from the frontier-validated record")]
29    MarkerBody,
30    /// Marker delivery sequence disagrees.
31    #[error("durable marker delivery sequence differs from the frontier-validated record")]
32    DeliverySequence,
33    /// Marker transaction order disagrees.
34    #[error("durable marker transaction order differs from the frontier-validated record")]
35    TransactionOrder,
36    /// Marker candidate phase disagrees.
37    #[error("durable marker candidate phase differs from the frontier-validated record")]
38    CandidatePhase,
39    /// Marker participant disagrees.
40    #[error("durable marker participant differs from the frontier-validated record")]
41    Participant,
42    /// Measured marker row charge disagrees.
43    #[error("durable marker retained charge differs from its measured canonical bytes")]
44    RetainedCharge,
45}
46
47/// Refused association retaining the move-only owner/recovery pair.
48#[derive(Debug)]
49pub struct MarkerSourceRefused {
50    retained: RetainedFencedMarkerSource,
51    marker_source_sequence: u64,
52    reason: MarkerSourceRefusalReason,
53}
54
55impl MarkerSourceRefused {
56    /// Returns the exact typed refusal.
57    #[must_use]
58    pub const fn reason(&self) -> MarkerSourceRefusalReason {
59        self.reason
60    }
61
62    /// Returns unchanged retained inputs and the selected sequence.
63    #[must_use]
64    pub fn into_parts(self) -> (RetainedFencedMarkerSource, u64) {
65        (self.retained, self.marker_source_sequence)
66    }
67}
68
69/// Source-associated inputs which still contain no one-use marker token.
70#[derive(Debug)]
71pub struct ValidatedFencedMarkerInputs {
72    retained: RetainedFencedMarkerSource,
73    marker_source_sequence: u64,
74}
75
76impl ValidatedFencedMarkerInputs {
77    /// Returns unchanged owner/recovery plus the validated source sequence.
78    #[must_use]
79    pub fn into_parts(
80        self,
81    ) -> (
82        LiveFrontierOwner,
83        liminal_protocol::lifecycle::DetachedCredentialRecovery,
84        u64,
85    ) {
86        let (owner, recovery) = self.retained.into_parts();
87        (owner, recovery, self.marker_source_sequence)
88    }
89}
90
91/// Point-reads and validates one durable marker source before the retained owner
92/// may proceed to its private one-use authority mint.
93///
94/// # Errors
95/// Returns a typed refusal containing the unchanged retained inputs when the row
96/// is absent, unreadable, has the wrong kind, or disagrees with frontier truth.
97pub async fn validate_marker_source(
98    store: Arc<dyn DurableStore>,
99    retained: RetainedFencedMarkerSource,
100    marker_source_sequence: u64,
101) -> Result<ValidatedFencedMarkerInputs, Box<MarkerSourceRefused>> {
102    let expectation = retained.expectation();
103    let log = OperationLog::new(store, expectation.conversation_id());
104    let result = read_and_validate(&log, marker_source_sequence, expectation).await;
105    match result {
106        Ok(()) => Ok(ValidatedFencedMarkerInputs {
107            retained,
108            marker_source_sequence,
109        }),
110        Err(reason) => Err(Box::new(MarkerSourceRefused {
111            retained,
112            marker_source_sequence,
113            reason,
114        })),
115    }
116}
117
118async fn read_and_validate(
119    log: &OperationLog,
120    marker_source_sequence: u64,
121    expectation: FencedMarkerSourceExpectation,
122) -> Result<(), MarkerSourceRefusalReason> {
123    let operation = log
124        .read_at(marker_source_sequence)
125        .await
126        .map_err(|_| MarkerSourceRefusalReason::DurableRead)?
127        .ok_or(MarkerSourceRefusalReason::Missing)?
128        .operation;
129    let row = match operation {
130        DecodedStoredOperation::V2(StoredOperationV2::MarkerDrained { row })
131        | DecodedStoredOperation::V3(StoredOperationV3::MarkerDrained { row }) => row,
132        DecodedStoredOperation::V2(_) | DecodedStoredOperation::V3(_) => {
133            return Err(MarkerSourceRefusalReason::WrongOperation);
134        }
135    };
136    validate_row(&row, expectation)
137}
138
139fn validate_row(
140    row: &StoredMarkerDrain,
141    expectation: FencedMarkerSourceExpectation,
142) -> Result<(), MarkerSourceRefusalReason> {
143    let canonical_marker = canonical_marker_bytes(expectation);
144    if row.marker != canonical_marker {
145        return Err(MarkerSourceRefusalReason::MarkerBody);
146    }
147    let charge = row.retained_charge;
148    let order = expectation.admission_order();
149    if charge.delivery_seq != expectation.marker_delivery_seq() {
150        return Err(MarkerSourceRefusalReason::DeliverySequence);
151    }
152    if charge.transaction_order != order.transaction_order() {
153        return Err(MarkerSourceRefusalReason::TransactionOrder);
154    }
155    if charge.candidate_phase != order.candidate_phase() as u8 {
156        return Err(MarkerSourceRefusalReason::CandidatePhase);
157    }
158    if charge.participant_id != expectation.participant_id() {
159        return Err(MarkerSourceRefusalReason::Participant);
160    }
161    let marker_bytes = u64::try_from(canonical_marker.len())
162        .map_err(|_| MarkerSourceRefusalReason::RetainedCharge)?;
163    if charge.charge.entries != 1 || charge.charge.bytes != marker_bytes {
164        return Err(MarkerSourceRefusalReason::RetainedCharge);
165    }
166    Ok(())
167}
168
169pub(super) fn canonical_marker_bytes(expectation: FencedMarkerSourceExpectation) -> Vec<u8> {
170    format!(
171        "MarkerCandidateAuthority {{ delivery_seq: {:?}, admission_order: {:?}, target_binding: {:?}, provenance: {:?}, current_owner: {:?} }}",
172        expectation.marker_delivery_seq(),
173        expectation.admission_order(),
174        expectation.target_binding(),
175        expectation.provenance(),
176        MarkerSequenceOwner::Marker,
177    )
178    .into_bytes()
179}
liminal_server/server/participant/production/marker_source.rs

liminal_server/server/participant/production/
marker_source.rs
