lace_rt/
runtime.rs

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

lace_rt/
runtime.rs
