liminal_server/server/connection/supervisor/supervisor_push.rs
1use super::{
2 Arc, ConnectionControl, ConnectionSupervisor, Duration, Instant, PushReplyAwaiter, ServerError,
3};
4
5impl ConnectionSupervisor {
6 /// Pushes an opaque payload to a specific connected client over that client's
7 /// existing connection and returns an awaiter for the client's correlated reply.
8 ///
9 /// This is the server-initiated leg (server-to-client), the inverse of every
10 /// other request frame. It allocates a correlation id, registers a one-shot
11 /// reply slot keyed by that id, and enqueues a [`ConnectionControl::Push`] for
12 /// the connection process owning `pid`; that process writes a [`Frame::Push`]
13 /// out on its socket. When the client answers with a `PushReply` carrying the
14 /// same correlation id, the connection process resolves the awaiter's slot. The
15 /// returned [`PushReplyAwaiter`] blocks (bounded) for that reply.
16 ///
17 /// The reply's lifetime belongs to the push, not to any one
18 /// [`PushReplyAwaiter::receive`] call: this no-deadline push reserves a slot
19 /// that is reclaimed only by (a) the reply being consumed or (b) the
20 /// connection closing. An elapsed `receive` poll is a benign re-arm, never a
21 /// failure and never a cancellation. The §5
22 /// `max_pending_pushes_per_connection` cap bounds abandonment; use
23 /// [`push_to_connection_with_deadline`](Self::push_to_connection_with_deadline)
24 /// when the reply must have an explicit expiry.
25 ///
26 /// # Errors
27 /// Returns [`ServerError`] when the correlation id cannot be allocated, the
28 /// reply slot cannot be registered, or the control message cannot be enqueued
29 /// for the (possibly already-gone or concurrently-closing) connection
30 /// process. PUBLICATION INVARIANT: an `Err` guarantees no `Push` control was
31 /// published — the client never sees a `Push` frame for a failed call.
32 /// Conversely `Ok` promises ADMISSION, not delivery: the awaiter's outcome
33 /// is the delivery truth (a push admitted just as its connection closes
34 /// resolves to the truthful disconnected outcome, never to a lost reply).
35 pub fn push_to_connection(
36 &self,
37 pid: u64,
38 payload: Vec<u8>,
39 ) -> Result<PushReplyAwaiter, ServerError> {
40 self.push_with_deadline(pid, payload, None, None)
41 }
42
43 /// Like [`push_to_connection`](Self::push_to_connection), with the push
44 /// made ON `subject`: the reply slot retains the subject, and a connection
45 /// admitted on a Lys pass has its `PushReply` decided as `serve` on that
46 /// subject before the slot is resolved (the access matrix's `push_binding`
47 /// source). A refused reply resolves the awaiter to
48 /// [`ServerError::PushReplyRefused`], so the pusher learns the outcome by
49 /// type rather than waiting out a deadline. A push made without a subject
50 /// to a Lys-admitted connection is refused the same way when its reply
51 /// arrives: the subject is recorded here, at push time, or not at all.
52 /// `deadline` is as in
53 /// [`push_to_connection_with_deadline`](Self::push_to_connection_with_deadline).
54 ///
55 /// # Errors
56 /// As [`push_to_connection_with_deadline`](Self::push_to_connection_with_deadline).
57 pub fn push_to_subject(
58 &self,
59 pid: u64,
60 subject: &str,
61 payload: Vec<u8>,
62 deadline: Option<Duration>,
63 ) -> Result<PushReplyAwaiter, ServerError> {
64 self.push_with_deadline(pid, payload, deadline, Some(subject.to_owned()))
65 }
66
67 /// Like [`push_to_connection`](Self::push_to_connection) but attaches an
68 /// explicit reply deadline to the reserved slot: `deadline` is a DURATION
69 /// FROM NOW bounding the reply's lifetime — a property of THIS push rather
70 /// than of any [`PushReplyAwaiter::receive`] wait quantum.
71 ///
72 /// Deadline expiry is evaluated HOST-SIDE and LAZILY — at the next `receive`
73 /// touch, and at connection close at the latest. It never wakes the connection
74 /// process, adds no timer thread, and runs no periodic sweeper: a push that is
75 /// abandoned and never polled resolves at the next host-side touch (connection
76 /// close). At expiry the slot resolves to [`ServerError::PushReplyExpired`],
77 /// is removed, and its §5 `max_pending_pushes_per_connection` cap admission is
78 /// released. An elapsed `receive` poll BEFORE the deadline is still a benign
79 /// re-arm. A `receive` call in flight when the deadline falls due returns
80 /// the terminal expiry PROMPTLY — it waits the earlier of its quantum and
81 /// the deadline, so a large quantum can never extend the reply's lifetime
82 /// and the terminal outcome is quantum-independent.
83 ///
84 /// The deadline is evaluated at OBSERVATION POINTS, not enforced against the
85 /// wall clock: a reply that arrives before expiry is observed is delivered
86 /// normally, even if it arrives after the deadline instant. The deadline
87 /// bounds waiting and slot occupancy; it is not a delivery-freshness
88 /// guarantee. (This is deliberate — a reply is checked for at every
89 /// observation point before the deadline is, so an answer in hand always
90 /// beats an expiry.)
91 ///
92 /// # Errors
93 /// Returns [`ServerError`] when `deadline` is not representable on the
94 /// monotonic clock (an extreme duration is refused, never a panic), the
95 /// correlation id cannot be allocated, the reply slot cannot be registered,
96 /// or the control message cannot be enqueued for the (possibly already-gone
97 /// or concurrently-closing) connection process. PUBLICATION INVARIANT: an
98 /// `Err` guarantees no `Push` control was published — the client never sees
99 /// a `Push` frame for a failed call. Conversely `Ok` promises ADMISSION,
100 /// not delivery: the awaiter's outcome is the delivery truth.
101 pub fn push_to_connection_with_deadline(
102 &self,
103 pid: u64,
104 payload: Vec<u8>,
105 deadline: Duration,
106 ) -> Result<PushReplyAwaiter, ServerError> {
107 self.push_with_deadline(pid, payload, Some(deadline), None)
108 }
109
110 /// Shared body for the no-deadline and explicit-deadline push paths. With
111 /// `deadline == None` this is byte-for-byte the historical
112 /// `push_to_connection` behaviour (no per-slot deadline); with `Some`, the
113 /// slot carries an absolute expiry evaluated lazily at `receive`. `subject`
114 /// is retained on the slot for the reply's access decision
115 /// ([`push_to_subject`](Self::push_to_subject)); `None` records no binding.
116 fn push_with_deadline(
117 &self,
118 pid: u64,
119 payload: Vec<u8>,
120 deadline: Option<Duration>,
121 subject: Option<String>,
122 ) -> Result<PushReplyAwaiter, ServerError> {
123 // S5: an extreme `Duration` must surface as this fallible API's typed
124 // error, not an `Instant` addition panic. Checked BEFORE any slot is
125 // registered so a refused deadline leaves nothing to roll back.
126 let deadline_at = match deadline {
127 None => None,
128 Some(window) => {
129 Some(
130 Instant::now()
131 .checked_add(window)
132 .ok_or_else(|| ServerError::ListenerAccept {
133 message: format!(
134 "cannot push to connection process {pid}: reply deadline of {window:?} overflows the monotonic clock"
135 ),
136 })?,
137 )
138 }
139 };
140 let correlation_id = self.inner.runtime.next_push_correlation_id();
141 let receiver =
142 self.inner
143 .runtime
144 .register_push(pid, correlation_id, deadline_at, subject, None)?;
145 // S3+S7 close-vs-register wall, ordered INSERT -> CONFIRM -> PUBLISH.
146 // The confirmation runs BEFORE the control is enqueued, which yields the
147 // PUBLICATION INVARIANT: an `Err` from this method guarantees no `Push`
148 // control was published — the client never sees a Push for a failed
149 // call. (Confirming after the enqueue was S7's non-linearizable race: a
150 // close could sweep, the published Push could already be answered and
151 // resolved, and the failed confirmation then returned `Err` for a push
152 // the client had received.) A close landing AFTER a successful confirm
153 // linearizes after push admission: the enqueue either fails (process
154 // gone — rollback below, `Err` truthful, nothing delivered) or succeeds
155 // with the slot already swept, and the awaiter then reads the truthful
156 // DISCONNECTED while a late client reply is the pinned harmless no-op.
157 // The exactly-one-side-observes argument lives at
158 // `confirm_push_registration`.
159 if !self
160 .inner
161 .runtime
162 .confirm_push_registration(pid, correlation_id)
163 {
164 return Err(ServerError::ListenerAccept {
165 message: format!(
166 "cannot push to connection process {pid}: the connection closed during push registration"
167 ),
168 });
169 }
170 let control = ConnectionControl::Push {
171 correlation_id,
172 payload,
173 };
174 if self.inner.enqueue_control(pid, control) {
175 Ok(PushReplyAwaiter {
176 correlation_id,
177 receiver,
178 deadline: deadline_at,
179 runtime: Arc::downgrade(&self.inner.runtime),
180 })
181 } else {
182 // The control provably never reached a consumer: `enqueue_control`
183 // returns false only when the control was never queued at all, or
184 // when its failed-wake rollback REMOVED the queued control (S8 — an
185 // entry a drain already consumed counts as published and returns
186 // true, with the slot lifecycle carrying the delivery truth).
187 // Dropping the now-unreachable reply slot here therefore keeps the
188 // publication invariant exact on every `Err` path. The message
189 // states that outcome rather than naming a cause this branch cannot
190 // distinguish — the refused-queue case has already warned with its
191 // own error.
192 self.inner.runtime.cancel_push(correlation_id);
193 Err(ServerError::ListenerAccept {
194 message: format!(
195 "cannot push to connection process {pid}: the control was not published"
196 ),
197 })
198 }
199 }
200
201 /// Server `Push` frames this supervisor's connections have queued for their
202 /// sockets since it started. A push that failed, was cancelled or was rolled
203 /// back never reaches a socket and is not counted.
204 #[must_use]
205 pub fn pushes_queued(&self) -> u64 {
206 self.inner.runtime.pushes_queued()
207 }
208}