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
//! The connection process's close and fate paths: a lost or failed
//! connection's `ConnectionLost` fate, a fate's close, and a normal close.
use super::{ConnectionTransport, SliceStep, TransportConnectionProcess};
use crate::ServerError;
use crate::server::participant::ConnectionFateClass;
use beamr::native::native_process::NativeOutcome;
use beamr::process::ExitReason;
impl<T: ConnectionTransport> TransportConnectionProcess<T> {
/// Tears the connection down on a fatal outbound condition: the
/// conversations are released, the crash is recorded, the process stops.
pub(super) fn tear_down_outbound(
&mut self,
pid: u64,
error: &dyn std::fmt::Display,
message: &str,
) -> NativeOutcome {
tracing::warn!(connection_pid = pid, %error, "{message}");
self.release_conversations();
self.runtime
.mark_crashed(pid, ExitReason::Error, self.peer_addr);
NativeOutcome::Stop(ExitReason::Error)
}
pub(super) fn fail_slice(&mut self, pid: u64, error: &ServerError) -> NativeOutcome {
// SRV-043 R2: the server's stop closing a connection before its
// readiness registered is no fault; the teardown is the same.
if matches!(error, ServerError::ReadinessClosedByStop { .. }) {
tracing::debug!(
connection_pid = pid,
"connection closed by the server's stop before its readiness registered"
);
} else {
tracing::error!(connection_pid = pid, %error, "connection readiness contract failed");
}
self.tear_down_lost(pid)
}
/// Tears down a connection the server can no longer serve, as a read
/// failure does: the conversations it tracks get their `ConnectionLost`
/// fate first, so no participant stays bound to a dead connection, then
/// they are released and the connection is marked crashed. A fate that
/// cannot complete takes the fatal fate path instead.
pub(super) fn tear_down_lost(&mut self, pid: u64) -> NativeOutcome {
if let Err(error) = self.complete_connection_fate(ConnectionFateClass::ConnectionLost) {
return match self.fail_fate(pid, &error) {
SliceStep::Stop(reason) => NativeOutcome::Stop(reason),
SliceStep::Continue => NativeOutcome::Continue,
};
}
self.release_conversations();
self.runtime
.mark_crashed(pid, ExitReason::Error, self.peer_addr);
NativeOutcome::Stop(ExitReason::Error)
}
/// Releases every conversation this connection opened, finalizing each so its
/// supervised actor and participant are terminated and their runtime
/// registrations dropped — an abrupt connection teardown never leaks them.
/// Finalization is bounded, non-blocking, and does not require the
/// conversation scheduler to run a slice (or even be live): teardown runs on
/// a connection scheduler worker and inside `Drop`, where waiting on another
/// scheduler would wedge the worker or hang reap/shutdown. The interactive
/// close round trip stays exclusively on the client-requested
/// `ConversationClose` frame path. Every in-handler termination path (EOF,
/// client `Close`, `ForceClose`, and each crash route) calls this before the
/// runtime teardown; the `Drop` backstop covers the external-termination/reap
/// and scheduler-shutdown paths that never run another slice. The
/// conversation map is drained, so a second call finds nothing — the explicit
/// paths and the backstop cannot double-finalize.
pub(super) fn release_conversations(&mut self) {
self.state.relay.take();
self.release_participant_publication();
// R1(vi)/§1.2(5): cancel every pending-reply entry BEFORE the conversation
// actors are torn down, so no entry (and no timeout write) outlives its
// connection. Finalizing each conversation below clears its reply notifier
// in the conversation core, so no marker fires after teardown.
self.state.pending_replies.cancel_all();
for (_conversation_id, conversation) in std::mem::take(&mut self.state.conversations) {
conversation.finalize();
}
}
/// Finishes a client-initiated or terminal-rejection close.
pub(super) fn complete_connection_fate(
&self,
class: ConnectionFateClass,
) -> Result<(), ServerError> {
let conversations = self.state.participant_conversations.tracked_conversations();
self.runtime.complete_connection_fate(
self.state.connection_incarnation,
class,
&conversations,
)
}
pub(super) fn fail_fate(&mut self, pid: u64, error: &ServerError) -> SliceStep {
tracing::error!(connection_pid = pid, %error, "connection fate fold failed");
self.release_conversations();
self.runtime
.mark_crashed(pid, ExitReason::Error, self.peer_addr);
SliceStep::Stop(ExitReason::Error)
}
pub(super) fn finish_fate_close(&mut self, pid: u64, class: ConnectionFateClass) -> SliceStep {
crate::agents::runtime::connection_fate_closed(self.runtime.services(), pid, class);
// One unbudgeted, nonblocking drain; the outcome is deliberately not
// consulted. What a `WouldBlock` loses is the unwritten tail of the
// terminal response — the client may read a truncated final frame. The
// close decision is already taken by the time we are here, so there is
// nothing to reconsider, and this path may not sleep, poll or retry to
// widen the attempt. The loss window closes with the peer's own
// half-close.
if let Err(error) = self.drain_outbound_for_close() {
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(class) {
return self.fail_fate(pid, &error);
}
self.release_conversations();
self.runtime.finish(pid);
SliceStep::Stop(ExitReason::Normal)
}
pub(super) fn finish_normal_close(&mut self, pid: u64) -> SliceStep {
crate::agents::runtime::connection_closed(self.runtime.services(), pid, "client_close");
// Multiple responses may already precede the terminal frame. Make one
// unbudgeted, nonblocking drain so the ordinary 8 KiB slice budget cannot
// deterministically truncate it. `WouldBlock` remains best-effort: never
// sleep, poll, or retry. What that forfeits is the residue after the
// socket buffer fills — the client may miss the tail of its own close
// response. Accepted: the close is already decided, and the loss window
// is the peer's half-close, not something this process can outwait.
if let Err(error) = self.drain_outbound_for_close() {
tracing::debug!(connection_pid = pid, %error, "the close path's best-effort drain did not finish; the peer's half-close ends it");
}
self.release_conversations();
self.runtime.finish(pid);
SliceStep::Stop(ExitReason::Normal)
}
}