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