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
#[cfg(test)]
use super::Barrier;
use super::{
Arc, ConnectionRuntime, ConnectionSupervisor, MountKind, RawFd, ReadinessToken, Receiver, Weak,
};
impl ConnectionSupervisor {
/// The runtime every connection process shares. Test-only.
#[cfg(test)]
pub(in crate::server::connection) fn runtime_for_test(&self) -> Arc<ConnectionRuntime> {
Arc::clone(&self.inner.runtime)
}
/// R7 test instrument: slices serviced by connection `pid` since spawn.
#[cfg(test)]
pub(crate) fn slice_count(&self, pid: u64) -> u64 {
self.inner.runtime.slice_count(pid)
}
/// The mount the admitting door stamped on `pid`'s registry record, or
/// `None` when no record is tracked (design §10).
#[cfg(test)]
pub(crate) fn connection_mount(&self, pid: u64) -> Option<MountKind> {
self.inner.runtime.connection_mount(pid)
}
/// Whether `pid`'s registry record holds an fd guard, or `None` when no
/// record is tracked. A loopback record must answer `Some(false)`: there is
/// no descriptor for the guard to keep alive.
#[cfg(test)]
pub(crate) fn connection_has_fd_guard(&self, pid: u64) -> Option<bool> {
self.inner.runtime.connection_has_fd_guard(pid)
}
/// Installs a one-use readiness marker for the next serviced slice of `pid`.
#[cfg(test)]
pub(crate) fn observe_next_slice(&self, pid: u64) -> Receiver<u64> {
self.inner.runtime.observe_next_slice(pid)
}
/// Installs a one-use readiness marker for the next genuine scheduler park of
/// `pid`. The delivered value is the process's slice count at the final probe
/// that selected `Wait`.
#[cfg(test)]
pub(crate) fn observe_next_park(&self, pid: u64) -> Receiver<u64> {
self.inner.runtime.observe_next_park(pid)
}
/// Returns a marker for the current park when `pid` is already settled, or
/// the next park when a coalesced readiness event has started another slice.
#[cfg(test)]
pub(crate) fn observe_settled_park(&self, pid: u64) -> Receiver<u64> {
self.inner.runtime.observe_settled_park(pid)
}
/// Queues an explicit outbound capacity for the next TCP process constructed.
#[cfg(test)]
pub(crate) fn queue_next_outbound_capacity(&self, capacity: usize) {
self.inner.runtime.queue_next_outbound_capacity(capacity);
}
#[cfg(test)]
pub(crate) fn install_participant_holdback_pause(&self, pid: u64) -> Receiver<()> {
self.inner.runtime.install_participant_holdback_pause(pid)
}
#[cfg(test)]
pub(crate) fn resume_test_process(&self, pid: u64) -> bool {
self.inner.runtime.ready_waker(pid).is_some_and(|waker| {
waker.fire();
true
})
}
/// Reserved push reply slots outstanding (test observability for the public
/// push paths — lets e2e tests assert slot reclamation and cap accounting).
#[cfg(test)]
pub(in crate::server::connection) fn pending_push_count(&self) -> usize {
self.inner.runtime.pending_push_count()
}
/// R6 test seam: a [`ReadyWaker`](super::wake::ReadyWaker) for `pid` — the same
/// handle a subscription-inbox or reply-availability notifier fires.
#[cfg(test)]
pub(in crate::server::connection) fn ready_waker(
&self,
pid: u64,
) -> Option<super::wake::ReadyWaker> {
self.inner.runtime.ready_waker(pid)
}
/// Registered readiness tokens held in host records (test observability).
#[cfg(test)]
pub(in crate::server::connection) fn readiness_registration_count(&self) -> usize {
self.inner.runtime.readiness_registration_count()
}
/// Kernel fd registered for `pid` (test observability for fd-reuse races).
#[cfg(test)]
pub(in crate::server::connection) fn readiness_fd(&self, pid: u64) -> Option<RawFd> {
self.inner.runtime.readiness_fd(pid)
}
/// The readiness token registered for `pid` (test observability). Lets the
/// fd-reuse successor oracle capture a stale token before reclamation and
/// replay its deregister afterwards, proving it is a keyed no-op.
#[cfg(test)]
pub(in crate::server::connection) fn readiness_token(
&self,
pid: u64,
) -> Option<ReadinessToken> {
self.inner.runtime.readiness_token(pid)
}
/// Installs the pid-specific reclamation gate (oracle 26) and returns its
/// `(reached, release, done)` endpoints.
#[cfg(test)]
pub(in crate::server::connection) fn install_reclaim_barrier(
&self,
pid: u64,
) -> (Arc<Barrier>, Arc<Barrier>, Arc<Barrier>) {
self.inner.runtime.install_reclaim_barrier(pid)
}
/// A weak handle to the connection runtime (test observability): lets a
/// lifetime test assert the runtime — and transitively the durable store's
/// writer lock — is released synchronously at supervisor drop rather than
/// held by the detached reclaim reactor.
#[cfg(test)]
pub(in crate::server::connection) fn runtime_weak(&self) -> Weak<ConnectionRuntime> {
Arc::downgrade(&self.inner.runtime)
}
/// Installs a one-use observation for the process-owned stream at `fd` being
/// dropped. External scheduler termination removes the process-table entry
/// before an executing native handler is destroyed, so table absence is not
/// sufficient evidence that the descriptor is reusable.
#[cfg(test)]
pub(in crate::server::connection) fn observe_process_stream_drop(
&self,
fd: RawFd,
) -> Receiver<()> {
self.inner.runtime.observe_process_stream_drop(fd)
}
/// Installs a one-use arm-to-probe barrier and returns its test endpoints.
#[cfg(test)]
pub(in crate::server::connection) fn install_pre_wait_barrier(
&self,
) -> (Arc<Barrier>, Arc<Barrier>) {
self.inner.runtime.install_pre_wait_barrier()
}
/// Barrier-staged final probes that observed newly arrived work.
#[cfg(test)]
pub(in crate::server::connection) fn pre_wait_probe_hits(&self) -> u64 {
self.inner.runtime.pre_wait_probe_hits()
}
/// Installs the one-use drain-park gate (oracles 18, 19) and returns its
/// `(armed, release)` endpoints.
#[cfg(test)]
pub(in crate::server::connection) fn install_drain_park_barrier(
&self,
) -> (Arc<Barrier>, Arc<Barrier>) {
self.inner.runtime.install_drain_park_barrier()
}
/// Drain waiter wakes that observed a real connection removal (oracle 12).
#[cfg(test)]
pub(in crate::server::connection) fn drain_exit_wakes(&self) -> u64 {
self.inner.runtime.drain_exit_wakes()
}
/// Drain waiter deadline expirations (oracles 12, 16).
#[cfg(test)]
pub(in crate::server::connection) fn drain_deadline_hits(&self) -> u64 {
self.inner.runtime.drain_deadline_hits()
}
}