runtime: ranges, health, logs, supervised loops, strict config

Media answers Range requests with 206 and Content-Range, so videos seek
and Safari plays them. Sealed files too: the vault's opener now decrypts
by record position (STREAM BE32 through StreamPrimitive, the same
construction and the same files), so a range starts at the record that
holds its first byte. Compression is limited to text, because a gzipped
body no longer matches the byte ranges a player asks for.

GET /api/health answers without a session once the database does, and
the compose healthcheck asks it instead of only opening the port.
Requests are logged at info with share tokens and keys left out of the
path and no query strings. A request that is not answered in 30 seconds
gets a 503, except uploads, imports, the export, re-sealing and backups.

The backup, mail, sweep and SFTP loops run under a supervisor that logs
a panic and starts the loop again. The backup loop starts without a
Proton session and picks up a later proton-login without a restart. A
lock file keeps backup-now and the server from mirroring at once, and an
unreadable session key file turns backups off instead of failing boot.

Numbers in the configuration are checked against bounds at startup, so
a session length or an upload limit can no longer overflow later, and a
switch that is neither on nor off is an error rather than off. Config
and MailConfig print with the password, secret and SMTP URL redacted.

A session epoch is signed into each cookie (migration 0014). :logout!
advances it, which ends every session, and a changed WORKSPACE_PASSWORD
advances it on the next start. Cookies issued before this keep working
in the first epoch.

Cargo.toml

@@ -62,6 +62,10 @@ russh-sftp = "3"
 # speaks rand_core 0.10, one major version past the `rand` used everywhere else.
 rand10 = { package = "rand", version = "0.10" }
 
+[dev-dependencies]
+# A paused clock, so tests of timed loops do not wait in real time.
+tokio = { version = "1", features = ["test-util"] }
+
 [features]
 default = []
 # Keep the Proton session in the OS keyring instead of a 0600 file. Needs a

README.md

@@ -37,7 +37,11 @@ spelling still accepted behind it, so a server configured before the rename
 keeps running untouched.
 
 `WORKSPACE_SECRET` signs the session cookie; generate one with
-`openssl rand -base64 32`. Changing it logs every device out. The database and
+`openssl rand -base64 32`. Changing it logs every device out, and so does
+changing `WORKSPACE_PASSWORD` or `:logout!`, which ends every session but
+needs no restart. A setting that is set but cannot be read — a number out of
+range, a switch that is neither on nor off — stops the start with the reason,
+rather than being taken for its default. The database and
 uploads live under `WORKSPACE_DATA_DIR` (`./data` by default) — that directory is
 the entire backup, and it is exactly what the Proton Drive mirror copies.
 
@@ -445,11 +449,14 @@ both halves.
 
 ## API
 
-Everything except the two share routes requires the session cookie.
+Everything except the login, the health check and the two share routes
+requires the session cookie. Media answers `Range` requests, so a video seeks,
+sealed or not.
 
 | method | path | purpose |
 | --- | --- | --- |
-| `POST` | `/api/login`, `/api/logout` | session |
+| `GET` | `/api/health` | public: up, and the database answers — for a supervisor |
+| `POST` | `/api/login`, `/api/logout` | session; `/api/logout?everywhere=true` ends every one |
 | `GET` | `/api/me` | current user |
 | `GET`/`POST` | `/api/spaces` | the spaces / create one |
 | `GET`/`POST` | `/api/nodes` | children (`?parent=`) or search (`?q=`, `?space=`) / create |
@@ -494,7 +501,17 @@ so a proxy must pass the `Host` header through unchanged (traefik and nginx's
 `proxy_set_header Host $host` both do). Only the upload routes accept a body as
 large as `WORKSPACE_MAX_UPLOAD_MB`; everything else stops at 16 MiB, and the
 login at 16 KiB, before anyone has proved who they are. `SIGTERM` gets in-flight
-requests eight seconds to finish, inside the ten `docker stop` waits.
+requests eight seconds to finish, inside the ten `docker stop` waits. Any
+other request that has not been answered after 30 seconds is given up on with
+a `503`, except the ones that move files around: uploads, imports, the export,
+re-sealing and a backup run.
+
+Every request is logged at `info`, path and status and time, never a query
+string, and with a share link's token and key left out. A background loop that
+panics — the mirror, the mail, the sweep, the SFTP server — is logged and
+started again rather than lost until the next restart. If
+`WORKSPACE_SESSION_KEY_FILE` cannot be read the server starts anyway with
+backups off, and `:backup` says why.
 
 ### Docker
 

compose.yml

@@ -28,8 +28,9 @@ services:
       # rather than relying on the reverse proxy — see compose.override.yml.example.
       # - "127.0.0.1:2222:2222"
     healthcheck:
-      # No extra tooling in the image, so probe the port with bash directly.
-      test: ["CMD", "bash", "-c", "exec 3<>/dev/tcp/127.0.0.1/4242"]
+      # The image has bash and nothing like curl: a request by hand, and a
+      # 200 from /api/health, which also asks the database.
+      test: ["CMD", "bash", "-c", "exec 3<>/dev/tcp/127.0.0.1/4242 && printf 'GET /api/health HTTP/1.0\r\n\r\n' >&3 && head -1 <&3 | grep -q ' 200 '"]
       interval: 30s
       timeout: 3s
       start_period: 5s

migrations/0014_session_epoch.sql

@@ -0,0 +1,10 @@
+-- Sessions are signed cookies, not rows: nothing on the server lists them, so
+-- nothing can revoke one. Signing a number into every cookie that has to match
+-- this one gives a single lever instead — raise it, and every cookie issued
+-- before is refused. It is raised by `:logout!` and whenever the login
+-- password changes, which is recognised by a keyed hash of it kept here.
+CREATE TABLE auth (
+    id            INTEGER PRIMARY KEY CHECK (id = 1),
+    epoch         INTEGER NOT NULL,
+    password_hash TEXT    NOT NULL
+);

src/auth.rs

