From 291d82f21bca2410e3a0046d59e9b76a79a18245 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 26 Sep 2026 17:21:14 -0700 Subject: [PATCH] M5b plan: task 12, instance lock in bxcore, loopd and inferproxy bxcore::instance::lock (File::try_lock, file 0600, missing directory 0700), LockError::line for the two lines the daemons print, and the already-running runbook fragment; loopd locks /run/loopd.lock before its socket and self-test, inferproxy .lock. Checks: given tests fail at ref 11 as the task says (bxcore E0432; loopd 3 and inferproxy 3 fail); the task's quoted edits applied to ref 11 reproduce ref 12 byte for byte; replay gate ok with 1113 tests (1099 + 14), /tmp unchanged; 25 mutations, 24 caught (the uncaught one maps TryLockError::Error to Held, not reachable from a test); given files rustfmt-clean, under 500 lines; no chat-template tags. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/plans/M5b/12-instance-lock.md | 327 ++++++++++++++++++ .../M5b/files/crates/bxcore/src/instance.rs | 70 ++++ .../M5b/files/crates/bxcore/tests/instance.rs | 137 ++++++++ .../files/crates/inferproxy/tests/instance.rs | 228 ++++++++++++ .../M5b/files/crates/loopd/tests/instance.rs | 311 +++++++++++++++++ .../M5b/files/runbook-already-running.md | 47 +++ 6 files changed, 1120 insertions(+) create mode 100644 docs/plans/M5b/12-instance-lock.md create mode 100644 docs/plans/M5b/files/crates/bxcore/src/instance.rs create mode 100644 docs/plans/M5b/files/crates/bxcore/tests/instance.rs create mode 100644 docs/plans/M5b/files/crates/inferproxy/tests/instance.rs create mode 100644 docs/plans/M5b/files/crates/loopd/tests/instance.rs create mode 100644 docs/plans/M5b/files/runbook-already-running.md diff --git a/docs/plans/M5b/12-instance-lock.md b/docs/plans/M5b/12-instance-lock.md new file mode 100644 index 0000000..b054f38 --- /dev/null +++ b/docs/plans/M5b/12-instance-lock.md @@ -0,0 +1,327 @@ +# M5b task 12: a lock file keeps `loopd` and `inferproxy` to one instance each + +**Branch:** `m5b` (run `git switch m5b`; `git status --short` must be empty, otherwise stop) +**Commit subject:** `bxcore, loopd, inferproxy: a lock file keeps each to one instance` + +## Goal + +Every daemon removes a socket it finds at its path before it binds, so a socket left by a daemon +that was killed never blocks the next start. But the removal is unconditional: a second copy of a +daemon, started by mistake, takes the path from the one that runs, which keeps running on a socket +nobody can reach, and nothing says so. From this task on, a daemon refuses to start instead (M5b +spec, section 9a): + +- **Before it touches its sockets**, each daemon takes an exclusive lock on a lock file with the + standard library's `File::try_lock`, and holds it for the life of the process. The system + releases it when the process ends, however it ends, so a killed daemon never leaves a lock + behind. The file itself stays where it is. +- **The lock file** is `/run/.lock` (here `loopd`'s: `/run/loopd.lock`), and + for `inferproxy`, which has no home, `.lock` beside its `--listen` socket. +- **Another instance holds it**: the daemon exits 1 with one line, and touches nothing else: + + ```text + loopd: another loopd is running (it holds /var/lib/boxmaker/run/loopd.lock); see docs/runbook.md#already-running + ``` + +- **Any other error** on the lock file (a `run/` it cannot write, a read-only socket directory): + it exits 1, naming the file and the error: + + ```text + inferproxy: cannot take the lock /run/bx/infer.sock.lock: Permission denied (os error 13); see docs/runbook.md#already-running + ``` + +This task writes the lock itself, `bxcore::instance`, and makes `loopd` and `inferproxy` take it. +Task 13 does the same for `brokerd`, `memoryd` and `gatewayd`. The runbook entry the pointer +names, `already-running`, is given as a fragment (it already covers all five daemons). + +| Name | What it is | +|---|---| +| `bxcore::instance` | new module (given as a skeleton; you fill two functions) | +| `instance::lock` | `pub fn lock(path: &Path) -> Result` | +| `instance::InstanceLock` | the held lock: the open file. Given. | +| `instance::LockError` | `Held`, or `Io(std::io::Error)`, with `Display` and `Error`. Given. | +| `LockError::line` | `pub fn line(&self, role: &str, path: &Path) -> String`: the line the daemon prints | +| `instance::RUNBOOK` | `"see docs/runbook.md#already-running"`. Given. | + +## Files + +- Copy, from `docs/plans/M5b/files/` (the same path under it): + - the skeleton `crates/bxcore/src/instance.rs` (you fill 2 functions in it) + - tests: `crates/bxcore/tests/instance.rs`, `crates/loopd/tests/instance.rs`, + `crates/inferproxy/tests/instance.rs` (all three new) +- Append, from `docs/plans/M5b/files/`, to the end of `docs/runbook.md`, with one blank line + before it: `runbook-already-running.md` (step 3 gives the command). +- Modify: `crates/bxcore/src/lib.rs` (L1, L2), `crates/loopd/src/main.rs` (M1, M2), + `crates/inferproxy/src/main.rs` (P1, P2), `docs/implementer-log.md`. +- Do not touch: any other source file, any `Cargo.toml`, any other test, and in the skeleton + anything but the 2 `todo!()` bodies. + +## Facts you need + +- **`File::try_lock`** (standard library, stable since Rust 1.89; the workspace needs 1.95, so it + is there): `pub fn try_lock(&self) -> Result<(), TryLockError>`. It tries once and never waits. + On Unix it is `flock` with `LOCK_EX | LOCK_NB`. From its documentation: it returns + `Err(TryLockError::WouldBlock)` "if a different lock is already held on this file (via another + handle/descriptor)", and "the lock will be released when this file (along with any other file + descriptors/handles duplicated or inherited from it) is closed". So: + - Two separate opens of the same file are two holders, in one process as in two. The bxcore + tests rely on that: they call `lock` twice in one test. + - The lock lives as long as the `File` does. `InstanceLock` holds the `File`, and each daemon + keeps its `InstanceLock` in a local variable of the function that serves, so it lives until + the process ends. `let _ = lock(..)` or a `match` arm `Ok(_) => {}` would drop it at once, + and the lock with it; the edits bind it to a name, `_instance`. + - Rust opens files with close-on-exec, so a program a daemon starts does not inherit the lock. +- **`std::fs::TryLockError`** (stable since 1.89): `pub enum TryLockError { Error(io::Error), + WouldBlock }`. `Error` never holds an error of kind `WouldBlock`. +- **The modes.** `OpenOptionsExt::mode(0o600)` and `DirBuilderExt::mode(0o700)` apply only when + the file or directory is created, and the process's umask can only remove bits. An existing + lock file is opened as it is, and never truncated (`truncate(false)`): it holds nothing, and a + killed daemon's file must be taken again unchanged. +- **The directory.** `lock` creates the lock file's directory when it is missing, with + `DirBuilder::new().recursive(true).mode(0o700)`, as `brokerd`, `memoryd` and `gatewayd` already + make their socket directories (and so `/run` on a first start). An existing directory is + fine, and an empty parent (a lock file named without a directory) is the current directory, + also fine. A directory that cannot be written is not made writable: the error is + `LockError::Io`, and the daemon stops with the second line of the Goal. +- **Where the lock is taken.** + - `loopd serve` (`run_serve` in `crates/loopd/src/main.rs`): right after the config loads, before + the stale socket is removed, and so before the self-test. A second `loopd` then never sends + the server a prompt either, which would take slot 0's cache from the one that runs. The home + is `cfg.paths.home` (`[paths] home`, or `$BOXMAKER_HOME`). `loopd selftest` takes no lock: it + binds no socket. + - `inferproxy` (`main` in `crates/inferproxy/src/main.rs`): right after the arguments are + parsed, before the stale socket is removed. The lock file is the `--listen` path with `.lock` + added. Since `lock` makes a missing directory, `inferproxy` now also starts when its socket's + directory does not exist yet; before, `bind` failed there. +- **The given tests start the daemons as processes.** Every wait has a deadline and every process + is killed by a guard. Before this task, the tests that expect a second instance to be refused + wait out that deadline (the second copy starts and serves), so the `loopd` file takes about 20 s + to fail and the `inferproxy` file about 10 s. A test that "hangs" for that long is this, not a + fault. The fake inference server in `crates/loopd/tests/support/mod.rs` does not change. +- The calls the bodies and edits make: + + | Call | Signature, and what it gives | + |---|---| + | `DirBuilder` | `DirBuilder::new().recursive(true).mode(0o700).create(dir) -> io::Result<()>` | + | `OpenOptions` | `OpenOptions::new().write(true).create(true).truncate(false).mode(0o600).open(path) -> io::Result` | + | `File::try_lock` | `file.try_lock() -> Result<(), TryLockError>` | + | `Result::map_err` | `.map_err(LockError::Io)?`: an `io::Error` becomes `LockError::Io` | + | `Path::parent` | `path.parent() -> Option<&Path>` | + | `format!`, `Path::display` | `format!("… {} …", path.display())`; `{RUNBOOK}` and `{e}` inline | + | `instance::lock` | `instance::lock(&lock_path) -> Result` | + | `LockError::line` | `e.line("loopd", &lock_path) -> String` | + | `OsString::push` | `let mut lock_path = path.clone().into_os_string(); lock_path.push(".lock");` | + +## The edits, exactly + +Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file. + +### `crates/bxcore/src/lib.rs` + +**L1.** The lines + +```rust +//! Shared data types and the frame codec for Boxmaker. No policy and no I/O beyond frames, except +//! the secret store (`secrets`), behind the feature of that name. +``` + +become + +```rust +//! Shared data types and the frame codec for Boxmaker. No policy and no I/O beyond frames, except +//! the secret store (`secrets`), behind the feature of that name, and the lock that keeps a daemon +//! to one instance (`instance`). +``` + +**L2.** The line + +```rust +pub mod ids; +``` + +becomes + +```rust +pub mod ids; +pub mod instance; +``` + +`instance` is not behind a feature: `loopd` uses it. + +### `crates/loopd/src/main.rs` + +**M1.** The lines + +```rust +use std::time::Duration; + +use loopd::broker_port::{BrokerPort, NoBroker, not_configured_line}; +``` + +become + +```rust +use std::time::Duration; + +use bxcore::instance; +use loopd::broker_port::{BrokerPort, NoBroker, not_configured_line}; +``` + +**M2.** In `run_serve`, the lines + +```rust + let socket = cfg.channel_socket(); + if socket.exists() +``` + +become + +```rust + // One loopd per home (M5b spec, section 9a). The lock comes before the socket is touched and + // before the self-test, so a second copy neither takes the socket of the one that runs nor + // sends its slot a prompt. Held until the process ends. + let lock_path = cfg.paths.home.join("run/loopd.lock"); + let _instance = match instance::lock(&lock_path) { + Ok(held) => held, + Err(e) => { + eprintln!("{}", e.line("loopd", &lock_path)); + return ExitCode::from(1); + } + }; + + let socket = cfg.channel_socket(); + if socket.exists() +``` + +### `crates/inferproxy/src/main.rs` + +**P1.** The line + +```rust +use inferproxy::{Limits, serve}; +``` + +becomes + +```rust +use bxcore::instance; +use inferproxy::{Limits, serve}; +``` + +`inferproxy` depends on `bxcore` already; its `Cargo.toml` does not change. + +**P2.** In `main`, the lines + +```rust + let path: PathBuf = listen.into(); + match fs::remove_file(&path) { +``` + +become + +```rust + let path: PathBuf = listen.into(); + // One inferproxy per socket (M5b spec, section 9a): the lock beside the socket comes before + // the stale socket is removed, so a second copy never takes the socket of the one that runs. + // Held until the process ends; `process::exit` leaves it to the system to release. + let mut lock_path = path.clone().into_os_string(); + lock_path.push(".lock"); + let lock_path = PathBuf::from(lock_path); + let _instance = match instance::lock(&lock_path) { + Ok(held) => held, + Err(e) => { + eprintln!("{}", e.line("inferproxy", &lock_path)); + process::exit(1); + } + }; + match fs::remove_file(&path) { +``` + +## Steps + +- [ ] **1. Copy the given tests.** From the repository root, three commands: + + ```sh + cp docs/plans/M5b/files/crates/bxcore/tests/instance.rs crates/bxcore/tests/ + cp docs/plans/M5b/files/crates/loopd/tests/instance.rs crates/loopd/tests/ + cp docs/plans/M5b/files/crates/inferproxy/tests/instance.rs crates/inferproxy/tests/ + ``` + +- [ ] **2. See them fail.** Three commands, one per file, each with a shell timeout of 600000 ms: + - `cargo test -p bxcore --test instance` does not compile: one `E0432`, unresolved import + `bxcore::instance`. + - `cargo test -p loopd --test instance` compiles, and its 3 tests fail (the run takes about + 20 s): + + | Test | Before this task | + |---|---| + | `a_second_loopd_on_the_same_home_is_refused_and_the_first_keeps_serving` | `the second loopd: …`, `left: None`, `right: Some(1)`: the second copy started and served | + | `after_the_first_is_killed_a_new_loopd_removes_the_stale_socket_and_serves` | `the lock file, mode 0600`, `left: None`, `right: Some(384)` | + | `a_run_directory_that_cannot_be_written_stops_loopd_naming_the_error` | `wanted the line` …: `loopd` ran its self-test and failed later, on the socket directory | + + - `cargo test -p inferproxy --test instance` compiles, and its 3 tests fail (about 10 s): the + same three shapes, `the second inferproxy: …` with `left: None`, `the lock file, beside the + socket` with `left: None`, and `wanted the line` for the read-only directory. +- [ ] **3. The runbook.** From the repository root: + + ```sh + { echo; cat docs/plans/M5b/files/runbook-already-running.md; } >> docs/runbook.md + ``` + + Then `grep -c '^## already-running$' docs/runbook.md` prints `1`. +- [ ] **4. The module.** Copy the skeleton, then make edits L1 and L2: + + ```sh + cp docs/plans/M5b/files/crates/bxcore/src/instance.rs crates/bxcore/src/ + ``` + + `cargo check -p bxcore`: no error, and 5 warnings, all from the unfilled bodies: two lines of + unused imports (`DirBuilder`, `OpenOptions`, `TryLockError`; `DirBuilderExt`, + `OpenOptionsExt`) and three unused variables (`path` twice, `role`). They go as you fill the + bodies. `cargo test -p bxcore --test instance` now compiles: 1 test passes + (`the_error_displays_and_has_a_source`: `Display` and `Error` are given), and 7 fail with + `not yet implemented`. +- [ ] **5. Fill the two functions of `instance.rs`, one at a time, in this order**, each from its + comment, and run `cargo check -p bxcore` after each: `LockError::line`, then `lock`. Change + nothing else in the file. Then `cargo test -p bxcore --test instance`: 8 passed. +- [ ] **6. `loopd`.** Make edits M1 and M2. `cargo check -p loopd`: no error. + `cargo test -p loopd --test instance`: 3 passed. +- [ ] **7. `inferproxy`.** Make edits P1 and P2. `cargo check -p inferproxy`: no error. + `cargo test -p inferproxy --test instance`: 3 passed. +- [ ] **8. Check.** `grep -rn 'runbook.md#already-running' crates/*/src` prints exactly one line, + the `RUNBOOK` constant in `crates/bxcore/src/instance.rs`: both daemons print the pointer + through `line`, never a copy of it. `git diff --stat -- '*Cargo.toml'` prints nothing. +- [ ] **9. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms. + Expected last line: `gate: ok`, with 1113 tests: 1099 before this task, plus 8 in + `crates/bxcore/tests/instance.rs`, 3 in `crates/loopd/tests/instance.rs` and 3 in + `crates/inferproxy/tests/instance.rs`. +- [ ] **10. Log and commit.** `git status --short` lists `crates/bxcore/src/lib.rs`, + `crates/loopd/src/main.rs`, `crates/inferproxy/src/main.rs`, `docs/runbook.md` and your log as + modified, and `crates/bxcore/src/instance.rs` and the three test files as new. Stage with + `git add crates/bxcore crates/loopd crates/inferproxy docs/runbook.md docs/implementer-log.md + Cargo.lock`, then commit as `AGENTS.md` says. `Cargo.lock` does not change in this task; the + `git add` is harmless. + +## Done when + +- `make gate` prints `gate: ok` with 1113 tests. +- `git diff HEAD~1 --stat` lists only `crates/bxcore/src/lib.rs`, `crates/bxcore/src/instance.rs`, + `crates/loopd/src/main.rs`, `crates/inferproxy/src/main.rs`, the three new test files, + `docs/runbook.md` and `docs/implementer-log.md`. + +## Stop and report if + +- A "before" block of L1, L2, M1, M2, P1 or P2 is not in its file exactly as shown, or is there + more than once. +- Step 2, 3 or 4 shows a different result: other errors, other failing tests, or other counts. +- `cargo check` reports an error after a body you filled or an edit you made, and the comment or + the edit does not tell you how to fix it. Quote the error, and stop. +- A comment in the skeleton does not fit the code under it. Quote it and stop. +- After step 5, 6 or 7 a test fails although each function follows its comment and each edit is + as shown. Do not change a test or the skeleton's given code: stop, and give the failing tests' + names and their messages. +- The fix seems to need a change anywhere else: another file of `loopd` or `inferproxy`, the + test support files, a `Cargo.toml`. It does not; stop and say why you think so. +- A test outside these three files fails in the gate: check once whether it fails without your + change, then stop and report both results. +- The gate passes with a count other than 1113. diff --git a/docs/plans/M5b/files/crates/bxcore/src/instance.rs b/docs/plans/M5b/files/crates/bxcore/src/instance.rs new file mode 100644 index 0000000..692cd26 --- /dev/null +++ b/docs/plans/M5b/files/crates/bxcore/src/instance.rs @@ -0,0 +1,70 @@ +//! One instance per daemon (M5b spec, section 9a). Every daemon removes a socket it finds at its +//! path before it binds, so a second copy started by mistake would take the sockets of the one +//! that runs. Before it touches them, a daemon takes an exclusive lock on its lock file with +//! `lock`, and keeps the `InstanceLock` until the process ends. + +use std::fs::{DirBuilder, File, OpenOptions, TryLockError}; +use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt}; +use std::path::Path; + +/// The runbook entry for both ways `lock` can fail. +pub const RUNBOOK: &str = "see docs/runbook.md#already-running"; + +/// A held lock: the open lock file. The system releases the lock when the file is closed, which is +/// when this is dropped or the process ends, however it ends. The file itself stays: removing it +/// would let a second copy lock a new file while the first still runs. +#[derive(Debug)] +#[must_use = "the lock is released when this value is dropped"] +pub struct InstanceLock { + _file: File, +} + +/// Why `lock` did not take the lock. +#[derive(Debug)] +pub enum LockError { + /// Another open file holds the lock: another instance is running. + Held, + /// The lock file, or its directory, could not be created or opened, or the lock failed. + Io(std::io::Error), +} + +impl std::fmt::Display for LockError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + LockError::Held => write!(f, "another instance holds the lock"), + LockError::Io(e) => write!(f, "{e}"), + } + } +} + +impl std::error::Error for LockError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + LockError::Held => None, + LockError::Io(e) => Some(e), + } + } +} + +impl LockError { + /// The line a daemon prints when it does not start: `role` is its name, `path` the lock file. + pub fn line(&self, role: &str, path: &Path) -> String { + // `Held`: "{role}: another {role} is running (it holds {path}); {RUNBOOK}". + // `Io(e)`: "{role}: cannot take the lock {path}: {e}; {RUNBOOK}". + // Each with `format!`, the path shown with `path.display()`. + todo!() + } +} + +/// Take the exclusive lock on `path`, creating the file with mode 0600, and its directory with +/// mode 0700 when it is missing, as the daemons make their socket directories. The lock is tried +/// once, never waited for. +pub fn lock(path: &Path) -> Result { + // 1. If `path.parent()` is `Some(dir)`: `DirBuilder::new().recursive(true).mode(0o700) + // .create(dir)`. 2. `OpenOptions::new().write(true).create(true).truncate(false) + // .mode(0o600).open(path)`. An error of either is `LockError::Io`, through `map_err`. + // 3. `file.try_lock()`: `Ok(())` is `Ok(InstanceLock { _file: file })`, + // `Err(TryLockError::WouldBlock)` is `Err(LockError::Held)`, and + // `Err(TryLockError::Error(e))` is `Err(LockError::Io(e))`. + todo!() +} diff --git a/docs/plans/M5b/files/crates/bxcore/tests/instance.rs b/docs/plans/M5b/files/crates/bxcore/tests/instance.rs new file mode 100644 index 0000000..b9fec52 --- /dev/null +++ b/docs/plans/M5b/files/crates/bxcore/tests/instance.rs @@ -0,0 +1,137 @@ +//! `bxcore::instance`: the lock that keeps a daemon to one instance (M5b spec, section 9a). Two +//! opens of the same file in one process are two holders, as two processes are. Do not edit. + +#[path = "support/tmp.rs"] +mod tmp; + +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; + +use bxcore::instance::{LockError, RUNBOOK, lock}; +use tmp::TempDir; + +/// Gives a directory back its owner's permissions when dropped, so the `TempDir` declared before +/// it can be removed even when an assertion fails (tip T44). +struct Restore(PathBuf); + +impl Drop for Restore { + fn drop(&mut self) { + let _ = std::fs::set_permissions(&self.0, std::fs::Permissions::from_mode(0o700)); + } +} + +fn mode(path: &Path) -> u32 { + std::fs::metadata(path).unwrap().permissions().mode() & 0o777 +} + +#[test] +fn a_second_lock_is_held_while_the_first_lives() { + let dir = TempDir::new("lock-held"); + let path = dir.path().join("run").join("loopd.lock"); + let first = lock(&path).expect("the first lock"); + for _ in 0..3 { + match lock(&path) { + Err(LockError::Held) => {} + other => panic!("a second lock while the first lives: {other:?}"), + } + } + drop(first); +} + +#[test] +fn the_lock_is_free_again_once_the_first_is_dropped() { + let dir = TempDir::new("lock-free"); + let path = dir.path().join("run").join("loopd.lock"); + let first = lock(&path).expect("the first lock"); + drop(first); + let second = lock(&path).expect("the lock after the first was dropped"); + assert!( + matches!(lock(&path), Err(LockError::Held)), + "the second holds it now" + ); + drop(second); + assert!(path.exists(), "the lock file is left in place"); +} + +#[test] +fn a_lock_file_left_behind_is_taken() { + // A killed daemon leaves its lock file, unlocked; the next start takes it and keeps its bytes. + let dir = TempDir::new("lock-left"); + let path = dir.write("run/memoryd.lock", "left\n"); + let held = lock(&path).expect("a lock file nobody holds"); + assert_eq!(std::fs::read_to_string(&path).unwrap(), "left\n"); + drop(held); +} + +#[test] +fn the_lock_file_is_created_with_mode_0600_and_its_directory_0700() { + let dir = TempDir::new("lock-mode"); + let run = dir.path().join("home").join("run"); + let path = run.join("brokerd.lock"); + let held = lock(&path).expect("a lock in a directory that does not exist yet"); + assert_eq!(mode(&path), 0o600, "the lock file"); + assert_eq!(mode(&run), 0o700, "the directory it made"); + assert_eq!(mode(&dir.path().join("home")), 0o700, "and the one above"); + assert_eq!( + std::fs::metadata(&path).unwrap().len(), + 0, + "nothing is written" + ); + drop(held); +} + +#[test] +fn a_directory_that_cannot_be_written_is_an_io_error() { + let dir = TempDir::new("lock-ro"); + let run = dir.path().join("run"); + std::fs::create_dir(&run).unwrap(); + let _restore = Restore(run.clone()); + std::fs::set_permissions(&run, std::fs::Permissions::from_mode(0o500)).unwrap(); + match lock(&run.join("gatewayd.lock")) { + Err(LockError::Io(e)) => { + assert_eq!(e.kind(), std::io::ErrorKind::PermissionDenied, "{e}"); + } + other => panic!("a lock in a read-only directory: {other:?}"), + } + assert!(!run.join("gatewayd.lock").exists()); +} + +#[test] +fn a_path_through_a_file_is_an_io_error() { + let dir = TempDir::new("lock-file"); + let file = dir.write("run", "not a directory\n"); + match lock(&file.join("loopd.lock")) { + Err(LockError::Io(_)) => {} + other => panic!("a lock under a file: {other:?}"), + } +} + +#[test] +fn the_lines_a_daemon_prints() { + let path = Path::new("/home/bx/run/loopd.lock"); + assert_eq!( + LockError::Held.line("loopd", path), + "loopd: another loopd is running (it holds /home/bx/run/loopd.lock); \ + see docs/runbook.md#already-running" + ); + let io = LockError::Io(std::io::Error::other("the disk is on fire")); + assert_eq!( + io.line("inferproxy", Path::new("/run/bx/infer.sock.lock")), + "inferproxy: cannot take the lock /run/bx/infer.sock.lock: the disk is on fire; \ + see docs/runbook.md#already-running" + ); + assert_eq!(RUNBOOK, "see docs/runbook.md#already-running"); +} + +#[test] +fn the_error_displays_and_has_a_source() { + use std::error::Error; + assert_eq!( + LockError::Held.to_string(), + "another instance holds the lock" + ); + assert!(LockError::Held.source().is_none()); + let io = LockError::Io(std::io::Error::other("no room")); + assert_eq!(io.to_string(), "no room"); + assert!(io.source().is_some()); +} diff --git a/docs/plans/M5b/files/crates/inferproxy/tests/instance.rs b/docs/plans/M5b/files/crates/inferproxy/tests/instance.rs new file mode 100644 index 0000000..41e4e5b --- /dev/null +++ b/docs/plans/M5b/files/crates/inferproxy/tests/instance.rs @@ -0,0 +1,228 @@ +//! One `inferproxy` per socket (M5b spec, section 9a): the lock is `.lock`, beside the +//! `--listen` socket. A second `inferproxy` on the same socket is refused, and the first keeps +//! forwarding; once the first is killed, a new one removes the stale socket and forwards. Every +//! wait has a deadline, and every process is killed and every path removed by a guard (tip T44). +//! Do not edit. + +use std::io::{Read, Write}; +use std::net::TcpListener; +use std::os::unix::fs::PermissionsExt; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::thread; +use std::time::{Duration, Instant}; + +const POINTER: &str = "see docs/runbook.md#already-running"; + +static NEXT: AtomicU32 = AtomicU32::new(0); + +/// A temporary directory for the socket, removed when dropped, first with its owner's permissions +/// put back. +struct Dir(PathBuf); + +impl Dir { + fn new() -> Dir { + let n = NEXT.fetch_add(1, Ordering::SeqCst); + let dir = std::env::temp_dir().join(format!("ip-lock-{}-{n}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + Dir(dir) + } + + fn socket(&self) -> PathBuf { + self.0.join("infer.sock") + } +} + +impl Drop for Dir { + fn drop(&mut self) { + let _ = std::fs::set_permissions(&self.0, std::fs::Permissions::from_mode(0o700)); + let _ = std::fs::remove_dir_all(&self.0); + } +} + +/// An upstream that reads four bytes from each connection, answers `pong` and closes. +fn upstream() -> String { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { break }; + thread::spawn(move || { + let mut ping = [0u8; 4]; + if stream.read_exact(&mut ping).is_ok() { + let _ = stream.write_all(b"pong"); + } + }); + } + }); + addr +} + +/// A running `inferproxy`, killed and waited for when dropped. Declared after the `Dir`, so it +/// drops first. +struct Proxy(Child); + +impl Drop for Proxy { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +impl Proxy { + fn running(&mut self) -> bool { + matches!(self.0.try_wait(), Ok(None)) + } +} + +fn command(socket: &Path, upstream: &str) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_inferproxy")); + command + .arg("--listen") + .arg(socket) + .args(["--upstream", upstream]) + .stdin(Stdio::null()) + .stdout(Stdio::null()); + command +} + +fn start(socket: &Path, upstream: &str) -> Proxy { + Proxy( + command(socket, upstream) + .stderr(Stdio::null()) + .spawn() + .unwrap(), + ) +} + +/// Runs `inferproxy` to its end, at most 10 s (then it is killed), and gives its exit code +/// (`None` if it had to be killed) and its stderr. +fn run_to_end(socket: &Path, upstream: &str) -> (Option, String) { + let mut child = Proxy( + command(socket, upstream) + .stderr(Stdio::piped()) + .spawn() + .unwrap(), + ); + let deadline = Instant::now() + Duration::from_secs(10); + let mut code = None; + while Instant::now() < deadline { + if let Ok(Some(status)) = child.0.try_wait() { + code = status.code(); + break; + } + thread::sleep(Duration::from_millis(20)); + } + let _ = child.0.kill(); + let _ = child.0.wait(); + let mut stderr = String::new(); + if let Some(mut pipe) = child.0.stderr.take() { + let _ = pipe.read_to_string(&mut stderr); + } + (code, stderr) +} + +/// True when a `ping` sent to `socket` comes back from the upstream as `pong` within 5 s. +fn forwards(socket: &Path) -> bool { + let Ok(mut stream) = UnixStream::connect(socket) else { + return false; + }; + let limit = Some(Duration::from_secs(5)); + if stream.set_read_timeout(limit).is_err() || stream.set_write_timeout(limit).is_err() { + return false; + } + if stream.write_all(b"ping").is_err() { + return false; + } + let mut answer = Vec::new(); + let _ = stream.read_to_end(&mut answer); + answer == b"pong" +} + +/// Waits up to 10 s for an `inferproxy` to forward on `socket`. +fn wait_until_it_forwards(socket: &Path) -> bool { + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline { + if forwards(socket) { + return true; + } + thread::sleep(Duration::from_millis(50)); + } + false +} + +fn inode(path: &Path) -> u64 { + use std::os::unix::fs::MetadataExt; + std::fs::symlink_metadata(path) + .map(|m| m.ino()) + .unwrap_or(0) +} + +#[test] +fn a_second_inferproxy_on_the_same_socket_is_refused_and_the_first_keeps_forwarding() { + let dir = Dir::new(); + let socket = dir.socket(); + let upstream = upstream(); + let mut first = start(&socket, &upstream); + assert!(wait_until_it_forwards(&socket), "the first never forwarded"); + let before = inode(&socket); + + let (code, stderr) = run_to_end(&socket, &upstream); + let line = format!( + "inferproxy: another inferproxy is running (it holds {}.lock); {POINTER}", + socket.display() + ); + assert_eq!(code, Some(1), "the second inferproxy: {stderr}"); + assert!( + stderr.lines().any(|l| l == line), + "wanted the line\n{line}\nstderr: {stderr}" + ); + + assert!(first.running(), "the first inferproxy is still running"); + assert_eq!(inode(&socket), before, "the first's socket is still there"); + assert!(forwards(&socket), "the first still forwards"); +} + +#[test] +fn after_the_first_is_killed_a_new_inferproxy_removes_the_stale_socket_and_forwards() { + let dir = Dir::new(); + let socket = dir.socket(); + let upstream = upstream(); + let mut first = start(&socket, &upstream); + assert!(wait_until_it_forwards(&socket), "the first never forwarded"); + let _ = first.0.kill(); + let _ = first.0.wait(); + assert!(socket.exists(), "a killed inferproxy leaves its socket"); + assert!(!forwards(&socket), "nobody serves the stale socket"); + + let _next = start(&socket, &upstream); + assert!( + wait_until_it_forwards(&socket), + "the new one never forwarded" + ); + let lock = dir.0.join("infer.sock.lock"); + let mode = std::fs::metadata(&lock).map(|m| m.permissions().mode() & 0o777); + assert_eq!(mode.ok(), Some(0o600), "the lock file, beside the socket"); +} + +#[test] +fn a_socket_directory_that_cannot_be_written_stops_inferproxy_naming_the_error() { + let dir = Dir::new(); + let socket = dir.socket(); + std::fs::set_permissions(&dir.0, std::fs::Permissions::from_mode(0o500)).unwrap(); + + let (code, stderr) = run_to_end(&socket, &upstream()); + let line = format!( + "inferproxy: cannot take the lock {}.lock: Permission denied (os error 13); {POINTER}", + socket.display() + ); + assert_eq!(code, Some(1), "{stderr}"); + assert!( + stderr.lines().any(|l| l == line), + "wanted the line\n{line}\nstderr: {stderr}" + ); + assert!(!socket.exists(), "no socket"); +} diff --git a/docs/plans/M5b/files/crates/loopd/tests/instance.rs b/docs/plans/M5b/files/crates/loopd/tests/instance.rs new file mode 100644 index 0000000..73ebaf0 --- /dev/null +++ b/docs/plans/M5b/files/crates/loopd/tests/instance.rs @@ -0,0 +1,311 @@ +//! One `loopd` per home (M5b spec, section 9a): a second `loopd serve` against the same home is +//! refused before it touches the socket or the server, and the first keeps serving; once the first +//! is killed, a new one removes the stale socket and serves. Every wait has a deadline, and every +//! process is killed and every path removed by a guard (tip T44). Do not edit. + +mod support; + +use std::os::unix::fs::PermissionsExt; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +use bxcore::{ + Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, WireError, read_frame, + write_frame, +}; +use support::{FakeServer, Home, Reply, tokenizer}; + +const POINTER: &str = "see docs/runbook.md#already-running"; + +fn config_file(home: &Home, server: &FakeServer) -> PathBuf { + let text = format!( + r#" +[infer] +socket = "{}" +model = "test-model" +[slots] +main = 0 +background = 1 +[expect] +template_sha256 = "f55f52930aa8bf44ab5cb85f99370fcc3c56e9a85640b812086d5330bce5d86b" +n_ctx = 131072 +slots = 2 +[limits] +poll_ms = 40 +liveness_ms = 500 +retry_backoff_ms = [10] +[paths] +home = "{}" +"#, + server.socket.display(), + home.dir.display() + ); + let path = home.dir.join("config.toml"); + std::fs::write(&path, text).unwrap(); + path +} + +/// A server that passes the self-test twice: once for the first `loopd`, once for the one started +/// after it is killed. +fn healthy_routes(server: &FakeServer) { + server.route("/props", vec![Reply::fixture("props")]); + server.respond("/tokenize", tokenizer); + server.route( + "/v1/chat/completions", + vec![ + Reply::fixture("tool_call"), + Reply::fixture("turn1"), + Reply::fixture("turn2"), + Reply::fixture("tool_call"), + Reply::fixture("turn1"), + Reply::fixture("turn2"), + ], + ); +} + +fn socket(home: &Home) -> PathBuf { + home.dir.join("run").join("loop").join("loop.sock") +} + +/// A running `loopd serve`, killed and waited for when dropped. Declared after the `Home` and the +/// `FakeServer`, so it drops first. +struct Daemon(Child); + +impl Drop for Daemon { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +impl Daemon { + fn running(&mut self) -> bool { + matches!(self.0.try_wait(), Ok(None)) + } +} + +/// `loopd serve` with its stderr in `/.err`, for the messages of a failing test. +fn start(home: &Home, config: &Path, name: &str) -> Daemon { + let err = std::fs::File::create(home.dir.join(format!("{name}.err"))).unwrap(); + let child = Command::new(env!("CARGO_BIN_EXE_loopd")) + .args(["serve", "--config"]) + .arg(config) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::from(err)) + .spawn() + .unwrap(); + Daemon(child) +} + +fn stderr_of(home: &Home, name: &str) -> String { + std::fs::read_to_string(home.dir.join(format!("{name}.err"))).unwrap_or_default() +} + +/// Runs `loopd serve` to its end, at most 20 s (then it is killed), and gives its exit code +/// (`None` if it had to be killed) and its stderr. +fn run_to_end(config: &Path) -> (Option, String) { + let mut child = Daemon( + Command::new(env!("CARGO_BIN_EXE_loopd")) + .args(["serve", "--config"]) + .arg(config) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(), + ); + let deadline = Instant::now() + Duration::from_secs(20); + let mut code = None; + while Instant::now() < deadline { + if let Ok(Some(status)) = child.0.try_wait() { + code = status.code(); + break; + } + std::thread::sleep(Duration::from_millis(20)); + } + let _ = child.0.kill(); + let _ = child.0.wait(); + let mut stderr = String::new(); + if let Some(mut pipe) = child.0.stderr.take() { + let _ = std::io::Read::read_to_string(&mut pipe, &mut stderr); + } + (code, stderr) +} + +/// True when `loopd` answers on `socket`: a turn that resumes a session that does not exist gets +/// the error frame for it. Any other answer, or none within 5 s, is false. +fn answers(socket: &Path) -> bool { + let Ok(mut stream) = UnixStream::connect(socket) else { + return false; + }; + let limit = Some(Duration::from_secs(5)); + if stream.set_read_timeout(limit).is_err() || stream.set_write_timeout(limit).is_err() { + return false; + } + let turn = Envelope { + v: PROTOCOL_VERSION, + id: 7, + r#final: true, + msg: Message::Turn(Turn { + session: SessionId::new("probe").unwrap(), + content: "are you there".to_string(), + resume: true, + }), + }; + if write_frame(&mut stream, &turn).is_err() { + return false; + } + matches!( + read_frame(&mut stream), + Ok(Envelope { + msg: Message::Error(WireError { + code: ErrorCode::NoSuchSession, + .. + }), + .. + }) + ) +} + +/// Waits up to 20 s for `loopd` to answer on `socket`. +fn wait_until_it_answers(socket: &Path) -> bool { + let deadline = Instant::now() + Duration::from_secs(20); + while Instant::now() < deadline { + if answers(socket) { + return true; + } + std::thread::sleep(Duration::from_millis(50)); + } + false +} + +fn inode(path: &Path) -> u64 { + use std::os::unix::fs::MetadataExt; + std::fs::symlink_metadata(path) + .map(|m| m.ino()) + .unwrap_or(0) +} + +#[test] +fn a_second_loopd_on_the_same_home_is_refused_and_the_first_keeps_serving() { + let home = Home::new(); + let server = FakeServer::start(); + healthy_routes(&server); + let config = config_file(&home, &server); + let socket = socket(&home); + let mut first = start(&home, &config, "first"); + assert!( + wait_until_it_answers(&socket), + "the first loopd never answered; stderr: {}", + stderr_of(&home, "first") + ); + let before = inode(&socket); + let completions = server.requests_to("/v1/chat/completions").len(); + let tokenized = server.requests_to("/tokenize").len(); + + let (code, stderr) = run_to_end(&config); + let lock = home.dir.join("run").join("loopd.lock"); + let line = format!( + "loopd: another loopd is running (it holds {}); {POINTER}", + lock.display() + ); + assert_eq!(code, Some(1), "the second loopd: {stderr}"); + assert!( + stderr.lines().any(|l| l == line), + "wanted the line\n{line}\nstderr: {stderr}" + ); + assert!( + !stderr.contains("selftest"), + "the second ran no self-test: {stderr}" + ); + assert_eq!( + ( + server.requests_to("/v1/chat/completions").len(), + server.requests_to("/tokenize").len() + ), + (completions, tokenized), + "the second sent the server nothing" + ); + + assert!(first.running(), "the first loopd is still running"); + assert_eq!(inode(&socket), before, "the first's socket is still there"); + assert!( + answers(&socket), + "the first loopd still answers; its stderr: {}", + stderr_of(&home, "first") + ); +} + +#[test] +fn after_the_first_is_killed_a_new_loopd_removes_the_stale_socket_and_serves() { + let home = Home::new(); + let server = FakeServer::start(); + healthy_routes(&server); + let config = config_file(&home, &server); + let socket = socket(&home); + let mut first = start(&home, &config, "first"); + assert!( + wait_until_it_answers(&socket), + "the first loopd never answered; stderr: {}", + stderr_of(&home, "first") + ); + let _ = first.0.kill(); + let _ = first.0.wait(); + assert!(socket.exists(), "a killed loopd leaves its socket"); + assert!(!answers(&socket), "nobody serves the stale socket"); + + let _next = start(&home, &config, "next"); + assert!( + wait_until_it_answers(&socket), + "the new loopd never answered; stderr: {}", + stderr_of(&home, "next") + ); + let lock = home.dir.join("run").join("loopd.lock"); + let mode = std::fs::metadata(&lock).map(|m| m.permissions().mode() & 0o777); + assert_eq!(mode.ok(), Some(0o600), "the lock file, mode 0600"); + assert_eq!( + server.requests_to("/v1/chat/completions").len(), + 6, + "each of the two ran the self-test" + ); +} + +/// Gives `/run` back its owner's permissions when dropped, before the `Home` is removed. +struct Restore(PathBuf); + +impl Drop for Restore { + fn drop(&mut self) { + let _ = std::fs::set_permissions(&self.0, std::fs::Permissions::from_mode(0o700)); + } +} + +#[test] +fn a_run_directory_that_cannot_be_written_stops_loopd_naming_the_error() { + let home = Home::new(); + let server = FakeServer::start(); + healthy_routes(&server); + let config = config_file(&home, &server); + let run = home.dir.join("run"); + std::fs::create_dir(&run).unwrap(); + let _restore = Restore(run.clone()); + std::fs::set_permissions(&run, std::fs::Permissions::from_mode(0o500)).unwrap(); + + let (code, stderr) = run_to_end(&config); + let line = format!( + "loopd: cannot take the lock {}: Permission denied (os error 13); {POINTER}", + run.join("loopd.lock").display() + ); + assert_eq!(code, Some(1), "{stderr}"); + assert!( + stderr.lines().any(|l| l == line), + "wanted the line\n{line}\nstderr: {stderr}" + ); + assert!( + server.requests().is_empty(), + "loopd sent the server nothing" + ); + assert!(!socket(&home).exists(), "no socket"); +} diff --git a/docs/plans/M5b/files/runbook-already-running.md b/docs/plans/M5b/files/runbook-already-running.md new file mode 100644 index 0000000..68dd720 --- /dev/null +++ b/docs/plans/M5b/files/runbook-already-running.md @@ -0,0 +1,47 @@ +## already-running + +**What you see.** A daemon exits 1 at start, before it touches its sockets, with one of these +lines: + +- `: another is running (it holds ); see docs/runbook.md#already-running` +- `: cannot take the lock : ; see docs/runbook.md#already-running` + +`` is `loopd`, `brokerd`, `memoryd`, `gatewayd` or `inferproxy`. The lock file is +`$BOXMAKER_HOME/run/.lock`; for `inferproxy` it is its `--listen` socket's path with +`.lock` added (`infer.sock.lock`, beside `infer.sock`). + +**Why.** Every daemon removes a socket it finds at its path before it binds, so that a socket +left by a daemon that was killed never blocks the next start. A second copy started by mistake +would therefore take the sockets of the one that runs, which would keep running, unreachable, +and nothing would say so. So each daemon first takes an exclusive lock on its lock file, and +holds it for as long as it runs (M5b spec, section 9a). The system releases the lock when the +process ends, however it ends, so a killed daemon never leaves a lock behind. The file stays in +place and holds nothing. + +**Confirm.** For `another is running`, find the copy that runs, with its command line +(the `--config` path says which home it serves): + +```sh +pgrep -a -x +ls -l "$BOXMAKER_HOME/run/.lock" +``` + +For `cannot take the lock`, the error says why: `Permission denied` means the daemon's user +cannot create or open the file, or create `run/` (`ls -ld "$BOXMAKER_HOME" "$BOXMAKER_HOME/run"` +and `ls -l` on the lock file show who owns them); `Not a directory` means a path above it is a +file; `Read-only file system` means what it says. + +**Fix.** Two copies of a role against the same home are always a mistake: use the one that runs. +To restart it, stop it (`pkill -x `, or Ctrl-C in its terminal), wait until `pgrep -x +` prints nothing, and start it again. Two copies with different homes are fine (each has +its own lock), as long as their configs do not name the same socket path. For a lock that cannot +be taken, give the daemon's user the directory (`chown`), or point its home somewhere it can +write. + +Never delete a lock file while its daemon runs: the lock is held on the open file, so a new copy +would create a new file, lock that, and take the sockets. Deleting it while the daemon is stopped +does no harm, and does not help either. + +**Check.** `pgrep -x ` prints one process id, and the daemon prints its start line +(`loopd: serving on …`, `inferproxy: -> `, `brokerd: serving tools on …`, +`memoryd: serving`, `gatewayd: connected to …`).