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
use super::{
AtomicBool, AtomicU64, Duration, Mutex, Ordering, ServedState, ServerError, TcpListener,
TcpStream, Write, acquire_reserve_descriptor, prepare_accepted_socket, response_for_request,
shed_on_fd_exhaustion,
};
pub(super) fn serve(
listener: &TcpListener,
served: &ServedState,
shutdown: &AtomicBool,
active_stream: &Mutex<Option<TcpStream>>,
accept_attempts: &AtomicU64,
shed_count: &AtomicU64,
requests_entered: &AtomicU64,
) -> Result<(), ServerError> {
// One reserve descriptor held for the shed-with-spare-fd EMFILE policy,
// reusing the leg-1 helper.
let mut reserve = acquire_reserve_descriptor(listener, "health endpoint listener");
while !shutdown.load(Ordering::SeqCst) {
accept_attempts.fetch_add(1, Ordering::SeqCst);
match listener.accept() {
Ok((stream, ..)) => {
// Register the in-flight stream and re-check shutdown atomically
// under the slot lock. If shutdown already fired, shed without
// serving (no request slips past the broadcast); otherwise the
// registered clone lets `stop_worker` interrupt this request's
// blocking read directly (TOLD). Pairing the flag load here with
// `stop_worker`'s flag-store-then-lock means a registration can
// never be missed: either the worker sees the flag and sheds, or
// shutdown finds the clone in the slot.
// A poisoned slot (a panic while it was held) is not trusted:
// the request is shed with a warning rather than served unseen
// by `stop_worker`.
let admitted = match active_stream.lock() {
Err(_) => {
tracing::warn!("health endpoint stream slot is poisoned; request shed");
false
}
Ok(_) if shutdown.load(Ordering::SeqCst) => false,
Ok(mut slot) => {
// A failed clone leaves the slot empty while the request
// is still served: `stop_worker` then finds nothing to
// interrupt, so shutdown must wait out this request's
// read timeout instead of cutting it short.
*slot = match stream.try_clone() {
Ok(clone) => Some(clone),
Err(error) => {
tracing::warn!(
%error,
"health endpoint request stream could not be retained; \
shutdown cannot interrupt this request"
);
None
}
};
true
}
};
if !admitted {
drop(stream);
continue;
}
// A `/metrics` body runs to many KB, so the response spans several
// segments and Nagle would hold its trailing partial segment for the
// scraper's delayed ACK. Same preparation as the leg-1 accept loops.
prepare_accepted_socket(&stream, None);
requests_entered.fetch_add(1, Ordering::SeqCst);
// A per-connection error (e.g. a TCP probe that connects but sends no HTTP
// data within the read timeout) must NOT terminate the serve loop — otherwise
// a single port probe kills the health server for the process lifetime and
// subsequent liveness/readiness probes get connection-refused. Only fatal
// listener-level accept errors (below) terminate serving.
let result = handle_connection(stream, served);
// The request is done: clear the slot so a later shutdown finds
// nothing to interrupt (and never shuts down an unrelated stream).
// A no-op if `stop_worker` already took the clone.
if let Ok(mut slot) = active_stream.lock() {
*slot = None;
}
if let Err(error) = result {
tracing::debug!(%error, "health endpoint connection error");
}
}
Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
Err(error) if is_transient_accept_error(&error) => {
shed_on_fd_exhaustion(listener, &mut reserve, shed_count, &error);
}
Err(error) => {
return Err(ServerError::HealthEndpoint {
message: format!("health endpoint accept failed: {error}"),
});
}
}
}
Ok(())
}
/// EMFILE/ENFILE resource exhaustion is transient, exactly as the leg-1 accept
/// loops treat it (mirrors the sibling WebSocket listener's local predicate).
pub(super) fn is_transient_accept_error(error: &std::io::Error) -> bool {
matches!(error.raw_os_error(), Some(code) if code == 24 || code == 23)
}
pub(super) fn handle_connection(
mut stream: TcpStream,
served: &ServedState,
) -> Result<(), ServerError> {
stream
.set_nonblocking(false)
.map_err(|error| ServerError::HealthEndpoint {
message: format!("failed to configure health request stream: {error}"),
})?;
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.map_err(|error| ServerError::HealthEndpoint {
message: format!("failed to set health request read timeout: {error}"),
})?;
let request = super::admin::read(&mut stream, served)?;
if request.is_empty() {
return Ok(());
}
let response = response_for_request(&request, served)?;
stream
.write_all(&response)
.map_err(|error| ServerError::HealthEndpoint {
message: format!("failed to write health response: {error}"),
})?;
stream.flush().map_err(|error| ServerError::HealthEndpoint {
message: format!("failed to flush health response: {error}"),
})
}