crux_core/bridge/
registry.rs

1use std::collections::HashMap;
2use std::sync::Mutex;
3
4use facet::Facet;
5use serde::{Deserialize, Serialize};
6
7use super::{BridgeError, FfiFormat, Request};
8use crate::bridge::request_serde::ResolveSerialized;
9use crate::{EffectFFI, ResolveError};
10
11/// Identifies one request across the FFI boundary, for as long as anything
12/// could still refer to it.
13///
14/// Ids are issued in ascending order and are not reused when a request
15/// completes, so an id that has been resolved stays unusable rather than being
16/// handed to some unrelated later request.
17#[allow(clippy::unsafe_derive_deserialize)]
18#[derive(Facet, Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
19#[serde(transparent)]
20#[facet(transparent)]
21pub struct EffectId(pub u32);
22
23pub struct ResolveRegistry<T: FfiFormat>(Mutex<Outstanding<T>>);
24
25/// The requests the shell could still resolve, keyed by the id it was given.
26struct Outstanding<T: FfiFormat> {
27    entries: HashMap<u32, ResolveSerialized<T>>,
28    next_id: u32,
29}
30
31impl<T: FfiFormat> Outstanding<T> {
32    /// Issue the next id.
33    ///
34    /// Ids ascend rather than filling gaps, so resolving a completed request is
35    /// a lookup miss instead of a hit on whichever request happened to inherit
36    /// its storage. The counter wraps after `u32::MAX` requests; ids still
37    /// outstanding are stepped over, so a live request can never be displaced
38    /// even then.
39    fn issue_id(&mut self) -> u32 {
40        loop {
41            let id = self.next_id;
42            self.next_id = self.next_id.wrapping_add(1);
43
44            if !self.entries.contains_key(&id) {
45                return id;
46            }
47        }
48    }
49}
50
51impl<T: FfiFormat> Default for ResolveRegistry<T> {
52    fn default() -> Self {
53        Self(Mutex::new(Outstanding {
54            entries: HashMap::new(),
55            next_id: 0,
56        }))
57    }
58}
59
60impl<T: FfiFormat> ResolveRegistry<T> {
61    /// Register an effect for future continuation, when it has been processed
62    /// and output given back to the core.
63    ///
64    /// The `effect` will be serialized into its FFI counterpart before being stored
65    /// and wrapped in a [`Request`].
66    ///
67    /// # Panics
68    ///
69    /// Panics if the internal mutex has been poisoned
70    // ANCHOR: register
71    pub fn register<Eff>(&self, effect: Eff) -> Request<Eff::Ffi>
72    where
73        Eff: EffectFFI,
74    {
75        let (effect, resolve) = effect.serialize();
76
77        let id = {
78            let mut outstanding = self.0.lock().expect("Registry Mutex poisoned.");
79            let id = outstanding.issue_id();
80
81            // A request that cannot be resolved has nothing worth keeping: storing
82            // one would add an entry per fire-and-forget effect — every render, for
83            // the life of the process — that nothing would ever remove.
84            if !matches!(resolve, ResolveSerialized::Never) {
85                outstanding.entries.insert(id, resolve);
86            }
87
88            id
89        };
90
91        Request {
92            id: EffectId(id),
93            effect,
94        }
95    }
96    // ANCHOR_END: register
97
98    /// Resume a previously registered effect.
99    ///
100    /// Fails with [`ResolveError::NotFound`] if `id` is not outstanding —
101    /// because it was never issued, because it has already been resolved, or
102    /// because it belongs to a request that never expected a response.
103    ///
104    /// # Errors
105    ///
106    /// Returns `BridgeError` if the stored request could not be resolved.
107    ///
108    /// # Panics
109    ///
110    /// Panics if the internal mutex has been poisoned
111    pub fn resume(&self, id: EffectId, response: &[u8]) -> Result<(), BridgeError<T>> {
112        let mut outstanding = self.0.lock().expect("Registry Mutex poisoned");
113
114        let Some(entry) = outstanding.entries.get_mut(&id.0) else {
115            return Err(BridgeError::ProcessResponse(ResolveError::NotFound(
116                id.0.into(),
117            )));
118        };
119
120        let resolved = entry.resolve(response);
121
122        // A `Once` turns itself into a `Never` as it resolves: the request is
123        // finished, and its id will not be issued again, so drop it.
124        if matches!(entry, ResolveSerialized::Never) {
125            outstanding.entries.remove(&id.0);
126        }
127
128        resolved
129    }
130}
131
132#[cfg(test)]
133mod tests {
134    use super::{Outstanding, ResolveSerialized};
135    use crate::bridge::JsonFfiFormat;
136    use std::collections::HashMap;
137
138    fn outstanding(next_id: u32) -> Outstanding<JsonFfiFormat> {
139        Outstanding {
140            entries: HashMap::new(),
141            next_id,
142        }
143    }
144
145    #[test]
146    fn ids_ascend_and_are_never_reused() {
147        let mut outstanding = outstanding(0);
148
149        let ids: Vec<_> = (0..4).map(|_| outstanding.issue_id()).collect();
150        assert_eq!(ids, vec![0, 1, 2, 3]);
151
152        // Finishing a request frees its entry, but not its id.
153        outstanding.entries.remove(&1);
154
155        assert_eq!(outstanding.issue_id(), 4);
156    }
157
158    #[test]
159    fn wrapping_steps_over_outstanding_ids() {
160        let mut outstanding = outstanding(u32::MAX);
161
162        // Still awaiting a response on 0 and 1 when the counter comes round.
163        outstanding.entries.insert(0, ResolveSerialized::Never);
164        outstanding.entries.insert(1, ResolveSerialized::Never);
165
166        assert_eq!(outstanding.issue_id(), u32::MAX);
167        assert_eq!(
168            outstanding.issue_id(),
169            2,
170            "wrapping displaced a request that was still outstanding"
171        );
172    }
173}
crux_core/bridge/registry.rs

crux_core/bridge/
registry.rs
