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
#[cfg(test)]
use super::CloseEventFdProbe;
#[cfg(test)]
use super::RawFd;
use super::{
ConnectionRecord, ConnectionRuntime, ExitReason, Ordering, Scheduler, SocketAddr, TimerRef,
};
impl ConnectionRuntime {
pub(in crate::server::connection) fn mark_crashed(
&self,
pid: u64,
reason: ExitReason,
peer_addr: Option<SocketAddr>,
) {
let removed = self.remove(pid);
if let Some(record) = removed.as_ref() {
self.fire_close_hooks(pid, record);
}
let removed_peer_addr = removed
.as_ref()
.and_then(|record| record.peer_addr)
.or(peer_addr);
// The mount names the door this connection came through — the one fact
// that tells a reader whether a crashed connection was a socket peer or
// a co-resident caller, which `peer_addr: None` alone cannot.
let mount = removed.as_ref().map(|record| record.mount);
tracing::warn!(
connection_pid = pid,
peer_addr = ?removed_peer_addr,
mount = ?mount,
reason = ?reason,
"connection process crashed"
);
}
/// Cancels deadline timers detached by reply completion or connection close.
pub(in crate::server::connection) fn cancel_deadline_timers(&self, timers: Vec<TimerRef>) {
let Some(scheduler) = self.scheduler.upgrade() else {
return;
};
if let Ok(mut wheel) = scheduler.timers().lock() {
for timer in timers {
wheel.cancel(timer);
}
}
}
pub(in crate::server::connection) fn finish(&self, pid: u64) {
if let Some(removed) = self.remove(pid) {
self.fire_close_hooks(pid, &removed);
}
}
/// Fires the close-side notifier hooks for a removed connection record:
/// `on_worker_unregistered` when it carried a worker registration, and
/// `on_pass_detached` (with the principal) when it carried a pass
/// principal. A record with neither — a plain bearer or open-access
/// connection, or a worker connection that never registered — is a no-op,
/// so each close hook fires for exactly the connections whose open-side
/// counterpart fired. Every teardown route (`finish`, `mark_crashed`,
/// `reclaim_terminated`, `reap_crashed`) reaches this through the
/// idempotent [`Self::remove`], so each fires at most once per connection.
pub(super) fn fire_close_hooks(&self, pid: u64, record: &ConnectionRecord) {
let Some(notifier) = self.notifier.as_ref() else {
return;
};
if record.registration.is_some() {
notifier.on_worker_unregistered(pid);
}
if let Some(principal) = record.pass_principal.as_ref() {
notifier.on_pass_detached(pid, principal);
}
}
pub(super) fn reap_crashed(&self, scheduler: &Scheduler) -> usize {
let pids = match self.records.lock() {
Ok(records) => records.connections.keys().copied().collect::<Vec<_>>(),
Err(error) => {
tracing::warn!(%error, "connection registry unavailable during crash reap");
return 0;
}
};
let mut reaped = 0;
for pid in pids {
if scheduler.process_table().get(pid).is_none() {
let removed = self.remove(pid);
if let Some(record) = removed.as_ref() {
self.fire_close_hooks(pid, record);
}
let peer_addr = removed.and_then(|record| record.peer_addr);
// This process exited without ever reaching `mark_crashed`/`finish`
// (e.g. the beamr scheduler terminated it externally). W4 leg 1
// retired this scan from the per-accept listener loop, and W4 leg 3
// retired the shutdown-drain reconciliation that also drove it: the
// reclaimer of these exits is now the TOLD exit-event reactor
// (the reclamation reactor), which also composes drain completion
// through the one `remove()` funnel. This scan now survives only as
// that reactor's exit-event overflow (`Lagged`) recovery, driven a
// bounded number of times, never periodically. beamr 0.15.4 exposes
// a public, non-blocking
// `peek_exit_reason` (and `take_exit_outcome`), so the real reason
// IS recoverable here rather than logged as an opaque literal.
let reason = scheduler.peek_exit_reason(pid);
tracing::warn!(
connection_pid = pid,
?peer_addr,
?reason,
"connection process exited without a final slice; reclaimed by reconciliation"
);
reaped += 1;
}
}
reaped
}
/// Removes the connection record for `pid` and, in the same close step, drops
/// every push reply slot that connection still owns so each waiting
/// [`PushReplyAwaiter`] wakes immediately with a disconnected error. This runs
/// on every close route — `finish`, `mark_crashed`, and `reap_crashed` all
/// remove through here — and fires regardless of whether the connection ever
/// registered a worker, so a plain push target is covered too.
///
/// ORDER MATTERS (S3/S7): the record is removed BEFORE the push sweep. A
/// push registering concurrently runs INSERT -> CONFIRM -> PUBLISH
/// (`confirm_push_registration` reads the record after inserting its slot
/// and before publishing its control), so with this ordering exactly one
/// side always observes a racing slot: a confirm that ran before this
/// removal implies the slot was inserted before the sweep below (which then
/// reaps it — a control published after that confirm is answered into a
/// swept slot, read as the truthful disconnected outcome); a confirm after
/// this removal sees the absence, rolls the slot back itself, and never
/// publishes. Sweeping first (the original order) left a window — sweep,
/// then insert+confirm, then record removal — where NEITHER side observed
/// the slot and it leaked past connection close. The two locks are taken
/// strictly sequentially (never nested), so no lock-order inversion.
pub(super) fn remove(&self, pid: u64) -> Option<ConnectionRecord> {
let mut removed = self.records.lock().ok().and_then(|mut records| {
let removed = records.connections.remove(&pid);
// SRV-027 R1: the directory entry leaves with its record, in
// this same lock take.
if let Some(registration) = removed
.as_ref()
.and_then(|record| record.registration.as_ref())
{
records.directory.forget(pid, ®istration.task_queue);
if let Err(error) = self.services.site_registration_closed(pid) {
tracing::error!(connection = pid, %error, "trial_registration_close_failed");
}
}
removed
});
#[cfg(test)]
let host_fd = removed
.as_ref()
.and_then(|record| record.fd_guard.as_ref())
.map(std::os::fd::AsRawFd::as_raw_fd);
self.cancel_pushes_for_connection(pid);
if let Some(registration) = removed.as_mut().and_then(|record| record.readiness.take()) {
if let Some(scheduler) = self.scheduler.upgrade() {
// This call is ACK'd: it returns only after the poll owner has
// removed the registration. `fd_guard` is still live here.
scheduler.readiness_deregister(registration.token);
tracing::debug!(
registered_fd = registration.fd,
"connection readiness deregistration acknowledged"
);
}
}
if let Some(record) = removed.as_mut() {
// Explicit after-deregister drop documents and enforces the fd wall.
drop(record.fd_guard.take());
}
// Decrement only when a record was actually present so a double-remove
// (e.g. `finish` after `reap_crashed`) cannot drive the gauge negative.
// The §5 admission slot is released on the same guard: the reservation
// acquired in `spawn_connection` converted into this record at
// `register`, so its removal is exactly one release per reservation.
if removed.is_some() {
self.link_life.remove(pid);
crate::metrics::connection_closed();
self.release_admission();
// ACCESS-005 R6: the connection's grant-index membership is
// retired on this same once-per-record guard, so revoke, EOF,
// crash and force-close all release it exactly once.
if let Some(revocation) = self.revocation.get()
&& revocation.release(pid)
{
tracing::debug!(connection_pid = pid, "grant-decision binding released");
}
if let Some(signals) = self.signals.get() {
#[cfg(test)]
self.probe_close_event_fd(host_fd);
signals.connection(pid, false);
signals.topology(crate::console::TopologyChange::ConnectionClosed { id: pid });
}
}
// TOLD drain-completion tell (W4 leg 3, §4.3): the record map above already
// reflects this removal, so bump the removal generation and wake the
// drain/settle waiter AFTER the observed state is updated. Ordering the
// state update before the generation bump — paired with the waiter arming
// its snapshot before observing `active_count` — is the arm-before-observe
// barrier that makes a concurrently delivered exit un-losable (oracle 18).
// Only a real removal tells, so a double-remove drives no spurious wake.
if removed.is_some() {
self.signal_connection_removed();
// Removal can complete delivery quiescence without another park.
// Tell the earlier flush waiter as well as the later drain waiter.
if self.settle_armed.load(Ordering::Acquire) {
self.signal_settle_changed();
}
}
removed
}
#[cfg(test)]
fn probe_close_event_fd(&self, host_fd: Option<RawFd>) {
let mut probe = match self.close_event_fd_probe.lock() {
Ok(probe) => probe,
Err(error) => {
tracing::error!(%error, "close-event fd probe lock poisoned");
return;
}
};
if !matches!(*probe, CloseEventFdProbe::Armed) {
return;
}
let observed = host_fd.map_or_else(
|| Err(std::io::Error::other("close-event probe has no owned fd")),
// Asked by lstat, never by opening: a socket fd's link refuses to open on
// Linux (ENXIO), and opening it on macOS duplicates the fd in question.
|fd| match std::fs::symlink_metadata(format!("/dev/fd/{fd}")) {
Ok(_) => Ok(false),
Err(error) if matches!(error.raw_os_error(), Some(2 | 9)) => Ok(true),
Err(error) => Err(error),
},
);
*probe = CloseEventFdProbe::Observed(observed);
}
}