michal/tit

Browse tree · Show commit · Download archive

Diff

ce1ffd96d52d → b48930ca1739

@@ -30,9 +30,16 @@
 `ssh_host_ed25519_key` with mode 600 during the first start and uses the same
 host key during subsequent starts. Keep this file with the instance data.
 
-The server owns the instance lock until it receives SIGINT or SIGTERM. Stop the
-server before you run an offline administrator command. Create a signup code
-through the control socket while the server runs:
+The server owns the instance lock until it receives SIGINT or SIGTERM. It does
+not put a process ID in the lock file. Stop the server before you run an offline
+administrator command.
+
+Use `GET /healthz` or `HEAD /healthz` as the readiness probe. The endpoint
+returns status 200 only after the HTTP and SSH listeners are ready. During
+shutdown, the server gives active connections 10 seconds to finish. It then
+cancels unfinished connections.
+
+Create a signup code through the control socket while the server runs:
 
 ```text
 tit --config /srv/tit/config.toml invite-code
@@ -476,3 +483,17 @@
 merges, concurrent ref updates, and recovery after an interrupted ref update.
 It also tests the Web workflow, the SSH checkout command, and historical schema
 migration.
+
+## Milestone 6.1 gate
+
+Run the process lifecycle gate:
+
+```text
+./scripts/check-m6-1
+```
+
+This command tests listener readiness, SIGTERM shutdown, the bounded connection
+drain, the empty advisory lock file, and rejection of a second server process.
+Read the
+[process lifecycle architectural decision record](docs/adr/0028-process-lifecycle.md)
+for the readiness, shutdown, and instance-lock contracts.

@@ -1,0 +1,53 @@
+# Architectural decision record 0028: process lifecycle
+
+Status: Accepted
+
+Date: 2026-07-23
+
+## Context
+
+The server already held one instance lock and stopped its listeners after
+SIGINT or SIGTERM. An active HTTP connection could stop shutdown for an
+unlimited time. An operator also had no endpoint that showed when both public
+listeners were ready.
+
+## Decision
+
+Keep the advisory lock in the empty `tit.lock` file for the full server run.
+Do not write a process ID to this file. Refuse an unsafe lock-file path and
+refuse a second process that requests the lock.
+
+Add `GET /healthz` and `HEAD /healthz`. Return status 503 until the HTTP and SSH
+listeners are bound. Then, return status 200 with `ready`. Set the state to not
+ready before shutdown starts. Always send `Cache-Control: no-store`.
+
+After SIGINT or SIGTERM, stop all listeners at the same time. Give HTTP, SSH,
+and control-socket tasks 10 seconds to finish. Cancel a task that does not
+finish during this time. Write one diagnostic to standard error if shutdown
+cancels an unfinished connection.
+
+## Failure and threat cases
+
+A process ID can be stale and can identify an unrelated process after PID
+reuse. The advisory file lock is the ownership authority, so the file stays
+empty.
+
+The readiness endpoint does not report ready while only HTTP is available. It
+does not use a cached response after shutdown starts. A slow or incomplete
+client cannot keep the process alive after the drain limit.
+
+## Evidence
+
+The production server test starts both listeners and checks the readiness
+response. It starts a second server for the same instance and checks that the
+instance lock rejects it. It also checks clean SIGTERM shutdown and restart.
+
+The Web server test holds an incomplete HTTP request open. It uses a short test
+drain limit and checks that shutdown cancels the connection after that limit.
+The instance-lock test checks that the lock file is empty.
+
+## Consequences
+
+Supervisors can use `/healthz` as the readiness probe. Shutdown permits current
+work to finish, but it has a fixed upper bound. A request that exceeds the
+bound can be canceled.

@@ -1,0 +1,7 @@
+#!/bin/sh
+set -eu
+
+./scripts/check
+cargo test --locked --release --bin tit instance::tests::serializes_instance_owners_and_rejects_unsafe_files
+cargo test --locked --release --test web_shell cancels_a_connection_after_the_shutdown_drain_limit
+cargo test --locked --release --test serve serves_an_imported_repository_through_http_and_ssh

@@ -7,7 +7,7 @@
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{UnixListener, UnixStream};
 use tokio::sync::oneshot;
-use tokio::task::JoinHandle;
+use tokio::task::{JoinHandle, JoinSet};
 
 use crate::account::{AccountError, AccountService};
 
@@ -66,18 +66,22 @@
         let (shutdown, mut receiver) = oneshot::channel();
         let task = tokio::spawn(async move {
             let _cleanup = cleanup;
+            let mut connections = JoinSet::new();
             loop {
                 tokio::select! {
-                    _ = &mut receiver => return Ok(()),
+                    _ = &mut receiver => break,
                     accepted = listener.accept() => {
                         let (stream, _) = accepted.map_err(ControlError::Accept)?;
                         let service = accounts.clone();
-                        tokio::spawn(async move {
+                        connections.spawn(async move {
                             let _ = handle(stream, service).await;
                         });
-                    }
+                    },
+                    _ = connections.join_next(), if !connections.is_empty() => {}
                 }
             }
+            while connections.join_next().await.is_some() {}
+            Ok(())
         });
         Ok(Self { shutdown, task })
     }
@@ -86,6 +90,21 @@
         let _ = self.shutdown.send(());
         self.task.await.map_err(|_| ControlError::Join)??;
         Ok(())
+    }
+
+    pub(crate) async fn shutdown_bounded(mut self, limit: Duration) -> Result<bool, ControlError> {
+        let _ = self.shutdown.send(());
+        match tokio::time::timeout(limit, &mut self.task).await {
+            Ok(result) => {
+                result.map_err(|_| ControlError::Join)??;
+                Ok(true)
+            }
+            Err(_) => {
+                self.task.abort();
+                let _ = self.task.await;
+                Ok(false)
+            }
+        }
     }
 }
 

@@ -8,6 +8,8 @@
 use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
 
 use askama::Template;
 use axum::Router;
@@ -57,6 +59,7 @@
     feeds: Option<FeedTokenService>,
     search: Option<MetadataSearchService>,
     watches: Option<WatchService>,
+    readiness: Option<ListenerReadiness>,
     secure_cookies: bool,
 }
 
@@ -70,6 +73,25 @@
     pub(crate) instance_dir: PathBuf,
     pub(crate) http_clone_base: String,
     pub(crate) ssh_clone_base: String,
+}
+
+#[derive(Clone, Default)]
+pub(crate) struct ListenerReadiness {
+    ready: Arc<AtomicBool>,
+}
+
+impl ListenerReadiness {
+    pub(crate) fn mark_ready(&self) {
+        self.ready.store(true, Ordering::Release);
+    }
+
+    pub(crate) fn mark_stopping(&self) {
+        self.ready.store(false, Ordering::Release);
+    }
+
+    fn is_ready(&self) -> bool {
+        self.ready.load(Ordering::Acquire)
+    }
 }
 
 pub(crate) struct RunningWebServer {
@@ -94,6 +116,7 @@
                 feeds: None,
                 search: None,
                 watches: None,
+                readiness: None,
                 secure_cookies: false,
             },
         )
@@ -104,21 +127,23 @@
         address: SocketAddr,
         config: PublicWebConfig,
     ) -> Result<Self, WebError> {
-        Self::start_public_inner(address, config, None).await
+        Self::start_public_inner(address, config, None, None).await
     }
 
     pub(crate) async fn start_public_with_key_reload(
         address: SocketAddr,
         config: PublicWebConfig,
         key_reloader: AccountKeyReloader,
+        readiness: ListenerReadiness,
     ) -> Result<Self, WebError> {
-        Self::start_public_inner(address, config, Some(key_reloader)).await
+        Self::start_public_inner(address, config, Some(key_reloader), Some(readiness)).await
     }
 
     async fn start_public_inner(
         address: SocketAddr,
         config: PublicWebConfig,
         key_reloader: Option<AccountKeyReloader>,
+        readiness: Option<ListenerReadiness>,
     ) -> Result<Self, WebError> {
         let jobs = Arc::new(Semaphore::new(MAX_BLOCKING_WEB_JOBS));
         let database = config.instance_dir.join(crate::store::DATABASE_FILE);
@@ -148,6 +173,7 @@
                 feeds: Some(feeds),
                 search: Some(search),
                 watches: Some(watches),
+                readiness,
                 secure_cookies,
             },
         )
@@ -181,6 +207,21 @@
         self.task.await.map_err(|_| WebError::Join)??;
         Ok(())
     }
+
+    pub(crate) async fn shutdown_bounded(mut self, limit: Duration) -> Result<bool, WebError> {
+        let _ = self.shutdown.send(());
+        match tokio::time::timeout(limit, &mut self.task).await {
+            Ok(result) => {
+                result.map_err(|_| WebError::Join)??;
+                Ok(true)
+            }
+            Err(_) => {
+                self.task.abort();
+                let _ = self.task.await;
+                Ok(false)
+            }
+        }
+    }
 }
 
 pub(crate) fn router() -> Router {
@@ -196,6 +237,7 @@
         feeds: None,
         search: None,
         watches: None,
+        readiness: None,
         secure_cookies: false,
     })
 }
@@ -212,6 +254,7 @@
         ));
     Router::new()
         .route("/", get(home))
+        .route("/healthz", get(health))
         .route("/go", get(go_to_repository))
         .route(
             "/signup",
@@ -275,6 +318,24 @@
 
 async fn home(Extension(request_id): Extension<RequestId>) -> Response {
     render_home(StatusCode::OK, &request_id.0, "", "", "")
+}
+
+async fn health(State(state): State<WebState>) -> Response {
+    let ready = state
+        .readiness
+        .as_ref()
+        .is_none_or(ListenerReadiness::is_ready);
+    let (status, body) = if ready {
+        (StatusCode::OK, "ready\n")
+    } else {
+        (StatusCode::SERVICE_UNAVAILABLE, "not ready\n")
+    };
+    Response::builder()
+        .status(status)
+        .header(header::CACHE_CONTROL, "no-store")
+        .header(header::CONTENT_TYPE, "text/plain; charset=utf-8")
+        .body(Body::from(body))
+        .expect("the readiness response is valid")
 }
 
 async fn go_to_repository(

@@ -190,6 +190,11 @@
         InstanceLock::acquire(directory.path()).expect("acquire the released instance lock");
 
         let lock_path = directory.path().join(LOCK_FILE);
+        assert!(
+            fs::read(&lock_path)
+                .expect("read the instance lock file")
+                .is_empty()
+        );
         fs::remove_file(&lock_path).expect("remove the lock file");
         symlink(directory.path().join("target"), &lock_path).expect("create a lock-file symlink");
         assert!(matches!(

@@ -2,6 +2,7 @@
 use std::io::{Read, Write};
 use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
 use std::path::{Path, PathBuf};
+use std::time::Duration;
 
 use rand::rng;
 use ssh_key::{Algorithm, LineEnding, PrivateKey};
@@ -12,12 +13,14 @@
 use crate::config::{Config, ConfigError};
 use crate::control::{ControlError, RunningControlServer};
 use crate::git::transport::{GitRepositories, RepositoryPathError};
-use crate::http::{PublicWebConfig, RunningWebServer, WebError};
+use crate::http::{ListenerReadiness, PublicWebConfig, RunningWebServer, WebError};
 use crate::instance::{InstanceError, InstanceLock, prepare_database, prepare_repository_root};
 use crate::policy::PolicyError;
 use crate::pull_request::{PullRequestError, PullRequestService};
 use crate::ssh::{AuthorizedSshKeys, RunningSshServer, SshServerError};
 use crate::store::{Store, StoreError};
+
+const SHUTDOWN_DRAIN_LIMIT: Duration = Duration::from_secs(10);
 
 pub(crate) async fn run(config: &Config) -> Result<(), ServeError> {
     let _lock = InstanceLock::acquire(&config.instance_dir)?;
@@ -32,6 +35,7 @@
     let accounts = AccountService::new(database);
     let control = RunningControlServer::start(&config.instance_dir, accounts.clone())?;
     let authorized_keys = AuthorizedSshKeys::for_accounts(keys);
+    let readiness = ListenerReadiness::default();
 
     let (http_clone_base, ssh_clone_base) = clone_bases(config)?;
     let host_key = load_or_create_host_key(&config.instance_dir)?;
@@ -52,6 +56,7 @@
             ssh_clone_base,
         },
         reload_keys,
+        readiness.clone(),
     )
     .await?;
     let ssh = match RunningSshServer::start_with_dynamic_keys(
@@ -69,13 +74,22 @@
             return Err(error.into());
         }
     };
+    readiness.mark_ready();
 
     let signal = shutdown_signal().await;
-    let (ssh_result, web_result, control_result) =
-        tokio::join!(ssh.shutdown(), web.shutdown(), control.shutdown());
-    ssh_result?;
-    web_result?;
-    control_result?;
+    readiness.mark_stopping();
+    let (ssh_result, web_result, control_result) = tokio::join!(
+        ssh.shutdown_bounded(SHUTDOWN_DRAIN_LIMIT),
+        web.shutdown_bounded(SHUTDOWN_DRAIN_LIMIT),
+        control.shutdown_bounded(SHUTDOWN_DRAIN_LIMIT)
+    );
+    let ssh_drained = ssh_result?;
+    let web_drained = web_result?;
+    let control_drained = control_result?;
+    let drained = ssh_drained && web_drained && control_drained;
+    if !drained {
+        eprintln!("tit: the shutdown drain limit expired; unfinished connections were canceled");
+    }
     signal.map_err(ServeError::Signal)
 }
 

@@ -259,6 +259,24 @@
         self.task.await.map_err(|_| SshServerError::Join)??;
         Ok(())
     }
+
+    pub(crate) async fn shutdown_bounded(
+        mut self,
+        limit: Duration,
+    ) -> Result<bool, SshServerError> {
+        self.handle.shutdown("tit shutdown".to_owned());
+        match tokio::time::timeout(limit, &mut self.task).await {
+            Ok(result) => {
+                result.map_err(|_| SshServerError::Join)??;
+                Ok(true)
+            }
+            Err(_) => {
+                self.task.abort();
+                let _ = self.task.await;
+                Ok(false)
+            }
+        }
+    }
 }
 
 async fn recover_pushes(repositories: &GitRepositories) -> Result<(), SshServerError> {

@@ -73,6 +73,21 @@
     let mut server = spawn_server(&config);
     wait_for_listener(http, &mut server);
     wait_for_listener(ssh, &mut server);
+    let health = http_get(http, "/healthz");
+    assert!(health.starts_with("HTTP/1.1 200"));
+    assert!(health.ends_with("\r\n\r\nready\n"));
+
+    let second = Command::new(env!("CARGO_BIN_EXE_tit"))
+        .args([
+            "--config",
+            config.to_str().expect("a UTF-8 configuration path"),
+            "serve",
+        ])
+        .output()
+        .expect("start a second tit server");
+    assert!(!second.status.success());
+    assert!(String::from_utf8_lossy(&second.stderr).contains("owns the instance lock"));
+
     let control_socket = instance.path().join("control.sock");
     assert_eq!(
         fs::symlink_metadata(&control_socket)

@@ -61,8 +61,10 @@
 use std::collections::BTreeMap;
 use std::io::{Read, Write};
 use std::net::{Ipv4Addr, SocketAddr, TcpStream};
+use std::time::Duration;
 
 use http::RunningWebServer;
+use tokio::io::AsyncWriteExt;
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn serves_the_semantic_shell_without_javascript() {
@@ -235,6 +237,26 @@
     assert_ne!(first.header("x-request-id"), second.header("x-request-id"));
 
     server.shutdown().await.expect("stop the Web server");
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn cancels_a_connection_after_the_shutdown_drain_limit() {
+    let server = start().await;
+    let mut stalled = tokio::net::TcpStream::connect(server.address())
+        .await
+        .expect("connect a stalled client");
+    stalled
+        .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n")
+        .await
+        .expect("write an incomplete request");
+    tokio::time::sleep(Duration::from_millis(20)).await;
+
+    assert!(
+        !server
+            .shutdown_bounded(Duration::from_millis(20))
+            .await
+            .expect("stop the Web server")
+    );
 }
 
 async fn start() -> RunningWebServer {