Milestone 1: workspace skeleton, wire protocol, socket round-trip
Three-crate workspace per SPECS.md. varde-proto defines the full JSON Lines protocol (requests, envelopes, structured errors, events) with string-typed hashes so the crate carries no iroh dependency. The daemon binds its unix socket, loads config with flags > env > file > defaults precedence, and answers status/list; everything else returns a structured "unimplemented" error. varde-ctl maps subcommands 1:1 onto requests and round-trips status against a real daemon in the tests. Dependencies: serde/serde_json (wire format), tokio (async runtime and unix sockets), tracing/tracing-subscriber (structured logging), toml (config file), anyhow (binary-edge errors), thiserror (reserved for library errors), clap (ctl flag parsing, per spec), tempfile (dev-only, ephemeral test dirs). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,251 @@
|
||||
//! Daemon configuration.
|
||||
//!
|
||||
//! Precedence: CLI flags > environment > config file > defaults.
|
||||
//! The daemon must run with an empty (or absent) config file.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use serde::Deserialize;
|
||||
|
||||
/// Fully resolved runtime configuration.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Config {
|
||||
/// Directory holding the blob store, metadata and secret key.
|
||||
pub store_dir: PathBuf,
|
||||
/// Path of the unix socket to serve the API on.
|
||||
pub socket_path: PathBuf,
|
||||
/// Upload rate cap in bytes/sec. 0 disables serving entirely.
|
||||
pub max_upload_bytes_per_sec: u64,
|
||||
/// Download rate cap in bytes/sec. 0 means unlimited.
|
||||
pub max_download_bytes_per_sec: u64,
|
||||
/// Whether LAN discovery (mDNS-style) is enabled.
|
||||
pub discovery: bool,
|
||||
/// Whether any WAN (non-link-local) upload is permitted. Default off:
|
||||
/// LAN-only posture with zero WAN upload.
|
||||
pub wan_upload: bool,
|
||||
}
|
||||
|
||||
/// Serde image of the TOML config file. Everything optional so an empty
|
||||
/// file (or none at all) is valid.
|
||||
#[derive(Debug, Default, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct FileConfig {
|
||||
store_dir: Option<PathBuf>,
|
||||
socket_path: Option<PathBuf>,
|
||||
max_upload_bytes_per_sec: Option<u64>,
|
||||
max_download_bytes_per_sec: Option<u64>,
|
||||
discovery: Option<bool>,
|
||||
wan_upload: Option<bool>,
|
||||
}
|
||||
|
||||
/// Values collected from CLI flags; `None` means "not given".
|
||||
#[derive(Debug, Default)]
|
||||
pub struct Overrides {
|
||||
pub config_path: Option<PathBuf>,
|
||||
pub store_dir: Option<PathBuf>,
|
||||
pub socket_path: Option<PathBuf>,
|
||||
}
|
||||
|
||||
/// Whether the daemon runs as a system service or a per-user service.
|
||||
/// Decides default paths only; everything can be overridden.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum Mode {
|
||||
System,
|
||||
User,
|
||||
}
|
||||
|
||||
impl Mode {
|
||||
/// System mode when running as root, user mode otherwise.
|
||||
pub fn detect() -> Mode {
|
||||
// SAFETY: geteuid has no preconditions and cannot fail.
|
||||
if unsafe { libc_geteuid() } == 0 {
|
||||
Mode::System
|
||||
} else {
|
||||
Mode::User
|
||||
}
|
||||
}
|
||||
|
||||
fn default_store_dir(self) -> PathBuf {
|
||||
match self {
|
||||
Mode::System => PathBuf::from("/var/lib/varde"),
|
||||
Mode::User => xdg_dir("XDG_DATA_HOME", ".local/share").join("varde"),
|
||||
}
|
||||
}
|
||||
|
||||
fn default_socket_path(self) -> PathBuf {
|
||||
match self {
|
||||
Mode::System => PathBuf::from("/run/varde/varde.sock"),
|
||||
Mode::User => match std::env::var_os("XDG_RUNTIME_DIR") {
|
||||
Some(dir) => PathBuf::from(dir).join("varde.sock"),
|
||||
// No XDG_RUNTIME_DIR (rare outside a session): fall back
|
||||
// next to the store so the daemon still starts.
|
||||
None => Mode::User.default_store_dir().join("varde.sock"),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fn default_config_path(self) -> PathBuf {
|
||||
match self {
|
||||
Mode::System => PathBuf::from("/etc/varde/config.toml"),
|
||||
Mode::User => xdg_dir("XDG_CONFIG_HOME", ".config").join("varde/config.toml"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn xdg_dir(var: &str, home_rel: &str) -> PathBuf {
|
||||
if let Some(dir) = std::env::var_os(var) {
|
||||
return PathBuf::from(dir);
|
||||
}
|
||||
let home = std::env::var_os("HOME").unwrap_or_else(|| "/".into());
|
||||
PathBuf::from(home).join(home_rel)
|
||||
}
|
||||
|
||||
// Minimal FFI shim instead of pulling in the libc crate for one call.
|
||||
extern "C" {
|
||||
#[link_name = "geteuid"]
|
||||
fn libc_geteuid() -> u32;
|
||||
}
|
||||
|
||||
impl Config {
|
||||
/// Resolve configuration with full precedence:
|
||||
/// `overrides` (flags) > environment > config file > mode defaults.
|
||||
pub fn load(mode: Mode, overrides: &Overrides) -> Result<Config> {
|
||||
let config_path = overrides
|
||||
.config_path
|
||||
.clone()
|
||||
.or_else(|| std::env::var_os("VARDE_CONFIG").map(PathBuf::from))
|
||||
.unwrap_or_else(|| mode.default_config_path());
|
||||
|
||||
// An explicitly named config file must exist; the default one is
|
||||
// allowed to be absent.
|
||||
let explicit =
|
||||
overrides.config_path.is_some() || std::env::var_os("VARDE_CONFIG").is_some();
|
||||
let file = Self::read_file(&config_path, explicit)?;
|
||||
|
||||
let env_path = |var: &str| std::env::var_os(var).map(PathBuf::from);
|
||||
let env_u64 = |var: &str| -> Result<Option<u64>> {
|
||||
match std::env::var(var) {
|
||||
Ok(v) => {
|
||||
Ok(Some(v.parse().with_context(|| {
|
||||
format!("{var} must be an integer, got {v:?}")
|
||||
})?))
|
||||
}
|
||||
Err(_) => Ok(None),
|
||||
}
|
||||
};
|
||||
let env_bool = |var: &str| -> Result<Option<bool>> {
|
||||
match std::env::var(var) {
|
||||
Ok(v) => {
|
||||
Ok(Some(v.parse().with_context(|| {
|
||||
format!("{var} must be true/false, got {v:?}")
|
||||
})?))
|
||||
}
|
||||
Err(_) => Ok(None),
|
||||
}
|
||||
};
|
||||
|
||||
Ok(Config {
|
||||
store_dir: overrides
|
||||
.store_dir
|
||||
.clone()
|
||||
.or_else(|| env_path("VARDE_STORE"))
|
||||
.or(file.store_dir)
|
||||
.unwrap_or_else(|| mode.default_store_dir()),
|
||||
socket_path: overrides
|
||||
.socket_path
|
||||
.clone()
|
||||
.or_else(|| env_path("VARDE_SOCKET"))
|
||||
.or(file.socket_path)
|
||||
.unwrap_or_else(|| mode.default_socket_path()),
|
||||
max_upload_bytes_per_sec: env_u64("VARDE_MAX_UPLOAD")?
|
||||
.or(file.max_upload_bytes_per_sec)
|
||||
// Conservative default: 10 MB/s on LAN.
|
||||
.unwrap_or(10 * 1024 * 1024),
|
||||
max_download_bytes_per_sec: env_u64("VARDE_MAX_DOWNLOAD")?
|
||||
.or(file.max_download_bytes_per_sec)
|
||||
.unwrap_or(0),
|
||||
discovery: env_bool("VARDE_DISCOVERY")?
|
||||
.or(file.discovery)
|
||||
.unwrap_or(true),
|
||||
wan_upload: env_bool("VARDE_WAN_UPLOAD")?
|
||||
.or(file.wan_upload)
|
||||
.unwrap_or(false),
|
||||
})
|
||||
}
|
||||
|
||||
fn read_file(path: &Path, must_exist: bool) -> Result<FileConfig> {
|
||||
match std::fs::read_to_string(path) {
|
||||
Ok(text) => toml::from_str(&text)
|
||||
.with_context(|| format!("parsing config file {}", path.display())),
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound && !must_exist => {
|
||||
Ok(FileConfig::default())
|
||||
}
|
||||
Err(e) => Err(e).with_context(|| format!("reading config file {}", path.display())),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
// Env-var manipulation is process-global, so these tests only use
|
||||
// overrides and files, never set_var.
|
||||
|
||||
#[test]
|
||||
fn empty_config_file_is_valid() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("config.toml");
|
||||
std::fs::write(&path, "").unwrap();
|
||||
let overrides = Overrides {
|
||||
config_path: Some(path),
|
||||
..Default::default()
|
||||
};
|
||||
let cfg = Config::load(Mode::User, &overrides).unwrap();
|
||||
assert!(!cfg.wan_upload, "WAN upload must default off");
|
||||
assert!(cfg.discovery, "discovery defaults on");
|
||||
assert_eq!(cfg.max_upload_bytes_per_sec, 10 * 1024 * 1024);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn missing_default_config_is_fine() {
|
||||
let cfg = Config::load(Mode::User, &Overrides::default()).unwrap();
|
||||
assert!(cfg.socket_path.to_string_lossy().ends_with("varde.sock"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn missing_explicit_config_errors() {
|
||||
let overrides = Overrides {
|
||||
config_path: Some(PathBuf::from("/nonexistent/varde.toml")),
|
||||
..Default::default()
|
||||
};
|
||||
assert!(Config::load(Mode::User, &overrides).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn flags_beat_file() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("config.toml");
|
||||
std::fs::write(&path, "store_dir = \"/from/file\"\n").unwrap();
|
||||
let overrides = Overrides {
|
||||
config_path: Some(path),
|
||||
store_dir: Some(PathBuf::from("/from/flag")),
|
||||
..Default::default()
|
||||
};
|
||||
let cfg = Config::load(Mode::User, &overrides).unwrap();
|
||||
assert_eq!(cfg.store_dir, PathBuf::from("/from/flag"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unknown_keys_rejected() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("config.toml");
|
||||
std::fs::write(&path, "no_such_key = 1\n").unwrap();
|
||||
let overrides = Overrides {
|
||||
config_path: Some(path),
|
||||
..Default::default()
|
||||
};
|
||||
assert!(Config::load(Mode::User, &overrides).is_err());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
//! Request dispatch and daemon state.
|
||||
|
||||
use tokio::sync::broadcast;
|
||||
use varde_proto::{Envelope, ErrorCode, Event, GlobalStatus, Request, ResponseData};
|
||||
|
||||
use crate::config::Config;
|
||||
|
||||
/// Shared daemon state. Handlers grow with the milestones; anything not
|
||||
/// yet wired returns [`ErrorCode::Unimplemented`] rather than pretending.
|
||||
pub struct Daemon {
|
||||
config: Config,
|
||||
events: broadcast::Sender<Event>,
|
||||
}
|
||||
|
||||
impl Daemon {
|
||||
pub fn new(config: Config) -> Self {
|
||||
// Capacity bounds memory if a subscriber stalls; laggards get a
|
||||
// Lagged error, not unbounded buffering.
|
||||
let (events, _) = broadcast::channel(1024);
|
||||
Daemon { config, events }
|
||||
}
|
||||
|
||||
pub fn config(&self) -> &Config {
|
||||
&self.config
|
||||
}
|
||||
|
||||
/// New receiver for the daemon event stream.
|
||||
pub fn subscribe_events(&self) -> broadcast::Receiver<Event> {
|
||||
self.events.subscribe()
|
||||
}
|
||||
|
||||
/// Handle one API request. Infallible at this level: all failures are
|
||||
/// mapped to structured error envelopes.
|
||||
pub async fn handle(&self, request: Request) -> Envelope {
|
||||
match request {
|
||||
Request::Status { hash: None } => Envelope::ok(ResponseData::Status(GlobalStatus {
|
||||
version: env!("CARGO_PKG_VERSION").to_string(),
|
||||
protocol: varde_proto::PROTOCOL_VERSION,
|
||||
node_id: None,
|
||||
store_path: self.config.store_dir.display().to_string(),
|
||||
pins: 0,
|
||||
connected_peers: 0,
|
||||
})),
|
||||
Request::List {} => Envelope::ok(ResponseData::Pins { pins: Vec::new() }),
|
||||
Request::Subscribe {} => Envelope::ok(ResponseData::Done {}),
|
||||
other => Envelope::err(
|
||||
ErrorCode::Unimplemented,
|
||||
format!("not implemented yet: {other:?}"),
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
//! varde-daemon library surface.
|
||||
//!
|
||||
//! The daemon is a binary, but its internals are exposed as a library so
|
||||
//! integration tests can run real daemons in-process on ephemeral sockets.
|
||||
|
||||
pub mod config;
|
||||
pub mod daemon;
|
||||
pub mod server;
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
|
||||
/// Run a daemon with the given config until the future is dropped or the
|
||||
/// listener fails. Binds the socket before returning control to the
|
||||
/// runtime so callers can rely on it existing once this future is polled.
|
||||
pub async fn run(config: config::Config) -> Result<()> {
|
||||
std::fs::create_dir_all(&config.store_dir)
|
||||
.with_context(|| format!("creating store directory {}", config.store_dir.display()))?;
|
||||
let listener = server::bind_socket(&config.socket_path)?;
|
||||
let daemon = Arc::new(daemon::Daemon::new(config));
|
||||
server::serve(listener, daemon).await
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
//! varde-daemon: content-addressed blob mirror daemon.
|
||||
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::{bail, Context, Result};
|
||||
use tracing::info;
|
||||
|
||||
use varde_daemon::config::{Config, Mode, Overrides};
|
||||
use varde_daemon::{daemon, server};
|
||||
|
||||
const USAGE: &str = "\
|
||||
varde-daemon — content-addressed blob mirror daemon
|
||||
|
||||
USAGE:
|
||||
varde-daemon [OPTIONS]
|
||||
|
||||
OPTIONS:
|
||||
--config <PATH> Config file (default: /etc/varde/config.toml or
|
||||
$XDG_CONFIG_HOME/varde/config.toml)
|
||||
--store <PATH> Store directory override
|
||||
--socket <PATH> API socket path override
|
||||
--system Force system-mode default paths
|
||||
--user Force user-mode default paths
|
||||
--version Print version and exit
|
||||
--help Print this help and exit
|
||||
";
|
||||
|
||||
// The spec reserves clap for varde-ctl; the daemon takes five flags, which
|
||||
// a hand parse covers without the dependency.
|
||||
fn parse_args() -> Result<(Mode, Overrides)> {
|
||||
let mut overrides = Overrides::default();
|
||||
let mut mode = Mode::detect();
|
||||
let mut args = std::env::args().skip(1);
|
||||
while let Some(arg) = args.next() {
|
||||
let mut value = |name: &str| -> Result<PathBuf> {
|
||||
args.next()
|
||||
.map(PathBuf::from)
|
||||
.with_context(|| format!("{name} requires a value"))
|
||||
};
|
||||
match arg.as_str() {
|
||||
"--config" => overrides.config_path = Some(value("--config")?),
|
||||
"--store" => overrides.store_dir = Some(value("--store")?),
|
||||
"--socket" => overrides.socket_path = Some(value("--socket")?),
|
||||
"--system" => mode = Mode::System,
|
||||
"--user" => mode = Mode::User,
|
||||
"--version" => {
|
||||
println!("varde-daemon {}", env!("CARGO_PKG_VERSION"));
|
||||
std::process::exit(0);
|
||||
}
|
||||
"--help" | "-h" => {
|
||||
print!("{USAGE}");
|
||||
std::process::exit(0);
|
||||
}
|
||||
other => bail!("unknown argument {other:?}\n\n{USAGE}"),
|
||||
}
|
||||
}
|
||||
Ok((mode, overrides))
|
||||
}
|
||||
|
||||
fn main() -> Result<()> {
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(
|
||||
tracing_subscriber::EnvFilter::try_from_default_env()
|
||||
.unwrap_or_else(|_| "varde_daemon=info".into()),
|
||||
)
|
||||
.init();
|
||||
|
||||
let (mode, overrides) = parse_args()?;
|
||||
let config = Config::load(mode, &overrides)?;
|
||||
info!(store = %config.store_dir.display(), socket = %config.socket_path.display(), "starting");
|
||||
|
||||
std::fs::create_dir_all(&config.store_dir)
|
||||
.with_context(|| format!("creating store directory {}", config.store_dir.display()))?;
|
||||
|
||||
tokio::runtime::Builder::new_multi_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.context("building tokio runtime")?
|
||||
.block_on(run(config))
|
||||
}
|
||||
|
||||
async fn run(config: Config) -> Result<()> {
|
||||
let socket_path = config.socket_path.clone();
|
||||
let listener = server::bind_socket(&socket_path)?;
|
||||
let daemon = Arc::new(daemon::Daemon::new(config));
|
||||
|
||||
let result = tokio::select! {
|
||||
r = server::serve(listener, daemon) => r,
|
||||
r = shutdown_signal() => r.map(|signal| info!(signal, "shutting down")),
|
||||
};
|
||||
|
||||
// Best-effort cleanup so the next start doesn't find a stale socket.
|
||||
let _ = std::fs::remove_file(&socket_path);
|
||||
result
|
||||
}
|
||||
|
||||
async fn shutdown_signal() -> Result<&'static str> {
|
||||
use tokio::signal::unix::{signal, SignalKind};
|
||||
let mut term = signal(SignalKind::terminate()).context("installing SIGTERM handler")?;
|
||||
let mut int = signal(SignalKind::interrupt()).context("installing SIGINT handler")?;
|
||||
tokio::select! {
|
||||
_ = term.recv() => Ok("SIGTERM"),
|
||||
_ = int.recv() => Ok("SIGINT"),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
//! Unix socket server: JSON Lines request/response plus event streaming.
|
||||
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||||
use tokio::net::{UnixListener, UnixStream};
|
||||
use tracing::{debug, info, warn};
|
||||
use varde_proto::{Envelope, ErrorCode, Request};
|
||||
|
||||
use crate::daemon::Daemon;
|
||||
|
||||
/// Bind the API socket, replacing a stale socket file if the previous
|
||||
/// daemon did not shut down cleanly.
|
||||
pub fn bind_socket(path: &Path) -> Result<UnixListener> {
|
||||
if let Some(parent) = path.parent() {
|
||||
std::fs::create_dir_all(parent)
|
||||
.with_context(|| format!("creating socket directory {}", parent.display()))?;
|
||||
}
|
||||
match std::fs::remove_file(path) {
|
||||
Ok(()) => debug!(path = %path.display(), "removed stale socket"),
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
|
||||
Err(e) => {
|
||||
return Err(e).with_context(|| format!("removing stale socket {}", path.display()))
|
||||
}
|
||||
}
|
||||
UnixListener::bind(path).with_context(|| format!("binding socket {}", path.display()))
|
||||
}
|
||||
|
||||
/// Accept connections until cancelled. Each connection is served by its
|
||||
/// own task; a connection failure never takes the daemon down.
|
||||
pub async fn serve(listener: UnixListener, daemon: Arc<Daemon>) -> Result<()> {
|
||||
info!("listening for clients");
|
||||
loop {
|
||||
let (stream, _addr) = listener.accept().await.context("accepting client")?;
|
||||
let daemon = daemon.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = handle_connection(stream, daemon).await {
|
||||
debug!(error = %e, "client connection ended with error");
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_connection(stream: UnixStream, daemon: Arc<Daemon>) -> Result<()> {
|
||||
let (read_half, mut write_half) = stream.into_split();
|
||||
let mut lines = BufReader::new(read_half).lines();
|
||||
|
||||
while let Some(line) = lines.next_line().await? {
|
||||
if line.trim().is_empty() {
|
||||
continue;
|
||||
}
|
||||
let request: Request = match serde_json::from_str(&line) {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
let reply = Envelope::err(ErrorCode::BadRequest, format!("invalid request: {e}"));
|
||||
write_line(&mut write_half, &reply).await?;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
debug!(?request, "request");
|
||||
|
||||
let subscribe = matches!(request, Request::Subscribe {});
|
||||
let reply = daemon.handle(request).await;
|
||||
write_line(&mut write_half, &reply).await?;
|
||||
|
||||
// Subscribe flips this connection into a one-way event stream;
|
||||
// no further requests are read from it.
|
||||
if subscribe && reply.ok {
|
||||
let mut events = daemon.subscribe_events();
|
||||
loop {
|
||||
match events.recv().await {
|
||||
Ok(event) => write_line(&mut write_half, &event).await?,
|
||||
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
|
||||
warn!(missed = n, "event subscriber lagged, events dropped");
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Closed) => return Ok(()),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn write_line<T: serde::Serialize>(
|
||||
writer: &mut (impl AsyncWriteExt + Unpin),
|
||||
value: &T,
|
||||
) -> Result<()> {
|
||||
let mut buf = serde_json::to_vec(value)?;
|
||||
buf.push(b'\n');
|
||||
writer.write_all(&buf).await?;
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user