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
use super::{
ConnectionRuntime, Entry, PushReplyDisposition, PushSlotResolution, WorkerRefusal, recover_lock,
};
impl ConnectionRuntime {
/// Drops every reply slot owned by connection `pid`, waking each awaiter with a
/// disconnected error (the dropped `Sender` disconnects the awaiter's
/// `Receiver`). Called from the close path so a connection that exits with
/// in-flight pushes signals worker death immediately instead of leaving each
/// awaiter to block the full push-reply timeout. A slot that [`resolve_push`]
/// already removed is gone, so it is untouched here; an unknown pid is a no-op.
pub(super) fn cancel_pushes_for_connection(&self, pid: u64) {
// S4: the close sweep is the reclamation of last resort ("connection
// close at the latest") — it must complete on a poisoned map too.
// SRV-027 R2: a requester waiting on a routed push to this connection
// is woken now, so its next slice names the worker gone instead of
// waiting out the reply deadline. The senders drop with the slots, so
// each requester reads the disconnected outcome.
let woken = {
let mut slots = recover_lock(&self.push_replies);
let mut woken = Vec::new();
slots.retain(|_correlation_id, pending| {
if pending.pid == pid {
woken.extend(pending.wake.take());
false
} else {
true
}
});
woken
};
for wake in woken {
wake.fire();
}
}
/// S3 second half (shape (b), check-after-insert): pre-publication
/// confirmation that the connection record for `pid` still exists, run in
/// the INSERT -> CONFIRM -> PUBLISH order (S7 — confirming after the
/// enqueue let a close-swept-then-answered push report `Err` for a Push the
/// client had received). `true` leaves the slot in place and the caller may
/// publish; `false` means a concurrent close already removed the record —
/// this call then removes the caller's own just-inserted slot (rolling back
/// its cap admission) so nothing is stranded, and the caller returns
/// WITHOUT publishing: an `Err` from the push methods guarantees no `Push`
/// control was published.
///
/// Why exactly one side always observes the slot: `remove` (the single
/// record-removal path) removes the host record BEFORE sweeping the pid's
/// push slots, and this check reads the record AFTER inserting the slot and
/// BEFORE the control is published. Both records accesses are serialized by
/// the `records` mutex, so either (i) this read precedes the record removal
/// — then the slot insert precedes the sweep (insert < read < removal <
/// sweep in the happens-before order) and the SWEEP observes and removes
/// the slot: if the control was published in the meantime the awaiter reads
/// the truthful disconnected outcome and a late client reply is a harmless
/// no-op; or (ii) this read follows the record removal — then THIS call
/// observes the absence, rolls the slot back itself, and nothing was
/// published. When both observe (a sweep and a rollback can both run in
/// case (ii) if the insert also preceded the sweep), removal is idempotent
/// and the cap is derived from map membership, so nothing double-releases.
///
/// Lock discipline: `records` and `push_replies` are NEVER held together —
/// here (`records` read, released, then `push_replies` on rollback), in
/// `remove` (`records` removal, released, then the sweep), and everywhere
/// else in this file the two mutexes are taken strictly sequentially, so no
/// lock-order inversion is possible. This adds ZERO work to the connection
/// slice path: the re-check runs on the push caller's thread only.
pub(in crate::server::connection) fn confirm_push_registration(
&self,
pid: u64,
correlation_id: u64,
) -> bool {
if self.is_registered(pid) {
return true;
}
self.cancel_push(correlation_id);
false
}
/// Number of reserved push reply slots outstanding. A benign wait-quantum
/// timeout must NOT change this (the slot survives); an explicit-deadline
/// expiry, a consumed reply, and connection close each release exactly one.
#[cfg(test)]
pub(in crate::server::connection) fn pending_push_count(&self) -> usize {
recover_lock(&self.push_replies).len()
}
/// Reserved push reply slots owned by connection `pid` — the exact quantity
/// the §5 `max_pending_pushes_per_connection` cap counts (test instrument).
#[cfg(test)]
pub(in crate::server::connection) fn pending_push_count_for(&self, pid: u64) -> usize {
recover_lock(&self.push_replies)
.values()
.filter(|pending| pending.pid == pid)
.count()
}
/// Resolves the reply slot for `correlation_id` with the client's reply
/// payload, waking the [`PushReplyAwaiter`]. Called by the connection process
/// when a correlated `PushReply` frame arrives. A different connection is
/// refused without consuming the slot. A missing slot — already
/// resolved, expired at its explicit deadline, dropped by connection close, or
/// an unknown id — is a harmless no-op: a late `PushReply` for a slot that is
/// gone is discarded here, never delivered and never a panic or desync.
pub(in crate::server::connection) fn resolve_push(
&self,
pid: u64,
correlation_id: u64,
payload: Vec<u8>,
) -> PushReplyDisposition {
self.settle_push(pid, correlation_id, PushSlotResolution::Reply(payload))
}
/// Uses the same owner check, removal, send and requester wake as an answer.
pub(in crate::server::connection) fn resolve_worker_refusal(
&self,
pid: u64,
correlation_id: u64,
refusal: WorkerRefusal,
) -> PushReplyDisposition {
self.settle_push(
pid,
correlation_id,
PushSlotResolution::WorkerRefused(refusal),
)
}
/// The station recorded by registration, never a name carried by the reply.
pub(in crate::server::connection) fn worker_station(&self, pid: u64) -> String {
recover_lock(&self.records)
.connections
.get(&pid)
.and_then(|record| record.registration.as_ref())
.map_or_else(
|| pid.to_string(),
|registration| registration.identity.clone(),
)
}
/// Settles the reply slot for `correlation_id` with the access refusal the
/// connection process decided for its `PushReply`: the slot is removed
/// exactly as for a delivered reply and the awaiter reads
/// [`ServerError::PushReplyRefused`] carrying `reason`. Same ownership and
/// missing-slot rules as [`resolve_push`](Self::resolve_push).
pub(in crate::server::connection) fn refuse_push(
&self,
pid: u64,
correlation_id: u64,
reason: String,
) -> PushReplyDisposition {
self.settle_push(
pid,
correlation_id,
PushSlotResolution::ReplyRefused { reason },
)
}
/// The one removal-and-send for a reply slot, shared by the delivered reply
/// and the refused one so neither can grow its own slot lifecycle.
fn settle_push(
&self,
pid: u64,
correlation_id: u64,
resolution: PushSlotResolution,
) -> PushReplyDisposition {
// Recover the guard so a poisoned registry still delivers and reclaims replies.
let mut slots = recover_lock(&self.push_replies);
let Entry::Occupied(entry) = slots.entry(correlation_id) else {
return PushReplyDisposition::UnknownCorrelation;
};
// The connection identity is checked before the slot or its credit is released.
if entry.get().pid != pid {
return PushReplyDisposition::SenderMismatch;
}
let pending = entry.remove();
// Keep removal and send atomic to awaiters checking whether the slot still exists.
// The unbounded send never blocks; an abandoned receiver is a named discard.
let disposition = match pending.sender.send(resolution) {
Ok(()) => PushReplyDisposition::Resolved,
Err(_) => PushReplyDisposition::ReceiverDropped,
};
drop(slots);
// SRV-027 R2: a routed push tells its requester, once, after the
// resolution is in the channel; the requester's slice reads it.
if let Some(wake) = pending.wake {
wake.fire();
}
disposition
}
}