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
use super::{
AdmissionRefusal, Arc, ConnectionHandle, ConnectionIncarnation, ConnectionRuntime,
ConnectionSupervisor, LoopbackConnectionProcess, LoopbackServerEnd, MountKind, Mutex,
NativeHandlerFactory, Scheduler, ServerError, SocketAddr, TcpStream,
};
impl ConnectionSupervisor {
/// Counts one refused admission, classified by reason.
///
/// Hung on the THREE public admission doors — TCP accept, the sibling
/// transport spawn (WebSocket), and the in-process loopback — because those
/// are the three places a connection can be turned away, and each of them
/// reaches the shared inner body exactly once. Recording deeper would
/// double-count the loopback (which calls the inner body directly);
/// recording shallower would miss the doors that only log.
fn record_admission_refusal(error: &ServerError) {
crate::metrics::admission_refused(AdmissionRefusal::classify(error));
}
/// Spawns one supervised beamr process that owns `stream`.
///
/// # Errors
/// Returns [`ServerError`] when stream configuration or beamr spawn fails.
pub fn spawn_connection(&self, stream: TcpStream) -> Result<ConnectionHandle, ServerError> {
self.inner
.spawn_connection(stream)
.inspect_err(Self::record_admission_refusal)
}
/// Returns the underlying beamr scheduler.
#[must_use]
pub fn scheduler(&self) -> Arc<Scheduler> {
Arc::clone(&self.inner.scheduler)
}
/// Reaps connection processes that have exited outside the normal handler path.
#[must_use]
pub fn reap_crashed_connections(&self) -> usize {
self.inner.runtime.reap_crashed(&self.inner.scheduler)
}
/// Returns true when `pid` is still tracked by the supervisor.
#[must_use]
pub fn is_tracked(&self, pid: u64) -> bool {
self.inner.runtime.contains(pid)
}
/// Returns the number of tracked live connections.
#[must_use]
pub fn active_connection_count(&self) -> usize {
self.inner.runtime.active_count()
}
/// Returns the beamr process ids of the currently tracked live connections.
///
/// Useful for addressing a specific connection — e.g. as the `pid` argument to
/// [`push_to_connection`](Self::push_to_connection) when the caller knows there
/// is a single connected client.
#[must_use]
pub fn active_connection_pids(&self) -> Vec<u64> {
self.inner
.runtime
.active_connections()
.into_iter()
.map(|connection| connection.pid)
.collect()
}
/// SRV-027 R4: how many keyed lookups the service directory has made,
/// read by the count pins.
#[must_use]
pub fn directory_lookups(&self) -> u64 {
self.inner.runtime.directory_lookups()
}
/// LP-WS-TRANSPORT R1.3 sibling-transport spawn seam (ADDITIVE ONLY).
///
/// Admits, allocates a durable connection incarnation for, spawns, and
/// registers a connection process whose handler is built by `build_factory`
/// over this supervisor's shared [`ConnectionRuntime`]. The WebSocket
/// sibling acceptor uses this so its connections share the ONE §5
/// `max_connections` admission bound, the one incarnation authority, the
/// one registry (controls, pushes, crash reap, drain, forced close), and the
/// one `apply_frame` seam with TCP connections. The TCP accept path above
/// (`spawn_connection`) is byte-for-byte untouched and never calls this.
///
/// `fd_guard` is a host-held duplicate of the connection's underlying
/// socket, exactly like the TCP path's: it keeps the fd alive until the
/// single record-removal path has synchronously deregistered readiness. It
/// is `None` for a transport that owns no descriptor.
///
/// `mount` is the admitting door's own name for itself (design §10), which
/// is why this seam takes it rather than deriving it: the caller IS the
/// door, and no other party — least of all the client — has any input.
///
/// This method exists because Rust module privacy makes the runtime,
/// admission counter, incarnation authority, and registry unreachable from
/// the sibling `websocket` module family; it is the narrow additive seam
/// that shares them without generalizing any TCP hot path.
///
/// # Errors
/// Returns [`ServerError`] when admission is refused
/// ([`ServerError::ConnectionLimitReached`]), incarnation allocation fails,
/// or beamr spawn/registration fails.
pub(in crate::server::connection) fn spawn_transport_connection(
&self,
peer_addr: Option<SocketAddr>,
fd_guard: Option<TcpStream>,
mount: MountKind,
build_factory: &dyn Fn(
Arc<ConnectionRuntime>,
Option<ConnectionIncarnation>,
) -> NativeHandlerFactory,
) -> Result<ConnectionHandle, ServerError> {
self.inner
.spawn_transport_connection(peer_addr, fd_guard, mount, build_factory)
.inspect_err(Self::record_admission_refusal)
}
/// Admits one in-process connection over `server_end` (design §8 step 3).
///
/// This replaces exactly the listener's `accept()` + `spawn_connection`
/// pair, and NOTHING else about admission. It runs the same
/// `try_reserve_admission` against the same §5 slot pool — an in-process
/// connect at capacity is refused with the same typed
/// [`ServerError::ConnectionLimitReached`] a socket connect is — allocates a
/// real durable [`ConnectionIncarnation`] from the same authority, so
/// participant binding, resume, and fate records work identically, and
/// registers the same record. Only two fields differ, and both are honest
/// descriptions rather than semantics: no fd guard (there is no descriptor
/// to keep alive) and `peer_addr: None` (there is no socket to have an
/// address). Nothing on this path reads a socket fact.
///
/// The connection's wake is installed by the process itself on its first
/// serviced slice, which is the earliest point at which its pid and host
/// record — the two things the wake names — both exist.
///
/// `pub(crate)` rather than `pub`: the embedding handle that grants
/// loopback connections lives in this crate, and widening the surface
/// further would hand an outside caller a way to reach the runtime that
/// module privacy currently denies it.
///
/// # Errors
/// Returns [`ServerError`] when admission is refused
/// ([`ServerError::ConnectionLimitReached`]), incarnation allocation fails,
/// or beamr spawn/registration fails.
pub(crate) fn spawn_loopback_connection(
&self,
server_end: LoopbackServerEnd,
) -> Result<ConnectionHandle, ServerError> {
// The same interior-mutability handoff the socket path uses: the native
// handler factory is `Fn + Send + Sync`, so the duplex end cannot be
// moved into it and the FIRST handler build takes it out exactly once.
let holder = Arc::new(Mutex::new(Some(server_end)));
let build = move |runtime: Arc<ConnectionRuntime>,
incarnation: Option<ConnectionIncarnation>|
-> NativeHandlerFactory {
let holder = Arc::clone(&holder);
Box::new(move || {
Box::new(LoopbackConnectionProcess::from_loopback_holder(
Arc::clone(&runtime),
&holder,
incarnation,
))
})
};
self.inner
.spawn_transport_connection(None, None, MountKind::Loopback, &build)
.inspect_err(Self::record_admission_refusal)
}
}