lace_sans_io/
runtime_values.rs

1//! Runtime-neutral application values shared by concrete Lace hosts.
2
3use crate::core::{
4    CloseReason, CoreError, ErrorCategory, InterlaceObservation, InterlaceResourceLimits,
5    InterlaceSide, MachineError, PeerProofAssurance, validate_expected_peer_by,
6};
7use crate::iltp::{PortText, StreamEncoding, StreamOffers};
8use crate::records::{BySecret, ByValue, Record, RecordHash};
9use std::{
10    collections::{HashMap, HashSet},
11    sync::Arc,
12};
13
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub enum ParticipantPosition {
16    First,
17    Second,
18}
19impl ParticipantPosition {
20    pub fn other(self) -> Self {
21        if self == Self::First {
22            Self::Second
23        } else {
24            Self::First
25        }
26    }
27}
28
29impl InterlaceSide {
30    /// The application argument position of this protocol side. A direct route
31    /// (`local` is `None`) maps side0 to `First`; a connected route maps its
32    /// local side to the local participant's position and the peer to the other.
33    #[doc(hidden)]
34    pub fn participant_position(
35        self,
36        local: Option<(InterlaceSide, ParticipantPosition)>,
37    ) -> ParticipantPosition {
38        match local {
39            Some((local_side, local_position)) if self == local_side => local_position,
40            Some((_, local_position)) => local_position.other(),
41            None if self == InterlaceSide::Side0 => ParticipantPosition::First,
42            None => ParticipantPosition::Second,
43        }
44    }
45}
46
47/// A bounded-outcome event; `RecordSaved` reports only the record hash.
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub enum InterlaceEvent {
50    RecordSent {
51        position: ParticipantPosition,
52        record_hash: RecordHash,
53    },
54    RecordSaved {
55        position: ParticipantPosition,
56        record_hash: RecordHash,
57    },
58    FixedPoint,
59    Closed {
60        reason: CloseReason,
61    },
62}
63
64/// An open-stream event. `RecordSaved` carries the admitted owned record;
65/// bounded outcomes project it to the hash-only `InterlaceEvent`.
66#[derive(Debug, Clone, PartialEq, Eq)]
67pub enum OpenEvent {
68    RecordSent {
69        position: ParticipantPosition,
70        record_hash: RecordHash,
71    },
72    RecordSaved {
73        position: ParticipantPosition,
74        record_hash: RecordHash,
75        record: Record,
76    },
77    FixedPoint,
78    Closed {
79        reason: CloseReason,
80    },
81}
82
83impl OpenEvent {
84    #[doc(hidden)]
85    pub fn from_observation(
86        observation: InterlaceObservation,
87        local: Option<(InterlaceSide, ParticipantPosition)>,
88    ) -> Self {
89        match observation {
90            InterlaceObservation::RecordSent {
91                side, record_hash, ..
92            } => Self::RecordSent {
93                position: side.participant_position(local),
94                record_hash,
95            },
96            InterlaceObservation::RecordSaved { side, record, .. } => Self::RecordSaved {
97                position: side.participant_position(local),
98                record_hash: record.hash(),
99                record,
100            },
101        }
102    }
103}
104
105impl From<OpenEvent> for InterlaceEvent {
106    fn from(event: OpenEvent) -> Self {
107        match event {
108            OpenEvent::RecordSent {
109                position,
110                record_hash,
111            } => Self::RecordSent {
112                position,
113                record_hash,
114            },
115            OpenEvent::RecordSaved {
116                position,
117                record_hash,
118                ..
119            } => Self::RecordSaved {
120                position,
121                record_hash,
122            },
123            OpenEvent::FixedPoint => Self::FixedPoint,
124            OpenEvent::Closed { reason } => Self::Closed { reason },
125        }
126    }
127}
128
129#[derive(Debug, Clone, Copy, PartialEq, Eq)]
130pub enum TimeoutPhase {
131    Prefacing,
132    Side0Proof,
133    Side1Proof,
134    Side0Operand,
135    Side1Operand,
136    Advertising,
137    Requesting,
138    Answering,
139    Fencing,
140}
141
142#[derive(Debug, Clone, PartialEq, Eq)]
143pub struct Error {
144    detail: Arc<ErrorDetail>,
145    position: Option<ParticipantPosition>,
146    additional_errors: Vec<Error>,
147}
148
149#[derive(Debug, PartialEq, Eq)]
150struct ErrorDetail {
151    category: ErrorCategory,
152    record_hash: Option<RecordHash>,
153    message: String,
154    phase: Option<TimeoutPhase>,
155}
156
157impl Error {
158    fn new(
159        category: ErrorCategory,
160        position: Option<ParticipantPosition>,
161        record_hash: Option<RecordHash>,
162        message: impl Into<String>,
163    ) -> Self {
164        Self {
165            detail: Arc::new(ErrorDetail {
166                category,
167                record_hash,
168                message: message.into(),
169                phase: None,
170            }),
171            position,
172            additional_errors: Vec::new(),
173        }
174    }
175
176    /// Retain an additional failure without replacing this error's primary detail.
177    /// Clones replay the same failure; independently created equal errors do not.
178    pub fn with_cleanup(self, cleanup: Self) -> Self {
179        Self::combine([self, cleanup]).expect("two failures")
180    }
181
182    #[doc(hidden)]
183    pub fn combine(errors: impl IntoIterator<Item = Self>) -> Option<Self> {
184        let mut errors = errors.into_iter().peekable();
185        let first = errors.next()?;
186        if errors.peek().is_none() {
187            return Some(first);
188        }
189        let mut positions = HashMap::new();
190        let mut combined: Vec<Self> = Vec::new();
191        for mut group in std::iter::once(first).chain(errors) {
192            let additional = std::mem::take(&mut group.additional_errors);
193            // Groups are flat: only this constructor adds additional errors.
194            for error in std::iter::once(group).chain(additional) {
195                let identity = Arc::as_ptr(&error.detail);
196                if let Some(&index) = positions.get(&identity) {
197                    let retained: &mut Self = &mut combined[index];
198                    retained.position = retained.position.or(error.position);
199                } else {
200                    positions.insert(identity, combined.len());
201                    combined.push(error);
202                }
203            }
204        }
205        let mut combined = combined.into_iter();
206        let mut first = combined.next()?;
207        first.additional_errors = combined.collect();
208        Some(first)
209    }
210
211    #[doc(hidden)]
212    pub fn with_position(mut self, position: ParticipantPosition) -> Self {
213        self.position.get_or_insert(position);
214        for error in &mut self.additional_errors {
215            error.position.get_or_insert(position);
216        }
217        self
218    }
219
220    /// Additional failures in occurrence order, each with its own typed detail.
221    /// The primary failure remains available through this error's usual accessors.
222    pub fn additional_errors(&self) -> &[Self] {
223        &self.additional_errors
224    }
225
226    #[doc(hidden)]
227    pub fn peer_protocol(message: impl Into<String>) -> Self {
228        Self::new(ErrorCategory::PeerProtocol, None, None, message)
229    }
230
231    pub fn policy(message: impl Into<String>) -> Self {
232        Self::new(ErrorCategory::Policy, None, None, message)
233    }
234
235    #[doc(hidden)]
236    pub fn transport(message: impl Into<String>) -> Self {
237        Self::new(ErrorCategory::Transport, None, None, message)
238    }
239
240    #[doc(hidden)]
241    pub fn integration(message: impl Into<String>) -> Self {
242        Self::new(ErrorCategory::Integration, None, None, message)
243    }
244
245    #[doc(hidden)]
246    pub fn detailed(
247        category: ErrorCategory,
248        position: Option<ParticipantPosition>,
249        record_hash: Option<RecordHash>,
250        message: impl Into<String>,
251    ) -> Self {
252        Self::new(category, position, record_hash, message)
253    }
254
255    #[doc(hidden)]
256    pub fn store(
257        position: Option<ParticipantPosition>,
258        record_hash: Option<RecordHash>,
259        message: impl Into<String>,
260    ) -> Self {
261        Self::new(ErrorCategory::Store, position, record_hash, message)
262    }
263
264    /// Project a terminal machine error, mapping its protocol side through
265    /// [`InterlaceSide::participant_position`].
266    #[doc(hidden)]
267    pub fn from_machine(
268        error: MachineError,
269        local: Option<(InterlaceSide, ParticipantPosition)>,
270    ) -> Self {
271        Self::new(
272            error.category,
273            error.side.map(|side| side.participant_position(local)),
274            error.record_hash,
275            error.detail,
276        )
277    }
278
279    #[doc(hidden)]
280    pub fn timeout(position: Option<ParticipantPosition>, phase: TimeoutPhase) -> Self {
281        Self {
282            detail: Arc::new(ErrorDetail {
283                category: ErrorCategory::Timeout,
284                record_hash: None,
285                message: format!("interlace phase timed out in {phase:?}"),
286                phase: Some(phase),
287            }),
288            position,
289            additional_errors: Vec::new(),
290        }
291    }
292
293    pub fn category(&self) -> ErrorCategory {
294        self.detail.category
295    }
296    pub fn position(&self) -> Option<ParticipantPosition> {
297        self.position
298    }
299    pub fn record_hash(&self) -> Option<RecordHash> {
300        self.detail.record_hash
301    }
302    pub fn message(&self) -> &str {
303        &self.detail.message
304    }
305    pub fn timeout_phase(&self) -> Option<TimeoutPhase> {
306        self.detail.phase
307    }
308}
309
310impl core::fmt::Display for Error {
311    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
312        write!(f, "{:?}: {}", self.category(), self.message())?;
313        for error in &self.additional_errors {
314            write!(f, "; additional: {error}")?;
315        }
316        Ok(())
317    }
318}
319#[cfg(test)]
320#[path = "runtime_values_tests.rs"]
321mod tests;
322
323impl std::error::Error for Error {
324    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
325        self.additional_errors.first().map(|error| error as _)
326    }
327}
328impl From<CoreError> for Error {
329    fn from(value: CoreError) -> Self {
330        Self::new(value.category(), None, None, value.detail())
331    }
332}
333impl From<crate::logic::PreparationError> for Error {
334    fn from(value: crate::logic::PreparationError) -> Self {
335        let category = if value.is_limit_exceeded() {
336            ErrorCategory::LimitExceeded
337        } else {
338            ErrorCategory::Policy
339        };
340        Self::new(category, None, None, value.to_string())
341    }
342}
343impl From<std::io::Error> for Error {
344    fn from(value: std::io::Error) -> Self {
345        Self::transport(value.to_string())
346    }
347}
348
349#[derive(Debug, Clone, PartialEq, Eq)]
350pub enum Terminal {
351    Complete,
352    StoppedByLimit,
353    Failed(Error),
354}
355
356#[derive(Debug, Clone, PartialEq, Eq)]
357pub struct BoundedInterlaceOutcome {
358    pub terminal: Terminal,
359    pub observations: Vec<InterlaceEvent>,
360    /// The By the live peer proved in the latest logical session of a
361    /// connected route; `None` on a direct pair or without such a proof.
362    pub peer_by: Option<ByValue>,
363    /// The 065 encoding of the local-to-peer direction the latest preface
364    /// pair selected; `None` on a direct pair or before a peer preface.
365    pub outgoing_encoding: Option<StreamEncoding>,
366    /// The 065 encoding of the peer-to-local direction, as above.
367    pub incoming_encoding: Option<StreamEncoding>,
368}
369
370/// The diagnostics a connected route reports in a bounded outcome; a direct
371/// pair reports none.
372#[doc(hidden)]
373#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
374pub struct PeerDiagnostics {
375    pub peer_by: Option<ByValue>,
376    pub outgoing_encoding: Option<StreamEncoding>,
377    pub incoming_encoding: Option<StreamEncoding>,
378}
379
380/// Classify a bounded operation's failure: a lone resource limit stops it;
381/// any other failure, or a limit combined with another failure, fails it.
382#[doc(hidden)]
383pub fn bounded_terminal(error: Error) -> Terminal {
384    if error.category() == ErrorCategory::LimitExceeded && error.additional_errors().is_empty() {
385        Terminal::StoppedByLimit
386    } else {
387        Terminal::Failed(error)
388    }
389}
390
391/// Settle a driven failure, if any, with independent cleanup failures.
392#[doc(hidden)]
393pub fn settled_terminal(
394    primary: Option<Error>,
395    cleanup: impl IntoIterator<Item = Error>,
396) -> Terminal {
397    match (primary, Error::combine(cleanup)) {
398        (Some(primary), Some(cleanup)) => bounded_terminal(primary.with_cleanup(cleanup)),
399        (None, Some(cleanup)) => Terminal::Failed(cleanup),
400        (Some(primary), None) => bounded_terminal(primary),
401        (None, None) => Terminal::Complete,
402    }
403}
404
405/// Build a bounded outcome from collected observations and the route's
406/// diagnostics; only a complete terminal keeps its `FixedPoint`.
407#[doc(hidden)]
408pub fn finish_bounded(
409    (mut observations, primary): (Vec<InterlaceEvent>, Option<Error>),
410    cleanup: impl IntoIterator<Item = Error>,
411    peer: PeerDiagnostics,
412) -> BoundedInterlaceOutcome {
413    let terminal = settled_terminal(primary, cleanup);
414    if !matches!(terminal, Terminal::Complete) {
415        observations.retain(|event| !matches!(event, InterlaceEvent::FixedPoint));
416    }
417    BoundedInterlaceOutcome {
418        terminal,
419        observations,
420        peer_by: peer.peer_by,
421        outgoing_encoding: peer.outgoing_encoding,
422        incoming_encoding: peer.incoming_encoding,
423    }
424}
425
426/// Assign each unique supplied record its Lace-100 store disposition from the
427/// provider's (`First`) `RecordSent` observations and the terminal.
428#[doc(hidden)]
429pub fn store_outcome(
430    outcome: BoundedInterlaceOutcome,
431    supplied: impl IntoIterator<Item = RecordHash>,
432) -> StoreOutcome {
433    let complete = matches!(outcome.terminal, Terminal::Complete);
434    let sent = outcome
435        .observations
436        .iter()
437        .filter_map(|event| match event {
438            InterlaceEvent::RecordSent {
439                position: ParticipantPosition::First,
440                record_hash,
441            } => Some(*record_hash),
442            _ => None,
443        })
444        .collect::<HashSet<_>>();
445    let mut seen = HashSet::new();
446    let records = supplied
447        .into_iter()
448        .filter(|hash| seen.insert(*hash))
449        .map(|record_hash| {
450            let provider_sent = sent.contains(&record_hash);
451            StoreRecordResult {
452                record_hash,
453                disposition: if !complete {
454                    StoreRecordDisposition::Unsettled
455                } else if provider_sent {
456                    StoreRecordDisposition::Saved
457                } else {
458                    StoreRecordDisposition::Skipped
459                },
460                provider_sent,
461                target_saved: complete && provider_sent,
462            }
463        })
464        .collect();
465    StoreOutcome {
466        terminal: outcome.terminal,
467        observations: outcome.observations,
468        records,
469    }
470}
471
472#[derive(Debug, Clone, PartialEq, Eq, Default)]
473pub struct InterlaceOperationOptions {
474    pub limits: InterlaceResourceLimits,
475}
476impl InterlaceOperationOptions {
477    pub fn new() -> Self {
478        Self::default()
479    }
480    pub fn with_limits(mut self, limits: InterlaceResourceLimits) -> Self {
481        self.limits = limits;
482        self
483    }
484}
485
486#[derive(Debug, Clone, PartialEq, Eq)]
487pub struct ListOutcome {
488    pub terminal: Terminal,
489    pub observations: Vec<InterlaceEvent>,
490    pub records: Vec<Record>,
491}
492
493#[derive(Debug, Clone, PartialEq, Eq)]
494pub struct GetOutcome {
495    pub terminal: Terminal,
496    pub observations: Vec<InterlaceEvent>,
497    pub record: Option<Record>,
498}
499
500#[derive(Debug, Clone, PartialEq, Eq)]
501pub enum StoreRecordDisposition {
502    Saved,
503    Skipped,
504    Unsettled,
505}
506
507#[derive(Debug, Clone, PartialEq, Eq)]
508pub struct StoreRecordResult {
509    pub record_hash: RecordHash,
510    pub disposition: StoreRecordDisposition,
511    pub provider_sent: bool,
512    pub target_saved: bool,
513}
514
515#[derive(Debug, Clone, PartialEq, Eq)]
516pub struct StoreOutcome {
517    pub terminal: Terminal,
518    pub observations: Vec<InterlaceEvent>,
519    pub records: Vec<StoreRecordResult>,
520}
521
522#[derive(Debug, Clone, PartialEq, Eq)]
523pub enum OpenBoundary {
524    FixedPoint,
525    Closed(CloseReason),
526}
527
528#[derive(Debug, Clone, PartialEq, Eq)]
529pub struct OpenBoundaryOutcome {
530    pub boundary: OpenBoundary,
531    pub observations: Vec<OpenEvent>,
532}
533
534#[derive(Debug, Clone)]
535pub struct ConnectedPeerOptions {
536    by_secret: Option<BySecret>,
537    peer_proof: PeerProofAssurance,
538    expected_peer_by: Option<ByValue>,
539    stream_encodings: Vec<StreamEncoding>,
540    port: Option<PortText>,
541}
542impl Default for ConnectedPeerOptions {
543    fn default() -> Self {
544        Self {
545            by_secret: None,
546            peer_proof: PeerProofAssurance::default(),
547            expected_peer_by: None,
548            stream_encodings: StreamEncoding::COMPILED.to_vec(),
549            port: None,
550        }
551    }
552}
553impl ConnectedPeerOptions {
554    pub fn new() -> Self {
555        Self::default()
556    }
557    pub fn with_by_secret(mut self, secret: BySecret) -> Self {
558        self.by_secret = Some(secret);
559        self
560    }
561    pub fn with_peer_proof(mut self, value: PeerProofAssurance) -> Self {
562        self.peer_proof = value;
563        self
564    }
565    pub fn with_expected_peer_by(mut self, value: ByValue) -> Self {
566        self.expected_peer_by = Some(value);
567        self
568    }
569    /// One ordered list of compressed stream encodings: the encode preference
570    /// and the decode set offered in every preface. `identity` is implied;
571    /// an empty list offers only `identity`. Defaults to every compiled-in
572    /// encoding, [`StreamEncoding::COMPILED`].
573    pub fn with_stream_encodings(
574        mut self,
575        encodings: impl IntoIterator<Item = StreamEncoding>,
576    ) -> Self {
577        self.stream_encodings = encodings.into_iter().collect();
578        self
579    }
580    /// What this side wants to talk to, sent in every preface of the carrier.
581    pub fn with_port(mut self, port: PortText) -> Self {
582        self.port = Some(port);
583        self
584    }
585    #[doc(hidden)]
586    pub fn validate(&self) -> Result<(), CoreError> {
587        validate_expected_peer_by(self.peer_proof, self.expected_peer_by)?;
588        StreamOffers::new(&self.stream_encodings).map_err(CoreError::InvalidConfig)?;
589        if matches!(self.peer_proof, PeerProofAssurance::ChannelBound(_))
590            && self.by_secret.is_none()
591        {
592            return Err(CoreError::InvalidConfig(
593                "channel-bound peer proof requires a local By-secret",
594            ));
595        }
596        Ok(())
597    }
598    #[doc(hidden)]
599    pub fn into_parts(
600        self,
601    ) -> (
602        Option<BySecret>,
603        PeerProofAssurance,
604        Option<ByValue>,
605        Vec<StreamEncoding>,
606        Option<PortText>,
607    ) {
608        (
609            self.by_secret,
610            self.peer_proof,
611            self.expected_peer_by,
612            self.stream_encodings,
613            self.port,
614        )
615    }
616}
lace_sans_io/runtime_values.rs

lace_sans_io/
runtime_values.rs