@@ -28,14 +28,16 @@ fn sign(secret: &[u8], payload: &str) -> String {
 
 /// A session token is `base64(expiry_unix_seconds).base64(hmac)`. There is no
 /// server-side session store: rotating `WORKSPACE_SECRET` invalidates every session.
-pub fn issue_token(config: &Config) -> String {
+/// A cookie is good until it expires and for as long as the epoch it was
+/// issued under is the current one.
+pub fn issue_token(config: &Config, epoch: i64) -> String {
     let expires = crate::now() + config.session_days * 86_400;
-    let payload = B64.encode(expires.to_string());
+    let payload = B64.encode(format!("{expires}:{epoch}"));
     let signature = sign(&config.secret, &payload);
     format!("{payload}.{signature}")
 }
 
-pub fn verify_token(config: &Config, token: &str) -> bool {
+pub fn verify_token(config: &Config, token: &str, epoch: i64) -> bool {
     let Some((payload, signature)) = token.split_once('.') else {
         return false;
     };
@@ -46,10 +48,61 @@ pub fn verify_token(config: &Config, token: &str) -> bool {
     let Ok(decoded) = B64.decode(payload) else {
         return false;
     };
-    let Ok(expires) = String::from_utf8_lossy(&decoded).parse::<i64>() else {
+    let decoded = String::from_utf8_lossy(&decoded);
+    // A cookie from before epochs existed carries only the expiry, and
+    // belongs to the first epoch.
+    let (expires, issued_in) = decoded.split_once(':').unwrap_or((&decoded, "0"));
+    let (Ok(expires), Ok(issued_in)) = (expires.parse::<i64>(), issued_in.parse::<i64>()) else {
         return false;
     };
-    expires > crate::now()
+    expires > crate::now() && issued_in == epoch
+}
+
+/// The session epoch: what every cookie must have been signed under.
+pub struct Epoch;
+
+impl Epoch {
+    /// A keyed hash of the login password, so a change to it can be told
+    /// from a restart without the password being written down anywhere.
+    fn password_hash(config: &Config) -> String {
+        sign(&config.secret, &format!("password:{}", config.password))
+    }
+
+    /// The epoch to start with: the stored one, raised when the password
+    /// is not the one it was stored with.
+    pub async fn load(db: &sqlx::SqlitePool, config: &Config) -> anyhow::Result<i64> {
+        let hash = Self::password_hash(config);
+        let stored: Option<(i64, String)> =
+            sqlx::query_as("SELECT epoch, password_hash FROM auth WHERE id = 1")
+                .fetch_optional(db)
+                .await?;
+        let epoch = match stored {
+            None => 0,
+            Some((epoch, previous)) if previous == hash => return Ok(epoch),
+            Some((epoch, _)) => {
+                tracing::info!("the login password changed: every device is signed out");
+                epoch + 1
+            }
+        };
+        sqlx::query(
+            "INSERT INTO auth (id, epoch, password_hash) VALUES (1, ?1, ?2)
+             ON CONFLICT (id) DO UPDATE SET epoch = excluded.epoch, password_hash = excluded.password_hash",
+        )
+        .bind(epoch)
+        .bind(&hash)
+        .execute(db)
+        .await?;
+        Ok(epoch)
+    }
+
+    /// Sign every device out: the epoch moves on, and no cookie has it yet.
+    pub async fn advance(state: &AppState) -> anyhow::Result<i64> {
+        let epoch: i64 = sqlx::query_scalar("UPDATE auth SET epoch = epoch + 1 WHERE id = 1 RETURNING epoch")
+            .fetch_one(&state.db)
+            .await?;
+        state.epoch.store(epoch, std::sync::atomic::Ordering::SeqCst);
+        Ok(epoch)
+    }
 }
 
 pub fn credentials_match(config: &Config, username: &str, password: &str) -> bool {
@@ -94,7 +147,8 @@ impl FromRequestParts<AppState> for Session {
         state: &AppState,
     ) -> Result<Self, Self::Rejection> {
         let token = cookie_value(parts, COOKIE_NAME).ok_or(AppError::Unauthorized)?;
-        if verify_token(&state.config, token) {
+        let epoch = state.epoch.load(std::sync::atomic::Ordering::SeqCst);
+        if verify_token(&state.config, token, epoch) {
             Ok(Session)
         } else {
             Err(AppError::Unauthorized)
@@ -199,45 +253,45 @@ mod tests {
     #[test]
     fn a_fresh_token_verifies() {
         let config = config(30);
-        assert!(verify_token(&config, &issue_token(&config)));
+        assert!(verify_token(&config, &issue_token(&config, 0), 0));
     }
 
     #[test]
     fn an_expired_token_does_not() {
         let config = config(-1);
-        assert!(!verify_token(&config, &issue_token(&config)));
+        assert!(!verify_token(&config, &issue_token(&config, 0), 0));
     }
 
     #[test]
     fn rotating_the_secret_invalidates_every_session() {
-        let token = issue_token(&config(30));
+        let token = issue_token(&config(30), 0);
         let mut rotated = config(30);
         rotated.secret = b"a-different-secret-entirely".to_vec();
-        assert!(!verify_token(&rotated, &token));
+        assert!(!verify_token(&rotated, &token, 0));
     }
 
     #[test]
     fn a_tampered_token_does_not_verify() {
         let config = config(30);
-        let token = issue_token(&config);
+        let token = issue_token(&config, 0);
         let (payload, signature) = token.split_once('.').unwrap();
 
         // A far-future expiry the holder signed themselves.
         let forged = base64::Engine::encode(&super::B64, "99999999999");
-        assert!(!verify_token(&config, &format!("{forged}.{signature}")));
+        assert!(!verify_token(&config, &format!("{forged}.{signature}"), 0));
 
         // The real expiry with the last byte of the signature flipped.
         let mut bytes = signature.as_bytes().to_vec();
         bytes[0] = if bytes[0] == b'A' { b'B' } else { b'A' };
         let flipped = String::from_utf8(bytes).unwrap();
-        assert!(!verify_token(&config, &format!("{payload}.{flipped}")));
+        assert!(!verify_token(&config, &format!("{payload}.{flipped}"), 0));
     }
 
     #[test]
     fn malformed_tokens_are_rejected_rather_than_panicking() {
         let config = config(30);
         for token in ["", ".", "no-dot", "!!!.!!!", "a.b.c", &".".repeat(100)] {
-            assert!(!verify_token(&config, token), "{token:?}");
+            assert!(!verify_token(&config, token, 0), "{token:?}");
         }
     }
 
@@ -273,4 +327,19 @@ mod tests {
         // curl, the CLI, a script with its own credentials.
         assert!(is_same_origin(&headers(&[host])));
     }
+
+    #[test]
+    fn a_cookie_from_an_earlier_epoch_is_refused() {
+        let config = config(30);
+        let token = issue_token(&config, 3);
+        assert!(verify_token(&config, &token, 3));
+        assert!(!verify_token(&config, &token, 4));
+
+        // Issued before epochs: only the expiry, and good in the first one.
+        let expires = crate::now() + 3600;
+        let payload = base64::Engine::encode(&super::B64, expires.to_string());
+        let legacy = format!("{payload}.{}", super::sign(&config.secret, &payload));
+        assert!(verify_token(&config, &legacy, 0));
+        assert!(!verify_token(&config, &legacy, 1));
+    }
 }

src/backup/mod.rs

@@ -62,16 +62,34 @@ pub struct Backup {
     /// Held for the duration of a run, so a manual backup and the timer cannot
     /// upload the same snapshot twice at once.
     running: tokio::sync::Mutex<()>,
+    /// Why backups cannot run at all — a session key file that is configured
+    /// but unreadable — when that is so. The server still starts; the mirror
+    /// says why it is off, every time it is asked.
+    disabled: Option<String>,
 }
 
 impl Backup {
     pub fn new(db: SqlitePool, config: Arc<Config>) -> Result<Arc<Self>> {
         let store = SessionStore::open(&config)?;
-        let configured = matches!(store.load(), Ok(Some(_)));
-        Ok(Arc::new(Self {
+        Ok(Self::with_store(db, config, store, None))
+    }
+
+    /// A mirror that is off for `reason`, for a server that should start
+    /// anyway: backups are worth a loud error, not an outage of the workspace.
+    pub fn unavailable(db: SqlitePool, config: Arc<Config>, reason: String) -> Arc<Self> {
+        let store = SessionStore::new(config.proton_session_path());
+        let backup = Self::with_store(db, config, store, Some(reason.clone()));
+        backup.set(|s| s.last_error = Some(reason));
+        backup
+    }
+
+    fn with_store(db: SqlitePool, config: Arc<Config>, store: SessionStore, disabled: Option<String>) -> Arc<Self> {
+        let configured = disabled.is_none() && matches!(store.load(), Ok(Some(_)));
+        Arc::new(Self {
             db,
             config,
             store,
+            disabled,
             wake: Notify::new(),
             // The workspace may have been written to while this process was down,
             // and a restart is the cheapest moment to find out.
@@ -81,11 +99,11 @@ impl Backup {
                 ..Status::default()
             }),
             running: tokio::sync::Mutex::new(()),
-        }))
+        })
     }
 
     pub fn status(&self) -> Status {
-        let mut status = self.status.lock().expect("backup status mutex poisoned").clone();
+        let mut status = self.status.lock().unwrap_or_else(|e| e.into_inner()).clone();
         status.pending = self.dirty.load(Ordering::Relaxed);
         status
     }
@@ -100,7 +118,20 @@ impl Backup {
     /// session has been stored — asking an unconfigured backup to run is a
     /// no-op, not a failure.
     pub async fn run_now(&self) -> Result<Option<Summary>> {
+        if let Some(reason) = &self.disabled {
+            anyhow::bail!("backups are off: {reason}");
+        }
         let _guard = self.running.lock().await;
+        // The same, across processes: `backup-now` beside a running server
+        // would otherwise share the snapshot file and the bookkeeping with it.
+        let _lock = match self.lock_data_dir() {
+            Ok(lock) => lock,
+            Err(e) => {
+                let message = format!("{e:#}");
+                self.set(move |s| s.last_error = Some(message));
+                return Err(e);
+            }
+        };
         self.dirty.store(false, Ordering::Relaxed);
         self.set(|s| {
             s.running = true;
@@ -145,6 +176,23 @@ impl Backup {
         outcome
     }
 
+    fn lock_data_dir(&self) -> Result<std::fs::File> {
+        let path = self.config.data_dir.join(".backup.lock");
+        let file = std::fs::OpenOptions::new()
+            .create(true)
+            .truncate(false)
+            .write(true)
+            .open(&path)
+            .with_context(|| format!("could not open {}", path.display()))?;
+        match file.try_lock() {
+            Ok(()) => Ok(file),
+            Err(std::fs::TryLockError::WouldBlock) => {
+                anyhow::bail!("another backup of this data directory is running — `backup-now` and the server take turns")
+            }
+            Err(std::fs::TryLockError::Error(e)) => Err(e).context("could not lock the data directory for a backup"),
+        }
+    }
+
     async fn mirror(&self) -> Result<Option<Summary>> {
         let Some(client) = proton::resume(&self.store).await? else {
             return Ok(None);
@@ -285,25 +333,39 @@ impl Backup {
     }
 
     fn set(&self, edit: impl FnOnce(&mut Status)) {
-        let mut status = self.status.lock().expect("backup status mutex poisoned");
+        let mut status = self.status.lock().unwrap_or_else(|e| e.into_inner());
         edit(&mut status);
     }
 }
 
-/// Run the mirror in the background for as long as the server lives.
+/// Run the mirror in the background for as long as the server lives. With
+/// no Proton session yet it waits for one: `proton-login` is a separate
+/// process, and the server picks its session up without a restart.
 pub fn spawn(backup: Arc<Backup>) {
     if backup.store.load().ok().flatten().is_none() {
         tracing::info!(
             "Proton Drive backups are not configured — run `narl-workspace proton-login` to enable them"
         );
+    }
+    crate::tasks::supervise("the backup", move || mirror_forever(Arc::clone(&backup)));
+}
+
+async fn mirror_forever(backup: Arc<Backup>) {
+    if backup.disabled.is_some() {
         return;
     }
-
-    tokio::spawn(async move {
     let interval = backup.config.backup.interval;
     let debounce = backup.config.backup.debounce;
 
     loop {
+        if backup.store.load().ok().flatten().is_none() {
+            tokio::select! {
+                _ = backup.wake.notified() => {}
+                _ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {}
+            }
+            continue;
+        }
+
         // Nothing to do: wait for a change, or for the backstop timer.
         while !backup.dirty.load(Ordering::Relaxed) {
             match interval {
@@ -328,7 +390,6 @@ pub fn spawn(backup: Arc<Backup>) {
 
         let _ = backup.run_now().await;
     }
-    });
 }
 
 /// One-shot mirror for `narl-workspace backup-now`, without starting a server.

src/config.rs

@@ -6,7 +6,7 @@ use chrono_tz::Tz;
 
 /// Runtime configuration. Everything comes from the environment (`.env` is
 /// loaded on startup), because this is a deliberately single-user application.
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub struct Config {
     pub username: String,
     pub password: String,
@@ -23,6 +23,61 @@ pub struct Config {
     pub sftp: SftpConfig,
 }
 
+/// Written by hand so that a `{:?}` anywhere — a log line, a panic — never
+/// prints the password, the cookie secret or the relay's credentials.
+impl std::fmt::Debug for Config {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Config")
+            .field("username", &self.username)
+            .field("password", &"<redacted>")
+            .field("secret", &"<redacted>")
+            .field("bind", &self.bind)
+            .field("data_dir", &self.data_dir)
+            .field("session_days", &self.session_days)
+            .field("max_upload_bytes", &self.max_upload_bytes)
+            .field("secure_cookie", &self.secure_cookie)
+            .field("vault_idle", &self.vault_idle)
+            .field("backup", &self.backup)
+            .field("mail", &self.mail)
+            .field("sftp", &self.sftp)
+            .finish()
+    }
+}
+
+/// A whole number from the environment, `default` when it is not set, and an
+/// error at startup — not an overflow later — when it is outside `allowed`.
+fn number(key: &str, default: u64, allowed: std::ops::RangeInclusive<u64>) -> Result<u64> {
+    let Some(raw) = var(key) else {
+        return Ok(default);
+    };
+    let value: u64 = raw
+        .trim()
+        .parse()
+        .with_context(|| format!("WORKSPACE_{key} must be a whole number, not {raw:?}"))?;
+    if !allowed.contains(&value) {
+        anyhow::bail!(
+            "WORKSPACE_{key} must be between {} and {}, not {value}",
+            allowed.start(),
+            allowed.end()
+        );
+    }
+    Ok(value)
+}
+
+/// A switch. Anything that is not clearly on or clearly off is an error:
+/// `WORKSPACE_SECURE_COOKIE=yes` quietly meaning "no" is how a cookie ends up
+/// going out over plain HTTP.
+fn flag(key: &str) -> Result<bool> {
+    let Some(raw) = var(key) else {
+        return Ok(false);
+    };
+    match raw.trim().to_ascii_lowercase().as_str() {
+        "1" | "true" | "yes" | "on" => Ok(true),
+        "0" | "false" | "no" | "off" => Ok(false),
+        _ => anyhow::bail!("WORKSPACE_{key} must be 1 or 0 (true/false, yes/no, on/off), not {raw:?}"),
+    }
+}
+
 /// The SFTP mount. Dormant until `WORKSPACE_SFTP_BIND` is set — same shape as
 /// the mail relay, because a workspace with no key registered yet should not
 /// refuse to start over a feature nobody has turned on.
@@ -51,7 +106,7 @@ pub struct TimeOfDay {
 /// Notification mail. Dormant until `WORKSPACE_SMTP_URL` is set — the same
 /// dormant-until-configured shape the Proton mirror has, because a workspace
 /// that refuses to start over an unconfigured mail relay would be a bad trade.
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub struct MailConfig {
     /// `smtp://host:587` or `smtps://user:pass@host:465`. `None` turns every
     /// notification off, including the queue that would otherwise pile up.
@@ -71,6 +126,22 @@ pub struct MailConfig {
     pub kinds: Vec<String>,
 }
 
+impl std::fmt::Debug for MailConfig {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("MailConfig")
+            // The URL carries the relay's user and password.
+            .field("url", &self.url.as_ref().map(|_| "<redacted>"))
+            .field("from", &self.from)
+            .field("to", &self.to)
+            .field("timezone", &self.timezone)
+            .field("reminder_at", &self.reminder_at)
+            .field("card_due_at", &self.card_due_at)
+            .field("digest_at", &self.digest_at)
+            .field("kinds", &self.kinds)
+            .finish()
+    }
+}
+
 impl MailConfig {
     /// Whether mail is configured at all.
     pub fn enabled(&self) -> bool {
@@ -190,32 +261,16 @@ impl Config {
 
         let data_dir = PathBuf::from(var("DATA_DIR").unwrap_or_else(|| "./data".into()));
 
-        let session_days = var("SESSION_DAYS")
-            .map(|v| v.parse())
-            .transpose()
-            .context("WORKSPACE_SESSION_DAYS must be a number")?
-            .unwrap_or(30);
-
-        let max_upload_bytes = var("MAX_UPLOAD_MB")
-            .map(|v| v.parse::<usize>())
-            .transpose()
-            .context("WORKSPACE_MAX_UPLOAD_MB must be a number")?
-            .unwrap_or(1024)
-            * 1024
-            * 1024;
-
-        let vault_idle = std::time::Duration::from_secs(
-            var("VAULT_IDLE_MIN")
-                .map(|v| v.parse::<u64>())
-                .transpose()
-                .context("WORKSPACE_VAULT_IDLE_MIN must be a number")?
-                .unwrap_or(15)
-                * 60,
-        );
-
-        let secure_cookie = var("SECURE_COOKIE")
-            .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
-            .unwrap_or(false);
+        // Ten years at most: a session is a cookie, and the expiry is
+        // arithmetic on it that must not overflow.
+        let session_days = number("SESSION_DAYS", 30, 1..=3650)? as i64;
+        // Up to a terabyte, in bytes.
+        let max_upload_bytes = usize::try_from(number("MAX_UPLOAD_MB", 1024, 1..=1024 * 1024)? * 1024 * 1024)
+            .context("WORKSPACE_MAX_UPLOAD_MB is more than this machine can address")?;
+        // A vault that locks the moment it opens could not be used; one that
+        // stays open for more than a day is not much of a vault.
+        let vault_idle = Duration::from_secs(number("VAULT_IDLE_MIN", 15, 1..=24 * 60)? * 60);
+        let secure_cookie = flag("SECURE_COOKIE")?;
 
         Ok(Self {
             username,
@@ -311,24 +366,11 @@ impl BackupConfig {
         // second device and a fresh upload of everything in it.
         let device_name = var("PROTON_DEVICE").unwrap_or_else(|| "narl-diary".to_string());
 
-        let minutes = var("BACKUP_INTERVAL_MIN")
-            .map(|v| v.parse::<u64>())
-            .transpose()
-            .context("WORKSPACE_BACKUP_INTERVAL_MIN must be a number of minutes")?
-            .unwrap_or(60);
+        // 0 turns the backstop off; a week is the longest it waits.
+        let minutes = number("BACKUP_INTERVAL_MIN", 60, 0..=7 * 24 * 60)?;
         let interval = (minutes > 0).then(|| Duration::from_secs(minutes * 60));
-
-        let debounce = Duration::from_secs(
-            var("BACKUP_DEBOUNCE_SEC")
-                .map(|v| v.parse::<u64>())
-                .transpose()
-                .context("WORKSPACE_BACKUP_DEBOUNCE_SEC must be a number of seconds")?
-                .unwrap_or(300),
-        );
-
-        let prune = var("BACKUP_PRUNE")
-            .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
-            .unwrap_or(false);
+        let debounce = Duration::from_secs(number("BACKUP_DEBOUNCE_SEC", 300, 0..=24 * 60 * 60)?);
+        let prune = flag("BACKUP_PRUNE")?;
 
         Ok(Self {
             device_name,
@@ -342,9 +384,34 @@ impl BackupConfig {
 
 #[cfg(test)]
 mod tests {
-    use super::{parse_digest, parse_time, TimeOfDay};
+    use super::{flag, number, parse_digest, parse_time, TimeOfDay};
     use chrono::Weekday;
 
+    /// Each case under a name of its own: tests run in parallel and share
+    /// one environment.
+    fn with(key: &str, value: &str) -> String {
+        std::env::set_var(format!("WORKSPACE_{key}"), value);
+        key.to_string()
+    }
+
+    #[test]
+    fn numbers_out_of_bounds_stop_the_start() {
+        assert_eq!(number("TEST_UNSET_NUMBER", 7, 1..=10).unwrap(), 7);
+        assert_eq!(number(&with("TEST_NUMBER_OK", " 9 "), 7, 1..=10).unwrap(), 9);
+        assert!(number(&with("TEST_NUMBER_ZERO", "0"), 7, 1..=10).is_err());
+        assert!(number(&with("TEST_NUMBER_HUGE", "99999999999999999999"), 7, 1..=10).is_err());
+        assert!(number(&with("TEST_NUMBER_NEGATIVE", "-1"), 7, 1..=10).is_err());
+    }
+
+    #[test]
+    fn a_switch_is_on_off_or_an_error() {
+        assert!(!flag("TEST_UNSET_FLAG").unwrap());
+        assert!(flag(&with("TEST_FLAG_YES", "yes")).unwrap());
+        assert!(flag(&with("TEST_FLAG_ONE", "1")).unwrap());
+        assert!(!flag(&with("TEST_FLAG_OFF", "off")).unwrap());
+        assert!(flag(&with("TEST_FLAG_TYPO", "ture")).is_err());
+    }
+
     #[test]
     fn a_time_of_day_is_read_as_written() {
         assert_eq!(

src/main.rs

@@ -11,6 +11,7 @@ mod sftp;
 mod state;
 mod static_files;
 mod sweep;
+mod tasks;
 mod throttle;
 mod vault;
 
@@ -23,7 +24,9 @@ use sqlx::sqlite::{
 };
 use axum::http::{header, HeaderValue};
 use tower_http::{
-    compression::CompressionLayer, set_header::SetResponseHeaderLayer, trace::TraceLayer,
+    compression::CompressionLayer,
+    set_header::SetResponseHeaderLayer,
+    trace::{DefaultOnResponse, TraceLayer},
 };
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
 
@@ -106,7 +109,14 @@ async fn main() -> Result<()> {
     let bind = config.bind;
     let max_upload = config.max_upload_bytes;
     let config = Arc::new(config);
-    let backup = backup::Backup::new(db.clone(), Arc::clone(&config))?;
+    let backup = match backup::Backup::new(db.clone(), Arc::clone(&config)) {
+        Ok(backup) => backup,
+        Err(e) => {
+            let reason = format!("{e:#}");
+            tracing::error!("Proton Drive backups are off: {reason}");
+            backup::Backup::unavailable(db.clone(), Arc::clone(&config), reason)
+        }
+    };
     backup::spawn(Arc::clone(&backup));
 
     let notify = notify::Notifier::new(db.clone(), Arc::clone(&config))?;
@@ -114,6 +124,7 @@ async fn main() -> Result<()> {
 
     sweep::spawn(db.clone(), config.uploads_dir());
 
+    let epoch = auth::Epoch::load(&db, &config).await?;
     let vault = vault::Vault::new(db.clone(), config.vault_idle);
     vault.spawn_idle_lock();
     let state = AppState {
@@ -124,13 +135,20 @@ async fn main() -> Result<()> {
         vault_throttle: Arc::default(),
         backup,
         notify,
+        epoch: Arc::new(std::sync::atomic::AtomicI64::new(epoch)),
     };
 
     if let Some(sftp_bind) = config.sftp.bind {
         let sftp_state = state.clone();
-        tokio::spawn(async move {
-            if let Err(e) = sftp::serve(sftp_state, sftp_bind).await {
-                tracing::error!("sftp server stopped: {e:#}");
+        tasks::supervise("the sftp server", move || {
+            let state = sftp_state.clone();
+            async move {
+                // A port that is taken now may be free in a minute; the web
+                // side carries on either way.
+                while let Err(e) = sftp::serve(state.clone(), sftp_bind).await {
+                    tracing::error!("sftp server stopped: {e:#}; trying again in a minute");
+                    tokio::time::sleep(std::time::Duration::from_secs(60)).await;
+                }
             }
         });
     }
@@ -222,9 +240,21 @@ async fn serve(bind: std::net::SocketAddr, max_upload: usize, state: AppState) -
         .nest("/api", routes::api_router(max_upload))
         .fallback(static_files::handler)
         .layer(axum::middleware::from_fn(auth::same_origin))
-        .layer(CompressionLayer::new())
+        .layer(axum::middleware::from_fn(deadline))
+        .layer(CompressionLayer::new().compress_when(compressible))
         .layer(security_headers())
-        .layer(TraceLayer::new_for_http())
+        .layer(
+            TraceLayer::new_for_http()
+                .make_span_with(|request: &axum::http::Request<axum::body::Body>| {
+                    tracing::info_span!(
+                        "request",
+                        method = %request.method(),
+                        path = %loggable(request.uri().path()),
+                    )
+                })
+                .on_request(())
+                .on_response(DefaultOnResponse::new().level(tracing::Level::INFO)),
+        )
         .with_state(state);
 
     let listener = tokio::net::TcpListener::bind(bind)
@@ -250,9 +280,90 @@ async fn serve(bind: std::net::SocketAddr, max_upload: usize, state: AppState) -
     Ok(())
 }
 
+/// How long an ordinary request may take to be answered. A request that takes
+/// this long is stuck — waiting on a lock it will not get — and is better
+/// answered with an error than left hanging.
+const REQUEST_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30);
+
+/// Requests that take as long as the bytes do: uploads, imports, re-sealing,
+/// a backup run, the export. Only the time to the first byte of a response is
+/// bounded anyway, so a long download of a file is never cut off.
+fn is_slow(method: &axum::http::Method, path: &str) -> bool {
+    use axum::http::Method;
+    match *method {
+        Method::POST => {
+            path == "/api/media"
+                || path == "/api/import"
+                || path == "/api/backup"
+                || path.ends_with("/import")
+                || path.ends_with("/files")
+                || path.ends_with("/encrypt")
+                || path.ends_with("/decrypt")
+        }
+        Method::GET => path == "/api/export",
+        _ => false,
+    }
+}
+
+async fn deadline(request: axum::extract::Request, next: axum::middleware::Next) -> axum::response::Response {
+    use axum::response::IntoResponse;
+    if is_slow(request.method(), request.uri().path()) {
+        return next.run(request).await;
+    }
+    let path = loggable(request.uri().path()).to_string();
+    match tokio::time::timeout(REQUEST_DEADLINE, next.run(request)).await {
+        Ok(response) => response,
+        Err(_) => {
+            tracing::error!("{path} took longer than {REQUEST_DEADLINE:?} and was given up on");
+            (
+                axum::http::StatusCode::SERVICE_UNAVAILABLE,
+                axum::Json(serde_json::json!({ "error": "the server took too long — try again" })),
+            )
+                .into_response()
+        }
+    }
+}
+
 /// Below the ten seconds `docker stop` waits before it sends SIGKILL.
 const SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(8);
 
+/// A request path as it may go into the log: a share link's token and key are
+/// the whole of its secret, and the key is kept out of every log on purpose —
+/// it travels in the URL fragment so that no server ever sees it, except this
+/// one's API, which must not write it down either. Query strings, which carry
+/// search terms, are never logged at all.
+fn loggable(path: &str) -> &str {
+    if path.starts_with("/api/share/") {
+        "/api/share/…"
+    } else if path.starts_with("/s/") {
+        "/s/…"
+    } else {
+        path
+    }
+}
+
+/// Text is worth compressing; uploads are not. Media is already compressed,
+/// and a compressed body no longer matches the byte ranges a video player
+/// asks for, so nothing that answers a range — or could — is touched.
+fn compressible(
+    status: axum::http::StatusCode,
+    _: axum::http::Version,
+    headers: &axum::http::HeaderMap,
+    _: &axum::http::Extensions,
+) -> bool {
+    if status == axum::http::StatusCode::PARTIAL_CONTENT || headers.contains_key(header::ACCEPT_RANGES) {
+        return false;
+    }
+    let kind = headers
+        .get(header::CONTENT_TYPE)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or_default();
+    kind.starts_with("text/")
+        || kind.starts_with("application/json")
+        || kind.starts_with("application/javascript")
+        || kind.starts_with("image/svg+xml")
+}
+
 /// Content-Security-Policy for the app shell. `unsafe-inline` is granted to
 /// styles only, because CodeMirror injects its theme as a <style> element at
 /// runtime; scripts stay confined to the bundled files.

src/notify/mod.rs

@@ -421,9 +421,11 @@ pub fn spawn(notifier: Arc<Notifier>) {
         return;
     }
 
-    tokio::spawn(async move {
-        // A minute is fine-grained enough for a configuration written as HH:MM
-        // and coarse enough to be free.
+    crate::tasks::supervise("the notifier", move || {
+        let notifier = Arc::clone(&notifier);
+        async move {
+            // A minute is fine-grained enough for a configuration written as
+            // HH:MM and coarse enough to be free.
             let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60));
             ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
             loop {
@@ -432,6 +434,7 @@ pub fn spawn(notifier: Arc<Notifier>) {
                     tracing::error!(error = %format!("{e:#}"), "the notification tick failed");
                 }
             }
+        }
     });
 }
 

src/routes/media.rs

@@ -3,7 +3,7 @@ use std::{collections::BTreeSet, sync::LazyLock};
 use axum::{
     body::Body,
     extract::{multipart::Field, Multipart, Path, Query, State},
-    http::header,
+    http::{header, StatusCode},
     response::{IntoResponse, Response},
     Json,
 };
@@ -385,8 +385,45 @@ pub async fn serve(
     _: Session,
     State(state): State<AppState>,
     Path(id): Path<String>,
+    headers: axum::http::HeaderMap,
 ) -> AppResult<Response> {
-    stream_media(&state, &id, true).await
+    stream_media(&state, &id, true, headers.get(header::RANGE)).await
+}
+
+/// The one `bytes=` range of a `Range` header, inclusive, against a file of
+/// `size` bytes. `Ok(None)` serves the whole file: no header, one that cannot
+/// be read, or several ranges — a whole file is always a valid answer.
+/// `Err` is a range that starts past the end.
+fn byte_range(header: Option<&axum::http::HeaderValue>, size: u64) -> Result<Option<(u64, u64)>, ()> {
+    let Some(spec) = header
+        .and_then(|h| h.to_str().ok())
+        .and_then(|h| h.trim().strip_prefix("bytes="))
+    else {
+        return Ok(None);
+    };
+    if spec.contains(',') {
+        return Ok(None);
+    }
+    let Some((from, to)) = spec.split_once('-') else {
+        return Ok(None);
+    };
+    let (from, to) = (from.trim(), to.trim());
+    let range = match (from.parse::<u64>(), to.parse::<u64>()) {
+        // `bytes=-500`: the last 500.
+        (Err(_), Ok(suffix)) if from.is_empty() => {
+            if suffix == 0 || size == 0 {
+                return Err(());
+            }
+            (size.saturating_sub(suffix), size - 1)
+        }
+        (Ok(start), Err(_)) if to.is_empty() => (start, size.saturating_sub(1)),
+        (Ok(start), Ok(end)) if start <= end => (start, end.min(size.saturating_sub(1))),
+        _ => return Ok(None),
+    };
+    if range.0 >= size {
+        return Err(());
+    }
+    Ok(Some(range))
 }
 
 pub async fn remove(
@@ -402,7 +439,12 @@ pub async fn remove(
 /// Shared by the private and the public-share media handlers. A sealed file
 /// is only ever served to the owner, and only while the vault is open — a
 /// share link is a door for strangers, and the vault is not behind it.
-pub async fn stream_media(state: &AppState, id: &str, private: bool) -> AppResult<Response> {
+pub async fn stream_media(
+    state: &AppState,
+    id: &str,
+    private: bool,
+    range: Option<&axum::http::HeaderValue>,
+) -> AppResult<Response> {
     let row = sqlx::query("SELECT mime, filename, size, encrypted FROM media WHERE id = ?1")
         .bind(id)
         .fetch_optional(&state.db)
@@ -423,18 +465,40 @@ pub async fn stream_media(state: &AppState, id: &str, private: bool) -> AppResul
         (true, true) => Some(state.vault.key().ok_or(AppError::Locked)?),
     };
 
-    let file = tokio::fs::File::open(state.config.uploads_dir().join(id))
+    let size = size.max(0) as u64;
+    // A video is seeked by asking for the bytes from a point on; without an
+    // answer to that, seeking fails, and Safari will not play it at all.
+    let (status, start, end) = match byte_range(range, size) {
+        Ok(None) => (StatusCode::OK, 0, size.saturating_sub(1)),
+        Ok(Some((start, end))) => (StatusCode::PARTIAL_CONTENT, start, end),
+        Err(()) => {
+            return Ok((
+                StatusCode::RANGE_NOT_SATISFIABLE,
+                [(header::CONTENT_RANGE, format!("bytes */{size}"))],
+            )
+                .into_response())
+        }
+    };
+    let length = if size == 0 { 0 } else { end - start + 1 };
+
+    let mut file = tokio::fs::File::open(state.config.uploads_dir().join(id))
         .await
         .map_err(|_| AppError::NotFound)?;
     let body = match key {
-        None => Body::from_stream(ReaderStream::new(file)),
-        Some(key) => Body::from_stream(open_stream(file, &key, id).await?),
+        None => {
+            use tokio::io::{AsyncReadExt, AsyncSeekExt};
+            file.seek(std::io::SeekFrom::Start(start)).await?;
+            Body::from_stream(ReaderStream::new(file.take(length)))
+        }
+        Some(key) => Body::from_stream(open_stream(file, &key, id, start, length).await?),
     };
 
-    Ok((
+    let mut response = (
+        status,
         [
             (header::CONTENT_TYPE, mime),
-            (header::CONTENT_LENGTH, size.to_string()),
+            (header::CONTENT_LENGTH, length.to_string()),
+            (header::ACCEPT_RANGES, "bytes".to_string()),
             // An upload never changes under its id, so the browser may keep
             // it — unless it was sealed: the plaintext would then sit in the
             // browser's cache long after the vault had been shut.
@@ -459,34 +523,59 @@ pub async fn stream_media(state: &AppState, id: &str, private: bool) -> AppResul
         ],
         body,
     )
-        .into_response())
+        .into_response();
+    if status == StatusCode::PARTIAL_CONTENT {
+        let value = format!("bytes {start}-{end}/{size}");
+        if let Ok(value) = axum::http::HeaderValue::from_str(&value) {
+            response.headers_mut().insert(header::CONTENT_RANGE, value);
+        }
+    }
+    Ok(response)
 }
 
-/// A sealed file as a stream of its plaintext, a record at a time. The
-/// header is checked before the first byte goes out, so a file that will not
-/// open is an error response rather than a download that breaks off.
+/// A sealed file as a stream of its plaintext, a record at a time, from
+/// byte `start` for `length` bytes. A range starts at the record that holds
+/// its first byte, so seeking into a sealed video costs one record, not the
+/// whole file before it. The header is checked before the first byte goes
+/// out, so a file that will not open is an error response rather than a
+/// download that breaks off.
 async fn open_stream(
     mut file: tokio::fs::File,
     key: &vault::Key,
     id: &str,
+    start: u64,
+    length: u64,
 ) -> AppResult<impl futures_util::Stream<Item = std::io::Result<axum::body::Bytes>>> {
-    use tokio::io::AsyncReadExt;
+    use tokio::io::{AsyncReadExt, AsyncSeekExt};
 
     let len = file.metadata().await?.len();
     let mut header = [0u8; vault::HEADER];
     file.read_exact(&mut header).await?;
-    let opener = vault::Opener::new(key, id, &header, len)?;
+    let mut opener = vault::Opener::new(key, id, &header, len)?;
+    let first = start / vault::CHUNK as u64;
+    if first > 0 {
+        opener.skip_to(first)?;
+        file.seek(std::io::SeekFrom::Start(vault::HEADER as u64 + first * vault::RECORD as u64))
+            .await?;
+    }
+    let skip = (start - first * vault::CHUNK as u64) as usize;
 
-    Ok(futures_util::stream::try_unfold((file, opener), |(mut file, mut opener)| async move {
+    Ok(futures_util::stream::try_unfold(
+        (file, opener, skip, length),
+        |(mut file, mut opener, skip, left)| async move {
             let len = opener.next_len();
-        if len == 0 {
+            if len == 0 || left == 0 {
                 return Ok(None);
             }
             let mut record = vec![0u8; len];
             file.read_exact(&mut record).await?;
             let plain = opener.open(&record).map_err(std::io::Error::other)?;
-        Ok(Some((axum::body::Bytes::from(plain), (file, opener))))
-    }))
+            let piece = &plain[skip.min(plain.len())..];
+            let piece = &piece[..piece.len().min(usize::try_from(left).unwrap_or(usize::MAX))];
+            let left = left - piece.len() as u64;
+            Ok(Some((axum::body::Bytes::copy_from_slice(piece), (file, opener, 0, left))))
+        },
+    ))
 }
 
 /// The id as it is stored and as it names the file on disk, or `None` for
@@ -627,7 +716,22 @@ fn urlencode(value: &str) -> String {
 
 #[cfg(test)]
 mod tests {
-    use super::{canonical_id, sanitize_mime};
+    use super::{byte_range, canonical_id, sanitize_mime};
+
+    #[test]
+    fn ranges_are_read_the_way_browsers_send_them() {
+        let range = |raw: &'static str| byte_range(Some(&axum::http::HeaderValue::from_static(raw)), 1000);
+        assert_eq!(byte_range(None, 1000), Ok(None));
+        assert_eq!(range("bytes=0-"), Ok(Some((0, 999))));
+        assert_eq!(range("bytes=100-199"), Ok(Some((100, 199))));
+        assert_eq!(range("bytes=900-5000"), Ok(Some((900, 999))));
+        assert_eq!(range("bytes=-100"), Ok(Some((900, 999))));
+        assert_eq!(range("bytes=1000-"), Err(()));
+        // Not understood, or several at once: the whole file is the answer.
+        assert_eq!(range("bytes=0-1,5-9"), Ok(None));
+        assert_eq!(range("items=0-1"), Ok(None));
+        assert_eq!(range("bytes=9-1"), Ok(None));
+    }
 
     #[test]
     fn only_a_uuid_is_an_id() {

src/routes/mod.rs

@@ -33,6 +33,7 @@ pub fn api_router(max_upload: usize) -> Router<AppState> {
     // Multipart framing on top of the files themselves.
     let uploads = DefaultBodyLimit::max(max_upload.saturating_add(1024 * 1024));
     Router::new()
+        .route("/health", get(session::health))
         .route("/me", get(session::me))
         .route("/login", post(session::login).layer(DefaultBodyLimit::max(LOGIN_LIMIT)))
         .route("/logout", post(session::logout))

src/routes/session.rs

@@ -1,5 +1,5 @@
 use axum::{
-    extract::State,
+    extract::{Query, State},
     http::{header, HeaderMap, StatusCode},
     response::IntoResponse,
     Json,
@@ -46,7 +46,7 @@ pub async fn login(
         .unwrap_or("unknown");
     state.notify.note_login(user_agent).await;
 
-    let token = auth::issue_token(&state.config);
+    let token = auth::issue_token(&state.config, state.epoch.load(std::sync::atomic::Ordering::SeqCst));
     Ok((
         StatusCode::OK,
         [(
@@ -57,20 +57,48 @@ pub async fn login(
     ))
 }
 
-pub async fn logout(session: Option<Session>, State(state): State<AppState>) -> impl IntoResponse {
+#[derive(Deserialize)]
+pub struct LogoutQuery {
+    /// Every device, not just this one: the session epoch moves on.
+    #[serde(default)]
+    pub everywhere: bool,
+}
+
+pub async fn logout(
+    session: Option<Session>,
+    State(state): State<AppState>,
+    Query(query): Query<LogoutQuery>,
+) -> AppResult<impl IntoResponse> {
     // Leaving is the moment the vault should not outlive — but only the owner
     // leaving: anyone may clear a cookie they do not have, and that must not
     // be a way to shut the vault on the one who does.
     if session.is_some() {
         state.vault.lock();
+        if query.everywhere {
+            auth::Epoch::advance(&state).await?;
+            tracing::info!("signed out everywhere");
         }
-    (
+    }
+    Ok((
         StatusCode::OK,
         [(header::SET_COOKIE, auth::clear_cookie(&state.config))],
         Json(json!({ "ok": true })),
-    )
+    ))
 }
 
 pub async fn me(_: Session, State(state): State<AppState>) -> impl IntoResponse {
     Json(json!({ "user": state.config.username }))
 }
+
+/// `GET /api/health` — for whatever supervises the process. Answers without a
+/// session and says nothing about the workspace, only that the server is up
+/// and its database answers.
+pub async fn health(State(state): State<AppState>) -> impl IntoResponse {
+    match sqlx::query("SELECT 1").execute(&state.db).await {
+        Ok(_) => (StatusCode::OK, Json(json!({ "ok": true }))),
+        Err(e) => {
+            tracing::error!("health check: the database does not answer: {e}");
+            (StatusCode::SERVICE_UNAVAILABLE, Json(json!({ "ok": false })))
+        }
+    }
+}

src/routes/share.rs

@@ -155,6 +155,7 @@ pub async fn read(
 pub async fn serve_media(
     State(state): State<AppState>,
     Path((token, key, id)): Path<(String, String, String)>,
+    headers: axum::http::HeaderMap,
 ) -> AppResult<Response> {
     let stored: Option<String> = sqlx::query(
         "SELECT n.share_key_hash
@@ -171,7 +172,7 @@ pub async fn serve_media(
     if !key_matches(stored.as_deref(), &key) {
         return Err(AppError::NotFound);
     }
-    media::stream_media(&state, &id, false).await
+    media::stream_media(&state, &id, false, headers.get(axum::http::header::RANGE)).await
 }
 
 #[cfg(test)]

src/state.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::sync::{atomic::AtomicI64, Arc};
 
 use sqlx::SqlitePool;
 
@@ -18,4 +18,7 @@ pub struct AppState {
     pub backup: Arc<Backup>,
     /// Notification mail. Always present; dormant until WORKSPACE_SMTP_URL is set.
     pub notify: Arc<Notifier>,
+    /// Signed into every session cookie, which must carry the current value
+    /// to be accepted; see [`crate::auth::Epoch`].
+    pub epoch: Arc<AtomicI64>,
 }

src/sweep.rs

@@ -19,7 +19,14 @@ const UNREFERENCED_DAYS: i64 = 7;
 const EVERY: Duration = Duration::from_secs(24 * 60 * 60);
 
 pub fn spawn(db: SqlitePool, uploads: std::path::PathBuf) {
-    tokio::spawn(async move {
+    crate::tasks::supervise("the uploads sweep", move || {
+        let (db, uploads) = (db.clone(), uploads.clone());
+        sweep_forever(db, uploads)
+    });
+}
+
+async fn sweep_forever(db: SqlitePool, uploads: std::path::PathBuf) {
+    {
         let mut ticker = tokio::time::interval(EVERY);
         loop {
             ticker.tick().await;
@@ -34,7 +41,7 @@ pub fn spawn(db: SqlitePool, uploads: std::path::PathBuf) {
                 Err(e) => tracing::warn!("could not sweep uploads: {e:#}"),
             }
         }
-    });
+    }
 }
 
 #[derive(Debug, Default, PartialEq)]

src/tasks.rs

@@ -0,0 +1,63 @@
+//! Background loops that outlive a panic in one of their passes.
+//!
+//! A task spawned and forgotten dies silently when it panics, and the server
+//! carries on without backups, or mail, with nothing in the log after the
+//! panic itself. Every long-running loop goes through [`supervise`] instead,
+//! which logs the panic and starts the loop again.
+
+use std::{future::Future, time::Duration};
+
+/// Long enough that a loop panicking at once does not spin.
+const RESTART: Duration = Duration::from_secs(10);
+
+/// Run `make()` for as long as the server lives, starting it again after a
+/// panic. A loop that returns has finished on purpose and is not restarted.
+pub fn supervise<F, Fut>(name: &'static str, make: F)
+where
+    F: Fn() -> Fut + Send + 'static,
+    Fut: Future<Output = ()> + Send + 'static,
+{
+    tokio::spawn(async move {
+        loop {
+            match tokio::spawn(make()).await {
+                Ok(()) => return,
+                Err(e) if e.is_panic() => {
+                    let payload = e.into_panic();
+                    let message = payload
+                        .downcast_ref::<&str>()
+                        .map(|s| s.to_string())
+                        .or_else(|| payload.downcast_ref::<String>().cloned())
+                        .unwrap_or_else(|| "(no message)".into());
+                    tracing::error!("{name} panicked: {message}; starting it again in {RESTART:?}");
+                    tokio::time::sleep(RESTART).await;
+                }
+                // Cancelled: the runtime is shutting down.
+                Err(_) => return,
+            }
+        }
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    };
+
+    #[tokio::test(start_paused = true)]
+    async fn a_loop_that_panics_is_started_again() {
+        let runs = Arc::new(AtomicUsize::new(0));
+        let counter = Arc::clone(&runs);
+        super::supervise("test", move || {
+            let counter = Arc::clone(&counter);
+            async move {
+                if counter.fetch_add(1, Ordering::SeqCst) == 0 {
+                    panic!("first pass");
+                }
+            }
+        });
+        tokio::time::sleep(super::RESTART * 2).await;
+        assert_eq!(runs.load(Ordering::SeqCst), 2);
+    }
+}

src/vault.rs

@@ -42,7 +42,7 @@ use std::{
 
 use aes_gcm::{
     aead::{
-        stream::{DecryptorBE32, EncryptorBE32},
+        stream::{EncryptorBE32, NewStream, StreamBE32, StreamPrimitive},
         Aead, KeyInit, Payload,
     },
     Aes256Gcm, Nonce,
@@ -57,7 +57,8 @@ pub const MAGIC: &[u8; 8] = b"NARLVLT1";
 /// small enough to hold one in memory per open download.
 pub const CHUNK: usize = 64 * 1024;
 const TAG: usize = 16;
-const RECORD: usize = CHUNK + TAG;
+/// A record on disk: a chunk of plaintext and its tag.
+pub const RECORD: usize = CHUNK + TAG;
 const WRAP_NONCE: usize = 12;
 const STREAM_NONCE: usize = 7;
 pub const HEADER: usize = MAGIC.len() + WRAP_NONCE + 32 + TAG + STREAM_NONCE;
@@ -185,11 +186,15 @@ fn file_aad(media_id: &str) -> Vec<u8> {
     aad
 }
 
-/// Decrypts one sealed file, a record at a time.
+/// Decrypts one sealed file, a record at a time, from the first or from any
+/// record a range asks to start at.
 pub struct Opener {
-    stream: Option<DecryptorBE32<Aes256Gcm>>,
-    /// Records still to come, the last of them included.
-    remaining: u64,
+    stream: StreamBE32<Aes256Gcm>,
+    /// The record `open` decrypts next. STREAM binds each record to its
+    /// position, so one moved within the file refuses to open.
+    position: u32,
+    /// Records in the file, the last of them included.
+    total: u64,
     /// The size of the last record, which is the only short one.
     last: usize,
 }
@@ -222,35 +227,46 @@ impl Opener {
                 )
                 .map_err(|_| anyhow!("this file was not sealed under this vault, or not as {media_id}"))?,
         );
-        let stream = DecryptorBE32::from_aead(
+        let stream = StreamBE32::from_aead(
             Aes256Gcm::new(file_key.as_slice().into()),
             stream_nonce.into(),
         );
-        Ok(Self { stream: Some(stream), remaining, last })
+        Ok(Self { stream, position: 0, total: remaining, last })
     }
 
     /// How many bytes the next record takes on disk; 0 once all are read.
     pub fn next_len(&self) -> usize {
-        match self.remaining {
+        match self.total.saturating_sub(self.position as u64) {
             0 => 0,
             1 => self.last,
             _ => RECORD,
         }
     }
 
+    /// Start at record `index` instead of the first — at byte
+    /// `HEADER + index * RECORD` of the file — for a read that wants only the
+    /// end of it.
+    pub fn skip_to(&mut self, index: u64) -> Result<()> {
+        if index >= self.total {
+            return Err(anyhow!("past the end of the sealed file"));
+        }
+        self.position = u32::try_from(index).map_err(|_| anyhow!("sealed file is too large"))?;
+        Ok(())
+    }
+
     /// Decrypt the record `next_len` asked for.
     pub fn open(&mut self, record: &[u8]) -> Result<Vec<u8>> {
-        let stream = self.stream.as_mut().ok_or_else(|| anyhow!("sealed file already read"))?;
-        let plain = if self.remaining == 1 {
-            self.stream
-                .take()
-                .expect("present: checked above")
-                .decrypt_last(record)
-        } else {
-            stream.decrypt_next(record)
+        let left = self.total.saturating_sub(self.position as u64);
+        if left == 0 {
+            return Err(anyhow!("sealed file already read"));
         }
+        // Only the record that really is the last may be opened as last, so
+        // a file cut short at a record boundary still refuses to open.
+        let plain = self
+            .stream
+            .decrypt(self.position, left == 1, record)
             .map_err(|_| anyhow!("sealed file is damaged or was tampered with"))?;
-        self.remaining -= 1;
+        self.position += 1;
         Ok(plain)
     }
 }
@@ -587,6 +603,33 @@ mod tests {
         assert!(vault.slot().is_none());
     }
 
+    /// Reading from record `k` on gives exactly the plaintext from
+    /// `k * CHUNK` on, and the last record still has to be the last.
+    #[test]
+    fn a_sealed_file_opens_from_any_record() {
+        let master = key(3);
+        let plain: Vec<u8> = (0..3 * CHUNK + 17).map(|i| (i % 249) as u8).collect();
+        let sealed = seal(&master, "id", &plain, CHUNK);
+        for first in 0..4u64 {
+            let mut opener = Opener::new(&master, "id", &sealed[..HEADER], sealed.len() as u64).unwrap();
+            opener.skip_to(first).unwrap();
+            let mut at = HEADER + first as usize * super::RECORD;
+            let mut out = Vec::new();
+            while opener.next_len() > 0 {
+                let len = opener.next_len();
+                out.extend(opener.open(&sealed[at..at + len]).unwrap());
+                at += len;
+            }
+            assert_eq!(out, plain[first as usize * CHUNK..], "from record {first}");
+        }
+        let mut opener = Opener::new(&master, "id", &sealed[..HEADER], sealed.len() as u64).unwrap();
+        assert!(opener.skip_to(4).is_err());
+        // A middle record presented where the last one should be does not open.
+        opener.skip_to(3).unwrap();
+        let middle = &sealed[HEADER + super::RECORD..HEADER + 2 * super::RECORD];
+        assert!(opener.open(&middle[..opener.next_len()]).is_err());
+    }
+
     #[test]
     fn files_of_every_awkward_size_round_trip() {
         let master = key(7);

web/src/lib/api.ts

@@ -201,7 +201,9 @@ export const api = {
   login: (username: string, password: string) =>
     request<{ user: string }>('/login', { method: 'POST', body: body({ username, password }) }),
 
-  logout: () => request<{ ok: true }>('/logout', { method: 'POST' }),
+  /** `everywhere`: every device's session ends, not only this one. */
+  logout: (everywhere = false) =>
+    request<{ ok: true }>(`/logout${everywhere ? '?everywhere=true' : ''}`, { method: 'POST' }),
 
   listSpaces: () => request<NodeSummary[]>('/spaces'),
 

web/src/lib/commands.ts

@@ -879,6 +879,10 @@ export const commands: CommandSpec[] = [
     group: 'view',
     help: 'end the session',
     run: () => workspace.guard(() => workspace.logout()),
+    bang: {
+      help: 'end every session, on every device — for a lost phone or a shared computer',
+      run: () => workspace.guard(() => workspace.logout(true)),
+    },
   },
 ]
 

web/src/lib/store.svelte.ts

@@ -400,8 +400,8 @@ class Workspace {
     await this.openAgendaIfDue()
   }
 
-  async logout() {
-    await api.logout()
+  async logout(everywhere = false) {
+    await api.logout(everywhere)
     this.user = null
     this.vault = null
     this.spaces = []