lace_tokio/
runtime.rs

1//! Runtime-owned stores, participant handles, and connected transport leases.
2
3use super::*;
4
5#[path = "work_failure.rs"]
6mod work_failure;
7use work_failure::LaceWorkFailure;
8
9use lace_sans_io::core::OperandCache;
10use lace_sans_io::runtime_support::io_budget::IoReservation;
11use lace_sans_io::runtime_support::record_staging::RecordStaging;
12use lace_sans_io::runtime_support::{StoreBackend, StoreExecution};
13
14/// The adopted store; started work holds a clone through durable classification.
15type AdoptedStore = Arc<Mutex<Box<dyn StoreBackend>>>;
16
17pub(super) const EXTERNAL_POSITION_POLL_INTERVAL: std::time::Duration =
18    std::time::Duration::from_secs(1);
19
20pub struct Lace {
21    pub(super) owner: Arc<LaceOwner>,
22}
23
24pub(super) struct LaceOwner {
25    pub(super) lifecycle: Mutex<LaceLifecycle>,
26    activity: tokio::sync::Notify,
27    store_execution: StoreExecution,
28    position_may_advance_externally: bool,
29    /// Where connected operations stage records received for this store.
30    pub(super) record_staging: Option<RecordStaging>,
31    /// Peer operands this Lace's connected operations received, so a later
32    /// session receives an operand it holds as its hash only (Lace-050).
33    pub(super) operand_cache: OperandCache,
34    polling: Mutex<PositionPolling>,
35    pub(super) store_position: StorePositionSignal,
36    pub(super) mutation: tokio::sync::Mutex<()>,
37}
38
39pub(super) struct LaceLifecycle {
40    store: Option<AdoptedStore>,
41    /// The store's copy buffer, returned once the store is closed.
42    copy_reservation: Option<IoReservation>,
43    pub(super) closing: bool,
44    pub(super) active: usize,
45    retained_failure: Option<Error>,
46    terminal_failure: Option<Error>,
47}
48
49struct PositionPolling {
50    next_poll_at: tokio::time::Instant,
51    poll_in_flight: bool,
52}
53
54struct ActiveStoreWork {
55    owner: Arc<LaceOwner>,
56    active: bool,
57}
58
59pub(super) struct LaceWorkCompletion<T> {
60    value: Option<T>,
61    failure: Arc<LaceWorkFailure>,
62}
63
64impl<T> LaceWorkCompletion<T> {
65    pub(super) fn retire(mut self) -> T {
66        self.failure.retire();
67        self.value.take().expect("Lace work completion value")
68    }
69
70    pub(super) fn acknowledge(mut self) -> T {
71        self.failure.acknowledge();
72        self.value.take().expect("Lace work completion value")
73    }
74}
75
76impl<T> Drop for LaceWorkCompletion<T> {
77    fn drop(&mut self) {
78        self.failure.retire();
79    }
80}
81
82pub(super) struct LaceWorkReceiver<T> {
83    receiver: Option<tokio::sync::oneshot::Receiver<LaceWorkCompletion<T>>>,
84    failure: Arc<LaceWorkFailure>,
85    owner: Arc<LaceOwner>,
86}
87
88impl<T> LaceWorkReceiver<T> {
89    pub(super) fn retire(&self) {
90        self.failure.retire();
91    }
92
93    pub(super) async fn receive_completion(&mut self) -> Result<LaceWorkCompletion<T>> {
94        let completion = self
95            .receiver
96            .as_mut()
97            .ok_or_else(|| Error::integration("Lace work completion was already consumed"))?
98            .await
99            .map_err(|_| Error::integration("Lace-owned store work ended without a completion"))?;
100        self.receiver = None;
101        Ok(completion)
102    }
103
104    pub(super) async fn receive(&mut self) -> Result<T> {
105        self.receive_completion()
106            .await
107            .map(LaceWorkCompletion::acknowledge)
108    }
109
110    pub(super) async fn receive_retired(&mut self) -> Result<T> {
111        self.retire();
112        match self.receive_completion().await {
113            Ok(mut completion) => Ok(completion.value.take().expect("Lace work completion value")),
114            Err(error) => {
115                self.owner.retain_failure(error.clone());
116                Err(error)
117            }
118        }
119    }
120}
121
122impl<T> Drop for LaceWorkReceiver<T> {
123    fn drop(&mut self) {
124        if self.receiver.is_some() {
125            self.retire();
126        }
127    }
128}
129
130impl ActiveStoreWork {
131    fn finish(mut self, late_failure: impl FnOnce() -> Option<Error>) {
132        let mut lifecycle = self.owner.lifecycle.lock().unwrap();
133        if lifecycle.closing
134            && let Some(error) = late_failure()
135        {
136            lifecycle.retained_failure =
137                Error::combine(lifecycle.retained_failure.take().into_iter().chain([error]));
138        }
139        lifecycle.active -= 1;
140        self.active = false;
141        if lifecycle.closing && lifecycle.active == 0 {
142            self.owner.activity.notify_waiters();
143        }
144    }
145}
146
147impl Drop for ActiveStoreWork {
148    fn drop(&mut self) {
149        if !self.active {
150            return;
151        }
152        let mut lifecycle = self.owner.lifecycle.lock().unwrap();
153        lifecycle.active -= 1;
154        if lifecycle.closing && lifecycle.active == 0 {
155            self.owner.activity.notify_waiters();
156        }
157    }
158}
159
160#[derive(Clone, Debug)]
161pub(super) enum LaceStoreState {
162    Position(StorePosition),
163    Failed(Error),
164}
165
166pub(super) struct StorePositionSignal {
167    state: Mutex<LaceStoreState>,
168    sender: watch::Sender<LaceStoreState>,
169}
170
171impl StorePositionSignal {
172    fn new(initial: StorePosition) -> Self {
173        let state = LaceStoreState::Position(initial);
174        let (sender, _) = watch::channel(state.clone());
175        Self {
176            state: Mutex::new(state),
177            sender,
178        }
179    }
180    pub(super) fn subscribe(&self) -> watch::Receiver<LaceStoreState> {
181        self.sender.subscribe()
182    }
183    pub(super) fn latest(&self) -> Result<StorePosition> {
184        match &*self.state.lock().unwrap() {
185            LaceStoreState::Position(position) => Ok(*position),
186            LaceStoreState::Failed(error) => Err(error.clone()),
187        }
188    }
189    fn publish(&self, position: StorePosition) -> Result<()> {
190        let mut state = self.state.lock().unwrap();
191        let previous = match &*state {
192            LaceStoreState::Position(previous) => *previous,
193            LaceStoreState::Failed(error) => return Err(error.clone()),
194        };
195        if position < previous {
196            return Err(Error::integration(format!(
197                "StorePosition publication regressed from {} to {}",
198                previous.raw(),
199                position.raw()
200            )));
201        }
202        if position == previous {
203            return Ok(());
204        }
205        *state = LaceStoreState::Position(position);
206        self.sender.send_replace(state.clone());
207        Ok(())
208    }
209    fn fail(&self, error: Error) {
210        let mut state = self.state.lock().unwrap();
211        let error = match &*state {
212            LaceStoreState::Failed(retained) => {
213                let combined = retained.clone().with_cleanup(error);
214                if &combined == retained {
215                    return;
216                }
217                combined
218            }
219            LaceStoreState::Position(_) => error,
220        };
221        *state = LaceStoreState::Failed(error);
222        self.sender.send_replace(state.clone());
223    }
224}
225
226pub(super) struct StoreExecutionOutcome {
227    /// Submitted to the machine, which owns store errors as terminal data.
228    /// `None` after a failed mandatory position refresh: the runtime fails the
229    /// operation with `failure` instead. A failed publication poisons the Lace
230    /// before the outcome is delivered.
231    pub(super) response: Option<std::result::Result<StoreResult, StoreError>>,
232    /// The operation failure, constructed once so delivery, retirement, and
233    /// Lace close retain one occurrence.
234    pub(super) failure: Option<Error>,
235}
236
237impl LaceOwner {
238    pub(super) fn retain_failure(&self, error: Error) {
239        let mut lifecycle = self.lifecycle.lock().unwrap();
240        lifecycle.retained_failure =
241            Error::combine(lifecycle.retained_failure.take().into_iter().chain([error]));
242    }
243
244    pub(super) fn terminal_failure(&self) -> Option<Error> {
245        self.lifecycle.lock().unwrap().terminal_failure.clone()
246    }
247
248    fn poison(&self, error: Error) -> Error {
249        let failure = {
250            let mut lifecycle = self.lifecycle.lock().unwrap();
251            let failure =
252                Error::combine(lifecycle.terminal_failure.take().into_iter().chain([error]))
253                    .expect("one terminal failure");
254            lifecycle.terminal_failure = Some(failure.clone());
255            failure
256        };
257        self.store_position.fail(failure.clone());
258        failure
259    }
260
261    fn begin_execution(
262        self: &Arc<Self>,
263    ) -> std::result::Result<(AdoptedStore, ActiveStoreWork), Error> {
264        let mut lifecycle = self.lifecycle.lock().unwrap();
265        if let Some(error) = &lifecycle.terminal_failure {
266            return Err(error.clone());
267        }
268        if lifecycle.closing {
269            return Err(Error::store(None, None, "Lace is closing"));
270        }
271        let store = lifecycle
272            .store
273            .as_ref()
274            .ok_or_else(|| Error::store(None, None, "Lace is closed"))?
275            .clone();
276        lifecycle.active += 1;
277        Ok((
278            store,
279            ActiveStoreWork {
280                owner: self.clone(),
281                active: true,
282            },
283        ))
284    }
285
286    pub(super) fn claim(&self) -> Result<()> {
287        let lifecycle = self.lifecycle.lock().unwrap();
288        if let Some(error) = &lifecycle.terminal_failure {
289            return Err(error.clone());
290        }
291        if lifecycle.closing || lifecycle.store.is_none() {
292            return Err(Error::policy("LaceHandle is closed"));
293        }
294        Ok(())
295    }
296
297    fn publish_confirmed_position(&self, position: StorePosition) -> Result<()> {
298        let mut lifecycle = self.lifecycle.lock().unwrap();
299        if let Some(error) = &lifecycle.terminal_failure {
300            return Err(error.clone());
301        }
302        match self.store_position.publish(position) {
303            Ok(()) => Ok(()),
304            Err(error) => {
305                lifecycle.terminal_failure = Some(error.clone());
306                drop(lifecycle);
307                self.store_position.fail(error.clone());
308                Err(error)
309            }
310        }
311    }
312
313    #[cfg(test)]
314    pub(super) fn force_external_poll_due(&self) {
315        self.polling.lock().unwrap().next_poll_at = tokio::time::Instant::now();
316    }
317
318    #[cfg(test)]
319    pub(super) fn external_poll_in_flight(&self) -> bool {
320        self.polling.lock().unwrap().poll_in_flight
321    }
322
323    #[cfg(test)]
324    pub(super) fn publish_position_for_test(&self, position: StorePosition) -> Result<()> {
325        self.publish_confirmed_position(position)
326    }
327
328    #[cfg(test)]
329    pub(super) fn start_store_execution(self: &Arc<Self>, op: StoreOp) -> StoreOpFuture {
330        self.start_store_execution_at(op, None)
331    }
332
333    pub(super) fn start_store_execution_at(
334        self: &Arc<Self>,
335        op: StoreOp,
336        position: Option<ParticipantPosition>,
337    ) -> StoreOpFuture {
338        let (sender, receiver) = tokio::sync::oneshot::channel();
339        let failure = LaceWorkFailure::new(self.clone());
340        let work_failure = failure.clone();
341        let owner = self.clone();
342        match self.begin_execution() {
343            Ok((store, execution)) => {
344                let mutation = matches!(op, StoreOp::StoreRecords { .. });
345                tokio::spawn(async move {
346                    let worker_owner = owner.clone();
347                    let worker = tokio::spawn(async move {
348                        worker_owner.execute_started(store, op, position).await
349                    });
350                    let (outcome, late_failure) = match worker.await {
351                        Ok(outcome) => {
352                            let late_failure = mutation.then(|| outcome.failure.clone()).flatten();
353                            (outcome, late_failure)
354                        }
355                        Err(error) => {
356                            let error = Error::integration(format!(
357                                "Lace-owned store execution task failed: {error}"
358                            ));
359                            (
360                                StoreExecutionOutcome {
361                                    response: None,
362                                    failure: Some(error.clone()),
363                                },
364                                Some(error),
365                            )
366                        }
367                    };
368                    work_failure.complete(late_failure);
369                    let completion = LaceWorkCompletion {
370                        value: Some(outcome),
371                        failure: work_failure.clone(),
372                    };
373                    let _ = sender.send(completion);
374                    execution.finish(|| work_failure.take_retired());
375                });
376            }
377            Err(error) => {
378                let _ = sender.send(LaceWorkCompletion {
379                    value: Some(StoreExecutionOutcome {
380                        response: None,
381                        failure: Some(error),
382                    }),
383                    failure: work_failure,
384                });
385            }
386        }
387        LaceWorkReceiver {
388            receiver: Some(receiver),
389            failure,
390            owner: self.clone(),
391        }
392    }
393
394    async fn execute_started(
395        &self,
396        store: AdoptedStore,
397        op: StoreOp,
398        position: Option<ParticipantPosition>,
399    ) -> StoreExecutionOutcome {
400        // Keep every same-Lace store call outside the mutation-to-authoritative-
401        // position interval. A retired native store's one-shot terminal handoff
402        // must be consumed by this mutation's mandatory refresh.
403        let _store_access = self.mutation.lock().await;
404        let mutation = matches!(op, StoreOp::StoreRecords { .. });
405        let response = {
406            let store = store.clone();
407            run_store(self.store_execution, move || {
408                lock_store(&store)?.execute(op)
409            })
410            .await
411        };
412        let store_error = |error: &StoreError| Error::store(position, None, format!("{error:?}"));
413        if !mutation {
414            let failure = response.as_ref().err().map(store_error);
415            return StoreExecutionOutcome {
416                response: Some(response),
417                failure,
418            };
419        }
420        match run_store(self.store_execution, move || {
421            lock_store(&store)?.current_position()
422        })
423        .await
424        {
425            Ok(confirmed) => {
426                let publication = self.publish_confirmed_position(confirmed).err();
427                let failure = Error::combine(
428                    publication
429                        .into_iter()
430                        .chain(response.as_ref().err().map(store_error)),
431                );
432                StoreExecutionOutcome {
433                    response: Some(response),
434                    failure,
435                }
436            }
437            Err(refresh) => StoreExecutionOutcome {
438                failure: Error::combine(
439                    [Error::store(
440                        position,
441                        None,
442                        format!("failed to establish CurrentStorePosition: {refresh:?}"),
443                    )]
444                    .into_iter()
445                    .chain(response.as_ref().err().map(store_error)),
446                ),
447                response: None,
448            },
449        }
450    }
451
452    #[cfg(test)]
453    pub(super) async fn execute(self: &Arc<Self>, op: StoreOp) -> StoreExecutionOutcome {
454        let mut work = self.start_store_execution_at(op, None);
455        work.receive()
456            .await
457            .unwrap_or_else(|error| StoreExecutionOutcome {
458                response: None,
459                failure: Some(error),
460            })
461    }
462
463    pub(super) fn coordinate_external_position_poll(
464        self: &Arc<Self>,
465        now: tokio::time::Instant,
466    ) -> Result<Option<tokio::time::Instant>> {
467        if !self.position_may_advance_externally {
468            return Ok(None);
469        }
470        let mut lifecycle = self.lifecycle.lock().unwrap();
471        if let Some(error) = &lifecycle.terminal_failure {
472            return Err(error.clone());
473        }
474        if lifecycle.closing || lifecycle.store.is_none() {
475            return Err(Error::store(None, None, "Lace is closing"));
476        }
477        let mut polling = self.polling.lock().unwrap();
478        if now < polling.next_poll_at {
479            return Ok(Some(polling.next_poll_at));
480        }
481        if polling.poll_in_flight {
482            return Ok(Some(now + EXTERNAL_POSITION_POLL_INTERVAL));
483        }
484
485        let store = lifecycle.store.as_ref().expect("checked store").clone();
486        lifecycle.active += 1;
487        let execution = ActiveStoreWork {
488            owner: self.clone(),
489            active: true,
490        };
491        polling.poll_in_flight = true;
492        polling.next_poll_at = now + EXTERNAL_POSITION_POLL_INTERVAL;
493        drop(polling);
494        drop(lifecycle);
495
496        let owner = self.clone();
497        tokio::spawn(async move {
498            let worker_owner = owner.clone();
499            let worker = tokio::spawn(async move {
500                let _mutation = worker_owner.mutation.lock().await;
501                let execution = worker_owner.store_execution;
502                match run_store(execution, move || lock_store(&store)?.current_position()).await {
503                    Ok(position) => worker_owner.publish_confirmed_position(position),
504                    Err(error) => {
505                        let error = store_runtime_error(error);
506                        Err(worker_owner.poison(error))
507                    }
508                }
509            });
510            let result = worker.await.unwrap_or_else(|error| {
511                Err(Error::integration(format!(
512                    "Lace-owned position refresh task failed: {error}"
513                )))
514            });
515            if let Err(error) = result {
516                owner.poison(error);
517            }
518            execution.finish(|| None);
519            {
520                let mut polling = owner.polling.lock().unwrap();
521                polling.poll_in_flight = false;
522                polling.next_poll_at =
523                    tokio::time::Instant::now() + EXTERNAL_POSITION_POLL_INTERVAL;
524            }
525            owner.activity.notify_waiters();
526        });
527        Ok(Some(now + EXTERNAL_POSITION_POLL_INTERVAL))
528    }
529
530    pub(super) async fn settle_external_position_poll(&self) -> Option<Error> {
531        loop {
532            let notified = self.activity.notified();
533            if !self.polling.lock().unwrap().poll_in_flight {
534                break;
535            }
536            notified.await;
537        }
538        self.terminal_failure()
539    }
540
541    async fn close(self: &Arc<Self>) -> Result<()> {
542        {
543            let mut lifecycle = self.lifecycle.lock().unwrap();
544            lifecycle.closing = true;
545        }
546        loop {
547            let notified = self.activity.notified();
548            if self.lifecycle.lock().unwrap().active == 0 {
549                break;
550            }
551            notified.await;
552        }
553        let (store, copy_reservation, terminal_failure, retained_failure) = {
554            let mut lifecycle = self.lifecycle.lock().unwrap();
555            (
556                lifecycle.store.take(),
557                lifecycle.copy_reservation.take(),
558                lifecycle.terminal_failure.clone(),
559                lifecycle.retained_failure.take(),
560            )
561        };
562        let close_result = match store {
563            Some(store) => match Arc::try_unwrap(store) {
564                Ok(store) => run_store(self.store_execution, move || {
565                    let closed = store
566                        .into_inner()
567                        .unwrap_or_else(std::sync::PoisonError::into_inner)
568                        .close();
569                    drop(copy_reservation);
570                    closed
571                })
572                .await
573                .map_err(store_runtime_error),
574                Err(_) => Err(Error::store(
575                    None,
576                    None,
577                    "store execution remained after Lace close drain",
578                )),
579            },
580            None => Ok(()),
581        };
582        Error::combine(
583            [terminal_failure, retained_failure, close_result.err()]
584                .into_iter()
585                .flatten(),
586        )
587        .map_or(Ok(()), Err)
588    }
589}
590
591/// Run store work inline or on a blocking worker, as the backend requires.
592async fn run_store<T: Send + 'static>(
593    execution: StoreExecution,
594    work: impl FnOnce() -> std::result::Result<T, StoreError> + Send + 'static,
595) -> std::result::Result<T, StoreError> {
596    match execution {
597        StoreExecution::Inline => work(),
598        StoreExecution::Blocking => match tokio::task::spawn_blocking(work).await {
599            Ok(result) => result,
600            // A store panic fails its caller exactly as inline work would.
601            Err(error) => match error.try_into_panic() {
602                Ok(panic) => std::panic::resume_unwind(panic),
603                Err(error) => Err(StoreError::StoreFailed(format!(
604                    "store task failed: {error}"
605                ))),
606            },
607        },
608    }
609}
610
611fn lock_store(
612    store: &AdoptedStore,
613) -> std::result::Result<std::sync::MutexGuard<'_, Box<dyn StoreBackend>>, StoreError> {
614    store
615        .lock()
616        .map_err(|_| StoreError::StoreFailed("store lock is poisoned".into()))
617}
618
619fn store_runtime_error(error: StoreError) -> Error {
620    Error::store(None, None, format!("{error:?}"))
621}
622
623pub(super) type StoreOpFuture = LaceWorkReceiver<StoreExecutionOutcome>;
624
625pub(super) fn connected_inputs(
626    options: ConnectedPeerOptions,
627) -> Result<Box<ConnectedCarrierInputs>> {
628    options.validate()?;
629    let (by_secret, peer_proof, expected_peer_by, stream_encodings, port) = options.into_parts();
630    let nonce_seed = lace_sans_io::core::NonceSeed::generate().map_err(|error| {
631        Error::integration(format!(
632            "failed to generate ILTP carrier nonce seed: {error}"
633        ))
634    })?;
635    Ok(Box::new(ConnectedCarrierInputs {
636        by_secret,
637        expected_peer_by,
638        peer_proof,
639        stream_encodings,
640        port,
641        nonce_seed,
642    }))
643}
644
645impl Lace {
646    /// Exclusively open a filesystem root until close or final owner drop.
647    /// Share Lace handles instead of opening the same root again.
648    #[cfg(all(target_pointer_width = "64", any(unix, windows)))]
649    pub fn filesystem(path: impl AsRef<std::path::Path>) -> Result<Self> {
650        let store =
651            lace_sans_io::runtime_support::native_filesystem_store::NativeFilesystemStore::open(
652                path,
653            )
654            .map_err(|error| store_runtime_error(error.into()))?;
655        Self::try_from_store(Box::new(store), false)
656    }
657    /// Exclusively open a portable file store root until close or final
658    /// owner drop, creating it with `format` when the root is empty or
659    /// absent; an existing root keeps the format it was created with.
660    /// Share Lace handles instead of opening the same root again.
661    pub fn portable(
662        path: impl AsRef<std::path::Path>,
663        format: crate::PortableStoreFormat,
664    ) -> Result<Self> {
665        let store = lace_sans_io::runtime_support::portable_file_store::PortableFileStore::open(
666            path, format,
667        )
668        .map_err(store_runtime_error)?;
669        Self::try_from_store(Box::new(store), false)
670    }
671    /// Open an LMDB store with authoritative external-position polling.
672    #[cfg(feature = "lmdb")]
673    pub fn lmdb(path: impl AsRef<std::path::Path>) -> Result<Self> {
674        let store = lmdb_lace::NativeLmdbStore::open(path).map_err(store_runtime_error)?;
675        Self::try_from_store(Box::new(store), true)
676    }
677    pub(super) fn try_from_store(
678        mut store: Box<dyn StoreBackend>,
679        position_may_advance_externally: bool,
680    ) -> Result<Self> {
681        let initial = store.current_position().map_err(store_runtime_error)?;
682        let store_execution = store.execution();
683        let chunk_bytes = crate::io::IO.chunk_bytes();
684        let record_staging = store.adopt(chunk_bytes);
685        let copy_reservation = match record_staging {
686            Some(_) => Some(crate::io::IO.reserve(chunk_bytes)?),
687            None => None,
688        };
689        Ok(Self {
690            owner: Arc::new(LaceOwner {
691                lifecycle: Mutex::new(LaceLifecycle {
692                    store: Some(Arc::new(Mutex::new(store))),
693                    copy_reservation,
694                    closing: false,
695                    active: 0,
696                    retained_failure: None,
697                    terminal_failure: None,
698                }),
699                activity: tokio::sync::Notify::new(),
700                store_execution,
701                position_may_advance_externally,
702                record_staging,
703                operand_cache: OperandCache::default(),
704                polling: Mutex::new(PositionPolling {
705                    next_poll_at: tokio::time::Instant::now() + EXTERNAL_POSITION_POLL_INTERVAL,
706                    poll_in_flight: false,
707                }),
708                store_position: StorePositionSignal::new(initial),
709                mutation: tokio::sync::Mutex::new(()),
710            }),
711        })
712    }
713    #[cfg(test)]
714    pub(super) fn from_store(store: Box<dyn StoreBackend>) -> Self {
715        Self::try_from_store(store, false).expect("test store must report its initial position")
716    }
717    #[cfg(test)]
718    pub(super) fn from_externally_advancing_store(store: Box<dyn StoreBackend>) -> Self {
719        Self::try_from_store(store, true).expect("test store must report its initial position")
720    }
721    pub fn memory(records: impl IntoIterator<Item = Record>) -> Self {
722        Self::try_from_store(
723            Box::new(InMemoryStore::source(records.into_iter().collect())),
724            false,
725        )
726        .expect("in-memory store must report its initial position")
727    }
728    pub fn empty_memory() -> Self {
729        Self::memory(Vec::<Record>::new())
730    }
731    /// A memory Lace holding at most `bytes` of canonical record bytes,
732    /// counted once per record, so a duplicate holds nothing; indexes are not
733    /// counted. Seeding `records` beyond it fails with `LimitExceeded`,
734    /// as does the operation whose admission would exceed it, before anything
735    /// is admitted; other operations continue. Close releases the held bytes;
736    /// a record a reader still holds stays readable on its own.
737    pub fn memory_with_capacity(
738        records: impl IntoIterator<Item = Record>,
739        bytes: usize,
740    ) -> Result<Self> {
741        let mut store = InMemoryStore::with_capacity(bytes);
742        store.seed(records.into_iter().collect()).map_err(|_| {
743            Error::detailed(
744                ErrorCategory::LimitExceeded,
745                None,
746                None,
747                format!("the records exceed the memory Lace's capacity of {bytes} bytes"),
748            )
749        })?;
750        Self::try_from_store(Box::new(store), false)
751    }
752    /// Hold at most `entries` received peer operands, most recently used
753    /// first (default [`OperandCache::DEFAULT_ENTRIES`]); `0` holds none, so
754    /// every peer operand crosses as a Blob.
755    pub fn with_operand_cache(self, entries: usize) -> Self {
756        self.owner.operand_cache.set_capacity(entries);
757        self
758    }
759    pub fn handle(&self) -> LaceHandle {
760        LaceHandle {
761            owner: self.owner.clone(),
762        }
763    }
764    /// The bytes a memory Lace with a capacity holds now, counted as its
765    /// constructor describes; `None` for any other store or once closed. Reads
766    /// the store between its operations, so it is meant for memory stores.
767    pub fn held_bytes(&self) -> Option<usize> {
768        let store = self.owner.lifecycle.lock().unwrap().store.clone()?;
769        let guard = store.lock().unwrap();
770        guard.held_bytes()
771    }
772
773    /// Stop new store access, drain already-started work, and release the
774    /// adopted store object.
775    pub async fn close(self) -> Result<()> {
776        self.owner.close().await
777    }
778}
779
780/// A clonable capability to one store-backed [`Lace`].
781#[derive(Clone)]
782pub struct LaceHandle {
783    pub(super) owner: Arc<LaceOwner>,
784}
785
786/// A move-only connected participant over host-established raw ILTP IO.
787pub struct ConnectedPeer {
788    pub(super) lease: ConnectedTransportLease,
789    pub(super) inputs: Box<ConnectedCarrierInputs>,
790}
791
792/// A clonable local participant that supplies its own prepared operand.
793///
794/// In an interlace, a bare [`LaceHandle`] position receives the operation's
795/// policy and a `LocalPeer` position receives this operand. At most one
796/// position may be a `LocalPeer`, and it cannot join a [`ConnectedPeer`].
797#[derive(Clone)]
798pub struct LocalPeer {
799    handle: LaceHandle,
800    operand: InterlaceModule,
801}
802impl LocalPeer {
803    /// Prepare `policy` as this participant's operand without performing IO.
804    pub fn new(handle: LaceHandle, policy: impl crate::IntoPolicy) -> Result<Self> {
805        Ok(Self {
806            handle,
807            operand: policy.into_policy()?,
808        })
809    }
810    pub(super) fn operand(&self) -> &InterlaceModule {
811        &self.operand
812    }
813}
814
815/// One application participant supplied to an interlace operation.
816pub enum Participant {
817    Local(LaceHandle),
818    LocalPeer(LocalPeer),
819    Connected(ConnectedPeer),
820}
821impl From<LaceHandle> for Participant {
822    fn from(value: LaceHandle) -> Self {
823        Self::Local(value)
824    }
825}
826impl From<LocalPeer> for Participant {
827    fn from(value: LocalPeer) -> Self {
828        Self::LocalPeer(value)
829    }
830}
831impl From<ConnectedPeer> for Participant {
832    fn from(value: ConnectedPeer) -> Self {
833        Self::Connected(value)
834    }
835}
836impl Participant {
837    pub(super) fn is_local(&self) -> bool {
838        !matches!(self, Self::Connected(_))
839    }
840}
841
842pub(super) struct ConnectedCarrierInputs {
843    pub(super) by_secret: Option<BySecret>,
844    pub(super) expected_peer_by: Option<crate::ByValue>,
845    pub(super) peer_proof: PeerProofAssurance,
846    pub(super) stream_encodings: Vec<lace_sans_io::iltp::StreamEncoding>,
847    pub(super) port: Option<crate::PortText>,
848    pub(super) nonce_seed: lace_sans_io::core::NonceSeed,
849}
850
851pub(super) struct ConnectedTransportLease {
852    pub(super) transport: Box<transport::ByteStreamCarrier>,
853    pub(super) role: CarrierRole,
854}
855
856pub(super) enum ParticipantLease {
857    Local(LaceHandle),
858    Connected {
859        peer: Box<ConnectedTransportLease>,
860        inputs: Box<ConnectedCarrierInputs>,
861    },
862}
863pub(super) fn claim_participant(participant: Participant) -> Result<ParticipantLease> {
864    match participant {
865        Participant::Local(handle) | Participant::LocalPeer(LocalPeer { handle, .. }) => {
866            handle.owner.claim()?;
867            Ok(ParticipantLease::Local(handle))
868        }
869        Participant::Connected(peer) => Ok(ParticipantLease::Connected {
870            peer: Box::new(peer.lease),
871            inputs: peer.inputs,
872        }),
873    }
874}
875
876pub(super) fn empty_direct_authorities() -> [SessionAuthority; 2] {
877    [SessionAuthority::default(), SessionAuthority::default()]
878}
879
880pub(super) enum OperationConfig {
881    Direct(Box<DirectConfig>),
882    Connected {
883        local_operand: lace_sans_io::logic::InterlaceModule,
884        limits: InterlaceResourceLimits,
885    },
886}
887
888/// Chooses each logical session's local Lace and operand for the peer
889/// preface's port; the Lace of the first session serves every later one.
890pub(super) type Selector =
891    Box<dyn FnMut(Option<&crate::PortText>) -> Option<(LaceHandle, crate::SessionOperand)> + Send>;
892
893/// A connected operation's local side: one Lace and operand for every
894/// session, or a selector that chooses the Lace with the first session's
895/// operand and each later session's operand.
896pub(super) enum ConnectedLocal {
897    Fixed(LaceHandle, lace_sans_io::logic::InterlaceModule),
898    Selected(Selector),
899}
lace_tokio/runtime.rs

lace_tokio/
runtime.rs
