liminal_server/server/connection/supervisor/supervisor_connections.rs
1use super::{
2 AdmissionRefusal, Arc, ConnectionHandle, ConnectionIncarnation, ConnectionRuntime,
3 ConnectionSupervisor, LoopbackConnectionProcess, LoopbackServerEnd, MountKind, Mutex,
4 NativeHandlerFactory, Scheduler, ServerError, SocketAddr, TcpStream,
5};
6
7impl ConnectionSupervisor {
8 /// Counts one refused admission, classified by reason.
9 ///
10 /// Hung on the THREE public admission doors — TCP accept, the sibling
11 /// transport spawn (WebSocket), and the in-process loopback — because those
12 /// are the three places a connection can be turned away, and each of them
13 /// reaches the shared inner body exactly once. Recording deeper would
14 /// double-count the loopback (which calls the inner body directly);
15 /// recording shallower would miss the doors that only log.
16 fn record_admission_refusal(error: &ServerError) {
17 crate::metrics::admission_refused(AdmissionRefusal::classify(error));
18 }
19
20 /// Spawns one supervised beamr process that owns `stream`.
21 ///
22 /// # Errors
23 /// Returns [`ServerError`] when stream configuration or beamr spawn fails.
24 pub fn spawn_connection(&self, stream: TcpStream) -> Result<ConnectionHandle, ServerError> {
25 self.inner
26 .spawn_connection(stream)
27 .inspect_err(Self::record_admission_refusal)
28 }
29
30 /// Returns the underlying beamr scheduler.
31 #[must_use]
32 pub fn scheduler(&self) -> Arc<Scheduler> {
33 Arc::clone(&self.inner.scheduler)
34 }
35
36 /// Reaps connection processes that have exited outside the normal handler path.
37 #[must_use]
38 pub fn reap_crashed_connections(&self) -> usize {
39 self.inner.runtime.reap_crashed(&self.inner.scheduler)
40 }
41
42 /// Returns true when `pid` is still tracked by the supervisor.
43 #[must_use]
44 pub fn is_tracked(&self, pid: u64) -> bool {
45 self.inner.runtime.contains(pid)
46 }
47
48 /// Returns the number of tracked live connections.
49 #[must_use]
50 pub fn active_connection_count(&self) -> usize {
51 self.inner.runtime.active_count()
52 }
53
54 /// Returns the beamr process ids of the currently tracked live connections.
55 ///
56 /// Useful for addressing a specific connection — e.g. as the `pid` argument to
57 /// [`push_to_connection`](Self::push_to_connection) when the caller knows there
58 /// is a single connected client.
59 #[must_use]
60 pub fn active_connection_pids(&self) -> Vec<u64> {
61 self.inner
62 .runtime
63 .active_connections()
64 .into_iter()
65 .map(|connection| connection.pid)
66 .collect()
67 }
68
69 /// SRV-027 R4: how many keyed lookups the service directory has made,
70 /// read by the count pins.
71 #[must_use]
72 pub fn directory_lookups(&self) -> u64 {
73 self.inner.runtime.directory_lookups()
74 }
75
76 /// LP-WS-TRANSPORT R1.3 sibling-transport spawn seam (ADDITIVE ONLY).
77 ///
78 /// Admits, allocates a durable connection incarnation for, spawns, and
79 /// registers a connection process whose handler is built by `build_factory`
80 /// over this supervisor's shared [`ConnectionRuntime`]. The WebSocket
81 /// sibling acceptor uses this so its connections share the ONE §5
82 /// `max_connections` admission bound, the one incarnation authority, the
83 /// one registry (controls, pushes, crash reap, drain, forced close), and the
84 /// one `apply_frame` seam with TCP connections. The TCP accept path above
85 /// (`spawn_connection`) is byte-for-byte untouched and never calls this.
86 ///
87 /// `fd_guard` is a host-held duplicate of the connection's underlying
88 /// socket, exactly like the TCP path's: it keeps the fd alive until the
89 /// single record-removal path has synchronously deregistered readiness. It
90 /// is `None` for a transport that owns no descriptor.
91 ///
92 /// `mount` is the admitting door's own name for itself (design §10), which
93 /// is why this seam takes it rather than deriving it: the caller IS the
94 /// door, and no other party — least of all the client — has any input.
95 ///
96 /// This method exists because Rust module privacy makes the runtime,
97 /// admission counter, incarnation authority, and registry unreachable from
98 /// the sibling `websocket` module family; it is the narrow additive seam
99 /// that shares them without generalizing any TCP hot path.
100 ///
101 /// # Errors
102 /// Returns [`ServerError`] when admission is refused
103 /// ([`ServerError::ConnectionLimitReached`]), incarnation allocation fails,
104 /// or beamr spawn/registration fails.
105 pub(in crate::server::connection) fn spawn_transport_connection(
106 &self,
107 peer_addr: Option<SocketAddr>,
108 fd_guard: Option<TcpStream>,
109 mount: MountKind,
110 build_factory: &dyn Fn(
111 Arc<ConnectionRuntime>,
112 Option<ConnectionIncarnation>,
113 ) -> NativeHandlerFactory,
114 ) -> Result<ConnectionHandle, ServerError> {
115 self.inner
116 .spawn_transport_connection(peer_addr, fd_guard, mount, build_factory)
117 .inspect_err(Self::record_admission_refusal)
118 }
119
120 /// Admits one in-process connection over `server_end` (design §8 step 3).
121 ///
122 /// This replaces exactly the listener's `accept()` + `spawn_connection`
123 /// pair, and NOTHING else about admission. It runs the same
124 /// `try_reserve_admission` against the same §5 slot pool — an in-process
125 /// connect at capacity is refused with the same typed
126 /// [`ServerError::ConnectionLimitReached`] a socket connect is — allocates a
127 /// real durable [`ConnectionIncarnation`] from the same authority, so
128 /// participant binding, resume, and fate records work identically, and
129 /// registers the same record. Only two fields differ, and both are honest
130 /// descriptions rather than semantics: no fd guard (there is no descriptor
131 /// to keep alive) and `peer_addr: None` (there is no socket to have an
132 /// address). Nothing on this path reads a socket fact.
133 ///
134 /// The connection's wake is installed by the process itself on its first
135 /// serviced slice, which is the earliest point at which its pid and host
136 /// record — the two things the wake names — both exist.
137 ///
138 /// `pub(crate)` rather than `pub`: the embedding handle that grants
139 /// loopback connections lives in this crate, and widening the surface
140 /// further would hand an outside caller a way to reach the runtime that
141 /// module privacy currently denies it.
142 ///
143 /// # Errors
144 /// Returns [`ServerError`] when admission is refused
145 /// ([`ServerError::ConnectionLimitReached`]), incarnation allocation fails,
146 /// or beamr spawn/registration fails.
147 pub(crate) fn spawn_loopback_connection(
148 &self,
149 server_end: LoopbackServerEnd,
150 ) -> Result<ConnectionHandle, ServerError> {
151 // The same interior-mutability handoff the socket path uses: the native
152 // handler factory is `Fn + Send + Sync`, so the duplex end cannot be
153 // moved into it and the FIRST handler build takes it out exactly once.
154 let holder = Arc::new(Mutex::new(Some(server_end)));
155 let build = move |runtime: Arc<ConnectionRuntime>,
156 incarnation: Option<ConnectionIncarnation>|
157 -> NativeHandlerFactory {
158 let holder = Arc::clone(&holder);
159 Box::new(move || {
160 Box::new(LoopbackConnectionProcess::from_loopback_holder(
161 Arc::clone(&runtime),
162 &holder,
163 incarnation,
164 ))
165 })
166 };
167 self.inner
168 .spawn_transport_connection(None, None, MountKind::Loopback, &build)
169 .inspect_err(Self::record_admission_refusal)
170 }
171}