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
use super::{
Arc, ConnectionControl, ConnectionSupervisor, Duration, Instant, PushReplyAwaiter, ServerError,
};
impl ConnectionSupervisor {
/// Pushes an opaque payload to a specific connected client over that client's
/// existing connection and returns an awaiter for the client's correlated reply.
///
/// This is the server-initiated leg (server-to-client), the inverse of every
/// other request frame. It allocates a correlation id, registers a one-shot
/// reply slot keyed by that id, and enqueues a [`ConnectionControl::Push`] for
/// the connection process owning `pid`; that process writes a [`Frame::Push`]
/// out on its socket. When the client answers with a `PushReply` carrying the
/// same correlation id, the connection process resolves the awaiter's slot. The
/// returned [`PushReplyAwaiter`] blocks (bounded) for that reply.
///
/// The reply's lifetime belongs to the push, not to any one
/// [`PushReplyAwaiter::receive`] call: this no-deadline push reserves a slot
/// that is reclaimed only by (a) the reply being consumed or (b) the
/// connection closing. An elapsed `receive` poll is a benign re-arm, never a
/// failure and never a cancellation. The §5
/// `max_pending_pushes_per_connection` cap bounds abandonment; use
/// [`push_to_connection_with_deadline`](Self::push_to_connection_with_deadline)
/// when the reply must have an explicit expiry.
///
/// # Errors
/// Returns [`ServerError`] when the correlation id cannot be allocated, the
/// reply slot cannot be registered, or the control message cannot be enqueued
/// for the (possibly already-gone or concurrently-closing) connection
/// process. PUBLICATION INVARIANT: an `Err` guarantees no `Push` control was
/// published — the client never sees a `Push` frame for a failed call.
/// Conversely `Ok` promises ADMISSION, not delivery: the awaiter's outcome
/// is the delivery truth (a push admitted just as its connection closes
/// resolves to the truthful disconnected outcome, never to a lost reply).
pub fn push_to_connection(
&self,
pid: u64,
payload: Vec<u8>,
) -> Result<PushReplyAwaiter, ServerError> {
self.push_with_deadline(pid, payload, None, None)
}
/// Like [`push_to_connection`](Self::push_to_connection), with the push
/// made ON `subject`: the reply slot retains the subject, and a connection
/// admitted on a Lys pass has its `PushReply` decided as `serve` on that
/// subject before the slot is resolved (the access matrix's `push_binding`
/// source). A refused reply resolves the awaiter to
/// [`ServerError::PushReplyRefused`], so the pusher learns the outcome by
/// type rather than waiting out a deadline. A push made without a subject
/// to a Lys-admitted connection is refused the same way when its reply
/// arrives: the subject is recorded here, at push time, or not at all.
/// `deadline` is as in
/// [`push_to_connection_with_deadline`](Self::push_to_connection_with_deadline).
///
/// # Errors
/// As [`push_to_connection_with_deadline`](Self::push_to_connection_with_deadline).
pub fn push_to_subject(
&self,
pid: u64,
subject: &str,
payload: Vec<u8>,
deadline: Option<Duration>,
) -> Result<PushReplyAwaiter, ServerError> {
self.push_with_deadline(pid, payload, deadline, Some(subject.to_owned()))
}
/// Like [`push_to_connection`](Self::push_to_connection) but attaches an
/// explicit reply deadline to the reserved slot: `deadline` is a DURATION
/// FROM NOW bounding the reply's lifetime — a property of THIS push rather
/// than of any [`PushReplyAwaiter::receive`] wait quantum.
///
/// Deadline expiry is evaluated HOST-SIDE and LAZILY — at the next `receive`
/// touch, and at connection close at the latest. It never wakes the connection
/// process, adds no timer thread, and runs no periodic sweeper: a push that is
/// abandoned and never polled resolves at the next host-side touch (connection
/// close). At expiry the slot resolves to [`ServerError::PushReplyExpired`],
/// is removed, and its §5 `max_pending_pushes_per_connection` cap admission is
/// released. An elapsed `receive` poll BEFORE the deadline is still a benign
/// re-arm. A `receive` call in flight when the deadline falls due returns
/// the terminal expiry PROMPTLY — it waits the earlier of its quantum and
/// the deadline, so a large quantum can never extend the reply's lifetime
/// and the terminal outcome is quantum-independent.
///
/// The deadline is evaluated at OBSERVATION POINTS, not enforced against the
/// wall clock: a reply that arrives before expiry is observed is delivered
/// normally, even if it arrives after the deadline instant. The deadline
/// bounds waiting and slot occupancy; it is not a delivery-freshness
/// guarantee. (This is deliberate — a reply is checked for at every
/// observation point before the deadline is, so an answer in hand always
/// beats an expiry.)
///
/// # Errors
/// Returns [`ServerError`] when `deadline` is not representable on the
/// monotonic clock (an extreme duration is refused, never a panic), the
/// correlation id cannot be allocated, the reply slot cannot be registered,
/// or the control message cannot be enqueued for the (possibly already-gone
/// or concurrently-closing) connection process. PUBLICATION INVARIANT: an
/// `Err` guarantees no `Push` control was published — the client never sees
/// a `Push` frame for a failed call. Conversely `Ok` promises ADMISSION,
/// not delivery: the awaiter's outcome is the delivery truth.
pub fn push_to_connection_with_deadline(
&self,
pid: u64,
payload: Vec<u8>,
deadline: Duration,
) -> Result<PushReplyAwaiter, ServerError> {
self.push_with_deadline(pid, payload, Some(deadline), None)
}
/// Shared body for the no-deadline and explicit-deadline push paths. With
/// `deadline == None` this is byte-for-byte the historical
/// `push_to_connection` behaviour (no per-slot deadline); with `Some`, the
/// slot carries an absolute expiry evaluated lazily at `receive`. `subject`
/// is retained on the slot for the reply's access decision
/// ([`push_to_subject`](Self::push_to_subject)); `None` records no binding.
fn push_with_deadline(
&self,
pid: u64,
payload: Vec<u8>,
deadline: Option<Duration>,
subject: Option<String>,
) -> Result<PushReplyAwaiter, ServerError> {
// S5: an extreme `Duration` must surface as this fallible API's typed
// error, not an `Instant` addition panic. Checked BEFORE any slot is
// registered so a refused deadline leaves nothing to roll back.
let deadline_at = match deadline {
None => None,
Some(window) => {
Some(
Instant::now()
.checked_add(window)
.ok_or_else(|| ServerError::ListenerAccept {
message: format!(
"cannot push to connection process {pid}: reply deadline of {window:?} overflows the monotonic clock"
),
})?,
)
}
};
let correlation_id = self.inner.runtime.next_push_correlation_id();
let receiver =
self.inner
.runtime
.register_push(pid, correlation_id, deadline_at, subject, None)?;
// S3+S7 close-vs-register wall, ordered INSERT -> CONFIRM -> PUBLISH.
// The confirmation runs BEFORE the control is enqueued, which yields the
// PUBLICATION INVARIANT: an `Err` from this method guarantees no `Push`
// control was published — the client never sees a Push for a failed
// call. (Confirming after the enqueue was S7's non-linearizable race: a
// close could sweep, the published Push could already be answered and
// resolved, and the failed confirmation then returned `Err` for a push
// the client had received.) A close landing AFTER a successful confirm
// linearizes after push admission: the enqueue either fails (process
// gone — rollback below, `Err` truthful, nothing delivered) or succeeds
// with the slot already swept, and the awaiter then reads the truthful
// DISCONNECTED while a late client reply is the pinned harmless no-op.
// The exactly-one-side-observes argument lives at
// `confirm_push_registration`.
if !self
.inner
.runtime
.confirm_push_registration(pid, correlation_id)
{
return Err(ServerError::ListenerAccept {
message: format!(
"cannot push to connection process {pid}: the connection closed during push registration"
),
});
}
let control = ConnectionControl::Push {
correlation_id,
payload,
};
if self.inner.enqueue_control(pid, control) {
Ok(PushReplyAwaiter {
correlation_id,
receiver,
deadline: deadline_at,
runtime: Arc::downgrade(&self.inner.runtime),
})
} else {
// The control provably never reached a consumer: `enqueue_control`
// returns false only when the control was never queued at all, or
// when its failed-wake rollback REMOVED the queued control (S8 — an
// entry a drain already consumed counts as published and returns
// true, with the slot lifecycle carrying the delivery truth).
// Dropping the now-unreachable reply slot here therefore keeps the
// publication invariant exact on every `Err` path. The message
// states that outcome rather than naming a cause this branch cannot
// distinguish — the refused-queue case has already warned with its
// own error.
self.inner.runtime.cancel_push(correlation_id);
Err(ServerError::ListenerAccept {
message: format!(
"cannot push to connection process {pid}: the control was not published"
),
})
}
}
/// Server `Push` frames this supervisor's connections have queued for their
/// sockets since it started. A push that failed, was cancelled or was rolled
/// back never reaches a socket and is not counted.
#[must_use]
pub fn pushes_queued(&self) -> u64 {
self.inner.runtime.pushes_queued()
}
}