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
use super::{
AtomicBool, AtomicU64, Duration, Mutex, Ordering, PoisonError, 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.
let admitted = {
let mut slot = active_stream.lock().unwrap_or_else(PoisonError::into_inner);
if shutdown.load(Ordering::SeqCst) {
false
} else {
// 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.
*active_stream.lock().unwrap_or_else(PoisonError::into_inner) = 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}"),
})
}