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
//! The connection process's control vocabulary, revocation, server pushes
//! and mailbox messages.
use super::{ConnectionTransport, PUSH_STREAM_ID, SliceStep, TransportConnectionProcess};
use crate::auth_revocation::CloseWrite;
use crate::server::connection::outbound::DrainOutcome;
use crate::server::connection::supervisor::runtime::ConnectionControl;
use crate::server::participant::ConnectionFateClass;
use beamr::atom::Atom;
use beamr::native::native_process::NativeOutcome;
use beamr::process::ExitReason;
use beamr::term::Term;
use liminal::protocol::Frame;
impl<T: ConnectionTransport> TransportConnectionProcess<T> {
pub(super) fn handle_control(
&mut self,
pid: u64,
control: ConnectionControl,
) -> Option<NativeOutcome> {
match control {
ConnectionControl::NotifyShutdown => {
self.notify_shutdown(pid, true);
None
}
ConnectionControl::ForceClose => {
crate::agents::runtime::connection_closed(
self.runtime.services(),
pid,
"server_shutdown",
);
self.notify_shutdown(pid, false);
// Flush the enqueued `Disconnect` (and any residue) before the
// stream is dropped; best-effort, since we are stopping regardless.
// A failure here loses the `Disconnect` itself, and the client
// learns of the shutdown from the dropped socket instead of from
// the frame. Accepted: the force-close decision is taken, and a
// retry or sleep on this path would hold the whole shutdown drain
// hostage to one unresponsive peer.
if let Err(error) = self.drain_outbound() {
tracing::debug!(connection_pid = pid, %error, "the close path's best-effort drain did not finish; the peer's half-close ends it");
}
if let Err(error) =
self.complete_connection_fate(ConnectionFateClass::ServerShutdown)
{
return Some(match self.fail_fate(pid, &error) {
SliceStep::Continue => NativeOutcome::Continue,
SliceStep::Stop(reason) => NativeOutcome::Stop(reason),
});
}
self.release_conversations();
// Host removal ACKs readiness deregistration while both the
// process stream and record fd guard are still live.
self.runtime.finish(pid);
self.transport.release();
Some(NativeOutcome::Stop(ExitReason::Normal))
}
ConnectionControl::Push {
correlation_id,
payload,
} => {
self.write_push(pid, correlation_id, payload);
None
}
ConnectionControl::Revocation => self.revocation_sweep(pid),
}
}
/// ACCESS-005 R4/R6: acts on this connection's rights after a grant
/// revocation changed them. A lost admitting right closes the connection
/// with its reason; a withdrawn right's subscriptions and conversations
/// are retired with named outcomes. Reached from the revocation control
/// and from every slice, so a control the queue refused is still acted on
/// at the connection's own wake.
pub(super) fn revocation_sweep(&mut self, pid: u64) -> Option<NativeOutcome> {
let close = self
.state
.revocation
.as_ref()
.and_then(crate::auth_revocation::ConnectionGrants::close_pending);
if let Some(reason) = close {
return Some(self.revocation_close(pid, reason));
}
for frame in super::super::apply::retire_withdrawn_rights(&self.runtime, &mut self.state) {
if let Err(error) = self.outbound.enqueue_frame(&frame) {
tracing::warn!(
connection_pid = pid,
%error,
"a withdrawn right's refusal could not be queued"
);
}
}
None
}
/// Closes the connection for a revocation (ACCESS-005 R5, R6): one
/// attempt to queue and flush the reason-bearing close, its outcome
/// recorded as attempted (never as received), then the same fate, cleanup
/// and single removal funnel a forced close takes.
pub(super) fn revocation_close(
&mut self,
pid: u64,
reason: crate::auth_revocation::RevocationRefusal,
) -> NativeOutcome {
crate::agents::runtime::connection_closed(self.runtime.services(), pid, reason.name());
let write = match reason.close_reason() {
Some(close) if self.transport.is_connected() => {
match self.outbound.enqueue_frame(&Frame::DisconnectReason {
flags: 0,
reason: close,
}) {
Ok(()) => match self.drain_outbound_for_close() {
Ok(DrainOutcome::Drained) => CloseWrite::Flushed,
Ok(DrainOutcome::Progress | DrainOutcome::WouldBlockWithResidue) => {
CloseWrite::Residue
}
Err(_) => CloseWrite::TransportFailed,
},
Err(_) => CloseWrite::NotQueued,
}
}
Some(_) => CloseWrite::NoStream,
None => CloseWrite::NotQueued,
};
if let Some(authority) = self.runtime.revocation() {
authority.record_close_write(pid, write);
}
tracing::info!(
connection_pid = pid,
reason = reason.name(),
write = ?write,
"closing connection: a grant revocation withdrew the right that admitted it"
);
if let Err(error) = self.complete_connection_fate(ConnectionFateClass::ServerShutdown) {
return match self.fail_fate(pid, &error) {
SliceStep::Continue => NativeOutcome::Continue,
SliceStep::Stop(reason) => NativeOutcome::Stop(reason),
};
}
self.release_conversations();
self.runtime.finish(pid);
self.transport.release();
NativeOutcome::Stop(ExitReason::Normal)
}
/// Enqueues a server-initiated [`Frame::Push`] into the outbound buffer.
///
/// The frame is flushed by the slice's outbound drain (this control message is
/// handled at the start of the slice, before `handle_slice` runs). A missing
/// stream or an encode failure cancels the push slot so the awaiter does not
/// block forever on a reply that can never arrive; the connection itself is
/// left to its normal lifecycle. An enqueue failure is settled by
/// [`ConnectionRuntime::settle_push_enqueue_failure`](crate::server::connection::supervisor::runtime::ConnectionRuntime::settle_push_enqueue_failure), which names the one
/// certain case — a frame larger than the whole outbound buffer — and
/// cancels every recoverable one exactly as before.
pub(super) fn write_push(&mut self, pid: u64, correlation_id: u64, payload: Vec<u8>) {
if !self.transport.is_connected() {
tracing::warn!(
connection_pid = pid,
correlation_id,
"server push skipped because connection stream is unavailable"
);
self.runtime.cancel_push(correlation_id);
return;
}
let frame = match Frame::new_push(PUSH_STREAM_ID, correlation_id, payload) {
Ok(frame) => frame,
Err(error) => {
tracing::warn!(
connection_pid = pid,
correlation_id,
%error,
"server push frame could not be constructed"
);
self.runtime.cancel_push(correlation_id);
return;
}
};
match self.outbound.enqueue_frame(&frame) {
Ok(()) => self.runtime.record_push_queued(),
Err(error) => self
.runtime
.settle_push_enqueue_failure(pid, correlation_id, &error),
}
}
pub(super) fn notify_shutdown(&mut self, pid: u64, subscribers_only: bool) {
if self.state.shutdown_notification_attempted {
return;
}
if subscribers_only && self.state.subscriptions.is_empty() {
return;
}
self.state.shutdown_notification_attempted = true;
if !self.transport.is_connected() {
tracing::warn!(
connection_pid = pid,
peer_addr = ?self.peer_addr,
"shutdown notification skipped because connection stream is unavailable"
);
return;
}
match self.outbound.enqueue_frame(&Frame::Disconnect { flags: 0 }) {
Ok(()) => {
tracing::debug!(
connection_pid = pid,
peer_addr = ?self.peer_addr,
subscriber_count = self.state.subscriptions.len(),
"enqueued shutdown notification to connection"
);
}
Err(error) => {
tracing::warn!(
connection_pid = pid,
peer_addr = ?self.peer_addr,
%error,
"shutdown notification could not be enqueued; connection will not be retried"
);
}
}
}
pub(super) fn handle_message(&mut self, pid: u64, message: Term) -> Option<NativeOutcome> {
if message == Term::atom(Atom::ERROR) {
self.release_conversations();
self.runtime
.mark_crashed(pid, ExitReason::Error, self.peer_addr);
return Some(NativeOutcome::Stop(ExitReason::Error));
}
if message.as_atom() == Some(self.runtime.control_atom()) {
while let Some(control) = self.runtime.pop_control(pid) {
if let Some(outcome) = self.handle_control(pid, control) {
return Some(outcome);
}
}
}
// R6 (§1.2(4)): a `READY` marker (from any wake source — subscription
// inbox R3, reply availability R1(vi), reply-deadline expiry) is a bare
// wake with no payload. It carries no lifecycle decision: the sole action
// is that ONE full slice runs, which the caller (`handle`) does exactly
// once after draining the whole mailbox — so N coalesced markers, or a
// duplicate marker, collapse to one slice and never double-apply work.
// Recognised explicitly (rather than falling through) so the discipline is
// legible and the park-flip inherits it unchanged. Returning `None` lets
// the drain continue and the single post-drain slice service every source.
if message.as_atom() == Some(self.runtime.ready_atom()) {
return None;
}
None
}
}