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
use std::time::{Duration, Instant};
use crate::ServerError;
use crate::server::connection::ConnectionSupervisor;
use super::{WEDGED_CONNECTION_STOP, flush_durable_state, wait_after_force_close};
pub(in crate::server) fn run(
supervisor: &ConnectionSupervisor,
drain_timeout: Duration,
stop_accepting: impl FnOnce() -> Result<(), ServerError>,
finalize: impl FnOnce() -> Result<(), ServerError>,
) -> Result<(), ServerError> {
let started = Instant::now();
tracing::info!(
ignored_drain_timeout = ?drain_timeout,
"starting shutdown sequence; the configured drain timeout has been ignored since 0.14.3 \
because every write is durable before it is acknowledged, so no request is in flight to \
drain"
);
// Stop accepting new connections first so none can slip into the accept
// window after shutdown begins and miss the notification broadcast below.
stop_accepting()?;
// FIX A-ii: flush accepted-but-unfanned-out publishes to their subscriber
// connections BEFORE broadcasting the shutdown Disconnect. Accept is now
// stopped, so the set of accepted publishes is bounded; this TOLD barrier
// parks on the delivery-quiescence signal (a connection parks only once every
// accepted publish has been pumped to its socket) and returns the instant it
// arrives. Without it, `notify_shutdown_subscribers` below could enqueue a
// subscriber's Disconnect ahead of an in-flight fan-out (measured 8-131 ms)
// and the subscriber's reader would exit before delivery. The bound is the
// hang stop, not a budget anything waits out; missing it is logged, not
// fatal — the close and flush legs below still run.
if !supervisor.wait_for_delivery_quiesced(Instant::now() + WEDGED_CONNECTION_STOP) {
tracing::warn!(
stop = ?WEDGED_CONNECTION_STOP,
"delivery flush barrier completion was not observed before the wedged-connection stop; \
proceeding to shutdown notification"
);
}
supervisor.notify_shutdown_subscribers();
// The close is the mechanism, and it is now the only route. Nothing waits
// for a peer to hang up first.
supervisor.force_close_active_connections();
wait_after_force_close(supervisor);
// A connection fate refused by candidate-lane precedence during the close
// is parked, and nothing will publish to drain its lane now. Settle it
// here, as the next boot otherwise would, so a clean stop leaves nothing
// to reconcile; a fate that cannot be settled is named and the stop is
// reported unclean once the flush and teardown below have still run.
let settled = supervisor.settle_parked_connection_fates();
supervisor.shutdown();
let quiesced = supervisor.finish_store_work();
let finalized = finalize();
supervisor.stop_participant_warm();
let flushed = flush_durable_state(supervisor);
let released = supervisor.release_store_holders();
tracing::info!(
elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
clean = settled.is_ok()
&& quiesced.is_ok()
&& finalized.is_ok()
&& flushed.is_ok()
&& released.is_ok(),
"shutdown sequence complete"
);
let mut failure = None;
for result in [finalized, flushed, settled, quiesced, released] {
if let Err(error) = result {
tracing::error!(%error, "shutdown phase failed");
if failure.is_none() {
failure = Some(error);
}
}
}
failure.map_or(Ok(()), Err)
}