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