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/// Exact settlement wake transferred to a connection whose attach or detach was
73/// refused `MarkerSettlementBackpressure` in THIS process lifetime.
74///
75/// ⛔ CONNECTION-SCOPED BY CONSTRUCTION (participant contract §0.16 build
76/// obligation 3). The lazy implementation pushes `MarkerSettled` to every
77/// connection on the conversation; that is a settlement-timing side channel to
78/// uninvolved parties and is OUTLAWED. There is no conversation-wide fan-out
79/// anywhere on this path: the only way a connection can receive one is for the
80/// registry to hold a waiter it installed from its own refusal. It is likewise
81/// never sent to a connection refused at the ENROLLMENT wrapper, whose refusal
82/// (`EnrollmentSettlementBackpressure`) carries no epoch and therefore cannot
83/// install a waiter at all.
84#[derive(Clone, Copy, Debug, PartialEq, Eq)]
85pub struct MarkerSettledPublication {
86    pub(crate) conversation_id: ConversationId,
87    pub(crate) refused_epoch: u64,
88}
89
90impl MarkerSettledPublication {
91    #[must_use]
92    pub(crate) const fn into_server_push(self) -> ServerPush {
93        ServerPush::MarkerSettled {
94            conversation_id: self.conversation_id,
95            refused_epoch: self.refused_epoch,
96        }
97    }
98}
99
100/// Weak exact-live-connection target captured when an observer arm is installed.
101///
102/// Cloning this value clones only weak/non-owning publication capability; it
103/// cannot keep the connection process or inbox alive.
104#[derive(Clone, Debug)]
105pub struct ObserverPublicationTarget {
106    inbox: Weak<Mutex<ReadyPublications>>,
107    waker: ReadyWaker,
108}
109
110impl ObserverPublicationTarget {
111    /// Transfers one fired payload to the exact live inbox. The queue keeps the
112    /// latest payload per conversation: later progress supersedes an undrained
113    /// older wake because the recovery consumer only needs the newest durable
114    /// progress. A dead weak target drops only this wake.
115    pub(crate) fn publish(
116        &self,
117        publication: ObserverPublication,
118    ) -> Result<bool, ParticipantPublicationError> {
119        let Some(inbox) = self.inbox.upgrade() else {
120            return Ok(false);
121        };
122        let should_wake = {
123            let mut inbox = inbox
124                .lock()
125                .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
126            let replacing = inbox
127                .observer_progressed
128                .contains_key(&publication.conversation_id);
129            if !replacing {
130                let occupied = u64::try_from(inbox.observer_progressed.len()).unwrap_or(u64::MAX);
131                if occupied >= inbox.limit {
132                    return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
133                }
134            }
135            let was_empty = inbox.is_empty();
136            inbox
137                .observer_progressed
138                .insert(publication.conversation_id, publication);
139            was_empty
140        };
141        if should_wake {
142            self.waker.fire();
143        }
144        Ok(true)
145    }
146}
147
148impl ObserverPublicationTarget {
149    /// Transfers one settlement wake to the exact live inbox that was refused.
150    ///
151    /// Deliberately the SAME weak connection-level capability the observer wake
152    /// uses: §0.16 rules the settlement wake "connection-scoped ... mirroring
153    /// `ObserverProgressed`'s connection-level delivery", and sharing the target
154    /// type is what makes that a structural property rather than a claim. Like
155    /// the observer lane it keeps the latest payload per conversation, which is
156    /// exact here because a connection waits on at most one settlement epoch per
157    /// conversation — its own most recent refusal.
158    pub(crate) fn publish_marker_settled(
159        &self,
160        publication: MarkerSettledPublication,
161    ) -> Result<bool, ParticipantPublicationError> {
162        let Some(inbox) = self.inbox.upgrade() else {
163            return Ok(false);
164        };
165        let should_wake = {
166            let mut inbox = inbox
167                .lock()
168                .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
169            let replacing = inbox
170                .marker_settled
171                .contains_key(&publication.conversation_id);
172            if !replacing {
173                let occupied = u64::try_from(inbox.marker_settled.len()).unwrap_or(u64::MAX);
174                if occupied >= inbox.limit {
175                    return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
176                }
177            }
178            let was_empty = inbox.is_empty();
179            inbox
180                .marker_settled
181                .insert(publication.conversation_id, publication);
182            was_empty
183        };
184        if should_wake {
185            self.waker.fire();
186        }
187        Ok(true)
188    }
189}
190
191#[derive(Clone, Copy, Debug, thiserror::Error)]
192pub enum ParticipantPublicationError {
193    /// A live incarnation was registered more than once.
194    #[error("participant publication incarnation {incarnation:?} is already registered")]
195    DuplicateRegistration {
196        /// Durable incarnation that already owns a live inbox.
197        incarnation: ConnectionIncarnation,
198    },
199    /// The connection-owned inbox mutex was poisoned.
200    #[error("participant publication inbox is poisoned")]
201    InboxPoisoned,
202    /// A new ready conversation would exceed the signed connection bound.
203    #[error("participant publication inbox exceeds its signed conversation bound {limit}")]
204    InboxCapacity {
205        /// Signed maximum semantic conversations for the connection.
206        limit: u64,
207    },
208}
209
210#[derive(Debug)]
211struct ReadyPublications {
212    limit: u64,
213    conversations: BTreeSet<ConversationId>,
214    observer_progressed: BTreeMap<ConversationId, ObserverPublication>,
215    marker_settled: BTreeMap<ConversationId, MarkerSettledPublication>,
216}
217
218impl ReadyPublications {
219    fn is_empty(&self) -> bool {
220        self.conversations.is_empty()
221            && self.observer_progressed.is_empty()
222            && self.marker_settled.is_empty()
223    }
224}
225
226/// All participant and observer work atomically removed for one shared push
227/// slice. The pump merges these collections by conversation before applying the
228/// single signed budget.
229#[derive(Debug)]
230pub struct ReadyPublicationBatch {
231    pub(crate) conversations: Vec<ConversationId>,
232    pub(crate) observer_progressed: Vec<ObserverPublication>,
233    pub(crate) marker_settled: Vec<MarkerSettledPublication>,
234}
235
236/// Inbox strongly owned by exactly one connection process.
237///
238/// The value is intentionally not `Clone`: the registry receives only a weak
239/// projection and cannot become another owner. Participant readiness coalesces
240/// conversation ids; observer readiness retains only the latest fired payload
241/// per conversation.
242#[derive(Debug)]
243pub struct ParticipantPublicationInbox {
244    inner: Arc<Mutex<ReadyPublications>>,
245}
246
247impl ParticipantPublicationInbox {
248    /// Creates the connection-owned bounded ready set from the signed semantic
249    /// conversation limit.
250    #[must_use]
251    pub(crate) fn new(limit: u64) -> Self {
252        Self {
253            inner: Arc::new(Mutex::new(ReadyPublications {
254                limit,
255                conversations: BTreeSet::new(),
256                observer_progressed: BTreeMap::new(),
257                marker_settled: BTreeMap::new(),
258            })),
259        }
260    }
261
262    fn weak(&self) -> Weak<Mutex<ReadyPublications>> {
263        Arc::downgrade(&self.inner)
264    }
265
266    /// Atomically removes all sorted, coalesced work for one shared push slice.
267    pub(crate) fn take_ready(&self) -> Result<ReadyPublicationBatch, ParticipantPublicationError> {
268        let mut inbox = self
269            .inner
270            .lock()
271            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
272        Ok(ReadyPublicationBatch {
273            conversations: std::mem::take(&mut inbox.conversations)
274                .into_iter()
275                .collect(),
276            observer_progressed: std::mem::take(&mut inbox.observer_progressed)
277                .into_values()
278                .collect(),
279            marker_settled: std::mem::take(&mut inbox.marker_settled)
280                .into_values()
281                .collect(),
282        })
283    }
284
285    /// Requeues deferred conversations after a budget-limited or held-back
286    /// slice. Existing ids remain coalesced.
287    pub(crate) fn requeue(
288        &self,
289        conversations: impl IntoIterator<Item = ConversationId>,
290    ) -> Result<(), ParticipantPublicationError> {
291        let mut inbox = self
292            .inner
293            .lock()
294            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
295        for conversation_id in conversations {
296            if inbox.conversations.contains(&conversation_id) {
297                continue;
298            }
299            let occupied = u64::try_from(inbox.conversations.len()).unwrap_or(u64::MAX);
300            if occupied >= inbox.limit {
301                return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
302            }
303            inbox.conversations.insert(conversation_id);
304        }
305        drop(inbox);
306        Ok(())
307    }
308
309    /// Requeues budget-deferred observer payloads only for vacant conversations.
310    /// Any incumbent arrived after the pump's take and supersedes the deferred
311    /// payload; skipping it consumes no signed conversation capacity.
312    pub(crate) fn requeue_observers(
313        &self,
314        publications: impl IntoIterator<Item = ObserverPublication>,
315    ) -> Result<(), ParticipantPublicationError> {
316        let mut inbox = self
317            .inner
318            .lock()
319            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
320        for publication in publications {
321            if inbox
322                .observer_progressed
323                .contains_key(&publication.conversation_id)
324            {
325                continue;
326            }
327            let occupied = u64::try_from(inbox.observer_progressed.len()).unwrap_or(u64::MAX);
328            if occupied >= inbox.limit {
329                return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
330            }
331            inbox
332                .observer_progressed
333                .insert(publication.conversation_id, publication);
334        }
335        drop(inbox);
336        Ok(())
337    }
338
339    /// Requeues budget-deferred settlement wakes only for vacant conversations.
340    /// An incumbent arrived after the pump's take and supersedes the deferred
341    /// payload, exactly as on the observer lane.
342    pub(crate) fn requeue_marker_settled(
343        &self,
344        publications: impl IntoIterator<Item = MarkerSettledPublication>,
345    ) -> Result<(), ParticipantPublicationError> {
346        let mut inbox = self
347            .inner
348            .lock()
349            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
350        for publication in publications {
351            if inbox
352                .marker_settled
353                .contains_key(&publication.conversation_id)
354            {
355                continue;
356            }
357            let occupied = u64::try_from(inbox.marker_settled.len()).unwrap_or(u64::MAX);
358            if occupied >= inbox.limit {
359                return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
360            }
361            inbox
362                .marker_settled
363                .insert(publication.conversation_id, publication);
364        }
365        drop(inbox);
366        Ok(())
367    }
368
369    /// Non-consuming final-probe fact used after socket readiness is armed.
370    pub(crate) fn has_pending(&self) -> Result<bool, ParticipantPublicationError> {
371        self.inner
372            .lock()
373            .map(|inbox| !inbox.is_empty())
374            .map_err(|_| ParticipantPublicationError::InboxPoisoned)
375    }
376}
377
378#[derive(Debug)]
379struct ParticipantPublicationHandle {
380    inbox: Weak<Mutex<ReadyPublications>>,
381    waker: ReadyWaker,
382}
383
384/// Server-wide incarnation-to-connection publication registry.
385#[derive(Debug, Default)]
386pub struct ParticipantPublicationRegistry {
387    registrations: Mutex<BTreeMap<ConnectionIncarnation, ParticipantPublicationHandle>>,
388    #[cfg(test)]
389    ready_fires: AtomicU64,
390}
391
392impl ParticipantPublicationRegistry {
393    /// Registers one connection-owned inbox. A stale weak entry may be replaced;
394    /// a second live owner for one durable incarnation is a typed invariant fault.
395    pub(crate) fn register(
396        &self,
397        incarnation: ConnectionIncarnation,
398        inbox: &ParticipantPublicationInbox,
399        waker: ReadyWaker,
400    ) -> Result<(), ParticipantPublicationError> {
401        let mut registrations = self
402            .registrations
403            .lock()
404            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
405        if registrations
406            .get(&incarnation)
407            .is_some_and(|existing| existing.inbox.strong_count() > 0)
408        {
409            return Err(ParticipantPublicationError::DuplicateRegistration { incarnation });
410        }
411        registrations.insert(
412            incarnation,
413            ParticipantPublicationHandle {
414                inbox: inbox.weak(),
415                waker,
416            },
417        );
418        drop(registrations);
419        Ok(())
420    }
421
422    /// Removes the registration at explicit process teardown. The weak handle
423    /// already makes stale delivery harmless; eager removal keeps lookup exact.
424    pub(crate) fn deregister(&self, incarnation: ConnectionIncarnation) {
425        if let Ok(mut registrations) = self.registrations.lock() {
426            registrations.remove(&incarnation);
427        }
428    }
429
430    /// Captures the weak exact-live-connection target for an accepted observer
431    /// recovery arm. Missing or already-dead registrations yield no target;
432    /// callers must never substitute or broadcast.
433    pub(crate) fn observer_target(
434        &self,
435        incarnation: ConnectionIncarnation,
436    ) -> Result<Option<ObserverPublicationTarget>, ParticipantPublicationError> {
437        let registrations = self
438            .registrations
439            .lock()
440            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
441        let target = registrations.get(&incarnation).and_then(|handle| {
442            (handle.inbox.strong_count() > 0).then(|| ObserverPublicationTarget {
443                inbox: Weak::clone(&handle.inbox),
444                waker: handle.waker.clone(),
445            })
446        });
447        drop(registrations);
448        Ok(target)
449    }
450
451    /// Coalesces one conversation into the exact live incarnation's inbox and
452    /// fires READY only on the empty-to-nonempty edge.
453    ///
454    /// Returns `false` when the incarnation or connection process is gone.
455    pub(crate) fn notify(
456        &self,
457        incarnation: ConnectionIncarnation,
458        conversation_id: ConversationId,
459    ) -> Result<bool, ParticipantPublicationError> {
460        let (weak_inbox, waker) = {
461            let registrations = self
462                .registrations
463                .lock()
464                .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
465            let Some(handle) = registrations.get(&incarnation) else {
466                return Ok(false);
467            };
468            let weak_inbox = Weak::clone(&handle.inbox);
469            let waker = handle.waker.clone();
470            drop(registrations);
471            (weak_inbox, waker)
472        };
473        let Some(inbox) = weak_inbox.upgrade() else {
474            self.deregister(incarnation);
475            return Ok(false);
476        };
477        let should_wake = {
478            let mut inbox = inbox
479                .lock()
480                .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
481            if inbox.conversations.contains(&conversation_id) {
482                return Ok(true);
483            }
484            let occupied = u64::try_from(inbox.conversations.len()).unwrap_or(u64::MAX);
485            if occupied >= inbox.limit {
486                return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
487            }
488            let was_empty = inbox.is_empty();
489            inbox.conversations.insert(conversation_id);
490            was_empty
491        };
492        if should_wake {
493            #[cfg(test)]
494            self.ready_fires.fetch_add(1, Ordering::SeqCst);
495            waker.fire();
496        }
497        Ok(true)
498    }
499
500    #[cfg(test)]
501    pub(crate) fn ready_fire_count(&self) -> u64 {
502        self.ready_fires.load(Ordering::SeqCst)
503    }
504}
505
506#[cfg(test)]
507mod tests {
508    use std::sync::Arc;
509    use std::sync::atomic::{AtomicU64, Ordering};
510
511    use liminal_protocol::wire::ConnectionIncarnation;
512
513    use super::{ParticipantPublicationInbox, ParticipantPublicationRegistry};
514    use crate::server::connection::ReadyWaker;
515
516    #[test]
517    fn parked_connection_wakes_on_outbox_and_no_polling_occurs()
518    -> Result<(), Box<dyn std::error::Error>> {
519        let incarnation = ConnectionIncarnation::new(12, 34);
520        let wake_count = Arc::new(AtomicU64::new(0));
521        let registry = ParticipantPublicationRegistry::default();
522        let inbox = ParticipantPublicationInbox::new(3);
523        registry.register(
524            incarnation,
525            &inbox,
526            ReadyWaker::for_test(Arc::clone(&wake_count)),
527        )?;
528
529        assert!(!inbox.has_pending()?);
530        assert!(registry.notify(incarnation, 7)?);
531        assert_eq!(wake_count.load(Ordering::SeqCst), 1);
532        assert!(inbox.has_pending()?);
533
534        // Duplicate and additional ready conversations coalesce behind the one
535        // empty-to-nonempty wake; no repeated probe or timer drives progress.
536        assert!(registry.notify(incarnation, 7)?);
537        assert!(registry.notify(incarnation, 8)?);
538        assert_eq!(wake_count.load(Ordering::SeqCst), 1);
539        let ready = inbox.take_ready()?;
540        assert_eq!(ready.conversations, vec![7, 8]);
541        assert!(ready.observer_progressed.is_empty());
542        assert!(ready.marker_settled.is_empty());
543        assert!(!inbox.has_pending()?);
544
545        // This notification models the execute-to-wait race: it lands after a
546        // drain but before the process final probe. The non-consuming probe sees
547        // it and the edge fires exactly one new READY.
548        assert!(registry.notify(incarnation, 9)?);
549        assert!(inbox.has_pending()?);
550        assert_eq!(wake_count.load(Ordering::SeqCst), 2);
551        let ready = inbox.take_ready()?;
552        assert_eq!(ready.conversations, vec![9]);
553        assert!(ready.observer_progressed.is_empty());
554        assert!(!inbox.has_pending()?);
555
556        let idle_count = wake_count.load(Ordering::SeqCst);
557        assert_eq!(wake_count.load(Ordering::SeqCst), idle_count);
558        registry.deregister(incarnation);
559        assert!(!registry.notify(incarnation, 10)?);
560        assert_eq!(wake_count.load(Ordering::SeqCst), idle_count);
561        Ok(())
562    }
563}
liminal_server/server/participant/publication.rs

liminal_server/server/participant/
publication.rs
