1use std::{
2 collections::{HashMap, VecDeque},
3 error::Error,
4 fmt, io,
5 path::PathBuf,
6 process::{ExitStatus, Stdio},
7 sync::{Arc, Mutex, OnceLock},
8 time::{Duration, SystemTime, UNIX_EPOCH},
9};
10
11use cortexkit_log::Retention;
12use serde_json::Value;
13use subc_control::{
14 ClientControlPush, LiveSpawn, ModuleProtocol, RouteCloseReason, SpawnCursor, SpawnEvent,
15 SpawnEventKind, SpawnSnapshot, SupervisorHealthStatus, TerminalDisposition, TerminalExitKind,
16};
17use subc_protocol::{
18 manifest::{SelfSignalKind, SignalAnchor},
19 session::{
20 HealthReport, HealthStatus, ModuleControlCommand, ModuleControlRequest,
21 MODULE_CONTROL_OP_HEALTH_CHECK,
22 },
23 Flags, FrameType, Priority, SUBC_LAUNCH_NONCE_ENV, SUBC_MODULE_ID_ENV,
24};
25use tokio::{
26 process::{Child, Command},
27 sync::{mpsc, oneshot, watch, Mutex as AsyncMutex},
28 task::JoinHandle,
29 time::{sleep, sleep_until, timeout, timeout_at, Instant},
30};
31use tracing::{debug, error, info, warn};
32
33use crate::{
34 child_roster::ChildRoster,
35 daemon_config::{
36 CAPTURE_KEEP_ENV, CAPTURE_MAX_AGE_DAYS_ENV, CAPTURE_MAX_FILE_MB_ENV, CK_LOG_ENV,
37 },
38 forwarding::{
39 CloseReason, ForwardingError, ForwardingTable, GoodbyeTarget, ModuleControlRpcOutcome,
40 ModuleDrainTarget, PendingModuleControlRpc,
41 },
42 provenance::{spawned_file_identity, ExecutableIdentityProbe, SpawnedFileIdentity},
43 registry::{ConnectionId, RegistryError},
44 stderr_tail::{
45 pump_stderr_to, pump_stdout_to, ChildOutputSink, StderrRing, StderrTailConfig,
46 StderrTailSnapshot,
47 },
48 terminal_ring::{TerminalHistorySnapshot, TerminalRecord, TerminalRing, TerminalRingConfig},
49 Frame, FrameSink, Registry,
50};
51
52#[path = "supervise_swap.rs"]
53mod swap;
54
55pub const SUBC_ARG: &str = "--subc";
61
62const DEFAULT_MAX_RESTARTS: u32 = 3;
63const DEFAULT_BACKOFF: Duration = Duration::from_millis(100);
64const DEFAULT_MAX_BACKOFF: Duration = Duration::from_secs(30);
65const DEFAULT_RESTART_WINDOW: Duration = Duration::from_secs(600);
69pub const DEFAULT_DRAIN_TIMEOUT: Duration = Duration::from_secs(30);
80const REGISTRY_RELEASE_TIMEOUT: Duration = Duration::from_secs(1);
81const REGISTRY_RELEASE_POLL: Duration = Duration::from_millis(10);
82const STDERR_PUMP_DRAIN_TIMEOUT: Duration = Duration::from_millis(250);
104pub const SPAWN_EVENT_RING_CAPACITY: usize = 4096;
106const SPAWN_SUBSCRIBER_BUFFER: usize = SPAWN_EVENT_RING_CAPACITY + 1;
107pub(crate) const SPAWN_SUBSCRIBER_LAGGED_CODE: &str = "spawn_subscriber_lagged";
112
113struct SupervisedChild {
114 child: Child,
115 #[cfg(target_os = "linux")]
118 module_id: String,
119 #[cfg(target_os = "linux")]
120 cgroup_placement: Option<subc_cgroup::Placement>,
121 stdout_pump: Option<JoinHandle<()>>,
122 stderr_pump: Option<StderrPump>,
123 stderr_ring: Arc<Mutex<StderrRing>>,
124 spawned_at_ms: u64,
125 spawned_from: PathBuf,
126 spawned_file_identity: Option<SpawnedFileIdentity>,
127 process_start_time: Option<u64>,
128 process_identity: Option<ProcessIdentity>,
129 pid: u32,
130 roster_guard: Option<crate::child_roster::RosterGuard>,
133}
134
135impl SupervisedChild {
136 fn id(&self) -> Option<u32> {
137 Some(self.pid)
138 }
139
140 fn process_identity(&self) -> Option<ProcessIdentity> {
141 self.process_identity
142 }
143
144 async fn wait(&mut self) -> io::Result<ExitStatus> {
145 let result = self.child.wait().await;
153 #[cfg(target_os = "linux")]
154 if result.is_ok() {
155 if let Some(placement) = self.cgroup_placement.take() {
156 remove_module_cgroup(&placement, &self.module_id);
157 }
158 }
159 result
160 }
161
162 fn release_roster(&mut self) {
166 self.roster_guard = None;
167 }
168
169 fn start_kill(&mut self) -> io::Result<()> {
170 self.child.start_kill()
171 }
172
173 async fn drain_stderr(&mut self, module_id: &str) {
174 if let Some(mut pump) = self.stdout_pump.take() {
175 match timeout(STDERR_PUMP_DRAIN_TIMEOUT, &mut pump).await {
176 Ok(Ok(())) => {}
177 Ok(Err(error)) => {
178 warn!(module_id, error = %error, "stdout pump ended unexpectedly");
179 }
180 Err(_) => {
181 pump.abort();
182 warn!(
183 module_id,
184 waited = ?STDERR_PUMP_DRAIN_TIMEOUT,
185 "stdout pump did not drain before restart; stopped it before the next process"
186 );
187 }
188 }
189 }
190
191 let Some(pump) = self.stderr_pump.take() else {
192 return;
193 };
194 settle_stderr_pump(
195 module_id,
196 &self.stderr_ring,
197 pump,
198 STDERR_PUMP_DRAIN_TIMEOUT,
199 )
200 .await;
201 }
202}
203
204struct StderrPump {
207 task: JoinHandle<()>,
208 generation: u64,
209}
210
211async fn settle_stderr_pump(
217 module_id: &str,
218 ring: &Arc<Mutex<StderrRing>>,
219 pump: StderrPump,
220 bound: Duration,
221) {
222 let lock = || ring.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
223 let StderrPump {
224 mut task,
225 generation,
226 } = pump;
227 lock().retire_pump(generation);
228 match timeout(bound, &mut task).await {
229 Ok(Ok(())) => {}
230 Ok(Err(err)) => {
231 let mut ring = lock();
232 ring.mark_incomplete(format!("stderr pump ended unexpectedly: {err}"));
233 ring.finish_pump(generation);
234 warn!(module_id, error = %err, "stderr pump ended before clean EOF");
235 }
236 Err(_) => {
237 drop(task);
239 lock().mark_pump_late(
240 generation,
241 format!(
242 "stderr of the exited process had not reached EOF {bound:?} after it was \
243 retired (a descendant may still hold the pipe open); lines it still \
244 writes are kept in that process's section"
245 ),
246 );
247 warn!(
248 module_id,
249 waited = ?bound,
250 "stderr pipe of the exited process is still open; its reader keeps running without delaying the restart"
251 );
252 }
253 }
254}
255
256fn registration_release_events() -> &'static watch::Sender<u64> {
257 static EVENTS: OnceLock<watch::Sender<u64>> = OnceLock::new();
258 EVENTS.get_or_init(|| {
259 let (sender, _receiver) = watch::channel(0);
260 sender
261 })
262}
263
264pub(crate) fn notify_registration_release() {
265 let events = registration_release_events();
266 let next_generation = (*events.borrow()).wrapping_add(1);
267 events.send_replace(next_generation);
268}
269
270#[derive(Debug, Clone, PartialEq, Eq)]
272pub struct ModuleSpec {
273 pub module_id: String,
274 pub program: PathBuf,
275 pub args: Vec<String>,
276 pub env: Vec<(String, String)>,
277 pub reserved: bool,
282 pub reserved_prefixes: Vec<String>,
287 pub protocol: ModuleProtocol,
303 pub overlap: ModuleOverlap,
308}
309
310#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
317pub enum ModuleOverlap {
318 #[default]
320 Exclusive,
321 Safe,
333}
334
335impl ModuleOverlap {
336 pub fn as_str(self) -> &'static str {
337 match self {
338 Self::Exclusive => "exclusive",
339 Self::Safe => "safe",
340 }
341 }
342}
343
344pub const SUBC_SPAWN_ROLE_ENV: &str = "SUBC_SPAWN_ROLE";
354pub const SPAWN_ROLE_SWAP_CANDIDATE: &str = "swap_candidate";
356pub const DEFAULT_SWAP_READY_TIMEOUT: Duration = Duration::from_secs(100);
361
362#[derive(Debug, Clone, Copy, PartialEq, Eq)]
380pub struct RestartPolicy {
381 pub max_restarts: u32,
382 pub backoff: Duration,
385 pub max_backoff: Duration,
387 pub window: Duration,
391}
392
393impl RestartPolicy {
394 pub fn new(max_restarts: u32, backoff: Duration) -> Self {
398 Self {
399 max_restarts,
400 backoff,
401 max_backoff: DEFAULT_MAX_BACKOFF,
402 window: DEFAULT_RESTART_WINDOW,
403 }
404 }
405
406 pub fn with_max_backoff(mut self, max_backoff: Duration) -> Self {
407 self.max_backoff = max_backoff;
408 self
409 }
410
411 pub fn with_window(mut self, window: Duration) -> Self {
412 self.window = window;
413 self
414 }
415
416 fn delay_for_restart(&self, restart_in_window: u32) -> Duration {
421 if self.backoff.is_zero() || self.max_backoff.is_zero() {
422 return Duration::ZERO;
423 }
424
425 let mut delay = self.backoff;
426 for _ in 0..restart_in_window {
427 if delay >= self.max_backoff {
428 return self.max_backoff;
429 }
430 delay = delay
431 .checked_mul(10)
432 .unwrap_or(self.max_backoff)
433 .min(self.max_backoff);
434 }
435 delay.min(self.max_backoff)
436 }
437
438 fn budget_exhausted_detail(&self) -> String {
443 format!(
444 "crash budget exhausted: max_restarts={} within window_secs={}",
445 self.max_restarts,
446 self.window.as_secs()
447 )
448 }
449}
450
451impl Default for RestartPolicy {
452 fn default() -> Self {
453 Self {
454 max_restarts: DEFAULT_MAX_RESTARTS,
455 backoff: DEFAULT_BACKOFF,
456 max_backoff: DEFAULT_MAX_BACKOFF,
457 window: DEFAULT_RESTART_WINDOW,
458 }
459 }
460}
461
462#[derive(Debug, Clone, Copy, PartialEq, Eq)]
463struct CrashRestartSchedule {
464 restart_in_window: u32,
465 delay: Duration,
466}
467
468fn daemon_will_restart(
475 state: &mut SupervisorSnapshot,
476 policy: &RestartPolicy,
477 now: Instant,
478) -> bool {
479 state.enabled && state.crash_restarts_in_window(policy.window, now) < policy.max_restarts
480}
481
482const DEFAULT_HEALTH_CADENCE: Duration = Duration::from_secs(30);
483const DEFAULT_HEALTH_DEADLINE: Duration = Duration::from_secs(5);
484const DEFAULT_HEALTH_FAILURE_THRESHOLD: u32 = 3;
485const MAX_HEALTH_METRICS_BYTES: usize = 16 * 1024;
486
487#[derive(Debug, Clone, Copy, PartialEq, Eq)]
488pub enum HealthAction {
489 Report,
490 Restart,
491 Alert,
492}
493
494impl fmt::Display for HealthAction {
495 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
496 f.write_str(match self {
497 Self::Report => "report",
498 Self::Restart => "restart",
499 Self::Alert => "alert",
500 })
501 }
502}
503
504#[derive(Debug, Clone, Copy, PartialEq, Eq)]
505pub struct HealthConfig {
506 pub cadence: Duration,
507 pub deadline: Duration,
508 pub failure_threshold: u32,
509 pub on_degraded: HealthAction,
510 pub on_failing: HealthAction,
511 pub critical: bool,
512}
513
514impl Default for HealthConfig {
515 fn default() -> Self {
516 Self {
517 cadence: DEFAULT_HEALTH_CADENCE,
518 deadline: DEFAULT_HEALTH_DEADLINE,
519 failure_threshold: DEFAULT_HEALTH_FAILURE_THRESHOLD,
520 on_degraded: HealthAction::Report,
521 on_failing: HealthAction::Report,
522 critical: false,
523 }
524 }
525}
526
527#[derive(Debug, Clone, PartialEq)]
545pub struct ModuleHealthStatus {
546 pub status: SupervisorHealthStatus,
547 pub last_probe_ms: Option<u64>,
548 pub detail: Option<String>,
549 pub metrics: Option<Value>,
550 pub consecutive_failures: u32,
551 pub late_answer_count: u64,
554 pub last_late_answer_latency_ms: Option<u64>,
556 pub last_action: Option<String>,
557 pub last_action_ms: Option<u64>,
561}
562
563impl Default for ModuleHealthStatus {
564 fn default() -> Self {
565 Self {
566 status: SupervisorHealthStatus::Unknown,
567 last_probe_ms: None,
568 detail: None,
569 metrics: None,
570 consecutive_failures: 0,
571 late_answer_count: 0,
572 last_late_answer_latency_ms: None,
573 last_action: None,
574 last_action_ms: None,
575 }
576 }
577}
578
579#[derive(Debug, Clone, Copy, PartialEq, Eq)]
581pub enum ModuleState {
582 Starting,
583 Running,
584 Unresponsive,
585 Restarting,
586 Draining,
587 Stopped,
588 Failed,
589 Disabled,
590}
591
592impl fmt::Display for ModuleState {
593 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
594 f.write_str(match self {
595 Self::Starting => "starting",
596 Self::Running => "running",
597 Self::Unresponsive => "unresponsive",
598 Self::Restarting => "restarting",
599 Self::Draining => "draining",
600 Self::Stopped => "stopped",
601 Self::Failed => "failed",
602 Self::Disabled => "disabled",
603 })
604 }
605}
606
607#[derive(Debug, Clone, Copy, PartialEq, Eq)]
609pub enum ExitKind {
610 Clean,
611 Crash,
612 DeliberateSeverance,
613}
614
615impl From<ExitKind> for TerminalExitKind {
616 fn from(kind: ExitKind) -> Self {
617 match kind {
618 ExitKind::Clean => Self::Clean,
619 ExitKind::Crash => Self::Crash,
620 ExitKind::DeliberateSeverance => Self::DeliberateSeverance,
621 }
622 }
623}
624
625#[derive(Debug, Clone, Copy, PartialEq, Eq)]
628pub(crate) struct ProcessIdentity {
629 pub(crate) pid: u32,
630 pub(crate) start_time: u64,
631}
632
633#[derive(Debug, Clone, PartialEq, Eq)]
635pub struct ExitReport {
636 pub kind: ExitKind,
637 pub code: Option<i32>,
638 pub signal: Option<i32>,
639 pub at_ms: u64,
640}
641
642#[derive(Debug, Clone, PartialEq)]
645pub struct ModuleStatus {
646 pub module_id: String,
647 pub state: ModuleState,
648 pub enabled: bool,
649 pub process_alive: bool,
650 pub registration_active: bool,
651 pub protocol: ModuleProtocol,
654 pub live: bool,
665 pub restart_count: u32,
669 pub lifetime_restarts: u32,
673 pub spawn_generation: u64,
674 pub max_restarts: u32,
679 pub restart_window: Duration,
683 pub drain_timeout: Duration,
687 pub restart_backoff: Duration,
688 pub restart_max_backoff: Duration,
689 pub pid: Option<u32>,
690 pub spawned_at_ms: Option<u64>,
691 pub spawned_from: Option<PathBuf>,
692 pub process_start_time: Option<u64>,
693 pub last_exit: Option<ExitReport>,
694 pub health: ModuleHealthStatus,
695}
696
697#[derive(Debug, Clone, PartialEq)]
698struct SupervisorSnapshot {
699 state: ModuleState,
700 enabled: bool,
701 process_alive: bool,
702 crash_restarts: VecDeque<Instant>,
708 lifetime_restarts: u32,
709 spawn_generation: u64,
718 pid: Option<u32>,
719 spawned_at_ms: Option<u64>,
720 spawned_from: Option<PathBuf>,
721 spawned_file_identity: Option<SpawnedFileIdentity>,
722 process_start_time: Option<u64>,
723 deliberate_severance: Option<ProcessIdentity>,
724 last_exit: Option<ExitReport>,
725 health: ModuleHealthStatus,
726 in_alternate_slot: bool,
731 draining_to_replace: bool,
738 configuration_updated_since_spawn: bool,
744}
745
746impl SupervisorSnapshot {
747 fn starting() -> Self {
748 Self::new(ModuleState::Starting, true)
749 }
750
751 fn disabled() -> Self {
752 Self::new(ModuleState::Disabled, false)
753 }
754
755 fn failed() -> Self {
756 Self::new(ModuleState::Failed, true)
757 }
758
759 fn crash_restarts_in_window(&mut self, window: Duration, now: Instant) -> u32 {
763 while let Some(oldest) = self.crash_restarts.front() {
764 if now.duration_since(*oldest) > window {
765 self.crash_restarts.pop_front();
766 } else {
767 break;
768 }
769 }
770 u32::try_from(self.crash_restarts.len()).unwrap_or(u32::MAX)
771 }
772
773 fn record_crash_restart(&mut self, policy: &RestartPolicy, now: Instant) {
779 self.crash_restarts.push_back(now);
780 while self.crash_restarts.len() > policy.max_restarts as usize {
781 self.crash_restarts.pop_front();
782 }
783 self.lifetime_restarts += 1;
784 }
785
786 fn next_crash_restart(
790 &mut self,
791 policy: &RestartPolicy,
792 now: Instant,
793 ) -> Option<CrashRestartSchedule> {
794 let restart_in_window = self.crash_restarts_in_window(policy.window, now);
795 if restart_in_window >= policy.max_restarts {
796 return None;
797 }
798 self.record_crash_restart(policy, now);
799 Some(CrashRestartSchedule {
800 restart_in_window,
801 delay: policy.delay_for_restart(restart_in_window),
802 })
803 }
804
805 fn clear_crash_restarts(&mut self) {
810 self.crash_restarts.clear();
811 }
812
813 fn new(state: ModuleState, enabled: bool) -> Self {
814 Self {
815 state,
816 enabled,
817 process_alive: false,
818 crash_restarts: VecDeque::new(),
819 lifetime_restarts: 0,
820 spawn_generation: 0,
821 pid: None,
822 spawned_at_ms: None,
823 spawned_from: None,
824 spawned_file_identity: None,
825 process_start_time: None,
826 deliberate_severance: None,
827 last_exit: None,
828 health: ModuleHealthStatus::default(),
829 in_alternate_slot: false,
830 draining_to_replace: false,
831 configuration_updated_since_spawn: false,
832 }
833 }
834}
835
836type SharedSnapshot = Arc<Mutex<SupervisorSnapshot>>;
837
838type SpawnSubscriberKey = (ConnectionId, u64);
839
840#[derive(Debug)]
841struct SpawnSubscriber {
842 version: u8,
843 frames: mpsc::Sender<Frame>,
844 lagged: Option<oneshot::Sender<SpawnCursor>>,
848}
849
850#[derive(Debug)]
851struct SpawnEventState {
852 daemon_incarnation: String,
853 seq: u64,
854 capacity: usize,
855 live: HashMap<String, LiveSpawn>,
856 generations: HashMap<String, u64>,
857 events: VecDeque<SpawnEvent>,
858 subscribers: HashMap<SpawnSubscriberKey, SpawnSubscriber>,
859}
860
861impl Default for SpawnEventState {
862 fn default() -> Self {
863 Self {
864 daemon_incarnation: "unconfigured".to_string(),
865 seq: 0,
866 capacity: SPAWN_EVENT_RING_CAPACITY,
867 live: HashMap::new(),
868 generations: HashMap::new(),
869 events: VecDeque::new(),
870 subscribers: HashMap::new(),
871 }
872 }
873}
874
875#[derive(Debug, Clone, Default)]
876struct SpawnEventFeed(Arc<Mutex<SpawnEventState>>);
877
878#[derive(Debug, Clone, PartialEq, Eq)]
879pub(crate) enum SpawnSubscribeRefusal {
880 ForeignIncarnation { current: String },
881 TooOld { oldest: SpawnCursor },
882 Frame(String),
883}
884
885impl SpawnEventFeed {
886 fn configure_incarnation(&self, daemon_incarnation: String) {
887 let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
888 state.daemon_incarnation = daemon_incarnation;
889 state.seq = 0;
890 state.live.clear();
891 state.generations.clear();
892 state.events.clear();
893 state.subscribers.clear();
894 }
895
896 fn cursor(state: &SpawnEventState) -> SpawnCursor {
897 SpawnCursor {
898 daemon_incarnation: state.daemon_incarnation.clone(),
899 seq: state.seq,
900 }
901 }
902
903 fn snapshot(&self) -> SpawnSnapshot {
904 let state = self.0.lock().unwrap_or_else(|p| p.into_inner());
905 let mut live = state.live.values().cloned().collect::<Vec<_>>();
906 live.sort_by(|left, right| left.module_id.cmp(&right.module_id));
907 SpawnSnapshot {
908 cursor: Self::cursor(&state),
909 ring_bound: state.capacity as u64,
910 live,
911 }
912 }
913
914 fn emit_spawned(&self, module_id: &str, pid: u32, spawned_at_ms: u64) -> u64 {
915 let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
916 let generation = state
917 .generations
918 .get(module_id)
919 .copied()
920 .unwrap_or(0)
921 .checked_add(1)
922 .expect("spawn generation exhausted");
923 state.generations.insert(module_id.to_string(), generation);
924 let live = LiveSpawn {
925 module_id: module_id.to_string(),
926 spawn_generation: generation,
927 pid,
928 spawned_at_ms,
929 };
930 state.live.insert(module_id.to_string(), live);
931 Self::emit_locked(
932 &mut state,
933 SpawnEventKind::Spawned,
934 module_id.to_string(),
935 generation,
936 pid,
937 None,
938 None,
939 );
940 generation
941 }
942
943 fn emit_exited(&self, module_id: &str, exit_code: Option<i32>, exit_signal: Option<i32>) {
944 let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
945 let Some(live) = state.live.remove(module_id) else {
946 warn!(
947 module_id,
948 "terminal record had no live spawn event identity"
949 );
950 return;
951 };
952 Self::emit_locked(
953 &mut state,
954 SpawnEventKind::Exited,
955 module_id.to_string(),
956 live.spawn_generation,
957 live.pid,
958 exit_code,
959 exit_signal,
960 );
961 }
962
963 fn emit_superseded_exited(
970 &self,
971 module_id: &str,
972 spawn_generation: u64,
973 pid: u32,
974 exit_code: Option<i32>,
975 exit_signal: Option<i32>,
976 ) {
977 let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
978 if state
979 .live
980 .get(module_id)
981 .is_some_and(|live| live.spawn_generation == spawn_generation)
982 {
983 state.live.remove(module_id);
984 }
985 Self::emit_locked(
986 &mut state,
987 SpawnEventKind::Exited,
988 module_id.to_string(),
989 spawn_generation,
990 pid,
991 exit_code,
992 exit_signal,
993 );
994 }
995
996 #[allow(clippy::too_many_arguments)]
997 fn emit_locked(
998 state: &mut SpawnEventState,
999 kind: SpawnEventKind,
1000 module_id: String,
1001 spawn_generation: u64,
1002 pid: u32,
1003 exit_code: Option<i32>,
1004 exit_signal: Option<i32>,
1005 ) {
1006 state.seq = state
1007 .seq
1008 .checked_add(1)
1009 .expect("spawn event sequence exhausted");
1010 let event = SpawnEvent {
1011 cursor: Self::cursor(state),
1012 kind,
1013 module_id,
1014 spawn_generation,
1015 pid,
1016 exit_code,
1017 exit_signal,
1018 };
1019 state.events.push_back(event.clone());
1020 while state.events.len() > state.capacity {
1021 state.events.pop_front();
1022 }
1023 let body = match serde_json::to_vec(&event) {
1024 Ok(body) => body,
1025 Err(error) => {
1026 error!(%error, "failed to serialize supervisor spawn event");
1027 return;
1028 }
1029 };
1030 state.subscribers.retain(|(connection_id, corr), subscriber| {
1031 let frame = Frame::build_with_version(
1032 subscriber.version,
1033 FrameType::StreamData,
1034 control_flags(),
1035 0,
1036 0,
1037 *corr,
1038 body.clone(),
1039 );
1040 match frame {
1041 Ok(frame) => {
1042 if subscriber.frames.try_send(frame).is_ok() {
1043 true
1044 } else {
1045 warn!(connection_id = connection_id.get(), corr, "dropping lagged supervisor spawn subscriber");
1046 if let Some(lagged) = subscriber.lagged.take() {
1047 let _ = lagged.send(event.cursor.clone());
1048 }
1049 false
1050 }
1051 }
1052 Err(error) => {
1053 warn!(connection_id = connection_id.get(), corr, %error, "dropping supervisor spawn subscriber after frame build failure");
1054 false
1055 }
1056 }
1057 });
1058 }
1059
1060 fn subscribe(
1061 &self,
1062 connection_id: ConnectionId,
1063 corr: u64,
1064 version: u8,
1065 since: Option<SpawnCursor>,
1066 sink: FrameSink,
1067 ) -> Result<(), SpawnSubscribeRefusal> {
1068 let (frames, mut receiver) = mpsc::channel(SPAWN_SUBSCRIBER_BUFFER);
1069 let (lagged, mut lagged_rx) = oneshot::channel::<SpawnCursor>();
1070 {
1071 let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
1072 let replay = if let Some(since) = since {
1073 if since.daemon_incarnation != state.daemon_incarnation {
1074 return Err(SpawnSubscribeRefusal::ForeignIncarnation {
1075 current: state.daemon_incarnation.clone(),
1076 });
1077 }
1078 if let Some(oldest) = state.events.front().map(|event| event.cursor.clone()) {
1079 if since.seq < oldest.seq.saturating_sub(1) {
1080 return Err(SpawnSubscribeRefusal::TooOld { oldest });
1081 }
1082 }
1083 state
1084 .events
1085 .iter()
1086 .filter(|event| event.cursor.seq > since.seq)
1087 .cloned()
1088 .collect::<Vec<_>>()
1089 } else {
1090 Vec::new()
1091 };
1092 for event in replay {
1093 let body = serde_json::to_vec(&event)
1094 .map_err(|error| SpawnSubscribeRefusal::Frame(error.to_string()))?;
1095 let frame = Frame::build_with_version(
1096 version,
1097 FrameType::StreamData,
1098 control_flags(),
1099 0,
1100 0,
1101 corr,
1102 body,
1103 )
1104 .map_err(|error| SpawnSubscribeRefusal::Frame(error.to_string()))?;
1105 frames
1106 .try_send(frame)
1107 .map_err(|error| SpawnSubscribeRefusal::Frame(error.to_string()))?;
1108 }
1109 state.subscribers.insert(
1110 (connection_id, corr),
1111 SpawnSubscriber {
1112 version,
1113 frames,
1114 lagged: Some(lagged),
1115 },
1116 );
1117 }
1118 tokio::spawn(async move {
1129 while let Some(frame) = receiver.recv().await {
1130 if sink.send(frame).await.is_err() {
1131 return;
1132 }
1133 }
1134 let Ok(first_undelivered) = lagged_rx.try_recv() else {
1135 return;
1136 };
1137 match spawn_subscriber_lagged_frame(version, corr, first_undelivered) {
1138 Ok(frame) => {
1139 let _ = sink.send(frame).await;
1140 }
1141 Err(error) => {
1142 error!(%error, corr, "failed to build lagged spawn subscriber terminal frame");
1143 }
1144 }
1145 });
1146 Ok(())
1147 }
1148
1149 fn cancel(&self, connection_id: ConnectionId, corr: u64) -> bool {
1150 let Some(subscriber) = self
1151 .0
1152 .lock()
1153 .unwrap_or_else(|p| p.into_inner())
1154 .subscribers
1155 .remove(&(connection_id, corr))
1156 else {
1157 return false;
1158 };
1159 if let Ok(frame) = Frame::build_with_version(
1160 subscriber.version,
1161 FrameType::StreamEnd,
1162 control_flags(),
1163 0,
1164 0,
1165 corr,
1166 Vec::new(),
1167 ) {
1168 tokio::spawn(async move {
1169 let _ = subscriber.frames.send(frame).await;
1170 });
1171 }
1172 true
1173 }
1174
1175 fn remove_connection(&self, connection_id: ConnectionId) {
1176 self.0
1177 .lock()
1178 .unwrap_or_else(|p| p.into_inner())
1179 .subscribers
1180 .retain(|(subscriber_connection, _), _| *subscriber_connection != connection_id);
1181 }
1182
1183 #[cfg(any(test, feature = "test-support"))]
1184 fn set_capacity(&self, capacity: usize) {
1185 self.0.lock().unwrap_or_else(|p| p.into_inner()).capacity = capacity;
1186 }
1187
1188 #[cfg(any(test, feature = "test-support"))]
1189 fn subscriber_count(&self) -> usize {
1190 self.0
1191 .lock()
1192 .unwrap_or_else(|p| p.into_inner())
1193 .subscribers
1194 .len()
1195 }
1196}
1197
1198fn spawn_subscriber_lagged_frame(
1201 version: u8,
1202 corr: u64,
1203 first_undelivered: SpawnCursor,
1204) -> Result<Frame, String> {
1205 let body = serde_json::to_vec(&subc_protocol::ErrorBody {
1206 code: SPAWN_SUBSCRIBER_LAGGED_CODE.to_string(),
1207 message: "spawn subscriber fell behind and was dropped; resubscribe from the last cursor received"
1208 .to_string(),
1209 detail: Some(serde_json::json!({
1210 "first_undelivered_cursor": first_undelivered
1211 })),
1212 })
1213 .map_err(|error| error.to_string())?;
1214 Frame::build_with_version(version, FrameType::Error, control_flags(), 0, 0, corr, body)
1215 .map_err(|error| error.to_string())
1216}
1217
1218pub trait ModuleProcessLiveness: Send + Sync {
1219 fn process_live(&self, module_id: &str) -> Option<bool>;
1220
1221 fn process_replacing(&self, _module_id: &str) -> bool {
1227 false
1228 }
1229}
1230
1231#[derive(Debug, Clone, Default)]
1233pub struct SupervisorProcessLiveness {
1234 snapshots: Arc<Mutex<HashMap<String, SharedSnapshot>>>,
1235}
1236
1237impl SupervisorProcessLiveness {
1238 pub fn new() -> Self {
1239 Self::default()
1240 }
1241
1242 fn track(&self, module_id: String, snapshot: SharedSnapshot) {
1243 let mut snapshots = self
1244 .snapshots
1245 .lock()
1246 .unwrap_or_else(|poisoned| poisoned.into_inner());
1247 snapshots.insert(module_id, snapshot);
1248 }
1249
1250 fn untrack_if_current(&self, module_id: &str, snapshot: &SharedSnapshot) {
1251 let mut snapshots = self
1252 .snapshots
1253 .lock()
1254 .unwrap_or_else(|poisoned| poisoned.into_inner());
1255 let is_current = snapshots
1256 .get(module_id)
1257 .map(|tracked| Arc::ptr_eq(tracked, snapshot))
1258 .unwrap_or(false);
1259 if is_current {
1260 snapshots.remove(module_id);
1261 }
1262 }
1263}
1264
1265impl ModuleProcessLiveness for SupervisorProcessLiveness {
1266 fn process_live(&self, module_id: &str) -> Option<bool> {
1267 let snapshot = {
1268 let snapshots = self
1269 .snapshots
1270 .lock()
1271 .unwrap_or_else(|poisoned| poisoned.into_inner());
1272 snapshots.get(module_id).cloned()
1273 }?;
1274 let snapshot = snapshot
1275 .lock()
1276 .unwrap_or_else(|poisoned| poisoned.into_inner());
1277 Some(snapshot.state == ModuleState::Running && snapshot.process_alive)
1278 }
1279
1280 fn process_replacing(&self, module_id: &str) -> bool {
1281 let Some(snapshot) = self
1282 .snapshots
1283 .lock()
1284 .unwrap_or_else(|poisoned| poisoned.into_inner())
1285 .get(module_id)
1286 .cloned()
1287 else {
1288 return false;
1289 };
1290 let snapshot = snapshot
1291 .lock()
1292 .unwrap_or_else(|poisoned| poisoned.into_inner());
1293 snapshot.enabled
1294 && match snapshot.state {
1295 ModuleState::Restarting => true,
1296 ModuleState::Draining => snapshot.draining_to_replace,
1297 ModuleState::Starting
1298 | ModuleState::Running
1299 | ModuleState::Unresponsive
1300 | ModuleState::Stopped
1301 | ModuleState::Failed
1302 | ModuleState::Disabled => false,
1303 }
1304 }
1305}
1306
1307#[derive(Debug, Clone)]
1308struct SupervisorRuntimeConfig {
1309 restart_policy: RestartPolicy,
1310 drain_timeout: Duration,
1313 effective_drain_timeout: Arc<Mutex<Duration>>,
1316 default_drain_timeout: Duration,
1319 health: HealthConfig,
1320 connection_file_path: Option<PathBuf>,
1321 capture_logs_dir: Option<PathBuf>,
1322 forwarding: Option<Arc<ForwardingTable>>,
1323 supervisor_handle: Option<SupervisorHandle>,
1326 stderr_ring: Arc<Mutex<StderrRing>>,
1333 terminal_ring: Arc<Mutex<TerminalRing>>,
1334 spawn_events: SpawnEventFeed,
1335 child_roster: ChildRoster,
1336 #[cfg(target_os = "linux")]
1337 cgroup_placement: Option<subc_cgroup::Placement>,
1338 #[cfg(test)]
1339 test_seed_stale_facts_before_enable_spawn: bool,
1340}
1341
1342#[derive(Debug, Clone, PartialEq, Eq)]
1343struct SupervisedConfiguration {
1344 spec: ModuleSpec,
1345 health: HealthConfig,
1346}
1347
1348#[derive(Debug, Clone, Default)]
1354pub struct SupervisorHandle {
1355 modules: Arc<Mutex<HashMap<String, SupervisedModule>>>,
1356 spawn_events: SpawnEventFeed,
1357 reserved_nonces: Arc<Mutex<HashMap<String, Option<String>>>>,
1368 removal_tombstones: Arc<Mutex<HashMap<String, u64>>>,
1374 spawn_nonces: Arc<Mutex<HashMap<String, String>>>,
1378 reserved_prefix_owners: Arc<Mutex<HashMap<String, String>>>,
1386 swaps: Arc<Mutex<HashMap<String, OpenSwap>>>,
1392 promotion_observer: PromotionObserverSlot,
1394 operation_lock: Arc<AsyncMutex<()>>,
1398}
1399
1400pub(crate) trait SwapPromotionObserver: Send + Sync {
1409 fn swap_promoted(&self, registration: &crate::registry::ModuleRegistration);
1410}
1411
1412#[derive(Clone, Default)]
1416struct PromotionObserverSlot(Arc<Mutex<Option<std::sync::Weak<dyn SwapPromotionObserver>>>>);
1417
1418impl fmt::Debug for PromotionObserverSlot {
1419 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1420 f.write_str("PromotionObserverSlot")
1421 }
1422}
1423
1424#[derive(Debug, Clone)]
1426struct OpenSwap {
1427 candidate_nonce: String,
1430 incumbent_nonce: Option<String>,
1435 candidate_admitted: bool,
1439}
1440
1441#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1444pub(crate) enum SwapHelloAdmission {
1445 NotSwapping,
1448 Candidate,
1450 Refused,
1453}
1454
1455#[derive(Debug, Clone, PartialEq, Eq)]
1456pub(crate) enum ReservedHelloRejection {
1457 Exact {
1458 module_id: String,
1459 },
1460 Prefix {
1461 prefix: String,
1462 owner_module_id: String,
1463 },
1464}
1465
1466impl SupervisorHandle {
1467 pub fn new() -> Self {
1468 Self::default()
1469 }
1470
1471 pub(crate) fn spawn_snapshot(&self) -> SpawnSnapshot {
1472 self.spawn_events.snapshot()
1473 }
1474
1475 pub(crate) fn subscribe_spawns(
1476 &self,
1477 connection_id: ConnectionId,
1478 corr: u64,
1479 version: u8,
1480 since: Option<SpawnCursor>,
1481 sink: FrameSink,
1482 ) -> Result<(), SpawnSubscribeRefusal> {
1483 self.spawn_events
1484 .subscribe(connection_id, corr, version, since, sink)
1485 }
1486
1487 pub(crate) fn cancel_spawn_subscription(&self, connection_id: ConnectionId, corr: u64) -> bool {
1488 self.spawn_events.cancel(connection_id, corr)
1489 }
1490
1491 pub(crate) fn remove_spawn_subscribers(&self, connection_id: ConnectionId) {
1492 self.spawn_events.remove_connection(connection_id);
1493 }
1494
1495 #[cfg(any(test, feature = "test-support"))]
1496 pub fn set_spawn_event_capacity_for_test(&self, capacity: usize) {
1497 assert!(capacity > 0, "spawn event capacity must be non-zero");
1498 self.spawn_events.set_capacity(capacity);
1499 }
1500
1501 #[cfg(any(test, feature = "test-support"))]
1502 pub fn spawn_subscriber_count_for_test(&self) -> usize {
1503 self.spawn_events.subscriber_count()
1504 }
1505
1506 pub fn set_spawn_nonce(&self, module_id: &str, nonce: String) {
1509 self.spawn_nonces
1510 .lock()
1511 .unwrap_or_else(|poisoned| poisoned.into_inner())
1512 .insert(module_id.to_string(), nonce);
1513 }
1514
1515 pub fn set_reserved_nonce(&self, module_id: &str, nonce: String) {
1518 self.reserved_nonces
1519 .lock()
1520 .unwrap_or_else(|poisoned| poisoned.into_inner())
1521 .insert(module_id.to_string(), Some(nonce));
1522 }
1523
1524 pub fn set_reserved_prefixes(&self, owner_module_id: &str, prefixes: &[String]) {
1526 let mut owners = self
1527 .reserved_prefix_owners
1528 .lock()
1529 .unwrap_or_else(|poisoned| poisoned.into_inner());
1530 owners.retain(|_, owner| owner != owner_module_id);
1531 for prefix in prefixes {
1532 owners.insert(prefix.clone(), owner_module_id.to_string());
1533 }
1534 }
1535
1536 #[cfg(test)]
1538 pub(crate) fn spawn_nonce(&self, module_id: &str) -> Option<String> {
1539 self.spawn_nonces
1540 .lock()
1541 .unwrap_or_else(|poisoned| poisoned.into_inner())
1542 .get(module_id)
1543 .cloned()
1544 }
1545
1546 fn apply_identity_configuration(&self, spec: &ModuleSpec) {
1547 self.set_reserved_prefixes(&spec.module_id, &spec.reserved_prefixes);
1548 let spawn_nonce = self
1549 .spawn_nonces
1550 .lock()
1551 .unwrap_or_else(|poisoned| poisoned.into_inner())
1552 .get(&spec.module_id)
1553 .cloned();
1554 let mut reserved_nonces = self
1555 .reserved_nonces
1556 .lock()
1557 .unwrap_or_else(|poisoned| poisoned.into_inner());
1558 if spec.reserved {
1559 reserved_nonces.insert(spec.module_id.clone(), spawn_nonce);
1564 }
1565 drop(reserved_nonces);
1566 self.removal_tombstones
1570 .lock()
1571 .unwrap_or_else(|poisoned| poisoned.into_inner())
1572 .remove(&spec.module_id);
1573 }
1574
1575 pub fn reserved_hello_authorized(&self, module_id: &str, presented: Option<&str>) -> bool {
1580 self.reserved_hello_rejection(module_id, presented)
1581 .is_none()
1582 }
1583
1584 pub(crate) fn reserved_hello_rejection(
1585 &self,
1586 module_id: &str,
1587 presented: Option<&str>,
1588 ) -> Option<ReservedHelloRejection> {
1589 let nonces = self
1590 .reserved_nonces
1591 .lock()
1592 .unwrap_or_else(|poisoned| poisoned.into_inner());
1593 if let Some(expected) = nonces.get(module_id) {
1594 let authorized = match expected {
1598 Some(expected) => {
1599 presented.is_some_and(|p| constant_time_eq(expected.as_bytes(), p.as_bytes()))
1600 }
1601 None => false,
1602 };
1603 if authorized {
1604 return None;
1605 }
1606 return Some(ReservedHelloRejection::Exact {
1607 module_id: module_id.to_string(),
1608 });
1609 }
1610 drop(nonces);
1611
1612 let matched_prefix = self
1613 .reserved_prefix_owners
1614 .lock()
1615 .unwrap_or_else(|poisoned| poisoned.into_inner())
1616 .iter()
1617 .filter(|(prefix, _)| module_id.starts_with(prefix.as_str()))
1618 .max_by_key(|(prefix, _)| prefix.len())
1619 .map(|(prefix, owner)| (prefix.clone(), owner.clone()));
1620 let (prefix, owner_module_id) = matched_prefix?;
1621
1622 let authorized = presented.is_some_and(|presented| {
1623 self.spawn_nonces
1624 .lock()
1625 .unwrap_or_else(|poisoned| poisoned.into_inner())
1626 .get(&owner_module_id)
1627 .is_some_and(|expected| constant_time_eq(expected.as_bytes(), presented.as_bytes()))
1628 || self.swap_nonce_matches(&owner_module_id, presented)
1631 });
1632 if authorized {
1633 None
1634 } else {
1635 Some(ReservedHelloRejection::Prefix {
1636 prefix,
1637 owner_module_id,
1638 })
1639 }
1640 }
1641
1642 pub fn spawned_consumer_authorized(&self, module_id: &str, presented: &str) -> bool {
1647 if presented.is_empty() {
1648 return false;
1649 }
1650 let nonces = self
1651 .spawn_nonces
1652 .lock()
1653 .unwrap_or_else(|poisoned| poisoned.into_inner());
1654 let current = nonces
1655 .get(module_id)
1656 .is_some_and(|expected| constant_time_eq(expected.as_bytes(), presented.as_bytes()));
1657 drop(nonces);
1658 current || self.swap_nonce_matches(module_id, presented)
1663 }
1664
1665 fn swap_nonce_matches(&self, module_id: &str, presented: &str) -> bool {
1667 let swaps = self
1668 .swaps
1669 .lock()
1670 .unwrap_or_else(|poisoned| poisoned.into_inner());
1671 swaps.get(module_id).is_some_and(|swap| {
1672 constant_time_eq(swap.candidate_nonce.as_bytes(), presented.as_bytes())
1673 || swap.incumbent_nonce.as_deref().is_some_and(|incumbent| {
1674 constant_time_eq(incumbent.as_bytes(), presented.as_bytes())
1675 })
1676 })
1677 }
1678
1679 pub(crate) fn open_swap(&self, module_id: &str, candidate_nonce: String) {
1682 let incumbent_nonce = self
1683 .spawn_nonces
1684 .lock()
1685 .unwrap_or_else(|poisoned| poisoned.into_inner())
1686 .get(module_id)
1687 .cloned();
1688 self.swaps
1689 .lock()
1690 .unwrap_or_else(|poisoned| poisoned.into_inner())
1691 .insert(
1692 module_id.to_string(),
1693 OpenSwap {
1694 candidate_nonce,
1695 incumbent_nonce,
1696 candidate_admitted: false,
1697 },
1698 );
1699 }
1700
1701 pub(crate) fn close_swap(&self, module_id: &str) {
1704 self.swaps
1705 .lock()
1706 .unwrap_or_else(|poisoned| poisoned.into_inner())
1707 .remove(module_id);
1708 }
1709
1710 pub(crate) fn set_swap_promotion_observer(
1713 &self,
1714 observer: std::sync::Weak<dyn SwapPromotionObserver>,
1715 ) {
1716 *self
1717 .promotion_observer
1718 .0
1719 .lock()
1720 .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(observer);
1721 }
1722
1723 fn notify_swap_promoted(&self, registration: &crate::registry::ModuleRegistration) {
1726 let observer = self
1727 .promotion_observer
1728 .0
1729 .lock()
1730 .unwrap_or_else(|poisoned| poisoned.into_inner())
1731 .as_ref()
1732 .and_then(std::sync::Weak::upgrade);
1733 if let Some(observer) = observer {
1734 observer.swap_promoted(registration);
1735 }
1736 }
1737
1738 pub(crate) fn swap_open(&self, module_id: &str) -> bool {
1740 self.swaps
1741 .lock()
1742 .unwrap_or_else(|poisoned| poisoned.into_inner())
1743 .contains_key(module_id)
1744 }
1745
1746 fn promote_swap_nonce(&self, module_id: &str, reserved: bool) {
1751 let candidate_nonce = self
1752 .swaps
1753 .lock()
1754 .unwrap_or_else(|poisoned| poisoned.into_inner())
1755 .get(module_id)
1756 .map(|swap| swap.candidate_nonce.clone());
1757 let Some(nonce) = candidate_nonce else {
1758 return;
1759 };
1760 self.set_spawn_nonce(module_id, nonce.clone());
1761 if reserved {
1762 self.set_reserved_nonce(module_id, nonce);
1763 }
1764 }
1765
1766 pub(crate) fn swap_hello_admission(
1781 &self,
1782 module_id: &str,
1783 presented: Option<&str>,
1784 ) -> SwapHelloAdmission {
1785 let swaps = self
1786 .swaps
1787 .lock()
1788 .unwrap_or_else(|poisoned| poisoned.into_inner());
1789 let Some(swap) = swaps.get(module_id) else {
1790 return SwapHelloAdmission::NotSwapping;
1791 };
1792 let Some(presented) = presented else {
1793 return SwapHelloAdmission::Refused;
1794 };
1795 if constant_time_eq(swap.candidate_nonce.as_bytes(), presented.as_bytes()) {
1796 return if swap.candidate_admitted {
1797 SwapHelloAdmission::Refused
1798 } else {
1799 SwapHelloAdmission::Candidate
1800 };
1801 }
1802 if swap
1803 .incumbent_nonce
1804 .as_deref()
1805 .is_some_and(|incumbent| constant_time_eq(incumbent.as_bytes(), presented.as_bytes()))
1806 {
1807 return SwapHelloAdmission::NotSwapping;
1808 }
1809 SwapHelloAdmission::Refused
1810 }
1811
1812 pub(crate) fn mark_swap_candidate_admitted(&self, module_id: &str) {
1815 if let Some(swap) = self
1816 .swaps
1817 .lock()
1818 .unwrap_or_else(|poisoned| poisoned.into_inner())
1819 .get_mut(module_id)
1820 {
1821 swap.candidate_admitted = true;
1822 }
1823 }
1824
1825 pub fn spawn_launch_nonce_for(&self, module_id: &str) -> Option<String> {
1827 self.spawn_nonces
1828 .lock()
1829 .unwrap_or_else(|poisoned| poisoned.into_inner())
1830 .get(module_id)
1831 .cloned()
1832 }
1833
1834 pub fn reserved_launch_nonce_for(&self, module_id: &str) -> Option<String> {
1836 self.reserved_nonces
1837 .lock()
1838 .unwrap_or_else(|poisoned| poisoned.into_inner())
1839 .get(module_id)
1840 .cloned()
1841 .flatten()
1842 }
1843
1844 pub fn insert(&self, module: SupervisedModule) -> Option<SupervisedModule> {
1845 let mut modules = self
1846 .modules
1847 .lock()
1848 .unwrap_or_else(|poisoned| poisoned.into_inner());
1849 modules.insert(module.module_id().to_string(), module)
1850 }
1851
1852 pub fn get(&self, module_id: &str) -> Option<SupervisedModule> {
1853 let modules = self
1854 .modules
1855 .lock()
1856 .unwrap_or_else(|poisoned| poisoned.into_inner());
1857 modules.get(module_id).cloned()
1858 }
1859
1860 pub(crate) fn record_late_health_answer(
1861 &self,
1862 module_id: &str,
1863 latency_ms: u64,
1864 ) -> Result<bool, SuperviseError> {
1865 let Some(module) = self.get(module_id) else {
1866 return Ok(false);
1867 };
1868 update_snapshot(&module.inner.snapshot, Some(module_id), |state| {
1869 state.health.late_answer_count = state.health.late_answer_count.saturating_add(1);
1870 state.health.last_late_answer_latency_ms = Some(latency_ms);
1871 state.health.consecutive_failures = 0;
1879 })?;
1880 Ok(true)
1881 }
1882
1883 pub fn record_deliberate_severance(&self, module_id: &str) -> Result<bool, SuperviseError> {
1889 let Some(module) = self.get(module_id) else {
1890 return Ok(false);
1891 };
1892 let status = module.status()?;
1893 let Some((pid, start_time)) = status.pid.zip(status.process_start_time) else {
1894 return Ok(false);
1895 };
1896 module.record_deliberate_severance(ProcessIdentity { pid, start_time })
1897 }
1898
1899 pub fn list(&self) -> Vec<SupervisedModule> {
1900 let modules = self
1901 .modules
1902 .lock()
1903 .unwrap_or_else(|poisoned| poisoned.into_inner());
1904 let mut modules = modules.values().cloned().collect::<Vec<_>>();
1905 modules.sort_by(|left, right| left.module_id().cmp(right.module_id()));
1906 modules
1907 }
1908
1909 pub(crate) fn retire(&self, module_id: &str) -> Option<SupervisedModule> {
1910 self.spawn_nonces
1911 .lock()
1912 .unwrap_or_else(|poisoned| poisoned.into_inner())
1913 .remove(module_id);
1914 self.close_swap(module_id);
1915 let mut reserved_nonces = self
1916 .reserved_nonces
1917 .lock()
1918 .unwrap_or_else(|poisoned| poisoned.into_inner());
1919 if reserved_nonces.contains_key(module_id) {
1920 reserved_nonces.insert(module_id.to_string(), None);
1923 }
1924 drop(reserved_nonces);
1925 self.reserved_prefix_owners
1926 .lock()
1927 .unwrap_or_else(|poisoned| poisoned.into_inner())
1928 .retain(|_, owner| owner != module_id);
1929 self.modules
1930 .lock()
1931 .unwrap_or_else(|poisoned| poisoned.into_inner())
1932 .remove(module_id)
1933 }
1934
1935 pub(crate) fn record_rescan_removal(&self, module_id: &str) {
1938 self.removal_tombstones
1939 .lock()
1940 .unwrap_or_else(|poisoned| poisoned.into_inner())
1941 .insert(module_id.to_string(), unix_ms_now());
1942 }
1943
1944 pub(crate) fn removal_tombstone_age_ms(&self, module_id: &str) -> Option<u64> {
1946 self.removal_tombstones
1947 .lock()
1948 .unwrap_or_else(|poisoned| poisoned.into_inner())
1949 .get(module_id)
1950 .copied()
1951 .map(|removed_at_ms| unix_ms_now().saturating_sub(removed_at_ms))
1952 }
1953
1954 pub(crate) fn release_retained_reserved_gate(&self, module_id: &str) -> bool {
1959 if self.get(module_id).is_some() {
1960 return false;
1961 }
1962 let mut reserved_nonces = self
1963 .reserved_nonces
1964 .lock()
1965 .unwrap_or_else(|poisoned| poisoned.into_inner());
1966 if !matches!(reserved_nonces.get(module_id), Some(None)) {
1967 return false;
1968 }
1969 reserved_nonces.remove(module_id);
1970 true
1971 }
1972
1973 pub(crate) fn operation_lock(&self) -> Arc<AsyncMutex<()>> {
1974 Arc::clone(&self.operation_lock)
1975 }
1976}
1977
1978#[derive(Debug, Clone)]
1980pub struct Supervisor {
1981 registry: Arc<Registry>,
1982 restart_policy: RestartPolicy,
1983 drain_timeout: Duration,
1984 connection_file_path: Option<PathBuf>,
1985 capture_logs_dir: Option<PathBuf>,
1986 forwarding: Option<Arc<ForwardingTable>>,
1987 process_liveness: Arc<SupervisorProcessLiveness>,
1988 supervisor_handle: Option<SupervisorHandle>,
1989 health: HealthConfig,
1990 daemon_start_clock: crate::clock::StartClock,
1991 terminal_journal: Option<Arc<crate::terminal_journal::TerminalJournal>>,
1992 spawn_events: SpawnEventFeed,
1993 provenance_probe: ExecutableIdentityProbe,
1994 child_roster: ChildRoster,
1997 #[cfg(target_os = "linux")]
1998 cgroup_placement: Option<subc_cgroup::Placement>,
1999}
2000
2001impl Supervisor {
2002 #[cfg(unix)]
2013 pub(crate) fn begin_daemon_shutdown(&self) {
2014 self.child_roster.close();
2015 if let Some(journal) = &self.terminal_journal {
2016 journal.stamp_shutdown();
2017 }
2018 }
2019
2020 #[cfg(unix)]
2024 pub(crate) async fn drain_for_daemon_shutdown(&self) -> Result<(), SuperviseError> {
2025 const NOTICE_BUDGET: Duration = Duration::from_millis(500);
2026 const DRAIN_BUDGET: Duration = Duration::from_secs(2);
2027 let Some(forwarding) = &self.forwarding else {
2028 return Ok(());
2029 };
2030 let module_ids = forwarding
2031 .begin_daemon_drain()
2032 .map_err(SuperviseError::Forwarding)?;
2033 let deadline_ms =
2034 unix_ms_now().saturating_add((NOTICE_BUDGET + DRAIN_BUDGET).as_millis() as u64);
2035 let mut notices = tokio::task::JoinSet::new();
2036 let mut drains = Vec::new();
2037 for module_id in module_ids {
2038 let Some(target) = forwarding
2039 .begin_module_drain(&module_id, RouteCloseReason::Restart)
2040 .map_err(SuperviseError::Forwarding)?
2041 else {
2042 continue;
2043 };
2044 let routes = forwarding
2045 .endpoint_routes(target.endpoint)
2046 .map_err(SuperviseError::Forwarding)?;
2047 let command = serde_json::to_vec(&ModuleControlCommand::Draining {
2053 reason: RouteCloseReason::Restart,
2054 deadline_ms,
2055 })
2056 .expect("module draining serializes");
2057 let closing = serde_json::to_vec(&ClientControlPush::RouteClosing {
2058 module_id: module_id.clone(),
2059 reason: RouteCloseReason::Restart,
2060 })
2061 .expect("route closing serializes");
2062 let mut recipients = vec![(target.sink.clone(), target.negotiated_ver, command)];
2063 let mut seen = std::collections::HashSet::new();
2064 for route in routes {
2065 let client = route.goodbye_target;
2066 if seen.insert(client.connection_id) {
2067 recipients.push((client.sink, client.negotiated_ver, closing.clone()));
2068 }
2069 }
2070 for (sink, version, body) in recipients {
2071 notices.spawn(async move {
2072 let frame = Frame::build_with_version(
2073 version,
2074 FrameType::Push,
2075 control_flags(),
2076 0,
2077 0,
2078 0,
2079 body,
2080 )
2081 .expect("bounded lifecycle notice frame builds");
2082 sink.send_flushed(frame).await
2083 });
2084 }
2085 let gauges = declared_busy_gauges(&self.registry, &module_id)?;
2086 drains.push((module_id, target.endpoint, gauges));
2087 }
2088 let notice_deadline = Instant::now() + NOTICE_BUDGET;
2091 while let Ok(Some(result)) = timeout_at(notice_deadline, notices.join_next()).await {
2092 if !matches!(result, Ok(Ok(()))) {
2093 warn!(?result, "daemon shutdown notice delivery failed");
2094 }
2095 }
2096 notices.abort_all();
2097 let deadline = Instant::now() + DRAIN_BUDGET;
2098 let mut waits = tokio::task::JoinSet::new();
2099 for (module_id, endpoint, gauges) in drains {
2100 let forwarding = Arc::clone(forwarding);
2101 let mut runtime = self.runtime_config();
2102 runtime.health.cadence = Duration::from_millis(100);
2103 waits.spawn(async move {
2104 wait_for_forwarding_quiescence(
2105 &forwarding,
2106 &module_id,
2107 &runtime,
2108 endpoint,
2109 deadline,
2110 &gauges,
2111 DrainScope::Active,
2112 )
2113 .await
2114 });
2115 }
2116 while let Ok(Some(result)) = timeout_at(deadline, waits.join_next()).await {
2117 if !matches!(result, Ok(Ok(true))) {
2118 warn!(?result, "daemon shutdown drain did not reach quiescence");
2119 }
2120 }
2121 Ok(())
2122 }
2123
2124 #[cfg(unix)]
2134 pub(crate) async fn end_children_for_daemon_shutdown(
2135 &self,
2136 already_escalated: bool,
2137 escalate: impl std::future::Future<Output = ()>,
2138 ) {
2139 if let Some(forwarding) = &self.forwarding {
2140 let closed = forwarding.close_all_connections(&CloseReason::new(
2141 "daemon_shutdown",
2142 "the daemon is exiting after its shutdown notice and drain",
2143 ));
2144 debug!(closed, "closed established connections for daemon shutdown");
2145 }
2146 crate::child_roster::end_children_for_daemon_shutdown(
2147 &self.child_roster,
2148 already_escalated,
2149 escalate,
2150 )
2151 .await;
2152 }
2153
2154 pub fn new(registry: Arc<Registry>, restart_policy: RestartPolicy) -> Self {
2155 Self {
2156 registry,
2157 restart_policy,
2158 drain_timeout: DEFAULT_DRAIN_TIMEOUT,
2159 connection_file_path: None,
2160 capture_logs_dir: None,
2161 forwarding: None,
2162 process_liveness: Arc::new(SupervisorProcessLiveness::default()),
2163 supervisor_handle: None,
2164 health: HealthConfig::default(),
2165 daemon_start_clock: crate::clock::StartClock::capture(),
2166 terminal_journal: None,
2167 spawn_events: SpawnEventFeed::default(),
2168 provenance_probe: ExecutableIdentityProbe::default(),
2169 child_roster: ChildRoster::default(),
2170 #[cfg(target_os = "linux")]
2171 cgroup_placement: None,
2172 }
2173 }
2174
2175 pub fn with_drain_timeout(mut self, drain_timeout: Duration) -> Self {
2176 self.drain_timeout = drain_timeout;
2177 self
2178 }
2179
2180 pub fn with_process_liveness(
2181 mut self,
2182 process_liveness: Arc<SupervisorProcessLiveness>,
2183 ) -> Self {
2184 self.process_liveness = process_liveness;
2185 self
2186 }
2187
2188 pub fn with_connection_file_path(mut self, connection_file_path: impl Into<PathBuf>) -> Self {
2189 self.connection_file_path = Some(connection_file_path.into());
2190 self
2191 }
2192
2193 pub fn with_capture_logs_dir(mut self, logs_dir: impl Into<PathBuf>) -> Self {
2195 self.capture_logs_dir = Some(logs_dir.into());
2196 self
2197 }
2198
2199 pub fn with_daemon_incarnation(self, daemon_incarnation: String) -> Self {
2202 self.spawn_events.configure_incarnation(daemon_incarnation);
2206 self
2207 }
2208
2209 pub fn with_terminal_journal(self, path: PathBuf, daemon_incarnation: String) -> Self {
2212 let mut this = self.with_daemon_incarnation(daemon_incarnation.clone());
2213 this.terminal_journal = Some(Arc::new(crate::terminal_journal::TerminalJournal::open(
2214 path,
2215 daemon_incarnation,
2216 )));
2217 this
2218 }
2219
2220 pub fn with_forwarding(mut self, forwarding: Arc<ForwardingTable>) -> Self {
2221 self.forwarding = Some(forwarding);
2222 self
2223 }
2224
2225 pub fn with_handle(mut self, supervisor_handle: SupervisorHandle) -> Self {
2226 self.spawn_events = supervisor_handle.spawn_events.clone();
2227 self.supervisor_handle = Some(supervisor_handle);
2228 self
2229 }
2230
2231 pub fn with_health_config(mut self, health: HealthConfig) -> Self {
2232 self.health = health;
2233 self
2234 }
2235
2236 #[cfg(target_os = "linux")]
2237 pub fn with_cgroup_placement(
2238 mut self,
2239 cgroup_placement: Option<subc_cgroup::Placement>,
2240 ) -> Self {
2241 self.cgroup_placement = cgroup_placement;
2242 self
2243 }
2244
2245 pub fn spawn(&self, spec: ModuleSpec) -> Result<SupervisedModule, SuperviseError> {
2251 validate_spec(&spec)?;
2252
2253 let runtime = self.runtime_config();
2254 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
2255 let child = spawn_child(
2256 &spec,
2257 runtime.connection_file_path.as_deref(),
2258 self.supervisor_handle.as_ref(),
2259 &runtime.stderr_ring,
2260 runtime.capture_logs_dir.as_deref(),
2261 &runtime.child_roster,
2262 #[cfg(target_os = "linux")]
2263 runtime.cgroup_placement.as_ref(),
2264 )?;
2265 set_running(&snapshot, &child, &spec.module_id, &runtime.spawn_events)?;
2266 self.process_liveness
2267 .track(spec.module_id.clone(), Arc::clone(&snapshot));
2268
2269 Ok(self.supervised_module(spec, runtime, snapshot, Some(child)))
2270 }
2271
2272 pub fn supervise_configured(
2278 &self,
2279 spec: ModuleSpec,
2280 enabled: bool,
2281 ) -> Result<SupervisedModule, SuperviseError> {
2282 validate_spec(&spec)?;
2283
2284 let runtime = self.runtime_config();
2285 if !enabled {
2286 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::disabled()));
2287 return Ok(self.supervised_module(spec, runtime, snapshot, None));
2288 }
2289
2290 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
2291 match spawn_child(
2292 &spec,
2293 runtime.connection_file_path.as_deref(),
2294 self.supervisor_handle.as_ref(),
2295 &runtime.stderr_ring,
2296 runtime.capture_logs_dir.as_deref(),
2297 &runtime.child_roster,
2298 #[cfg(target_os = "linux")]
2299 runtime.cgroup_placement.as_ref(),
2300 ) {
2301 Ok(child) => {
2302 set_running(&snapshot, &child, &spec.module_id, &runtime.spawn_events)?;
2303 self.process_liveness
2304 .track(spec.module_id.clone(), Arc::clone(&snapshot));
2305 Ok(self.supervised_module(spec, runtime, snapshot, Some(child)))
2306 }
2307 Err(err) => {
2308 error!(
2309 module_id = %spec.module_id,
2310 program = %spec.program.display(),
2311 error = %err,
2312 "configured module failed to spawn; marking failed and continuing"
2313 );
2314 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::failed()));
2315 Ok(self.supervised_module(spec, runtime, snapshot, None))
2316 }
2317 }
2318 }
2319
2320 pub fn supervise_configured_with_health(
2326 &self,
2327 spec: ModuleSpec,
2328 enabled: bool,
2329 health: HealthConfig,
2330 drain_timeout_ms: Option<u64>,
2331 restart_policy: RestartPolicy,
2332 ) -> Result<SupervisedModule, SuperviseError> {
2333 validate_spec(&spec)?;
2334
2335 let mut runtime = self.runtime_config();
2336 runtime.health = health;
2337 runtime.restart_policy = restart_policy;
2338 if let Some(ms) = drain_timeout_ms {
2339 runtime.drain_timeout = Duration::from_millis(ms);
2340 *runtime
2341 .effective_drain_timeout
2342 .lock()
2343 .unwrap_or_else(|poisoned| poisoned.into_inner()) = runtime.drain_timeout;
2344 }
2345 if !enabled {
2346 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::disabled()));
2347 return Ok(self.supervised_module(spec, runtime, snapshot, None));
2348 }
2349
2350 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
2351 match spawn_child(
2352 &spec,
2353 runtime.connection_file_path.as_deref(),
2354 self.supervisor_handle.as_ref(),
2355 &runtime.stderr_ring,
2356 runtime.capture_logs_dir.as_deref(),
2357 &runtime.child_roster,
2358 #[cfg(target_os = "linux")]
2359 runtime.cgroup_placement.as_ref(),
2360 ) {
2361 Ok(child) => {
2362 set_running(&snapshot, &child, &spec.module_id, &runtime.spawn_events)?;
2363 self.process_liveness
2364 .track(spec.module_id.clone(), Arc::clone(&snapshot));
2365 Ok(self.supervised_module(spec, runtime, snapshot, Some(child)))
2366 }
2367 Err(err) => {
2368 if health.critical {
2369 error!(
2370 module_id = %spec.module_id,
2371 program = %spec.program.display(),
2372 error = %err,
2373 "critical configured module failed to spawn; marking failed and alerting"
2374 );
2375 } else {
2376 error!(
2377 module_id = %spec.module_id,
2378 program = %spec.program.display(),
2379 error = %err,
2380 "configured module failed to spawn; marking failed and continuing"
2381 );
2382 }
2383 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::failed()));
2384 Ok(self.supervised_module(spec, runtime, snapshot, None))
2385 }
2386 }
2387 }
2388
2389 fn runtime_config(&self) -> SupervisorRuntimeConfig {
2390 let effective_drain_timeout = Arc::new(Mutex::new(self.drain_timeout));
2391 SupervisorRuntimeConfig {
2392 restart_policy: self.restart_policy,
2393 drain_timeout: self.drain_timeout,
2394 child_roster: self
2397 .child_roster
2398 .for_module(Arc::clone(&effective_drain_timeout)),
2399 effective_drain_timeout,
2400 default_drain_timeout: self.drain_timeout,
2401 health: self.health,
2402 connection_file_path: self.connection_file_path.clone(),
2403 capture_logs_dir: self.capture_logs_dir.clone(),
2404 forwarding: self.forwarding.clone(),
2405 supervisor_handle: self.supervisor_handle.clone(),
2406 stderr_ring: Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default()))),
2407 terminal_ring: Arc::new(Mutex::new(
2408 TerminalRing::new(
2409 TerminalRingConfig::default(),
2410 self.daemon_start_clock.started_at_ms(),
2411 )
2412 .with_start_clock(self.daemon_start_clock)
2413 .with_journal(self.terminal_journal.clone())
2414 .with_daemon_shutdown(self.child_roster.shutdown_flag()),
2415 )),
2416 spawn_events: self.spawn_events.clone(),
2417 #[cfg(target_os = "linux")]
2418 cgroup_placement: self.cgroup_placement.clone(),
2419 #[cfg(test)]
2420 test_seed_stale_facts_before_enable_spawn: false,
2421 }
2422 }
2423
2424 fn supervised_module(
2425 &self,
2426 spec: ModuleSpec,
2427 runtime: SupervisorRuntimeConfig,
2428 snapshot: SharedSnapshot,
2429 child: Option<SupervisedChild>,
2430 ) -> SupervisedModule {
2431 let configuration = Arc::new(Mutex::new(SupervisedConfiguration {
2432 spec: spec.clone(),
2433 health: runtime.health,
2434 }));
2435 let stderr_ring = Arc::clone(&runtime.stderr_ring);
2436 let terminal_ring = Arc::clone(&runtime.terminal_ring);
2437 let restart_policy = runtime.restart_policy;
2441 let effective_drain_timeout = Arc::clone(&runtime.effective_drain_timeout);
2442 let (tx, rx) = mpsc::channel(4);
2443 let monitor = tokio::spawn(supervise_loop(
2444 spec.clone(),
2445 runtime,
2446 Arc::clone(&self.registry),
2447 Arc::clone(&self.process_liveness),
2448 Arc::clone(&snapshot),
2449 child,
2450 rx,
2451 ));
2452
2453 let module_id = spec.module_id.clone();
2454 let module = SupervisedModule {
2455 inner: Arc::new(SupervisedModuleInner {
2456 module_id: module_id.clone(),
2457 registry: Arc::clone(&self.registry),
2458 snapshot,
2459 configuration,
2460 stderr_ring,
2461 terminal_ring,
2462 commands: tx,
2463 monitor: Mutex::new(Some(monitor)),
2464 restart_policy,
2465 effective_drain_timeout,
2466 provenance_probe: self.provenance_probe.clone(),
2467 }),
2468 };
2469 if let Some(supervisor_handle) = &self.supervisor_handle {
2470 supervisor_handle.apply_identity_configuration(&spec);
2471 supervisor_handle.insert(module.clone());
2472 }
2473 module
2474 }
2475}
2476
2477impl Default for Supervisor {
2478 fn default() -> Self {
2479 Self::new(Arc::new(Registry::default()), RestartPolicy::default())
2480 }
2481}
2482
2483#[derive(Clone)]
2485pub struct SupervisedModule {
2486 inner: Arc<SupervisedModuleInner>,
2487}
2488
2489struct SupervisedModuleInner {
2490 module_id: String,
2491 registry: Arc<Registry>,
2492 snapshot: SharedSnapshot,
2493 configuration: Arc<Mutex<SupervisedConfiguration>>,
2494 stderr_ring: Arc<Mutex<StderrRing>>,
2495 terminal_ring: Arc<Mutex<TerminalRing>>,
2496 commands: mpsc::Sender<SupervisorCommand>,
2497 monitor: Mutex<Option<JoinHandle<()>>>,
2498 restart_policy: RestartPolicy,
2502 effective_drain_timeout: Arc<Mutex<Duration>>,
2503 provenance_probe: ExecutableIdentityProbe,
2504}
2505
2506impl fmt::Debug for SupervisedModule {
2507 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2508 f.debug_struct("SupervisedModule")
2509 .field("module_id", &self.inner.module_id)
2510 .field("status", &self.status())
2511 .finish_non_exhaustive()
2512 }
2513}
2514
2515impl SupervisedModule {
2516 pub fn module_id(&self) -> &str {
2517 &self.inner.module_id
2518 }
2519
2520 #[cfg(test)]
2524 pub(crate) fn record_health_probe_failure_for_test(
2525 &self,
2526 detail: &str,
2527 ) -> Result<(), SuperviseError> {
2528 update_snapshot(&self.inner.snapshot, Some(&self.inner.module_id), |state| {
2529 state.health.consecutive_failures = state.health.consecutive_failures.saturating_add(1);
2530 state.health.detail = Some(detail.to_string());
2531 })
2532 }
2533
2534 pub fn state(&self) -> Result<ModuleState, SuperviseError> {
2535 Ok(lock_snapshot(&self.inner.snapshot)?.state)
2536 }
2537
2538 pub fn stderr_tail(
2545 &self,
2546 max_lines: Option<usize>,
2547 max_bytes: Option<usize>,
2548 ) -> StderrTailSnapshot {
2549 self.inner
2550 .stderr_ring
2551 .lock()
2552 .unwrap_or_else(|poisoned| poisoned.into_inner())
2553 .snapshot(max_lines, max_bytes)
2554 }
2555
2556 pub fn terminal_history(&self) -> TerminalHistorySnapshot {
2561 self.inner
2562 .terminal_ring
2563 .lock()
2564 .unwrap_or_else(|poisoned| poisoned.into_inner())
2565 .snapshot()
2566 }
2567
2568 pub fn durable_terminal_history(&self) -> subc_control::TerminalHistory {
2573 durable_terminal_history_of(&self.inner.terminal_ring, &self.inner.module_id)
2574 }
2575
2576 pub(crate) async fn read_durable_terminal_history(
2581 &self,
2582 ) -> Result<subc_control::TerminalHistory, tokio::task::JoinError> {
2583 let terminal_ring = Arc::clone(&self.inner.terminal_ring);
2584 let module_id = self.inner.module_id.clone();
2585 tokio::task::spawn_blocking(move || durable_terminal_history_of(&terminal_ring, &module_id))
2586 .await
2587 }
2588
2589 pub fn status(&self) -> Result<ModuleStatus, SuperviseError> {
2590 self.status_with_snapshot_lock(&self.inner.snapshot, None)
2591 }
2592
2593 pub(crate) fn record_deliberate_severance(
2594 &self,
2595 identity: ProcessIdentity,
2596 ) -> Result<bool, SuperviseError> {
2597 let mut snapshot = lock_snapshot(&self.inner.snapshot)?;
2598 if snapshot.pid != Some(identity.pid)
2599 || snapshot.process_start_time != Some(identity.start_time)
2600 {
2601 return Ok(false);
2602 }
2603 snapshot.deliberate_severance = Some(identity);
2604 Ok(true)
2605 }
2606
2607 pub(crate) fn status_for_control(
2612 &self,
2613 caller: &'static str,
2614 ) -> Result<ModuleStatus, SuperviseError> {
2615 self.status_with_snapshot_lock(&self.inner.snapshot, Some(caller))
2616 }
2617
2618 fn status_with_snapshot_lock(
2619 &self,
2620 snapshot: &SharedSnapshot,
2621 caller: Option<&'static str>,
2622 ) -> Result<ModuleStatus, SuperviseError> {
2623 let mut guard = match caller {
2624 Some(caller) => lock_snapshot_for_control(snapshot, &self.inner.module_id, caller)?,
2625 None => lock_snapshot(snapshot)?,
2626 };
2627 let restart_count =
2630 guard.crash_restarts_in_window(self.inner.restart_policy.window, Instant::now());
2631 let snapshot = guard.clone();
2632 drop(guard);
2633 let drain_timeout = *self.inner.effective_drain_timeout.lock().map_err(|_| {
2634 SuperviseError::StatePoisoned {
2635 module_id: Some(self.inner.module_id.clone()),
2636 }
2637 })?;
2638 let registration_active = self
2639 .inner
2640 .registry
2641 .get_module(&self.inner.module_id)
2642 .map_err(SuperviseError::Registry)?
2643 .is_some();
2644 let protocol = self.declared_protocol()?;
2645 let running_process =
2646 snapshot.enabled && snapshot.state == ModuleState::Running && snapshot.process_alive;
2647 let live = match protocol {
2653 ModuleProtocol::Subc => running_process && registration_active,
2654 ModuleProtocol::None => running_process,
2655 };
2656
2657 Ok(ModuleStatus {
2658 module_id: self.inner.module_id.clone(),
2659 state: snapshot.state,
2660 enabled: snapshot.enabled,
2661 process_alive: snapshot.process_alive,
2662 registration_active,
2663 protocol,
2664 live,
2665 restart_count,
2666 lifetime_restarts: snapshot.lifetime_restarts,
2667 spawn_generation: snapshot.spawn_generation,
2668 max_restarts: self.inner.restart_policy.max_restarts,
2669 restart_window: self.inner.restart_policy.window,
2670 drain_timeout,
2671 restart_backoff: self.inner.restart_policy.backoff,
2672 restart_max_backoff: self.inner.restart_policy.max_backoff,
2673 pid: snapshot.pid,
2674 spawned_at_ms: snapshot.spawned_at_ms,
2675 spawned_from: snapshot.spawned_from,
2676 process_start_time: snapshot.process_start_time,
2677 last_exit: snapshot.last_exit,
2678 health: snapshot.health,
2679 })
2680 }
2681
2682 #[cfg(test)]
2683 pub(crate) fn hold_snapshot_for_test(
2684 &self,
2685 acquired: std::sync::mpsc::Sender<()>,
2686 hold: Duration,
2687 ) -> std::thread::JoinHandle<()> {
2688 let snapshot = Arc::clone(&self.inner.snapshot);
2689 std::thread::spawn(move || {
2690 let _guard = snapshot.lock().expect("test snapshot lock is not poisoned");
2691 acquired
2692 .send(())
2693 .expect("test receiver waits for snapshot lock");
2694 std::thread::sleep(hold);
2695 })
2696 }
2697
2698 pub(crate) async fn running_image_agreement(&self) -> subc_control::RunningImageAgreement {
2699 let snapshot = match lock_snapshot(&self.inner.snapshot) {
2700 Ok(snapshot) => snapshot.clone(),
2701 Err(_) => {
2702 return subc_control::RunningImageAgreement::Unavailable {
2703 reason: subc_control::RunningImageUnavailableReason::NotRunning,
2704 };
2705 }
2706 };
2707 self.inner
2708 .provenance_probe
2709 .observe(
2710 snapshot.pid,
2711 snapshot.spawned_from.as_deref(),
2712 snapshot.spawned_file_identity,
2713 snapshot.process_start_time,
2714 )
2715 .await
2716 }
2717
2718 pub(crate) fn will_recover_after_connection_loss(&self) -> Result<bool, SuperviseError> {
2719 let mut snapshot = lock_snapshot(&self.inner.snapshot)?;
2720 Ok(match snapshot.state {
2721 ModuleState::Restarting => true,
2722 ModuleState::Failed | ModuleState::Disabled => false,
2723 _ => daemon_will_restart(&mut snapshot, &self.inner.restart_policy, Instant::now()),
2724 })
2725 }
2726
2727 #[cfg(test)]
2728 pub(crate) fn is_warming(&self) -> Result<bool, SuperviseError> {
2729 self.is_warming_with_snapshot_lock(None)
2730 }
2731
2732 pub(crate) fn is_warming_for_control(
2733 &self,
2734 caller: &'static str,
2735 ) -> Result<bool, SuperviseError> {
2736 self.is_warming_with_snapshot_lock(Some(caller))
2737 }
2738
2739 fn is_warming_with_snapshot_lock(
2740 &self,
2741 caller: Option<&'static str>,
2742 ) -> Result<bool, SuperviseError> {
2743 let snapshot = match caller {
2744 Some(caller) => {
2745 lock_snapshot_for_control(&self.inner.snapshot, &self.inner.module_id, caller)?
2746 }
2747 None => lock_snapshot(&self.inner.snapshot)?,
2748 }
2749 .clone();
2750 Ok(matches!(
2751 snapshot.state,
2752 ModuleState::Starting | ModuleState::Running | ModuleState::Restarting
2753 ))
2754 }
2755
2756 pub async fn drain(&self) -> Result<(), SuperviseError> {
2758 self.stop().await
2759 }
2760
2761 pub(crate) async fn retire(&self) -> Result<(), SuperviseError> {
2762 match self.state()? {
2763 ModuleState::Stopped | ModuleState::Failed => return Ok(()),
2764 ModuleState::Starting
2765 | ModuleState::Running
2766 | ModuleState::Unresponsive
2767 | ModuleState::Restarting
2768 | ModuleState::Draining
2769 | ModuleState::Disabled => {}
2770 }
2771
2772 let (reply_tx, reply_rx) = oneshot::channel();
2773 self.inner
2774 .commands
2775 .send(SupervisorCommand::Retire { reply: reply_tx })
2776 .await
2777 .map_err(|_| SuperviseError::CommandClosed {
2778 module_id: self.inner.module_id.clone(),
2779 })?;
2780 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2781 module_id: self.inner.module_id.clone(),
2782 })?
2783 }
2784
2785 pub async fn stop(&self) -> Result<(), SuperviseError> {
2786 match self.state()? {
2787 ModuleState::Stopped | ModuleState::Failed => return Ok(()),
2788 ModuleState::Starting
2789 | ModuleState::Running
2790 | ModuleState::Unresponsive
2791 | ModuleState::Restarting
2792 | ModuleState::Draining
2793 | ModuleState::Disabled => {}
2794 }
2795
2796 let (reply_tx, reply_rx) = oneshot::channel();
2797 self.inner
2798 .commands
2799 .send(SupervisorCommand::Drain { reply: reply_tx })
2800 .await
2801 .map_err(|_| SuperviseError::CommandClosed {
2802 module_id: self.inner.module_id.clone(),
2803 })?;
2804 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2805 module_id: self.inner.module_id.clone(),
2806 })?
2807 }
2808
2809 pub async fn restart(&self, drain_timeout_ms: Option<u64>) -> Result<(), SuperviseError> {
2810 let received_at_generation = lock_snapshot(&self.inner.snapshot)?.spawn_generation;
2811 let (reply_tx, reply_rx) = oneshot::channel();
2812 self.inner
2813 .commands
2814 .send(SupervisorCommand::Restart {
2815 drain_timeout_ms,
2816 received_at_generation,
2817 queued_at: Instant::now(),
2818 reply: reply_tx,
2819 })
2820 .await
2821 .map_err(|_| SuperviseError::CommandClosed {
2822 module_id: self.inner.module_id.clone(),
2823 })?;
2824 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2825 module_id: self.inner.module_id.clone(),
2826 })?
2827 }
2828
2829 pub async fn swap(&self, ready_timeout: Option<Duration>) -> Result<(), SuperviseError> {
2834 let (reply_tx, reply_rx) = oneshot::channel();
2835 self.inner
2836 .commands
2837 .send(SupervisorCommand::Swap {
2838 ready_timeout,
2839 reply: reply_tx,
2840 })
2841 .await
2842 .map_err(|_| SuperviseError::CommandClosed {
2843 module_id: self.inner.module_id.clone(),
2844 })?;
2845 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2846 module_id: self.inner.module_id.clone(),
2847 })?
2848 }
2849
2850 pub async fn reload(&self) -> Result<(), SuperviseError> {
2851 let (reply_tx, reply_rx) = oneshot::channel();
2852 self.inner
2853 .commands
2854 .send(SupervisorCommand::Reload { reply: reply_tx })
2855 .await
2856 .map_err(|_| SuperviseError::CommandClosed {
2857 module_id: self.inner.module_id.clone(),
2858 })?;
2859 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2860 module_id: self.inner.module_id.clone(),
2861 })?
2862 }
2863
2864 pub async fn set_enabled(&self, enabled: bool) -> Result<bool, SuperviseError> {
2865 let (reply_tx, reply_rx) = oneshot::channel();
2866 self.inner
2867 .commands
2868 .send(SupervisorCommand::SetEnabled {
2869 enabled,
2870 reply: reply_tx,
2871 })
2872 .await
2873 .map_err(|_| SuperviseError::CommandClosed {
2874 module_id: self.inner.module_id.clone(),
2875 })?;
2876 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2877 module_id: self.inner.module_id.clone(),
2878 })?
2879 }
2880
2881 pub(crate) fn declared_protocol(&self) -> Result<ModuleProtocol, SuperviseError> {
2886 Ok(self
2887 .inner
2888 .configuration
2889 .lock()
2890 .map_err(|_| SuperviseError::StatePoisoned {
2891 module_id: Some(self.inner.module_id.clone()),
2892 })?
2893 .spec
2894 .protocol)
2895 }
2896
2897 pub(crate) fn configuration(&self) -> Result<(ModuleSpec, HealthConfig), SuperviseError> {
2898 let configuration =
2899 self.inner
2900 .configuration
2901 .lock()
2902 .map_err(|_| SuperviseError::StatePoisoned {
2903 module_id: Some(self.inner.module_id.clone()),
2904 })?;
2905 Ok((configuration.spec.clone(), configuration.health))
2906 }
2907
2908 #[cfg(any(test, feature = "test-support"))]
2912 pub async fn update_spec_for_test(&self, spec: ModuleSpec) -> Result<(), SuperviseError> {
2913 let (_, health) = self.configuration()?;
2914 let drain_timeout_ms = u64::try_from(
2915 self.inner
2916 .effective_drain_timeout
2917 .lock()
2918 .unwrap_or_else(|poisoned| poisoned.into_inner())
2919 .as_millis(),
2920 )
2921 .ok();
2922 self.update_configuration(spec, health, drain_timeout_ms)
2923 .await
2924 }
2925
2926 pub(crate) async fn update_configuration(
2927 &self,
2928 spec: ModuleSpec,
2929 health: HealthConfig,
2930 drain_timeout_ms: Option<u64>,
2931 ) -> Result<(), SuperviseError> {
2932 if spec.module_id != self.inner.module_id {
2933 return Err(SuperviseError::InvalidSpec {
2934 reason: "a supervised module's module_id cannot be changed".to_string(),
2935 });
2936 }
2937 validate_spec(&spec)?;
2938 let (reply_tx, reply_rx) = oneshot::channel();
2939 self.inner
2940 .commands
2941 .send(SupervisorCommand::UpdateConfiguration {
2942 spec: spec.clone(),
2943 health,
2944 drain_timeout_ms,
2945 reply: reply_tx,
2946 })
2947 .await
2948 .map_err(|_| SuperviseError::CommandClosed {
2949 module_id: self.inner.module_id.clone(),
2950 })?;
2951 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2952 module_id: self.inner.module_id.clone(),
2953 })?;
2954 let mut configuration =
2955 self.inner
2956 .configuration
2957 .lock()
2958 .map_err(|_| SuperviseError::StatePoisoned {
2959 module_id: Some(self.inner.module_id.clone()),
2960 })?;
2961 configuration.spec = spec;
2962 configuration.health = health;
2963 Ok(())
2964 }
2965}
2966
2967impl Drop for SupervisedModuleInner {
2968 fn drop(&mut self) {
2969 let Ok(mut monitor) = self.monitor.lock() else {
2970 return;
2971 };
2972 if let Some(monitor) = monitor.as_ref().filter(|monitor| !monitor.is_finished()) {
2973 let _ = update_snapshot(&self.snapshot, Some(&self.module_id), |state| {
2974 state.state = ModuleState::Stopped;
2975 clear_current_process_facts(state);
2976 });
2977 monitor.abort();
2978 }
2979 let _ = monitor.take();
2980 }
2981}
2982
2983#[derive(Debug)]
2984enum SupervisorCommand {
2985 Drain {
2986 reply: oneshot::Sender<Result<(), SuperviseError>>,
2987 },
2988 Retire {
2989 reply: oneshot::Sender<Result<(), SuperviseError>>,
2990 },
2991 Restart {
2992 drain_timeout_ms: Option<u64>,
2997 received_at_generation: u64,
3001 queued_at: Instant,
3004 reply: oneshot::Sender<Result<(), SuperviseError>>,
3005 },
3006 Reload {
3007 reply: oneshot::Sender<Result<(), SuperviseError>>,
3008 },
3009 SetEnabled {
3010 enabled: bool,
3011 reply: oneshot::Sender<Result<bool, SuperviseError>>,
3012 },
3013 UpdateConfiguration {
3014 spec: ModuleSpec,
3015 health: HealthConfig,
3016 drain_timeout_ms: Option<u64>,
3019 reply: oneshot::Sender<()>,
3020 },
3021 Swap {
3022 ready_timeout: Option<Duration>,
3025 reply: oneshot::Sender<Result<(), SuperviseError>>,
3027 },
3028}
3029
3030#[derive(Debug)]
3031pub enum SuperviseError {
3032 InvalidSpec {
3033 reason: String,
3034 },
3035 Spawn {
3036 program: PathBuf,
3037 source: io::Error,
3038 cgroup_path: Option<PathBuf>,
3039 },
3040 Cgroup {
3041 module_id: String,
3042 source: io::Error,
3043 },
3044 LaunchNonce {
3047 reason: String,
3048 },
3049 Wait {
3050 module_id: String,
3051 source: io::Error,
3052 },
3053 Kill {
3054 module_id: String,
3055 source: io::Error,
3056 },
3057 Forwarding(ForwardingError),
3058 Registry(RegistryError),
3059 ReloadUnavailable {
3060 module_id: String,
3061 reason: String,
3062 },
3063 Disabled {
3068 module_id: String,
3069 },
3070 ReloadFailed {
3071 module_id: String,
3072 reason: String,
3073 },
3074 RegistrationStillActive {
3075 module_id: String,
3076 waited: Duration,
3077 },
3078 StatePoisoned {
3079 module_id: Option<String>,
3080 },
3081 CommandClosed {
3082 module_id: String,
3083 },
3084 SwapInProgress {
3088 module_id: String,
3089 },
3090 SwapRefused {
3092 module_id: String,
3093 reason: SwapRefusal,
3094 },
3095 SwapFailed {
3099 module_id: String,
3100 arm: SwapFailureArm,
3101 detail: String,
3102 candidate_exit: Option<ExitReport>,
3105 },
3106}
3107
3108#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3110pub enum SwapRefusal {
3111 OverlapExclusive,
3113 NotRegistered,
3116 ProtocolNone,
3119 NotConfigured,
3122 AlreadySwapping,
3124}
3125
3126impl SwapRefusal {
3127 pub fn as_str(self) -> &'static str {
3128 match self {
3129 Self::OverlapExclusive => "overlap_exclusive",
3130 Self::NotRegistered => "not_registered",
3131 Self::ProtocolNone => "protocol_none",
3132 Self::NotConfigured => "not_configured",
3133 Self::AlreadySwapping => "already_swapping",
3134 }
3135 }
3136}
3137
3138#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3141pub enum SwapFailureArm {
3142 SpawnFailed,
3144 NeverRegistered,
3146 NeverReady,
3148 CandidateExited,
3150 CandidateUnhealthy,
3152 Interrupted,
3156 CutoverLost,
3161}
3162
3163impl SwapFailureArm {
3164 pub fn as_str(self) -> &'static str {
3165 match self {
3166 Self::SpawnFailed => "spawn_failed",
3167 Self::NeverRegistered => "never_registered",
3168 Self::NeverReady => "never_ready",
3169 Self::CandidateExited => "candidate_exited",
3170 Self::CandidateUnhealthy => "candidate_unhealthy",
3171 Self::Interrupted => "interrupted",
3172 Self::CutoverLost => "cutover_lost",
3173 }
3174 }
3175}
3176
3177impl fmt::Display for SuperviseError {
3178 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3179 match self {
3180 Self::InvalidSpec { reason } => write!(f, "invalid module spec: {reason}"),
3181 Self::Spawn {
3182 program,
3183 source,
3184 cgroup_path: Some(cgroup_path),
3185 } => write!(
3186 f,
3187 "failed to place module in cgroup '{}' while spawning '{}': {source}",
3188 cgroup_path.display(),
3189 program.display()
3190 ),
3191 Self::Spawn {
3192 program,
3193 source,
3194 cgroup_path: None,
3195 } => write!(
3196 f,
3197 "failed to spawn module '{}': {source}",
3198 program.display()
3199 ),
3200 Self::Cgroup { module_id, source } => {
3201 write!(
3202 f,
3203 "failed to prepare cgroup for module '{module_id}': {source}"
3204 )
3205 }
3206 Self::LaunchNonce { reason } => {
3207 write!(
3208 f,
3209 "failed to generate reserved-module launch nonce: {reason}"
3210 )
3211 }
3212 Self::Wait { module_id, source } => {
3213 write!(f, "failed to wait for module '{module_id}': {source}")
3214 }
3215 Self::Kill { module_id, source } => {
3216 write!(f, "failed to kill module '{module_id}': {source}")
3217 }
3218 Self::Forwarding(err) => write!(f, "forwarding error: {err}"),
3219 Self::Registry(err) => write!(f, "registry error: {err}"),
3220 Self::ReloadUnavailable { module_id, reason } => {
3221 write!(f, "reload unavailable for module '{module_id}': {reason}")
3222 }
3223 Self::Disabled { module_id } => {
3224 write!(
3225 f,
3226 "module '{module_id}' is disabled; enable it before restart or reload"
3227 )
3228 }
3229 Self::ReloadFailed { module_id, reason } => {
3230 write!(f, "reload failed for module '{module_id}': {reason}")
3231 }
3232 Self::RegistrationStillActive { module_id, waited } => write!(
3233 f,
3234 "module '{module_id}' registration remained active after waiting {waited:?}"
3235 ),
3236 Self::StatePoisoned { module_id } => match module_id {
3237 Some(module_id) => {
3238 write!(f, "supervisor state for module '{module_id}' was poisoned")
3239 }
3240 None => write!(f, "supervisor state was poisoned"),
3241 },
3242 Self::CommandClosed { module_id } => {
3243 write!(
3244 f,
3245 "supervisor command channel for module '{module_id}' is closed"
3246 )
3247 }
3248 Self::SwapInProgress { module_id } => write!(
3249 f,
3250 "module '{module_id}' is being swapped; retry once the swap has cut over or failed, or stop the module to abort the swap"
3251 ),
3252 Self::SwapRefused { module_id, reason } => match reason {
3253 SwapRefusal::OverlapExclusive => write!(
3254 f,
3255 "module '{module_id}' is declared overlap: \"exclusive\" (the default): two processes of it must not run at once, so it cannot be swapped; use a plain restart, or declare overlap: \"safe\" in its config if it really tolerates a second process"
3256 ),
3257 SwapRefusal::NotRegistered => write!(
3258 f,
3259 "module '{module_id}' is not registered, so there is no serving process to keep while a replacement warms; use a plain restart"
3260 ),
3261 SwapRefusal::ProtocolNone => write!(
3262 f,
3263 "module '{module_id}' is protocol: \"none\" and never registers, so a swap could never see its replacement become ready; use a plain restart"
3264 ),
3265 SwapRefusal::NotConfigured => write!(
3266 f,
3267 "module '{module_id}' cannot be swapped: the supervisor was built without the forwarding table or shared handle a swap needs"
3268 ),
3269 SwapRefusal::AlreadySwapping => {
3270 write!(f, "module '{module_id}' is already being swapped")
3271 }
3272 },
3273 Self::SwapFailed {
3274 module_id,
3275 arm,
3276 detail,
3277 ..
3278 } => write!(
3279 f,
3280 "swap of module '{module_id}' failed ({}): {detail}; the running process was left serving",
3281 arm.as_str()
3282 ),
3283 }
3284 }
3285}
3286
3287impl Error for SuperviseError {
3288 fn source(&self) -> Option<&(dyn Error + 'static)> {
3289 match self {
3290 Self::Spawn { source, .. }
3291 | Self::Cgroup { source, .. }
3292 | Self::Wait { source, .. }
3293 | Self::Kill { source, .. } => Some(source),
3294 Self::Forwarding(err) => Some(err),
3295 Self::Registry(err) => Some(err),
3296 Self::LaunchNonce { .. }
3297 | Self::InvalidSpec { .. }
3298 | Self::ReloadUnavailable { .. }
3299 | Self::Disabled { .. }
3300 | Self::ReloadFailed { .. }
3301 | Self::RegistrationStillActive { .. }
3302 | Self::StatePoisoned { .. }
3303 | Self::CommandClosed { .. }
3304 | Self::SwapInProgress { .. }
3305 | Self::SwapRefused { .. }
3306 | Self::SwapFailed { .. } => None,
3307 }
3308 }
3309}
3310
3311pub(crate) fn validate_spec(spec: &ModuleSpec) -> Result<(), SuperviseError> {
3312 if spec.module_id.trim().is_empty() {
3313 return Err(SuperviseError::InvalidSpec {
3314 reason: "module_id must not be empty".to_string(),
3315 });
3316 }
3317
3318 Ok(())
3319}
3320
3321#[derive(Debug, Default)]
3322struct HealthProbeRuntime {
3323 registered_connection: Option<crate::ConnectionId>,
3324 advertised: bool,
3325 next_probe_at: Option<Instant>,
3326 probe_index: u64,
3327}
3328
3329impl HealthProbeRuntime {
3330 fn refresh_registration(
3331 &mut self,
3332 spec: &ModuleSpec,
3333 runtime: &SupervisorRuntimeConfig,
3334 registry: &Registry,
3335 snapshot: &SharedSnapshot,
3336 ) {
3337 if spec.protocol == ModuleProtocol::None {
3349 self.registered_connection = None;
3350 self.advertised = false;
3351 self.next_probe_at = None;
3352 return;
3353 }
3354
3355 let registration = match registry.get_module(&spec.module_id) {
3356 Ok(registration) => registration,
3357 Err(err) => {
3358 warn!(module_id = %spec.module_id, error = %err, "health prober could not read registry");
3359 self.advertised = false;
3360 self.next_probe_at = None;
3361 return;
3362 }
3363 };
3364
3365 let Some(registration) = registration else {
3366 self.registered_connection = None;
3367 self.advertised = false;
3368 self.next_probe_at = None;
3369 return;
3370 };
3371
3372 let advertised = registration
3373 .control_ops
3374 .iter()
3375 .any(|op| op == MODULE_CONTROL_OP_HEALTH_CHECK);
3376 if !advertised {
3377 self.registered_connection = Some(registration.connection_id);
3378 self.advertised = false;
3379 self.next_probe_at = None;
3380 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3381 state.health.status = SupervisorHealthStatus::Unknown;
3382 state.health.consecutive_failures = 0;
3383 state.health.last_probe_ms = None;
3384 state.health.detail = None;
3385 state.health.metrics = None;
3386 });
3387 return;
3388 }
3389
3390 let reregistered = self.registered_connection != Some(registration.connection_id);
3391 self.registered_connection = Some(registration.connection_id);
3392 self.advertised = true;
3393 if reregistered || self.next_probe_at.is_none() {
3394 self.probe_index = 0;
3395 self.next_probe_at = Some(
3396 Instant::now() + jittered_health_delay(&spec.module_id, 0, runtime.health.cadence),
3397 );
3398 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3399 state.health.status = SupervisorHealthStatus::Unknown;
3400 state.health.consecutive_failures = 0;
3401 state.health.detail = None;
3402 state.health.metrics = None;
3403 });
3404 }
3405 }
3406
3407 fn wake_after(&self) -> Duration {
3408 if !self.advertised {
3409 return REGISTRY_RELEASE_POLL;
3410 }
3411 self.next_probe_at
3412 .map(|next| next.saturating_duration_since(Instant::now()))
3413 .unwrap_or(REGISTRY_RELEASE_POLL)
3414 }
3415
3416 fn due(&self) -> bool {
3417 self.advertised
3418 && self
3419 .next_probe_at
3420 .is_some_and(|next| Instant::now() >= next)
3421 }
3422
3423 fn schedule_next(&mut self, spec: &ModuleSpec, cadence: Duration) {
3424 self.probe_index = self.probe_index.wrapping_add(1);
3425 self.next_probe_at = Some(
3426 Instant::now() + jittered_health_delay(&spec.module_id, self.probe_index, cadence),
3427 );
3428 }
3429}
3430
3431#[derive(Debug)]
3466enum HealthProbeEvidence {
3467 LaneDead,
3469 NoAnswer,
3471 BadAnswer,
3473 Misconfigured,
3475}
3476
3477#[derive(Debug)]
3478struct HealthProbeError {
3479 evidence: HealthProbeEvidence,
3480 message: String,
3481}
3482
3483impl HealthProbeError {
3484 fn lane_dead(message: impl Into<String>) -> Self {
3485 Self::with(HealthProbeEvidence::LaneDead, message)
3486 }
3487
3488 fn no_answer(message: impl Into<String>) -> Self {
3489 Self::with(HealthProbeEvidence::NoAnswer, message)
3490 }
3491
3492 fn bad_answer(message: impl Into<String>) -> Self {
3493 Self::with(HealthProbeEvidence::BadAnswer, message)
3494 }
3495
3496 fn misconfigured(message: impl Into<String>) -> Self {
3497 Self::with(HealthProbeEvidence::Misconfigured, message)
3498 }
3499
3500 fn with(evidence: HealthProbeEvidence, message: impl Into<String>) -> Self {
3501 Self {
3502 evidence,
3503 message: message.into(),
3504 }
3505 }
3506
3507 #[allow(dead_code)]
3521 fn is_proof_of_death(&self) -> bool {
3522 matches!(self.evidence, HealthProbeEvidence::LaneDead)
3523 }
3524
3525 fn label(&self) -> &'static str {
3533 match self.evidence {
3534 HealthProbeEvidence::LaneDead => "lane-dead",
3535 HealthProbeEvidence::NoAnswer => "no-answer",
3536 HealthProbeEvidence::BadAnswer => "bad-answer",
3537 HealthProbeEvidence::Misconfigured => "daemon-misconfigured",
3538 }
3539 }
3540}
3541
3542impl fmt::Display for HealthProbeError {
3543 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3544 f.write_str(&self.message)
3545 }
3546}
3547
3548async fn run_health_probe_cycle(
3549 spec: &ModuleSpec,
3550 runtime: &SupervisorRuntimeConfig,
3551 registry: &Registry,
3552 process_liveness: &SupervisorProcessLiveness,
3553 snapshot: &SharedSnapshot,
3554 child: &mut Option<SupervisedChild>,
3555) {
3556 let now_ms = unix_ms_now();
3557 match probe_module_health(&spec.module_id, runtime, None).await {
3558 Ok(report) => {
3559 handle_health_report(
3560 spec,
3561 runtime,
3562 registry,
3563 process_liveness,
3564 snapshot,
3565 child,
3566 report,
3567 now_ms,
3568 )
3569 .await;
3570 }
3571 Err(err) => {
3572 handle_health_probe_failure(
3573 spec,
3574 runtime,
3575 registry,
3576 process_liveness,
3577 snapshot,
3578 child,
3579 err,
3580 now_ms,
3581 )
3582 .await;
3583 }
3584 }
3585}
3586
3587async fn probe_module_health(
3588 module_id: &str,
3589 runtime: &SupervisorRuntimeConfig,
3590 drain_deadline: Option<Instant>,
3591) -> Result<HealthReport, HealthProbeError> {
3592 let Some(forwarding) = runtime.forwarding.as_ref() else {
3593 return Err(HealthProbeError::misconfigured(
3594 "supervisor was not configured with a forwarding table",
3595 ));
3596 };
3597 let probe_started_at = Instant::now();
3598 let mut deadline = probe_started_at + runtime.health.deadline;
3599 if let Some(drain_deadline) = drain_deadline {
3600 deadline = deadline.min(drain_deadline);
3601 }
3602 let pending = if drain_deadline.is_some() {
3603 forwarding.begin_drain_health_probe_rpc_for(
3604 module_id,
3605 MODULE_CONTROL_OP_HEALTH_CHECK,
3606 probe_started_at,
3607 deadline,
3608 )
3609 } else {
3610 forwarding.begin_health_probe_rpc_for(
3611 module_id,
3612 MODULE_CONTROL_OP_HEALTH_CHECK,
3613 probe_started_at,
3614 deadline,
3615 )
3616 }
3617 .map_err(|err| {
3618 HealthProbeError::lane_dead(format!("failed to begin health.check RPC: {err}"))
3621 })?;
3622 await_health_probe(forwarding, pending, deadline, runtime.health.deadline).await
3623}
3624
3625async fn probe_endpoint_health(
3632 endpoint: crate::ModuleEndpointId,
3633 runtime: &SupervisorRuntimeConfig,
3634 deadline_cap: Option<Instant>,
3635) -> Result<HealthReport, HealthProbeError> {
3636 let Some(forwarding) = runtime.forwarding.as_ref() else {
3637 return Err(HealthProbeError::misconfigured(
3638 "supervisor was not configured with a forwarding table",
3639 ));
3640 };
3641 let probe_started_at = Instant::now();
3642 let mut deadline = probe_started_at + runtime.health.deadline;
3643 if let Some(cap) = deadline_cap {
3644 deadline = deadline.min(cap);
3645 }
3646 let pending = forwarding
3647 .begin_endpoint_health_probe_rpc_for(
3648 endpoint,
3649 MODULE_CONTROL_OP_HEALTH_CHECK,
3650 probe_started_at,
3651 deadline,
3652 )
3653 .map_err(|err| {
3654 HealthProbeError::lane_dead(format!("failed to begin health.check RPC: {err}"))
3655 })?;
3656 await_health_probe(forwarding, pending, deadline, runtime.health.deadline).await
3657}
3658
3659async fn await_health_probe(
3661 forwarding: &ForwardingTable,
3662 pending: PendingModuleControlRpc,
3663 deadline: Instant,
3664 probe_budget: Duration,
3665) -> Result<HealthReport, HealthProbeError> {
3666 let PendingModuleControlRpc {
3667 endpoint,
3668 module_sink,
3669 negotiated_ver,
3670 corr,
3671 receiver,
3672 } = pending;
3673 let body = serde_json::to_vec(&ModuleControlRequest::HealthCheck {}).map_err(|err| {
3674 HealthProbeError::misconfigured(format!("failed to encode health.check: {err}"))
3675 })?;
3676 let frame = Frame::build_with_version(
3677 negotiated_ver,
3678 FrameType::Request,
3679 control_flags(),
3680 0,
3681 0,
3682 corr,
3683 body,
3684 )
3685 .map_err(|err| {
3686 HealthProbeError::misconfigured(format!("failed to build health.check frame: {err}"))
3687 })?;
3688
3689 match timeout_at(deadline, module_sink.send(frame)).await {
3695 Ok(Ok(())) => {}
3696 Ok(Err(err)) => {
3697 let _ = forwarding.cancel_module_control_rpc(endpoint, corr);
3698 return Err(HealthProbeError::lane_dead(format!(
3701 "failed to send health.check: {err}"
3702 )));
3703 }
3704 Err(_elapsed) => {
3705 let _ = forwarding.cancel_module_control_rpc(endpoint, corr);
3706 return Err(HealthProbeError::no_answer(
3710 "health.check send timed out before enqueue (module egress full)",
3711 ));
3712 }
3713 }
3714
3715 match timeout_at(deadline, receiver).await {
3716 Ok(Ok(ModuleControlRpcOutcome::Response(response))) => {
3720 response.health_report().ok_or_else(|| {
3721 HealthProbeError::bad_answer("health.check RPC returned a non-health response")
3722 })
3723 }
3724 Ok(Ok(ModuleControlRpcOutcome::Rejected(body))) => Err(HealthProbeError::bad_answer(
3725 format!("health.check rejected: {}", body.message),
3726 )),
3727 Ok(Ok(ModuleControlRpcOutcome::ModuleGone(message))) => {
3728 Err(HealthProbeError::lane_dead(message))
3729 }
3730 Ok(Ok(ModuleControlRpcOutcome::MalformedResponse(message))) => {
3731 Err(HealthProbeError::bad_answer(message))
3732 }
3733 Ok(Ok(ModuleControlRpcOutcome::UnexpectedOp { expected, actual })) => {
3734 Err(HealthProbeError::bad_answer(format!(
3735 "expected module-control op '{expected}', got '{actual}'"
3736 )))
3737 }
3738 Ok(Ok(ModuleControlRpcOutcome::DeadlineElapsed)) => Err(HealthProbeError::bad_answer(
3742 "module answered health.check after its daemon deadline",
3743 )),
3744 Ok(Err(_)) => Err(HealthProbeError::misconfigured(
3745 "health.check waiter was canceled before the module responded",
3746 )),
3747 Err(_) => {
3748 let _ = forwarding.tombstone_health_probe_rpc(endpoint, corr);
3749 Err(HealthProbeError::no_answer(format!(
3750 "module did not answer health.check within {probe_budget:?}"
3751 )))
3752 }
3753 }
3754}
3755
3756#[allow(clippy::too_many_arguments)]
3757async fn handle_health_report(
3758 spec: &ModuleSpec,
3759 runtime: &SupervisorRuntimeConfig,
3760 registry: &Registry,
3761 process_liveness: &SupervisorProcessLiveness,
3762 snapshot: &SharedSnapshot,
3763 child: &mut Option<SupervisedChild>,
3764 report: HealthReport,
3765 now_ms: u64,
3766) {
3767 let status = supervisor_health_status(report.status);
3768 let detail = report.detail.clone();
3769 let metrics = truncate_health_metrics(report.metrics);
3770 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3771 state.health.status = status;
3772 state.health.last_probe_ms = Some(now_ms);
3773 state.health.detail = detail.clone();
3774 state.health.metrics = metrics.clone();
3775 state.health.consecutive_failures = 0;
3776 });
3777
3778 let action = match report.status {
3779 HealthStatus::Ok => return,
3780 HealthStatus::Degraded => runtime.health.on_degraded,
3781 HealthStatus::Failing => runtime.health.on_failing,
3782 };
3783 apply_l3_health_action(
3784 spec,
3785 runtime,
3786 registry,
3787 process_liveness,
3788 snapshot,
3789 child,
3790 status,
3791 detail.as_deref(),
3792 action,
3793 now_ms,
3794 )
3795 .await;
3796}
3797
3798#[allow(clippy::too_many_arguments)]
3799async fn handle_health_probe_failure(
3800 spec: &ModuleSpec,
3801 runtime: &SupervisorRuntimeConfig,
3802 registry: &Registry,
3803 process_liveness: &SupervisorProcessLiveness,
3804 snapshot: &SharedSnapshot,
3805 child: &mut Option<SupervisedChild>,
3806 err: HealthProbeError,
3807 now_ms: u64,
3808) {
3809 let threshold = runtime.health.failure_threshold.max(1);
3810 let mut failures = 0;
3811 let detail = format!("[{}] {err}", err.label());
3816 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3817 state.health.last_probe_ms = Some(now_ms);
3818 state.health.consecutive_failures = state.health.consecutive_failures.saturating_add(1);
3819 state.health.detail = Some(detail.clone());
3820 state.health.metrics = None;
3821 failures = state.health.consecutive_failures;
3822 });
3823
3824 if failures < threshold {
3825 warn!(
3826 module_id = %spec.module_id,
3827 consecutive_failures = failures,
3828 threshold,
3829 evidence = err.label(),
3830 detail = %detail,
3831 "health.check probe failed"
3832 );
3833 return;
3834 }
3835
3836 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3837 state.state = ModuleState::Unresponsive;
3838 state.health.status = SupervisorHealthStatus::Unresponsive;
3839 });
3840 if runtime.health.critical {
3844 error!(
3845 module_id = %spec.module_id,
3846 status = "unresponsive",
3847 evidence = err.label(),
3848 detail = %detail,
3849 "critical module health alert"
3850 );
3851 } else {
3852 warn!(
3853 module_id = %spec.module_id,
3854 status = "unresponsive",
3855 evidence = err.label(),
3856 detail = %detail,
3857 "module health threshold breached"
3858 );
3859 }
3860 if let Err(err) = health_restart_child(
3861 spec,
3862 runtime,
3863 registry,
3864 process_liveness,
3865 snapshot,
3866 child,
3867 SupervisorHealthStatus::Unresponsive,
3868 Some(&detail),
3869 now_ms,
3870 )
3871 .await
3872 {
3873 error!(module_id = %spec.module_id, error = %err, "health-triggered restart failed");
3874 }
3875}
3876
3877#[allow(clippy::too_many_arguments)]
3878async fn apply_l3_health_action(
3879 spec: &ModuleSpec,
3880 runtime: &SupervisorRuntimeConfig,
3881 registry: &Registry,
3882 process_liveness: &SupervisorProcessLiveness,
3883 snapshot: &SharedSnapshot,
3884 child: &mut Option<SupervisedChild>,
3885 status: SupervisorHealthStatus,
3886 detail: Option<&str>,
3887 action: HealthAction,
3888 now_ms: u64,
3889) {
3890 record_health_action(snapshot, &spec.module_id, action.to_string(), now_ms);
3891 match action {
3892 HealthAction::Report => {
3893 info!(
3894 module_id = %spec.module_id,
3895 status = ?status,
3896 detail,
3897 "module reported non-ok health"
3898 );
3899 }
3900 HealthAction::Alert => {
3901 error!(
3902 module_id = %spec.module_id,
3903 status = ?status,
3904 detail,
3905 "module health alert"
3906 );
3907 }
3908 HealthAction::Restart => {
3909 if let Err(err) = health_restart_child(
3910 spec,
3911 runtime,
3912 registry,
3913 process_liveness,
3914 snapshot,
3915 child,
3916 status,
3917 detail,
3918 now_ms,
3919 )
3920 .await
3921 {
3922 error!(module_id = %spec.module_id, error = %err, "health-triggered restart failed");
3923 }
3924 }
3925 }
3926}
3927
3928#[allow(clippy::too_many_arguments)]
3929async fn health_restart_child(
3930 spec: &ModuleSpec,
3931 runtime: &SupervisorRuntimeConfig,
3932 registry: &Registry,
3933 process_liveness: &SupervisorProcessLiveness,
3934 snapshot: &SharedSnapshot,
3935 child: &mut Option<SupervisedChild>,
3936 status: SupervisorHealthStatus,
3937 detail: Option<&str>,
3938 now_ms: u64,
3939) -> Result<(), SuperviseError> {
3940 let (enabled, schedule) = {
3941 let mut state = lock_snapshot(snapshot)?;
3942 let enabled = state.enabled;
3943 let schedule = if enabled {
3944 state.next_crash_restart(&runtime.restart_policy, Instant::now())
3945 } else {
3946 None
3947 };
3948 (enabled, schedule)
3949 };
3950
3951 if !enabled {
3952 return Err(SuperviseError::Disabled {
3953 module_id: spec.module_id.clone(),
3954 });
3955 }
3956
3957 if schedule.is_none() {
3958 record_health_action(snapshot, &spec.module_id, "disabled".to_string(), now_ms);
3959 error!(
3960 module_id = %spec.module_id,
3961 status = ?status,
3962 detail,
3963 max_restarts = runtime.restart_policy.max_restarts,
3964 window_secs = runtime.restart_policy.window.as_secs(),
3965 "health restart budget exhausted; disabling module"
3966 );
3967 let stop_notice = begin_forwarding_drain_if_configured(
3968 spec,
3969 runtime,
3970 registry,
3971 snapshot,
3972 Some(false),
3973 RouteCloseReason::Disable,
3974 )
3975 .await?;
3976 drain_optional_child(
3977 &spec.module_id,
3978 spec.protocol,
3979 stop_notice,
3980 registry,
3981 snapshot,
3982 &runtime.terminal_ring,
3983 &runtime.spawn_events,
3984 child,
3985 runtime.drain_timeout,
3986 ModuleState::Disabled,
3987 Some(false),
3988 )
3989 .await?;
3990 process_liveness.untrack_if_current(&spec.module_id, snapshot);
3991 return Ok(());
3992 }
3993
3994 let schedule = schedule.expect("a health restart must have a crash-restart schedule");
3995 let mut restart_count = 0;
3996 update_snapshot(snapshot, Some(&spec.module_id), |state| {
3997 restart_count = state.crash_restarts.len();
3998 state.state = ModuleState::Unresponsive;
3999 state.health.status = status;
4000 state.health.last_action = Some(HealthAction::Restart.to_string());
4001 state.health.last_action_ms = Some(now_ms);
4002 })?;
4003 warn!(
4004 module_id = %spec.module_id,
4005 status = ?status,
4006 detail,
4007 restart_count,
4008 restart_in_window = schedule.restart_in_window,
4009 delay_ms = schedule.delay.as_millis() as u64,
4010 "health-triggered module restart"
4011 );
4012
4013 let stop_notice = begin_forwarding_drain_if_configured(
4014 spec,
4015 runtime,
4016 registry,
4017 snapshot,
4018 Some(true),
4019 RouteCloseReason::Restart,
4020 )
4021 .await?;
4022 drain_optional_child(
4023 &spec.module_id,
4024 spec.protocol,
4025 stop_notice,
4026 registry,
4027 snapshot,
4028 &runtime.terminal_ring,
4029 &runtime.spawn_events,
4030 child,
4031 runtime.drain_timeout,
4032 ModuleState::Restarting,
4033 Some(true),
4034 )
4035 .await?;
4036 sleep(schedule.delay).await;
4037 if !respawn_still_pending(snapshot) {
4041 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4042 return Ok(());
4043 }
4044 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
4045 match spawn_and_mark_running(spec, runtime, snapshot) {
4046 Ok(next_child) => {
4047 *child = Some(next_child);
4048 Ok(())
4049 }
4050 Err(err) => {
4051 fail_snapshot(snapshot, Some(&spec.module_id), None);
4052 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4053 *child = None;
4054 Err(err)
4055 }
4056 }
4057}
4058
4059fn record_health_action(snapshot: &SharedSnapshot, module_id: &str, action: String, now_ms: u64) {
4060 let _ = update_snapshot(snapshot, Some(module_id), |state| {
4061 state.health.last_action = Some(action);
4062 state.health.last_action_ms = Some(now_ms);
4063 });
4064}
4065
4066fn supervisor_health_status(status: HealthStatus) -> SupervisorHealthStatus {
4067 match status {
4068 HealthStatus::Ok => SupervisorHealthStatus::Ok,
4069 HealthStatus::Degraded => SupervisorHealthStatus::Degraded,
4070 HealthStatus::Failing => SupervisorHealthStatus::Failing,
4071 }
4072}
4073
4074fn truncate_health_metrics(metrics: Option<Value>) -> Option<Value> {
4086 let metrics = metrics?;
4087 match serde_json::to_vec(&metrics) {
4088 Ok(encoded) if encoded.len() > MAX_HEALTH_METRICS_BYTES => Some(serde_json::json!({
4089 "truncated": true,
4090 "original_bytes": encoded.len(),
4091 })),
4092 Ok(_) | Err(_) => Some(metrics),
4093 }
4094}
4095
4096fn jittered_health_delay(module_id: &str, probe_index: u64, cadence: Duration) -> Duration {
4102 if cadence.is_zero() {
4103 return Duration::ZERO;
4104 }
4105 let cadence_ms = cadence.as_millis() as u64;
4106 if cadence_ms == 0 {
4122 return cadence;
4123 }
4124 let jitter_span = (cadence_ms / 10).max(1);
4139 let hash = module_id.as_bytes().iter().fold(
4140 probe_index.wrapping_mul(0x9E37_79B9_7F4A_7C15),
4141 |acc, byte| {
4142 acc.wrapping_mul(1099511628211)
4143 .wrapping_add(u64::from(*byte))
4144 },
4145 );
4146 cadence + Duration::from_millis(hash % jitter_span)
4147}
4148
4149#[cfg(test)]
4150mod tests {
4151 use super::*;
4152
4153 #[test]
4154 fn readding_a_module_clears_its_rescan_removal_tombstone() {
4155 let handle = SupervisorHandle::new();
4156 let module_id = "readded-tombstone";
4157 handle.record_rescan_removal(module_id);
4158 assert!(handle.removal_tombstone_age_ms(module_id).is_some());
4159
4160 handle.apply_identity_configuration(&ModuleSpec {
4161 module_id: module_id.to_string(),
4162 program: PathBuf::from("/test/module"),
4163 args: Vec::new(),
4164 env: Vec::new(),
4165 reserved: false,
4166 reserved_prefixes: Vec::new(),
4167 protocol: ModuleProtocol::Subc,
4168 overlap: Default::default(),
4169 });
4170
4171 assert!(
4172 handle.removal_tombstone_age_ms(module_id).is_none(),
4173 "a re-added module must not retain a stale removal tombstone"
4174 );
4175 }
4176
4177 fn stale_process_snapshot(state: ModuleState, enabled: bool) -> SharedSnapshot {
4178 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::new(state, enabled)));
4179 update_snapshot(&snapshot, Some("stale-process-facts"), |snapshot| {
4180 snapshot.process_alive = true;
4181 snapshot.pid = Some(41);
4182 snapshot.spawned_at_ms = Some(42);
4183 snapshot.spawned_from = Some(PathBuf::from("/spawned/module"));
4184 snapshot.spawned_file_identity = Some(SpawnedFileIdentity {
4185 device: 43,
4186 inode: 44,
4187 });
4188 })
4189 .unwrap();
4190 snapshot
4191 }
4192
4193 fn assert_snapshot_process_facts_cleared(snapshot: &SharedSnapshot) {
4194 let snapshot = lock_snapshot(snapshot).unwrap();
4195 assert!(!snapshot.process_alive);
4196 assert_eq!(snapshot.pid, None);
4197 assert_eq!(snapshot.spawned_at_ms, None);
4198 assert_eq!(snapshot.spawned_from, None);
4199 assert_eq!(snapshot.spawned_file_identity, None);
4200 }
4201
4202 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4203 async fn failed_enable_spawn_clears_preexisting_current_process_facts() {
4204 let supervisor = Supervisor::default();
4205 let mut runtime = supervisor.runtime_config();
4206 runtime.test_seed_stale_facts_before_enable_spawn = true;
4207 let snapshot = stale_process_snapshot(ModuleState::Disabled, false);
4208 let mut child = None;
4209 let spec = ModuleSpec {
4210 module_id: "failed-enable-clears-facts".to_string(),
4211 program: PathBuf::from("/definitely/missing/failed-enable-module"),
4212 args: Vec::new(),
4213 env: Vec::new(),
4214 reserved: false,
4215 reserved_prefixes: Vec::new(),
4216 protocol: ModuleProtocol::Subc,
4217 overlap: Default::default(),
4218 };
4219
4220 let result = set_child_enabled(
4221 &spec,
4222 &runtime,
4223 &supervisor.registry,
4224 &supervisor.process_liveness,
4225 &snapshot,
4226 &mut child,
4227 true,
4228 )
4229 .await;
4230
4231 assert!(matches!(result, Err(SuperviseError::Spawn { .. })));
4232 assert_eq!(lock_snapshot(&snapshot).unwrap().state, ModuleState::Failed);
4233 assert_snapshot_process_facts_cleared(&snapshot);
4234 }
4235
4236 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4237 async fn failed_reload_spawn_clears_current_process_facts() {
4238 let supervisor = Supervisor::default();
4239 let mut runtime = supervisor.runtime_config();
4240 runtime.restart_policy = RestartPolicy::new(0, Duration::ZERO);
4241 let snapshot = stale_process_snapshot(ModuleState::Running, true);
4242 let mut child = None;
4243 let spec = ModuleSpec {
4244 module_id: "failed-reload-clears-facts".to_string(),
4245 program: PathBuf::from("/unused/failed-reload-module"),
4246 args: Vec::new(),
4247 env: Vec::new(),
4248 reserved: false,
4249 reserved_prefixes: Vec::new(),
4250 protocol: ModuleProtocol::Subc,
4251 overlap: Default::default(),
4252 };
4253
4254 let result = handle_reload_spawn_failure(
4255 &spec,
4256 &runtime,
4257 &supervisor.process_liveness,
4258 &snapshot,
4259 &mut child,
4260 "forced reload spawn failure".to_string(),
4261 )
4262 .await;
4263
4264 assert!(matches!(result, Err(SuperviseError::ReloadFailed { .. })));
4265 assert_eq!(lock_snapshot(&snapshot).unwrap().state, ModuleState::Failed);
4266 assert_snapshot_process_facts_cleared(&snapshot);
4267 }
4268
4269 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4270 async fn dropping_a_module_with_an_active_monitor_clears_current_process_facts() {
4271 let supervisor = Supervisor::default();
4272 let snapshot = stale_process_snapshot(ModuleState::Running, true);
4273 let module = supervisor.supervised_module(
4274 ModuleSpec {
4275 module_id: "drop-clears-facts".to_string(),
4276 program: PathBuf::from("/unused/drop-module"),
4277 args: Vec::new(),
4278 env: Vec::new(),
4279 reserved: false,
4280 reserved_prefixes: Vec::new(),
4281 protocol: ModuleProtocol::Subc,
4282 overlap: Default::default(),
4283 },
4284 supervisor.runtime_config(),
4285 Arc::clone(&snapshot),
4286 None,
4287 );
4288 assert!(!module
4289 .inner
4290 .monitor
4291 .lock()
4292 .unwrap()
4293 .as_ref()
4294 .unwrap()
4295 .is_finished());
4296
4297 drop(module);
4298
4299 assert_eq!(
4300 lock_snapshot(&snapshot).unwrap().state,
4301 ModuleState::Stopped
4302 );
4303 assert_snapshot_process_facts_cleared(&snapshot);
4304 }
4305
4306 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4307 async fn configuration_update_does_not_replace_captured_running_process_facts() {
4308 let supervisor = Supervisor::default();
4309 let snapshot = stale_process_snapshot(ModuleState::Running, true);
4310 let initial = ModuleSpec {
4311 module_id: "rescan-preserves-spawn-facts".to_string(),
4312 program: PathBuf::from("/spawned/module"),
4313 args: Vec::new(),
4314 env: Vec::new(),
4315 reserved: false,
4316 reserved_prefixes: Vec::new(),
4317 protocol: ModuleProtocol::Subc,
4318 overlap: Default::default(),
4319 };
4320 let module = supervisor.supervised_module(
4321 initial.clone(),
4322 supervisor.runtime_config(),
4323 snapshot,
4324 None,
4325 );
4326 let before = module.status().unwrap();
4327 let mut replacement = initial;
4328 replacement.program = PathBuf::from("/rescanned/replacement-module");
4329
4330 module
4331 .update_configuration(replacement, HealthConfig::default(), None)
4332 .await
4333 .unwrap();
4334
4335 let after = module.status().unwrap();
4336 assert_eq!(after.pid, before.pid);
4337 assert_eq!(after.spawned_at_ms, before.spawned_at_ms);
4338 assert_eq!(after.spawned_from, before.spawned_from);
4339 drop(module);
4340 }
4341}
4342
4343fn unix_ms_now() -> u64 {
4344 SystemTime::now()
4345 .duration_since(UNIX_EPOCH)
4346 .map(|duration| duration.as_millis().min(u128::from(u64::MAX)) as u64)
4347 .unwrap_or(0)
4348}
4349
4350async fn supervise_loop(
4351 mut spec: ModuleSpec,
4352 mut runtime: SupervisorRuntimeConfig,
4353 registry: Arc<Registry>,
4354 process_liveness: Arc<SupervisorProcessLiveness>,
4355 snapshot: SharedSnapshot,
4356 mut child: Option<SupervisedChild>,
4357 mut commands: mpsc::Receiver<SupervisorCommand>,
4358) {
4359 let mut health_probe = HealthProbeRuntime::default();
4360 let mut pending_respawn: Option<Instant> = None;
4364 let mut requeued: VecDeque<SupervisorCommand> = VecDeque::new();
4367 loop {
4368 if let Some(command) = requeued.pop_front() {
4369 if !handle_supervisor_command(
4370 command,
4371 &mut spec,
4372 &mut runtime,
4373 ®istry,
4374 &process_liveness,
4375 &snapshot,
4376 &mut child,
4377 &mut commands,
4378 &mut requeued,
4379 )
4380 .await
4381 {
4382 return;
4383 }
4384 if child.is_some() || !respawn_still_pending(&snapshot) {
4385 pending_respawn = None;
4386 }
4387 continue;
4388 }
4389 if child.is_some() {
4390 health_probe.refresh_registration(&spec, &runtime, ®istry, &snapshot);
4391 let probe_sleep = sleep(health_probe.wake_after());
4392 tokio::pin!(probe_sleep);
4393 let active_child = child.as_mut().expect("child checked above");
4394 tokio::select! {
4395 wait_result = active_child.wait() => {
4396 let exit_report = match wait_result {
4405 Ok(status) => classify_reaped_child_exit(&snapshot, active_child, &status),
4406 Err(err) => {
4407 active_child.drain_stderr(&spec.module_id).await;
4408 fail_snapshot(&snapshot, Some(&spec.module_id), None);
4409 record_wait_error_terminal(
4415 &spec.module_id,
4416 &runtime.terminal_ring,
4417 &runtime.spawn_events,
4418 );
4419 untrack_if_registration_released(
4420 &process_liveness,
4421 ®istry,
4422 &spec.module_id,
4423 &snapshot,
4424 );
4425 error!(module_id = %spec.module_id, error = %err, "failed to wait for supervised module");
4426 child = None;
4427 continue;
4428 }
4429 };
4430 active_child.drain_stderr(&spec.module_id).await;
4431
4432 let next = on_child_exit(
4433 &spec,
4434 runtime.restart_policy,
4435 ®istry,
4436 &snapshot,
4437 &runtime.terminal_ring,
4438 &runtime.spawn_events,
4439 &runtime.child_roster,
4440 exit_report,
4441 ).await;
4442 active_child.release_roster();
4445 match next {
4446 NextAction::Stop { registration_released } => {
4447 if registration_released {
4448 process_liveness.untrack_if_current(&spec.module_id, &snapshot);
4449 }
4450 child = None;
4451 }
4452 NextAction::Restart { schedule } => {
4453 let delay = schedule.map_or(
4454 runtime.restart_policy.delay_for_restart(0),
4455 |schedule| schedule.delay,
4456 );
4457 if let Some(schedule) = schedule {
4458 log_crash_respawn(&spec.module_id, schedule);
4459 }
4460 child = None;
4468 pending_respawn = Some(Instant::now() + delay);
4469 }
4470 }
4471 }
4472 command = commands.recv() => {
4473 let Some(command) = command else {
4474 return;
4475 };
4476 if !handle_supervisor_command(
4477 command,
4478 &mut spec,
4479 &mut runtime,
4480 ®istry,
4481 &process_liveness,
4482 &snapshot,
4483 &mut child,
4484 &mut commands,
4485 &mut requeued,
4486 ).await {
4487 return;
4488 }
4489 }
4490 _ = &mut probe_sleep => {
4491 if health_probe.due() {
4492 run_health_probe_cycle(
4493 &spec,
4494 &runtime,
4495 ®istry,
4496 &process_liveness,
4497 &snapshot,
4498 &mut child,
4499 ).await;
4500 if child.is_some() {
4501 health_probe.schedule_next(&spec, runtime.health.cadence);
4502 }
4503 }
4504 }
4505 }
4506 } else if let Some(deadline) = pending_respawn {
4507 tokio::select! {
4508 _ = sleep_until(deadline) => {
4509 pending_respawn = None;
4510 if !respawn_still_pending(&snapshot) {
4514 continue;
4515 }
4516 if runtime.child_roster.is_closed() {
4521 let _ = update_snapshot(&snapshot, Some(&spec.module_id), |state| {
4522 state.state = ModuleState::Stopped;
4523 });
4524 debug!(module_id = %spec.module_id, "crash respawn cancelled by daemon shutdown");
4525 continue;
4526 }
4527 if let Err(err) = wait_for_registration_release(
4528 ®istry,
4529 &spec.module_id,
4530 REGISTRY_RELEASE_TIMEOUT,
4531 ).await {
4532 fail_snapshot(&snapshot, Some(&spec.module_id), None);
4533 error!(module_id = %spec.module_id, error = %err, "registration did not release before restart");
4534 continue;
4535 }
4536
4537 match spawn_and_mark_running(&spec, &runtime, &snapshot) {
4538 Ok(next_child) => {
4539 child = Some(next_child);
4540 debug!(module_id = %spec.module_id, "supervised module restarted after crash");
4541 }
4542 Err(err) => {
4543 fail_snapshot(&snapshot, Some(&spec.module_id), None);
4544 process_liveness.untrack_if_current(&spec.module_id, &snapshot);
4545 error!(module_id = %spec.module_id, error = %err, "failed to restart supervised module");
4546 }
4547 }
4548 }
4549 command = commands.recv() => {
4550 let Some(command) = command else {
4551 return;
4552 };
4553 if !handle_supervisor_command(
4554 command,
4555 &mut spec,
4556 &mut runtime,
4557 ®istry,
4558 &process_liveness,
4559 &snapshot,
4560 &mut child,
4561 &mut commands,
4562 &mut requeued,
4563 ).await {
4564 return;
4565 }
4566 if child.is_some() || !respawn_still_pending(&snapshot) {
4571 pending_respawn = None;
4572 }
4573 }
4574 }
4575 } else {
4576 let Some(command) = commands.recv().await else {
4577 return;
4578 };
4579 if !handle_supervisor_command(
4580 command,
4581 &mut spec,
4582 &mut runtime,
4583 ®istry,
4584 &process_liveness,
4585 &snapshot,
4586 &mut child,
4587 &mut commands,
4588 &mut requeued,
4589 )
4590 .await
4591 {
4592 return;
4593 }
4594 }
4595 }
4596}
4597
4598fn log_crash_respawn(module_id: &str, schedule: CrashRestartSchedule) {
4599 info!(
4600 module_id,
4601 restart_in_window = schedule.restart_in_window,
4602 delay_ms = schedule.delay.as_millis() as u64,
4603 "respawning after crash"
4604 );
4605}
4606
4607fn respawn_still_pending(snapshot: &SharedSnapshot) -> bool {
4613 matches!(
4614 lock_snapshot(snapshot),
4615 Ok(state) if state.enabled && state.state == ModuleState::Restarting
4616 )
4617}
4618
4619enum NextAction {
4620 Stop {
4621 registration_released: bool,
4622 },
4623 Restart {
4624 schedule: Option<CrashRestartSchedule>,
4625 },
4626}
4627
4628#[allow(clippy::too_many_arguments)]
4629async fn handle_supervisor_command(
4630 command: SupervisorCommand,
4631 spec: &mut ModuleSpec,
4632 runtime: &mut SupervisorRuntimeConfig,
4633 registry: &Registry,
4634 process_liveness: &SupervisorProcessLiveness,
4635 snapshot: &SharedSnapshot,
4636 child: &mut Option<SupervisedChild>,
4637 commands: &mut mpsc::Receiver<SupervisorCommand>,
4638 requeued: &mut VecDeque<SupervisorCommand>,
4639) -> bool {
4640 match command {
4641 SupervisorCommand::Drain { reply } => {
4642 let result = drain_optional_child(
4645 &spec.module_id,
4646 spec.protocol,
4647 StopNotice::NotSent,
4648 registry,
4649 snapshot,
4650 &runtime.terminal_ring,
4651 &runtime.spawn_events,
4652 child,
4653 runtime.drain_timeout,
4654 ModuleState::Stopped,
4655 None,
4656 )
4657 .await;
4658 let registration_released = result.is_ok();
4659 let _ = reply.send(result);
4660 if registration_released {
4661 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4662 }
4663 false
4664 }
4665 SupervisorCommand::Retire { reply } => {
4666 let result = async {
4667 let stop_notice = begin_forwarding_drain_if_configured(
4668 spec,
4669 runtime,
4670 registry,
4671 snapshot,
4672 None,
4673 RouteCloseReason::Disable,
4674 )
4675 .await?;
4676 drain_optional_child(
4677 &spec.module_id,
4678 spec.protocol,
4679 stop_notice,
4680 registry,
4681 snapshot,
4682 &runtime.terminal_ring,
4683 &runtime.spawn_events,
4684 child,
4685 runtime.drain_timeout,
4686 ModuleState::Stopped,
4687 None,
4688 )
4689 .await
4690 }
4691 .await;
4692 let registration_released = result.is_ok();
4693 let _ = reply.send(result);
4694 if registration_released {
4695 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4696 }
4697 false
4698 }
4699 SupervisorCommand::Restart {
4700 drain_timeout_ms,
4701 received_at_generation,
4702 queued_at,
4703 reply,
4704 } => {
4705 info!(
4709 module_id = %spec.module_id,
4710 queued_ms = u64::try_from(queued_at.elapsed().as_millis()).unwrap_or(u64::MAX),
4711 "restart command dequeued"
4712 );
4713 let validation = match lock_snapshot(snapshot) {
4725 Ok(state) if !state.enabled => Err(SuperviseError::Disabled {
4726 module_id: spec.module_id.clone(),
4727 }),
4728 Ok(_) => Ok(()),
4729 Err(err) => Err(err),
4730 };
4731 let initiated = validation.is_ok();
4732 let _ = reply.send(validation);
4733 let satisfied_by_generation = if initiated && child.is_some() {
4744 lock_snapshot(snapshot).ok().and_then(|state| {
4745 (state.spawn_generation > received_at_generation
4746 && !state.configuration_updated_since_spawn)
4747 .then_some(state.spawn_generation)
4748 })
4749 } else {
4750 None
4751 };
4752 if let Some(generation) = satisfied_by_generation {
4753 info!(
4754 module_id = %spec.module_id,
4755 received_at_generation,
4756 "restart already satisfied by generation {generation}; not restarting again"
4757 );
4758 } else if initiated {
4759 let drain_timeout = drain_timeout_ms
4762 .map(Duration::from_millis)
4763 .unwrap_or(runtime.drain_timeout);
4764 if let Err(err) = restart_child(
4765 spec,
4766 runtime,
4767 registry,
4768 process_liveness,
4769 snapshot,
4770 child,
4771 drain_timeout,
4772 )
4773 .await
4774 {
4775 warn!(
4776 module_id = %spec.module_id,
4777 error = %err,
4778 "operator restart failed after initiation ack; module state carries the outcome"
4779 );
4780 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
4781 state.state = ModuleState::Failed;
4782 clear_current_process_facts(state);
4783 });
4784 }
4785 }
4786 true
4787 }
4788 SupervisorCommand::Reload { reply } => {
4789 let result =
4790 reload_child(spec, runtime, registry, process_liveness, snapshot, child).await;
4791 let _ = reply.send(result);
4792 true
4793 }
4794 SupervisorCommand::SetEnabled { enabled, reply } => {
4795 let result = set_child_enabled(
4796 spec,
4797 runtime,
4798 registry,
4799 process_liveness,
4800 snapshot,
4801 child,
4802 enabled,
4803 )
4804 .await;
4805 let _ = reply.send(result);
4806 true
4807 }
4808 SupervisorCommand::UpdateConfiguration {
4809 spec: next_spec,
4810 health,
4811 drain_timeout_ms,
4812 reply,
4813 } => {
4814 if let Some(handle) = &runtime.supervisor_handle {
4815 handle.apply_identity_configuration(&next_spec);
4816 }
4817 *spec = next_spec;
4818 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
4819 state.configuration_updated_since_spawn = true;
4820 });
4821 runtime.health = health;
4822 runtime.drain_timeout = drain_timeout_ms
4823 .map(Duration::from_millis)
4824 .unwrap_or(runtime.default_drain_timeout);
4825 *runtime
4826 .effective_drain_timeout
4827 .lock()
4828 .unwrap_or_else(|poisoned| poisoned.into_inner()) = runtime.drain_timeout;
4829 let _ = reply.send(());
4830 true
4831 }
4832 SupervisorCommand::Swap {
4833 ready_timeout,
4834 reply,
4835 } => {
4836 let end = swap::run_swap(
4837 spec,
4838 runtime,
4839 registry,
4840 process_liveness,
4841 snapshot,
4842 child,
4843 commands,
4844 ready_timeout.unwrap_or(DEFAULT_SWAP_READY_TIMEOUT),
4845 reply,
4846 )
4847 .await;
4848 requeued.extend(end.requeue);
4849 true
4850 }
4851 }
4852}
4853
4854async fn restart_child(
4855 spec: &ModuleSpec,
4856 runtime: &SupervisorRuntimeConfig,
4857 registry: &Registry,
4858 process_liveness: &SupervisorProcessLiveness,
4859 snapshot: &SharedSnapshot,
4860 child: &mut Option<SupervisedChild>,
4861 drain_timeout: Duration,
4862) -> Result<(), SuperviseError> {
4863 if !lock_snapshot(snapshot)?.enabled {
4865 return Err(SuperviseError::Disabled {
4866 module_id: spec.module_id.clone(),
4867 });
4868 }
4869 let stop_notice = begin_forwarding_drain_with_timeout(
4870 spec,
4871 runtime,
4872 registry,
4873 snapshot,
4874 None,
4875 RouteCloseReason::Restart,
4876 drain_timeout,
4877 )
4878 .await?;
4879
4880 if child.is_some() {
4881 drain_optional_child(
4882 &spec.module_id,
4883 spec.protocol,
4884 stop_notice,
4885 registry,
4886 snapshot,
4887 &runtime.terminal_ring,
4888 &runtime.spawn_events,
4889 child,
4890 drain_timeout,
4891 ModuleState::Restarting,
4892 Some(true),
4893 )
4894 .await?;
4895 } else {
4896 update_snapshot(snapshot, Some(&spec.module_id), |state| {
4897 state.enabled = true;
4898 state.state = ModuleState::Restarting;
4899 clear_current_process_facts(state);
4900 })?;
4901 wait_for_registration_release(registry, &spec.module_id, REGISTRY_RELEASE_TIMEOUT).await?;
4902 }
4903
4904 reset_restart_count(snapshot, &spec.module_id)?;
4905 sleep(runtime.restart_policy.backoff).await;
4906 if !respawn_still_pending(snapshot) {
4909 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4910 return Ok(());
4911 }
4912 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
4913 match spawn_and_mark_running(spec, runtime, snapshot) {
4919 Ok(next_child) => {
4920 *child = Some(next_child);
4921 debug!(module_id = %spec.module_id, "supervised module restarted by operator request");
4922 Ok(())
4923 }
4924 Err(err) => {
4925 fail_snapshot(snapshot, Some(&spec.module_id), None);
4926 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4927 *child = None;
4928 Err(err)
4929 }
4930 }
4931}
4932
4933async fn reload_child(
4934 spec: &ModuleSpec,
4935 runtime: &SupervisorRuntimeConfig,
4936 registry: &Registry,
4937 process_liveness: &SupervisorProcessLiveness,
4938 snapshot: &SharedSnapshot,
4939 child: &mut Option<SupervisedChild>,
4940) -> Result<(), SuperviseError> {
4941 if !lock_snapshot(snapshot)?.enabled {
4943 return Err(SuperviseError::Disabled {
4944 module_id: spec.module_id.clone(),
4945 });
4946 }
4947 let stop_notice = begin_forwarding_drain(
4948 spec,
4949 runtime,
4950 registry,
4951 snapshot,
4952 Some(true),
4953 RouteCloseReason::Reload,
4954 )
4955 .await?;
4956
4957 if child.is_some() {
4958 drain_optional_child(
4959 &spec.module_id,
4960 spec.protocol,
4961 stop_notice,
4962 registry,
4963 snapshot,
4964 &runtime.terminal_ring,
4965 &runtime.spawn_events,
4966 child,
4967 runtime.drain_timeout,
4968 ModuleState::Restarting,
4969 Some(true),
4970 )
4971 .await?;
4972 } else {
4973 update_snapshot(snapshot, Some(&spec.module_id), |state| {
4974 state.enabled = true;
4975 state.state = ModuleState::Restarting;
4976 clear_current_process_facts(state);
4977 })?;
4978 wait_for_registration_release(registry, &spec.module_id, REGISTRY_RELEASE_TIMEOUT).await?;
4979 }
4980
4981 reset_restart_count(snapshot, &spec.module_id)?;
4982 sleep(runtime.restart_policy.backoff).await;
4983 if !respawn_still_pending(snapshot) {
4986 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4987 return Ok(());
4988 }
4989 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
4990 let next_child = match spawn_and_mark_running(spec, runtime, snapshot) {
4991 Ok(next_child) => next_child,
4992 Err(err) => {
4993 return handle_reload_spawn_failure(
4994 spec,
4995 runtime,
4996 process_liveness,
4997 snapshot,
4998 child,
4999 format!("new child failed to spawn: {err}"),
5000 )
5001 .await;
5002 }
5003 };
5004 *child = Some(next_child);
5005
5006 let wait_outcome = {
5007 let active_child = child.as_mut().expect("new reload child was just stored");
5008 wait_for_registration_after_reload(
5009 registry,
5010 &spec.module_id,
5011 snapshot,
5012 active_child,
5013 REGISTRY_RELEASE_TIMEOUT,
5014 )
5015 .await?
5016 };
5017
5018 match wait_outcome {
5019 RegistrationWaitOutcome::Registered => {
5020 debug!(module_id = %spec.module_id, "supervised module reloaded and registered");
5021 Ok(())
5022 }
5023 RegistrationWaitOutcome::Exited(exit_report) => {
5024 if let Some(active_child) = child.as_mut() {
5025 active_child.drain_stderr(&spec.module_id).await;
5026 }
5027 *child = None;
5028 handle_reload_child_registration_failure(
5029 spec,
5030 runtime,
5031 registry,
5032 process_liveness,
5033 snapshot,
5034 child,
5035 ReloadRegistrationFailure {
5036 exit_report: registration_failure_exit_report(exit_report),
5037 reason: "new child exited before registering".to_string(),
5038 },
5039 )
5040 .await
5041 }
5042 RegistrationWaitOutcome::TimedOut => {
5043 let mut timed_out_child = child
5044 .take()
5045 .expect("timed-out reload child is still running");
5046 timed_out_child
5047 .start_kill()
5048 .map_err(|source| SuperviseError::Kill {
5049 module_id: spec.module_id.clone(),
5050 source,
5051 })?;
5052 let status = timed_out_child
5053 .wait()
5054 .await
5055 .map_err(|source| SuperviseError::Wait {
5056 module_id: spec.module_id.clone(),
5057 source,
5058 })?;
5059 timed_out_child.drain_stderr(&spec.module_id).await;
5060 handle_reload_child_registration_failure(
5061 spec,
5062 runtime,
5063 registry,
5064 process_liveness,
5065 snapshot,
5066 child,
5067 ReloadRegistrationFailure {
5068 exit_report: registration_failure_exit_report(classify_reaped_child_exit(
5069 snapshot,
5070 &timed_out_child,
5071 &status,
5072 )),
5073 reason: format!(
5074 "new child did not register within {:?}",
5075 REGISTRY_RELEASE_TIMEOUT
5076 ),
5077 },
5078 )
5079 .await
5080 }
5081 }
5082}
5083
5084async fn set_child_enabled(
5085 spec: &ModuleSpec,
5086 runtime: &SupervisorRuntimeConfig,
5087 registry: &Registry,
5088 process_liveness: &SupervisorProcessLiveness,
5089 snapshot: &SharedSnapshot,
5090 child: &mut Option<SupervisedChild>,
5091 enabled: bool,
5092) -> Result<bool, SuperviseError> {
5093 let (current_enabled, current_state) = {
5094 let state = lock_snapshot(snapshot)?;
5095 (state.enabled, state.state)
5096 };
5097 let revive_terminal = enabled
5105 && current_enabled
5106 && child.is_none()
5107 && matches!(current_state, ModuleState::Failed | ModuleState::Stopped);
5108 if current_enabled == enabled && !revive_terminal {
5109 return Ok(false);
5110 }
5111
5112 if enabled {
5113 update_snapshot(snapshot, Some(&spec.module_id), |state| {
5114 state.enabled = true;
5115 state.state = ModuleState::Starting;
5116 clear_current_process_facts(state);
5117 })?;
5118 #[cfg(test)]
5119 if runtime.test_seed_stale_facts_before_enable_spawn {
5120 update_snapshot(snapshot, Some(&spec.module_id), |state| {
5121 state.process_alive = true;
5122 state.pid = Some(41);
5123 state.spawned_at_ms = Some(42);
5124 state.spawned_from = Some(PathBuf::from("/spawned/module"));
5125 state.spawned_file_identity = Some(SpawnedFileIdentity {
5126 device: 43,
5127 inode: 44,
5128 });
5129 })?;
5130 }
5131 wait_for_registration_release(registry, &spec.module_id, REGISTRY_RELEASE_TIMEOUT).await?;
5132 reset_restart_count(snapshot, &spec.module_id)?;
5133 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
5134 let next_child = match spawn_and_mark_running(spec, runtime, snapshot) {
5135 Ok(next_child) => next_child,
5136 Err(err) => {
5137 if let Err(state_err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5138 state.state = ModuleState::Failed;
5139 clear_current_process_facts(state);
5140 }) {
5141 error!(module_id = %spec.module_id, error = %state_err, "failed to record enable spawn failure");
5142 }
5143 process_liveness.untrack_if_current(&spec.module_id, snapshot);
5144 return Err(err);
5145 }
5146 };
5147 *child = Some(next_child);
5148 debug!(module_id = %spec.module_id, "supervised module enabled");
5149 Ok(true)
5150 } else {
5151 let stop_notice = begin_forwarding_drain_if_configured(
5152 spec,
5153 runtime,
5154 registry,
5155 snapshot,
5156 Some(false),
5157 RouteCloseReason::Disable,
5158 )
5159 .await?;
5160 drain_optional_child(
5161 &spec.module_id,
5162 spec.protocol,
5163 stop_notice,
5164 registry,
5165 snapshot,
5166 &runtime.terminal_ring,
5167 &runtime.spawn_events,
5168 child,
5169 runtime.drain_timeout,
5170 ModuleState::Disabled,
5171 Some(false),
5172 )
5173 .await?;
5174 debug!(module_id = %spec.module_id, "supervised module disabled");
5175 Ok(true)
5176 }
5177}
5178
5179#[allow(clippy::too_many_arguments)]
5180async fn on_child_exit(
5181 spec: &ModuleSpec,
5182 policy: RestartPolicy,
5183 registry: &Registry,
5184 snapshot: &SharedSnapshot,
5185 terminal_ring: &Arc<Mutex<TerminalRing>>,
5186 spawn_events: &SpawnEventFeed,
5187 roster: &ChildRoster,
5188 exit_report: ExitReport,
5189) -> NextAction {
5190 if roster.is_closed() {
5196 return on_child_exit_during_daemon_shutdown(
5197 spec,
5198 registry,
5199 snapshot,
5200 terminal_ring,
5201 spawn_events,
5202 exit_report,
5203 )
5204 .await;
5205 }
5206 match exit_report.kind {
5207 ExitKind::Clean => {
5208 info!(
5209 module_id = %spec.module_id,
5210 exit_code = ?exit_report.code,
5211 exit_signal = ?exit_report.signal,
5212 "supervised module exited cleanly"
5213 );
5214 if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5215 state.state = ModuleState::Stopped;
5216 clear_current_process_facts(state);
5217 state.last_exit = Some(exit_report.clone());
5218 }) {
5219 error!(module_id = %spec.module_id, error = %err, "failed to record clean module exit");
5220 }
5221 record_terminal(
5222 &spec.module_id,
5223 terminal_ring,
5224 spawn_events,
5225 &exit_report,
5226 TerminalDisposition::Stopped,
5227 );
5228 let registration_released = match wait_for_registration_release(
5229 registry,
5230 &spec.module_id,
5231 REGISTRY_RELEASE_TIMEOUT,
5232 )
5233 .await
5234 {
5235 Ok(()) => true,
5236 Err(err) => {
5237 warn!(module_id = %spec.module_id, error = %err, "registration still active after clean exit");
5238 false
5239 }
5240 };
5241 NextAction::Stop {
5242 registration_released,
5243 }
5244 }
5245 ExitKind::Crash => {
5246 warn!(
5247 module_id = %spec.module_id,
5248 exit_code = ?exit_report.code,
5249 exit_signal = ?exit_report.signal,
5250 "supervised module exited abnormally (crash)"
5251 );
5252 let mut restart_schedule = None;
5253 let mut disposition = TerminalDisposition::Disabled;
5254 let mut disposition_detail = None;
5258 let now = Instant::now();
5259 if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5260 clear_current_process_facts(state);
5261 state.last_exit = Some(exit_report.clone());
5262 if state.enabled {
5263 if let Some(schedule) = state.next_crash_restart(&policy, now) {
5264 state.state = ModuleState::Restarting;
5265 restart_schedule = Some(schedule);
5266 disposition = TerminalDisposition::Restarting;
5267 } else {
5268 state.state = ModuleState::Failed;
5269 disposition = TerminalDisposition::Failed;
5270 disposition_detail = Some(policy.budget_exhausted_detail());
5271 }
5272 } else {
5273 state.state = ModuleState::Disabled;
5274 disposition = TerminalDisposition::Disabled;
5275 }
5276 }) {
5277 error!(module_id = %spec.module_id, error = %err, "failed to record crashed module exit");
5278 return NextAction::Stop {
5279 registration_released: false,
5280 };
5281 }
5282 if disposition_detail.is_some() {
5283 error!(
5288 module_id = %spec.module_id,
5289 max_restarts = policy.max_restarts,
5290 window_secs = policy.window.as_secs(),
5291 "module stopped: {}",
5292 policy.budget_exhausted_detail()
5293 );
5294 }
5295 record_terminal_with_detail(
5296 &spec.module_id,
5297 terminal_ring,
5298 spawn_events,
5299 &exit_report,
5300 disposition,
5301 disposition_detail,
5302 );
5303
5304 if let Some(schedule) = restart_schedule {
5305 NextAction::Restart {
5306 schedule: Some(schedule),
5307 }
5308 } else {
5309 let registration_released = match wait_for_registration_release(
5310 registry,
5311 &spec.module_id,
5312 REGISTRY_RELEASE_TIMEOUT,
5313 )
5314 .await
5315 {
5316 Ok(()) => true,
5317 Err(err) => {
5318 warn!(module_id = %spec.module_id, error = %err, "registration still active after failed module");
5319 false
5320 }
5321 };
5322 NextAction::Stop {
5323 registration_released,
5324 }
5325 }
5326 }
5327 ExitKind::DeliberateSeverance => {
5328 warn!(
5329 module_id = %spec.module_id,
5330 exit_code = ?exit_report.code,
5331 exit_signal = ?exit_report.signal,
5332 "supervised module exited after deliberate connection severance"
5333 );
5334 let mut should_restart = false;
5335 let mut disposition = TerminalDisposition::Disabled;
5336 if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5337 clear_current_process_facts(state);
5338 state.last_exit = Some(exit_report.clone());
5339 state.lifetime_restarts += 1;
5340 if state.enabled {
5341 state.state = ModuleState::Restarting;
5342 should_restart = true;
5343 disposition = TerminalDisposition::Restarting;
5344 } else {
5345 state.state = ModuleState::Disabled;
5346 }
5347 }) {
5348 error!(module_id = %spec.module_id, error = %err, "failed to record deliberately severed module exit");
5349 return NextAction::Stop {
5350 registration_released: false,
5351 };
5352 }
5353 record_terminal(
5354 &spec.module_id,
5355 terminal_ring,
5356 spawn_events,
5357 &exit_report,
5358 disposition,
5359 );
5360
5361 if should_restart {
5362 NextAction::Restart { schedule: None }
5363 } else {
5364 let registration_released = match wait_for_registration_release(
5365 registry,
5366 &spec.module_id,
5367 REGISTRY_RELEASE_TIMEOUT,
5368 )
5369 .await
5370 {
5371 Ok(()) => true,
5372 Err(err) => {
5373 warn!(module_id = %spec.module_id, error = %err, "registration still active after deliberately severed module exit");
5374 false
5375 }
5376 };
5377 NextAction::Stop {
5378 registration_released,
5379 }
5380 }
5381 }
5382 }
5383}
5384
5385async fn on_child_exit_during_daemon_shutdown(
5386 spec: &ModuleSpec,
5387 registry: &Registry,
5388 snapshot: &SharedSnapshot,
5389 terminal_ring: &Arc<Mutex<TerminalRing>>,
5390 spawn_events: &SpawnEventFeed,
5391 exit_report: ExitReport,
5392) -> NextAction {
5393 info!(
5394 module_id = %spec.module_id,
5395 exit_code = ?exit_report.code,
5396 exit_signal = ?exit_report.signal,
5397 exit_kind = ?exit_report.kind,
5398 "supervised module exited during daemon shutdown; not restarting it"
5399 );
5400 if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5401 state.state = ModuleState::Stopped;
5402 clear_current_process_facts(state);
5403 state.last_exit = Some(exit_report.clone());
5404 }) {
5405 error!(module_id = %spec.module_id, error = %err, "failed to record module exit during daemon shutdown");
5406 }
5407 record_terminal(
5408 &spec.module_id,
5409 terminal_ring,
5410 spawn_events,
5411 &exit_report,
5412 TerminalDisposition::DaemonShutdown,
5413 );
5414 let registration_released =
5415 wait_for_registration_release(registry, &spec.module_id, REGISTRY_RELEASE_TIMEOUT)
5416 .await
5417 .is_ok();
5418 NextAction::Stop {
5419 registration_released,
5420 }
5421}
5422
5423fn record_wait_error_terminal(
5424 module_id: &str,
5425 terminal_ring: &Arc<Mutex<TerminalRing>>,
5426 spawn_events: &SpawnEventFeed,
5427) {
5428 record_terminal(
5429 module_id,
5430 terminal_ring,
5431 spawn_events,
5432 &wait_error_exit_report(),
5433 TerminalDisposition::Failed,
5434 );
5435}
5436
5437fn record_terminal(
5438 module_id: &str,
5439 terminal_ring: &Arc<Mutex<TerminalRing>>,
5440 spawn_events: &SpawnEventFeed,
5441 exit_report: &ExitReport,
5442 disposition: TerminalDisposition,
5443) {
5444 record_terminal_with_detail(
5445 module_id,
5446 terminal_ring,
5447 spawn_events,
5448 exit_report,
5449 disposition,
5450 None,
5451 );
5452}
5453
5454fn durable_terminal_history_of(
5458 terminal_ring: &Mutex<TerminalRing>,
5459 module_id: &str,
5460) -> subc_control::TerminalHistory {
5461 let read = terminal_ring
5462 .lock()
5463 .unwrap_or_else(|p| p.into_inner())
5464 .capture_durable_history();
5465 read.read(module_id)
5466}
5467
5468fn record_terminal_with_detail(
5469 module_id: &str,
5470 terminal_ring: &Arc<Mutex<TerminalRing>>,
5471 spawn_events: &SpawnEventFeed,
5472 exit_report: &ExitReport,
5473 disposition: TerminalDisposition,
5474 disposition_detail: Option<String>,
5475) {
5476 spawn_events.emit_exited(module_id, exit_report.code, exit_report.signal);
5477 let record = TerminalRecord {
5478 exit_code: exit_report.code,
5479 exit_signal: exit_report.signal,
5480 at_ms: exit_report.at_ms,
5481 disposition,
5482 exit_kind: exit_report.kind.into(),
5483 disposition_detail,
5484 };
5485 terminal_ring
5486 .lock()
5487 .unwrap_or_else(|poisoned| poisoned.into_inner())
5488 .record_exit(module_id, record);
5489}
5490
5491fn untrack_if_registration_released(
5492 process_liveness: &SupervisorProcessLiveness,
5493 registry: &Registry,
5494 module_id: &str,
5495 snapshot: &SharedSnapshot,
5496) {
5497 match registry.get_module(module_id) {
5498 Ok(None) => process_liveness.untrack_if_current(module_id, snapshot),
5499 Ok(Some(_)) => {}
5500 Err(err) => {
5501 warn!(module_id, error = %err, "could not determine whether supervisor liveness can be untracked");
5502 }
5503 }
5504}
5505
5506#[cfg(test)]
5520fn apply_wire_spawn_args(
5521 command: &mut Command,
5522 spec: &ModuleSpec,
5523 connection_file_path: Option<&std::path::Path>,
5524 handle: Option<&SupervisorHandle>,
5525) -> Result<(), SuperviseError> {
5526 apply_wire_spawn_args_for_role(
5527 command,
5528 spec,
5529 connection_file_path,
5530 handle,
5531 SpawnRole::Plain,
5532 )
5533}
5534
5535fn apply_wire_spawn_args_for_role(
5544 command: &mut Command,
5545 spec: &ModuleSpec,
5546 connection_file_path: Option<&std::path::Path>,
5547 handle: Option<&SupervisorHandle>,
5548 role: SpawnRole,
5549) -> Result<(), SuperviseError> {
5550 command.env(SUBC_MODULE_ID_ENV, &spec.module_id);
5551 if spec.protocol == ModuleProtocol::None {
5552 return Ok(());
5553 }
5554 if let Some(connection_file_path) = connection_file_path {
5555 command.arg(SUBC_ARG).arg(connection_file_path);
5556 }
5557
5558 let nonce = generate_launch_nonce()?;
5562 if let Some(handle) = handle {
5563 match role {
5564 SpawnRole::Plain => {
5565 handle.set_spawn_nonce(&spec.module_id, nonce.clone());
5566 if spec.reserved {
5567 handle.set_reserved_nonce(&spec.module_id, nonce.clone());
5568 }
5569 }
5570 SpawnRole::SwapCandidate => handle.open_swap(&spec.module_id, nonce.clone()),
5571 }
5572 }
5573 command.env(SUBC_LAUNCH_NONCE_ENV, nonce);
5574 Ok(())
5575}
5576
5577fn apply_child_env(command: &mut Command, spec: &ModuleSpec) {
5578 command.env_remove(CK_LOG_ENV);
5579 command.env_remove(SUBC_SPAWN_ROLE_ENV);
5586 for (key, value) in &spec.env {
5587 if matches!(
5591 key.as_str(),
5592 CAPTURE_MAX_FILE_MB_ENV | CAPTURE_KEEP_ENV | CAPTURE_MAX_AGE_DAYS_ENV
5593 ) || key == SUBC_SPAWN_ROLE_ENV
5594 {
5595 continue;
5596 }
5597 command.env(key, value);
5598 }
5599}
5600
5601#[derive(Debug, Clone, Copy, PartialEq, Eq)]
5604enum SpawnRole {
5605 Plain,
5606 SwapCandidate,
5607}
5608
5609fn apply_spawn_role(command: &mut Command, role: SpawnRole) {
5612 if role == SpawnRole::SwapCandidate {
5613 command.env(SUBC_SPAWN_ROLE_ENV, SPAWN_ROLE_SWAP_CANDIDATE);
5614 }
5615}
5616
5617fn spawn_child(
5618 spec: &ModuleSpec,
5619 connection_file_path: Option<&std::path::Path>,
5620 handle: Option<&SupervisorHandle>,
5621 ring: &Arc<Mutex<StderrRing>>,
5622 capture_logs_dir: Option<&std::path::Path>,
5623 roster: &ChildRoster,
5624 #[cfg(target_os = "linux")] cgroup_placement: Option<&subc_cgroup::Placement>,
5625) -> Result<SupervisedChild, SuperviseError> {
5626 spawn_child_in_slot(
5627 spec,
5628 connection_file_path,
5629 handle,
5630 ring,
5631 capture_logs_dir,
5632 roster,
5633 #[cfg(target_os = "linux")]
5634 cgroup_placement,
5635 SpawnRole::Plain,
5636 false,
5637 )
5638}
5639
5640#[allow(clippy::too_many_arguments)]
5653fn spawn_child_in_slot(
5654 spec: &ModuleSpec,
5655 connection_file_path: Option<&std::path::Path>,
5656 handle: Option<&SupervisorHandle>,
5657 ring: &Arc<Mutex<StderrRing>>,
5658 capture_logs_dir: Option<&std::path::Path>,
5659 roster: &ChildRoster,
5660 #[cfg(target_os = "linux")] cgroup_placement: Option<&subc_cgroup::Placement>,
5661 role: SpawnRole,
5662 alternate_slot: bool,
5663) -> Result<SupervisedChild, SuperviseError> {
5664 if roster.is_closed() {
5665 return Err(SuperviseError::Spawn {
5666 program: spec.program.clone(),
5667 source: io::Error::other("the daemon is shutting down; not starting a new process"),
5668 cgroup_path: None,
5669 });
5670 }
5671 #[cfg(target_os = "linux")]
5672 let cgroup_name = swap::cgroup_name(&spec.module_id, alternate_slot);
5673 #[cfg(not(target_os = "linux"))]
5674 let _ = alternate_slot;
5675 let mut command = Command::new(&spec.program);
5676 command.args(&spec.args);
5677 apply_child_env(&mut command, spec);
5707 apply_spawn_role(&mut command, role);
5708 apply_wire_spawn_args_for_role(&mut command, spec, connection_file_path, handle, role)?;
5709
5710 #[cfg(target_os = "linux")]
5711 let cgroup_path = cgroup_placement
5712 .map(|placement| placement.module_path(&cgroup_name))
5713 .transpose()
5714 .map_err(|source| SuperviseError::Cgroup {
5715 module_id: spec.module_id.clone(),
5716 source,
5717 })?;
5718 #[cfg(not(target_os = "linux"))]
5719 let cgroup_path: Option<PathBuf> = None;
5720 #[cfg(target_os = "linux")]
5721 if let Some(path) = &cgroup_path {
5722 if let Err(error) = apply_cgroup_placement(&mut command, spec, path) {
5723 if let Some(placement) = cgroup_placement {
5724 remove_module_cgroup(placement, &cgroup_name);
5725 }
5726 return Err(error);
5727 }
5728 }
5729
5730 let output_sink = if let Some(logs_dir) = capture_logs_dir {
5731 let path = logs_dir.join(format!("{}.stderr.log", spec.module_id));
5732 match ChildOutputSink::open(&path, capture_retention(spec)) {
5733 Ok(sink) => sink,
5734 Err(error) => {
5735 warn!(
5736 module_id = %spec.module_id,
5737 path = %path.display(),
5738 error = %error,
5739 "could not open child output capture file; forwarding to stderr"
5740 );
5741 ChildOutputSink::Stderr
5742 }
5743 }
5744 } else {
5745 ChildOutputSink::Stderr
5746 };
5747
5748 command.stdout(Stdio::piped());
5749 command.stderr(Stdio::piped());
5750 command.kill_on_drop(true);
5751 #[cfg(unix)]
5768 command.process_group(0);
5769 command.stdin(Stdio::null());
5770 let mut child = match command.spawn() {
5771 Ok(child) => child,
5772 Err(source) => {
5773 #[cfg(target_os = "linux")]
5774 if let Some(placement) = cgroup_placement {
5775 remove_module_cgroup(placement, &cgroup_name);
5776 }
5777 return Err(SuperviseError::Spawn {
5778 program: spec.program.clone(),
5779 source,
5780 cgroup_path,
5781 });
5782 }
5783 };
5784 let spawned_at_ms = unix_ms_now();
5785 let spawned_from = spec.program.clone();
5786 let spawned_file_identity = spawned_file_identity(&spawned_from);
5787 let pid = child.id().ok_or_else(|| SuperviseError::Spawn {
5788 program: spec.program.clone(),
5789 source: io::Error::other("spawned child exposed no live pid"),
5790 cgroup_path: cgroup_path.clone(),
5791 })?;
5792 let process_start_time = crate::provenance::process_start_time(pid);
5793 let process_identity = process_start_time.map(|start_time| ProcessIdentity { pid, start_time });
5794 let roster_guard = roster.admit(
5795 spec.module_id.clone(),
5796 pid,
5797 spec.protocol,
5798 process_start_time,
5799 );
5800 if roster.is_closed() {
5809 if let Err(error) = child.start_kill() {
5810 debug!(module_id = %spec.module_id, pid, %error, "kill of a process spawned during daemon shutdown failed; it may already have exited");
5811 }
5812 drop(roster_guard);
5813 return Err(SuperviseError::Spawn {
5814 program: spec.program.clone(),
5815 source: io::Error::other(
5816 "the daemon began shutting down while this process was starting; ended it",
5817 ),
5818 cgroup_path,
5819 });
5820 }
5821
5822 let stdout_pump = match child.stdout.take() {
5823 Some(stdout) => Some(tokio::spawn(pump_stdout_to(stdout, output_sink.clone()))),
5824 None => {
5825 warn!(
5826 module_id = %spec.module_id,
5827 "spawned child exposed no stdout pipe; file capture will be incomplete"
5828 );
5829 None
5830 }
5831 };
5832 let stderr_pump = match child.stderr.take() {
5833 Some(stderr) => {
5834 let generation = ring
5835 .lock()
5836 .unwrap_or_else(|poisoned| poisoned.into_inner())
5837 .begin_process();
5838 Some(StderrPump {
5839 task: tokio::spawn(pump_stderr_to(
5840 stderr,
5841 Arc::clone(ring),
5842 generation,
5843 output_sink,
5844 )),
5845 generation,
5846 })
5847 }
5848 None => {
5849 ring.lock()
5853 .unwrap_or_else(|poisoned| poisoned.into_inner())
5854 .mark_not_captured("stderr pipe was not available on spawn");
5855 warn!(
5856 module_id = %spec.module_id,
5857 "spawned child exposed no stderr pipe; tail will be unavailable"
5858 );
5859 None
5860 }
5861 };
5862
5863 Ok(SupervisedChild {
5864 child,
5865 #[cfg(target_os = "linux")]
5866 module_id: cgroup_name,
5867 #[cfg(target_os = "linux")]
5868 cgroup_placement: cgroup_placement.cloned(),
5869 stdout_pump,
5870 stderr_pump,
5871 stderr_ring: Arc::clone(ring),
5872 spawned_at_ms,
5873 spawned_from,
5874 spawned_file_identity,
5875 process_start_time,
5876 process_identity,
5877 pid,
5878 roster_guard: Some(roster_guard),
5879 })
5880}
5881
5882#[cfg(target_os = "linux")]
5883fn remove_module_cgroup(placement: &subc_cgroup::Placement, module_id: &str) {
5884 match placement.remove_module(module_id) {
5885 Ok(()) => debug!(module_id, "removed module cgroup after process exit"),
5886 Err(error) => warn!(
5887 module_id,
5888 error = %error,
5889 "could not remove module cgroup after process exit; continuing teardown"
5890 ),
5891 }
5892}
5893
5894#[cfg(target_os = "linux")]
5895fn apply_cgroup_placement(
5896 command: &mut Command,
5897 spec: &ModuleSpec,
5898 path: &std::path::Path,
5899) -> Result<(), SuperviseError> {
5900 subc_cgroup::apply(command, path).map_err(|source| SuperviseError::Cgroup {
5901 module_id: spec.module_id.clone(),
5902 source,
5903 })
5904}
5905
5906fn capture_retention(spec: &ModuleSpec) -> Retention {
5907 let defaults = Retention::default();
5908 let value = |name: &str| {
5909 spec.env
5910 .iter()
5911 .rev()
5912 .find_map(|(key, value)| (key == name).then_some(value.as_str()))
5913 };
5914 Retention {
5915 max_file_mb: value(CAPTURE_MAX_FILE_MB_ENV)
5916 .and_then(|value| value.parse().ok())
5917 .unwrap_or(defaults.max_file_mb),
5918 keep: value(CAPTURE_KEEP_ENV)
5919 .and_then(|value| value.parse().ok())
5920 .unwrap_or(defaults.keep),
5921 max_age_days: value(CAPTURE_MAX_AGE_DAYS_ENV)
5922 .and_then(|value| value.parse().ok())
5923 .unwrap_or(defaults.max_age_days),
5924 }
5925}
5926
5927fn generate_launch_nonce() -> Result<String, SuperviseError> {
5930 let mut bytes = [0u8; 32];
5931 getrandom::getrandom(&mut bytes).map_err(|source| SuperviseError::LaunchNonce {
5932 reason: source.to_string(),
5933 })?;
5934 let mut hex = String::with_capacity(64);
5935 for b in bytes {
5936 use std::fmt::Write;
5937 let _ = write!(hex, "{b:02x}");
5938 }
5939 Ok(hex)
5940}
5941
5942fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
5945 if a.len() != b.len() {
5946 return false;
5947 }
5948 let mut diff = 0u8;
5949 for (x, y) in a.iter().zip(b.iter()) {
5950 diff |= x ^ y;
5951 }
5952 diff == 0
5953}
5954
5955fn spawn_and_mark_running(
5956 spec: &ModuleSpec,
5957 runtime: &SupervisorRuntimeConfig,
5958 snapshot: &SharedSnapshot,
5959) -> Result<SupervisedChild, SuperviseError> {
5960 let child = spawn_child(
5961 spec,
5962 runtime.connection_file_path.as_deref(),
5963 runtime.supervisor_handle.as_ref(),
5964 &runtime.stderr_ring,
5965 runtime.capture_logs_dir.as_deref(),
5966 &runtime.child_roster,
5967 #[cfg(target_os = "linux")]
5968 runtime.cgroup_placement.as_ref(),
5969 )?;
5970 set_running(snapshot, &child, &spec.module_id, &runtime.spawn_events)?;
5971 Ok(child)
5972}
5973
5974enum RegistrationWaitOutcome {
5975 Registered,
5976 Exited(ExitReport),
5977 TimedOut,
5978}
5979
5980struct ReloadRegistrationFailure {
5981 exit_report: ExitReport,
5982 reason: String,
5983}
5984
5985#[derive(Debug, Clone, Copy, PartialEq, Eq)]
5986enum BusyGaugeObservation {
5987 Quiescent,
5988 Busy,
5989 Omitted,
5990}
5991
5992fn busy_gauge_observation(metrics: Option<&Value>, gauges: &[String]) -> BusyGaugeObservation {
5993 let Some(metrics) = metrics.and_then(Value::as_object) else {
5994 return BusyGaugeObservation::Omitted;
5995 };
5996 let mut sum = 0u128;
5997 for gauge in gauges {
5998 let Some(value) = metrics.get(gauge) else {
5999 return BusyGaugeObservation::Omitted;
6000 };
6001 let Some(value) = value.as_u64() else {
6002 return BusyGaugeObservation::Busy;
6003 };
6004 sum = sum.saturating_add(u128::from(value));
6005 }
6006 if sum == 0 {
6007 BusyGaugeObservation::Quiescent
6008 } else {
6009 BusyGaugeObservation::Busy
6010 }
6011}
6012
6013fn declared_busy_gauges(
6014 registry: &Registry,
6015 module_id: &str,
6016) -> Result<Vec<String>, SuperviseError> {
6017 busy_gauges_of(
6018 registry
6019 .get_module(module_id)
6020 .map_err(SuperviseError::Registry)?,
6021 )
6022}
6023
6024fn declared_busy_gauges_for_connection(
6028 registry: &Registry,
6029 connection_id: ConnectionId,
6030) -> Result<Vec<String>, SuperviseError> {
6031 busy_gauges_of(
6032 registry
6033 .get_module_by_connection(connection_id)
6034 .map_err(SuperviseError::Registry)?,
6035 )
6036}
6037
6038fn busy_gauges_of(
6039 registration: Option<crate::registry::ModuleRegistration>,
6040) -> Result<Vec<String>, SuperviseError> {
6041 let Some(registration) = registration else {
6042 return Ok(Vec::new());
6043 };
6044 let Some(self_signals) = registration.manifest.self_signals else {
6045 return Ok(Vec::new());
6046 };
6047
6048 let mut gauges = Vec::new();
6049 for declaration in self_signals {
6050 if declaration.kind != SelfSignalKind::Busy {
6051 continue;
6052 }
6053 match declaration.anchored_to {
6054 SignalAnchor::HealthGauges { gauges: declared } if !declared.is_empty() => {
6055 gauges.extend(declared)
6056 }
6057 _ => {
6058 gauges.push(String::new());
6061 }
6062 }
6063 }
6064 Ok(gauges)
6065}
6066
6067async fn wait_for_forwarding_quiescence(
6072 forwarding: &ForwardingTable,
6073 module_id: &str,
6074 runtime: &SupervisorRuntimeConfig,
6075 endpoint: crate::ModuleEndpointId,
6076 deadline: Instant,
6077 busy_gauges: &[String],
6078 scope: DrainScope,
6079) -> Result<bool, SuperviseError> {
6080 let mut gauges_quiescent = busy_gauges.is_empty();
6081 let mut next_probe_at = Instant::now();
6082 let mut omission_counted = false;
6083
6084 loop {
6085 let now = Instant::now();
6086 if !busy_gauges.is_empty() && now >= next_probe_at && now < deadline {
6087 let report = match scope {
6088 DrainScope::Active => probe_module_health(module_id, runtime, Some(deadline)).await,
6089 DrainScope::Endpoint(endpoint) => {
6090 probe_endpoint_health(endpoint, runtime, Some(deadline)).await
6091 }
6092 };
6093 gauges_quiescent = match report {
6094 Ok(report) => match busy_gauge_observation(report.metrics.as_ref(), busy_gauges) {
6095 BusyGaugeObservation::Quiescent => true,
6096 BusyGaugeObservation::Busy => false,
6097 BusyGaugeObservation::Omitted => {
6098 if !omission_counted {
6099 forwarding
6100 .counters()
6101 .increment_drains_with_undeclared_gauge();
6102 omission_counted = true;
6103 }
6104 false
6105 }
6106 },
6107 Err(err) => {
6108 warn!(
6109 module_id,
6110 error = %err,
6111 "drain health.check did not produce declared busy gauges; treating module as busy"
6112 );
6113 false
6114 }
6115 };
6116 next_probe_at = Instant::now() + runtime.health.cadence.max(REGISTRY_RELEASE_POLL);
6117 }
6118
6119 let in_flight = forwarding
6120 .endpoint_in_flight_count(endpoint)
6121 .map_err(SuperviseError::Forwarding)?;
6122 if in_flight == 0 && gauges_quiescent {
6123 return Ok(true);
6124 }
6125
6126 let now = Instant::now();
6127 if now >= deadline {
6128 return Ok(false);
6129 }
6130 let mut wait = deadline
6131 .saturating_duration_since(now)
6132 .min(REGISTRY_RELEASE_POLL);
6133 if !busy_gauges.is_empty() {
6134 wait = wait.min(next_probe_at.saturating_duration_since(now));
6135 }
6136 sleep(wait).await;
6137 }
6138}
6139
6140fn drained_after_quiescence_wait(wait_result: &Result<bool, SuperviseError>) -> bool {
6148 match wait_result {
6149 Ok(drained) => *drained,
6150 Err(_) => false,
6151 }
6152}
6153
6154fn send_route_goodbyes(forwarding: &ForwardingTable, released_routes: Vec<GoodbyeTarget>) {
6155 for released in released_routes {
6156 let frame = match Frame::build_with_version(
6157 released.negotiated_ver,
6158 FrameType::Goodbye,
6159 control_flags(),
6160 released.channel,
6161 released.epoch,
6162 0,
6163 Vec::new(),
6164 ) {
6165 Ok(frame) => frame,
6166 Err(err) => {
6167 warn!(
6168 route_channel = released.channel,
6169 error = %err,
6170 "failed to build supervisor drain route GOODBYE frame"
6171 );
6172 continue;
6173 }
6174 };
6175 if !released.close_on_delivery_failure() {
6176 crate::forwarding::send_module_route_goodbye(
6177 &forwarding.counters(),
6178 &released.sink,
6179 frame,
6180 released.module_id.as_deref(),
6181 "supervisor drain",
6182 );
6183 continue;
6184 }
6185 if let Err(err) = released.sink.try_send(frame) {
6186 warn!(
6187 target_connection_id = released.connection_id.get(),
6188 route_channel = released.channel,
6189 error = %err,
6190 "supervisor drain route GOODBYE was not delivered to client; closing target connection"
6191 );
6192 let _ = forwarding.escalate_client_delivery_failure(
6193 released.connection_id,
6194 released.channel,
6195 released.epoch,
6196 CloseReason::new(
6197 "route_goodbye_delivery_failed",
6198 format!(
6199 "failed to enqueue supervisor drain route GOODBYE for channel {}: {err}",
6200 released.channel
6201 ),
6202 ),
6203 crate::forwarding::UndeliveredFrame {
6204 module_id: released.module_id.as_deref(),
6205 sink: &released.sink,
6206 },
6207 );
6208 }
6209 }
6210}
6211
6212fn send_module_draining(
6213 module_id: &str,
6214 reason: RouteCloseReason,
6215 deadline_ms: u64,
6216 target: &ModuleDrainTarget,
6217) {
6218 let body = match serde_json::to_vec(&ModuleControlCommand::Draining {
6219 reason,
6220 deadline_ms,
6221 }) {
6222 Ok(body) => body,
6223 Err(err) => {
6224 warn!(
6225 module_id,
6226 error = %err,
6227 "failed to encode module draining command"
6228 );
6229 return;
6230 }
6231 };
6232 let frame = match Frame::build_with_version(
6233 target.negotiated_ver,
6234 FrameType::Push,
6235 control_flags(),
6236 0,
6237 0,
6238 0,
6239 body,
6240 ) {
6241 Ok(frame) => frame,
6242 Err(err) => {
6243 warn!(
6244 module_id,
6245 error = %err,
6246 "failed to build module draining command frame"
6247 );
6248 return;
6249 }
6250 };
6251 if let Err(err) = target.sink.try_send(frame) {
6252 warn!(
6253 module_id,
6254 target_connection_id = target.endpoint.connection_id.get(),
6255 error = %err,
6256 "module draining command was not delivered to peer"
6257 );
6258 }
6259}
6260
6261fn send_module_goodbye(module_id: &str, forwarding: &ForwardingTable, target: &ModuleDrainTarget) {
6262 let frame = match Frame::build_with_version(
6263 target.negotiated_ver,
6264 FrameType::Goodbye,
6265 control_flags(),
6266 0,
6267 0,
6268 0,
6269 Vec::new(),
6270 ) {
6271 Ok(frame) => frame,
6272 Err(err) => {
6273 warn!(
6274 module_id,
6275 error = %err,
6276 "failed to build supervisor drain module GOODBYE frame"
6277 );
6278 return;
6279 }
6280 };
6281 if let Err(err) = target.sink.try_send(frame) {
6282 warn!(
6283 module_id,
6284 target_connection_id = target.endpoint.connection_id.get(),
6285 error = %err,
6286 "supervisor drain module GOODBYE was not delivered to peer; closing module connection"
6287 );
6288 forwarding.request_connection_close(
6289 target.endpoint.connection_id,
6290 CloseReason::new(
6291 "module_goodbye_delivery_failed",
6292 format!("failed to enqueue supervisor drain module GOODBYE for module '{module_id}': {err}"),
6293 ),
6294 );
6295 }
6296}
6297
6298#[derive(Clone, Copy)]
6299struct ForwardingDrainContext<'a> {
6300 spec: &'a ModuleSpec,
6301 runtime: &'a SupervisorRuntimeConfig,
6302 registry: &'a Registry,
6303 scope: DrainScope,
6304}
6305
6306#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6308enum DrainScope {
6309 Active,
6312 Endpoint(crate::ModuleEndpointId),
6317}
6318
6319#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6327enum StopNotice {
6328 SentOverConnection,
6331 NoConnection,
6335 NotSent,
6339}
6340
6341async fn begin_forwarding_drain(
6342 spec: &ModuleSpec,
6343 runtime: &SupervisorRuntimeConfig,
6344 registry: &Registry,
6345 snapshot: &SharedSnapshot,
6346 enabled: Option<bool>,
6347 reason: RouteCloseReason,
6348) -> Result<StopNotice, SuperviseError> {
6349 let Some(forwarding) = runtime.forwarding.as_ref() else {
6350 return Err(SuperviseError::ReloadUnavailable {
6351 module_id: spec.module_id.clone(),
6352 reason: "supervisor was not configured with a forwarding table".to_string(),
6353 });
6354 };
6355
6356 begin_forwarding_drain_with(
6357 forwarding,
6358 ForwardingDrainContext {
6359 spec,
6360 runtime,
6361 registry,
6362 scope: DrainScope::Active,
6363 },
6364 snapshot,
6365 enabled,
6366 reason,
6367 runtime.drain_timeout,
6368 )
6369 .await
6370}
6371
6372async fn begin_forwarding_drain_if_configured(
6373 spec: &ModuleSpec,
6374 runtime: &SupervisorRuntimeConfig,
6375 registry: &Registry,
6376 snapshot: &SharedSnapshot,
6377 enabled: Option<bool>,
6378 reason: RouteCloseReason,
6379) -> Result<StopNotice, SuperviseError> {
6380 begin_forwarding_drain_with_timeout(
6381 spec,
6382 runtime,
6383 registry,
6384 snapshot,
6385 enabled,
6386 reason,
6387 runtime.drain_timeout,
6388 )
6389 .await
6390}
6391
6392async fn begin_forwarding_drain_with_timeout(
6396 spec: &ModuleSpec,
6397 runtime: &SupervisorRuntimeConfig,
6398 registry: &Registry,
6399 snapshot: &SharedSnapshot,
6400 enabled: Option<bool>,
6401 reason: RouteCloseReason,
6402 drain_timeout: Duration,
6403) -> Result<StopNotice, SuperviseError> {
6404 let Some(forwarding) = runtime.forwarding.as_ref() else {
6405 return Ok(StopNotice::NotSent);
6406 };
6407
6408 begin_forwarding_drain_with(
6409 forwarding,
6410 ForwardingDrainContext {
6411 spec,
6412 runtime,
6413 registry,
6414 scope: DrainScope::Active,
6415 },
6416 snapshot,
6417 enabled,
6418 reason,
6419 drain_timeout,
6420 )
6421 .await
6422}
6423
6424async fn begin_forwarding_drain_with(
6425 forwarding: &ForwardingTable,
6426 context: ForwardingDrainContext<'_>,
6427 snapshot: &SharedSnapshot,
6428 enabled: Option<bool>,
6429 reason: RouteCloseReason,
6430 drain_timeout: Duration,
6431) -> Result<StopNotice, SuperviseError> {
6432 let ForwardingDrainContext {
6433 spec,
6434 runtime,
6435 registry,
6436 scope,
6437 } = context;
6438 debug_assert_ne!(reason, RouteCloseReason::Crash);
6439 let terminal = matches!(reason, RouteCloseReason::Disable);
6440 let drain_started_at = Instant::now();
6441 let drain_deadline = drain_started_at + drain_timeout;
6442 let deadline_ms =
6443 unix_ms_now().saturating_add(u64::try_from(drain_timeout.as_millis()).unwrap_or(u64::MAX));
6444 let busy_gauges = match scope {
6445 DrainScope::Active => declared_busy_gauges(registry, &spec.module_id)?,
6446 DrainScope::Endpoint(endpoint) => {
6447 declared_busy_gauges_for_connection(registry, endpoint.connection_id)?
6448 }
6449 };
6450
6451 let gate_started = Instant::now();
6454 let drain_target = match scope {
6455 DrainScope::Active => forwarding.begin_module_drain(&spec.module_id, reason),
6456 DrainScope::Endpoint(endpoint) => forwarding.begin_endpoint_drain(endpoint, reason),
6457 }
6458 .map_err(SuperviseError::Forwarding)?;
6459 info!(
6464 module_id = %spec.module_id,
6465 ?reason,
6466 gate_ms = u64::try_from(gate_started.elapsed().as_millis()).unwrap_or(u64::MAX),
6467 connected = drain_target.is_some(),
6468 "module drain began; route admission closed"
6469 );
6470 if scope == DrainScope::Active {
6471 update_snapshot(snapshot, Some(&spec.module_id), |state| {
6472 state.state = ModuleState::Draining;
6473 state.draining_to_replace =
6474 matches!(reason, RouteCloseReason::Restart | RouteCloseReason::Reload);
6475 if let Some(enabled) = enabled {
6476 state.enabled = enabled;
6477 }
6478 })?;
6479 }
6480
6481 let Some(target) = drain_target.as_ref() else {
6482 return Ok(StopNotice::NoConnection);
6486 };
6487 {
6488 send_module_draining(&spec.module_id, reason, deadline_ms, target);
6489 let routes = forwarding
6490 .endpoint_routes(target.endpoint)
6491 .map_err(SuperviseError::Forwarding)?;
6492 let routes_notified = routes.len();
6493 crate::control::send_route_control_pushes(
6494 forwarding,
6495 routes.clone(),
6496 ClientControlPush::RouteClosing {
6497 module_id: spec.module_id.clone(),
6498 reason,
6499 },
6500 );
6501 send_route_goodbyes(forwarding, target.abandoned_bindings.clone());
6502
6503 let wait_result = wait_for_forwarding_quiescence(
6509 forwarding,
6510 &spec.module_id,
6511 runtime,
6512 target.endpoint,
6513 drain_deadline,
6514 &busy_gauges,
6515 scope,
6516 )
6517 .await;
6518 let drained = drained_after_quiescence_wait(&wait_result);
6519 if let Err(err) = &wait_result {
6520 error!(
6521 module_id = %spec.module_id,
6522 ?reason,
6523 error = %err,
6524 "forwarding quiescence wait failed after route.closing; forcing route.closed(drained: false) so the client is not left waiting on an unfulfilled promise"
6525 );
6526 } else if !drained {
6527 let holdouts = forwarding
6533 .endpoint_drain_holdouts(target.endpoint)
6534 .unwrap_or_default();
6535 warn!(
6536 module_id = %spec.module_id,
6537 waited = ?drain_timeout,
6538 ?reason,
6539 held_requests = holdouts.requests,
6540 held_routes = holdouts.routes,
6541 total_routes = holdouts.total_routes,
6542 top_connections = ?holdouts.top_connections,
6543 held = %holdouts
6546 .held
6547 .iter()
6548 .map(|(channel, corr)| format!("{channel}:{corr}"))
6549 .collect::<Vec<_>>()
6550 .join(","),
6551 "route drain timed out before request quiescence; forcing teardown"
6552 );
6553 }
6554 crate::control::send_route_control_pushes(
6555 forwarding,
6556 routes,
6557 ClientControlPush::RouteClosed {
6558 module_id: spec.module_id.clone(),
6559 reason,
6560 drained,
6561 abandoned: target.abandoned_bindings.len() as u32,
6562 excluded_subscriptions: target.excluded_subscriptions,
6563 terminal: Some(terminal),
6564 },
6565 );
6566 wait_result?;
6567
6568 let released_routes = match forwarding.release_module_endpoint_routes(target.endpoint) {
6574 Ok(routes) => routes,
6575 Err(err) => {
6576 warn!(
6577 module_id = %spec.module_id,
6578 ?reason,
6579 error = %err,
6580 "failed to release module endpoint routes after route.closed; module GOODBYE will still be sent"
6581 );
6582 send_module_goodbye(&spec.module_id, forwarding, target);
6583 return Err(SuperviseError::Forwarding(err));
6584 }
6585 };
6586 let route_goodbye_count = released_routes.len();
6587 send_route_goodbyes(forwarding, released_routes);
6588 send_module_goodbye(&spec.module_id, forwarding, target);
6589
6590 info!(
6596 module_id = %spec.module_id,
6597 ?reason,
6598 routes_notified,
6599 route_goodbyes = route_goodbye_count,
6600 abandoned_reservations = target.abandoned_bindings.len(),
6601 excluded_subscriptions = target.excluded_subscriptions,
6602 drained,
6603 "module drain complete; consumers notified via route.closing/route.closed pushes and per-route GOODBYE frames"
6604 );
6605 }
6606
6607 Ok(StopNotice::SentOverConnection)
6608}
6609
6610async fn wait_for_registration_after_reload(
6613 registry: &Registry,
6614 module_id: &str,
6615 snapshot: &SharedSnapshot,
6616 child: &mut SupervisedChild,
6617 wait: Duration,
6618) -> Result<RegistrationWaitOutcome, SuperviseError> {
6619 wait_for_slot_registration(
6620 registry,
6621 crate::registry::RegistrationSlot::Active(module_id),
6622 module_id,
6623 snapshot,
6624 child,
6625 wait,
6626 )
6627 .await
6628}
6629
6630async fn wait_for_slot_registration(
6638 registry: &Registry,
6639 slot: crate::registry::RegistrationSlot<'_>,
6640 module_id: &str,
6641 snapshot: &SharedSnapshot,
6642 child: &mut SupervisedChild,
6643 wait: Duration,
6644) -> Result<RegistrationWaitOutcome, SuperviseError> {
6645 let deadline = Instant::now() + wait;
6646 loop {
6647 if registry
6648 .registration(slot)
6649 .map_err(SuperviseError::Registry)?
6650 .is_some()
6651 {
6652 return Ok(RegistrationWaitOutcome::Registered);
6653 }
6654
6655 let now = Instant::now();
6656 if now >= deadline {
6657 return Ok(RegistrationWaitOutcome::TimedOut);
6658 }
6659 let remaining = deadline.saturating_duration_since(now);
6660 let poll = remaining.min(REGISTRY_RELEASE_POLL);
6661
6662 tokio::select! {
6663 wait_result = child.wait() => {
6664 let status = wait_result.map_err(|source| SuperviseError::Wait {
6665 module_id: module_id.to_string(),
6666 source,
6667 })?;
6668 return Ok(RegistrationWaitOutcome::Exited(classify_reaped_child_exit(
6669 snapshot,
6670 child,
6671 &status,
6672 )));
6673 }
6674 _ = sleep(poll) => {}
6675 }
6676 }
6677}
6678
6679fn registration_failure_exit_report(mut exit_report: ExitReport) -> ExitReport {
6680 if exit_report.kind != ExitKind::DeliberateSeverance {
6683 exit_report.kind = ExitKind::Crash;
6684 }
6685 exit_report
6686}
6687
6688async fn handle_reload_child_registration_failure(
6689 spec: &ModuleSpec,
6690 runtime: &SupervisorRuntimeConfig,
6691 registry: &Registry,
6692 process_liveness: &SupervisorProcessLiveness,
6693 snapshot: &SharedSnapshot,
6694 child: &mut Option<SupervisedChild>,
6695 failure: ReloadRegistrationFailure,
6696) -> Result<(), SuperviseError> {
6697 let ReloadRegistrationFailure {
6698 exit_report,
6699 reason,
6700 } = failure;
6701 match on_child_exit(
6702 spec,
6703 runtime.restart_policy,
6704 registry,
6705 snapshot,
6706 &runtime.terminal_ring,
6707 &runtime.spawn_events,
6708 &runtime.child_roster,
6709 exit_report,
6710 )
6711 .await
6712 {
6713 NextAction::Stop {
6714 registration_released,
6715 } => {
6716 if registration_released {
6717 process_liveness.untrack_if_current(&spec.module_id, snapshot);
6718 }
6719 }
6720 NextAction::Restart { schedule } => {
6721 let delay = schedule.map_or(runtime.restart_policy.delay_for_restart(0), |schedule| {
6722 schedule.delay
6723 });
6724 if let Some(schedule) = schedule {
6725 log_crash_respawn(&spec.module_id, schedule);
6726 }
6727 sleep(delay).await;
6728 if respawn_still_pending(snapshot) {
6732 if let Err(err) = wait_for_registration_release(
6733 registry,
6734 &spec.module_id,
6735 REGISTRY_RELEASE_TIMEOUT,
6736 )
6737 .await
6738 {
6739 fail_snapshot(snapshot, Some(&spec.module_id), None);
6740 process_liveness.untrack_if_current(&spec.module_id, snapshot);
6741 return Err(SuperviseError::ReloadFailed {
6742 module_id: spec.module_id.clone(),
6743 reason: format!(
6744 "{reason}; registration did not release before policy retry: {err}"
6745 ),
6746 });
6747 }
6748 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
6749 match spawn_and_mark_running(spec, runtime, snapshot) {
6750 Ok(next_child) => {
6751 *child = Some(next_child);
6752 }
6753 Err(err) => {
6754 fail_snapshot(snapshot, Some(&spec.module_id), None);
6755 process_liveness.untrack_if_current(&spec.module_id, snapshot);
6756 return Err(SuperviseError::ReloadFailed {
6757 module_id: spec.module_id.clone(),
6758 reason: format!("{reason}; policy retry spawn failed: {err}"),
6759 });
6760 }
6761 }
6762 }
6763 }
6764 }
6765
6766 Err(SuperviseError::ReloadFailed {
6767 module_id: spec.module_id.clone(),
6768 reason,
6769 })
6770}
6771
6772async fn handle_reload_spawn_failure(
6773 spec: &ModuleSpec,
6774 runtime: &SupervisorRuntimeConfig,
6775 process_liveness: &SupervisorProcessLiveness,
6776 snapshot: &SharedSnapshot,
6777 child: &mut Option<SupervisedChild>,
6778 reason: String,
6779) -> Result<(), SuperviseError> {
6780 let mut should_retry = false;
6781 let now = Instant::now();
6782 update_snapshot(snapshot, Some(&spec.module_id), |state| {
6783 clear_current_process_facts(state);
6784 if daemon_will_restart(state, &runtime.restart_policy, now) {
6785 state.record_crash_restart(&runtime.restart_policy, now);
6786 state.state = ModuleState::Restarting;
6787 should_retry = true;
6788 } else if state.enabled {
6789 state.state = ModuleState::Failed;
6790 } else {
6791 state.state = ModuleState::Disabled;
6792 }
6793 })?;
6794
6795 if should_retry {
6796 sleep(runtime.restart_policy.backoff).await;
6797 if respawn_still_pending(snapshot) {
6801 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
6802 match spawn_and_mark_running(spec, runtime, snapshot) {
6803 Ok(next_child) => {
6804 *child = Some(next_child);
6805 }
6806 Err(err) => {
6807 fail_snapshot(snapshot, Some(&spec.module_id), None);
6808 process_liveness.untrack_if_current(&spec.module_id, snapshot);
6809 return Err(SuperviseError::ReloadFailed {
6810 module_id: spec.module_id.clone(),
6811 reason: format!("{reason}; policy retry spawn failed: {err}"),
6812 });
6813 }
6814 }
6815 }
6816 } else {
6817 process_liveness.untrack_if_current(&spec.module_id, snapshot);
6818 }
6819
6820 Err(SuperviseError::ReloadFailed {
6821 module_id: spec.module_id.clone(),
6822 reason,
6823 })
6824}
6825
6826fn control_flags() -> Flags {
6827 Flags::new(false, Priority::Passive, false)
6828}
6829
6830#[allow(clippy::too_many_arguments)]
6831async fn drain_optional_child(
6832 module_id: &str,
6833 protocol: ModuleProtocol,
6834 stop_notice: StopNotice,
6835 registry: &Registry,
6836 snapshot: &SharedSnapshot,
6837 terminal_ring: &Arc<Mutex<TerminalRing>>,
6838 spawn_events: &SpawnEventFeed,
6839 child: &mut Option<SupervisedChild>,
6840 drain_timeout: Duration,
6841 final_state: ModuleState,
6842 enabled: Option<bool>,
6843) -> Result<(), SuperviseError> {
6844 if let Some(child) = child.take() {
6845 drain_child_to_state(
6846 module_id,
6847 protocol,
6848 stop_notice,
6849 registry,
6850 snapshot,
6851 terminal_ring,
6852 spawn_events,
6853 child,
6854 drain_timeout,
6855 final_state,
6856 enabled,
6857 )
6858 .await
6859 } else {
6860 update_snapshot(snapshot, Some(module_id), |state| {
6861 state.state = final_state;
6862 if let Some(enabled) = enabled {
6863 state.enabled = enabled;
6864 }
6865 clear_current_process_facts(state);
6866 })?;
6867 wait_for_registration_release(registry, module_id, REGISTRY_RELEASE_TIMEOUT).await
6868 }
6869}
6870
6871#[allow(clippy::too_many_arguments)]
6872async fn drain_child_to_state(
6873 module_id: &str,
6874 protocol: ModuleProtocol,
6875 stop_notice: StopNotice,
6876 registry: &Registry,
6877 snapshot: &SharedSnapshot,
6878 terminal_ring: &Arc<Mutex<TerminalRing>>,
6879 spawn_events: &SpawnEventFeed,
6880 mut child: SupervisedChild,
6881 drain_timeout: Duration,
6882 final_state: ModuleState,
6883 enabled: Option<bool>,
6884) -> Result<(), SuperviseError> {
6885 update_snapshot(snapshot, Some(module_id), |state| {
6886 state.state = ModuleState::Draining;
6887 state.draining_to_replace = final_state == ModuleState::Restarting;
6888 if let Some(enabled) = enabled {
6889 state.enabled = enabled;
6890 }
6891 })?;
6892
6893 if stop_notice != StopNotice::SentOverConnection {
6904 if protocol == ModuleProtocol::Subc && stop_notice == StopNotice::NoConnection {
6905 info!(
6906 module_id,
6907 pid = child.pid,
6908 budget_ms = u64::try_from(drain_timeout.as_millis()).unwrap_or(u64::MAX),
6909 "module has no connection yet; requesting stop by signal"
6910 );
6911 }
6912 request_graceful_stop(module_id, &child);
6913 }
6914
6915 let exit_report = match timeout(drain_timeout, child.wait()).await {
6916 Ok(Ok(status)) => classify_reaped_child_exit(snapshot, &child, &status),
6917 Ok(Err(source)) => {
6918 fail_snapshot(snapshot, Some(module_id), None);
6919 return Err(SuperviseError::Wait {
6920 module_id: module_id.to_string(),
6921 source,
6922 });
6923 }
6924 Err(_) => {
6925 warn!(
6938 module_id,
6939 pid = child.pid,
6940 budget_ms = u64::try_from(drain_timeout.as_millis()).unwrap_or(u64::MAX),
6941 reason = ?final_state,
6942 ?stop_notice,
6943 "drain budget expired before the module exited; killing it"
6944 );
6945 child.start_kill().map_err(|source| {
6946 fail_snapshot(snapshot, Some(module_id), None);
6947 SuperviseError::Kill {
6948 module_id: module_id.to_string(),
6949 source,
6950 }
6951 })?;
6952 let status = child.wait().await.map_err(|source| {
6953 fail_snapshot(snapshot, Some(module_id), None);
6954 SuperviseError::Wait {
6955 module_id: module_id.to_string(),
6956 source,
6957 }
6958 })?;
6959 classify_reaped_child_exit(snapshot, &child, &status)
6960 }
6961 };
6962
6963 update_snapshot(snapshot, Some(module_id), |state| {
6964 state.state = final_state;
6965 if let Some(enabled) = enabled {
6966 state.enabled = enabled;
6967 }
6968 clear_current_process_facts(state);
6969 state.last_exit = Some(exit_report.clone());
6970 if exit_report.kind == ExitKind::DeliberateSeverance {
6971 state.lifetime_restarts += 1;
6972 }
6973 })?;
6974 record_terminal(
6975 module_id,
6976 terminal_ring,
6977 spawn_events,
6978 &exit_report,
6979 terminal_disposition(final_state),
6980 );
6981 child.drain_stderr(module_id).await;
6982
6983 wait_for_registration_release(registry, module_id, REGISTRY_RELEASE_TIMEOUT).await
6984}
6985
6986#[cfg(unix)]
7006fn request_graceful_stop(module_id: &str, child: &SupervisedChild) {
7007 let Some(pid) = child
7008 .id()
7009 .and_then(|pid| i32::try_from(pid).ok())
7010 .and_then(rustix::process::Pid::from_raw)
7011 else {
7012 debug!(
7013 module_id,
7014 "no pid to signal for teardown; falling through to the drain wait"
7015 );
7016 return;
7017 };
7018 match rustix::process::kill_process(pid, rustix::process::Signal::TERM) {
7019 Ok(()) => debug!(
7020 module_id,
7021 "sent SIGTERM to a module nothing else asked to stop"
7022 ),
7023 Err(err) => debug!(
7024 module_id,
7025 error = %err,
7026 "SIGTERM to module failed; the drain wait and kill still apply"
7027 ),
7028 }
7029}
7030
7031#[cfg(not(unix))]
7039fn request_graceful_stop(module_id: &str, _child: &SupervisedChild) {
7040 debug!(
7041 module_id,
7042 "no graceful stop signal exists on this platform; teardown of a module nothing asked to stop waits, then kills"
7043 );
7044}
7045
7046fn terminal_disposition(final_state: ModuleState) -> TerminalDisposition {
7047 match final_state {
7048 ModuleState::Stopped => TerminalDisposition::Stopped,
7049 ModuleState::Disabled => TerminalDisposition::Disabled,
7050 ModuleState::Restarting => TerminalDisposition::Restarting,
7051 ModuleState::Failed => TerminalDisposition::Failed,
7052 ModuleState::Starting
7053 | ModuleState::Running
7054 | ModuleState::Unresponsive
7055 | ModuleState::Draining => {
7056 unreachable!("terminal exits only finish in terminal or restarting states")
7057 }
7058 }
7059}
7060
7061async fn wait_for_registration_release(
7064 registry: &Registry,
7065 module_id: &str,
7066 wait: Duration,
7067) -> Result<(), SuperviseError> {
7068 wait_for_slot_registration_release(
7069 registry,
7070 crate::registry::RegistrationSlot::Active(module_id),
7071 wait,
7072 )
7073 .await
7074}
7075
7076async fn wait_for_slot_registration_release(
7084 registry: &Registry,
7085 slot: crate::registry::RegistrationSlot<'_>,
7086 wait: Duration,
7087) -> Result<(), SuperviseError> {
7088 let deadline = Instant::now() + wait;
7089 let mut release_events = registration_release_events().subscribe();
7090 let still_active = |registration: &crate::registry::ModuleRegistration| {
7091 SuperviseError::RegistrationStillActive {
7092 module_id: registration.manifest.module_id.clone(),
7093 waited: wait,
7094 }
7095 };
7096 loop {
7097 let _observed_generation = *release_events.borrow_and_update();
7098 let Some(registration) = registry
7099 .registration(slot)
7100 .map_err(SuperviseError::Registry)?
7101 else {
7102 return Ok(());
7103 };
7104
7105 let now = Instant::now();
7106 if now >= deadline {
7107 return Err(still_active(®istration));
7108 }
7109
7110 let remaining = deadline.saturating_duration_since(now);
7111 match timeout(remaining, release_events.changed()).await {
7112 Ok(Ok(())) | Ok(Err(_)) => {}
7113 Err(_) => return Err(still_active(®istration)),
7114 }
7115 }
7116}
7117
7118#[cfg(test)]
7119mod slot_registration_wait_tests {
7120 use super::*;
7121 use crate::registry::{ConnectionId, RegistrationSlot};
7122 use subc_protocol::manifest::ModuleManifest;
7123
7124 const INCUMBENT: u64 = 1;
7125 const CANDIDATE: u64 = 2;
7126
7127 fn swapped_registry() -> Arc<Registry> {
7128 let registry = Arc::new(Registry::default());
7129 let manifest = ModuleManifest::builder("m", "0.1.0").build();
7130 registry
7131 .register_with_control_ops(
7132 manifest.clone(),
7133 1,
7134 ConnectionId::new(INCUMBENT),
7135 Vec::new(),
7136 )
7137 .unwrap();
7138 registry
7139 .register_candidate_with_control_ops(
7140 manifest,
7141 1,
7142 ConnectionId::new(CANDIDATE),
7143 Vec::new(),
7144 )
7145 .unwrap();
7146 registry
7147 }
7148
7149 #[tokio::test]
7153 async fn incumbent_release_is_awaited_by_connection_not_by_module_id() {
7154 let registry = swapped_registry();
7155 registry.promote_candidate("m").unwrap().unwrap();
7156
7157 assert!(matches!(
7158 wait_for_registration_release(®istry, "m", Duration::from_millis(50)).await,
7159 Err(SuperviseError::RegistrationStillActive { .. })
7160 ));
7161
7162 assert!(matches!(
7164 wait_for_slot_registration_release(
7165 ®istry,
7166 RegistrationSlot::Connection(ConnectionId::new(INCUMBENT)),
7167 Duration::from_millis(50),
7168 )
7169 .await,
7170 Err(SuperviseError::RegistrationStillActive { .. })
7171 ));
7172
7173 let releaser = Arc::clone(®istry);
7174 let release = tokio::spawn(async move {
7175 sleep(Duration::from_millis(20)).await;
7176 releaser
7177 .deregister_connection(ConnectionId::new(INCUMBENT))
7178 .unwrap();
7179 notify_registration_release();
7180 });
7181 wait_for_slot_registration_release(
7182 ®istry,
7183 RegistrationSlot::Connection(ConnectionId::new(INCUMBENT)),
7184 Duration::from_secs(5),
7185 )
7186 .await
7187 .expect("the incumbent's own registration is released");
7188 release.await.unwrap();
7189 assert!(registry.get_module("m").unwrap().is_some());
7190 }
7191
7192 #[tokio::test]
7195 async fn candidate_slot_wait_ignores_the_incumbents_registration() {
7196 let registry = swapped_registry();
7197 assert!(matches!(
7198 wait_for_slot_registration_release(
7199 ®istry,
7200 RegistrationSlot::Candidate("m"),
7201 Duration::from_millis(50),
7202 )
7203 .await,
7204 Err(SuperviseError::RegistrationStillActive { .. })
7205 ));
7206 registry
7207 .deregister_connection(ConnectionId::new(CANDIDATE))
7208 .unwrap();
7209 wait_for_slot_registration_release(
7210 ®istry,
7211 RegistrationSlot::Candidate("m"),
7212 Duration::from_millis(50),
7213 )
7214 .await
7215 .expect("a candidate slot with no candidate is released");
7216 assert!(registry
7217 .registration(RegistrationSlot::Active("m"))
7218 .unwrap()
7219 .is_some());
7220 }
7221}
7222
7223fn classify_exit(status: &ExitStatus) -> ExitReport {
7224 ExitReport {
7225 kind: if status.success() {
7226 ExitKind::Clean
7227 } else {
7228 ExitKind::Crash
7229 },
7230 code: status.code(),
7231 signal: exit_signal(status),
7232 at_ms: unix_ms_now(),
7233 }
7234}
7235
7236fn wait_error_exit_report() -> ExitReport {
7242 ExitReport {
7243 kind: ExitKind::Crash,
7244 code: None,
7245 signal: None,
7246 at_ms: unix_ms_now(),
7247 }
7248}
7249
7250#[cfg(unix)]
7251fn exit_signal(status: &ExitStatus) -> Option<i32> {
7252 use std::os::unix::process::ExitStatusExt;
7253
7254 status.signal()
7255}
7256
7257#[cfg(not(unix))]
7258fn exit_signal(_status: &ExitStatus) -> Option<i32> {
7259 None
7260}
7261
7262fn reset_restart_count(snapshot: &SharedSnapshot, module_id: &str) -> Result<(), SuperviseError> {
7268 update_snapshot(snapshot, Some(module_id), |state| {
7269 state.clear_crash_restarts();
7270 })
7271}
7272
7273fn set_running(
7274 snapshot: &SharedSnapshot,
7275 child: &SupervisedChild,
7276 module_id: &str,
7277 spawn_events: &SpawnEventFeed,
7278) -> Result<(), SuperviseError> {
7279 let mut state = snapshot.lock().map_err(|_| SuperviseError::StatePoisoned {
7280 module_id: Some(module_id.to_string()),
7281 })?;
7282 state.spawn_generation = spawn_events.emit_spawned(module_id, child.pid, child.spawned_at_ms);
7283 state.in_alternate_slot = false;
7286 state.configuration_updated_since_spawn = false;
7287 state.state = ModuleState::Running;
7288 state.enabled = true;
7289 state.process_alive = true;
7290 state.pid = child.id();
7291 state.spawned_at_ms = Some(child.spawned_at_ms);
7292 state.spawned_from = Some(child.spawned_from.clone());
7293 state.spawned_file_identity = child.spawned_file_identity;
7294 state.process_start_time = child.process_start_time;
7295 Ok(())
7296}
7297
7298fn clear_current_process_facts(state: &mut SupervisorSnapshot) {
7299 state.process_alive = false;
7300 state.pid = None;
7301 state.spawned_at_ms = None;
7302 state.spawned_from = None;
7303 state.spawned_file_identity = None;
7304 state.process_start_time = None;
7305 state.deliberate_severance = None;
7306}
7307
7308#[cfg(test)]
7309fn record_deliberate_severance(
7310 snapshot: &SharedSnapshot,
7311 identity: ProcessIdentity,
7312) -> Result<(), SuperviseError> {
7313 update_snapshot(snapshot, None, |state| {
7314 state.deliberate_severance = Some(identity);
7315 })
7316}
7317
7318fn apply_deliberate_severance_marker(
7319 snapshot: &SharedSnapshot,
7320 exited_identity: Option<ProcessIdentity>,
7321 mut exit_report: ExitReport,
7322) -> ExitReport {
7323 let marker = lock_snapshot(snapshot)
7324 .ok()
7325 .and_then(|mut state| state.deliberate_severance.take());
7326 if marker.is_some() && marker == exited_identity {
7327 exit_report.kind = ExitKind::DeliberateSeverance;
7328 }
7329 exit_report
7330}
7331
7332fn classify_reaped_child_exit(
7333 snapshot: &SharedSnapshot,
7334 child: &SupervisedChild,
7335 status: &ExitStatus,
7336) -> ExitReport {
7337 apply_deliberate_severance_marker(snapshot, child.process_identity(), classify_exit(status))
7338}
7339
7340fn fail_snapshot(
7341 snapshot: &SharedSnapshot,
7342 module_id: Option<&str>,
7343 last_exit: Option<ExitReport>,
7344) {
7345 if let Err(err) = update_snapshot(snapshot, module_id, |state| {
7346 state.state = ModuleState::Failed;
7347 clear_current_process_facts(state);
7348 if let Some(last_exit) = last_exit {
7349 state.last_exit = Some(last_exit);
7350 }
7351 }) {
7352 error!(error = %err, "failed to mark supervisor state failed");
7353 }
7354}
7355
7356fn update_snapshot(
7357 snapshot: &SharedSnapshot,
7358 module_id: Option<&str>,
7359 update: impl FnOnce(&mut SupervisorSnapshot),
7360) -> Result<(), SuperviseError> {
7361 let mut state = snapshot.lock().map_err(|_| SuperviseError::StatePoisoned {
7362 module_id: module_id.map(ToOwned::to_owned),
7363 })?;
7364 update(&mut state);
7365 Ok(())
7366}
7367
7368const SLOW_SNAPSHOT_LOCK_THRESHOLD: Duration = Duration::from_millis(250);
7369
7370fn lock_snapshot_for_control<'a>(
7371 snapshot: &'a SharedSnapshot,
7372 module_id: &str,
7373 caller: &'static str,
7374) -> Result<std::sync::MutexGuard<'a, SupervisorSnapshot>, SuperviseError> {
7375 let started_at = Instant::now();
7376 let guard = lock_snapshot(snapshot)?;
7377 let waited = started_at.elapsed();
7378 if waited >= SLOW_SNAPSHOT_LOCK_THRESHOLD {
7379 warn!(
7380 module_id = %module_id,
7381 waited_ms = waited.as_millis() as u64,
7382 caller = %caller,
7383 "slow snapshot lock"
7384 );
7385 }
7386 Ok(guard)
7387}
7388
7389fn lock_snapshot(
7390 snapshot: &SharedSnapshot,
7391) -> Result<std::sync::MutexGuard<'_, SupervisorSnapshot>, SuperviseError> {
7392 snapshot
7393 .lock()
7394 .map_err(|_| SuperviseError::StatePoisoned { module_id: None })
7395}
7396
7397#[cfg(test)]
7398mod terminal_history_tests {
7399 use std::{
7400 path::PathBuf,
7401 sync::Arc,
7402 time::{Duration, Instant},
7403 };
7404
7405 use tokio::time::sleep;
7406
7407 use super::{
7408 apply_deliberate_severance_marker, daemon_will_restart, drain_child_to_state,
7409 drained_after_quiescence_wait, handle_reload_spawn_failure, health_restart_child,
7410 lock_snapshot, on_child_exit, record_deliberate_severance, record_wait_error_terminal,
7411 reset_restart_count, spawn_and_mark_running, update_snapshot, wait_error_exit_report,
7412 ExitKind, ExitReport, ModuleProtocol, ModuleSpec, ModuleState, NextAction, ProcessIdentity,
7413 RestartPolicy, SpawnEventKind, StopNotice, SuperviseError, SupervisedModule, Supervisor,
7414 SupervisorHandle, SupervisorHealthStatus, SupervisorSnapshot,
7415 };
7416 use super::Instant as ClockInstant;
7421 use crate::{
7422 registry::Registry,
7423 terminal_ring::{TerminalRing, TerminalRingConfig},
7424 };
7425 use std::sync::Mutex;
7426 use subc_control::TerminalDisposition;
7427
7428 fn fake_aft_stub_path() -> PathBuf {
7433 let mut path = std::env::current_exe().expect("current_exe available in tests");
7434 path.pop();
7435 path.pop();
7436 path.push(if cfg!(windows) {
7437 "fake-aft-stub.exe"
7438 } else {
7439 "fake-aft-stub"
7440 });
7441 assert!(
7442 path.exists(),
7443 "fake-aft-stub not built at {}: run `cargo test -p subc-core` (which builds \
7444 [[bin]] targets) rather than `cargo test -p subc-core --lib` (which does not)",
7445 path.display()
7446 );
7447 path
7448 }
7449
7450 #[test]
7451 fn reserved_never_spawned_refuses_every_hello() {
7452 let supervisor = SupervisorHandle::default();
7457 supervisor.apply_identity_configuration(&ModuleSpec {
7458 module_id: "never-spawned".to_string(),
7459 program: PathBuf::from("/usr/bin/false"),
7460 args: Vec::new(),
7461 env: Vec::new(),
7462 reserved: true,
7463 reserved_prefixes: Vec::new(),
7464 protocol: ModuleProtocol::Subc,
7465 overlap: Default::default(),
7466 });
7467 assert!(
7468 supervisor
7469 .reserved_hello_rejection("never-spawned", Some("any-forged-nonce"))
7470 .is_some(),
7471 "forged nonce must refuse on a reserved never-spawned id"
7472 );
7473 assert!(
7474 supervisor
7475 .reserved_hello_rejection("never-spawned", None)
7476 .is_some(),
7477 "absent nonce must refuse on a reserved never-spawned id"
7478 );
7479 supervisor.set_spawn_nonce("never-spawned", "minted".to_string());
7481 supervisor.apply_identity_configuration(&ModuleSpec {
7482 module_id: "never-spawned".to_string(),
7483 program: PathBuf::from("/usr/bin/false"),
7484 args: Vec::new(),
7485 env: Vec::new(),
7486 reserved: true,
7487 reserved_prefixes: Vec::new(),
7488 protocol: ModuleProtocol::Subc,
7489 overlap: Default::default(),
7490 });
7491 assert!(supervisor
7492 .reserved_hello_rejection("never-spawned", Some("minted"))
7493 .is_none());
7494 assert!(supervisor
7495 .reserved_hello_rejection("never-spawned", Some("forged"))
7496 .is_some());
7497 }
7498
7499 fn seed_crash_restarts(state: &mut SupervisorSnapshot, count: u32) {
7502 let now = ClockInstant::now();
7503 for _ in 0..count {
7504 state.crash_restarts.push_back(now);
7505 }
7506 }
7507
7508 fn age_oldest_crash_restart_out_of_window(state: &mut SupervisorSnapshot, window: Duration) {
7512 let aged = state
7513 .crash_restarts
7514 .front()
7515 .expect("a crash restart must be recorded before it can be aged")
7516 .checked_sub(window + Duration::from_secs(1))
7517 .expect("the test clock is far enough from its origin to age an instant");
7518 state.crash_restarts[0] = aged;
7519 }
7520
7521 fn snapshot_with_restarts(enabled: bool, count: u32) -> SupervisorSnapshot {
7522 let mut state = SupervisorSnapshot::new(ModuleState::Running, enabled);
7523 seed_crash_restarts(&mut state, count);
7524 state
7525 }
7526
7527 #[test]
7528 fn daemon_owned_recovery_predicate_uses_the_pre_increment_budget() {
7529 let policy = RestartPolicy::new(3, Duration::ZERO);
7530 let now = ClockInstant::now();
7531 assert!(daemon_will_restart(
7532 &mut snapshot_with_restarts(true, 2),
7533 &policy,
7534 now
7535 ));
7536 assert!(!daemon_will_restart(
7537 &mut snapshot_with_restarts(true, 3),
7538 &policy,
7539 now
7540 ));
7541 assert!(!daemon_will_restart(
7542 &mut snapshot_with_restarts(false, 0),
7543 &policy,
7544 now
7545 ));
7546 }
7547
7548 #[test]
7549 fn crash_restart_backoff_escalates_with_in_window_count() {
7550 let policy = RestartPolicy::new(4, Duration::from_millis(100))
7551 .with_max_backoff(Duration::from_secs(30));
7552 let now = ClockInstant::now();
7553 let mut state = SupervisorSnapshot::new(ModuleState::Running, true);
7554 let schedules = (0..4)
7555 .map(|_| {
7556 state
7557 .next_crash_restart(&policy, now)
7558 .expect("the test policy allows four crash restarts")
7559 })
7560 .collect::<Vec<_>>();
7561
7562 assert_eq!(
7563 schedules
7564 .iter()
7565 .map(|schedule| schedule.restart_in_window)
7566 .collect::<Vec<_>>(),
7567 vec![0, 1, 2, 3]
7568 );
7569 assert_eq!(
7570 schedules
7571 .iter()
7572 .map(|schedule| schedule.delay)
7573 .collect::<Vec<_>>(),
7574 vec![
7575 Duration::from_millis(100),
7576 Duration::from_secs(1),
7577 Duration::from_secs(10),
7578 Duration::from_secs(30),
7579 ]
7580 );
7581 }
7582
7583 #[test]
7584 fn crash_restart_backoff_resets_after_ring_clear() {
7585 let policy = RestartPolicy::new(3, Duration::from_millis(100));
7586 let now = ClockInstant::now();
7587 let mut state = SupervisorSnapshot::new(ModuleState::Running, true);
7588 assert_eq!(
7589 state.next_crash_restart(&policy, now).unwrap().delay,
7590 Duration::from_millis(100)
7591 );
7592 assert_eq!(
7593 state.next_crash_restart(&policy, now).unwrap().delay,
7594 Duration::from_secs(1)
7595 );
7596
7597 state.clear_crash_restarts();
7598 let schedule = state
7599 .next_crash_restart(&policy, now)
7600 .expect("a cleared ring must allow another restart");
7601 assert_eq!(schedule.restart_in_window, 0);
7602 assert_eq!(schedule.delay, Duration::from_millis(100));
7603 }
7604
7605 #[test]
7606 fn crash_restart_backoff_ignores_aged_restarts() {
7607 let policy = RestartPolicy::new(3, Duration::from_millis(100));
7608 let now = ClockInstant::now();
7609 let mut state = SupervisorSnapshot::new(ModuleState::Running, true);
7610 state
7611 .next_crash_restart(&policy, now)
7612 .expect("the first restart is allowed");
7613 state
7614 .next_crash_restart(&policy, now)
7615 .expect("the second restart is allowed");
7616 state.crash_restarts[0] = now
7617 .checked_sub(policy.window + Duration::from_secs(1))
7618 .expect("the fake clock can age a restart past the window");
7619
7620 let schedule = state
7621 .next_crash_restart(&policy, now)
7622 .expect("an aged restart must release its slot");
7623 assert_eq!(schedule.restart_in_window, 1);
7624 assert_eq!(schedule.delay, Duration::from_secs(1));
7625 assert_eq!(state.crash_restarts.len(), 2);
7626 }
7627
7628 #[test]
7632 fn a_budget_spent_before_the_window_no_longer_refuses() {
7633 let policy = RestartPolicy::new(3, Duration::ZERO);
7634 let mut state = snapshot_with_restarts(true, 3);
7635 let now = ClockInstant::now();
7636 assert!(!daemon_will_restart(&mut state, &policy, now));
7637
7638 assert!(daemon_will_restart(
7639 &mut state,
7640 &policy,
7641 now + policy.window + Duration::from_secs(1)
7642 ));
7643 assert!(
7644 state.crash_restarts.is_empty(),
7645 "reading the budget must drop the instants that left the window"
7646 );
7647 }
7648
7649 fn module_with_recovery_snapshot(
7650 state: ModuleState,
7651 enabled: bool,
7652 restart_count: u32,
7653 ) -> SupervisedModule {
7654 let registry = Arc::new(Registry::default());
7655 let supervisor =
7656 Supervisor::new(Arc::clone(®istry), RestartPolicy::new(3, Duration::ZERO));
7657 let module = supervisor
7658 .spawn(ModuleSpec {
7659 module_id: "recovery-snapshot".to_string(),
7660 program: fake_aft_stub_path(),
7661 args: Vec::new(),
7662 env: Vec::new(),
7663 reserved: false,
7664 reserved_prefixes: Vec::new(),
7665 protocol: ModuleProtocol::Subc,
7666 overlap: Default::default(),
7667 })
7668 .unwrap();
7669 update_snapshot(
7670 &module.inner.snapshot,
7671 Some("recovery-snapshot"),
7672 |snapshot| {
7673 snapshot.state = state;
7674 snapshot.enabled = enabled;
7675 seed_crash_restarts(snapshot, restart_count);
7676 },
7677 )
7678 .unwrap();
7679 module
7680 }
7681
7682 #[cfg(target_os = "linux")]
7683 #[tokio::test]
7684 async fn no_cgroup_placement_does_not_block_fake_aft_stub_spawn() {
7685 let supervisor = Supervisor::new(Arc::new(Registry::default()), RestartPolicy::default())
7686 .with_cgroup_placement(None);
7687 let result = supervisor.spawn(ModuleSpec {
7688 module_id: "no-cgroup-placement".to_string(),
7689 program: fake_aft_stub_path(),
7690 args: Vec::new(),
7691 env: Vec::new(),
7692 reserved: false,
7693 reserved_prefixes: Vec::new(),
7694 protocol: ModuleProtocol::Subc,
7695 overlap: Default::default(),
7696 });
7697
7698 assert!(
7699 result.is_ok(),
7700 "no delegation must not turn an otherwise valid spawn into a failure: {result:?}"
7701 );
7702 }
7703
7704 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7705 async fn undecided_snapshot_uses_shared_restart_predicate() {
7706 assert!(module_with_recovery_snapshot(ModuleState::Running, true, 2)
7707 .will_recover_after_connection_loss()
7708 .unwrap());
7709 assert!(
7710 !module_with_recovery_snapshot(ModuleState::Running, true, 3)
7711 .will_recover_after_connection_loss()
7712 .unwrap()
7713 );
7714 }
7715
7716 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7717 async fn restarting_snapshot_at_exhausted_budget_is_non_terminal() {
7718 assert!(
7719 module_with_recovery_snapshot(ModuleState::Restarting, true, 3)
7720 .will_recover_after_connection_loss()
7721 .unwrap()
7722 );
7723 }
7724
7725 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7726 async fn terminal_phase_snapshots_are_terminal_before_budget_exhaustion() {
7727 assert!(!module_with_recovery_snapshot(ModuleState::Failed, true, 0)
7728 .will_recover_after_connection_loss()
7729 .unwrap());
7730 assert!(
7731 !module_with_recovery_snapshot(ModuleState::Disabled, true, 0)
7732 .will_recover_after_connection_loss()
7733 .unwrap()
7734 );
7735 }
7736
7737 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7738 async fn warming_snapshot_is_limited_to_startup_phases() {
7739 for state in [
7740 ModuleState::Starting,
7741 ModuleState::Running,
7742 ModuleState::Restarting,
7743 ] {
7744 assert!(
7745 module_with_recovery_snapshot(state, true, 0)
7746 .is_warming()
7747 .unwrap(),
7748 "{state:?} should be warming"
7749 );
7750 }
7751 for state in [
7752 ModuleState::Unresponsive,
7753 ModuleState::Draining,
7754 ModuleState::Stopped,
7755 ModuleState::Failed,
7756 ModuleState::Disabled,
7757 ] {
7758 assert!(
7759 !module_with_recovery_snapshot(state, true, 0)
7760 .is_warming()
7761 .unwrap(),
7762 "{state:?} should not be warming"
7763 );
7764 }
7765 }
7766
7767 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7768 async fn terminal_history_survives_respawn_and_keeps_both_crashes_in_order() {
7769 let registry = Arc::new(Registry::default());
7770 let supervisor =
7771 Supervisor::new(Arc::clone(®istry), RestartPolicy::new(1, Duration::ZERO));
7772 let module = supervisor
7773 .spawn(ModuleSpec {
7774 module_id: "terminal-history".to_string(),
7775 program: fake_aft_stub_path(),
7776 args: Vec::new(),
7777 env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
7778 reserved: false,
7779 reserved_prefixes: Vec::new(),
7780 protocol: ModuleProtocol::Subc,
7781 overlap: Default::default(),
7782 })
7783 .unwrap();
7784
7785 let deadline = Instant::now() + Duration::from_secs(5);
7786 loop {
7787 let history = module.terminal_history();
7788 if history.entries.len() == 2 {
7789 assert_eq!(module.status().unwrap().state, ModuleState::Failed);
7790 assert_eq!(history.dropped, 0);
7791 assert_eq!(
7792 history
7793 .entries
7794 .iter()
7795 .map(|entry| entry.exit_code)
7796 .collect::<Vec<_>>(),
7797 vec![Some(23), Some(23)]
7798 );
7799 assert!(history.entries[0].at_ms <= history.entries[1].at_ms);
7800 return;
7801 }
7802 assert!(
7803 Instant::now() < deadline,
7804 "module did not retain two terminal exits: {history:?}"
7805 );
7806 sleep(Duration::from_millis(10)).await;
7807 }
7808 }
7809
7810 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7814 async fn disable_during_crash_backoff_cancels_pending_respawn() {
7815 let backoff = Duration::from_secs(2);
7816 let supervisor = Supervisor::new(
7817 Arc::new(Registry::default()),
7818 RestartPolicy::new(10, backoff),
7819 );
7820 let module = supervisor
7821 .spawn(ModuleSpec {
7822 module_id: "disable-during-backoff".to_string(),
7823 program: fake_aft_stub_path(),
7824 args: Vec::new(),
7825 env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
7826 reserved: false,
7827 reserved_prefixes: Vec::new(),
7828 protocol: ModuleProtocol::Subc,
7829 overlap: Default::default(),
7830 })
7831 .unwrap();
7832
7833 let deadline = Instant::now() + Duration::from_secs(5);
7835 loop {
7836 if module.status().unwrap().state == ModuleState::Restarting {
7837 break;
7838 }
7839 assert!(
7840 Instant::now() < deadline,
7841 "module never entered the crash backoff"
7842 );
7843 sleep(Duration::from_millis(10)).await;
7844 }
7845
7846 let started = Instant::now();
7847 module.set_enabled(false).await.unwrap();
7848 let waited = started.elapsed();
7849
7850 assert!(
7851 waited < backoff / 2,
7852 "disable waited {waited:?} behind the {backoff:?} crash backoff; the operator command must preempt the pending respawn"
7853 );
7854 assert_eq!(module.status().unwrap().state, ModuleState::Disabled);
7855
7856 sleep(backoff + Duration::from_millis(500)).await;
7858 let status = module.status().unwrap();
7859 assert_eq!(status.state, ModuleState::Disabled);
7860 assert_eq!(
7861 status.spawn_generation, 1,
7862 "module respawned after the operator disabled it"
7863 );
7864 }
7865
7866 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7870 async fn every_restart_increment_path_advances_lifetime_count() {
7871 let supervisor = Supervisor::new(
7872 Arc::new(Registry::default()),
7873 RestartPolicy::new(1, Duration::ZERO),
7874 );
7875 let runtime = supervisor.runtime_config();
7876 let spec = ModuleSpec {
7877 module_id: "lifetime-increment-path".to_string(),
7878 program: PathBuf::from("/unused/lifetime-increment-path"),
7879 args: Vec::new(),
7880 env: Vec::new(),
7881 reserved: false,
7882 reserved_prefixes: Vec::new(),
7883 protocol: ModuleProtocol::Subc,
7884 overlap: Default::default(),
7885 };
7886
7887 let crash_snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
7888 assert!(matches!(
7889 on_child_exit(
7890 &spec,
7891 runtime.restart_policy,
7892 &supervisor.registry,
7893 &crash_snapshot,
7894 &runtime.terminal_ring,
7895 &runtime.spawn_events,
7896 &runtime.child_roster,
7897 ExitReport {
7898 kind: ExitKind::Crash,
7899 code: Some(1),
7900 signal: None,
7901 at_ms: 1,
7902 },
7903 )
7904 .await,
7905 NextAction::Restart { schedule: _ }
7906 ));
7907 let (crash_restarts, crash_lifetime) = {
7908 let state = lock_snapshot(&crash_snapshot).unwrap();
7909 (state.crash_restarts.len(), state.lifetime_restarts)
7910 };
7911 assert_eq!(crash_restarts, 1);
7912 assert_eq!(crash_lifetime, 1);
7913
7914 let health_snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
7915 let mut health_child = None;
7916 assert!(matches!(
7917 health_restart_child(
7918 &spec,
7919 &runtime,
7920 &supervisor.registry,
7921 &supervisor.process_liveness,
7922 &health_snapshot,
7923 &mut health_child,
7924 SupervisorHealthStatus::Failing,
7925 None,
7926 2,
7927 )
7928 .await,
7929 Err(SuperviseError::Spawn { .. })
7930 ));
7931 let (health_restarts, health_lifetime) = {
7932 let state = lock_snapshot(&health_snapshot).unwrap();
7933 (state.crash_restarts.len(), state.lifetime_restarts)
7934 };
7935 assert_eq!(health_restarts, 1);
7936 assert_eq!(health_lifetime, 1);
7937
7938 let reload_snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
7939 let mut reload_child = None;
7940 assert!(matches!(
7941 handle_reload_spawn_failure(
7942 &spec,
7943 &runtime,
7944 &supervisor.process_liveness,
7945 &reload_snapshot,
7946 &mut reload_child,
7947 "forced reload spawn failure".to_string(),
7948 )
7949 .await,
7950 Err(SuperviseError::ReloadFailed { .. })
7951 ));
7952 let (reload_restarts, reload_lifetime) = {
7953 let state = lock_snapshot(&reload_snapshot).unwrap();
7954 (state.crash_restarts.len(), state.lifetime_restarts)
7955 };
7956 assert_eq!(reload_restarts, 1);
7957 assert_eq!(reload_lifetime, 1);
7958 }
7959
7960 #[tokio::test]
7961 async fn deliberately_severed_live_child_records_lifetime_without_spending_restart_budget() {
7962 let supervisor = Supervisor::new(
7963 Arc::new(Registry::default()),
7964 RestartPolicy::new(3, Duration::ZERO),
7965 );
7966 let runtime = supervisor.runtime_config();
7967 let spec = ModuleSpec {
7968 module_id: "deliberately-severed".to_string(),
7969 program: PathBuf::from("/unused/deliberately-severed"),
7970 args: Vec::new(),
7971 env: Vec::new(),
7972 reserved: false,
7973 reserved_prefixes: Vec::new(),
7974 protocol: ModuleProtocol::Subc,
7975 overlap: Default::default(),
7976 };
7977 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
7978 let process = ProcessIdentity {
7979 pid: 41,
7980 start_time: 101,
7981 };
7982 record_deliberate_severance(&snapshot, process).unwrap();
7983 let exit_report = apply_deliberate_severance_marker(
7984 &snapshot,
7985 Some(process),
7986 ExitReport {
7987 kind: ExitKind::Crash,
7988 code: Some(1),
7989 signal: None,
7990 at_ms: 1,
7991 },
7992 );
7993 assert_eq!(exit_report.kind, ExitKind::DeliberateSeverance);
7994
7995 assert!(matches!(
7996 on_child_exit(
7997 &spec,
7998 runtime.restart_policy,
7999 &supervisor.registry,
8000 &snapshot,
8001 &runtime.terminal_ring,
8002 &runtime.spawn_events,
8003 &runtime.child_roster,
8004 exit_report,
8005 )
8006 .await,
8007 NextAction::Restart { schedule: _ }
8008 ));
8009 let state = lock_snapshot(&snapshot).unwrap();
8010 assert_eq!(state.lifetime_restarts, 1);
8011 assert_eq!(state.crash_restarts.len(), 0);
8012 }
8013
8014 #[tokio::test]
8015 async fn genuine_crash_spends_restart_budget_and_records_lifetime() {
8016 let supervisor = Supervisor::new(
8017 Arc::new(Registry::default()),
8018 RestartPolicy::new(3, Duration::ZERO),
8019 );
8020 let runtime = supervisor.runtime_config();
8021 let spec = ModuleSpec {
8022 module_id: "genuine-crash".to_string(),
8023 program: PathBuf::from("/unused/genuine-crash"),
8024 args: Vec::new(),
8025 env: Vec::new(),
8026 reserved: false,
8027 reserved_prefixes: Vec::new(),
8028 protocol: ModuleProtocol::Subc,
8029 overlap: Default::default(),
8030 };
8031 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8032
8033 assert!(matches!(
8034 on_child_exit(
8035 &spec,
8036 runtime.restart_policy,
8037 &supervisor.registry,
8038 &snapshot,
8039 &runtime.terminal_ring,
8040 &runtime.spawn_events,
8041 &runtime.child_roster,
8042 ExitReport {
8043 kind: ExitKind::Crash,
8044 code: Some(1),
8045 signal: None,
8046 at_ms: 1,
8047 },
8048 )
8049 .await,
8050 NextAction::Restart { schedule: _ }
8051 ));
8052 let state = lock_snapshot(&snapshot).unwrap();
8053 assert_eq!(state.lifetime_restarts, 1);
8054 assert_eq!(state.crash_restarts.len(), 1);
8055 }
8056
8057 fn crash_exit_report(at_ms: u64) -> ExitReport {
8058 ExitReport {
8059 kind: ExitKind::Crash,
8060 code: Some(1),
8061 signal: None,
8062 at_ms,
8063 }
8064 }
8065
8066 fn windowed_crash_spec(module_id: &str) -> ModuleSpec {
8067 ModuleSpec {
8068 module_id: module_id.to_string(),
8069 program: PathBuf::from("/unused").join(module_id),
8070 args: Vec::new(),
8071 env: Vec::new(),
8072 reserved: false,
8073 reserved_prefixes: Vec::new(),
8074 protocol: ModuleProtocol::Subc,
8075 overlap: Default::default(),
8076 }
8077 }
8078
8079 #[tokio::test]
8085 async fn three_crashes_inside_the_window_stop_the_module_and_name_the_window() {
8086 let (logs, _guard) = crate::router::test_log::log_capture(tracing::Level::ERROR);
8087 let supervisor = Supervisor::new(
8088 Arc::new(Registry::default()),
8089 RestartPolicy::new(2, Duration::ZERO),
8090 );
8091 let runtime = supervisor.runtime_config();
8092 let spec = windowed_crash_spec("crash-loop-in-window");
8093 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8094
8095 for attempt in 1..=2 {
8096 assert!(
8097 matches!(
8098 on_child_exit(
8099 &spec,
8100 runtime.restart_policy,
8101 &supervisor.registry,
8102 &snapshot,
8103 &runtime.terminal_ring,
8104 &runtime.spawn_events,
8105 &runtime.child_roster,
8106 crash_exit_report(attempt),
8107 )
8108 .await,
8109 NextAction::Restart { schedule: _ }
8110 ),
8111 "crash {attempt} is inside the budget and must respawn"
8112 );
8113 }
8114
8115 assert!(matches!(
8116 on_child_exit(
8117 &spec,
8118 runtime.restart_policy,
8119 &supervisor.registry,
8120 &snapshot,
8121 &runtime.terminal_ring,
8122 &runtime.spawn_events,
8123 &runtime.child_roster,
8124 crash_exit_report(3),
8125 )
8126 .await,
8127 NextAction::Stop { .. }
8128 ));
8129
8130 {
8131 let state = lock_snapshot(&snapshot).unwrap();
8132 assert_eq!(state.state, ModuleState::Failed);
8133 assert_eq!(state.crash_restarts.len(), 2);
8134 assert_eq!(state.lifetime_restarts, 2);
8135 }
8136
8137 let history = runtime
8138 .terminal_ring
8139 .lock()
8140 .expect("terminal ring is not poisoned")
8141 .snapshot();
8142 let last = history
8143 .entries
8144 .last()
8145 .expect("the refused crash is retained");
8146 assert_eq!(last.disposition, TerminalDisposition::Failed);
8147 assert_eq!(
8148 last.disposition_detail.as_deref(),
8149 Some("crash budget exhausted: max_restarts=2 within window_secs=600")
8150 );
8151
8152 let captured = crate::router::test_log::captured_logs(&logs);
8153 assert!(
8154 captured.contains("crash budget exhausted: max_restarts=2 within window_secs=600"),
8155 "the stop must be logged with its window: {captured}"
8156 );
8157 }
8158
8159 #[tokio::test]
8167 async fn a_crash_older_than_the_window_frees_its_slot_for_a_later_crash() {
8168 let supervisor = Supervisor::new(
8169 Arc::new(Registry::default()),
8170 RestartPolicy::new(2, Duration::ZERO),
8171 );
8172 let runtime = supervisor.runtime_config();
8173 let spec = windowed_crash_spec("crash-across-windows");
8174 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8175
8176 for attempt in 1..=2 {
8177 assert!(matches!(
8178 on_child_exit(
8179 &spec,
8180 runtime.restart_policy,
8181 &supervisor.registry,
8182 &snapshot,
8183 &runtime.terminal_ring,
8184 &runtime.spawn_events,
8185 &runtime.child_roster,
8186 crash_exit_report(attempt),
8187 )
8188 .await,
8189 NextAction::Restart { schedule: _ }
8190 ));
8191 }
8192
8193 update_snapshot(&snapshot, Some(&spec.module_id), |state| {
8196 age_oldest_crash_restart_out_of_window(state, runtime.restart_policy.window);
8197 })
8198 .unwrap();
8199
8200 assert!(
8201 matches!(
8202 on_child_exit(
8203 &spec,
8204 runtime.restart_policy,
8205 &supervisor.registry,
8206 &snapshot,
8207 &runtime.terminal_ring,
8208 &runtime.spawn_events,
8209 &runtime.child_roster,
8210 crash_exit_report(3),
8211 )
8212 .await,
8213 NextAction::Restart { schedule: _ }
8214 ),
8215 "a crash older than the window must not hold a budget slot"
8216 );
8217
8218 let state = lock_snapshot(&snapshot).unwrap();
8219 assert_eq!(state.state, ModuleState::Restarting);
8220 assert_eq!(
8221 state.crash_restarts.len(),
8222 2,
8223 "the aged instant is dropped and the new one takes its place"
8224 );
8225 assert_eq!(
8226 state.lifetime_restarts, 3,
8227 "the ledger counts every restart, including the ones the window forgot"
8228 );
8229 }
8230
8231 #[tokio::test]
8236 async fn an_operator_restart_clears_the_ring_and_leaves_the_ledger_alone() {
8237 let supervisor = Supervisor::new(
8238 Arc::new(Registry::default()),
8239 RestartPolicy::new(2, Duration::ZERO),
8240 );
8241 let runtime = supervisor.runtime_config();
8242 let spec = windowed_crash_spec("operator-cleared-budget");
8243 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8244
8245 for attempt in 1..=2 {
8246 assert!(matches!(
8247 on_child_exit(
8248 &spec,
8249 runtime.restart_policy,
8250 &supervisor.registry,
8251 &snapshot,
8252 &runtime.terminal_ring,
8253 &runtime.spawn_events,
8254 &runtime.child_roster,
8255 crash_exit_report(attempt),
8256 )
8257 .await,
8258 NextAction::Restart { schedule: _ }
8259 ));
8260 }
8261
8262 reset_restart_count(&snapshot, &spec.module_id).unwrap();
8263 {
8264 let state = lock_snapshot(&snapshot).unwrap();
8265 assert!(
8266 state.crash_restarts.is_empty(),
8267 "an operator restart returns the full budget"
8268 );
8269 assert_eq!(
8270 state.lifetime_restarts, 2,
8271 "clearing the budget must not unmake the crashes"
8272 );
8273 }
8274
8275 assert!(
8276 matches!(
8277 on_child_exit(
8278 &spec,
8279 runtime.restart_policy,
8280 &supervisor.registry,
8281 &snapshot,
8282 &runtime.terminal_ring,
8283 &runtime.spawn_events,
8284 &runtime.child_roster,
8285 crash_exit_report(3),
8286 )
8287 .await,
8288 NextAction::Restart { schedule: _ }
8289 ),
8290 "the cleared budget must be spendable again"
8291 );
8292 let state = lock_snapshot(&snapshot).unwrap();
8293 assert_eq!(state.crash_restarts.len(), 1);
8294 assert_eq!(state.lifetime_restarts, 3);
8295 }
8296
8297 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
8298 async fn severance_marker_for_a_dead_child_does_not_label_its_successor() {
8299 let severed = ProcessIdentity {
8300 pid: 41,
8301 start_time: 101,
8302 };
8303 let successor = ProcessIdentity {
8304 pid: 41,
8305 start_time: 202,
8306 };
8307 let module = module_with_recovery_snapshot(ModuleState::Running, true, 0);
8308 update_snapshot(&module.inner.snapshot, Some("recovery-snapshot"), |state| {
8309 state.pid = Some(successor.pid);
8310 state.process_start_time = Some(successor.start_time);
8311 })
8312 .unwrap();
8313 assert!(!module.record_deliberate_severance(severed).unwrap());
8314
8315 let exit_report = apply_deliberate_severance_marker(
8316 &module.inner.snapshot,
8317 Some(successor),
8318 ExitReport {
8319 kind: ExitKind::Crash,
8320 code: Some(1),
8321 signal: None,
8322 at_ms: 1,
8323 },
8324 );
8325
8326 assert_eq!(exit_report.kind, ExitKind::Crash);
8327 }
8328
8329 #[tokio::test]
8330 async fn drain_reap_marks_deliberate_severance_and_records_lifetime_without_budget() {
8331 let registry = Registry::default();
8332 let supervisor = Supervisor::new(
8333 Arc::new(Registry::default()),
8334 RestartPolicy::new(3, Duration::ZERO),
8335 );
8336 let runtime = supervisor.runtime_config();
8337 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8338 let spec = ModuleSpec {
8339 module_id: "drain-deliberate-severance".to_string(),
8340 program: fake_aft_stub_path(),
8341 args: Vec::new(),
8342 env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
8343 reserved: false,
8344 reserved_prefixes: Vec::new(),
8345 protocol: ModuleProtocol::Subc,
8346 overlap: Default::default(),
8347 };
8348 let mut child = spawn_and_mark_running(&spec, &runtime, &snapshot).unwrap();
8349 let process = ProcessIdentity {
8350 pid: 41,
8351 start_time: 101,
8352 };
8353 child.process_identity = Some(process);
8354 update_snapshot(&snapshot, Some(&spec.module_id), |state| {
8355 state.pid = Some(process.pid);
8356 state.process_start_time = Some(process.start_time);
8357 })
8358 .unwrap();
8359 record_deliberate_severance(&snapshot, process).unwrap();
8360
8361 drain_child_to_state(
8362 &spec.module_id,
8363 spec.protocol,
8364 StopNotice::SentOverConnection,
8367 ®istry,
8368 &snapshot,
8369 &runtime.terminal_ring,
8370 &runtime.spawn_events,
8371 child,
8372 Duration::from_secs(1),
8373 ModuleState::Stopped,
8374 Some(false),
8375 )
8376 .await
8377 .unwrap();
8378
8379 let state = lock_snapshot(&snapshot).unwrap();
8380 assert_eq!(
8381 state.last_exit.as_ref().map(|exit| exit.kind),
8382 Some(ExitKind::DeliberateSeverance)
8383 );
8384 assert_eq!(state.lifetime_restarts, 1);
8385 assert_eq!(state.crash_restarts.len(), 0);
8386 drop(state);
8387 let history = runtime.terminal_ring.lock().unwrap().snapshot();
8388 assert_eq!(
8389 history.entries[0].exit_kind,
8390 subc_control::TerminalExitKind::DeliberateSeverance
8391 );
8392 }
8393
8394 #[tokio::test]
8395 async fn ordinary_drain_reap_does_not_record_a_lifetime_restart() {
8396 let registry = Registry::default();
8397 let supervisor = Supervisor::new(
8398 Arc::new(Registry::default()),
8399 RestartPolicy::new(3, Duration::ZERO),
8400 );
8401 let runtime = supervisor.runtime_config();
8402 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8403 let spec = ModuleSpec {
8404 module_id: "ordinary-drain".to_string(),
8405 program: fake_aft_stub_path(),
8406 args: Vec::new(),
8407 env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
8408 reserved: false,
8409 reserved_prefixes: Vec::new(),
8410 protocol: ModuleProtocol::Subc,
8411 overlap: Default::default(),
8412 };
8413 let child = spawn_and_mark_running(&spec, &runtime, &snapshot).unwrap();
8414
8415 drain_child_to_state(
8416 &spec.module_id,
8417 spec.protocol,
8418 StopNotice::SentOverConnection,
8421 ®istry,
8422 &snapshot,
8423 &runtime.terminal_ring,
8424 &runtime.spawn_events,
8425 child,
8426 Duration::from_secs(1),
8427 ModuleState::Stopped,
8428 Some(false),
8429 )
8430 .await
8431 .unwrap();
8432
8433 let state = lock_snapshot(&snapshot).unwrap();
8434 assert_eq!(
8435 state.last_exit.as_ref().map(|exit| exit.kind),
8436 Some(ExitKind::Crash)
8437 );
8438 assert_eq!(state.lifetime_restarts, 0);
8439 assert_eq!(state.crash_restarts.len(), 0);
8440 }
8441
8442 #[test]
8443 fn fatal_connection_teardown_cannot_arm_a_marker_for_a_surviving_process() {
8444 assert!(!include_str!("server.rs")
8450 .contains("router.record_deliberate_connection_severance(ctx.connection_id)"));
8451 }
8452
8453 #[test]
8460 fn drained_after_quiescence_wait_passes_ok_through_and_forces_false_on_err() {
8461 assert!(drained_after_quiescence_wait(&Ok(true)));
8462 assert!(!drained_after_quiescence_wait(&Ok(false)));
8463 assert!(!drained_after_quiescence_wait(&Err(
8464 SuperviseError::StatePoisoned { module_id: None }
8465 )));
8466 }
8467
8468 #[test]
8477 fn wait_error_exit_report_records_a_failed_terminal_with_no_code_or_signal() {
8478 let ring = Arc::new(Mutex::new(TerminalRing::new(
8479 TerminalRingConfig::default(),
8480 0,
8481 )));
8482 record_wait_error_terminal("wait-error", &ring, &super::SpawnEventFeed::default());
8483
8484 let snapshot = ring.lock().unwrap().snapshot();
8485 assert_eq!(snapshot.entries.len(), 1);
8486 let entry = &snapshot.entries[0];
8487 assert_eq!(entry.exit_code, None);
8488 assert_eq!(entry.exit_signal, None);
8489 assert_eq!(entry.disposition, TerminalDisposition::Failed);
8490 }
8491
8492 #[test]
8493 fn wait_error_exit_path_preserves_spawn_event_density() {
8494 let feed = super::SpawnEventFeed::default();
8495 feed.configure_incarnation("wait-error-density".to_string());
8496 feed.emit_spawned("wait-error", 41, 1);
8497 let ring = Arc::new(Mutex::new(TerminalRing::new(
8498 TerminalRingConfig::default(),
8499 0,
8500 )));
8501
8502 record_wait_error_terminal("wait-error", &ring, &feed);
8503 feed.emit_spawned("after-wait-error", 42, 2);
8504
8505 let state = feed.0.lock().unwrap();
8506 let sequences = state
8507 .events
8508 .iter()
8509 .map(|event| event.cursor.seq)
8510 .collect::<Vec<_>>();
8511 assert_eq!(sequences, vec![1, 2, 3]);
8512 assert_eq!(state.events[1].kind, SpawnEventKind::Exited);
8513 assert_eq!(state.events[1].exit_code, None);
8514 assert_eq!(state.events[1].exit_signal, None);
8515 }
8516
8517 #[test]
8521 fn wait_error_exit_report_is_classified_as_a_crash() {
8522 assert_eq!(wait_error_exit_report().kind, ExitKind::Crash);
8523 }
8524}
8525
8526#[cfg(test)]
8527mod health_evidence_tests {
8528 use super::{HealthProbeError, HealthProbeEvidence};
8529 use std::collections::HashSet;
8530
8531 #[test]
8539 fn only_a_dead_lane_is_proof_of_death() {
8540 assert!(HealthProbeError::lane_dead("gone").is_proof_of_death());
8541 assert!(!HealthProbeError::no_answer("timed out").is_proof_of_death());
8545 assert!(!HealthProbeError::bad_answer("garbage").is_proof_of_death());
8546 assert!(!HealthProbeError::misconfigured("no table").is_proof_of_death());
8547 }
8548
8549 #[test]
8555 fn every_evidence_class_has_a_distinct_label() {
8556 let labels = [
8557 HealthProbeError::lane_dead("").label(),
8558 HealthProbeError::no_answer("").label(),
8559 HealthProbeError::bad_answer("").label(),
8560 HealthProbeError::misconfigured("").label(),
8561 ];
8562 let unique: HashSet<_> = labels.iter().collect();
8563 assert_eq!(unique.len(), labels.len(), "labels collided: {labels:?}");
8564 }
8565
8566 #[test]
8572 fn classification_preserves_the_original_message() {
8573 let err = HealthProbeError::no_answer("module did not answer within 5s");
8574 assert_eq!(err.to_string(), "module did not answer within 5s");
8575 assert!(matches!(err.evidence, HealthProbeEvidence::NoAnswer));
8576 }
8577}
8578
8579#[cfg(test)]
8580mod health_tombstone_tests {
8581 use std::{path::PathBuf, sync::Arc, time::Duration};
8582
8583 use subc_protocol::{
8584 manifest::Concurrency,
8585 session::{HealthStatus, ModuleControlResponse},
8586 };
8587 use tokio::sync::mpsc;
8588
8589 use super::{
8590 probe_module_health, HealthAction, HealthConfig, HealthProbeEvidence, ModuleProtocol,
8591 ModuleSpec, RestartPolicy, Supervisor, SupervisorRuntimeConfig,
8592 };
8593 use crate::{
8594 control::ControlHandler,
8595 forwarding::{ForwardingTable, ModuleControlRpcCompletion, ModuleControlRpcOutcome},
8596 registry::{ConnectionId, Registry},
8597 router::FrameSink,
8598 };
8599
8600 struct ProbeHarness {
8601 spec: ModuleSpec,
8602 runtime: SupervisorRuntimeConfig,
8603 forwarding: Arc<ForwardingTable>,
8604 module_connection: ConnectionId,
8605 module_rx: mpsc::Receiver<crate::router::OutboundFrame>,
8606 handler: ControlHandler,
8607 module: super::SupervisedModule,
8608 }
8609
8610 fn probe_harness() -> ProbeHarness {
8611 let registry = Arc::new(Registry::default());
8612 let forwarding = Arc::new(ForwardingTable::default());
8613 let supervisor_handle = super::SupervisorHandle::new();
8614 let health = HealthConfig {
8615 cadence: Duration::from_secs(30),
8616 deadline: Duration::from_secs(5),
8617 failure_threshold: 3,
8618 on_degraded: HealthAction::Report,
8619 on_failing: HealthAction::Report,
8620 critical: false,
8621 };
8622 let supervisor = Supervisor::new(Arc::clone(®istry), RestartPolicy::default())
8623 .with_forwarding(Arc::clone(&forwarding))
8624 .with_handle(supervisor_handle.clone())
8625 .with_health_config(health);
8626 let spec = ModuleSpec {
8627 module_id: "late-health-module".to_string(),
8628 program: PathBuf::from("disabled-module"),
8629 args: Vec::new(),
8630 env: Vec::new(),
8631 reserved: false,
8632 reserved_prefixes: Vec::new(),
8633 protocol: ModuleProtocol::Subc,
8634 overlap: Default::default(),
8635 };
8636 let module = supervisor
8637 .supervise_configured(spec.clone(), false)
8638 .unwrap();
8639 let runtime = supervisor.runtime_config();
8640 let handler = ControlHandler::with_forwarding(registry, Arc::clone(&forwarding))
8641 .with_supervisor(supervisor_handle);
8642 let module_connection = ConnectionId::new(700);
8643 let (module_tx, module_rx) = mpsc::channel(8);
8644 forwarding
8645 .register_module_connection(
8646 module_connection,
8647 spec.module_id.clone(),
8648 subc_protocol::PROTOCOL_VERSION,
8649 Concurrency::ModuleManaged,
8650 FrameSink::new(module_tx),
8651 )
8652 .unwrap();
8653
8654 ProbeHarness {
8655 spec,
8656 runtime,
8657 forwarding,
8658 module_connection,
8659 module_rx,
8660 handler,
8661 module,
8662 }
8663 }
8664
8665 async fn finish_after(
8666 harness: &mut ProbeHarness,
8667 stall: Duration,
8668 ) -> ModuleControlRpcCompletion {
8669 assert!(stall > harness.runtime.health.deadline);
8670 let deadline = harness.runtime.health.deadline;
8671 let probe = probe_module_health(&harness.spec.module_id, &harness.runtime, None);
8672 let answer = async {
8673 let frame = harness.module_rx.recv().await.expect("health.check frame");
8674 tokio::time::advance(deadline).await;
8675 tokio::task::yield_now().await;
8676 tokio::time::advance(stall - deadline).await;
8677 harness
8678 .forwarding
8679 .complete_module_control_rpc(
8680 harness.module_connection,
8681 frame.header.corr,
8682 Some("health.check"),
8683 ModuleControlRpcOutcome::Response(ModuleControlResponse::HealthCheck {
8684 status: HealthStatus::Ok,
8685 detail: None,
8686 metrics: None,
8687 }),
8688 )
8689 .unwrap()
8690 };
8691 let (probe_result, completion) = tokio::join!(probe, answer);
8692 let err = probe_result.expect_err("probe must miss its deadline");
8693 assert!(matches!(err.evidence, HealthProbeEvidence::NoAnswer));
8694 completion
8695 }
8696
8697 async fn time_out_without_answer(harness: &mut ProbeHarness) {
8698 let deadline = harness.runtime.health.deadline;
8699 let probe = probe_module_health(&harness.spec.module_id, &harness.runtime, None);
8700 let exhaust_deadline = async {
8701 let _frame = harness.module_rx.recv().await.expect("health.check frame");
8702 tokio::time::advance(deadline).await;
8703 tokio::task::yield_now().await;
8704 };
8705 let (probe_result, ()) = tokio::join!(probe, exhaust_deadline);
8706 let err = probe_result.expect_err("probe must miss its deadline");
8707 assert!(matches!(err.evidence, HealthProbeEvidence::NoAnswer));
8708 }
8709
8710 #[tokio::test(start_paused = true)]
8711 async fn late_health_answers_record_start_anchored_latency_for_two_stalls() {
8712 let mut harness = probe_harness();
8713
8714 let first = finish_after(&mut harness, Duration::from_secs(8)).await;
8715 let first_latency = match &first {
8716 ModuleControlRpcCompletion::LateHealthAnswer { latency, .. } => *latency,
8717 other => panic!("late answer was not retained: {other:?}"),
8718 };
8719 assert!(harness.handler.observe_module_control_completion(first));
8720
8721 let second = finish_after(&mut harness, Duration::from_secs(11)).await;
8722 let second_latency = match &second {
8723 ModuleControlRpcCompletion::LateHealthAnswer { latency, .. } => *latency,
8724 other => panic!("late answer was not retained: {other:?}"),
8725 };
8726 assert!(harness.handler.observe_module_control_completion(second));
8727
8728 assert_eq!(first_latency, Duration::from_secs(8));
8729 assert_eq!(
8730 second_latency - first_latency,
8731 Duration::from_secs(3),
8732 "latency must grow linearly with the additional stall"
8733 );
8734 let health = harness.module.status().unwrap().health;
8735 assert_eq!(health.late_answer_count, 2);
8736 assert_eq!(health.last_late_answer_latency_ms, Some(11_000));
8737 }
8738
8739 #[tokio::test(start_paused = true)]
8747 async fn late_answer_clears_the_consecutive_failure_streak() {
8748 let mut harness = probe_harness();
8749
8750 time_out_without_answer(&mut harness).await;
8752 harness
8753 .module
8754 .record_health_probe_failure_for_test("[no-answer] test miss")
8755 .unwrap();
8756 assert_eq!(
8757 harness.module.status().unwrap().health.consecutive_failures,
8758 1,
8759 "precondition: the miss must be on the streak before the late answer"
8760 );
8761
8762 let late = finish_after(&mut harness, Duration::from_secs(9)).await;
8764 assert!(matches!(
8765 late,
8766 ModuleControlRpcCompletion::LateHealthAnswer { .. }
8767 ));
8768 assert!(harness.handler.observe_module_control_completion(late));
8769
8770 let health = harness.module.status().unwrap().health;
8771 assert_eq!(
8772 health.consecutive_failures, 0,
8773 "a late answer is an answer: the streak must reset"
8774 );
8775 assert_eq!(health.late_answer_count, 1);
8776 }
8777
8778 #[tokio::test(start_paused = true)]
8779 async fn repeated_serial_probe_cycles_keep_one_tombstone_per_endpoint() {
8780 let mut harness = probe_harness();
8781
8782 for _ in 0..20 {
8783 time_out_without_answer(&mut harness).await;
8784 assert_eq!(
8785 harness.forwarding.health_probe_tombstone_count().unwrap(),
8786 1
8787 );
8788 }
8789 }
8790}
8791
8792#[cfg(test)]
8793mod child_env_tests {
8794 use super::{
8795 apply_child_env, apply_spawn_role, apply_wire_spawn_args, ModuleProtocol, ModuleSpec,
8796 SpawnRole, SupervisorHandle, SPAWN_ROLE_SWAP_CANDIDATE, SUBC_ARG, SUBC_LAUNCH_NONCE_ENV,
8797 SUBC_MODULE_ID_ENV, SUBC_SPAWN_ROLE_ENV,
8798 };
8799 use std::{ffi::OsStr, path::PathBuf};
8800 use tokio::process::Command;
8801
8802 fn spec(env: Vec<(String, String)>) -> ModuleSpec {
8803 ModuleSpec {
8804 module_id: "env-plan".to_string(),
8805 program: PathBuf::from("/nonexistent"),
8806 args: Vec::new(),
8807 env,
8808 reserved: false,
8809 reserved_prefixes: Vec::new(),
8810 protocol: ModuleProtocol::Subc,
8811 overlap: Default::default(),
8812 }
8813 }
8814
8815 #[test]
8829 fn ambient_ck_log_is_removed_and_a_configured_one_survives() {
8830 let mut command = Command::new("/nonexistent");
8831 apply_child_env(&mut command, &spec(Vec::new()));
8832 let removed = command
8833 .as_std()
8834 .get_envs()
8835 .any(|(key, value)| key == OsStr::new("CK_LOG") && value.is_none());
8836 assert!(
8837 removed,
8838 "ambient CK_LOG must be explicitly removed for an unconfigured module"
8839 );
8840
8841 let mut configured = Command::new("/nonexistent");
8842 apply_child_env(
8843 &mut configured,
8844 &spec(vec![("CK_LOG".to_string(), "debug".to_string())]),
8845 );
8846 let effective = configured
8847 .as_std()
8848 .get_envs()
8849 .filter(|(key, _)| *key == OsStr::new("CK_LOG"))
8850 .last()
8851 .map(|(_, value)| value.map(|v| v.to_string_lossy().into_owned()));
8852 assert_eq!(
8853 effective,
8854 Some(Some("debug".to_string())),
8855 "a module's configured CK_LOG must survive the ambient removal"
8856 );
8857 }
8858
8859 #[test]
8868 fn protocol_none_spawn_carries_no_subc_argument_and_no_nonce() {
8869 let connection_file = std::path::Path::new("/run/subc-connection.json");
8870 let handle = SupervisorHandle::new();
8871
8872 let mut none_spec = spec(Vec::new());
8873 none_spec.protocol = ModuleProtocol::None;
8874 let mut none = Command::new("/nonexistent");
8875 apply_wire_spawn_args(&mut none, &none_spec, Some(connection_file), Some(&handle))
8876 .expect("protocol-none spawn args apply");
8877 let none_args: Vec<String> = none
8878 .as_std()
8879 .get_args()
8880 .map(|a| a.to_string_lossy().into_owned())
8881 .collect();
8882 assert!(
8883 !none_args.iter().any(|a| a == SUBC_ARG),
8884 "protocol:none argv must not carry --subc; got {none_args:?}"
8885 );
8886 let none_has_nonce = none
8887 .as_std()
8888 .get_envs()
8889 .any(|(key, value)| key == OsStr::new(SUBC_LAUNCH_NONCE_ENV) && value.is_some());
8890 assert!(
8891 !none_has_nonce,
8892 "protocol:none spawn must not receive a launch nonce"
8893 );
8894 let none_has_module_id = none
8895 .as_std()
8896 .get_envs()
8897 .any(|(key, value)| key == OsStr::new(SUBC_MODULE_ID_ENV) && value.is_some());
8898 assert!(
8899 none_has_module_id,
8900 "SUBC_MODULE_ID is inert and stays on every path"
8901 );
8902 assert!(
8903 handle.spawn_nonce(&none_spec.module_id).is_none(),
8904 "no nonce record for a process that will never present one"
8905 );
8906
8907 let wire_spec = spec(Vec::new());
8909 let mut wire = Command::new("/nonexistent");
8910 apply_wire_spawn_args(&mut wire, &wire_spec, Some(connection_file), Some(&handle))
8911 .expect("subc-wire spawn args apply");
8912 let wire_args: Vec<String> = wire
8913 .as_std()
8914 .get_args()
8915 .map(|a| a.to_string_lossy().into_owned())
8916 .collect();
8917 assert_eq!(
8918 wire_args,
8919 vec![
8920 SUBC_ARG.to_string(),
8921 connection_file.to_string_lossy().into_owned()
8922 ],
8923 "a subc-wire spawn still carries --subc <path>"
8924 );
8925 assert!(wire
8926 .as_std()
8927 .get_envs()
8928 .any(|(key, value)| key == OsStr::new(SUBC_LAUNCH_NONCE_ENV) && value.is_some()));
8929 assert!(handle.spawn_nonce(&wire_spec.module_id).is_some());
8930 }
8931
8932 #[test]
8942 fn plain_spawn_removes_the_spawn_role_even_when_the_spec_sets_it() {
8943 let role = |command: &Command| {
8944 command
8945 .as_std()
8946 .get_envs()
8947 .filter(|(key, _)| *key == OsStr::new(SUBC_SPAWN_ROLE_ENV))
8948 .last()
8949 .map(|(_, value)| value.map(|v| v.to_string_lossy().into_owned()))
8950 };
8951 let forged = spec(vec![(
8952 SUBC_SPAWN_ROLE_ENV.to_string(),
8953 SPAWN_ROLE_SWAP_CANDIDATE.to_string(),
8954 )]);
8955
8956 let mut plain = Command::new("/nonexistent");
8957 apply_child_env(&mut plain, &forged);
8958 apply_spawn_role(&mut plain, SpawnRole::Plain);
8959 assert_eq!(
8960 role(&plain),
8961 Some(None),
8962 "a plain spawn must remove SUBC_SPAWN_ROLE, whatever the spec says"
8963 );
8964
8965 let mut candidate = Command::new("/nonexistent");
8966 apply_child_env(&mut candidate, &spec(Vec::new()));
8967 apply_spawn_role(&mut candidate, SpawnRole::SwapCandidate);
8968 assert_eq!(
8969 role(&candidate),
8970 Some(Some(SPAWN_ROLE_SWAP_CANDIDATE.to_string()))
8971 );
8972 }
8973
8974 #[test]
8980 fn daemon_private_capture_keys_are_not_passed_to_the_child() {
8981 let mut command = Command::new("/nonexistent");
8982 apply_child_env(
8983 &mut command,
8984 &spec(vec![
8985 (super::CAPTURE_KEEP_ENV.to_string(), "5".to_string()),
8986 ("KEPT".to_string(), "yes".to_string()),
8987 ]),
8988 );
8989 let keys: Vec<String> = command
8990 .as_std()
8991 .get_envs()
8992 .filter(|(_, value)| value.is_some())
8993 .map(|(key, _)| key.to_string_lossy().into_owned())
8994 .collect();
8995 assert!(keys.contains(&"KEPT".to_string()), "got {keys:?}");
8996 assert!(
8997 !keys.contains(&super::CAPTURE_KEEP_ENV.to_string()),
8998 "daemon-private capture key leaked to the child: {keys:?}"
8999 );
9000 }
9001}
9002
9003#[cfg(test)]
9004mod jitter_tests {
9005 use super::jittered_health_delay;
9006 use std::{collections::HashSet, time::Duration};
9007
9008 const FLEET: [&str; 14] = [
9017 "aft",
9018 "alfonso-core",
9019 "magic-context",
9020 "broca",
9021 "thalamus",
9022 "quota",
9023 "engram",
9024 "plexus",
9025 "cerebellum",
9026 "astrocyte",
9027 "synapse",
9028 "subc-mcp",
9029 "cortexkit-credentials",
9030 "subc-federation",
9031 ];
9032
9033 #[test]
9041 fn probe_delays_disperse_across_the_fleet() {
9042 let cadence = Duration::from_secs(30);
9043 let delays: HashSet<Duration> = FLEET
9044 .iter()
9045 .map(|id| jittered_health_delay(id, 0, cadence))
9046 .collect();
9047 assert_eq!(
9048 delays.len(),
9049 FLEET.len(),
9050 "every supervised module must land on its own probe offset"
9051 );
9052 }
9053
9054 #[test]
9060 fn jitter_only_delays_and_stays_within_one_tenth_of_cadence() {
9061 let cadence = Duration::from_secs(30);
9062 let span = cadence / 10;
9063 for id in FLEET {
9064 for probe_index in 0..8 {
9065 let delay = jittered_health_delay(id, probe_index, cadence);
9066 assert!(
9067 delay >= cadence,
9068 "{id}#{probe_index}: jitter must not shorten the cadence"
9069 );
9070 assert!(
9071 delay < cadence + span,
9072 "{id}#{probe_index}: jitter must stay inside one tenth of the cadence"
9073 );
9074 }
9075 }
9076 }
9077
9078 #[test]
9084 fn a_module_offset_is_stable_across_restarts() {
9085 let cadence = Duration::from_secs(30);
9086 for id in FLEET {
9087 assert_eq!(
9088 jittered_health_delay(id, 0, cadence),
9089 jittered_health_delay(id, 0, cadence),
9090 "{id}: the same module and probe index must produce the same offset"
9091 );
9092 }
9093 }
9094
9095 #[test]
9097 fn zero_cadence_yields_zero_delay() {
9098 assert_eq!(
9099 jittered_health_delay("aft", 0, Duration::ZERO),
9100 Duration::ZERO
9101 );
9102 }
9103}
9104
9105#[cfg(all(test, target_os = "linux"))]
9106mod cgroup_placement_tests {
9107 use super::{
9108 apply_cgroup_placement, remove_module_cgroup, ModuleProtocol, ModuleSpec, SuperviseError,
9109 SupervisedChild,
9110 };
9111 use crate::{
9112 stderr_tail::{StderrRing, StderrTailConfig},
9113 test_support::TestTempDir,
9114 };
9115 use std::{
9116 fs, io,
9117 path::{Path, PathBuf},
9118 sync::{Arc, Mutex},
9119 };
9120 use tokio::process::Command;
9121
9122 #[test]
9123 fn failed_parent_cgroup_open_is_a_cgroup_supervision_error() {
9124 let path = Path::new("/definitely-missing-subc-cgroup");
9125 let mut command = Command::new("true");
9126 let error = apply_cgroup_placement(
9127 &mut command,
9128 &ModuleSpec {
9129 module_id: "broken-cgroup".to_string(),
9130 program: PathBuf::from("true"),
9131 args: Vec::new(),
9132 env: Vec::new(),
9133 reserved: false,
9134 reserved_prefixes: Vec::new(),
9135 protocol: ModuleProtocol::Subc,
9136 overlap: Default::default(),
9137 },
9138 path,
9139 )
9140 .expect_err("a parent cgroup open failure must reject the supervised spawn");
9141 let reason = error.to_string();
9142
9143 assert!(
9144 matches!(error, SuperviseError::Cgroup { .. }),
9145 "parent cgroup open must be reported as a cgroup supervision error: {reason}"
9146 );
9147 assert!(
9148 reason.contains("/definitely-missing-subc-cgroup/cgroup.procs"),
9149 "parent cgroup open failure must name cgroup.procs: {reason}"
9150 );
9151 }
9152
9153 #[tokio::test]
9154 async fn reaping_a_child_removes_its_empty_module_cgroup() {
9155 let root = TestTempDir::new("supervisor-reap-cgroup");
9156 fs::write(root.join("cgroup.procs"), b"").expect("write scratch cgroup marker");
9157 let placement = subc_cgroup::prepare_at(&root)
9158 .expect("prepare scratch cgroup root")
9159 .expect("scratch root has a cgroup.procs marker");
9160 let module_id = "reaped-module";
9161 let module = placement
9162 .module_path(module_id)
9163 .expect("create scratch module cgroup");
9164 let child = Command::new("true")
9165 .spawn()
9166 .expect("spawn short-lived child");
9167 let pid = child.id().expect("spawned child has pid");
9168 let mut child = SupervisedChild {
9169 child,
9170 module_id: module_id.to_string(),
9171 cgroup_placement: Some(placement),
9172 stdout_pump: None,
9173 stderr_pump: None,
9174 stderr_ring: Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default()))),
9175 spawned_at_ms: 0,
9176 spawned_from: PathBuf::from("true"),
9177 spawned_file_identity: None,
9178 process_start_time: None,
9179 process_identity: None,
9180 pid,
9181 roster_guard: None,
9182 };
9183
9184 child.wait().await.expect("reap short-lived child");
9185
9186 assert!(
9187 !module.exists(),
9188 "reaping the supervised child must remove its empty cgroup"
9189 );
9190 }
9191
9192 #[test]
9193 fn non_empty_cgroup_removal_is_reported_without_blocking_teardown() {
9194 let root = TestTempDir::new("supervisor-non-empty-cgroup");
9195 fs::write(root.join("cgroup.procs"), b"").expect("write scratch cgroup marker");
9196 let placement = subc_cgroup::prepare_at(&root)
9197 .expect("prepare scratch cgroup root")
9198 .expect("scratch root has a cgroup.procs marker");
9199 let module = placement
9200 .module_path("surviving-module")
9201 .expect("create scratch module cgroup");
9202 fs::write(module.join("surviving-process"), b"still present")
9203 .expect("make scratch cgroup non-empty");
9204 let (logs, _guard) = crate::router::test_log::log_capture(tracing::Level::WARN);
9205
9206 remove_module_cgroup(&placement, "surviving-module");
9207
9208 let logs = crate::router::test_log::captured_logs(&logs);
9209 assert!(
9210 module.exists(),
9211 "failed removal must leave the cgroup intact"
9212 );
9213 assert!(
9214 logs.contains("could not remove module cgroup after process exit; continuing teardown")
9215 && logs.contains("surviving-module"),
9216 "best-effort removal must report the failure without returning it: {logs}"
9217 );
9218 }
9219
9220 #[test]
9221 fn cgroup_pre_exec_spawn_failure_names_the_cgroup_path() {
9222 let cgroup_path = PathBuf::from("/sys/fs/cgroup/subc-modules/broken-module");
9223 let reason = SuperviseError::Spawn {
9224 program: PathBuf::from("/bin/true"),
9225 source: io::Error::from_raw_os_error(13),
9226 cgroup_path: Some(cgroup_path.clone()),
9227 }
9228 .to_string();
9229
9230 assert!(
9231 reason.contains(&cgroup_path.display().to_string()),
9232 "a pre_exec spawn failure must name the cgroup path: {reason}"
9233 );
9234 }
9235}
9236
9237#[cfg(test)]
9238mod spawn_subscriber_lag_tests {
9239 use super::*;
9240
9241 #[tokio::test]
9246 async fn lagged_spawn_subscriber_receives_a_terminal_lagged_error_after_its_queued_frames() {
9247 let feed = SpawnEventFeed::default();
9248 feed.configure_incarnation("lag-incarnation".to_string());
9249 let (tx, mut rx) = mpsc::channel(1);
9252 feed.subscribe(ConnectionId::new(1), 7, 1, None, FrameSink::new(tx))
9253 .expect("subscribe");
9254 let emitted = SPAWN_SUBSCRIBER_BUFFER + 16;
9255 for index in 0..emitted {
9256 feed.emit_spawned(&format!("lag-module-{index}"), 1000, 0);
9257 tokio::task::yield_now().await;
9260 }
9261 assert_eq!(
9262 feed.subscriber_count(),
9263 0,
9264 "the lagged subscriber must be removed"
9265 );
9266
9267 let mut data = Vec::new();
9268 let mut last = None;
9269 loop {
9270 let next = tokio::time::timeout(Duration::from_secs(5), rx.recv())
9271 .await
9272 .expect("the forwarder must finish once the subscriber is dropped");
9273 let Some(outbound) = next else { break };
9274 let frame = outbound.frame;
9275 if frame.header.ty == FrameType::StreamData {
9276 assert!(last.is_none(), "no data may follow the terminal frame");
9277 let event: SpawnEvent = serde_json::from_slice(&frame.body).unwrap();
9278 data.push(event.cursor.seq);
9279 } else {
9280 assert!(last.is_none(), "exactly one terminal frame");
9281 last = Some(frame);
9282 }
9283 }
9284 assert!(!data.is_empty(), "queued frames drain before the terminal");
9285 for pair in data.windows(2) {
9286 assert_eq!(
9287 pair[1],
9288 pair[0] + 1,
9289 "queued frames arrive dense and in order"
9290 );
9291 }
9292 let terminal = last.expect("a lagged subscriber must receive a terminal frame");
9293 assert_eq!(terminal.header.ty, FrameType::Error);
9294 assert_eq!(terminal.header.corr, 7);
9295 let body: subc_protocol::ErrorBody = serde_json::from_slice(&terminal.body).unwrap();
9296 assert_eq!(body.code, SPAWN_SUBSCRIBER_LAGGED_CODE);
9297 let detail = body.detail.expect("lagged error carries detail");
9298 assert_eq!(
9299 detail["first_undelivered_cursor"]["seq"],
9300 data.last().unwrap() + 1,
9301 "the named cursor is the first event the subscriber did not receive"
9302 );
9303 assert_eq!(
9304 detail["first_undelivered_cursor"]["daemon_incarnation"],
9305 "lag-incarnation"
9306 );
9307 }
9308}
9309
9310#[cfg(test)]
9311mod terminal_history_read_concurrency_tests {
9312 use super::*;
9313 use crate::{terminal_journal::read_pause, test_support::TestTempDir};
9314 use std::sync::mpsc as std_mpsc;
9315
9316 fn journaled_ring(
9317 journal: &Arc<crate::terminal_journal::TerminalJournal>,
9318 ) -> Arc<Mutex<TerminalRing>> {
9319 Arc::new(Mutex::new(
9320 TerminalRing::new(TerminalRingConfig::default(), 1)
9321 .with_journal(Some(Arc::clone(journal))),
9322 ))
9323 }
9324
9325 fn crash(at_ms: u64) -> ExitReport {
9326 ExitReport {
9327 kind: ExitKind::Crash,
9328 code: Some(1),
9329 signal: None,
9330 at_ms,
9331 }
9332 }
9333
9334 fn record_within(
9337 module_id: &'static str,
9338 ring: &Arc<Mutex<TerminalRing>>,
9339 at_ms: u64,
9340 bound: Duration,
9341 ) -> bool {
9342 let ring = Arc::clone(ring);
9343 let (done, done_rx) = std_mpsc::channel();
9344 std::thread::spawn(move || {
9345 record_terminal(
9346 module_id,
9347 &ring,
9348 &SpawnEventFeed::default(),
9349 &crash(at_ms),
9350 TerminalDisposition::Restarting,
9351 );
9352 let _ = done.send(());
9353 });
9354 done_rx.recv_timeout(bound).is_ok()
9355 }
9356
9357 #[test]
9362 fn exits_recorded_during_a_paused_history_read_are_not_blocked_or_half_merged() {
9363 let dir = TestTempDir::new("terminal-history-concurrent-read");
9364 let path = dir.join("terminals.jsonl");
9365 let journal = Arc::new(crate::terminal_journal::TerminalJournal::open(
9366 path.clone(),
9367 "daemon".into(),
9368 ));
9369 let reader_ring = journaled_ring(&journal);
9370 let other_ring = journaled_ring(&journal);
9371 assert!(record_within(
9372 "reader-module",
9373 &reader_ring,
9374 10,
9375 Duration::from_secs(5)
9376 ));
9377
9378 let (started, release) = read_pause::install(&path);
9379 let reading = {
9380 let ring = Arc::clone(&reader_ring);
9381 std::thread::spawn(move || durable_terminal_history_of(&ring, "reader-module"))
9382 };
9383 started
9384 .recv_timeout(Duration::from_secs(5))
9385 .expect("the history read reached its pause");
9386
9387 let bound = Duration::from_secs(1);
9388 assert!(
9389 record_within("other-module", &other_ring, 20, bound),
9390 "another module's exit waited on a history read (journal writer held)"
9391 );
9392 assert!(
9393 record_within("reader-module", &reader_ring, 30, bound),
9394 "the read module's own exit waited on its history read (ring held)"
9395 );
9396
9397 drop(release);
9398 let paused = reading.join().unwrap();
9399 assert_eq!(
9400 paused.entries.iter().map(|e| e.at_ms).collect::<Vec<_>>(),
9401 vec![10],
9402 "an exit recorded after the read began lands in neither half of it"
9403 );
9404 assert_eq!(paused.journal_skipped_lines, 0);
9405 assert_eq!(paused.journal_read_errors, 0);
9406
9407 let after = durable_terminal_history_of(&reader_ring, "reader-module");
9408 assert_eq!(
9409 after.entries.iter().map(|e| e.at_ms).collect::<Vec<_>>(),
9410 vec![10, 30],
9411 "the next read merges ring and journal with no duplicate"
9412 );
9413 assert_eq!(after.journal_skipped_lines, 0);
9414 }
9415}
9416
9417#[cfg(test)]
9422mod stderr_settle_tests {
9423 use std::{
9424 future::Future,
9425 io,
9426 pin::Pin,
9427 sync::{Arc, Mutex},
9428 task::{Context, Poll},
9429 time::Duration,
9430 };
9431
9432 use tokio::{
9433 io::{AsyncRead, ReadBuf},
9434 sync::oneshot,
9435 time::Instant,
9436 };
9437
9438 use super::{settle_stderr_pump, StderrPump};
9439 use crate::stderr_tail::{
9440 pump_stderr_to, CaptureState, OutputSink, StderrRing, StderrTailConfig, TailEntry,
9441 };
9442
9443 const BOUND: Duration = Duration::from_millis(250);
9444
9445 struct HeldReader {
9449 before: Option<Vec<u8>>,
9450 gate: Option<oneshot::Receiver<()>>,
9451 after: io::Cursor<Vec<u8>>,
9452 }
9453
9454 impl AsyncRead for HeldReader {
9455 fn poll_read(
9456 mut self: Pin<&mut Self>,
9457 cx: &mut Context<'_>,
9458 buf: &mut ReadBuf<'_>,
9459 ) -> Poll<io::Result<()>> {
9460 if let Some(bytes) = self.before.take() {
9461 buf.put_slice(&bytes);
9462 return Poll::Ready(Ok(()));
9463 }
9464 if let Some(gate) = self.gate.as_mut() {
9465 match Pin::new(gate).poll(cx) {
9466 Poll::Pending => return Poll::Pending,
9467 Poll::Ready(_) => self.gate = None,
9468 }
9469 }
9470 Pin::new(&mut self.after).poll_read(cx, buf)
9471 }
9472 }
9473
9474 struct DiscardSink;
9475
9476 impl OutputSink for DiscardSink {
9477 fn write_line(&mut self, _line: &[u8]) {}
9478 }
9479
9480 fn line(text: &str) -> TailEntry {
9481 TailEntry::Line {
9482 text: text.to_string(),
9483 truncated: false,
9484 }
9485 }
9486
9487 fn lock(ring: &Arc<Mutex<StderrRing>>) -> std::sync::MutexGuard<'_, StderrRing> {
9488 ring.lock().unwrap()
9489 }
9490
9491 fn held_pump(
9495 ring: &Arc<Mutex<StderrRing>>,
9496 before: &str,
9497 after: &str,
9498 ) -> (StderrPump, oneshot::Sender<()>) {
9499 let generation = lock(ring).begin_process();
9500 let (release, gate) = oneshot::channel();
9501 let reader = HeldReader {
9502 before: Some(before.as_bytes().to_vec()),
9503 gate: Some(gate),
9504 after: io::Cursor::new(after.as_bytes().to_vec()),
9505 };
9506 let task = tokio::spawn(pump_stderr_to(
9507 reader,
9508 Arc::clone(ring),
9509 generation,
9510 DiscardSink,
9511 ));
9512 (StderrPump { task, generation }, release)
9513 }
9514
9515 async fn wait_until(ring: &Arc<Mutex<StderrRing>>, done: impl Fn(&StderrRing) -> bool) {
9516 for _ in 0..1000 {
9517 if done(&lock(ring)) {
9518 return;
9519 }
9520 tokio::time::sleep(Duration::from_millis(1)).await;
9521 }
9522 panic!(
9523 "ring never reached the expected state: {:?}",
9524 lock(ring).snapshot(None, None)
9525 );
9526 }
9527
9528 #[tokio::test(start_paused = true)]
9529 async fn a_crash_line_the_reader_had_not_reached_by_the_bound_is_kept_before_the_restart() {
9530 let ring = Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default())));
9531 let (pump, release) = held_pump(&ring, "booting\n", "config error: missing storage\n");
9532
9533 settle_stderr_pump("crasher", &ring, pump, BOUND).await;
9534 let before_release = lock(&ring).snapshot(None, None);
9535 assert!(
9536 matches!(before_release.capture, CaptureState::Incomplete { .. }),
9537 "a reader that has not reached EOF cannot claim a whole tail: {before_release:?}"
9538 );
9539
9540 let next = lock(&ring).begin_process();
9543 lock(&ring).push_line_from(next, "next process booting");
9544 release.send(()).unwrap();
9545 wait_until(&ring, |ring| {
9546 ring.snapshot(None, None).capture == CaptureState::Captured
9547 })
9548 .await;
9549
9550 assert_eq!(
9551 lock(&ring).snapshot(None, None).entries,
9552 vec![
9553 line("booting"),
9554 line("config error: missing storage"),
9555 TailEntry::ProcessStart,
9556 line("next process booting"),
9557 ],
9558 "the crash's last line must survive a slow reader and stay in the crashed process's section"
9559 );
9560 }
9561
9562 #[tokio::test(start_paused = true)]
9563 async fn a_pipe_held_open_by_a_descendant_reads_incomplete_without_delaying_the_restart_past_the_bound(
9564 ) {
9565 let ring = Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default())));
9566 let (pump, _held) = held_pump(&ring, "parent exiting\n", "");
9569
9570 let started = Instant::now();
9571 settle_stderr_pump("orphaning", &ring, pump, BOUND).await;
9572 assert_eq!(
9573 started.elapsed(),
9574 BOUND,
9575 "the restart must wait exactly the bound for a pipe that stays open, no longer"
9576 );
9577
9578 let next = lock(&ring).begin_process();
9579 lock(&ring).push_line_from(next, "next process booting");
9580 tokio::time::sleep(Duration::from_secs(60)).await;
9581
9582 let snapshot = lock(&ring).snapshot(None, None);
9583 match &snapshot.capture {
9584 CaptureState::Incomplete { reason } => assert!(
9585 reason.contains("had not reached EOF") && reason.contains("250ms"),
9586 "the reason must say what is missing and after how long: {reason}"
9587 ),
9588 other => panic!("expected Incomplete while the pipe is held open, got {other:?}"),
9589 }
9590 assert_eq!(
9591 snapshot.entries,
9592 vec![
9593 line("parent exiting"),
9594 TailEntry::ProcessStart,
9595 line("next process booting"),
9596 ]
9597 );
9598 }
9599
9600 #[tokio::test(start_paused = true)]
9601 async fn a_reader_that_reaches_eof_within_the_bound_leaves_the_tail_captured() {
9602 let ring = Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default())));
9603 let (pump, release) = held_pump(&ring, "one\n", "two\n");
9604 release.send(()).unwrap();
9605
9606 settle_stderr_pump("clean", &ring, pump, BOUND).await;
9607
9608 let snapshot = lock(&ring).snapshot(None, None);
9609 assert_eq!(snapshot.capture, CaptureState::Captured);
9610 assert_eq!(snapshot.entries, vec![line("one"), line("two")]);
9611 }
9612}