lace_rt/
operations.rs

1use crate::driver::Driver;
2use crate::policy::{all_records, private_operands};
3use crate::runtime::*;
4use crate::*;
5use lace_sans_io::core::{
6    DirectConfig, InterlaceResourceLimits, SessionAuthority, StoreOp, StoreRecordOutcome,
7    StoreResult, validate_operand_blob_size,
8};
9use lace_sans_io::logic::{rx_disabled, tx_disabled};
10use lace_sans_io::runtime_values::{OpenEvent, PeerDiagnostics, finish_bounded, store_outcome};
11
12#[cfg(test)]
13#[path = "store_stream_tests.rs"]
14mod store_stream_tests;
15#[cfg(test)]
16#[path = "operations_tests.rs"]
17mod tests;
18
19#[must_use = "drive or drop the open interlace"]
20pub struct OpenInterlace {
21    pub(crate) driver: Driver,
22}
23pub struct ForEachStream {
24    pub(crate) interlace: OpenInterlace,
25    observer: Lace,
26}
27/// The open counterpart of [`store`]: offers records one at a time; see
28/// [`store_stream`].
29pub struct StoreStream {
30    interlace: OpenInterlace,
31    provider: Lace,
32    /// The failure that ended the stream, returned by later calls.
33    failure: Option<Error>,
34}
35
36/// One route for the bare-position operand pair by participant position. A
37/// `LocalPeer` position supplies its own operand instead; direct places both in
38/// slot order; connected keeps the local participant's operand.
39pub(crate) fn route(
40    first: Participant,
41    second: Participant,
42    mut operands: [InterlaceModule; 2],
43    limits: InterlaceResourceLimits,
44) -> Result<OpenInterlace> {
45    let local_peer = match (&first, &second) {
46        (Participant::LocalPeer(_), Participant::LocalPeer(_)) => {
47            return Err(Error::policy("an interlace accepts at most one LocalPeer"));
48        }
49        (Participant::LocalPeer(_), Participant::Connected(_))
50        | (Participant::Connected(_), Participant::LocalPeer(_)) => {
51            return Err(Error::policy(
52                "a LocalPeer cannot interlace with a ConnectedPeer",
53            ));
54        }
55        (Participant::LocalPeer(peer), _) => Some((0, peer.operand().clone())),
56        (_, Participant::LocalPeer(peer)) => Some((1, peer.operand().clone())),
57        _ => None,
58    };
59    if let Some((slot, operand)) = &local_peer {
60        operands[*slot] = operand.clone();
61    }
62    let config = if first.is_local() && second.is_local() {
63        OperationConfig::Direct(Box::new(
64            DirectConfig::new(
65                operands,
66                [SessionAuthority::default(), SessionAuthority::default()],
67            )
68            .with_limits(limits),
69        ))
70    } else {
71        let [first_operand, second_operand] = operands;
72        let local_operand = if first.is_local() {
73            first_operand
74        } else {
75            second_operand
76        };
77        validate_operand_blob_size(&local_operand)?;
78        OperationConfig::Connected {
79            local_operand,
80            limits,
81        }
82    };
83    Ok(OpenInterlace {
84        driver: Driver::new(claim(first)?, claim(second)?, config, local_peer)?,
85    })
86}
87fn exhausted() -> Error {
88    Error::integration("interlace exhausted before fixed point")
89}
90/// Drive to the bounded terminal, projecting public events and handing each
91/// `First` admitted record to `saved`.
92fn drive_bounded(
93    driver: &mut Driver,
94    mut saved: impl FnMut(Record),
95) -> (Vec<InterlaceEvent>, Option<Error>) {
96    let mut observations = Vec::new();
97    let error = loop {
98        match driver.next_event() {
99            Ok(Some(OpenEvent::FixedPoint)) => match driver.settle_route() {
100                Ok(()) => {
101                    observations.push(InterlaceEvent::FixedPoint);
102                    break None;
103                }
104                Err(error) => break Some(error),
105            },
106            Ok(Some(OpenEvent::Closed { .. })) | Ok(None) => break Some(exhausted()),
107            Ok(Some(OpenEvent::RecordSaved {
108                position,
109                record_hash,
110                record,
111            })) => {
112                observations.push(InterlaceEvent::RecordSaved {
113                    position,
114                    record_hash,
115                });
116                if position == ParticipantPosition::First {
117                    saved(record);
118                }
119            }
120            Ok(Some(event)) => observations.push(event.into()),
121            Err(error) => break Some(error),
122        }
123    };
124    (observations, error)
125}
126
127/// Drive a private capture against `target`, handing each record admitted to the
128/// capture to `saved`, then close the capture.
129fn capture(
130    capture: Lace,
131    target: Participant,
132    policy: impl IntoPolicy,
133    limits: InterlaceResourceLimits,
134    mut saved: impl FnMut(Record),
135) -> Result<BoundedInterlaceOutcome> {
136    let operands = private_operands(policy.into_policy()?, tx_disabled())?;
137    let first = Participant::from(capture.handle());
138    let mut driver = route(first, target, operands, limits)?.driver;
139    let collected = drive_bounded(&mut driver, &mut saved);
140    drop(driver);
141    Ok(finish_bounded(
142        collected,
143        capture.close().err(),
144        PeerDiagnostics::default(),
145    ))
146}
147
148pub fn bounded_interlace(
149    first: impl Into<Participant>,
150    second: impl Into<Participant>,
151    policy: impl IntoPolicy,
152) -> Result<BoundedInterlaceOutcome> {
153    bounded_interlace_with_options(first, second, policy, InterlaceOperationOptions::default())
154}
155pub fn bounded_interlace_with_options(
156    first: impl Into<Participant>,
157    second: impl Into<Participant>,
158    policy: impl IntoPolicy,
159    operation: InterlaceOperationOptions,
160) -> Result<BoundedInterlaceOutcome> {
161    let module = policy.into_policy()?;
162    let operands = [module.clone(), module];
163    let mut driver = route(first.into(), second.into(), operands, operation.limits)?.driver;
164    let collected = drive_bounded(&mut driver, drop);
165    Ok(finish_bounded(collected, None, driver.peer_diagnostics()))
166}
167pub fn open_interlace(
168    first: impl Into<Participant>,
169    second: impl Into<Participant>,
170    policy: impl IntoPolicy,
171) -> Result<OpenInterlace> {
172    open_interlace_with_options(first, second, policy, InterlaceOperationOptions::default())
173}
174pub fn open_interlace_with_options(
175    first: impl Into<Participant>,
176    second: impl Into<Participant>,
177    policy: impl IntoPolicy,
178    operation: InterlaceOperationOptions,
179) -> Result<OpenInterlace> {
180    let module = policy.into_policy()?;
181    let operands = [module.clone(), module];
182    route(first.into(), second.into(), operands, operation.limits)
183}
184/// Open an interlace between a local Lace `selector` chooses, position First,
185/// and `peer`, position Second. Every logical session takes its operand and
186/// optional expected peer By from `selector`, called with the port of that
187/// session's peer preface; the first call also chooses the Lace, which serves
188/// the whole carrier, so a later call must return it again. `None`, or
189/// another Lace, refuses the session as `PeerProtocol`. Until the first
190/// selection the operation owns the transport and its prefacing deadline but
191/// no Lace. `peer` must not carry an expected peer By, and the operation takes
192/// no policy replacement.
193pub fn open_interlace_with_selector(
194    peer: ConnectedPeer,
195    selector: impl FnMut(Option<&PortText>) -> Option<(LaceHandle, SessionOperand)> + Send + 'static,
196    operation: InterlaceOperationOptions,
197) -> Result<OpenInterlace> {
198    let peer = claim(peer.into())?;
199    Ok(OpenInterlace {
200        driver: Driver::selecting(peer, Box::new(selector), operation.limits)?,
201    })
202}
203impl OpenInterlace {
204    /// The port the peer's latest preface sent; `None` on a direct pair,
205    /// before the first peer preface, or when the peer sent none.
206    pub fn peer_port(&self) -> Option<PortText> {
207        self.driver.peer_port()
208    }
209    /// The By the live peer proved in the latest logical session under
210    /// confidential or channel-bound assurance, for diagnostics; `None` on a
211    /// direct pair, before that proof validates, or without one.
212    pub fn peer_by(&self) -> Option<ByValue> {
213        self.driver.peer_by()
214    }
215    /// The 065 encoding of the local-to-peer direction the latest preface
216    /// pair selected, for diagnostics; `None` on a direct pair or before the
217    /// first peer preface.
218    pub fn outgoing_encoding(&self) -> Option<StreamEncoding> {
219        self.driver.outgoing_encoding()
220    }
221    /// The 065 encoding of the peer-to-local direction, as
222    /// [`Self::outgoing_encoding`].
223    pub fn incoming_encoding(&self) -> Option<StreamEncoding> {
224        self.driver.incoming_encoding()
225    }
226    pub fn replace_policy(&mut self, policy: impl IntoPolicy) -> Result<()> {
227        let module = policy.into_policy()?;
228        self.driver.replace_policy(module)
229    }
230    pub fn next_event(&mut self) -> Result<Option<OpenEvent>> {
231        self.driver.next_event()
232    }
233    /// Cancel this operation, settle started work and connected transport
234    /// ownership, and return its final local-close boundary.
235    pub fn close(self) -> Result<OpenBoundaryOutcome> {
236        let reason = self.driver.close_cancelled()?;
237        Ok(OpenBoundaryOutcome {
238            boundary: OpenBoundary::Closed(reason.clone()),
239            observations: vec![OpenEvent::Closed { reason }],
240        })
241    }
242    pub fn run_until_fixed_point(&mut self) -> Result<OpenBoundaryOutcome> {
243        let mut observations = Vec::new();
244        loop {
245            match self.next_event()? {
246                Some(event @ OpenEvent::FixedPoint) => {
247                    observations.push(event);
248                    return Ok(OpenBoundaryOutcome {
249                        boundary: OpenBoundary::FixedPoint,
250                        observations,
251                    });
252                }
253                Some(OpenEvent::Closed { reason }) => {
254                    observations.push(OpenEvent::Closed {
255                        reason: reason.clone(),
256                    });
257                    return Ok(OpenBoundaryOutcome {
258                        boundary: OpenBoundary::Closed(reason),
259                        observations,
260                    });
261                }
262                Some(event) => observations.push(event),
263                None => return Err(exhausted()),
264            }
265        }
266    }
267}
268
269pub fn list(target: impl Into<Participant>, policy: impl IntoPolicy) -> Result<ListOutcome> {
270    list_with_options(target, policy, InterlaceOperationOptions::default())
271}
272pub fn list_with_options(
273    target: impl Into<Participant>,
274    policy: impl IntoPolicy,
275    operation: InterlaceOperationOptions,
276) -> Result<ListOutcome> {
277    let mut records = Vec::new();
278    let outcome = capture(
279        Lace::empty_memory(),
280        target.into(),
281        policy,
282        operation.limits,
283        |record| records.push(record),
284    )?;
285    Ok(ListOutcome {
286        terminal: outcome.terminal,
287        observations: outcome.observations,
288        records,
289    })
290}
291pub fn get(target: impl Into<Participant>, policy: impl IntoPolicy) -> Result<GetOutcome> {
292    get_with_options(target, policy, InterlaceOperationOptions::default())
293}
294pub fn get_with_options(
295    target: impl Into<Participant>,
296    policy: impl IntoPolicy,
297    operation: InterlaceOperationOptions,
298) -> Result<GetOutcome> {
299    let mut first = None;
300    let outcome = capture(
301        Lace::empty_memory(),
302        target.into(),
303        policy,
304        operation.limits,
305        |record| {
306            first.get_or_insert(record);
307        },
308    )?;
309    Ok(GetOutcome {
310        terminal: outcome.terminal,
311        observations: outcome.observations,
312        record: first,
313    })
314}
315
316pub fn store(
317    target: impl Into<Participant>,
318    records: impl IntoRecordBatch,
319) -> Result<StoreOutcome> {
320    store_with_options(target, records, InterlaceOperationOptions::default())
321}
322pub fn store_with_options(
323    target: impl Into<Participant>,
324    records: impl IntoRecordBatch,
325    operation: InterlaceOperationOptions,
326) -> Result<StoreOutcome> {
327    store_inner(target.into(), records.into_records(), operation.limits)
328}
329fn store_inner(
330    target: Participant,
331    records: Vec<Record>,
332    limits: InterlaceResourceLimits,
333) -> Result<StoreOutcome> {
334    if records.is_empty() {
335        return Ok(StoreOutcome {
336            terminal: Terminal::Complete,
337            observations: Vec::new(),
338            records: Vec::new(),
339        });
340    }
341    let supplied = records
342        .iter()
343        .map(|record| record.hash())
344        .collect::<Vec<_>>();
345    let operands = private_operands(all_records().clone(), rx_disabled())?;
346    let source = Lace::memory(records);
347    let first = Participant::from(source.handle());
348    let mut driver = route(first, target, operands, limits)?.driver;
349    let collected = drive_bounded(&mut driver, drop);
350    drop(driver);
351    Ok(store_outcome(
352        finish_bounded(collected, source.close().err(), PeerDiagnostics::default()),
353        supplied,
354    ))
355}
356
357/// Open `store`'s interlace against `target` with an empty private provider
358/// and drive it to its first fixed point; then [`StoreStream::offer`] each
359/// record.
360pub fn store_stream(target: impl Into<Participant>) -> Result<StoreStream> {
361    store_stream_with_options(target, InterlaceOperationOptions::default())
362}
363pub fn store_stream_with_options(
364    target: impl Into<Participant>,
365    operation: InterlaceOperationOptions,
366) -> Result<StoreStream> {
367    let target = target.into();
368    let operands = private_operands(all_records().clone(), rx_disabled())?;
369    let provider = Lace::empty_memory();
370    let first = Participant::from(provider.handle());
371    let interlace = match route(first, target, operands, operation.limits) {
372        Ok(interlace) => interlace,
373        Err(error) => {
374            return Err(Error::combine([error].into_iter().chain(provider.close().err())).unwrap());
375        }
376    };
377    let mut stream = StoreStream {
378        interlace,
379        provider,
380        failure: None,
381    };
382    match stream.fixed_point(None) {
383        Ok(_) => Ok(stream),
384        Err(error) => {
385            stream.failure = Some(error.clone());
386            Err(stream.close().err().unwrap_or(error))
387        }
388    }
389}
390impl StoreStream {
391    /// Store `record` into the provider and drive to the next fixed point:
392    /// `Saved` when the provider sent it, `Skipped` when it did not or the
393    /// provider already held it. An error ends the stream; later calls
394    /// return it, and records saved before it stay saved.
395    pub fn offer(&mut self, record: Record) -> Result<StoreRecordDisposition> {
396        if let Some(error) = &self.failure {
397            return Err(error.clone());
398        }
399        let result = self.offer_inner(record);
400        if let Err(error) = &result {
401            self.failure = Some(error.clone());
402        }
403        result
404    }
405
406    fn offer_inner(&mut self, record: Record) -> Result<StoreRecordDisposition> {
407        let hash = record.hash();
408        let outcome = self.provider.owner.execute(StoreOp::StoreRecords {
409            records: vec![record],
410        });
411        if let Some(error) = outcome.failure {
412            return Err(error);
413        }
414        match outcome.response {
415            Some(Ok(StoreResult::StoredRecords { outcomes, .. })) => match outcomes.as_slice() {
416                [StoreRecordOutcome::Duplicate { .. }] => Ok(StoreRecordDisposition::Skipped),
417                [StoreRecordOutcome::Stored { .. }] => Ok(if self.fixed_point(Some(hash))? {
418                    StoreRecordDisposition::Saved
419                } else {
420                    StoreRecordDisposition::Skipped
421                }),
422                _ => Err(Error::integration(
423                    "provider store returned the wrong outcomes",
424                )),
425            },
426            _ => Err(Error::integration(
427                "provider store returned the wrong result",
428            )),
429        }
430    }
431
432    /// Drive to the next fixed point; whether the provider sent `hash`
433    /// before it.
434    fn fixed_point(&mut self, hash: Option<RecordHash>) -> Result<bool> {
435        let mut sent = false;
436        loop {
437            match self.interlace.next_event()? {
438                Some(OpenEvent::FixedPoint) => return Ok(sent),
439                Some(OpenEvent::RecordSent {
440                    position: ParticipantPosition::First,
441                    record_hash,
442                }) => sent |= Some(record_hash) == hash,
443                Some(OpenEvent::Closed { reason }) => {
444                    return Err(Error::peer_protocol(format!(
445                        "store stream closed before an offer's fixed point: {reason:?}"
446                    )));
447                }
448                Some(_) => {}
449                None => return Err(exhausted()),
450            }
451        }
452    }
453
454    /// Cancel the interlace and close the provider, returning the failure
455    /// that ended the stream with any cleanup failure.
456    pub fn close(self) -> Result<()> {
457        let StoreStream {
458            interlace,
459            provider,
460            failure,
461        } = self;
462        let interlace = interlace.close().err();
463        let provider = provider.close().err();
464        Error::combine(failure.into_iter().chain(interlace).chain(provider)).map_or(Ok(()), Err)
465    }
466}
467
468pub fn for_each(target: impl Into<Participant>, policy: impl IntoPolicy) -> Result<ForEachStream> {
469    for_each_with_options(target, policy, InterlaceOperationOptions::default())
470}
471pub fn for_each_with_options(
472    target: impl Into<Participant>,
473    policy: impl IntoPolicy,
474    operation: InterlaceOperationOptions,
475) -> Result<ForEachStream> {
476    let target = target.into();
477    let observer = Lace::empty_memory();
478    let operands = private_operands(policy.into_policy()?, tx_disabled())?;
479    let first = Participant::from(observer.handle());
480    Ok(ForEachStream {
481        interlace: route(first, target, operands, operation.limits)?,
482        observer,
483    })
484}
485impl ForEachStream {
486    /// Cancel the observer operation and return its final local-close event.
487    pub fn close(self) -> Result<OpenEvent> {
488        let ForEachStream {
489            interlace,
490            observer,
491        } = self;
492        let interlace_result = interlace.driver.close_cancelled();
493        let observer_result = observer.close();
494        if let Some(error) = Error::combine(
495            interlace_result
496                .as_ref()
497                .err()
498                .cloned()
499                .into_iter()
500                .chain(observer_result.err()),
501        ) {
502            return Err(error);
503        }
504        Ok(OpenEvent::Closed {
505            reason: interlace_result?,
506        })
507    }
508
509    pub fn next_event(&mut self) -> Result<Option<OpenEvent>> {
510        loop {
511            match self.interlace.driver.next_event()? {
512                Some(
513                    event @ (OpenEvent::RecordSaved {
514                        position: ParticipantPosition::First,
515                        ..
516                    }
517                    | OpenEvent::FixedPoint
518                    | OpenEvent::Closed { .. }),
519                ) => return Ok(Some(event)),
520                Some(_) => {}
521                None => return Ok(None),
522            }
523        }
524    }
525}
lace_rt/operations.rs

lace_rt/
operations.rs
