liminal_server/server/connection/notifier.rs
1//! Connection-keyed notifier hook for worker registration lifecycle.
2//!
3//! This is the application seam for self-describing worker registration. When a
4//! worker sends a [`Frame::WorkerRegister`](liminal::protocol::Frame) over its
5//! established connection, the server associates the registration with the
6//! connection's beamr process id and invokes the configured
7//! [`ConnectionNotifier`]; on connection close it invokes the matching
8//! deregistration. The notifier is connection-keyed (by pid), which is distinct
9//! from the subject-keyed responder registry in [`super::services`].
10//!
11//! Keeping the hook a `liminal-server` trait — rather than a liminal-core
12//! concern — preserves liminal's generality: liminal still runs standalone with
13//! no notifier configured, and the application (aion, in Stage 2) plugs its
14//! registry in without liminal depending on it.
15//!
16//! The same trait carries the presence seam for pass-stamped connections: a
17//! connection whose `Connect` verified a registry pass (a browser or agent
18//! seat, which sends no `WorkerRegister`) announces itself once on attach and
19//! once on detach through [`ConnectionNotifier::on_pass_attached`] and
20//! [`ConnectionNotifier::on_pass_detached`], so an application derives
21//! presence from those two events and never polls for it.
22
23use liminal::protocol::WorkerRegistration;
24
25use crate::ServerError;
26use crate::auth_pass::{PassPrincipal, VerifiedHolder};
27
28/// Application hook invoked when a worker registers or unregisters on a
29/// connection.
30///
31/// Implementations associate the connection's beamr process id (`pid`) with the
32/// worker's declared [`WorkerRegistration`] so the application can route work to
33/// it, and release that association on disconnect. The hook is synchronous: a
34/// registration is acknowledged to the worker only after
35/// [`on_worker_registered`](Self::on_worker_registered) returns, so a rejecting
36/// application surfaces a `Rejected` ack instead of leaving the worker silently
37/// connected but never dispatched-to.
38pub trait ConnectionNotifier: std::fmt::Debug + Send + Sync {
39 /// Called when a worker registers on the connection identified by `pid`.
40 ///
41 /// Returning `Ok(())` accepts the registration (the worker receives an
42 /// `Accepted` ack). Returning [`ServerError`] rejects it (the worker receives
43 /// a `Rejected` ack carrying the error text), so a failed association never
44 /// leaves the worker believing it is registered.
45 ///
46 /// # Errors
47 /// Returns [`ServerError`] when the application declines the registration.
48 fn on_worker_registered(
49 &self,
50 pid: u64,
51 registration: &WorkerRegistration,
52 ) -> Result<(), ServerError>;
53
54 /// Called when a worker registers on the connection identified by `pid`,
55 /// with the holder that connection's `Connect` verified, or `None` for a
56 /// connection that proved no one (a bearer, or an open server). The
57 /// server calls this hook, never [`on_worker_registered`] directly
58 /// (ACCESS-004 R4).
59 ///
60 /// When it is called, the registration's name is the holder's: a
61 /// mismatch was refused before any hook ran. The default forwards to
62 /// [`on_worker_registered`], so an existing implementor compiles and
63 /// behaves unchanged; one that decides rights by holder overrides this.
64 ///
65 /// [`on_worker_registered`]: Self::on_worker_registered
66 ///
67 /// # Errors
68 /// Returns [`ServerError`] when the application declines the registration.
69 fn on_worker_registered_by(
70 &self,
71 pid: u64,
72 registration: &WorkerRegistration,
73 _holder: Option<&VerifiedHolder>,
74 ) -> Result<(), ServerError> {
75 self.on_worker_registered(pid, registration)
76 }
77
78 /// Called when the connection identified by `pid` — which had a stored
79 /// registration — closes, so the application can release the association.
80 ///
81 /// Deregistration is best-effort and infallible from the connection's
82 /// perspective: it runs on the close path where there is no peer to report an
83 /// error to.
84 fn on_worker_unregistered(&self, pid: u64);
85
86 /// Called when the connection identified by `pid` publishes to `channel`,
87 /// carrying the opaque envelope `payload`, BEFORE the normal channel fan-out.
88 ///
89 /// Returns `true` when the application CONSUMED the publish out-of-band (an
90 /// observability-drain tap): the connection process then does NOT route it to the
91 /// channel-fan-out cluster and answers with no wire response, so a tapped channel
92 /// need not be a declared fan-out channel. Returns `false` (the default) to let
93 /// the publish flow through the normal channel machinery unchanged.
94 ///
95 /// This is the observability-drain hook: a worker publishing an agent transcript
96 /// event to the reserved observability channel is consumed here — the hosting
97 /// application (aion) persists and live-fans-out the event without a second
98 /// connection. It is fire-and-forget: a publish is a one-way notification, so
99 /// there is no reply and a failed persist is the application's concern to log.
100 ///
101 /// The default returns `false`, so liminal still runs standalone: with no
102 /// notifier, or a notifier that does not recognise the channel, every publish
103 /// routes to the normal fan-out exactly as before.
104 fn on_channel_publish(&self, pid: u64, channel: &str, payload: &[u8]) -> bool {
105 let _ = (pid, channel, payload);
106 false
107 }
108
109 /// Called exactly once when a `Connect` carrying a registry pass has
110 /// SUCCEEDED on the connection identified by `pid` — after the pass
111 /// verified, version negotiation passed, and the handshake was admitted —
112 /// carrying the principal the pass stamped: participant, public key, live
113 /// prefix, conversation scope, `may_enroll`. `pid` is the connection's own
114 /// id, the same one every other hook on this trait is keyed by, so the
115 /// matching [`on_pass_detached`](Self::on_pass_detached) can be paired.
116 ///
117 /// A `Connect` that presents a pass and is refused (or a bearer connection,
118 /// which stamps no principal) never reaches this hook. The default does
119 /// nothing, so existing implementors compile unchanged and liminal still
120 /// runs standalone.
121 fn on_pass_attached(&self, pid: u64, principal: &PassPrincipal) {
122 let _ = (pid, principal);
123 }
124
125 /// Called exactly once when a connection that announced
126 /// [`on_pass_attached`](Self::on_pass_attached) closes — for any reason: a
127 /// clean `Disconnect`, the peer ending the transport, a transport error, a
128 /// process crash, or supervisor shutdown — carrying the same `pid` and the
129 /// principal the attach carried. Best-effort and infallible like
130 /// [`on_worker_unregistered`](Self::on_worker_unregistered): it runs on the
131 /// close path where there is no peer to report to. The default does
132 /// nothing.
133 fn on_pass_detached(&self, pid: u64, principal: &PassPrincipal) {
134 let _ = (pid, principal);
135 }
136
137 /// Called after a conversation's checkpoint crossing commits, every
138 /// `CHECKPOINT_EVERY` operations, with what the conversation's outbox
139 /// still owes each member. It is the layer above's one signal for its
140 /// wedged-participant policy (PARTICIPANT-CONTRACT R-B5): a member owed
141 /// more than the room's retention can hold is the application's to
142 /// retire, by that member's own detached Leave.
143 ///
144 /// It runs on the conversation's write path, under its owner: it must
145 /// return at once and must not call the server from inside itself. An
146 /// implementation that acts hands the crossing to its own task. The
147 /// default does nothing.
148 fn on_checkpoint_crossing(&self, crossing: &CheckpointCrossing) {
149 let _ = crossing;
150 }
151}
152
153/// One committed checkpoint crossing and its outbox census
154/// ([`ConnectionNotifier::on_checkpoint_crossing`]).
155#[derive(Clone, Debug, PartialEq, Eq)]
156pub struct CheckpointCrossing {
157 /// The conversation that crossed.
158 pub conversation_id: u64,
159 /// The log sequence the crossing's checkpoint covers through.
160 pub covered_through: u64,
161 /// The highest delivery sequence the conversation has assigned.
162 pub delivery_head: u64,
163 /// Every member the outbox still owes a record, ascending by id.
164 pub recipients: Vec<CrossingRecipient>,
165}
166
167/// What the outbox owes one member at a crossing.
168#[derive(Clone, Copy, Debug, PartialEq, Eq)]
169pub struct CrossingRecipient {
170 /// The member.
171 pub participant_id: u64,
172 /// The member's binding at the crossing.
173 pub binding: CrossingBinding,
174 /// Records still owed to it.
175 pub owed_records: u64,
176 /// Encoded push bytes of those records.
177 pub owed_bytes: u64,
178 /// The member's durable cumulative acknowledgement.
179 pub acknowledged_through: u64,
180}
181
182/// A member's binding at a crossing.
183#[derive(Clone, Copy, Debug, PartialEq, Eq)]
184pub enum CrossingBinding {
185 /// A live connection holds the member.
186 Bound,
187 /// No connection holds the member; it may resume.
188 Detached,
189 /// A terminal awaits its finalization.
190 PendingFinalization,
191}