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
use super::{
Atom, ConnectionControl, ConnectionRuntime, QueuedConnectionControl, Scheduler, ServerError,
lock,
};
impl ConnectionRuntime {
/// Queues `control` for connection `pid` and wakes it on `scheduler`;
/// `true` means the control was published (queued and the wake landed,
/// or a drain consumed it first). The one publication path for a control,
/// shared by the supervisor's pushes and a routed request's (SRV-027 R2).
pub(in crate::server::connection) fn publish_control(
&self,
scheduler: &Scheduler,
pid: u64,
control: ConnectionControl,
) -> bool {
// Keep a key for the failure-path removal before the control is moved into
// the queue, so a non-`Copy` (push) control can still be located and pulled
// back out if the scheduler wakeup fails.
let removal_key = control.clone();
if let Err(error) = self.push_control(pid, control) {
// The cause is only knowable HERE. Callers see a bare `false` and
// used to render every one of them as "process is not live", which
// is a lie about a live process whose control queue refused the
// entry — and it discarded the one error that said what happened.
tracing::warn!(
connection_pid = pid,
%error,
"connection control queue refused the entry; the control was never queued"
);
return false;
}
// Deterministic test seam in the insert->wake window (S8 staging).
#[cfg(test)]
self.run_pre_wake_barrier();
if scheduler.enqueue_atom_message(pid, self.control_atom()) {
true
} else {
// Failed wake: the entry's fate is the publication verdict. Removed
// here => nobody consumed it => unpublished. Already gone => a
// drain consumed it before the wake check => published. (A poisoned
// queue lock reads as not-removed => published — the safe
// direction: the slot lifecycle then reports the truthful outcome,
// whereas claiming "unpublished" could be a lie.)
!self.remove_control(pid, &removal_key)
}
}
pub(in crate::server::connection) const fn control_atom(&self) -> Atom {
self.control_atom
}
fn push_control(&self, pid: u64, control: ConnectionControl) -> Result<(), ServerError> {
lock(&self.controls, "connection control queue")?
.push(QueuedConnectionControl { pid, control });
Ok(())
}
pub(in crate::server::connection) fn pop_control(&self, pid: u64) -> Option<ConnectionControl> {
let mut controls = self.controls.lock().ok()?;
let index = controls
.iter()
.position(|queued| queued.pid == pid && queued.control == ConnectionControl::ForceClose)
.or_else(|| controls.iter().position(|queued| queued.pid == pid))?;
Some(controls.remove(index).control)
}
/// Non-consuming final-probe query for controls enqueued after mailbox drain.
pub(in crate::server::connection) fn has_control(&self, pid: u64) -> bool {
self.controls
.lock()
.is_ok_and(|controls| controls.iter().any(|queued| queued.pid == pid))
}
/// Pulls a queued-but-unconsumed control back out of the queue. Returns
/// whether THIS call removed it — `false` means the entry already left the
/// queue, and since only the consumer removes a published control, a
/// consumer drain took it (S8's publication disambiguator). Matching is
/// `pid` + full control equality; a `Push` control embeds its
/// runtime-unique correlation id, so this can never remove a different
/// push's entry and misreport.
fn remove_control(&self, pid: u64, control: &ConnectionControl) -> bool {
let Ok(mut controls) = self.controls.lock() else {
return false;
};
let Some(index) = controls
.iter()
.position(|queued| queued.pid == pid && &queued.control == control)
else {
return false;
};
controls.remove(index);
true
}
}