supervised mode: run the services as child processes

SCROLL_MODE=supervised spawns auth-service, game-service and gateway, waits for
each rpc port before starting the next, and restarts whatever dies with backoff
from 500ms to 30s. the counter resets once a service has been up a minute.

single process mode is still the default and unchanged.

kill_on_drop means ctrl-c takes the children with it, no orphans.

Cargo.toml

@@ -19,7 +19,7 @@ game-service = { path = "crates/game-service" }
 gateway = { path = "crates/gateway" }
 storage = { path = "crates/storage" }
 
-tokio = { version = "1.45", features = ["rt-multi-thread", "macros", "net", "io-util", "sync", "time", "signal", "fs"] }
+tokio = { version = "1.45", features = ["rt-multi-thread", "macros", "net", "io-util", "sync", "time", "signal", "fs", "process"] }
 thiserror = "2"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }

README.md

@@ -24,7 +24,15 @@ on first boot. `DATABASE_URL` defaults to `postgres:///scroll`.
 
 On macOS a Homebrew cluster may need `LC_ALL=C` to start at all.
 
-Services can also run separately:
+That runs everything in one process. For real isolation, `SCROLL_MODE=supervised`
+makes it spawn the three services as child processes and restart any that dies,
+with backoff:
+
+```bash
+SCROLL_MODE=supervised ./target/release/scroll-server
+```
+
+They can also be started by hand, in any order:
 
 ```bash
 cargo run --release -p auth-service
@@ -55,6 +63,10 @@ nothing around it moves.
 | variable | default |
 | --- | --- |
 | `DATABASE_URL` | `postgres:///scroll` |
+| `SCROLL_MODE` | `single`, or `supervised` |
+| `SCROLL_BIN_DIR` | next to the running binary |
+| `SCROLL_AUTH_LISTEN` | `127.0.0.1:9401` |
+| `SCROLL_GAME_LISTEN` | `127.0.0.1:9402` |
 | `SCROLL_GATEWAY_LISTEN` | `0.0.0.0:9339` |
 | `SCROLL_CSV_ROOT` | `assets` |
 | `SCROLL_SHOP` | `config/shop.json` |

crates/scroll-server/src/main.rs

@@ -1,7 +1,14 @@
+mod supervisor;
 use std::sync::Arc;
 use auth_service::{AuthConfig, AuthService};
 use game_service::{GameConfig, GameService};
 use gateway::{Backends, GatewayConfig};
+use crate::supervisor::ServiceSpec;
+fn supervised() -> bool {
+    std::env::var("SCROLL_MODE")
+        .map(|mode| mode.eq_ignore_ascii_case("supervised"))
+        .unwrap_or(false)
+}
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     tracing_subscriber::fmt()
@@ -10,6 +17,32 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
         )
         .init();
