lace_sans_io/core/
carrier.rs

1//! Connected carrier ownership, nonce prefacing, and hard session replacement.
2
3use super::machine::{
4    ConnectedSessionConfig, core_machine_error, integration_machine_error, peer_protocol,
5};
6use super::*;
7use crate::iltp::{
8    IltpItem, PortText, SessionNonceText, SessionPreface, StreamEncoding, StreamOffers,
9};
10use crate::records::BySecret;
11
12const NONCE_DOMAIN: &[u8] = b"lace-iltp/1.0/session-nonce/v1\0";
13
14/// Zeroizing owner of one connected carrier's 32-byte nonce seed. Each clone
15/// owns and erases an independent copy.
16#[derive(Clone)]
17pub struct NonceSeed([u8; 32]);
18impl NonceSeed {
19    pub fn new(bytes: [u8; 32]) -> Self {
20        Self(bytes)
21    }
22    /// Draw a fresh unpredictable nonzero seed from the operating system.
23    pub fn generate() -> Result<Self, getrandom::Error> {
24        crate::records::random_aux_rand32().map(Self)
25    }
26    pub(super) fn expose(&self) -> &[u8; 32] {
27        &self.0
28    }
29}
30impl std::fmt::Debug for NonceSeed {
31    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
32        f.write_str("NonceSeed([REDACTED])")
33    }
34}
35impl Drop for NonceSeed {
36    fn drop(&mut self) {
37        crate::records::erasure::erase_secret_bytes(&mut self.0);
38    }
39}
40
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub enum PrefaceCause {
43    Initial,
44    Reset,
45}
46
47#[derive(Debug)]
48struct PrefacingState {
49    cause: PrefaceCause,
50    /// `None` until the host selects it for the peer preface's port.
51    local_operand: Option<SessionOperand>,
52    local_nonce: SessionNonceText,
53    local_preface_accepted: bool,
54    peer_nonce: Option<SessionNonceText>,
55    retained_output: Option<Vec<IltpItem>>,
56    discarded_control_bytes: u64,
57}
58
59#[derive(Debug)]
60pub enum CarrierState {
61    Prefacing,
62    Running,
63    Failed(MachineError),
64    Closed(CloseReason),
65}
66
67#[derive(Debug)]
68enum CarrierOwner {
69    Prefacing(PrefacingState),
70    Running {
71        session: Box<SessionMachine>,
72        staged_replacement: Option<crate::logic::InterlaceModule>,
73    },
74    Failed(MachineError),
75    Closed(CloseReason),
76}
77
78/// Where each logical session's operand comes from: the fixed operand's
79/// expected peer By, which survives replacement, or host selection.
80#[derive(Debug)]
81enum OperandSource {
82    Fixed {
83        expected_peer_by: Option<crate::records::ByValue>,
84    },
85    Selected,
86}
87
88/// One connected byte carrier above exactly one immutable logical session.
89#[derive(Debug)]
90pub struct ConnectedCarrier {
91    role: CarrierRole,
92    peer_proof: PeerProofAssurance,
93    operand_source: OperandSource,
94    port: Option<PortText>,
95    /// The port of the latest peer preface.
96    peer_port: Option<PortText>,
97    /// The By the live peer proved in the latest logical session.
98    peer_by: Option<crate::records::ByValue>,
99    /// The outgoing and incoming encodings the latest preface pair selects.
100    encodings: Option<[StreamEncoding; 2]>,
101    by_secret: Option<BySecret>,
102    local_by_value: Option<crate::records::ByValue>,
103    limits: InterlaceResourceLimits,
104    operand_cache: OperandCache,
105    stream_offers: StreamOffers,
106    nonce_seed: NonceSeed,
107    next_nonce_sequence: Option<u64>,
108    owner: CarrierOwner,
109}
110
111impl ConnectedCarrier {
112    pub fn new(role: CarrierRole, config: ConnectedConfig) -> Result<Self, CoreError> {
113        config.validate()?;
114        let ConnectedConfig {
115            operand,
116            by_secret,
117            peer_proof,
118            port,
119            stream_encodings,
120            nonce_seed,
121            limits,
122            operand_cache,
123        } = config;
124        let stream_offers =
125            StreamOffers::new(&stream_encodings).map_err(CoreError::InvalidConfig)?;
126        let local_by_value = by_secret.as_ref().map(BySecret::by_value);
127        let (operand_source, local_operand) = match operand {
128            ConnectedOperand::Fixed(operand) => (
129                OperandSource::Fixed {
130                    expected_peer_by: operand.expected_peer_by,
131                },
132                Some(operand),
133            ),
134            ConnectedOperand::Selected => (OperandSource::Selected, None),
135        };
136        let mut carrier = Self {
137            role,
138            peer_proof,
139            operand_source,
140            port,
141            peer_port: None,
142            peer_by: None,
143            encodings: None,
144            by_secret,
145            local_by_value,
146            limits,
147            operand_cache,
148            stream_offers,
149            nonce_seed,
150            next_nonce_sequence: Some(0),
151            owner: CarrierOwner::Closed(CloseReason::Cancelled),
152        };
153        let local_nonce = carrier.derive_nonce(None)?;
154        carrier.owner = CarrierOwner::Prefacing(PrefacingState::new(
155            PrefaceCause::Initial,
156            local_operand,
157            local_nonce,
158        ));
159        Ok(carrier)
160    }
161
162    pub(super) fn local_protocol_side(&self) -> InterlaceSide {
163        self.role.protocol_side()
164    }
165
166    /// The port the latest peer preface sent, for diagnostics.
167    pub fn peer_port(&self) -> Option<&PortText> {
168        self.peer_port.as_ref()
169    }
170
171    /// The By the live peer proved in the latest logical session, for
172    /// diagnostics: its `PeerConfidential` key. Absent until that proof
173    /// validates under confidential or channel-bound assurance, and from the
174    /// next preface cut until the following session's proof does.
175    pub fn peer_by(&self) -> Option<&crate::records::ByValue> {
176        self.peer_by.as_ref()
177    }
178
179    /// The 065 encoding of the local-to-peer direction that the latest
180    /// preface pair selects, for diagnostics. Absent before the first peer
181    /// preface; each peer preface selects again.
182    pub fn outgoing_encoding(&self) -> Option<StreamEncoding> {
183        self.encodings.map(|[outgoing, _]| outgoing)
184    }
185
186    /// The 065 encoding of the peer-to-local direction that the latest
187    /// preface pair selects, for diagnostics; absent like
188    /// [`Self::outgoing_encoding`].
189    pub fn incoming_encoding(&self) -> Option<StreamEncoding> {
190        self.encodings.map(|[_, incoming]| incoming)
191    }
192
193    pub fn state(&self) -> CarrierState {
194        match &self.owner {
195            CarrierOwner::Prefacing(_) => CarrierState::Prefacing,
196            CarrierOwner::Running { .. } => CarrierState::Running,
197            CarrierOwner::Failed(error) => CarrierState::Failed(error.clone()),
198            CarrierOwner::Closed(reason) => CarrierState::Closed(reason.clone()),
199        }
200    }
201
202    pub fn session(&self) -> Option<&SessionMachine> {
203        match &self.owner {
204            CarrierOwner::Running { session, .. } => Some(session),
205            _ => None,
206        }
207    }
208    pub fn protocol_phase(&self) -> MachineProtocolPhase {
209        match &self.owner {
210            CarrierOwner::Prefacing(_) => MachineProtocolPhase::Prefacing,
211            CarrierOwner::Running { session, .. } => session.protocol_phase(),
212            CarrierOwner::Failed(_) => MachineProtocolPhase::Failed,
213            CarrierOwner::Closed(_) => MachineProtocolPhase::Closed,
214        }
215    }
216    pub fn protocol_phase_token(&self) -> ProtocolPhaseToken {
217        ProtocolPhaseToken::new(
218            self.session().map(SessionMachine::id),
219            self.protocol_phase(),
220        )
221    }
222    pub fn active_store_call(&self) -> Option<&StoreCall> {
223        self.session().and_then(SessionMachine::active_store_call)
224    }
225    pub fn active_store_call_mut(&mut self) -> Option<&mut StoreCall> {
226        match &mut self.owner {
227            CarrierOwner::Running { session, .. } => session.active_store_call_mut(),
228            _ => None,
229        }
230    }
231    /// Whether the host may decode another peer item without violating the
232    /// carrier or endpoint's current ownership boundary.
233    pub fn can_read_peer_input(&self) -> bool {
234        if self.active_store_call().is_some() {
235            return false;
236        }
237        match &self.owner {
238            CarrierOwner::Prefacing(prefacing) => prefacing.peer_nonce.is_none(),
239            CarrierOwner::Running { session, .. } => session.has_free_peer_slot(),
240            CarrierOwner::Failed(_) | CarrierOwner::Closed(_) => false,
241        }
242    }
243
244    /// How the host receives the next peer record: establishment records
245    /// into memory within their bounds, an outstanding response through the
246    /// store, an old session's record during reset prefacing discarded within
247    /// the reset control budget, and no other record.
248    pub fn peer_record_ingress(&self) -> RecordIngress {
249        match &self.owner {
250            CarrierOwner::Running { session, .. } => session.record_ingress(),
251            CarrierOwner::Prefacing(prefacing)
252                if prefacing.cause == PrefaceCause::Reset && prefacing.peer_nonce.is_none() =>
253            {
254                let remaining = self
255                    .limits
256                    .max_received_control_bytes
257                    .saturating_sub(prefacing.discarded_control_bytes);
258                RecordIngress::Discard {
259                    limit: usize::try_from(remaining).unwrap_or(usize::MAX),
260                }
261            }
262            _ => RecordIngress::Reject,
263        }
264    }
265
266    /// Stage a canonical replacement without polling, emitting output, or
267    /// retiring a running session.
268    pub fn replace_policy(&mut self, next: crate::logic::InterlaceModule) -> Result<(), CoreError> {
269        self.promote_terminal_session();
270        if matches!(
271            self.owner,
272            CarrierOwner::Failed(_) | CarrierOwner::Closed(_)
273        ) {
274            return Err(CoreError::ReplacementUnavailable("interlace is not open"));
275        }
276        if matches!(self.operand_source, OperandSource::Selected) {
277            return Err(CoreError::ReplacementUnavailable(
278                "a port-selected interlace takes each session's policy from its selector",
279            ));
280        }
281        validate_operand_blob_size(&next)?;
282        match &mut self.owner {
283            CarrierOwner::Prefacing(prefacing) => {
284                if let Some(operand) = &mut prefacing.local_operand {
285                    operand.operand = next;
286                }
287                Ok(())
288            }
289            CarrierOwner::Running {
290                session,
291                staged_replacement,
292            } => {
293                if session.local_operand() == Some(&next) {
294                    *staged_replacement = None;
295                    return Ok(());
296                }
297                if staged_replacement.as_ref() == Some(&next) {
298                    return Ok(());
299                }
300                if session.active_store_call().is_some() {
301                    return Err(CoreError::ReplacementUnavailable(
302                        "connected policy replacement is unavailable while a store call is active",
303                    ));
304                }
305                *staged_replacement = Some(next);
306                Ok(())
307            }
308            CarrierOwner::Failed(_) | CarrierOwner::Closed(_) => {
309                Err(CoreError::ReplacementUnavailable("interlace is not open"))
310            }
311        }
312    }
313
314    /// Classifies one already-decoded input without making the host inspect
315    /// carrier, session, or endpoint state. `RejectNow` means the host should
316    /// hand the input to `push_input` so the core produces the protocol failure.
317    pub fn peer_input_disposition(&self, input: &PeerInput) -> PeerInputDisposition {
318        if self.active_store_call().is_some() {
319            return PeerInputDisposition::Backpressure;
320        }
321        let item = match input {
322            PeerInput::Item(item) => Some(item),
323            PeerInput::DiscardedRecord(_) => None,
324        };
325        match (&self.owner, item) {
326            (CarrierOwner::Prefacing(prefacing), Some(IltpItem::SessionPreface(_))) => {
327                if prefacing.peer_nonce.is_none() {
328                    PeerInputDisposition::Accept
329                } else {
330                    PeerInputDisposition::RejectNow
331                }
332            }
333            (CarrierOwner::Prefacing(prefacing), _) => {
334                if prefacing.cause == PrefaceCause::Reset && prefacing.peer_nonce.is_none() {
335                    PeerInputDisposition::Accept
336                } else if prefacing.cause == PrefaceCause::Initial && prefacing.peer_nonce.is_none()
337                {
338                    PeerInputDisposition::RejectNow
339                } else {
340                    PeerInputDisposition::Backpressure
341                }
342            }
343            (CarrierOwner::Running { .. }, Some(IltpItem::SessionPreface(_))) => {
344                PeerInputDisposition::Accept
345            }
346            (CarrierOwner::Running { session, .. }, Some(item)) => {
347                if session.can_accept_item(item) {
348                    PeerInputDisposition::Accept
349                } else if session.has_free_peer_slot() {
350                    PeerInputDisposition::RejectNow
351                } else {
352                    PeerInputDisposition::Backpressure
353                }
354            }
355            (CarrierOwner::Running { .. }, None) => PeerInputDisposition::RejectNow,
356            (CarrierOwner::Failed(_) | CarrierOwner::Closed(_), _) => {
357                PeerInputDisposition::Backpressure
358            }
359        }
360    }
361
362    pub fn push_input(&mut self, input: impl Into<PeerInput>) -> Result<(), MachineError> {
363        self.consume_staged_or_fail()?;
364        if self.active_store_call().is_some() {
365            return Err(integration_machine_error(
366                "peer input is blocked while a store call is active",
367            ));
368        }
369        let ordinary = match input.into() {
370            PeerInput::Item(IltpItem::SessionPreface(preface)) => {
371                return self.push_preface(preface);
372            }
373            ordinary => ordinary,
374        };
375        match &mut self.owner {
376            CarrierOwner::Prefacing(prefacing)
377                if prefacing.cause == PrefaceCause::Reset && prefacing.peer_nonce.is_none() =>
378            {
379                let size = match &ordinary {
380                    PeerInput::Item(item) => crate::iltp::canonical_item_size(item)
381                        .map_err(|error| integration_machine_error(error.to_string()))?
382                        as u64,
383                    PeerInput::DiscardedRecord(len) => *len as u64,
384                };
385                prefacing.discarded_control_bytes = prefacing
386                    .discarded_control_bytes
387                    .checked_add(size)
388                    .ok_or_else(|| {
389                        integration_machine_error("discarded control-byte accounting overflow")
390                    })?;
391                if prefacing.discarded_control_bytes > self.limits.max_received_control_bytes {
392                    return self.fail(MachineError {
393                        category: ErrorCategory::LimitExceeded,
394                        side: Some(self.role.protocol_side()),
395                        record_hash: None,
396                        detail: "received reset control-byte limit exceeded".into(),
397                        source: None,
398                    });
399                }
400                Ok(())
401            }
402            CarrierOwner::Prefacing(_) => {
403                self.fail(peer_protocol("ordinary input before session preface pair"))
404            }
405            CarrierOwner::Running { session, .. } => {
406                let PeerInput::Item(item) = ordinary else {
407                    return Err(integration_machine_error(
408                        "a discarded record reached a running session",
409                    ));
410                };
411                let result = session.push_item(item);
412                match &result {
413                    Ok(()) => self.peer_by = session.peer_by(),
414                    Err(error) => self.owner = CarrierOwner::Failed(error.clone()),
415                }
416                result
417            }
418            CarrierOwner::Failed(error) => Err(error.clone()),
419            CarrierOwner::Closed(_) => Err(integration_machine_error(
420                "cannot push input to closed carrier",
421            )),
422        }
423    }
424
425    /// Fails with the retained terminal error when staged replacement or fresh
426    /// session construction fails the carrier; host misuse is `Integration`
427    /// without a transition.
428    pub fn accept_peer_output(&mut self) -> Result<Vec<IltpItem>, MachineError> {
429        self.consume_staged_or_fail()?;
430        let result = match &mut self.owner {
431            CarrierOwner::Prefacing(prefacing) => {
432                let items = prefacing.retained_output.take().ok_or_else(|| {
433                    integration_machine_error("carrier has no retained preface output")
434                })?;
435                prefacing.local_preface_accepted = true;
436                self.construct_session_if_ready().map(|()| items)
437            }
438            CarrierOwner::Running { session, .. } => session.accept_peer_output(),
439            CarrierOwner::Failed(error) => Err(error.clone()),
440            CarrierOwner::Closed(_) => Err(integration_machine_error("interlace is not open")),
441        };
442        self.promote_terminal_session();
443        result
444    }
445
446    pub fn submit_store_call(&mut self, id: u64) -> Result<(), MachineError> {
447        let result = match &mut self.owner {
448            CarrierOwner::Running { session, .. } => session.submit_store_call(id),
449            _ => Err(integration_machine_error("carrier has no active session")),
450        };
451        self.promote_terminal_session();
452        result
453    }
454
455    pub fn try_store_position_advanced(
456        &mut self,
457        boundary: &FixedPointWakeBoundary,
458        position: StorePosition,
459    ) -> StorePositionAdvance {
460        if self.consume_staged_or_fail().is_err() {
461            return StorePositionAdvance::Stale;
462        }
463        match &mut self.owner {
464            CarrierOwner::Running { session, .. } => {
465                session.try_store_position_advanced(boundary, position)
466            }
467            _ => StorePositionAdvance::Stale,
468        }
469    }
470
471    pub fn cancel(&mut self) {
472        match &mut self.owner {
473            CarrierOwner::Running { session, .. } => session.cancel(),
474            CarrierOwner::Prefacing(_) => self.owner = CarrierOwner::Closed(CloseReason::Cancelled),
475            CarrierOwner::Failed(_) | CarrierOwner::Closed(_) => {}
476        }
477        self.promote_terminal_session();
478    }
479
480    pub fn timeout(&mut self) {
481        match &mut self.owner {
482            CarrierOwner::Prefacing(_) => {
483                let error = MachineError {
484                    category: ErrorCategory::Timeout,
485                    side: Some(self.role.protocol_side()),
486                    record_hash: None,
487                    detail: "phase timeout in Prefacing".into(),
488                    source: None,
489                };
490                self.owner = CarrierOwner::Failed(error);
491            }
492            CarrierOwner::Running { session, .. } => session.timeout(),
493            _ => {}
494        }
495        self.promote_terminal_session();
496    }
497
498    pub fn peer_eof(&mut self) -> Result<(), MachineError> {
499        let result = match &mut self.owner {
500            CarrierOwner::Running { session, .. } => session.peer_eof(),
501            CarrierOwner::Prefacing(_) => {
502                return self.fail(peer_protocol("peer EOF during prefacing"));
503            }
504            CarrierOwner::Failed(error) => return Err(error.clone()),
505            CarrierOwner::Closed(_) => return Ok(()),
506        };
507        self.promote_terminal_session();
508        result
509    }
510
511    pub fn poll(&mut self) -> MachinePoll<'_> {
512        // A failed replacement is retained and reported as `Failed` below.
513        let _ = self.consume_staged_or_fail();
514        self.promote_terminal_session();
515        match &mut self.owner {
516            CarrierOwner::Prefacing(prefacing) => {
517                if prefacing.retained_output.is_none() && !prefacing.local_preface_accepted {
518                    prefacing.retained_output =
519                        Some(vec![IltpItem::SessionPreface(SessionPreface {
520                            nonce: prefacing.local_nonce.clone(),
521                            offers: self.stream_offers.clone(),
522                            port: self.port.clone(),
523                        })]);
524                }
525                if let Some(items) = &prefacing.retained_output {
526                    let effect = EndpointEffectRef::PeerOutput(PeerOutputRef { items });
527                    return MachinePoll::Effects(if self.role == CarrierRole::Listener {
528                        ReadyEffectsRef {
529                            side0: Some(effect),
530                            side1: None,
531                        }
532                    } else {
533                        ReadyEffectsRef {
534                            side0: None,
535                            side1: Some(effect),
536                        }
537                    });
538                }
539                if prefacing.peer_nonce.is_some() && prefacing.local_operand.is_none() {
540                    return MachinePoll::SelectOperand {
541                        port: self.peer_port.as_ref(),
542                    };
543                }
544                MachinePoll::NeedPeerInput
545            }
546            CarrierOwner::Running { session, .. } => session.poll(),
547            CarrierOwner::Failed(error) => MachinePoll::Failed(error),
548            CarrierOwner::Closed(reason) => MachinePoll::Closed(reason.clone()),
549        }
550    }
551
552    fn promote_terminal_session(&mut self) {
553        let terminal = match &self.owner {
554            CarrierOwner::Running { session, .. } => match session.state() {
555                SessionState::Failed(error) => Some(Err(error.clone())),
556                SessionState::Closed(reason) => Some(Ok(reason.clone())),
557                _ => None,
558            },
559            _ => None,
560        };
561        match terminal {
562            Some(Err(error)) => self.owner = CarrierOwner::Failed(error),
563            Some(Ok(reason)) => self.owner = CarrierOwner::Closed(reason),
564            None => {}
565        }
566    }
567
568    /// Share `cache` with this and every later logical session. A carrier
569    /// opened without a local participant adopts the selected participant's
570    /// cache while the first [`MachinePoll::SelectOperand`] is unanswered.
571    pub fn adopt_operand_cache(&mut self, cache: OperandCache) -> Result<(), MachineError> {
572        match &self.owner {
573            CarrierOwner::Prefacing(prefacing)
574                if prefacing.peer_nonce.is_some() && prefacing.local_operand.is_none() =>
575            {
576                self.operand_cache = cache;
577                Ok(())
578            }
579            CarrierOwner::Failed(error) => Err(error.clone()),
580            _ => Err(integration_machine_error(
581                "the carrier is not selecting an operand",
582            )),
583        }
584    }
585
586    /// Answer [`MachinePoll::SelectOperand`]: this logical session's operand,
587    /// or `None` to refuse the peer's port, which fails the carrier.
588    pub fn select_operand(
589        &mut self,
590        selection: Option<SessionOperand>,
591    ) -> Result<(), MachineError> {
592        match &self.owner {
593            CarrierOwner::Prefacing(prefacing)
594                if prefacing.peer_nonce.is_some() && prefacing.local_operand.is_none() => {}
595            CarrierOwner::Failed(error) => return Err(error.clone()),
596            _ => {
597                return Err(integration_machine_error(
598                    "the carrier is not selecting an operand",
599                ));
600            }
601        }
602        let Some(selected) = selection else {
603            return self.fail(peer_protocol("port refused"));
604        };
605        if let Err(error) = selected.validate(self.peer_proof) {
606            return self.fail(core_machine_error(error));
607        }
608        if let CarrierOwner::Prefacing(prefacing) = &mut self.owner {
609            prefacing.local_operand = Some(selected);
610        }
611        self.construct_session_if_ready()
612    }
613
614    fn push_preface(&mut self, preface: SessionPreface) -> Result<(), MachineError> {
615        self.promote_terminal_session();
616        let SessionPreface {
617            nonce: peer_nonce,
618            port,
619            offers,
620        } = preface;
621        let prior = std::mem::replace(
622            &mut self.owner,
623            CarrierOwner::Closed(CloseReason::Cancelled),
624        );
625        if matches!(
626            prior,
627            CarrierOwner::Running { .. } | CarrierOwner::Prefacing(_)
628        ) {
629            self.peer_port = port;
630            self.peer_by = None;
631            self.encodings = Some([
632                StreamOffers::select(&self.stream_offers, &offers),
633                StreamOffers::select(&offers, &self.stream_offers),
634            ]);
635        }
636        match prior {
637            CarrierOwner::Running { session, .. } => {
638                let local_operand = match &self.operand_source {
639                    OperandSource::Selected => None,
640                    OperandSource::Fixed { expected_peer_by } => {
641                        let expected_peer_by = *expected_peer_by;
642                        match (*session).into_local_operand() {
643                            Some(operand) => Some(SessionOperand {
644                                operand,
645                                expected_peer_by,
646                            }),
647                            None => {
648                                return self.fail(integration_machine_error(
649                                    "running session lost its local operand",
650                                ));
651                            }
652                        }
653                    }
654                };
655                let local_nonce = match self.derive_nonce(Some(&peer_nonce)) {
656                    Ok(nonce) => nonce,
657                    Err(_) => {
658                        return self.fail(integration_machine_error(
659                            "carrier nonce generator exhausted",
660                        ));
661                    }
662                };
663                self.owner = CarrierOwner::Prefacing(PrefacingState::with_peer(
664                    PrefaceCause::Reset,
665                    local_operand,
666                    local_nonce,
667                    peer_nonce,
668                ));
669                Ok(())
670            }
671            CarrierOwner::Prefacing(mut prefacing) => {
672                if prefacing.peer_nonce.is_some() {
673                    self.owner = CarrierOwner::Prefacing(prefacing);
674                    return self.fail(peer_protocol(
675                        "duplicate session preface before fresh session construction",
676                    ));
677                }
678                if prefacing.local_nonce == peer_nonce {
679                    self.owner = CarrierOwner::Prefacing(prefacing);
680                    return self.fail(peer_protocol("side0 and side1 session nonces must differ"));
681                }
682                prefacing.peer_nonce = Some(peer_nonce);
683                self.owner = CarrierOwner::Prefacing(prefacing);
684                self.construct_session_if_ready()
685            }
686            CarrierOwner::Failed(error) => {
687                self.owner = CarrierOwner::Failed(error.clone());
688                Err(error)
689            }
690            CarrierOwner::Closed(reason) => {
691                self.owner = CarrierOwner::Closed(reason);
692                Err(integration_machine_error("cannot preface a closed carrier"))
693            }
694        }
695    }
696
697    fn consume_staged_or_fail(&mut self) -> Result<(), MachineError> {
698        match self.consume_staged_replacement() {
699            Ok(()) => Ok(()),
700            Err(error) => {
701                self.owner = CarrierOwner::Failed(error.clone());
702                Err(error)
703            }
704        }
705    }
706
707    fn consume_staged_replacement(&mut self) -> Result<(), MachineError> {
708        let staged = match &mut self.owner {
709            CarrierOwner::Running {
710                session,
711                staged_replacement,
712            } if session.active_store_call().is_none() => staged_replacement.take(),
713            _ => None,
714        };
715        let Some(next) = staged else { return Ok(()) };
716        let OperandSource::Fixed { expected_peer_by } = self.operand_source else {
717            return Err(integration_machine_error(
718                "a port-selected carrier staged a replacement",
719            ));
720        };
721        let local_nonce = self
722            .derive_nonce(None)
723            .map_err(|_| integration_machine_error("carrier nonce generator exhausted"))?;
724        self.peer_by = None;
725        self.owner = CarrierOwner::Prefacing(PrefacingState::new(
726            PrefaceCause::Reset,
727            Some(SessionOperand {
728                operand: next,
729                expected_peer_by,
730            }),
731            local_nonce,
732        ));
733        Ok(())
734    }
735
736    fn construct_session_if_ready(&mut self) -> Result<(), MachineError> {
737        let ready = matches!(&self.owner, CarrierOwner::Prefacing(prefacing) if prefacing.local_preface_accepted && prefacing.peer_nonce.is_some() && prefacing.local_operand.is_some());
738        if !ready {
739            return Ok(());
740        }
741        let prior = std::mem::replace(
742            &mut self.owner,
743            CarrierOwner::Closed(CloseReason::Cancelled),
744        );
745        let CarrierOwner::Prefacing(prefacing) = prior else {
746            unreachable!()
747        };
748        let peer_nonce = prefacing.peer_nonce.expect("ready preface owns peer nonce");
749        let local_operand = prefacing
750            .local_operand
751            .expect("ready preface owns its operand");
752        if peer_nonce == prefacing.local_nonce {
753            return self.fail(peer_protocol("side0 and side1 session nonces must differ"));
754        }
755        let (side0_nonce, side1_nonce) = if self.role == CarrierRole::Listener {
756            (prefacing.local_nonce, peer_nonce)
757        } else {
758            (peer_nonce, prefacing.local_nonce)
759        };
760        let local_proof = if !matches!(self.peer_proof, PeerProofAssurance::Disabled) {
761            match self
762                .by_secret
763                .as_ref()
764                .map(|by_secret| {
765                    peer_proof::create_proof_record(
766                        by_secret,
767                        self.role,
768                        side0_nonce.as_str(),
769                        side1_nonce.as_str(),
770                        &if self.role == CarrierRole::Listener {
771                            side0_nonce.bytes()
772                        } else {
773                            side1_nonce.bytes()
774                        },
775                        self.peer_proof,
776                    )
777                })
778                .transpose()
779            {
780                Ok(proof) => proof,
781                Err(error) => return self.fail(integration_machine_error(error)),
782            }
783        } else {
784            None
785        };
786        let session = SessionMachine::new(
787            self.role,
788            ConnectedSessionConfig {
789                local_operand: local_operand.operand,
790                local_proof,
791                local_by_value: self.local_by_value,
792                peer_proof: self.peer_proof,
793                expected_peer_by: local_operand.expected_peer_by,
794                side0_nonce,
795                side1_nonce,
796                limits: self.limits.clone(),
797                operand_cache: self.operand_cache.clone(),
798            },
799        );
800        self.owner = CarrierOwner::Running {
801            session: Box::new(session),
802            staged_replacement: None,
803        };
804        Ok(())
805    }
806
807    fn derive_nonce(
808        &mut self,
809        peer: Option<&SessionNonceText>,
810    ) -> Result<SessionNonceText, CoreError> {
811        loop {
812            let sequence = self
813                .next_nonce_sequence
814                .ok_or(CoreError::InvalidConfig("carrier nonce sequence exhausted"))?;
815            self.next_nonce_sequence = sequence.checked_add(1);
816            let mut hasher = blake3::Hasher::new_keyed(self.nonce_seed.expose());
817            hasher.update(NONCE_DOMAIN);
818            hasher.update(&sequence.to_be_bytes());
819            let bytes = *hasher.finalize().as_bytes();
820            if bytes.iter().all(|byte| *byte == 0) || peer.is_some_and(|peer| peer.bytes() == bytes)
821            {
822                continue;
823            }
824            return SessionNonceText::from_bytes(bytes)
825                .map_err(|_| CoreError::InvalidConfig("derived carrier nonce is invalid"));
826        }
827    }
828
829    fn fail<T>(&mut self, error: MachineError) -> Result<T, MachineError> {
830        self.owner = CarrierOwner::Failed(error.clone());
831        Err(error)
832    }
833}
834
835impl PrefacingState {
836    fn new(
837        cause: PrefaceCause,
838        local_operand: Option<SessionOperand>,
839        local_nonce: SessionNonceText,
840    ) -> Self {
841        Self {
842            cause,
843            local_operand,
844            local_nonce,
845            local_preface_accepted: false,
846            peer_nonce: None,
847            retained_output: None,
848            discarded_control_bytes: 0,
849        }
850    }
851    fn with_peer(
852        cause: PrefaceCause,
853        local_operand: Option<SessionOperand>,
854        local_nonce: SessionNonceText,
855        peer_nonce: SessionNonceText,
856    ) -> Self {
857        let mut value = Self::new(cause, local_operand, local_nonce);
858        value.peer_nonce = Some(peer_nonce);
859        value
860    }
861}
862
863#[cfg(test)]
864#[path = "carrier/erasure_tests.rs"]
865mod erasure_tests;
866
867#[cfg(test)]
868#[path = "carrier/tests.rs"]
869mod tests;
lace_sans_io/core/carrier.rs

lace_sans_io/core/
carrier.rs
