lace_sans_io/core/
machine.rs

1//! Connected-session semantic ownership: establishment, the endpoint, the
2//! first round's listing reconciliation, and the 060 items of listings,
3//! requests, responses, and fences.
4
5use super::*;
6use crate::iltp::{IltpItem, ListingForm, RecordLabel};
7
8mod establishment;
9mod reconciliation;
10mod store_result;
11mod wire;
12
13pub use establishment::EstablishmentState;
14use establishment::Progress;
15use reconciliation::{PeerBlock, Reconciliation};
16pub use store_result::validate_store_result;
17use wire::{apply_peer_delta, encode_output, listed_entries, withdrawn_bytes};
18
19#[derive(Debug)]
20pub(super) struct ConnectedSessionConfig {
21    pub(super) local_operand: crate::logic::InterlaceModule,
22    pub(super) local_proof: Option<Record>,
23    pub(super) local_by_value: Option<crate::records::ByValue>,
24    pub(super) peer_proof: PeerProofAssurance,
25    pub(super) expected_peer_by: Option<crate::records::ByValue>,
26    pub(super) side0_nonce: crate::iltp::SessionNonceText,
27    pub(super) side1_nonce: crate::iltp::SessionNonceText,
28    pub(super) limits: InterlaceResourceLimits,
29    pub(super) operand_cache: OperandCache,
30}
31
32/// One owning connected-session semantic state.
33#[derive(Debug)]
34pub enum SessionState {
35    Establishing(Box<EstablishmentState>),
36    Round(Box<low_level::EndpointMachine>),
37    FixedPoint(Box<low_level::EndpointMachine>),
38    Failed(MachineError),
39    Closed(CloseReason),
40}
41
42/// One logical session on a connected carrier. The preface nonce pair stays
43/// with the session: proof validation reads it during establishment and
44/// listing reconciliation derives its session key from it.
45#[derive(Debug)]
46pub struct SessionMachine {
47    id: InterlaceId,
48    role: CarrierRole,
49    side0_nonce: crate::iltp::SessionNonceText,
50    side1_nonce: crate::iltp::SessionNonceText,
51    state: SessionState,
52    /// The first round's listing reconciliation; `None` once its complete peer
53    /// listing is published.
54    reconciliation: Option<Reconciliation>,
55    /// The last complete peer listing, the base the peer's next delta applies
56    /// to. The endpoint shares it until it freezes its request, so the delta
57    /// changes it in place.
58    previous_peer_listing: AdvertisementListing,
59    /// The peer-offered entries of this side's previous request, which extend
60    /// the base of its next listing delta; empty when the plan omits this
61    /// side's `MaySend`.
62    local_base_additions: Vec<Arc<AdvertisementRecord>>,
63    /// This side's entries the peer requested in the previous round, which
64    /// extend the base of the peer's next listing delta; empty when the plan
65    /// omits the peer's `MaySend`.
66    peer_base_additions: Vec<Arc<AdvertisementRecord>>,
67    /// Whether the plan omits `MaySend` for this side and for the peer.
68    never_sends: [bool; 2],
69    retained_output: Option<Vec<IltpItem>>,
70    active_store_call: Option<StoreCall>,
71    store_response: Option<Result<StoreResult, StoreError>>,
72}
73
74impl SessionMachine {
75    pub(super) fn new(role: CarrierRole, config: ConnectedSessionConfig) -> Self {
76        Self {
77            id: InterlaceId(NEXT_INTERLACE_ID.next()),
78            role,
79            side0_nonce: config.side0_nonce.clone(),
80            side1_nonce: config.side1_nonce.clone(),
81            state: SessionState::Establishing(Box::new(EstablishmentState::new(config))),
82            reconciliation: None,
83            previous_peer_listing: AdvertisementListing::default(),
84            local_base_additions: Vec::new(),
85            peer_base_additions: Vec::new(),
86            never_sends: [false; 2],
87            retained_output: None,
88            active_store_call: None,
89            store_response: None,
90        }
91    }
92    #[cfg(test)]
93    pub(super) fn previous_peer_listing(&self) -> &AdvertisementListing {
94        &self.previous_peer_listing
95    }
96    /// The role-ordered preface nonces of this logical session.
97    #[cfg(test)]
98    pub(super) fn session_nonces(&self) -> [&crate::iltp::SessionNonceText; 2] {
99        [&self.side0_nonce, &self.side1_nonce]
100    }
101    pub fn id(&self) -> InterlaceId {
102        self.id
103    }
104    pub fn state(&self) -> &SessionState {
105        &self.state
106    }
107    /// `(sent, received)` reconciliation step items while reconciling.
108    #[cfg(test)]
109    pub(super) fn test_reconciliation_steps(&self) -> Option<(u32, u32)> {
110        self.reconciliation.as_ref().map(Reconciliation::steps)
111    }
112    #[cfg(test)]
113    pub(super) fn test_plan_id(&self) -> Option<RecordHash> {
114        self.endpoint()
115            .map(low_level::EndpointMachine::test_plan_id)
116    }
117    pub(super) fn authority(&self) -> Option<&SessionAuthority> {
118        self.endpoint().map(low_level::EndpointMachine::authority)
119    }
120    /// The By the live peer proved in this session under confidential or
121    /// channel-bound assurance: the `PeerConfidential` key, once validated.
122    pub(super) fn peer_by(&self) -> Option<crate::records::ByValue> {
123        match &self.state {
124            SessionState::Establishing(est) => est.peer_confidential(),
125            _ => self.authority()?.peer_confidential().copied(),
126        }
127    }
128    pub(super) fn local_operand(&self) -> Option<&crate::logic::InterlaceModule> {
129        match &self.state {
130            SessionState::Establishing(est) => Some(est.local_operand()),
131            SessionState::Round(endpoint) | SessionState::FixedPoint(endpoint) => {
132                Some(endpoint.local_operand())
133            }
134            SessionState::Failed(_) | SessionState::Closed(_) => None,
135        }
136    }
137    pub(super) fn into_local_operand(self) -> Option<crate::logic::InterlaceModule> {
138        match self.state {
139            SessionState::Establishing(est) => Some(est.into_local_operand()),
140            SessionState::Round(endpoint) | SessionState::FixedPoint(endpoint) => {
141                Some(endpoint.local_operand().clone())
142            }
143            SessionState::Failed(_) | SessionState::Closed(_) => None,
144        }
145    }
146    pub fn active_store_call(&self) -> Option<&StoreCall> {
147        if self.is_terminating() {
148            None
149        } else {
150            self.active_store_call.as_ref()
151        }
152    }
153    pub(super) fn can_accept_item(&self, item: &IltpItem) -> bool {
154        if self.active_store_call.is_some() {
155            return false;
156        }
157        match (&self.state, item) {
158            (SessionState::Establishing(_), IltpItem::SessionPreface(_)) => false,
159            (SessionState::Establishing(est), _) => est.peer_turn(self.role),
160            (SessionState::Round(_), IltpItem::SessionPreface(_)) => false,
161            (SessionState::Round(endpoint), IltpItem::Record(..)) => {
162                endpoint.can_accept_record_response()
163            }
164            (SessionState::Round(endpoint), _) => self.free_listing_slot(endpoint),
165            (SessionState::FixedPoint(_), item) => {
166                !matches!(item, IltpItem::Record(..) | IltpItem::SessionPreface(_))
167            }
168            _ => false,
169        }
170    }
171    pub(super) fn has_free_peer_slot(&self) -> bool {
172        match &self.state {
173            SessionState::Establishing(est) => est.peer_turn(self.role),
174            SessionState::Round(endpoint) | SessionState::FixedPoint(endpoint) => {
175                self.free_listing_slot(endpoint)
176            }
177            SessionState::Failed(_) | SessionState::Closed(_) => false,
178        }
179    }
180    /// During reconciliation the carrier-owned step item slot stands in for
181    /// the endpoint's peer-listing slot.
182    fn free_listing_slot(&self, endpoint: &low_level::EndpointMachine) -> bool {
183        match &self.reconciliation {
184            Some(reconciliation) => reconciliation.expects_peer_item(),
185            None => endpoint.has_free_peer_slot(),
186        }
187    }
188    pub fn active_store_call_mut(&mut self) -> Option<&mut StoreCall> {
189        if self.is_terminating() {
190            None
191        } else {
192            self.active_store_call.as_mut()
193        }
194    }
195    pub fn is_terminating(&self) -> bool {
196        matches!(
197            self.state,
198            SessionState::Failed(_) | SessionState::Closed(_)
199        )
200    }
201    pub fn protocol_phase(&self) -> MachineProtocolPhase {
202        match &self.state {
203            SessionState::Establishing(est) => MachineProtocolPhase::Establishing(est.phase),
204            SessionState::Round(endpoint) => endpoint_protocol_phase(endpoint),
205            SessionState::FixedPoint(endpoint) => MachineProtocolPhase::FixedPoint {
206                round: endpoint.round(),
207            },
208            SessionState::Failed(_) => MachineProtocolPhase::Failed,
209            SessionState::Closed(_) => MachineProtocolPhase::Closed,
210        }
211    }
212
213    pub fn push_item(&mut self, item: IltpItem) -> Result<(), MachineError> {
214        if self.is_terminating() {
215            return Err(integration_machine_error(
216                "cannot push input to terminal session",
217            ));
218        }
219        if self.active_store_call.is_some() {
220            return Err(integration_machine_error(
221                "peer input is blocked while a store call is active",
222            ));
223        }
224        let result = if matches!(self.state, SessionState::Establishing(_)) {
225            self.consume_establishment(item)
226        } else {
227            self.consume_round_item(item)
228        };
229        if let Err(error) = result {
230            self.set_failure(error.clone());
231            return Err(error);
232        }
233        Ok(())
234    }
235    pub fn peer_eof(&mut self) -> Result<(), MachineError> {
236        match &self.state {
237            SessionState::FixedPoint(_) => {
238                self.set_closed(CloseReason::PeerClosed);
239                Ok(())
240            }
241            SessionState::Closed(_) => Ok(()),
242            SessionState::Failed(error) => Err(error.clone()),
243            SessionState::Establishing(_) | SessionState::Round(_) => {
244                let error = peer_protocol("peer EOF before fixed point");
245                self.set_failure(error.clone());
246                Err(error)
247            }
248        }
249    }
250    pub fn cancel(&mut self) {
251        if matches!(
252            self.state,
253            SessionState::Establishing(_) | SessionState::Round(_) | SessionState::FixedPoint(_)
254        ) {
255            self.set_closed(CloseReason::Cancelled);
256        }
257    }
258    pub fn timeout(&mut self) {
259        if !self.protocol_phase().timeout_phase().is_some() {
260            return;
261        }
262        let error = MachineError {
263            category: ErrorCategory::Timeout,
264            side: Some(self.local_protocol_side()),
265            record_hash: None,
266            detail: format!("phase timeout in {:?}", self.protocol_phase()),
267            source: None,
268        };
269        self.set_failure(error);
270    }
271    pub fn try_store_position_advanced(
272        &mut self,
273        boundary: &FixedPointWakeBoundary,
274        position: StorePosition,
275    ) -> StorePositionAdvance {
276        let prior = std::mem::replace(
277            &mut self.state,
278            SessionState::Closed(CloseReason::Cancelled),
279        );
280        let mut result = StorePositionAdvance::Stale;
281        self.state = match prior {
282            SessionState::FixedPoint(mut endpoint) => {
283                result = endpoint.try_store_position_advanced(boundary, position);
284                if result == StorePositionAdvance::Accepted {
285                    SessionState::Round(endpoint)
286                } else {
287                    SessionState::FixedPoint(endpoint)
288                }
289            }
290            other => other,
291        };
292        result
293    }
294    /// Accept the retained peer output, handing its items to the writer.
295    pub fn accept_peer_output(&mut self) -> Result<Vec<IltpItem>, MachineError> {
296        let items = self
297            .retained_output
298            .take()
299            .ok_or_else(|| integration_machine_error("session has no retained peer output"))?;
300        self.accept_retained_output()?;
301        Ok(items)
302    }
303
304    fn accept_retained_output(&mut self) -> Result<(), MachineError> {
305        if let SessionState::Establishing(est) = &mut self.state {
306            let result = est.accept_output(self.role).and_then(|progress| {
307                if progress == Progress::Complete {
308                    self.activate_from_establishment()
309                } else {
310                    Ok(())
311                }
312            });
313            if let Err(error) = &result {
314                self.set_failure(error.clone());
315            }
316            return result;
317        }
318        if self.reconciliation.is_some() {
319            self.reconciliation
320                .as_mut()
321                .expect("checked")
322                .accept_output();
323            if let Err(error) = self.advance_reconciliation() {
324                self.set_failure(error.clone());
325                return Err(error);
326            }
327        } else if let SessionState::Round(endpoint) = &mut self.state {
328            endpoint.accept_peer_output();
329        }
330        Ok(())
331    }
332    pub fn submit_store_call(&mut self, id: u64) -> Result<(), MachineError> {
333        let Some(mut active) = self.active_store_call.take() else {
334            return Err(integration_machine_error(
335                "session has no active store call",
336            ));
337        };
338        if active.id != id || active.owner != self.id {
339            self.active_store_call = Some(active);
340            return Err(integration_machine_error(
341                "store call does not belong to this session",
342            ));
343        }
344        let response = match active.response.take() {
345            Some(response) => response,
346            None => {
347                return self.fail_store_submission(CoreError::InvalidConfig(
348                    "store-call response is incomplete",
349                ));
350            }
351        };
352        if let Ok(result) = &response
353            && let Err(error) = validate_store_result(active.op(), result)
354        {
355            return self.fail_store_submission(error);
356        }
357        self.endpoint_mut()
358            .ok_or_else(|| integration_machine_error("session has no active endpoint"))?
359            .restore_pending_store(active.continuation);
360        self.store_response = Some(response);
361        Ok(())
362    }
363
364    pub fn poll(&mut self) -> MachinePoll<'_> {
365        let mut observation = None;
366        let mut entered_fixed = None;
367        for _ in 0..1024 {
368            // While its output waits for the writer, the endpoint still
369            // admits a received response, so the peer's writer progresses.
370            let admission_only = self.retained_output.is_some();
371            if matches!(
372                self.state,
373                SessionState::Failed(_) | SessionState::Closed(_) | SessionState::FixedPoint(_)
374            ) || self.active_store_call.is_some()
375                || admission_only
376                    && !matches!(&self.state, SessionState::Round(endpoint) if endpoint.has_admission_work())
377            {
378                break;
379            }
380            if matches!(self.state, SessionState::Establishing(_)) {
381                if let Err(error) = self.prepare_establishment_output() {
382                    self.set_failure(error);
383                }
384                break;
385            }
386            let poll = {
387                let response = &mut self.store_response;
388                let SessionState::Round(endpoint) = &mut self.state else {
389                    unreachable!()
390                };
391                if admission_only {
392                    endpoint.step_admission(response)
393                } else {
394                    endpoint.step(response)
395                }
396            };
397            match poll {
398                EndpointPoll::PeerOutput { message } => {
399                    let result = match message {
400                        EndpointMessage::Advertisement(listing, _)
401                            if self.reconciliation.is_some() =>
402                        {
403                            self.reconciliation
404                                .as_mut()
405                                .expect("checked")
406                                .freeze_local(listing)
407                                .and_then(|()| self.advance_reconciliation())
408                        }
409                        message => self
410                            .encode_round_output(message)
411                            .map(|item| self.retained_output = Some(vec![item])),
412                    };
413                    if let Err(error) = result {
414                        self.set_failure(error);
415                    }
416                }
417                EndpointPoll::StoreRequest(request) => self.begin_store_call(request),
418                EndpointPoll::Observation(value) => {
419                    observation = Some(value);
420                    break;
421                }
422                EndpointPoll::FixedPoint(boundary) => {
423                    let prior = std::mem::replace(
424                        &mut self.state,
425                        SessionState::Closed(CloseReason::Cancelled),
426                    );
427                    let SessionState::Round(endpoint) = prior else {
428                        unreachable!()
429                    };
430                    self.state = SessionState::FixedPoint(endpoint);
431                    entered_fixed = Some(boundary);
432                    break;
433                }
434                EndpointPoll::Closed(reason) => {
435                    self.set_closed(reason);
436                    break;
437                }
438                EndpointPoll::LogicError(error) => {
439                    self.set_failure(machine_error_from_logic(error))
440                }
441                EndpointPoll::StoreFailed {
442                    side,
443                    record_hash,
444                    error,
445                } => self.set_failure(store_machine_error(side, record_hash, error)),
446                EndpointPoll::Progress => continue,
447                EndpointPoll::Idle => break,
448            }
449        }
450        if let Some(value) = observation {
451            return MachinePoll::Observation(value);
452        }
453        if let Some(value) = entered_fixed {
454            return MachinePoll::EnteredFixedPoint(value);
455        }
456        match &self.state {
457            SessionState::Failed(error) => MachinePoll::Failed(error),
458            SessionState::Closed(reason) => MachinePoll::Closed(reason.clone()),
459            SessionState::FixedPoint(endpoint) => MachinePoll::NeedStoreAdvance(
460                endpoint
461                    .fixed_point()
462                    .expect("fixed-point session owns fixed-point endpoint")
463                    .clone(),
464            ),
465            SessionState::Establishing(_) | SessionState::Round(_) => {
466                let role = self.carrier_role();
467                let effect = if let Some(call) = &self.active_store_call {
468                    EndpointEffectRef::StoreCall(call)
469                } else if let Some(items) = &self.retained_output {
470                    EndpointEffectRef::PeerOutput(PeerOutputRef { items })
471                } else {
472                    return MachinePoll::NeedPeerInput;
473                };
474                MachinePoll::Effects(if role == CarrierRole::Listener {
475                    ReadyEffectsRef {
476                        side0: Some(effect),
477                        side1: None,
478                    }
479                } else {
480                    ReadyEffectsRef {
481                        side0: None,
482                        side1: Some(effect),
483                    }
484                })
485            }
486        }
487    }
488
489    fn prepare_establishment_output(&mut self) -> Result<(), MachineError> {
490        let SessionState::Establishing(est) = &self.state else {
491            return Ok(());
492        };
493        self.retained_output = est.output(self.role)?;
494        Ok(())
495    }
496
497    fn consume_establishment(&mut self, item: IltpItem) -> Result<(), MachineError> {
498        let SessionState::Establishing(est) = &mut self.state else {
499            unreachable!()
500        };
501        let nonces = [self.side0_nonce.as_str(), self.side1_nonce.as_str()];
502        if est.consume(self.role, nonces, item)? == Progress::Complete {
503            self.activate_from_establishment()?;
504        }
505        Ok(())
506    }
507
508    /// How the host receives the next peer record in this session's state.
509    pub(super) fn record_ingress(&self) -> RecordIngress {
510        match &self.state {
511            SessionState::Establishing(est) => est.record_ingress(),
512            SessionState::Round(endpoint) => endpoint.record_ingress(),
513            SessionState::FixedPoint(_) | SessionState::Failed(_) | SessionState::Closed(_) => {
514                RecordIngress::Reject
515            }
516        }
517    }
518
519    fn activate_from_establishment(&mut self) -> Result<(), MachineError> {
520        let prior = std::mem::replace(
521            &mut self.state,
522            SessionState::Closed(CloseReason::Cancelled),
523        );
524        let SessionState::Establishing(est) = prior else {
525            self.state = prior;
526            return Err(integration_machine_error("session is not establishing"));
527        };
528        let established = est.into_established()?;
529        let operands = if self.role == CarrierRole::Listener {
530            [established.local_operand, established.peer_operand]
531        } else {
532            [established.peer_operand, established.local_operand]
533        };
534        let plan = Arc::new(compile_exchange_plan(operands));
535        let slot = if self.role == CarrierRole::Listener {
536            crate::logic::ExchangePlanSlot::Slot0
537        } else {
538            crate::logic::ExchangePlanSlot::Slot1
539        };
540        self.never_sends = [plan.never_sends(slot), plan.never_sends(slot.other())];
541        self.state = SessionState::Round(Box::new(low_level::EndpointMachine::new(
542            self.id,
543            SideExecutionConfig::new(plan, slot, established.authority, established.limits),
544        )));
545        self.reconciliation = Some(Reconciliation::new(
546            crate::core::listing_symbols::SessionKey::derive(
547                &self.side0_nonce.bytes(),
548                &self.side1_nonce.bytes(),
549            ),
550        ));
551        Ok(())
552    }
553    /// One peer listing or symbol item during the first round's
554    /// reconciliation: charge its decoded bytes, advance the step state, and
555    /// publish the complete peer listing once nothing more is owed in either
556    /// direction.
557    fn consume_reconciliation_item(&mut self, item: IltpItem) -> Result<(), MachineError> {
558        let block = PeerBlock::from_item(item)?;
559        let SessionState::Round(endpoint) = &mut self.state else {
560            unreachable!()
561        };
562        endpoint
563            .charge_received_control(block.charged_bytes())
564            .map_err(machine_error_from_logic)?;
565        self.reconciliation
566            .as_mut()
567            .expect("reconciliation item")
568            .receive(block)?;
569        self.advance_reconciliation()
570    }
571    fn advance_reconciliation(&mut self) -> Result<(), MachineError> {
572        let Some(reconciliation) = &mut self.reconciliation else {
573            return Ok(());
574        };
575        if self.retained_output.is_none()
576            && let Some(item) = reconciliation.next_output()?
577        {
578            self.retained_output = Some(vec![item]);
579        }
580        let Some(listing) = reconciliation.complete() else {
581            return Ok(());
582        };
583        self.reconciliation = None;
584        self.previous_peer_listing = listing.clone();
585        let endpoint = self
586            .endpoint_mut()
587            .ok_or_else(|| integration_machine_error("session has no endpoint"))?;
588        endpoint.accept_peer_output();
589        // The session's first peer listing adds every entry.
590        push_endpoint_message(
591            endpoint,
592            EndpointMessage::Advertisement(listing, ListingChange::All),
593        )
594    }
595    /// Interpret one decoded peer item from the endpoint's current state. A
596    /// listing at fixed point is the next round's delta and starts that round
597    /// first, so its control bytes charge the new round.
598    fn consume_round_item(&mut self, item: IltpItem) -> Result<(), MachineError> {
599        if matches!(self.state, SessionState::FixedPoint(_))
600            && matches!(&item, IltpItem::Listing { .. })
601        {
602            let prior = std::mem::replace(
603                &mut self.state,
604                SessionState::Closed(CloseReason::Cancelled),
605            );
606            let SessionState::FixedPoint(mut endpoint) = prior else {
607                unreachable!()
608            };
609            endpoint.start_open_round();
610            self.state = SessionState::Round(endpoint);
611        }
612        if self.reconciliation.is_some()
613            && matches!(item, IltpItem::Listing { .. } | IltpItem::Symbols(_))
614        {
615            return self.consume_reconciliation_item(item);
616        }
617        let (SessionState::Round(endpoint) | SessionState::FixedPoint(endpoint)) = &mut self.state
618        else {
619            return Err(integration_machine_error("session has no endpoint"));
620        };
621        let message = match (endpoint.phase(), item) {
622            (_, IltpItem::SessionPreface(_)) => {
623                return Err(peer_protocol("session preface belongs to the carrier"));
624            }
625            (_, IltpItem::Record(RecordLabel::Response, record)) => EndpointMessage::Record(record),
626            (
627                Some(RoundPhase::Advertising),
628                IltpItem::Listing {
629                    form: ListingForm::Delta,
630                    rows,
631                },
632            ) => {
633                endpoint
634                    .charge_received_control(withdrawn_bytes(&rows))
635                    .map_err(machine_error_from_logic)?;
636                let (listing, change) = apply_peer_delta(
637                    std::mem::take(&mut self.previous_peer_listing),
638                    std::mem::take(&mut self.peer_base_additions),
639                    rows,
640                )?;
641                self.previous_peer_listing = listing.clone();
642                EndpointMessage::Advertisement(listing, change)
643            }
644            (Some(RoundPhase::Requesting), IltpItem::Want(record_hashes)) => {
645                if !self.never_sends[1] {
646                    self.peer_base_additions = peer_requested_entries(endpoint, &record_hashes)?;
647                }
648                EndpointMessage::Request(RequestListing { record_hashes })
649            }
650            (Some(RoundPhase::Answering | RoundPhase::Fencing), IltpItem::RoundEnd) => {
651                EndpointMessage::RoundComplete
652            }
653            (_, item) => {
654                return Err(peer_protocol(format!(
655                    "{} is invalid in the current phase",
656                    item_label(&item)
657                )));
658            }
659        };
660        push_endpoint_message(endpoint, message)
661    }
662    /// Encode one endpoint output after a session's first round's listing. A
663    /// listing delta consumes the base additions of this side's previous
664    /// request; a request records its own.
665    fn encode_round_output(&mut self, message: EndpointMessage) -> Result<IltpItem, MachineError> {
666        match &message {
667            EndpointMessage::Advertisement(..) => {
668                let additions = std::mem::take(&mut self.local_base_additions);
669                return encode_output(message, &additions);
670            }
671            EndpointMessage::Request(request) if !self.never_sends[0] => {
672                self.local_base_additions =
673                    listed_entries(&self.previous_peer_listing.entries, &request.record_hashes)
674                        .map_err(|hash| {
675                            integration_machine_error(format!(
676                                "request for {hash}, which the peer listing lacks"
677                            ))
678                        })?;
679            }
680            _ => {}
681        }
682        encode_output(message, &[])
683    }
684    fn begin_store_call(&mut self, request: StoreRequest) {
685        let id = NEXT_STORE_CALL_ID.next();
686        let continuation = self
687            .endpoint_mut()
688            .expect("round endpoint")
689            .take_pending_store()
690            .expect("StoreRequest owns continuation");
691        self.active_store_call = Some(StoreCall::new(id, self.id, request, continuation));
692    }
693    fn endpoint(&self) -> Option<&low_level::EndpointMachine> {
694        match &self.state {
695            SessionState::Round(e) | SessionState::FixedPoint(e) => Some(e),
696            _ => None,
697        }
698    }
699    fn endpoint_mut(&mut self) -> Option<&mut low_level::EndpointMachine> {
700        match &mut self.state {
701            SessionState::Round(e) | SessionState::FixedPoint(e) => Some(e),
702            _ => None,
703        }
704    }
705    fn carrier_role(&self) -> CarrierRole {
706        self.role
707    }
708    fn local_protocol_side(&self) -> InterlaceSide {
709        self.role.protocol_side()
710    }
711    fn fail_store_submission<T>(&mut self, error: CoreError) -> Result<T, MachineError> {
712        let error = core_machine_error(error);
713        self.set_failure(error.clone());
714        Err(error)
715    }
716    fn retire_effects(&mut self) {
717        self.reconciliation = None;
718        self.retained_output = None;
719        self.active_store_call = None;
720        self.store_response = None;
721    }
722    fn set_failure(&mut self, error: MachineError) {
723        self.retire_effects();
724        self.state = SessionState::Failed(error);
725    }
726    fn set_closed(&mut self, reason: CloseReason) {
727        self.retire_effects();
728        self.state = SessionState::Closed(reason);
729    }
730}
731
732/// The local entries the peer's request names: they extend the base of its
733/// next listing delta.
734fn peer_requested_entries(
735    endpoint: &low_level::EndpointMachine,
736    hashes: &[RecordHash],
737) -> Result<Vec<Arc<AdvertisementRecord>>, MachineError> {
738    let local = endpoint
739        .frozen_local_listing()
740        .ok_or_else(|| integration_machine_error("peer request before a frozen listing"))?;
741    listed_entries(&local.entries, hashes)
742        .map_err(|hash| peer_protocol(format!("request is outside frozen commitment: {hash}")))
743}
744
745fn push_endpoint_message(
746    endpoint: &mut low_level::EndpointMachine,
747    message: EndpointMessage,
748) -> Result<(), MachineError> {
749    match endpoint.push_input(message) {
750        Ok(()) => Ok(()),
751        Err(low_level::PushInputError::Unavailable) => {
752            Err(peer_protocol("peer input is invalid in the current state"))
753        }
754        Err(low_level::PushInputError::Logic(error)) => Err(machine_error_from_logic(error)),
755    }
756}
757
758/// The label of a peer item, for protocol failure details.
759pub(in crate::core) fn item_label(item: &IltpItem) -> &'static str {
760    match item {
761        IltpItem::SessionPreface(_) => "session preface",
762        IltpItem::NoProof => "::as unknown",
763        IltpItem::Record(RecordLabel::Proof, _) => "::as",
764        IltpItem::Policy(_) => "::policy",
765        IltpItem::PolicyKnown => "::policy-known",
766        IltpItem::PolicyMissing => "::policy-missing",
767        IltpItem::Record(RecordLabel::PolicyBody, _) => "::policy-body",
768        IltpItem::Listing { form, .. } => form.label().trim_end(),
769        IltpItem::Symbols(_) => "::symbols",
770        IltpItem::Want(_) => "::want",
771        IltpItem::Record(RecordLabel::Response, _) => "::response",
772        IltpItem::RoundEnd => "::round-end",
773    }
774}
775
776pub(super) fn endpoint_protocol_phase(
777    endpoint: &low_level::EndpointMachine,
778) -> MachineProtocolPhase {
779    match endpoint.endpoint_state() {
780        EndpointState::Round(round) => MachineProtocolPhase::Round {
781            round: round.round,
782            phase: round.phase,
783        },
784        EndpointState::FixedPoint(fixed) => MachineProtocolPhase::FixedPoint {
785            round: fixed.boundary.round,
786        },
787        EndpointState::Failed(_) => MachineProtocolPhase::Failed,
788        EndpointState::Closed(_) => MachineProtocolPhase::Closed,
789    }
790}
791pub(super) fn machine_error_from_logic(error: LogicError) -> MachineError {
792    let (category, side, detail) = match &error {
793        LogicError::Protocol(detail) => (ErrorCategory::PeerProtocol, None, detail.clone()),
794        LogicError::ResourceLimit(detail) => (ErrorCategory::LimitExceeded, None, detail.clone()),
795        LogicError::Policy(detail) => (ErrorCategory::Policy, None, detail.clone()),
796        LogicError::MissingStore(side) => (
797            ErrorCategory::Policy,
798            Some(*side),
799            "policy evaluation has no store".to_string(),
800        ),
801        LogicError::Integration(detail) => (ErrorCategory::Integration, None, detail.clone()),
802    };
803    MachineError {
804        category,
805        side,
806        record_hash: None,
807        detail,
808        source: Some(MachineErrorSource::Logic(error)),
809    }
810}
811pub(super) fn store_machine_error(
812    side: InterlaceSide,
813    record_hash: Option<RecordHash>,
814    error: StoreError,
815) -> MachineError {
816    // A store refusing admission at its capacity is a configured local
817    // resource limit, not a store fault.
818    let category = match error {
819        StoreError::CapacityExceeded => ErrorCategory::LimitExceeded,
820        _ => ErrorCategory::Store,
821    };
822    MachineError {
823        category,
824        side: Some(side),
825        record_hash,
826        detail: format!("{error:?}"),
827        source: Some(MachineErrorSource::Store(error)),
828    }
829}
830pub(super) fn peer_protocol(detail: impl Into<String>) -> MachineError {
831    MachineError {
832        category: ErrorCategory::PeerProtocol,
833        side: None,
834        record_hash: None,
835        detail: detail.into(),
836        source: None,
837    }
838}
839pub(super) fn core_machine_error(error: CoreError) -> MachineError {
840    MachineError {
841        category: error.category(),
842        side: None,
843        record_hash: None,
844        detail: error.detail().into(),
845        source: None,
846    }
847}
848pub(super) fn integration_machine_error(detail: impl Into<String>) -> MachineError {
849    MachineError {
850        category: ErrorCategory::Integration,
851        side: None,
852        record_hash: None,
853        detail: detail.into(),
854        source: None,
855    }
856}
857
858#[cfg(test)]
859mod connected_harness;
860#[cfg(test)]
861mod listing_base_tests;
862#[cfg(test)]
863mod listing_exchange_tests;
864#[cfg(test)]
865mod reconciliation_tests;
866#[cfg(test)]
867mod record_ingress_tests;
868#[cfg(test)]
869#[path = "machine/terminal_store_call_tests.rs"]
870mod terminal_store_call_tests;
lace_sans_io/core/machine.rs

lace_sans_io/core/
machine.rs
