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
#[cfg(test)]
use super::WorkerRegistration;
use super::{
ActiveConnection, Arc, AtomicBool, ConnectionIncarnation, ConnectionRecord, ConnectionRuntime,
Entry, ExitReason, MountKind, Scheduler, ServerError, SocketAddr, TcpStream, Weak, lock,
};
impl ConnectionRuntime {
/// SRV-027 R1: the connection serving `subject`, by one keyed lookup in
/// the directory under the registry lock; `None` when no worker serves it.
pub(in crate::server::connection) fn serving(&self, subject: &str) -> Option<u64> {
self.records
.lock()
.ok()
.and_then(|records| records.directory.serving(subject))
}
/// How many directory lookups `serving` has made (R4 count pins).
pub(in crate::server::connection) fn directory_lookups(&self) -> u64 {
self.records
.lock()
.map_or(0, |records| records.directory.lookups())
}
/// Sole registration path for a connection: the spawn thread inserts the
/// record synchronously, before `spawn_connection` returns the handle, so
/// `is_tracked`/`active_connection_count` reflect the connection
/// immediately. The connection handler never writes the registry (it only
/// reads via `mark_crashed`/`finish`), so there is a single writer here and
/// no register/ensure-register race.
///
/// Ordering note: `spawn_native` only enqueues the process, so its first
/// slice may run on another worker thread before this insert lands. If that
/// first slice exits immediately (e.g. a missing-stream crash) its
/// `mark_crashed`/`finish` removes nothing and this insert then leaves a
/// record for an already-dead pid. W4 leg 1 retires the per-accept
/// `reap_crashed` scan that used to self-heal that orphan continuously;
/// instead [`Self::reconcile_register_orphan`] closes the race with a SINGLE
/// point check on the registration event itself — never a loop.
/// `fd_guard` is `None` for a transport that owns no descriptor. The guard
/// exists to keep an fd alive until readiness deregistration has been
/// acknowledged; a loopback connection registers no readiness and holds no
/// descriptor, so there is nothing to guard and the slot is honestly empty
/// rather than filled with a placeholder.
pub(super) fn register_connection(
&self,
pid: u64,
peer_addr: Option<SocketAddr>,
connection_incarnation: Option<ConnectionIncarnation>,
mount: MountKind,
fd_guard: Option<TcpStream>,
) -> Result<(), ServerError> {
self.register_record(pid, peer_addr, connection_incarnation, mount, fd_guard)?;
self.reconcile_register_orphan(pid);
Ok(())
}
/// Closes the register-orphan race (see [`Self::register_with_fd`]) with one
/// point check driven by the registration event — not a periodic scan. If
/// the just-registered pid is already absent from the scheduler process
/// table, its first slice has run and exited, so the record this
/// registration inserted is an orphan the retiring reap scan used to sweep;
/// reclaim it immediately through the ordinary `remove()` funnel. A pid still
/// present is live and needs nothing here: a later external termination rides
/// the exit-event reactor and an ordinary exit its own final slice.
pub(super) fn reconcile_register_orphan(&self, pid: u64) {
let Some(scheduler) = self.scheduler.upgrade() else {
return;
};
// The process-table lookup returns a sharded guard; bind only the
// presence bool so the guard is released before `reclaim_terminated`
// takes the connection registry lock (no cross-lock hold).
let already_exited = scheduler.process_table().get(pid).is_none();
if !already_exited {
return;
}
let reason = scheduler
.peek_exit_reason(pid)
.unwrap_or(ExitReason::Normal);
self.reclaim_terminated(pid, reason);
}
/// TOLD reclamation of a connection whose process exited WITHOUT running a
/// final handler slice — external/panic termination, where
/// [`ConnectionProcess::Drop`] runs but no `mark_crashed`/`finish` does, and
/// the register-orphan race above. Delivered the instant beamr publishes the
/// process's exit event (via the reclamation reactor) or at the
/// registration point check, and routed through the SAME [`Self::remove`]
/// funnel as every other teardown — no third funnel, no periodic scan.
/// Idempotent: a record already removed in-slice, by the orphan reconcile, or
/// by a duplicate delivery is a no-op here (remove returns `None`), so the §5
/// admission gauge is released exactly once.
fn reclaim_terminated(&self, pid: u64, reason: ExitReason) {
let Some(record) = self.remove(pid) else {
return;
};
self.fire_close_hooks(pid, &record);
tracing::warn!(
connection_pid = pid,
peer_addr = ?record.peer_addr,
mount = ?record.mount,
reason = ?reason,
"connection process exited without a final slice; host record reclaimed by delivery"
);
}
/// One exit-event delivery: drain beamr's retained outcome (sole drainer,
/// bounding its store) then reclaim through [`Self::reclaim_terminated`]. The
/// only non-production element is the `#[cfg(test)]` reclamation gate, which
/// is staged for at most one targeted pid and compiled out entirely in
/// production — the delivery semantics are identical with or without it.
pub(super) fn deliver_reclamation(
&self,
scheduler: &Weak<Scheduler>,
pid: u64,
reason: ExitReason,
) {
if let Some(scheduler) = scheduler.upgrade() {
// The reason is already in-hand from the event; the drained outcome
// is discarded, its purpose being only to bound beamr's store.
drop(scheduler.take_exit_outcome(pid));
}
#[cfg(test)]
let staged = self.stage_reclaim_barrier(pid);
#[cfg(test)]
if let Some(barrier) = staged.as_ref() {
// Rendezvous: the pid is now dead but its record is still tracked —
// the S8 reclamation window (oracle 26). Then wait for the harness to
// release the reclaim.
barrier.reached.wait();
barrier.release.wait();
}
self.reclaim_terminated(pid, reason);
#[cfg(test)]
if let Some(barrier) = staged.as_ref() {
// Signal the funnel completed so the harness can observe the record
// gone without sampling.
barrier.done.wait();
}
}
#[cfg(test)]
pub(super) fn register(
&self,
pid: u64,
peer_addr: Option<SocketAddr>,
) -> Result<(), ServerError> {
self.register_record(pid, peer_addr, None, MountKind::Tcp, None)
}
/// Gives connection `pid` a record, as a spawned connection has, so a
/// unit test through `apply_frame` can see what is stored on it.
#[cfg(test)]
pub(in crate::server::connection) fn register_for_tests(
&self,
pid: u64,
) -> Result<(), ServerError> {
self.register(pid, None)
}
/// The worker registration stored on connection `pid`'s record. A
/// registration is stored before its topology change is emitted, and
/// only then (`set_registration`), so `None` also means no
/// `TopologyChange::Worker` was emitted for it.
#[cfg(test)]
pub(in crate::server::connection) fn stored_registration_for_tests(
&self,
pid: u64,
) -> Result<Option<WorkerRegistration>, ServerError> {
Ok(lock(&self.records, "connection registry")?
.connections
.get(&pid)
.and_then(|record| record.registration.clone()))
}
pub(super) fn register_record(
&self,
pid: u64,
peer_addr: Option<SocketAddr>,
connection_incarnation: Option<ConnectionIncarnation>,
mount: MountKind,
fd_guard: Option<TcpStream>,
) -> Result<(), ServerError> {
match lock(&self.records, "connection registry")?
.connections
.entry(pid)
{
// ENFORCED (not comment-only): pids are fresh per spawn and a record
// is reclaimed through the single `remove` funnel before its pid can
// recycle, so an occupied slot here is a supervision defect. Refuse
// the whole registration rather than silently replacing — a replace
// would drop the displaced record's `fd_guard` outside the teardown
// funnel (orphaning a live connection's stream) and increment the
// `liminal_connections_active` gauge a second time with no paired
// decrement. The prior record stays intact; both spawn paths roll the
// fresh process back on this error. Same idiom as
// `StreamTable::insert` (entry-vacant → typed error) and the
// fail-closed duplicate-conversation refusal in `apply.rs`.
Entry::Occupied(_) => return Err(ServerError::ConnectionPidCollision { pid }),
Entry::Vacant(entry) => {
entry.insert(ConnectionRecord {
peer_addr,
mount,
connection_incarnation,
registration: None,
pass_principal: None,
readiness: None,
ready_pending: Arc::new(AtomicBool::new(false)),
parked: AtomicBool::new(false),
fd_guard,
});
}
}
// Single-writer vacant-only insert (see doc above and the refusal arm)
// pairs EXACTLY one gauge increment with the decrement in `remove`,
// keeping `liminal_connections_active` equal to the live record count on
// every teardown route; a refused duplicate increments nothing.
crate::metrics::connection_spawned();
if let Some(signals) = self.signals.get() {
signals.connection(pid, true);
signals.topology(crate::console::TopologyChange::ConnectionOpened {
id: pid,
mount: mount.into(),
participant: None,
});
}
Ok(())
}
/// Whether the spawn thread has installed the host record. A first native
/// slice can win the enqueue-vs-record race and must remain runnable until it
/// has somewhere host-reachable to publish its readiness token.
pub(in crate::server::connection) fn is_registered(&self, pid: u64) -> bool {
self.contains(pid)
}
pub(super) fn contains(&self, pid: u64) -> bool {
self.records
.lock()
.is_ok_and(|records| records.connections.contains_key(&pid))
}
pub(super) fn active_connections(&self) -> Vec<ActiveConnection> {
self.records.lock().map_or_else(
|_| Vec::new(),
|records| {
records
.connections
.iter()
.map(|(&pid, record)| ActiveConnection {
pid,
peer_addr: record.peer_addr,
})
.collect()
},
)
}
/// Reads the complete active-connection incarnation set under the
/// registry lock — the bounded defense-in-depth collision-skip input to
/// incarnation allocation (uniqueness itself comes from allocator-log
/// monotonicity; see `allocate_connection_incarnation`). Poisoning fails
/// admission closed: treating an unreadable registry as empty would
/// silently drop the defense layer.
pub(super) fn complete_active_incarnation_references(
&self,
) -> Result<Vec<ConnectionIncarnation>, ServerError> {
Ok(
lock(&self.records, "connection incarnation reference registry")?
.connections
.values()
.filter_map(|record| record.connection_incarnation)
.collect(),
)
}
pub(super) fn active_count(&self) -> usize {
self.records
.lock()
.map_or(0, |records| records.connections.len())
}
/// Test read of the record's stamped mount (design §10).
#[cfg(test)]
pub(super) fn connection_mount(&self, pid: u64) -> Option<MountKind> {
self.records
.lock()
.ok()?
.connections
.get(&pid)
.map(|record| record.mount)
}
/// Test read of whether the record holds an fd guard.
#[cfg(test)]
pub(super) fn connection_has_fd_guard(&self, pid: u64) -> Option<bool> {
self.records
.lock()
.ok()?
.connections
.get(&pid)
.map(|record| record.fd_guard.is_some())
}
}