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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
#[cfg(test)]
use super::Sender;
use super::{
AdmissionReadiness, AdmissionReservation, Arc, AtomTable, CONNECTION_READY_ATOM,
CONNECTION_SCHEDULER_THREADS, CONNECTION_SHUTDOWN_CONTROL_ATOM, CandidateLaneListener,
ConnectionControl, ConnectionHandle, ConnectionIncarnation, ConnectionIncarnationAuthority,
ConnectionNotifier, ConnectionProcess, ConnectionRuntime, ConnectionRuntimeInstallation,
ConnectionServiceInstallation, ConnectionServices, ExitReason, IncarnationStartInputs,
LimitsConfig, LysInstall, ModuleRegistry, MountKind, Mutex, NativeBifs, NativeHandlerFactory,
NativeRetirements, PassVerifier, Scheduler, SchedulerConfig, SchedulerServices, ServerError,
ShutdownHandle, SocketAddr, TcpStream, Weak, install_lys, lifecycle, reclaim,
};
pub(super) struct SupervisorInner {
pub(super) scheduler: Arc<Scheduler>,
pub(super) runtime: Arc<ConnectionRuntime>,
pub(super) reclaim: reclaim::ReclaimReactor,
pub(super) retirements: Arc<NativeRetirements>,
pub(super) background: Mutex<lifecycle::BackgroundStart>,
pub(super) incarnations: Option<Arc<ConnectionIncarnationAuthority>>,
/// P0 #56 R4: the readiness probe's view of whether this server can admit.
/// Shared with the incarnation authority when one is installed.
pub(super) admission_readiness: AdmissionReadiness,
#[cfg(test)]
pub(super) reclaim_stop_observer: Mutex<Option<Sender<bool>>>,
}
impl std::fmt::Debug for SupervisorInner {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("SupervisorInner")
.field("runtime", &self.runtime)
.finish_non_exhaustive()
}
}
impl SupervisorInner {
pub(super) fn new(
services: Arc<dyn ConnectionServices>,
notifier: Option<Arc<dyn ConnectionNotifier>>,
auth_token: Option<Vec<u8>>,
pass_verifier: Option<PassVerifier>,
lys: Option<LysInstall>,
limits: LimitsConfig,
fatal_shutdown: Option<ShutdownHandle>,
) -> Result<Self, ServerError> {
let installed_services = ConnectionServiceInstallation::capture(services);
// P0 #56 R4. Owned by the supervisor so the readiness probe has a stable
// handle whether or not a participant service (and therefore an
// incarnation authority) is installed: a server with no authority has
// nothing that can hold admission, and reports available.
let admission_readiness = AdmissionReadiness::available();
let incarnations = installed_services
.participant_service
.as_ref()
.map(
|service| -> Result<Arc<ConnectionIncarnationAuthority>, ServerError> {
// SRV-041 R2: nothing is read here; the start runs behind
// the bind, begun by `listening()`.
ConnectionIncarnationAuthority::starting(
IncarnationStartInputs {
store: service.durable_store(),
maximum_references: limits.max_connections,
maximum_conversations: service.publication_conversation_limit(),
handler: Arc::new(service.clone()),
},
admission_readiness.clone(),
)
.map(Arc::new)
},
)
.transpose()?;
let atoms = AtomTable::with_common_atoms();
let control_atom = atoms.intern(CONNECTION_SHUTDOWN_CONTROL_ATOM);
let registry = Arc::new(ModuleRegistry::new());
let scheduler_starting = std::time::Instant::now();
let scheduler = Scheduler::with_services(
SchedulerConfig {
thread_count: Some(CONNECTION_SCHEDULER_THREADS),
..SchedulerConfig::default()
},
SchedulerServices::from_config().owned_readiness(),
registry,
// The strongest of the three cases: `registry` here is a bare
// `ModuleRegistry::new()` with NO module ever inserted, so this
// scheduler cannot execute a single bytecode instruction, let alone
// a guard BIF. Connections run as native processes via
// `NativeHandlerFactory`. beamr 0.19 requires the answer be written
// down rather than inherited; the answer is none.
NativeBifs::none(),
)
.map_err(|message| ServerError::ListenerAccept {
message: format!("failed to start connection scheduler: {message}"),
})?;
crate::server::boot_steps::finished("connection_scheduler_start", scheduler_starting);
let ready_atom = atoms.intern(CONNECTION_READY_ATOM);
let scheduler = Arc::new(scheduler);
// The runtime captures a WEAK handle to the connection scheduler so
// notifier wakes (R3/R1(vi)) can be fired from another actor's slice
// without a strong scheduler↔process↔runtime cycle that would leak the
// whole connection scheduler.
let runtime = Arc::new(ConnectionRuntime::new(
ConnectionRuntimeInstallation {
services: installed_services,
incarnations: incarnations.clone(),
fatal_shutdown,
},
control_atom,
ready_atom,
Arc::downgrade(&scheduler),
notifier,
(auth_token, pass_verifier),
limits,
));
if let Some(lys) = lys {
install_lys(&runtime, lys)?;
}
// A connection fate parked on candidate-lane `Precedence` is retried by
// the drain that clears it, so the runtime is told about every drain.
if let (Some(service), Some(notifier)) = (runtime.participant_service(), runtime.notifier())
{
service.install_checkpoint_crossing_observer(Arc::clone(notifier));
}
if let Some(service) = runtime.participant_service() {
let runtime_weak: Weak<ConnectionRuntime> = Arc::downgrade(&runtime);
let listener: Weak<dyn CandidateLaneListener> = runtime_weak;
service
.install_candidate_lane_listener(listener)
.map_err(|error| ServerError::ParticipantIncarnation {
phase: "candidate-lane listener installation",
message: error.to_string(),
})?;
}
let reclaim = crate::server::boot_steps::timed("reclaim_start", || {
reclaim::ReclaimReactor::start(&scheduler, &runtime)
})?;
Ok(Self {
scheduler,
runtime,
reclaim,
retirements: Arc::new(NativeRetirements::default()),
background: Mutex::new(lifecycle::BackgroundStart::default()),
incarnations,
admission_readiness,
#[cfg(test)]
reclaim_stop_observer: Mutex::new(None),
})
}
pub(super) fn spawn_connection(
self: &Arc<Self>,
stream: TcpStream,
) -> Result<ConnectionHandle, ServerError> {
// §5 `max_connections`: ATOMIC admission reservation acquired BEFORE any
// process construction (review round 1 item 7 — a signed bound must not
// be exceedable by concurrent callers; check-then-spawn across an
// unlocked window was). The CAS reservation is released on every failure
// path below and converts into the connection record at `register`;
// thereafter the single record-removal path (`remove`) releases it. An
// over-cap accept therefore costs nothing and the bound holds under any
// concurrency.
self.runtime.try_reserve_admission()?;
let reservation = AdmissionReservation {
runtime: &self.runtime,
armed: true,
};
stream
.set_nonblocking(true)
.map_err(|error| ServerError::ListenerAccept {
message: format!("failed to configure connection stream: {error}"),
})?;
// A failed `peer_addr` is folded to `None` rather than refusing the
// accept: an unidentifiable peer is still a serviceable connection. The
// cost is paid in triage, not in service — every later log line for this
// connection reports `peer_addr=None`, so nothing that happens on it can
// be tied back to a remote endpoint.
let peer_addr = stream.peer_addr().ok();
// The host-held duplicate keeps the fd alive until the single record-removal
// path has synchronously deregistered readiness. External process death can
// therefore never let fd reuse overtake host-side deregistration.
let fd_guard = stream
.try_clone()
.map_err(|error| ServerError::ListenerAccept {
message: format!("failed to retain connection fd for teardown: {error}"),
})?;
let connection_incarnation = self.allocate_connection_incarnation()?;
let holder = Arc::new(Mutex::new(Some(stream)));
let runtime = Arc::clone(&self.runtime);
let process_holder = Arc::clone(&holder);
let factory: NativeHandlerFactory = Box::new(move || {
Box::new(ConnectionProcess::from_holder(
Arc::clone(&runtime),
peer_addr,
&process_holder,
connection_incarnation,
))
});
let pid = self.spawn_native_connection(factory, "failed to spawn connection process")?;
if let Err(error) = self.runtime.register_connection(
pid,
peer_addr,
connection_incarnation,
MountKind::Tcp,
Some(fd_guard),
) {
// Registration failure leaves no host record to reap. Terminate the
// just-spawned process explicitly so neither its stream nor admission
// reservation can escape this failed spawn.
self.scheduler.terminate_process(pid, ExitReason::Error);
return Err(error);
}
// The reservation is now owned by the registered record: `remove` (the
// single record-removal path — finish/mark_crashed/reap all funnel
// through it) releases the admission when the record goes away.
reservation.convert();
Ok(ConnectionHandle {
pid,
peer_addr,
connection_incarnation,
supervisor: Arc::clone(self),
})
}
/// LP-WS-TRANSPORT R1.3: the sibling-transport spawn body. Mirrors
/// [`Self::spawn_connection`]'s admission → incarnation → spawn → register →
/// convert sequence exactly (same reservation guard, same failure rollback,
/// same single record-removal ownership), differing only in that the caller
/// supplies the native handler factory and the host-held fd guard instead of
/// a raw `TcpStream`. Purely additive; the TCP path never calls this.
pub(super) fn spawn_transport_connection(
self: &Arc<Self>,
peer_addr: Option<SocketAddr>,
fd_guard: Option<TcpStream>,
mount: MountKind,
build_factory: &dyn Fn(
Arc<ConnectionRuntime>,
Option<ConnectionIncarnation>,
) -> NativeHandlerFactory,
) -> Result<ConnectionHandle, ServerError> {
self.runtime.try_reserve_admission()?;
let reservation = AdmissionReservation {
runtime: &self.runtime,
armed: true,
};
let connection_incarnation = self.allocate_connection_incarnation()?;
let factory = build_factory(Arc::clone(&self.runtime), connection_incarnation);
let pid = self.spawn_native_connection(factory, "failed to spawn connection process")?;
if let Err(error) = self.runtime.register_connection(
pid,
peer_addr,
connection_incarnation,
mount,
fd_guard,
) {
self.scheduler.terminate_process(pid, ExitReason::Error);
return Err(error);
}
reservation.convert();
Ok(ConnectionHandle {
pid,
peer_addr,
connection_incarnation,
supervisor: Arc::clone(self),
})
}
pub(super) fn allocate_connection_incarnation(
&self,
) -> Result<Option<ConnectionIncarnation>, ServerError> {
let Some(authority) = self.incarnations.as_ref() else {
return Ok(None);
};
// Production-era uniqueness invariant: every published incarnation is
// unique against ALL durable references — binding epochs committed
// into conversation logs included — by allocator-log monotonicity
// alone, not by the completeness of the reference set below.
//
// 1. Startup replays the durable allocator stream and STRICTLY
// increments the server incarnation, fsyncing the Startup event
// before any listener becomes ready
// (`IncarnationStream::startup`); a server value is never wrapped
// or reused, so no two process lifetimes share one.
// 2. Within a lifetime, allocations are serialized under this
// authority's mutex, candidates start strictly above the durable
// `last_examined_connection_ordinal`
// (`allocate_connection_incarnation`), and every allocation's
// event is appended and flushed BEFORE its pair is published
// (`StartedIncarnationStream::allocate`), so ordinals never
// repeat within a lifetime and replay restores a head at or
// above every published ordinal.
// 3. A durable reference can only name a pair this allocator
// previously PUBLISHED (binding epochs are committed only after
// their connection was admitted), and the same store's flush
// barrier orders the allocator event before any conversation-log
// entry that references it.
//
// The live-connection reference set below is therefore defense in
// depth — a bounded collision skip against a rolled-back or divergent
// allocator stream — never the uniqueness foundation, and never a raw
// caller-supplied matrix.
let references = self.runtime.complete_active_incarnation_references()?;
authority.allocate(&references).map(Some)
}
pub(super) fn broadcast_control(&self, control: &ConnectionControl) {
for connection in self.runtime.active_connections() {
if !self.enqueue_control(connection.pid, control.clone()) {
// Observation, not diagnosis: the routine cause is a process
// that is no longer live, but a refused control queue reaches
// here too and has already warned with its own error.
tracing::debug!(
connection_pid = connection.pid,
peer_addr = ?connection.peer_addr,
?control,
"connection control message was not published to the process"
);
}
}
}
/// Queues `control` for `pid` and wakes the process. Returns whether the
/// control was PUBLISHED (left in the queue with a successful wake, or
/// already consumed by a drain) — `false` guarantees no consumer ever saw
/// it.
///
/// S8: a failed wake does NOT prove the queued control was never consumed.
/// The insert releases the queue lock before the wake attempt, and a
/// process already executing a control drain (each control atom drains ALL
/// queued controls for the pid) can pop the just-inserted entry in that
/// window, then exit before the wake check. Publication is therefore
/// disambiguated BY OBSERVATION on the failed-wake path: `remove_control`
/// finding and removing the entry proves no consumer saw it (truly
/// unpublished — `false`); finding nothing proves a drain consumed it
/// (`pop_control` is the only other remover of queue entries, and the
/// removal key embeds the push's runtime-unique correlation id, so it can
/// never match a different entry) — the control was published and the
/// caller's slot lifecycle carries the delivery truth (`true`).
pub(super) fn enqueue_control(&self, pid: u64, control: ConnectionControl) -> bool {
self.runtime.publish_control(&self.scheduler, pid, control)
}
}