From 5f48ba65b0eeb26a914b2b8c2fcc293feec57650 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 26 Sep 2026 17:21:14 -0700 Subject: [PATCH] M5b plan: task 13, instance lock in brokerd, memoryd and gatewayd Each locks /run/.lock in main (gatewayd after its secret, so nothing is made before the secret is read). brokerd's old second-instance test in serve.rs expected the audit lock's message, so serve.rs has a new version; task 01 now copies serve.rs-task01. Checks: given tests fail at ref 12 as the task says (2 each in brokerd, memoryd, gatewayd; serve.rs passes); the task's quoted edits applied to ref 12 reproduce ref 13's tree exactly; replay gate ok with 1118 tests (1113 - 1 + 6), /tmp unchanged; 14 mutations, all caught; the six instance test files stressed 25 runs under 16 busy loops and 20 under 32, no failure. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/plans/M5b/01-invalid-arguments-detail.md | 7 +- docs/plans/M5b/13-instance-lock-rest.md | 280 ++++++++++++ .../files/crates/brokerd/tests/instance.rs | 218 ++++++++++ .../M5b/files/crates/brokerd/tests/serve.rs | 18 - .../crates/brokerd/tests/serve.rs-task01 | 411 ++++++++++++++++++ .../files/crates/gatewayd/tests/instance.rs | 218 ++++++++++ .../files/crates/memoryd/tests/instance.rs | 223 ++++++++++ 7 files changed, 1354 insertions(+), 21 deletions(-) create mode 100644 docs/plans/M5b/13-instance-lock-rest.md create mode 100644 docs/plans/M5b/files/crates/brokerd/tests/instance.rs create mode 100644 docs/plans/M5b/files/crates/brokerd/tests/serve.rs-task01 create mode 100644 docs/plans/M5b/files/crates/gatewayd/tests/instance.rs create mode 100644 docs/plans/M5b/files/crates/memoryd/tests/instance.rs diff --git a/docs/plans/M5b/01-invalid-arguments-detail.md b/docs/plans/M5b/01-invalid-arguments-detail.md index 6ed6a80..5d3be08 100644 --- a/docs/plans/M5b/01-invalid-arguments-detail.md +++ b/docs/plans/M5b/01-invalid-arguments-detail.md @@ -53,7 +53,8 @@ This is the complete list. Each one is covered by an edit below or by a given fi - **new versions** of: `crates/bxcore/tests/wire.rs`, `crates/bxcore/tests/strict.rs`; `crates/brokerd/tests/admin.rs`, `broker.rs`, `broker_pending.rs`, `courier.rs`, `ledger.rs`, `serve.rs`; `crates/loopd/tests/broker_port.rs`, `tools.rs`, `turn_broker.rs`. Your - `broker.rs` is the given `broker.rs-task01`: the given `broker.rs` is task 09's. + `broker.rs` is the given `broker.rs-task01`: the given `broker.rs` is task 09's. Your + `serve.rs` is the given `serve.rs-task01`: the given `serve.rs` is task 13's. - Modify: `crates/bxcore/src/wire.rs`, `crates/brokerd/src/args.rs`, `crates/brokerd/src/policy/mod.rs`, `crates/brokerd/src/ledger/mod.rs`, `crates/brokerd/src/broker.rs`, `crates/loopd/src/tools.rs`, `crates/loopd/src/turn.rs`, @@ -498,8 +499,8 @@ The event keeps only the reason: the owner's texts do not change. crates/brokerd/tests/ cp docs/plans/M5b/files/crates/brokerd/tests/ledger.rs \ crates/brokerd/tests/ - cp docs/plans/M5b/files/crates/brokerd/tests/serve.rs \ - crates/brokerd/tests/ + cp docs/plans/M5b/files/crates/brokerd/tests/serve.rs-task01 \ + crates/brokerd/tests/serve.rs cp docs/plans/M5b/files/crates/loopd/tests/broker_port.rs \ crates/loopd/tests/ cp docs/plans/M5b/files/crates/loopd/tests/tools.rs \ diff --git a/docs/plans/M5b/13-instance-lock-rest.md b/docs/plans/M5b/13-instance-lock-rest.md new file mode 100644 index 0000000..a67374c --- /dev/null +++ b/docs/plans/M5b/13-instance-lock-rest.md @@ -0,0 +1,280 @@ +# M5b task 13: `brokerd`, `memoryd` and `gatewayd` take the instance lock too + +**Branch:** `m5b` (run `git switch m5b`; `git status --short` must be empty, otherwise stop) +**Commit subject:** `brokerd, memoryd, gatewayd: a lock file keeps each to one instance` + +## Goal + +Task 12 wrote `bxcore::instance`, the lock that keeps a daemon to one instance (M5b spec, +section 9a), and made `loopd` and `inferproxy` take it. This task makes the other three daemons +take it the same way: before they touch their sockets, each locks `/run/.lock` and +holds the lock for the life of the process. A second copy against the same home exits 1 with one +line and touches nothing else: + +```text +brokerd: another brokerd is running (it holds /var/lib/boxmaker/run/brokerd.lock); see docs/runbook.md#already-running +memoryd: another memoryd is running (it holds /var/lib/boxmaker/run/memoryd.lock); see docs/runbook.md#already-running +gatewayd: another gatewayd is running (it holds /var/lib/boxmaker/run/gatewayd.lock); see docs/runbook.md#already-running +``` + +Any other error on the lock file stops the daemon too, with the line +`: cannot take the lock : ; see docs/runbook.md#already-running`. Both +lines come from `LockError::line`; the runbook entry `already-running` is there since task 12, and +covers all five daemons. + +## Files + +- Copy, from `docs/plans/M5b/files/` (the same path under it): + - new tests: `crates/brokerd/tests/instance.rs`, `crates/memoryd/tests/instance.rs`, + `crates/gatewayd/tests/instance.rs` + - a **new version** of `crates/brokerd/tests/serve.rs` (it replaces the file in the tree). Do + not copy `serve.rs-task01`: it is the older version. +- Modify: `crates/brokerd/src/main.rs` (B1, B2), `crates/memoryd/src/main.rs` (D1, D2), + `crates/gatewayd/src/main.rs` (G1 to G3), `docs/implementer-log.md`. +- Do not touch: `crates/bxcore` (`instance` is done), any other source file, any `Cargo.toml`, + any other test, `docs/runbook.md`. + +## Facts you need + +- **What `bxcore::instance` gives** (task 12; do not change it): + + | Name | What it is | + |---|---| + | `instance::lock` | `pub fn lock(path: &Path) -> Result`: creates the file (0600) and a missing directory (0700), and tries the lock once | + | `InstanceLock` | the held lock; the lock is released when it is dropped or the process ends | + | `LockError` | `Held` (another instance holds it) or `Io(std::io::Error)` | + | `LockError::line` | `e.line("brokerd", &lock_path) -> String`: the whole line to print, pointer included | + +- **Keep the lock alive.** Each daemon binds the `InstanceLock` to a name, `_instance`, in `main` + (or `serve` for `gatewayd`), the function that serves until the process ends. `let _ = …`, or a + `match` arm `Ok(_) => {}`, would drop it, and the lock with it, at once. The edits show the form. +- **Where each daemon takes it**, and why there: + - `brokerd` (`main` in `crates/brokerd/src/main.rs`): right after the config loads, before + `serve::start`, which opens the audit log and binds both sockets. The home is + `cfg.paths.home`. + - `memoryd` (`main` in `crates/memoryd/src/main.rs`): right after the config loads, before + `serve::start`, which creates `memory/` and the index and binds three sockets. The home is + `config.paths.home`. + - `gatewayd` (`serve` in `crates/gatewayd/src/main.rs`): after the secret is loaded, before the + state directory is made and before `run`, which binds `gateway.sock` and then connects to + Mattermost. Not earlier: the test `a_missing_secret_stops_at_start` in + `crates/gatewayd/tests/main.rs` checks that nothing under the home is made before the secret + is read, and the lock creates `/run`. The home is `config.paths.home`. +- **`brokerd`'s audit lock stays.** `brokerd` has held a lock on `/audit/.lock` since M3a, + and a second `brokerd` used to stop there, with `brokerd is already running` and the pointer + `brokerd-already-running`. It stays: it guards the log itself, and its own test + (`a_second_writer_is_refused` in `crates/brokerd/tests/audit.rs`) still passes. But the instance + lock now comes first, so a second `brokerd` stops with the new line instead. That is why + `serve.rs` has a new version: its test `a_second_brokerd_on_the_same_home_refuses_to_start` + expected the old message, and it is gone; `instance.rs` checks the second `brokerd` now. Nothing + else in `serve.rs` changes. +- **The given tests start each daemon as a process**, as `brokerd`'s `serve.rs` does. The + `gatewayd` one runs it against the fake Mattermost of `tests/support/fake_mm.rs`, with its token + in the environment (`GW_TEST_TOKEN`), and checks that it answers on `gateway.sock` by asking it + to post an approval, as `brokerd` would. Every wait has a deadline and every process is killed by + a guard. Before this task, the `memoryd` and `gatewayd` tests that expect a second instance to be + refused wait out their deadline (the second copy starts and serves), so each of those files takes + about 10 s to fail. That is this, not a hang. +- The calls the edits make: `instance::lock` and `LockError::line` from the table above; + `Path::join` for the lock file's path; `eprintln!` and `ExitCode::from(1)` as the lines around + them already do. + +## The edits, exactly + +Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file. + +### `crates/brokerd/src/main.rs` + +**B1.** The line + +```rust +use brokerd::serve::{self, ServeError}; +``` + +becomes + +```rust +use brokerd::serve::{self, ServeError}; +use bxcore::instance; +``` + +**B2.** In `main`, the line + +```rust + // Keep the socket paths for the serving notice, before `cfg` is moved into `start`. +``` + +becomes + +```rust + // One brokerd per home (M5b spec, section 9a). The lock comes before anything is made, the + // audit log's own lock included, so a second copy never takes the sockets of the one that + // runs. Held until the process ends. + let lock_path = cfg.paths.home.join("run/brokerd.lock"); + let _instance = match instance::lock(&lock_path) { + Ok(held) => held, + Err(e) => { + eprintln!("{}", e.line("brokerd", &lock_path)); + return ExitCode::from(1); + } + }; + + // Keep the socket paths for the serving notice, before `cfg` is moved into `start`. +``` + +### `crates/memoryd/src/main.rs` + +**D1.** The line + +```rust +use memoryd::config::Config; +``` + +becomes + +```rust +use bxcore::instance; +use memoryd::config::Config; +``` + +**D2.** In `main`, the line + +```rust + let log: serve::Log = Arc::new(|line: &str| eprintln!("{line}")); +``` + +becomes + +```rust + // One memoryd per home (M5b spec, section 9a). The lock comes before anything under the home + // is made or bound, so a second copy never takes the sockets of the one that runs. Held until + // the process ends. + let lock_path = config.paths.home.join("run/memoryd.lock"); + let _instance = match instance::lock(&lock_path) { + Ok(held) => held, + Err(e) => { + eprintln!("{}", e.line("memoryd", &lock_path)); + return ExitCode::from(1); + } + }; + let log: serve::Log = Arc::new(|line: &str| eprintln!("{line}")); +``` + +### `crates/gatewayd/src/main.rs` + +**G1.** The line + +```rust +use bxcore::secrets; +``` + +becomes + +```rust +use bxcore::{instance, secrets}; +``` + +**G2.** In the comment at the top of `serve`, the line + +```rust + // Then `if let Some(warning) = &loaded.warning { eprintln!("{warning}"); }`. +``` + +becomes + +```rust + // Then `if let Some(warning) = &loaded.warning { eprintln!("{warning}"); }`. + // 3b. The instance lock, `/run/gatewayd.lock` (M5b spec, section 9a): after the secret, + // which is read before anything is made, and before the directory and the socket. +``` + +**G3.** In `serve`, the lines + +```rust + if let Some(warning) = &loaded.warning { + eprintln!("{warning}"); + } +``` + +become + +```rust + if let Some(warning) = &loaded.warning { + eprintln!("{warning}"); + } + let lock_path = config.paths.home.join("run/gatewayd.lock"); + let _instance = match instance::lock(&lock_path) { + Ok(held) => held, + Err(e) => { + eprintln!("{}", e.line("gatewayd", &lock_path)); + return ExitCode::from(1); + } + }; +``` + +## Steps + +- [ ] **1. Copy the given tests.** From the repository root, four commands: + + ```sh + cp docs/plans/M5b/files/crates/brokerd/tests/instance.rs crates/brokerd/tests/ + cp docs/plans/M5b/files/crates/brokerd/tests/serve.rs crates/brokerd/tests/ + cp docs/plans/M5b/files/crates/memoryd/tests/instance.rs crates/memoryd/tests/ + cp docs/plans/M5b/files/crates/gatewayd/tests/instance.rs crates/gatewayd/tests/ + ``` + +- [ ] **2. See them fail.** Three commands, each with a shell timeout of 600000 ms. Each file + compiles; `serve.rs` passes. + - `cargo test -p brokerd --no-fail-fast --test instance --test serve` (`--no-fail-fast`, or + cargo stops at the first failing file): `serve` 8 passed. `instance`: 2 failed. + `a_second_brokerd_on_the_same_home_is_refused_and_the_first_keeps_serving` fails on `the + lock's line, and nothing else` (the second stopped at the audit lock, with `brokerd is already + running`); `after_the_first_is_killed_a_new_brokerd_removes_the_stale_sockets_and_serves` + fails on `the lock file, mode 0600`, `left: None`, `right: Some(384)`. + - `cargo test -p memoryd --test instance`: 2 failed (about 10 s). + `a_second_memoryd_on_the_same_home_is_refused_and_the_first_keeps_serving` fails on `the + second memoryd: …` with `left: None`, `right: Some(1)` (it started and served); + `after_the_first_is_killed_a_new_memoryd_removes_the_stale_sockets_and_serves` on `the lock + file, mode 0600`, `left: None`. + - `cargo test -p gatewayd --test instance`: 2 failed (about 10 s), the same two shapes: + `the second gatewayd: …` with `left: None`, and `the lock file, mode 0600` with `left: None`. +- [ ] **3. `brokerd`.** Make edits B1 and B2. `cargo check -p brokerd`: no error. + `cargo test -p brokerd --test instance --test serve`: `instance` 2 passed, `serve` 8 passed. +- [ ] **4. `memoryd`.** Make edits D1 and D2. `cargo check -p memoryd`: no error. + `cargo test -p memoryd --test instance`: 2 passed. +- [ ] **5. `gatewayd`.** Make edits G1 to G3. `cargo check -p gatewayd`: no error. + `cargo test -p gatewayd --test instance --test main`: `instance` 2 passed, `main` 5 passed. +- [ ] **6. Check.** `grep -rn 'instance::lock(' crates/*/src` prints exactly five lines, one in + the `main.rs` of each of `brokerd`, `gatewayd`, `inferproxy`, `loopd` and `memoryd`. + `grep -rn 'runbook.md#already-running' crates/*/src` still prints exactly one line, in + `crates/bxcore/src/instance.rs`. `git diff --stat -- crates/bxcore '*Cargo.toml'` prints nothing. +- [ ] **7. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms. + Expected last line: `gate: ok`, with 1118 tests: 1113 before this task, minus 1 in the new + `crates/brokerd/tests/serve.rs` (its old second-instance test), plus 2 in each of the three new + `instance.rs` files. +- [ ] **8. Log and commit.** `git status --short` lists `crates/brokerd/src/main.rs`, + `crates/memoryd/src/main.rs`, `crates/gatewayd/src/main.rs`, `crates/brokerd/tests/serve.rs` + and your log as modified, and the three `instance.rs` test files as new. Stage with + `git add crates/brokerd crates/memoryd crates/gatewayd 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 1118 tests. +- `git diff HEAD~1 --stat` lists only the three `main.rs` files, `crates/brokerd/tests/serve.rs`, + the three new `instance.rs` test files and `docs/implementer-log.md`. + +## Stop and report if + +- A "before" block of B1, B2, D1, D2 or G1 to G3 is not in its file exactly as shown, or is there + more than once. +- Step 2 shows a different result: other failing tests, other messages, or other counts. +- `cargo check` reports an error after an edit made as shown. Quote it and stop. +- After step 3, 4 or 5 a test fails although the edits are as shown. Do not change a test or + `bxcore`: stop, and give the failing tests' names and their messages. +- The fix seems to need a change anywhere else: `bxcore`, another file of a daemon, the test + support files, a `Cargo.toml`, the runbook. It does not; stop and say why you think so. +- A test outside these 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 1118. diff --git a/docs/plans/M5b/files/crates/brokerd/tests/instance.rs b/docs/plans/M5b/files/crates/brokerd/tests/instance.rs new file mode 100644 index 0000000..3305dc1 --- /dev/null +++ b/docs/plans/M5b/files/crates/brokerd/tests/instance.rs @@ -0,0 +1,218 @@ +//! One `brokerd` per home (M5b spec, section 9a): a second `brokerd serve` against the same home is +//! refused with the lock's line, and the first keeps serving; once the first is killed, a new one +//! removes the stale sockets and serves. Every wait has a deadline, and every process is killed +//! and every path removed by a guard (tip T44). Do not edit. + +#[path = "support/tmp.rs"] +mod tmp; + +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::{Empty, Envelope, Message, PROTOCOL_VERSION, read_frame, write_frame}; +use tmp::TempDir; + +const POINTER: &str = "see docs/runbook.md#already-running"; + +struct Home { + dir: TempDir, + config: PathBuf, +} + +impl Home { + fn new(tag: &str) -> Home { + let dir = TempDir::new(tag); + std::fs::create_dir_all(dir.path().join("grants")).unwrap(); + let text = format!( + "[paths]\nhome = \"{home}\"\ngrants = \"{home}/grants\"\n", + home = dir.path().display() + ); + let config = dir.write("brokerd.toml", &text); + Home { dir, config } + } + + fn path(&self, relative: &str) -> PathBuf { + self.dir.path().join(relative) + } + + fn command(&self) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_brokerd")); + command + .args(["serve", "--config"]) + .arg(&self.config) + .stdin(Stdio::null()) + .stdout(Stdio::null()); + command + } + + /// `brokerd serve` with its stderr in `/.err`, for the messages of a failing test. + fn start(&self, name: &str) -> Daemon { + let err = std::fs::File::create(self.path(&format!("{name}.err"))).unwrap(); + Daemon(self.command().stderr(Stdio::from(err)).spawn().unwrap()) + } + + fn stderr_of(&self, name: &str) -> String { + std::fs::read_to_string(self.path(&format!("{name}.err"))).unwrap_or_default() + } + + /// Runs `brokerd serve` 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(&self) -> (Option, String) { + let mut child = Daemon(self.command().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; + } + 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) + } + + fn tools(&self) -> PathBuf { + self.path("run/loop-broker/broker.sock") + } + + fn admin(&self) -> PathBuf { + self.path("run/owner-broker/admin.sock") + } +} + +/// A running `brokerd serve`, killed and waited for when dropped. Declared after the `Home`, 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)) + } +} + +/// True when `brokerd` answers on its owner's socket: `approvals` gets the list of approvals. Any +/// other answer, or none within 5 s, is false. +fn answers(admin: &Path) -> bool { + let Ok(mut stream) = UnixStream::connect(admin) 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 ask = Envelope { + v: PROTOCOL_VERSION, + id: 1, + r#final: true, + msg: Message::Approvals(Empty {}), + }; + if write_frame(&mut stream, &ask).is_err() { + return false; + } + matches!( + read_frame(&mut stream), + Ok(Envelope { + msg: Message::ApprovalList(_), + .. + }) + ) +} + +/// Waits up to 10 s for `brokerd` to answer on `admin` and to listen on `tools`. +fn wait_until_it_answers(home: &Home) -> bool { + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline { + if answers(&home.admin()) && UnixStream::connect(home.tools()).is_ok() { + return true; + } + std::thread::sleep(Duration::from_millis(20)); + } + 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_brokerd_on_the_same_home_is_refused_and_the_first_keeps_serving() { + let home = Home::new("inst-twice"); + let mut first = home.start("first"); + assert!( + wait_until_it_answers(&home), + "the first brokerd never answered; stderr: {}", + home.stderr_of("first") + ); + let before = (inode(&home.tools()), inode(&home.admin())); + + let (code, stderr) = home.run_to_end(); + let line = format!( + "brokerd: another brokerd is running (it holds {}); {POINTER}", + home.path("run/brokerd.lock").display() + ); + assert_eq!(code, Some(1), "the second brokerd: {stderr}"); + assert_eq!( + stderr, + format!("{line}\n"), + "the lock's line, and nothing else" + ); + + assert!(first.running(), "the first brokerd is still running"); + assert_eq!( + (inode(&home.tools()), inode(&home.admin())), + before, + "the first's sockets are still there" + ); + assert!( + answers(&home.admin()), + "the first brokerd still answers; its stderr: {}", + home.stderr_of("first") + ); +} + +#[test] +fn after_the_first_is_killed_a_new_brokerd_removes_the_stale_sockets_and_serves() { + let home = Home::new("inst-restart"); + let mut first = home.start("first"); + assert!( + wait_until_it_answers(&home), + "the first brokerd never answered; stderr: {}", + home.stderr_of("first") + ); + let _ = first.0.kill(); + let _ = first.0.wait(); + assert!( + home.tools().exists() && home.admin().exists(), + "a killed brokerd leaves its sockets" + ); + assert!(!answers(&home.admin()), "nobody serves the stale socket"); + + let _next = home.start("next"); + assert!( + wait_until_it_answers(&home), + "the new brokerd never answered; stderr: {}", + home.stderr_of("next") + ); + let lock = home.path("run/brokerd.lock"); + let mode = std::fs::metadata(&lock).map(|m| m.permissions().mode() & 0o777); + assert_eq!(mode.ok(), Some(0o600), "the lock file, mode 0600"); +} diff --git a/docs/plans/M5b/files/crates/brokerd/tests/serve.rs b/docs/plans/M5b/files/crates/brokerd/tests/serve.rs index bcdffac..5ef23ed 100644 --- a/docs/plans/M5b/files/crates/brokerd/tests/serve.rs +++ b/docs/plans/M5b/files/crates/brokerd/tests/serve.rs @@ -216,24 +216,6 @@ fn a_stale_socket_is_replaced() { ); } -#[test] -fn a_second_brokerd_on_the_same_home_refuses_to_start() { - let home = Home::new("serve-twice", 900_000); - let _running = home.serve(); - let second = home.run(&[]); - assert_eq!(second.status.code(), Some(1)); - let text = stderr(&second); - assert!(text.contains("brokerd is already running"), "{text}"); - assert!( - text.trim_end() - .ends_with("see docs/runbook.md#brokerd-already-running"), - "{text}" - ); - // The first still has its sockets. - let frames = exchange(&home.admin(), 1, Message::Approvals(Empty {})); - assert!(matches!(frames[0].msg, Message::ApprovalList(_))); -} - #[test] fn it_answers_each_socket_and_refuses_the_other_kinds() { let home = Home::new("serve-kinds", 900_000); diff --git a/docs/plans/M5b/files/crates/brokerd/tests/serve.rs-task01 b/docs/plans/M5b/files/crates/brokerd/tests/serve.rs-task01 new file mode 100644 index 0000000..bcdffac --- /dev/null +++ b/docs/plans/M5b/files/crates/brokerd/tests/serve.rs-task01 @@ -0,0 +1,411 @@ +//! `brokerd serve` as a process: its startup, both sockets, and the expiry thread. Do not edit. + +#[path = "support/tmp.rs"] +mod tmp; + +use std::os::unix::fs::PermissionsExt; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Output, Stdio}; +use std::time::{Duration, Instant}; + +use brokerd::audit::RECOVERED_NOTICE; +use brokerd::runner::REFUSING; +use bxcore::{ + AuditEvent, CallId, DenyReason, Empty, Envelope, ErrorCode, Message, PROTOCOL_VERSION, + ResultStatus, SessionId, ToolRequest, ToolResponse, +}; +use tmp::TempDir; + +struct Home { + dir: TempDir, + config: PathBuf, +} + +impl Home { + fn new(tag: &str, ttl_ms: u64) -> Home { + let dir = TempDir::new(tag); + std::fs::create_dir_all(dir.path().join("grants")).unwrap(); + let text = format!( + "[paths]\nhome = \"{home}\"\ngrants = \"{home}/grants\"\n[approvals]\nttl_ms = {ttl_ms}\n", + home = dir.path().display() + ); + let config = dir.write("brokerd.toml", &text); + Home { dir, config } + } + + fn path(&self, relative: &str) -> PathBuf { + self.dir.path().join(relative) + } + + fn tools(&self) -> PathBuf { + self.path("run/loop-broker/broker.sock") + } + + fn admin(&self) -> PathBuf { + self.path("run/owner-broker/admin.sock") + } + + fn grant(&self, id: &str, mode: &str) { + let text = format!( + "tool = \"read_file\"\nmode = \"{mode}\"\nmax_taint = \"secret\"\n\ + result_class = \"private\"\nuntrusted = false\n[constraints]\npaths = [\"/n\"]\n" + ); + std::fs::write(self.path(&format!("grants/{id}.toml")), text).unwrap(); + } + + fn command(&self, extra: &[&str]) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_brokerd")); + command + .args(["serve", "--config"]) + .arg(&self.config) + .args(extra); + command + } + + /// Starts `brokerd serve` and waits until both sockets answer. + fn serve(&self) -> Running { + let child = self + .command(&[]) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let running = Running(Some(child)); + let until = Instant::now() + Duration::from_secs(10); + while UnixStream::connect(self.tools()).is_err() + || UnixStream::connect(self.admin()).is_err() + { + assert!(Instant::now() < until, "brokerd never listened"); + std::thread::sleep(Duration::from_millis(20)); + } + running + } + + /// Runs `brokerd serve` expecting it to exit by itself. + fn run(&self, extra: &[&str]) -> Output { + self.command(extra).output().unwrap() + } + + fn events(&self) -> Vec { + let dir = self.path("audit"); + let mut names: Vec = std::fs::read_dir(&dir) + .unwrap() + .map(|e| e.unwrap().file_name().into_string().unwrap()) + .filter(|n| n.ends_with(".jsonl")) + .collect(); + names.sort(); + names + .iter() + .flat_map(|n| { + let text = std::fs::read_to_string(dir.join(n)).unwrap(); + text.lines() + .map(|l| { + serde_json::from_str::(l) + .unwrap() + .event + }) + .collect::>() + }) + .collect() + } +} + +/// Kills the daemon when dropped; `stop` returns what it printed. +struct Running(Option); + +impl Running { + fn stop(mut self) -> String { + let mut child = self.0.take().unwrap(); + child.kill().unwrap(); + let output = child.wait_with_output().unwrap(); + String::from_utf8_lossy(&output.stderr).to_string() + } +} + +impl Drop for Running { + fn drop(&mut self) { + if let Some(child) = &mut self.0 { + let _ = child.kill(); + let _ = child.wait(); + } + } +} + +fn mode(path: &Path) -> u32 { + std::fs::metadata(path).unwrap().permissions().mode() & 0o777 +} + +fn exchange(socket: &Path, id: u64, msg: Message) -> Vec { + let mut stream = UnixStream::connect(socket).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + let env = Envelope { + v: PROTOCOL_VERSION, + id, + r#final: true, + msg, + }; + bxcore::write_frame(&mut stream, &env).unwrap(); + let mut frames = Vec::new(); + loop { + let env = bxcore::read_frame(&mut stream).unwrap(); + let last = env.r#final; + frames.push(env); + if last { + return frames; + } + } +} + +fn read_notes(call: u64) -> Message { + Message::ToolRequest(ToolRequest { + session: SessionId::new("s1").unwrap(), + call: CallId(call), + tool: "read_file".to_string(), + arguments: r#"{"path":"/n/a"}"#.to_string(), + }) +} + +fn last_response(frames: &[Envelope]) -> &ToolResponse { + match &frames.last().unwrap().msg { + Message::ToolResponse(r) => r, + other => panic!("{other:?}"), + } +} + +fn stderr(output: &Output) -> String { + String::from_utf8_lossy(&output.stderr).to_string() +} + +#[test] +fn it_makes_its_directories_0700_and_its_sockets_0600() { + let home = Home::new("serve-modes", 900_000); + // One directory found too open, one made. + std::fs::create_dir_all(home.path("run/owner-broker")).unwrap(); + std::fs::set_permissions( + home.path("run/owner-broker"), + std::fs::Permissions::from_mode(0o755), + ) + .unwrap(); + let running = home.serve(); + assert_eq!(mode(&home.path("run/loop-broker")), 0o700); + assert_eq!(mode(&home.path("run/owner-broker")), 0o700); + assert_eq!(mode(&home.tools()), 0o600); + assert_eq!(mode(&home.admin()), 0o600); + assert_eq!(mode(&home.path("audit")), 0o700); + let printed = running.stop(); + assert!(printed.contains("serving"), "{printed}"); +} + +#[test] +fn a_stale_socket_is_replaced() { + let home = Home::new("serve-stale", 900_000); + std::fs::create_dir_all(home.path("run/loop-broker")).unwrap(); + drop(UnixListener::bind(home.tools()).unwrap()); + assert!(home.tools().exists(), "the stale socket file is there"); + let _running = home.serve(); + let frames = exchange(&home.tools(), 3, read_notes(3)); + assert_eq!( + last_response(&frames), + &ToolResponse::Denied { + reason: DenyReason::NoGrant, + detail: None, + } + ); +} + +#[test] +fn a_second_brokerd_on_the_same_home_refuses_to_start() { + let home = Home::new("serve-twice", 900_000); + let _running = home.serve(); + let second = home.run(&[]); + assert_eq!(second.status.code(), Some(1)); + let text = stderr(&second); + assert!(text.contains("brokerd is already running"), "{text}"); + assert!( + text.trim_end() + .ends_with("see docs/runbook.md#brokerd-already-running"), + "{text}" + ); + // The first still has its sockets. + let frames = exchange(&home.admin(), 1, Message::Approvals(Empty {})); + assert!(matches!(frames[0].msg, Message::ApprovalList(_))); +} + +#[test] +fn it_answers_each_socket_and_refuses_the_other_kinds() { + let home = Home::new("serve-kinds", 900_000); + home.grant("notes", "auto"); + let running = home.serve(); + let frames = exchange(&home.tools(), 7, read_notes(7)); + assert_eq!(frames[0].id, 7); + // The production runtime runs nothing. + assert_eq!( + last_response(&frames), + &ToolResponse::Failed { + message: REFUSING.to_string() + } + ); + let wrong = exchange(&home.tools(), 8, Message::Approvals(Empty {})); + assert!(matches!(&wrong[0].msg, Message::Error(e) if e.code == ErrorCode::Forbidden)); + let wrong = exchange(&home.admin(), 9, read_notes(9)); + assert!(matches!(&wrong[0].msg, Message::Error(e) if e.code == ErrorCode::Forbidden)); + let printed = running.stop(); + assert!( + printed.contains("approvals on broker.sock\nsee docs/runbook.md#socket-forbidden"), + "{printed}" + ); + assert!( + printed.contains("tool_request on admin.sock\nsee docs/runbook.md#socket-forbidden"), + "{printed}" + ); + assert!(matches!( + home.events().as_slice(), + [ + AuditEvent::Decision { .. }, + AuditEvent::Result { + status: ResultStatus::Failed, + .. + } + ] + )); +} + +#[test] +fn an_approval_nobody_answers_expires() { + let home = Home::new("serve-expire", 100); + home.grant("notes", "ask"); + let _running = home.serve(); + let started = Instant::now(); + let frames = exchange(&home.tools(), 2, read_notes(2)); + assert_eq!(frames.len(), 2, "{frames:?}"); + assert!(matches!( + &frames[0].msg, + Message::ToolResponse(ToolResponse::PendingApproval { approval: 0, .. }) + )); + assert_eq!( + last_response(&frames), + &ToolResponse::Denied { + reason: DenyReason::ApprovalExpired, + detail: None, + } + ); + // The expiry thread looks every second. + assert!( + started.elapsed() < Duration::from_secs(5), + "{:?}", + started.elapsed() + ); +} + +fn copy_case(home: &Home, case: &str, only: &[&str]) { + let from = format!( + "{}/../bxcore/tests/fixtures/audit/{case}", + env!("CARGO_MANIFEST_DIR") + ); + std::fs::create_dir_all(home.path("audit")).unwrap(); + for name in only { + std::fs::copy( + format!("{from}/{name}"), + home.path(&format!("audit/{name}")), + ) + .unwrap(); + } +} + +fn snapshot(dir: &Path) -> Vec<(String, Vec)> { + let mut all: Vec<(String, Vec)> = std::fs::read_dir(dir) + .unwrap() + .map(|e| e.unwrap()) + .filter(|e| e.file_name().to_string_lossy().ends_with(".jsonl")) + .map(|e| { + ( + e.file_name().into_string().unwrap(), + std::fs::read(e.path()).unwrap(), + ) + }) + .collect(); + all.sort(); + all +} + +#[test] +fn a_broken_chain_stops_it_before_any_socket_and_nothing_is_written() { + let home = Home::new("serve-broken", 900_000); + copy_case(&home, "changed-byte", &["2026-09-17.jsonl"]); + let before = snapshot(&home.path("audit")); + let output = home.run(&[]); + assert_eq!(output.status.code(), Some(1)); + let text = stderr(&output); + assert!(text.contains("2026-09-17.jsonl:4: "), "{text}"); + assert!( + text.trim_end() + .ends_with("see docs/runbook.md#audit-chain-broken"), + "{text}" + ); + assert_eq!(snapshot(&home.path("audit")), before); + assert!(!home.tools().exists() && !home.admin().exists()); +} + +#[test] +fn a_torn_tail_is_recovered_and_it_serves() { + let home = Home::new("serve-torn", 900_000); + copy_case( + &home, + "torn-tail", + &["2026-09-17.jsonl", "2026-09-18.jsonl"], + ); + let running = home.serve(); + let printed = running.stop(); + assert!(printed.contains(RECOVERED_NOTICE), "{printed}"); + assert!(RECOVERED_NOTICE.ends_with("see docs/runbook.md#audit-recovered")); +} + +#[test] +fn accept_break_with_nothing_to_accept_exits_2() { + let home = Home::new("serve-nothing", 900_000); + let output = home.run(&["--accept-break"]); + assert_eq!(output.status.code(), Some(2)); + assert!( + stderr(&output).contains("nothing to accept"), + "{}", + stderr(&output) + ); +} + +#[test] +fn bad_arguments_and_bad_configs_do_not_start() { + let brokerd = env!("CARGO_BIN_EXE_brokerd"); + for args in [ + &[][..], + &["serve"][..], + &["serve", "--config"][..], + &["serve", "--config", "a", "--config", "b"][..], + &["serve", "--config", "a", "--loud"][..], + &["run", "--config", "a"][..], + ] { + let output = Command::new(brokerd).args(args).output().unwrap(); + assert_eq!(output.status.code(), Some(2), "{args:?}"); + assert!( + stderr(&output).starts_with("usage: brokerd serve"), + "{args:?}" + ); + } + let home = Home::new("serve-config", 900_000); + std::fs::write(&home.config, "[paths]\nhoem = \"/x\"\n").unwrap(); + let output = home.run(&[]); + assert_eq!(output.status.code(), Some(1)); + assert!( + stderr(&output).contains("brokerd.toml"), + "{}", + stderr(&output) + ); + std::fs::remove_file(&home.config).unwrap(); + assert_eq!(home.run(&[]).status.code(), Some(1)); + assert!( + !home.path("audit").exists(), + "nothing made before the config is read" + ); +} diff --git a/docs/plans/M5b/files/crates/gatewayd/tests/instance.rs b/docs/plans/M5b/files/crates/gatewayd/tests/instance.rs new file mode 100644 index 0000000..766940f --- /dev/null +++ b/docs/plans/M5b/files/crates/gatewayd/tests/instance.rs @@ -0,0 +1,218 @@ +//! One `gatewayd` per home (M5b spec, section 9a): a second `gatewayd serve` against the same home +//! is refused with the lock's line, and the first keeps serving `gateway.sock`; once the first is +//! killed, a new one removes the stale socket and serves. Each runs against the fake Mattermost. +//! Every wait has a deadline, and every process is killed and every path removed by a guard (tip +//! T44). Do not edit. + +#[path = "support/brokerd_client.rs"] +mod brokerd_client; +#[path = "support/fake_mm.rs"] +mod fake_mm; +#[path = "support/tmp.rs"] +mod tmp; + +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +use brokerd_client::{ask, pending}; +use fake_mm::{FakeMm, KYLE}; +use tmp::TempDir; + +const POINTER: &str = "see docs/runbook.md#already-running"; +const TOKEN: &str = "tok-instance-test"; +const WAIT: Duration = Duration::from_secs(10); + +struct Home { + dir: TempDir, + config: PathBuf, +} + +impl Home { + fn new(tag: &str, mm: &FakeMm) -> Home { + let dir = TempDir::new(tag); + let text = format!( + "[mattermost]\nurl = \"{}\"\n[secrets.mattermost_token]\nenv = \"GW_TEST_TOKEN\"\n\ + [allow]\nusers = [\"{KYLE}\"]\n[paths]\nhome = \"{}\"\n", + mm.url(), + dir.path().display() + ); + let config = dir.write("gatewayd.toml", &text); + Home { dir, config } + } + + fn path(&self, relative: &str) -> PathBuf { + self.dir.path().join(relative) + } + + fn socket(&self) -> PathBuf { + self.path("run/gateway/gateway.sock") + } + + fn command(&self) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_gatewayd")); + command + .args(["serve", "--config"]) + .arg(&self.config) + .env_remove("CREDENTIALS_DIRECTORY") + .env("GW_TEST_TOKEN", TOKEN) + .stdin(Stdio::null()) + .stdout(Stdio::null()); + command + } + + /// `gatewayd serve` with its stderr in `/.err`, for the messages of a failing test. + fn start(&self, name: &str) -> Daemon { + let err = std::fs::File::create(self.path(&format!("{name}.err"))).unwrap(); + Daemon(self.command().stderr(Stdio::from(err)).spawn().unwrap()) + } + + fn stderr_of(&self, name: &str) -> String { + std::fs::read_to_string(self.path(&format!("{name}.err"))).unwrap_or_default() + } + + /// Runs `gatewayd serve` 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(&self) -> (Option, String) { + let mut child = Daemon(self.command().stderr(Stdio::piped()).spawn().unwrap()); + let deadline = Instant::now() + WAIT; + 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) + } +} + +/// A running `gatewayd serve`, killed and waited for when dropped. Declared after the `Home` and +/// the `FakeMm`, 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)) + } +} + +/// True when `gatewayd` answers on `gateway.sock` as it does for `brokerd`: approval `approval`, +/// of a `bxctl chat` session, is posted to the first approver and the posts come back. The +/// connection is then closed, which ends the approval. +fn answers(socket: &Path, approval: u64) -> bool { + match ask(socket, pending(approval, "chat-1789700000-1", "{}")) { + Ok(asking) => { + let answered = !asking.posts.is_empty(); + asking.close(); + answered + } + Err(_) => false, + } +} + +/// Waits up to 10 s for `gatewayd` to answer on `socket`. +fn wait_until_it_answers(socket: &Path, approval: u64) -> bool { + let deadline = Instant::now() + WAIT; + while Instant::now() < deadline { + if answers(socket, approval) { + 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_gatewayd_on_the_same_home_is_refused_and_the_first_keeps_serving() { + let mm = FakeMm::start(); + let home = Home::new("inst-twice", &mm); + let mut first = home.start("first"); + let _ws = mm.next_ws(WAIT); + assert!( + wait_until_it_answers(&home.socket(), 1), + "the first gatewayd never answered; stderr: {}", + home.stderr_of("first") + ); + let before = inode(&home.socket()); + + let (code, stderr) = home.run_to_end(); + let line = format!( + "gatewayd: another gatewayd is running (it holds {}); {POINTER}", + home.path("run/gatewayd.lock").display() + ); + assert_eq!(code, Some(1), "the second gatewayd: {stderr}"); + assert_eq!( + stderr, + format!("{line}\n"), + "the lock's line, and nothing else" + ); + + assert!(first.running(), "the first gatewayd is still running"); + assert_eq!( + inode(&home.socket()), + before, + "the first's socket is still there" + ); + assert!( + answers(&home.socket(), 2), + "the first gatewayd still answers; its stderr: {}", + home.stderr_of("first") + ); +} + +#[test] +fn after_the_first_is_killed_a_new_gatewayd_removes_the_stale_socket_and_serves() { + let mm = FakeMm::start(); + let home = Home::new("inst-restart", &mm); + let mut first = home.start("first"); + let _ws = mm.next_ws(WAIT); + assert!( + wait_until_it_answers(&home.socket(), 1), + "the first gatewayd never answered; stderr: {}", + home.stderr_of("first") + ); + let _ = first.0.kill(); + let _ = first.0.wait(); + assert!( + home.socket().exists(), + "a killed gatewayd leaves its socket" + ); + assert!( + !answers(&home.socket(), 2), + "nobody serves the stale socket" + ); + + let _next = home.start("next"); + let _ws = mm.next_ws(WAIT); + assert!( + wait_until_it_answers(&home.socket(), 3), + "the new gatewayd never answered; stderr: {}", + home.stderr_of("next") + ); + let lock = home.path("run/gatewayd.lock"); + let mode = std::fs::metadata(&lock).map(|m| m.permissions().mode() & 0o777); + assert_eq!(mode.ok(), Some(0o600), "the lock file, mode 0600"); +} diff --git a/docs/plans/M5b/files/crates/memoryd/tests/instance.rs b/docs/plans/M5b/files/crates/memoryd/tests/instance.rs new file mode 100644 index 0000000..03ade34 --- /dev/null +++ b/docs/plans/M5b/files/crates/memoryd/tests/instance.rs @@ -0,0 +1,223 @@ +//! One `memoryd` per home (M5b spec, section 9a): a second `memoryd serve` against the same home is +//! refused with the lock's line, and the first keeps serving; once the first is killed, a new one +//! removes the stale sockets and serves. Every wait has a deadline, and every process is killed +//! and every path removed by a guard (tip T44). Do not edit. + +#[path = "support/tmp.rs"] +mod tmp; + +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::{Empty, Envelope, Message, PROTOCOL_VERSION, read_frame, write_frame}; +use tmp::TempDir; + +const POINTER: &str = "see docs/runbook.md#already-running"; +const SOCKETS: [&str; 3] = [ + "run/loop-memory/memory.sock", + "run/broker-memory/memory.sock", + "run/owner-memory/admin.sock", +]; + +struct Home { + dir: TempDir, + config: PathBuf, +} + +impl Home { + fn new(tag: &str) -> Home { + let dir = TempDir::new(tag); + let soul = dir.write("etc/soul.md", "You are Boxmaker.\n"); + let text = format!( + "[paths]\nhome = \"{}\"\n[memory]\nsoul_starter = \"{}\"\n", + dir.path().display(), + soul.display() + ); + let config = dir.write("memoryd.toml", &text); + Home { dir, config } + } + + fn path(&self, relative: &str) -> PathBuf { + self.dir.path().join(relative) + } + + fn command(&self) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_memoryd")); + command + .args(["serve", "--config"]) + .arg(&self.config) + .stdin(Stdio::null()) + .stdout(Stdio::null()); + command + } + + /// `memoryd serve` with its stderr in `/.err`, for the messages of a failing test. + fn start(&self, name: &str) -> Daemon { + let err = std::fs::File::create(self.path(&format!("{name}.err"))).unwrap(); + Daemon(self.command().stderr(Stdio::from(err)).spawn().unwrap()) + } + + fn stderr_of(&self, name: &str) -> String { + std::fs::read_to_string(self.path(&format!("{name}.err"))).unwrap_or_default() + } + + /// Runs `memoryd serve` 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(&self) -> (Option, String) { + let mut child = Daemon(self.command().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; + } + 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) + } + + fn inodes(&self) -> Vec { + SOCKETS.iter().map(|s| inode(&self.path(s))).collect() + } +} + +/// A running `memoryd serve`, killed and waited for when dropped. Declared after the `Home`, 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)) + } +} + +/// True when `memoryd` answers on `socket`, `loopd`'s: `memory_block` gets the block. 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 ask = Envelope { + v: PROTOCOL_VERSION, + id: 1, + r#final: true, + msg: Message::MemoryBlock(Empty {}), + }; + if write_frame(&mut stream, &ask).is_err() { + return false; + } + matches!( + read_frame(&mut stream), + Ok(Envelope { + msg: Message::MemoryBlockReply(_), + .. + }) + ) +} + +/// Waits up to 10 s for `memoryd` to answer on `loopd`'s socket and to listen on the other two. +fn wait_until_it_answers(home: &Home) -> bool { + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline { + if answers(&home.path(SOCKETS[0])) + && SOCKETS[1..] + .iter() + .all(|s| UnixStream::connect(home.path(s)).is_ok()) + { + return true; + } + std::thread::sleep(Duration::from_millis(20)); + } + 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_memoryd_on_the_same_home_is_refused_and_the_first_keeps_serving() { + let home = Home::new("inst-twice"); + let mut first = home.start("first"); + assert!( + wait_until_it_answers(&home), + "the first memoryd never answered; stderr: {}", + home.stderr_of("first") + ); + let before = home.inodes(); + + let (code, stderr) = home.run_to_end(); + let line = format!( + "memoryd: another memoryd is running (it holds {}); {POINTER}", + home.path("run/memoryd.lock").display() + ); + assert_eq!(code, Some(1), "the second memoryd: {stderr}"); + assert_eq!( + stderr, + format!("{line}\n"), + "the lock's line, and nothing else" + ); + + assert!(first.running(), "the first memoryd is still running"); + assert_eq!(home.inodes(), before, "the first's sockets are still there"); + assert!( + answers(&home.path(SOCKETS[0])), + "the first memoryd still answers; its stderr: {}", + home.stderr_of("first") + ); +} + +#[test] +fn after_the_first_is_killed_a_new_memoryd_removes_the_stale_sockets_and_serves() { + let home = Home::new("inst-restart"); + let mut first = home.start("first"); + assert!( + wait_until_it_answers(&home), + "the first memoryd never answered; stderr: {}", + home.stderr_of("first") + ); + let _ = first.0.kill(); + let _ = first.0.wait(); + assert!( + SOCKETS.iter().all(|s| home.path(s).exists()), + "a killed memoryd leaves its sockets" + ); + assert!( + !answers(&home.path(SOCKETS[0])), + "nobody serves the stale socket" + ); + + let _next = home.start("next"); + assert!( + wait_until_it_answers(&home), + "the new memoryd never answered; stderr: {}", + home.stderr_of("next") + ); + let lock = home.path("run/memoryd.lock"); + let mode = std::fs::metadata(&lock).map(|m| m.permissions().mode() & 0o777); + assert_eq!(mode.ok(), Some(0o600), "the lock file, mode 0600"); +}