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            warn_if_undelivered(self.waker.fire(), "observer progress");
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            warn_if_undelivered(self.waker.fire(), "marker settled");
186        }
187        Ok(true)
188    }
189}
190
191/// Reports a READY marker the connection scheduler did not accept.
192///
193/// The caller returns `Ok(true)` — the publication IS queued in the inbox — so
194/// an undelivered wake leaves that work with nothing scheduled to drain it.
195/// `fire()` returning `false` normally means the connection is already tearing
196/// down, where the lost wake is correct (R5); the point of saying so is that
197/// the benign case and a genuinely stranded publication are otherwise
198/// indistinguishable from outside.
199fn warn_if_undelivered(delivered: bool, publication_kind: &'static str) {
200    if !delivered {
201        tracing::warn!(
202            publication_kind,
203            "participant ready wake was not delivered; the publication is queued with \
204             no scheduled drain (usual cause: the connection is already tearing down)"
205        );
206    }
207}
208
209#[derive(Clone, Copy, Debug, thiserror::Error)]
210pub enum ParticipantPublicationError {
211    /// A live incarnation was registered more than once.
212    #[error("participant publication incarnation {incarnation:?} is already registered")]
213    DuplicateRegistration {
214        /// Durable incarnation that already owns a live inbox.
215        incarnation: ConnectionIncarnation,
216    },
217    /// The connection-owned inbox mutex was poisoned.
218    #[error("participant publication inbox is poisoned")]
219    InboxPoisoned,
220    /// A new ready conversation would exceed the signed connection bound.
221    #[error("participant publication inbox exceeds its signed conversation bound {limit}")]
222    InboxCapacity {
223        /// Signed maximum semantic conversations for the connection.
224        limit: u64,
225    },
226}
227
228#[derive(Debug)]
229struct ReadyPublications {
230    limit: u64,
231    conversations: BTreeSet<ConversationId>,
232    observer_progressed: BTreeMap<ConversationId, ObserverPublication>,
233    marker_settled: BTreeMap<ConversationId, MarkerSettledPublication>,
234}
235
236impl ReadyPublications {
237    fn is_empty(&self) -> bool {
238        self.conversations.is_empty()
239            && self.observer_progressed.is_empty()
240            && self.marker_settled.is_empty()
241    }
242}
243
244/// All participant and observer work atomically removed for one shared push
245/// slice. The pump merges these collections by conversation before applying the
246/// single signed budget.
247#[derive(Debug)]
248pub struct ReadyPublicationBatch {
249    pub(crate) conversations: Vec<ConversationId>,
250    pub(crate) observer_progressed: Vec<ObserverPublication>,
251    pub(crate) marker_settled: Vec<MarkerSettledPublication>,
252}
253
254/// Inbox strongly owned by exactly one connection process.
255///
256/// The value is intentionally not `Clone`: the registry receives only a weak
257/// projection and cannot become another owner. Participant readiness coalesces
258/// conversation ids; observer readiness retains only the latest fired payload
259/// per conversation.
260#[derive(Debug)]
261pub struct ParticipantPublicationInbox {
262    inner: Arc<Mutex<ReadyPublications>>,
263}
264
265impl ParticipantPublicationInbox {
266    /// Creates the connection-owned bounded ready set from the signed semantic
267    /// conversation limit.
268    #[must_use]
269    pub(crate) fn new(limit: u64) -> Self {
270        Self {
271            inner: Arc::new(Mutex::new(ReadyPublications {
272                limit,
273                conversations: BTreeSet::new(),
274                observer_progressed: BTreeMap::new(),
275                marker_settled: BTreeMap::new(),
276            })),
277        }
278    }
279
280    fn weak(&self) -> Weak<Mutex<ReadyPublications>> {
281        Arc::downgrade(&self.inner)
282    }
283
284    /// Atomically removes all sorted, coalesced work for one shared push slice.
285    pub(crate) fn take_ready(&self) -> Result<ReadyPublicationBatch, ParticipantPublicationError> {
286        let mut inbox = self
287            .inner
288            .lock()
289            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
290        Ok(ReadyPublicationBatch {
291            conversations: std::mem::take(&mut inbox.conversations)
292                .into_iter()
293                .collect(),
294            observer_progressed: std::mem::take(&mut inbox.observer_progressed)
295                .into_values()
296                .collect(),
297            marker_settled: std::mem::take(&mut inbox.marker_settled)
298                .into_values()
299                .collect(),
300        })
301    }
302
303    /// Requeues deferred conversations after a budget-limited or held-back
304    /// slice. Existing ids remain coalesced.
305    pub(crate) fn requeue(
306        &self,
307        conversations: impl IntoIterator<Item = ConversationId>,
308    ) -> Result<(), ParticipantPublicationError> {
309        let mut inbox = self
310            .inner
311            .lock()
312            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
313        for conversation_id in conversations {
314            if inbox.conversations.contains(&conversation_id) {
315                continue;
316            }
317            let occupied = u64::try_from(inbox.conversations.len()).unwrap_or(u64::MAX);
318            if occupied >= inbox.limit {
319                return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
320            }
321            inbox.conversations.insert(conversation_id);
322        }
323        drop(inbox);
324        Ok(())
325    }
326
327    /// Requeues budget-deferred observer payloads only for vacant conversations.
328    /// Any incumbent arrived after the pump's take and supersedes the deferred
329    /// payload; skipping it consumes no signed conversation capacity.
330    pub(crate) fn requeue_observers(
331        &self,
332        publications: impl IntoIterator<Item = ObserverPublication>,
333    ) -> Result<(), ParticipantPublicationError> {
334        let mut inbox = self
335            .inner
336            .lock()
337            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
338        for publication in publications {
339            if inbox
340                .observer_progressed
341                .contains_key(&publication.conversation_id)
342            {
343                continue;
344            }
345            let occupied = u64::try_from(inbox.observer_progressed.len()).unwrap_or(u64::MAX);
346            if occupied >= inbox.limit {
347                return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
348            }
349            inbox
350                .observer_progressed
351                .insert(publication.conversation_id, publication);
352        }
353        drop(inbox);
354        Ok(())
355    }
356
357    /// Requeues budget-deferred settlement wakes only for vacant conversations.
358    /// An incumbent arrived after the pump's take and supersedes the deferred
359    /// payload, exactly as on the observer lane.
360    pub(crate) fn requeue_marker_settled(
361        &self,
362        publications: impl IntoIterator<Item = MarkerSettledPublication>,
363    ) -> Result<(), ParticipantPublicationError> {
364        let mut inbox = self
365            .inner
366            .lock()
367            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
368        for publication in publications {
369            if inbox
370                .marker_settled
371                .contains_key(&publication.conversation_id)
372            {
373                continue;
374            }
375            let occupied = u64::try_from(inbox.marker_settled.len()).unwrap_or(u64::MAX);
376            if occupied >= inbox.limit {
377                return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
378            }
379            inbox
380                .marker_settled
381                .insert(publication.conversation_id, publication);
382        }
383        drop(inbox);
384        Ok(())
385    }
386
387    /// Non-consuming final-probe fact used after socket readiness is armed.
388    pub(crate) fn has_pending(&self) -> Result<bool, ParticipantPublicationError> {
389        self.inner
390            .lock()
391            .map(|inbox| !inbox.is_empty())
392            .map_err(|_| ParticipantPublicationError::InboxPoisoned)
393    }
394}
395
396#[derive(Debug)]
397struct ParticipantPublicationHandle {
398    inbox: Weak<Mutex<ReadyPublications>>,
399    waker: ReadyWaker,
400}
401
402/// Server-wide incarnation-to-connection publication registry.
403#[derive(Debug, Default)]
404pub struct ParticipantPublicationRegistry {
405    registrations: Mutex<BTreeMap<ConnectionIncarnation, ParticipantPublicationHandle>>,
406    #[cfg(test)]
407    ready_fires: AtomicU64,
408}
409
410impl ParticipantPublicationRegistry {
411    /// Registers one connection-owned inbox. A stale weak entry may be replaced;
412    /// a second live owner for one durable incarnation is a typed invariant fault.
413    pub(crate) fn register(
414        &self,
415        incarnation: ConnectionIncarnation,
416        inbox: &ParticipantPublicationInbox,
417        waker: ReadyWaker,
418    ) -> Result<(), ParticipantPublicationError> {
419        let mut registrations = self
420            .registrations
421            .lock()
422            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
423        if registrations
424            .get(&incarnation)
425            .is_some_and(|existing| existing.inbox.strong_count() > 0)
426        {
427            return Err(ParticipantPublicationError::DuplicateRegistration { incarnation });
428        }
429        registrations.insert(
430            incarnation,
431            ParticipantPublicationHandle {
432                inbox: inbox.weak(),
433                waker,
434            },
435        );
436        drop(registrations);
437        Ok(())
438    }
439
440    /// Removes the registration at explicit process teardown. The weak handle
441    /// already makes stale delivery harmless; eager removal keeps lookup exact.
442    pub(crate) fn deregister(&self, incarnation: ConnectionIncarnation) {
443        if let Ok(mut registrations) = self.registrations.lock() {
444            registrations.remove(&incarnation);
445        }
446    }
447
448    /// Captures the weak exact-live-connection target for an accepted observer
449    /// recovery arm. Missing or already-dead registrations yield no target;
450    /// callers must never substitute or broadcast.
451    pub(crate) fn observer_target(
452        &self,
453        incarnation: ConnectionIncarnation,
454    ) -> Result<Option<ObserverPublicationTarget>, ParticipantPublicationError> {
455        let registrations = self
456            .registrations
457            .lock()
458            .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
459        let target = registrations.get(&incarnation).and_then(|handle| {
460            (handle.inbox.strong_count() > 0).then(|| ObserverPublicationTarget {
461                inbox: Weak::clone(&handle.inbox),
462                waker: handle.waker.clone(),
463            })
464        });
465        drop(registrations);
466        Ok(target)
467    }
468
469    /// Coalesces one conversation into the exact live incarnation's inbox and
470    /// fires READY only on the empty-to-nonempty edge.
471    ///
472    /// Returns `false` when the incarnation or connection process is gone.
473    pub(crate) fn notify(
474        &self,
475        incarnation: ConnectionIncarnation,
476        conversation_id: ConversationId,
477    ) -> Result<bool, ParticipantPublicationError> {
478        let (weak_inbox, waker) = {
479            let registrations = self
480                .registrations
481                .lock()
482                .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
483            let Some(handle) = registrations.get(&incarnation) else {
484                return Ok(false);
485            };
486            let weak_inbox = Weak::clone(&handle.inbox);
487            let waker = handle.waker.clone();
488            drop(registrations);
489            (weak_inbox, waker)
490        };
491        let Some(inbox) = weak_inbox.upgrade() else {
492            self.deregister(incarnation);
493            return Ok(false);
494        };
495        let should_wake = {
496            let mut inbox = inbox
497                .lock()
498                .map_err(|_| ParticipantPublicationError::InboxPoisoned)?;
499            if inbox.conversations.contains(&conversation_id) {
500                return Ok(true);
501            }
502            let occupied = u64::try_from(inbox.conversations.len()).unwrap_or(u64::MAX);
503            if occupied >= inbox.limit {
504                return Err(ParticipantPublicationError::InboxCapacity { limit: inbox.limit });
505            }
506            let was_empty = inbox.is_empty();
507            inbox.conversations.insert(conversation_id);
508            was_empty
509        };
510        if should_wake {
511            #[cfg(test)]
512            self.ready_fires.fetch_add(1, Ordering::SeqCst);
513            warn_if_undelivered(waker.fire(), "conversation ready");
514        }
515        Ok(true)
516    }
517
518    #[cfg(test)]
519    pub(crate) fn ready_fire_count(&self) -> u64 {
520        self.ready_fires.load(Ordering::SeqCst)
521    }
522}
523
524#[cfg(test)]
525mod tests {
526    use std::sync::Arc;
527    use std::sync::atomic::{AtomicU64, Ordering};
528
529    use liminal_protocol::wire::ConnectionIncarnation;
530
531    use super::{ParticipantPublicationInbox, ParticipantPublicationRegistry};
532    use crate::server::connection::ReadyWaker;
533
534    #[test]
535    fn parked_connection_wakes_on_outbox_and_no_polling_occurs()
536    -> Result<(), Box<dyn std::error::Error>> {
537        let incarnation = ConnectionIncarnation::new(12, 34);
538        let wake_count = Arc::new(AtomicU64::new(0));
539        let registry = ParticipantPublicationRegistry::default();
540        let inbox = ParticipantPublicationInbox::new(3);
541        registry.register(
542            incarnation,
543            &inbox,
544            ReadyWaker::for_test(Arc::clone(&wake_count)),
545        )?;
546
547        assert!(!inbox.has_pending()?);
548        assert!(registry.notify(incarnation, 7)?);
549        assert_eq!(wake_count.load(Ordering::SeqCst), 1);
550        assert!(inbox.has_pending()?);
551
552        // Duplicate and additional ready conversations coalesce behind the one
553        // empty-to-nonempty wake; no repeated probe or timer drives progress.
554        assert!(registry.notify(incarnation, 7)?);
555        assert!(registry.notify(incarnation, 8)?);
556        assert_eq!(wake_count.load(Ordering::SeqCst), 1);
557        let ready = inbox.take_ready()?;
558        assert_eq!(ready.conversations, vec![7, 8]);
559        assert!(ready.observer_progressed.is_empty());
560        assert!(ready.marker_settled.is_empty());
561        assert!(!inbox.has_pending()?);
562
563        // This notification models the execute-to-wait race: it lands after a
564        // drain but before the process final probe. The non-consuming probe sees
565        // it and the edge fires exactly one new READY.
566        assert!(registry.notify(incarnation, 9)?);
567        assert!(inbox.has_pending()?);
568        assert_eq!(wake_count.load(Ordering::SeqCst), 2);
569        let ready = inbox.take_ready()?;
570        assert_eq!(ready.conversations, vec![9]);
571        assert!(ready.observer_progressed.is_empty());
572        assert!(!inbox.has_pending()?);
573
574        let idle_count = wake_count.load(Ordering::SeqCst);
575        assert_eq!(wake_count.load(Ordering::SeqCst), idle_count);
576        registry.deregister(incarnation);
577        assert!(!registry.notify(incarnation, 10)?);
578        assert_eq!(wake_count.load(Ordering::SeqCst), idle_count);
579        Ok(())
580    }
581
582    /// A waker whose connection scheduler is gone: `fire()` reports `false`.
583    /// This is the real production shape — the waker holds the scheduler weakly
584    /// precisely so it can outlive it — reached here by dropping the scheduler.
585    fn dead_waker() -> Result<ReadyWaker, Box<dyn std::error::Error>> {
586        let scheduler = Arc::new(
587            beamr::scheduler::Scheduler::with_services(
588                beamr::scheduler::SchedulerConfig {
589                    thread_count: Some(1),
590                    ..beamr::scheduler::SchedulerConfig::default()
591                },
592                beamr::scheduler::SchedulerServices::from_config().owned_readiness(),
593                Arc::new(beamr::module::ModuleRegistry::new()),
594                beamr::scheduler::NativeBifs::none(),
595            )
596            .map_err(|message| format!("scheduler startup failed: {message}"))?,
597        );
598        let waker = ReadyWaker::new(
599            &scheduler,
600            1,
601            beamr::atom::Atom::OK,
602            Arc::new(std::sync::atomic::AtomicBool::new(false)),
603        );
604        drop(scheduler);
605        assert!(
606            !waker.fire(),
607            "a scheduler-less waker must report no delivery"
608        );
609        Ok(waker)
610    }
611
612    /// Both publication paths return `Ok(true)` — "accepted, and the connection
613    /// will be woken" — while discarding `fire()`'s report that the wake was NOT
614    /// delivered. The queued work then sits in the inbox with nothing scheduled
615    /// to drain it. Usually that is a connection already tearing down and the
616    /// lost wake is correct, but it is never currently DISTINGUISHABLE from a
617    /// stranded publication, because nothing is said either way.
618    #[test]
619    fn an_undelivered_ready_wake_is_reported() -> Result<(), Box<dyn std::error::Error>> {
620        let incarnation = ConnectionIncarnation::new(21, 5);
621        let registry = ParticipantPublicationRegistry::default();
622        let inbox = ParticipantPublicationInbox::new(3);
623        registry.register(incarnation, &inbox, dead_waker()?)?;
624
625        // Leg 1: the conversation-ready notification.
626        let log = crate::test_log::CapturedLog::default();
627        let accepted = log.capture(|| registry.notify(incarnation, 7))?;
628        assert!(accepted, "the publication is still accepted into the inbox");
629        assert!(inbox.has_pending()?, "and it is genuinely queued");
630        let output = log.contents();
631        assert!(
632            output.contains("WARN") && output.contains("ready wake was not delivered"),
633            "an undelivered wake must be reported, got: {output}"
634        );
635
636        // Leg 2: the observer-progress transfer, which fires the same waker.
637        let ready = inbox.take_ready()?;
638        assert_eq!(ready.conversations, vec![7]);
639        let target = registry
640            .observer_target(incarnation)?
641            .ok_or("the live registration must yield an observer target")?;
642        let log = crate::test_log::CapturedLog::default();
643        let accepted = log.capture(|| {
644            target.publish(super::ObserverPublication {
645                conversation_id: 7,
646                refused_epoch: 1,
647                observer_progress: 2,
648            })
649        })?;
650        assert!(accepted, "the observer publication is still accepted");
651        let output = log.contents();
652        assert!(
653            output.contains("WARN") && output.contains("ready wake was not delivered"),
654            "an undelivered observer wake must be reported, got: {output}"
655        );
656        Ok(())
657    }
658}
liminal_server/server/participant/publication.rs

liminal_server/server/participant/
publication.rs
