diff --git a/crates/xy-supervisor/src/supervisor.rs b/crates/xy-supervisor/src/supervisor.rs index 42edcf9..329490c 100644 --- a/crates/xy-supervisor/src/supervisor.rs +++ b/crates/xy-supervisor/src/supervisor.rs @@ -134,7 +134,7 @@ impl SupervisorTask { if child.is_some() { let _ = ack.send(StartAck::AlreadyRunning); } else { - match self.do_start().await { + match self.do_start(StartCause::Initial).await { Ok(c) => { child = Some(c); let _ = ack.send(StartAck::Started); @@ -161,7 +161,7 @@ impl SupervisorTask { self.do_stop(c).await; } - match self.do_start().await { + match self.do_start(StartCause::Restart).await { Ok(c) => child = Some(c), Err(err) => { warn!(name = %self.cfg.name, error = %err, "restart spawn failed"); @@ -267,7 +267,7 @@ impl SupervisorTask { match action { Action::RetryNow => { - match self.do_start().await { + match self.do_start(StartCause::Restart).await { Ok(c) => child = Some(c), Err(err) => { warn!(name = %self.cfg.name, error = %err, "restart spawn failed"); @@ -288,12 +288,15 @@ impl SupervisorTask { } } - async fn do_start(&mut self) -> std::io::Result { + async fn do_start(&mut self, cause: StartCause) -> std::io::Result { self.set_state(ServerState::Starting); let c = self.spawner.spawn(&self.cfg, self.log_sink.clone()).await?; - self.restart_count = self.restart_count.saturating_add(1); + if cause == StartCause::Restart { + self.restart_count = self.restart_count.saturating_add(1); + } + self.started_at = Some(Instant::now()); self.current_pid = Some(c.pid()); self.backoff.reset(); @@ -332,6 +335,12 @@ async fn wait_child(slot: &mut Option) -> Option { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum StartCause { + Initial, + Restart, +} + pub struct RealSpawner; #[async_trait::async_trait] @@ -404,6 +413,19 @@ mod tests { } } + async fn wait_for_pid(rx: &mut watch::Receiver, want: u32) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + loop { + if rx.borrow().pid == Some(want) { + return; + } + tokio::select! { + _ = rx.changed() => {} + _ = tokio::time::sleep_until(deadline) => panic!("never saw pid {want}, last={:?}", rx.borrow().pid), + } + } + } + async fn wait_for(rx: &mut watch::Receiver, want: ServerState) { let deadline = tokio::time::Instant::now() + Duration::from_secs(2); loop { @@ -448,4 +470,86 @@ mod tests { ack_rx.await.unwrap(); h.await.unwrap(); } + + #[tokio::test] + async fn first_start_is_not_counted_as_a_restart() { + let cfg = cfg("x", RestartPolicy::Never, 5); + let (mock, ctl) = MockChild::new(1); + let queue = Arc::new(Mutex::new(vec![mock])); + let spawner = QueueSpawner { queue }; + + let (status_tx, mut status_rx) = watch::channel(initial_status(&cfg)); + let (cmd_tx, cmd_rx) = mpsc::channel(8); + let task = SupervisorTask::new(cfg, sink("x"), spawner, status_tx, cmd_rx); + let h = tokio::spawn(task.run()); + + let (ack_tx, ack_rx) = oneshot::channel(); + cmd_tx + .send(SupervisorCmd::Start { ack: ack_tx }) + .await + .unwrap(); + assert_eq!(ack_rx.await.unwrap(), StartAck::Started); + wait_for(&mut status_rx, ServerState::Running).await; + + assert_eq!( + status_rx.borrow().restart_count, + 0, + "a server started once has not restarted" + ); + + // do_stop's post-kill wait() is unbounded, and MockChild only reports an + // exit when its controller does; dropping it closes the channel so + // shutdown can finish. + drop(ctl); + + let (ack_tx, ack_rx) = oneshot::channel(); + cmd_tx + .send(SupervisorCmd::Shutdown { ack: ack_tx }) + .await + .unwrap(); + ack_rx.await.unwrap(); + h.await.unwrap(); + } + + #[tokio::test] + async fn an_automatic_restart_after_a_crash_is_counted() { + let cfg = cfg("x", RestartPolicy::Always, 5); + let (first, mut ctl) = MockChild::new(1); + let (second, ctl2) = MockChild::new(2); + let queue = Arc::new(Mutex::new(vec![first, second])); + let spawner = QueueSpawner { queue }; + + let (status_tx, mut status_rx) = watch::channel(initial_status(&cfg)); + let (cmd_tx, cmd_rx) = mpsc::channel(8); + let task = SupervisorTask::new(cfg, sink("x"), spawner, status_tx, cmd_rx); + let h = tokio::spawn(task.run()); + + let (ack_tx, ack_rx) = oneshot::channel(); + cmd_tx + .send(SupervisorCmd::Start { ack: ack_tx }) + .await + .unwrap(); + assert_eq!(ack_rx.await.unwrap(), StartAck::Started); + wait_for(&mut status_rx, ServerState::Running).await; + + ctl.exit_tx.take().unwrap().send(Some(1)).unwrap(); + + wait_for_pid(&mut status_rx, 2).await; + + assert_eq!( + status_rx.borrow().restart_count, + 1, + "one crash-and-respawn is one restart" + ); + + drop(ctl2); + + let (ack_tx, ack_rx) = oneshot::channel(); + cmd_tx + .send(SupervisorCmd::Shutdown { ack: ack_tx }) + .await + .unwrap(); + ack_rx.await.unwrap(); + h.await.unwrap(); + } }