liminal_server/server/participant/
publication.rs

1//! Server-wide participant publication registry and connection-owned ready inboxes.
2//!
3//! Durable recipient snapshots name participant ids. Production resolves each
4//! participant's current binding to a durable connection incarnation and uses
5//! this registry only to wake that exact live connection. The registry stores a
6//! weak inbox and a weak-scheduler READY waker, so it cannot keep a connection
7//! process or scheduler alive.
8
9use std::collections::{BTreeMap, BTreeSet};
10use std::sync::{Arc, Mutex, Weak};
11
12use liminal_protocol::wire::{
13    BindingEpoch, ConnectionIncarnation, ConversationId, ParticipantDelivery, ParticipantId,
14    ServerPush,
15};
16
17use crate::server::connection::ReadyWaker;
18
19/// Connection-local volatile offer cursor for one conversation and exact
20/// binding. A different binding epoch discards this progress and restarts from
21/// the durable recipient acknowledgement frontier.
22#[derive(Clone, Copy, Debug, PartialEq, Eq)]
23pub struct ParticipantOfferedProgress {
24    pub(crate) binding_epoch: BindingEpoch,
25    pub(crate) through_seq: u64,
26}
27
28/// One exact durable obligation selected for the connection's current binding.
29#[derive(Clone, Debug, PartialEq, Eq)]
30pub struct ParticipantPublication {
31    pub(crate) participant_id: ParticipantId,
32    pub(crate) binding_epoch: BindingEpoch,
33    pub(crate) delivery: ParticipantDelivery,
34}
35
36impl ParticipantPublication {
37    #[must_use]
38    pub(crate) const fn conversation_id(&self) -> ConversationId {
39        self.delivery.conversation_id
40    }
41
42    #[must_use]
43    pub(crate) const fn delivery_seq(&self) -> u64 {
44        self.delivery.delivery_seq
45    }
46}
47
48/// Exact refusal-arm wake transferred by the observer owner after its durable
49/// `Advance` flush. It is volatile connection work, not a participant outbox
50/// record, and deliberately carries no participant recipient or delivery
51/// sequence.
52#[derive(Clone, Copy, Debug, PartialEq, Eq)]
53pub struct ObserverPublication {
54    pub(crate) conversation_id: ConversationId,
55    pub(crate) refused_epoch: u64,
56    pub(crate) observer_progress: u64,
57}
58
59impl ObserverPublication {
60    #[must_use]
61    pub(crate) const fn into_server_push(self) -> ServerPush {
62        ServerPush::ObserverProgressed {
63            conversation_id: self.conversation_id,
64            refused_epoch: self.refused_epoch,
65            observer_progress: self.observer_progress,
66        }
67    }
68}
69
70/// Weak exact-live-connection target captured when an observer arm is installed.
71///
72/// Cloning this value clones only weak/non-owning publication capability; it
73/// cannot keep the connection process or inbox alive.
74#[derive(Clone, Debug)]
75pub struct ObserverPublicationTarget {
76    inbox: Weak<Mutex<ReadyPublications>>,
77    waker: ReadyWaker,
78}
79
80impl ObserverPublicationTarget {
81    /// Transfers one fired payload to the exact live inbox. The queue keeps the
82    /// latest payload per conversation: later progress supersedes an undrained
83    /// older wake because the recovery consumer only needs the newest durable
84    /// progress. A dead weak target drops only this wake.
85    pub(crate) fn publish(
86        &self,
87        publication: ObserverPublication,
88    ) -> Result<bool, ParticipantPublicationError> {
89        let Some(inbox) = self.inbox.upgrade() else {
90            return Ok(false);
91        };
92        let should_wake = {
93            let mut inbox = inbox
94                .lock()
95                .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
96            let replacing = inbox
97                .observer_progressed
98                .contains_key(&publication.conversation_id);
99            if !replacing {
100                let occupied = u64::try_from(inbox.observer_progressed.len()).unwrap_or(u64::MAX);
101                if occupied >= inbox.limit {
102                    return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
103                }
104            }
105            let was_empty = inbox.is_empty();
106            inbox
107                .observer_progressed
108                .insert(publication.conversation_id, publication);
109            was_empty
110        };
111        if should_wake {
112            self.waker.fire();
113        }
114        Ok(true)
115    }
116}
117
118#[derive(Clone, Copy, Debug, thiserror::Error)]
119pub enum ParticipantPublicationError {
120    /// A live incarnation was registered more than once.
121    #[error("participant publication incarnation {incarnation:?} is already registered")]
122    DuplicateRegistration {
123        /// Durable incarnation that already owns a live inbox.
124        incarnation: ConnectionIncarnation,
125    },
126    /// The connection-owned inbox mutex was poisoned.
127    #[error("participant publication inbox is poisoned")]
128    InboxPoisoned,
129    /// A new ready conversation would exceed the signed connection bound.
130    #[error("participant publication inbox exceeds its signed conversation bound {limit}")]
131    InboxCapacity {
132        /// Signed maximum semantic conversations for the connection.
133        limit: u64,
134    },
135}
136
137#[derive(Debug)]
138struct ReadyPublications {
139    limit: u64,
140    conversations: BTreeSet<ConversationId>,
141    observer_progressed: BTreeMap<ConversationId, ObserverPublication>,
142}
143
144impl ReadyPublications {
145    fn is_empty(&self) -> bool {
146        self.conversations.is_empty() && self.observer_progressed.is_empty()
147    }
148}
149
150/// All participant and observer work atomically removed for one shared push
151/// slice. The pump merges these collections by conversation before applying the
152/// single signed budget.
153#[derive(Debug)]
154pub struct ReadyPublicationBatch {
155    pub(crate) conversations: Vec<ConversationId>,
156    pub(crate) observer_progressed: Vec<ObserverPublication>,
157}
158
159/// Inbox strongly owned by exactly one connection process.
160///
161/// The value is intentionally not `Clone`: the registry receives only a weak
162/// projection and cannot become another owner. Participant readiness coalesces
163/// conversation ids; observer readiness retains only the latest fired payload
164/// per conversation.
165#[derive(Debug)]
166pub struct ParticipantPublicationInbox {
167    inner: Arc<Mutex<ReadyPublications>>,
168}
169
170impl ParticipantPublicationInbox {
171    /// Creates the connection-owned bounded ready set from the signed semantic
172    /// conversation limit.
173    #[must_use]
174    pub(crate) fn new(limit: u64) -> Self {
175        Self {
176            inner: Arc::new(Mutex::new(ReadyPublications {
177                limit,
178                conversations: BTreeSet::new(),
179                observer_progressed: BTreeMap::new(),
180            })),
181        }
182    }
183
184    fn weak(&self) -> Weak<Mutex<ReadyPublications>> {
185        Arc::downgrade(&self.inner)
186    }
187
188    /// Atomically removes all sorted, coalesced work for one shared push slice.
189    pub(crate) fn take_ready(&self) -> Result<ReadyPublicationBatch, ParticipantPublicationError> {
190        let mut inbox = self
191            .inner
192            .lock()
193            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
194        Ok(ReadyPublicationBatch {
195            conversations: std::mem::take(&mut inbox.conversations)
196                .into_iter()
197                .collect(),
198            observer_progressed: std::mem::take(&mut inbox.observer_progressed)
199                .into_values()
200                .collect(),
201        })
202    }
203
204    /// Requeues deferred conversations after a budget-limited or held-back
205    /// slice. Existing ids remain coalesced.
206    pub(crate) fn requeue(
207        &self,
208        conversations: impl IntoIterator<Item = ConversationId>,
209    ) -> Result<(), ParticipantPublicationError> {
210        let mut inbox = self
211            .inner
212            .lock()
213            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
214        for conversation_id in conversations {
215            if inbox.conversations.contains(&conversation_id) {
216                continue;
217            }
218            let occupied = u64::try_from(inbox.conversations.len()).unwrap_or(u64::MAX);
219            if occupied >= inbox.limit {
220                return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
221            }
222            inbox.conversations.insert(conversation_id);
223        }
224        drop(inbox);
225        Ok(())
226    }
227
228    /// Requeues budget-deferred observer payloads only for vacant conversations.
229    /// Any incumbent arrived after the pump's take and supersedes the deferred
230    /// payload; skipping it consumes no signed conversation capacity.
231    pub(crate) fn requeue_observers(
232        &self,
233        publications: impl IntoIterator<Item = ObserverPublication>,
234    ) -> Result<(), ParticipantPublicationError> {
235        let mut inbox = self
236            .inner
237            .lock()
238            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
239        for publication in publications {
240            if inbox
241                .observer_progressed
242                .contains_key(&publication.conversation_id)
243            {
244                continue;
245            }
246            let occupied = u64::try_from(inbox.observer_progressed.len()).unwrap_or(u64::MAX);
247            if occupied >= inbox.limit {
248                return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
249            }
250            inbox
251                .observer_progressed
252                .insert(publication.conversation_id, publication);
253        }
254        drop(inbox);
255        Ok(())
256    }
257
258    /// Non-consuming final-probe fact used after socket readiness is armed.
259    pub(crate) fn has_pending(&self) -> Result<bool, ParticipantPublicationError> {
260        self.inner
261            .lock()
262            .map(|inbox| !inbox.is_empty())
263            .map_err(|_| ParticipantPublicationError::InboxPoisoned)
264    }
265}
266
267#[derive(Debug)]
268struct ParticipantPublicationHandle {
269    inbox: Weak<Mutex<ReadyPublications>>,
270    waker: ReadyWaker,
271}
272
273/// Server-wide incarnation-to-connection publication registry.
274#[derive(Debug, Default)]
275pub struct ParticipantPublicationRegistry {
276    registrations: Mutex<BTreeMap<ConnectionIncarnation, ParticipantPublicationHandle>>,
277}
278
279impl ParticipantPublicationRegistry {
280    /// Registers one connection-owned inbox. A stale weak entry may be replaced;
281    /// a second live owner for one durable incarnation is a typed invariant fault.
282    pub(crate) fn register(
283        &self,
284        incarnation: ConnectionIncarnation,
285        inbox: &ParticipantPublicationInbox,
286        waker: ReadyWaker,
287    ) -> Result<(), ParticipantPublicationError> {
288        let mut registrations = self
289            .registrations
290            .lock()
291            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
292        if registrations
293            .get(&incarnation)
294            .is_some_and(|existing| existing.inbox.strong_count() > 0)
295        {
296            return Err(ParticipantPublicationError::DuplicateRegistration { incarnation });
297        }
298        registrations.insert(
299            incarnation,
300            ParticipantPublicationHandle {
301                inbox: inbox.weak(),
302                waker,
303            },
304        );
305        drop(registrations);
306        Ok(())
307    }
308
309    /// Removes the registration at explicit process teardown. The weak handle
310    /// already makes stale delivery harmless; eager removal keeps lookup exact.
311    pub(crate) fn deregister(&self, incarnation: ConnectionIncarnation) {
312        if let Ok(mut registrations) = self.registrations.lock() {
313            registrations.remove(&incarnation);
314        }
315    }
316
317    /// Captures the weak exact-live-connection target for an accepted observer
318    /// recovery arm. Missing or already-dead registrations yield no target;
319    /// callers must never substitute or broadcast.
320    pub(crate) fn observer_target(
321        &self,
322        incarnation: ConnectionIncarnation,
323    ) -> Result<Option<ObserverPublicationTarget>, ParticipantPublicationError> {
324        let registrations = self
325            .registrations
326            .lock()
327            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
328        let target = registrations.get(&incarnation).and_then(|handle| {
329            (handle.inbox.strong_count() > 0).then(|| ObserverPublicationTarget {
330                inbox: Weak::clone(&handle.inbox),
331                waker: handle.waker.clone(),
332            })
333        });
334        drop(registrations);
335        Ok(target)
336    }
337
338    /// Coalesces one conversation into the exact live incarnation's inbox and
339    /// fires READY only on the empty-to-nonempty edge.
340    ///
341    /// Returns `false` when the incarnation or connection process is gone.
342    pub(crate) fn notify(
343        &self,
344        incarnation: ConnectionIncarnation,
345        conversation_id: ConversationId,
346    ) -> Result<bool, ParticipantPublicationError> {
347        let (weak_inbox, waker) = {
348            let registrations = self
349                .registrations
350                .lock()
351                .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
352            let Some(handle) = registrations.get(&incarnation) else {
353                return Ok(false);
354            };
355            let weak_inbox = Weak::clone(&handle.inbox);
356            let waker = handle.waker.clone();
357            drop(registrations);
358            (weak_inbox, waker)
359        };
360        let Some(inbox) = weak_inbox.upgrade() else {
361            self.deregister(incarnation);
362            return Ok(false);
363        };
364        let should_wake = {
365            let mut inbox = inbox
366                .lock()
367                .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
368            if inbox.conversations.contains(&conversation_id) {
369                return Ok(true);
370            }
371            let occupied = u64::try_from(inbox.conversations.len()).unwrap_or(u64::MAX);
372            if occupied >= inbox.limit {
373                return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
374            }
375            let was_empty = inbox.is_empty();
376            inbox.conversations.insert(conversation_id);
377            was_empty
378        };
379        if should_wake {
380            waker.fire();
381        }
382        Ok(true)
383    }
384}
385
386#[cfg(test)]
387mod tests {
388    use std::sync::Arc;
389    use std::sync::atomic::{AtomicU64, Ordering};
390
391    use liminal_protocol::wire::ConnectionIncarnation;
392
393    use super::{ParticipantPublicationInbox, ParticipantPublicationRegistry};
394    use crate::server::connection::ReadyWaker;
395
396    #[test]
397    fn parked_connection_wakes_on_outbox_and_no_polling_occurs()
398    -> Result<(), Box<dyn std::error::Error>> {
399        let incarnation = ConnectionIncarnation::new(12, 34);
400        let wake_count = Arc::new(AtomicU64::new(0));
401        let registry = ParticipantPublicationRegistry::default();
402        let inbox = ParticipantPublicationInbox::new(3);
403        registry.register(
404            incarnation,
405            &inbox,
406            ReadyWaker::for_test(Arc::clone(&wake_count)),
407        )?;
408
409        assert!(!inbox.has_pending()?);
410        assert!(registry.notify(incarnation, 7)?);
411        assert_eq!(wake_count.load(Ordering::SeqCst), 1);
412        assert!(inbox.has_pending()?);
413
414        // Duplicate and additional ready conversations coalesce behind the one
415        // empty-to-nonempty wake; no repeated probe or timer drives progress.
416        assert!(registry.notify(incarnation, 7)?);
417        assert!(registry.notify(incarnation, 8)?);
418        assert_eq!(wake_count.load(Ordering::SeqCst), 1);
419        let ready = inbox.take_ready()?;
420        assert_eq!(ready.conversations, vec![7, 8]);
421        assert!(ready.observer_progressed.is_empty());
422        assert!(!inbox.has_pending()?);
423
424        // This notification models the execute-to-wait race: it lands after a
425        // drain but before the process final probe. The non-consuming probe sees
426        // it and the edge fires exactly one new READY.
427        assert!(registry.notify(incarnation, 9)?);
428        assert!(inbox.has_pending()?);
429        assert_eq!(wake_count.load(Ordering::SeqCst), 2);
430        let ready = inbox.take_ready()?;
431        assert_eq!(ready.conversations, vec![9]);
432        assert!(ready.observer_progressed.is_empty());
433        assert!(!inbox.has_pending()?);
434
435        let idle_count = wake_count.load(Ordering::SeqCst);
436        assert_eq!(wake_count.load(Ordering::SeqCst), idle_count);
437        registry.deregister(incarnation);
438        assert!(!registry.notify(incarnation, 10)?);
439        assert_eq!(wake_count.load(Ordering::SeqCst), idle_count);
440        Ok(())
441    }
442}
liminal_server/server/participant/publication.rs

liminal_server/server/participant/
publication.rs
