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

liminal_server/server/participant/
publication.rs
