Each locks <home>/run/<role>.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) <noreply@anthropic.com>
219 lines
6.8 KiB
Rust
219 lines
6.8 KiB
Rust
//! 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 `<home>/<name>.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<i32>, 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");
|
|
}
|