+    if supervised() {
+        return run_supervised().await;
+    }
+    run_single_process().await
+}
+async fn run_supervised() -> Result<(), Box<dyn std::error::Error>> {
+    let directory = supervisor::binary_directory()?;
+    let auth = AuthConfig::from_env();
+    let game = GameConfig::from_env();
+    let gateway = GatewayConfig::from_env();
+    tracing::info!(
+        directory = %directory.display(),
+        auth = %auth.listen,
+        game = %game.listen,
+        listen = %gateway.listen,
+        "scroll server starting in supervised mode"
+    );
+    supervisor::run(vec![
+        ServiceSpec::new("auth-service", &directory, Some(auth.listen)),
+        ServiceSpec::new("game-service", &directory, Some(game.listen)),
+        ServiceSpec::new("gateway", &directory, None),
+    ])
+    .await?;
+    Ok(())
+}
+async fn run_single_process() -> Result<(), Box<dyn std::error::Error>> {
     let auth = AuthService::bootstrap(AuthConfig::from_env()).await?;
     let game = GameService::bootstrap(GameConfig::from_env()).await?;
     let gateway_config = GatewayConfig::from_env();

crates/scroll-server/src/supervisor.rs

@@ -0,0 +1,129 @@
+use std::path::{Path, PathBuf};
+use std::process::Stdio;
+use std::time::Duration;
+use tokio::process::{Child, Command};
+use tokio::time::Instant;
+pub const RESTART_DELAY_MIN: Duration = Duration::from_millis(500);
+pub const RESTART_DELAY_MAX: Duration = Duration::from_secs(30);
+pub const HEALTHY_AFTER: Duration = Duration::from_secs(60);
+pub const READY_TIMEOUT: Duration = Duration::from_secs(30);
+pub const READY_POLL: Duration = Duration::from_millis(100);
+#[derive(Debug, Clone)]
+pub struct ServiceSpec {
+    pub name: &'static str,
+    pub program: PathBuf,
+    pub endpoint: Option<String>,
+}
+impl ServiceSpec {
+    pub fn new(name: &'static str, directory: &Path, endpoint: Option<String>) -> Self {
+        Self {
+            name,
+            program: directory.join(name),
+            endpoint,
+        }
+    }
+}
+pub fn binary_directory() -> std::io::Result<PathBuf> {
+    if let Some(directory) = std::env::var_os("SCROLL_BIN_DIR") {
+        return Ok(PathBuf::from(directory));
+    }
+    let executable = std::env::current_exe()?;
+    executable
+        .parent()
+        .map(Path::to_path_buf)
+        .ok_or_else(|| std::io::Error::other("the running binary has no parent directory"))
+}
+fn spawn(spec: &ServiceSpec) -> std::io::Result<Child> {
+    Command::new(&spec.program)
+        .stdout(Stdio::inherit())
+        .stderr(Stdio::inherit())
+        .kill_on_drop(true)
+        .spawn()
+}
+async fn wait_until_ready(spec: &ServiceSpec) -> bool {
+    let Some(endpoint) = spec.endpoint.as_deref() else {
+        return true;
+    };
+    let deadline = Instant::now() + READY_TIMEOUT;
+    while Instant::now() < deadline {
+        if tokio::net::TcpStream::connect(endpoint).await.is_ok() {
+            return true;
+        }
+        tokio::time::sleep(READY_POLL).await;
+    }
+    false
+}
+async fn supervise(spec: ServiceSpec) {
+    let mut delay = RESTART_DELAY_MIN;
+    let mut generation: u32 = 0;
+    loop {
+        generation += 1;
+        let started = Instant::now();
+        let mut child = match spawn(&spec) {
+            Ok(child) => child,
+            Err(error) => {
+                tracing::error!(
+                    service = spec.name,
+                    program = %spec.program.display(),
+                    %error,
+                    "could not start the service, retrying"
+                );
+                tokio::time::sleep(delay).await;
+                delay = (delay * 2).min(RESTART_DELAY_MAX);
+                continue;
+            }
+        };
+        tracing::info!(
+            service = spec.name,
+            pid = child.id().unwrap_or(0),
+            generation,
+            "service started"
+        );
+        let status = child.wait().await;
+        let uptime = started.elapsed();
+        if uptime >= HEALTHY_AFTER {
+            delay = RESTART_DELAY_MIN;
+        }
+        match status {
+            Ok(status) => tracing::warn!(
+                service = spec.name,
+                code = status.code().unwrap_or(-1),
+                uptime_seconds = uptime.as_secs(),
+                restart_in_ms = delay.as_millis() as u64,
+                "service exited, restarting"
+            ),
+            Err(error) => tracing::error!(
+                service = spec.name,
+                %error,
+                restart_in_ms = delay.as_millis() as u64,
+                "lost track of the service, restarting"
+            ),
+        }
+        tokio::time::sleep(delay).await;
+        delay = (delay * 2).min(RESTART_DELAY_MAX);
+    }
+}
+pub async fn run(specs: Vec<ServiceSpec>) -> std::io::Result<()> {
+    for spec in &specs {
+        if !spec.program.is_file() {
+            return Err(std::io::Error::other(format!(
+                "{} is not next to the supervisor, set SCROLL_BIN_DIR",
+                spec.program.display()
+            )));
+        }
+    }
+    let mut tasks = Vec::new();
+    for spec in specs {
+        tasks.push(tokio::spawn(supervise(spec.clone())));
+        if !wait_until_ready(&spec).await {
+            tracing::warn!(service = spec.name, "service did not open its port in time");
+        }
+    }
+    tokio::signal::ctrl_c().await?;
+    tracing::info!("shutdown requested, stopping services");
+    for task in tasks {
+        task.abort();
+    }
+    tokio::time::sleep(RESTART_DELAY_MIN).await;
+    Ok(())
+}