monoloop_loop/
subscription.rs

1//! Lossless, gap-detecting canonical event subscription for The Loop.
2
3use monoloop_contracts::InterpreterOutputEvent;
4use tokio::sync::mpsc;
5
6/// Subscriber identity (correlation only).
7#[derive(Clone, Debug, PartialEq, Eq, Hash)]
8pub struct SubscriberId(String);
9
10impl SubscriberId {
11    /// Create a subscriber id.
12    pub fn new(value: impl Into<String>) -> Self {
13        Self(value.into())
14    }
15
16    /// Borrow the id.
17    pub fn as_str(&self) -> &str {
18        &self.0
19    }
20}
21
22/// Subscription status / gap notification.
23#[derive(Clone, Debug, PartialEq, Eq)]
24pub enum SubscriptionStatus {
25    /// Source opened.
26    Opened,
27    /// Source ending cleanly.
28    Closing,
29    /// Delivery sequence gap detected.
30    Gap(SubscriptionGap),
31    /// Source lost without clean end.
32    Lost,
33}
34
35/// Gap details.
36#[derive(Clone, Debug, PartialEq, Eq)]
37pub struct SubscriptionGap {
38    /// Expected next sequence.
39    pub expected: u64,
40    /// Observed sequence (if any).
41    pub observed: Option<u64>,
42}
43
44/// One delivered event with explicit delivery sequence.
45#[derive(Clone, Debug)]
46pub struct DeliveredEvent {
47    /// Monotonic delivery sequence (1-based) from the distributor.
48    pub delivery_sequence: u64,
49    /// Canonical interpreter event (unit or end).
50    pub event: InterpreterOutputEvent,
51}
52
53/// Lossless subscription owned by one Loop (never shared with Console).
54pub struct CanonicalEventSubscription {
55    /// Subscriber id.
56    pub subscriber_id: SubscriberId,
57    rx: mpsc::Receiver<Result<DeliveredEvent, SubscriptionStatus>>,
58}
59
60impl CanonicalEventSubscription {
61    /// Create from a channel end.
62    pub fn new(
63        subscriber_id: SubscriberId,
64        rx: mpsc::Receiver<Result<DeliveredEvent, SubscriptionStatus>>,
65    ) -> Self {
66        Self { subscriber_id, rx }
67    }
68
69    /// Receive next delivery or status. `None` = channel closed.
70    pub async fn recv(&mut self) -> Option<Result<DeliveredEvent, SubscriptionStatus>> {
71        self.rx.recv().await
72    }
73}
74
75/// Publisher half used by the event distributor (testkit / composition).
76#[derive(Clone)]
77pub struct SubscriptionPublisher {
78    tx: mpsc::Sender<Result<DeliveredEvent, SubscriptionStatus>>,
79    next_seq: std::sync::Arc<std::sync::atomic::AtomicU64>,
80}
81
82impl SubscriptionPublisher {
83    /// Create a bounded subscription pair.
84    pub fn channel(
85        subscriber_id: impl Into<String>,
86        capacity: usize,
87    ) -> (Self, CanonicalEventSubscription) {
88        let (tx, rx) = mpsc::channel(capacity.max(1));
89        (
90            Self {
91                tx,
92                next_seq: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(1)),
93            },
94            CanonicalEventSubscription::new(SubscriberId::new(subscriber_id), rx),
95        )
96    }
97
98    /// Publish one event with the next delivery sequence (lossless backpressure).
99    pub async fn publish(
100        &self,
101        event: InterpreterOutputEvent,
102    ) -> Result<(), mpsc::error::SendError<Result<DeliveredEvent, SubscriptionStatus>>> {
103        let seq = self
104            .next_seq
105            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
106        self.tx
107            .send(Ok(DeliveredEvent {
108                delivery_sequence: seq,
109                event,
110            }))
111            .await
112    }
113
114    /// Signal a gap (fail-closed for Loop).
115    pub async fn signal_gap(&self, expected: u64, observed: Option<u64>) -> Result<(), ()> {
116        self.tx
117            .send(Err(SubscriptionStatus::Gap(SubscriptionGap {
118                expected,
119                observed,
120            })))
121            .await
122            .map_err(|_| ())
123    }
124
125    /// Signal source lost.
126    pub async fn signal_lost(&self) -> Result<(), ()> {
127        self.tx
128            .send(Err(SubscriptionStatus::Lost))
129            .await
130            .map_err(|_| ())
131    }
132}
monoloop_loop/subscription.rs

monoloop_loop/
subscription.rs
