M5b plan: task 13, instance lock in brokerd, memoryd and gatewayd
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>
This commit is contained in:
Claude Opus 5.5
@@ -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 `<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");
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
@@ -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<AuditEvent> {
|
||||
let dir = self.path("audit");
|
||||
let mut names: Vec<String> = 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::<bxcore::AuditRecord>(l)
|
||||
.unwrap()
|
||||
.event
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
/// Kills the daemon when dropped; `stop` returns what it printed.
|
||||
struct Running(Option<Child>);
|
||||
|
||||
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<Envelope> {
|
||||
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<u8>)> {
|
||||
let mut all: Vec<(String, Vec<u8>)> = 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"
|
||||
);
|
||||
}
|
||||
@@ -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 `<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 `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<i32>, 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");
|
||||
}
|
||||
@@ -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 `<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 `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<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 inodes(&self) -> Vec<u64> {
|
||||
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");
|
||||
}
|
||||
Reference in New Issue
Block a user