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)]
2180 pub(crate) async fn end_children_for_daemon_shutdown(
2181 &self,
2182 already_escalated: bool,
2183 escalate: impl std::future::Future<Output = ()>,
2184 ) {
2185 tokio::pin!(escalate);
2186 let mut escalated = already_escalated;
2187 if let Some(forwarding) = &self.forwarding {
2188 let reason = CloseReason::new(
2189 "daemon_shutdown",
2190 "the daemon is exiting after its shutdown notice and drain",
2191 );
2192 if escalated {
2193 send_module_goodbyes_for_daemon_shutdown(forwarding, &reason, false).await;
2196 } else {
2197 tokio::select! {
2198 biased;
2199 _ = escalate.as_mut() => {
2200 info!("second SIGTERM: abandoning module GOODBYE delivery");
2201 escalated = true;
2202 }
2203 _ = send_module_goodbyes_for_daemon_shutdown(forwarding, &reason, true) => {}
2204 }
2205 }
2206 let closed = forwarding.close_all_connections(&reason);
2207 debug!(closed, "closed established connections for daemon shutdown");
2208 }
2209 let escalated_here = escalated && !already_escalated;
2213 let remaining_escalate = async move {
2214 if escalated_here {
2215 std::future::pending::<()>().await;
2216 } else {
2217 escalate.await;
2218 }
2219 };
2220 crate::child_roster::end_children_for_daemon_shutdown(
2221 &self.child_roster,
2222 escalated,
2223 remaining_escalate,
2224 )
2225 .await;
2226 }
2227
2228 pub fn new(registry: Arc<Registry>, restart_policy: RestartPolicy) -> Self {
2229 Self {
2230 registry,
2231 restart_policy,
2232 drain_timeout: DEFAULT_DRAIN_TIMEOUT,
2233 connection_file_path: None,
2234 capture_logs_dir: None,
2235 forwarding: None,
2236 process_liveness: Arc::new(SupervisorProcessLiveness::default()),
2237 supervisor_handle: None,
2238 health: HealthConfig::default(),
2239 daemon_start_clock: crate::clock::StartClock::capture(),
2240 terminal_journal: None,
2241 spawn_events: SpawnEventFeed::default(),
2242 provenance_probe: ExecutableIdentityProbe::default(),
2243 child_roster: ChildRoster::default(),
2244 #[cfg(target_os = "linux")]
2245 cgroup_placement: None,
2246 }
2247 }
2248
2249 pub fn with_drain_timeout(mut self, drain_timeout: Duration) -> Self {
2250 self.drain_timeout = drain_timeout;
2251 self
2252 }
2253
2254 pub fn with_process_liveness(
2255 mut self,
2256 process_liveness: Arc<SupervisorProcessLiveness>,
2257 ) -> Self {
2258 self.process_liveness = process_liveness;
2259 self
2260 }
2261
2262 pub fn with_connection_file_path(mut self, connection_file_path: impl Into<PathBuf>) -> Self {
2263 self.connection_file_path = Some(connection_file_path.into());
2264 self
2265 }
2266
2267 pub fn with_capture_logs_dir(mut self, logs_dir: impl Into<PathBuf>) -> Self {
2269 self.capture_logs_dir = Some(logs_dir.into());
2270 self
2271 }
2272
2273 pub fn with_daemon_incarnation(self, daemon_incarnation: String) -> Self {
2276 self.spawn_events.configure_incarnation(daemon_incarnation);
2280 self
2281 }
2282
2283 pub fn with_terminal_journal(self, path: PathBuf, daemon_incarnation: String) -> Self {
2286 let mut this = self.with_daemon_incarnation(daemon_incarnation.clone());
2287 this.terminal_journal = Some(Arc::new(crate::terminal_journal::TerminalJournal::open(
2288 path,
2289 daemon_incarnation,
2290 )));
2291 this
2292 }
2293
2294 pub fn with_forwarding(mut self, forwarding: Arc<ForwardingTable>) -> Self {
2295 self.forwarding = Some(forwarding);
2296 self
2297 }
2298
2299 pub fn with_handle(mut self, supervisor_handle: SupervisorHandle) -> Self {
2300 self.spawn_events = supervisor_handle.spawn_events.clone();
2301 self.supervisor_handle = Some(supervisor_handle);
2302 self
2303 }
2304
2305 pub fn with_health_config(mut self, health: HealthConfig) -> Self {
2306 self.health = health;
2307 self
2308 }
2309
2310 pub fn with_live_children_record(self, path: impl Into<PathBuf>) -> Self {
2314 self.child_roster.record_to(path.into());
2315 self
2316 }
2317
2318 #[cfg(target_os = "linux")]
2319 pub fn with_cgroup_placement(
2320 mut self,
2321 cgroup_placement: Option<subc_cgroup::Placement>,
2322 ) -> Self {
2323 self.cgroup_placement = cgroup_placement;
2324 self
2325 }
2326
2327 pub fn spawn(&self, spec: ModuleSpec) -> Result<SupervisedModule, SuperviseError> {
2333 validate_spec(&spec)?;
2334
2335 let runtime = self.runtime_config();
2336 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
2337 let child = spawn_child(
2338 &spec,
2339 runtime.connection_file_path.as_deref(),
2340 self.supervisor_handle.as_ref(),
2341 &runtime.stderr_ring,
2342 runtime.capture_logs_dir.as_deref(),
2343 &runtime.child_roster,
2344 #[cfg(target_os = "linux")]
2345 runtime.cgroup_placement.as_ref(),
2346 )?;
2347 set_running(&snapshot, &child, &spec.module_id, &runtime.spawn_events)?;
2348 self.process_liveness
2349 .track(spec.module_id.clone(), Arc::clone(&snapshot));
2350
2351 Ok(self.supervised_module(spec, runtime, snapshot, Some(child)))
2352 }
2353
2354 pub fn supervise_configured(
2360 &self,
2361 spec: ModuleSpec,
2362 enabled: bool,
2363 ) -> Result<SupervisedModule, SuperviseError> {
2364 validate_spec(&spec)?;
2365
2366 let runtime = self.runtime_config();
2367 if !enabled {
2368 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::disabled()));
2369 return Ok(self.supervised_module(spec, runtime, snapshot, None));
2370 }
2371
2372 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
2373 match spawn_child(
2374 &spec,
2375 runtime.connection_file_path.as_deref(),
2376 self.supervisor_handle.as_ref(),
2377 &runtime.stderr_ring,
2378 runtime.capture_logs_dir.as_deref(),
2379 &runtime.child_roster,
2380 #[cfg(target_os = "linux")]
2381 runtime.cgroup_placement.as_ref(),
2382 ) {
2383 Ok(child) => {
2384 set_running(&snapshot, &child, &spec.module_id, &runtime.spawn_events)?;
2385 self.process_liveness
2386 .track(spec.module_id.clone(), Arc::clone(&snapshot));
2387 Ok(self.supervised_module(spec, runtime, snapshot, Some(child)))
2388 }
2389 Err(err) => {
2390 error!(
2391 module_id = %spec.module_id,
2392 program = %spec.program.display(),
2393 error = %err,
2394 "configured module failed to spawn; marking failed and continuing"
2395 );
2396 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::failed()));
2397 Ok(self.supervised_module(spec, runtime, snapshot, None))
2398 }
2399 }
2400 }
2401
2402 pub fn supervise_configured_with_health(
2408 &self,
2409 spec: ModuleSpec,
2410 enabled: bool,
2411 health: HealthConfig,
2412 drain_timeout_ms: Option<u64>,
2413 restart_policy: RestartPolicy,
2414 ) -> Result<SupervisedModule, SuperviseError> {
2415 validate_spec(&spec)?;
2416
2417 let mut runtime = self.runtime_config();
2418 runtime.health = health;
2419 runtime.restart_policy = restart_policy;
2420 if let Some(ms) = drain_timeout_ms {
2421 runtime.drain_timeout = Duration::from_millis(ms);
2422 *runtime
2423 .effective_drain_timeout
2424 .lock()
2425 .unwrap_or_else(|poisoned| poisoned.into_inner()) = runtime.drain_timeout;
2426 }
2427 if !enabled {
2428 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::disabled()));
2429 return Ok(self.supervised_module(spec, runtime, snapshot, None));
2430 }
2431
2432 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
2433 match spawn_child(
2434 &spec,
2435 runtime.connection_file_path.as_deref(),
2436 self.supervisor_handle.as_ref(),
2437 &runtime.stderr_ring,
2438 runtime.capture_logs_dir.as_deref(),
2439 &runtime.child_roster,
2440 #[cfg(target_os = "linux")]
2441 runtime.cgroup_placement.as_ref(),
2442 ) {
2443 Ok(child) => {
2444 set_running(&snapshot, &child, &spec.module_id, &runtime.spawn_events)?;
2445 self.process_liveness
2446 .track(spec.module_id.clone(), Arc::clone(&snapshot));
2447 Ok(self.supervised_module(spec, runtime, snapshot, Some(child)))
2448 }
2449 Err(err) => {
2450 if health.critical {
2451 error!(
2452 module_id = %spec.module_id,
2453 program = %spec.program.display(),
2454 error = %err,
2455 "critical configured module failed to spawn; marking failed and alerting"
2456 );
2457 } else {
2458 error!(
2459 module_id = %spec.module_id,
2460 program = %spec.program.display(),
2461 error = %err,
2462 "configured module failed to spawn; marking failed and continuing"
2463 );
2464 }
2465 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::failed()));
2466 Ok(self.supervised_module(spec, runtime, snapshot, None))
2467 }
2468 }
2469 }
2470
2471 fn runtime_config(&self) -> SupervisorRuntimeConfig {
2472 let effective_drain_timeout = Arc::new(Mutex::new(self.drain_timeout));
2473 SupervisorRuntimeConfig {
2474 restart_policy: self.restart_policy,
2475 drain_timeout: self.drain_timeout,
2476 child_roster: self
2479 .child_roster
2480 .for_module(Arc::clone(&effective_drain_timeout)),
2481 effective_drain_timeout,
2482 default_drain_timeout: self.drain_timeout,
2483 health: self.health,
2484 connection_file_path: self.connection_file_path.clone(),
2485 capture_logs_dir: self.capture_logs_dir.clone(),
2486 forwarding: self.forwarding.clone(),
2487 supervisor_handle: self.supervisor_handle.clone(),
2488 stderr_ring: Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default()))),
2489 terminal_ring: Arc::new(Mutex::new(
2490 TerminalRing::new(
2491 TerminalRingConfig::default(),
2492 self.daemon_start_clock.started_at_ms(),
2493 )
2494 .with_start_clock(self.daemon_start_clock)
2495 .with_journal(self.terminal_journal.clone())
2496 .with_daemon_shutdown(self.child_roster.shutdown_flag()),
2497 )),
2498 spawn_events: self.spawn_events.clone(),
2499 #[cfg(target_os = "linux")]
2500 cgroup_placement: self.cgroup_placement.clone(),
2501 #[cfg(test)]
2502 test_seed_stale_facts_before_enable_spawn: false,
2503 }
2504 }
2505
2506 fn supervised_module(
2507 &self,
2508 spec: ModuleSpec,
2509 runtime: SupervisorRuntimeConfig,
2510 snapshot: SharedSnapshot,
2511 child: Option<SupervisedChild>,
2512 ) -> SupervisedModule {
2513 let configuration = Arc::new(Mutex::new(SupervisedConfiguration {
2514 spec: spec.clone(),
2515 health: runtime.health,
2516 }));
2517 let stderr_ring = Arc::clone(&runtime.stderr_ring);
2518 let terminal_ring = Arc::clone(&runtime.terminal_ring);
2519 let restart_policy = runtime.restart_policy;
2523 let effective_drain_timeout = Arc::clone(&runtime.effective_drain_timeout);
2524 let (tx, rx) = mpsc::channel(4);
2525 let monitor = tokio::spawn(supervise_loop(
2526 spec.clone(),
2527 runtime,
2528 Arc::clone(&self.registry),
2529 Arc::clone(&self.process_liveness),
2530 Arc::clone(&snapshot),
2531 child,
2532 rx,
2533 ));
2534
2535 let module_id = spec.module_id.clone();
2536 let module = SupervisedModule {
2537 inner: Arc::new(SupervisedModuleInner {
2538 module_id: module_id.clone(),
2539 registry: Arc::clone(&self.registry),
2540 snapshot,
2541 configuration,
2542 stderr_ring,
2543 terminal_ring,
2544 commands: tx,
2545 monitor: Mutex::new(Some(monitor)),
2546 restart_policy,
2547 effective_drain_timeout,
2548 provenance_probe: self.provenance_probe.clone(),
2549 }),
2550 };
2551 if let Some(supervisor_handle) = &self.supervisor_handle {
2552 supervisor_handle.apply_identity_configuration(&spec);
2553 supervisor_handle.insert(module.clone());
2554 }
2555 module
2556 }
2557}
2558
2559impl Default for Supervisor {
2560 fn default() -> Self {
2561 Self::new(Arc::new(Registry::default()), RestartPolicy::default())
2562 }
2563}
2564
2565#[derive(Clone)]
2567pub struct SupervisedModule {
2568 inner: Arc<SupervisedModuleInner>,
2569}
2570
2571struct SupervisedModuleInner {
2572 module_id: String,
2573 registry: Arc<Registry>,
2574 snapshot: SharedSnapshot,
2575 configuration: Arc<Mutex<SupervisedConfiguration>>,
2576 stderr_ring: Arc<Mutex<StderrRing>>,
2577 terminal_ring: Arc<Mutex<TerminalRing>>,
2578 commands: mpsc::Sender<SupervisorCommand>,
2579 monitor: Mutex<Option<JoinHandle<()>>>,
2580 restart_policy: RestartPolicy,
2584 effective_drain_timeout: Arc<Mutex<Duration>>,
2585 provenance_probe: ExecutableIdentityProbe,
2586}
2587
2588impl fmt::Debug for SupervisedModule {
2589 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2590 f.debug_struct("SupervisedModule")
2591 .field("module_id", &self.inner.module_id)
2592 .field("status", &self.status())
2593 .finish_non_exhaustive()
2594 }
2595}
2596
2597impl SupervisedModule {
2598 pub fn module_id(&self) -> &str {
2599 &self.inner.module_id
2600 }
2601
2602 #[cfg(test)]
2606 pub(crate) fn record_health_probe_failure_for_test(
2607 &self,
2608 detail: &str,
2609 ) -> Result<(), SuperviseError> {
2610 update_snapshot(&self.inner.snapshot, Some(&self.inner.module_id), |state| {
2611 state.health.consecutive_failures = state.health.consecutive_failures.saturating_add(1);
2612 state.health.detail = Some(detail.to_string());
2613 })
2614 }
2615
2616 pub fn state(&self) -> Result<ModuleState, SuperviseError> {
2617 Ok(lock_snapshot(&self.inner.snapshot)?.state)
2618 }
2619
2620 pub fn stderr_tail(
2627 &self,
2628 max_lines: Option<usize>,
2629 max_bytes: Option<usize>,
2630 ) -> StderrTailSnapshot {
2631 self.inner
2632 .stderr_ring
2633 .lock()
2634 .unwrap_or_else(|poisoned| poisoned.into_inner())
2635 .snapshot(max_lines, max_bytes)
2636 }
2637
2638 pub fn terminal_history(&self) -> TerminalHistorySnapshot {
2643 self.inner
2644 .terminal_ring
2645 .lock()
2646 .unwrap_or_else(|poisoned| poisoned.into_inner())
2647 .snapshot()
2648 }
2649
2650 pub fn durable_terminal_history(&self) -> subc_control::TerminalHistory {
2655 durable_terminal_history_of(&self.inner.terminal_ring, &self.inner.module_id)
2656 }
2657
2658 pub(crate) async fn read_durable_terminal_history(
2663 &self,
2664 ) -> Result<subc_control::TerminalHistory, tokio::task::JoinError> {
2665 let terminal_ring = Arc::clone(&self.inner.terminal_ring);
2666 let module_id = self.inner.module_id.clone();
2667 tokio::task::spawn_blocking(move || durable_terminal_history_of(&terminal_ring, &module_id))
2668 .await
2669 }
2670
2671 pub fn status(&self) -> Result<ModuleStatus, SuperviseError> {
2672 self.status_with_snapshot_lock(&self.inner.snapshot, None)
2673 }
2674
2675 pub(crate) fn record_deliberate_severance(
2676 &self,
2677 identity: ProcessIdentity,
2678 ) -> Result<bool, SuperviseError> {
2679 let mut snapshot = lock_snapshot(&self.inner.snapshot)?;
2680 if snapshot.pid != Some(identity.pid)
2681 || snapshot.process_start_time != Some(identity.start_time)
2682 {
2683 return Ok(false);
2684 }
2685 snapshot.deliberate_severance = Some(identity);
2686 Ok(true)
2687 }
2688
2689 pub(crate) fn status_for_control(
2694 &self,
2695 caller: &'static str,
2696 ) -> Result<ModuleStatus, SuperviseError> {
2697 self.status_with_snapshot_lock(&self.inner.snapshot, Some(caller))
2698 }
2699
2700 fn status_with_snapshot_lock(
2701 &self,
2702 snapshot: &SharedSnapshot,
2703 caller: Option<&'static str>,
2704 ) -> Result<ModuleStatus, SuperviseError> {
2705 let mut guard = match caller {
2706 Some(caller) => lock_snapshot_for_control(snapshot, &self.inner.module_id, caller)?,
2707 None => lock_snapshot(snapshot)?,
2708 };
2709 let restart_count =
2712 guard.crash_restarts_in_window(self.inner.restart_policy.window, Instant::now());
2713 let snapshot = guard.clone();
2714 drop(guard);
2715 let drain_timeout = *self.inner.effective_drain_timeout.lock().map_err(|_| {
2716 SuperviseError::StatePoisoned {
2717 module_id: Some(self.inner.module_id.clone()),
2718 }
2719 })?;
2720 let registration_active = self
2721 .inner
2722 .registry
2723 .get_module(&self.inner.module_id)
2724 .map_err(SuperviseError::Registry)?
2725 .is_some();
2726 let protocol = self.declared_protocol()?;
2727 let running_process =
2728 snapshot.enabled && snapshot.state == ModuleState::Running && snapshot.process_alive;
2729 let live = match protocol {
2735 ModuleProtocol::Subc => running_process && registration_active,
2736 ModuleProtocol::None => running_process,
2737 };
2738
2739 Ok(ModuleStatus {
2740 module_id: self.inner.module_id.clone(),
2741 state: snapshot.state,
2742 enabled: snapshot.enabled,
2743 process_alive: snapshot.process_alive,
2744 registration_active,
2745 protocol,
2746 live,
2747 restart_count,
2748 lifetime_restarts: snapshot.lifetime_restarts,
2749 spawn_generation: snapshot.spawn_generation,
2750 max_restarts: self.inner.restart_policy.max_restarts,
2751 restart_window: self.inner.restart_policy.window,
2752 drain_timeout,
2753 restart_backoff: self.inner.restart_policy.backoff,
2754 restart_max_backoff: self.inner.restart_policy.max_backoff,
2755 pid: snapshot.pid,
2756 spawned_at_ms: snapshot.spawned_at_ms,
2757 spawned_from: snapshot.spawned_from,
2758 process_start_time: snapshot.process_start_time,
2759 last_exit: snapshot.last_exit,
2760 health: snapshot.health,
2761 })
2762 }
2763
2764 #[cfg(test)]
2765 pub(crate) fn hold_snapshot_for_test(
2766 &self,
2767 acquired: std::sync::mpsc::Sender<()>,
2768 hold: Duration,
2769 ) -> std::thread::JoinHandle<()> {
2770 let snapshot = Arc::clone(&self.inner.snapshot);
2771 std::thread::spawn(move || {
2772 let _guard = snapshot.lock().expect("test snapshot lock is not poisoned");
2773 acquired
2774 .send(())
2775 .expect("test receiver waits for snapshot lock");
2776 std::thread::sleep(hold);
2777 })
2778 }
2779
2780 pub(crate) async fn running_image_agreement(&self) -> subc_control::RunningImageAgreement {
2781 let snapshot = match lock_snapshot(&self.inner.snapshot) {
2782 Ok(snapshot) => snapshot.clone(),
2783 Err(_) => {
2784 return subc_control::RunningImageAgreement::Unavailable {
2785 reason: subc_control::RunningImageUnavailableReason::NotRunning,
2786 };
2787 }
2788 };
2789 self.inner
2790 .provenance_probe
2791 .observe(
2792 snapshot.pid,
2793 snapshot.spawned_from.as_deref(),
2794 snapshot.spawned_file_identity,
2795 snapshot.process_start_time,
2796 )
2797 .await
2798 }
2799
2800 pub(crate) fn child_resource_usage(&self) -> subc_control::ChildResourceUsage {
2803 let (pid, start_time) = match lock_snapshot(&self.inner.snapshot) {
2804 Ok(snapshot) => (snapshot.pid, snapshot.process_start_time),
2805 Err(_) => {
2806 return subc_control::ChildResourceUsage::Unavailable {
2807 reason: subc_control::ChildResourceUnavailableReason::Unreadable,
2808 }
2809 }
2810 };
2811 crate::child_resources::read(pid, start_time)
2812 }
2813
2814 pub(crate) fn will_recover_after_connection_loss(&self) -> Result<bool, SuperviseError> {
2815 let mut snapshot = lock_snapshot(&self.inner.snapshot)?;
2816 Ok(match snapshot.state {
2817 ModuleState::Restarting => true,
2818 ModuleState::Failed | ModuleState::Disabled => false,
2819 _ => daemon_will_restart(&mut snapshot, &self.inner.restart_policy, Instant::now()),
2820 })
2821 }
2822
2823 #[cfg(test)]
2824 pub(crate) fn is_warming(&self) -> Result<bool, SuperviseError> {
2825 self.is_warming_with_snapshot_lock(None)
2826 }
2827
2828 pub(crate) fn is_warming_for_control(
2829 &self,
2830 caller: &'static str,
2831 ) -> Result<bool, SuperviseError> {
2832 self.is_warming_with_snapshot_lock(Some(caller))
2833 }
2834
2835 fn is_warming_with_snapshot_lock(
2836 &self,
2837 caller: Option<&'static str>,
2838 ) -> Result<bool, SuperviseError> {
2839 let snapshot = match caller {
2840 Some(caller) => {
2841 lock_snapshot_for_control(&self.inner.snapshot, &self.inner.module_id, caller)?
2842 }
2843 None => lock_snapshot(&self.inner.snapshot)?,
2844 }
2845 .clone();
2846 Ok(matches!(
2847 snapshot.state,
2848 ModuleState::Starting | ModuleState::Running | ModuleState::Restarting
2849 ))
2850 }
2851
2852 pub async fn drain(&self) -> Result<(), SuperviseError> {
2854 self.stop().await
2855 }
2856
2857 pub(crate) async fn retire(&self) -> Result<(), SuperviseError> {
2858 match self.state()? {
2859 ModuleState::Stopped | ModuleState::Failed => return Ok(()),
2860 ModuleState::Starting
2861 | ModuleState::Running
2862 | ModuleState::Unresponsive
2863 | ModuleState::Restarting
2864 | ModuleState::Draining
2865 | ModuleState::Disabled => {}
2866 }
2867
2868 let (reply_tx, reply_rx) = oneshot::channel();
2869 self.inner
2870 .commands
2871 .send(SupervisorCommand::Retire { reply: reply_tx })
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 stop(&self) -> Result<(), SuperviseError> {
2882 match self.state()? {
2883 ModuleState::Stopped | ModuleState::Failed => return Ok(()),
2884 ModuleState::Starting
2885 | ModuleState::Running
2886 | ModuleState::Unresponsive
2887 | ModuleState::Restarting
2888 | ModuleState::Draining
2889 | ModuleState::Disabled => {}
2890 }
2891
2892 let (reply_tx, reply_rx) = oneshot::channel();
2893 self.inner
2894 .commands
2895 .send(SupervisorCommand::Drain { reply: reply_tx })
2896 .await
2897 .map_err(|_| SuperviseError::CommandClosed {
2898 module_id: self.inner.module_id.clone(),
2899 })?;
2900 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2901 module_id: self.inner.module_id.clone(),
2902 })?
2903 }
2904
2905 pub async fn restart(&self, drain_timeout_ms: Option<u64>) -> Result<(), SuperviseError> {
2906 let received_at_generation = lock_snapshot(&self.inner.snapshot)?.spawn_generation;
2907 let (reply_tx, reply_rx) = oneshot::channel();
2908 self.inner
2909 .commands
2910 .send(SupervisorCommand::Restart {
2911 drain_timeout_ms,
2912 received_at_generation,
2913 queued_at: Instant::now(),
2914 reply: reply_tx,
2915 })
2916 .await
2917 .map_err(|_| SuperviseError::CommandClosed {
2918 module_id: self.inner.module_id.clone(),
2919 })?;
2920 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2921 module_id: self.inner.module_id.clone(),
2922 })?
2923 }
2924
2925 pub async fn swap(&self, ready_timeout: Option<Duration>) -> Result<(), SuperviseError> {
2930 let (reply_tx, reply_rx) = oneshot::channel();
2931 self.inner
2932 .commands
2933 .send(SupervisorCommand::Swap {
2934 ready_timeout,
2935 reply: reply_tx,
2936 })
2937 .await
2938 .map_err(|_| SuperviseError::CommandClosed {
2939 module_id: self.inner.module_id.clone(),
2940 })?;
2941 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2942 module_id: self.inner.module_id.clone(),
2943 })?
2944 }
2945
2946 pub async fn reload(&self) -> Result<(), SuperviseError> {
2947 let (reply_tx, reply_rx) = oneshot::channel();
2948 self.inner
2949 .commands
2950 .send(SupervisorCommand::Reload { reply: reply_tx })
2951 .await
2952 .map_err(|_| SuperviseError::CommandClosed {
2953 module_id: self.inner.module_id.clone(),
2954 })?;
2955 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2956 module_id: self.inner.module_id.clone(),
2957 })?
2958 }
2959
2960 pub async fn set_enabled(&self, enabled: bool) -> Result<bool, SuperviseError> {
2961 let (reply_tx, reply_rx) = oneshot::channel();
2962 self.inner
2963 .commands
2964 .send(SupervisorCommand::SetEnabled {
2965 enabled,
2966 reply: reply_tx,
2967 })
2968 .await
2969 .map_err(|_| SuperviseError::CommandClosed {
2970 module_id: self.inner.module_id.clone(),
2971 })?;
2972 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
2973 module_id: self.inner.module_id.clone(),
2974 })?
2975 }
2976
2977 pub(crate) fn declared_protocol(&self) -> Result<ModuleProtocol, SuperviseError> {
2982 Ok(self
2983 .inner
2984 .configuration
2985 .lock()
2986 .map_err(|_| SuperviseError::StatePoisoned {
2987 module_id: Some(self.inner.module_id.clone()),
2988 })?
2989 .spec
2990 .protocol)
2991 }
2992
2993 pub(crate) fn configuration(&self) -> Result<(ModuleSpec, HealthConfig), SuperviseError> {
2994 let configuration =
2995 self.inner
2996 .configuration
2997 .lock()
2998 .map_err(|_| SuperviseError::StatePoisoned {
2999 module_id: Some(self.inner.module_id.clone()),
3000 })?;
3001 Ok((configuration.spec.clone(), configuration.health))
3002 }
3003
3004 #[cfg(any(test, feature = "test-support"))]
3008 pub async fn update_spec_for_test(&self, spec: ModuleSpec) -> Result<(), SuperviseError> {
3009 let (_, health) = self.configuration()?;
3010 let drain_timeout_ms = u64::try_from(
3011 self.inner
3012 .effective_drain_timeout
3013 .lock()
3014 .unwrap_or_else(|poisoned| poisoned.into_inner())
3015 .as_millis(),
3016 )
3017 .ok();
3018 self.update_configuration(spec, health, drain_timeout_ms)
3019 .await
3020 }
3021
3022 pub(crate) async fn update_configuration(
3023 &self,
3024 spec: ModuleSpec,
3025 health: HealthConfig,
3026 drain_timeout_ms: Option<u64>,
3027 ) -> Result<(), SuperviseError> {
3028 if spec.module_id != self.inner.module_id {
3029 return Err(SuperviseError::InvalidSpec {
3030 reason: "a supervised module's module_id cannot be changed".to_string(),
3031 });
3032 }
3033 validate_spec(&spec)?;
3034 let (reply_tx, reply_rx) = oneshot::channel();
3035 self.inner
3036 .commands
3037 .send(SupervisorCommand::UpdateConfiguration {
3038 spec: spec.clone(),
3039 health,
3040 drain_timeout_ms,
3041 reply: reply_tx,
3042 })
3043 .await
3044 .map_err(|_| SuperviseError::CommandClosed {
3045 module_id: self.inner.module_id.clone(),
3046 })?;
3047 reply_rx.await.map_err(|_| SuperviseError::CommandClosed {
3048 module_id: self.inner.module_id.clone(),
3049 })?;
3050 let mut configuration =
3051 self.inner
3052 .configuration
3053 .lock()
3054 .map_err(|_| SuperviseError::StatePoisoned {
3055 module_id: Some(self.inner.module_id.clone()),
3056 })?;
3057 configuration.spec = spec;
3058 configuration.health = health;
3059 Ok(())
3060 }
3061}
3062
3063impl Drop for SupervisedModuleInner {
3064 fn drop(&mut self) {
3065 let Ok(mut monitor) = self.monitor.lock() else {
3066 return;
3067 };
3068 if let Some(monitor) = monitor.as_ref().filter(|monitor| !monitor.is_finished()) {
3069 let _ = update_snapshot(&self.snapshot, Some(&self.module_id), |state| {
3070 state.state = ModuleState::Stopped;
3071 clear_current_process_facts(state);
3072 });
3073 monitor.abort();
3074 }
3075 let _ = monitor.take();
3076 }
3077}
3078
3079#[derive(Debug)]
3080enum SupervisorCommand {
3081 Drain {
3082 reply: oneshot::Sender<Result<(), SuperviseError>>,
3083 },
3084 Retire {
3085 reply: oneshot::Sender<Result<(), SuperviseError>>,
3086 },
3087 Restart {
3088 drain_timeout_ms: Option<u64>,
3093 received_at_generation: u64,
3097 queued_at: Instant,
3100 reply: oneshot::Sender<Result<(), SuperviseError>>,
3101 },
3102 Reload {
3103 reply: oneshot::Sender<Result<(), SuperviseError>>,
3104 },
3105 SetEnabled {
3106 enabled: bool,
3107 reply: oneshot::Sender<Result<bool, SuperviseError>>,
3108 },
3109 UpdateConfiguration {
3110 spec: ModuleSpec,
3111 health: HealthConfig,
3112 drain_timeout_ms: Option<u64>,
3115 reply: oneshot::Sender<()>,
3116 },
3117 Swap {
3118 ready_timeout: Option<Duration>,
3121 reply: oneshot::Sender<Result<(), SuperviseError>>,
3123 },
3124}
3125
3126#[derive(Debug)]
3127pub enum SuperviseError {
3128 InvalidSpec {
3129 reason: String,
3130 },
3131 Spawn {
3132 program: PathBuf,
3133 source: io::Error,
3134 cgroup_path: Option<PathBuf>,
3135 },
3136 Cgroup {
3137 module_id: String,
3138 source: io::Error,
3139 },
3140 LaunchNonce {
3143 reason: String,
3144 },
3145 Wait {
3146 module_id: String,
3147 source: io::Error,
3148 },
3149 Kill {
3150 module_id: String,
3151 source: io::Error,
3152 },
3153 Forwarding(ForwardingError),
3154 Registry(RegistryError),
3155 ReloadUnavailable {
3156 module_id: String,
3157 reason: String,
3158 },
3159 Disabled {
3164 module_id: String,
3165 },
3166 ReloadFailed {
3167 module_id: String,
3168 reason: String,
3169 },
3170 RegistrationStillActive {
3171 module_id: String,
3172 waited: Duration,
3173 },
3174 StatePoisoned {
3175 module_id: Option<String>,
3176 },
3177 CommandClosed {
3178 module_id: String,
3179 },
3180 SwapInProgress {
3184 module_id: String,
3185 },
3186 SwapRefused {
3188 module_id: String,
3189 reason: SwapRefusal,
3190 },
3191 SwapFailed {
3195 module_id: String,
3196 arm: SwapFailureArm,
3197 detail: String,
3198 candidate_exit: Option<ExitReport>,
3201 },
3202}
3203
3204#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3206pub enum SwapRefusal {
3207 OverlapExclusive,
3209 NotRegistered,
3212 ProtocolNone,
3215 NotConfigured,
3218 AlreadySwapping,
3220}
3221
3222impl SwapRefusal {
3223 pub fn as_str(self) -> &'static str {
3224 match self {
3225 Self::OverlapExclusive => "overlap_exclusive",
3226 Self::NotRegistered => "not_registered",
3227 Self::ProtocolNone => "protocol_none",
3228 Self::NotConfigured => "not_configured",
3229 Self::AlreadySwapping => "already_swapping",
3230 }
3231 }
3232}
3233
3234#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3237pub enum SwapFailureArm {
3238 SpawnFailed,
3240 NeverRegistered,
3242 NeverReady,
3244 CandidateExited,
3246 CandidateUnhealthy,
3248 Interrupted,
3252 CutoverLost,
3257}
3258
3259impl SwapFailureArm {
3260 pub fn as_str(self) -> &'static str {
3261 match self {
3262 Self::SpawnFailed => "spawn_failed",
3263 Self::NeverRegistered => "never_registered",
3264 Self::NeverReady => "never_ready",
3265 Self::CandidateExited => "candidate_exited",
3266 Self::CandidateUnhealthy => "candidate_unhealthy",
3267 Self::Interrupted => "interrupted",
3268 Self::CutoverLost => "cutover_lost",
3269 }
3270 }
3271}
3272
3273impl fmt::Display for SuperviseError {
3274 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3275 match self {
3276 Self::InvalidSpec { reason } => write!(f, "invalid module spec: {reason}"),
3277 Self::Spawn {
3278 program,
3279 source,
3280 cgroup_path: Some(cgroup_path),
3281 } => write!(
3282 f,
3283 "failed to place module in cgroup '{}' while spawning '{}': {source}",
3284 cgroup_path.display(),
3285 program.display()
3286 ),
3287 Self::Spawn {
3288 program,
3289 source,
3290 cgroup_path: None,
3291 } => write!(
3292 f,
3293 "failed to spawn module '{}': {source}",
3294 program.display()
3295 ),
3296 Self::Cgroup { module_id, source } => {
3297 write!(
3298 f,
3299 "failed to prepare cgroup for module '{module_id}': {source}"
3300 )
3301 }
3302 Self::LaunchNonce { reason } => {
3303 write!(
3304 f,
3305 "failed to generate reserved-module launch nonce: {reason}"
3306 )
3307 }
3308 Self::Wait { module_id, source } => {
3309 write!(f, "failed to wait for module '{module_id}': {source}")
3310 }
3311 Self::Kill { module_id, source } => {
3312 write!(f, "failed to kill module '{module_id}': {source}")
3313 }
3314 Self::Forwarding(err) => write!(f, "forwarding error: {err}"),
3315 Self::Registry(err) => write!(f, "registry error: {err}"),
3316 Self::ReloadUnavailable { module_id, reason } => {
3317 write!(f, "reload unavailable for module '{module_id}': {reason}")
3318 }
3319 Self::Disabled { module_id } => {
3320 write!(
3321 f,
3322 "module '{module_id}' is disabled; enable it before restart or reload"
3323 )
3324 }
3325 Self::ReloadFailed { module_id, reason } => {
3326 write!(f, "reload failed for module '{module_id}': {reason}")
3327 }
3328 Self::RegistrationStillActive { module_id, waited } => write!(
3329 f,
3330 "module '{module_id}' registration remained active after waiting {waited:?}"
3331 ),
3332 Self::StatePoisoned { module_id } => match module_id {
3333 Some(module_id) => {
3334 write!(f, "supervisor state for module '{module_id}' was poisoned")
3335 }
3336 None => write!(f, "supervisor state was poisoned"),
3337 },
3338 Self::CommandClosed { module_id } => {
3339 write!(
3340 f,
3341 "supervisor command channel for module '{module_id}' is closed"
3342 )
3343 }
3344 Self::SwapInProgress { module_id } => write!(
3345 f,
3346 "module '{module_id}' is being swapped; retry once the swap has cut over or failed, or stop the module to abort the swap"
3347 ),
3348 Self::SwapRefused { module_id, reason } => match reason {
3349 SwapRefusal::OverlapExclusive => write!(
3350 f,
3351 "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"
3352 ),
3353 SwapRefusal::NotRegistered => write!(
3354 f,
3355 "module '{module_id}' is not registered, so there is no serving process to keep while a replacement warms; use a plain restart"
3356 ),
3357 SwapRefusal::ProtocolNone => write!(
3358 f,
3359 "module '{module_id}' is protocol: \"none\" and never registers, so a swap could never see its replacement become ready; use a plain restart"
3360 ),
3361 SwapRefusal::NotConfigured => write!(
3362 f,
3363 "module '{module_id}' cannot be swapped: the supervisor was built without the forwarding table or shared handle a swap needs"
3364 ),
3365 SwapRefusal::AlreadySwapping => {
3366 write!(f, "module '{module_id}' is already being swapped")
3367 }
3368 },
3369 Self::SwapFailed {
3370 module_id,
3371 arm,
3372 detail,
3373 ..
3374 } => write!(
3375 f,
3376 "swap of module '{module_id}' failed ({}): {detail}; the running process was left serving",
3377 arm.as_str()
3378 ),
3379 }
3380 }
3381}
3382
3383impl Error for SuperviseError {
3384 fn source(&self) -> Option<&(dyn Error + 'static)> {
3385 match self {
3386 Self::Spawn { source, .. }
3387 | Self::Cgroup { source, .. }
3388 | Self::Wait { source, .. }
3389 | Self::Kill { source, .. } => Some(source),
3390 Self::Forwarding(err) => Some(err),
3391 Self::Registry(err) => Some(err),
3392 Self::LaunchNonce { .. }
3393 | Self::InvalidSpec { .. }
3394 | Self::ReloadUnavailable { .. }
3395 | Self::Disabled { .. }
3396 | Self::ReloadFailed { .. }
3397 | Self::RegistrationStillActive { .. }
3398 | Self::StatePoisoned { .. }
3399 | Self::CommandClosed { .. }
3400 | Self::SwapInProgress { .. }
3401 | Self::SwapRefused { .. }
3402 | Self::SwapFailed { .. } => None,
3403 }
3404 }
3405}
3406
3407pub(crate) fn validate_spec(spec: &ModuleSpec) -> Result<(), SuperviseError> {
3408 if spec.module_id.trim().is_empty() {
3409 return Err(SuperviseError::InvalidSpec {
3410 reason: "module_id must not be empty".to_string(),
3411 });
3412 }
3413
3414 Ok(())
3415}
3416
3417#[derive(Debug, Default)]
3418struct HealthProbeRuntime {
3419 registered_connection: Option<crate::ConnectionId>,
3420 advertised: bool,
3421 next_probe_at: Option<Instant>,
3422 probe_index: u64,
3423}
3424
3425impl HealthProbeRuntime {
3426 fn refresh_registration(
3427 &mut self,
3428 spec: &ModuleSpec,
3429 runtime: &SupervisorRuntimeConfig,
3430 registry: &Registry,
3431 snapshot: &SharedSnapshot,
3432 ) {
3433 if spec.protocol == ModuleProtocol::None {
3445 self.registered_connection = None;
3446 self.advertised = false;
3447 self.next_probe_at = None;
3448 return;
3449 }
3450
3451 let registration = match registry.get_module(&spec.module_id) {
3452 Ok(registration) => registration,
3453 Err(err) => {
3454 warn!(module_id = %spec.module_id, error = %err, "health prober could not read registry");
3455 self.advertised = false;
3456 self.next_probe_at = None;
3457 return;
3458 }
3459 };
3460
3461 let Some(registration) = registration else {
3462 self.registered_connection = None;
3463 self.advertised = false;
3464 self.next_probe_at = None;
3465 return;
3466 };
3467
3468 let advertised = registration
3469 .control_ops
3470 .iter()
3471 .any(|op| op == MODULE_CONTROL_OP_HEALTH_CHECK);
3472 if !advertised {
3473 self.registered_connection = Some(registration.connection_id);
3474 self.advertised = false;
3475 self.next_probe_at = None;
3476 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3477 state.health.status = SupervisorHealthStatus::Unknown;
3478 state.health.consecutive_failures = 0;
3479 state.health.last_probe_ms = None;
3480 state.health.detail = None;
3481 state.health.metrics = None;
3482 });
3483 return;
3484 }
3485
3486 let reregistered = self.registered_connection != Some(registration.connection_id);
3487 self.registered_connection = Some(registration.connection_id);
3488 self.advertised = true;
3489 if reregistered || self.next_probe_at.is_none() {
3490 self.probe_index = 0;
3491 self.next_probe_at = Some(
3492 Instant::now() + jittered_health_delay(&spec.module_id, 0, runtime.health.cadence),
3493 );
3494 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3495 state.health.status = SupervisorHealthStatus::Unknown;
3496 state.health.consecutive_failures = 0;
3497 state.health.detail = None;
3498 state.health.metrics = None;
3499 });
3500 }
3501 }
3502
3503 fn wake_after(&self) -> Duration {
3504 if !self.advertised {
3505 return REGISTRY_RELEASE_POLL;
3506 }
3507 self.next_probe_at
3508 .map(|next| next.saturating_duration_since(Instant::now()))
3509 .unwrap_or(REGISTRY_RELEASE_POLL)
3510 }
3511
3512 fn due(&self) -> bool {
3513 self.advertised
3514 && self
3515 .next_probe_at
3516 .is_some_and(|next| Instant::now() >= next)
3517 }
3518
3519 fn schedule_next(&mut self, spec: &ModuleSpec, cadence: Duration) {
3520 self.probe_index = self.probe_index.wrapping_add(1);
3521 self.next_probe_at = Some(
3522 Instant::now() + jittered_health_delay(&spec.module_id, self.probe_index, cadence),
3523 );
3524 }
3525}
3526
3527#[derive(Debug)]
3562enum HealthProbeEvidence {
3563 LaneDead,
3565 NoAnswer,
3567 BadAnswer,
3569 Misconfigured,
3571}
3572
3573#[derive(Debug)]
3574struct HealthProbeError {
3575 evidence: HealthProbeEvidence,
3576 message: String,
3577}
3578
3579impl HealthProbeError {
3580 fn lane_dead(message: impl Into<String>) -> Self {
3581 Self::with(HealthProbeEvidence::LaneDead, message)
3582 }
3583
3584 fn no_answer(message: impl Into<String>) -> Self {
3585 Self::with(HealthProbeEvidence::NoAnswer, message)
3586 }
3587
3588 fn bad_answer(message: impl Into<String>) -> Self {
3589 Self::with(HealthProbeEvidence::BadAnswer, message)
3590 }
3591
3592 fn misconfigured(message: impl Into<String>) -> Self {
3593 Self::with(HealthProbeEvidence::Misconfigured, message)
3594 }
3595
3596 fn with(evidence: HealthProbeEvidence, message: impl Into<String>) -> Self {
3597 Self {
3598 evidence,
3599 message: message.into(),
3600 }
3601 }
3602
3603 #[allow(dead_code)]
3617 fn is_proof_of_death(&self) -> bool {
3618 matches!(self.evidence, HealthProbeEvidence::LaneDead)
3619 }
3620
3621 fn label(&self) -> &'static str {
3629 match self.evidence {
3630 HealthProbeEvidence::LaneDead => "lane-dead",
3631 HealthProbeEvidence::NoAnswer => "no-answer",
3632 HealthProbeEvidence::BadAnswer => "bad-answer",
3633 HealthProbeEvidence::Misconfigured => "daemon-misconfigured",
3634 }
3635 }
3636}
3637
3638impl fmt::Display for HealthProbeError {
3639 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3640 f.write_str(&self.message)
3641 }
3642}
3643
3644async fn run_health_probe_cycle(
3645 spec: &ModuleSpec,
3646 runtime: &SupervisorRuntimeConfig,
3647 registry: &Registry,
3648 process_liveness: &SupervisorProcessLiveness,
3649 snapshot: &SharedSnapshot,
3650 child: &mut Option<SupervisedChild>,
3651) {
3652 let now_ms = unix_ms_now();
3653 match probe_module_health(&spec.module_id, runtime, None).await {
3654 Ok(report) => {
3655 handle_health_report(
3656 spec,
3657 runtime,
3658 registry,
3659 process_liveness,
3660 snapshot,
3661 child,
3662 report,
3663 now_ms,
3664 )
3665 .await;
3666 }
3667 Err(err) => {
3668 handle_health_probe_failure(
3669 spec,
3670 runtime,
3671 registry,
3672 process_liveness,
3673 snapshot,
3674 child,
3675 err,
3676 now_ms,
3677 )
3678 .await;
3679 }
3680 }
3681}
3682
3683async fn probe_module_health(
3684 module_id: &str,
3685 runtime: &SupervisorRuntimeConfig,
3686 drain_deadline: 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(drain_deadline) = drain_deadline {
3696 deadline = deadline.min(drain_deadline);
3697 }
3698 let pending = if drain_deadline.is_some() {
3699 forwarding.begin_drain_health_probe_rpc_for(
3700 module_id,
3701 MODULE_CONTROL_OP_HEALTH_CHECK,
3702 probe_started_at,
3703 deadline,
3704 )
3705 } else {
3706 forwarding.begin_health_probe_rpc_for(
3707 module_id,
3708 MODULE_CONTROL_OP_HEALTH_CHECK,
3709 probe_started_at,
3710 deadline,
3711 )
3712 }
3713 .map_err(|err| {
3714 HealthProbeError::lane_dead(format!("failed to begin health.check RPC: {err}"))
3717 })?;
3718 await_health_probe(forwarding, pending, deadline, runtime.health.deadline).await
3719}
3720
3721async fn probe_endpoint_health(
3728 endpoint: crate::ModuleEndpointId,
3729 runtime: &SupervisorRuntimeConfig,
3730 deadline_cap: Option<Instant>,
3731) -> Result<HealthReport, HealthProbeError> {
3732 let Some(forwarding) = runtime.forwarding.as_ref() else {
3733 return Err(HealthProbeError::misconfigured(
3734 "supervisor was not configured with a forwarding table",
3735 ));
3736 };
3737 let probe_started_at = Instant::now();
3738 let mut deadline = probe_started_at + runtime.health.deadline;
3739 if let Some(cap) = deadline_cap {
3740 deadline = deadline.min(cap);
3741 }
3742 let pending = forwarding
3743 .begin_endpoint_health_probe_rpc_for(
3744 endpoint,
3745 MODULE_CONTROL_OP_HEALTH_CHECK,
3746 probe_started_at,
3747 deadline,
3748 )
3749 .map_err(|err| {
3750 HealthProbeError::lane_dead(format!("failed to begin health.check RPC: {err}"))
3751 })?;
3752 await_health_probe(forwarding, pending, deadline, runtime.health.deadline).await
3753}
3754
3755async fn await_health_probe(
3757 forwarding: &ForwardingTable,
3758 pending: PendingModuleControlRpc,
3759 deadline: Instant,
3760 probe_budget: Duration,
3761) -> Result<HealthReport, HealthProbeError> {
3762 let PendingModuleControlRpc {
3763 endpoint,
3764 module_sink,
3765 negotiated_ver,
3766 corr,
3767 receiver,
3768 } = pending;
3769 let body = serde_json::to_vec(&ModuleControlRequest::HealthCheck {}).map_err(|err| {
3770 HealthProbeError::misconfigured(format!("failed to encode health.check: {err}"))
3771 })?;
3772 let frame = Frame::build_with_version(
3773 negotiated_ver,
3774 FrameType::Request,
3775 control_flags(),
3776 0,
3777 0,
3778 corr,
3779 body,
3780 )
3781 .map_err(|err| {
3782 HealthProbeError::misconfigured(format!("failed to build health.check frame: {err}"))
3783 })?;
3784
3785 match timeout_at(deadline, module_sink.send(frame)).await {
3791 Ok(Ok(())) => {}
3792 Ok(Err(err)) => {
3793 let _ = forwarding.cancel_module_control_rpc(endpoint, corr);
3794 return Err(HealthProbeError::lane_dead(format!(
3797 "failed to send health.check: {err}"
3798 )));
3799 }
3800 Err(_elapsed) => {
3801 let _ = forwarding.cancel_module_control_rpc(endpoint, corr);
3802 return Err(HealthProbeError::no_answer(
3806 "health.check send timed out before enqueue (module egress full)",
3807 ));
3808 }
3809 }
3810
3811 match timeout_at(deadline, receiver).await {
3812 Ok(Ok(ModuleControlRpcOutcome::Response(response))) => {
3816 response.health_report().ok_or_else(|| {
3817 HealthProbeError::bad_answer("health.check RPC returned a non-health response")
3818 })
3819 }
3820 Ok(Ok(ModuleControlRpcOutcome::Rejected(body))) => Err(HealthProbeError::bad_answer(
3821 format!("health.check rejected: {}", body.message),
3822 )),
3823 Ok(Ok(ModuleControlRpcOutcome::ModuleGone(message))) => {
3824 Err(HealthProbeError::lane_dead(message))
3825 }
3826 Ok(Ok(ModuleControlRpcOutcome::MalformedResponse(message))) => {
3827 Err(HealthProbeError::bad_answer(message))
3828 }
3829 Ok(Ok(ModuleControlRpcOutcome::UnexpectedOp { expected, actual })) => {
3830 Err(HealthProbeError::bad_answer(format!(
3831 "expected module-control op '{expected}', got '{actual}'"
3832 )))
3833 }
3834 Ok(Ok(ModuleControlRpcOutcome::DeadlineElapsed)) => Err(HealthProbeError::bad_answer(
3838 "module answered health.check after its daemon deadline",
3839 )),
3840 Ok(Err(_)) => Err(HealthProbeError::misconfigured(
3841 "health.check waiter was canceled before the module responded",
3842 )),
3843 Err(_) => {
3844 let _ = forwarding.tombstone_health_probe_rpc(endpoint, corr);
3845 Err(HealthProbeError::no_answer(format!(
3846 "module did not answer health.check within {probe_budget:?}"
3847 )))
3848 }
3849 }
3850}
3851
3852#[allow(clippy::too_many_arguments)]
3853async fn handle_health_report(
3854 spec: &ModuleSpec,
3855 runtime: &SupervisorRuntimeConfig,
3856 registry: &Registry,
3857 process_liveness: &SupervisorProcessLiveness,
3858 snapshot: &SharedSnapshot,
3859 child: &mut Option<SupervisedChild>,
3860 report: HealthReport,
3861 now_ms: u64,
3862) {
3863 let status = supervisor_health_status(report.status);
3864 let detail = report.detail.clone();
3865 let metrics = truncate_health_metrics(report.metrics);
3866 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3867 state.health.status = status;
3868 state.health.last_probe_ms = Some(now_ms);
3869 state.health.detail = detail.clone();
3870 state.health.metrics = metrics.clone();
3871 state.health.consecutive_failures = 0;
3872 });
3873
3874 let action = match report.status {
3875 HealthStatus::Ok => return,
3876 HealthStatus::Degraded => runtime.health.on_degraded,
3877 HealthStatus::Failing => runtime.health.on_failing,
3878 };
3879 apply_l3_health_action(
3880 spec,
3881 runtime,
3882 registry,
3883 process_liveness,
3884 snapshot,
3885 child,
3886 status,
3887 detail.as_deref(),
3888 action,
3889 now_ms,
3890 )
3891 .await;
3892}
3893
3894#[allow(clippy::too_many_arguments)]
3895async fn handle_health_probe_failure(
3896 spec: &ModuleSpec,
3897 runtime: &SupervisorRuntimeConfig,
3898 registry: &Registry,
3899 process_liveness: &SupervisorProcessLiveness,
3900 snapshot: &SharedSnapshot,
3901 child: &mut Option<SupervisedChild>,
3902 err: HealthProbeError,
3903 now_ms: u64,
3904) {
3905 let threshold = runtime.health.failure_threshold.max(1);
3906 let mut failures = 0;
3907 let detail = format!("[{}] {err}", err.label());
3912 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3913 state.health.last_probe_ms = Some(now_ms);
3914 state.health.consecutive_failures = state.health.consecutive_failures.saturating_add(1);
3915 state.health.detail = Some(detail.clone());
3916 state.health.metrics = None;
3917 failures = state.health.consecutive_failures;
3918 });
3919
3920 if failures < threshold {
3921 warn!(
3922 module_id = %spec.module_id,
3923 consecutive_failures = failures,
3924 threshold,
3925 evidence = err.label(),
3926 detail = %detail,
3927 "health.check probe failed"
3928 );
3929 return;
3930 }
3931
3932 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
3933 state.state = ModuleState::Unresponsive;
3934 state.health.status = SupervisorHealthStatus::Unresponsive;
3935 });
3936 if runtime.health.critical {
3940 error!(
3941 module_id = %spec.module_id,
3942 status = "unresponsive",
3943 evidence = err.label(),
3944 detail = %detail,
3945 "critical module health alert"
3946 );
3947 } else {
3948 warn!(
3949 module_id = %spec.module_id,
3950 status = "unresponsive",
3951 evidence = err.label(),
3952 detail = %detail,
3953 "module health threshold breached"
3954 );
3955 }
3956 if let Err(err) = health_restart_child(
3957 spec,
3958 runtime,
3959 registry,
3960 process_liveness,
3961 snapshot,
3962 child,
3963 SupervisorHealthStatus::Unresponsive,
3964 Some(&detail),
3965 now_ms,
3966 )
3967 .await
3968 {
3969 error!(module_id = %spec.module_id, error = %err, "health-triggered restart failed");
3970 }
3971}
3972
3973#[allow(clippy::too_many_arguments)]
3974async fn apply_l3_health_action(
3975 spec: &ModuleSpec,
3976 runtime: &SupervisorRuntimeConfig,
3977 registry: &Registry,
3978 process_liveness: &SupervisorProcessLiveness,
3979 snapshot: &SharedSnapshot,
3980 child: &mut Option<SupervisedChild>,
3981 status: SupervisorHealthStatus,
3982 detail: Option<&str>,
3983 action: HealthAction,
3984 now_ms: u64,
3985) {
3986 record_health_action(snapshot, &spec.module_id, action.to_string(), now_ms);
3987 match action {
3988 HealthAction::Report => {
3989 info!(
3990 module_id = %spec.module_id,
3991 status = ?status,
3992 detail,
3993 "module reported non-ok health"
3994 );
3995 }
3996 HealthAction::Alert => {
3997 error!(
3998 module_id = %spec.module_id,
3999 status = ?status,
4000 detail,
4001 "module health alert"
4002 );
4003 }
4004 HealthAction::Restart => {
4005 if let Err(err) = health_restart_child(
4006 spec,
4007 runtime,
4008 registry,
4009 process_liveness,
4010 snapshot,
4011 child,
4012 status,
4013 detail,
4014 now_ms,
4015 )
4016 .await
4017 {
4018 error!(module_id = %spec.module_id, error = %err, "health-triggered restart failed");
4019 }
4020 }
4021 }
4022}
4023
4024#[allow(clippy::too_many_arguments)]
4025async fn health_restart_child(
4026 spec: &ModuleSpec,
4027 runtime: &SupervisorRuntimeConfig,
4028 registry: &Registry,
4029 process_liveness: &SupervisorProcessLiveness,
4030 snapshot: &SharedSnapshot,
4031 child: &mut Option<SupervisedChild>,
4032 status: SupervisorHealthStatus,
4033 detail: Option<&str>,
4034 now_ms: u64,
4035) -> Result<(), SuperviseError> {
4036 let (enabled, schedule) = {
4037 let mut state = lock_snapshot(snapshot)?;
4038 let enabled = state.enabled;
4039 let schedule = if enabled {
4040 state.next_crash_restart(&runtime.restart_policy, Instant::now())
4041 } else {
4042 None
4043 };
4044 (enabled, schedule)
4045 };
4046
4047 if !enabled {
4048 return Err(SuperviseError::Disabled {
4049 module_id: spec.module_id.clone(),
4050 });
4051 }
4052
4053 if schedule.is_none() {
4054 record_health_action(snapshot, &spec.module_id, "disabled".to_string(), now_ms);
4055 error!(
4056 module_id = %spec.module_id,
4057 status = ?status,
4058 detail,
4059 max_restarts = runtime.restart_policy.max_restarts,
4060 window_secs = runtime.restart_policy.window.as_secs(),
4061 "health restart budget exhausted; disabling module"
4062 );
4063 let stop_notice = begin_forwarding_drain_if_configured(
4064 spec,
4065 runtime,
4066 registry,
4067 snapshot,
4068 Some(false),
4069 RouteCloseReason::Disable,
4070 )
4071 .await?;
4072 drain_optional_child(
4073 &spec.module_id,
4074 spec.protocol,
4075 stop_notice,
4076 registry,
4077 snapshot,
4078 &runtime.terminal_ring,
4079 &runtime.spawn_events,
4080 child,
4081 runtime.drain_timeout,
4082 ModuleState::Disabled,
4083 Some(false),
4084 )
4085 .await?;
4086 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4087 return Ok(());
4088 }
4089
4090 let schedule = schedule.expect("a health restart must have a crash-restart schedule");
4091 let mut restart_count = 0;
4092 update_snapshot(snapshot, Some(&spec.module_id), |state| {
4093 restart_count = state.crash_restarts.len();
4094 state.state = ModuleState::Unresponsive;
4095 state.health.status = status;
4096 state.health.last_action = Some(HealthAction::Restart.to_string());
4097 state.health.last_action_ms = Some(now_ms);
4098 })?;
4099 warn!(
4100 module_id = %spec.module_id,
4101 status = ?status,
4102 detail,
4103 restart_count,
4104 restart_in_window = schedule.restart_in_window,
4105 delay_ms = schedule.delay.as_millis() as u64,
4106 "health-triggered module restart"
4107 );
4108
4109 let stop_notice = begin_forwarding_drain_if_configured(
4110 spec,
4111 runtime,
4112 registry,
4113 snapshot,
4114 Some(true),
4115 RouteCloseReason::Restart,
4116 )
4117 .await?;
4118 drain_optional_child(
4119 &spec.module_id,
4120 spec.protocol,
4121 stop_notice,
4122 registry,
4123 snapshot,
4124 &runtime.terminal_ring,
4125 &runtime.spawn_events,
4126 child,
4127 runtime.drain_timeout,
4128 ModuleState::Restarting,
4129 Some(true),
4130 )
4131 .await?;
4132 sleep(schedule.delay).await;
4133 if !respawn_still_pending(snapshot) {
4137 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4138 return Ok(());
4139 }
4140 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
4141 match spawn_and_mark_running(spec, runtime, snapshot) {
4142 Ok(next_child) => {
4143 *child = Some(next_child);
4144 Ok(())
4145 }
4146 Err(err) => {
4147 fail_snapshot(snapshot, Some(&spec.module_id), None);
4148 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4149 *child = None;
4150 Err(err)
4151 }
4152 }
4153}
4154
4155fn record_health_action(snapshot: &SharedSnapshot, module_id: &str, action: String, now_ms: u64) {
4156 let _ = update_snapshot(snapshot, Some(module_id), |state| {
4157 state.health.last_action = Some(action);
4158 state.health.last_action_ms = Some(now_ms);
4159 });
4160}
4161
4162fn supervisor_health_status(status: HealthStatus) -> SupervisorHealthStatus {
4163 match status {
4164 HealthStatus::Ok => SupervisorHealthStatus::Ok,
4165 HealthStatus::Degraded => SupervisorHealthStatus::Degraded,
4166 HealthStatus::Failing => SupervisorHealthStatus::Failing,
4167 }
4168}
4169
4170fn truncate_health_metrics(metrics: Option<Value>) -> Option<Value> {
4182 let metrics = metrics?;
4183 match serde_json::to_vec(&metrics) {
4184 Ok(encoded) if encoded.len() > MAX_HEALTH_METRICS_BYTES => Some(serde_json::json!({
4185 "truncated": true,
4186 "original_bytes": encoded.len(),
4187 })),
4188 Ok(_) | Err(_) => Some(metrics),
4189 }
4190}
4191
4192fn jittered_health_delay(module_id: &str, probe_index: u64, cadence: Duration) -> Duration {
4198 if cadence.is_zero() {
4199 return Duration::ZERO;
4200 }
4201 let cadence_ms = cadence.as_millis() as u64;
4202 if cadence_ms == 0 {
4218 return cadence;
4219 }
4220 let jitter_span = (cadence_ms / 10).max(1);
4235 let hash = module_id.as_bytes().iter().fold(
4236 probe_index.wrapping_mul(0x9E37_79B9_7F4A_7C15),
4237 |acc, byte| {
4238 acc.wrapping_mul(1099511628211)
4239 .wrapping_add(u64::from(*byte))
4240 },
4241 );
4242 cadence + Duration::from_millis(hash % jitter_span)
4243}
4244
4245#[cfg(test)]
4246mod tests {
4247 use super::*;
4248
4249 #[test]
4250 fn readding_a_module_clears_its_rescan_removal_tombstone() {
4251 let handle = SupervisorHandle::new();
4252 let module_id = "readded-tombstone";
4253 handle.record_rescan_removal(module_id);
4254 assert!(handle.removal_tombstone_age_ms(module_id).is_some());
4255
4256 handle.apply_identity_configuration(&ModuleSpec {
4257 module_id: module_id.to_string(),
4258 program: PathBuf::from("/test/module"),
4259 args: Vec::new(),
4260 env: Vec::new(),
4261 reserved: false,
4262 reserved_prefixes: Vec::new(),
4263 protocol: ModuleProtocol::Subc,
4264 overlap: Default::default(),
4265 });
4266
4267 assert!(
4268 handle.removal_tombstone_age_ms(module_id).is_none(),
4269 "a re-added module must not retain a stale removal tombstone"
4270 );
4271 }
4272
4273 fn stale_process_snapshot(state: ModuleState, enabled: bool) -> SharedSnapshot {
4274 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::new(state, enabled)));
4275 update_snapshot(&snapshot, Some("stale-process-facts"), |snapshot| {
4276 snapshot.process_alive = true;
4277 snapshot.pid = Some(41);
4278 snapshot.spawned_at_ms = Some(42);
4279 snapshot.spawned_from = Some(PathBuf::from("/spawned/module"));
4280 snapshot.spawned_file_identity = Some(SpawnedFileIdentity {
4281 device: 43,
4282 inode: 44,
4283 });
4284 })
4285 .unwrap();
4286 snapshot
4287 }
4288
4289 fn assert_snapshot_process_facts_cleared(snapshot: &SharedSnapshot) {
4290 let snapshot = lock_snapshot(snapshot).unwrap();
4291 assert!(!snapshot.process_alive);
4292 assert_eq!(snapshot.pid, None);
4293 assert_eq!(snapshot.spawned_at_ms, None);
4294 assert_eq!(snapshot.spawned_from, None);
4295 assert_eq!(snapshot.spawned_file_identity, None);
4296 }
4297
4298 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4299 async fn failed_enable_spawn_clears_preexisting_current_process_facts() {
4300 let supervisor = Supervisor::default();
4301 let mut runtime = supervisor.runtime_config();
4302 runtime.test_seed_stale_facts_before_enable_spawn = true;
4303 let snapshot = stale_process_snapshot(ModuleState::Disabled, false);
4304 let mut child = None;
4305 let spec = ModuleSpec {
4306 module_id: "failed-enable-clears-facts".to_string(),
4307 program: PathBuf::from("/definitely/missing/failed-enable-module"),
4308 args: Vec::new(),
4309 env: Vec::new(),
4310 reserved: false,
4311 reserved_prefixes: Vec::new(),
4312 protocol: ModuleProtocol::Subc,
4313 overlap: Default::default(),
4314 };
4315
4316 let result = set_child_enabled(
4317 &spec,
4318 &runtime,
4319 &supervisor.registry,
4320 &supervisor.process_liveness,
4321 &snapshot,
4322 &mut child,
4323 true,
4324 )
4325 .await;
4326
4327 assert!(matches!(result, Err(SuperviseError::Spawn { .. })));
4328 assert_eq!(lock_snapshot(&snapshot).unwrap().state, ModuleState::Failed);
4329 assert_snapshot_process_facts_cleared(&snapshot);
4330 }
4331
4332 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4333 async fn failed_reload_spawn_clears_current_process_facts() {
4334 let supervisor = Supervisor::default();
4335 let mut runtime = supervisor.runtime_config();
4336 runtime.restart_policy = RestartPolicy::new(0, Duration::ZERO);
4337 let snapshot = stale_process_snapshot(ModuleState::Running, true);
4338 let mut child = None;
4339 let spec = ModuleSpec {
4340 module_id: "failed-reload-clears-facts".to_string(),
4341 program: PathBuf::from("/unused/failed-reload-module"),
4342 args: Vec::new(),
4343 env: Vec::new(),
4344 reserved: false,
4345 reserved_prefixes: Vec::new(),
4346 protocol: ModuleProtocol::Subc,
4347 overlap: Default::default(),
4348 };
4349
4350 let result = handle_reload_spawn_failure(
4351 &spec,
4352 &runtime,
4353 &supervisor.process_liveness,
4354 &snapshot,
4355 &mut child,
4356 "forced reload spawn failure".to_string(),
4357 )
4358 .await;
4359
4360 assert!(matches!(result, Err(SuperviseError::ReloadFailed { .. })));
4361 assert_eq!(lock_snapshot(&snapshot).unwrap().state, ModuleState::Failed);
4362 assert_snapshot_process_facts_cleared(&snapshot);
4363 }
4364
4365 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4366 async fn dropping_a_module_with_an_active_monitor_clears_current_process_facts() {
4367 let supervisor = Supervisor::default();
4368 let snapshot = stale_process_snapshot(ModuleState::Running, true);
4369 let module = supervisor.supervised_module(
4370 ModuleSpec {
4371 module_id: "drop-clears-facts".to_string(),
4372 program: PathBuf::from("/unused/drop-module"),
4373 args: Vec::new(),
4374 env: Vec::new(),
4375 reserved: false,
4376 reserved_prefixes: Vec::new(),
4377 protocol: ModuleProtocol::Subc,
4378 overlap: Default::default(),
4379 },
4380 supervisor.runtime_config(),
4381 Arc::clone(&snapshot),
4382 None,
4383 );
4384 assert!(!module
4385 .inner
4386 .monitor
4387 .lock()
4388 .unwrap()
4389 .as_ref()
4390 .unwrap()
4391 .is_finished());
4392
4393 drop(module);
4394
4395 assert_eq!(
4396 lock_snapshot(&snapshot).unwrap().state,
4397 ModuleState::Stopped
4398 );
4399 assert_snapshot_process_facts_cleared(&snapshot);
4400 }
4401
4402 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
4403 async fn configuration_update_does_not_replace_captured_running_process_facts() {
4404 let supervisor = Supervisor::default();
4405 let snapshot = stale_process_snapshot(ModuleState::Running, true);
4406 let initial = ModuleSpec {
4407 module_id: "rescan-preserves-spawn-facts".to_string(),
4408 program: PathBuf::from("/spawned/module"),
4409 args: Vec::new(),
4410 env: Vec::new(),
4411 reserved: false,
4412 reserved_prefixes: Vec::new(),
4413 protocol: ModuleProtocol::Subc,
4414 overlap: Default::default(),
4415 };
4416 let module = supervisor.supervised_module(
4417 initial.clone(),
4418 supervisor.runtime_config(),
4419 snapshot,
4420 None,
4421 );
4422 let before = module.status().unwrap();
4423 let mut replacement = initial;
4424 replacement.program = PathBuf::from("/rescanned/replacement-module");
4425
4426 module
4427 .update_configuration(replacement, HealthConfig::default(), None)
4428 .await
4429 .unwrap();
4430
4431 let after = module.status().unwrap();
4432 assert_eq!(after.pid, before.pid);
4433 assert_eq!(after.spawned_at_ms, before.spawned_at_ms);
4434 assert_eq!(after.spawned_from, before.spawned_from);
4435 drop(module);
4436 }
4437}
4438
4439fn unix_ms_now() -> u64 {
4440 SystemTime::now()
4441 .duration_since(UNIX_EPOCH)
4442 .map(|duration| duration.as_millis().min(u128::from(u64::MAX)) as u64)
4443 .unwrap_or(0)
4444}
4445
4446async fn supervise_loop(
4447 mut spec: ModuleSpec,
4448 mut runtime: SupervisorRuntimeConfig,
4449 registry: Arc<Registry>,
4450 process_liveness: Arc<SupervisorProcessLiveness>,
4451 snapshot: SharedSnapshot,
4452 mut child: Option<SupervisedChild>,
4453 mut commands: mpsc::Receiver<SupervisorCommand>,
4454) {
4455 let mut health_probe = HealthProbeRuntime::default();
4456 let mut pending_respawn: Option<Instant> = None;
4460 let mut requeued: VecDeque<SupervisorCommand> = VecDeque::new();
4463 loop {
4464 if let Some(command) = requeued.pop_front() {
4465 if !handle_supervisor_command(
4466 command,
4467 &mut spec,
4468 &mut runtime,
4469 ®istry,
4470 &process_liveness,
4471 &snapshot,
4472 &mut child,
4473 &mut commands,
4474 &mut requeued,
4475 )
4476 .await
4477 {
4478 return;
4479 }
4480 if child.is_some() || !respawn_still_pending(&snapshot) {
4481 pending_respawn = None;
4482 }
4483 continue;
4484 }
4485 if child.is_some() {
4486 health_probe.refresh_registration(&spec, &runtime, ®istry, &snapshot);
4487 let probe_sleep = sleep(health_probe.wake_after());
4488 tokio::pin!(probe_sleep);
4489 let active_child = child.as_mut().expect("child checked above");
4490 tokio::select! {
4491 wait_result = active_child.wait() => {
4492 let exit_report = match wait_result {
4501 Ok(status) => classify_reaped_child_exit(&snapshot, active_child, &status),
4502 Err(err) => {
4503 active_child.drain_stderr(&spec.module_id).await;
4504 fail_snapshot(&snapshot, Some(&spec.module_id), None);
4505 record_wait_error_terminal(
4511 &spec.module_id,
4512 &runtime.terminal_ring,
4513 &runtime.spawn_events,
4514 );
4515 untrack_if_registration_released(
4516 &process_liveness,
4517 ®istry,
4518 &spec.module_id,
4519 &snapshot,
4520 );
4521 error!(module_id = %spec.module_id, error = %err, "failed to wait for supervised module");
4522 child = None;
4523 continue;
4524 }
4525 };
4526 active_child.drain_stderr(&spec.module_id).await;
4527
4528 let next = on_child_exit(
4529 &spec,
4530 runtime.restart_policy,
4531 ®istry,
4532 &snapshot,
4533 &runtime.terminal_ring,
4534 &runtime.spawn_events,
4535 &runtime.child_roster,
4536 exit_report,
4537 ).await;
4538 active_child.release_roster();
4541 match next {
4542 NextAction::Stop { registration_released } => {
4543 if registration_released {
4544 process_liveness.untrack_if_current(&spec.module_id, &snapshot);
4545 }
4546 child = None;
4547 }
4548 NextAction::Restart { schedule } => {
4549 let delay = schedule.map_or(
4550 runtime.restart_policy.delay_for_restart(0),
4551 |schedule| schedule.delay,
4552 );
4553 if let Some(schedule) = schedule {
4554 log_crash_respawn(&spec.module_id, schedule);
4555 }
4556 child = None;
4564 pending_respawn = Some(Instant::now() + delay);
4565 }
4566 }
4567 }
4568 command = commands.recv() => {
4569 let Some(command) = command else {
4570 return;
4571 };
4572 if !handle_supervisor_command(
4573 command,
4574 &mut spec,
4575 &mut runtime,
4576 ®istry,
4577 &process_liveness,
4578 &snapshot,
4579 &mut child,
4580 &mut commands,
4581 &mut requeued,
4582 ).await {
4583 return;
4584 }
4585 }
4586 _ = &mut probe_sleep => {
4587 if health_probe.due() {
4588 run_health_probe_cycle(
4589 &spec,
4590 &runtime,
4591 ®istry,
4592 &process_liveness,
4593 &snapshot,
4594 &mut child,
4595 ).await;
4596 if child.is_some() {
4597 health_probe.schedule_next(&spec, runtime.health.cadence);
4598 }
4599 }
4600 }
4601 }
4602 } else if let Some(deadline) = pending_respawn {
4603 tokio::select! {
4604 _ = sleep_until(deadline) => {
4605 pending_respawn = None;
4606 if !respawn_still_pending(&snapshot) {
4610 continue;
4611 }
4612 if runtime.child_roster.is_closed() {
4617 let _ = update_snapshot(&snapshot, Some(&spec.module_id), |state| {
4618 state.state = ModuleState::Stopped;
4619 });
4620 debug!(module_id = %spec.module_id, "crash respawn cancelled by daemon shutdown");
4621 continue;
4622 }
4623 if let Err(err) = wait_for_registration_release(
4624 ®istry,
4625 &spec.module_id,
4626 REGISTRY_RELEASE_TIMEOUT,
4627 ).await {
4628 fail_snapshot(&snapshot, Some(&spec.module_id), None);
4629 error!(module_id = %spec.module_id, error = %err, "registration did not release before restart");
4630 continue;
4631 }
4632
4633 match spawn_and_mark_running(&spec, &runtime, &snapshot) {
4634 Ok(next_child) => {
4635 child = Some(next_child);
4636 debug!(module_id = %spec.module_id, "supervised module restarted after crash");
4637 }
4638 Err(err) => {
4639 fail_snapshot(&snapshot, Some(&spec.module_id), None);
4640 process_liveness.untrack_if_current(&spec.module_id, &snapshot);
4641 error!(module_id = %spec.module_id, error = %err, "failed to restart supervised module");
4642 }
4643 }
4644 }
4645 command = commands.recv() => {
4646 let Some(command) = command else {
4647 return;
4648 };
4649 if !handle_supervisor_command(
4650 command,
4651 &mut spec,
4652 &mut runtime,
4653 ®istry,
4654 &process_liveness,
4655 &snapshot,
4656 &mut child,
4657 &mut commands,
4658 &mut requeued,
4659 ).await {
4660 return;
4661 }
4662 if child.is_some() || !respawn_still_pending(&snapshot) {
4667 pending_respawn = None;
4668 }
4669 }
4670 }
4671 } else {
4672 let Some(command) = commands.recv().await else {
4673 return;
4674 };
4675 if !handle_supervisor_command(
4676 command,
4677 &mut spec,
4678 &mut runtime,
4679 ®istry,
4680 &process_liveness,
4681 &snapshot,
4682 &mut child,
4683 &mut commands,
4684 &mut requeued,
4685 )
4686 .await
4687 {
4688 return;
4689 }
4690 }
4691 }
4692}
4693
4694fn log_crash_respawn(module_id: &str, schedule: CrashRestartSchedule) {
4695 info!(
4696 module_id,
4697 restart_in_window = schedule.restart_in_window,
4698 delay_ms = schedule.delay.as_millis() as u64,
4699 "respawning after crash"
4700 );
4701}
4702
4703fn respawn_still_pending(snapshot: &SharedSnapshot) -> bool {
4709 matches!(
4710 lock_snapshot(snapshot),
4711 Ok(state) if state.enabled && state.state == ModuleState::Restarting
4712 )
4713}
4714
4715enum NextAction {
4716 Stop {
4717 registration_released: bool,
4718 },
4719 Restart {
4720 schedule: Option<CrashRestartSchedule>,
4721 },
4722}
4723
4724#[allow(clippy::too_many_arguments)]
4725async fn handle_supervisor_command(
4726 command: SupervisorCommand,
4727 spec: &mut ModuleSpec,
4728 runtime: &mut SupervisorRuntimeConfig,
4729 registry: &Registry,
4730 process_liveness: &SupervisorProcessLiveness,
4731 snapshot: &SharedSnapshot,
4732 child: &mut Option<SupervisedChild>,
4733 commands: &mut mpsc::Receiver<SupervisorCommand>,
4734 requeued: &mut VecDeque<SupervisorCommand>,
4735) -> bool {
4736 match command {
4737 SupervisorCommand::Drain { reply } => {
4738 let result = drain_optional_child(
4741 &spec.module_id,
4742 spec.protocol,
4743 StopNotice::NotSent,
4744 registry,
4745 snapshot,
4746 &runtime.terminal_ring,
4747 &runtime.spawn_events,
4748 child,
4749 runtime.drain_timeout,
4750 ModuleState::Stopped,
4751 None,
4752 )
4753 .await;
4754 let registration_released = result.is_ok();
4755 let _ = reply.send(result);
4756 if registration_released {
4757 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4758 }
4759 false
4760 }
4761 SupervisorCommand::Retire { reply } => {
4762 let result = async {
4763 let stop_notice = begin_forwarding_drain_if_configured(
4764 spec,
4765 runtime,
4766 registry,
4767 snapshot,
4768 None,
4769 RouteCloseReason::Disable,
4770 )
4771 .await?;
4772 drain_optional_child(
4773 &spec.module_id,
4774 spec.protocol,
4775 stop_notice,
4776 registry,
4777 snapshot,
4778 &runtime.terminal_ring,
4779 &runtime.spawn_events,
4780 child,
4781 runtime.drain_timeout,
4782 ModuleState::Stopped,
4783 None,
4784 )
4785 .await
4786 }
4787 .await;
4788 let registration_released = result.is_ok();
4789 let _ = reply.send(result);
4790 if registration_released {
4791 process_liveness.untrack_if_current(&spec.module_id, snapshot);
4792 }
4793 false
4794 }
4795 SupervisorCommand::Restart {
4796 drain_timeout_ms,
4797 received_at_generation,
4798 queued_at,
4799 reply,
4800 } => {
4801 info!(
4805 module_id = %spec.module_id,
4806 queued_ms = u64::try_from(queued_at.elapsed().as_millis()).unwrap_or(u64::MAX),
4807 "restart command dequeued"
4808 );
4809 let validation = match lock_snapshot(snapshot) {
4821 Ok(state) if !state.enabled => Err(SuperviseError::Disabled {
4822 module_id: spec.module_id.clone(),
4823 }),
4824 Ok(_) => Ok(()),
4825 Err(err) => Err(err),
4826 };
4827 let initiated = validation.is_ok();
4828 let _ = reply.send(validation);
4829 let satisfied_by_generation = if initiated && child.is_some() {
4840 lock_snapshot(snapshot).ok().and_then(|state| {
4841 (state.spawn_generation > received_at_generation
4842 && !state.configuration_updated_since_spawn)
4843 .then_some(state.spawn_generation)
4844 })
4845 } else {
4846 None
4847 };
4848 if let Some(generation) = satisfied_by_generation {
4849 info!(
4850 module_id = %spec.module_id,
4851 received_at_generation,
4852 "restart already satisfied by generation {generation}; not restarting again"
4853 );
4854 } else if initiated {
4855 let drain_timeout = drain_timeout_ms
4858 .map(Duration::from_millis)
4859 .unwrap_or(runtime.drain_timeout);
4860 if let Err(err) = restart_child(
4861 spec,
4862 runtime,
4863 registry,
4864 process_liveness,
4865 snapshot,
4866 child,
4867 drain_timeout,
4868 )
4869 .await
4870 {
4871 warn!(
4872 module_id = %spec.module_id,
4873 error = %err,
4874 "operator restart failed after initiation ack; module state carries the outcome"
4875 );
4876 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
4877 state.state = ModuleState::Failed;
4878 clear_current_process_facts(state);
4879 });
4880 }
4881 }
4882 true
4883 }
4884 SupervisorCommand::Reload { reply } => {
4885 let result =
4886 reload_child(spec, runtime, registry, process_liveness, snapshot, child).await;
4887 let _ = reply.send(result);
4888 true
4889 }
4890 SupervisorCommand::SetEnabled { enabled, reply } => {
4891 let result = set_child_enabled(
4892 spec,
4893 runtime,
4894 registry,
4895 process_liveness,
4896 snapshot,
4897 child,
4898 enabled,
4899 )
4900 .await;
4901 let _ = reply.send(result);
4902 true
4903 }
4904 SupervisorCommand::UpdateConfiguration {
4905 spec: next_spec,
4906 health,
4907 drain_timeout_ms,
4908 reply,
4909 } => {
4910 if let Some(handle) = &runtime.supervisor_handle {
4911 handle.apply_identity_configuration(&next_spec);
4912 }
4913 *spec = next_spec;
4914 let _ = update_snapshot(snapshot, Some(&spec.module_id), |state| {
4915 state.configuration_updated_since_spawn = true;
4916 });
4917 runtime.health = health;
4918 runtime.drain_timeout = drain_timeout_ms
4919 .map(Duration::from_millis)
4920 .unwrap_or(runtime.default_drain_timeout);
4921 *runtime
4922 .effective_drain_timeout
4923 .lock()
4924 .unwrap_or_else(|poisoned| poisoned.into_inner()) = runtime.drain_timeout;
4925 let _ = reply.send(());
4926 true
4927 }
4928 SupervisorCommand::Swap {
4929 ready_timeout,
4930 reply,
4931 } => {
4932 let end = swap::run_swap(
4933 spec,
4934 runtime,
4935 registry,
4936 process_liveness,
4937 snapshot,
4938 child,
4939 commands,
4940 ready_timeout.unwrap_or(DEFAULT_SWAP_READY_TIMEOUT),
4941 reply,
4942 )
4943 .await;
4944 requeued.extend(end.requeue);
4945 true
4946 }
4947 }
4948}
4949
4950async fn restart_child(
4951 spec: &ModuleSpec,
4952 runtime: &SupervisorRuntimeConfig,
4953 registry: &Registry,
4954 process_liveness: &SupervisorProcessLiveness,
4955 snapshot: &SharedSnapshot,
4956 child: &mut Option<SupervisedChild>,
4957 drain_timeout: Duration,
4958) -> Result<(), SuperviseError> {
4959 if !lock_snapshot(snapshot)?.enabled {
4961 return Err(SuperviseError::Disabled {
4962 module_id: spec.module_id.clone(),
4963 });
4964 }
4965 let stop_notice = begin_forwarding_drain_with_timeout(
4966 spec,
4967 runtime,
4968 registry,
4969 snapshot,
4970 None,
4971 RouteCloseReason::Restart,
4972 drain_timeout,
4973 )
4974 .await?;
4975
4976 if child.is_some() {
4977 drain_optional_child(
4978 &spec.module_id,
4979 spec.protocol,
4980 stop_notice,
4981 registry,
4982 snapshot,
4983 &runtime.terminal_ring,
4984 &runtime.spawn_events,
4985 child,
4986 drain_timeout,
4987 ModuleState::Restarting,
4988 Some(true),
4989 )
4990 .await?;
4991 } else {
4992 update_snapshot(snapshot, Some(&spec.module_id), |state| {
4993 state.enabled = true;
4994 state.state = ModuleState::Restarting;
4995 clear_current_process_facts(state);
4996 })?;
4997 wait_for_registration_release(registry, &spec.module_id, REGISTRY_RELEASE_TIMEOUT).await?;
4998 }
4999
5000 reset_restart_count(snapshot, &spec.module_id)?;
5001 sleep(runtime.restart_policy.backoff).await;
5002 if !respawn_still_pending(snapshot) {
5005 process_liveness.untrack_if_current(&spec.module_id, snapshot);
5006 return Ok(());
5007 }
5008 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
5009 match spawn_and_mark_running(spec, runtime, snapshot) {
5015 Ok(next_child) => {
5016 *child = Some(next_child);
5017 debug!(module_id = %spec.module_id, "supervised module restarted by operator request");
5018 Ok(())
5019 }
5020 Err(err) => {
5021 fail_snapshot(snapshot, Some(&spec.module_id), None);
5022 process_liveness.untrack_if_current(&spec.module_id, snapshot);
5023 *child = None;
5024 Err(err)
5025 }
5026 }
5027}
5028
5029async fn reload_child(
5030 spec: &ModuleSpec,
5031 runtime: &SupervisorRuntimeConfig,
5032 registry: &Registry,
5033 process_liveness: &SupervisorProcessLiveness,
5034 snapshot: &SharedSnapshot,
5035 child: &mut Option<SupervisedChild>,
5036) -> Result<(), SuperviseError> {
5037 if !lock_snapshot(snapshot)?.enabled {
5039 return Err(SuperviseError::Disabled {
5040 module_id: spec.module_id.clone(),
5041 });
5042 }
5043 let stop_notice = begin_forwarding_drain(
5044 spec,
5045 runtime,
5046 registry,
5047 snapshot,
5048 Some(true),
5049 RouteCloseReason::Reload,
5050 )
5051 .await?;
5052
5053 if child.is_some() {
5054 drain_optional_child(
5055 &spec.module_id,
5056 spec.protocol,
5057 stop_notice,
5058 registry,
5059 snapshot,
5060 &runtime.terminal_ring,
5061 &runtime.spawn_events,
5062 child,
5063 runtime.drain_timeout,
5064 ModuleState::Restarting,
5065 Some(true),
5066 )
5067 .await?;
5068 } else {
5069 update_snapshot(snapshot, Some(&spec.module_id), |state| {
5070 state.enabled = true;
5071 state.state = ModuleState::Restarting;
5072 clear_current_process_facts(state);
5073 })?;
5074 wait_for_registration_release(registry, &spec.module_id, REGISTRY_RELEASE_TIMEOUT).await?;
5075 }
5076
5077 reset_restart_count(snapshot, &spec.module_id)?;
5078 sleep(runtime.restart_policy.backoff).await;
5079 if !respawn_still_pending(snapshot) {
5082 process_liveness.untrack_if_current(&spec.module_id, snapshot);
5083 return Ok(());
5084 }
5085 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
5086 let next_child = match spawn_and_mark_running(spec, runtime, snapshot) {
5087 Ok(next_child) => next_child,
5088 Err(err) => {
5089 return handle_reload_spawn_failure(
5090 spec,
5091 runtime,
5092 process_liveness,
5093 snapshot,
5094 child,
5095 format!("new child failed to spawn: {err}"),
5096 )
5097 .await;
5098 }
5099 };
5100 *child = Some(next_child);
5101
5102 let wait_outcome = {
5103 let active_child = child.as_mut().expect("new reload child was just stored");
5104 wait_for_registration_after_reload(
5105 registry,
5106 &spec.module_id,
5107 snapshot,
5108 active_child,
5109 REGISTRY_RELEASE_TIMEOUT,
5110 )
5111 .await?
5112 };
5113
5114 match wait_outcome {
5115 RegistrationWaitOutcome::Registered => {
5116 debug!(module_id = %spec.module_id, "supervised module reloaded and registered");
5117 Ok(())
5118 }
5119 RegistrationWaitOutcome::Exited(exit_report) => {
5120 if let Some(active_child) = child.as_mut() {
5121 active_child.drain_stderr(&spec.module_id).await;
5122 }
5123 *child = None;
5124 handle_reload_child_registration_failure(
5125 spec,
5126 runtime,
5127 registry,
5128 process_liveness,
5129 snapshot,
5130 child,
5131 ReloadRegistrationFailure {
5132 exit_report: registration_failure_exit_report(exit_report),
5133 reason: "new child exited before registering".to_string(),
5134 },
5135 )
5136 .await
5137 }
5138 RegistrationWaitOutcome::TimedOut => {
5139 let mut timed_out_child = child
5140 .take()
5141 .expect("timed-out reload child is still running");
5142 timed_out_child
5143 .start_kill()
5144 .map_err(|source| SuperviseError::Kill {
5145 module_id: spec.module_id.clone(),
5146 source,
5147 })?;
5148 let status = timed_out_child
5149 .wait()
5150 .await
5151 .map_err(|source| SuperviseError::Wait {
5152 module_id: spec.module_id.clone(),
5153 source,
5154 })?;
5155 timed_out_child.drain_stderr(&spec.module_id).await;
5156 handle_reload_child_registration_failure(
5157 spec,
5158 runtime,
5159 registry,
5160 process_liveness,
5161 snapshot,
5162 child,
5163 ReloadRegistrationFailure {
5164 exit_report: registration_failure_exit_report(classify_reaped_child_exit(
5165 snapshot,
5166 &timed_out_child,
5167 &status,
5168 )),
5169 reason: format!(
5170 "new child did not register within {:?}",
5171 REGISTRY_RELEASE_TIMEOUT
5172 ),
5173 },
5174 )
5175 .await
5176 }
5177 }
5178}
5179
5180async fn set_child_enabled(
5181 spec: &ModuleSpec,
5182 runtime: &SupervisorRuntimeConfig,
5183 registry: &Registry,
5184 process_liveness: &SupervisorProcessLiveness,
5185 snapshot: &SharedSnapshot,
5186 child: &mut Option<SupervisedChild>,
5187 enabled: bool,
5188) -> Result<bool, SuperviseError> {
5189 let (current_enabled, current_state) = {
5190 let state = lock_snapshot(snapshot)?;
5191 (state.enabled, state.state)
5192 };
5193 let revive_terminal = enabled
5201 && current_enabled
5202 && child.is_none()
5203 && matches!(current_state, ModuleState::Failed | ModuleState::Stopped);
5204 if current_enabled == enabled && !revive_terminal {
5205 return Ok(false);
5206 }
5207
5208 if enabled {
5209 update_snapshot(snapshot, Some(&spec.module_id), |state| {
5210 state.enabled = true;
5211 state.state = ModuleState::Starting;
5212 clear_current_process_facts(state);
5213 })?;
5214 #[cfg(test)]
5215 if runtime.test_seed_stale_facts_before_enable_spawn {
5216 update_snapshot(snapshot, Some(&spec.module_id), |state| {
5217 state.process_alive = true;
5218 state.pid = Some(41);
5219 state.spawned_at_ms = Some(42);
5220 state.spawned_from = Some(PathBuf::from("/spawned/module"));
5221 state.spawned_file_identity = Some(SpawnedFileIdentity {
5222 device: 43,
5223 inode: 44,
5224 });
5225 })?;
5226 }
5227 wait_for_registration_release(registry, &spec.module_id, REGISTRY_RELEASE_TIMEOUT).await?;
5228 reset_restart_count(snapshot, &spec.module_id)?;
5229 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
5230 let next_child = match spawn_and_mark_running(spec, runtime, snapshot) {
5231 Ok(next_child) => next_child,
5232 Err(err) => {
5233 if let Err(state_err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5234 state.state = ModuleState::Failed;
5235 clear_current_process_facts(state);
5236 }) {
5237 error!(module_id = %spec.module_id, error = %state_err, "failed to record enable spawn failure");
5238 }
5239 process_liveness.untrack_if_current(&spec.module_id, snapshot);
5240 return Err(err);
5241 }
5242 };
5243 *child = Some(next_child);
5244 debug!(module_id = %spec.module_id, "supervised module enabled");
5245 Ok(true)
5246 } else {
5247 let stop_notice = begin_forwarding_drain_if_configured(
5248 spec,
5249 runtime,
5250 registry,
5251 snapshot,
5252 Some(false),
5253 RouteCloseReason::Disable,
5254 )
5255 .await?;
5256 drain_optional_child(
5257 &spec.module_id,
5258 spec.protocol,
5259 stop_notice,
5260 registry,
5261 snapshot,
5262 &runtime.terminal_ring,
5263 &runtime.spawn_events,
5264 child,
5265 runtime.drain_timeout,
5266 ModuleState::Disabled,
5267 Some(false),
5268 )
5269 .await?;
5270 debug!(module_id = %spec.module_id, "supervised module disabled");
5271 Ok(true)
5272 }
5273}
5274
5275#[allow(clippy::too_many_arguments)]
5276async fn on_child_exit(
5277 spec: &ModuleSpec,
5278 policy: RestartPolicy,
5279 registry: &Registry,
5280 snapshot: &SharedSnapshot,
5281 terminal_ring: &Arc<Mutex<TerminalRing>>,
5282 spawn_events: &SpawnEventFeed,
5283 roster: &ChildRoster,
5284 exit_report: ExitReport,
5285) -> NextAction {
5286 if roster.is_closed() {
5292 return on_child_exit_during_daemon_shutdown(
5293 spec,
5294 registry,
5295 snapshot,
5296 terminal_ring,
5297 spawn_events,
5298 exit_report,
5299 )
5300 .await;
5301 }
5302 match exit_report.kind {
5303 ExitKind::Clean => {
5304 info!(
5305 module_id = %spec.module_id,
5306 exit_code = ?exit_report.code,
5307 exit_signal = ?exit_report.signal,
5308 "supervised module exited cleanly"
5309 );
5310 if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5311 state.state = ModuleState::Stopped;
5312 clear_current_process_facts(state);
5313 state.last_exit = Some(exit_report.clone());
5314 }) {
5315 error!(module_id = %spec.module_id, error = %err, "failed to record clean module exit");
5316 }
5317 record_terminal(
5318 &spec.module_id,
5319 terminal_ring,
5320 spawn_events,
5321 &exit_report,
5322 TerminalDisposition::Stopped,
5323 );
5324 let registration_released = match wait_for_registration_release(
5325 registry,
5326 &spec.module_id,
5327 REGISTRY_RELEASE_TIMEOUT,
5328 )
5329 .await
5330 {
5331 Ok(()) => true,
5332 Err(err) => {
5333 warn!(module_id = %spec.module_id, error = %err, "registration still active after clean exit");
5334 false
5335 }
5336 };
5337 NextAction::Stop {
5338 registration_released,
5339 }
5340 }
5341 ExitKind::Crash => {
5342 warn!(
5343 module_id = %spec.module_id,
5344 exit_code = ?exit_report.code,
5345 exit_signal = ?exit_report.signal,
5346 "supervised module exited abnormally (crash)"
5347 );
5348 let mut restart_schedule = None;
5349 let mut disposition = TerminalDisposition::Disabled;
5350 let mut disposition_detail = None;
5354 let now = Instant::now();
5355 if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5356 clear_current_process_facts(state);
5357 state.last_exit = Some(exit_report.clone());
5358 if state.enabled {
5359 if let Some(schedule) = state.next_crash_restart(&policy, now) {
5360 state.state = ModuleState::Restarting;
5361 restart_schedule = Some(schedule);
5362 disposition = TerminalDisposition::Restarting;
5363 } else {
5364 state.state = ModuleState::Failed;
5365 disposition = TerminalDisposition::Failed;
5366 disposition_detail = Some(policy.budget_exhausted_detail());
5367 }
5368 } else {
5369 state.state = ModuleState::Disabled;
5370 disposition = TerminalDisposition::Disabled;
5371 }
5372 }) {
5373 error!(module_id = %spec.module_id, error = %err, "failed to record crashed module exit");
5374 return NextAction::Stop {
5375 registration_released: false,
5376 };
5377 }
5378 if disposition_detail.is_some() {
5379 error!(
5384 module_id = %spec.module_id,
5385 max_restarts = policy.max_restarts,
5386 window_secs = policy.window.as_secs(),
5387 "module stopped: {}",
5388 policy.budget_exhausted_detail()
5389 );
5390 }
5391 record_terminal_with_detail(
5392 &spec.module_id,
5393 terminal_ring,
5394 spawn_events,
5395 &exit_report,
5396 disposition,
5397 disposition_detail,
5398 );
5399
5400 if let Some(schedule) = restart_schedule {
5401 NextAction::Restart {
5402 schedule: Some(schedule),
5403 }
5404 } else {
5405 let registration_released = match wait_for_registration_release(
5406 registry,
5407 &spec.module_id,
5408 REGISTRY_RELEASE_TIMEOUT,
5409 )
5410 .await
5411 {
5412 Ok(()) => true,
5413 Err(err) => {
5414 warn!(module_id = %spec.module_id, error = %err, "registration still active after failed module");
5415 false
5416 }
5417 };
5418 NextAction::Stop {
5419 registration_released,
5420 }
5421 }
5422 }
5423 ExitKind::DeliberateSeverance => {
5424 warn!(
5425 module_id = %spec.module_id,
5426 exit_code = ?exit_report.code,
5427 exit_signal = ?exit_report.signal,
5428 "supervised module exited after deliberate connection severance"
5429 );
5430 let mut should_restart = false;
5431 let mut disposition = TerminalDisposition::Disabled;
5432 if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5433 clear_current_process_facts(state);
5434 state.last_exit = Some(exit_report.clone());
5435 state.lifetime_restarts += 1;
5436 if state.enabled {
5437 state.state = ModuleState::Restarting;
5438 should_restart = true;
5439 disposition = TerminalDisposition::Restarting;
5440 } else {
5441 state.state = ModuleState::Disabled;
5442 }
5443 }) {
5444 error!(module_id = %spec.module_id, error = %err, "failed to record deliberately severed module exit");
5445 return NextAction::Stop {
5446 registration_released: false,
5447 };
5448 }
5449 record_terminal(
5450 &spec.module_id,
5451 terminal_ring,
5452 spawn_events,
5453 &exit_report,
5454 disposition,
5455 );
5456
5457 if should_restart {
5458 NextAction::Restart { schedule: None }
5459 } else {
5460 let registration_released = match wait_for_registration_release(
5461 registry,
5462 &spec.module_id,
5463 REGISTRY_RELEASE_TIMEOUT,
5464 )
5465 .await
5466 {
5467 Ok(()) => true,
5468 Err(err) => {
5469 warn!(module_id = %spec.module_id, error = %err, "registration still active after deliberately severed module exit");
5470 false
5471 }
5472 };
5473 NextAction::Stop {
5474 registration_released,
5475 }
5476 }
5477 }
5478 }
5479}
5480
5481async fn on_child_exit_during_daemon_shutdown(
5482 spec: &ModuleSpec,
5483 registry: &Registry,
5484 snapshot: &SharedSnapshot,
5485 terminal_ring: &Arc<Mutex<TerminalRing>>,
5486 spawn_events: &SpawnEventFeed,
5487 exit_report: ExitReport,
5488) -> NextAction {
5489 info!(
5490 module_id = %spec.module_id,
5491 exit_code = ?exit_report.code,
5492 exit_signal = ?exit_report.signal,
5493 exit_kind = ?exit_report.kind,
5494 "supervised module exited during daemon shutdown; not restarting it"
5495 );
5496 if let Err(err) = update_snapshot(snapshot, Some(&spec.module_id), |state| {
5497 state.state = ModuleState::Stopped;
5498 clear_current_process_facts(state);
5499 state.last_exit = Some(exit_report.clone());
5500 }) {
5501 error!(module_id = %spec.module_id, error = %err, "failed to record module exit during daemon shutdown");
5502 }
5503 record_terminal(
5504 &spec.module_id,
5505 terminal_ring,
5506 spawn_events,
5507 &exit_report,
5508 TerminalDisposition::DaemonShutdown,
5509 );
5510 let registration_released =
5511 wait_for_registration_release(registry, &spec.module_id, REGISTRY_RELEASE_TIMEOUT)
5512 .await
5513 .is_ok();
5514 NextAction::Stop {
5515 registration_released,
5516 }
5517}
5518
5519fn record_wait_error_terminal(
5520 module_id: &str,
5521 terminal_ring: &Arc<Mutex<TerminalRing>>,
5522 spawn_events: &SpawnEventFeed,
5523) {
5524 record_terminal(
5525 module_id,
5526 terminal_ring,
5527 spawn_events,
5528 &wait_error_exit_report(),
5529 TerminalDisposition::Failed,
5530 );
5531}
5532
5533fn record_terminal(
5534 module_id: &str,
5535 terminal_ring: &Arc<Mutex<TerminalRing>>,
5536 spawn_events: &SpawnEventFeed,
5537 exit_report: &ExitReport,
5538 disposition: TerminalDisposition,
5539) {
5540 record_terminal_with_detail(
5541 module_id,
5542 terminal_ring,
5543 spawn_events,
5544 exit_report,
5545 disposition,
5546 None,
5547 );
5548}
5549
5550fn durable_terminal_history_of(
5554 terminal_ring: &Mutex<TerminalRing>,
5555 module_id: &str,
5556) -> subc_control::TerminalHistory {
5557 let read = terminal_ring
5558 .lock()
5559 .unwrap_or_else(|p| p.into_inner())
5560 .capture_durable_history();
5561 read.read(module_id)
5562}
5563
5564fn record_terminal_with_detail(
5565 module_id: &str,
5566 terminal_ring: &Arc<Mutex<TerminalRing>>,
5567 spawn_events: &SpawnEventFeed,
5568 exit_report: &ExitReport,
5569 disposition: TerminalDisposition,
5570 disposition_detail: Option<String>,
5571) {
5572 spawn_events.emit_exited(module_id, exit_report.code, exit_report.signal);
5573 let record = TerminalRecord {
5574 exit_code: exit_report.code,
5575 exit_signal: exit_report.signal,
5576 at_ms: exit_report.at_ms,
5577 disposition,
5578 exit_kind: exit_report.kind.into(),
5579 disposition_detail,
5580 };
5581 terminal_ring
5582 .lock()
5583 .unwrap_or_else(|poisoned| poisoned.into_inner())
5584 .record_exit(module_id, record);
5585}
5586
5587fn untrack_if_registration_released(
5588 process_liveness: &SupervisorProcessLiveness,
5589 registry: &Registry,
5590 module_id: &str,
5591 snapshot: &SharedSnapshot,
5592) {
5593 match registry.get_module(module_id) {
5594 Ok(None) => process_liveness.untrack_if_current(module_id, snapshot),
5595 Ok(Some(_)) => {}
5596 Err(err) => {
5597 warn!(module_id, error = %err, "could not determine whether supervisor liveness can be untracked");
5598 }
5599 }
5600}
5601
5602#[cfg(test)]
5616fn apply_wire_spawn_args(
5617 command: &mut Command,
5618 spec: &ModuleSpec,
5619 connection_file_path: Option<&std::path::Path>,
5620 handle: Option<&SupervisorHandle>,
5621) -> Result<(), SuperviseError> {
5622 apply_wire_spawn_args_for_role(
5623 command,
5624 spec,
5625 connection_file_path,
5626 handle,
5627 SpawnRole::Plain,
5628 )
5629}
5630
5631fn apply_wire_spawn_args_for_role(
5640 command: &mut Command,
5641 spec: &ModuleSpec,
5642 connection_file_path: Option<&std::path::Path>,
5643 handle: Option<&SupervisorHandle>,
5644 role: SpawnRole,
5645) -> Result<(), SuperviseError> {
5646 command.env(SUBC_MODULE_ID_ENV, &spec.module_id);
5647 if spec.protocol == ModuleProtocol::None {
5648 return Ok(());
5649 }
5650 if let Some(connection_file_path) = connection_file_path {
5651 command.arg(SUBC_ARG).arg(connection_file_path);
5652 }
5653
5654 let nonce = generate_launch_nonce()?;
5658 if let Some(handle) = handle {
5659 match role {
5660 SpawnRole::Plain => {
5661 handle.set_spawn_nonce(&spec.module_id, nonce.clone());
5662 if spec.reserved {
5663 handle.set_reserved_nonce(&spec.module_id, nonce.clone());
5664 }
5665 }
5666 SpawnRole::SwapCandidate => handle.open_swap(&spec.module_id, nonce.clone()),
5667 }
5668 }
5669 command.env(SUBC_LAUNCH_NONCE_ENV, nonce);
5670 Ok(())
5671}
5672
5673fn apply_child_env(command: &mut Command, spec: &ModuleSpec) {
5674 command.env_remove(CK_LOG_ENV);
5675 command.env_remove(SUBC_SPAWN_ROLE_ENV);
5682 for (key, value) in &spec.env {
5683 if matches!(
5687 key.as_str(),
5688 CAPTURE_MAX_FILE_MB_ENV | CAPTURE_KEEP_ENV | CAPTURE_MAX_AGE_DAYS_ENV
5689 ) || key == SUBC_SPAWN_ROLE_ENV
5690 {
5691 continue;
5692 }
5693 command.env(key, value);
5694 }
5695}
5696
5697#[derive(Debug, Clone, Copy, PartialEq, Eq)]
5700enum SpawnRole {
5701 Plain,
5702 SwapCandidate,
5703}
5704
5705fn apply_spawn_role(command: &mut Command, role: SpawnRole) {
5708 if role == SpawnRole::SwapCandidate {
5709 command.env(SUBC_SPAWN_ROLE_ENV, SPAWN_ROLE_SWAP_CANDIDATE);
5710 }
5711}
5712
5713fn spawn_child(
5714 spec: &ModuleSpec,
5715 connection_file_path: Option<&std::path::Path>,
5716 handle: Option<&SupervisorHandle>,
5717 ring: &Arc<Mutex<StderrRing>>,
5718 capture_logs_dir: Option<&std::path::Path>,
5719 roster: &ChildRoster,
5720 #[cfg(target_os = "linux")] cgroup_placement: Option<&subc_cgroup::Placement>,
5721) -> Result<SupervisedChild, SuperviseError> {
5722 spawn_child_in_slot(
5723 spec,
5724 connection_file_path,
5725 handle,
5726 ring,
5727 capture_logs_dir,
5728 roster,
5729 #[cfg(target_os = "linux")]
5730 cgroup_placement,
5731 SpawnRole::Plain,
5732 false,
5733 )
5734}
5735
5736#[allow(clippy::too_many_arguments)]
5749fn spawn_child_in_slot(
5750 spec: &ModuleSpec,
5751 connection_file_path: Option<&std::path::Path>,
5752 handle: Option<&SupervisorHandle>,
5753 ring: &Arc<Mutex<StderrRing>>,
5754 capture_logs_dir: Option<&std::path::Path>,
5755 roster: &ChildRoster,
5756 #[cfg(target_os = "linux")] cgroup_placement: Option<&subc_cgroup::Placement>,
5757 role: SpawnRole,
5758 alternate_slot: bool,
5759) -> Result<SupervisedChild, SuperviseError> {
5760 if roster.is_closed() {
5761 return Err(SuperviseError::Spawn {
5762 program: spec.program.clone(),
5763 source: io::Error::other("the daemon is shutting down; not starting a new process"),
5764 cgroup_path: None,
5765 });
5766 }
5767 #[cfg(target_os = "linux")]
5768 let cgroup_name = swap::cgroup_name(&spec.module_id, alternate_slot);
5769 #[cfg(not(target_os = "linux"))]
5770 let _ = alternate_slot;
5771 let mut command = Command::new(&spec.program);
5772 command.args(&spec.args);
5773 apply_child_env(&mut command, spec);
5803 apply_spawn_role(&mut command, role);
5804 apply_wire_spawn_args_for_role(&mut command, spec, connection_file_path, handle, role)?;
5805
5806 #[cfg(target_os = "linux")]
5807 let cgroup_path = cgroup_placement
5808 .map(|placement| placement.module_path(&cgroup_name))
5809 .transpose()
5810 .map_err(|source| SuperviseError::Cgroup {
5811 module_id: spec.module_id.clone(),
5812 source,
5813 })?;
5814 #[cfg(not(target_os = "linux"))]
5815 let cgroup_path: Option<PathBuf> = None;
5816 #[cfg(target_os = "linux")]
5817 if let Some(path) = &cgroup_path {
5818 if let Err(error) = apply_cgroup_placement(&mut command, spec, path) {
5819 if let Some(placement) = cgroup_placement {
5820 remove_module_cgroup(placement, &cgroup_name);
5821 }
5822 return Err(error);
5823 }
5824 }
5825
5826 let output_sink = if let Some(logs_dir) = capture_logs_dir {
5827 let path = logs_dir.join(format!("{}.stderr.log", spec.module_id));
5828 match ChildOutputSink::open(&path, capture_retention(spec)) {
5829 Ok(sink) => sink,
5830 Err(error) => {
5831 warn!(
5832 module_id = %spec.module_id,
5833 path = %path.display(),
5834 error = %error,
5835 "could not open child output capture file; forwarding to stderr"
5836 );
5837 ChildOutputSink::Stderr
5838 }
5839 }
5840 } else {
5841 ChildOutputSink::Stderr
5842 };
5843
5844 command.stdout(Stdio::piped());
5845 command.stderr(Stdio::piped());
5846 command.kill_on_drop(true);
5847 #[cfg(unix)]
5864 command.process_group(0);
5865 command.stdin(Stdio::null());
5866
5867 #[cfg(windows)]
5872 subc_jobobject::suspend_on_create_async(&mut command);
5873 let mut child = match command.spawn() {
5874 Ok(child) => child,
5875 Err(source) => {
5876 #[cfg(target_os = "linux")]
5877 if let Some(placement) = cgroup_placement {
5878 remove_module_cgroup(placement, &cgroup_name);
5879 }
5880 return Err(SuperviseError::Spawn {
5881 program: spec.program.clone(),
5882 source,
5883 cgroup_path,
5884 });
5885 }
5886 };
5887
5888 #[cfg(windows)]
5890 let job = contain_spawned_child(&child, spec)?;
5891 let spawned_at_ms = unix_ms_now();
5892 let spawned_from = spec.program.clone();
5893 let spawned_file_identity = spawned_file_identity(&spawned_from);
5894 let pid = child.id().ok_or_else(|| SuperviseError::Spawn {
5895 program: spec.program.clone(),
5896 source: io::Error::other("spawned child exposed no live pid"),
5897 cgroup_path: cgroup_path.clone(),
5898 })?;
5899 let process_start_time = crate::provenance::process_start_time(pid);
5900 let process_identity = process_start_time.map(|start_time| ProcessIdentity { pid, start_time });
5901 #[cfg(target_os = "linux")]
5905 let recorded_cgroup_name = cgroup_path.as_ref().map(|_| cgroup_name.clone());
5906 #[cfg(not(target_os = "linux"))]
5907 let recorded_cgroup_name = None;
5908 let roster_guard = roster.admit(
5909 spec.module_id.clone(),
5910 pid,
5911 spec.protocol,
5912 process_start_time,
5913 crate::child_roster::RecordedIdentity {
5914 start_time: subc_os::start_time(pid),
5915 executable: spawned_file_identity.map(|identity| {
5916 crate::live_children::ExecutableIdentity {
5917 device: identity.device,
5918 inode: identity.inode,
5919 }
5920 }),
5921 cgroup_name: recorded_cgroup_name,
5922 },
5923 );
5924 if roster.is_closed() {
5933 if let Err(error) = child.start_kill() {
5934 debug!(module_id = %spec.module_id, pid, %error, "kill of a process spawned during daemon shutdown failed; it may already have exited");
5935 }
5936 drop(roster_guard);
5937 return Err(SuperviseError::Spawn {
5938 program: spec.program.clone(),
5939 source: io::Error::other(
5940 "the daemon began shutting down while this process was starting; ended it",
5941 ),
5942 cgroup_path,
5943 });
5944 }
5945
5946 let stdout_pump = match child.stdout.take() {
5947 Some(stdout) => Some(tokio::spawn(pump_stdout_to(stdout, output_sink.clone()))),
5948 None => {
5949 warn!(
5950 module_id = %spec.module_id,
5951 "spawned child exposed no stdout pipe; file capture will be incomplete"
5952 );
5953 None
5954 }
5955 };
5956 let stderr_pump = match child.stderr.take() {
5957 Some(stderr) => {
5958 let generation = ring
5959 .lock()
5960 .unwrap_or_else(|poisoned| poisoned.into_inner())
5961 .begin_process();
5962 Some(StderrPump {
5963 task: tokio::spawn(pump_stderr_to(
5964 stderr,
5965 Arc::clone(ring),
5966 generation,
5967 output_sink,
5968 )),
5969 generation,
5970 })
5971 }
5972 None => {
5973 ring.lock()
5977 .unwrap_or_else(|poisoned| poisoned.into_inner())
5978 .mark_not_captured("stderr pipe was not available on spawn");
5979 warn!(
5980 module_id = %spec.module_id,
5981 "spawned child exposed no stderr pipe; tail will be unavailable"
5982 );
5983 None
5984 }
5985 };
5986
5987 Ok(SupervisedChild {
5988 child,
5989 #[cfg(target_os = "linux")]
5990 module_id: cgroup_name,
5991 #[cfg(target_os = "linux")]
5992 cgroup_placement: cgroup_placement.cloned(),
5993 #[cfg(windows)]
5994 job,
5995 stdout_pump,
5996 stderr_pump,
5997 stderr_ring: Arc::clone(ring),
5998 spawned_at_ms,
5999 spawned_from,
6000 spawned_file_identity,
6001 process_start_time,
6002 process_identity,
6003 pid,
6004 roster_guard: Some(roster_guard),
6005 })
6006}
6007
6008#[cfg(windows)]
6022fn contain_spawned_child(
6023 child: &Child,
6024 spec: &ModuleSpec,
6025) -> Result<Option<subc_jobobject::JobObject>, SuperviseError> {
6026 let module_id = spec.module_id.as_str();
6027 let Some(pid) = child.id() else {
6028 warn!(
6031 module_id,
6032 "spawned child had already exited before containment; no job object attached"
6033 );
6034 return Ok(None);
6035 };
6036
6037 let job = match subc_jobobject::JobObject::new() {
6038 Ok(job) => job,
6039 Err(source) => {
6040 warn!(
6041 module_id,
6042 error = %source,
6043 "could not create a job object; this module's helper processes will not be \
6044 reaped on teardown"
6045 );
6046 resume_suspended_child(pid, spec)?;
6049 return Ok(None);
6050 }
6051 };
6052
6053 if let Err(source) = job.assign(child) {
6054 warn!(
6055 module_id,
6056 error = %source,
6057 "could not assign the child to its job object; this module's helper processes \
6058 will not be reaped on teardown"
6059 );
6060 resume_suspended_child(pid, spec)?;
6061 return Ok(None);
6062 }
6063
6064 resume_suspended_child(pid, spec)?;
6065 Ok(Some(job))
6066}
6067
6068#[cfg(windows)]
6073fn resume_suspended_child(pid: u32, spec: &ModuleSpec) -> Result<(), SuperviseError> {
6074 if let Err(source) = subc_jobobject::resume_main_thread(pid) {
6075 let _ = std::process::Command::new("taskkill.exe")
6079 .args(["/PID", &pid.to_string(), "/T", "/F"])
6080 .stdin(Stdio::null())
6081 .stdout(Stdio::null())
6082 .stderr(Stdio::null())
6083 .status();
6084 return Err(SuperviseError::Spawn {
6085 program: spec.program.clone(),
6086 source,
6087 cgroup_path: None,
6088 });
6089 }
6090 Ok(())
6091}
6092
6093#[cfg(target_os = "linux")]
6094fn remove_module_cgroup(placement: &subc_cgroup::Placement, module_id: &str) {
6095 match placement.remove_module(module_id) {
6096 Ok(()) => debug!(module_id, "removed module cgroup after process exit"),
6097 Err(error) => warn!(
6098 module_id,
6099 error = %error,
6100 "could not remove module cgroup after process exit; continuing teardown"
6101 ),
6102 }
6103}
6104
6105#[cfg(target_os = "linux")]
6106fn apply_cgroup_placement(
6107 command: &mut Command,
6108 spec: &ModuleSpec,
6109 path: &std::path::Path,
6110) -> Result<(), SuperviseError> {
6111 subc_cgroup::apply(command, path).map_err(|source| SuperviseError::Cgroup {
6112 module_id: spec.module_id.clone(),
6113 source,
6114 })
6115}
6116
6117fn capture_retention(spec: &ModuleSpec) -> Retention {
6118 let defaults = Retention::default();
6119 let value = |name: &str| {
6120 spec.env
6121 .iter()
6122 .rev()
6123 .find_map(|(key, value)| (key == name).then_some(value.as_str()))
6124 };
6125 Retention {
6126 max_file_mb: value(CAPTURE_MAX_FILE_MB_ENV)
6127 .and_then(|value| value.parse().ok())
6128 .unwrap_or(defaults.max_file_mb),
6129 keep: value(CAPTURE_KEEP_ENV)
6130 .and_then(|value| value.parse().ok())
6131 .unwrap_or(defaults.keep),
6132 max_age_days: value(CAPTURE_MAX_AGE_DAYS_ENV)
6133 .and_then(|value| value.parse().ok())
6134 .unwrap_or(defaults.max_age_days),
6135 }
6136}
6137
6138fn generate_launch_nonce() -> Result<String, SuperviseError> {
6141 let mut bytes = [0u8; 32];
6142 getrandom::getrandom(&mut bytes).map_err(|source| SuperviseError::LaunchNonce {
6143 reason: source.to_string(),
6144 })?;
6145 let mut hex = String::with_capacity(64);
6146 for b in bytes {
6147 use std::fmt::Write;
6148 let _ = write!(hex, "{b:02x}");
6149 }
6150 Ok(hex)
6151}
6152
6153fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
6156 if a.len() != b.len() {
6157 return false;
6158 }
6159 let mut diff = 0u8;
6160 for (x, y) in a.iter().zip(b.iter()) {
6161 diff |= x ^ y;
6162 }
6163 diff == 0
6164}
6165
6166fn spawn_and_mark_running(
6167 spec: &ModuleSpec,
6168 runtime: &SupervisorRuntimeConfig,
6169 snapshot: &SharedSnapshot,
6170) -> Result<SupervisedChild, SuperviseError> {
6171 let child = spawn_child(
6172 spec,
6173 runtime.connection_file_path.as_deref(),
6174 runtime.supervisor_handle.as_ref(),
6175 &runtime.stderr_ring,
6176 runtime.capture_logs_dir.as_deref(),
6177 &runtime.child_roster,
6178 #[cfg(target_os = "linux")]
6179 runtime.cgroup_placement.as_ref(),
6180 )?;
6181 set_running(snapshot, &child, &spec.module_id, &runtime.spawn_events)?;
6182 Ok(child)
6183}
6184
6185enum RegistrationWaitOutcome {
6186 Registered,
6187 Exited(ExitReport),
6188 TimedOut,
6189}
6190
6191struct ReloadRegistrationFailure {
6192 exit_report: ExitReport,
6193 reason: String,
6194}
6195
6196#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6197enum BusyGaugeObservation {
6198 Quiescent,
6199 Busy,
6200 Omitted,
6201}
6202
6203fn busy_gauge_observation(metrics: Option<&Value>, gauges: &[String]) -> BusyGaugeObservation {
6204 let Some(metrics) = metrics.and_then(Value::as_object) else {
6205 return BusyGaugeObservation::Omitted;
6206 };
6207 let mut sum = 0u128;
6208 for gauge in gauges {
6209 let Some(value) = metrics.get(gauge) else {
6210 return BusyGaugeObservation::Omitted;
6211 };
6212 let Some(value) = value.as_u64() else {
6213 return BusyGaugeObservation::Busy;
6214 };
6215 sum = sum.saturating_add(u128::from(value));
6216 }
6217 if sum == 0 {
6218 BusyGaugeObservation::Quiescent
6219 } else {
6220 BusyGaugeObservation::Busy
6221 }
6222}
6223
6224fn declared_busy_gauges(
6225 registry: &Registry,
6226 module_id: &str,
6227) -> Result<Vec<String>, SuperviseError> {
6228 busy_gauges_of(
6229 registry
6230 .get_module(module_id)
6231 .map_err(SuperviseError::Registry)?,
6232 )
6233}
6234
6235fn declared_busy_gauges_for_connection(
6239 registry: &Registry,
6240 connection_id: ConnectionId,
6241) -> Result<Vec<String>, SuperviseError> {
6242 busy_gauges_of(
6243 registry
6244 .get_module_by_connection(connection_id)
6245 .map_err(SuperviseError::Registry)?,
6246 )
6247}
6248
6249fn busy_gauges_of(
6250 registration: Option<crate::registry::ModuleRegistration>,
6251) -> Result<Vec<String>, SuperviseError> {
6252 let Some(registration) = registration else {
6253 return Ok(Vec::new());
6254 };
6255 let Some(self_signals) = registration.manifest.self_signals else {
6256 return Ok(Vec::new());
6257 };
6258
6259 let mut gauges = Vec::new();
6260 for declaration in self_signals {
6261 if declaration.kind != SelfSignalKind::Busy {
6262 continue;
6263 }
6264 match declaration.anchored_to {
6265 SignalAnchor::HealthGauges { gauges: declared } if !declared.is_empty() => {
6266 gauges.extend(declared)
6267 }
6268 _ => {
6269 gauges.push(String::new());
6272 }
6273 }
6274 }
6275 Ok(gauges)
6276}
6277
6278async fn wait_for_forwarding_quiescence(
6283 forwarding: &ForwardingTable,
6284 module_id: &str,
6285 runtime: &SupervisorRuntimeConfig,
6286 endpoint: crate::ModuleEndpointId,
6287 deadline: Instant,
6288 busy_gauges: &[String],
6289 scope: DrainScope,
6290) -> Result<bool, SuperviseError> {
6291 let mut gauges_quiescent = busy_gauges.is_empty();
6292 let mut next_probe_at = Instant::now();
6293 let mut omission_counted = false;
6294
6295 loop {
6296 let now = Instant::now();
6297 if !busy_gauges.is_empty() && now >= next_probe_at && now < deadline {
6298 let report = match scope {
6299 DrainScope::Active => probe_module_health(module_id, runtime, Some(deadline)).await,
6300 DrainScope::Endpoint(endpoint) => {
6301 probe_endpoint_health(endpoint, runtime, Some(deadline)).await
6302 }
6303 };
6304 gauges_quiescent = match report {
6305 Ok(report) => match busy_gauge_observation(report.metrics.as_ref(), busy_gauges) {
6306 BusyGaugeObservation::Quiescent => true,
6307 BusyGaugeObservation::Busy => false,
6308 BusyGaugeObservation::Omitted => {
6309 if !omission_counted {
6310 forwarding
6311 .counters()
6312 .increment_drains_with_undeclared_gauge();
6313 omission_counted = true;
6314 }
6315 false
6316 }
6317 },
6318 Err(err) => {
6319 warn!(
6320 module_id,
6321 error = %err,
6322 "drain health.check did not produce declared busy gauges; treating module as busy"
6323 );
6324 false
6325 }
6326 };
6327 next_probe_at = Instant::now() + runtime.health.cadence.max(REGISTRY_RELEASE_POLL);
6328 }
6329
6330 let in_flight = forwarding
6331 .endpoint_in_flight_count(endpoint)
6332 .map_err(SuperviseError::Forwarding)?;
6333 if in_flight == 0 && gauges_quiescent {
6334 return Ok(true);
6335 }
6336
6337 let now = Instant::now();
6338 if now >= deadline {
6339 return Ok(false);
6340 }
6341 let mut wait = deadline
6342 .saturating_duration_since(now)
6343 .min(REGISTRY_RELEASE_POLL);
6344 if !busy_gauges.is_empty() {
6345 wait = wait.min(next_probe_at.saturating_duration_since(now));
6346 }
6347 sleep(wait).await;
6348 }
6349}
6350
6351fn drained_after_quiescence_wait(wait_result: &Result<bool, SuperviseError>) -> bool {
6359 match wait_result {
6360 Ok(drained) => *drained,
6361 Err(_) => false,
6362 }
6363}
6364
6365fn send_route_goodbyes(forwarding: &ForwardingTable, released_routes: Vec<GoodbyeTarget>) {
6366 for released in released_routes {
6367 let frame = match Frame::build_with_version(
6368 released.negotiated_ver,
6369 FrameType::Goodbye,
6370 control_flags(),
6371 released.channel,
6372 released.epoch,
6373 0,
6374 Vec::new(),
6375 ) {
6376 Ok(frame) => frame,
6377 Err(err) => {
6378 warn!(
6379 route_channel = released.channel,
6380 error = %err,
6381 "failed to build supervisor drain route GOODBYE frame"
6382 );
6383 continue;
6384 }
6385 };
6386 if !released.close_on_delivery_failure() {
6387 crate::forwarding::send_module_route_goodbye(
6388 &forwarding.counters(),
6389 &released.sink,
6390 frame,
6391 released.module_id.as_deref(),
6392 "supervisor drain",
6393 );
6394 continue;
6395 }
6396 if let Err(err) = released.sink.try_send(frame) {
6397 warn!(
6398 target_connection_id = released.connection_id.get(),
6399 route_channel = released.channel,
6400 error = %err,
6401 "supervisor drain route GOODBYE was not delivered to client; closing target connection"
6402 );
6403 let _ = forwarding.escalate_client_delivery_failure(
6404 released.connection_id,
6405 released.channel,
6406 released.epoch,
6407 CloseReason::new(
6408 "route_goodbye_delivery_failed",
6409 format!(
6410 "failed to enqueue supervisor drain route GOODBYE for channel {}: {err}",
6411 released.channel
6412 ),
6413 ),
6414 crate::forwarding::UndeliveredFrame {
6415 module_id: released.module_id.as_deref(),
6416 sink: &released.sink,
6417 },
6418 );
6419 }
6420 }
6421}
6422
6423fn send_module_draining(
6424 module_id: &str,
6425 reason: RouteCloseReason,
6426 deadline_ms: u64,
6427 target: &ModuleDrainTarget,
6428) {
6429 let body = match serde_json::to_vec(&ModuleControlCommand::Draining {
6430 reason,
6431 deadline_ms,
6432 }) {
6433 Ok(body) => body,
6434 Err(err) => {
6435 warn!(
6436 module_id,
6437 error = %err,
6438 "failed to encode module draining command"
6439 );
6440 return;
6441 }
6442 };
6443 let frame = match Frame::build_with_version(
6444 target.negotiated_ver,
6445 FrameType::Push,
6446 control_flags(),
6447 0,
6448 0,
6449 0,
6450 body,
6451 ) {
6452 Ok(frame) => frame,
6453 Err(err) => {
6454 warn!(
6455 module_id,
6456 error = %err,
6457 "failed to build module draining command frame"
6458 );
6459 return;
6460 }
6461 };
6462 if let Err(err) = target.sink.try_send(frame) {
6463 warn!(
6464 module_id,
6465 target_connection_id = target.endpoint.connection_id.get(),
6466 error = %err,
6467 "module draining command was not delivered to peer"
6468 );
6469 }
6470}
6471
6472fn module_goodbye_frame(module_id: &str, negotiated_ver: u8) -> Option<Frame> {
6474 match Frame::build_with_version(
6475 negotiated_ver,
6476 FrameType::Goodbye,
6477 control_flags(),
6478 0,
6479 0,
6480 0,
6481 Vec::new(),
6482 ) {
6483 Ok(frame) => Some(frame),
6484 Err(err) => {
6485 warn!(
6486 module_id,
6487 error = %err,
6488 "failed to build module GOODBYE frame"
6489 );
6490 None
6491 }
6492 }
6493}
6494
6495#[cfg(unix)]
6509async fn send_module_goodbyes_for_daemon_shutdown(
6510 forwarding: &Arc<ForwardingTable>,
6511 reason: &CloseReason,
6512 wait_for_flush: bool,
6513) {
6514 const GOODBYE_BUDGET: Duration = Duration::from_millis(500);
6515 let targets = match forwarding.module_connections() {
6516 Ok(targets) => targets,
6517 Err(err) => {
6518 warn!(error = %err, "could not list module connections for shutdown GOODBYE");
6519 return;
6520 }
6521 };
6522 let deadline = Instant::now() + GOODBYE_BUDGET;
6523 let mut sends = tokio::task::JoinSet::new();
6524 for target in targets {
6525 let Some(frame) = module_goodbye_frame(&target.module_id, target.negotiated_ver) else {
6526 continue;
6527 };
6528 if !wait_for_flush {
6529 if let Err(err) = target.sink.try_send(frame) {
6530 debug!(
6531 module_id = %target.module_id,
6532 error = %err,
6533 "shutdown module GOODBYE was not queued"
6534 );
6535 }
6536 continue;
6537 }
6538 let forwarding = Arc::clone(forwarding);
6539 let reason = reason.clone();
6540 sends.spawn(async move {
6541 match timeout_at(deadline, target.sink.send_flushed(frame)).await {
6542 Ok(Ok(())) => {}
6543 Ok(Err(err)) => debug!(
6544 module_id = %target.module_id,
6545 error = %err,
6546 "module connection closed before its shutdown GOODBYE was written"
6547 ),
6548 Err(_) => warn!(
6549 module_id = %target.module_id,
6550 budget = ?GOODBYE_BUDGET,
6551 "shutdown module GOODBYE was not written within its budget; closing anyway"
6552 ),
6553 }
6554 forwarding.request_connection_close(target.endpoint.connection_id, reason);
6555 });
6556 }
6557 while sends.join_next().await.is_some() {}
6559}
6560
6561fn send_module_goodbye(module_id: &str, forwarding: &ForwardingTable, target: &ModuleDrainTarget) {
6562 let Some(frame) = module_goodbye_frame(module_id, target.negotiated_ver) else {
6563 return;
6564 };
6565 if let Err(err) = target.sink.try_send(frame) {
6566 warn!(
6567 module_id,
6568 target_connection_id = target.endpoint.connection_id.get(),
6569 error = %err,
6570 "supervisor drain module GOODBYE was not delivered to peer; closing module connection"
6571 );
6572 forwarding.request_connection_close(
6573 target.endpoint.connection_id,
6574 CloseReason::new(
6575 "module_goodbye_delivery_failed",
6576 format!("failed to enqueue supervisor drain module GOODBYE for module '{module_id}': {err}"),
6577 ),
6578 );
6579 }
6580}
6581
6582#[derive(Clone, Copy)]
6583struct ForwardingDrainContext<'a> {
6584 spec: &'a ModuleSpec,
6585 runtime: &'a SupervisorRuntimeConfig,
6586 registry: &'a Registry,
6587 scope: DrainScope,
6588}
6589
6590#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6592enum DrainScope {
6593 Active,
6596 Endpoint(crate::ModuleEndpointId),
6601}
6602
6603#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6611enum StopNotice {
6612 SentOverConnection,
6615 NoConnection,
6619 NotSent,
6623}
6624
6625async fn begin_forwarding_drain(
6626 spec: &ModuleSpec,
6627 runtime: &SupervisorRuntimeConfig,
6628 registry: &Registry,
6629 snapshot: &SharedSnapshot,
6630 enabled: Option<bool>,
6631 reason: RouteCloseReason,
6632) -> Result<StopNotice, SuperviseError> {
6633 let Some(forwarding) = runtime.forwarding.as_ref() else {
6634 return Err(SuperviseError::ReloadUnavailable {
6635 module_id: spec.module_id.clone(),
6636 reason: "supervisor was not configured with a forwarding table".to_string(),
6637 });
6638 };
6639
6640 begin_forwarding_drain_with(
6641 forwarding,
6642 ForwardingDrainContext {
6643 spec,
6644 runtime,
6645 registry,
6646 scope: DrainScope::Active,
6647 },
6648 snapshot,
6649 enabled,
6650 reason,
6651 runtime.drain_timeout,
6652 )
6653 .await
6654}
6655
6656async fn begin_forwarding_drain_if_configured(
6657 spec: &ModuleSpec,
6658 runtime: &SupervisorRuntimeConfig,
6659 registry: &Registry,
6660 snapshot: &SharedSnapshot,
6661 enabled: Option<bool>,
6662 reason: RouteCloseReason,
6663) -> Result<StopNotice, SuperviseError> {
6664 begin_forwarding_drain_with_timeout(
6665 spec,
6666 runtime,
6667 registry,
6668 snapshot,
6669 enabled,
6670 reason,
6671 runtime.drain_timeout,
6672 )
6673 .await
6674}
6675
6676async fn begin_forwarding_drain_with_timeout(
6680 spec: &ModuleSpec,
6681 runtime: &SupervisorRuntimeConfig,
6682 registry: &Registry,
6683 snapshot: &SharedSnapshot,
6684 enabled: Option<bool>,
6685 reason: RouteCloseReason,
6686 drain_timeout: Duration,
6687) -> Result<StopNotice, SuperviseError> {
6688 let Some(forwarding) = runtime.forwarding.as_ref() else {
6689 return Ok(StopNotice::NotSent);
6690 };
6691
6692 begin_forwarding_drain_with(
6693 forwarding,
6694 ForwardingDrainContext {
6695 spec,
6696 runtime,
6697 registry,
6698 scope: DrainScope::Active,
6699 },
6700 snapshot,
6701 enabled,
6702 reason,
6703 drain_timeout,
6704 )
6705 .await
6706}
6707
6708async fn begin_forwarding_drain_with(
6709 forwarding: &ForwardingTable,
6710 context: ForwardingDrainContext<'_>,
6711 snapshot: &SharedSnapshot,
6712 enabled: Option<bool>,
6713 reason: RouteCloseReason,
6714 drain_timeout: Duration,
6715) -> Result<StopNotice, SuperviseError> {
6716 let ForwardingDrainContext {
6717 spec,
6718 runtime,
6719 registry,
6720 scope,
6721 } = context;
6722 debug_assert_ne!(reason, RouteCloseReason::Crash);
6723 let terminal = matches!(reason, RouteCloseReason::Disable);
6724 let drain_started_at = Instant::now();
6725 let drain_deadline = drain_started_at + drain_timeout;
6726 let deadline_ms =
6727 unix_ms_now().saturating_add(u64::try_from(drain_timeout.as_millis()).unwrap_or(u64::MAX));
6728 let busy_gauges = match scope {
6729 DrainScope::Active => declared_busy_gauges(registry, &spec.module_id)?,
6730 DrainScope::Endpoint(endpoint) => {
6731 declared_busy_gauges_for_connection(registry, endpoint.connection_id)?
6732 }
6733 };
6734
6735 let gate_started = Instant::now();
6738 let drain_target = match scope {
6739 DrainScope::Active => forwarding.begin_module_drain(&spec.module_id, reason),
6740 DrainScope::Endpoint(endpoint) => forwarding.begin_endpoint_drain(endpoint, reason),
6741 }
6742 .map_err(SuperviseError::Forwarding)?;
6743 info!(
6748 module_id = %spec.module_id,
6749 ?reason,
6750 gate_ms = u64::try_from(gate_started.elapsed().as_millis()).unwrap_or(u64::MAX),
6751 connected = drain_target.is_some(),
6752 "module drain began; route admission closed"
6753 );
6754 if scope == DrainScope::Active {
6755 update_snapshot(snapshot, Some(&spec.module_id), |state| {
6756 state.state = ModuleState::Draining;
6757 state.draining_to_replace =
6758 matches!(reason, RouteCloseReason::Restart | RouteCloseReason::Reload);
6759 if let Some(enabled) = enabled {
6760 state.enabled = enabled;
6761 }
6762 })?;
6763 }
6764
6765 let Some(target) = drain_target.as_ref() else {
6766 return Ok(StopNotice::NoConnection);
6770 };
6771 {
6772 send_module_draining(&spec.module_id, reason, deadline_ms, target);
6773 let routes = forwarding
6774 .endpoint_routes(target.endpoint)
6775 .map_err(SuperviseError::Forwarding)?;
6776 let routes_notified = routes.len();
6777 crate::control::send_route_control_pushes(
6778 forwarding,
6779 routes.clone(),
6780 ClientControlPush::RouteClosing {
6781 module_id: spec.module_id.clone(),
6782 reason,
6783 },
6784 );
6785 send_route_goodbyes(forwarding, target.abandoned_bindings.clone());
6786
6787 let wait_result = wait_for_forwarding_quiescence(
6793 forwarding,
6794 &spec.module_id,
6795 runtime,
6796 target.endpoint,
6797 drain_deadline,
6798 &busy_gauges,
6799 scope,
6800 )
6801 .await;
6802 let drained = drained_after_quiescence_wait(&wait_result);
6803 if let Err(err) = &wait_result {
6804 error!(
6805 module_id = %spec.module_id,
6806 ?reason,
6807 error = %err,
6808 "forwarding quiescence wait failed after route.closing; forcing route.closed(drained: false) so the client is not left waiting on an unfulfilled promise"
6809 );
6810 } else if !drained {
6811 let holdouts = forwarding
6817 .endpoint_drain_holdouts(target.endpoint)
6818 .unwrap_or_default();
6819 warn!(
6820 module_id = %spec.module_id,
6821 waited = ?drain_timeout,
6822 ?reason,
6823 held_requests = holdouts.requests,
6824 held_routes = holdouts.routes,
6825 total_routes = holdouts.total_routes,
6826 top_connections = ?holdouts.top_connections,
6827 held = %holdouts
6830 .held
6831 .iter()
6832 .map(|(channel, corr)| format!("{channel}:{corr}"))
6833 .collect::<Vec<_>>()
6834 .join(","),
6835 "route drain timed out before request quiescence; forcing teardown"
6836 );
6837 }
6838 crate::control::send_route_control_pushes(
6839 forwarding,
6840 routes,
6841 ClientControlPush::RouteClosed {
6842 module_id: spec.module_id.clone(),
6843 reason,
6844 drained,
6845 abandoned: target.abandoned_bindings.len() as u32,
6846 excluded_subscriptions: target.excluded_subscriptions,
6847 terminal: Some(terminal),
6848 },
6849 );
6850 wait_result?;
6851
6852 let released_routes = match forwarding.release_module_endpoint_routes(target.endpoint) {
6858 Ok(routes) => routes,
6859 Err(err) => {
6860 warn!(
6861 module_id = %spec.module_id,
6862 ?reason,
6863 error = %err,
6864 "failed to release module endpoint routes after route.closed; module GOODBYE will still be sent"
6865 );
6866 send_module_goodbye(&spec.module_id, forwarding, target);
6867 return Err(SuperviseError::Forwarding(err));
6868 }
6869 };
6870 let route_goodbye_count = released_routes.len();
6871 send_route_goodbyes(forwarding, released_routes);
6872 send_module_goodbye(&spec.module_id, forwarding, target);
6873
6874 info!(
6880 module_id = %spec.module_id,
6881 ?reason,
6882 routes_notified,
6883 route_goodbyes = route_goodbye_count,
6884 abandoned_reservations = target.abandoned_bindings.len(),
6885 excluded_subscriptions = target.excluded_subscriptions,
6886 drained,
6887 "module drain complete; consumers notified via route.closing/route.closed pushes and per-route GOODBYE frames"
6888 );
6889 }
6890
6891 Ok(StopNotice::SentOverConnection)
6892}
6893
6894async fn wait_for_registration_after_reload(
6897 registry: &Registry,
6898 module_id: &str,
6899 snapshot: &SharedSnapshot,
6900 child: &mut SupervisedChild,
6901 wait: Duration,
6902) -> Result<RegistrationWaitOutcome, SuperviseError> {
6903 wait_for_slot_registration(
6904 registry,
6905 crate::registry::RegistrationSlot::Active(module_id),
6906 module_id,
6907 snapshot,
6908 child,
6909 wait,
6910 )
6911 .await
6912}
6913
6914async fn wait_for_slot_registration(
6922 registry: &Registry,
6923 slot: crate::registry::RegistrationSlot<'_>,
6924 module_id: &str,
6925 snapshot: &SharedSnapshot,
6926 child: &mut SupervisedChild,
6927 wait: Duration,
6928) -> Result<RegistrationWaitOutcome, SuperviseError> {
6929 let deadline = Instant::now() + wait;
6930 loop {
6931 if registry
6932 .registration(slot)
6933 .map_err(SuperviseError::Registry)?
6934 .is_some()
6935 {
6936 return Ok(RegistrationWaitOutcome::Registered);
6937 }
6938
6939 let now = Instant::now();
6940 if now >= deadline {
6941 return Ok(RegistrationWaitOutcome::TimedOut);
6942 }
6943 let remaining = deadline.saturating_duration_since(now);
6944 let poll = remaining.min(REGISTRY_RELEASE_POLL);
6945
6946 tokio::select! {
6947 wait_result = child.wait() => {
6948 let status = wait_result.map_err(|source| SuperviseError::Wait {
6949 module_id: module_id.to_string(),
6950 source,
6951 })?;
6952 return Ok(RegistrationWaitOutcome::Exited(classify_reaped_child_exit(
6953 snapshot,
6954 child,
6955 &status,
6956 )));
6957 }
6958 _ = sleep(poll) => {}
6959 }
6960 }
6961}
6962
6963fn registration_failure_exit_report(mut exit_report: ExitReport) -> ExitReport {
6964 if exit_report.kind != ExitKind::DeliberateSeverance {
6967 exit_report.kind = ExitKind::Crash;
6968 }
6969 exit_report
6970}
6971
6972async fn handle_reload_child_registration_failure(
6973 spec: &ModuleSpec,
6974 runtime: &SupervisorRuntimeConfig,
6975 registry: &Registry,
6976 process_liveness: &SupervisorProcessLiveness,
6977 snapshot: &SharedSnapshot,
6978 child: &mut Option<SupervisedChild>,
6979 failure: ReloadRegistrationFailure,
6980) -> Result<(), SuperviseError> {
6981 let ReloadRegistrationFailure {
6982 exit_report,
6983 reason,
6984 } = failure;
6985 match on_child_exit(
6986 spec,
6987 runtime.restart_policy,
6988 registry,
6989 snapshot,
6990 &runtime.terminal_ring,
6991 &runtime.spawn_events,
6992 &runtime.child_roster,
6993 exit_report,
6994 )
6995 .await
6996 {
6997 NextAction::Stop {
6998 registration_released,
6999 } => {
7000 if registration_released {
7001 process_liveness.untrack_if_current(&spec.module_id, snapshot);
7002 }
7003 }
7004 NextAction::Restart { schedule } => {
7005 let delay = schedule.map_or(runtime.restart_policy.delay_for_restart(0), |schedule| {
7006 schedule.delay
7007 });
7008 if let Some(schedule) = schedule {
7009 log_crash_respawn(&spec.module_id, schedule);
7010 }
7011 sleep(delay).await;
7012 if respawn_still_pending(snapshot) {
7016 if let Err(err) = wait_for_registration_release(
7017 registry,
7018 &spec.module_id,
7019 REGISTRY_RELEASE_TIMEOUT,
7020 )
7021 .await
7022 {
7023 fail_snapshot(snapshot, Some(&spec.module_id), None);
7024 process_liveness.untrack_if_current(&spec.module_id, snapshot);
7025 return Err(SuperviseError::ReloadFailed {
7026 module_id: spec.module_id.clone(),
7027 reason: format!(
7028 "{reason}; registration did not release before policy retry: {err}"
7029 ),
7030 });
7031 }
7032 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
7033 match spawn_and_mark_running(spec, runtime, snapshot) {
7034 Ok(next_child) => {
7035 *child = Some(next_child);
7036 }
7037 Err(err) => {
7038 fail_snapshot(snapshot, Some(&spec.module_id), None);
7039 process_liveness.untrack_if_current(&spec.module_id, snapshot);
7040 return Err(SuperviseError::ReloadFailed {
7041 module_id: spec.module_id.clone(),
7042 reason: format!("{reason}; policy retry spawn failed: {err}"),
7043 });
7044 }
7045 }
7046 }
7047 }
7048 }
7049
7050 Err(SuperviseError::ReloadFailed {
7051 module_id: spec.module_id.clone(),
7052 reason,
7053 })
7054}
7055
7056async fn handle_reload_spawn_failure(
7057 spec: &ModuleSpec,
7058 runtime: &SupervisorRuntimeConfig,
7059 process_liveness: &SupervisorProcessLiveness,
7060 snapshot: &SharedSnapshot,
7061 child: &mut Option<SupervisedChild>,
7062 reason: String,
7063) -> Result<(), SuperviseError> {
7064 let mut should_retry = false;
7065 let now = Instant::now();
7066 update_snapshot(snapshot, Some(&spec.module_id), |state| {
7067 clear_current_process_facts(state);
7068 if daemon_will_restart(state, &runtime.restart_policy, now) {
7069 state.record_crash_restart(&runtime.restart_policy, now);
7070 state.state = ModuleState::Restarting;
7071 should_retry = true;
7072 } else if state.enabled {
7073 state.state = ModuleState::Failed;
7074 } else {
7075 state.state = ModuleState::Disabled;
7076 }
7077 })?;
7078
7079 if should_retry {
7080 sleep(runtime.restart_policy.backoff).await;
7081 if respawn_still_pending(snapshot) {
7085 process_liveness.track(spec.module_id.clone(), Arc::clone(snapshot));
7086 match spawn_and_mark_running(spec, runtime, snapshot) {
7087 Ok(next_child) => {
7088 *child = Some(next_child);
7089 }
7090 Err(err) => {
7091 fail_snapshot(snapshot, Some(&spec.module_id), None);
7092 process_liveness.untrack_if_current(&spec.module_id, snapshot);
7093 return Err(SuperviseError::ReloadFailed {
7094 module_id: spec.module_id.clone(),
7095 reason: format!("{reason}; policy retry spawn failed: {err}"),
7096 });
7097 }
7098 }
7099 }
7100 } else {
7101 process_liveness.untrack_if_current(&spec.module_id, snapshot);
7102 }
7103
7104 Err(SuperviseError::ReloadFailed {
7105 module_id: spec.module_id.clone(),
7106 reason,
7107 })
7108}
7109
7110fn control_flags() -> Flags {
7111 Flags::new(false, Priority::Passive, false)
7112}
7113
7114#[allow(clippy::too_many_arguments)]
7115async fn drain_optional_child(
7116 module_id: &str,
7117 protocol: ModuleProtocol,
7118 stop_notice: StopNotice,
7119 registry: &Registry,
7120 snapshot: &SharedSnapshot,
7121 terminal_ring: &Arc<Mutex<TerminalRing>>,
7122 spawn_events: &SpawnEventFeed,
7123 child: &mut Option<SupervisedChild>,
7124 drain_timeout: Duration,
7125 final_state: ModuleState,
7126 enabled: Option<bool>,
7127) -> Result<(), SuperviseError> {
7128 if let Some(child) = child.take() {
7129 drain_child_to_state(
7130 module_id,
7131 protocol,
7132 stop_notice,
7133 registry,
7134 snapshot,
7135 terminal_ring,
7136 spawn_events,
7137 child,
7138 drain_timeout,
7139 final_state,
7140 enabled,
7141 )
7142 .await
7143 } else {
7144 update_snapshot(snapshot, Some(module_id), |state| {
7145 state.state = final_state;
7146 if let Some(enabled) = enabled {
7147 state.enabled = enabled;
7148 }
7149 clear_current_process_facts(state);
7150 })?;
7151 wait_for_registration_release(registry, module_id, REGISTRY_RELEASE_TIMEOUT).await
7152 }
7153}
7154
7155#[allow(clippy::too_many_arguments)]
7156async fn drain_child_to_state(
7157 module_id: &str,
7158 protocol: ModuleProtocol,
7159 stop_notice: StopNotice,
7160 registry: &Registry,
7161 snapshot: &SharedSnapshot,
7162 terminal_ring: &Arc<Mutex<TerminalRing>>,
7163 spawn_events: &SpawnEventFeed,
7164 mut child: SupervisedChild,
7165 drain_timeout: Duration,
7166 final_state: ModuleState,
7167 enabled: Option<bool>,
7168) -> Result<(), SuperviseError> {
7169 update_snapshot(snapshot, Some(module_id), |state| {
7170 state.state = ModuleState::Draining;
7171 state.draining_to_replace = final_state == ModuleState::Restarting;
7172 if let Some(enabled) = enabled {
7173 state.enabled = enabled;
7174 }
7175 })?;
7176
7177 if stop_notice != StopNotice::SentOverConnection {
7188 if protocol == ModuleProtocol::Subc && stop_notice == StopNotice::NoConnection {
7189 info!(
7190 module_id,
7191 pid = child.pid,
7192 budget_ms = u64::try_from(drain_timeout.as_millis()).unwrap_or(u64::MAX),
7193 "module has no connection yet; requesting stop by signal"
7194 );
7195 }
7196 request_graceful_stop(module_id, &child);
7197 }
7198
7199 let exit_report = match timeout(drain_timeout, child.wait()).await {
7200 Ok(Ok(status)) => classify_reaped_child_exit(snapshot, &child, &status),
7201 Ok(Err(source)) => {
7202 fail_snapshot(snapshot, Some(module_id), None);
7203 return Err(SuperviseError::Wait {
7204 module_id: module_id.to_string(),
7205 source,
7206 });
7207 }
7208 Err(_) => {
7209 warn!(
7222 module_id,
7223 pid = child.pid,
7224 budget_ms = u64::try_from(drain_timeout.as_millis()).unwrap_or(u64::MAX),
7225 reason = ?final_state,
7226 ?stop_notice,
7227 "drain budget expired before the module exited; killing it"
7228 );
7229 child.start_kill().map_err(|source| {
7230 fail_snapshot(snapshot, Some(module_id), None);
7231 SuperviseError::Kill {
7232 module_id: module_id.to_string(),
7233 source,
7234 }
7235 })?;
7236 let status = child.wait().await.map_err(|source| {
7237 fail_snapshot(snapshot, Some(module_id), None);
7238 SuperviseError::Wait {
7239 module_id: module_id.to_string(),
7240 source,
7241 }
7242 })?;
7243 classify_reaped_child_exit(snapshot, &child, &status)
7244 }
7245 };
7246
7247 update_snapshot(snapshot, Some(module_id), |state| {
7248 state.state = final_state;
7249 if let Some(enabled) = enabled {
7250 state.enabled = enabled;
7251 }
7252 clear_current_process_facts(state);
7253 state.last_exit = Some(exit_report.clone());
7254 if exit_report.kind == ExitKind::DeliberateSeverance {
7255 state.lifetime_restarts += 1;
7256 }
7257 })?;
7258 record_terminal(
7259 module_id,
7260 terminal_ring,
7261 spawn_events,
7262 &exit_report,
7263 terminal_disposition(final_state),
7264 );
7265 child.drain_stderr(module_id).await;
7266
7267 wait_for_registration_release(registry, module_id, REGISTRY_RELEASE_TIMEOUT).await
7268}
7269
7270#[cfg(unix)]
7290fn request_graceful_stop(module_id: &str, child: &SupervisedChild) {
7291 let Some(pid) = child
7292 .id()
7293 .and_then(|pid| i32::try_from(pid).ok())
7294 .and_then(rustix::process::Pid::from_raw)
7295 else {
7296 debug!(
7297 module_id,
7298 "no pid to signal for teardown; falling through to the drain wait"
7299 );
7300 return;
7301 };
7302 match rustix::process::kill_process(pid, rustix::process::Signal::TERM) {
7303 Ok(()) => debug!(
7304 module_id,
7305 "sent SIGTERM to a module nothing else asked to stop"
7306 ),
7307 Err(err) => debug!(
7308 module_id,
7309 error = %err,
7310 "SIGTERM to module failed; the drain wait and kill still apply"
7311 ),
7312 }
7313}
7314
7315#[cfg(not(unix))]
7323fn request_graceful_stop(module_id: &str, _child: &SupervisedChild) {
7324 debug!(
7325 module_id,
7326 "no graceful stop signal exists on this platform; teardown of a module nothing asked to stop waits, then kills"
7327 );
7328}
7329
7330fn terminal_disposition(final_state: ModuleState) -> TerminalDisposition {
7331 match final_state {
7332 ModuleState::Stopped => TerminalDisposition::Stopped,
7333 ModuleState::Disabled => TerminalDisposition::Disabled,
7334 ModuleState::Restarting => TerminalDisposition::Restarting,
7335 ModuleState::Failed => TerminalDisposition::Failed,
7336 ModuleState::Starting
7337 | ModuleState::Running
7338 | ModuleState::Unresponsive
7339 | ModuleState::Draining => {
7340 unreachable!("terminal exits only finish in terminal or restarting states")
7341 }
7342 }
7343}
7344
7345async fn wait_for_registration_release(
7348 registry: &Registry,
7349 module_id: &str,
7350 wait: Duration,
7351) -> Result<(), SuperviseError> {
7352 wait_for_slot_registration_release(
7353 registry,
7354 crate::registry::RegistrationSlot::Active(module_id),
7355 wait,
7356 )
7357 .await
7358}
7359
7360async fn wait_for_slot_registration_release(
7368 registry: &Registry,
7369 slot: crate::registry::RegistrationSlot<'_>,
7370 wait: Duration,
7371) -> Result<(), SuperviseError> {
7372 let deadline = Instant::now() + wait;
7373 let mut release_events = registration_release_events().subscribe();
7374 let still_active = |registration: &crate::registry::ModuleRegistration| {
7375 SuperviseError::RegistrationStillActive {
7376 module_id: registration.manifest.module_id.clone(),
7377 waited: wait,
7378 }
7379 };
7380 loop {
7381 let _observed_generation = *release_events.borrow_and_update();
7382 let Some(registration) = registry
7383 .registration(slot)
7384 .map_err(SuperviseError::Registry)?
7385 else {
7386 return Ok(());
7387 };
7388
7389 let now = Instant::now();
7390 if now >= deadline {
7391 return Err(still_active(®istration));
7392 }
7393
7394 let remaining = deadline.saturating_duration_since(now);
7395 match timeout(remaining, release_events.changed()).await {
7396 Ok(Ok(())) | Ok(Err(_)) => {}
7397 Err(_) => return Err(still_active(®istration)),
7398 }
7399 }
7400}
7401
7402#[cfg(test)]
7403mod slot_registration_wait_tests {
7404 use super::*;
7405 use crate::registry::{ConnectionId, RegistrationSlot};
7406 use subc_protocol::manifest::ModuleManifest;
7407
7408 const INCUMBENT: u64 = 1;
7409 const CANDIDATE: u64 = 2;
7410
7411 fn swapped_registry() -> Arc<Registry> {
7412 let registry = Arc::new(Registry::default());
7413 let manifest = ModuleManifest::builder("m", "0.1.0").build();
7414 registry
7415 .register_with_control_ops(
7416 manifest.clone(),
7417 1,
7418 ConnectionId::new(INCUMBENT),
7419 Vec::new(),
7420 )
7421 .unwrap();
7422 registry
7423 .register_candidate_with_control_ops(
7424 manifest,
7425 1,
7426 ConnectionId::new(CANDIDATE),
7427 Vec::new(),
7428 )
7429 .unwrap();
7430 registry
7431 }
7432
7433 #[tokio::test]
7437 async fn incumbent_release_is_awaited_by_connection_not_by_module_id() {
7438 let registry = swapped_registry();
7439 registry.promote_candidate("m").unwrap().unwrap();
7440
7441 assert!(matches!(
7442 wait_for_registration_release(®istry, "m", Duration::from_millis(50)).await,
7443 Err(SuperviseError::RegistrationStillActive { .. })
7444 ));
7445
7446 assert!(matches!(
7448 wait_for_slot_registration_release(
7449 ®istry,
7450 RegistrationSlot::Connection(ConnectionId::new(INCUMBENT)),
7451 Duration::from_millis(50),
7452 )
7453 .await,
7454 Err(SuperviseError::RegistrationStillActive { .. })
7455 ));
7456
7457 let releaser = Arc::clone(®istry);
7458 let release = tokio::spawn(async move {
7459 sleep(Duration::from_millis(20)).await;
7460 releaser
7461 .deregister_connection(ConnectionId::new(INCUMBENT))
7462 .unwrap();
7463 notify_registration_release();
7464 });
7465 wait_for_slot_registration_release(
7466 ®istry,
7467 RegistrationSlot::Connection(ConnectionId::new(INCUMBENT)),
7468 Duration::from_secs(5),
7469 )
7470 .await
7471 .expect("the incumbent's own registration is released");
7472 release.await.unwrap();
7473 assert!(registry.get_module("m").unwrap().is_some());
7474 }
7475
7476 #[tokio::test]
7479 async fn candidate_slot_wait_ignores_the_incumbents_registration() {
7480 let registry = swapped_registry();
7481 assert!(matches!(
7482 wait_for_slot_registration_release(
7483 ®istry,
7484 RegistrationSlot::Candidate("m"),
7485 Duration::from_millis(50),
7486 )
7487 .await,
7488 Err(SuperviseError::RegistrationStillActive { .. })
7489 ));
7490 registry
7491 .deregister_connection(ConnectionId::new(CANDIDATE))
7492 .unwrap();
7493 wait_for_slot_registration_release(
7494 ®istry,
7495 RegistrationSlot::Candidate("m"),
7496 Duration::from_millis(50),
7497 )
7498 .await
7499 .expect("a candidate slot with no candidate is released");
7500 assert!(registry
7501 .registration(RegistrationSlot::Active("m"))
7502 .unwrap()
7503 .is_some());
7504 }
7505}
7506
7507fn classify_exit(status: &ExitStatus) -> ExitReport {
7508 ExitReport {
7509 kind: if status.success() {
7510 ExitKind::Clean
7511 } else {
7512 ExitKind::Crash
7513 },
7514 code: status.code(),
7515 signal: exit_signal(status),
7516 at_ms: unix_ms_now(),
7517 }
7518}
7519
7520fn wait_error_exit_report() -> ExitReport {
7526 ExitReport {
7527 kind: ExitKind::Crash,
7528 code: None,
7529 signal: None,
7530 at_ms: unix_ms_now(),
7531 }
7532}
7533
7534#[cfg(unix)]
7535fn exit_signal(status: &ExitStatus) -> Option<i32> {
7536 use std::os::unix::process::ExitStatusExt;
7537
7538 status.signal()
7539}
7540
7541#[cfg(not(unix))]
7542fn exit_signal(_status: &ExitStatus) -> Option<i32> {
7543 None
7544}
7545
7546fn reset_restart_count(snapshot: &SharedSnapshot, module_id: &str) -> Result<(), SuperviseError> {
7552 update_snapshot(snapshot, Some(module_id), |state| {
7553 state.clear_crash_restarts();
7554 })
7555}
7556
7557fn set_running(
7558 snapshot: &SharedSnapshot,
7559 child: &SupervisedChild,
7560 module_id: &str,
7561 spawn_events: &SpawnEventFeed,
7562) -> Result<(), SuperviseError> {
7563 let mut state = snapshot.lock().map_err(|_| SuperviseError::StatePoisoned {
7564 module_id: Some(module_id.to_string()),
7565 })?;
7566 state.spawn_generation = spawn_events.emit_spawned(module_id, child.pid, child.spawned_at_ms);
7567 state.in_alternate_slot = false;
7570 state.configuration_updated_since_spawn = false;
7571 state.state = ModuleState::Running;
7572 state.enabled = true;
7573 state.process_alive = true;
7574 state.pid = child.id();
7575 state.spawned_at_ms = Some(child.spawned_at_ms);
7576 state.spawned_from = Some(child.spawned_from.clone());
7577 state.spawned_file_identity = child.spawned_file_identity;
7578 state.process_start_time = child.process_start_time;
7579 Ok(())
7580}
7581
7582fn clear_current_process_facts(state: &mut SupervisorSnapshot) {
7583 state.process_alive = false;
7584 state.pid = None;
7585 state.spawned_at_ms = None;
7586 state.spawned_from = None;
7587 state.spawned_file_identity = None;
7588 state.process_start_time = None;
7589 state.deliberate_severance = None;
7590}
7591
7592#[cfg(test)]
7593fn record_deliberate_severance(
7594 snapshot: &SharedSnapshot,
7595 identity: ProcessIdentity,
7596) -> Result<(), SuperviseError> {
7597 update_snapshot(snapshot, None, |state| {
7598 state.deliberate_severance = Some(identity);
7599 })
7600}
7601
7602fn apply_deliberate_severance_marker(
7603 snapshot: &SharedSnapshot,
7604 exited_identity: Option<ProcessIdentity>,
7605 mut exit_report: ExitReport,
7606) -> ExitReport {
7607 let marker = lock_snapshot(snapshot)
7608 .ok()
7609 .and_then(|mut state| state.deliberate_severance.take());
7610 if marker.is_some() && marker == exited_identity {
7611 exit_report.kind = ExitKind::DeliberateSeverance;
7612 }
7613 exit_report
7614}
7615
7616fn classify_reaped_child_exit(
7617 snapshot: &SharedSnapshot,
7618 child: &SupervisedChild,
7619 status: &ExitStatus,
7620) -> ExitReport {
7621 apply_deliberate_severance_marker(snapshot, child.process_identity(), classify_exit(status))
7622}
7623
7624fn fail_snapshot(
7625 snapshot: &SharedSnapshot,
7626 module_id: Option<&str>,
7627 last_exit: Option<ExitReport>,
7628) {
7629 if let Err(err) = update_snapshot(snapshot, module_id, |state| {
7630 state.state = ModuleState::Failed;
7631 clear_current_process_facts(state);
7632 if let Some(last_exit) = last_exit {
7633 state.last_exit = Some(last_exit);
7634 }
7635 }) {
7636 error!(error = %err, "failed to mark supervisor state failed");
7637 }
7638}
7639
7640fn update_snapshot(
7641 snapshot: &SharedSnapshot,
7642 module_id: Option<&str>,
7643 update: impl FnOnce(&mut SupervisorSnapshot),
7644) -> Result<(), SuperviseError> {
7645 let mut state = snapshot.lock().map_err(|_| SuperviseError::StatePoisoned {
7646 module_id: module_id.map(ToOwned::to_owned),
7647 })?;
7648 update(&mut state);
7649 Ok(())
7650}
7651
7652const SLOW_SNAPSHOT_LOCK_THRESHOLD: Duration = Duration::from_millis(250);
7653
7654fn lock_snapshot_for_control<'a>(
7655 snapshot: &'a SharedSnapshot,
7656 module_id: &str,
7657 caller: &'static str,
7658) -> Result<std::sync::MutexGuard<'a, SupervisorSnapshot>, SuperviseError> {
7659 let started_at = Instant::now();
7660 let guard = lock_snapshot(snapshot)?;
7661 let waited = started_at.elapsed();
7662 if waited >= SLOW_SNAPSHOT_LOCK_THRESHOLD {
7663 warn!(
7664 module_id = %module_id,
7665 waited_ms = waited.as_millis() as u64,
7666 caller = %caller,
7667 "slow snapshot lock"
7668 );
7669 }
7670 Ok(guard)
7671}
7672
7673fn lock_snapshot(
7674 snapshot: &SharedSnapshot,
7675) -> Result<std::sync::MutexGuard<'_, SupervisorSnapshot>, SuperviseError> {
7676 snapshot
7677 .lock()
7678 .map_err(|_| SuperviseError::StatePoisoned { module_id: None })
7679}
7680
7681#[cfg(test)]
7682mod terminal_history_tests {
7683 use std::{
7684 path::PathBuf,
7685 sync::Arc,
7686 time::{Duration, Instant},
7687 };
7688
7689 use tokio::time::sleep;
7690
7691 use super::{
7692 apply_deliberate_severance_marker, daemon_will_restart, drain_child_to_state,
7693 drained_after_quiescence_wait, handle_reload_spawn_failure, health_restart_child,
7694 lock_snapshot, on_child_exit, record_deliberate_severance, record_wait_error_terminal,
7695 reset_restart_count, spawn_and_mark_running, update_snapshot, wait_error_exit_report,
7696 ExitKind, ExitReport, ModuleProtocol, ModuleSpec, ModuleState, NextAction, ProcessIdentity,
7697 RestartPolicy, SpawnEventKind, StopNotice, SuperviseError, SupervisedModule, Supervisor,
7698 SupervisorHandle, SupervisorHealthStatus, SupervisorSnapshot,
7699 };
7700 use super::Instant as ClockInstant;
7705 use crate::{
7706 registry::Registry,
7707 terminal_ring::{TerminalRing, TerminalRingConfig},
7708 };
7709 use std::sync::Mutex;
7710 use subc_control::TerminalDisposition;
7711
7712 fn fake_aft_stub_path() -> PathBuf {
7717 let mut path = std::env::current_exe().expect("current_exe available in tests");
7718 path.pop();
7719 path.pop();
7720 path.push(if cfg!(windows) {
7721 "fake-aft-stub.exe"
7722 } else {
7723 "fake-aft-stub"
7724 });
7725 assert!(
7726 path.exists(),
7727 "fake-aft-stub not built at {}: run `cargo test -p subc-core` (which builds \
7728 [[bin]] targets) rather than `cargo test -p subc-core --lib` (which does not)",
7729 path.display()
7730 );
7731 path
7732 }
7733
7734 #[test]
7735 fn reserved_never_spawned_refuses_every_hello() {
7736 let supervisor = SupervisorHandle::default();
7741 supervisor.apply_identity_configuration(&ModuleSpec {
7742 module_id: "never-spawned".to_string(),
7743 program: PathBuf::from("/usr/bin/false"),
7744 args: Vec::new(),
7745 env: Vec::new(),
7746 reserved: true,
7747 reserved_prefixes: Vec::new(),
7748 protocol: ModuleProtocol::Subc,
7749 overlap: Default::default(),
7750 });
7751 assert!(
7752 supervisor
7753 .reserved_hello_rejection("never-spawned", Some("any-forged-nonce"))
7754 .is_some(),
7755 "forged nonce must refuse on a reserved never-spawned id"
7756 );
7757 assert!(
7758 supervisor
7759 .reserved_hello_rejection("never-spawned", None)
7760 .is_some(),
7761 "absent nonce must refuse on a reserved never-spawned id"
7762 );
7763 supervisor.set_spawn_nonce("never-spawned", "minted".to_string());
7765 supervisor.apply_identity_configuration(&ModuleSpec {
7766 module_id: "never-spawned".to_string(),
7767 program: PathBuf::from("/usr/bin/false"),
7768 args: Vec::new(),
7769 env: Vec::new(),
7770 reserved: true,
7771 reserved_prefixes: Vec::new(),
7772 protocol: ModuleProtocol::Subc,
7773 overlap: Default::default(),
7774 });
7775 assert!(supervisor
7776 .reserved_hello_rejection("never-spawned", Some("minted"))
7777 .is_none());
7778 assert!(supervisor
7779 .reserved_hello_rejection("never-spawned", Some("forged"))
7780 .is_some());
7781 }
7782
7783 fn seed_crash_restarts(state: &mut SupervisorSnapshot, count: u32) {
7786 let now = ClockInstant::now();
7787 for _ in 0..count {
7788 state.crash_restarts.push_back(now);
7789 }
7790 }
7791
7792 fn age_oldest_crash_restart_out_of_window(state: &mut SupervisorSnapshot, window: Duration) {
7796 let aged = state
7797 .crash_restarts
7798 .front()
7799 .expect("a crash restart must be recorded before it can be aged")
7800 .checked_sub(window + Duration::from_secs(1))
7801 .expect("the test clock is far enough from its origin to age an instant");
7802 state.crash_restarts[0] = aged;
7803 }
7804
7805 fn snapshot_with_restarts(enabled: bool, count: u32) -> SupervisorSnapshot {
7806 let mut state = SupervisorSnapshot::new(ModuleState::Running, enabled);
7807 seed_crash_restarts(&mut state, count);
7808 state
7809 }
7810
7811 #[test]
7812 fn daemon_owned_recovery_predicate_uses_the_pre_increment_budget() {
7813 let policy = RestartPolicy::new(3, Duration::ZERO);
7814 let now = ClockInstant::now();
7815 assert!(daemon_will_restart(
7816 &mut snapshot_with_restarts(true, 2),
7817 &policy,
7818 now
7819 ));
7820 assert!(!daemon_will_restart(
7821 &mut snapshot_with_restarts(true, 3),
7822 &policy,
7823 now
7824 ));
7825 assert!(!daemon_will_restart(
7826 &mut snapshot_with_restarts(false, 0),
7827 &policy,
7828 now
7829 ));
7830 }
7831
7832 #[test]
7833 fn crash_restart_backoff_escalates_with_in_window_count() {
7834 let policy = RestartPolicy::new(4, Duration::from_millis(100))
7835 .with_max_backoff(Duration::from_secs(30));
7836 let now = ClockInstant::now();
7837 let mut state = SupervisorSnapshot::new(ModuleState::Running, true);
7838 let schedules = (0..4)
7839 .map(|_| {
7840 state
7841 .next_crash_restart(&policy, now)
7842 .expect("the test policy allows four crash restarts")
7843 })
7844 .collect::<Vec<_>>();
7845
7846 assert_eq!(
7847 schedules
7848 .iter()
7849 .map(|schedule| schedule.restart_in_window)
7850 .collect::<Vec<_>>(),
7851 vec![0, 1, 2, 3]
7852 );
7853 assert_eq!(
7854 schedules
7855 .iter()
7856 .map(|schedule| schedule.delay)
7857 .collect::<Vec<_>>(),
7858 vec![
7859 Duration::from_millis(100),
7860 Duration::from_secs(1),
7861 Duration::from_secs(10),
7862 Duration::from_secs(30),
7863 ]
7864 );
7865 }
7866
7867 #[test]
7868 fn crash_restart_backoff_resets_after_ring_clear() {
7869 let policy = RestartPolicy::new(3, Duration::from_millis(100));
7870 let now = ClockInstant::now();
7871 let mut state = SupervisorSnapshot::new(ModuleState::Running, true);
7872 assert_eq!(
7873 state.next_crash_restart(&policy, now).unwrap().delay,
7874 Duration::from_millis(100)
7875 );
7876 assert_eq!(
7877 state.next_crash_restart(&policy, now).unwrap().delay,
7878 Duration::from_secs(1)
7879 );
7880
7881 state.clear_crash_restarts();
7882 let schedule = state
7883 .next_crash_restart(&policy, now)
7884 .expect("a cleared ring must allow another restart");
7885 assert_eq!(schedule.restart_in_window, 0);
7886 assert_eq!(schedule.delay, Duration::from_millis(100));
7887 }
7888
7889 #[test]
7890 fn crash_restart_backoff_ignores_aged_restarts() {
7891 let policy = RestartPolicy::new(3, Duration::from_millis(100));
7892 let now = ClockInstant::now();
7893 let mut state = SupervisorSnapshot::new(ModuleState::Running, true);
7894 state
7895 .next_crash_restart(&policy, now)
7896 .expect("the first restart is allowed");
7897 state
7898 .next_crash_restart(&policy, now)
7899 .expect("the second restart is allowed");
7900 state.crash_restarts[0] = now
7901 .checked_sub(policy.window + Duration::from_secs(1))
7902 .expect("the fake clock can age a restart past the window");
7903
7904 let schedule = state
7905 .next_crash_restart(&policy, now)
7906 .expect("an aged restart must release its slot");
7907 assert_eq!(schedule.restart_in_window, 1);
7908 assert_eq!(schedule.delay, Duration::from_secs(1));
7909 assert_eq!(state.crash_restarts.len(), 2);
7910 }
7911
7912 #[test]
7916 fn a_budget_spent_before_the_window_no_longer_refuses() {
7917 let policy = RestartPolicy::new(3, Duration::ZERO);
7918 let mut state = snapshot_with_restarts(true, 3);
7919 let now = ClockInstant::now();
7920 assert!(!daemon_will_restart(&mut state, &policy, now));
7921
7922 assert!(daemon_will_restart(
7923 &mut state,
7924 &policy,
7925 now + policy.window + Duration::from_secs(1)
7926 ));
7927 assert!(
7928 state.crash_restarts.is_empty(),
7929 "reading the budget must drop the instants that left the window"
7930 );
7931 }
7932
7933 fn module_with_recovery_snapshot(
7934 state: ModuleState,
7935 enabled: bool,
7936 restart_count: u32,
7937 ) -> SupervisedModule {
7938 let registry = Arc::new(Registry::default());
7939 let supervisor =
7940 Supervisor::new(Arc::clone(®istry), RestartPolicy::new(3, Duration::ZERO));
7941 let module = supervisor
7942 .spawn(ModuleSpec {
7943 module_id: "recovery-snapshot".to_string(),
7944 program: fake_aft_stub_path(),
7945 args: Vec::new(),
7946 env: Vec::new(),
7947 reserved: false,
7948 reserved_prefixes: Vec::new(),
7949 protocol: ModuleProtocol::Subc,
7950 overlap: Default::default(),
7951 })
7952 .unwrap();
7953 update_snapshot(
7954 &module.inner.snapshot,
7955 Some("recovery-snapshot"),
7956 |snapshot| {
7957 snapshot.state = state;
7958 snapshot.enabled = enabled;
7959 seed_crash_restarts(snapshot, restart_count);
7960 },
7961 )
7962 .unwrap();
7963 module
7964 }
7965
7966 #[cfg(target_os = "linux")]
7967 #[tokio::test]
7968 async fn no_cgroup_placement_does_not_block_fake_aft_stub_spawn() {
7969 let supervisor = Supervisor::new(Arc::new(Registry::default()), RestartPolicy::default())
7970 .with_cgroup_placement(None);
7971 let result = supervisor.spawn(ModuleSpec {
7972 module_id: "no-cgroup-placement".to_string(),
7973 program: fake_aft_stub_path(),
7974 args: Vec::new(),
7975 env: Vec::new(),
7976 reserved: false,
7977 reserved_prefixes: Vec::new(),
7978 protocol: ModuleProtocol::Subc,
7979 overlap: Default::default(),
7980 });
7981
7982 assert!(
7983 result.is_ok(),
7984 "no delegation must not turn an otherwise valid spawn into a failure: {result:?}"
7985 );
7986 }
7987
7988 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
7989 async fn undecided_snapshot_uses_shared_restart_predicate() {
7990 assert!(module_with_recovery_snapshot(ModuleState::Running, true, 2)
7991 .will_recover_after_connection_loss()
7992 .unwrap());
7993 assert!(
7994 !module_with_recovery_snapshot(ModuleState::Running, true, 3)
7995 .will_recover_after_connection_loss()
7996 .unwrap()
7997 );
7998 }
7999
8000 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
8001 async fn restarting_snapshot_at_exhausted_budget_is_non_terminal() {
8002 assert!(
8003 module_with_recovery_snapshot(ModuleState::Restarting, true, 3)
8004 .will_recover_after_connection_loss()
8005 .unwrap()
8006 );
8007 }
8008
8009 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
8010 async fn terminal_phase_snapshots_are_terminal_before_budget_exhaustion() {
8011 assert!(!module_with_recovery_snapshot(ModuleState::Failed, true, 0)
8012 .will_recover_after_connection_loss()
8013 .unwrap());
8014 assert!(
8015 !module_with_recovery_snapshot(ModuleState::Disabled, true, 0)
8016 .will_recover_after_connection_loss()
8017 .unwrap()
8018 );
8019 }
8020
8021 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
8022 async fn warming_snapshot_is_limited_to_startup_phases() {
8023 for state in [
8024 ModuleState::Starting,
8025 ModuleState::Running,
8026 ModuleState::Restarting,
8027 ] {
8028 assert!(
8029 module_with_recovery_snapshot(state, true, 0)
8030 .is_warming()
8031 .unwrap(),
8032 "{state:?} should be warming"
8033 );
8034 }
8035 for state in [
8036 ModuleState::Unresponsive,
8037 ModuleState::Draining,
8038 ModuleState::Stopped,
8039 ModuleState::Failed,
8040 ModuleState::Disabled,
8041 ] {
8042 assert!(
8043 !module_with_recovery_snapshot(state, true, 0)
8044 .is_warming()
8045 .unwrap(),
8046 "{state:?} should not be warming"
8047 );
8048 }
8049 }
8050
8051 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
8052 async fn terminal_history_survives_respawn_and_keeps_both_crashes_in_order() {
8053 let registry = Arc::new(Registry::default());
8054 let supervisor =
8055 Supervisor::new(Arc::clone(®istry), RestartPolicy::new(1, Duration::ZERO));
8056 let module = supervisor
8057 .spawn(ModuleSpec {
8058 module_id: "terminal-history".to_string(),
8059 program: fake_aft_stub_path(),
8060 args: Vec::new(),
8061 env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
8062 reserved: false,
8063 reserved_prefixes: Vec::new(),
8064 protocol: ModuleProtocol::Subc,
8065 overlap: Default::default(),
8066 })
8067 .unwrap();
8068
8069 let deadline = Instant::now() + Duration::from_secs(5);
8070 loop {
8071 let history = module.terminal_history();
8072 if history.entries.len() == 2 {
8073 assert_eq!(module.status().unwrap().state, ModuleState::Failed);
8074 assert_eq!(history.dropped, 0);
8075 assert_eq!(
8076 history
8077 .entries
8078 .iter()
8079 .map(|entry| entry.exit_code)
8080 .collect::<Vec<_>>(),
8081 vec![Some(23), Some(23)]
8082 );
8083 assert!(history.entries[0].at_ms <= history.entries[1].at_ms);
8084 return;
8085 }
8086 assert!(
8087 Instant::now() < deadline,
8088 "module did not retain two terminal exits: {history:?}"
8089 );
8090 sleep(Duration::from_millis(10)).await;
8091 }
8092 }
8093
8094 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
8098 async fn disable_during_crash_backoff_cancels_pending_respawn() {
8099 let backoff = Duration::from_secs(2);
8100 let supervisor = Supervisor::new(
8101 Arc::new(Registry::default()),
8102 RestartPolicy::new(10, backoff),
8103 );
8104 let module = supervisor
8105 .spawn(ModuleSpec {
8106 module_id: "disable-during-backoff".to_string(),
8107 program: fake_aft_stub_path(),
8108 args: Vec::new(),
8109 env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
8110 reserved: false,
8111 reserved_prefixes: Vec::new(),
8112 protocol: ModuleProtocol::Subc,
8113 overlap: Default::default(),
8114 })
8115 .unwrap();
8116
8117 let deadline = Instant::now() + Duration::from_secs(5);
8119 loop {
8120 if module.status().unwrap().state == ModuleState::Restarting {
8121 break;
8122 }
8123 assert!(
8124 Instant::now() < deadline,
8125 "module never entered the crash backoff"
8126 );
8127 sleep(Duration::from_millis(10)).await;
8128 }
8129
8130 let started = Instant::now();
8131 module.set_enabled(false).await.unwrap();
8132 let waited = started.elapsed();
8133
8134 assert!(
8135 waited < backoff / 2,
8136 "disable waited {waited:?} behind the {backoff:?} crash backoff; the operator command must preempt the pending respawn"
8137 );
8138 assert_eq!(module.status().unwrap().state, ModuleState::Disabled);
8139
8140 sleep(backoff + Duration::from_millis(500)).await;
8142 let status = module.status().unwrap();
8143 assert_eq!(status.state, ModuleState::Disabled);
8144 assert_eq!(
8145 status.spawn_generation, 1,
8146 "module respawned after the operator disabled it"
8147 );
8148 }
8149
8150 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
8154 async fn every_restart_increment_path_advances_lifetime_count() {
8155 let supervisor = Supervisor::new(
8156 Arc::new(Registry::default()),
8157 RestartPolicy::new(1, Duration::ZERO),
8158 );
8159 let runtime = supervisor.runtime_config();
8160 let spec = ModuleSpec {
8161 module_id: "lifetime-increment-path".to_string(),
8162 program: PathBuf::from("/unused/lifetime-increment-path"),
8163 args: Vec::new(),
8164 env: Vec::new(),
8165 reserved: false,
8166 reserved_prefixes: Vec::new(),
8167 protocol: ModuleProtocol::Subc,
8168 overlap: Default::default(),
8169 };
8170
8171 let crash_snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8172 assert!(matches!(
8173 on_child_exit(
8174 &spec,
8175 runtime.restart_policy,
8176 &supervisor.registry,
8177 &crash_snapshot,
8178 &runtime.terminal_ring,
8179 &runtime.spawn_events,
8180 &runtime.child_roster,
8181 ExitReport {
8182 kind: ExitKind::Crash,
8183 code: Some(1),
8184 signal: None,
8185 at_ms: 1,
8186 },
8187 )
8188 .await,
8189 NextAction::Restart { schedule: _ }
8190 ));
8191 let (crash_restarts, crash_lifetime) = {
8192 let state = lock_snapshot(&crash_snapshot).unwrap();
8193 (state.crash_restarts.len(), state.lifetime_restarts)
8194 };
8195 assert_eq!(crash_restarts, 1);
8196 assert_eq!(crash_lifetime, 1);
8197
8198 let health_snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8199 let mut health_child = None;
8200 assert!(matches!(
8201 health_restart_child(
8202 &spec,
8203 &runtime,
8204 &supervisor.registry,
8205 &supervisor.process_liveness,
8206 &health_snapshot,
8207 &mut health_child,
8208 SupervisorHealthStatus::Failing,
8209 None,
8210 2,
8211 )
8212 .await,
8213 Err(SuperviseError::Spawn { .. })
8214 ));
8215 let (health_restarts, health_lifetime) = {
8216 let state = lock_snapshot(&health_snapshot).unwrap();
8217 (state.crash_restarts.len(), state.lifetime_restarts)
8218 };
8219 assert_eq!(health_restarts, 1);
8220 assert_eq!(health_lifetime, 1);
8221
8222 let reload_snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8223 let mut reload_child = None;
8224 assert!(matches!(
8225 handle_reload_spawn_failure(
8226 &spec,
8227 &runtime,
8228 &supervisor.process_liveness,
8229 &reload_snapshot,
8230 &mut reload_child,
8231 "forced reload spawn failure".to_string(),
8232 )
8233 .await,
8234 Err(SuperviseError::ReloadFailed { .. })
8235 ));
8236 let (reload_restarts, reload_lifetime) = {
8237 let state = lock_snapshot(&reload_snapshot).unwrap();
8238 (state.crash_restarts.len(), state.lifetime_restarts)
8239 };
8240 assert_eq!(reload_restarts, 1);
8241 assert_eq!(reload_lifetime, 1);
8242 }
8243
8244 #[tokio::test]
8245 async fn deliberately_severed_live_child_records_lifetime_without_spending_restart_budget() {
8246 let supervisor = Supervisor::new(
8247 Arc::new(Registry::default()),
8248 RestartPolicy::new(3, Duration::ZERO),
8249 );
8250 let runtime = supervisor.runtime_config();
8251 let spec = ModuleSpec {
8252 module_id: "deliberately-severed".to_string(),
8253 program: PathBuf::from("/unused/deliberately-severed"),
8254 args: Vec::new(),
8255 env: Vec::new(),
8256 reserved: false,
8257 reserved_prefixes: Vec::new(),
8258 protocol: ModuleProtocol::Subc,
8259 overlap: Default::default(),
8260 };
8261 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8262 let process = ProcessIdentity {
8263 pid: 41,
8264 start_time: 101,
8265 };
8266 record_deliberate_severance(&snapshot, process).unwrap();
8267 let exit_report = apply_deliberate_severance_marker(
8268 &snapshot,
8269 Some(process),
8270 ExitReport {
8271 kind: ExitKind::Crash,
8272 code: Some(1),
8273 signal: None,
8274 at_ms: 1,
8275 },
8276 );
8277 assert_eq!(exit_report.kind, ExitKind::DeliberateSeverance);
8278
8279 assert!(matches!(
8280 on_child_exit(
8281 &spec,
8282 runtime.restart_policy,
8283 &supervisor.registry,
8284 &snapshot,
8285 &runtime.terminal_ring,
8286 &runtime.spawn_events,
8287 &runtime.child_roster,
8288 exit_report,
8289 )
8290 .await,
8291 NextAction::Restart { schedule: _ }
8292 ));
8293 let state = lock_snapshot(&snapshot).unwrap();
8294 assert_eq!(state.lifetime_restarts, 1);
8295 assert_eq!(state.crash_restarts.len(), 0);
8296 }
8297
8298 #[tokio::test]
8299 async fn genuine_crash_spends_restart_budget_and_records_lifetime() {
8300 let supervisor = Supervisor::new(
8301 Arc::new(Registry::default()),
8302 RestartPolicy::new(3, Duration::ZERO),
8303 );
8304 let runtime = supervisor.runtime_config();
8305 let spec = ModuleSpec {
8306 module_id: "genuine-crash".to_string(),
8307 program: PathBuf::from("/unused/genuine-crash"),
8308 args: Vec::new(),
8309 env: Vec::new(),
8310 reserved: false,
8311 reserved_prefixes: Vec::new(),
8312 protocol: ModuleProtocol::Subc,
8313 overlap: Default::default(),
8314 };
8315 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8316
8317 assert!(matches!(
8318 on_child_exit(
8319 &spec,
8320 runtime.restart_policy,
8321 &supervisor.registry,
8322 &snapshot,
8323 &runtime.terminal_ring,
8324 &runtime.spawn_events,
8325 &runtime.child_roster,
8326 ExitReport {
8327 kind: ExitKind::Crash,
8328 code: Some(1),
8329 signal: None,
8330 at_ms: 1,
8331 },
8332 )
8333 .await,
8334 NextAction::Restart { schedule: _ }
8335 ));
8336 let state = lock_snapshot(&snapshot).unwrap();
8337 assert_eq!(state.lifetime_restarts, 1);
8338 assert_eq!(state.crash_restarts.len(), 1);
8339 }
8340
8341 fn crash_exit_report(at_ms: u64) -> ExitReport {
8342 ExitReport {
8343 kind: ExitKind::Crash,
8344 code: Some(1),
8345 signal: None,
8346 at_ms,
8347 }
8348 }
8349
8350 fn windowed_crash_spec(module_id: &str) -> ModuleSpec {
8351 ModuleSpec {
8352 module_id: module_id.to_string(),
8353 program: PathBuf::from("/unused").join(module_id),
8354 args: Vec::new(),
8355 env: Vec::new(),
8356 reserved: false,
8357 reserved_prefixes: Vec::new(),
8358 protocol: ModuleProtocol::Subc,
8359 overlap: Default::default(),
8360 }
8361 }
8362
8363 #[tokio::test]
8369 async fn three_crashes_inside_the_window_stop_the_module_and_name_the_window() {
8370 let (logs, _guard) = crate::router::test_log::log_capture(tracing::Level::ERROR);
8371 let supervisor = Supervisor::new(
8372 Arc::new(Registry::default()),
8373 RestartPolicy::new(2, Duration::ZERO),
8374 );
8375 let runtime = supervisor.runtime_config();
8376 let spec = windowed_crash_spec("crash-loop-in-window");
8377 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8378
8379 for attempt in 1..=2 {
8380 assert!(
8381 matches!(
8382 on_child_exit(
8383 &spec,
8384 runtime.restart_policy,
8385 &supervisor.registry,
8386 &snapshot,
8387 &runtime.terminal_ring,
8388 &runtime.spawn_events,
8389 &runtime.child_roster,
8390 crash_exit_report(attempt),
8391 )
8392 .await,
8393 NextAction::Restart { schedule: _ }
8394 ),
8395 "crash {attempt} is inside the budget and must respawn"
8396 );
8397 }
8398
8399 assert!(matches!(
8400 on_child_exit(
8401 &spec,
8402 runtime.restart_policy,
8403 &supervisor.registry,
8404 &snapshot,
8405 &runtime.terminal_ring,
8406 &runtime.spawn_events,
8407 &runtime.child_roster,
8408 crash_exit_report(3),
8409 )
8410 .await,
8411 NextAction::Stop { .. }
8412 ));
8413
8414 {
8415 let state = lock_snapshot(&snapshot).unwrap();
8416 assert_eq!(state.state, ModuleState::Failed);
8417 assert_eq!(state.crash_restarts.len(), 2);
8418 assert_eq!(state.lifetime_restarts, 2);
8419 }
8420
8421 let history = runtime
8422 .terminal_ring
8423 .lock()
8424 .expect("terminal ring is not poisoned")
8425 .snapshot();
8426 let last = history
8427 .entries
8428 .last()
8429 .expect("the refused crash is retained");
8430 assert_eq!(last.disposition, TerminalDisposition::Failed);
8431 assert_eq!(
8432 last.disposition_detail.as_deref(),
8433 Some("crash budget exhausted: max_restarts=2 within window_secs=600")
8434 );
8435
8436 let captured = crate::router::test_log::captured_logs(&logs);
8437 assert!(
8438 captured.contains("crash budget exhausted: max_restarts=2 within window_secs=600"),
8439 "the stop must be logged with its window: {captured}"
8440 );
8441 }
8442
8443 #[tokio::test]
8451 async fn a_crash_older_than_the_window_frees_its_slot_for_a_later_crash() {
8452 let supervisor = Supervisor::new(
8453 Arc::new(Registry::default()),
8454 RestartPolicy::new(2, Duration::ZERO),
8455 );
8456 let runtime = supervisor.runtime_config();
8457 let spec = windowed_crash_spec("crash-across-windows");
8458 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8459
8460 for attempt in 1..=2 {
8461 assert!(matches!(
8462 on_child_exit(
8463 &spec,
8464 runtime.restart_policy,
8465 &supervisor.registry,
8466 &snapshot,
8467 &runtime.terminal_ring,
8468 &runtime.spawn_events,
8469 &runtime.child_roster,
8470 crash_exit_report(attempt),
8471 )
8472 .await,
8473 NextAction::Restart { schedule: _ }
8474 ));
8475 }
8476
8477 update_snapshot(&snapshot, Some(&spec.module_id), |state| {
8480 age_oldest_crash_restart_out_of_window(state, runtime.restart_policy.window);
8481 })
8482 .unwrap();
8483
8484 assert!(
8485 matches!(
8486 on_child_exit(
8487 &spec,
8488 runtime.restart_policy,
8489 &supervisor.registry,
8490 &snapshot,
8491 &runtime.terminal_ring,
8492 &runtime.spawn_events,
8493 &runtime.child_roster,
8494 crash_exit_report(3),
8495 )
8496 .await,
8497 NextAction::Restart { schedule: _ }
8498 ),
8499 "a crash older than the window must not hold a budget slot"
8500 );
8501
8502 let state = lock_snapshot(&snapshot).unwrap();
8503 assert_eq!(state.state, ModuleState::Restarting);
8504 assert_eq!(
8505 state.crash_restarts.len(),
8506 2,
8507 "the aged instant is dropped and the new one takes its place"
8508 );
8509 assert_eq!(
8510 state.lifetime_restarts, 3,
8511 "the ledger counts every restart, including the ones the window forgot"
8512 );
8513 }
8514
8515 #[tokio::test]
8520 async fn an_operator_restart_clears_the_ring_and_leaves_the_ledger_alone() {
8521 let supervisor = Supervisor::new(
8522 Arc::new(Registry::default()),
8523 RestartPolicy::new(2, Duration::ZERO),
8524 );
8525 let runtime = supervisor.runtime_config();
8526 let spec = windowed_crash_spec("operator-cleared-budget");
8527 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8528
8529 for attempt in 1..=2 {
8530 assert!(matches!(
8531 on_child_exit(
8532 &spec,
8533 runtime.restart_policy,
8534 &supervisor.registry,
8535 &snapshot,
8536 &runtime.terminal_ring,
8537 &runtime.spawn_events,
8538 &runtime.child_roster,
8539 crash_exit_report(attempt),
8540 )
8541 .await,
8542 NextAction::Restart { schedule: _ }
8543 ));
8544 }
8545
8546 reset_restart_count(&snapshot, &spec.module_id).unwrap();
8547 {
8548 let state = lock_snapshot(&snapshot).unwrap();
8549 assert!(
8550 state.crash_restarts.is_empty(),
8551 "an operator restart returns the full budget"
8552 );
8553 assert_eq!(
8554 state.lifetime_restarts, 2,
8555 "clearing the budget must not unmake the crashes"
8556 );
8557 }
8558
8559 assert!(
8560 matches!(
8561 on_child_exit(
8562 &spec,
8563 runtime.restart_policy,
8564 &supervisor.registry,
8565 &snapshot,
8566 &runtime.terminal_ring,
8567 &runtime.spawn_events,
8568 &runtime.child_roster,
8569 crash_exit_report(3),
8570 )
8571 .await,
8572 NextAction::Restart { schedule: _ }
8573 ),
8574 "the cleared budget must be spendable again"
8575 );
8576 let state = lock_snapshot(&snapshot).unwrap();
8577 assert_eq!(state.crash_restarts.len(), 1);
8578 assert_eq!(state.lifetime_restarts, 3);
8579 }
8580
8581 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
8582 async fn severance_marker_for_a_dead_child_does_not_label_its_successor() {
8583 let severed = ProcessIdentity {
8584 pid: 41,
8585 start_time: 101,
8586 };
8587 let successor = ProcessIdentity {
8588 pid: 41,
8589 start_time: 202,
8590 };
8591 let module = module_with_recovery_snapshot(ModuleState::Running, true, 0);
8592 update_snapshot(&module.inner.snapshot, Some("recovery-snapshot"), |state| {
8593 state.pid = Some(successor.pid);
8594 state.process_start_time = Some(successor.start_time);
8595 })
8596 .unwrap();
8597 assert!(!module.record_deliberate_severance(severed).unwrap());
8598
8599 let exit_report = apply_deliberate_severance_marker(
8600 &module.inner.snapshot,
8601 Some(successor),
8602 ExitReport {
8603 kind: ExitKind::Crash,
8604 code: Some(1),
8605 signal: None,
8606 at_ms: 1,
8607 },
8608 );
8609
8610 assert_eq!(exit_report.kind, ExitKind::Crash);
8611 }
8612
8613 #[tokio::test]
8614 async fn drain_reap_marks_deliberate_severance_and_records_lifetime_without_budget() {
8615 let registry = Registry::default();
8616 let supervisor = Supervisor::new(
8617 Arc::new(Registry::default()),
8618 RestartPolicy::new(3, Duration::ZERO),
8619 );
8620 let runtime = supervisor.runtime_config();
8621 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8622 let spec = ModuleSpec {
8623 module_id: "drain-deliberate-severance".to_string(),
8624 program: fake_aft_stub_path(),
8625 args: Vec::new(),
8626 env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
8627 reserved: false,
8628 reserved_prefixes: Vec::new(),
8629 protocol: ModuleProtocol::Subc,
8630 overlap: Default::default(),
8631 };
8632 let mut child = spawn_and_mark_running(&spec, &runtime, &snapshot).unwrap();
8633 let process = ProcessIdentity {
8634 pid: 41,
8635 start_time: 101,
8636 };
8637 child.process_identity = Some(process);
8638 update_snapshot(&snapshot, Some(&spec.module_id), |state| {
8639 state.pid = Some(process.pid);
8640 state.process_start_time = Some(process.start_time);
8641 })
8642 .unwrap();
8643 record_deliberate_severance(&snapshot, process).unwrap();
8644
8645 drain_child_to_state(
8646 &spec.module_id,
8647 spec.protocol,
8648 StopNotice::SentOverConnection,
8651 ®istry,
8652 &snapshot,
8653 &runtime.terminal_ring,
8654 &runtime.spawn_events,
8655 child,
8656 Duration::from_secs(1),
8657 ModuleState::Stopped,
8658 Some(false),
8659 )
8660 .await
8661 .unwrap();
8662
8663 let state = lock_snapshot(&snapshot).unwrap();
8664 assert_eq!(
8665 state.last_exit.as_ref().map(|exit| exit.kind),
8666 Some(ExitKind::DeliberateSeverance)
8667 );
8668 assert_eq!(state.lifetime_restarts, 1);
8669 assert_eq!(state.crash_restarts.len(), 0);
8670 drop(state);
8671 let history = runtime.terminal_ring.lock().unwrap().snapshot();
8672 assert_eq!(
8673 history.entries[0].exit_kind,
8674 subc_control::TerminalExitKind::DeliberateSeverance
8675 );
8676 }
8677
8678 #[tokio::test]
8679 async fn ordinary_drain_reap_does_not_record_a_lifetime_restart() {
8680 let registry = Registry::default();
8681 let supervisor = Supervisor::new(
8682 Arc::new(Registry::default()),
8683 RestartPolicy::new(3, Duration::ZERO),
8684 );
8685 let runtime = supervisor.runtime_config();
8686 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
8687 let spec = ModuleSpec {
8688 module_id: "ordinary-drain".to_string(),
8689 program: fake_aft_stub_path(),
8690 args: Vec::new(),
8691 env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
8692 reserved: false,
8693 reserved_prefixes: Vec::new(),
8694 protocol: ModuleProtocol::Subc,
8695 overlap: Default::default(),
8696 };
8697 let child = spawn_and_mark_running(&spec, &runtime, &snapshot).unwrap();
8698
8699 drain_child_to_state(
8700 &spec.module_id,
8701 spec.protocol,
8702 StopNotice::SentOverConnection,
8705 ®istry,
8706 &snapshot,
8707 &runtime.terminal_ring,
8708 &runtime.spawn_events,
8709 child,
8710 Duration::from_secs(1),
8711 ModuleState::Stopped,
8712 Some(false),
8713 )
8714 .await
8715 .unwrap();
8716
8717 let state = lock_snapshot(&snapshot).unwrap();
8718 assert_eq!(
8719 state.last_exit.as_ref().map(|exit| exit.kind),
8720 Some(ExitKind::Crash)
8721 );
8722 assert_eq!(state.lifetime_restarts, 0);
8723 assert_eq!(state.crash_restarts.len(), 0);
8724 }
8725
8726 #[test]
8727 fn fatal_connection_teardown_cannot_arm_a_marker_for_a_surviving_process() {
8728 assert!(!include_str!("server.rs")
8734 .contains("router.record_deliberate_connection_severance(ctx.connection_id)"));
8735 }
8736
8737 #[test]
8744 fn drained_after_quiescence_wait_passes_ok_through_and_forces_false_on_err() {
8745 assert!(drained_after_quiescence_wait(&Ok(true)));
8746 assert!(!drained_after_quiescence_wait(&Ok(false)));
8747 assert!(!drained_after_quiescence_wait(&Err(
8748 SuperviseError::StatePoisoned { module_id: None }
8749 )));
8750 }
8751
8752 #[test]
8761 fn wait_error_exit_report_records_a_failed_terminal_with_no_code_or_signal() {
8762 let ring = Arc::new(Mutex::new(TerminalRing::new(
8763 TerminalRingConfig::default(),
8764 0,
8765 )));
8766 record_wait_error_terminal("wait-error", &ring, &super::SpawnEventFeed::default());
8767
8768 let snapshot = ring.lock().unwrap().snapshot();
8769 assert_eq!(snapshot.entries.len(), 1);
8770 let entry = &snapshot.entries[0];
8771 assert_eq!(entry.exit_code, None);
8772 assert_eq!(entry.exit_signal, None);
8773 assert_eq!(entry.disposition, TerminalDisposition::Failed);
8774 }
8775
8776 #[test]
8777 fn wait_error_exit_path_preserves_spawn_event_density() {
8778 let feed = super::SpawnEventFeed::default();
8779 feed.configure_incarnation("wait-error-density".to_string());
8780 feed.emit_spawned("wait-error", 41, 1);
8781 let ring = Arc::new(Mutex::new(TerminalRing::new(
8782 TerminalRingConfig::default(),
8783 0,
8784 )));
8785
8786 record_wait_error_terminal("wait-error", &ring, &feed);
8787 feed.emit_spawned("after-wait-error", 42, 2);
8788
8789 let state = feed.0.lock().unwrap();
8790 let sequences = state
8791 .events
8792 .iter()
8793 .map(|event| event.cursor.seq)
8794 .collect::<Vec<_>>();
8795 assert_eq!(sequences, vec![1, 2, 3]);
8796 assert_eq!(state.events[1].kind, SpawnEventKind::Exited);
8797 assert_eq!(state.events[1].exit_code, None);
8798 assert_eq!(state.events[1].exit_signal, None);
8799 }
8800
8801 #[test]
8805 fn wait_error_exit_report_is_classified_as_a_crash() {
8806 assert_eq!(wait_error_exit_report().kind, ExitKind::Crash);
8807 }
8808}
8809
8810#[cfg(test)]
8811mod health_evidence_tests {
8812 use super::{HealthProbeError, HealthProbeEvidence};
8813 use std::collections::HashSet;
8814
8815 #[test]
8823 fn only_a_dead_lane_is_proof_of_death() {
8824 assert!(HealthProbeError::lane_dead("gone").is_proof_of_death());
8825 assert!(!HealthProbeError::no_answer("timed out").is_proof_of_death());
8829 assert!(!HealthProbeError::bad_answer("garbage").is_proof_of_death());
8830 assert!(!HealthProbeError::misconfigured("no table").is_proof_of_death());
8831 }
8832
8833 #[test]
8839 fn every_evidence_class_has_a_distinct_label() {
8840 let labels = [
8841 HealthProbeError::lane_dead("").label(),
8842 HealthProbeError::no_answer("").label(),
8843 HealthProbeError::bad_answer("").label(),
8844 HealthProbeError::misconfigured("").label(),
8845 ];
8846 let unique: HashSet<_> = labels.iter().collect();
8847 assert_eq!(unique.len(), labels.len(), "labels collided: {labels:?}");
8848 }
8849
8850 #[test]
8856 fn classification_preserves_the_original_message() {
8857 let err = HealthProbeError::no_answer("module did not answer within 5s");
8858 assert_eq!(err.to_string(), "module did not answer within 5s");
8859 assert!(matches!(err.evidence, HealthProbeEvidence::NoAnswer));
8860 }
8861}
8862
8863#[cfg(test)]
8864mod health_tombstone_tests {
8865 use std::{path::PathBuf, sync::Arc, time::Duration};
8866
8867 use subc_protocol::{
8868 manifest::Concurrency,
8869 session::{HealthStatus, ModuleControlResponse},
8870 };
8871 use tokio::sync::mpsc;
8872
8873 use super::{
8874 probe_module_health, HealthAction, HealthConfig, HealthProbeEvidence, ModuleProtocol,
8875 ModuleSpec, RestartPolicy, Supervisor, SupervisorRuntimeConfig,
8876 };
8877 use crate::{
8878 control::ControlHandler,
8879 forwarding::{ForwardingTable, ModuleControlRpcCompletion, ModuleControlRpcOutcome},
8880 registry::{ConnectionId, Registry},
8881 router::FrameSink,
8882 };
8883
8884 struct ProbeHarness {
8885 spec: ModuleSpec,
8886 runtime: SupervisorRuntimeConfig,
8887 forwarding: Arc<ForwardingTable>,
8888 module_connection: ConnectionId,
8889 module_rx: mpsc::Receiver<crate::router::OutboundFrame>,
8890 handler: ControlHandler,
8891 module: super::SupervisedModule,
8892 }
8893
8894 fn probe_harness() -> ProbeHarness {
8895 let registry = Arc::new(Registry::default());
8896 let forwarding = Arc::new(ForwardingTable::default());
8897 let supervisor_handle = super::SupervisorHandle::new();
8898 let health = HealthConfig {
8899 cadence: Duration::from_secs(30),
8900 deadline: Duration::from_secs(5),
8901 failure_threshold: 3,
8902 on_degraded: HealthAction::Report,
8903 on_failing: HealthAction::Report,
8904 critical: false,
8905 };
8906 let supervisor = Supervisor::new(Arc::clone(®istry), RestartPolicy::default())
8907 .with_forwarding(Arc::clone(&forwarding))
8908 .with_handle(supervisor_handle.clone())
8909 .with_health_config(health);
8910 let spec = ModuleSpec {
8911 module_id: "late-health-module".to_string(),
8912 program: PathBuf::from("disabled-module"),
8913 args: Vec::new(),
8914 env: Vec::new(),
8915 reserved: false,
8916 reserved_prefixes: Vec::new(),
8917 protocol: ModuleProtocol::Subc,
8918 overlap: Default::default(),
8919 };
8920 let module = supervisor
8921 .supervise_configured(spec.clone(), false)
8922 .unwrap();
8923 let runtime = supervisor.runtime_config();
8924 let handler = ControlHandler::with_forwarding(registry, Arc::clone(&forwarding))
8925 .with_supervisor(supervisor_handle);
8926 let module_connection = ConnectionId::new(700);
8927 let (module_tx, module_rx) = mpsc::channel(8);
8928 forwarding
8929 .register_module_connection(
8930 module_connection,
8931 spec.module_id.clone(),
8932 subc_protocol::PROTOCOL_VERSION,
8933 Concurrency::ModuleManaged,
8934 FrameSink::new(module_tx),
8935 )
8936 .unwrap();
8937
8938 ProbeHarness {
8939 spec,
8940 runtime,
8941 forwarding,
8942 module_connection,
8943 module_rx,
8944 handler,
8945 module,
8946 }
8947 }
8948
8949 async fn finish_after(
8950 harness: &mut ProbeHarness,
8951 stall: Duration,
8952 ) -> ModuleControlRpcCompletion {
8953 assert!(stall > harness.runtime.health.deadline);
8954 let deadline = harness.runtime.health.deadline;
8955 let probe = probe_module_health(&harness.spec.module_id, &harness.runtime, None);
8956 let answer = async {
8957 let frame = harness.module_rx.recv().await.expect("health.check frame");
8958 tokio::time::advance(deadline).await;
8959 tokio::task::yield_now().await;
8960 tokio::time::advance(stall - deadline).await;
8961 harness
8962 .forwarding
8963 .complete_module_control_rpc(
8964 harness.module_connection,
8965 frame.header.corr,
8966 Some("health.check"),
8967 ModuleControlRpcOutcome::Response(ModuleControlResponse::HealthCheck {
8968 status: HealthStatus::Ok,
8969 detail: None,
8970 metrics: None,
8971 }),
8972 )
8973 .unwrap()
8974 };
8975 let (probe_result, completion) = tokio::join!(probe, answer);
8976 let err = probe_result.expect_err("probe must miss its deadline");
8977 assert!(matches!(err.evidence, HealthProbeEvidence::NoAnswer));
8978 completion
8979 }
8980
8981 async fn time_out_without_answer(harness: &mut ProbeHarness) {
8982 let deadline = harness.runtime.health.deadline;
8983 let probe = probe_module_health(&harness.spec.module_id, &harness.runtime, None);
8984 let exhaust_deadline = async {
8985 let _frame = harness.module_rx.recv().await.expect("health.check frame");
8986 tokio::time::advance(deadline).await;
8987 tokio::task::yield_now().await;
8988 };
8989 let (probe_result, ()) = tokio::join!(probe, exhaust_deadline);
8990 let err = probe_result.expect_err("probe must miss its deadline");
8991 assert!(matches!(err.evidence, HealthProbeEvidence::NoAnswer));
8992 }
8993
8994 #[tokio::test(start_paused = true)]
8995 async fn late_health_answers_record_start_anchored_latency_for_two_stalls() {
8996 let mut harness = probe_harness();
8997
8998 let first = finish_after(&mut harness, Duration::from_secs(8)).await;
8999 let first_latency = match &first {
9000 ModuleControlRpcCompletion::LateHealthAnswer { latency, .. } => *latency,
9001 other => panic!("late answer was not retained: {other:?}"),
9002 };
9003 assert!(harness.handler.observe_module_control_completion(first));
9004
9005 let second = finish_after(&mut harness, Duration::from_secs(11)).await;
9006 let second_latency = match &second {
9007 ModuleControlRpcCompletion::LateHealthAnswer { latency, .. } => *latency,
9008 other => panic!("late answer was not retained: {other:?}"),
9009 };
9010 assert!(harness.handler.observe_module_control_completion(second));
9011
9012 assert_eq!(first_latency, Duration::from_secs(8));
9013 assert_eq!(
9014 second_latency - first_latency,
9015 Duration::from_secs(3),
9016 "latency must grow linearly with the additional stall"
9017 );
9018 let health = harness.module.status().unwrap().health;
9019 assert_eq!(health.late_answer_count, 2);
9020 assert_eq!(health.last_late_answer_latency_ms, Some(11_000));
9021 }
9022
9023 #[tokio::test(start_paused = true)]
9031 async fn late_answer_clears_the_consecutive_failure_streak() {
9032 let mut harness = probe_harness();
9033
9034 time_out_without_answer(&mut harness).await;
9036 harness
9037 .module
9038 .record_health_probe_failure_for_test("[no-answer] test miss")
9039 .unwrap();
9040 assert_eq!(
9041 harness.module.status().unwrap().health.consecutive_failures,
9042 1,
9043 "precondition: the miss must be on the streak before the late answer"
9044 );
9045
9046 let late = finish_after(&mut harness, Duration::from_secs(9)).await;
9048 assert!(matches!(
9049 late,
9050 ModuleControlRpcCompletion::LateHealthAnswer { .. }
9051 ));
9052 assert!(harness.handler.observe_module_control_completion(late));
9053
9054 let health = harness.module.status().unwrap().health;
9055 assert_eq!(
9056 health.consecutive_failures, 0,
9057 "a late answer is an answer: the streak must reset"
9058 );
9059 assert_eq!(health.late_answer_count, 1);
9060 }
9061
9062 #[tokio::test(start_paused = true)]
9063 async fn repeated_serial_probe_cycles_keep_one_tombstone_per_endpoint() {
9064 let mut harness = probe_harness();
9065
9066 for _ in 0..20 {
9067 time_out_without_answer(&mut harness).await;
9068 assert_eq!(
9069 harness.forwarding.health_probe_tombstone_count().unwrap(),
9070 1
9071 );
9072 }
9073 }
9074}
9075
9076#[cfg(test)]
9077mod child_env_tests {
9078 use super::{
9079 apply_child_env, apply_spawn_role, apply_wire_spawn_args, ModuleProtocol, ModuleSpec,
9080 SpawnRole, SupervisorHandle, SPAWN_ROLE_SWAP_CANDIDATE, SUBC_ARG, SUBC_LAUNCH_NONCE_ENV,
9081 SUBC_MODULE_ID_ENV, SUBC_SPAWN_ROLE_ENV,
9082 };
9083 use std::{ffi::OsStr, path::PathBuf};
9084 use tokio::process::Command;
9085
9086 fn spec(env: Vec<(String, String)>) -> ModuleSpec {
9087 ModuleSpec {
9088 module_id: "env-plan".to_string(),
9089 program: PathBuf::from("/nonexistent"),
9090 args: Vec::new(),
9091 env,
9092 reserved: false,
9093 reserved_prefixes: Vec::new(),
9094 protocol: ModuleProtocol::Subc,
9095 overlap: Default::default(),
9096 }
9097 }
9098
9099 #[test]
9113 fn ambient_ck_log_is_removed_and_a_configured_one_survives() {
9114 let mut command = Command::new("/nonexistent");
9115 apply_child_env(&mut command, &spec(Vec::new()));
9116 let removed = command
9117 .as_std()
9118 .get_envs()
9119 .any(|(key, value)| key == OsStr::new("CK_LOG") && value.is_none());
9120 assert!(
9121 removed,
9122 "ambient CK_LOG must be explicitly removed for an unconfigured module"
9123 );
9124
9125 let mut configured = Command::new("/nonexistent");
9126 apply_child_env(
9127 &mut configured,
9128 &spec(vec![("CK_LOG".to_string(), "debug".to_string())]),
9129 );
9130 let effective = configured
9131 .as_std()
9132 .get_envs()
9133 .filter(|(key, _)| *key == OsStr::new("CK_LOG"))
9134 .last()
9135 .map(|(_, value)| value.map(|v| v.to_string_lossy().into_owned()));
9136 assert_eq!(
9137 effective,
9138 Some(Some("debug".to_string())),
9139 "a module's configured CK_LOG must survive the ambient removal"
9140 );
9141 }
9142
9143 #[test]
9152 fn protocol_none_spawn_carries_no_subc_argument_and_no_nonce() {
9153 let connection_file = std::path::Path::new("/run/subc-connection.json");
9154 let handle = SupervisorHandle::new();
9155
9156 let mut none_spec = spec(Vec::new());
9157 none_spec.protocol = ModuleProtocol::None;
9158 let mut none = Command::new("/nonexistent");
9159 apply_wire_spawn_args(&mut none, &none_spec, Some(connection_file), Some(&handle))
9160 .expect("protocol-none spawn args apply");
9161 let none_args: Vec<String> = none
9162 .as_std()
9163 .get_args()
9164 .map(|a| a.to_string_lossy().into_owned())
9165 .collect();
9166 assert!(
9167 !none_args.iter().any(|a| a == SUBC_ARG),
9168 "protocol:none argv must not carry --subc; got {none_args:?}"
9169 );
9170 let none_has_nonce = none
9171 .as_std()
9172 .get_envs()
9173 .any(|(key, value)| key == OsStr::new(SUBC_LAUNCH_NONCE_ENV) && value.is_some());
9174 assert!(
9175 !none_has_nonce,
9176 "protocol:none spawn must not receive a launch nonce"
9177 );
9178 let none_has_module_id = none
9179 .as_std()
9180 .get_envs()
9181 .any(|(key, value)| key == OsStr::new(SUBC_MODULE_ID_ENV) && value.is_some());
9182 assert!(
9183 none_has_module_id,
9184 "SUBC_MODULE_ID is inert and stays on every path"
9185 );
9186 assert!(
9187 handle.spawn_nonce(&none_spec.module_id).is_none(),
9188 "no nonce record for a process that will never present one"
9189 );
9190
9191 let wire_spec = spec(Vec::new());
9193 let mut wire = Command::new("/nonexistent");
9194 apply_wire_spawn_args(&mut wire, &wire_spec, Some(connection_file), Some(&handle))
9195 .expect("subc-wire spawn args apply");
9196 let wire_args: Vec<String> = wire
9197 .as_std()
9198 .get_args()
9199 .map(|a| a.to_string_lossy().into_owned())
9200 .collect();
9201 assert_eq!(
9202 wire_args,
9203 vec![
9204 SUBC_ARG.to_string(),
9205 connection_file.to_string_lossy().into_owned()
9206 ],
9207 "a subc-wire spawn still carries --subc <path>"
9208 );
9209 assert!(wire
9210 .as_std()
9211 .get_envs()
9212 .any(|(key, value)| key == OsStr::new(SUBC_LAUNCH_NONCE_ENV) && value.is_some()));
9213 assert!(handle.spawn_nonce(&wire_spec.module_id).is_some());
9214 }
9215
9216 #[test]
9226 fn plain_spawn_removes_the_spawn_role_even_when_the_spec_sets_it() {
9227 let role = |command: &Command| {
9228 command
9229 .as_std()
9230 .get_envs()
9231 .filter(|(key, _)| *key == OsStr::new(SUBC_SPAWN_ROLE_ENV))
9232 .last()
9233 .map(|(_, value)| value.map(|v| v.to_string_lossy().into_owned()))
9234 };
9235 let forged = spec(vec![(
9236 SUBC_SPAWN_ROLE_ENV.to_string(),
9237 SPAWN_ROLE_SWAP_CANDIDATE.to_string(),
9238 )]);
9239
9240 let mut plain = Command::new("/nonexistent");
9241 apply_child_env(&mut plain, &forged);
9242 apply_spawn_role(&mut plain, SpawnRole::Plain);
9243 assert_eq!(
9244 role(&plain),
9245 Some(None),
9246 "a plain spawn must remove SUBC_SPAWN_ROLE, whatever the spec says"
9247 );
9248
9249 let mut candidate = Command::new("/nonexistent");
9250 apply_child_env(&mut candidate, &spec(Vec::new()));
9251 apply_spawn_role(&mut candidate, SpawnRole::SwapCandidate);
9252 assert_eq!(
9253 role(&candidate),
9254 Some(Some(SPAWN_ROLE_SWAP_CANDIDATE.to_string()))
9255 );
9256 }
9257
9258 #[test]
9264 fn daemon_private_capture_keys_are_not_passed_to_the_child() {
9265 let mut command = Command::new("/nonexistent");
9266 apply_child_env(
9267 &mut command,
9268 &spec(vec![
9269 (super::CAPTURE_KEEP_ENV.to_string(), "5".to_string()),
9270 ("KEPT".to_string(), "yes".to_string()),
9271 ]),
9272 );
9273 let keys: Vec<String> = command
9274 .as_std()
9275 .get_envs()
9276 .filter(|(_, value)| value.is_some())
9277 .map(|(key, _)| key.to_string_lossy().into_owned())
9278 .collect();
9279 assert!(keys.contains(&"KEPT".to_string()), "got {keys:?}");
9280 assert!(
9281 !keys.contains(&super::CAPTURE_KEEP_ENV.to_string()),
9282 "daemon-private capture key leaked to the child: {keys:?}"
9283 );
9284 }
9285}
9286
9287#[cfg(test)]
9288mod jitter_tests {
9289 use super::jittered_health_delay;
9290 use std::{collections::HashSet, time::Duration};
9291
9292 const FLEET: [&str; 14] = [
9301 "aft",
9302 "alfonso-core",
9303 "magic-context",
9304 "broca",
9305 "thalamus",
9306 "quota",
9307 "engram",
9308 "plexus",
9309 "cerebellum",
9310 "astrocyte",
9311 "synapse",
9312 "subc-mcp",
9313 "cortexkit-credentials",
9314 "subc-federation",
9315 ];
9316
9317 #[test]
9325 fn probe_delays_disperse_across_the_fleet() {
9326 let cadence = Duration::from_secs(30);
9327 let delays: HashSet<Duration> = FLEET
9328 .iter()
9329 .map(|id| jittered_health_delay(id, 0, cadence))
9330 .collect();
9331 assert_eq!(
9332 delays.len(),
9333 FLEET.len(),
9334 "every supervised module must land on its own probe offset"
9335 );
9336 }
9337
9338 #[test]
9344 fn jitter_only_delays_and_stays_within_one_tenth_of_cadence() {
9345 let cadence = Duration::from_secs(30);
9346 let span = cadence / 10;
9347 for id in FLEET {
9348 for probe_index in 0..8 {
9349 let delay = jittered_health_delay(id, probe_index, cadence);
9350 assert!(
9351 delay >= cadence,
9352 "{id}#{probe_index}: jitter must not shorten the cadence"
9353 );
9354 assert!(
9355 delay < cadence + span,
9356 "{id}#{probe_index}: jitter must stay inside one tenth of the cadence"
9357 );
9358 }
9359 }
9360 }
9361
9362 #[test]
9368 fn a_module_offset_is_stable_across_restarts() {
9369 let cadence = Duration::from_secs(30);
9370 for id in FLEET {
9371 assert_eq!(
9372 jittered_health_delay(id, 0, cadence),
9373 jittered_health_delay(id, 0, cadence),
9374 "{id}: the same module and probe index must produce the same offset"
9375 );
9376 }
9377 }
9378
9379 #[test]
9381 fn zero_cadence_yields_zero_delay() {
9382 assert_eq!(
9383 jittered_health_delay("aft", 0, Duration::ZERO),
9384 Duration::ZERO
9385 );
9386 }
9387}
9388
9389#[cfg(all(test, target_os = "linux"))]
9390mod cgroup_placement_tests {
9391 use super::{
9392 apply_cgroup_placement, remove_module_cgroup, ModuleProtocol, ModuleSpec, SuperviseError,
9393 SupervisedChild,
9394 };
9395 use crate::stderr_tail::{StderrRing, StderrTailConfig};
9396 use std::{
9397 fs, io,
9398 path::{Path, PathBuf},
9399 sync::{Arc, Mutex},
9400 };
9401 use subc_test_support::TestTempDir;
9402 use tokio::process::Command;
9403
9404 #[test]
9405 fn failed_parent_cgroup_open_is_a_cgroup_supervision_error() {
9406 let path = Path::new("/definitely-missing-subc-cgroup");
9407 let mut command = Command::new("true");
9408 let error = apply_cgroup_placement(
9409 &mut command,
9410 &ModuleSpec {
9411 module_id: "broken-cgroup".to_string(),
9412 program: PathBuf::from("true"),
9413 args: Vec::new(),
9414 env: Vec::new(),
9415 reserved: false,
9416 reserved_prefixes: Vec::new(),
9417 protocol: ModuleProtocol::Subc,
9418 overlap: Default::default(),
9419 },
9420 path,
9421 )
9422 .expect_err("a parent cgroup open failure must reject the supervised spawn");
9423 let reason = error.to_string();
9424
9425 assert!(
9426 matches!(error, SuperviseError::Cgroup { .. }),
9427 "parent cgroup open must be reported as a cgroup supervision error: {reason}"
9428 );
9429 assert!(
9430 reason.contains("/definitely-missing-subc-cgroup/cgroup.procs"),
9431 "parent cgroup open failure must name cgroup.procs: {reason}"
9432 );
9433 }
9434
9435 #[tokio::test]
9436 async fn reaping_a_child_removes_its_empty_module_cgroup() {
9437 let root = TestTempDir::new("supervisor-reap-cgroup");
9438 fs::write(root.join("cgroup.procs"), b"").expect("write scratch cgroup marker");
9439 let placement = subc_cgroup::prepare_at(&root)
9440 .expect("prepare scratch cgroup root")
9441 .expect("scratch root has a cgroup.procs marker");
9442 let module_id = "reaped-module";
9443 let module = placement
9444 .module_path(module_id)
9445 .expect("create scratch module cgroup");
9446 let child = Command::new("true")
9447 .spawn()
9448 .expect("spawn short-lived child");
9449 let pid = child.id().expect("spawned child has pid");
9450 let mut child = SupervisedChild {
9451 child,
9452 module_id: module_id.to_string(),
9453 cgroup_placement: Some(placement),
9454 stdout_pump: None,
9455 stderr_pump: None,
9456 stderr_ring: Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default()))),
9457 spawned_at_ms: 0,
9458 spawned_from: PathBuf::from("true"),
9459 spawned_file_identity: None,
9460 process_start_time: None,
9461 process_identity: None,
9462 pid,
9463 roster_guard: None,
9464 };
9465
9466 child.wait().await.expect("reap short-lived child");
9467
9468 assert!(
9469 !module.exists(),
9470 "reaping the supervised child must remove its empty cgroup"
9471 );
9472 }
9473
9474 #[test]
9475 fn non_empty_cgroup_removal_is_reported_without_blocking_teardown() {
9476 let root = TestTempDir::new("supervisor-non-empty-cgroup");
9477 fs::write(root.join("cgroup.procs"), b"").expect("write scratch cgroup marker");
9478 let placement = subc_cgroup::prepare_at(&root)
9479 .expect("prepare scratch cgroup root")
9480 .expect("scratch root has a cgroup.procs marker");
9481 let module = placement
9482 .module_path("surviving-module")
9483 .expect("create scratch module cgroup");
9484 fs::write(module.join("surviving-process"), b"still present")
9485 .expect("make scratch cgroup non-empty");
9486 let (logs, _guard) = crate::router::test_log::log_capture(tracing::Level::WARN);
9487
9488 remove_module_cgroup(&placement, "surviving-module");
9489
9490 let logs = crate::router::test_log::captured_logs(&logs);
9491 assert!(
9492 module.exists(),
9493 "failed removal must leave the cgroup intact"
9494 );
9495 assert!(
9496 logs.contains("could not remove module cgroup after process exit; continuing teardown")
9497 && logs.contains("surviving-module"),
9498 "best-effort removal must report the failure without returning it: {logs}"
9499 );
9500 }
9501
9502 #[test]
9503 fn cgroup_pre_exec_spawn_failure_names_the_cgroup_path() {
9504 let cgroup_path = PathBuf::from("/sys/fs/cgroup/subc-modules/broken-module");
9505 let reason = SuperviseError::Spawn {
9506 program: PathBuf::from("/bin/true"),
9507 source: io::Error::from_raw_os_error(13),
9508 cgroup_path: Some(cgroup_path.clone()),
9509 }
9510 .to_string();
9511
9512 assert!(
9513 reason.contains(&cgroup_path.display().to_string()),
9514 "a pre_exec spawn failure must name the cgroup path: {reason}"
9515 );
9516 }
9517}
9518
9519#[cfg(test)]
9520mod spawn_subscriber_lag_tests {
9521 use super::*;
9522
9523 #[tokio::test]
9528 async fn lagged_spawn_subscriber_receives_a_terminal_lagged_error_after_its_queued_frames() {
9529 let feed = SpawnEventFeed::default();
9530 feed.configure_incarnation("lag-incarnation".to_string());
9531 let (tx, mut rx) = mpsc::channel(1);
9534 feed.subscribe(ConnectionId::new(1), 7, 1, None, FrameSink::new(tx))
9535 .expect("subscribe");
9536 let emitted = SPAWN_SUBSCRIBER_BUFFER + 16;
9537 for index in 0..emitted {
9538 feed.emit_spawned(&format!("lag-module-{index}"), 1000, 0);
9539 tokio::task::yield_now().await;
9542 }
9543 assert_eq!(
9544 feed.subscriber_count(),
9545 0,
9546 "the lagged subscriber must be removed"
9547 );
9548
9549 let mut data = Vec::new();
9550 let mut last = None;
9551 loop {
9552 let next = tokio::time::timeout(Duration::from_secs(5), rx.recv())
9553 .await
9554 .expect("the forwarder must finish once the subscriber is dropped");
9555 let Some(outbound) = next else { break };
9556 let frame = outbound.frame;
9557 if frame.header.ty == FrameType::StreamData {
9558 assert!(last.is_none(), "no data may follow the terminal frame");
9559 let event: SpawnEvent = serde_json::from_slice(&frame.body).unwrap();
9560 data.push(event.cursor.seq);
9561 } else {
9562 assert!(last.is_none(), "exactly one terminal frame");
9563 last = Some(frame);
9564 }
9565 }
9566 assert!(!data.is_empty(), "queued frames drain before the terminal");
9567 for pair in data.windows(2) {
9568 assert_eq!(
9569 pair[1],
9570 pair[0] + 1,
9571 "queued frames arrive dense and in order"
9572 );
9573 }
9574 let terminal = last.expect("a lagged subscriber must receive a terminal frame");
9575 assert_eq!(terminal.header.ty, FrameType::Error);
9576 assert_eq!(terminal.header.corr, 7);
9577 let body: subc_protocol::ErrorBody = serde_json::from_slice(&terminal.body).unwrap();
9578 assert_eq!(body.code, SPAWN_SUBSCRIBER_LAGGED_CODE);
9579 let detail = body.detail.expect("lagged error carries detail");
9580 assert_eq!(
9581 detail["first_undelivered_cursor"]["seq"],
9582 data.last().unwrap() + 1,
9583 "the named cursor is the first event the subscriber did not receive"
9584 );
9585 assert_eq!(
9586 detail["first_undelivered_cursor"]["daemon_incarnation"],
9587 "lag-incarnation"
9588 );
9589 }
9590}
9591
9592#[cfg(test)]
9593mod terminal_history_read_concurrency_tests {
9594 use super::*;
9595 use crate::terminal_journal::read_pause;
9596 use std::sync::mpsc as std_mpsc;
9597 use subc_test_support::TestTempDir;
9598
9599 fn journaled_ring(
9600 journal: &Arc<crate::terminal_journal::TerminalJournal>,
9601 ) -> Arc<Mutex<TerminalRing>> {
9602 Arc::new(Mutex::new(
9603 TerminalRing::new(TerminalRingConfig::default(), 1)
9604 .with_journal(Some(Arc::clone(journal))),
9605 ))
9606 }
9607
9608 fn crash(at_ms: u64) -> ExitReport {
9609 ExitReport {
9610 kind: ExitKind::Crash,
9611 code: Some(1),
9612 signal: None,
9613 at_ms,
9614 }
9615 }
9616
9617 fn record_within(
9620 module_id: &'static str,
9621 ring: &Arc<Mutex<TerminalRing>>,
9622 at_ms: u64,
9623 bound: Duration,
9624 ) -> bool {
9625 let ring = Arc::clone(ring);
9626 let (done, done_rx) = std_mpsc::channel();
9627 std::thread::spawn(move || {
9628 record_terminal(
9629 module_id,
9630 &ring,
9631 &SpawnEventFeed::default(),
9632 &crash(at_ms),
9633 TerminalDisposition::Restarting,
9634 );
9635 let _ = done.send(());
9636 });
9637 done_rx.recv_timeout(bound).is_ok()
9638 }
9639
9640 #[test]
9645 fn exits_recorded_during_a_paused_history_read_are_not_blocked_or_half_merged() {
9646 let dir = TestTempDir::new("terminal-history-concurrent-read");
9647 let path = dir.join("terminals.jsonl");
9648 let journal = Arc::new(crate::terminal_journal::TerminalJournal::open(
9649 path.clone(),
9650 "daemon".into(),
9651 ));
9652 let reader_ring = journaled_ring(&journal);
9653 let other_ring = journaled_ring(&journal);
9654 assert!(record_within(
9655 "reader-module",
9656 &reader_ring,
9657 10,
9658 Duration::from_secs(5)
9659 ));
9660
9661 let (started, release) = read_pause::install(&path);
9662 let reading = {
9663 let ring = Arc::clone(&reader_ring);
9664 std::thread::spawn(move || durable_terminal_history_of(&ring, "reader-module"))
9665 };
9666 started
9667 .recv_timeout(Duration::from_secs(5))
9668 .expect("the history read reached its pause");
9669
9670 let bound = Duration::from_secs(1);
9671 assert!(
9672 record_within("other-module", &other_ring, 20, bound),
9673 "another module's exit waited on a history read (journal writer held)"
9674 );
9675 assert!(
9676 record_within("reader-module", &reader_ring, 30, bound),
9677 "the read module's own exit waited on its history read (ring held)"
9678 );
9679
9680 drop(release);
9681 let paused = reading.join().unwrap();
9682 assert_eq!(
9683 paused.entries.iter().map(|e| e.at_ms).collect::<Vec<_>>(),
9684 vec![10],
9685 "an exit recorded after the read began lands in neither half of it"
9686 );
9687 assert_eq!(paused.journal_skipped_lines, 0);
9688 assert_eq!(paused.journal_read_errors, 0);
9689
9690 let after = durable_terminal_history_of(&reader_ring, "reader-module");
9691 assert_eq!(
9692 after.entries.iter().map(|e| e.at_ms).collect::<Vec<_>>(),
9693 vec![10, 30],
9694 "the next read merges ring and journal with no duplicate"
9695 );
9696 assert_eq!(after.journal_skipped_lines, 0);
9697 }
9698}
9699
9700#[cfg(test)]
9705mod stderr_settle_tests {
9706 use std::{
9707 future::Future,
9708 io,
9709 pin::Pin,
9710 sync::{Arc, Mutex},
9711 task::{Context, Poll},
9712 time::Duration,
9713 };
9714
9715 use tokio::{
9716 io::{AsyncRead, ReadBuf},
9717 sync::oneshot,
9718 time::Instant,
9719 };
9720
9721 use super::{settle_stderr_pump, StderrPump};
9722 use crate::stderr_tail::{
9723 pump_stderr_to, CaptureState, OutputSink, StderrRing, StderrTailConfig, TailEntry,
9724 };
9725
9726 const BOUND: Duration = Duration::from_millis(250);
9727
9728 struct HeldReader {
9732 before: Option<Vec<u8>>,
9733 gate: Option<oneshot::Receiver<()>>,
9734 after: io::Cursor<Vec<u8>>,
9735 }
9736
9737 impl AsyncRead for HeldReader {
9738 fn poll_read(
9739 mut self: Pin<&mut Self>,
9740 cx: &mut Context<'_>,
9741 buf: &mut ReadBuf<'_>,
9742 ) -> Poll<io::Result<()>> {
9743 if let Some(bytes) = self.before.take() {
9744 buf.put_slice(&bytes);
9745 return Poll::Ready(Ok(()));
9746 }
9747 if let Some(gate) = self.gate.as_mut() {
9748 match Pin::new(gate).poll(cx) {
9749 Poll::Pending => return Poll::Pending,
9750 Poll::Ready(_) => self.gate = None,
9751 }
9752 }
9753 Pin::new(&mut self.after).poll_read(cx, buf)
9754 }
9755 }
9756
9757 struct DiscardSink;
9758
9759 impl OutputSink for DiscardSink {
9760 fn write_line(&mut self, _line: &[u8]) {}
9761 }
9762
9763 fn line(text: &str) -> TailEntry {
9764 TailEntry::Line {
9765 text: text.to_string(),
9766 truncated: false,
9767 }
9768 }
9769
9770 fn lock(ring: &Arc<Mutex<StderrRing>>) -> std::sync::MutexGuard<'_, StderrRing> {
9771 ring.lock().unwrap()
9772 }
9773
9774 fn held_pump(
9778 ring: &Arc<Mutex<StderrRing>>,
9779 before: &str,
9780 after: &str,
9781 ) -> (StderrPump, oneshot::Sender<()>) {
9782 let generation = lock(ring).begin_process();
9783 let (release, gate) = oneshot::channel();
9784 let reader = HeldReader {
9785 before: Some(before.as_bytes().to_vec()),
9786 gate: Some(gate),
9787 after: io::Cursor::new(after.as_bytes().to_vec()),
9788 };
9789 let task = tokio::spawn(pump_stderr_to(
9790 reader,
9791 Arc::clone(ring),
9792 generation,
9793 DiscardSink,
9794 ));
9795 (StderrPump { task, generation }, release)
9796 }
9797
9798 async fn wait_until(ring: &Arc<Mutex<StderrRing>>, done: impl Fn(&StderrRing) -> bool) {
9799 for _ in 0..1000 {
9800 if done(&lock(ring)) {
9801 return;
9802 }
9803 tokio::time::sleep(Duration::from_millis(1)).await;
9804 }
9805 panic!(
9806 "ring never reached the expected state: {:?}",
9807 lock(ring).snapshot(None, None)
9808 );
9809 }
9810
9811 #[tokio::test(start_paused = true)]
9812 async fn a_crash_line_the_reader_had_not_reached_by_the_bound_is_kept_before_the_restart() {
9813 let ring = Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default())));
9814 let (pump, release) = held_pump(&ring, "booting\n", "config error: missing storage\n");
9815
9816 settle_stderr_pump("crasher", &ring, pump, BOUND).await;
9817 let before_release = lock(&ring).snapshot(None, None);
9818 assert!(
9819 matches!(before_release.capture, CaptureState::Incomplete { .. }),
9820 "a reader that has not reached EOF cannot claim a whole tail: {before_release:?}"
9821 );
9822
9823 let next = lock(&ring).begin_process();
9826 lock(&ring).push_line_from(next, "next process booting");
9827 release.send(()).unwrap();
9828 wait_until(&ring, |ring| {
9829 ring.snapshot(None, None).capture == CaptureState::Captured
9830 })
9831 .await;
9832
9833 assert_eq!(
9834 lock(&ring).snapshot(None, None).entries,
9835 vec![
9836 line("booting"),
9837 line("config error: missing storage"),
9838 TailEntry::ProcessStart,
9839 line("next process booting"),
9840 ],
9841 "the crash's last line must survive a slow reader and stay in the crashed process's section"
9842 );
9843 }
9844
9845 #[tokio::test(start_paused = true)]
9846 async fn a_pipe_held_open_by_a_descendant_reads_incomplete_without_delaying_the_restart_past_the_bound(
9847 ) {
9848 let ring = Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default())));
9849 let (pump, _held) = held_pump(&ring, "parent exiting\n", "");
9852
9853 let started = Instant::now();
9854 settle_stderr_pump("orphaning", &ring, pump, BOUND).await;
9855 assert_eq!(
9856 started.elapsed(),
9857 BOUND,
9858 "the restart must wait exactly the bound for a pipe that stays open, no longer"
9859 );
9860
9861 let next = lock(&ring).begin_process();
9862 lock(&ring).push_line_from(next, "next process booting");
9863 tokio::time::sleep(Duration::from_secs(60)).await;
9864
9865 let snapshot = lock(&ring).snapshot(None, None);
9866 match &snapshot.capture {
9867 CaptureState::Incomplete { reason } => assert!(
9868 reason.contains("had not reached EOF") && reason.contains("250ms"),
9869 "the reason must say what is missing and after how long: {reason}"
9870 ),
9871 other => panic!("expected Incomplete while the pipe is held open, got {other:?}"),
9872 }
9873 assert_eq!(
9874 snapshot.entries,
9875 vec![
9876 line("parent exiting"),
9877 TailEntry::ProcessStart,
9878 line("next process booting"),
9879 ]
9880 );
9881 }
9882
9883 #[tokio::test(start_paused = true)]
9884 async fn a_reader_that_reaches_eof_within_the_bound_leaves_the_tail_captured() {
9885 let ring = Arc::new(Mutex::new(StderrRing::new(StderrTailConfig::default())));
9886 let (pump, release) = held_pump(&ring, "one\n", "two\n");
9887 release.send(()).unwrap();
9888
9889 settle_stderr_pump("clean", &ring, pump, BOUND).await;
9890
9891 let snapshot = lock(&ring).snapshot(None, None);
9892 assert_eq!(snapshot.capture, CaptureState::Captured);
9893 assert_eq!(snapshot.entries, vec![line("one"), line("two")]);
9894 }
9895}
9896
9897#[cfg(all(test, windows))]
9911mod job_containment_tests {
9912 use super::*;
9913 use std::{
9914 path::{Path, PathBuf},
9915 sync::{Arc, Mutex},
9916 time::{Duration, Instant},
9917 };
9918 use subc_test_support::TestTempDir;
9919
9920 fn stub_path() -> PathBuf {
9926 let mut path = std::env::current_exe().expect("current_exe available in tests");
9927 path.pop();
9928 path.pop();
9929 path.push("fake-aft-stub.exe");
9930 assert!(
9931 path.exists(),
9932 "fake-aft-stub not built at {}: run `cargo test -p subc-core` (which builds \
9933 [[bin]] targets) rather than `cargo test -p subc-core --lib` (which does not)",
9934 path.display()
9935 );
9936 path
9937 }
9938
9939 fn read_grandchild_pid(path: &Path) -> u32 {
9941 let deadline = Instant::now() + Duration::from_secs(10);
9942 loop {
9943 if let Ok(contents) = std::fs::read_to_string(path) {
9944 if let Ok(pid) = contents.trim().parse() {
9945 return pid;
9946 }
9947 }
9948 assert!(
9949 Instant::now() < deadline,
9950 "the stub never recorded a grandchild pid at {}",
9951 path.display()
9952 );
9953 std::thread::sleep(Duration::from_millis(10));
9954 }
9955 }
9956
9957 struct Fixture {
9960 _dir: TestTempDir,
9961 module_id: String,
9962 grandchild: u32,
9963 child: Option<SupervisedChild>,
9964 registry: Arc<Registry>,
9965 snapshot: Arc<Mutex<SupervisorSnapshot>>,
9966 terminal_ring: Arc<Mutex<TerminalRing>>,
9967 spawn_events: SpawnEventFeed,
9968 }
9969
9970 fn fixture(label: &str, module_id: &str) -> Fixture {
9971 let dir = TestTempDir::new(label);
9972 let pid_file = dir.join("grandchild.pid");
9973 let supervisor = Supervisor::new(
9974 Arc::new(Registry::default()),
9975 RestartPolicy::new(3, Duration::ZERO),
9976 );
9977 let runtime = supervisor.runtime_config();
9978 let snapshot = Arc::new(Mutex::new(SupervisorSnapshot::starting()));
9979 let spec = ModuleSpec {
9980 module_id: module_id.to_string(),
9981 program: stub_path(),
9982 args: Vec::new(),
9986 env: vec![
9987 ("FAKE_AFT_NEVER_CONNECT".to_string(), "1".to_string()),
9988 (
9989 "FAKE_AFT_GRANDCHILD_PID_FILE".to_string(),
9990 pid_file.display().to_string(),
9991 ),
9992 ],
9993 reserved: false,
9994 reserved_prefixes: Vec::new(),
9995 protocol: ModuleProtocol::Subc,
9996 overlap: Default::default(),
9997 };
9998 let child = spawn_and_mark_running(&spec, &runtime, &snapshot)
9999 .expect("spawn the supervised fixture");
10000 let grandchild = read_grandchild_pid(&pid_file);
10001 Fixture {
10002 _dir: dir,
10003 module_id: module_id.to_string(),
10004 grandchild,
10005 child: Some(child),
10006 registry: Arc::new(Registry::default()),
10007 snapshot,
10008 terminal_ring: Arc::clone(&runtime.terminal_ring),
10009 spawn_events: SpawnEventFeed::default(),
10010 }
10011 }
10012
10013 impl Fixture {
10014 async fn drain(&mut self) {
10016 let child = self
10017 .child
10018 .take()
10019 .expect("the fixture child is still present");
10020 drain_child_to_state(
10021 &self.module_id,
10022 ModuleProtocol::Subc,
10023 StopNotice::NotSent,
10026 &self.registry,
10027 &self.snapshot,
10028 &self.terminal_ring,
10029 &self.spawn_events,
10030 child,
10031 Duration::from_millis(500),
10032 ModuleState::Stopped,
10033 Some(false),
10034 )
10035 .await
10036 .expect("drain the supervised fixture");
10037 }
10038 }
10039
10040 #[tokio::test]
10046 async fn teardown_reaps_the_grandchild() {
10047 let mut fixture = fixture("teardown-grandchild", "tree-teardown");
10048 let grandchild = fixture.grandchild;
10049
10050 assert!(
10051 subc_jobobject::process_exists(grandchild),
10052 "grandchild {grandchild} must be alive before teardown, or this proves nothing"
10053 );
10054
10055 fixture.drain().await;
10056
10057 assert!(
10058 subc_jobobject::wait_for_process_exit(grandchild, Duration::from_secs(10)),
10059 "grandchild {grandchild} outlived module teardown: the tree was not contained"
10060 );
10061 }
10062
10063 #[test]
10076 fn an_uncontained_grandchild_survives_a_direct_child_kill() {
10077 let dir = TestTempDir::new("teardown-uncontained");
10078 let pid_file = dir.join("grandchild.pid");
10079 let mut child = std::process::Command::new(stub_path())
10080 .env("FAKE_AFT_NEVER_CONNECT", "1")
10081 .env(
10082 "FAKE_AFT_GRANDCHILD_PID_FILE",
10083 pid_file.display().to_string(),
10084 )
10085 .stdin(std::process::Stdio::null())
10086 .stdout(std::process::Stdio::null())
10087 .stderr(std::process::Stdio::null())
10088 .spawn()
10089 .expect("spawn the uncontained fixture");
10090 let grandchild = read_grandchild_pid(&pid_file);
10091
10092 child.kill().expect("kill the direct child");
10094 let _ = child.wait();
10095
10096 assert!(
10097 subc_jobobject::process_exists(grandchild),
10098 "grandchild {grandchild} died with the direct child, so this control no longer \
10099 distinguishes contained from uncontained teardown and the regression test is \
10100 passing vacuously"
10101 );
10102
10103 kill_tree(grandchild);
10106 }
10107
10108 #[tokio::test]
10117 async fn dropping_containment_reaps_the_grandchild() {
10118 let mut fixture = fixture("drop-containment", "tree-drop");
10119 let grandchild = fixture.grandchild;
10120
10121 assert!(subc_jobobject::process_exists(grandchild));
10122
10123 fixture.child.as_mut().expect("child present").job = None;
10125
10126 assert!(
10127 subc_jobobject::wait_for_process_exit(grandchild, Duration::from_secs(10)),
10128 "grandchild {grandchild} survived the containment handle closing, so a daemon \
10129 crash would leave the tree behind"
10130 );
10131 }
10132
10133 fn kill_tree(pid: u32) {
10135 let _ = std::process::Command::new("taskkill.exe")
10136 .args(["/PID", &pid.to_string(), "/T", "/F"])
10137 .stdin(std::process::Stdio::null())
10138 .stdout(std::process::Stdio::null())
10139 .stderr(std::process::Stdio::null())
10140 .status();
10141 assert!(
10142 subc_jobobject::wait_for_process_exit(pid, Duration::from_secs(10)),
10143 "could not clean up grandchild {pid}"
10144 );
10145 }
10146}