saluki_core/runtime/state/
dataspace.rs

1//! A type-erased, async-aware dataspace registry for inter-process coordination.
2//!
3//! The [`DataspaceRegistry`] allows processes to assert and retract typed values by identifier, and subscribe to
4//! receive notifications of those assertions and retractions. Multiple subscribers can observe the same updates.
5//!
6//! - **Assertion**: a value of type `T` becomes available, associated with a given identifier.
7//! - **Retraction**: the value of type `T` associated with a given identifier is withdrawn.
8//!
9//! Subscribers can listen for updates matching a specific identifier, a prefix, or all identifiers for a given type.
10//!
11//! This enables decoupled coordination where processes don't need to know about each other, only the identifier and
12//! type of the values they're exchanging.
13//!
14//! # Example
15//!
16//! ```
17//! use saluki_core::runtime::state::{AssertionUpdate, Identifier, IdentifierFilter, DataspaceRegistry};
18//!
19//! # #[tokio::main]
20//! # async fn main() {
21//! let registry = DataspaceRegistry::new();
22//! let id = Identifier::named("my_value");
23//!
24//! // Subscribe before asserting:
25//! let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
26//!
27//! // Assert a value:
28//! registry.assert(42u32, id.clone());
29//!
30//! // Receive the assertion:
31//! let value = sub.recv().await;
32//! assert_eq!(value, Some(AssertionUpdate::Asserted(id, 42)));
33//! # }
34//! ```
35
36use std::{
37    any::{Any, TypeId},
38    collections::{HashMap, HashSet, VecDeque},
39    hash::Hash,
40    sync::{Arc, Mutex},
41};
42
43use tokio::sync::broadcast;
44
45use super::{Identifier, IdentifierFilter};
46use crate::runtime::process::Id;
47
48const DEFAULT_CHANNEL_CAPACITY: usize = 16;
49
50tokio::task_local! {
51    pub(crate) static CURRENT_DATASPACE: DataspaceRegistry;
52}
53
54/// An update received by a subscription, indicating that a value was asserted or retracted.
55#[derive(Clone, Debug, PartialEq, Eq)]
56pub enum AssertionUpdate<T> {
57    /// A value was asserted (made available), along with the identifier it's associated with.
58    Asserted(Identifier, T),
59
60    /// The value associated with the given identifier was retracted (withdrawn).
61    Retracted(Identifier),
62}
63
64/// Internal key combining type and identifier.
65#[derive(Clone, PartialEq, Eq, Hash)]
66struct StorageKey {
67    type_id: TypeId,
68    identifier: Identifier,
69}
70
71impl StorageKey {
72    fn new<T: 'static>(identifier: Identifier) -> Self {
73        Self {
74            type_id: TypeId::of::<T>(),
75            identifier,
76        }
77    }
78}
79
80/// A type-erased broadcast channel that supports sending retraction notifications without knowing `T`.
81trait AnyChannel: Send + Sync {
82    /// Sends a retraction for the given identifier.
83    fn send_retraction(&self, id: &Identifier);
84
85    /// Returns a downcasted reference to `self` that can be fallibly upcasted back to the original type.
86    fn as_any(&self) -> &dyn Any;
87}
88
89/// Concrete implementation of [`AnyChannel`] that wraps a `broadcast::Sender<AssertionUpdate<T>>`.
90struct TypedChannel<T: Clone + Send + Sync + 'static> {
91    tx: broadcast::Sender<AssertionUpdate<T>>,
92}
93
94impl<T: Clone + Send + Sync + 'static> AnyChannel for TypedChannel<T> {
95    fn send_retraction(&self, id: &Identifier) {
96        let _ = self.tx.send(AssertionUpdate::Retracted(id.clone()));
97    }
98
99    fn as_any(&self) -> &dyn Any {
100        self
101    }
102}
103
104/// A type-erased filtered subscription channel.
105struct FilteredChannel {
106    type_id: TypeId,
107    filter: IdentifierFilter,
108    sender: Box<dyn AnyChannel>,
109}
110
111/// A stored assertion value along with the process that owns it.
112struct StoredValue {
113    value: Box<dyn Any + Send + Sync>,
114    owner: Id,
115}
116
117/// Internal registry state protected by a mutex.
118struct RegistryState {
119    /// Broadcast senders for exact-match subscriptions, keyed by (type, identifier).
120    channels: HashMap<StorageKey, Box<dyn AnyChannel>>,
121
122    /// Filtered subscription channels that are evaluated on every assert/retract.
123    filtered_channels: Vec<FilteredChannel>,
124
125    /// Current assertion values for each (type, identifier) pair, stored type-erased.
126    ///
127    /// Entries are removed on retraction. Used to replay current state to new subscribers.
128    current_values: HashMap<StorageKey, StoredValue>,
129
130    /// Tracks which storage keys each process has asserted, for automatic retraction on process exit.
131    process_assertions: HashMap<Id, HashSet<StorageKey>>,
132
133    /// Default capacity for new broadcast channels.
134    channel_capacity: usize,
135}
136
137impl RegistryState {
138    fn new(channel_capacity: usize) -> Self {
139        Self {
140            channels: HashMap::new(),
141            filtered_channels: Vec::new(),
142            current_values: HashMap::new(),
143            process_assertions: HashMap::new(),
144            channel_capacity,
145        }
146    }
147
148    /// Gets or creates a broadcast sender for the given key, returning a new receiver.
149    fn get_or_create_exact_sender<T>(&mut self, key: StorageKey) -> broadcast::Receiver<AssertionUpdate<T>>
150    where
151        T: Clone + Send + Sync + 'static,
152    {
153        let channel = self.channels.entry(key).or_insert_with(|| {
154            let (tx, _) = broadcast::channel::<AssertionUpdate<T>>(self.channel_capacity);
155            Box::new(TypedChannel { tx })
156        });
157
158        let typed = channel
159            .as_any()
160            .downcast_ref::<TypedChannel<T>>()
161            // `StorageKey` includes `TypeId::of::<T>()`, so a channel stored under a
162            // given key is always a `TypedChannel<T>` for the same `T`.
163            .unwrap_or_else(|| unreachable!("type mismatch in dataspace registry"));
164
165        typed.tx.subscribe()
166    }
167
168    /// Creates a new filtered subscription channel, returning a new receiver.
169    fn create_filtered_sender<T>(&mut self, filter: IdentifierFilter) -> broadcast::Receiver<AssertionUpdate<T>>
170    where
171        T: Clone + Send + Sync + 'static,
172    {
173        let (tx, rx) = broadcast::channel::<AssertionUpdate<T>>(self.channel_capacity);
174
175        self.filtered_channels.push(FilteredChannel {
176            type_id: TypeId::of::<T>(),
177            filter,
178            sender: Box::new(TypedChannel { tx }),
179        });
180
181        rx
182    }
183
184    /// Sends a retraction notification on all channels (exact + filtered) matching the given key.
185    fn notify_retraction(&self, key: &StorageKey) {
186        if let Some(ch) = self.channels.get(key) {
187            ch.send_retraction(&key.identifier);
188        }
189
190        for filtered in &self.filtered_channels {
191            if filtered.type_id == key.type_id && filtered.filter.matches(&key.identifier) {
192                filtered.sender.send_retraction(&key.identifier);
193            }
194        }
195    }
196}
197
198/// Shared inner state of the registry.
199struct DataspaceRegistryInner {
200    state: Mutex<RegistryState>,
201}
202
203/// A dataspace registry for async coordination between processes.
204///
205/// The registry stores broadcast channels indexed by type and [`Identifier`]. Processes can subscribe to receive
206/// assertion and retraction updates for a given type and identifier filter, and other processes can assert or retract
207/// values that are delivered to all matching subscribers.
208///
209/// # Thread Safety
210///
211/// `DataspaceRegistry` is `Clone` and can be safely shared across threads and tasks. All operations are thread-safe.
212#[derive(Clone)]
213pub struct DataspaceRegistry {
214    inner: Arc<DataspaceRegistryInner>,
215}
216
217impl Default for DataspaceRegistry {
218    fn default() -> Self {
219        Self::new()
220    }
221}
222
223impl DataspaceRegistry {
224    /// Creates a new empty registry with the default channel capacity.
225    pub fn new() -> Self {
226        Self::with_channel_capacity(DEFAULT_CHANNEL_CAPACITY)
227    }
228
229    /// Returns the dataspace registry for the current supervision tree, if one exists.
230    pub fn try_current() -> Option<Self> {
231        CURRENT_DATASPACE.try_with(|ds| ds.clone()).ok()
232    }
233
234    /// Creates a new empty registry with the given channel capacity for broadcast channels.
235    pub fn with_channel_capacity(capacity: usize) -> Self {
236        Self {
237            inner: Arc::new(DataspaceRegistryInner {
238                state: Mutex::new(RegistryState::new(capacity)),
239            }),
240        }
241    }
242
243    /// Asserts a value with the given identifier, notifying all matching subscribers.
244    ///
245    /// The assertion is automatically associated with the current process, and will be automatically retracted when
246    /// that process exists. Only the owning process may update an existing assertion for a given type/identifier
247    /// combination.
248    pub fn assert<T>(&self, value: T, id: impl Into<Identifier>)
249    where
250        T: Clone + Send + Sync + 'static,
251    {
252        let id = id.into();
253        let key = StorageKey::new::<T>(id.clone());
254        let caller = Id::current();
255        let mut state = self.inner.state.lock().unwrap();
256
257        // If an assertion already exists for this key, only the owning process may update it.
258        if let Some(existing) = state.current_values.get(&key) {
259            debug_assert_eq!(
260                existing.owner, caller,
261                "process {caller:?} attempted to update assertion owned by {:?}",
262                existing.owner
263            );
264            if existing.owner != caller {
265                return;
266            }
267        }
268
269        // Store the current value for future subscribers, along with the owning process.
270        state.current_values.insert(
271            key.clone(),
272            StoredValue {
273                value: Box::new(value.clone()),
274                owner: caller,
275            },
276        );
277
278        // Track this assertion against the owning process.
279        state.process_assertions.entry(caller).or_default().insert(key.clone());
280
281        // Notify exact-match subscribers.
282        if let Some(ch) = state.channels.get(&key) {
283            if let Some(typed) = ch.as_any().downcast_ref::<TypedChannel<T>>() {
284                let _ = typed.tx.send(AssertionUpdate::Asserted(id.clone(), value.clone()));
285            }
286        }
287
288        // Notify filtered subscribers.
289        let type_id = TypeId::of::<T>();
290        for channel in &state.filtered_channels {
291            if channel.type_id == type_id && channel.filter.matches(&id) {
292                if let Some(typed) = channel.sender.as_any().downcast_ref::<TypedChannel<T>>() {
293                    let _ = typed.tx.send(AssertionUpdate::Asserted(id.clone(), value.clone()));
294                }
295            }
296        }
297    }
298
299    /// Retracts the value of the given type and identifier, notifying all matching subscribers.
300    ///
301    /// Only the process that originally asserted the value may retract it.
302    pub fn retract<T>(&self, id: impl Into<Identifier>)
303    where
304        T: Clone + Send + Sync + 'static,
305    {
306        let id = id.into();
307        let key = StorageKey::new::<T>(id.clone());
308        let caller = Id::current();
309        let mut state = self.inner.state.lock().unwrap();
310
311        // Check that the assertion exists and is owned by the calling process.
312        let Some(stored) = state.current_values.get(&key) else {
313            return;
314        };
315
316        debug_assert_eq!(
317            stored.owner, caller,
318            "process {caller:?} attempted to retract assertion owned by {:?}",
319            stored.owner
320        );
321        if stored.owner != caller {
322            return;
323        }
324
325        // Remove the stored value and clean up process tracking.
326        state.current_values.remove(&key);
327
328        if let Some(keys) = state.process_assertions.get_mut(&caller) {
329            keys.remove(&key);
330            if keys.is_empty() {
331                state.process_assertions.remove(&caller);
332            }
333        }
334
335        state.notify_retraction(&key);
336    }
337
338    /// Retracts all assertions made by the given process.
339    ///
340    /// This is called automatically when a process exits (via [`FutureProcess`] drop) to ensure that no stale
341    /// assertions remain in the registry after the owning process is gone.
342    pub(crate) fn retract_all_for_process(&self, process_id: Id) {
343        let mut state = self.inner.state.lock().unwrap();
344
345        let Some(keys) = state.process_assertions.remove(&process_id) else {
346            return;
347        };
348
349        for key in keys {
350            state.current_values.remove(&key);
351            state.notify_retraction(&key);
352        }
353    }
354
355    /// This is a synchronous point-in-time read that locks the registry, collects matching values,
356    /// and returns immediately without creating any subscription or channel. Use this when you need
357    /// a snapshot of what is currently asserted rather than ongoing notifications of future changes.
358    ///
359    pub fn current_values<T>(&self, filter: IdentifierFilter) -> Vec<T>
360    where
361        T: Clone + Send + Sync + 'static,
362    {
363        let type_id = TypeId::of::<T>();
364        let state = self.inner.state.lock().unwrap();
365        state
366            .current_values
367            .iter()
368            .filter(|(key, _)| key.type_id == type_id && filter.matches(&key.identifier))
369            .filter_map(|(_, stored)| stored.value.downcast_ref::<T>().cloned())
370            .collect()
371    }
372
373    /// Subscribes to assertion and retraction updates matching the given filter.
374    ///
375    /// Returns a [`Subscription`] that can be used to asynchronously receive updates. Any
376    /// assertions that match the filter at the time of subscribing will be immediately replayed
377    /// into the subscription's pending queue.
378    pub fn subscribe<T>(&self, filter: IdentifierFilter) -> Subscription<T>
379    where
380        T: Clone + Send + Sync + 'static,
381    {
382        let mut state = self.inner.state.lock().unwrap();
383
384        match filter {
385            IdentifierFilter::Exact(ref id) => {
386                let key = StorageKey::new::<T>(id.clone());
387                let rx = state.get_or_create_exact_sender::<T>(key.clone());
388
389                // Replay current value if one exists.
390                let pending: VecDeque<_> = state
391                    .current_values
392                    .get(&key)
393                    .and_then(|stored| stored.value.downcast_ref::<T>())
394                    .map(|value| AssertionUpdate::Asserted(id.clone(), value.clone()))
395                    .into_iter()
396                    .collect();
397
398                Subscription { pending, rx }
399            }
400            filter @ (IdentifierFilter::All | IdentifierFilter::Prefix(_)) => {
401                let rx = state.create_filtered_sender::<T>(filter.clone());
402
403                // Replay all matching current values.
404                let type_id = TypeId::of::<T>();
405                let pending: VecDeque<_> = state
406                    .current_values
407                    .iter()
408                    .filter(|(key, _)| key.type_id == type_id && filter.matches(&key.identifier))
409                    .filter_map(|(key, stored)| {
410                        stored
411                            .value
412                            .downcast_ref::<T>()
413                            .map(|value| AssertionUpdate::Asserted(key.identifier.clone(), value.clone()))
414                    })
415                    .collect();
416
417                Subscription { pending, rx }
418            }
419        }
420    }
421}
422
423/// A subscription to updates for a specific type/identifier combination.
424pub struct Subscription<T> {
425    pending: VecDeque<AssertionUpdate<T>>,
426    rx: broadcast::Receiver<AssertionUpdate<T>>,
427}
428
429impl<T> Subscription<T>
430where
431    T: Clone + Send + Sync + 'static,
432{
433    /// Receives the next assertion or retraction update.
434    ///
435    /// Returns `Some(update)` when an update is available, or `None` if the channel has been closed (all senders
436    /// dropped). If messages were missed due to the subscriber falling behind, the missed messages are skipped and the
437    /// next available update is returned.
438    pub async fn recv(&mut self) -> Option<AssertionUpdate<T>> {
439        // TODO: Switch to bounded MPSC channels for delivering assertion/retraction updates.
440
441        if let Some(value) = self.pending.pop_front() {
442            return Some(value);
443        }
444
445        loop {
446            match self.rx.recv().await {
447                Ok(value) => return Some(value),
448                Err(broadcast::error::RecvError::Lagged(_)) => continue,
449                Err(broadcast::error::RecvError::Closed) => return None,
450            }
451        }
452    }
453}
454
455#[cfg(test)]
456mod tests {
457    use std::future::pending;
458
459    use tokio_test::{assert_pending, assert_ready, assert_ready_eq, task::spawn as test_spawn};
460
461    use super::*;
462    use crate::runtime::process::Process;
463
464    #[test]
465    fn subscribe_then_assert() {
466        let registry = DataspaceRegistry::new();
467        let id = Identifier::numeric(1);
468
469        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
470        registry.assert(42u32, id.clone());
471
472        let mut recv = test_spawn(sub.recv());
473        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id, 42)));
474    }
475
476    #[test]
477    fn multiple_subscribers_receive_same_assertion() {
478        let registry = DataspaceRegistry::new();
479        let id = Identifier::numeric(1);
480
481        let mut sub1 = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
482        let mut sub2 = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
483
484        registry.assert(42u32, id.clone());
485
486        let mut recv1 = test_spawn(sub1.recv());
487        assert_ready_eq!(recv1.poll(), Some(AssertionUpdate::Asserted(id.clone(), 42)));
488
489        let mut recv2 = test_spawn(sub2.recv());
490        assert_ready_eq!(recv2.poll(), Some(AssertionUpdate::Asserted(id, 42)));
491    }
492
493    #[test]
494    fn assert_without_subscribers_stores_value() {
495        let registry = DataspaceRegistry::new();
496        let id = Identifier::numeric(1);
497
498        // Assert before any subscriber exists -- the value is stored.
499        registry.assert(42u32, id.clone());
500
501        // A later subscriber should receive the current value.
502        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
503        let mut recv = test_spawn(sub.recv());
504        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id, 42)));
505    }
506
507    #[test]
508    fn different_types_same_identifier() {
509        let registry = DataspaceRegistry::new();
510        let id = Identifier::numeric(1);
511
512        let mut sub_u32 = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
513        let mut sub_string = registry.subscribe::<String>(IdentifierFilter::exact(id.clone()));
514
515        registry.assert(42u32, id.clone());
516        registry.assert("hello".to_string(), id.clone());
517
518        let mut recv_u32 = test_spawn(sub_u32.recv());
519        assert_ready_eq!(recv_u32.poll(), Some(AssertionUpdate::Asserted(id.clone(), 42)));
520
521        let mut recv_string = test_spawn(sub_string.recv());
522        assert_ready_eq!(
523            recv_string.poll(),
524            Some(AssertionUpdate::Asserted(id, "hello".to_string()))
525        );
526    }
527
528    #[test]
529    fn process_identifier() {
530        let registry = DataspaceRegistry::new();
531        let pid = crate::runtime::process::Id::new();
532        let id = Identifier::from(pid);
533
534        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
535        registry.assert(42u32, id.clone());
536
537        let mut recv = test_spawn(sub.recv());
538        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id, 42)));
539    }
540
541    #[test]
542    fn channel_closed_returns_none() {
543        let registry = DataspaceRegistry::new();
544        let id = Identifier::numeric(1);
545
546        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id));
547
548        // Drop the registry, which drops the Arc. Since we only have one reference, the sender is dropped.
549        drop(registry);
550
551        let mut recv = test_spawn(sub.recv());
552        assert_ready_eq!(recv.poll(), None);
553    }
554
555    #[test]
556    fn lagged_subscriber_recovers() {
557        // Create a registry with a tiny buffer so we can force lag.
558        let registry = DataspaceRegistry::with_channel_capacity(2);
559        let id = Identifier::numeric(1);
560
561        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
562
563        // Assert more values than the channel can hold.
564        for i in 0..10 {
565            registry.assert(i as u32, id.clone());
566        }
567
568        // The subscriber should skip lagged messages and still receive a value.
569        let mut recv = test_spawn(sub.recv());
570        let value = assert_ready!(recv.poll());
571        assert!(value.is_some());
572    }
573
574    #[test]
575    fn multiple_values_received_in_order() {
576        let registry = DataspaceRegistry::new();
577        let id = Identifier::numeric(1);
578
579        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
580
581        registry.assert(1u32, id.clone());
582        registry.assert(2u32, id.clone());
583        registry.assert(3u32, id.clone());
584
585        let mut recv = test_spawn(sub.recv());
586        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id.clone(), 1)));
587        drop(recv);
588
589        let mut recv = test_spawn(sub.recv());
590        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id.clone(), 2)));
591        drop(recv);
592
593        let mut recv = test_spawn(sub.recv());
594        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id, 3)));
595    }
596
597    #[test]
598    fn assert_then_retract() {
599        let registry = DataspaceRegistry::new();
600        let id = Identifier::numeric(1);
601
602        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
603
604        registry.assert(42u32, id.clone());
605        registry.retract::<u32>(id.clone());
606
607        let mut recv = test_spawn(sub.recv());
608        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id.clone(), 42)));
609        drop(recv);
610
611        let mut recv = test_spawn(sub.recv());
612        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Retracted(id)));
613    }
614
615    #[test]
616    fn retract_without_subscribers_succeeds() {
617        let registry = DataspaceRegistry::new();
618        let id = Identifier::numeric(1);
619
620        // Retract without any subscribers -- should not panic.
621        registry.retract::<u32>(id);
622    }
623
624    #[test]
625    fn multiple_subscribers_receive_retraction() {
626        let registry = DataspaceRegistry::new();
627        let id = Identifier::numeric(1);
628
629        let mut sub1 = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
630        let mut sub2 = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
631
632        registry.assert(42u32, id.clone());
633        registry.retract::<u32>(id.clone());
634
635        // Drain assertion notifications.
636        let mut recv1 = test_spawn(sub1.recv());
637        assert_ready_eq!(recv1.poll(), Some(AssertionUpdate::Asserted(id.clone(), 42)));
638        drop(recv1);
639
640        let mut recv2 = test_spawn(sub2.recv());
641        assert_ready_eq!(recv2.poll(), Some(AssertionUpdate::Asserted(id.clone(), 42)));
642        drop(recv2);
643
644        // Check retraction notifications.
645        let mut recv1 = test_spawn(sub1.recv());
646        assert_ready_eq!(recv1.poll(), Some(AssertionUpdate::Retracted(id.clone())));
647
648        let mut recv2 = test_spawn(sub2.recv());
649        assert_ready_eq!(recv2.poll(), Some(AssertionUpdate::Retracted(id)));
650    }
651
652    #[test]
653    fn assert_retract_assert_sequence() {
654        let registry = DataspaceRegistry::new();
655        let id = Identifier::numeric(1);
656
657        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
658
659        registry.assert(1u32, id.clone());
660        registry.retract::<u32>(id.clone());
661        registry.assert(2u32, id.clone());
662
663        let mut recv = test_spawn(sub.recv());
664        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id.clone(), 1)));
665        drop(recv);
666
667        let mut recv = test_spawn(sub.recv());
668        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Retracted(id.clone())));
669        drop(recv);
670
671        let mut recv = test_spawn(sub.recv());
672        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id, 2)));
673    }
674
675    #[test]
676    fn all_filter_receives_from_multiple_identifiers() {
677        let registry = DataspaceRegistry::new();
678        let id1 = Identifier::numeric(1);
679        let id2 = Identifier::numeric(2);
680
681        let mut sub = registry.subscribe::<u32>(IdentifierFilter::all());
682
683        registry.assert(1u32, id1.clone());
684        registry.assert(2u32, id2.clone());
685
686        let mut recv = test_spawn(sub.recv());
687        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id1, 1)));
688        drop(recv);
689
690        let mut recv = test_spawn(sub.recv());
691        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id2, 2)));
692    }
693
694    #[test]
695    fn all_filter_receives_retraction() {
696        let registry = DataspaceRegistry::new();
697        let id = Identifier::numeric(1);
698
699        let mut sub = registry.subscribe::<u32>(IdentifierFilter::all());
700
701        registry.assert(42u32, id.clone());
702        registry.retract::<u32>(id.clone());
703
704        let mut recv = test_spawn(sub.recv());
705        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id.clone(), 42)));
706        drop(recv);
707
708        let mut recv = test_spawn(sub.recv());
709        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Retracted(id)));
710    }
711
712    #[test]
713    fn all_filter_and_exact_both_receive() {
714        let registry = DataspaceRegistry::new();
715        let id = Identifier::numeric(1);
716
717        let mut specific = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
718        let mut all = registry.subscribe::<u32>(IdentifierFilter::all());
719
720        registry.assert(42u32, id.clone());
721
722        let mut recv_specific = test_spawn(specific.recv());
723        assert_ready_eq!(recv_specific.poll(), Some(AssertionUpdate::Asserted(id.clone(), 42)));
724
725        let mut recv_all = test_spawn(all.recv());
726        assert_ready_eq!(recv_all.poll(), Some(AssertionUpdate::Asserted(id, 42)));
727    }
728
729    #[test]
730    fn all_filter_created_before_exact_channels() {
731        let registry = DataspaceRegistry::new();
732
733        // Subscribe to all identifiers before any specific-identifier activity exists.
734        let mut all = registry.subscribe::<u32>(IdentifierFilter::all());
735
736        let id = Identifier::numeric(1);
737        registry.assert(42u32, id.clone());
738
739        let mut recv = test_spawn(all.recv());
740        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id, 42)));
741    }
742
743    #[test]
744    fn subscribe_receives_current_value() {
745        let registry = DataspaceRegistry::new();
746        let id = Identifier::numeric(1);
747
748        // Assert before subscribing.
749        registry.assert(42u32, id.clone());
750
751        // Subscribe after asserting -- should immediately get the current value.
752        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
753        let mut recv = test_spawn(sub.recv());
754        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id, 42)));
755    }
756
757    #[test]
758    fn subscribe_after_retract_gets_nothing_pending() {
759        let registry = DataspaceRegistry::new();
760        let id = Identifier::numeric(1);
761
762        // Assert then retract -- no current value.
763        registry.assert(42u32, id.clone());
764        registry.retract::<u32>(id.clone());
765
766        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id));
767
768        // Drop the registry so the channel closes, proving no pending value is delivered.
769        drop(registry);
770
771        let mut recv = test_spawn(sub.recv());
772        assert_ready_eq!(recv.poll(), None);
773    }
774
775    #[test]
776    fn all_filter_receives_current_values() {
777        let registry = DataspaceRegistry::new();
778        let id1 = Identifier::numeric(1);
779        let id2 = Identifier::numeric(2);
780
781        // Assert on two identifiers before subscribing.
782        registry.assert(1u32, id1.clone());
783        registry.assert(2u32, id2.clone());
784
785        let mut sub = registry.subscribe::<u32>(IdentifierFilter::all());
786
787        // Should receive both current values (order is not guaranteed since HashMap iteration is unordered).
788        let mut recv = test_spawn(sub.recv());
789        let v1 = assert_ready!(recv.poll());
790        drop(recv);
791
792        let mut recv = test_spawn(sub.recv());
793        let v2 = assert_ready!(recv.poll());
794
795        let mut received = [v1.unwrap(), v2.unwrap()];
796        received.sort_by_key(|update| match update {
797            AssertionUpdate::Asserted(_, v) => *v,
798            AssertionUpdate::Retracted(_) => 0,
799        });
800
801        assert_eq!(received[0], AssertionUpdate::Asserted(id1, 1));
802        assert_eq!(received[1], AssertionUpdate::Asserted(id2, 2));
803    }
804
805    #[test]
806    fn subscribe_receives_current_then_future() {
807        let registry = DataspaceRegistry::new();
808        let id = Identifier::numeric(1);
809
810        // Assert before subscribing.
811        registry.assert(1u32, id.clone());
812
813        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
814
815        // Assert again after subscribing.
816        registry.assert(2u32, id.clone());
817
818        // First recv should return the initial value, second should return the broadcast value.
819        let mut recv = test_spawn(sub.recv());
820        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id.clone(), 1)));
821        drop(recv);
822
823        let mut recv = test_spawn(sub.recv());
824        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id, 2)));
825    }
826
827    #[test]
828    fn prefix_filter_matches_named_identifiers() {
829        let registry = DataspaceRegistry::new();
830        let id1 = Identifier::named("metrics.cpu");
831        let id2 = Identifier::named("metrics.mem");
832        let id3 = Identifier::named("logs.error");
833
834        let mut sub = registry.subscribe::<u32>(IdentifierFilter::prefix("metrics."));
835
836        registry.assert(1u32, id1.clone());
837        registry.assert(2u32, id2.clone());
838        registry.assert(3u32, id3);
839
840        // Should only receive the two metrics identifiers.
841        let mut recv = test_spawn(sub.recv());
842        let v1 = assert_ready!(recv.poll());
843        drop(recv);
844
845        let mut recv = test_spawn(sub.recv());
846        let v2 = assert_ready!(recv.poll());
847
848        let mut received = [v1.unwrap(), v2.unwrap()];
849        received.sort_by_key(|update| match update {
850            AssertionUpdate::Asserted(_, v) => *v,
851            AssertionUpdate::Retracted(_) => 0,
852        });
853
854        assert_eq!(received[0], AssertionUpdate::Asserted(id1, 1));
855        assert_eq!(received[1], AssertionUpdate::Asserted(id2, 2));
856    }
857
858    #[test]
859    fn prefix_filter_does_not_match_numeric() {
860        let registry = DataspaceRegistry::new();
861        let id = Identifier::numeric(42);
862
863        let mut sub = registry.subscribe::<u32>(IdentifierFilter::prefix("any"));
864
865        registry.assert(1u32, id);
866
867        // Drop registry to close channel, proving no value was delivered.
868        drop(registry);
869
870        let mut recv = test_spawn(sub.recv());
871        assert_ready_eq!(recv.poll(), None);
872    }
873
874    #[test]
875    fn prefix_filter_replays_matching_current_values() {
876        let registry = DataspaceRegistry::new();
877        let id1 = Identifier::named("svc.alpha");
878        let id2 = Identifier::named("svc.beta");
879        let id3 = Identifier::named("other.gamma");
880
881        // Assert before subscribing.
882        registry.assert(1u32, id1.clone());
883        registry.assert(2u32, id2.clone());
884        registry.assert(3u32, id3);
885
886        let mut sub = registry.subscribe::<u32>(IdentifierFilter::prefix("svc."));
887
888        // Should replay only the two matching values.
889        let mut recv = test_spawn(sub.recv());
890        let v1 = assert_ready!(recv.poll());
891        drop(recv);
892
893        let mut recv = test_spawn(sub.recv());
894        let v2 = assert_ready!(recv.poll());
895
896        let mut received = [v1.unwrap(), v2.unwrap()];
897        received.sort_by_key(|update| match update {
898            AssertionUpdate::Asserted(_, v) => *v,
899            AssertionUpdate::Retracted(_) => 0,
900        });
901
902        assert_eq!(received[0], AssertionUpdate::Asserted(id1, 1));
903        assert_eq!(received[1], AssertionUpdate::Asserted(id2, 2));
904    }
905
906    #[test]
907    fn try_current_returns_none_outside_context() {
908        assert!(DataspaceRegistry::try_current().is_none());
909    }
910
911    #[test]
912    fn current_and_try_current_work_within_scope() {
913        let registry = DataspaceRegistry::new();
914        let registry_clone = registry.clone();
915
916        let mut scope_fut = test_spawn(CURRENT_DATASPACE.scope(registry, async {
917            let current = DataspaceRegistry::try_current();
918            assert!(current.is_some());
919
920            // Verify it's the same registry by asserting in one and reading from the other.
921            let current = current.unwrap();
922            let id = Identifier::named("test");
923            current.assert(42u32, id.clone());
924
925            let mut sub = registry_clone.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
926            let value = sub.recv().await;
927            assert_eq!(value, Some(AssertionUpdate::Asserted(id, 42)));
928        }));
929        assert_ready!(scope_fut.poll());
930    }
931
932    #[test]
933    fn retract_all_for_process_retracts_all_owned_assertions() {
934        let registry = DataspaceRegistry::new();
935        let process_id = Id::new();
936        let id1 = Identifier::named("val1");
937        let id2 = Identifier::named("val2");
938
939        // Assert two values as if from the given process.
940        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(process_id, || {
941            registry.assert(1u32, id1.clone());
942            registry.assert(2u32, id2.clone());
943        });
944
945        // Subscribe to both after assertion to get current values replayed.
946        let mut sub1 = registry.subscribe::<u32>(IdentifierFilter::exact(id1.clone()));
947        let mut sub2 = registry.subscribe::<u32>(IdentifierFilter::exact(id2.clone()));
948
949        // Drain the initial replayed values.
950        let mut recv = test_spawn(sub1.recv());
951        let _ = assert_ready!(recv.poll());
952        drop(recv);
953
954        let mut recv = test_spawn(sub2.recv());
955        let _ = assert_ready!(recv.poll());
956        drop(recv);
957
958        // Retract all assertions for the process.
959        registry.retract_all_for_process(process_id);
960
961        // Both subscribers should receive retraction notifications.
962        let mut recv = test_spawn(sub1.recv());
963        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Retracted(id1)));
964        drop(recv);
965
966        let mut recv = test_spawn(sub2.recv());
967        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Retracted(id2)));
968    }
969
970    #[test]
971    fn retract_all_for_process_does_not_affect_other_processes() {
972        let registry = DataspaceRegistry::new();
973        let pid_a = Id::new();
974        let pid_b = Id::new();
975        let id_a = Identifier::named("a");
976        let id_b = Identifier::named("b");
977
978        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(pid_a, || {
979            registry.assert(1u32, id_a.clone());
980        });
981        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(pid_b, || {
982            registry.assert(2u32, id_b.clone());
983        });
984
985        // Retract only process A's assertions.
986        registry.retract_all_for_process(pid_a);
987
988        // Process B's value should still be available to new subscribers.
989        let mut sub_b = registry.subscribe::<u32>(IdentifierFilter::exact(id_b.clone()));
990        let mut recv = test_spawn(sub_b.recv());
991        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id_b, 2)));
992        drop(recv);
993
994        // Process A's value should not be available.
995        let mut sub_a = registry.subscribe::<u32>(IdentifierFilter::exact(id_a));
996        drop(registry);
997        let mut recv = test_spawn(sub_a.recv());
998        assert_ready_eq!(recv.poll(), None);
999    }
1000
1001    #[test]
1002    fn retract_all_for_process_notifies_filtered_subscribers() {
1003        let registry = DataspaceRegistry::new();
1004        let process_id = Id::new();
1005        let id = Identifier::named("val");
1006
1007        let mut all_sub = registry.subscribe::<u32>(IdentifierFilter::all());
1008
1009        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(process_id, || {
1010            registry.assert(42u32, id.clone());
1011        });
1012
1013        // Receive the assertion.
1014        let mut recv = test_spawn(all_sub.recv());
1015        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id.clone(), 42)));
1016        drop(recv);
1017
1018        // Retract all for the process.
1019        registry.retract_all_for_process(process_id);
1020
1021        // Should receive the retraction on the wildcard subscription.
1022        let mut recv = test_spawn(all_sub.recv());
1023        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Retracted(id)));
1024    }
1025
1026    #[test]
1027    fn manual_retract_cleans_up_process_tracking() {
1028        let registry = DataspaceRegistry::new();
1029        let process_id = Id::new();
1030        let id = Identifier::named("val");
1031
1032        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
1033
1034        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(process_id, || {
1035            registry.assert(42u32, id.clone());
1036        });
1037
1038        // Manually retract (must be from the owning process).
1039        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(process_id, || {
1040            registry.retract::<u32>(id.clone());
1041        });
1042
1043        // Drain the assertion and retraction.
1044        let mut recv = test_spawn(sub.recv());
1045        let _ = assert_ready!(recv.poll());
1046        drop(recv);
1047
1048        let mut recv = test_spawn(sub.recv());
1049        let _ = assert_ready!(recv.poll());
1050        drop(recv);
1051
1052        // Now retract_all_for_process should be a no-op (no duplicate retraction).
1053        registry.retract_all_for_process(process_id);
1054
1055        // Drop the registry to close the channel, proving no further messages are pending.
1056        drop(registry);
1057        let mut recv = test_spawn(sub.recv());
1058        assert_ready_eq!(recv.poll(), None);
1059    }
1060
1061    #[test]
1062    fn retract_all_for_unknown_process_is_noop() {
1063        let registry = DataspaceRegistry::new();
1064        // Should not panic.
1065        registry.retract_all_for_process(Id::new());
1066    }
1067
1068    #[test]
1069    fn instrumented_process_retracts_on_normal_completion() {
1070        let registry = DataspaceRegistry::new();
1071        let process = Process::supervisor_with_dataspace("test_worker", None, Some(registry.clone())).unwrap();
1072        let id = "from_worker";
1073
1074        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id));
1075
1076        // Spawn a subscriber future, asserting that it is quiesced.
1077        let mut recv_fut = test_spawn(sub.recv());
1078        assert!(!recv_fut.is_woken());
1079        assert_pending!(recv_fut.poll());
1080
1081        // Create an instrumented future that asserts a value, then completes.
1082        let mut worker = test_spawn(process.into_process_future(async {
1083            DataspaceRegistry::try_current()
1084                .expect("dataspace registry should be available")
1085                .assert(42u32, "from_worker");
1086        }));
1087
1088        // Poll the worker to completion — the assertion happens during this poll.
1089        assert_ready!(worker.poll());
1090
1091        // The subscriber should now be woken and have the assertion update.
1092        assert!(recv_fut.is_woken());
1093        assert_ready_eq!(recv_fut.poll(), Some(AssertionUpdate::Asserted(id.into(), 42)));
1094
1095        drop(recv_fut);
1096
1097        // Set up a new subscriber call for getting the retraction.
1098        let mut recv_fut = test_spawn(sub.recv());
1099        assert_pending!(recv_fut.poll());
1100        assert!(!recv_fut.is_woken());
1101
1102        // Drop the InstrumentedProcess — this triggers automatic retraction.
1103        drop(worker);
1104
1105        // The drop should have woken the subscriber.
1106        assert!(recv_fut.is_woken());
1107        assert_ready_eq!(recv_fut.poll(), Some(AssertionUpdate::Retracted(id.into())));
1108    }
1109
1110    #[test]
1111    fn instrumented_process_retracts_on_drop_before_completion() {
1112        let registry = DataspaceRegistry::new();
1113        let process = Process::supervisor_with_dataspace("test_worker", None, Some(registry.clone())).unwrap();
1114        let id = "from_worker";
1115
1116        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id));
1117
1118        // Spawn a subscriber future, asserting that it is quiesced.
1119        let mut recv_fut = test_spawn(sub.recv());
1120        assert!(!recv_fut.is_woken());
1121        assert_pending!(recv_fut.poll());
1122
1123        // Create an instrumented future that asserts a value, then pends forever.
1124        let mut worker = test_spawn(process.into_process_future(async {
1125            DataspaceRegistry::try_current()
1126                .expect("dataspace registry should be available")
1127                .assert(42u32, "from_worker");
1128            pending::<()>().await;
1129        }));
1130
1131        // Drive the worker to assert the value.
1132        assert_pending!(worker.poll());
1133
1134        // The subscriber should now be woken and have the assertion update.
1135        assert!(recv_fut.is_woken());
1136        assert_ready_eq!(recv_fut.poll(), Some(AssertionUpdate::Asserted(id.into(), 42)));
1137
1138        drop(recv_fut);
1139
1140        // Set up a new subscriber call for getting the retraction.
1141        let mut recv_fut = test_spawn(sub.recv());
1142        assert_pending!(recv_fut.poll());
1143        assert!(!recv_fut.is_woken());
1144
1145        // Drop the worker (simulates abort/cancel) — triggers automatic retraction.
1146        drop(worker);
1147
1148        // The drop should have woken the subscriber with a retraction.
1149        assert!(recv_fut.is_woken());
1150        assert_ready_eq!(recv_fut.poll(), Some(AssertionUpdate::Retracted(id.into())));
1151    }
1152
1153    #[test]
1154    fn instrumented_process_retracts_multiple_types_on_drop() {
1155        let registry = DataspaceRegistry::new();
1156        let process = Process::supervisor_with_dataspace("test_worker", None, Some(registry.clone())).unwrap();
1157        let id_num = "number";
1158        let id_str = "text";
1159
1160        let mut sub_u32 = registry.subscribe::<u32>(IdentifierFilter::exact(id_num));
1161        let mut sub_str = registry.subscribe::<String>(IdentifierFilter::exact(id_str));
1162
1163        // Spawn futures for both subscribers, asserting that they are quiesced.
1164        let mut recv_u32_fut = test_spawn(sub_u32.recv());
1165        let mut recv_str_fut = test_spawn(sub_str.recv());
1166        assert!(!recv_u32_fut.is_woken());
1167        assert!(!recv_str_fut.is_woken());
1168        assert_pending!(recv_u32_fut.poll());
1169        assert_pending!(recv_str_fut.poll());
1170
1171        // Create an instrumented future that asserts values of two different types.
1172        let mut worker = test_spawn(process.into_process_future(async {
1173            let ds = DataspaceRegistry::try_current().expect("dataspace registry should be available");
1174            ds.assert(42u32, id_num);
1175            ds.assert("hello".to_string(), id_str);
1176            pending::<()>().await;
1177        }));
1178
1179        // Drive the worker to assert the values.
1180        assert_pending!(worker.poll());
1181
1182        // Both subscribers should now be woken and have the assertion updates.
1183        assert!(recv_u32_fut.is_woken());
1184        assert!(recv_str_fut.is_woken());
1185        assert_ready_eq!(recv_u32_fut.poll(), Some(AssertionUpdate::Asserted(id_num.into(), 42)));
1186        assert_ready_eq!(
1187            recv_str_fut.poll(),
1188            Some(AssertionUpdate::Asserted(id_str.into(), "hello".to_string()))
1189        );
1190
1191        drop(recv_u32_fut);
1192        drop(recv_str_fut);
1193
1194        // Set up new subscriber calls for getting the retractions.
1195        let mut recv_u32 = test_spawn(sub_u32.recv());
1196        let mut recv_str = test_spawn(sub_str.recv());
1197        assert_pending!(recv_u32.poll());
1198        assert_pending!(recv_str.poll());
1199        assert!(!recv_u32.is_woken());
1200        assert!(!recv_str.is_woken());
1201
1202        // Drop the worker — both types should be retracted.
1203        drop(worker);
1204
1205        assert!(recv_u32.is_woken());
1206        assert!(recv_str.is_woken());
1207        assert_ready_eq!(recv_u32.poll(), Some(AssertionUpdate::Retracted(id_num.into())));
1208        assert_ready_eq!(recv_str.poll(), Some(AssertionUpdate::Retracted(id_str.into())));
1209    }
1210
1211    #[test]
1212    #[cfg(not(debug_assertions))]
1213    fn retract_from_non_owner_is_ignored() {
1214        let registry = DataspaceRegistry::new();
1215        let pid_a = Id::new();
1216        let pid_b = Id::new();
1217        let id = Identifier::named("owned_by_a");
1218
1219        // Assert as process A.
1220        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(pid_a, || {
1221            registry.assert(42u32, id.clone());
1222        });
1223
1224        // Attempt to retract as process B -- should be silently ignored.
1225        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(pid_b, || {
1226            registry.retract::<u32>(id.clone());
1227        });
1228
1229        // Value should still be present for new subscribers.
1230        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
1231        let mut recv = test_spawn(sub.recv());
1232        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id.clone(), 42)));
1233        drop(recv);
1234
1235        // Retract as process A -- should succeed.
1236        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(pid_a, || {
1237            registry.retract::<u32>(id.clone());
1238        });
1239
1240        // Subscriber should receive the retraction.
1241        let mut recv = test_spawn(sub.recv());
1242        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Retracted(id)));
1243    }
1244
1245    #[test]
1246    #[cfg(debug_assertions)]
1247    #[should_panic(expected = "attempted to retract assertion owned by")]
1248    fn retract_from_non_owner_panics_in_debug() {
1249        let registry = DataspaceRegistry::new();
1250        let pid_a = Id::new();
1251        let pid_b = Id::new();
1252        let id = Identifier::named("owned_by_a");
1253
1254        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(pid_a, || {
1255            registry.assert(42u32, id.clone());
1256        });
1257
1258        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(pid_b, || {
1259            registry.retract::<u32>(id.clone());
1260        });
1261    }
1262
1263    #[test]
1264    #[cfg(not(debug_assertions))]
1265    fn reassert_from_non_owner_is_ignored() {
1266        let registry = DataspaceRegistry::new();
1267        let pid_a = Id::new();
1268        let pid_b = Id::new();
1269        let id = Identifier::named("owned_by_a");
1270
1271        // Assert as process A.
1272        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(pid_a, || {
1273            registry.assert(42u32, id.clone());
1274        });
1275
1276        // Attempt to overwrite as process B -- should be silently ignored.
1277        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(pid_b, || {
1278            registry.assert(99u32, id.clone());
1279        });
1280
1281        // Value should still be the original.
1282        let mut sub = registry.subscribe::<u32>(IdentifierFilter::exact(id.clone()));
1283        let mut recv = test_spawn(sub.recv());
1284        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id.clone(), 42)));
1285        drop(recv);
1286
1287        // Update as process A -- should succeed.
1288        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(pid_a, || {
1289            registry.assert(100u32, id.clone());
1290        });
1291
1292        // Subscriber should receive the update.
1293        let mut recv = test_spawn(sub.recv());
1294        assert_ready_eq!(recv.poll(), Some(AssertionUpdate::Asserted(id, 100)));
1295    }
1296
1297    #[test]
1298    #[cfg(debug_assertions)]
1299    #[should_panic(expected = "attempted to update assertion owned by")]
1300    fn reassert_from_non_owner_panics_in_debug() {
1301        let registry = DataspaceRegistry::new();
1302        let pid_a = Id::new();
1303        let pid_b = Id::new();
1304        let id = Identifier::named("owned_by_a");
1305
1306        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(pid_a, || {
1307            registry.assert(42u32, id.clone());
1308        });
1309
1310        crate::runtime::process::CURRENT_PROCESS_ID.sync_scope(pid_b, || {
1311            registry.assert(99u32, id.clone());
1312        });
1313    }
1314}
saluki_core/runtime/state/dataspace.rs

saluki_core/runtime/state/
dataspace.rs
