M5b plan: task 12, instance lock in bxcore, loopd and inferproxy
bxcore::instance::lock (File::try_lock, file 0600, missing directory 0700), LockError::line for the two lines the daemons print, and the already-running runbook fragment; loopd locks <home>/run/loopd.lock before its socket and self-test, inferproxy <socket>.lock. Checks: given tests fail at ref 11 as the task says (bxcore E0432; loopd 3 and inferproxy 3 fail); the task's quoted edits applied to ref 11 reproduce ref 12 byte for byte; replay gate ok with 1113 tests (1099 + 14), /tmp unchanged; 25 mutations, 24 caught (the uncaught one maps TryLockError::Error to Held, not reachable from a test); given files rustfmt-clean, under 500 lines; no chat-template tags. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,70 @@
|
||||
//! One instance per daemon (M5b spec, section 9a). Every daemon removes a socket it finds at its
|
||||
//! path before it binds, so a second copy started by mistake would take the sockets of the one
|
||||
//! that runs. Before it touches them, a daemon takes an exclusive lock on its lock file with
|
||||
//! `lock`, and keeps the `InstanceLock` until the process ends.
|
||||
|
||||
use std::fs::{DirBuilder, File, OpenOptions, TryLockError};
|
||||
use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt};
|
||||
use std::path::Path;
|
||||
|
||||
/// The runbook entry for both ways `lock` can fail.
|
||||
pub const RUNBOOK: &str = "see docs/runbook.md#already-running";
|
||||
|
||||
/// A held lock: the open lock file. The system releases the lock when the file is closed, which is
|
||||
/// when this is dropped or the process ends, however it ends. The file itself stays: removing it
|
||||
/// would let a second copy lock a new file while the first still runs.
|
||||
#[derive(Debug)]
|
||||
#[must_use = "the lock is released when this value is dropped"]
|
||||
pub struct InstanceLock {
|
||||
_file: File,
|
||||
}
|
||||
|
||||
/// Why `lock` did not take the lock.
|
||||
#[derive(Debug)]
|
||||
pub enum LockError {
|
||||
/// Another open file holds the lock: another instance is running.
|
||||
Held,
|
||||
/// The lock file, or its directory, could not be created or opened, or the lock failed.
|
||||
Io(std::io::Error),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for LockError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
LockError::Held => write!(f, "another instance holds the lock"),
|
||||
LockError::Io(e) => write!(f, "{e}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for LockError {
|
||||
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
|
||||
match self {
|
||||
LockError::Held => None,
|
||||
LockError::Io(e) => Some(e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl LockError {
|
||||
/// The line a daemon prints when it does not start: `role` is its name, `path` the lock file.
|
||||
pub fn line(&self, role: &str, path: &Path) -> String {
|
||||
// `Held`: "{role}: another {role} is running (it holds {path}); {RUNBOOK}".
|
||||
// `Io(e)`: "{role}: cannot take the lock {path}: {e}; {RUNBOOK}".
|
||||
// Each with `format!`, the path shown with `path.display()`.
|
||||
todo!()
|
||||
}
|
||||
}
|
||||
|
||||
/// Take the exclusive lock on `path`, creating the file with mode 0600, and its directory with
|
||||
/// mode 0700 when it is missing, as the daemons make their socket directories. The lock is tried
|
||||
/// once, never waited for.
|
||||
pub fn lock(path: &Path) -> Result<InstanceLock, LockError> {
|
||||
// 1. If `path.parent()` is `Some(dir)`: `DirBuilder::new().recursive(true).mode(0o700)
|
||||
// .create(dir)`. 2. `OpenOptions::new().write(true).create(true).truncate(false)
|
||||
// .mode(0o600).open(path)`. An error of either is `LockError::Io`, through `map_err`.
|
||||
// 3. `file.try_lock()`: `Ok(())` is `Ok(InstanceLock { _file: file })`,
|
||||
// `Err(TryLockError::WouldBlock)` is `Err(LockError::Held)`, and
|
||||
// `Err(TryLockError::Error(e))` is `Err(LockError::Io(e))`.
|
||||
todo!()
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
//! `bxcore::instance`: the lock that keeps a daemon to one instance (M5b spec, section 9a). Two
|
||||
//! opens of the same file in one process are two holders, as two processes are. Do not edit.
|
||||
|
||||
#[path = "support/tmp.rs"]
|
||||
mod tmp;
|
||||
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use bxcore::instance::{LockError, RUNBOOK, lock};
|
||||
use tmp::TempDir;
|
||||
|
||||
/// Gives a directory back its owner's permissions when dropped, so the `TempDir` declared before
|
||||
/// it can be removed even when an assertion fails (tip T44).
|
||||
struct Restore(PathBuf);
|
||||
|
||||
impl Drop for Restore {
|
||||
fn drop(&mut self) {
|
||||
let _ = std::fs::set_permissions(&self.0, std::fs::Permissions::from_mode(0o700));
|
||||
}
|
||||
}
|
||||
|
||||
fn mode(path: &Path) -> u32 {
|
||||
std::fs::metadata(path).unwrap().permissions().mode() & 0o777
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_second_lock_is_held_while_the_first_lives() {
|
||||
let dir = TempDir::new("lock-held");
|
||||
let path = dir.path().join("run").join("loopd.lock");
|
||||
let first = lock(&path).expect("the first lock");
|
||||
for _ in 0..3 {
|
||||
match lock(&path) {
|
||||
Err(LockError::Held) => {}
|
||||
other => panic!("a second lock while the first lives: {other:?}"),
|
||||
}
|
||||
}
|
||||
drop(first);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_lock_is_free_again_once_the_first_is_dropped() {
|
||||
let dir = TempDir::new("lock-free");
|
||||
let path = dir.path().join("run").join("loopd.lock");
|
||||
let first = lock(&path).expect("the first lock");
|
||||
drop(first);
|
||||
let second = lock(&path).expect("the lock after the first was dropped");
|
||||
assert!(
|
||||
matches!(lock(&path), Err(LockError::Held)),
|
||||
"the second holds it now"
|
||||
);
|
||||
drop(second);
|
||||
assert!(path.exists(), "the lock file is left in place");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_lock_file_left_behind_is_taken() {
|
||||
// A killed daemon leaves its lock file, unlocked; the next start takes it and keeps its bytes.
|
||||
let dir = TempDir::new("lock-left");
|
||||
let path = dir.write("run/memoryd.lock", "left\n");
|
||||
let held = lock(&path).expect("a lock file nobody holds");
|
||||
assert_eq!(std::fs::read_to_string(&path).unwrap(), "left\n");
|
||||
drop(held);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_lock_file_is_created_with_mode_0600_and_its_directory_0700() {
|
||||
let dir = TempDir::new("lock-mode");
|
||||
let run = dir.path().join("home").join("run");
|
||||
let path = run.join("brokerd.lock");
|
||||
let held = lock(&path).expect("a lock in a directory that does not exist yet");
|
||||
assert_eq!(mode(&path), 0o600, "the lock file");
|
||||
assert_eq!(mode(&run), 0o700, "the directory it made");
|
||||
assert_eq!(mode(&dir.path().join("home")), 0o700, "and the one above");
|
||||
assert_eq!(
|
||||
std::fs::metadata(&path).unwrap().len(),
|
||||
0,
|
||||
"nothing is written"
|
||||
);
|
||||
drop(held);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_directory_that_cannot_be_written_is_an_io_error() {
|
||||
let dir = TempDir::new("lock-ro");
|
||||
let run = dir.path().join("run");
|
||||
std::fs::create_dir(&run).unwrap();
|
||||
let _restore = Restore(run.clone());
|
||||
std::fs::set_permissions(&run, std::fs::Permissions::from_mode(0o500)).unwrap();
|
||||
match lock(&run.join("gatewayd.lock")) {
|
||||
Err(LockError::Io(e)) => {
|
||||
assert_eq!(e.kind(), std::io::ErrorKind::PermissionDenied, "{e}");
|
||||
}
|
||||
other => panic!("a lock in a read-only directory: {other:?}"),
|
||||
}
|
||||
assert!(!run.join("gatewayd.lock").exists());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_path_through_a_file_is_an_io_error() {
|
||||
let dir = TempDir::new("lock-file");
|
||||
let file = dir.write("run", "not a directory\n");
|
||||
match lock(&file.join("loopd.lock")) {
|
||||
Err(LockError::Io(_)) => {}
|
||||
other => panic!("a lock under a file: {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_lines_a_daemon_prints() {
|
||||
let path = Path::new("/home/bx/run/loopd.lock");
|
||||
assert_eq!(
|
||||
LockError::Held.line("loopd", path),
|
||||
"loopd: another loopd is running (it holds /home/bx/run/loopd.lock); \
|
||||
see docs/runbook.md#already-running"
|
||||
);
|
||||
let io = LockError::Io(std::io::Error::other("the disk is on fire"));
|
||||
assert_eq!(
|
||||
io.line("inferproxy", Path::new("/run/bx/infer.sock.lock")),
|
||||
"inferproxy: cannot take the lock /run/bx/infer.sock.lock: the disk is on fire; \
|
||||
see docs/runbook.md#already-running"
|
||||
);
|
||||
assert_eq!(RUNBOOK, "see docs/runbook.md#already-running");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_error_displays_and_has_a_source() {
|
||||
use std::error::Error;
|
||||
assert_eq!(
|
||||
LockError::Held.to_string(),
|
||||
"another instance holds the lock"
|
||||
);
|
||||
assert!(LockError::Held.source().is_none());
|
||||
let io = LockError::Io(std::io::Error::other("no room"));
|
||||
assert_eq!(io.to_string(), "no room");
|
||||
assert!(io.source().is_some());
|
||||
}
|
||||
@@ -0,0 +1,228 @@
|
||||
//! One `inferproxy` per socket (M5b spec, section 9a): the lock is `<socket>.lock`, beside the
|
||||
//! `--listen` socket. A second `inferproxy` on the same socket is refused, and the first keeps
|
||||
//! forwarding; once the first is killed, a new one removes the stale socket and forwards. Every
|
||||
//! wait has a deadline, and every process is killed and every path removed by a guard (tip T44).
|
||||
//! Do not edit.
|
||||
|
||||
use std::io::{Read, Write};
|
||||
use std::net::TcpListener;
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::os::unix::net::UnixStream;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::{Child, Command, Stdio};
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
use std::thread;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
const POINTER: &str = "see docs/runbook.md#already-running";
|
||||
|
||||
static NEXT: AtomicU32 = AtomicU32::new(0);
|
||||
|
||||
/// A temporary directory for the socket, removed when dropped, first with its owner's permissions
|
||||
/// put back.
|
||||
struct Dir(PathBuf);
|
||||
|
||||
impl Dir {
|
||||
fn new() -> Dir {
|
||||
let n = NEXT.fetch_add(1, Ordering::SeqCst);
|
||||
let dir = std::env::temp_dir().join(format!("ip-lock-{}-{n}", std::process::id()));
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
Dir(dir)
|
||||
}
|
||||
|
||||
fn socket(&self) -> PathBuf {
|
||||
self.0.join("infer.sock")
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Dir {
|
||||
fn drop(&mut self) {
|
||||
let _ = std::fs::set_permissions(&self.0, std::fs::Permissions::from_mode(0o700));
|
||||
let _ = std::fs::remove_dir_all(&self.0);
|
||||
}
|
||||
}
|
||||
|
||||
/// An upstream that reads four bytes from each connection, answers `pong` and closes.
|
||||
fn upstream() -> String {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
let addr = listener.local_addr().unwrap().to_string();
|
||||
thread::spawn(move || {
|
||||
for stream in listener.incoming() {
|
||||
let Ok(mut stream) = stream else { break };
|
||||
thread::spawn(move || {
|
||||
let mut ping = [0u8; 4];
|
||||
if stream.read_exact(&mut ping).is_ok() {
|
||||
let _ = stream.write_all(b"pong");
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
addr
|
||||
}
|
||||
|
||||
/// A running `inferproxy`, killed and waited for when dropped. Declared after the `Dir`, so it
|
||||
/// drops first.
|
||||
struct Proxy(Child);
|
||||
|
||||
impl Drop for Proxy {
|
||||
fn drop(&mut self) {
|
||||
let _ = self.0.kill();
|
||||
let _ = self.0.wait();
|
||||
}
|
||||
}
|
||||
|
||||
impl Proxy {
|
||||
fn running(&mut self) -> bool {
|
||||
matches!(self.0.try_wait(), Ok(None))
|
||||
}
|
||||
}
|
||||
|
||||
fn command(socket: &Path, upstream: &str) -> Command {
|
||||
let mut command = Command::new(env!("CARGO_BIN_EXE_inferproxy"));
|
||||
command
|
||||
.arg("--listen")
|
||||
.arg(socket)
|
||||
.args(["--upstream", upstream])
|
||||
.stdin(Stdio::null())
|
||||
.stdout(Stdio::null());
|
||||
command
|
||||
}
|
||||
|
||||
fn start(socket: &Path, upstream: &str) -> Proxy {
|
||||
Proxy(
|
||||
command(socket, upstream)
|
||||
.stderr(Stdio::null())
|
||||
.spawn()
|
||||
.unwrap(),
|
||||
)
|
||||
}
|
||||
|
||||
/// Runs `inferproxy` to its end, at most 10 s (then it is killed), and gives its exit code
|
||||
/// (`None` if it had to be killed) and its stderr.
|
||||
fn run_to_end(socket: &Path, upstream: &str) -> (Option<i32>, String) {
|
||||
let mut child = Proxy(
|
||||
command(socket, upstream)
|
||||
.stderr(Stdio::piped())
|
||||
.spawn()
|
||||
.unwrap(),
|
||||
);
|
||||
let deadline = Instant::now() + Duration::from_secs(10);
|
||||
let mut code = None;
|
||||
while Instant::now() < deadline {
|
||||
if let Ok(Some(status)) = child.0.try_wait() {
|
||||
code = status.code();
|
||||
break;
|
||||
}
|
||||
thread::sleep(Duration::from_millis(20));
|
||||
}
|
||||
let _ = child.0.kill();
|
||||
let _ = child.0.wait();
|
||||
let mut stderr = String::new();
|
||||
if let Some(mut pipe) = child.0.stderr.take() {
|
||||
let _ = pipe.read_to_string(&mut stderr);
|
||||
}
|
||||
(code, stderr)
|
||||
}
|
||||
|
||||
/// True when a `ping` sent to `socket` comes back from the upstream as `pong` within 5 s.
|
||||
fn forwards(socket: &Path) -> bool {
|
||||
let Ok(mut stream) = UnixStream::connect(socket) else {
|
||||
return false;
|
||||
};
|
||||
let limit = Some(Duration::from_secs(5));
|
||||
if stream.set_read_timeout(limit).is_err() || stream.set_write_timeout(limit).is_err() {
|
||||
return false;
|
||||
}
|
||||
if stream.write_all(b"ping").is_err() {
|
||||
return false;
|
||||
}
|
||||
let mut answer = Vec::new();
|
||||
let _ = stream.read_to_end(&mut answer);
|
||||
answer == b"pong"
|
||||
}
|
||||
|
||||
/// Waits up to 10 s for an `inferproxy` to forward on `socket`.
|
||||
fn wait_until_it_forwards(socket: &Path) -> bool {
|
||||
let deadline = Instant::now() + Duration::from_secs(10);
|
||||
while Instant::now() < deadline {
|
||||
if forwards(socket) {
|
||||
return true;
|
||||
}
|
||||
thread::sleep(Duration::from_millis(50));
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
fn inode(path: &Path) -> u64 {
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
std::fs::symlink_metadata(path)
|
||||
.map(|m| m.ino())
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_second_inferproxy_on_the_same_socket_is_refused_and_the_first_keeps_forwarding() {
|
||||
let dir = Dir::new();
|
||||
let socket = dir.socket();
|
||||
let upstream = upstream();
|
||||
let mut first = start(&socket, &upstream);
|
||||
assert!(wait_until_it_forwards(&socket), "the first never forwarded");
|
||||
let before = inode(&socket);
|
||||
|
||||
let (code, stderr) = run_to_end(&socket, &upstream);
|
||||
let line = format!(
|
||||
"inferproxy: another inferproxy is running (it holds {}.lock); {POINTER}",
|
||||
socket.display()
|
||||
);
|
||||
assert_eq!(code, Some(1), "the second inferproxy: {stderr}");
|
||||
assert!(
|
||||
stderr.lines().any(|l| l == line),
|
||||
"wanted the line\n{line}\nstderr: {stderr}"
|
||||
);
|
||||
|
||||
assert!(first.running(), "the first inferproxy is still running");
|
||||
assert_eq!(inode(&socket), before, "the first's socket is still there");
|
||||
assert!(forwards(&socket), "the first still forwards");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn after_the_first_is_killed_a_new_inferproxy_removes_the_stale_socket_and_forwards() {
|
||||
let dir = Dir::new();
|
||||
let socket = dir.socket();
|
||||
let upstream = upstream();
|
||||
let mut first = start(&socket, &upstream);
|
||||
assert!(wait_until_it_forwards(&socket), "the first never forwarded");
|
||||
let _ = first.0.kill();
|
||||
let _ = first.0.wait();
|
||||
assert!(socket.exists(), "a killed inferproxy leaves its socket");
|
||||
assert!(!forwards(&socket), "nobody serves the stale socket");
|
||||
|
||||
let _next = start(&socket, &upstream);
|
||||
assert!(
|
||||
wait_until_it_forwards(&socket),
|
||||
"the new one never forwarded"
|
||||
);
|
||||
let lock = dir.0.join("infer.sock.lock");
|
||||
let mode = std::fs::metadata(&lock).map(|m| m.permissions().mode() & 0o777);
|
||||
assert_eq!(mode.ok(), Some(0o600), "the lock file, beside the socket");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_socket_directory_that_cannot_be_written_stops_inferproxy_naming_the_error() {
|
||||
let dir = Dir::new();
|
||||
let socket = dir.socket();
|
||||
std::fs::set_permissions(&dir.0, std::fs::Permissions::from_mode(0o500)).unwrap();
|
||||
|
||||
let (code, stderr) = run_to_end(&socket, &upstream());
|
||||
let line = format!(
|
||||
"inferproxy: cannot take the lock {}.lock: Permission denied (os error 13); {POINTER}",
|
||||
socket.display()
|
||||
);
|
||||
assert_eq!(code, Some(1), "{stderr}");
|
||||
assert!(
|
||||
stderr.lines().any(|l| l == line),
|
||||
"wanted the line\n{line}\nstderr: {stderr}"
|
||||
);
|
||||
assert!(!socket.exists(), "no socket");
|
||||
}
|
||||
@@ -0,0 +1,311 @@
|
||||
//! One `loopd` per home (M5b spec, section 9a): a second `loopd serve` against the same home is
|
||||
//! refused before it touches the socket or the server, and the first keeps serving; once the first
|
||||
//! is killed, a new one removes the stale socket and serves. Every wait has a deadline, and every
|
||||
//! process is killed and every path removed by a guard (tip T44). Do not edit.
|
||||
|
||||
mod support;
|
||||
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::os::unix::net::UnixStream;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::{Child, Command, Stdio};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use bxcore::{
|
||||
Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, WireError, read_frame,
|
||||
write_frame,
|
||||
};
|
||||
use support::{FakeServer, Home, Reply, tokenizer};
|
||||
|
||||
const POINTER: &str = "see docs/runbook.md#already-running";
|
||||
|
||||
fn config_file(home: &Home, server: &FakeServer) -> PathBuf {
|
||||
let text = format!(
|
||||
r#"
|
||||
[infer]
|
||||
socket = "{}"
|
||||
model = "test-model"
|
||||
[slots]
|
||||
main = 0
|
||||
background = 1
|
||||
[expect]
|
||||
template_sha256 = "f55f52930aa8bf44ab5cb85f99370fcc3c56e9a85640b812086d5330bce5d86b"
|
||||
n_ctx = 131072
|
||||
slots = 2
|
||||
[limits]
|
||||
poll_ms = 40
|
||||
liveness_ms = 500
|
||||
retry_backoff_ms = [10]
|
||||
[paths]
|
||||
home = "{}"
|
||||
"#,
|
||||
server.socket.display(),
|
||||
home.dir.display()
|
||||
);
|
||||
let path = home.dir.join("config.toml");
|
||||
std::fs::write(&path, text).unwrap();
|
||||
path
|
||||
}
|
||||
|
||||
/// A server that passes the self-test twice: once for the first `loopd`, once for the one started
|
||||
/// after it is killed.
|
||||
fn healthy_routes(server: &FakeServer) {
|
||||
server.route("/props", vec![Reply::fixture("props")]);
|
||||
server.respond("/tokenize", tokenizer);
|
||||
server.route(
|
||||
"/v1/chat/completions",
|
||||
vec![
|
||||
Reply::fixture("tool_call"),
|
||||
Reply::fixture("turn1"),
|
||||
Reply::fixture("turn2"),
|
||||
Reply::fixture("tool_call"),
|
||||
Reply::fixture("turn1"),
|
||||
Reply::fixture("turn2"),
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
fn socket(home: &Home) -> PathBuf {
|
||||
home.dir.join("run").join("loop").join("loop.sock")
|
||||
}
|
||||
|
||||
/// A running `loopd serve`, killed and waited for when dropped. Declared after the `Home` and the
|
||||
/// `FakeServer`, so it drops first.
|
||||
struct Daemon(Child);
|
||||
|
||||
impl Drop for Daemon {
|
||||
fn drop(&mut self) {
|
||||
let _ = self.0.kill();
|
||||
let _ = self.0.wait();
|
||||
}
|
||||
}
|
||||
|
||||
impl Daemon {
|
||||
fn running(&mut self) -> bool {
|
||||
matches!(self.0.try_wait(), Ok(None))
|
||||
}
|
||||
}
|
||||
|
||||
/// `loopd serve` with its stderr in `<home>/<name>.err`, for the messages of a failing test.
|
||||
fn start(home: &Home, config: &Path, name: &str) -> Daemon {
|
||||
let err = std::fs::File::create(home.dir.join(format!("{name}.err"))).unwrap();
|
||||
let child = Command::new(env!("CARGO_BIN_EXE_loopd"))
|
||||
.args(["serve", "--config"])
|
||||
.arg(config)
|
||||
.stdin(Stdio::null())
|
||||
.stdout(Stdio::null())
|
||||
.stderr(Stdio::from(err))
|
||||
.spawn()
|
||||
.unwrap();
|
||||
Daemon(child)
|
||||
}
|
||||
|
||||
fn stderr_of(home: &Home, name: &str) -> String {
|
||||
std::fs::read_to_string(home.dir.join(format!("{name}.err"))).unwrap_or_default()
|
||||
}
|
||||
|
||||
/// Runs `loopd serve` to its end, at most 20 s (then it is killed), and gives its exit code
|
||||
/// (`None` if it had to be killed) and its stderr.
|
||||
fn run_to_end(config: &Path) -> (Option<i32>, String) {
|
||||
let mut child = Daemon(
|
||||
Command::new(env!("CARGO_BIN_EXE_loopd"))
|
||||
.args(["serve", "--config"])
|
||||
.arg(config)
|
||||
.stdin(Stdio::null())
|
||||
.stdout(Stdio::null())
|
||||
.stderr(Stdio::piped())
|
||||
.spawn()
|
||||
.unwrap(),
|
||||
);
|
||||
let deadline = Instant::now() + Duration::from_secs(20);
|
||||
let mut code = None;
|
||||
while Instant::now() < deadline {
|
||||
if let Ok(Some(status)) = child.0.try_wait() {
|
||||
code = status.code();
|
||||
break;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(20));
|
||||
}
|
||||
let _ = child.0.kill();
|
||||
let _ = child.0.wait();
|
||||
let mut stderr = String::new();
|
||||
if let Some(mut pipe) = child.0.stderr.take() {
|
||||
let _ = std::io::Read::read_to_string(&mut pipe, &mut stderr);
|
||||
}
|
||||
(code, stderr)
|
||||
}
|
||||
|
||||
/// True when `loopd` answers on `socket`: a turn that resumes a session that does not exist gets
|
||||
/// the error frame for it. Any other answer, or none within 5 s, is false.
|
||||
fn answers(socket: &Path) -> bool {
|
||||
let Ok(mut stream) = UnixStream::connect(socket) else {
|
||||
return false;
|
||||
};
|
||||
let limit = Some(Duration::from_secs(5));
|
||||
if stream.set_read_timeout(limit).is_err() || stream.set_write_timeout(limit).is_err() {
|
||||
return false;
|
||||
}
|
||||
let turn = Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
id: 7,
|
||||
r#final: true,
|
||||
msg: Message::Turn(Turn {
|
||||
session: SessionId::new("probe").unwrap(),
|
||||
content: "are you there".to_string(),
|
||||
resume: true,
|
||||
}),
|
||||
};
|
||||
if write_frame(&mut stream, &turn).is_err() {
|
||||
return false;
|
||||
}
|
||||
matches!(
|
||||
read_frame(&mut stream),
|
||||
Ok(Envelope {
|
||||
msg: Message::Error(WireError {
|
||||
code: ErrorCode::NoSuchSession,
|
||||
..
|
||||
}),
|
||||
..
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
/// Waits up to 20 s for `loopd` to answer on `socket`.
|
||||
fn wait_until_it_answers(socket: &Path) -> bool {
|
||||
let deadline = Instant::now() + Duration::from_secs(20);
|
||||
while Instant::now() < deadline {
|
||||
if answers(socket) {
|
||||
return true;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(50));
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
fn inode(path: &Path) -> u64 {
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
std::fs::symlink_metadata(path)
|
||||
.map(|m| m.ino())
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_second_loopd_on_the_same_home_is_refused_and_the_first_keeps_serving() {
|
||||
let home = Home::new();
|
||||
let server = FakeServer::start();
|
||||
healthy_routes(&server);
|
||||
let config = config_file(&home, &server);
|
||||
let socket = socket(&home);
|
||||
let mut first = start(&home, &config, "first");
|
||||
assert!(
|
||||
wait_until_it_answers(&socket),
|
||||
"the first loopd never answered; stderr: {}",
|
||||
stderr_of(&home, "first")
|
||||
);
|
||||
let before = inode(&socket);
|
||||
let completions = server.requests_to("/v1/chat/completions").len();
|
||||
let tokenized = server.requests_to("/tokenize").len();
|
||||
|
||||
let (code, stderr) = run_to_end(&config);
|
||||
let lock = home.dir.join("run").join("loopd.lock");
|
||||
let line = format!(
|
||||
"loopd: another loopd is running (it holds {}); {POINTER}",
|
||||
lock.display()
|
||||
);
|
||||
assert_eq!(code, Some(1), "the second loopd: {stderr}");
|
||||
assert!(
|
||||
stderr.lines().any(|l| l == line),
|
||||
"wanted the line\n{line}\nstderr: {stderr}"
|
||||
);
|
||||
assert!(
|
||||
!stderr.contains("selftest"),
|
||||
"the second ran no self-test: {stderr}"
|
||||
);
|
||||
assert_eq!(
|
||||
(
|
||||
server.requests_to("/v1/chat/completions").len(),
|
||||
server.requests_to("/tokenize").len()
|
||||
),
|
||||
(completions, tokenized),
|
||||
"the second sent the server nothing"
|
||||
);
|
||||
|
||||
assert!(first.running(), "the first loopd is still running");
|
||||
assert_eq!(inode(&socket), before, "the first's socket is still there");
|
||||
assert!(
|
||||
answers(&socket),
|
||||
"the first loopd still answers; its stderr: {}",
|
||||
stderr_of(&home, "first")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn after_the_first_is_killed_a_new_loopd_removes_the_stale_socket_and_serves() {
|
||||
let home = Home::new();
|
||||
let server = FakeServer::start();
|
||||
healthy_routes(&server);
|
||||
let config = config_file(&home, &server);
|
||||
let socket = socket(&home);
|
||||
let mut first = start(&home, &config, "first");
|
||||
assert!(
|
||||
wait_until_it_answers(&socket),
|
||||
"the first loopd never answered; stderr: {}",
|
||||
stderr_of(&home, "first")
|
||||
);
|
||||
let _ = first.0.kill();
|
||||
let _ = first.0.wait();
|
||||
assert!(socket.exists(), "a killed loopd leaves its socket");
|
||||
assert!(!answers(&socket), "nobody serves the stale socket");
|
||||
|
||||
let _next = start(&home, &config, "next");
|
||||
assert!(
|
||||
wait_until_it_answers(&socket),
|
||||
"the new loopd never answered; stderr: {}",
|
||||
stderr_of(&home, "next")
|
||||
);
|
||||
let lock = home.dir.join("run").join("loopd.lock");
|
||||
let mode = std::fs::metadata(&lock).map(|m| m.permissions().mode() & 0o777);
|
||||
assert_eq!(mode.ok(), Some(0o600), "the lock file, mode 0600");
|
||||
assert_eq!(
|
||||
server.requests_to("/v1/chat/completions").len(),
|
||||
6,
|
||||
"each of the two ran the self-test"
|
||||
);
|
||||
}
|
||||
|
||||
/// Gives `<home>/run` back its owner's permissions when dropped, before the `Home` is removed.
|
||||
struct Restore(PathBuf);
|
||||
|
||||
impl Drop for Restore {
|
||||
fn drop(&mut self) {
|
||||
let _ = std::fs::set_permissions(&self.0, std::fs::Permissions::from_mode(0o700));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_run_directory_that_cannot_be_written_stops_loopd_naming_the_error() {
|
||||
let home = Home::new();
|
||||
let server = FakeServer::start();
|
||||
healthy_routes(&server);
|
||||
let config = config_file(&home, &server);
|
||||
let run = home.dir.join("run");
|
||||
std::fs::create_dir(&run).unwrap();
|
||||
let _restore = Restore(run.clone());
|
||||
std::fs::set_permissions(&run, std::fs::Permissions::from_mode(0o500)).unwrap();
|
||||
|
||||
let (code, stderr) = run_to_end(&config);
|
||||
let line = format!(
|
||||
"loopd: cannot take the lock {}: Permission denied (os error 13); {POINTER}",
|
||||
run.join("loopd.lock").display()
|
||||
);
|
||||
assert_eq!(code, Some(1), "{stderr}");
|
||||
assert!(
|
||||
stderr.lines().any(|l| l == line),
|
||||
"wanted the line\n{line}\nstderr: {stderr}"
|
||||
);
|
||||
assert!(
|
||||
server.requests().is_empty(),
|
||||
"loopd sent the server nothing"
|
||||
);
|
||||
assert!(!socket(&home).exists(), "no socket");
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
## already-running
|
||||
|
||||
**What you see.** A daemon exits 1 at start, before it touches its sockets, with one of these
|
||||
lines:
|
||||
|
||||
- `<role>: another <role> is running (it holds <lock file>); see docs/runbook.md#already-running`
|
||||
- `<role>: cannot take the lock <lock file>: <error>; see docs/runbook.md#already-running`
|
||||
|
||||
`<role>` is `loopd`, `brokerd`, `memoryd`, `gatewayd` or `inferproxy`. The lock file is
|
||||
`$BOXMAKER_HOME/run/<role>.lock`; for `inferproxy` it is its `--listen` socket's path with
|
||||
`.lock` added (`infer.sock.lock`, beside `infer.sock`).
|
||||
|
||||
**Why.** Every daemon removes a socket it finds at its path before it binds, so that a socket
|
||||
left by a daemon that was killed never blocks the next start. A second copy started by mistake
|
||||
would therefore take the sockets of the one that runs, which would keep running, unreachable,
|
||||
and nothing would say so. So each daemon first takes an exclusive lock on its lock file, and
|
||||
holds it for as long as it runs (M5b spec, section 9a). The system releases the lock when the
|
||||
process ends, however it ends, so a killed daemon never leaves a lock behind. The file stays in
|
||||
place and holds nothing.
|
||||
|
||||
**Confirm.** For `another <role> is running`, find the copy that runs, with its command line
|
||||
(the `--config` path says which home it serves):
|
||||
|
||||
```sh
|
||||
pgrep -a -x <role>
|
||||
ls -l "$BOXMAKER_HOME/run/<role>.lock"
|
||||
```
|
||||
|
||||
For `cannot take the lock`, the error says why: `Permission denied` means the daemon's user
|
||||
cannot create or open the file, or create `run/` (`ls -ld "$BOXMAKER_HOME" "$BOXMAKER_HOME/run"`
|
||||
and `ls -l` on the lock file show who owns them); `Not a directory` means a path above it is a
|
||||
file; `Read-only file system` means what it says.
|
||||
|
||||
**Fix.** Two copies of a role against the same home are always a mistake: use the one that runs.
|
||||
To restart it, stop it (`pkill -x <role>`, or Ctrl-C in its terminal), wait until `pgrep -x
|
||||
<role>` prints nothing, and start it again. Two copies with different homes are fine (each has
|
||||
its own lock), as long as their configs do not name the same socket path. For a lock that cannot
|
||||
be taken, give the daemon's user the directory (`chown`), or point its home somewhere it can
|
||||
write.
|
||||
|
||||
Never delete a lock file while its daemon runs: the lock is held on the open file, so a new copy
|
||||
would create a new file, lock that, and take the sockets. Deleting it while the daemon is stopped
|
||||
does no harm, and does not help either.
|
||||
|
||||
**Check.** `pgrep -x <role>` prints one process id, and the daemon prints its start line
|
||||
(`loopd: serving on …`, `inferproxy: <socket> -> <upstream>`, `brokerd: serving tools on …`,
|
||||
`memoryd: serving`, `gatewayd: connected to …`).
|
||||
Reference in New Issue
Block a user