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
use super::{
ConnectionRuntime, Instant, Ordering, OutboundError, PendingPush, PushBinding,
PushSlotDisposition, PushSlotResolution, Receiver, ServerError, channel, lock, recover_lock,
};
#[cfg(test)]
use super::{PushReplyAwaiter, Weak};
impl ConnectionRuntime {
/// Allocates the next monotonic push correlation id.
pub(super) fn next_push_correlation_id(&self) -> u64 {
self.next_push_id.fetch_add(1, Ordering::Relaxed)
}
/// Registers a one-shot reply slot for `correlation_id`, owned by connection
/// `pid`, and returns its receiver. `deadline` is the slot's optional absolute
/// reply expiry (`None` = the default no-deadline shape). The connection
/// process resolves the slot via [`resolve_push`]; the close path drops the
/// connection's outstanding slots via [`cancel_pushes_for_connection`]; an
/// explicit deadline resolves it via [`expire_push_if_due`].
///
/// # Errors
/// Returns [`ServerError`] when the correlation registry mutex is poisoned.
pub(super) fn register_push(
&self,
pid: u64,
correlation_id: u64,
deadline: Option<Instant>,
subject: Option<String>,
wake: Option<super::wake::ReadyWaker>,
) -> Result<Receiver<PushSlotResolution>, ServerError> {
let (sender, receiver) = channel();
let limit = self.limits.max_pending_pushes_per_connection;
{
let mut slots = lock(&self.push_replies, "push correlation registry")?;
// §5 `max_pending_pushes_per_connection`: refuse a new in-flight push
// once this connection already holds the cap. Counted per owning pid so
// one connection cannot exhaust the shared registry; slots free on
// reply, deadline expiry, or connection close. The count-and-insert
// stays under the one lock so the cap is enforced atomically.
let outstanding = slots.values().filter(|pending| pending.pid == pid).count();
if outstanding >= limit {
return Err(ServerError::ConnectionCapReached {
operation: "server push".to_owned(),
cap: "max_pending_pushes_per_connection",
limit,
});
}
slots.insert(
correlation_id,
PendingPush {
pid,
subject,
sender,
deadline,
wake,
},
);
}
Ok(receiver)
}
/// Registers a confirmed, no-deadline push slot for `pid` without a
/// supervisor, so a unit fixture can answer it with a `PushReply` frame
/// through `apply_frame` and read the outcome on the awaiter. The awaiter's
/// runtime handle is empty: the no-deadline receive path never upgrades it.
#[cfg(test)]
pub(in crate::server::connection) fn push_for_tests(
&self,
pid: u64,
subject: Option<&str>,
) -> Result<PushReplyAwaiter, ServerError> {
let correlation_id = self.next_push_correlation_id();
let receiver =
self.register_push(pid, correlation_id, None, subject.map(str::to_owned), None)?;
if !self.confirm_push_registration(pid, correlation_id) {
return Err(ServerError::ListenerAccept {
message: format!("connection process {pid} is not registered"),
});
}
Ok(PushReplyAwaiter {
correlation_id,
receiver,
deadline: None,
runtime: Weak::new(),
})
}
/// The subject retained by `pid`'s reply slot for `correlation_id`, read
/// before the reply is decided. Read and settle are two lock takes on
/// purpose (never nested with a record lock): a slot swept between them is
/// named by the settle as `UnknownCorrelation`, which is the truth then.
pub(in crate::server::connection) fn push_binding(
&self,
pid: u64,
correlation_id: u64,
) -> PushBinding {
let slots = recover_lock(&self.push_replies);
match slots.get(&correlation_id) {
Some(pending) if pending.pid == pid => pending
.subject
.clone()
.map_or(PushBinding::Subjectless, PushBinding::Subject),
Some(_) | None => PushBinding::Unbound,
}
}
/// Host-side, lazy evaluation of a push's reply deadline, called from an
/// elapsed [`PushReplyAwaiter::receive`] quantum. This NEVER wakes the
/// connection process and runs no timer — it inspects supervisor-owned state
/// under the registry lock only.
///
/// A slot with an explicit deadline that has passed is removed here (dropping
/// its `Sender` and releasing its §5 `max_pending_pushes_per_connection` cap
/// admission, since the cap is the per-pid slot count) and reported
/// [`PushSlotDisposition::Expired`]. A slot with no deadline, or a deadline
/// still in the future, is left UNTOUCHED and reported
/// [`PushSlotDisposition::Live`] — the elapsed quantum is a benign re-arm. A
/// missing slot is [`PushSlotDisposition::Absent`].
pub(super) fn expire_push_if_due(&self, correlation_id: u64) -> PushSlotDisposition {
// S4: a poisoned registry must NOT read as slot absence — the slot (and
// its cap admission) may still be in the map. Reclamation recovers the
// guard: removal-only operations are sound on a recovered map (a panic
// in another critical section cannot leave the HashMap itself in a
// partial state; only our bookkeeping invariants could be stale, and
// removal restores them). Admission (`register_push`) stays fail-closed.
let mut slots = recover_lock(&self.push_replies);
let Some(pending) = slots.get(&correlation_id) else {
return PushSlotDisposition::Absent;
};
// Copy the deadline out so the immutable borrow of `slots` ends before the
// conditional `remove` below takes a mutable one.
let deadline = pending.deadline;
match deadline {
Some(at) if Instant::now() >= at => {
slots.remove(&correlation_id);
PushSlotDisposition::Expired
}
_ => PushSlotDisposition::Live,
}
}
/// Drops a registered reply slot without resolving it, used on the
/// push-enqueue failure path (the control could not be delivered to a
/// now-gone process, so the just-reserved slot is unreachable). Dropping the
/// slot's `Sender` wakes a still-waiting awaiter with a disconnected error.
///
/// Returns whether THIS call removed the slot. Removal under the registry
/// mutex is the atomic resolved-vs-cancelled transition: `false` means
/// another path won — [`resolve_push`](Self::resolve_push) already sent the
/// reply (its send happens under the same lock, so the payload is already in
/// the channel when this returns), or the connection's close path dropped
/// the slot (sender gone, channel disconnected).
/// Records one server `Push` frame queued for a connection's socket.
pub(in crate::server::connection) fn record_push_queued(&self) {
self.pushes_queued.fetch_add(1, Ordering::Relaxed);
}
/// Server `Push` frames queued for sockets so far.
pub(in crate::server::connection) fn pushes_queued(&self) -> u64 {
self.pushes_queued.load(Ordering::Relaxed)
}
pub(in crate::server::connection) fn cancel_push(&self, correlation_id: u64) -> bool {
// S4: reclamation recovers a poisoned guard — a rollback that silently
// skipped its removal would strand the slot and its cap admission.
recover_lock(&self.push_replies)
.remove(&correlation_id)
.is_some()
}
/// Settles the reply slot for a push whose frame could not be enqueued,
/// applying the one ruling that separates a certain failure from a
/// recoverable one.
///
/// A frame larger than the whole outbound buffer can never be carried on
/// this connection at this bound, so its slot resolves to the named terminal
/// [`ServerError::PushFrameExceedsOutboundCapacity`]. EVERYTHING ELSE —
/// congestion behind traffic still draining, an encode failure, a socket
/// write error — cancels the slot exactly as before, and its dropped
/// `Sender` keeps the retryable [`ServerError::PushReplyDisconnected`]
/// reading, because retrying is the correct answer to all three.
///
/// This is the sole branch between the two, deliberately: a caller that
/// widened it would turn a busy moment into a permanent failure, which is a
/// worse defect than the one the named outcome exists to fix.
pub(in crate::server::connection) fn settle_push_enqueue_failure(
&self,
pid: u64,
correlation_id: u64,
error: &OutboundError,
) {
if let Some(too_large) = error.frame_exceeds_capacity() {
tracing::warn!(
connection_pid = pid,
correlation_id,
needed = too_large.needed,
capacity = too_large.capacity,
queued = too_large.queued,
"server push frame exceeds the connection's whole outbound capacity; \
it can never be sent on this connection and its reply slot is failed by name"
);
self.fail_push_outbound_frame_too_large(
correlation_id,
too_large.needed,
too_large.capacity,
too_large.queued,
);
return;
}
tracing::warn!(
connection_pid = pid,
correlation_id,
%error,
"server push could not be enqueued; the push reply slot is cancelled"
);
self.cancel_push(correlation_id);
}
/// Resolves the reply slot for `correlation_id` with the ONE named terminal
/// outcome the outbound path can prove: the push's encoded frame is larger
/// than the whole outbound buffer, so it can never be carried on this
/// connection at this bound and no retry could change that.
///
/// Callers must have established `needed > capacity` first —
/// [`settle_push_enqueue_failure`](Self::settle_push_enqueue_failure) is the
/// only one, and it does. A frame that merely does not fit the CURRENT free
/// space is congestion and belongs on [`cancel_push`](Self::cancel_push),
/// whose dropped `Sender` keeps its retryable disconnect reading.
///
/// A slot that is already gone — answered, expired, or swept by connection
/// close — is a harmless no-op, exactly as a late reply is to
/// [`resolve_push`](Self::resolve_push): the first path to remove it under
/// the registry lock wins and the outcome it chose stands.
///
/// Removal and delivery stay under the ONE registry lock for the same reason
/// `resolve_push` does it — an awaiter that observes the slot gone is then
/// guaranteed to find the outcome already in the channel, instead of seeing
/// the removal, finding the channel empty, and reporting something else.
fn fail_push_outbound_frame_too_large(
&self,
correlation_id: u64,
needed: usize,
capacity: usize,
queued: usize,
) {
// S4: as `cancel_push` — a skipped removal here would strand the slot and
// its cap admission, and lose the only report the caller will ever get.
let mut slots = recover_lock(&self.push_replies);
if let Some(pending) = slots.remove(&correlation_id) {
pending
.sender
.send(PushSlotResolution::OutboundFrameTooLarge {
needed,
capacity,
queued,
})
.ok();
}
}
}