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
16use liminal::protocol::WorkerRegistration;
17
18use crate::ServerError;
19
20/// Application hook invoked when a worker registers or unregisters on a
21/// connection.
22///
23/// Implementations associate the connection's beamr process id (`pid`) with the
24/// worker's declared [`WorkerRegistration`] so the application can route work to
25/// it, and release that association on disconnect. The hook is synchronous: a
26/// registration is acknowledged to the worker only after
27/// [`on_worker_registered`](Self::on_worker_registered) returns, so a rejecting
28/// application surfaces a `Rejected` ack instead of leaving the worker silently
29/// connected but never dispatched-to.
30pub trait ConnectionNotifier: std::fmt::Debug + Send + Sync {
31    /// Called when a worker registers on the connection identified by `pid`.
32    ///
33    /// Returning `Ok(())` accepts the registration (the worker receives an
34    /// `Accepted` ack). Returning [`ServerError`] rejects it (the worker receives
35    /// a `Rejected` ack carrying the error text), so a failed association never
36    /// leaves the worker believing it is registered.
37    ///
38    /// # Errors
39    /// Returns [`ServerError`] when the application declines the registration.
40    fn on_worker_registered(
41        &self,
42        pid: u64,
43        registration: &WorkerRegistration,
44    ) -> Result<(), ServerError>;
45
46    /// Called when the connection identified by `pid` — which had a stored
47    /// registration — closes, so the application can release the association.
48    ///
49    /// Deregistration is best-effort and infallible from the connection's
50    /// perspective: it runs on the close path where there is no peer to report an
51    /// error to.
52    fn on_worker_unregistered(&self, pid: u64);
53
54    /// Called when the connection identified by `pid` publishes to `channel`,
55    /// carrying the opaque envelope `payload`, BEFORE the normal channel fan-out.
56    ///
57    /// Returns `true` when the application CONSUMED the publish out-of-band (an
58    /// observability-drain tap): the connection process then does NOT route it to the
59    /// channel-fan-out cluster and answers with no wire response, so a tapped channel
60    /// need not be a declared fan-out channel. Returns `false` (the default) to let
61    /// the publish flow through the normal channel machinery unchanged.
62    ///
63    /// This is the observability-drain hook: a worker publishing an agent transcript
64    /// event to the reserved observability channel is consumed here — the hosting
65    /// application (aion) persists and live-fans-out the event without a second
66    /// connection. It is fire-and-forget: a publish is a one-way notification, so
67    /// there is no reply and a failed persist is the application's concern to log.
68    ///
69    /// The default returns `false`, so liminal still runs standalone: with no
70    /// notifier, or a notifier that does not recognise the channel, every publish
71    /// routes to the normal fan-out exactly as before.
72    fn on_channel_publish(&self, pid: u64, channel: &str, payload: &[u8]) -> bool {
73        let _ = (pid, channel, payload);
74        false
75    }
76}
liminal_server/server/connection/notifier.rs

liminal_server/server/connection/
notifier.rs
