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 #[cfg(windows)]
143 job: Option<subc_jobobject::JobObject>,
144 stdout_pump: Option<JoinHandle<()>>,
145 stderr_pump: Option<StderrPump>,
146 stderr_ring: Arc<Mutex<StderrRing>>,
147 spawned_at_ms: u64,
148 spawned_from: PathBuf,
149 spawned_file_identity: Option<SpawnedFileIdentity>,
150 process_start_time: Option<u64>,
151 process_identity: Option<ProcessIdentity>,
152 pid: u32,
153 roster_guard: Option<crate::child_roster::RosterGuard>,
156}
157
158impl SupervisedChild {
159 fn id(&self) -> Option<u32> {
160 Some(self.pid)
161 }
162
163 fn process_identity(&self) -> Option<ProcessIdentity> {
164 self.process_identity
165 }
166
167 async fn wait(&mut self) -> io::Result<ExitStatus> {
168 let result = self.child.wait().await;
176 #[cfg(target_os = "linux")]
177 if result.is_ok() {
178 if let Some(placement) = self.cgroup_placement.take() {
179 remove_module_cgroup(&placement, &self.module_id);
180 }
181 }
182 result
183 }
184
185 fn release_roster(&mut self) {
189 self.roster_guard = None;
190 }
191
192 fn start_kill(&mut self) -> io::Result<()> {
205 #[cfg(windows)]
206 if let Some(job) = &self.job {
207 if let Err(error) = job.terminate() {
208 debug!(
209 error = %error,
210 "job termination failed; the direct-child kill still owns the outcome"
211 );
212 }
213 }
214 self.child.start_kill()
215 }
216
217 async fn drain_stderr(&mut self, module_id: &str) {
218 if let Some(mut pump) = self.stdout_pump.take() {
219 match timeout(STDERR_PUMP_DRAIN_TIMEOUT, &mut pump).await {
220 Ok(Ok(())) => {}
221 Ok(Err(error)) => {
222 warn!(module_id, error = %error, "stdout pump ended unexpectedly");
223 }
224 Err(_) => {
225 pump.abort();
226 warn!(
227 module_id,
228 waited = ?STDERR_PUMP_DRAIN_TIMEOUT,
229 "stdout pump did not drain before restart; stopped it before the next process"
230 );
231 }
232 }
233 }
234
235 let Some(pump) = self.stderr_pump.take() else {
236 return;
237 };
238 settle_stderr_pump(
239 module_id,
240 &self.stderr_ring,
241 pump,
242 STDERR_PUMP_DRAIN_TIMEOUT,
243 )
244 .await;
245 }
246}
247
248struct StderrPump {
251 task: JoinHandle<()>,
252 generation: u64,
253}
254
255async fn settle_stderr_pump(
261 module_id: &str,
262 ring: &Arc<Mutex<StderrRing>>,
263 pump: StderrPump,
264 bound: Duration,
265) {
266 let lock = || ring.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
267 let StderrPump {
268 mut task,
269 generation,
270 } = pump;
271 lock().retire_pump(generation);
272 match timeout(bound, &mut task).await {
273 Ok(Ok(())) => {}
274 Ok(Err(err)) => {
275 let mut ring = lock();
276 ring.mark_incomplete(format!("stderr pump ended unexpectedly: {err}"));
277 ring.finish_pump(generation);
278 warn!(module_id, error = %err, "stderr pump ended before clean EOF");
279 }
280 Err(_) => {
281 drop(task);
283 lock().mark_pump_late(
284 generation,
285 format!(
286 "stderr of the exited process had not reached EOF {bound:?} after it was \
287 retired (a descendant may still hold the pipe open); lines it still \
288 writes are kept in that process's section"
289 ),
290 );
291 warn!(
292 module_id,
293 waited = ?bound,
294 "stderr pipe of the exited process is still open; its reader keeps running without delaying the restart"
295 );
296 }
297 }
298}
299
300fn registration_release_events() -> &'static watch::Sender<u64> {
301 static EVENTS: OnceLock<watch::Sender<u64>> = OnceLock::new();
302 EVENTS.get_or_init(|| {
303 let (sender, _receiver) = watch::channel(0);
304 sender
305 })
306}
307
308pub(crate) fn notify_registration_release() {
309 let events = registration_release_events();
310 let next_generation = (*events.borrow()).wrapping_add(1);
311 events.send_replace(next_generation);
312}
313
314#[derive(Debug, Clone, PartialEq, Eq)]
316pub struct ModuleSpec {
317 pub module_id: String,
318 pub program: PathBuf,
319 pub args: Vec<String>,
320 pub env: Vec<(String, String)>,
321 pub reserved: bool,
326 pub reserved_prefixes: Vec<String>,
331 pub protocol: ModuleProtocol,
347 pub overlap: ModuleOverlap,
352}
353
354#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
361pub enum ModuleOverlap {
362 #[default]
364 Exclusive,
365 Safe,
377}
378
379impl ModuleOverlap {
380 pub fn as_str(self) -> &'static str {
381 match self {
382 Self::Exclusive => "exclusive",
383 Self::Safe => "safe",
384 }
385 }
386}
387
388pub const SUBC_SPAWN_ROLE_ENV: &str = "SUBC_SPAWN_ROLE";
398pub const SPAWN_ROLE_SWAP_CANDIDATE: &str = "swap_candidate";
400pub const DEFAULT_SWAP_READY_TIMEOUT: Duration = Duration::from_secs(100);
405
406#[derive(Debug, Clone, Copy, PartialEq, Eq)]
424pub struct RestartPolicy {
425 pub max_restarts: u32,
426 pub backoff: Duration,
429 pub max_backoff: Duration,
431 pub window: Duration,
435}
436
437impl RestartPolicy {
438 pub fn new(max_restarts: u32, backoff: Duration) -> Self {
442 Self {
443 max_restarts,
444 backoff,
445 max_backoff: DEFAULT_MAX_BACKOFF,
446 window: DEFAULT_RESTART_WINDOW,
447 }
448 }
449
450 pub fn with_max_backoff(mut self, max_backoff: Duration) -> Self {
451 self.max_backoff = max_backoff;
452 self
453 }
454
455 pub fn with_window(mut self, window: Duration) -> Self {
456 self.window = window;
457 self
458 }
459
460 fn delay_for_restart(&self, restart_in_window: u32) -> Duration {
465 if self.backoff.is_zero() || self.max_backoff.is_zero() {
466 return Duration::ZERO;
467 }
468
469 let mut delay = self.backoff;
470 for _ in 0..restart_in_window {
471 if delay >= self.max_backoff {
472 return self.max_backoff;
473 }
474 delay = delay
475 .checked_mul(10)
476 .unwrap_or(self.max_backoff)
477 .min(self.max_backoff);
478 }
479 delay.min(self.max_backoff)
480 }
481
482 fn budget_exhausted_detail(&self) -> String {
487 format!(
488 "crash budget exhausted: max_restarts={} within window_secs={}",
489 self.max_restarts,
490 self.window.as_secs()
491 )
492 }
493}
494
495impl Default for RestartPolicy {
496 fn default() -> Self {
497 Self {
498 max_restarts: DEFAULT_MAX_RESTARTS,
499 backoff: DEFAULT_BACKOFF,
500 max_backoff: DEFAULT_MAX_BACKOFF,
501 window: DEFAULT_RESTART_WINDOW,
502 }
503 }
504}
505
506#[derive(Debug, Clone, Copy, PartialEq, Eq)]
507struct CrashRestartSchedule {
508 restart_in_window: u32,
509 delay: Duration,
510}
511
512fn daemon_will_restart(
519 state: &mut SupervisorSnapshot,
520 policy: &RestartPolicy,
521 now: Instant,
522) -> bool {
523 state.enabled && state.crash_restarts_in_window(policy.window, now) < policy.max_restarts
524}
525
526const DEFAULT_HEALTH_CADENCE: Duration = Duration::from_secs(30);
527const DEFAULT_HEALTH_DEADLINE: Duration = Duration::from_secs(5);
528const DEFAULT_HEALTH_FAILURE_THRESHOLD: u32 = 3;
529const MAX_HEALTH_METRICS_BYTES: usize = 16 * 1024;
530
531#[derive(Debug, Clone, Copy, PartialEq, Eq)]
532pub enum HealthAction {
533 Report,
534 Restart,
535 Alert,
536}
537
538impl fmt::Display for HealthAction {
539 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
540 f.write_str(match self {
541 Self::Report => "report",
542 Self::Restart => "restart",
543 Self::Alert => "alert",
544 })
545 }
546}
547
548#[derive(Debug, Clone, Copy, PartialEq, Eq)]
549pub struct HealthConfig {
550 pub cadence: Duration,
551 pub deadline: Duration,
552 pub failure_threshold: u32,
553 pub on_degraded: HealthAction,
554 pub on_failing: HealthAction,
555 pub critical: bool,
556}
557
558impl Default for HealthConfig {
559 fn default() -> Self {
560 Self {
561 cadence: DEFAULT_HEALTH_CADENCE,
562 deadline: DEFAULT_HEALTH_DEADLINE,
563 failure_threshold: DEFAULT_HEALTH_FAILURE_THRESHOLD,
564 on_degraded: HealthAction::Report,
565 on_failing: HealthAction::Report,
566 critical: false,
567 }
568 }
569}
570
571#[derive(Debug, Clone, PartialEq)]
589pub struct ModuleHealthStatus {
590 pub status: SupervisorHealthStatus,
591 pub last_probe_ms: Option<u64>,
592 pub detail: Option<String>,
593 pub metrics: Option<Value>,
594 pub consecutive_failures: u32,
595 pub late_answer_count: u64,
598 pub last_late_answer_latency_ms: Option<u64>,
600 pub last_action: Option<String>,
601 pub last_action_ms: Option<u64>,
605}
606
607impl Default for ModuleHealthStatus {
608 fn default() -> Self {
609 Self {
610 status: SupervisorHealthStatus::Unknown,
611 last_probe_ms: None,
612 detail: None,
613 metrics: None,
614 consecutive_failures: 0,
615 late_answer_count: 0,
616 last_late_answer_latency_ms: None,
617 last_action: None,
618 last_action_ms: None,
619 }
620 }
621}
622
623#[derive(Debug, Clone, Copy, PartialEq, Eq)]
625pub enum ModuleState {
626 Starting,
627 Running,
628 Unresponsive,
629 Restarting,
630 Draining,
631 Stopped,
632 Failed,
633 Disabled,
634}
635
636impl fmt::Display for ModuleState {
637 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
638 f.write_str(match self {
639 Self::Starting => "starting",
640 Self::Running => "running",
641 Self::Unresponsive => "unresponsive",
642 Self::Restarting => "restarting",
643 Self::Draining => "draining",
644 Self::Stopped => "stopped",
645 Self::Failed => "failed",
646 Self::Disabled => "disabled",
647 })
648 }
649}
650
651#[derive(Debug, Clone, Copy, PartialEq, Eq)]
653pub enum ExitKind {
654 Clean,
655 Crash,
656 DeliberateSeverance,
657}
658
659impl From<ExitKind> for TerminalExitKind {
660 fn from(kind: ExitKind) -> Self {
661 match kind {
662 ExitKind::Clean => Self::Clean,
663 ExitKind::Crash => Self::Crash,
664 ExitKind::DeliberateSeverance => Self::DeliberateSeverance,
665 }
666 }
667}
668
669#[derive(Debug, Clone, Copy, PartialEq, Eq)]
672pub(crate) struct ProcessIdentity {
673 pub(crate) pid: u32,
674 pub(crate) start_time: u64,
675}
676
677#[derive(Debug, Clone, PartialEq, Eq)]
679pub struct ExitReport {
680 pub kind: ExitKind,
681 pub code: Option<i32>,
682 pub signal: Option<i32>,
683 pub at_ms: u64,
684}
685
686#[derive(Debug, Clone, PartialEq)]
689pub struct ModuleStatus {
690 pub module_id: String,
691 pub state: ModuleState,
692 pub enabled: bool,
693 pub process_alive: bool,
694 pub registration_active: bool,
695 pub protocol: ModuleProtocol,
698 pub live: bool,
709 pub restart_count: u32,
713 pub lifetime_restarts: u32,
717 pub spawn_generation: u64,
718 pub max_restarts: u32,
723 pub restart_window: Duration,
727 pub drain_timeout: Duration,
731 pub restart_backoff: Duration,
732 pub restart_max_backoff: Duration,
733 pub pid: Option<u32>,
734 pub spawned_at_ms: Option<u64>,
735 pub spawned_from: Option<PathBuf>,
736 pub process_start_time: Option<u64>,
737 pub last_exit: Option<ExitReport>,
738 pub health: ModuleHealthStatus,
739}
740
741#[derive(Debug, Clone, PartialEq)]
742struct SupervisorSnapshot {
743 state: ModuleState,
744 enabled: bool,
745 process_alive: bool,
746 crash_restarts: VecDeque<Instant>,
752 lifetime_restarts: u32,
753 spawn_generation: u64,
762 pid: Option<u32>,
763 spawned_at_ms: Option<u64>,
764 spawned_from: Option<PathBuf>,
765 spawned_file_identity: Option<SpawnedFileIdentity>,
766 process_start_time: Option<u64>,
767 deliberate_severance: Option<ProcessIdentity>,
768 last_exit: Option<ExitReport>,
769 health: ModuleHealthStatus,
770 in_alternate_slot: bool,
775 draining_to_replace: bool,
782 configuration_updated_since_spawn: bool,
788}
789
790impl SupervisorSnapshot {
791 fn starting() -> Self {
792 Self::new(ModuleState::Starting, true)
793 }
794
795 fn disabled() -> Self {
796 Self::new(ModuleState::Disabled, false)
797 }
798
799 fn failed() -> Self {
800 Self::new(ModuleState::Failed, true)
801 }
802
803 fn crash_restarts_in_window(&mut self, window: Duration, now: Instant) -> u32 {
807 while let Some(oldest) = self.crash_restarts.front() {
808 if now.duration_since(*oldest) > window {
809 self.crash_restarts.pop_front();
810 } else {
811 break;
812 }
813 }
814 u32::try_from(self.crash_restarts.len()).unwrap_or(u32::MAX)
815 }
816
817 fn record_crash_restart(&mut self, policy: &RestartPolicy, now: Instant) {
823 self.crash_restarts.push_back(now);
824 while self.crash_restarts.len() > policy.max_restarts as usize {
825 self.crash_restarts.pop_front();
826 }
827 self.lifetime_restarts += 1;
828 }
829
830 fn next_crash_restart(
834 &mut self,
835 policy: &RestartPolicy,
836 now: Instant,
837 ) -> Option<CrashRestartSchedule> {
838 let restart_in_window = self.crash_restarts_in_window(policy.window, now);
839 if restart_in_window >= policy.max_restarts {
840 return None;
841 }
842 self.record_crash_restart(policy, now);
843 Some(CrashRestartSchedule {
844 restart_in_window,
845 delay: policy.delay_for_restart(restart_in_window),
846 })
847 }
848
849 fn clear_crash_restarts(&mut self) {
854 self.crash_restarts.clear();
855 }
856
857 fn new(state: ModuleState, enabled: bool) -> Self {
858 Self {
859 state,
860 enabled,
861 process_alive: false,
862 crash_restarts: VecDeque::new(),
863 lifetime_restarts: 0,
864 spawn_generation: 0,
865 pid: None,
866 spawned_at_ms: None,
867 spawned_from: None,
868 spawned_file_identity: None,
869 process_start_time: None,
870 deliberate_severance: None,
871 last_exit: None,
872 health: ModuleHealthStatus::default(),
873 in_alternate_slot: false,
874 draining_to_replace: false,
875 configuration_updated_since_spawn: false,
876 }
877 }
878}
879
880type SharedSnapshot = Arc<Mutex<SupervisorSnapshot>>;
881
882type SpawnSubscriberKey = (ConnectionId, u64);
883
884#[derive(Debug)]
885struct SpawnSubscriber {
886 version: u8,
887 frames: mpsc::Sender<Frame>,
888 lagged: Option<oneshot::Sender<SpawnCursor>>,
892}
893
894#[derive(Debug)]
895struct SpawnEventState {
896 daemon_incarnation: String,
897 seq: u64,
898 capacity: usize,
899 live: HashMap<String, LiveSpawn>,
900 generations: HashMap<String, u64>,
901 events: VecDeque<SpawnEvent>,
902 subscribers: HashMap<SpawnSubscriberKey, SpawnSubscriber>,
903}
904
905impl Default for SpawnEventState {
906 fn default() -> Self {
907 Self {
908 daemon_incarnation: "unconfigured".to_string(),
909 seq: 0,
910 capacity: SPAWN_EVENT_RING_CAPACITY,
911 live: HashMap::new(),
912 generations: HashMap::new(),
913 events: VecDeque::new(),
914 subscribers: HashMap::new(),
915 }
916 }
917}
918
919#[derive(Debug, Clone, Default)]
920struct SpawnEventFeed(Arc<Mutex<SpawnEventState>>);
921
922#[derive(Debug, Clone, PartialEq, Eq)]
923pub(crate) enum SpawnSubscribeRefusal {
924 ForeignIncarnation { current: String },
925 TooOld { oldest: SpawnCursor },
926 Frame(String),
927}
928
929impl SpawnEventFeed {
930 fn configure_incarnation(&self, daemon_incarnation: String) {
931 let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
932 state.daemon_incarnation = daemon_incarnation;
933 state.seq = 0;
934 state.live.clear();
935 state.generations.clear();
936 state.events.clear();
937 state.subscribers.clear();
938 }
939
940 fn cursor(state: &SpawnEventState) -> SpawnCursor {
941 SpawnCursor {
942 daemon_incarnation: state.daemon_incarnation.clone(),
943 seq: state.seq,
944 }
945 }
946
947 fn snapshot(&self) -> SpawnSnapshot {
948 let state = self.0.lock().unwrap_or_else(|p| p.into_inner());
949 let mut live = state.live.values().cloned().collect::<Vec<_>>();
950 live.sort_by(|left, right| left.module_id.cmp(&right.module_id));
951 SpawnSnapshot {
952 cursor: Self::cursor(&state),
953 ring_bound: state.capacity as u64,
954 live,
955 }
956 }
957
958 fn emit_spawned(&self, module_id: &str, pid: u32, spawned_at_ms: u64) -> u64 {
959 let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
960 let generation = state
961 .generations
962 .get(module_id)
963 .copied()
964 .unwrap_or(0)
965 .checked_add(1)
966 .expect("spawn generation exhausted");
967 state.generations.insert(module_id.to_string(), generation);
968 let live = LiveSpawn {
969 module_id: module_id.to_string(),
970 spawn_generation: generation,
971 pid,
972 spawned_at_ms,
973 };
974 state.live.insert(module_id.to_string(), live);
975 Self::emit_locked(
976 &mut state,
977 SpawnEventKind::Spawned,
978 module_id.to_string(),
979 generation,
980 pid,
981 None,
982 None,
983 );
984 generation
985 }
986
987 fn emit_exited(&self, module_id: &str, exit_code: Option<i32>, exit_signal: Option<i32>) {
988 let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
989 let Some(live) = state.live.remove(module_id) else {
990 warn!(
991 module_id,
992 "terminal record had no live spawn event identity"
993 );
994 return;
995 };
996 Self::emit_locked(
997 &mut state,
998 SpawnEventKind::Exited,
999 module_id.to_string(),
1000 live.spawn_generation,
1001 live.pid,
1002 exit_code,
1003 exit_signal,
1004 );
1005 }
1006
1007 fn emit_superseded_exited(
1014 &self,
1015 module_id: &str,
1016 spawn_generation: u64,
1017 pid: u32,
1018 exit_code: Option<i32>,
1019 exit_signal: Option<i32>,
1020 ) {
1021 let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
1022 if state
1023 .live
1024 .get(module_id)
1025 .is_some_and(|live| live.spawn_generation == spawn_generation)
1026 {
1027 state.live.remove(module_id);
1028 }
1029 Self::emit_locked(
1030 &mut state,
1031 SpawnEventKind::Exited,
1032 module_id.to_string(),
1033 spawn_generation,
1034 pid,
1035 exit_code,
1036 exit_signal,
1037 );
1038 }
1039
1040 #[allow(clippy::too_many_arguments)]
1041 fn emit_locked(
1042 state: &mut SpawnEventState,
1043 kind: SpawnEventKind,
1044 module_id: String,
1045 spawn_generation: u64,
1046 pid: u32,
1047 exit_code: Option<i32>,
1048 exit_signal: Option<i32>,
1049 ) {
1050 state.seq = state
1051 .seq
1052 .checked_add(1)
1053 .expect("spawn event sequence exhausted");
1054 let event = SpawnEvent {
1055 cursor: Self::cursor(state),
1056 kind,
1057 module_id,
1058 spawn_generation,
1059 pid,
1060 exit_code,
1061 exit_signal,
1062 };
1063 state.events.push_back(event.clone());
1064 while state.events.len() > state.capacity {
1065 state.events.pop_front();
1066 }
1067 let body = match serde_json::to_vec(&event) {
1068 Ok(body) => body,
1069 Err(error) => {
1070 error!(%error, "failed to serialize supervisor spawn event");
1071 return;
1072 }
1073 };
1074 state.subscribers.retain(|(connection_id, corr), subscriber| {
1075 let frame = Frame::build_with_version(
1076 subscriber.version,
1077 FrameType::StreamData,
1078 control_flags(),
1079 0,
1080 0,
1081 *corr,
1082 body.clone(),
1083 );
1084 match frame {
1085 Ok(frame) => {
1086 if subscriber.frames.try_send(frame).is_ok() {
1087 true
1088 } else {
1089 warn!(connection_id = connection_id.get(), corr, "dropping lagged supervisor spawn subscriber");
1090 if let Some(lagged) = subscriber.lagged.take() {
1091 let _ = lagged.send(event.cursor.clone());
1092 }
1093 false
1094 }
1095 }
1096 Err(error) => {
1097 warn!(connection_id = connection_id.get(), corr, %error, "dropping supervisor spawn subscriber after frame build failure");
1098 false
1099 }
1100 }
1101 });
1102 }
1103
1104 fn subscribe(
1105 &self,
1106 connection_id: ConnectionId,
1107 corr: u64,
1108 version: u8,
1109 since: Option<SpawnCursor>,
1110 sink: FrameSink,
1111 ) -> Result<(), SpawnSubscribeRefusal> {
1112 let (frames, mut receiver) = mpsc::channel(SPAWN_SUBSCRIBER_BUFFER);
1113 let (lagged, mut lagged_rx) = oneshot::channel::<SpawnCursor>();
1114 {
1115 let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
1116 let replay = if let Some(since) = since {
1117 if since.daemon_incarnation != state.daemon_incarnation {
1118 return Err(SpawnSubscribeRefusal::ForeignIncarnation {
1119 current: state.daemon_incarnation.clone(),
1120 });
1121 }
1122 if let Some(oldest) = state.events.front().map(|event| event.cursor.clone()) {
1123 if since.seq < oldest.seq.saturating_sub(1) {
1124 return Err(SpawnSubscribeRefusal::TooOld { oldest });
1125 }
1126 }
1127 state
1128 .events
1129 .iter()
1130 .filter(|event| event.cursor.seq > since.seq)
1131 .cloned()
1132 .collect::<Vec<_>>()
1133 } else {
1134 Vec::new()
1135 };
1136 for event in replay {
1137 let body = serde_json::to_vec(&event)
1138 .map_err(|error| SpawnSubscribeRefusal::Frame(error.to_string()))?;
1139 let frame = Frame::build_with_version(
1140 version,
1141 FrameType::StreamData,
1142 control_flags(),
1143 0,
1144 0,
1145 corr,
1146 body,
1147 )
1148 .map_err(|error| SpawnSubscribeRefusal::Frame(error.to_string()))?;
1149 frames
1150 .try_send(frame)
1151 .map_err(|error| SpawnSubscribeRefusal::Frame(error.to_string()))?;
1152 }
1153 state.subscribers.insert(
1154 (connection_id, corr),
1155 SpawnSubscriber {
1156 version,
1157 frames,
1158 lagged: Some(lagged),
1159 },
1160 );
1161 }
1162 tokio::spawn(async move {
1173 while let Some(frame) = receiver.recv().await {
1174 if sink.send(frame).await.is_err() {
1175 return;
1176 }
1177 }
1178 let Ok(first_undelivered) = lagged_rx.try_recv() else {
1179 return;
1180 };
1181 match spawn_subscriber_lagged_frame(version, corr, first_undelivered) {
1182 Ok(frame) => {
1183 let _ = sink.send(frame).await;
1184 }
1185 Err(error) => {
1186 error!(%error, corr, "failed to build lagged spawn subscriber terminal frame");
1187 }
1188 }
1189 });
1190 Ok(())
1191 }
1192
1193 fn cancel(&self, connection_id: ConnectionId, corr: u64) -> bool {
1194 let Some(subscriber) = self
1195 .0
1196 .lock()
1197 .unwrap_or_else(|p| p.into_inner())
1198 .subscribers
1199 .remove(&(connection_id, corr))
1200 else {
1201 return false;
1202 };
1203 if let Ok(frame) = Frame::build_with_version(
1204 subscriber.version,
1205 FrameType::StreamEnd,
1206 control_flags(),
1207 0,
1208 0,
1209 corr,
1210 Vec::new(),
1211 ) {
1212 tokio::spawn(async move {
1213 let _ = subscriber.frames.send(frame).await;
1214 });
1215 }
1216 true
1217 }
1218
1219 fn remove_connection(&self, connection_id: ConnectionId) {
1220 self.0
1221 .lock()
1222 .unwrap_or_else(|p| p.into_inner())
1223 .subscribers
1224 .retain(|(subscriber_connection, _), _| *subscriber_connection != connection_id);
1225 }
1226
1227 #[cfg(any(test, feature = "test-support"))]
1228 fn set_capacity(&self, capacity: usize) {
1229 self.0.lock().unwrap_or_else(|p| p.into_inner()).capacity = capacity;
1230 }
1231
1232 #[cfg(any(test, feature = "test-support"))]
1233 fn subscriber_count(&self) -> usize {
1234 self.0
1235 .lock()
1236 .unwrap_or_else(|p| p.into_inner())
1237 .subscribers
1238 .len()
1239 }
1240}
1241
1242fn spawn_subscriber_lagged_frame(
1245 version: u8,
1246 corr: u64,
1247 first_undelivered: SpawnCursor,
1248) -> Result<Frame, String> {
1249 let body = serde_json::to_vec(&subc_protocol::ErrorBody {
1250 code: SPAWN_SUBSCRIBER_LAGGED_CODE.to_string(),
1251 message: "spawn subscriber fell behind and was dropped; resubscribe from the last cursor received"
1252 .to_string(),
1253 detail: Some(serde_json::json!({
1254 "first_undelivered_cursor": first_undelivered
1255 })),
1256 })
1257 .map_err(|error| error.to_string())?;
1258 Frame::build_with_version(version, FrameType::Error, control_flags(), 0, 0, corr, body)
1259 .map_err(|error| error.to_string())
1260}
1261
1262pub trait ModuleProcessLiveness: Send + Sync {
1263 fn process_live(&self, module_id: &str) -> Option<bool>;
1264
1265 fn process_replacing(&self, _module_id: &str) -> bool {
1271 false
1272 }
1273}
1274
1275#[derive(Debug, Clone, Default)]
1277pub struct SupervisorProcessLiveness {
1278 snapshots: Arc<Mutex<HashMap<String, SharedSnapshot>>>,
1279}
1280
1281impl SupervisorProcessLiveness {
1282 pub fn new() -> Self {
1283 Self::default()
1284 }
1285
1286 fn track(&self, module_id: String, snapshot: SharedSnapshot) {
1287 let mut snapshots = self
1288 .snapshots
1289 .lock()
1290 .unwrap_or_else(|poisoned| poisoned.into_inner());
1291 snapshots.insert(module_id, snapshot);
1292 }
1293
1294 fn untrack_if_current(&self, module_id: &str, snapshot: &SharedSnapshot) {
1295 let mut snapshots = self
1296 .snapshots
1297 .lock()
1298 .unwrap_or_else(|poisoned| poisoned.into_inner());
1299 let is_current = snapshots
1300 .get(module_id)
1301 .map(|tracked| Arc::ptr_eq(tracked, snapshot))
1302 .unwrap_or(false);
1303 if is_current {
1304 snapshots.remove(module_id);
1305 }
1306 }
1307}
1308
1309impl ModuleProcessLiveness for SupervisorProcessLiveness {
1310 fn process_live(&self, module_id: &str) -> Option<bool> {
1311 let snapshot = {
1312 let snapshots = self
1313 .snapshots
1314 .lock()
1315 .unwrap_or_else(|poisoned| poisoned.into_inner());
1316 snapshots.get(module_id).cloned()
1317 }?;
1318 let snapshot = snapshot
1319 .lock()
1320 .unwrap_or_else(|poisoned| poisoned.into_inner());
1321 Some(snapshot.state == ModuleState::Running && snapshot.process_alive)
1322 }
1323
1324 fn process_replacing(&self, module_id: &str) -> bool {
1325 let Some(snapshot) = self
1326 .snapshots
1327 .lock()
1328 .unwrap_or_else(|poisoned| poisoned.into_inner())
1329 .get(module_id)
1330 .cloned()
1331 else {
1332 return false;
1333 };
1334 let snapshot = snapshot
1335 .lock()
1336 .unwrap_or_else(|poisoned| poisoned.into_inner());
1337 snapshot.enabled
1338 && match snapshot.state {
1339 ModuleState::Restarting => true,
1340 ModuleState::Draining => snapshot.draining_to_replace,
1341 ModuleState::Starting
1342 | ModuleState::Running
1343 | ModuleState::Unresponsive
1344 | ModuleState::Stopped
1345 | ModuleState::Failed
1346 | ModuleState::Disabled => false,
1347 }
1348 }
1349}
1350
1351#[derive(Debug, Clone)]
1352struct SupervisorRuntimeConfig {
1353 restart_policy: RestartPolicy,
1354 drain_timeout: Duration,
1357 effective_drain_timeout: Arc<Mutex<Duration>>,
1360 default_drain_timeout: Duration,
1363 health: HealthConfig,
1364 connection_file_path: Option<PathBuf>,
1365 capture_logs_dir: Option<PathBuf>,
1366 forwarding: Option<Arc<ForwardingTable>>,
1367 supervisor_handle: Option<SupervisorHandle>,
1370 stderr_ring: Arc<Mutex<StderrRing>>,
1377 terminal_ring: Arc<Mutex<TerminalRing>>,
1378 spawn_events: SpawnEventFeed,
1379 child_roster: ChildRoster,
1380 #[cfg(target_os = "linux")]
1381 cgroup_placement: Option<subc_cgroup::Placement>,
1382 #[cfg(test)]
1383 test_seed_stale_facts_before_enable_spawn: bool,
1384}
1385
1386#[derive(Debug, Clone, PartialEq, Eq)]
1387struct SupervisedConfiguration {
1388 spec: ModuleSpec,
1389 health: HealthConfig,
1390}
1391
1392#[derive(Debug, Clone, Default)]
1398pub struct SupervisorHandle {
1399 modules: Arc<Mutex<HashMap<String, SupervisedModule>>>,
1400 spawn_events: SpawnEventFeed,
1401 reserved_nonces: Arc<Mutex<HashMap<String, Option<String>>>>,
1412 removal_tombstones: Arc<Mutex<HashMap<String, u64>>>,
1418 spawn_nonces: Arc<Mutex<HashMap<String, String>>>,
1422 reserved_prefix_owners: Arc<Mutex<HashMap<String, String>>>,
1430 swaps: Arc<Mutex<HashMap<String, OpenSwap>>>,
1436 promotion_observer: PromotionObserverSlot,
1438 operation_lock: Arc<AsyncMutex<()>>,
1442}
1443
1444pub(crate) trait SwapPromotionObserver: Send + Sync {
1453 fn swap_promoted(&self, registration: &crate::registry::ModuleRegistration);
1454}
1455
1456#[derive(Clone, Default)]
1460struct PromotionObserverSlot(Arc<Mutex<Option<std::sync::Weak<dyn SwapPromotionObserver>>>>);
1461
1462impl fmt::Debug for PromotionObserverSlot {
1463 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1464 f.write_str("PromotionObserverSlot")
1465 }
1466}
1467
1468#[derive(Debug, Clone)]
1470struct OpenSwap {
1471 candidate_nonce: String,
1474 incumbent_nonce: Option<String>,
1479 candidate_admitted: bool,
1483}
1484
1485#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1488pub(crate) enum SwapHelloAdmission {
1489 NotSwapping,
1492 Candidate,
1494 Refused,
1497}
1498
1499#[derive(Debug, Clone, PartialEq, Eq)]
1500pub(crate) enum ReservedHelloRejection {
1501 Exact {
1502 module_id: String,
1503 },
1504 Prefix {
1505 prefix: String,
1506 owner_module_id: String,
1507 },
1508}
1509
1510impl SupervisorHandle {
1511 pub fn new() -> Self {
1512 Self::default()
1513 }
1514
1515 pub(crate) fn spawn_snapshot(&self) -> SpawnSnapshot {
1516 self.spawn_events.snapshot()
1517 }
1518
1519 pub(crate) fn subscribe_spawns(
1520 &self,
1521 connection_id: ConnectionId,
1522 corr: u64,
1523 version: u8,
1524 since: Option<SpawnCursor>,
1525 sink: FrameSink,
1526 ) -> Result<(), SpawnSubscribeRefusal> {
1527 self.spawn_events
1528 .subscribe(connection_id, corr, version, since, sink)
1529 }
1530
1531 pub(crate) fn cancel_spawn_subscription(&self, connection_id: ConnectionId, corr: u64) -> bool {
1532 self.spawn_events.cancel(connection_id, corr)
1533 }
1534
1535 pub(crate) fn remove_spawn_subscribers(&self, connection_id: ConnectionId) {
1536 self.spawn_events.remove_connection(connection_id);
1537 }
1538
1539 #[cfg(any(test, feature = "test-support"))]
1540 pub fn set_spawn_event_capacity_for_test(&self, capacity: usize) {
1541 assert!(capacity > 0, "spawn event capacity must be non-zero");
1542 self.spawn_events.set_capacity(capacity);
1543 }
1544
1545 #[cfg(any(test, feature = "test-support"))]
1546 pub fn spawn_subscriber_count_for_test(&self) -> usize {
1547 self.spawn_events.subscriber_count()
1548 }
1549
1550 pub fn set_spawn_nonce(&self, module_id: &str, nonce: String) {
1553 self.spawn_nonces
1554 .lock()
1555 .unwrap_or_else(|poisoned| poisoned.into_inner())
1556 .insert(module_id.to_string(), nonce);
1557 }
1558
1559 pub fn set_reserved_nonce(&self, module_id: &str, nonce: String) {
1562 self.reserved_nonces
1563 .lock()
1564 .unwrap_or_else(|poisoned| poisoned.into_inner())
1565 .insert(module_id.to_string(), Some(nonce));
1566 }
1567
1568 pub fn set_reserved_prefixes(&self, owner_module_id: &str, prefixes: &[String]) {
1570 let mut owners = self
1571 .reserved_prefix_owners
1572 .lock()
1573 .unwrap_or_else(|poisoned| poisoned.into_inner());
1574 owners.retain(|_, owner| owner != owner_module_id);
1575 for prefix in prefixes {
1576 owners.insert(prefix.clone(), owner_module_id.to_string());
1577 }
1578 }
1579
1580 #[cfg(test)]
1582 pub(crate) fn spawn_nonce(&self, module_id: &str) -> Option<String> {
1583 self.spawn_nonces
1584 .lock()
1585 .unwrap_or_else(|poisoned| poisoned.into_inner())
1586 .get(module_id)
1587 .cloned()
1588 }
1589
1590 fn apply_identity_configuration(&self, spec: &ModuleSpec) {
1591 self.set_reserved_prefixes(&spec.module_id, &spec.reserved_prefixes);
1592 let spawn_nonce = self
1593 .spawn_nonces
1594 .lock()
1595 .unwrap_or_else(|poisoned| poisoned.into_inner())
1596 .get(&spec.module_id)
1597 .cloned();
1598 let mut reserved_nonces = self
1599 .reserved_nonces
1600 .lock()
1601 .unwrap_or_else(|poisoned| poisoned.into_inner());
1602 if spec.reserved {
1603 reserved_nonces.insert(spec.module_id.clone(), spawn_nonce);
1608 }
1609 drop(reserved_nonces);
1610 self.removal_tombstones
1614 .lock()
1615 .unwrap_or_else(|poisoned| poisoned.into_inner())
1616 .remove(&spec.module_id);
1617 }
1618
1619 pub fn reserved_hello_authorized(&self, module_id: &str, presented: Option<&str>) -> bool {
1624 self.reserved_hello_rejection(module_id, presented)
1625 .is_none()
1626 }
1627
1628 pub(crate) fn reserved_hello_rejection(
1629 &self,
1630 module_id: &str,
1631 presented: Option<&str>,
1632 ) -> Option<ReservedHelloRejection> {
1633 let nonces = self
1634 .reserved_nonces
1635 .lock()
1636 .unwrap_or_else(|poisoned| poisoned.into_inner());
1637 if let Some(expected) = nonces.get(module_id) {
1638 let authorized = match expected {
1642 Some(expected) => {
1643 presented.is_some_and(|p| constant_time_eq(expected.as_bytes(), p.as_bytes()))
1644 }
1645 None => false,
1646 };
1647 if authorized {
1648 return None;
1649 }
1650 return Some(ReservedHelloRejection::Exact {
1651 module_id: module_id.to_string(),
1652 });
1653 }
1654 drop(nonces);
1655
1656 let matched_prefix = self
1657 .reserved_prefix_owners
1658 .lock()
1659 .unwrap_or_else(|poisoned| poisoned.into_inner())
1660 .iter()
1661 .filter(|(prefix, _)| module_id.starts_with(prefix.as_str()))
1662 .max_by_key(|(prefix, _)| prefix.len())
1663 .map(|(prefix, owner)| (prefix.clone(), owner.clone()));
1664 let (prefix, owner_module_id) = matched_prefix?;
1665
1666 let authorized = presented.is_some_and(|presented| {
1667 self.spawn_nonces
1668 .lock()
1669 .unwrap_or_else(|poisoned| poisoned.into_inner())
1670 .get(&owner_module_id)
1671 .is_some_and(|expected| constant_time_eq(expected.as_bytes(), presented.as_bytes()))
1672 || self.swap_nonce_matches(&owner_module_id, presented)
1675 });
1676 if authorized {
1677 None
1678 } else {
1679 Some(ReservedHelloRejection::Prefix {
1680 prefix,
1681 owner_module_id,
1682 })
1683 }
1684 }
1685
1686 pub fn spawned_consumer_authorized(&self, module_id: &str, presented: &str) -> bool {
1691 if presented.is_empty() {
1692 return false;
1693 }
1694 let nonces = self
1695 .spawn_nonces
1696 .lock()
1697 .unwrap_or_else(|poisoned| poisoned.into_inner());
1698 let current = nonces
1699 .get(module_id)
1700 .is_some_and(|expected| constant_time_eq(expected.as_bytes(), presented.as_bytes()));
1701 drop(nonces);
1702 current || self.swap_nonce_matches(module_id, presented)
1707 }
1708
1709 fn swap_nonce_matches(&self, module_id: &str, presented: &str) -> bool {
1711 let swaps = self
1712 .swaps
1713 .lock()
1714 .unwrap_or_else(|poisoned| poisoned.into_inner());
1715 swaps.get(module_id).is_some_and(|swap| {
1716 constant_time_eq(swap.candidate_nonce.as_bytes(), presented.as_bytes())
1717 || swap.incumbent_nonce.as_deref().is_some_and(|incumbent| {
1718 constant_time_eq(incumbent.as_bytes(), presented.as_bytes())
1719 })
1720 })
1721 }
1722
1723 pub(crate) fn open_swap(&self, module_id: &str, candidate_nonce: String) {
1726 let incumbent_nonce = self
1727 .spawn_nonces
1728 .lock()
1729 .unwrap_or_else(|poisoned| poisoned.into_inner())
1730 .get(module_id)
1731 .cloned();
1732 self.swaps
1733 .lock()
1734 .unwrap_or_else(|poisoned| poisoned.into_inner())
1735 .insert(
1736 module_id.to_string(),
1737 OpenSwap {
1738 candidate_nonce,
1739 incumbent_nonce,
1740 candidate_admitted: false,
1741 },
1742 );
1743 }
1744
1745 pub(crate) fn close_swap(&self, module_id: &str) {
1748 self.swaps
1749 .lock()
1750 .unwrap_or_else(|poisoned| poisoned.into_inner())
1751 .remove(module_id);
1752 }
1753
1754 pub(crate) fn set_swap_promotion_observer(
1757 &self,
1758 observer: std::sync::Weak<dyn SwapPromotionObserver>,
1759 ) {
1760 *self
1761 .promotion_observer
1762 .0
1763 .lock()
1764 .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(observer);
1765 }
1766
1767 fn notify_swap_promoted(&self, registration: &crate::registry::ModuleRegistration) {
1770 let observer = self
1771 .promotion_observer
1772 .0
1773 .lock()
1774 .unwrap_or_else(|poisoned| poisoned.into_inner())
1775 .as_ref()
1776 .and_then(std::sync::Weak::upgrade);
1777 if let Some(observer) = observer {
1778 observer.swap_promoted(registration);
1779 }
1780 }
1781
1782 pub(crate) fn swap_open(&self, module_id: &str) -> bool {
1784 self.swaps
1785 .lock()
1786 .unwrap_or_else(|poisoned| poisoned.into_inner())
1787 .contains_key(module_id)
1788 }
1789
1790 fn promote_swap_nonce(&self, module_id: &str, reserved: bool) {
1795 let candidate_nonce = self
1796 .swaps
1797 .lock()
1798 .unwrap_or_else(|poisoned| poisoned.into_inner())
1799 .get(module_id)
1800 .map(|swap| swap.candidate_nonce.clone());
1801 let Some(nonce) = candidate_nonce else {
1802 return;
1803 };
1804 self.set_spawn_nonce(module_id, nonce.clone());
1805 if reserved {
1806 self.set_reserved_nonce(module_id, nonce);
1807 }
1808 }
1809
1810 pub(crate) fn swap_hello_admission(
1825 &self,
1826 module_id: &str,
1827 presented: Option<&str>,
1828 ) -> SwapHelloAdmission {
1829 let swaps = self
1830 .swaps
1831 .lock()
1832 .unwrap_or_else(|poisoned| poisoned.into_inner());
1833 let Some(swap) = swaps.get(module_id) else {
1834 return SwapHelloAdmission::NotSwapping;
1835 };
1836 let Some(presented) = presented else {
1837 return SwapHelloAdmission::Refused;
1838 };
1839 if constant_time_eq(swap.candidate_nonce.as_bytes(), presented.as_bytes()) {
1840 return if swap.candidate_admitted {
1841 SwapHelloAdmission::Refused
1842 } else {
1843 SwapHelloAdmission::Candidate
1844 };
1845 }
1846 if swap
1847 .incumbent_nonce
1848 .as_deref()
1849 .is_some_and(|incumbent| constant_time_eq(incumbent.as_bytes(), presented.as_bytes()))
1850 {
1851 return SwapHelloAdmission::NotSwapping;
1852 }
1853 SwapHelloAdmission::Refused
1854 }
1855
1856 pub(crate) fn mark_swap_candidate_admitted(&self, module_id: &str) {
1859 if let Some(swap) = self
1860 .swaps
1861 .lock()
1862 .unwrap_or_else(|poisoned| poisoned.into_inner())
1863 .get_mut(module_id)
1864 {
1865 swap.candidate_admitted = true;
1866 }
1867 }
1868
1869 pub fn spawn_launch_nonce_for(&self, module_id: &str) -> Option<String> {
1871 self.spawn_nonces
1872 .lock()
1873 .unwrap_or_else(|poisoned| poisoned.into_inner())
1874 .get(module_id)
1875 .cloned()
1876 }
1877
1878 pub fn reserved_launch_nonce_for(&self, module_id: &str) -> Option<String> {
1880 self.reserved_nonces
1881 .lock()
1882 .unwrap_or_else(|poisoned| poisoned.into_inner())
1883 .get(module_id)
1884 .cloned()
1885 .flatten()
1886 }
1887
1888 pub fn insert(&self, module: SupervisedModule) -> Option<SupervisedModule> {
1889 let mut modules = self
1890 .modules
1891 .lock()
1892 .unwrap_or_else(|poisoned| poisoned.into_inner());
1893 modules.insert(module.module_id().to_string(), module)
1894 }
1895
1896 pub fn get(&self, module_id: &str) -> Option<SupervisedModule> {
1897 let modules = self
1898 .modules
1899 .lock()
1900 .unwrap_or_else(|poisoned| poisoned.into_inner());
1901 modules.get(module_id).cloned()
1902 }
1903
1904 pub(crate) fn record_late_health_answer(
1905 &self,
1906 module_id: &str,
1907 latency_ms: u64,
1908 ) -> Result<bool, SuperviseError> {
1909 let Some(module) = self.get(module_id) else {
1910 return Ok(false);
1911 };
1912 update_snapshot(&module.inner.snapshot, Some(module_id), |state| {
1913 state.health.late_answer_count = state.health.late_answer_count.saturating_add(1);
1914 state.health.last_late_answer_latency_ms = Some(latency_ms);
1915 state.health.consecutive_failures = 0;
1923 })?;
1924 Ok(true)
1925 }
1926
1927 pub fn record_deliberate_severance(&self, module_id: &str) -> Result<bool, SuperviseError> {
1933 let Some(module) = self.get(module_id) else {
1934 return Ok(false);
1935 };
1936 let status = module.status()?;
1937 let Some((pid, start_time)) = status.pid.zip(status.process_start_time) else {
1938 return Ok(false);
1939 };
1940 module.record_deliberate_severance(ProcessIdentity { pid, start_time })
1941 }
1942
1943 pub fn list(&self) -> Vec<SupervisedModule> {
1944 let modules = self
1945 .modules
1946 .lock()
1947 .unwrap_or_else(|poisoned| poisoned.into_inner());
1948 let mut modules = modules.values().cloned().collect::<Vec<_>>();
1949 modules.sort_by(|left, right| left.module_id().cmp(right.module_id()));
1950 modules
1951 }
1952
1953 pub(crate) fn retire(&self, module_id: &str) -> Option<SupervisedModule> {
1954 self.spawn_nonces
1955 .lock()
1956 .unwrap_or_else(|poisoned| poisoned.into_inner())
1957 .remove(module_id);
1958 self.close_swap(module_id);
1959 let mut reserved_nonces = self
1960 .reserved_nonces
1961 .lock()
1962 .unwrap_or_else(|poisoned| poisoned.into_inner());
1963 if reserved_nonces.contains_key(module_id) {
1964 reserved_nonces.insert(module_id.to_string(), None);
1967 }
1968 drop(reserved_nonces);
1969 self.reserved_prefix_owners
1970 .lock()
1971 .unwrap_or_else(|poisoned| poisoned.into_inner())
1972 .retain(|_, owner| owner != module_id);
1973 self.modules
1974 .lock()
1975 .unwrap_or_else(|poisoned| poisoned.into_inner())
1976 .remove(module_id)
1977 }
1978
1979 pub(crate) fn record_rescan_removal(&self, module_id: &str) {
1982 self.removal_tombstones
1983 .lock()
1984 .unwrap_or_else(|poisoned| poisoned.into_inner())
1985 .insert(module_id.to_string(), unix_ms_now());
1986 }
1987
1988 pub(crate) fn removal_tombstone_age_ms(&self, module_id: &str) -> Option<u64> {
1990 self.removal_tombstones
1991 .lock()
1992 .unwrap_or_else(|poisoned| poisoned.into_inner())
1993 .get(module_id)
1994 .copied()
1995 .map(|removed_at_ms| unix_ms_now().saturating_sub(removed_at_ms))
1996 }
1997
1998 pub(crate) fn release_retained_reserved_gate(&self, module_id: &str) -> bool {
2003 if self.get(module_id).is_some() {
2004 return false;
2005 }
2006 let mut reserved_nonces = self
2007 .reserved_nonces
2008 .lock()
2009 .unwrap_or_else(|poisoned| poisoned.into_inner());
2010 if !matches!(reserved_nonces.get(module_id), Some(None)) {
2011 return false;
2012 }
2013 reserved_nonces.remove(module_id);
2014 true
2015 }
2016
2017 pub(crate) fn operation_lock(&self) -> Arc<AsyncMutex<()>> {
2018 Arc::clone(&self.operation_lock)
2019 }
2020}
2021
2022#[derive(Debug, Clone)]
2024pub struct Supervisor {
2025 registry: Arc<Registry>,
2026 restart_policy: RestartPolicy,
2027 drain_timeout: Duration,
2028 connection_file_path: Option<PathBuf>,
2029 capture_logs_dir: Option<PathBuf>,
2030 forwarding: Option<Arc<ForwardingTable>>,
2031 process_liveness: Arc<SupervisorProcessLiveness>,
2032 supervisor_handle: Option<SupervisorHandle>,
2033 health: HealthConfig,
2034 daemon_start_clock: crate::clock::StartClock,
2035 terminal_journal: Option<Arc<crate::terminal_journal::TerminalJournal>>,
2036 spawn_events: SpawnEventFeed,
2037 provenance_probe: ExecutableIdentityProbe,
2038 child_roster: ChildRoster,
2041 #[cfg(target_os = "linux")]
2042 cgroup_placement: Option<subc_cgroup::Placement>,
2043}
2044
2045impl Supervisor {
2046 #[cfg(unix)]
2057 pub(crate) fn begin_daemon_shutdown(&self) {
2058 self.child_roster.close();
2059 if let Some(journal) = &self.terminal_journal {
2060 journal.stamp_shutdown();
2061 }
2062 }
2063
2064 #[cfg(unix)]
2068 pub(crate) async fn drain_for_daemon_shutdown(&self) -> Result<(), SuperviseError> {
2069 const NOTICE_BUDGET: Duration = Duration::from_millis(500);
2070 const DRAIN_BUDGET: Duration = Duration::from_secs(2);
2071 let Some(forwarding) = &self.forwarding else {
2072 return Ok(());
2073 };
2074 let module_ids = forwarding
2075 .begin_daemon_drain()
2076 .map_err(SuperviseError::Forwarding)?;
2077 let deadline_ms =
2078 unix_ms_now().saturating_add((NOTICE_BUDGET + DRAIN_BUDGET).as_millis() as u64);
2079 let mut notices = tokio::task::JoinSet::new();
2080 let mut drains = Vec::new();
2081 for module_id in module_ids {
2082 let Some(target) = forwarding
2083 .begin_module_drain(&module_id, RouteCloseReason::Restart)
2084 .map_err(SuperviseError::Forwarding)?
2085 else {
2086 continue;
2087 };
2088 let routes = forwarding
2089 .endpoint_routes(target.endpoint)
2090 .map_err(SuperviseError::Forwarding)?;
2091 let command = serde_json::to_vec(&ModuleControlCommand::Draining {
2097 reason: RouteCloseReason::Restart,
2098 deadline_ms,
2099 })
2100 .expect("module draining serializes");
2101 let closing = serde_json::to_vec(&ClientControlPush::RouteClosing {
2102 module_id: module_id.clone(),
2103 reason: RouteCloseReason::Restart,
2104 })
2105 .expect("route closing serializes");
2106 let mut recipients = vec![(target.sink.clone(), target.negotiated_ver, command)];
2107 let mut seen = std::collections::HashSet::new();
2108 for route in routes {
2109 let client = route.goodbye_target;
2110 if seen.insert(client.connection_id) {
2111 recipients.push((client.sink, client.negotiated_ver, closing.clone()));
2112 }
2113 }
2114 for (sink, version, body) in recipients {
2115 notices.spawn(async move {
2116 let frame = Frame::build_with_version(
2117 version,
2118 FrameType::Push,
2119 control_flags(),
2120 0,
2121 0,
2122 0,
2123 body,
2124 )
2125 .expect("bounded lifecycle notice frame builds");
2126 sink.send_flushed(frame).await
2127 });
2128 }
2129 let gauges = declared_busy_gauges(&self.registry, &module_id)?;
2130 drains.push((module_id, target.endpoint, gauges));
2131 }
2132 let notice_deadline = Instant::now() + NOTICE_BUDGET;
2135 while let Ok(Some(result)) = timeout_at(notice_deadline, notices.join_next()).await {
2136 if !matches!(result, Ok(Ok(()))) {
2137 warn!(?result, "daemon shutdown notice delivery failed");
2138 }
2139 }
2140 notices.abort_all();
2141 let deadline = Instant::now() + DRAIN_BUDGET;
2142 let mut waits = tokio::task::JoinSet::new();
2143 for (module_id, endpoint, gauges) in drains {
2144 let forwarding = Arc::clone(forwarding);
2145 let mut runtime = self.runtime_config();
2146 runtime.health.cadence = Duration::from_millis(100);
2147 waits.spawn(async move {
2148 wait_for_forwarding_quiescence(
2149 &forwarding,
2150 &module_id,
2151 &runtime,
2152 endpoint,
2153 deadline,
2154 &gauges,
2155 DrainScope::Active,
2156 )
2157 .await
2158 });
2159 }
2160 while let Ok(Some(result)) = timeout_at(deadline, waits.join_next()).await {
2161 if !matches!(result, Ok(Ok(true))) {
2162 warn!(?result, "daemon shutdown drain did not reach quiescence");
2163 }
2164 }
2165 Ok(())
2166 }
2167
2168 #[cfg(unix)]
2178 pub(crate) async fn end_children_for_daemon_shutdown(
2179 &self,
2180 already_escalated: bool,
2181 escalate: impl std::future::Future<Output = ()>,
2182 ) {
2183 if let Some(forwarding) = &self.forwarding {
2184 let closed = forwarding.close_all_connections(&CloseReason::new(
2185 "daemon_shutdown",
2186 "the daemon is exiting after its shutdown notice and drain",
2187 ));
2188 debug!(closed, "closed established connections for daemon shutdown");
2189 }
2190 crate::child_roster::end_children_for_daemon_shutdown(
2191 &self.child_roster,
2192 already_escalated,
2193 escalate,
2194 )
2195 .await;
2196 }
2197
2198 pub fn new(registry: Arc<Registry>, restart_policy: RestartPolicy) -> Self {
2199 Self {
2200 registry,
2201 restart_policy,
2202 drain_timeout: DEFAULT_DRAIN_TIMEOUT,
2203 connection_file_path: None,
2204 capture_logs_dir: None,
2205 forwarding: None,
2206 process_liveness: Arc::new(SupervisorProcessLiveness::default()),
2207 supervisor_handle: None,
2208 health: HealthConfig::default(),
2209 daemon_start_clock: crate::clock::StartClock::capture(),
2210 terminal_journal: None,
2211 spawn_events: SpawnEventFeed::default(),
2212 provenance_probe: ExecutableIdentityProbe::default(),
2213 child_roster: ChildRoster::default(),
2214 #[cfg(target_os = "linux")]
2215 cgroup_placement: None,
2216 }
2217 }
2218
2219 pub fn with_drain_timeout(mut self, drain_timeout: Duration) -> Self {
2220 self.drain_timeout = drain_timeout;
2221 self
2222 }
2223
2224 pub fn with_process_liveness(
2225 mut self,
2226 process_liveness: Arc<SupervisorProcessLiveness>,
2227 ) -> Self {
2228 self.process_liveness = process_liveness;
2229 self
2230 }
2231
2232 pub fn with_connection_file_path(mut self, connection_file_path: impl Into<PathBuf>) -> Self {
2233 self.connection_file_path = Some(connection_file_path.into());
2234 self
2235 }
2236
2237 pub fn with_capture_logs_dir(mut self, logs_dir: impl Into<PathBuf>) -> Self {
2239 self.capture_logs_dir = Some(logs_dir.into());
2240 self
2241 }
2242
2243 pub fn with_daemon_incarnation(self, daemon_incarnation: String) -> Self {
2246 self.spawn_events.configure_incarnation(daemon_incarnation);
2250 self
2251 }
2252
2253 pub fn with_terminal_journal(self, path: PathBuf, daemon_incarnation: String) -> Self {
2256 let mut this = self.with_daemon_incarnation(daemon_incarnation.clone());
2257 this.terminal_journal = Some(Arc::new(crate::terminal_journal::TerminalJournal::open(
2258 path,
2259 daemon_incarnation,
2260 )));
2261 this
2262 }
2263
2264 pub fn with_forwarding(mut self, forwarding: Arc<ForwardingTable>) -> Self {
2265 self.forwarding = Some(forwarding);
2266 self
2267 }
2268
2269 pub fn with_handle(mut self, supervisor_handle: SupervisorHandle) -> Self {
2270 self.spawn_events = supervisor_handle.spawn_events.clone();
2271 self.supervisor_handle = Some(supervisor_handle);
2272 self
2273 }
2274
2275 pub fn with_health_config(mut self, health: HealthConfig) -> Self {
2276 self.health = health;
2277 self
2278 }
2279
2280 pub fn with_live_children_record(self, path: impl Into<PathBuf>) -> Self {
2284 self.child_roster.record_to(path.into());
2285 self
2286 }
2287
2288 #[cfg(target_os = "linux")]
2289 pub fn with_cgroup_placement(
2290 mut self,
2291 cgroup_placement: Option<subc_cgroup::Placement>,
2292 ) -> Self {
2293 self.cgroup_placement = cgroup_placement;
2294 self
2295 }
2296
2297 pub fn spawn(&self, spec: ModuleSpec) -> Result<SupervisedModule, SuperviseError> {
2303 validate_spec(&spec)?;
2304
2305 let runtime = self.runtime_config();
2306 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
2307 let child = spawn_child(
2308 &spec,
2309 runtime.connection_file_path.as_deref(),
2310 self.supervisor_handle.as_ref(),
2311 &runtime.stderr_ring,
2312 runtime.capture_logs_dir.as_deref(),
2313 &runtime.child_roster,
2314 #[cfg(target_os = "linux")]
2315 runtime.cgroup_placement.as_ref(),
2316 )?;
2317 set_running(&snapshot, &child, &spec.module_id, &runtime.spawn_events)?;
2318 self.process_liveness
2319 .track(spec.module_id.clone(), Arc::clone(&snapshot));
2320
2321 Ok(self.supervised_module(spec, runtime, snapshot, Some(child)))
2322 }
2323
2324 pub fn supervise_configured(
2330 &self,
2331 spec: ModuleSpec,
2332 enabled: bool,
2333 ) -> Result<SupervisedModule, SuperviseError> {
2334 validate_spec(&spec)?;
2335
2336 let runtime = self.runtime_config();
2337 if !enabled {
2338 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::disabled()));
2339 return Ok(self.supervised_module(spec, runtime, snapshot, None));
2340 }
2341
2342 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
2343 match spawn_child(
2344 &spec,
2345 runtime.connection_file_path.as_deref(),
2346 self.supervisor_handle.as_ref(),
2347 &runtime.stderr_ring,
2348 runtime.capture_logs_dir.as_deref(),
2349 &runtime.child_roster,
2350 #[cfg(target_os = "linux")]
2351 runtime.cgroup_placement.as_ref(),
2352 ) {
2353 Ok(child) => {
2354 set_running(&snapshot, &child, &spec.module_id, &runtime.spawn_events)?;
2355 self.process_liveness
2356 .track(spec.module_id.clone(), Arc::clone(&snapshot));
2357 Ok(self.supervised_module(spec, runtime, snapshot, Some(child)))
2358 }
2359 Err(err) => {
2360 error!(
2361 module_id = %spec.module_id,
2362 program = %spec.program.display(),
2363 error = %err,
2364 "configured module failed to spawn; marking failed and continuing"
2365 );
2366 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::failed()));
2367 Ok(self.supervised_module(spec, runtime, snapshot, None))
2368 }
2369 }
2370 }
2371
2372 pub fn supervise_configured_with_health(
2378 &self,
2379 spec: ModuleSpec,
2380 enabled: bool,
2381 health: HealthConfig,
2382 drain_timeout_ms: Option<u64>,
2383 restart_policy: RestartPolicy,
2384 ) -> Result<SupervisedModule, SuperviseError> {
2385 validate_spec(&spec)?;
2386
2387 let mut runtime = self.runtime_config();
2388 runtime.health = health;
2389 runtime.restart_policy = restart_policy;
2390 if let Some(ms) = drain_timeout_ms {
2391 runtime.drain_timeout = Duration::from_millis(ms);
2392 *runtime
2393 .effective_drain_timeout
2394 .lock()
2395 .unwrap_or_else(|poisoned| poisoned.into_inner()) = runtime.drain_timeout;
2396 }
2397 if !enabled {
2398 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::disabled()));
2399 return Ok(self.supervised_module(spec, runtime, snapshot, None));
2400 }
2401
2402 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
2403 match spawn_child(
2404 &spec,
2405 runtime.connection_file_path.as_deref(),
2406 self.supervisor_handle.as_ref(),
2407 &runtime.stderr_ring,
2408 runtime.capture_logs_dir.as_deref(),
2409 &runtime.child_roster,
2410 #[cfg(target_os = "linux")]
2411 runtime.cgroup_placement.as_ref(),
2412 ) {
2413 Ok(child) => {
2414 set_running(&snapshot, &child, &spec.module_id, &runtime.spawn_events)?;
2415 self.process_liveness
2416 .track(spec.module_id.clone(), Arc::clone(&snapshot));
2417 Ok(self.supervised_module(spec, runtime, snapshot, Some(child)))
2418 }
2419 Err(err) => {
2420 if health.critical {
2421 error!(
2422 module_id = %spec.module_id,
2423 program = %spec.program.display(),
2424 error = %err,
2425 "critical configured module failed to spawn; marking failed and alerting"
2426 );
2427 } else {
2428 error!(
2429 module_id = %spec.module_id,
2430 program = %spec.program.display(),
2431 error = %err,
2432 "configured module failed to spawn; marking failed and continuing"
2433 );
2434 }
2435 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::failed()));
2436 Ok(self.supervised_module(spec, runtime, snapshot, None))
2437 }
2438 }
2439 }
2440
2441 fn runtime_config(&self) -> SupervisorRuntimeConfig {
2442 let effective_drain_timeout = Arc::new(Mutex::new(self.drain_timeout));
2443 SupervisorRuntimeConfig {
2444 restart_policy: self.restart_policy,
2445 drain_timeout: self.drain_timeout,
2446 child_roster: self
2449 .child_roster
2450 .for_module(Arc::clone(&effective_drain_timeout)),
2451 effective_drain_timeout,
2452 default_drain_timeout: self.drain_timeout,
2453 health: self.health,
2454 connection_file_path: self.connection_file_path.clone(),
2455 capture_logs_dir: self.capture_logs_dir.clone(),
2456 forwarding: self.forwarding.clone(),
2457 supervisor_handle: self.supervisor_handle.clone(),
2458 stderr_ring: Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default()))),
2459 terminal_ring: Arc::new(Mutex::new(
2460 TerminalRing::new(
2461 TerminalRingConfig::default(),
2462 self.daemon_start_clock.started_at_ms(),
2463 )
2464 .with_start_clock(self.daemon_start_clock)
2465 .with_journal(self.terminal_journal.clone())
2466 .with_daemon_shutdown(self.child_roster.shutdown_flag()),
2467 )),
2468 spawn_events: self.spawn_events.clone(),
2469 #[cfg(target_os = "linux")]
2470 cgroup_placement: self.cgroup_placement.clone(),
2471 #[cfg(test)]
2472 test_seed_stale_facts_before_enable_spawn: false,
2473 }
2474 }
2475
2476 fn supervised_module(
2477 &self,
2478 spec: ModuleSpec,
2479 runtime: SupervisorRuntimeConfig,
2480 snapshot: SharedSnapshot,
2481 child: Option<SupervisedChild>,
2482 ) -> SupervisedModule {
2483 let configuration = Arc::new(Mutex::new(SupervisedConfiguration {
2484 spec: spec.clone(),
2485 health: runtime.health,
2486 }));
2487 let stderr_ring = Arc::clone(&runtime.stderr_ring);
2488 let terminal_ring = Arc::clone(&runtime.terminal_ring);
2489 let restart_policy = runtime.restart_policy;
2493 let effective_drain_timeout = Arc::clone(&runtime.effective_drain_timeout);
2494 let (tx, rx) = mpsc::channel(4);
2495 let monitor = tokio::spawn(supervise_loop(
2496 spec.clone(),
2497 runtime,
2498 Arc::clone(&self.registry),
2499 Arc::clone(&self.process_liveness),
2500 Arc::clone(&snapshot),
2501 child,
2502 rx,
2503 ));
2504
2505 let module_id = spec.module_id.clone();
2506 let module = SupervisedModule {
2507 inner: Arc::new(SupervisedModuleInner {
2508 module_id: module_id.clone(),
2509 registry: Arc::clone(&self.registry),
2510 snapshot,
2511 configuration,
2512 stderr_ring,
2513 terminal_ring,
2514 commands: tx,
2515 monitor: Mutex::new(Some(monitor)),
2516 restart_policy,
2517 effective_drain_timeout,
2518 provenance_probe: self.provenance_probe.clone(),
2519 }),
2520 };
2521 if let Some(supervisor_handle) = &self.supervisor_handle {
2522 supervisor_handle.apply_identity_configuration(&spec);
2523 supervisor_handle.insert(module.clone());
2524 }
2525 module
2526 }
2527}
2528
2529impl Default for Supervisor {
2530 fn default() -> Self {
2531 Self::new(Arc::new(Registry::default()), RestartPolicy::default())
2532 }
2533}
2534
2535#[derive(Clone)]
2537pub struct SupervisedModule {
2538 inner: Arc<SupervisedModuleInner>,
2539}
2540
2541struct SupervisedModuleInner {
2542 module_id: String,
2543 registry: Arc<Registry>,
2544 snapshot: SharedSnapshot,
2545 configuration: Arc<Mutex<SupervisedConfiguration>>,
2546 stderr_ring: Arc<Mutex<StderrRing>>,
2547 terminal_ring: Arc<Mutex<TerminalRing>>,
2548 commands: mpsc::Sender<SupervisorCommand>,
2549 monitor: Mutex<Option<JoinHandle<()>>>,
2550 restart_policy: RestartPolicy,
2554 effective_drain_timeout: Arc<Mutex<Duration>>,
2555 provenance_probe: ExecutableIdentityProbe,
2556}
2557
2558impl fmt::Debug for SupervisedModule {
2559 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2560 f.debug_struct("SupervisedModule")
2561 .field("module_id", &self.inner.module_id)
2562 .field("status", &self.status())
2563 .finish_non_exhaustive()
2564 }
2565}
2566
2567impl SupervisedModule {
2568 pub fn module_id(&self) -> &str {
2569 &self.inner.module_id
2570 }
2571
2572 #[cfg(test)]
2576 pub(crate) fn record_health_probe_failure_for_test(
2577 &self,
2578 detail: &str,
2579 ) -> Result<(), SuperviseError> {
2580 update_snapshot(&self.inner.snapshot, Some(&self.inner.module_id), |state| {
2581 state.health.consecutive_failures = state.health.consecutive_failures.saturating_add(1);
2582 state.health.detail = Some(detail.to_string());
2583 })
2584 }
2585
2586 pub fn state(&self) -> Result<ModuleState, SuperviseError> {
2587 Ok(lock_snapshot(&self.inner.snapshot)?.state)
2588 }
2589
2590 pub fn stderr_tail(
2597 &self,
2598 max_lines: Option<usize>,
2599 max_bytes: Option<usize>,
2600 ) -> StderrTailSnapshot {
2601 self.inner
2602 .stderr_ring
2603 .lock()
2604 .unwrap_or_else(|poisoned| poisoned.into_inner())
2605 .snapshot(max_lines, max_bytes)
2606 }
2607
2608 pub fn terminal_history(&self) -> TerminalHistorySnapshot {
2613 self.inner
2614 .terminal_ring
2615 .lock()
2616 .unwrap_or_else(|poisoned| poisoned.into_inner())
2617 .snapshot()
2618 }
2619
2620 pub fn durable_terminal_history(&self) -> subc_control::TerminalHistory {
2625 durable_terminal_history_of(&self.inner.terminal_ring, &self.inner.module_id)
2626 }
2627
2628 pub(crate) async fn read_durable_terminal_history(
2633 &self,
2634 ) -> Result<subc_control::TerminalHistory, tokio::task::JoinError> {
2635 let terminal_ring = Arc::clone(&self.inner.terminal_ring);
2636 let module_id = self.inner.module_id.clone();
2637 tokio::task::spawn_blocking(move || durable_terminal_history_of(&terminal_ring, &module_id))
2638 .await
2639 }
2640
2641 pub fn status(&self) -> Result<ModuleStatus, SuperviseError> {
2642 self.status_with_snapshot_lock(&self.inner.snapshot, None)
2643 }
2644
2645 pub(crate) fn record_deliberate_severance(
2646 &self,
2647 identity: ProcessIdentity,
2648 ) -> Result<bool, SuperviseError> {
2649 let mut snapshot = lock_snapshot(&self.inner.snapshot)?;
2650 if snapshot.pid != Some(identity.pid)
2651 || snapshot.process_start_time != Some(identity.start_time)
2652 {
2653 return Ok(false);
2654 }
2655 snapshot.deliberate_severance = Some(identity);
2656 Ok(true)
2657 }
2658
2659 pub(crate) fn status_for_control(
2664 &self,
2665 caller: &'static str,
2666 ) -> Result<ModuleStatus, SuperviseError> {
2667 self.status_with_snapshot_lock(&self.inner.snapshot, Some(caller))
2668 }
2669
2670 fn status_with_snapshot_lock(
2671 &self,
2672 snapshot: &SharedSnapshot,
2673 caller: Option<&'static str>,
2674 ) -> Result<ModuleStatus, SuperviseError> {
2675 let mut guard = match caller {
2676 Some(caller) => lock_snapshot_for_control(snapshot, &self.inner.module_id, caller)?,
2677 None => lock_snapshot(snapshot)?,
2678 };
2679 let restart_count =
2682 guard.crash_restarts_in_window(self.inner.restart_policy.window, Instant::now());
2683 let snapshot = guard.clone();
2684 drop(guard);
2685 let drain_timeout = *self.inner.effective_drain_timeout.lock().map_err(|_| {
2686 SuperviseError::StatePoisoned {
2687 module_id: Some(self.inner.module_id.clone()),
2688 }
2689 })?;
2690 let registration_active = self
2691 .inner
2692 .registry
2693 .get_module(&self.inner.module_id)
2694 .map_err(SuperviseError::Registry)?
2695 .is_some();
2696 let protocol = self.declared_protocol()?;
2697 let running_process =
2698 snapshot.enabled && snapshot.state == ModuleState::Running && snapshot.process_alive;
2699 let live = match protocol {
2705 ModuleProtocol::Subc => running_process && registration_active,
2706 ModuleProtocol::None => running_process,
2707 };
2708
2709 Ok(ModuleStatus {
2710 module_id: self.inner.module_id.clone(),
2711 state: snapshot.state,
2712 enabled: snapshot.enabled,
2713 process_alive: snapshot.process_alive,
2714 registration_active,
2715 protocol,
2716 live,
2717 restart_count,
2718 lifetime_restarts: snapshot.lifetime_restarts,
2719 spawn_generation: snapshot.spawn_generation,
2720 max_restarts: self.inner.restart_policy.max_restarts,
2721 restart_window: self.inner.restart_policy.window,
2722 drain_timeout,
2723 restart_backoff: self.inner.restart_policy.backoff,
2724 restart_max_backoff: self.inner.restart_policy.max_backoff,
2725 pid: snapshot.pid,
2726 spawned_at_ms: snapshot.spawned_at_ms,
2727 spawned_from: snapshot.spawned_from,
2728 process_start_time: snapshot.process_start_time,
2729 last_exit: snapshot.last_exit,
2730 health: snapshot.health,
2731 })
2732 }
2733
2734 #[cfg(test)]
2735 pub(crate) fn hold_snapshot_for_test(
2736 &self,
2737 acquired: std::sync::mpsc::Sender<()>,
2738 hold: Duration,
2739 ) -> std::thread::JoinHandle<()> {
2740 let snapshot = Arc::clone(&self.inner.snapshot);
2741 std::thread::spawn(move || {
2742 let _guard = snapshot.lock().expect("test snapshot lock is not poisoned");
2743 acquired
2744 .send(())
2745 .expect("test receiver waits for snapshot lock");
2746 std::thread::sleep(hold);
2747 })
2748 }
2749
2750 pub(crate) async fn running_image_agreement(&self) -> subc_control::RunningImageAgreement {
2751 let snapshot = match lock_snapshot(&self.inner.snapshot) {
2752 Ok(snapshot) => snapshot.clone(),
2753 Err(_) => {
2754 return subc_control::RunningImageAgreement::Unavailable {
2755 reason: subc_control::RunningImageUnavailableReason::NotRunning,
2756 };
2757 }
2758 };
2759 self.inner
2760 .provenance_probe
2761 .observe(
2762 snapshot.pid,
2763 snapshot.spawned_from.as_deref(),
2764 snapshot.spawned_file_identity,
2765 snapshot.process_start_time,
2766 )
2767 .await
2768 }
2769
2770 pub(crate) fn will_recover_after_connection_loss(&self) -> Result<bool, SuperviseError> {
2771 let mut snapshot = lock_snapshot(&self.inner.snapshot)?;
2772 Ok(match snapshot.state {
2773 ModuleState::Restarting => true,
2774 ModuleState::Failed | ModuleState::Disabled => false,
2775 _ => daemon_will_restart(&mut snapshot, &self.inner.restart_policy, Instant::now()),
2776 })
2777 }
2778
2779 #[cfg(test)]
2780 pub(crate) fn is_warming(&self) -> Result<bool, SuperviseError> {
2781 self.is_warming_with_snapshot_lock(None)
2782 }
2783
2784 pub(crate) fn is_warming_for_control(
2785 &self,
2786 caller: &'static str,
2787 ) -> Result<bool, SuperviseError> {
2788 self.is_warming_with_snapshot_lock(Some(caller))
2789 }
2790
2791 fn is_warming_with_snapshot_lock(
2792 &self,
2793 caller: Option<&'static str>,
2794 ) -> Result<bool, SuperviseError> {
2795 let snapshot = match caller {
2796 Some(caller) => {
2797 lock_snapshot_for_control(&self.inner.snapshot, &self.inner.module_id, caller)?
2798 }
2799 None => lock_snapshot(&self.inner.snapshot)?,
2800 }
2801 .clone();
2802 Ok(matches!(
2803 snapshot.state,
2804 ModuleState::Starting | ModuleState::Running | ModuleState::Restarting
2805 ))
2806 }
2807
2808 pub async fn drain(&self) -> Result<(), SuperviseError> {
2810 self.stop().await
2811 }
2812
2813 pub(crate) async fn retire(&self) -> Result<(), SuperviseError> {
2814 match self.state()? {
2815 ModuleState::Stopped | ModuleState::Failed => return Ok(()),
2816 ModuleState::Starting
2817 | ModuleState::Running
2818 | ModuleState::Unresponsive
2819 | ModuleState::Restarting
2820 | ModuleState::Draining
2821 | ModuleState::Disabled => {}
2822 }
2823
2824 let (reply_tx, reply_rx) = oneshot::channel();
2825 self.inner
2826 .commands
2827 .send(SupervisorCommand::Retire { reply: reply_tx })
2828 .await
2829 .map_err(|_| SuperviseError::CommandClosed {
2830 module_id: self.inner.module_id.clone(),
2831 })?;
2832 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2833 module_id: self.inner.module_id.clone(),
2834 })?
2835 }
2836
2837 pub async fn stop(&self) -> Result<(), SuperviseError> {
2838 match self.state()? {
2839 ModuleState::Stopped | ModuleState::Failed => return Ok(()),
2840 ModuleState::Starting
2841 | ModuleState::Running
2842 | ModuleState::Unresponsive
2843 | ModuleState::Restarting
2844 | ModuleState::Draining
2845 | ModuleState::Disabled => {}
2846 }
2847
2848 let (reply_tx, reply_rx) = oneshot::channel();
2849 self.inner
2850 .commands
2851 .send(SupervisorCommand::Drain { reply: reply_tx })
2852 .await
2853 .map_err(|_| SuperviseError::CommandClosed {
2854 module_id: self.inner.module_id.clone(),
2855 })?;
2856 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2857 module_id: self.inner.module_id.clone(),
2858 })?
2859 }
2860
2861 pub async fn restart(&self, drain_timeout_ms: Option<u64>) -> Result<(), SuperviseError> {
2862 let received_at_generation = lock_snapshot(&self.inner.snapshot)?.spawn_generation;
2863 let (reply_tx, reply_rx) = oneshot::channel();
2864 self.inner
2865 .commands
2866 .send(SupervisorCommand::Restart {
2867 drain_timeout_ms,
2868 received_at_generation,
2869 queued_at: Instant::now(),
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 async fn swap(&self, ready_timeout: Option<Duration>) -> Result<(), SuperviseError> {
2886 let (reply_tx, reply_rx) = oneshot::channel();
2887 self.inner
2888 .commands
2889 .send(SupervisorCommand::Swap {
2890 ready_timeout,
2891 reply: reply_tx,
2892 })
2893 .await
2894 .map_err(|_| SuperviseError::CommandClosed {
2895 module_id: self.inner.module_id.clone(),
2896 })?;
2897 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2898 module_id: self.inner.module_id.clone(),
2899 })?
2900 }
2901
2902 pub async fn reload(&self) -> Result<(), SuperviseError> {
2903 let (reply_tx, reply_rx) = oneshot::channel();
2904 self.inner
2905 .commands
2906 .send(SupervisorCommand::Reload { reply: reply_tx })
2907 .await
2908 .map_err(|_| SuperviseError::CommandClosed {
2909 module_id: self.inner.module_id.clone(),
2910 })?;
2911 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2912 module_id: self.inner.module_id.clone(),
2913 })?
2914 }
2915
2916 pub async fn set_enabled(&self, enabled: bool) -> Result<bool, SuperviseError> {
2917 let (reply_tx, reply_rx) = oneshot::channel();
2918 self.inner
2919 .commands
2920 .send(SupervisorCommand::SetEnabled {
2921 enabled,
2922 reply: reply_tx,
2923 })
2924 .await
2925 .map_err(|_| SuperviseError::CommandClosed {
2926 module_id: self.inner.module_id.clone(),
2927 })?;
2928 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2929 module_id: self.inner.module_id.clone(),
2930 })?
2931 }
2932
2933 pub(crate) fn declared_protocol(&self) -> Result<ModuleProtocol, SuperviseError> {
2938 Ok(self
2939 .inner
2940 .configuration
2941 .lock()
2942 .map_err(|_| SuperviseError::StatePoisoned {
2943 module_id: Some(self.inner.module_id.clone()),
2944 })?
2945 .spec
2946 .protocol)
2947 }
2948
2949 pub(crate) fn configuration(&self) -> Result<(ModuleSpec, HealthConfig), SuperviseError> {
2950 let configuration =
2951 self.inner
2952 .configuration
2953 .lock()
2954 .map_err(|_| SuperviseError::StatePoisoned {
2955 module_id: Some(self.inner.module_id.clone()),
2956 })?;
2957 Ok((configuration.spec.clone(), configuration.health))
2958 }
2959
2960 #[cfg(any(test, feature = "test-support"))]
2964 pub async fn update_spec_for_test(&self, spec: ModuleSpec) -> Result<(), SuperviseError> {
2965 let (_, health) = self.configuration()?;
2966 let drain_timeout_ms = u64::try_from(
2967 self.inner
2968 .effective_drain_timeout
2969 .lock()
2970 .unwrap_or_else(|poisoned| poisoned.into_inner())
2971 .as_millis(),
2972 )
2973 .ok();
2974 self.update_configuration(spec, health, drain_timeout_ms)
2975 .await
2976 }
2977
2978 pub(crate) async fn update_configuration(
2979 &self,
2980 spec: ModuleSpec,
2981 health: HealthConfig,
2982 drain_timeout_ms: Option<u64>,
2983 ) -> Result<(), SuperviseError> {
2984 if spec.module_id != self.inner.module_id {
2985 return Err(SuperviseError::InvalidSpec {
2986 reason: "a supervised module's module_id cannot be changed".to_string(),
2987 });
2988 }
2989 validate_spec(&spec)?;
2990 let (reply_tx, reply_rx) = oneshot::channel();
2991 self.inner
2992 .commands
2993 .send(SupervisorCommand::UpdateConfiguration {
2994 spec: spec.clone(),
2995 health,
2996 drain_timeout_ms,
2997 reply: reply_tx,
2998 })
2999 .await
3000 .map_err(|_| SuperviseError::CommandClosed {
3001 module_id: self.inner.module_id.clone(),
3002 })?;
3003 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
3004 module_id: self.inner.module_id.clone(),
3005 })?;
3006 let mut configuration =
3007 self.inner
3008 .configuration
3009 .lock()
3010 .map_err(|_| SuperviseError::StatePoisoned {
3011 module_id: Some(self.inner.module_id.clone()),
3012 })?;
3013 configuration.spec = spec;
3014 configuration.health = health;
3015 Ok(())
3016 }
3017}
3018
3019impl Drop for SupervisedModuleInner {
3020 fn drop(&mut self) {
3021 let Ok(mut monitor) = self.monitor.lock() else {
3022 return;
3023 };
3024 if let Some(monitor) = monitor.as_ref().filter(|monitor| !monitor.is_finished()) {
3025 let _ = update_snapshot(&self.snapshot, Some(&self.module_id), |state| {
3026 state.state = ModuleState::Stopped;
3027 clear_current_process_facts(state);
3028 });
3029 monitor.abort();
3030 }
3031 let _ = monitor.take();
3032 }
3033}
3034
3035#[derive(Debug)]
3036enum SupervisorCommand {
3037 Drain {
3038 reply: oneshot::Sender<Result<(), SuperviseError>>,
3039 },
3040 Retire {
3041 reply: oneshot::Sender<Result<(), SuperviseError>>,
3042 },
3043 Restart {
3044 drain_timeout_ms: Option<u64>,
3049 received_at_generation: u64,
3053 queued_at: Instant,
3056 reply: oneshot::Sender<Result<(), SuperviseError>>,
3057 },
3058 Reload {
3059 reply: oneshot::Sender<Result<(), SuperviseError>>,
3060 },
3061 SetEnabled {
3062 enabled: bool,
3063 reply: oneshot::Sender<Result<bool, SuperviseError>>,
3064 },
3065 UpdateConfiguration {
3066 spec: ModuleSpec,
3067 health: HealthConfig,
3068 drain_timeout_ms: Option<u64>,
3071 reply: oneshot::Sender<()>,
3072 },
3073 Swap {
3074 ready_timeout: Option<Duration>,
3077 reply: oneshot::Sender<Result<(), SuperviseError>>,
3079 },
3080}
3081
3082#[derive(Debug)]
3083pub enum SuperviseError {
3084 InvalidSpec {
3085 reason: String,
3086 },
3087 Spawn {
3088 program: PathBuf,
3089 source: io::Error,
3090 cgroup_path: Option<PathBuf>,
3091 },
3092 Cgroup {
3093 module_id: String,
3094 source: io::Error,
3095 },
3096 LaunchNonce {
3099 reason: String,
3100 },
3101 Wait {
3102 module_id: String,
3103 source: io::Error,
3104 },
3105 Kill {
3106 module_id: String,
3107 source: io::Error,
3108 },
3109 Forwarding(ForwardingError),
3110 Registry(RegistryError),
3111 ReloadUnavailable {
3112 module_id: String,
3113 reason: String,
3114 },
3115 Disabled {
3120 module_id: String,
3121 },
3122 ReloadFailed {
3123 module_id: String,
3124 reason: String,
3125 },
3126 RegistrationStillActive {
3127 module_id: String,
3128 waited: Duration,
3129 },
3130 StatePoisoned {
3131 module_id: Option<String>,
3132 },
3133 CommandClosed {
3134 module_id: String,
3135 },
3136 SwapInProgress {
3140 module_id: String,
3141 },
3142 SwapRefused {
3144 module_id: String,
3145 reason: SwapRefusal,
3146 },
3147 SwapFailed {
3151 module_id: String,
3152 arm: SwapFailureArm,
3153 detail: String,
3154 candidate_exit: Option<ExitReport>,
3157 },
3158}
3159
3160#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3162pub enum SwapRefusal {
3163 OverlapExclusive,
3165 NotRegistered,
3168 ProtocolNone,
3171 NotConfigured,
3174 AlreadySwapping,
3176}
3177
3178impl SwapRefusal {
3179 pub fn as_str(self) -> &'static str {
3180 match self {
3181 Self::OverlapExclusive => "overlap_exclusive",
3182 Self::NotRegistered => "not_registered",
3183 Self::ProtocolNone => "protocol_none",
3184 Self::NotConfigured => "not_configured",
3185 Self::AlreadySwapping => "already_swapping",
3186 }
3187 }
3188}
3189
3190#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3193pub enum SwapFailureArm {
3194 SpawnFailed,
3196 NeverRegistered,
3198 NeverReady,
3200 CandidateExited,
3202 CandidateUnhealthy,
3204 Interrupted,
3208 CutoverLost,
3213}
3214
3215impl SwapFailureArm {
3216 pub fn as_str(self) -> &'static str {
3217 match self {
3218 Self::SpawnFailed => "spawn_failed",
3219 Self::NeverRegistered => "never_registered",
3220 Self::NeverReady => "never_ready",
3221 Self::CandidateExited => "candidate_exited",
3222 Self::CandidateUnhealthy => "candidate_unhealthy",
3223 Self::Interrupted => "interrupted",
3224 Self::CutoverLost => "cutover_lost",
3225 }
3226 }
3227}
3228
3229impl fmt::Display for SuperviseError {
3230 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3231 match self {
3232 Self::InvalidSpec { reason } => write!(f, "invalid module spec: {reason}"),
3233 Self::Spawn {
3234 program,
3235 source,
3236 cgroup_path: Some(cgroup_path),
3237 } => write!(
3238 f,
3239 "failed to place module in cgroup '{}' while spawning '{}': {source}",
3240 cgroup_path.display(),
3241 program.display()
3242 ),
3243 Self::Spawn {
3244 program,
3245 source,
3246 cgroup_path: None,
3247 } => write!(
3248 f,
3249 "failed to spawn module '{}': {source}",
3250 program.display()
3251 ),
3252 Self::Cgroup { module_id, source } => {
3253 write!(
3254 f,
3255 "failed to prepare cgroup for module '{module_id}': {source}"
3256 )
3257 }
3258 Self::LaunchNonce { reason } => {
3259 write!(
3260 f,
3261 "failed to generate reserved-module launch nonce: {reason}"
3262 )
3263 }
3264 Self::Wait { module_id, source } => {
3265 write!(f, "failed to wait for module '{module_id}': {source}")
3266 }
3267 Self::Kill { module_id, source } => {
3268 write!(f, "failed to kill module '{module_id}': {source}")
3269 }
3270 Self::Forwarding(err) => write!(f, "forwarding error: {err}"),
3271 Self::Registry(err) => write!(f, "registry error: {err}"),
3272 Self::ReloadUnavailable { module_id, reason } => {
3273 write!(f, "reload unavailable for module '{module_id}': {reason}")
3274 }
3275 Self::Disabled { module_id } => {
3276 write!(
3277 f,
3278 "module '{module_id}' is disabled; enable it before restart or reload"
3279 )
3280 }
3281 Self::ReloadFailed { module_id, reason } => {
3282 write!(f, "reload failed for module '{module_id}': {reason}")
3283 }
3284 Self::RegistrationStillActive { module_id, waited } => write!(
3285 f,
3286 "module '{module_id}' registration remained active after waiting {waited:?}"
3287 ),
3288 Self::StatePoisoned { module_id } => match module_id {
3289 Some(module_id) => {
3290 write!(f, "supervisor state for module '{module_id}' was poisoned")
3291 }
3292 None => write!(f, "supervisor state was poisoned"),
3293 },
3294 Self::CommandClosed { module_id } => {
3295 write!(
3296 f,
3297 "supervisor command channel for module '{module_id}' is closed"
3298 )
3299 }
3300 Self::SwapInProgress { module_id } => write!(
3301 f,
3302 "module '{module_id}' is being swapped; retry once the swap has cut over or failed, or stop the module to abort the swap"
3303 ),
3304 Self::SwapRefused { module_id, reason } => match reason {
3305 SwapRefusal::OverlapExclusive => write!(
3306 f,
3307 "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"
3308 ),
3309 SwapRefusal::NotRegistered => write!(
3310 f,
3311 "module '{module_id}' is not registered, so there is no serving process to keep while a replacement warms; use a plain restart"
3312 ),
3313 SwapRefusal::ProtocolNone => write!(
3314 f,
3315 "module '{module_id}' is protocol: \"none\" and never registers, so a swap could never see its replacement become ready; use a plain restart"
3316 ),
3317 SwapRefusal::NotConfigured => write!(
3318 f,
3319 "module '{module_id}' cannot be swapped: the supervisor was built without the forwarding table or shared handle a swap needs"
3320 ),
3321 SwapRefusal::AlreadySwapping => {
3322 write!(f, "module '{module_id}' is already being swapped")
3323 }
3324 },
3325 Self::SwapFailed {
3326 module_id,
3327 arm,
3328 detail,
3329 ..
3330 } => write!(
3331 f,
3332 "swap of module '{module_id}' failed ({}): {detail}; the running process was left serving",
3333 arm.as_str()
3334 ),
3335 }
3336 }
3337}
3338
3339impl Error for SuperviseError {
3340 fn source(&self) -> Option<&(dyn Error + 'static)> {
3341 match self {
3342 Self::Spawn { source, .. }
3343 | Self::Cgroup { source, .. }
3344 | Self::Wait { source, .. }
3345 | Self::Kill { source, .. } => Some(source),
3346 Self::Forwarding(err) => Some(err),
3347 Self::Registry(err) => Some(err),
3348 Self::LaunchNonce { .. }
3349 | Self::InvalidSpec { .. }
3350 | Self::ReloadUnavailable { .. }
3351 | Self::Disabled { .. }
3352 | Self::ReloadFailed { .. }
3353 | Self::RegistrationStillActive { .. }
3354 | Self::StatePoisoned { .. }
3355 | Self::CommandClosed { .. }
3356 | Self::SwapInProgress { .. }
3357 | Self::SwapRefused { .. }
3358 | Self::SwapFailed { .. } => None,
3359 }
3360 }
3361}
3362
3363pub(crate) fn validate_spec(spec: &ModuleSpec) -> Result<(), SuperviseError> {
3364 if spec.module_id.trim().is_empty() {
3365 return Err(SuperviseError::InvalidSpec {
3366 reason: "module_id must not be empty".to_string(),
3367 });
3368 }
3369
3370 Ok(())
3371}
3372
3373#[derive(Debug, Default)]
3374struct HealthProbeRuntime {
3375 registered_connection: Option<crate::ConnectionId>,
3376 advertised: bool,
3377 next_probe_at: Option<Instant>,
3378 probe_index: u64,
3379}
3380
3381impl HealthProbeRuntime {
3382 fn refresh_registration(
3383 &mut self,
3384 spec: &ModuleSpec,
3385 runtime: &SupervisorRuntimeConfig,
3386 registry: &Registry,
3387 snapshot: &SharedSnapshot,
3388 ) {
3389 if spec.protocol == ModuleProtocol::None {
3401 self.registered_connection = None;
3402 self.advertised = false;
3403 self.next_probe_at = None;
3404 return;
3405 }
3406
3407 let registration = match registry.get_module(&spec.module_id) {
3408 Ok(registration) => registration,
3409 Err(err) => {
3410 warn!(module_id = %spec.module_id, error = %err, "health prober could not read registry");
3411 self.advertised = false;
3412 self.next_probe_at = None;
3413 return;
3414 }
3415 };
3416
3417 let Some(registration) = registration else {
3418 self.registered_connection = None;
3419 self.advertised = false;
3420 self.next_probe_at = None;
3421 return;
3422 };
3423
3424 let advertised = registration
3425 .control_ops
3426 .iter()
3427 .any(|op| op == MODULE_CONTROL_OP_HEALTH_CHECK);
3428 if !advertised {
3429 self.registered_connection = Some(registration.connection_id);
3430 self.advertised = false;
3431 self.next_probe_at = None;
3432 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3433 state.health.status = SupervisorHealthStatus::Unknown;
3434 state.health.consecutive_failures = 0;
3435 state.health.last_probe_ms = None;
3436 state.health.detail = None;
3437 state.health.metrics = None;
3438 });
3439 return;
3440 }
3441
3442 let reregistered = self.registered_connection != Some(registration.connection_id);
3443 self.registered_connection = Some(registration.connection_id);
3444 self.advertised = true;
3445 if reregistered || self.next_probe_at.is_none() {
3446 self.probe_index = 0;
3447 self.next_probe_at = Some(
3448 Instant::now() + jittered_health_delay(&spec.module_id, 0, runtime.health.cadence),
3449 );
3450 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3451 state.health.status = SupervisorHealthStatus::Unknown;
3452 state.health.consecutive_failures = 0;
3453 state.health.detail = None;
3454 state.health.metrics = None;
3455 });
3456 }
3457 }
3458
3459 fn wake_after(&self) -> Duration {
3460 if !self.advertised {
3461 return REGISTRY_RELEASE_POLL;
3462 }
3463 self.next_probe_at
3464 .map(|next| next.saturating_duration_since(Instant::now()))
3465 .unwrap_or(REGISTRY_RELEASE_POLL)
3466 }
3467
3468 fn due(&self) -> bool {
3469 self.advertised
3470 && self
3471 .next_probe_at
3472 .is_some_and(|next| Instant::now() >= next)
3473 }
3474
3475 fn schedule_next(&mut self, spec: &ModuleSpec, cadence: Duration) {
3476 self.probe_index = self.probe_index.wrapping_add(1);
3477 self.next_probe_at = Some(
3478 Instant::now() + jittered_health_delay(&spec.module_id, self.probe_index, cadence),
3479 );
3480 }
3481}
3482
3483#[derive(Debug)]
3518enum HealthProbeEvidence {
3519 LaneDead,
3521 NoAnswer,
3523 BadAnswer,
3525 Misconfigured,
3527}
3528
3529#[derive(Debug)]
3530struct HealthProbeError {
3531 evidence: HealthProbeEvidence,
3532 message: String,
3533}
3534
3535impl HealthProbeError {
3536 fn lane_dead(message: impl Into<String>) -> Self {
3537 Self::with(HealthProbeEvidence::LaneDead, message)
3538 }
3539
3540 fn no_answer(message: impl Into<String>) -> Self {
3541 Self::with(HealthProbeEvidence::NoAnswer, message)
3542 }
3543
3544 fn bad_answer(message: impl Into<String>) -> Self {
3545 Self::with(HealthProbeEvidence::BadAnswer, message)
3546 }
3547
3548 fn misconfigured(message: impl Into<String>) -> Self {
3549 Self::with(HealthProbeEvidence::Misconfigured, message)
3550 }
3551
3552 fn with(evidence: HealthProbeEvidence, message: impl Into<String>) -> Self {
3553 Self {
3554 evidence,
3555 message: message.into(),
3556 }
3557 }
3558
3559 #[allow(dead_code)]
3573 fn is_proof_of_death(&self) -> bool {
3574 matches!(self.evidence, HealthProbeEvidence::LaneDead)
3575 }
3576
3577 fn label(&self) -> &'static str {
3585 match self.evidence {
3586 HealthProbeEvidence::LaneDead => "lane-dead",
3587 HealthProbeEvidence::NoAnswer => "no-answer",
3588 HealthProbeEvidence::BadAnswer => "bad-answer",
3589 HealthProbeEvidence::Misconfigured => "daemon-misconfigured",
3590 }
3591 }
3592}
3593
3594impl fmt::Display for HealthProbeError {
3595 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3596 f.write_str(&self.message)
3597 }
3598}
3599
3600async fn run_health_probe_cycle(
3601 spec: &ModuleSpec,
3602 runtime: &SupervisorRuntimeConfig,
3603 registry: &Registry,
3604 process_liveness: &SupervisorProcessLiveness,
3605 snapshot: &SharedSnapshot,
3606 child: &mut Option<SupervisedChild>,
3607) {
3608 let now_ms = unix_ms_now();
3609 match probe_module_health(&spec.module_id, runtime, None).await {
3610 Ok(report) => {
3611 handle_health_report(
3612 spec,
3613 runtime,
3614 registry,
3615 process_liveness,
3616 snapshot,
3617 child,
3618 report,
3619 now_ms,
3620 )
3621 .await;
3622 }
3623 Err(err) => {
3624 handle_health_probe_failure(
3625 spec,
3626 runtime,
3627 registry,
3628 process_liveness,
3629 snapshot,
3630 child,
3631 err,
3632 now_ms,
3633 )
3634 .await;
3635 }
3636 }
3637}
3638
3639async fn probe_module_health(
3640 module_id: &str,
3641 runtime: &SupervisorRuntimeConfig,
3642 drain_deadline: Option<Instant>,
3643) -> Result<HealthReport, HealthProbeError> {
3644 let Some(forwarding) = runtime.forwarding.as_ref() else {
3645 return Err(HealthProbeError::misconfigured(
3646 "supervisor was not configured with a forwarding table",
3647 ));
3648 };
3649 let probe_started_at = Instant::now();
3650 let mut deadline = probe_started_at + runtime.health.deadline;
3651 if let Some(drain_deadline) = drain_deadline {
3652 deadline = deadline.min(drain_deadline);
3653 }
3654 let pending = if drain_deadline.is_some() {
3655 forwarding.begin_drain_health_probe_rpc_for(
3656 module_id,
3657 MODULE_CONTROL_OP_HEALTH_CHECK,
3658 probe_started_at,
3659 deadline,
3660 )
3661 } else {
3662 forwarding.begin_health_probe_rpc_for(
3663 module_id,
3664 MODULE_CONTROL_OP_HEALTH_CHECK,
3665 probe_started_at,
3666 deadline,
3667 )
3668 }
3669 .map_err(|err| {
3670 HealthProbeError::lane_dead(format!("failed to begin health.check RPC: {err}"))
3673 })?;
3674 await_health_probe(forwarding, pending, deadline, runtime.health.deadline).await
3675}
3676
3677async fn probe_endpoint_health(
3684 endpoint: crate::ModuleEndpointId,
3685 runtime: &SupervisorRuntimeConfig,
3686 deadline_cap: Option<Instant>,
3687) -> Result<HealthReport, HealthProbeError> {
3688 let Some(forwarding) = runtime.forwarding.as_ref() else {
3689 return Err(HealthProbeError::misconfigured(
3690 "supervisor was not configured with a forwarding table",
3691 ));
3692 };
3693 let probe_started_at = Instant::now();
3694 let mut deadline = probe_started_at + runtime.health.deadline;
3695 if let Some(cap) = deadline_cap {
3696 deadline = deadline.min(cap);
3697 }
3698 let pending = forwarding
3699 .begin_endpoint_health_probe_rpc_for(
3700 endpoint,
3701 MODULE_CONTROL_OP_HEALTH_CHECK,
3702 probe_started_at,
3703 deadline,
3704 )
3705 .map_err(|err| {
3706 HealthProbeError::lane_dead(format!("failed to begin health.check RPC: {err}"))
3707 })?;
3708 await_health_probe(forwarding, pending, deadline, runtime.health.deadline).await
3709}
3710
3711async fn await_health_probe(
3713 forwarding: &ForwardingTable,
3714 pending: PendingModuleControlRpc,
3715 deadline: Instant,
3716 probe_budget: Duration,
3717) -> Result<HealthReport, HealthProbeError> {
3718 let PendingModuleControlRpc {
3719 endpoint,
3720 module_sink,
3721 negotiated_ver,
3722 corr,
3723 receiver,
3724 } = pending;
3725 let body = serde_json::to_vec(&ModuleControlRequest::HealthCheck {}).map_err(|err| {
3726 HealthProbeError::misconfigured(format!("failed to encode health.check: {err}"))
3727 })?;
3728 let frame = Frame::build_with_version(
3729 negotiated_ver,
3730 FrameType::Request,
3731 control_flags(),
3732 0,
3733 0,
3734 corr,
3735 body,
3736 )
3737 .map_err(|err| {
3738 HealthProbeError::misconfigured(format!("failed to build health.check frame: {err}"))
3739 })?;
3740
3741 match timeout_at(deadline, module_sink.send(frame)).await {
3747 Ok(Ok(())) => {}
3748 Ok(Err(err)) => {
3749 let _ = forwarding.cancel_module_control_rpc(endpoint, corr);
3750 return Err(HealthProbeError::lane_dead(format!(
3753 "failed to send health.check: {err}"
3754 )));
3755 }
3756 Err(_elapsed) => {
3757 let _ = forwarding.cancel_module_control_rpc(endpoint, corr);
3758 return Err(HealthProbeError::no_answer(
3762 "health.check send timed out before enqueue (module egress full)",
3763 ));
3764 }
3765 }
3766
3767 match timeout_at(deadline, receiver).await {
3768 Ok(Ok(ModuleControlRpcOutcome::Response(response))) => {
3772 response.health_report().ok_or_else(|| {
3773 HealthProbeError::bad_answer("health.check RPC returned a non-health response")
3774 })
3775 }
3776 Ok(Ok(ModuleControlRpcOutcome::Rejected(body))) => Err(HealthProbeError::bad_answer(
3777 format!("health.check rejected: {}", body.message),
3778 )),
3779 Ok(Ok(ModuleControlRpcOutcome::ModuleGone(message))) => {
3780 Err(HealthProbeError::lane_dead(message))
3781 }
3782 Ok(Ok(ModuleControlRpcOutcome::MalformedResponse(message))) => {
3783 Err(HealthProbeError::bad_answer(message))
3784 }
3785 Ok(Ok(ModuleControlRpcOutcome::UnexpectedOp { expected, actual })) => {
3786 Err(HealthProbeError::bad_answer(format!(
3787 "expected module-control op '{expected}', got '{actual}'"
3788 )))
3789 }
3790 Ok(Ok(ModuleControlRpcOutcome::DeadlineElapsed)) => Err(HealthProbeError::bad_answer(
3794 "module answered health.check after its daemon deadline",
3795 )),
3796 Ok(Err(_)) => Err(HealthProbeError::misconfigured(
3797 "health.check waiter was canceled before the module responded",
3798 )),
3799 Err(_) => {
3800 let _ = forwarding.tombstone_health_probe_rpc(endpoint, corr);
3801 Err(HealthProbeError::no_answer(format!(
3802 "module did not answer health.check within {probe_budget:?}"
3803 )))
3804 }
3805 }
3806}
3807
3808#[allow(clippy::too_many_arguments)]
3809async fn handle_health_report(
3810 spec: &ModuleSpec,
3811 runtime: &SupervisorRuntimeConfig,
3812 registry: &Registry,
3813 process_liveness: &SupervisorProcessLiveness,
3814 snapshot: &SharedSnapshot,
3815 child: &mut Option<SupervisedChild>,
3816 report: HealthReport,
3817 now_ms: u64,
3818) {
3819 let status = supervisor_health_status(report.status);
3820 let detail = report.detail.clone();
3821 let metrics = truncate_health_metrics(report.metrics);
3822 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3823 state.health.status = status;
3824 state.health.last_probe_ms = Some(now_ms);
3825 state.health.detail = detail.clone();
3826 state.health.metrics = metrics.clone();
3827 state.health.consecutive_failures = 0;
3828 });
3829
3830 let action = match report.status {
3831 HealthStatus::Ok => return,
3832 HealthStatus::Degraded => runtime.health.on_degraded,
3833 HealthStatus::Failing => runtime.health.on_failing,
3834 };
3835 apply_l3_health_action(
3836 spec,
3837 runtime,
3838 registry,
3839 process_liveness,
3840 snapshot,
3841 child,
3842 status,
3843 detail.as_deref(),
3844 action,
3845 now_ms,
3846 )
3847 .await;
3848}
3849
3850#[allow(clippy::too_many_arguments)]
3851async fn handle_health_probe_failure(
3852 spec: &ModuleSpec,
3853 runtime: &SupervisorRuntimeConfig,
3854 registry: &Registry,
3855 process_liveness: &SupervisorProcessLiveness,
3856 snapshot: &SharedSnapshot,
3857 child: &mut Option<SupervisedChild>,
3858 err: HealthProbeError,
3859 now_ms: u64,
3860) {
3861 let threshold = runtime.health.failure_threshold.max(1);
3862 let mut failures = 0;
3863 let detail = format!("[{}] {err}", err.label());
3868 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3869 state.health.last_probe_ms = Some(now_ms);
3870 state.health.consecutive_failures = state.health.consecutive_failures.saturating_add(1);
3871 state.health.detail = Some(detail.clone());
3872 state.health.metrics = None;
3873 failures = state.health.consecutive_failures;
3874 });
3875
3876 if failures < threshold {
3877 warn!(
3878 module_id = %spec.module_id,
3879 consecutive_failures = failures,
3880 threshold,
3881 evidence = err.label(),
3882 detail = %detail,
3883 "health.check probe failed"
3884 );
3885 return;
3886 }
3887
3888 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3889 state.state = ModuleState::Unresponsive;
3890 state.health.status = SupervisorHealthStatus::Unresponsive;
3891 });
3892 if runtime.health.critical {
3896 error!(
3897 module_id = %spec.module_id,
3898 status = "unresponsive",
3899 evidence = err.label(),
3900 detail = %detail,
3901 "critical module health alert"
3902 );
3903 } else {
3904 warn!(
3905 module_id = %spec.module_id,
3906 status = "unresponsive",
3907 evidence = err.label(),
3908 detail = %detail,
3909 "module health threshold breached"
3910 );
3911 }
3912 if let Err(err) = health_restart_child(
3913 spec,
3914 runtime,
3915 registry,
3916 process_liveness,
3917 snapshot,
3918 child,
3919 SupervisorHealthStatus::Unresponsive,
3920 Some(&detail),
3921 now_ms,
3922 )
3923 .await
3924 {
3925 error!(module_id = %spec.module_id, error = %err, "health-triggered restart failed");
3926 }
3927}
3928
3929#[allow(clippy::too_many_arguments)]
3930async fn apply_l3_health_action(
3931 spec: &ModuleSpec,
3932 runtime: &SupervisorRuntimeConfig,
3933 registry: &Registry,
3934 process_liveness: &SupervisorProcessLiveness,
3935 snapshot: &SharedSnapshot,
3936 child: &mut Option<SupervisedChild>,
3937 status: SupervisorHealthStatus,
3938 detail: Option<&str>,
3939 action: HealthAction,
3940 now_ms: u64,
3941) {
3942 record_health_action(snapshot, &spec.module_id, action.to_string(), now_ms);
3943 match action {
3944 HealthAction::Report => {
3945 info!(
3946 module_id = %spec.module_id,
3947 status = ?status,
3948 detail,
3949 "module reported non-ok health"
3950 );
3951 }
3952 HealthAction::Alert => {
3953 error!(
3954 module_id = %spec.module_id,
3955 status = ?status,
3956 detail,
3957 "module health alert"
3958 );
3959 }
3960 HealthAction::Restart => {
3961 if let Err(err) = health_restart_child(
3962 spec,
3963 runtime,
3964 registry,
3965 process_liveness,
3966 snapshot,
3967 child,
3968 status,
3969 detail,
3970 now_ms,
3971 )
3972 .await
3973 {
3974 error!(module_id = %spec.module_id, error = %err, "health-triggered restart failed");
3975 }
3976 }
3977 }
3978}
3979
3980#[allow(clippy::too_many_arguments)]
3981async fn health_restart_child(
3982 spec: &ModuleSpec,
3983 runtime: &SupervisorRuntimeConfig,
3984 registry: &Registry,
3985 process_liveness: &SupervisorProcessLiveness,
3986 snapshot: &SharedSnapshot,
3987 child: &mut Option<SupervisedChild>,
3988 status: SupervisorHealthStatus,
3989 detail: Option<&str>,
3990 now_ms: u64,
3991) -> Result<(), SuperviseError> {
3992 let (enabled, schedule) = {
3993 let mut state = lock_snapshot(snapshot)?;
3994 let enabled = state.enabled;
3995 let schedule = if enabled {
3996 state.next_crash_restart(&runtime.restart_policy, Instant::now())
3997 } else {
3998 None
3999 };
4000 (enabled, schedule)
4001 };
4002
4003 if !enabled {
4004 return Err(SuperviseError::Disabled {
4005 module_id: spec.module_id.clone(),
4006 });
4007 }
4008
4009 if schedule.is_none() {
4010 record_health_action(snapshot, &spec.module_id, "disabled".to_string(), now_ms);
4011 error!(
4012 module_id = %spec.module_id,
4013 status = ?status,
4014 detail,
4015 max_restarts = runtime.restart_policy.max_restarts,
4016 window_secs = runtime.restart_policy.window.as_secs(),
4017 "health restart budget exhausted; disabling module"
4018 );
4019 let stop_notice = begin_forwarding_drain_if_configured(
4020 spec,
4021 runtime,
4022 registry,
4023 snapshot,
4024 Some(false),
4025 RouteCloseReason::Disable,
4026 )
4027 .await?;
4028 drain_optional_child(
4029 &spec.module_id,
4030 spec.protocol,
4031 stop_notice,
4032 registry,
4033 snapshot,
4034 &runtime.terminal_ring,
4035 &runtime.spawn_events,
4036 child,
4037 runtime.drain_timeout,
4038 ModuleState::Disabled,
4039 Some(false),
4040 )
4041 .await?;
4042 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4043 return Ok(());
4044 }
4045
4046 let schedule = schedule.expect("a health restart must have a crash-restart schedule");
4047 let mut restart_count = 0;
4048 update_snapshot(snapshot, Some(&spec.module_id), |state| {
4049 restart_count = state.crash_restarts.len();
4050 state.state = ModuleState::Unresponsive;
4051 state.health.status = status;
4052 state.health.last_action = Some(HealthAction::Restart.to_string());
4053 state.health.last_action_ms = Some(now_ms);
4054 })?;
4055 warn!(
4056 module_id = %spec.module_id,
4057 status = ?status,
4058 detail,
4059 restart_count,
4060 restart_in_window = schedule.restart_in_window,
4061 delay_ms = schedule.delay.as_millis() as u64,
4062 "health-triggered module restart"
4063 );
4064
4065 let stop_notice = begin_forwarding_drain_if_configured(
4066 spec,
4067 runtime,
4068 registry,
4069 snapshot,
4070 Some(true),
4071 RouteCloseReason::Restart,
4072 )
4073 .await?;
4074 drain_optional_child(
4075 &spec.module_id,
4076 spec.protocol,
4077 stop_notice,
4078 registry,
4079 snapshot,
4080 &runtime.terminal_ring,
4081 &runtime.spawn_events,
4082 child,
4083 runtime.drain_timeout,
4084 ModuleState::Restarting,
4085 Some(true),
4086 )
4087 .await?;
4088 sleep(schedule.delay).await;
4089 if !respawn_still_pending(snapshot) {
4093 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4094 return Ok(());
4095 }
4096 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
4097 match spawn_and_mark_running(spec, runtime, snapshot) {
4098 Ok(next_child) => {
4099 *child = Some(next_child);
4100 Ok(())
4101 }
4102 Err(err) => {
4103 fail_snapshot(snapshot, Some(&spec.module_id), None);
4104 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4105 *child = None;
4106 Err(err)
4107 }
4108 }
4109}
4110
4111fn record_health_action(snapshot: &SharedSnapshot, module_id: &str, action: String, now_ms: u64) {
4112 let _ = update_snapshot(snapshot, Some(module_id), |state| {
4113 state.health.last_action = Some(action);
4114 state.health.last_action_ms = Some(now_ms);
4115 });
4116}
4117
4118fn supervisor_health_status(status: HealthStatus) -> SupervisorHealthStatus {
4119 match status {
4120 HealthStatus::Ok => SupervisorHealthStatus::Ok,
4121 HealthStatus::Degraded => SupervisorHealthStatus::Degraded,
4122 HealthStatus::Failing => SupervisorHealthStatus::Failing,
4123 }
4124}
4125
4126fn truncate_health_metrics(metrics: Option<Value>) -> Option<Value> {
4138 let metrics = metrics?;
4139 match serde_json::to_vec(&metrics) {
4140 Ok(encoded) if encoded.len() > MAX_HEALTH_METRICS_BYTES => Some(serde_json::json!({
4141 "truncated": true,
4142 "original_bytes": encoded.len(),
4143 })),
4144 Ok(_) | Err(_) => Some(metrics),
4145 }
4146}
4147
4148fn jittered_health_delay(module_id: &str, probe_index: u64, cadence: Duration) -> Duration {
4154 if cadence.is_zero() {
4155 return Duration::ZERO;
4156 }
4157 let cadence_ms = cadence.as_millis() as u64;
4158 if cadence_ms == 0 {
4174 return cadence;
4175 }
4176 let jitter_span = (cadence_ms / 10).max(1);
4191 let hash = module_id.as_bytes().iter().fold(
4192 probe_index.wrapping_mul(0x9E37_79B9_7F4A_7C15),
4193 |acc, byte| {
4194 acc.wrapping_mul(1099511628211)
4195 .wrapping_add(u64::from(*byte))
4196 },
4197 );
4198 cadence + Duration::from_millis(hash % jitter_span)
4199}
4200
4201#[cfg(test)]
4202mod tests {
4203 use super::*;
4204
4205 #[test]
4206 fn readding_a_module_clears_its_rescan_removal_tombstone() {
4207 let handle = SupervisorHandle::new();
4208 let module_id = "readded-tombstone";
4209 handle.record_rescan_removal(module_id);
4210 assert!(handle.removal_tombstone_age_ms(module_id).is_some());
4211
4212 handle.apply_identity_configuration(&ModuleSpec {
4213 module_id: module_id.to_string(),
4214 program: PathBuf::from("/test/module"),
4215 args: Vec::new(),
4216 env: Vec::new(),
4217 reserved: false,
4218 reserved_prefixes: Vec::new(),
4219 protocol: ModuleProtocol::Subc,
4220 overlap: Default::default(),
4221 });
4222
4223 assert!(
4224 handle.removal_tombstone_age_ms(module_id).is_none(),
4225 "a re-added module must not retain a stale removal tombstone"
4226 );
4227 }
4228
4229 fn stale_process_snapshot(state: ModuleState, enabled: bool) -> SharedSnapshot {
4230 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::new(state, enabled)));
4231 update_snapshot(&snapshot, Some("stale-process-facts"), |snapshot| {
4232 snapshot.process_alive = true;
4233 snapshot.pid = Some(41);
4234 snapshot.spawned_at_ms = Some(42);
4235 snapshot.spawned_from = Some(PathBuf::from("/spawned/module"));
4236 snapshot.spawned_file_identity = Some(SpawnedFileIdentity {
4237 device: 43,
4238 inode: 44,
4239 });
4240 })
4241 .unwrap();
4242 snapshot
4243 }
4244
4245 fn assert_snapshot_process_facts_cleared(snapshot: &SharedSnapshot) {
4246 let snapshot = lock_snapshot(snapshot).unwrap();
4247 assert!(!snapshot.process_alive);
4248 assert_eq!(snapshot.pid, None);
4249 assert_eq!(snapshot.spawned_at_ms, None);
4250 assert_eq!(snapshot.spawned_from, None);
4251 assert_eq!(snapshot.spawned_file_identity, None);
4252 }
4253
4254 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4255 async fn failed_enable_spawn_clears_preexisting_current_process_facts() {
4256 let supervisor = Supervisor::default();
4257 let mut runtime = supervisor.runtime_config();
4258 runtime.test_seed_stale_facts_before_enable_spawn = true;
4259 let snapshot = stale_process_snapshot(ModuleState::Disabled, false);
4260 let mut child = None;
4261 let spec = ModuleSpec {
4262 module_id: "failed-enable-clears-facts".to_string(),
4263 program: PathBuf::from("/definitely/missing/failed-enable-module"),
4264 args: Vec::new(),
4265 env: Vec::new(),
4266 reserved: false,
4267 reserved_prefixes: Vec::new(),
4268 protocol: ModuleProtocol::Subc,
4269 overlap: Default::default(),
4270 };
4271
4272 let result = set_child_enabled(
4273 &spec,
4274 &runtime,
4275 &supervisor.registry,
4276 &supervisor.process_liveness,
4277 &snapshot,
4278 &mut child,
4279 true,
4280 )
4281 .await;
4282
4283 assert!(matches!(result, Err(SuperviseError::Spawn { .. })));
4284 assert_eq!(lock_snapshot(&snapshot).unwrap().state, ModuleState::Failed);
4285 assert_snapshot_process_facts_cleared(&snapshot);
4286 }
4287
4288 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4289 async fn failed_reload_spawn_clears_current_process_facts() {
4290 let supervisor = Supervisor::default();
4291 let mut runtime = supervisor.runtime_config();
4292 runtime.restart_policy = RestartPolicy::new(0, Duration::ZERO);
4293 let snapshot = stale_process_snapshot(ModuleState::Running, true);
4294 let mut child = None;
4295 let spec = ModuleSpec {
4296 module_id: "failed-reload-clears-facts".to_string(),
4297 program: PathBuf::from("/unused/failed-reload-module"),
4298 args: Vec::new(),
4299 env: Vec::new(),
4300 reserved: false,
4301 reserved_prefixes: Vec::new(),
4302 protocol: ModuleProtocol::Subc,
4303 overlap: Default::default(),
4304 };
4305
4306 let result = handle_reload_spawn_failure(
4307 &spec,
4308 &runtime,
4309 &supervisor.process_liveness,
4310 &snapshot,
4311 &mut child,
4312 "forced reload spawn failure".to_string(),
4313 )
4314 .await;
4315
4316 assert!(matches!(result, Err(SuperviseError::ReloadFailed { .. })));
4317 assert_eq!(lock_snapshot(&snapshot).unwrap().state, ModuleState::Failed);
4318 assert_snapshot_process_facts_cleared(&snapshot);
4319 }
4320
4321 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4322 async fn dropping_a_module_with_an_active_monitor_clears_current_process_facts() {
4323 let supervisor = Supervisor::default();
4324 let snapshot = stale_process_snapshot(ModuleState::Running, true);
4325 let module = supervisor.supervised_module(
4326 ModuleSpec {
4327 module_id: "drop-clears-facts".to_string(),
4328 program: PathBuf::from("/unused/drop-module"),
4329 args: Vec::new(),
4330 env: Vec::new(),
4331 reserved: false,
4332 reserved_prefixes: Vec::new(),
4333 protocol: ModuleProtocol::Subc,
4334 overlap: Default::default(),
4335 },
4336 supervisor.runtime_config(),
4337 Arc::clone(&snapshot),
4338 None,
4339 );
4340 assert!(!module
4341 .inner
4342 .monitor
4343 .lock()
4344 .unwrap()
4345 .as_ref()
4346 .unwrap()
4347 .is_finished());
4348
4349 drop(module);
4350
4351 assert_eq!(
4352 lock_snapshot(&snapshot).unwrap().state,
4353 ModuleState::Stopped
4354 );
4355 assert_snapshot_process_facts_cleared(&snapshot);
4356 }
4357
4358 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4359 async fn configuration_update_does_not_replace_captured_running_process_facts() {
4360 let supervisor = Supervisor::default();
4361 let snapshot = stale_process_snapshot(ModuleState::Running, true);
4362 let initial = ModuleSpec {
4363 module_id: "rescan-preserves-spawn-facts".to_string(),
4364 program: PathBuf::from("/spawned/module"),
4365 args: Vec::new(),
4366 env: Vec::new(),
4367 reserved: false,
4368 reserved_prefixes: Vec::new(),
4369 protocol: ModuleProtocol::Subc,
4370 overlap: Default::default(),
4371 };
4372 let module = supervisor.supervised_module(
4373 initial.clone(),
4374 supervisor.runtime_config(),
4375 snapshot,
4376 None,
4377 );
4378 let before = module.status().unwrap();
4379 let mut replacement = initial;
4380 replacement.program = PathBuf::from("/rescanned/replacement-module");
4381
4382 module
4383 .update_configuration(replacement, HealthConfig::default(), None)
4384 .await
4385 .unwrap();
4386
4387 let after = module.status().unwrap();
4388 assert_eq!(after.pid, before.pid);
4389 assert_eq!(after.spawned_at_ms, before.spawned_at_ms);
4390 assert_eq!(after.spawned_from, before.spawned_from);
4391 drop(module);
4392 }
4393}
4394
4395fn unix_ms_now() -> u64 {
4396 SystemTime::now()
4397 .duration_since(UNIX_EPOCH)
4398 .map(|duration| duration.as_millis().min(u128::from(u64::MAX)) as u64)
4399 .unwrap_or(0)
4400}
4401
4402async fn supervise_loop(
4403 mut spec: ModuleSpec,
4404 mut runtime: SupervisorRuntimeConfig,
4405 registry: Arc<Registry>,
4406 process_liveness: Arc<SupervisorProcessLiveness>,
4407 snapshot: SharedSnapshot,
4408 mut child: Option<SupervisedChild>,
4409 mut commands: mpsc::Receiver<SupervisorCommand>,
4410) {
4411 let mut health_probe = HealthProbeRuntime::default();
4412 let mut pending_respawn: Option<Instant> = None;
4416 let mut requeued: VecDeque<SupervisorCommand> = VecDeque::new();
4419 loop {
4420 if let Some(command) = requeued.pop_front() {
4421 if !handle_supervisor_command(
4422 command,
4423 &mut spec,
4424 &mut runtime,
4425 ®istry,
4426 &process_liveness,
4427 &snapshot,
4428 &mut child,
4429 &mut commands,
4430 &mut requeued,
4431 )
4432 .await
4433 {
4434 return;
4435 }
4436 if child.is_some() || !respawn_still_pending(&snapshot) {
4437 pending_respawn = None;
4438 }
4439 continue;
4440 }
4441 if child.is_some() {
4442 health_probe.refresh_registration(&spec, &runtime, ®istry, &snapshot);
4443 let probe_sleep = sleep(health_probe.wake_after());
4444 tokio::pin!(probe_sleep);
4445 let active_child = child.as_mut().expect("child checked above");
4446 tokio::select! {
4447 wait_result = active_child.wait() => {
4448 let exit_report = match wait_result {
4457 Ok(status) => classify_reaped_child_exit(&snapshot, active_child, &status),
4458 Err(err) => {
4459 active_child.drain_stderr(&spec.module_id).await;
4460 fail_snapshot(&snapshot, Some(&spec.module_id), None);
4461 record_wait_error_terminal(
4467 &spec.module_id,
4468 &runtime.terminal_ring,
4469 &runtime.spawn_events,
4470 );
4471 untrack_if_registration_released(
4472 &process_liveness,
4473 ®istry,
4474 &spec.module_id,
4475 &snapshot,
4476 );
4477 error!(module_id = %spec.module_id, error = %err, "failed to wait for supervised module");
4478 child = None;
4479 continue;
4480 }
4481 };
4482 active_child.drain_stderr(&spec.module_id).await;
4483
4484 let next = on_child_exit(
4485 &spec,
4486 runtime.restart_policy,
4487 ®istry,
4488 &snapshot,
4489 &runtime.terminal_ring,
4490 &runtime.spawn_events,
4491 &runtime.child_roster,
4492 exit_report,
4493 ).await;
4494 active_child.release_roster();
4497 match next {
4498 NextAction::Stop { registration_released } => {
4499 if registration_released {
4500 process_liveness.untrack_if_current(&spec.module_id, &snapshot);
4501 }
4502 child = None;
4503 }
4504 NextAction::Restart { schedule } => {
4505 let delay = schedule.map_or(
4506 runtime.restart_policy.delay_for_restart(0),
4507 |schedule| schedule.delay,
4508 );
4509 if let Some(schedule) = schedule {
4510 log_crash_respawn(&spec.module_id, schedule);
4511 }
4512 child = None;
4520 pending_respawn = Some(Instant::now() + delay);
4521 }
4522 }
4523 }
4524 command = commands.recv() => {
4525 let Some(command) = command else {
4526 return;
4527 };
4528 if !handle_supervisor_command(
4529 command,
4530 &mut spec,
4531 &mut runtime,
4532 ®istry,
4533 &process_liveness,
4534 &snapshot,
4535 &mut child,
4536 &mut commands,
4537 &mut requeued,
4538 ).await {
4539 return;
4540 }
4541 }
4542 _ = &mut probe_sleep => {
4543 if health_probe.due() {
4544 run_health_probe_cycle(
4545 &spec,
4546 &runtime,
4547 ®istry,
4548 &process_liveness,
4549 &snapshot,
4550 &mut child,
4551 ).await;
4552 if child.is_some() {
4553 health_probe.schedule_next(&spec, runtime.health.cadence);
4554 }
4555 }
4556 }
4557 }
4558 } else if let Some(deadline) = pending_respawn {
4559 tokio::select! {
4560 _ = sleep_until(deadline) => {
4561 pending_respawn = None;
4562 if !respawn_still_pending(&snapshot) {
4566 continue;
4567 }
4568 if runtime.child_roster.is_closed() {
4573 let _ = update_snapshot(&snapshot, Some(&spec.module_id), |state| {
4574 state.state = ModuleState::Stopped;
4575 });
4576 debug!(module_id = %spec.module_id, "crash respawn cancelled by daemon shutdown");
4577 continue;
4578 }
4579 if let Err(err) = wait_for_registration_release(
4580 ®istry,
4581 &spec.module_id,
4582 REGISTRY_RELEASE_TIMEOUT,
4583 ).await {
4584 fail_snapshot(&snapshot, Some(&spec.module_id), None);
4585 error!(module_id = %spec.module_id, error = %err, "registration did not release before restart");
4586 continue;
4587 }
4588
4589 match spawn_and_mark_running(&spec, &runtime, &snapshot) {
4590 Ok(next_child) => {
4591 child = Some(next_child);
4592 debug!(module_id = %spec.module_id, "supervised module restarted after crash");
4593 }
4594 Err(err) => {
4595 fail_snapshot(&snapshot, Some(&spec.module_id), None);
4596 process_liveness.untrack_if_current(&spec.module_id, &snapshot);
4597 error!(module_id = %spec.module_id, error = %err, "failed to restart supervised module");
4598 }
4599 }
4600 }
4601 command = commands.recv() => {
4602 let Some(command) = command else {
4603 return;
4604 };
4605 if !handle_supervisor_command(
4606 command,
4607 &mut spec,
4608 &mut runtime,
4609 ®istry,
4610 &process_liveness,
4611 &snapshot,
4612 &mut child,
4613 &mut commands,
4614 &mut requeued,
4615 ).await {
4616 return;
4617 }
4618 if child.is_some() || !respawn_still_pending(&snapshot) {
4623 pending_respawn = None;
4624 }
4625 }
4626 }
4627 } else {
4628 let Some(command) = commands.recv().await else {
4629 return;
4630 };
4631 if !handle_supervisor_command(
4632 command,
4633 &mut spec,
4634 &mut runtime,
4635 ®istry,
4636 &process_liveness,
4637 &snapshot,
4638 &mut child,
4639 &mut commands,
4640 &mut requeued,
4641 )
4642 .await
4643 {
4644 return;
4645 }
4646 }
4647 }
4648}
4649
4650fn log_crash_respawn(module_id: &str, schedule: CrashRestartSchedule) {
4651 info!(
4652 module_id,
4653 restart_in_window = schedule.restart_in_window,
4654 delay_ms = schedule.delay.as_millis() as u64,
4655 "respawning after crash"
4656 );
4657}
4658
4659fn respawn_still_pending(snapshot: &SharedSnapshot) -> bool {
4665 matches!(
4666 lock_snapshot(snapshot),
4667 Ok(state) if state.enabled && state.state == ModuleState::Restarting
4668 )
4669}
4670
4671enum NextAction {
4672 Stop {
4673 registration_released: bool,
4674 },
4675 Restart {
4676 schedule: Option<CrashRestartSchedule>,
4677 },
4678}
4679
4680#[allow(clippy::too_many_arguments)]
4681async fn handle_supervisor_command(
4682 command: SupervisorCommand,
4683 spec: &mut ModuleSpec,
4684 runtime: &mut SupervisorRuntimeConfig,
4685 registry: &Registry,
4686 process_liveness: &SupervisorProcessLiveness,
4687 snapshot: &SharedSnapshot,
4688 child: &mut Option<SupervisedChild>,
4689 commands: &mut mpsc::Receiver<SupervisorCommand>,
4690 requeued: &mut VecDeque<SupervisorCommand>,
4691) -> bool {
4692 match command {
4693 SupervisorCommand::Drain { reply } => {
4694 let result = drain_optional_child(
4697 &spec.module_id,
4698 spec.protocol,
4699 StopNotice::NotSent,
4700 registry,
4701 snapshot,
4702 &runtime.terminal_ring,
4703 &runtime.spawn_events,
4704 child,
4705 runtime.drain_timeout,
4706 ModuleState::Stopped,
4707 None,
4708 )
4709 .await;
4710 let registration_released = result.is_ok();
4711 let _ = reply.send(result);
4712 if registration_released {
4713 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4714 }
4715 false
4716 }
4717 SupervisorCommand::Retire { reply } => {
4718 let result = async {
4719 let stop_notice = begin_forwarding_drain_if_configured(
4720 spec,
4721 runtime,
4722 registry,
4723 snapshot,
4724 None,
4725 RouteCloseReason::Disable,
4726 )
4727 .await?;
4728 drain_optional_child(
4729 &spec.module_id,
4730 spec.protocol,
4731 stop_notice,
4732 registry,
4733 snapshot,
4734 &runtime.terminal_ring,
4735 &runtime.spawn_events,
4736 child,
4737 runtime.drain_timeout,
4738 ModuleState::Stopped,
4739 None,
4740 )
4741 .await
4742 }
4743 .await;
4744 let registration_released = result.is_ok();
4745 let _ = reply.send(result);
4746 if registration_released {
4747 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4748 }
4749 false
4750 }
4751 SupervisorCommand::Restart {
4752 drain_timeout_ms,
4753 received_at_generation,
4754 queued_at,
4755 reply,
4756 } => {
4757 info!(
4761 module_id = %spec.module_id,
4762 queued_ms = u64::try_from(queued_at.elapsed().as_millis()).unwrap_or(u64::MAX),
4763 "restart command dequeued"
4764 );
4765 let validation = match lock_snapshot(snapshot) {
4777 Ok(state) if !state.enabled => Err(SuperviseError::Disabled {
4778 module_id: spec.module_id.clone(),
4779 }),
4780 Ok(_) => Ok(()),
4781 Err(err) => Err(err),
4782 };
4783 let initiated = validation.is_ok();
4784 let _ = reply.send(validation);
4785 let satisfied_by_generation = if initiated && child.is_some() {
4796 lock_snapshot(snapshot).ok().and_then(|state| {
4797 (state.spawn_generation > received_at_generation
4798 && !state.configuration_updated_since_spawn)
4799 .then_some(state.spawn_generation)
4800 })
4801 } else {
4802 None
4803 };
4804 if let Some(generation) = satisfied_by_generation {
4805 info!(
4806 module_id = %spec.module_id,
4807 received_at_generation,
4808 "restart already satisfied by generation {generation}; not restarting again"
4809 );
4810 } else if initiated {
4811 let drain_timeout = drain_timeout_ms
4814 .map(Duration::from_millis)
4815 .unwrap_or(runtime.drain_timeout);
4816 if let Err(err) = restart_child(
4817 spec,
4818 runtime,
4819 registry,
4820 process_liveness,
4821 snapshot,
4822 child,
4823 drain_timeout,
4824 )
4825 .await
4826 {
4827 warn!(
4828 module_id = %spec.module_id,
4829 error = %err,
4830 "operator restart failed after initiation ack; module state carries the outcome"
4831 );
4832 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
4833 state.state = ModuleState::Failed;
4834 clear_current_process_facts(state);
4835 });
4836 }
4837 }
4838 true
4839 }
4840 SupervisorCommand::Reload { reply } => {
4841 let result =
4842 reload_child(spec, runtime, registry, process_liveness, snapshot, child).await;
4843 let _ = reply.send(result);
4844 true
4845 }
4846 SupervisorCommand::SetEnabled { enabled, reply } => {
4847 let result = set_child_enabled(
4848 spec,
4849 runtime,
4850 registry,
4851 process_liveness,
4852 snapshot,
4853 child,
4854 enabled,
4855 )
4856 .await;
4857 let _ = reply.send(result);
4858 true
4859 }
4860 SupervisorCommand::UpdateConfiguration {
4861 spec: next_spec,
4862 health,
4863 drain_timeout_ms,
4864 reply,
4865 } => {
4866 if let Some(handle) = &runtime.supervisor_handle {
4867 handle.apply_identity_configuration(&next_spec);
4868 }
4869 *spec = next_spec;
4870 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
4871 state.configuration_updated_since_spawn = true;
4872 });
4873 runtime.health = health;
4874 runtime.drain_timeout = drain_timeout_ms
4875 .map(Duration::from_millis)
4876 .unwrap_or(runtime.default_drain_timeout);
4877 *runtime
4878 .effective_drain_timeout
4879 .lock()
4880 .unwrap_or_else(|poisoned| poisoned.into_inner()) = runtime.drain_timeout;
4881 let _ = reply.send(());
4882 true
4883 }
4884 SupervisorCommand::Swap {
4885 ready_timeout,
4886 reply,
4887 } => {
4888 let end = swap::run_swap(
4889 spec,
4890 runtime,
4891 registry,
4892 process_liveness,
4893 snapshot,
4894 child,
4895 commands,
4896 ready_timeout.unwrap_or(DEFAULT_SWAP_READY_TIMEOUT),
4897 reply,
4898 )
4899 .await;
4900 requeued.extend(end.requeue);
4901 true
4902 }
4903 }
4904}
4905
4906async fn restart_child(
4907 spec: &ModuleSpec,
4908 runtime: &SupervisorRuntimeConfig,
4909 registry: &Registry,
4910 process_liveness: &SupervisorProcessLiveness,
4911 snapshot: &SharedSnapshot,
4912 child: &mut Option<SupervisedChild>,
4913 drain_timeout: Duration,
4914) -> Result<(), SuperviseError> {
4915 if !lock_snapshot(snapshot)?.enabled {
4917 return Err(SuperviseError::Disabled {
4918 module_id: spec.module_id.clone(),
4919 });
4920 }
4921 let stop_notice = begin_forwarding_drain_with_timeout(
4922 spec,
4923 runtime,
4924 registry,
4925 snapshot,
4926 None,
4927 RouteCloseReason::Restart,
4928 drain_timeout,
4929 )
4930 .await?;
4931
4932 if child.is_some() {
4933 drain_optional_child(
4934 &spec.module_id,
4935 spec.protocol,
4936 stop_notice,
4937 registry,
4938 snapshot,
4939 &runtime.terminal_ring,
4940 &runtime.spawn_events,
4941 child,
4942 drain_timeout,
4943 ModuleState::Restarting,
4944 Some(true),
4945 )
4946 .await?;
4947 } else {
4948 update_snapshot(snapshot, Some(&spec.module_id), |state| {
4949 state.enabled = true;
4950 state.state = ModuleState::Restarting;
4951 clear_current_process_facts(state);
4952 })?;
4953 wait_for_registration_release(registry, &spec.module_id, REGISTRY_RELEASE_TIMEOUT).await?;
4954 }
4955
4956 reset_restart_count(snapshot, &spec.module_id)?;
4957 sleep(runtime.restart_policy.backoff).await;
4958 if !respawn_still_pending(snapshot) {
4961 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4962 return Ok(());
4963 }
4964 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
4965 match spawn_and_mark_running(spec, runtime, snapshot) {
4971 Ok(next_child) => {
4972 *child = Some(next_child);
4973 debug!(module_id = %spec.module_id, "supervised module restarted by operator request");
4974 Ok(())
4975 }
4976 Err(err) => {
4977 fail_snapshot(snapshot, Some(&spec.module_id), None);
4978 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4979 *child = None;
4980 Err(err)
4981 }
4982 }
4983}
4984
4985async fn reload_child(
4986 spec: &ModuleSpec,
4987 runtime: &SupervisorRuntimeConfig,
4988 registry: &Registry,
4989 process_liveness: &SupervisorProcessLiveness,
4990 snapshot: &SharedSnapshot,
4991 child: &mut Option<SupervisedChild>,
4992) -> Result<(), SuperviseError> {
4993 if !lock_snapshot(snapshot)?.enabled {
4995 return Err(SuperviseError::Disabled {
4996 module_id: spec.module_id.clone(),
4997 });
4998 }
4999 let stop_notice = begin_forwarding_drain(
5000 spec,
5001 runtime,
5002 registry,
5003 snapshot,
5004 Some(true),
5005 RouteCloseReason::Reload,
5006 )
5007 .await?;
5008
5009 if child.is_some() {
5010 drain_optional_child(
5011 &spec.module_id,
5012 spec.protocol,
5013 stop_notice,
5014 registry,
5015 snapshot,
5016 &runtime.terminal_ring,
5017 &runtime.spawn_events,
5018 child,
5019 runtime.drain_timeout,
5020 ModuleState::Restarting,
5021 Some(true),
5022 )
5023 .await?;
5024 } else {
5025 update_snapshot(snapshot, Some(&spec.module_id), |state| {
5026 state.enabled = true;
5027 state.state = ModuleState::Restarting;
5028 clear_current_process_facts(state);
5029 })?;
5030 wait_for_registration_release(registry, &spec.module_id, REGISTRY_RELEASE_TIMEOUT).await?;
5031 }
5032
5033 reset_restart_count(snapshot, &spec.module_id)?;
5034 sleep(runtime.restart_policy.backoff).await;
5035 if !respawn_still_pending(snapshot) {
5038 process_liveness.untrack_if_current(&spec.module_id, snapshot);
5039 return Ok(());
5040 }
5041 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
5042 let next_child = match spawn_and_mark_running(spec, runtime, snapshot) {
5043 Ok(next_child) => next_child,
5044 Err(err) => {
5045 return handle_reload_spawn_failure(
5046 spec,
5047 runtime,
5048 process_liveness,
5049 snapshot,
5050 child,
5051 format!("new child failed to spawn: {err}"),
5052 )
5053 .await;
5054 }
5055 };
5056 *child = Some(next_child);
5057
5058 let wait_outcome = {
5059 let active_child = child.as_mut().expect("new reload child was just stored");
5060 wait_for_registration_after_reload(
5061 registry,
5062 &spec.module_id,
5063 snapshot,
5064 active_child,
5065 REGISTRY_RELEASE_TIMEOUT,
5066 )
5067 .await?
5068 };
5069
5070 match wait_outcome {
5071 RegistrationWaitOutcome::Registered => {
5072 debug!(module_id = %spec.module_id, "supervised module reloaded and registered");
5073 Ok(())
5074 }
5075 RegistrationWaitOutcome::Exited(exit_report) => {
5076 if let Some(active_child) = child.as_mut() {
5077 active_child.drain_stderr(&spec.module_id).await;
5078 }
5079 *child = None;
5080 handle_reload_child_registration_failure(
5081 spec,
5082 runtime,
5083 registry,
5084 process_liveness,
5085 snapshot,
5086 child,
5087 ReloadRegistrationFailure {
5088 exit_report: registration_failure_exit_report(exit_report),
5089 reason: "new child exited before registering".to_string(),
5090 },
5091 )
5092 .await
5093 }
5094 RegistrationWaitOutcome::TimedOut => {
5095 let mut timed_out_child = child
5096 .take()
5097 .expect("timed-out reload child is still running");
5098 timed_out_child
5099 .start_kill()
5100 .map_err(|source| SuperviseError::Kill {
5101 module_id: spec.module_id.clone(),
5102 source,
5103 })?;
5104 let status = timed_out_child
5105 .wait()
5106 .await
5107 .map_err(|source| SuperviseError::Wait {
5108 module_id: spec.module_id.clone(),
5109 source,
5110 })?;
5111 timed_out_child.drain_stderr(&spec.module_id).await;
5112 handle_reload_child_registration_failure(
5113 spec,
5114 runtime,
5115 registry,
5116 process_liveness,
5117 snapshot,
5118 child,
5119 ReloadRegistrationFailure {
5120 exit_report: registration_failure_exit_report(classify_reaped_child_exit(
5121 snapshot,
5122 &timed_out_child,
5123 &status,
5124 )),
5125 reason: format!(
5126 "new child did not register within {:?}",
5127 REGISTRY_RELEASE_TIMEOUT
5128 ),
5129 },
5130 )
5131 .await
5132 }
5133 }
5134}
5135
5136async fn set_child_enabled(
5137 spec: &ModuleSpec,
5138 runtime: &SupervisorRuntimeConfig,
5139 registry: &Registry,
5140 process_liveness: &SupervisorProcessLiveness,
5141 snapshot: &SharedSnapshot,
5142 child: &mut Option<SupervisedChild>,
5143 enabled: bool,
5144) -> Result<bool, SuperviseError> {
5145 let (current_enabled, current_state) = {
5146 let state = lock_snapshot(snapshot)?;
5147 (state.enabled, state.state)
5148 };
5149 let revive_terminal = enabled
5157 && current_enabled
5158 && child.is_none()
5159 && matches!(current_state, ModuleState::Failed | ModuleState::Stopped);
5160 if current_enabled == enabled && !revive_terminal {
5161 return Ok(false);
5162 }
5163
5164 if enabled {
5165 update_snapshot(snapshot, Some(&spec.module_id), |state| {
5166 state.enabled = true;
5167 state.state = ModuleState::Starting;
5168 clear_current_process_facts(state);
5169 })?;
5170 #[cfg(test)]
5171 if runtime.test_seed_stale_facts_before_enable_spawn {
5172 update_snapshot(snapshot, Some(&spec.module_id), |state| {
5173 state.process_alive = true;
5174 state.pid = Some(41);
5175 state.spawned_at_ms = Some(42);
5176 state.spawned_from = Some(PathBuf::from("/spawned/module"));
5177 state.spawned_file_identity = Some(SpawnedFileIdentity {
5178 device: 43,
5179 inode: 44,
5180 });
5181 })?;
5182 }
5183 wait_for_registration_release(registry, &spec.module_id, REGISTRY_RELEASE_TIMEOUT).await?;
5184 reset_restart_count(snapshot, &spec.module_id)?;
5185 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
5186 let next_child = match spawn_and_mark_running(spec, runtime, snapshot) {
5187 Ok(next_child) => next_child,
5188 Err(err) => {
5189 if let Err(state_err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5190 state.state = ModuleState::Failed;
5191 clear_current_process_facts(state);
5192 }) {
5193 error!(module_id = %spec.module_id, error = %state_err, "failed to record enable spawn failure");
5194 }
5195 process_liveness.untrack_if_current(&spec.module_id, snapshot);
5196 return Err(err);
5197 }
5198 };
5199 *child = Some(next_child);
5200 debug!(module_id = %spec.module_id, "supervised module enabled");
5201 Ok(true)
5202 } else {
5203 let stop_notice = begin_forwarding_drain_if_configured(
5204 spec,
5205 runtime,
5206 registry,
5207 snapshot,
5208 Some(false),
5209 RouteCloseReason::Disable,
5210 )
5211 .await?;
5212 drain_optional_child(
5213 &spec.module_id,
5214 spec.protocol,
5215 stop_notice,
5216 registry,
5217 snapshot,
5218 &runtime.terminal_ring,
5219 &runtime.spawn_events,
5220 child,
5221 runtime.drain_timeout,
5222 ModuleState::Disabled,
5223 Some(false),
5224 )
5225 .await?;
5226 debug!(module_id = %spec.module_id, "supervised module disabled");
5227 Ok(true)
5228 }
5229}
5230
5231#[allow(clippy::too_many_arguments)]
5232async fn on_child_exit(
5233 spec: &ModuleSpec,
5234 policy: RestartPolicy,
5235 registry: &Registry,
5236 snapshot: &SharedSnapshot,
5237 terminal_ring: &Arc<Mutex<TerminalRing>>,
5238 spawn_events: &SpawnEventFeed,
5239 roster: &ChildRoster,
5240 exit_report: ExitReport,
5241) -> NextAction {
5242 if roster.is_closed() {
5248 return on_child_exit_during_daemon_shutdown(
5249 spec,
5250 registry,
5251 snapshot,
5252 terminal_ring,
5253 spawn_events,
5254 exit_report,
5255 )
5256 .await;
5257 }
5258 match exit_report.kind {
5259 ExitKind::Clean => {
5260 info!(
5261 module_id = %spec.module_id,
5262 exit_code = ?exit_report.code,
5263 exit_signal = ?exit_report.signal,
5264 "supervised module exited cleanly"
5265 );
5266 if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5267 state.state = ModuleState::Stopped;
5268 clear_current_process_facts(state);
5269 state.last_exit = Some(exit_report.clone());
5270 }) {
5271 error!(module_id = %spec.module_id, error = %err, "failed to record clean module exit");
5272 }
5273 record_terminal(
5274 &spec.module_id,
5275 terminal_ring,
5276 spawn_events,
5277 &exit_report,
5278 TerminalDisposition::Stopped,
5279 );
5280 let registration_released = match wait_for_registration_release(
5281 registry,
5282 &spec.module_id,
5283 REGISTRY_RELEASE_TIMEOUT,
5284 )
5285 .await
5286 {
5287 Ok(()) => true,
5288 Err(err) => {
5289 warn!(module_id = %spec.module_id, error = %err, "registration still active after clean exit");
5290 false
5291 }
5292 };
5293 NextAction::Stop {
5294 registration_released,
5295 }
5296 }
5297 ExitKind::Crash => {
5298 warn!(
5299 module_id = %spec.module_id,
5300 exit_code = ?exit_report.code,
5301 exit_signal = ?exit_report.signal,
5302 "supervised module exited abnormally (crash)"
5303 );
5304 let mut restart_schedule = None;
5305 let mut disposition = TerminalDisposition::Disabled;
5306 let mut disposition_detail = None;
5310 let now = Instant::now();
5311 if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5312 clear_current_process_facts(state);
5313 state.last_exit = Some(exit_report.clone());
5314 if state.enabled {
5315 if let Some(schedule) = state.next_crash_restart(&policy, now) {
5316 state.state = ModuleState::Restarting;
5317 restart_schedule = Some(schedule);
5318 disposition = TerminalDisposition::Restarting;
5319 } else {
5320 state.state = ModuleState::Failed;
5321 disposition = TerminalDisposition::Failed;
5322 disposition_detail = Some(policy.budget_exhausted_detail());
5323 }
5324 } else {
5325 state.state = ModuleState::Disabled;
5326 disposition = TerminalDisposition::Disabled;
5327 }
5328 }) {
5329 error!(module_id = %spec.module_id, error = %err, "failed to record crashed module exit");
5330 return NextAction::Stop {
5331 registration_released: false,
5332 };
5333 }
5334 if disposition_detail.is_some() {
5335 error!(
5340 module_id = %spec.module_id,
5341 max_restarts = policy.max_restarts,
5342 window_secs = policy.window.as_secs(),
5343 "module stopped: {}",
5344 policy.budget_exhausted_detail()
5345 );
5346 }
5347 record_terminal_with_detail(
5348 &spec.module_id,
5349 terminal_ring,
5350 spawn_events,
5351 &exit_report,
5352 disposition,
5353 disposition_detail,
5354 );
5355
5356 if let Some(schedule) = restart_schedule {
5357 NextAction::Restart {
5358 schedule: Some(schedule),
5359 }
5360 } else {
5361 let registration_released = match wait_for_registration_release(
5362 registry,
5363 &spec.module_id,
5364 REGISTRY_RELEASE_TIMEOUT,
5365 )
5366 .await
5367 {
5368 Ok(()) => true,
5369 Err(err) => {
5370 warn!(module_id = %spec.module_id, error = %err, "registration still active after failed module");
5371 false
5372 }
5373 };
5374 NextAction::Stop {
5375 registration_released,
5376 }
5377 }
5378 }
5379 ExitKind::DeliberateSeverance => {
5380 warn!(
5381 module_id = %spec.module_id,
5382 exit_code = ?exit_report.code,
5383 exit_signal = ?exit_report.signal,
5384 "supervised module exited after deliberate connection severance"
5385 );
5386 let mut should_restart = false;
5387 let mut disposition = TerminalDisposition::Disabled;
5388 if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5389 clear_current_process_facts(state);
5390 state.last_exit = Some(exit_report.clone());
5391 state.lifetime_restarts += 1;
5392 if state.enabled {
5393 state.state = ModuleState::Restarting;
5394 should_restart = true;
5395 disposition = TerminalDisposition::Restarting;
5396 } else {
5397 state.state = ModuleState::Disabled;
5398 }
5399 }) {
5400 error!(module_id = %spec.module_id, error = %err, "failed to record deliberately severed module exit");
5401 return NextAction::Stop {
5402 registration_released: false,
5403 };
5404 }
5405 record_terminal(
5406 &spec.module_id,
5407 terminal_ring,
5408 spawn_events,
5409 &exit_report,
5410 disposition,
5411 );
5412
5413 if should_restart {
5414 NextAction::Restart { schedule: None }
5415 } else {
5416 let registration_released = match wait_for_registration_release(
5417 registry,
5418 &spec.module_id,
5419 REGISTRY_RELEASE_TIMEOUT,
5420 )
5421 .await
5422 {
5423 Ok(()) => true,
5424 Err(err) => {
5425 warn!(module_id = %spec.module_id, error = %err, "registration still active after deliberately severed module exit");
5426 false
5427 }
5428 };
5429 NextAction::Stop {
5430 registration_released,
5431 }
5432 }
5433 }
5434 }
5435}
5436
5437async fn on_child_exit_during_daemon_shutdown(
5438 spec: &ModuleSpec,
5439 registry: &Registry,
5440 snapshot: &SharedSnapshot,
5441 terminal_ring: &Arc<Mutex<TerminalRing>>,
5442 spawn_events: &SpawnEventFeed,
5443 exit_report: ExitReport,
5444) -> NextAction {
5445 info!(
5446 module_id = %spec.module_id,
5447 exit_code = ?exit_report.code,
5448 exit_signal = ?exit_report.signal,
5449 exit_kind = ?exit_report.kind,
5450 "supervised module exited during daemon shutdown; not restarting it"
5451 );
5452 if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5453 state.state = ModuleState::Stopped;
5454 clear_current_process_facts(state);
5455 state.last_exit = Some(exit_report.clone());
5456 }) {
5457 error!(module_id = %spec.module_id, error = %err, "failed to record module exit during daemon shutdown");
5458 }
5459 record_terminal(
5460 &spec.module_id,
5461 terminal_ring,
5462 spawn_events,
5463 &exit_report,
5464 TerminalDisposition::DaemonShutdown,
5465 );
5466 let registration_released =
5467 wait_for_registration_release(registry, &spec.module_id, REGISTRY_RELEASE_TIMEOUT)
5468 .await
5469 .is_ok();
5470 NextAction::Stop {
5471 registration_released,
5472 }
5473}
5474
5475fn record_wait_error_terminal(
5476 module_id: &str,
5477 terminal_ring: &Arc<Mutex<TerminalRing>>,
5478 spawn_events: &SpawnEventFeed,
5479) {
5480 record_terminal(
5481 module_id,
5482 terminal_ring,
5483 spawn_events,
5484 &wait_error_exit_report(),
5485 TerminalDisposition::Failed,
5486 );
5487}
5488
5489fn record_terminal(
5490 module_id: &str,
5491 terminal_ring: &Arc<Mutex<TerminalRing>>,
5492 spawn_events: &SpawnEventFeed,
5493 exit_report: &ExitReport,
5494 disposition: TerminalDisposition,
5495) {
5496 record_terminal_with_detail(
5497 module_id,
5498 terminal_ring,
5499 spawn_events,
5500 exit_report,
5501 disposition,
5502 None,
5503 );
5504}
5505
5506fn durable_terminal_history_of(
5510 terminal_ring: &Mutex<TerminalRing>,
5511 module_id: &str,
5512) -> subc_control::TerminalHistory {
5513 let read = terminal_ring
5514 .lock()
5515 .unwrap_or_else(|p| p.into_inner())
5516 .capture_durable_history();
5517 read.read(module_id)
5518}
5519
5520fn record_terminal_with_detail(
5521 module_id: &str,
5522 terminal_ring: &Arc<Mutex<TerminalRing>>,
5523 spawn_events: &SpawnEventFeed,
5524 exit_report: &ExitReport,
5525 disposition: TerminalDisposition,
5526 disposition_detail: Option<String>,
5527) {
5528 spawn_events.emit_exited(module_id, exit_report.code, exit_report.signal);
5529 let record = TerminalRecord {
5530 exit_code: exit_report.code,
5531 exit_signal: exit_report.signal,
5532 at_ms: exit_report.at_ms,
5533 disposition,
5534 exit_kind: exit_report.kind.into(),
5535 disposition_detail,
5536 };
5537 terminal_ring
5538 .lock()
5539 .unwrap_or_else(|poisoned| poisoned.into_inner())
5540 .record_exit(module_id, record);
5541}
5542
5543fn untrack_if_registration_released(
5544 process_liveness: &SupervisorProcessLiveness,
5545 registry: &Registry,
5546 module_id: &str,
5547 snapshot: &SharedSnapshot,
5548) {
5549 match registry.get_module(module_id) {
5550 Ok(None) => process_liveness.untrack_if_current(module_id, snapshot),
5551 Ok(Some(_)) => {}
5552 Err(err) => {
5553 warn!(module_id, error = %err, "could not determine whether supervisor liveness can be untracked");
5554 }
5555 }
5556}
5557
5558#[cfg(test)]
5572fn apply_wire_spawn_args(
5573 command: &mut Command,
5574 spec: &ModuleSpec,
5575 connection_file_path: Option<&std::path::Path>,
5576 handle: Option<&SupervisorHandle>,
5577) -> Result<(), SuperviseError> {
5578 apply_wire_spawn_args_for_role(
5579 command,
5580 spec,
5581 connection_file_path,
5582 handle,
5583 SpawnRole::Plain,
5584 )
5585}
5586
5587fn apply_wire_spawn_args_for_role(
5596 command: &mut Command,
5597 spec: &ModuleSpec,
5598 connection_file_path: Option<&std::path::Path>,
5599 handle: Option<&SupervisorHandle>,
5600 role: SpawnRole,
5601) -> Result<(), SuperviseError> {
5602 command.env(SUBC_MODULE_ID_ENV, &spec.module_id);
5603 if spec.protocol == ModuleProtocol::None {
5604 return Ok(());
5605 }
5606 if let Some(connection_file_path) = connection_file_path {
5607 command.arg(SUBC_ARG).arg(connection_file_path);
5608 }
5609
5610 let nonce = generate_launch_nonce()?;
5614 if let Some(handle) = handle {
5615 match role {
5616 SpawnRole::Plain => {
5617 handle.set_spawn_nonce(&spec.module_id, nonce.clone());
5618 if spec.reserved {
5619 handle.set_reserved_nonce(&spec.module_id, nonce.clone());
5620 }
5621 }
5622 SpawnRole::SwapCandidate => handle.open_swap(&spec.module_id, nonce.clone()),
5623 }
5624 }
5625 command.env(SUBC_LAUNCH_NONCE_ENV, nonce);
5626 Ok(())
5627}
5628
5629fn apply_child_env(command: &mut Command, spec: &ModuleSpec) {
5630 command.env_remove(CK_LOG_ENV);
5631 command.env_remove(SUBC_SPAWN_ROLE_ENV);
5638 for (key, value) in &spec.env {
5639 if matches!(
5643 key.as_str(),
5644 CAPTURE_MAX_FILE_MB_ENV | CAPTURE_KEEP_ENV | CAPTURE_MAX_AGE_DAYS_ENV
5645 ) || key == SUBC_SPAWN_ROLE_ENV
5646 {
5647 continue;
5648 }
5649 command.env(key, value);
5650 }
5651}
5652
5653#[derive(Debug, Clone, Copy, PartialEq, Eq)]
5656enum SpawnRole {
5657 Plain,
5658 SwapCandidate,
5659}
5660
5661fn apply_spawn_role(command: &mut Command, role: SpawnRole) {
5664 if role == SpawnRole::SwapCandidate {
5665 command.env(SUBC_SPAWN_ROLE_ENV, SPAWN_ROLE_SWAP_CANDIDATE);
5666 }
5667}
5668
5669fn spawn_child(
5670 spec: &ModuleSpec,
5671 connection_file_path: Option<&std::path::Path>,
5672 handle: Option<&SupervisorHandle>,
5673 ring: &Arc<Mutex<StderrRing>>,
5674 capture_logs_dir: Option<&std::path::Path>,
5675 roster: &ChildRoster,
5676 #[cfg(target_os = "linux")] cgroup_placement: Option<&subc_cgroup::Placement>,
5677) -> Result<SupervisedChild, SuperviseError> {
5678 spawn_child_in_slot(
5679 spec,
5680 connection_file_path,
5681 handle,
5682 ring,
5683 capture_logs_dir,
5684 roster,
5685 #[cfg(target_os = "linux")]
5686 cgroup_placement,
5687 SpawnRole::Plain,
5688 false,
5689 )
5690}
5691
5692#[allow(clippy::too_many_arguments)]
5705fn spawn_child_in_slot(
5706 spec: &ModuleSpec,
5707 connection_file_path: Option<&std::path::Path>,
5708 handle: Option<&SupervisorHandle>,
5709 ring: &Arc<Mutex<StderrRing>>,
5710 capture_logs_dir: Option<&std::path::Path>,
5711 roster: &ChildRoster,
5712 #[cfg(target_os = "linux")] cgroup_placement: Option<&subc_cgroup::Placement>,
5713 role: SpawnRole,
5714 alternate_slot: bool,
5715) -> Result<SupervisedChild, SuperviseError> {
5716 if roster.is_closed() {
5717 return Err(SuperviseError::Spawn {
5718 program: spec.program.clone(),
5719 source: io::Error::other("the daemon is shutting down; not starting a new process"),
5720 cgroup_path: None,
5721 });
5722 }
5723 #[cfg(target_os = "linux")]
5724 let cgroup_name = swap::cgroup_name(&spec.module_id, alternate_slot);
5725 #[cfg(not(target_os = "linux"))]
5726 let _ = alternate_slot;
5727 let mut command = Command::new(&spec.program);
5728 command.args(&spec.args);
5729 apply_child_env(&mut command, spec);
5759 apply_spawn_role(&mut command, role);
5760 apply_wire_spawn_args_for_role(&mut command, spec, connection_file_path, handle, role)?;
5761
5762 #[cfg(target_os = "linux")]
5763 let cgroup_path = cgroup_placement
5764 .map(|placement| placement.module_path(&cgroup_name))
5765 .transpose()
5766 .map_err(|source| SuperviseError::Cgroup {
5767 module_id: spec.module_id.clone(),
5768 source,
5769 })?;
5770 #[cfg(not(target_os = "linux"))]
5771 let cgroup_path: Option<PathBuf> = None;
5772 #[cfg(target_os = "linux")]
5773 if let Some(path) = &cgroup_path {
5774 if let Err(error) = apply_cgroup_placement(&mut command, spec, path) {
5775 if let Some(placement) = cgroup_placement {
5776 remove_module_cgroup(placement, &cgroup_name);
5777 }
5778 return Err(error);
5779 }
5780 }
5781
5782 let output_sink = if let Some(logs_dir) = capture_logs_dir {
5783 let path = logs_dir.join(format!("{}.stderr.log", spec.module_id));
5784 match ChildOutputSink::open(&path, capture_retention(spec)) {
5785 Ok(sink) => sink,
5786 Err(error) => {
5787 warn!(
5788 module_id = %spec.module_id,
5789 path = %path.display(),
5790 error = %error,
5791 "could not open child output capture file; forwarding to stderr"
5792 );
5793 ChildOutputSink::Stderr
5794 }
5795 }
5796 } else {
5797 ChildOutputSink::Stderr
5798 };
5799
5800 command.stdout(Stdio::piped());
5801 command.stderr(Stdio::piped());
5802 command.kill_on_drop(true);
5803 #[cfg(unix)]
5820 command.process_group(0);
5821 command.stdin(Stdio::null());
5822
5823 #[cfg(windows)]
5828 subc_jobobject::suspend_on_create_async(&mut command);
5829 let mut child = match command.spawn() {
5830 Ok(child) => child,
5831 Err(source) => {
5832 #[cfg(target_os = "linux")]
5833 if let Some(placement) = cgroup_placement {
5834 remove_module_cgroup(placement, &cgroup_name);
5835 }
5836 return Err(SuperviseError::Spawn {
5837 program: spec.program.clone(),
5838 source,
5839 cgroup_path,
5840 });
5841 }
5842 };
5843
5844 #[cfg(windows)]
5846 let job = contain_spawned_child(&child, spec)?;
5847 let spawned_at_ms = unix_ms_now();
5848 let spawned_from = spec.program.clone();
5849 let spawned_file_identity = spawned_file_identity(&spawned_from);
5850 let pid = child.id().ok_or_else(|| SuperviseError::Spawn {
5851 program: spec.program.clone(),
5852 source: io::Error::other("spawned child exposed no live pid"),
5853 cgroup_path: cgroup_path.clone(),
5854 })?;
5855 let process_start_time = crate::provenance::process_start_time(pid);
5856 let process_identity = process_start_time.map(|start_time| ProcessIdentity { pid, start_time });
5857 #[cfg(target_os = "linux")]
5861 let recorded_cgroup_name = cgroup_path.as_ref().map(|_| cgroup_name.clone());
5862 #[cfg(not(target_os = "linux"))]
5863 let recorded_cgroup_name = None;
5864 let roster_guard = roster.admit(
5865 spec.module_id.clone(),
5866 pid,
5867 spec.protocol,
5868 process_start_time,
5869 crate::child_roster::RecordedIdentity {
5870 start_time: subc_os::start_time(pid),
5871 executable: spawned_file_identity.map(|identity| {
5872 crate::live_children::ExecutableIdentity {
5873 device: identity.device,
5874 inode: identity.inode,
5875 }
5876 }),
5877 cgroup_name: recorded_cgroup_name,
5878 },
5879 );
5880 if roster.is_closed() {
5889 if let Err(error) = child.start_kill() {
5890 debug!(module_id = %spec.module_id, pid, %error, "kill of a process spawned during daemon shutdown failed; it may already have exited");
5891 }
5892 drop(roster_guard);
5893 return Err(SuperviseError::Spawn {
5894 program: spec.program.clone(),
5895 source: io::Error::other(
5896 "the daemon began shutting down while this process was starting; ended it",
5897 ),
5898 cgroup_path,
5899 });
5900 }
5901
5902 let stdout_pump = match child.stdout.take() {
5903 Some(stdout) => Some(tokio::spawn(pump_stdout_to(stdout, output_sink.clone()))),
5904 None => {
5905 warn!(
5906 module_id = %spec.module_id,
5907 "spawned child exposed no stdout pipe; file capture will be incomplete"
5908 );
5909 None
5910 }
5911 };
5912 let stderr_pump = match child.stderr.take() {
5913 Some(stderr) => {
5914 let generation = ring
5915 .lock()
5916 .unwrap_or_else(|poisoned| poisoned.into_inner())
5917 .begin_process();
5918 Some(StderrPump {
5919 task: tokio::spawn(pump_stderr_to(
5920 stderr,
5921 Arc::clone(ring),
5922 generation,
5923 output_sink,
5924 )),
5925 generation,
5926 })
5927 }
5928 None => {
5929 ring.lock()
5933 .unwrap_or_else(|poisoned| poisoned.into_inner())
5934 .mark_not_captured("stderr pipe was not available on spawn");
5935 warn!(
5936 module_id = %spec.module_id,
5937 "spawned child exposed no stderr pipe; tail will be unavailable"
5938 );
5939 None
5940 }
5941 };
5942
5943 Ok(SupervisedChild {
5944 child,
5945 #[cfg(target_os = "linux")]
5946 module_id: cgroup_name,
5947 #[cfg(target_os = "linux")]
5948 cgroup_placement: cgroup_placement.cloned(),
5949 #[cfg(windows)]
5950 job,
5951 stdout_pump,
5952 stderr_pump,
5953 stderr_ring: Arc::clone(ring),
5954 spawned_at_ms,
5955 spawned_from,
5956 spawned_file_identity,
5957 process_start_time,
5958 process_identity,
5959 pid,
5960 roster_guard: Some(roster_guard),
5961 })
5962}
5963
5964#[cfg(windows)]
5978fn contain_spawned_child(
5979 child: &Child,
5980 spec: &ModuleSpec,
5981) -> Result<Option<subc_jobobject::JobObject>, SuperviseError> {
5982 let module_id = spec.module_id.as_str();
5983 let Some(pid) = child.id() else {
5984 warn!(
5987 module_id,
5988 "spawned child had already exited before containment; no job object attached"
5989 );
5990 return Ok(None);
5991 };
5992
5993 let job = match subc_jobobject::JobObject::new() {
5994 Ok(job) => job,
5995 Err(source) => {
5996 warn!(
5997 module_id,
5998 error = %source,
5999 "could not create a job object; this module's helper processes will not be \
6000 reaped on teardown"
6001 );
6002 resume_suspended_child(pid, spec)?;
6005 return Ok(None);
6006 }
6007 };
6008
6009 if let Err(source) = job.assign(child) {
6010 warn!(
6011 module_id,
6012 error = %source,
6013 "could not assign the child to its job object; this module's helper processes \
6014 will not be reaped on teardown"
6015 );
6016 resume_suspended_child(pid, spec)?;
6017 return Ok(None);
6018 }
6019
6020 resume_suspended_child(pid, spec)?;
6021 Ok(Some(job))
6022}
6023
6024#[cfg(windows)]
6029fn resume_suspended_child(pid: u32, spec: &ModuleSpec) -> Result<(), SuperviseError> {
6030 if let Err(source) = subc_jobobject::resume_main_thread(pid) {
6031 let _ = std::process::Command::new("taskkill.exe")
6035 .args(["/PID", &pid.to_string(), "/T", "/F"])
6036 .stdin(Stdio::null())
6037 .stdout(Stdio::null())
6038 .stderr(Stdio::null())
6039 .status();
6040 return Err(SuperviseError::Spawn {
6041 program: spec.program.clone(),
6042 source,
6043 cgroup_path: None,
6044 });
6045 }
6046 Ok(())
6047}
6048
6049#[cfg(target_os = "linux")]
6050fn remove_module_cgroup(placement: &subc_cgroup::Placement, module_id: &str) {
6051 match placement.remove_module(module_id) {
6052 Ok(()) => debug!(module_id, "removed module cgroup after process exit"),
6053 Err(error) => warn!(
6054 module_id,
6055 error = %error,
6056 "could not remove module cgroup after process exit; continuing teardown"
6057 ),
6058 }
6059}
6060
6061#[cfg(target_os = "linux")]
6062fn apply_cgroup_placement(
6063 command: &mut Command,
6064 spec: &ModuleSpec,
6065 path: &std::path::Path,
6066) -> Result<(), SuperviseError> {
6067 subc_cgroup::apply(command, path).map_err(|source| SuperviseError::Cgroup {
6068 module_id: spec.module_id.clone(),
6069 source,
6070 })
6071}
6072
6073fn capture_retention(spec: &ModuleSpec) -> Retention {
6074 let defaults = Retention::default();
6075 let value = |name: &str| {
6076 spec.env
6077 .iter()
6078 .rev()
6079 .find_map(|(key, value)| (key == name).then_some(value.as_str()))
6080 };
6081 Retention {
6082 max_file_mb: value(CAPTURE_MAX_FILE_MB_ENV)
6083 .and_then(|value| value.parse().ok())
6084 .unwrap_or(defaults.max_file_mb),
6085 keep: value(CAPTURE_KEEP_ENV)
6086 .and_then(|value| value.parse().ok())
6087 .unwrap_or(defaults.keep),
6088 max_age_days: value(CAPTURE_MAX_AGE_DAYS_ENV)
6089 .and_then(|value| value.parse().ok())
6090 .unwrap_or(defaults.max_age_days),
6091 }
6092}
6093
6094fn generate_launch_nonce() -> Result<String, SuperviseError> {
6097 let mut bytes = [0u8; 32];
6098 getrandom::getrandom(&mut bytes).map_err(|source| SuperviseError::LaunchNonce {
6099 reason: source.to_string(),
6100 })?;
6101 let mut hex = String::with_capacity(64);
6102 for b in bytes {
6103 use std::fmt::Write;
6104 let _ = write!(hex, "{b:02x}");
6105 }
6106 Ok(hex)
6107}
6108
6109fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
6112 if a.len() != b.len() {
6113 return false;
6114 }
6115 let mut diff = 0u8;
6116 for (x, y) in a.iter().zip(b.iter()) {
6117 diff |= x ^ y;
6118 }
6119 diff == 0
6120}
6121
6122fn spawn_and_mark_running(
6123 spec: &ModuleSpec,
6124 runtime: &SupervisorRuntimeConfig,
6125 snapshot: &SharedSnapshot,
6126) -> Result<SupervisedChild, SuperviseError> {
6127 let child = spawn_child(
6128 spec,
6129 runtime.connection_file_path.as_deref(),
6130 runtime.supervisor_handle.as_ref(),
6131 &runtime.stderr_ring,
6132 runtime.capture_logs_dir.as_deref(),
6133 &runtime.child_roster,
6134 #[cfg(target_os = "linux")]
6135 runtime.cgroup_placement.as_ref(),
6136 )?;
6137 set_running(snapshot, &child, &spec.module_id, &runtime.spawn_events)?;
6138 Ok(child)
6139}
6140
6141enum RegistrationWaitOutcome {
6142 Registered,
6143 Exited(ExitReport),
6144 TimedOut,
6145}
6146
6147struct ReloadRegistrationFailure {
6148 exit_report: ExitReport,
6149 reason: String,
6150}
6151
6152#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6153enum BusyGaugeObservation {
6154 Quiescent,
6155 Busy,
6156 Omitted,
6157}
6158
6159fn busy_gauge_observation(metrics: Option<&Value>, gauges: &[String]) -> BusyGaugeObservation {
6160 let Some(metrics) = metrics.and_then(Value::as_object) else {
6161 return BusyGaugeObservation::Omitted;
6162 };
6163 let mut sum = 0u128;
6164 for gauge in gauges {
6165 let Some(value) = metrics.get(gauge) else {
6166 return BusyGaugeObservation::Omitted;
6167 };
6168 let Some(value) = value.as_u64() else {
6169 return BusyGaugeObservation::Busy;
6170 };
6171 sum = sum.saturating_add(u128::from(value));
6172 }
6173 if sum == 0 {
6174 BusyGaugeObservation::Quiescent
6175 } else {
6176 BusyGaugeObservation::Busy
6177 }
6178}
6179
6180fn declared_busy_gauges(
6181 registry: &Registry,
6182 module_id: &str,
6183) -> Result<Vec<String>, SuperviseError> {
6184 busy_gauges_of(
6185 registry
6186 .get_module(module_id)
6187 .map_err(SuperviseError::Registry)?,
6188 )
6189}
6190
6191fn declared_busy_gauges_for_connection(
6195 registry: &Registry,
6196 connection_id: ConnectionId,
6197) -> Result<Vec<String>, SuperviseError> {
6198 busy_gauges_of(
6199 registry
6200 .get_module_by_connection(connection_id)
6201 .map_err(SuperviseError::Registry)?,
6202 )
6203}
6204
6205fn busy_gauges_of(
6206 registration: Option<crate::registry::ModuleRegistration>,
6207) -> Result<Vec<String>, SuperviseError> {
6208 let Some(registration) = registration else {
6209 return Ok(Vec::new());
6210 };
6211 let Some(self_signals) = registration.manifest.self_signals else {
6212 return Ok(Vec::new());
6213 };
6214
6215 let mut gauges = Vec::new();
6216 for declaration in self_signals {
6217 if declaration.kind != SelfSignalKind::Busy {
6218 continue;
6219 }
6220 match declaration.anchored_to {
6221 SignalAnchor::HealthGauges { gauges: declared } if !declared.is_empty() => {
6222 gauges.extend(declared)
6223 }
6224 _ => {
6225 gauges.push(String::new());
6228 }
6229 }
6230 }
6231 Ok(gauges)
6232}
6233
6234async fn wait_for_forwarding_quiescence(
6239 forwarding: &ForwardingTable,
6240 module_id: &str,
6241 runtime: &SupervisorRuntimeConfig,
6242 endpoint: crate::ModuleEndpointId,
6243 deadline: Instant,
6244 busy_gauges: &[String],
6245 scope: DrainScope,
6246) -> Result<bool, SuperviseError> {
6247 let mut gauges_quiescent = busy_gauges.is_empty();
6248 let mut next_probe_at = Instant::now();
6249 let mut omission_counted = false;
6250
6251 loop {
6252 let now = Instant::now();
6253 if !busy_gauges.is_empty() && now >= next_probe_at && now < deadline {
6254 let report = match scope {
6255 DrainScope::Active => probe_module_health(module_id, runtime, Some(deadline)).await,
6256 DrainScope::Endpoint(endpoint) => {
6257 probe_endpoint_health(endpoint, runtime, Some(deadline)).await
6258 }
6259 };
6260 gauges_quiescent = match report {
6261 Ok(report) => match busy_gauge_observation(report.metrics.as_ref(), busy_gauges) {
6262 BusyGaugeObservation::Quiescent => true,
6263 BusyGaugeObservation::Busy => false,
6264 BusyGaugeObservation::Omitted => {
6265 if !omission_counted {
6266 forwarding
6267 .counters()
6268 .increment_drains_with_undeclared_gauge();
6269 omission_counted = true;
6270 }
6271 false
6272 }
6273 },
6274 Err(err) => {
6275 warn!(
6276 module_id,
6277 error = %err,
6278 "drain health.check did not produce declared busy gauges; treating module as busy"
6279 );
6280 false
6281 }
6282 };
6283 next_probe_at = Instant::now() + runtime.health.cadence.max(REGISTRY_RELEASE_POLL);
6284 }
6285
6286 let in_flight = forwarding
6287 .endpoint_in_flight_count(endpoint)
6288 .map_err(SuperviseError::Forwarding)?;
6289 if in_flight == 0 && gauges_quiescent {
6290 return Ok(true);
6291 }
6292
6293 let now = Instant::now();
6294 if now >= deadline {
6295 return Ok(false);
6296 }
6297 let mut wait = deadline
6298 .saturating_duration_since(now)
6299 .min(REGISTRY_RELEASE_POLL);
6300 if !busy_gauges.is_empty() {
6301 wait = wait.min(next_probe_at.saturating_duration_since(now));
6302 }
6303 sleep(wait).await;
6304 }
6305}
6306
6307fn drained_after_quiescence_wait(wait_result: &Result<bool, SuperviseError>) -> bool {
6315 match wait_result {
6316 Ok(drained) => *drained,
6317 Err(_) => false,
6318 }
6319}
6320
6321fn send_route_goodbyes(forwarding: &ForwardingTable, released_routes: Vec<GoodbyeTarget>) {
6322 for released in released_routes {
6323 let frame = match Frame::build_with_version(
6324 released.negotiated_ver,
6325 FrameType::Goodbye,
6326 control_flags(),
6327 released.channel,
6328 released.epoch,
6329 0,
6330 Vec::new(),
6331 ) {
6332 Ok(frame) => frame,
6333 Err(err) => {
6334 warn!(
6335 route_channel = released.channel,
6336 error = %err,
6337 "failed to build supervisor drain route GOODBYE frame"
6338 );
6339 continue;
6340 }
6341 };
6342 if !released.close_on_delivery_failure() {
6343 crate::forwarding::send_module_route_goodbye(
6344 &forwarding.counters(),
6345 &released.sink,
6346 frame,
6347 released.module_id.as_deref(),
6348 "supervisor drain",
6349 );
6350 continue;
6351 }
6352 if let Err(err) = released.sink.try_send(frame) {
6353 warn!(
6354 target_connection_id = released.connection_id.get(),
6355 route_channel = released.channel,
6356 error = %err,
6357 "supervisor drain route GOODBYE was not delivered to client; closing target connection"
6358 );
6359 let _ = forwarding.escalate_client_delivery_failure(
6360 released.connection_id,
6361 released.channel,
6362 released.epoch,
6363 CloseReason::new(
6364 "route_goodbye_delivery_failed",
6365 format!(
6366 "failed to enqueue supervisor drain route GOODBYE for channel {}: {err}",
6367 released.channel
6368 ),
6369 ),
6370 crate::forwarding::UndeliveredFrame {
6371 module_id: released.module_id.as_deref(),
6372 sink: &released.sink,
6373 },
6374 );
6375 }
6376 }
6377}
6378
6379fn send_module_draining(
6380 module_id: &str,
6381 reason: RouteCloseReason,
6382 deadline_ms: u64,
6383 target: &ModuleDrainTarget,
6384) {
6385 let body = match serde_json::to_vec(&ModuleControlCommand::Draining {
6386 reason,
6387 deadline_ms,
6388 }) {
6389 Ok(body) => body,
6390 Err(err) => {
6391 warn!(
6392 module_id,
6393 error = %err,
6394 "failed to encode module draining command"
6395 );
6396 return;
6397 }
6398 };
6399 let frame = match Frame::build_with_version(
6400 target.negotiated_ver,
6401 FrameType::Push,
6402 control_flags(),
6403 0,
6404 0,
6405 0,
6406 body,
6407 ) {
6408 Ok(frame) => frame,
6409 Err(err) => {
6410 warn!(
6411 module_id,
6412 error = %err,
6413 "failed to build module draining command frame"
6414 );
6415 return;
6416 }
6417 };
6418 if let Err(err) = target.sink.try_send(frame) {
6419 warn!(
6420 module_id,
6421 target_connection_id = target.endpoint.connection_id.get(),
6422 error = %err,
6423 "module draining command was not delivered to peer"
6424 );
6425 }
6426}
6427
6428fn send_module_goodbye(module_id: &str, forwarding: &ForwardingTable, target: &ModuleDrainTarget) {
6429 let frame = match Frame::build_with_version(
6430 target.negotiated_ver,
6431 FrameType::Goodbye,
6432 control_flags(),
6433 0,
6434 0,
6435 0,
6436 Vec::new(),
6437 ) {
6438 Ok(frame) => frame,
6439 Err(err) => {
6440 warn!(
6441 module_id,
6442 error = %err,
6443 "failed to build supervisor drain module GOODBYE frame"
6444 );
6445 return;
6446 }
6447 };
6448 if let Err(err) = target.sink.try_send(frame) {
6449 warn!(
6450 module_id,
6451 target_connection_id = target.endpoint.connection_id.get(),
6452 error = %err,
6453 "supervisor drain module GOODBYE was not delivered to peer; closing module connection"
6454 );
6455 forwarding.request_connection_close(
6456 target.endpoint.connection_id,
6457 CloseReason::new(
6458 "module_goodbye_delivery_failed",
6459 format!("failed to enqueue supervisor drain module GOODBYE for module '{module_id}': {err}"),
6460 ),
6461 );
6462 }
6463}
6464
6465#[derive(Clone, Copy)]
6466struct ForwardingDrainContext<'a> {
6467 spec: &'a ModuleSpec,
6468 runtime: &'a SupervisorRuntimeConfig,
6469 registry: &'a Registry,
6470 scope: DrainScope,
6471}
6472
6473#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6475enum DrainScope {
6476 Active,
6479 Endpoint(crate::ModuleEndpointId),
6484}
6485
6486#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6494enum StopNotice {
6495 SentOverConnection,
6498 NoConnection,
6502 NotSent,
6506}
6507
6508async fn begin_forwarding_drain(
6509 spec: &ModuleSpec,
6510 runtime: &SupervisorRuntimeConfig,
6511 registry: &Registry,
6512 snapshot: &SharedSnapshot,
6513 enabled: Option<bool>,
6514 reason: RouteCloseReason,
6515) -> Result<StopNotice, SuperviseError> {
6516 let Some(forwarding) = runtime.forwarding.as_ref() else {
6517 return Err(SuperviseError::ReloadUnavailable {
6518 module_id: spec.module_id.clone(),
6519 reason: "supervisor was not configured with a forwarding table".to_string(),
6520 });
6521 };
6522
6523 begin_forwarding_drain_with(
6524 forwarding,
6525 ForwardingDrainContext {
6526 spec,
6527 runtime,
6528 registry,
6529 scope: DrainScope::Active,
6530 },
6531 snapshot,
6532 enabled,
6533 reason,
6534 runtime.drain_timeout,
6535 )
6536 .await
6537}
6538
6539async fn begin_forwarding_drain_if_configured(
6540 spec: &ModuleSpec,
6541 runtime: &SupervisorRuntimeConfig,
6542 registry: &Registry,
6543 snapshot: &SharedSnapshot,
6544 enabled: Option<bool>,
6545 reason: RouteCloseReason,
6546) -> Result<StopNotice, SuperviseError> {
6547 begin_forwarding_drain_with_timeout(
6548 spec,
6549 runtime,
6550 registry,
6551 snapshot,
6552 enabled,
6553 reason,
6554 runtime.drain_timeout,
6555 )
6556 .await
6557}
6558
6559async fn begin_forwarding_drain_with_timeout(
6563 spec: &ModuleSpec,
6564 runtime: &SupervisorRuntimeConfig,
6565 registry: &Registry,
6566 snapshot: &SharedSnapshot,
6567 enabled: Option<bool>,
6568 reason: RouteCloseReason,
6569 drain_timeout: Duration,
6570) -> Result<StopNotice, SuperviseError> {
6571 let Some(forwarding) = runtime.forwarding.as_ref() else {
6572 return Ok(StopNotice::NotSent);
6573 };
6574
6575 begin_forwarding_drain_with(
6576 forwarding,
6577 ForwardingDrainContext {
6578 spec,
6579 runtime,
6580 registry,
6581 scope: DrainScope::Active,
6582 },
6583 snapshot,
6584 enabled,
6585 reason,
6586 drain_timeout,
6587 )
6588 .await
6589}
6590
6591async fn begin_forwarding_drain_with(
6592 forwarding: &ForwardingTable,
6593 context: ForwardingDrainContext<'_>,
6594 snapshot: &SharedSnapshot,
6595 enabled: Option<bool>,
6596 reason: RouteCloseReason,
6597 drain_timeout: Duration,
6598) -> Result<StopNotice, SuperviseError> {
6599 let ForwardingDrainContext {
6600 spec,
6601 runtime,
6602 registry,
6603 scope,
6604 } = context;
6605 debug_assert_ne!(reason, RouteCloseReason::Crash);
6606 let terminal = matches!(reason, RouteCloseReason::Disable);
6607 let drain_started_at = Instant::now();
6608 let drain_deadline = drain_started_at + drain_timeout;
6609 let deadline_ms =
6610 unix_ms_now().saturating_add(u64::try_from(drain_timeout.as_millis()).unwrap_or(u64::MAX));
6611 let busy_gauges = match scope {
6612 DrainScope::Active => declared_busy_gauges(registry, &spec.module_id)?,
6613 DrainScope::Endpoint(endpoint) => {
6614 declared_busy_gauges_for_connection(registry, endpoint.connection_id)?
6615 }
6616 };
6617
6618 let gate_started = Instant::now();
6621 let drain_target = match scope {
6622 DrainScope::Active => forwarding.begin_module_drain(&spec.module_id, reason),
6623 DrainScope::Endpoint(endpoint) => forwarding.begin_endpoint_drain(endpoint, reason),
6624 }
6625 .map_err(SuperviseError::Forwarding)?;
6626 info!(
6631 module_id = %spec.module_id,
6632 ?reason,
6633 gate_ms = u64::try_from(gate_started.elapsed().as_millis()).unwrap_or(u64::MAX),
6634 connected = drain_target.is_some(),
6635 "module drain began; route admission closed"
6636 );
6637 if scope == DrainScope::Active {
6638 update_snapshot(snapshot, Some(&spec.module_id), |state| {
6639 state.state = ModuleState::Draining;
6640 state.draining_to_replace =
6641 matches!(reason, RouteCloseReason::Restart | RouteCloseReason::Reload);
6642 if let Some(enabled) = enabled {
6643 state.enabled = enabled;
6644 }
6645 })?;
6646 }
6647
6648 let Some(target) = drain_target.as_ref() else {
6649 return Ok(StopNotice::NoConnection);
6653 };
6654 {
6655 send_module_draining(&spec.module_id, reason, deadline_ms, target);
6656 let routes = forwarding
6657 .endpoint_routes(target.endpoint)
6658 .map_err(SuperviseError::Forwarding)?;
6659 let routes_notified = routes.len();
6660 crate::control::send_route_control_pushes(
6661 forwarding,
6662 routes.clone(),
6663 ClientControlPush::RouteClosing {
6664 module_id: spec.module_id.clone(),
6665 reason,
6666 },
6667 );
6668 send_route_goodbyes(forwarding, target.abandoned_bindings.clone());
6669
6670 let wait_result = wait_for_forwarding_quiescence(
6676 forwarding,
6677 &spec.module_id,
6678 runtime,
6679 target.endpoint,
6680 drain_deadline,
6681 &busy_gauges,
6682 scope,
6683 )
6684 .await;
6685 let drained = drained_after_quiescence_wait(&wait_result);
6686 if let Err(err) = &wait_result {
6687 error!(
6688 module_id = %spec.module_id,
6689 ?reason,
6690 error = %err,
6691 "forwarding quiescence wait failed after route.closing; forcing route.closed(drained: false) so the client is not left waiting on an unfulfilled promise"
6692 );
6693 } else if !drained {
6694 let holdouts = forwarding
6700 .endpoint_drain_holdouts(target.endpoint)
6701 .unwrap_or_default();
6702 warn!(
6703 module_id = %spec.module_id,
6704 waited = ?drain_timeout,
6705 ?reason,
6706 held_requests = holdouts.requests,
6707 held_routes = holdouts.routes,
6708 total_routes = holdouts.total_routes,
6709 top_connections = ?holdouts.top_connections,
6710 held = %holdouts
6713 .held
6714 .iter()
6715 .map(|(channel, corr)| format!("{channel}:{corr}"))
6716 .collect::<Vec<_>>()
6717 .join(","),
6718 "route drain timed out before request quiescence; forcing teardown"
6719 );
6720 }
6721 crate::control::send_route_control_pushes(
6722 forwarding,
6723 routes,
6724 ClientControlPush::RouteClosed {
6725 module_id: spec.module_id.clone(),
6726 reason,
6727 drained,
6728 abandoned: target.abandoned_bindings.len() as u32,
6729 excluded_subscriptions: target.excluded_subscriptions,
6730 terminal: Some(terminal),
6731 },
6732 );
6733 wait_result?;
6734
6735 let released_routes = match forwarding.release_module_endpoint_routes(target.endpoint) {
6741 Ok(routes) => routes,
6742 Err(err) => {
6743 warn!(
6744 module_id = %spec.module_id,
6745 ?reason,
6746 error = %err,
6747 "failed to release module endpoint routes after route.closed; module GOODBYE will still be sent"
6748 );
6749 send_module_goodbye(&spec.module_id, forwarding, target);
6750 return Err(SuperviseError::Forwarding(err));
6751 }
6752 };
6753 let route_goodbye_count = released_routes.len();
6754 send_route_goodbyes(forwarding, released_routes);
6755 send_module_goodbye(&spec.module_id, forwarding, target);
6756
6757 info!(
6763 module_id = %spec.module_id,
6764 ?reason,
6765 routes_notified,
6766 route_goodbyes = route_goodbye_count,
6767 abandoned_reservations = target.abandoned_bindings.len(),
6768 excluded_subscriptions = target.excluded_subscriptions,
6769 drained,
6770 "module drain complete; consumers notified via route.closing/route.closed pushes and per-route GOODBYE frames"
6771 );
6772 }
6773
6774 Ok(StopNotice::SentOverConnection)
6775}
6776
6777async fn wait_for_registration_after_reload(
6780 registry: &Registry,
6781 module_id: &str,
6782 snapshot: &SharedSnapshot,
6783 child: &mut SupervisedChild,
6784 wait: Duration,
6785) -> Result<RegistrationWaitOutcome, SuperviseError> {
6786 wait_for_slot_registration(
6787 registry,
6788 crate::registry::RegistrationSlot::Active(module_id),
6789 module_id,
6790 snapshot,
6791 child,
6792 wait,
6793 )
6794 .await
6795}
6796
6797async fn wait_for_slot_registration(
6805 registry: &Registry,
6806 slot: crate::registry::RegistrationSlot<'_>,
6807 module_id: &str,
6808 snapshot: &SharedSnapshot,
6809 child: &mut SupervisedChild,
6810 wait: Duration,
6811) -> Result<RegistrationWaitOutcome, SuperviseError> {
6812 let deadline = Instant::now() + wait;
6813 loop {
6814 if registry
6815 .registration(slot)
6816 .map_err(SuperviseError::Registry)?
6817 .is_some()
6818 {
6819 return Ok(RegistrationWaitOutcome::Registered);
6820 }
6821
6822 let now = Instant::now();
6823 if now >= deadline {
6824 return Ok(RegistrationWaitOutcome::TimedOut);
6825 }
6826 let remaining = deadline.saturating_duration_since(now);
6827 let poll = remaining.min(REGISTRY_RELEASE_POLL);
6828
6829 tokio::select! {
6830 wait_result = child.wait() => {
6831 let status = wait_result.map_err(|source| SuperviseError::Wait {
6832 module_id: module_id.to_string(),
6833 source,
6834 })?;
6835 return Ok(RegistrationWaitOutcome::Exited(classify_reaped_child_exit(
6836 snapshot,
6837 child,
6838 &status,
6839 )));
6840 }
6841 _ = sleep(poll) => {}
6842 }
6843 }
6844}
6845
6846fn registration_failure_exit_report(mut exit_report: ExitReport) -> ExitReport {
6847 if exit_report.kind != ExitKind::DeliberateSeverance {
6850 exit_report.kind = ExitKind::Crash;
6851 }
6852 exit_report
6853}
6854
6855async fn handle_reload_child_registration_failure(
6856 spec: &ModuleSpec,
6857 runtime: &SupervisorRuntimeConfig,
6858 registry: &Registry,
6859 process_liveness: &SupervisorProcessLiveness,
6860 snapshot: &SharedSnapshot,
6861 child: &mut Option<SupervisedChild>,
6862 failure: ReloadRegistrationFailure,
6863) -> Result<(), SuperviseError> {
6864 let ReloadRegistrationFailure {
6865 exit_report,
6866 reason,
6867 } = failure;
6868 match on_child_exit(
6869 spec,
6870 runtime.restart_policy,
6871 registry,
6872 snapshot,
6873 &runtime.terminal_ring,
6874 &runtime.spawn_events,
6875 &runtime.child_roster,
6876 exit_report,
6877 )
6878 .await
6879 {
6880 NextAction::Stop {
6881 registration_released,
6882 } => {
6883 if registration_released {
6884 process_liveness.untrack_if_current(&spec.module_id, snapshot);
6885 }
6886 }
6887 NextAction::Restart { schedule } => {
6888 let delay = schedule.map_or(runtime.restart_policy.delay_for_restart(0), |schedule| {
6889 schedule.delay
6890 });
6891 if let Some(schedule) = schedule {
6892 log_crash_respawn(&spec.module_id, schedule);
6893 }
6894 sleep(delay).await;
6895 if respawn_still_pending(snapshot) {
6899 if let Err(err) = wait_for_registration_release(
6900 registry,
6901 &spec.module_id,
6902 REGISTRY_RELEASE_TIMEOUT,
6903 )
6904 .await
6905 {
6906 fail_snapshot(snapshot, Some(&spec.module_id), None);
6907 process_liveness.untrack_if_current(&spec.module_id, snapshot);
6908 return Err(SuperviseError::ReloadFailed {
6909 module_id: spec.module_id.clone(),
6910 reason: format!(
6911 "{reason}; registration did not release before policy retry: {err}"
6912 ),
6913 });
6914 }
6915 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
6916 match spawn_and_mark_running(spec, runtime, snapshot) {
6917 Ok(next_child) => {
6918 *child = Some(next_child);
6919 }
6920 Err(err) => {
6921 fail_snapshot(snapshot, Some(&spec.module_id), None);
6922 process_liveness.untrack_if_current(&spec.module_id, snapshot);
6923 return Err(SuperviseError::ReloadFailed {
6924 module_id: spec.module_id.clone(),
6925 reason: format!("{reason}; policy retry spawn failed: {err}"),
6926 });
6927 }
6928 }
6929 }
6930 }
6931 }
6932
6933 Err(SuperviseError::ReloadFailed {
6934 module_id: spec.module_id.clone(),
6935 reason,
6936 })
6937}
6938
6939async fn handle_reload_spawn_failure(
6940 spec: &ModuleSpec,
6941 runtime: &SupervisorRuntimeConfig,
6942 process_liveness: &SupervisorProcessLiveness,
6943 snapshot: &SharedSnapshot,
6944 child: &mut Option<SupervisedChild>,
6945 reason: String,
6946) -> Result<(), SuperviseError> {
6947 let mut should_retry = false;
6948 let now = Instant::now();
6949 update_snapshot(snapshot, Some(&spec.module_id), |state| {
6950 clear_current_process_facts(state);
6951 if daemon_will_restart(state, &runtime.restart_policy, now) {
6952 state.record_crash_restart(&runtime.restart_policy, now);
6953 state.state = ModuleState::Restarting;
6954 should_retry = true;
6955 } else if state.enabled {
6956 state.state = ModuleState::Failed;
6957 } else {
6958 state.state = ModuleState::Disabled;
6959 }
6960 })?;
6961
6962 if should_retry {
6963 sleep(runtime.restart_policy.backoff).await;
6964 if respawn_still_pending(snapshot) {
6968 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
6969 match spawn_and_mark_running(spec, runtime, snapshot) {
6970 Ok(next_child) => {
6971 *child = Some(next_child);
6972 }
6973 Err(err) => {
6974 fail_snapshot(snapshot, Some(&spec.module_id), None);
6975 process_liveness.untrack_if_current(&spec.module_id, snapshot);
6976 return Err(SuperviseError::ReloadFailed {
6977 module_id: spec.module_id.clone(),
6978 reason: format!("{reason}; policy retry spawn failed: {err}"),
6979 });
6980 }
6981 }
6982 }
6983 } else {
6984 process_liveness.untrack_if_current(&spec.module_id, snapshot);
6985 }
6986
6987 Err(SuperviseError::ReloadFailed {
6988 module_id: spec.module_id.clone(),
6989 reason,
6990 })
6991}
6992
6993fn control_flags() -> Flags {
6994 Flags::new(false, Priority::Passive, false)
6995}
6996
6997#[allow(clippy::too_many_arguments)]
6998async fn drain_optional_child(
6999 module_id: &str,
7000 protocol: ModuleProtocol,
7001 stop_notice: StopNotice,
7002 registry: &Registry,
7003 snapshot: &SharedSnapshot,
7004 terminal_ring: &Arc<Mutex<TerminalRing>>,
7005 spawn_events: &SpawnEventFeed,
7006 child: &mut Option<SupervisedChild>,
7007 drain_timeout: Duration,
7008 final_state: ModuleState,
7009 enabled: Option<bool>,
7010) -> Result<(), SuperviseError> {
7011 if let Some(child) = child.take() {
7012 drain_child_to_state(
7013 module_id,
7014 protocol,
7015 stop_notice,
7016 registry,
7017 snapshot,
7018 terminal_ring,
7019 spawn_events,
7020 child,
7021 drain_timeout,
7022 final_state,
7023 enabled,
7024 )
7025 .await
7026 } else {
7027 update_snapshot(snapshot, Some(module_id), |state| {
7028 state.state = final_state;
7029 if let Some(enabled) = enabled {
7030 state.enabled = enabled;
7031 }
7032 clear_current_process_facts(state);
7033 })?;
7034 wait_for_registration_release(registry, module_id, REGISTRY_RELEASE_TIMEOUT).await
7035 }
7036}
7037
7038#[allow(clippy::too_many_arguments)]
7039async fn drain_child_to_state(
7040 module_id: &str,
7041 protocol: ModuleProtocol,
7042 stop_notice: StopNotice,
7043 registry: &Registry,
7044 snapshot: &SharedSnapshot,
7045 terminal_ring: &Arc<Mutex<TerminalRing>>,
7046 spawn_events: &SpawnEventFeed,
7047 mut child: SupervisedChild,
7048 drain_timeout: Duration,
7049 final_state: ModuleState,
7050 enabled: Option<bool>,
7051) -> Result<(), SuperviseError> {
7052 update_snapshot(snapshot, Some(module_id), |state| {
7053 state.state = ModuleState::Draining;
7054 state.draining_to_replace = final_state == ModuleState::Restarting;
7055 if let Some(enabled) = enabled {
7056 state.enabled = enabled;
7057 }
7058 })?;
7059
7060 if stop_notice != StopNotice::SentOverConnection {
7071 if protocol == ModuleProtocol::Subc && stop_notice == StopNotice::NoConnection {
7072 info!(
7073 module_id,
7074 pid = child.pid,
7075 budget_ms = u64::try_from(drain_timeout.as_millis()).unwrap_or(u64::MAX),
7076 "module has no connection yet; requesting stop by signal"
7077 );
7078 }
7079 request_graceful_stop(module_id, &child);
7080 }
7081
7082 let exit_report = match timeout(drain_timeout, child.wait()).await {
7083 Ok(Ok(status)) => classify_reaped_child_exit(snapshot, &child, &status),
7084 Ok(Err(source)) => {
7085 fail_snapshot(snapshot, Some(module_id), None);
7086 return Err(SuperviseError::Wait {
7087 module_id: module_id.to_string(),
7088 source,
7089 });
7090 }
7091 Err(_) => {
7092 warn!(
7105 module_id,
7106 pid = child.pid,
7107 budget_ms = u64::try_from(drain_timeout.as_millis()).unwrap_or(u64::MAX),
7108 reason = ?final_state,
7109 ?stop_notice,
7110 "drain budget expired before the module exited; killing it"
7111 );
7112 child.start_kill().map_err(|source| {
7113 fail_snapshot(snapshot, Some(module_id), None);
7114 SuperviseError::Kill {
7115 module_id: module_id.to_string(),
7116 source,
7117 }
7118 })?;
7119 let status = child.wait().await.map_err(|source| {
7120 fail_snapshot(snapshot, Some(module_id), None);
7121 SuperviseError::Wait {
7122 module_id: module_id.to_string(),
7123 source,
7124 }
7125 })?;
7126 classify_reaped_child_exit(snapshot, &child, &status)
7127 }
7128 };
7129
7130 update_snapshot(snapshot, Some(module_id), |state| {
7131 state.state = final_state;
7132 if let Some(enabled) = enabled {
7133 state.enabled = enabled;
7134 }
7135 clear_current_process_facts(state);
7136 state.last_exit = Some(exit_report.clone());
7137 if exit_report.kind == ExitKind::DeliberateSeverance {
7138 state.lifetime_restarts += 1;
7139 }
7140 })?;
7141 record_terminal(
7142 module_id,
7143 terminal_ring,
7144 spawn_events,
7145 &exit_report,
7146 terminal_disposition(final_state),
7147 );
7148 child.drain_stderr(module_id).await;
7149
7150 wait_for_registration_release(registry, module_id, REGISTRY_RELEASE_TIMEOUT).await
7151}
7152
7153#[cfg(unix)]
7173fn request_graceful_stop(module_id: &str, child: &SupervisedChild) {
7174 let Some(pid) = child
7175 .id()
7176 .and_then(|pid| i32::try_from(pid).ok())
7177 .and_then(rustix::process::Pid::from_raw)
7178 else {
7179 debug!(
7180 module_id,
7181 "no pid to signal for teardown; falling through to the drain wait"
7182 );
7183 return;
7184 };
7185 match rustix::process::kill_process(pid, rustix::process::Signal::TERM) {
7186 Ok(()) => debug!(
7187 module_id,
7188 "sent SIGTERM to a module nothing else asked to stop"
7189 ),
7190 Err(err) => debug!(
7191 module_id,
7192 error = %err,
7193 "SIGTERM to module failed; the drain wait and kill still apply"
7194 ),
7195 }
7196}
7197
7198#[cfg(not(unix))]
7206fn request_graceful_stop(module_id: &str, _child: &SupervisedChild) {
7207 debug!(
7208 module_id,
7209 "no graceful stop signal exists on this platform; teardown of a module nothing asked to stop waits, then kills"
7210 );
7211}
7212
7213fn terminal_disposition(final_state: ModuleState) -> TerminalDisposition {
7214 match final_state {
7215 ModuleState::Stopped => TerminalDisposition::Stopped,
7216 ModuleState::Disabled => TerminalDisposition::Disabled,
7217 ModuleState::Restarting => TerminalDisposition::Restarting,
7218 ModuleState::Failed => TerminalDisposition::Failed,
7219 ModuleState::Starting
7220 | ModuleState::Running
7221 | ModuleState::Unresponsive
7222 | ModuleState::Draining => {
7223 unreachable!("terminal exits only finish in terminal or restarting states")
7224 }
7225 }
7226}
7227
7228async fn wait_for_registration_release(
7231 registry: &Registry,
7232 module_id: &str,
7233 wait: Duration,
7234) -> Result<(), SuperviseError> {
7235 wait_for_slot_registration_release(
7236 registry,
7237 crate::registry::RegistrationSlot::Active(module_id),
7238 wait,
7239 )
7240 .await
7241}
7242
7243async fn wait_for_slot_registration_release(
7251 registry: &Registry,
7252 slot: crate::registry::RegistrationSlot<'_>,
7253 wait: Duration,
7254) -> Result<(), SuperviseError> {
7255 let deadline = Instant::now() + wait;
7256 let mut release_events = registration_release_events().subscribe();
7257 let still_active = |registration: &crate::registry::ModuleRegistration| {
7258 SuperviseError::RegistrationStillActive {
7259 module_id: registration.manifest.module_id.clone(),
7260 waited: wait,
7261 }
7262 };
7263 loop {
7264 let _observed_generation = *release_events.borrow_and_update();
7265 let Some(registration) = registry
7266 .registration(slot)
7267 .map_err(SuperviseError::Registry)?
7268 else {
7269 return Ok(());
7270 };
7271
7272 let now = Instant::now();
7273 if now >= deadline {
7274 return Err(still_active(®istration));
7275 }
7276
7277 let remaining = deadline.saturating_duration_since(now);
7278 match timeout(remaining, release_events.changed()).await {
7279 Ok(Ok(())) | Ok(Err(_)) => {}
7280 Err(_) => return Err(still_active(®istration)),
7281 }
7282 }
7283}
7284
7285#[cfg(test)]
7286mod slot_registration_wait_tests {
7287 use super::*;
7288 use crate::registry::{ConnectionId, RegistrationSlot};
7289 use subc_protocol::manifest::ModuleManifest;
7290
7291 const INCUMBENT: u64 = 1;
7292 const CANDIDATE: u64 = 2;
7293
7294 fn swapped_registry() -> Arc<Registry> {
7295 let registry = Arc::new(Registry::default());
7296 let manifest = ModuleManifest::builder("m", "0.1.0").build();
7297 registry
7298 .register_with_control_ops(
7299 manifest.clone(),
7300 1,
7301 ConnectionId::new(INCUMBENT),
7302 Vec::new(),
7303 )
7304 .unwrap();
7305 registry
7306 .register_candidate_with_control_ops(
7307 manifest,
7308 1,
7309 ConnectionId::new(CANDIDATE),
7310 Vec::new(),
7311 )
7312 .unwrap();
7313 registry
7314 }
7315
7316 #[tokio::test]
7320 async fn incumbent_release_is_awaited_by_connection_not_by_module_id() {
7321 let registry = swapped_registry();
7322 registry.promote_candidate("m").unwrap().unwrap();
7323
7324 assert!(matches!(
7325 wait_for_registration_release(®istry, "m", Duration::from_millis(50)).await,
7326 Err(SuperviseError::RegistrationStillActive { .. })
7327 ));
7328
7329 assert!(matches!(
7331 wait_for_slot_registration_release(
7332 ®istry,
7333 RegistrationSlot::Connection(ConnectionId::new(INCUMBENT)),
7334 Duration::from_millis(50),
7335 )
7336 .await,
7337 Err(SuperviseError::RegistrationStillActive { .. })
7338 ));
7339
7340 let releaser = Arc::clone(®istry);
7341 let release = tokio::spawn(async move {
7342 sleep(Duration::from_millis(20)).await;
7343 releaser
7344 .deregister_connection(ConnectionId::new(INCUMBENT))
7345 .unwrap();
7346 notify_registration_release();
7347 });
7348 wait_for_slot_registration_release(
7349 ®istry,
7350 RegistrationSlot::Connection(ConnectionId::new(INCUMBENT)),
7351 Duration::from_secs(5),
7352 )
7353 .await
7354 .expect("the incumbent's own registration is released");
7355 release.await.unwrap();
7356 assert!(registry.get_module("m").unwrap().is_some());
7357 }
7358
7359 #[tokio::test]
7362 async fn candidate_slot_wait_ignores_the_incumbents_registration() {
7363 let registry = swapped_registry();
7364 assert!(matches!(
7365 wait_for_slot_registration_release(
7366 ®istry,
7367 RegistrationSlot::Candidate("m"),
7368 Duration::from_millis(50),
7369 )
7370 .await,
7371 Err(SuperviseError::RegistrationStillActive { .. })
7372 ));
7373 registry
7374 .deregister_connection(ConnectionId::new(CANDIDATE))
7375 .unwrap();
7376 wait_for_slot_registration_release(
7377 ®istry,
7378 RegistrationSlot::Candidate("m"),
7379 Duration::from_millis(50),
7380 )
7381 .await
7382 .expect("a candidate slot with no candidate is released");
7383 assert!(registry
7384 .registration(RegistrationSlot::Active("m"))
7385 .unwrap()
7386 .is_some());
7387 }
7388}
7389
7390fn classify_exit(status: &ExitStatus) -> ExitReport {
7391 ExitReport {
7392 kind: if status.success() {
7393 ExitKind::Clean
7394 } else {
7395 ExitKind::Crash
7396 },
7397 code: status.code(),
7398 signal: exit_signal(status),
7399 at_ms: unix_ms_now(),
7400 }
7401}
7402
7403fn wait_error_exit_report() -> ExitReport {
7409 ExitReport {
7410 kind: ExitKind::Crash,
7411 code: None,
7412 signal: None,
7413 at_ms: unix_ms_now(),
7414 }
7415}
7416
7417#[cfg(unix)]
7418fn exit_signal(status: &ExitStatus) -> Option<i32> {
7419 use std::os::unix::process::ExitStatusExt;
7420
7421 status.signal()
7422}
7423
7424#[cfg(not(unix))]
7425fn exit_signal(_status: &ExitStatus) -> Option<i32> {
7426 None
7427}
7428
7429fn reset_restart_count(snapshot: &SharedSnapshot, module_id: &str) -> Result<(), SuperviseError> {
7435 update_snapshot(snapshot, Some(module_id), |state| {
7436 state.clear_crash_restarts();
7437 })
7438}
7439
7440fn set_running(
7441 snapshot: &SharedSnapshot,
7442 child: &SupervisedChild,
7443 module_id: &str,
7444 spawn_events: &SpawnEventFeed,
7445) -> Result<(), SuperviseError> {
7446 let mut state = snapshot.lock().map_err(|_| SuperviseError::StatePoisoned {
7447 module_id: Some(module_id.to_string()),
7448 })?;
7449 state.spawn_generation = spawn_events.emit_spawned(module_id, child.pid, child.spawned_at_ms);
7450 state.in_alternate_slot = false;
7453 state.configuration_updated_since_spawn = false;
7454 state.state = ModuleState::Running;
7455 state.enabled = true;
7456 state.process_alive = true;
7457 state.pid = child.id();
7458 state.spawned_at_ms = Some(child.spawned_at_ms);
7459 state.spawned_from = Some(child.spawned_from.clone());
7460 state.spawned_file_identity = child.spawned_file_identity;
7461 state.process_start_time = child.process_start_time;
7462 Ok(())
7463}
7464
7465fn clear_current_process_facts(state: &mut SupervisorSnapshot) {
7466 state.process_alive = false;
7467 state.pid = None;
7468 state.spawned_at_ms = None;
7469 state.spawned_from = None;
7470 state.spawned_file_identity = None;
7471 state.process_start_time = None;
7472 state.deliberate_severance = None;
7473}
7474
7475#[cfg(test)]
7476fn record_deliberate_severance(
7477 snapshot: &SharedSnapshot,
7478 identity: ProcessIdentity,
7479) -> Result<(), SuperviseError> {
7480 update_snapshot(snapshot, None, |state| {
7481 state.deliberate_severance = Some(identity);
7482 })
7483}
7484
7485fn apply_deliberate_severance_marker(
7486 snapshot: &SharedSnapshot,
7487 exited_identity: Option<ProcessIdentity>,
7488 mut exit_report: ExitReport,
7489) -> ExitReport {
7490 let marker = lock_snapshot(snapshot)
7491 .ok()
7492 .and_then(|mut state| state.deliberate_severance.take());
7493 if marker.is_some() && marker == exited_identity {
7494 exit_report.kind = ExitKind::DeliberateSeverance;
7495 }
7496 exit_report
7497}
7498
7499fn classify_reaped_child_exit(
7500 snapshot: &SharedSnapshot,
7501 child: &SupervisedChild,
7502 status: &ExitStatus,
7503) -> ExitReport {
7504 apply_deliberate_severance_marker(snapshot, child.process_identity(), classify_exit(status))
7505}
7506
7507fn fail_snapshot(
7508 snapshot: &SharedSnapshot,
7509 module_id: Option<&str>,
7510 last_exit: Option<ExitReport>,
7511) {
7512 if let Err(err) = update_snapshot(snapshot, module_id, |state| {
7513 state.state = ModuleState::Failed;
7514 clear_current_process_facts(state);
7515 if let Some(last_exit) = last_exit {
7516 state.last_exit = Some(last_exit);
7517 }
7518 }) {
7519 error!(error = %err, "failed to mark supervisor state failed");
7520 }
7521}
7522
7523fn update_snapshot(
7524 snapshot: &SharedSnapshot,
7525 module_id: Option<&str>,
7526 update: impl FnOnce(&mut SupervisorSnapshot),
7527) -> Result<(), SuperviseError> {
7528 let mut state = snapshot.lock().map_err(|_| SuperviseError::StatePoisoned {
7529 module_id: module_id.map(ToOwned::to_owned),
7530 })?;
7531 update(&mut state);
7532 Ok(())
7533}
7534
7535const SLOW_SNAPSHOT_LOCK_THRESHOLD: Duration = Duration::from_millis(250);
7536
7537fn lock_snapshot_for_control<'a>(
7538 snapshot: &'a SharedSnapshot,
7539 module_id: &str,
7540 caller: &'static str,
7541) -> Result<std::sync::MutexGuard<'a, SupervisorSnapshot>, SuperviseError> {
7542 let started_at = Instant::now();
7543 let guard = lock_snapshot(snapshot)?;
7544 let waited = started_at.elapsed();
7545 if waited >= SLOW_SNAPSHOT_LOCK_THRESHOLD {
7546 warn!(
7547 module_id = %module_id,
7548 waited_ms = waited.as_millis() as u64,
7549 caller = %caller,
7550 "slow snapshot lock"
7551 );
7552 }
7553 Ok(guard)
7554}
7555
7556fn lock_snapshot(
7557 snapshot: &SharedSnapshot,
7558) -> Result<std::sync::MutexGuard<'_, SupervisorSnapshot>, SuperviseError> {
7559 snapshot
7560 .lock()
7561 .map_err(|_| SuperviseError::StatePoisoned { module_id: None })
7562}
7563
7564#[cfg(test)]
7565mod terminal_history_tests {
7566 use std::{
7567 path::PathBuf,
7568 sync::Arc,
7569 time::{Duration, Instant},
7570 };
7571
7572 use tokio::time::sleep;
7573
7574 use super::{
7575 apply_deliberate_severance_marker, daemon_will_restart, drain_child_to_state,
7576 drained_after_quiescence_wait, handle_reload_spawn_failure, health_restart_child,
7577 lock_snapshot, on_child_exit, record_deliberate_severance, record_wait_error_terminal,
7578 reset_restart_count, spawn_and_mark_running, update_snapshot, wait_error_exit_report,
7579 ExitKind, ExitReport, ModuleProtocol, ModuleSpec, ModuleState, NextAction, ProcessIdentity,
7580 RestartPolicy, SpawnEventKind, StopNotice, SuperviseError, SupervisedModule, Supervisor,
7581 SupervisorHandle, SupervisorHealthStatus, SupervisorSnapshot,
7582 };
7583 use super::Instant as ClockInstant;
7588 use crate::{
7589 registry::Registry,
7590 terminal_ring::{TerminalRing, TerminalRingConfig},
7591 };
7592 use std::sync::Mutex;
7593 use subc_control::TerminalDisposition;
7594
7595 fn fake_aft_stub_path() -> PathBuf {
7600 let mut path = std::env::current_exe().expect("current_exe available in tests");
7601 path.pop();
7602 path.pop();
7603 path.push(if cfg!(windows) {
7604 "fake-aft-stub.exe"
7605 } else {
7606 "fake-aft-stub"
7607 });
7608 assert!(
7609 path.exists(),
7610 "fake-aft-stub not built at {}: run `cargo test -p subc-core` (which builds \
7611 [[bin]] targets) rather than `cargo test -p subc-core --lib` (which does not)",
7612 path.display()
7613 );
7614 path
7615 }
7616
7617 #[test]
7618 fn reserved_never_spawned_refuses_every_hello() {
7619 let supervisor = SupervisorHandle::default();
7624 supervisor.apply_identity_configuration(&ModuleSpec {
7625 module_id: "never-spawned".to_string(),
7626 program: PathBuf::from("/usr/bin/false"),
7627 args: Vec::new(),
7628 env: Vec::new(),
7629 reserved: true,
7630 reserved_prefixes: Vec::new(),
7631 protocol: ModuleProtocol::Subc,
7632 overlap: Default::default(),
7633 });
7634 assert!(
7635 supervisor
7636 .reserved_hello_rejection("never-spawned", Some("any-forged-nonce"))
7637 .is_some(),
7638 "forged nonce must refuse on a reserved never-spawned id"
7639 );
7640 assert!(
7641 supervisor
7642 .reserved_hello_rejection("never-spawned", None)
7643 .is_some(),
7644 "absent nonce must refuse on a reserved never-spawned id"
7645 );
7646 supervisor.set_spawn_nonce("never-spawned", "minted".to_string());
7648 supervisor.apply_identity_configuration(&ModuleSpec {
7649 module_id: "never-spawned".to_string(),
7650 program: PathBuf::from("/usr/bin/false"),
7651 args: Vec::new(),
7652 env: Vec::new(),
7653 reserved: true,
7654 reserved_prefixes: Vec::new(),
7655 protocol: ModuleProtocol::Subc,
7656 overlap: Default::default(),
7657 });
7658 assert!(supervisor
7659 .reserved_hello_rejection("never-spawned", Some("minted"))
7660 .is_none());
7661 assert!(supervisor
7662 .reserved_hello_rejection("never-spawned", Some("forged"))
7663 .is_some());
7664 }
7665
7666 fn seed_crash_restarts(state: &mut SupervisorSnapshot, count: u32) {
7669 let now = ClockInstant::now();
7670 for _ in 0..count {
7671 state.crash_restarts.push_back(now);
7672 }
7673 }
7674
7675 fn age_oldest_crash_restart_out_of_window(state: &mut SupervisorSnapshot, window: Duration) {
7679 let aged = state
7680 .crash_restarts
7681 .front()
7682 .expect("a crash restart must be recorded before it can be aged")
7683 .checked_sub(window + Duration::from_secs(1))
7684 .expect("the test clock is far enough from its origin to age an instant");
7685 state.crash_restarts[0] = aged;
7686 }
7687
7688 fn snapshot_with_restarts(enabled: bool, count: u32) -> SupervisorSnapshot {
7689 let mut state = SupervisorSnapshot::new(ModuleState::Running, enabled);
7690 seed_crash_restarts(&mut state, count);
7691 state
7692 }
7693
7694 #[test]
7695 fn daemon_owned_recovery_predicate_uses_the_pre_increment_budget() {
7696 let policy = RestartPolicy::new(3, Duration::ZERO);
7697 let now = ClockInstant::now();
7698 assert!(daemon_will_restart(
7699 &mut snapshot_with_restarts(true, 2),
7700 &policy,
7701 now
7702 ));
7703 assert!(!daemon_will_restart(
7704 &mut snapshot_with_restarts(true, 3),
7705 &policy,
7706 now
7707 ));
7708 assert!(!daemon_will_restart(
7709 &mut snapshot_with_restarts(false, 0),
7710 &policy,
7711 now
7712 ));
7713 }
7714
7715 #[test]
7716 fn crash_restart_backoff_escalates_with_in_window_count() {
7717 let policy = RestartPolicy::new(4, Duration::from_millis(100))
7718 .with_max_backoff(Duration::from_secs(30));
7719 let now = ClockInstant::now();
7720 let mut state = SupervisorSnapshot::new(ModuleState::Running, true);
7721 let schedules = (0..4)
7722 .map(|_| {
7723 state
7724 .next_crash_restart(&policy, now)
7725 .expect("the test policy allows four crash restarts")
7726 })
7727 .collect::<Vec<_>>();
7728
7729 assert_eq!(
7730 schedules
7731 .iter()
7732 .map(|schedule| schedule.restart_in_window)
7733 .collect::<Vec<_>>(),
7734 vec![0, 1, 2, 3]
7735 );
7736 assert_eq!(
7737 schedules
7738 .iter()
7739 .map(|schedule| schedule.delay)
7740 .collect::<Vec<_>>(),
7741 vec![
7742 Duration::from_millis(100),
7743 Duration::from_secs(1),
7744 Duration::from_secs(10),
7745 Duration::from_secs(30),
7746 ]
7747 );
7748 }
7749
7750 #[test]
7751 fn crash_restart_backoff_resets_after_ring_clear() {
7752 let policy = RestartPolicy::new(3, Duration::from_millis(100));
7753 let now = ClockInstant::now();
7754 let mut state = SupervisorSnapshot::new(ModuleState::Running, true);
7755 assert_eq!(
7756 state.next_crash_restart(&policy, now).unwrap().delay,
7757 Duration::from_millis(100)
7758 );
7759 assert_eq!(
7760 state.next_crash_restart(&policy, now).unwrap().delay,
7761 Duration::from_secs(1)
7762 );
7763
7764 state.clear_crash_restarts();
7765 let schedule = state
7766 .next_crash_restart(&policy, now)
7767 .expect("a cleared ring must allow another restart");
7768 assert_eq!(schedule.restart_in_window, 0);
7769 assert_eq!(schedule.delay, Duration::from_millis(100));
7770 }
7771
7772 #[test]
7773 fn crash_restart_backoff_ignores_aged_restarts() {
7774 let policy = RestartPolicy::new(3, Duration::from_millis(100));
7775 let now = ClockInstant::now();
7776 let mut state = SupervisorSnapshot::new(ModuleState::Running, true);
7777 state
7778 .next_crash_restart(&policy, now)
7779 .expect("the first restart is allowed");
7780 state
7781 .next_crash_restart(&policy, now)
7782 .expect("the second restart is allowed");
7783 state.crash_restarts[0] = now
7784 .checked_sub(policy.window + Duration::from_secs(1))
7785 .expect("the fake clock can age a restart past the window");
7786
7787 let schedule = state
7788 .next_crash_restart(&policy, now)
7789 .expect("an aged restart must release its slot");
7790 assert_eq!(schedule.restart_in_window, 1);
7791 assert_eq!(schedule.delay, Duration::from_secs(1));
7792 assert_eq!(state.crash_restarts.len(), 2);
7793 }
7794
7795 #[test]
7799 fn a_budget_spent_before_the_window_no_longer_refuses() {
7800 let policy = RestartPolicy::new(3, Duration::ZERO);
7801 let mut state = snapshot_with_restarts(true, 3);
7802 let now = ClockInstant::now();
7803 assert!(!daemon_will_restart(&mut state, &policy, now));
7804
7805 assert!(daemon_will_restart(
7806 &mut state,
7807 &policy,
7808 now + policy.window + Duration::from_secs(1)
7809 ));
7810 assert!(
7811 state.crash_restarts.is_empty(),
7812 "reading the budget must drop the instants that left the window"
7813 );
7814 }
7815
7816 fn module_with_recovery_snapshot(
7817 state: ModuleState,
7818 enabled: bool,
7819 restart_count: u32,
7820 ) -> SupervisedModule {
7821 let registry = Arc::new(Registry::default());
7822 let supervisor =
7823 Supervisor::new(Arc::clone(®istry), RestartPolicy::new(3, Duration::ZERO));
7824 let module = supervisor
7825 .spawn(ModuleSpec {
7826 module_id: "recovery-snapshot".to_string(),
7827 program: fake_aft_stub_path(),
7828 args: Vec::new(),
7829 env: Vec::new(),
7830 reserved: false,
7831 reserved_prefixes: Vec::new(),
7832 protocol: ModuleProtocol::Subc,
7833 overlap: Default::default(),
7834 })
7835 .unwrap();
7836 update_snapshot(
7837 &module.inner.snapshot,
7838 Some("recovery-snapshot"),
7839 |snapshot| {
7840 snapshot.state = state;
7841 snapshot.enabled = enabled;
7842 seed_crash_restarts(snapshot, restart_count);
7843 },
7844 )
7845 .unwrap();
7846 module
7847 }
7848
7849 #[cfg(target_os = "linux")]
7850 #[tokio::test]
7851 async fn no_cgroup_placement_does_not_block_fake_aft_stub_spawn() {
7852 let supervisor = Supervisor::new(Arc::new(Registry::default()), RestartPolicy::default())
7853 .with_cgroup_placement(None);
7854 let result = supervisor.spawn(ModuleSpec {
7855 module_id: "no-cgroup-placement".to_string(),
7856 program: fake_aft_stub_path(),
7857 args: Vec::new(),
7858 env: Vec::new(),
7859 reserved: false,
7860 reserved_prefixes: Vec::new(),
7861 protocol: ModuleProtocol::Subc,
7862 overlap: Default::default(),
7863 });
7864
7865 assert!(
7866 result.is_ok(),
7867 "no delegation must not turn an otherwise valid spawn into a failure: {result:?}"
7868 );
7869 }
7870
7871 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7872 async fn undecided_snapshot_uses_shared_restart_predicate() {
7873 assert!(module_with_recovery_snapshot(ModuleState::Running, true, 2)
7874 .will_recover_after_connection_loss()
7875 .unwrap());
7876 assert!(
7877 !module_with_recovery_snapshot(ModuleState::Running, true, 3)
7878 .will_recover_after_connection_loss()
7879 .unwrap()
7880 );
7881 }
7882
7883 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7884 async fn restarting_snapshot_at_exhausted_budget_is_non_terminal() {
7885 assert!(
7886 module_with_recovery_snapshot(ModuleState::Restarting, true, 3)
7887 .will_recover_after_connection_loss()
7888 .unwrap()
7889 );
7890 }
7891
7892 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7893 async fn terminal_phase_snapshots_are_terminal_before_budget_exhaustion() {
7894 assert!(!module_with_recovery_snapshot(ModuleState::Failed, true, 0)
7895 .will_recover_after_connection_loss()
7896 .unwrap());
7897 assert!(
7898 !module_with_recovery_snapshot(ModuleState::Disabled, true, 0)
7899 .will_recover_after_connection_loss()
7900 .unwrap()
7901 );
7902 }
7903
7904 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7905 async fn warming_snapshot_is_limited_to_startup_phases() {
7906 for state in [
7907 ModuleState::Starting,
7908 ModuleState::Running,
7909 ModuleState::Restarting,
7910 ] {
7911 assert!(
7912 module_with_recovery_snapshot(state, true, 0)
7913 .is_warming()
7914 .unwrap(),
7915 "{state:?} should be warming"
7916 );
7917 }
7918 for state in [
7919 ModuleState::Unresponsive,
7920 ModuleState::Draining,
7921 ModuleState::Stopped,
7922 ModuleState::Failed,
7923 ModuleState::Disabled,
7924 ] {
7925 assert!(
7926 !module_with_recovery_snapshot(state, true, 0)
7927 .is_warming()
7928 .unwrap(),
7929 "{state:?} should not be warming"
7930 );
7931 }
7932 }
7933
7934 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7935 async fn terminal_history_survives_respawn_and_keeps_both_crashes_in_order() {
7936 let registry = Arc::new(Registry::default());
7937 let supervisor =
7938 Supervisor::new(Arc::clone(®istry), RestartPolicy::new(1, Duration::ZERO));
7939 let module = supervisor
7940 .spawn(ModuleSpec {
7941 module_id: "terminal-history".to_string(),
7942 program: fake_aft_stub_path(),
7943 args: Vec::new(),
7944 env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
7945 reserved: false,
7946 reserved_prefixes: Vec::new(),
7947 protocol: ModuleProtocol::Subc,
7948 overlap: Default::default(),
7949 })
7950 .unwrap();
7951
7952 let deadline = Instant::now() + Duration::from_secs(5);
7953 loop {
7954 let history = module.terminal_history();
7955 if history.entries.len() == 2 {
7956 assert_eq!(module.status().unwrap().state, ModuleState::Failed);
7957 assert_eq!(history.dropped, 0);
7958 assert_eq!(
7959 history
7960 .entries
7961 .iter()
7962 .map(|entry| entry.exit_code)
7963 .collect::<Vec<_>>(),
7964 vec![Some(23), Some(23)]
7965 );
7966 assert!(history.entries[0].at_ms <= history.entries[1].at_ms);
7967 return;
7968 }
7969 assert!(
7970 Instant::now() < deadline,
7971 "module did not retain two terminal exits: {history:?}"
7972 );
7973 sleep(Duration::from_millis(10)).await;
7974 }
7975 }
7976
7977 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7981 async fn disable_during_crash_backoff_cancels_pending_respawn() {
7982 let backoff = Duration::from_secs(2);
7983 let supervisor = Supervisor::new(
7984 Arc::new(Registry::default()),
7985 RestartPolicy::new(10, backoff),
7986 );
7987 let module = supervisor
7988 .spawn(ModuleSpec {
7989 module_id: "disable-during-backoff".to_string(),
7990 program: fake_aft_stub_path(),
7991 args: Vec::new(),
7992 env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
7993 reserved: false,
7994 reserved_prefixes: Vec::new(),
7995 protocol: ModuleProtocol::Subc,
7996 overlap: Default::default(),
7997 })
7998 .unwrap();
7999
8000 let deadline = Instant::now() + Duration::from_secs(5);
8002 loop {
8003 if module.status().unwrap().state == ModuleState::Restarting {
8004 break;
8005 }
8006 assert!(
8007 Instant::now() < deadline,
8008 "module never entered the crash backoff"
8009 );
8010 sleep(Duration::from_millis(10)).await;
8011 }
8012
8013 let started = Instant::now();
8014 module.set_enabled(false).await.unwrap();
8015 let waited = started.elapsed();
8016
8017 assert!(
8018 waited < backoff / 2,
8019 "disable waited {waited:?} behind the {backoff:?} crash backoff; the operator command must preempt the pending respawn"
8020 );
8021 assert_eq!(module.status().unwrap().state, ModuleState::Disabled);
8022
8023 sleep(backoff + Duration::from_millis(500)).await;
8025 let status = module.status().unwrap();
8026 assert_eq!(status.state, ModuleState::Disabled);
8027 assert_eq!(
8028 status.spawn_generation, 1,
8029 "module respawned after the operator disabled it"
8030 );
8031 }
8032
8033 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
8037 async fn every_restart_increment_path_advances_lifetime_count() {
8038 let supervisor = Supervisor::new(
8039 Arc::new(Registry::default()),
8040 RestartPolicy::new(1, Duration::ZERO),
8041 );
8042 let runtime = supervisor.runtime_config();
8043 let spec = ModuleSpec {
8044 module_id: "lifetime-increment-path".to_string(),
8045 program: PathBuf::from("/unused/lifetime-increment-path"),
8046 args: Vec::new(),
8047 env: Vec::new(),
8048 reserved: false,
8049 reserved_prefixes: Vec::new(),
8050 protocol: ModuleProtocol::Subc,
8051 overlap: Default::default(),
8052 };
8053
8054 let crash_snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8055 assert!(matches!(
8056 on_child_exit(
8057 &spec,
8058 runtime.restart_policy,
8059 &supervisor.registry,
8060 &crash_snapshot,
8061 &runtime.terminal_ring,
8062 &runtime.spawn_events,
8063 &runtime.child_roster,
8064 ExitReport {
8065 kind: ExitKind::Crash,
8066 code: Some(1),
8067 signal: None,
8068 at_ms: 1,
8069 },
8070 )
8071 .await,
8072 NextAction::Restart { schedule: _ }
8073 ));
8074 let (crash_restarts, crash_lifetime) = {
8075 let state = lock_snapshot(&crash_snapshot).unwrap();
8076 (state.crash_restarts.len(), state.lifetime_restarts)
8077 };
8078 assert_eq!(crash_restarts, 1);
8079 assert_eq!(crash_lifetime, 1);
8080
8081 let health_snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8082 let mut health_child = None;
8083 assert!(matches!(
8084 health_restart_child(
8085 &spec,
8086 &runtime,
8087 &supervisor.registry,
8088 &supervisor.process_liveness,
8089 &health_snapshot,
8090 &mut health_child,
8091 SupervisorHealthStatus::Failing,
8092 None,
8093 2,
8094 )
8095 .await,
8096 Err(SuperviseError::Spawn { .. })
8097 ));
8098 let (health_restarts, health_lifetime) = {
8099 let state = lock_snapshot(&health_snapshot).unwrap();
8100 (state.crash_restarts.len(), state.lifetime_restarts)
8101 };
8102 assert_eq!(health_restarts, 1);
8103 assert_eq!(health_lifetime, 1);
8104
8105 let reload_snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8106 let mut reload_child = None;
8107 assert!(matches!(
8108 handle_reload_spawn_failure(
8109 &spec,
8110 &runtime,
8111 &supervisor.process_liveness,
8112 &reload_snapshot,
8113 &mut reload_child,
8114 "forced reload spawn failure".to_string(),
8115 )
8116 .await,
8117 Err(SuperviseError::ReloadFailed { .. })
8118 ));
8119 let (reload_restarts, reload_lifetime) = {
8120 let state = lock_snapshot(&reload_snapshot).unwrap();
8121 (state.crash_restarts.len(), state.lifetime_restarts)
8122 };
8123 assert_eq!(reload_restarts, 1);
8124 assert_eq!(reload_lifetime, 1);
8125 }
8126
8127 #[tokio::test]
8128 async fn deliberately_severed_live_child_records_lifetime_without_spending_restart_budget() {
8129 let supervisor = Supervisor::new(
8130 Arc::new(Registry::default()),
8131 RestartPolicy::new(3, Duration::ZERO),
8132 );
8133 let runtime = supervisor.runtime_config();
8134 let spec = ModuleSpec {
8135 module_id: "deliberately-severed".to_string(),
8136 program: PathBuf::from("/unused/deliberately-severed"),
8137 args: Vec::new(),
8138 env: Vec::new(),
8139 reserved: false,
8140 reserved_prefixes: Vec::new(),
8141 protocol: ModuleProtocol::Subc,
8142 overlap: Default::default(),
8143 };
8144 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8145 let process = ProcessIdentity {
8146 pid: 41,
8147 start_time: 101,
8148 };
8149 record_deliberate_severance(&snapshot, process).unwrap();
8150 let exit_report = apply_deliberate_severance_marker(
8151 &snapshot,
8152 Some(process),
8153 ExitReport {
8154 kind: ExitKind::Crash,
8155 code: Some(1),
8156 signal: None,
8157 at_ms: 1,
8158 },
8159 );
8160 assert_eq!(exit_report.kind, ExitKind::DeliberateSeverance);
8161
8162 assert!(matches!(
8163 on_child_exit(
8164 &spec,
8165 runtime.restart_policy,
8166 &supervisor.registry,
8167 &snapshot,
8168 &runtime.terminal_ring,
8169 &runtime.spawn_events,
8170 &runtime.child_roster,
8171 exit_report,
8172 )
8173 .await,
8174 NextAction::Restart { schedule: _ }
8175 ));
8176 let state = lock_snapshot(&snapshot).unwrap();
8177 assert_eq!(state.lifetime_restarts, 1);
8178 assert_eq!(state.crash_restarts.len(), 0);
8179 }
8180
8181 #[tokio::test]
8182 async fn genuine_crash_spends_restart_budget_and_records_lifetime() {
8183 let supervisor = Supervisor::new(
8184 Arc::new(Registry::default()),
8185 RestartPolicy::new(3, Duration::ZERO),
8186 );
8187 let runtime = supervisor.runtime_config();
8188 let spec = ModuleSpec {
8189 module_id: "genuine-crash".to_string(),
8190 program: PathBuf::from("/unused/genuine-crash"),
8191 args: Vec::new(),
8192 env: Vec::new(),
8193 reserved: false,
8194 reserved_prefixes: Vec::new(),
8195 protocol: ModuleProtocol::Subc,
8196 overlap: Default::default(),
8197 };
8198 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8199
8200 assert!(matches!(
8201 on_child_exit(
8202 &spec,
8203 runtime.restart_policy,
8204 &supervisor.registry,
8205 &snapshot,
8206 &runtime.terminal_ring,
8207 &runtime.spawn_events,
8208 &runtime.child_roster,
8209 ExitReport {
8210 kind: ExitKind::Crash,
8211 code: Some(1),
8212 signal: None,
8213 at_ms: 1,
8214 },
8215 )
8216 .await,
8217 NextAction::Restart { schedule: _ }
8218 ));
8219 let state = lock_snapshot(&snapshot).unwrap();
8220 assert_eq!(state.lifetime_restarts, 1);
8221 assert_eq!(state.crash_restarts.len(), 1);
8222 }
8223
8224 fn crash_exit_report(at_ms: u64) -> ExitReport {
8225 ExitReport {
8226 kind: ExitKind::Crash,
8227 code: Some(1),
8228 signal: None,
8229 at_ms,
8230 }
8231 }
8232
8233 fn windowed_crash_spec(module_id: &str) -> ModuleSpec {
8234 ModuleSpec {
8235 module_id: module_id.to_string(),
8236 program: PathBuf::from("/unused").join(module_id),
8237 args: Vec::new(),
8238 env: Vec::new(),
8239 reserved: false,
8240 reserved_prefixes: Vec::new(),
8241 protocol: ModuleProtocol::Subc,
8242 overlap: Default::default(),
8243 }
8244 }
8245
8246 #[tokio::test]
8252 async fn three_crashes_inside_the_window_stop_the_module_and_name_the_window() {
8253 let (logs, _guard) = crate::router::test_log::log_capture(tracing::Level::ERROR);
8254 let supervisor = Supervisor::new(
8255 Arc::new(Registry::default()),
8256 RestartPolicy::new(2, Duration::ZERO),
8257 );
8258 let runtime = supervisor.runtime_config();
8259 let spec = windowed_crash_spec("crash-loop-in-window");
8260 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8261
8262 for attempt in 1..=2 {
8263 assert!(
8264 matches!(
8265 on_child_exit(
8266 &spec,
8267 runtime.restart_policy,
8268 &supervisor.registry,
8269 &snapshot,
8270 &runtime.terminal_ring,
8271 &runtime.spawn_events,
8272 &runtime.child_roster,
8273 crash_exit_report(attempt),
8274 )
8275 .await,
8276 NextAction::Restart { schedule: _ }
8277 ),
8278 "crash {attempt} is inside the budget and must respawn"
8279 );
8280 }
8281
8282 assert!(matches!(
8283 on_child_exit(
8284 &spec,
8285 runtime.restart_policy,
8286 &supervisor.registry,
8287 &snapshot,
8288 &runtime.terminal_ring,
8289 &runtime.spawn_events,
8290 &runtime.child_roster,
8291 crash_exit_report(3),
8292 )
8293 .await,
8294 NextAction::Stop { .. }
8295 ));
8296
8297 {
8298 let state = lock_snapshot(&snapshot).unwrap();
8299 assert_eq!(state.state, ModuleState::Failed);
8300 assert_eq!(state.crash_restarts.len(), 2);
8301 assert_eq!(state.lifetime_restarts, 2);
8302 }
8303
8304 let history = runtime
8305 .terminal_ring
8306 .lock()
8307 .expect("terminal ring is not poisoned")
8308 .snapshot();
8309 let last = history
8310 .entries
8311 .last()
8312 .expect("the refused crash is retained");
8313 assert_eq!(last.disposition, TerminalDisposition::Failed);
8314 assert_eq!(
8315 last.disposition_detail.as_deref(),
8316 Some("crash budget exhausted: max_restarts=2 within window_secs=600")
8317 );
8318
8319 let captured = crate::router::test_log::captured_logs(&logs);
8320 assert!(
8321 captured.contains("crash budget exhausted: max_restarts=2 within window_secs=600"),
8322 "the stop must be logged with its window: {captured}"
8323 );
8324 }
8325
8326 #[tokio::test]
8334 async fn a_crash_older_than_the_window_frees_its_slot_for_a_later_crash() {
8335 let supervisor = Supervisor::new(
8336 Arc::new(Registry::default()),
8337 RestartPolicy::new(2, Duration::ZERO),
8338 );
8339 let runtime = supervisor.runtime_config();
8340 let spec = windowed_crash_spec("crash-across-windows");
8341 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8342
8343 for attempt in 1..=2 {
8344 assert!(matches!(
8345 on_child_exit(
8346 &spec,
8347 runtime.restart_policy,
8348 &supervisor.registry,
8349 &snapshot,
8350 &runtime.terminal_ring,
8351 &runtime.spawn_events,
8352 &runtime.child_roster,
8353 crash_exit_report(attempt),
8354 )
8355 .await,
8356 NextAction::Restart { schedule: _ }
8357 ));
8358 }
8359
8360 update_snapshot(&snapshot, Some(&spec.module_id), |state| {
8363 age_oldest_crash_restart_out_of_window(state, runtime.restart_policy.window);
8364 })
8365 .unwrap();
8366
8367 assert!(
8368 matches!(
8369 on_child_exit(
8370 &spec,
8371 runtime.restart_policy,
8372 &supervisor.registry,
8373 &snapshot,
8374 &runtime.terminal_ring,
8375 &runtime.spawn_events,
8376 &runtime.child_roster,
8377 crash_exit_report(3),
8378 )
8379 .await,
8380 NextAction::Restart { schedule: _ }
8381 ),
8382 "a crash older than the window must not hold a budget slot"
8383 );
8384
8385 let state = lock_snapshot(&snapshot).unwrap();
8386 assert_eq!(state.state, ModuleState::Restarting);
8387 assert_eq!(
8388 state.crash_restarts.len(),
8389 2,
8390 "the aged instant is dropped and the new one takes its place"
8391 );
8392 assert_eq!(
8393 state.lifetime_restarts, 3,
8394 "the ledger counts every restart, including the ones the window forgot"
8395 );
8396 }
8397
8398 #[tokio::test]
8403 async fn an_operator_restart_clears_the_ring_and_leaves_the_ledger_alone() {
8404 let supervisor = Supervisor::new(
8405 Arc::new(Registry::default()),
8406 RestartPolicy::new(2, Duration::ZERO),
8407 );
8408 let runtime = supervisor.runtime_config();
8409 let spec = windowed_crash_spec("operator-cleared-budget");
8410 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8411
8412 for attempt in 1..=2 {
8413 assert!(matches!(
8414 on_child_exit(
8415 &spec,
8416 runtime.restart_policy,
8417 &supervisor.registry,
8418 &snapshot,
8419 &runtime.terminal_ring,
8420 &runtime.spawn_events,
8421 &runtime.child_roster,
8422 crash_exit_report(attempt),
8423 )
8424 .await,
8425 NextAction::Restart { schedule: _ }
8426 ));
8427 }
8428
8429 reset_restart_count(&snapshot, &spec.module_id).unwrap();
8430 {
8431 let state = lock_snapshot(&snapshot).unwrap();
8432 assert!(
8433 state.crash_restarts.is_empty(),
8434 "an operator restart returns the full budget"
8435 );
8436 assert_eq!(
8437 state.lifetime_restarts, 2,
8438 "clearing the budget must not unmake the crashes"
8439 );
8440 }
8441
8442 assert!(
8443 matches!(
8444 on_child_exit(
8445 &spec,
8446 runtime.restart_policy,
8447 &supervisor.registry,
8448 &snapshot,
8449 &runtime.terminal_ring,
8450 &runtime.spawn_events,
8451 &runtime.child_roster,
8452 crash_exit_report(3),
8453 )
8454 .await,
8455 NextAction::Restart { schedule: _ }
8456 ),
8457 "the cleared budget must be spendable again"
8458 );
8459 let state = lock_snapshot(&snapshot).unwrap();
8460 assert_eq!(state.crash_restarts.len(), 1);
8461 assert_eq!(state.lifetime_restarts, 3);
8462 }
8463
8464 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
8465 async fn severance_marker_for_a_dead_child_does_not_label_its_successor() {
8466 let severed = ProcessIdentity {
8467 pid: 41,
8468 start_time: 101,
8469 };
8470 let successor = ProcessIdentity {
8471 pid: 41,
8472 start_time: 202,
8473 };
8474 let module = module_with_recovery_snapshot(ModuleState::Running, true, 0);
8475 update_snapshot(&module.inner.snapshot, Some("recovery-snapshot"), |state| {
8476 state.pid = Some(successor.pid);
8477 state.process_start_time = Some(successor.start_time);
8478 })
8479 .unwrap();
8480 assert!(!module.record_deliberate_severance(severed).unwrap());
8481
8482 let exit_report = apply_deliberate_severance_marker(
8483 &module.inner.snapshot,
8484 Some(successor),
8485 ExitReport {
8486 kind: ExitKind::Crash,
8487 code: Some(1),
8488 signal: None,
8489 at_ms: 1,
8490 },
8491 );
8492
8493 assert_eq!(exit_report.kind, ExitKind::Crash);
8494 }
8495
8496 #[tokio::test]
8497 async fn drain_reap_marks_deliberate_severance_and_records_lifetime_without_budget() {
8498 let registry = Registry::default();
8499 let supervisor = Supervisor::new(
8500 Arc::new(Registry::default()),
8501 RestartPolicy::new(3, Duration::ZERO),
8502 );
8503 let runtime = supervisor.runtime_config();
8504 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8505 let spec = ModuleSpec {
8506 module_id: "drain-deliberate-severance".to_string(),
8507 program: fake_aft_stub_path(),
8508 args: Vec::new(),
8509 env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
8510 reserved: false,
8511 reserved_prefixes: Vec::new(),
8512 protocol: ModuleProtocol::Subc,
8513 overlap: Default::default(),
8514 };
8515 let mut child = spawn_and_mark_running(&spec, &runtime, &snapshot).unwrap();
8516 let process = ProcessIdentity {
8517 pid: 41,
8518 start_time: 101,
8519 };
8520 child.process_identity = Some(process);
8521 update_snapshot(&snapshot, Some(&spec.module_id), |state| {
8522 state.pid = Some(process.pid);
8523 state.process_start_time = Some(process.start_time);
8524 })
8525 .unwrap();
8526 record_deliberate_severance(&snapshot, process).unwrap();
8527
8528 drain_child_to_state(
8529 &spec.module_id,
8530 spec.protocol,
8531 StopNotice::SentOverConnection,
8534 ®istry,
8535 &snapshot,
8536 &runtime.terminal_ring,
8537 &runtime.spawn_events,
8538 child,
8539 Duration::from_secs(1),
8540 ModuleState::Stopped,
8541 Some(false),
8542 )
8543 .await
8544 .unwrap();
8545
8546 let state = lock_snapshot(&snapshot).unwrap();
8547 assert_eq!(
8548 state.last_exit.as_ref().map(|exit| exit.kind),
8549 Some(ExitKind::DeliberateSeverance)
8550 );
8551 assert_eq!(state.lifetime_restarts, 1);
8552 assert_eq!(state.crash_restarts.len(), 0);
8553 drop(state);
8554 let history = runtime.terminal_ring.lock().unwrap().snapshot();
8555 assert_eq!(
8556 history.entries[0].exit_kind,
8557 subc_control::TerminalExitKind::DeliberateSeverance
8558 );
8559 }
8560
8561 #[tokio::test]
8562 async fn ordinary_drain_reap_does_not_record_a_lifetime_restart() {
8563 let registry = Registry::default();
8564 let supervisor = Supervisor::new(
8565 Arc::new(Registry::default()),
8566 RestartPolicy::new(3, Duration::ZERO),
8567 );
8568 let runtime = supervisor.runtime_config();
8569 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8570 let spec = ModuleSpec {
8571 module_id: "ordinary-drain".to_string(),
8572 program: fake_aft_stub_path(),
8573 args: Vec::new(),
8574 env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
8575 reserved: false,
8576 reserved_prefixes: Vec::new(),
8577 protocol: ModuleProtocol::Subc,
8578 overlap: Default::default(),
8579 };
8580 let child = spawn_and_mark_running(&spec, &runtime, &snapshot).unwrap();
8581
8582 drain_child_to_state(
8583 &spec.module_id,
8584 spec.protocol,
8585 StopNotice::SentOverConnection,
8588 ®istry,
8589 &snapshot,
8590 &runtime.terminal_ring,
8591 &runtime.spawn_events,
8592 child,
8593 Duration::from_secs(1),
8594 ModuleState::Stopped,
8595 Some(false),
8596 )
8597 .await
8598 .unwrap();
8599
8600 let state = lock_snapshot(&snapshot).unwrap();
8601 assert_eq!(
8602 state.last_exit.as_ref().map(|exit| exit.kind),
8603 Some(ExitKind::Crash)
8604 );
8605 assert_eq!(state.lifetime_restarts, 0);
8606 assert_eq!(state.crash_restarts.len(), 0);
8607 }
8608
8609 #[test]
8610 fn fatal_connection_teardown_cannot_arm_a_marker_for_a_surviving_process() {
8611 assert!(!include_str!("server.rs")
8617 .contains("router.record_deliberate_connection_severance(ctx.connection_id)"));
8618 }
8619
8620 #[test]
8627 fn drained_after_quiescence_wait_passes_ok_through_and_forces_false_on_err() {
8628 assert!(drained_after_quiescence_wait(&Ok(true)));
8629 assert!(!drained_after_quiescence_wait(&Ok(false)));
8630 assert!(!drained_after_quiescence_wait(&Err(
8631 SuperviseError::StatePoisoned { module_id: None }
8632 )));
8633 }
8634
8635 #[test]
8644 fn wait_error_exit_report_records_a_failed_terminal_with_no_code_or_signal() {
8645 let ring = Arc::new(Mutex::new(TerminalRing::new(
8646 TerminalRingConfig::default(),
8647 0,
8648 )));
8649 record_wait_error_terminal("wait-error", &ring, &super::SpawnEventFeed::default());
8650
8651 let snapshot = ring.lock().unwrap().snapshot();
8652 assert_eq!(snapshot.entries.len(), 1);
8653 let entry = &snapshot.entries[0];
8654 assert_eq!(entry.exit_code, None);
8655 assert_eq!(entry.exit_signal, None);
8656 assert_eq!(entry.disposition, TerminalDisposition::Failed);
8657 }
8658
8659 #[test]
8660 fn wait_error_exit_path_preserves_spawn_event_density() {
8661 let feed = super::SpawnEventFeed::default();
8662 feed.configure_incarnation("wait-error-density".to_string());
8663 feed.emit_spawned("wait-error", 41, 1);
8664 let ring = Arc::new(Mutex::new(TerminalRing::new(
8665 TerminalRingConfig::default(),
8666 0,
8667 )));
8668
8669 record_wait_error_terminal("wait-error", &ring, &feed);
8670 feed.emit_spawned("after-wait-error", 42, 2);
8671
8672 let state = feed.0.lock().unwrap();
8673 let sequences = state
8674 .events
8675 .iter()
8676 .map(|event| event.cursor.seq)
8677 .collect::<Vec<_>>();
8678 assert_eq!(sequences, vec![1, 2, 3]);
8679 assert_eq!(state.events[1].kind, SpawnEventKind::Exited);
8680 assert_eq!(state.events[1].exit_code, None);
8681 assert_eq!(state.events[1].exit_signal, None);
8682 }
8683
8684 #[test]
8688 fn wait_error_exit_report_is_classified_as_a_crash() {
8689 assert_eq!(wait_error_exit_report().kind, ExitKind::Crash);
8690 }
8691}
8692
8693#[cfg(test)]
8694mod health_evidence_tests {
8695 use super::{HealthProbeError, HealthProbeEvidence};
8696 use std::collections::HashSet;
8697
8698 #[test]
8706 fn only_a_dead_lane_is_proof_of_death() {
8707 assert!(HealthProbeError::lane_dead("gone").is_proof_of_death());
8708 assert!(!HealthProbeError::no_answer("timed out").is_proof_of_death());
8712 assert!(!HealthProbeError::bad_answer("garbage").is_proof_of_death());
8713 assert!(!HealthProbeError::misconfigured("no table").is_proof_of_death());
8714 }
8715
8716 #[test]
8722 fn every_evidence_class_has_a_distinct_label() {
8723 let labels = [
8724 HealthProbeError::lane_dead("").label(),
8725 HealthProbeError::no_answer("").label(),
8726 HealthProbeError::bad_answer("").label(),
8727 HealthProbeError::misconfigured("").label(),
8728 ];
8729 let unique: HashSet<_> = labels.iter().collect();
8730 assert_eq!(unique.len(), labels.len(), "labels collided: {labels:?}");
8731 }
8732
8733 #[test]
8739 fn classification_preserves_the_original_message() {
8740 let err = HealthProbeError::no_answer("module did not answer within 5s");
8741 assert_eq!(err.to_string(), "module did not answer within 5s");
8742 assert!(matches!(err.evidence, HealthProbeEvidence::NoAnswer));
8743 }
8744}
8745
8746#[cfg(test)]
8747mod health_tombstone_tests {
8748 use std::{path::PathBuf, sync::Arc, time::Duration};
8749
8750 use subc_protocol::{
8751 manifest::Concurrency,
8752 session::{HealthStatus, ModuleControlResponse},
8753 };
8754 use tokio::sync::mpsc;
8755
8756 use super::{
8757 probe_module_health, HealthAction, HealthConfig, HealthProbeEvidence, ModuleProtocol,
8758 ModuleSpec, RestartPolicy, Supervisor, SupervisorRuntimeConfig,
8759 };
8760 use crate::{
8761 control::ControlHandler,
8762 forwarding::{ForwardingTable, ModuleControlRpcCompletion, ModuleControlRpcOutcome},
8763 registry::{ConnectionId, Registry},
8764 router::FrameSink,
8765 };
8766
8767 struct ProbeHarness {
8768 spec: ModuleSpec,
8769 runtime: SupervisorRuntimeConfig,
8770 forwarding: Arc<ForwardingTable>,
8771 module_connection: ConnectionId,
8772 module_rx: mpsc::Receiver<crate::router::OutboundFrame>,
8773 handler: ControlHandler,
8774 module: super::SupervisedModule,
8775 }
8776
8777 fn probe_harness() -> ProbeHarness {
8778 let registry = Arc::new(Registry::default());
8779 let forwarding = Arc::new(ForwardingTable::default());
8780 let supervisor_handle = super::SupervisorHandle::new();
8781 let health = HealthConfig {
8782 cadence: Duration::from_secs(30),
8783 deadline: Duration::from_secs(5),
8784 failure_threshold: 3,
8785 on_degraded: HealthAction::Report,
8786 on_failing: HealthAction::Report,
8787 critical: false,
8788 };
8789 let supervisor = Supervisor::new(Arc::clone(®istry), RestartPolicy::default())
8790 .with_forwarding(Arc::clone(&forwarding))
8791 .with_handle(supervisor_handle.clone())
8792 .with_health_config(health);
8793 let spec = ModuleSpec {
8794 module_id: "late-health-module".to_string(),
8795 program: PathBuf::from("disabled-module"),
8796 args: Vec::new(),
8797 env: Vec::new(),
8798 reserved: false,
8799 reserved_prefixes: Vec::new(),
8800 protocol: ModuleProtocol::Subc,
8801 overlap: Default::default(),
8802 };
8803 let module = supervisor
8804 .supervise_configured(spec.clone(), false)
8805 .unwrap();
8806 let runtime = supervisor.runtime_config();
8807 let handler = ControlHandler::with_forwarding(registry, Arc::clone(&forwarding))
8808 .with_supervisor(supervisor_handle);
8809 let module_connection = ConnectionId::new(700);
8810 let (module_tx, module_rx) = mpsc::channel(8);
8811 forwarding
8812 .register_module_connection(
8813 module_connection,
8814 spec.module_id.clone(),
8815 subc_protocol::PROTOCOL_VERSION,
8816 Concurrency::ModuleManaged,
8817 FrameSink::new(module_tx),
8818 )
8819 .unwrap();
8820
8821 ProbeHarness {
8822 spec,
8823 runtime,
8824 forwarding,
8825 module_connection,
8826 module_rx,
8827 handler,
8828 module,
8829 }
8830 }
8831
8832 async fn finish_after(
8833 harness: &mut ProbeHarness,
8834 stall: Duration,
8835 ) -> ModuleControlRpcCompletion {
8836 assert!(stall > harness.runtime.health.deadline);
8837 let deadline = harness.runtime.health.deadline;
8838 let probe = probe_module_health(&harness.spec.module_id, &harness.runtime, None);
8839 let answer = async {
8840 let frame = harness.module_rx.recv().await.expect("health.check frame");
8841 tokio::time::advance(deadline).await;
8842 tokio::task::yield_now().await;
8843 tokio::time::advance(stall - deadline).await;
8844 harness
8845 .forwarding
8846 .complete_module_control_rpc(
8847 harness.module_connection,
8848 frame.header.corr,
8849 Some("health.check"),
8850 ModuleControlRpcOutcome::Response(ModuleControlResponse::HealthCheck {
8851 status: HealthStatus::Ok,
8852 detail: None,
8853 metrics: None,
8854 }),
8855 )
8856 .unwrap()
8857 };
8858 let (probe_result, completion) = tokio::join!(probe, answer);
8859 let err = probe_result.expect_err("probe must miss its deadline");
8860 assert!(matches!(err.evidence, HealthProbeEvidence::NoAnswer));
8861 completion
8862 }
8863
8864 async fn time_out_without_answer(harness: &mut ProbeHarness) {
8865 let deadline = harness.runtime.health.deadline;
8866 let probe = probe_module_health(&harness.spec.module_id, &harness.runtime, None);
8867 let exhaust_deadline = async {
8868 let _frame = harness.module_rx.recv().await.expect("health.check frame");
8869 tokio::time::advance(deadline).await;
8870 tokio::task::yield_now().await;
8871 };
8872 let (probe_result, ()) = tokio::join!(probe, exhaust_deadline);
8873 let err = probe_result.expect_err("probe must miss its deadline");
8874 assert!(matches!(err.evidence, HealthProbeEvidence::NoAnswer));
8875 }
8876
8877 #[tokio::test(start_paused = true)]
8878 async fn late_health_answers_record_start_anchored_latency_for_two_stalls() {
8879 let mut harness = probe_harness();
8880
8881 let first = finish_after(&mut harness, Duration::from_secs(8)).await;
8882 let first_latency = match &first {
8883 ModuleControlRpcCompletion::LateHealthAnswer { latency, .. } => *latency,
8884 other => panic!("late answer was not retained: {other:?}"),
8885 };
8886 assert!(harness.handler.observe_module_control_completion(first));
8887
8888 let second = finish_after(&mut harness, Duration::from_secs(11)).await;
8889 let second_latency = match &second {
8890 ModuleControlRpcCompletion::LateHealthAnswer { latency, .. } => *latency,
8891 other => panic!("late answer was not retained: {other:?}"),
8892 };
8893 assert!(harness.handler.observe_module_control_completion(second));
8894
8895 assert_eq!(first_latency, Duration::from_secs(8));
8896 assert_eq!(
8897 second_latency - first_latency,
8898 Duration::from_secs(3),
8899 "latency must grow linearly with the additional stall"
8900 );
8901 let health = harness.module.status().unwrap().health;
8902 assert_eq!(health.late_answer_count, 2);
8903 assert_eq!(health.last_late_answer_latency_ms, Some(11_000));
8904 }
8905
8906 #[tokio::test(start_paused = true)]
8914 async fn late_answer_clears_the_consecutive_failure_streak() {
8915 let mut harness = probe_harness();
8916
8917 time_out_without_answer(&mut harness).await;
8919 harness
8920 .module
8921 .record_health_probe_failure_for_test("[no-answer] test miss")
8922 .unwrap();
8923 assert_eq!(
8924 harness.module.status().unwrap().health.consecutive_failures,
8925 1,
8926 "precondition: the miss must be on the streak before the late answer"
8927 );
8928
8929 let late = finish_after(&mut harness, Duration::from_secs(9)).await;
8931 assert!(matches!(
8932 late,
8933 ModuleControlRpcCompletion::LateHealthAnswer { .. }
8934 ));
8935 assert!(harness.handler.observe_module_control_completion(late));
8936
8937 let health = harness.module.status().unwrap().health;
8938 assert_eq!(
8939 health.consecutive_failures, 0,
8940 "a late answer is an answer: the streak must reset"
8941 );
8942 assert_eq!(health.late_answer_count, 1);
8943 }
8944
8945 #[tokio::test(start_paused = true)]
8946 async fn repeated_serial_probe_cycles_keep_one_tombstone_per_endpoint() {
8947 let mut harness = probe_harness();
8948
8949 for _ in 0..20 {
8950 time_out_without_answer(&mut harness).await;
8951 assert_eq!(
8952 harness.forwarding.health_probe_tombstone_count().unwrap(),
8953 1
8954 );
8955 }
8956 }
8957}
8958
8959#[cfg(test)]
8960mod child_env_tests {
8961 use super::{
8962 apply_child_env, apply_spawn_role, apply_wire_spawn_args, ModuleProtocol, ModuleSpec,
8963 SpawnRole, SupervisorHandle, SPAWN_ROLE_SWAP_CANDIDATE, SUBC_ARG, SUBC_LAUNCH_NONCE_ENV,
8964 SUBC_MODULE_ID_ENV, SUBC_SPAWN_ROLE_ENV,
8965 };
8966 use std::{ffi::OsStr, path::PathBuf};
8967 use tokio::process::Command;
8968
8969 fn spec(env: Vec<(String, String)>) -> ModuleSpec {
8970 ModuleSpec {
8971 module_id: "env-plan".to_string(),
8972 program: PathBuf::from("/nonexistent"),
8973 args: Vec::new(),
8974 env,
8975 reserved: false,
8976 reserved_prefixes: Vec::new(),
8977 protocol: ModuleProtocol::Subc,
8978 overlap: Default::default(),
8979 }
8980 }
8981
8982 #[test]
8996 fn ambient_ck_log_is_removed_and_a_configured_one_survives() {
8997 let mut command = Command::new("/nonexistent");
8998 apply_child_env(&mut command, &spec(Vec::new()));
8999 let removed = command
9000 .as_std()
9001 .get_envs()
9002 .any(|(key, value)| key == OsStr::new("CK_LOG") && value.is_none());
9003 assert!(
9004 removed,
9005 "ambient CK_LOG must be explicitly removed for an unconfigured module"
9006 );
9007
9008 let mut configured = Command::new("/nonexistent");
9009 apply_child_env(
9010 &mut configured,
9011 &spec(vec![("CK_LOG".to_string(), "debug".to_string())]),
9012 );
9013 let effective = configured
9014 .as_std()
9015 .get_envs()
9016 .filter(|(key, _)| *key == OsStr::new("CK_LOG"))
9017 .last()
9018 .map(|(_, value)| value.map(|v| v.to_string_lossy().into_owned()));
9019 assert_eq!(
9020 effective,
9021 Some(Some("debug".to_string())),
9022 "a module's configured CK_LOG must survive the ambient removal"
9023 );
9024 }
9025
9026 #[test]
9035 fn protocol_none_spawn_carries_no_subc_argument_and_no_nonce() {
9036 let connection_file = std::path::Path::new("/run/subc-connection.json");
9037 let handle = SupervisorHandle::new();
9038
9039 let mut none_spec = spec(Vec::new());
9040 none_spec.protocol = ModuleProtocol::None;
9041 let mut none = Command::new("/nonexistent");
9042 apply_wire_spawn_args(&mut none, &none_spec, Some(connection_file), Some(&handle))
9043 .expect("protocol-none spawn args apply");
9044 let none_args: Vec<String> = none
9045 .as_std()
9046 .get_args()
9047 .map(|a| a.to_string_lossy().into_owned())
9048 .collect();
9049 assert!(
9050 !none_args.iter().any(|a| a == SUBC_ARG),
9051 "protocol:none argv must not carry --subc; got {none_args:?}"
9052 );
9053 let none_has_nonce = none
9054 .as_std()
9055 .get_envs()
9056 .any(|(key, value)| key == OsStr::new(SUBC_LAUNCH_NONCE_ENV) && value.is_some());
9057 assert!(
9058 !none_has_nonce,
9059 "protocol:none spawn must not receive a launch nonce"
9060 );
9061 let none_has_module_id = none
9062 .as_std()
9063 .get_envs()
9064 .any(|(key, value)| key == OsStr::new(SUBC_MODULE_ID_ENV) && value.is_some());
9065 assert!(
9066 none_has_module_id,
9067 "SUBC_MODULE_ID is inert and stays on every path"
9068 );
9069 assert!(
9070 handle.spawn_nonce(&none_spec.module_id).is_none(),
9071 "no nonce record for a process that will never present one"
9072 );
9073
9074 let wire_spec = spec(Vec::new());
9076 let mut wire = Command::new("/nonexistent");
9077 apply_wire_spawn_args(&mut wire, &wire_spec, Some(connection_file), Some(&handle))
9078 .expect("subc-wire spawn args apply");
9079 let wire_args: Vec<String> = wire
9080 .as_std()
9081 .get_args()
9082 .map(|a| a.to_string_lossy().into_owned())
9083 .collect();
9084 assert_eq!(
9085 wire_args,
9086 vec![
9087 SUBC_ARG.to_string(),
9088 connection_file.to_string_lossy().into_owned()
9089 ],
9090 "a subc-wire spawn still carries --subc <path>"
9091 );
9092 assert!(wire
9093 .as_std()
9094 .get_envs()
9095 .any(|(key, value)| key == OsStr::new(SUBC_LAUNCH_NONCE_ENV) && value.is_some()));
9096 assert!(handle.spawn_nonce(&wire_spec.module_id).is_some());
9097 }
9098
9099 #[test]
9109 fn plain_spawn_removes_the_spawn_role_even_when_the_spec_sets_it() {
9110 let role = |command: &Command| {
9111 command
9112 .as_std()
9113 .get_envs()
9114 .filter(|(key, _)| *key == OsStr::new(SUBC_SPAWN_ROLE_ENV))
9115 .last()
9116 .map(|(_, value)| value.map(|v| v.to_string_lossy().into_owned()))
9117 };
9118 let forged = spec(vec![(
9119 SUBC_SPAWN_ROLE_ENV.to_string(),
9120 SPAWN_ROLE_SWAP_CANDIDATE.to_string(),
9121 )]);
9122
9123 let mut plain = Command::new("/nonexistent");
9124 apply_child_env(&mut plain, &forged);
9125 apply_spawn_role(&mut plain, SpawnRole::Plain);
9126 assert_eq!(
9127 role(&plain),
9128 Some(None),
9129 "a plain spawn must remove SUBC_SPAWN_ROLE, whatever the spec says"
9130 );
9131
9132 let mut candidate = Command::new("/nonexistent");
9133 apply_child_env(&mut candidate, &spec(Vec::new()));
9134 apply_spawn_role(&mut candidate, SpawnRole::SwapCandidate);
9135 assert_eq!(
9136 role(&candidate),
9137 Some(Some(SPAWN_ROLE_SWAP_CANDIDATE.to_string()))
9138 );
9139 }
9140
9141 #[test]
9147 fn daemon_private_capture_keys_are_not_passed_to_the_child() {
9148 let mut command = Command::new("/nonexistent");
9149 apply_child_env(
9150 &mut command,
9151 &spec(vec![
9152 (super::CAPTURE_KEEP_ENV.to_string(), "5".to_string()),
9153 ("KEPT".to_string(), "yes".to_string()),
9154 ]),
9155 );
9156 let keys: Vec<String> = command
9157 .as_std()
9158 .get_envs()
9159 .filter(|(_, value)| value.is_some())
9160 .map(|(key, _)| key.to_string_lossy().into_owned())
9161 .collect();
9162 assert!(keys.contains(&"KEPT".to_string()), "got {keys:?}");
9163 assert!(
9164 !keys.contains(&super::CAPTURE_KEEP_ENV.to_string()),
9165 "daemon-private capture key leaked to the child: {keys:?}"
9166 );
9167 }
9168}
9169
9170#[cfg(test)]
9171mod jitter_tests {
9172 use super::jittered_health_delay;
9173 use std::{collections::HashSet, time::Duration};
9174
9175 const FLEET: [&str; 14] = [
9184 "aft",
9185 "alfonso-core",
9186 "magic-context",
9187 "broca",
9188 "thalamus",
9189 "quota",
9190 "engram",
9191 "plexus",
9192 "cerebellum",
9193 "astrocyte",
9194 "synapse",
9195 "subc-mcp",
9196 "cortexkit-credentials",
9197 "subc-federation",
9198 ];
9199
9200 #[test]
9208 fn probe_delays_disperse_across_the_fleet() {
9209 let cadence = Duration::from_secs(30);
9210 let delays: HashSet<Duration> = FLEET
9211 .iter()
9212 .map(|id| jittered_health_delay(id, 0, cadence))
9213 .collect();
9214 assert_eq!(
9215 delays.len(),
9216 FLEET.len(),
9217 "every supervised module must land on its own probe offset"
9218 );
9219 }
9220
9221 #[test]
9227 fn jitter_only_delays_and_stays_within_one_tenth_of_cadence() {
9228 let cadence = Duration::from_secs(30);
9229 let span = cadence / 10;
9230 for id in FLEET {
9231 for probe_index in 0..8 {
9232 let delay = jittered_health_delay(id, probe_index, cadence);
9233 assert!(
9234 delay >= cadence,
9235 "{id}#{probe_index}: jitter must not shorten the cadence"
9236 );
9237 assert!(
9238 delay < cadence + span,
9239 "{id}#{probe_index}: jitter must stay inside one tenth of the cadence"
9240 );
9241 }
9242 }
9243 }
9244
9245 #[test]
9251 fn a_module_offset_is_stable_across_restarts() {
9252 let cadence = Duration::from_secs(30);
9253 for id in FLEET {
9254 assert_eq!(
9255 jittered_health_delay(id, 0, cadence),
9256 jittered_health_delay(id, 0, cadence),
9257 "{id}: the same module and probe index must produce the same offset"
9258 );
9259 }
9260 }
9261
9262 #[test]
9264 fn zero_cadence_yields_zero_delay() {
9265 assert_eq!(
9266 jittered_health_delay("aft", 0, Duration::ZERO),
9267 Duration::ZERO
9268 );
9269 }
9270}
9271
9272#[cfg(all(test, target_os = "linux"))]
9273mod cgroup_placement_tests {
9274 use super::{
9275 apply_cgroup_placement, remove_module_cgroup, ModuleProtocol, ModuleSpec, SuperviseError,
9276 SupervisedChild,
9277 };
9278 use crate::stderr_tail::{StderrRing, StderrTailConfig};
9279 use std::{
9280 fs, io,
9281 path::{Path, PathBuf},
9282 sync::{Arc, Mutex},
9283 };
9284 use subc_test_support::TestTempDir;
9285 use tokio::process::Command;
9286
9287 #[test]
9288 fn failed_parent_cgroup_open_is_a_cgroup_supervision_error() {
9289 let path = Path::new("/definitely-missing-subc-cgroup");
9290 let mut command = Command::new("true");
9291 let error = apply_cgroup_placement(
9292 &mut command,
9293 &ModuleSpec {
9294 module_id: "broken-cgroup".to_string(),
9295 program: PathBuf::from("true"),
9296 args: Vec::new(),
9297 env: Vec::new(),
9298 reserved: false,
9299 reserved_prefixes: Vec::new(),
9300 protocol: ModuleProtocol::Subc,
9301 overlap: Default::default(),
9302 },
9303 path,
9304 )
9305 .expect_err("a parent cgroup open failure must reject the supervised spawn");
9306 let reason = error.to_string();
9307
9308 assert!(
9309 matches!(error, SuperviseError::Cgroup { .. }),
9310 "parent cgroup open must be reported as a cgroup supervision error: {reason}"
9311 );
9312 assert!(
9313 reason.contains("/definitely-missing-subc-cgroup/cgroup.procs"),
9314 "parent cgroup open failure must name cgroup.procs: {reason}"
9315 );
9316 }
9317
9318 #[tokio::test]
9319 async fn reaping_a_child_removes_its_empty_module_cgroup() {
9320 let root = TestTempDir::new("supervisor-reap-cgroup");
9321 fs::write(root.join("cgroup.procs"), b"").expect("write scratch cgroup marker");
9322 let placement = subc_cgroup::prepare_at(&root)
9323 .expect("prepare scratch cgroup root")
9324 .expect("scratch root has a cgroup.procs marker");
9325 let module_id = "reaped-module";
9326 let module = placement
9327 .module_path(module_id)
9328 .expect("create scratch module cgroup");
9329 let child = Command::new("true")
9330 .spawn()
9331 .expect("spawn short-lived child");
9332 let pid = child.id().expect("spawned child has pid");
9333 let mut child = SupervisedChild {
9334 child,
9335 module_id: module_id.to_string(),
9336 cgroup_placement: Some(placement),
9337 stdout_pump: None,
9338 stderr_pump: None,
9339 stderr_ring: Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default()))),
9340 spawned_at_ms: 0,
9341 spawned_from: PathBuf::from("true"),
9342 spawned_file_identity: None,
9343 process_start_time: None,
9344 process_identity: None,
9345 pid,
9346 roster_guard: None,
9347 };
9348
9349 child.wait().await.expect("reap short-lived child");
9350
9351 assert!(
9352 !module.exists(),
9353 "reaping the supervised child must remove its empty cgroup"
9354 );
9355 }
9356
9357 #[test]
9358 fn non_empty_cgroup_removal_is_reported_without_blocking_teardown() {
9359 let root = TestTempDir::new("supervisor-non-empty-cgroup");
9360 fs::write(root.join("cgroup.procs"), b"").expect("write scratch cgroup marker");
9361 let placement = subc_cgroup::prepare_at(&root)
9362 .expect("prepare scratch cgroup root")
9363 .expect("scratch root has a cgroup.procs marker");
9364 let module = placement
9365 .module_path("surviving-module")
9366 .expect("create scratch module cgroup");
9367 fs::write(module.join("surviving-process"), b"still present")
9368 .expect("make scratch cgroup non-empty");
9369 let (logs, _guard) = crate::router::test_log::log_capture(tracing::Level::WARN);
9370
9371 remove_module_cgroup(&placement, "surviving-module");
9372
9373 let logs = crate::router::test_log::captured_logs(&logs);
9374 assert!(
9375 module.exists(),
9376 "failed removal must leave the cgroup intact"
9377 );
9378 assert!(
9379 logs.contains("could not remove module cgroup after process exit; continuing teardown")
9380 && logs.contains("surviving-module"),
9381 "best-effort removal must report the failure without returning it: {logs}"
9382 );
9383 }
9384
9385 #[test]
9386 fn cgroup_pre_exec_spawn_failure_names_the_cgroup_path() {
9387 let cgroup_path = PathBuf::from("/sys/fs/cgroup/subc-modules/broken-module");
9388 let reason = SuperviseError::Spawn {
9389 program: PathBuf::from("/bin/true"),
9390 source: io::Error::from_raw_os_error(13),
9391 cgroup_path: Some(cgroup_path.clone()),
9392 }
9393 .to_string();
9394
9395 assert!(
9396 reason.contains(&cgroup_path.display().to_string()),
9397 "a pre_exec spawn failure must name the cgroup path: {reason}"
9398 );
9399 }
9400}
9401
9402#[cfg(test)]
9403mod spawn_subscriber_lag_tests {
9404 use super::*;
9405
9406 #[tokio::test]
9411 async fn lagged_spawn_subscriber_receives_a_terminal_lagged_error_after_its_queued_frames() {
9412 let feed = SpawnEventFeed::default();
9413 feed.configure_incarnation("lag-incarnation".to_string());
9414 let (tx, mut rx) = mpsc::channel(1);
9417 feed.subscribe(ConnectionId::new(1), 7, 1, None, FrameSink::new(tx))
9418 .expect("subscribe");
9419 let emitted = SPAWN_SUBSCRIBER_BUFFER + 16;
9420 for index in 0..emitted {
9421 feed.emit_spawned(&format!("lag-module-{index}"), 1000, 0);
9422 tokio::task::yield_now().await;
9425 }
9426 assert_eq!(
9427 feed.subscriber_count(),
9428 0,
9429 "the lagged subscriber must be removed"
9430 );
9431
9432 let mut data = Vec::new();
9433 let mut last = None;
9434 loop {
9435 let next = tokio::time::timeout(Duration::from_secs(5), rx.recv())
9436 .await
9437 .expect("the forwarder must finish once the subscriber is dropped");
9438 let Some(outbound) = next else { break };
9439 let frame = outbound.frame;
9440 if frame.header.ty == FrameType::StreamData {
9441 assert!(last.is_none(), "no data may follow the terminal frame");
9442 let event: SpawnEvent = serde_json::from_slice(&frame.body).unwrap();
9443 data.push(event.cursor.seq);
9444 } else {
9445 assert!(last.is_none(), "exactly one terminal frame");
9446 last = Some(frame);
9447 }
9448 }
9449 assert!(!data.is_empty(), "queued frames drain before the terminal");
9450 for pair in data.windows(2) {
9451 assert_eq!(
9452 pair[1],
9453 pair[0] + 1,
9454 "queued frames arrive dense and in order"
9455 );
9456 }
9457 let terminal = last.expect("a lagged subscriber must receive a terminal frame");
9458 assert_eq!(terminal.header.ty, FrameType::Error);
9459 assert_eq!(terminal.header.corr, 7);
9460 let body: subc_protocol::ErrorBody = serde_json::from_slice(&terminal.body).unwrap();
9461 assert_eq!(body.code, SPAWN_SUBSCRIBER_LAGGED_CODE);
9462 let detail = body.detail.expect("lagged error carries detail");
9463 assert_eq!(
9464 detail["first_undelivered_cursor"]["seq"],
9465 data.last().unwrap() + 1,
9466 "the named cursor is the first event the subscriber did not receive"
9467 );
9468 assert_eq!(
9469 detail["first_undelivered_cursor"]["daemon_incarnation"],
9470 "lag-incarnation"
9471 );
9472 }
9473}
9474
9475#[cfg(test)]
9476mod terminal_history_read_concurrency_tests {
9477 use super::*;
9478 use crate::terminal_journal::read_pause;
9479 use std::sync::mpsc as std_mpsc;
9480 use subc_test_support::TestTempDir;
9481
9482 fn journaled_ring(
9483 journal: &Arc<crate::terminal_journal::TerminalJournal>,
9484 ) -> Arc<Mutex<TerminalRing>> {
9485 Arc::new(Mutex::new(
9486 TerminalRing::new(TerminalRingConfig::default(), 1)
9487 .with_journal(Some(Arc::clone(journal))),
9488 ))
9489 }
9490
9491 fn crash(at_ms: u64) -> ExitReport {
9492 ExitReport {
9493 kind: ExitKind::Crash,
9494 code: Some(1),
9495 signal: None,
9496 at_ms,
9497 }
9498 }
9499
9500 fn record_within(
9503 module_id: &'static str,
9504 ring: &Arc<Mutex<TerminalRing>>,
9505 at_ms: u64,
9506 bound: Duration,
9507 ) -> bool {
9508 let ring = Arc::clone(ring);
9509 let (done, done_rx) = std_mpsc::channel();
9510 std::thread::spawn(move || {
9511 record_terminal(
9512 module_id,
9513 &ring,
9514 &SpawnEventFeed::default(),
9515 &crash(at_ms),
9516 TerminalDisposition::Restarting,
9517 );
9518 let _ = done.send(());
9519 });
9520 done_rx.recv_timeout(bound).is_ok()
9521 }
9522
9523 #[test]
9528 fn exits_recorded_during_a_paused_history_read_are_not_blocked_or_half_merged() {
9529 let dir = TestTempDir::new("terminal-history-concurrent-read");
9530 let path = dir.join("terminals.jsonl");
9531 let journal = Arc::new(crate::terminal_journal::TerminalJournal::open(
9532 path.clone(),
9533 "daemon".into(),
9534 ));
9535 let reader_ring = journaled_ring(&journal);
9536 let other_ring = journaled_ring(&journal);
9537 assert!(record_within(
9538 "reader-module",
9539 &reader_ring,
9540 10,
9541 Duration::from_secs(5)
9542 ));
9543
9544 let (started, release) = read_pause::install(&path);
9545 let reading = {
9546 let ring = Arc::clone(&reader_ring);
9547 std::thread::spawn(move || durable_terminal_history_of(&ring, "reader-module"))
9548 };
9549 started
9550 .recv_timeout(Duration::from_secs(5))
9551 .expect("the history read reached its pause");
9552
9553 let bound = Duration::from_secs(1);
9554 assert!(
9555 record_within("other-module", &other_ring, 20, bound),
9556 "another module's exit waited on a history read (journal writer held)"
9557 );
9558 assert!(
9559 record_within("reader-module", &reader_ring, 30, bound),
9560 "the read module's own exit waited on its history read (ring held)"
9561 );
9562
9563 drop(release);
9564 let paused = reading.join().unwrap();
9565 assert_eq!(
9566 paused.entries.iter().map(|e| e.at_ms).collect::<Vec<_>>(),
9567 vec![10],
9568 "an exit recorded after the read began lands in neither half of it"
9569 );
9570 assert_eq!(paused.journal_skipped_lines, 0);
9571 assert_eq!(paused.journal_read_errors, 0);
9572
9573 let after = durable_terminal_history_of(&reader_ring, "reader-module");
9574 assert_eq!(
9575 after.entries.iter().map(|e| e.at_ms).collect::<Vec<_>>(),
9576 vec![10, 30],
9577 "the next read merges ring and journal with no duplicate"
9578 );
9579 assert_eq!(after.journal_skipped_lines, 0);
9580 }
9581}
9582
9583#[cfg(test)]
9588mod stderr_settle_tests {
9589 use std::{
9590 future::Future,
9591 io,
9592 pin::Pin,
9593 sync::{Arc, Mutex},
9594 task::{Context, Poll},
9595 time::Duration,
9596 };
9597
9598 use tokio::{
9599 io::{AsyncRead, ReadBuf},
9600 sync::oneshot,
9601 time::Instant,
9602 };
9603
9604 use super::{settle_stderr_pump, StderrPump};
9605 use crate::stderr_tail::{
9606 pump_stderr_to, CaptureState, OutputSink, StderrRing, StderrTailConfig, TailEntry,
9607 };
9608
9609 const BOUND: Duration = Duration::from_millis(250);
9610
9611 struct HeldReader {
9615 before: Option<Vec<u8>>,
9616 gate: Option<oneshot::Receiver<()>>,
9617 after: io::Cursor<Vec<u8>>,
9618 }
9619
9620 impl AsyncRead for HeldReader {
9621 fn poll_read(
9622 mut self: Pin<&mut Self>,
9623 cx: &mut Context<'_>,
9624 buf: &mut ReadBuf<'_>,
9625 ) -> Poll<io::Result<()>> {
9626 if let Some(bytes) = self.before.take() {
9627 buf.put_slice(&bytes);
9628 return Poll::Ready(Ok(()));
9629 }
9630 if let Some(gate) = self.gate.as_mut() {
9631 match Pin::new(gate).poll(cx) {
9632 Poll::Pending => return Poll::Pending,
9633 Poll::Ready(_) => self.gate = None,
9634 }
9635 }
9636 Pin::new(&mut self.after).poll_read(cx, buf)
9637 }
9638 }
9639
9640 struct DiscardSink;
9641
9642 impl OutputSink for DiscardSink {
9643 fn write_line(&mut self, _line: &[u8]) {}
9644 }
9645
9646 fn line(text: &str) -> TailEntry {
9647 TailEntry::Line {
9648 text: text.to_string(),
9649 truncated: false,
9650 }
9651 }
9652
9653 fn lock(ring: &Arc<Mutex<StderrRing>>) -> std::sync::MutexGuard<'_, StderrRing> {
9654 ring.lock().unwrap()
9655 }
9656
9657 fn held_pump(
9661 ring: &Arc<Mutex<StderrRing>>,
9662 before: &str,
9663 after: &str,
9664 ) -> (StderrPump, oneshot::Sender<()>) {
9665 let generation = lock(ring).begin_process();
9666 let (release, gate) = oneshot::channel();
9667 let reader = HeldReader {
9668 before: Some(before.as_bytes().to_vec()),
9669 gate: Some(gate),
9670 after: io::Cursor::new(after.as_bytes().to_vec()),
9671 };
9672 let task = tokio::spawn(pump_stderr_to(
9673 reader,
9674 Arc::clone(ring),
9675 generation,
9676 DiscardSink,
9677 ));
9678 (StderrPump { task, generation }, release)
9679 }
9680
9681 async fn wait_until(ring: &Arc<Mutex<StderrRing>>, done: impl Fn(&StderrRing) -> bool) {
9682 for _ in 0..1000 {
9683 if done(&lock(ring)) {
9684 return;
9685 }
9686 tokio::time::sleep(Duration::from_millis(1)).await;
9687 }
9688 panic!(
9689 "ring never reached the expected state: {:?}",
9690 lock(ring).snapshot(None, None)
9691 );
9692 }
9693
9694 #[tokio::test(start_paused = true)]
9695 async fn a_crash_line_the_reader_had_not_reached_by_the_bound_is_kept_before_the_restart() {
9696 let ring = Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default())));
9697 let (pump, release) = held_pump(&ring, "booting\n", "config error: missing storage\n");
9698
9699 settle_stderr_pump("crasher", &ring, pump, BOUND).await;
9700 let before_release = lock(&ring).snapshot(None, None);
9701 assert!(
9702 matches!(before_release.capture, CaptureState::Incomplete { .. }),
9703 "a reader that has not reached EOF cannot claim a whole tail: {before_release:?}"
9704 );
9705
9706 let next = lock(&ring).begin_process();
9709 lock(&ring).push_line_from(next, "next process booting");
9710 release.send(()).unwrap();
9711 wait_until(&ring, |ring| {
9712 ring.snapshot(None, None).capture == CaptureState::Captured
9713 })
9714 .await;
9715
9716 assert_eq!(
9717 lock(&ring).snapshot(None, None).entries,
9718 vec![
9719 line("booting"),
9720 line("config error: missing storage"),
9721 TailEntry::ProcessStart,
9722 line("next process booting"),
9723 ],
9724 "the crash's last line must survive a slow reader and stay in the crashed process's section"
9725 );
9726 }
9727
9728 #[tokio::test(start_paused = true)]
9729 async fn a_pipe_held_open_by_a_descendant_reads_incomplete_without_delaying_the_restart_past_the_bound(
9730 ) {
9731 let ring = Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default())));
9732 let (pump, _held) = held_pump(&ring, "parent exiting\n", "");
9735
9736 let started = Instant::now();
9737 settle_stderr_pump("orphaning", &ring, pump, BOUND).await;
9738 assert_eq!(
9739 started.elapsed(),
9740 BOUND,
9741 "the restart must wait exactly the bound for a pipe that stays open, no longer"
9742 );
9743
9744 let next = lock(&ring).begin_process();
9745 lock(&ring).push_line_from(next, "next process booting");
9746 tokio::time::sleep(Duration::from_secs(60)).await;
9747
9748 let snapshot = lock(&ring).snapshot(None, None);
9749 match &snapshot.capture {
9750 CaptureState::Incomplete { reason } => assert!(
9751 reason.contains("had not reached EOF") && reason.contains("250ms"),
9752 "the reason must say what is missing and after how long: {reason}"
9753 ),
9754 other => panic!("expected Incomplete while the pipe is held open, got {other:?}"),
9755 }
9756 assert_eq!(
9757 snapshot.entries,
9758 vec![
9759 line("parent exiting"),
9760 TailEntry::ProcessStart,
9761 line("next process booting"),
9762 ]
9763 );
9764 }
9765
9766 #[tokio::test(start_paused = true)]
9767 async fn a_reader_that_reaches_eof_within_the_bound_leaves_the_tail_captured() {
9768 let ring = Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default())));
9769 let (pump, release) = held_pump(&ring, "one\n", "two\n");
9770 release.send(()).unwrap();
9771
9772 settle_stderr_pump("clean", &ring, pump, BOUND).await;
9773
9774 let snapshot = lock(&ring).snapshot(None, None);
9775 assert_eq!(snapshot.capture, CaptureState::Captured);
9776 assert_eq!(snapshot.entries, vec![line("one"), line("two")]);
9777 }
9778}
9779
9780#[cfg(all(test, windows))]
9794mod job_containment_tests {
9795 use super::*;
9796 use std::{
9797 path::{Path, PathBuf},
9798 sync::{Arc, Mutex},
9799 time::{Duration, Instant},
9800 };
9801 use subc_test_support::TestTempDir;
9802
9803 fn stub_path() -> PathBuf {
9809 let mut path = std::env::current_exe().expect("current_exe available in tests");
9810 path.pop();
9811 path.pop();
9812 path.push("fake-aft-stub.exe");
9813 assert!(
9814 path.exists(),
9815 "fake-aft-stub not built at {}: run `cargo test -p subc-core` (which builds \
9816 [[bin]] targets) rather than `cargo test -p subc-core --lib` (which does not)",
9817 path.display()
9818 );
9819 path
9820 }
9821
9822 fn read_grandchild_pid(path: &Path) -> u32 {
9824 let deadline = Instant::now() + Duration::from_secs(10);
9825 loop {
9826 if let Ok(contents) = std::fs::read_to_string(path) {
9827 if let Ok(pid) = contents.trim().parse() {
9828 return pid;
9829 }
9830 }
9831 assert!(
9832 Instant::now() < deadline,
9833 "the stub never recorded a grandchild pid at {}",
9834 path.display()
9835 );
9836 std::thread::sleep(Duration::from_millis(10));
9837 }
9838 }
9839
9840 struct Fixture {
9843 _dir: TestTempDir,
9844 module_id: String,
9845 grandchild: u32,
9846 child: Option<SupervisedChild>,
9847 registry: Arc<Registry>,
9848 snapshot: Arc<Mutex<SupervisorSnapshot>>,
9849 terminal_ring: Arc<Mutex<TerminalRing>>,
9850 spawn_events: SpawnEventFeed,
9851 }
9852
9853 fn fixture(label: &str, module_id: &str) -> Fixture {
9854 let dir = TestTempDir::new(label);
9855 let pid_file = dir.join("grandchild.pid");
9856 let supervisor = Supervisor::new(
9857 Arc::new(Registry::default()),
9858 RestartPolicy::new(3, Duration::ZERO),
9859 );
9860 let runtime = supervisor.runtime_config();
9861 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
9862 let spec = ModuleSpec {
9863 module_id: module_id.to_string(),
9864 program: stub_path(),
9865 args: Vec::new(),
9869 env: vec![
9870 ("FAKE_AFT_NEVER_CONNECT".to_string(), "1".to_string()),
9871 (
9872 "FAKE_AFT_GRANDCHILD_PID_FILE".to_string(),
9873 pid_file.display().to_string(),
9874 ),
9875 ],
9876 reserved: false,
9877 reserved_prefixes: Vec::new(),
9878 protocol: ModuleProtocol::Subc,
9879 overlap: Default::default(),
9880 };
9881 let child = spawn_and_mark_running(&spec, &runtime, &snapshot)
9882 .expect("spawn the supervised fixture");
9883 let grandchild = read_grandchild_pid(&pid_file);
9884 Fixture {
9885 _dir: dir,
9886 module_id: module_id.to_string(),
9887 grandchild,
9888 child: Some(child),
9889 registry: Arc::new(Registry::default()),
9890 snapshot,
9891 terminal_ring: Arc::clone(&runtime.terminal_ring),
9892 spawn_events: SpawnEventFeed::default(),
9893 }
9894 }
9895
9896 impl Fixture {
9897 async fn drain(&mut self) {
9899 let child = self
9900 .child
9901 .take()
9902 .expect("the fixture child is still present");
9903 drain_child_to_state(
9904 &self.module_id,
9905 ModuleProtocol::Subc,
9906 StopNotice::NotSent,
9909 &self.registry,
9910 &self.snapshot,
9911 &self.terminal_ring,
9912 &self.spawn_events,
9913 child,
9914 Duration::from_millis(500),
9915 ModuleState::Stopped,
9916 Some(false),
9917 )
9918 .await
9919 .expect("drain the supervised fixture");
9920 }
9921 }
9922
9923 #[tokio::test]
9929 async fn teardown_reaps_the_grandchild() {
9930 let mut fixture = fixture("teardown-grandchild", "tree-teardown");
9931 let grandchild = fixture.grandchild;
9932
9933 assert!(
9934 subc_jobobject::process_exists(grandchild),
9935 "grandchild {grandchild} must be alive before teardown, or this proves nothing"
9936 );
9937
9938 fixture.drain().await;
9939
9940 assert!(
9941 subc_jobobject::wait_for_process_exit(grandchild, Duration::from_secs(10)),
9942 "grandchild {grandchild} outlived module teardown: the tree was not contained"
9943 );
9944 }
9945
9946 #[test]
9959 fn an_uncontained_grandchild_survives_a_direct_child_kill() {
9960 let dir = TestTempDir::new("teardown-uncontained");
9961 let pid_file = dir.join("grandchild.pid");
9962 let mut child = std::process::Command::new(stub_path())
9963 .env("FAKE_AFT_NEVER_CONNECT", "1")
9964 .env(
9965 "FAKE_AFT_GRANDCHILD_PID_FILE",
9966 pid_file.display().to_string(),
9967 )
9968 .stdin(std::process::Stdio::null())
9969 .stdout(std::process::Stdio::null())
9970 .stderr(std::process::Stdio::null())
9971 .spawn()
9972 .expect("spawn the uncontained fixture");
9973 let grandchild = read_grandchild_pid(&pid_file);
9974
9975 child.kill().expect("kill the direct child");
9977 let _ = child.wait();
9978
9979 assert!(
9980 subc_jobobject::process_exists(grandchild),
9981 "grandchild {grandchild} died with the direct child, so this control no longer \
9982 distinguishes contained from uncontained teardown and the regression test is \
9983 passing vacuously"
9984 );
9985
9986 kill_tree(grandchild);
9989 }
9990
9991 #[tokio::test]
10000 async fn dropping_containment_reaps_the_grandchild() {
10001 let mut fixture = fixture("drop-containment", "tree-drop");
10002 let grandchild = fixture.grandchild;
10003
10004 assert!(subc_jobobject::process_exists(grandchild));
10005
10006 fixture.child.as_mut().expect("child present").job = None;
10008
10009 assert!(
10010 subc_jobobject::wait_for_process_exit(grandchild, Duration::from_secs(10)),
10011 "grandchild {grandchild} survived the containment handle closing, so a daemon \
10012 crash would leave the tree behind"
10013 );
10014 }
10015
10016 fn kill_tree(pid: u32) {
10018 let _ = std::process::Command::new("taskkill.exe")
10019 .args(["/PID", &pid.to_string(), "/T", "/F"])
10020 .stdin(std::process::Stdio::null())
10021 .stdout(std::process::Stdio::null())
10022 .stderr(std::process::Stdio::null())
10023 .status();
10024 assert!(
10025 subc_jobobject::wait_for_process_exit(pid, Duration::from_secs(10)),
10026 "could not clean up grandchild {pid}"
10027 );
10028 }
10029}