brokerd, memoryd, gatewayd: a lock file keeps each to one instance
Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>
This commit is contained in:
@@ -11,6 +11,7 @@ use brokerd::container::Podman;
|
||||
use brokerd::memory_port::{MEMORY_WAIT, MemoryPort, Routed};
|
||||
use brokerd::runner::{Refusing, Runtime};
|
||||
use brokerd::serve::{self, ServeError};
|
||||
use bxcore::instance;
|
||||
|
||||
const USAGE: &str = "usage: brokerd serve --config <path> [--accept-break]";
|
||||
/// Where the owner looks when brokerd cannot start: its config, a directory or a socket.
|
||||
@@ -42,6 +43,18 @@ fn main() -> ExitCode {
|
||||
}
|
||||
};
|
||||
|
||||
// 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`.
|
||||
let broker_path = cfg.broker_socket();
|
||||
let admin_path = cfg.admin_socket();
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -7,7 +7,7 @@ use std::process::ExitCode;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::AtomicBool;
|
||||
|
||||
use bxcore::secrets;
|
||||
use bxcore::{instance, secrets};
|
||||
use gatewayd::config::{Config, MATTERMOST_TOKEN};
|
||||
use gatewayd::serve::{START_FAILED, Tuning, run};
|
||||
|
||||
@@ -34,6 +34,8 @@ fn serve(path: &Path) -> ExitCode {
|
||||
// `let loaded = match secrets::load("gatewayd", MATTERMOST_TOKEN, &source, &env)`: an Err(e)
|
||||
// prints "gatewayd: {e}" (the error carries its own pointer).
|
||||
// Then `if let Some(warning) = &loaded.warning { eprintln!("{warning}"); }`.
|
||||
// 3b. The instance lock, `<home>/run/gatewayd.lock` (M5b spec, section 9a): after the secret,
|
||||
// which is read before anything is made, and before the directory and the socket.
|
||||
// 4. The state file's directory:
|
||||
// `let dir = config.state_path().parent().map(Path::to_path_buf).unwrap_or_default();`
|
||||
// `std::fs::DirBuilder::new().recursive(true).mode(0o700).create(&dir)`: an Err(e) prints
|
||||
@@ -66,6 +68,14 @@ fn serve(path: &Path) -> ExitCode {
|
||||
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);
|
||||
}
|
||||
};
|
||||
let dir = config
|
||||
.state_path()
|
||||
.parent()
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
@@ -5,6 +5,7 @@ use std::path::PathBuf;
|
||||
use std::process::ExitCode;
|
||||
use std::sync::Arc;
|
||||
|
||||
use bxcore::instance;
|
||||
use memoryd::config::Config;
|
||||
use memoryd::serve;
|
||||
|
||||
@@ -24,6 +25,17 @@ fn main() -> ExitCode {
|
||||
return ExitCode::from(1);
|
||||
}
|
||||
};
|
||||
// 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}"));
|
||||
let running = match serve::start(config, log) {
|
||||
Ok(running) => running,
|
||||
|
||||
@@ -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