commit 4e1a95613bb817fa70c92eaadf6f4eaa416df162 Author: Bendik Lynghaug Date: Tue Jul 14 20:10:29 2026 +0200 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 diff --git a/.ci/woodpecker.yaml b/.ci/woodpecker.yaml new file mode 100644 index 0000000..7721aa2 --- /dev/null +++ b/.ci/woodpecker.yaml @@ -0,0 +1,22 @@ +# Woodpecker CI pipeline. Also parseable by eye as the definition of the +# project quality bar: fmt, clippy -D warnings, full test suite. +when: + - event: [push, pull_request] + +steps: + fmt: + image: rust:1 + commands: + - rustup component add rustfmt + - cargo fmt --all --check + + clippy: + image: rust:1 + commands: + - rustup component add clippy + - cargo clippy --workspace --all-targets -- -D warnings + + test: + image: rust:1 + commands: + - cargo test --workspace diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..3e8e408 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +target/ +.claude/ diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000..8f7f64b --- /dev/null +++ b/Cargo.lock @@ -0,0 +1,696 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "aho-corasick" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +dependencies = [ + "memchr", +] + +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" +dependencies = [ + "windows-sys", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" +dependencies = [ + "anstyle", + "once_cell_polyfill", + "windows-sys", +] + +[[package]] +name = "anyhow" +version = "1.0.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" + +[[package]] +name = "bitflags" +version = "2.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "clap" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51" +dependencies = [ + "clap_builder", + "clap_derive", +] + +[[package]] +name = "clap_builder" +version = "4.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "714a53001bf66416adb0e2ef5ac857140e7dc3a0c48fb28b2f10762fc4b5069f" +dependencies = [ + "anstream", + "anstyle", + "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "clap_lex" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" + +[[package]] +name = "colorchoice" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" + +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "fastrand" +version = "2.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", +] + +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + +[[package]] +name = "indexmap" +version = "2.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +dependencies = [ + "equivalent", + "hashbrown", +] + +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + +[[package]] +name = "libc" +version = "0.2.186" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" + +[[package]] +name = "linux-raw-sys" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" + +[[package]] +name = "log" +version = "0.4.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" + +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "mio" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "regex-automata" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f388202e4b80542a0921078cc23b6333bcf1409c1e3f86404cae4766a6131db" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" + +[[package]] +name = "rustix" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys", +] + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.150" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "serde_spanned" +version = "0.6.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3" +dependencies = [ + "serde", +] + +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "strsim" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" + +[[package]] +name = "syn" +version = "2.0.118" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "tempfile" +version = "3.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" +dependencies = [ + "fastrand", + "getrandom", + "once_cell", + "rustix", + "windows-sys", +] + +[[package]] +name = "thiserror" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "tokio" +version = "1.52.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys", +] + +[[package]] +name = "tokio-macros" +version = "2.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "toml" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362" +dependencies = [ + "serde", + "serde_spanned", + "toml_datetime", + "toml_edit", +] + +[[package]] +name = "toml_datetime" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c" +dependencies = [ + "serde", +] + +[[package]] +name = "toml_edit" +version = "0.22.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a" +dependencies = [ + "indexmap", + "serde", + "serde_spanned", + "toml_datetime", + "toml_write", + "winnow", +] + +[[package]] +name = "toml_write" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + +[[package]] +name = "varde-ctl" +version = "0.1.0" +dependencies = [ + "anyhow", + "clap", + "serde_json", + "tempfile", + "tokio", + "varde-daemon", + "varde-proto", +] + +[[package]] +name = "varde-daemon" +version = "0.1.0" +dependencies = [ + "anyhow", + "serde", + "serde_json", + "tempfile", + "thiserror", + "tokio", + "toml", + "tracing", + "tracing-subscriber", + "varde-proto", +] + +[[package]] +name = "varde-proto" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", +] + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "winnow" +version = "0.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df79d97927682d2fd8adb29682d1140b343be4ac0f08fd68b7765d9c059d3945" +dependencies = [ + "memchr", +] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..8c76b01 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,26 @@ +[workspace] +resolver = "2" +members = ["varde-proto", "varde-daemon", "varde-ctl"] + +[workspace.package] +version = "0.1.0" +edition = "2021" +license = "MIT OR Apache-2.0" +repository = "https://github.com/bendiklh/varde" +rust-version = "1.85" + +[workspace.dependencies] +varde-proto = { path = "varde-proto" } + +serde = { version = "1", features = ["derive"] } +serde_json = "1" +thiserror = "2" +anyhow = "1" +tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "io-util", "signal", "sync", "time", "fs"] } +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } +clap = { version = "4", features = ["derive", "env"] } +toml = "0.8" + +[profile.release] +lto = "thin" diff --git a/SPECS.md b/SPECS.md new file mode 100644 index 0000000..610db95 --- /dev/null +++ b/SPECS.md @@ -0,0 +1,129 @@ +# varde — a system blob mirror daemon on iroh + +> Working name: **varde** (Norwegian: a stone cairn used as a waypoint/beacon). Rename freely; the crate prefix is used consistently below so a find-replace suffices. + +This document is a build brief for Claude Code. Execute the milestones in order. Each milestone must compile, pass tests, and be committed before starting the next. Ask before deviating from the architecture; do not ask about implementation minutiae. + +## 1. Vision + +A small, boring, distro-packageable Linux daemon that owns a content-addressed blob store and mirrors content between consenting peers. Applications talk to it over a unix socket: "add this path", "pin this hash", "materialize hash X at path Y". The daemon handles storage, dedup, verification, LAN peer discovery, and replication policy. Think: what Windows Delivery Optimization is for updates, generalized, open, and built on iroh. + +**Design ethos (non-negotiable):** +- Infrastructure, not product. No GUI, no self-updater, no bundled runtime. Small composable pieces with a stable socket interface, a man page, and systemd units. +- Legible to the network. The daemon never camouflages its traffic. Default posture is LAN-only with zero WAN upload. All background transfer is rate-limited and clearly attributable. +- Consent-tiered. Discovery is open; replication is explicit. Nothing is fetched or served without a standing policy the user created. + +**Non-goals (v1):** gesture/DFT discovery (that arrives later as a separate discovery provider — see §6), WAN swarming, ISP policy channels (ALTO), Windows/macOS support, quotas/multi-tenant accounting. + +## 2. Architecture + +Cargo workspace, three crates: + +``` +varde/ +├── varde-proto # API types: requests, responses, events. serde. No I/O. +├── varde-daemon # the service: store, endpoint, policy engine, socket server +└── varde-ctl # CLI client speaking the socket protocol (also serves as API reference impl) +``` + +**Key dependencies and versions:** +- `iroh-blobs = "0.35"` — pin this. Per upstream, the post-0.35 rewrite is not yet production quality; 0.35 is the recommended production line. Use its persistent fs store (`iroh_blobs::store::fs`). +- `iroh` — matching version compatible with iroh-blobs 0.35. +- `tokio`, `serde`, `serde_json`, `tracing`, `clap` (ctl only), `anyhow`/`thiserror`. +- `zbus` for NetworkManager metered-status (feature-gated, `default-features` on Linux only). + +**Store layout:** `/var/lib/varde/` in system mode, `$XDG_DATA_HOME/varde/` in user mode. Contains the iroh-blobs store directory, a `meta.redb` (or JSON for MVP) for pins/policies/trust, and the endpoint secret key (`0600`). + +**Materialization:** when a client asks for a hash at a path, export from the store. Attempt reflink first (`FICLONE` ioctl via the `reflink-copy` crate) so btrfs/XFS users get zero-copy; fall back to copy. Never hardlink out of the store (store files must stay immutable). Document that the store dir and materialization targets should share a filesystem for reflink to work. + +## 3. Socket API + +JSON Lines over a unix stream socket. One request per line, one response per line, plus an event subscription mode. No varlink/D-Bus in v1 — keep the dependency surface minimal; the protocol is trivially wrappable later. + +Socket paths: `/run/varde/varde.sock` (system), `$XDG_RUNTIME_DIR/varde.sock` (user). Support systemd socket activation (`sd_notify` + `LISTEN_FDS`) but also plain standalone bind for non-systemd distros. + +Requests (define these as enums in `varde-proto`): + +| Request | Semantics | +|---|---| +| `Add { path, recursive }` | Import file/dir into store. Returns root hash (HashSeq for dirs). | +| `Pin { hash, policy }` | Standing intent: keep this content, fetch if absent, serve to peers per policy. | +| `Unpin { hash }` | Remove intent. GC is a separate explicit op. | +| `Materialize { hash, dest, mode }` | Export to path. `mode: reflink-or-copy \| copy`. | +| `Status { hash? }` | Global or per-hash: have/missing bytes, peers, transfer rates. | +| `List {}` | All pins with policies and completeness. | +| `TicketExport { hash }` / `TicketImport { ticket, pin_policy }` | iroh blob tickets — the v1 out-of-band sharing mechanism. | +| `Gc {}` | Drop unpinned blobs. | +| `Subscribe {}` | Switch connection to event stream (transfer progress, peer joined, pin complete). | + +Every response carries `{"ok": bool, ...}`. Errors are structured (`code`, `message`), never bare strings. + +## 4. Peer discovery and transfer (v1) + +Two mechanisms only: + +1. **Tickets** — manual, explicit, works today. `varde ctl ticket export ` on machine A, `ticket import` on machine B. +2. **LAN discovery** — iroh's local discovery (mDNS-style). Advertise under a stable, documented service so the daemon is *identifiable* on the network, not anonymous noise. Peers on the same LAN that hold overlapping pins exchange content automatically. + +**Trust model v1:** a peer allowlist keyed by iroh NodeId. `varde ctl peer trust `. Content is only *served* to trusted peers and only *fetched* from trusted peers. An `open-lan` policy flag exists per-pin for deliberately public content but defaults off. All fetched data is BLAKE3-verified by iroh-blobs regardless — trust gates participation, not integrity. + +**Traffic posture:** +- Token-bucket rate limiters on upload and download, configurable, with conservative defaults (e.g. 10 MB/s up on LAN, 0 on WAN). +- Set DSCP CS1/LE on the endpoint socket (best-effort; document that it may require `CAP_NET_ADMIN` or be silently ignored — that's fine). +- LEDBAT is not available off-the-shelf in the Rust QUIC stack; do not attempt it in v1. The rate limiter + DSCP + LAN-only default is the v1 answer to "don't anger anyone." Leave a `transport.rs` seam where scavenger congestion control could later slot in. +- **Metered awareness:** query NetworkManager over D-Bus for the metered flag on the active connection. If metered: no background fetching, no serving, period. Feature-gated so the daemon still builds without D-Bus. + +## 5. systemd and packaging + +Provide in `dist/`: +- `varde.service` + `varde.socket` (system) and user-unit variants. Hardening: `DynamicUser=yes` or dedicated `varde` user, `ProtectSystem=strict`, `StateDirectory=varde`, `RuntimeDirectory=varde`, `NoNewPrivileges=yes`, `RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6 AF_NETLINK`. +- A man page `varde.8` and `varde-ctl.1` (scdoc or raw roff, your choice — scdoc preferred). +- Example config `/etc/varde/config.toml`: store path, socket path, rate limits, discovery on/off, WAN policy. +- A `PKGBUILD` skeleton for Arch (the author's daily driver) — best-effort, clearly marked untested. + +Config precedence: CLI flags > env > config file > defaults. The daemon must run with an empty config. + +## 6. Discovery provider seam (the redoal hook) + +Define a trait in `varde-daemon`: + +```rust +/// A source of content announcements the daemon may subscribe to. +/// v1 ships LanDiscovery + TicketImport. A future provider maps +/// gesture-derived gossip TopicIds (redoal) to announcement streams. +trait DiscoveryProvider { + fn subscribe(&self, topic: TopicKey) -> BoxStream; + fn announce(&self, topic: TopicKey, ann: Announcement) -> Result<()>; +} + +struct Announcement { + root: Hash, // HashSeq root + author: PublicKey, // ed25519, signs the announcement + meta: BTreeMap, + providers: Vec, +} +``` + +Announcements are signed; the daemon indexes all announcements on subscribed topics but only auto-fetches from trusted author keys. This is the consent tiering: discovery open, replication explicit. Do not implement a gossip provider in v1 — just make `LanDiscovery` conform to the trait so the seam is proven. + +## 7. Milestones + +Each milestone = compiling code + tests + a short `docs/milestone-N.md` note on decisions made. + +1. **Skeleton.** Workspace, `varde-proto` types, daemon binds socket, `varde-ctl status` round-trips against an empty daemon. Config loading. Tracing setup. +2. **Store.** iroh-blobs 0.35 fs store wired in. `Add`, `List`, `Materialize` (reflink + fallback), `Gc`. Integration test: add a directory tree, materialize it elsewhere, byte-identical, reflink verified on btrfs (skip-if-unsupported in CI). +3. **Transfer.** iroh endpoint, ticket export/import, `Pin` triggers fetch from ticket-embedded provider. Two-daemon integration test over localhost. +4. **LAN + trust.** Local discovery, peer allowlist, auto-sync of overlapping pins between two trusted daemons. Rate limiting. Event subscription streaming progress. +5. **Citizenship.** Metered detection, DSCP attempt, systemd units + socket activation, man pages, config example, hardening pass, `varde-ctl` polish (human + `--json` output). +6. **Seam proof.** `DiscoveryProvider` trait, LanDiscovery refactored onto it, signed `Announcement` type with verification tests. Write `docs/redoal-integration.md` sketching the gesture-topic provider without implementing it. + +## 8. Testing and quality bar + +- `cargo clippy -- -D warnings`, `cargo fmt --check` in CI (a simple Woodpecker/Gitea Actions yaml in `.ci/`). +- Integration tests spawn real daemons on ephemeral sockets/ports; no mocking iroh. +- Property test: any file round-trips add→materialize bit-identically (proptest, bounded sizes). +- The daemon must survive `kill -9` mid-transfer and resume cleanly on restart (iroh-blobs handles partial state; write a test proving it). + +## 9. Style + +Plain Rust, minimal generics, no async-trait acrobatics where a concrete enum suffices. Errors: `thiserror` in libraries, `anyhow` at binary edges. Every public item in `varde-proto` documented. Comments explain *why*, not *what*. No dependency added without a one-line justification in the commit message. diff --git a/docs/milestone-1.md b/docs/milestone-1.md new file mode 100644 index 0000000..b18d99b --- /dev/null +++ b/docs/milestone-1.md @@ -0,0 +1,47 @@ +# Milestone 1 — Skeleton + +Workspace with the three crates from the spec, the full wire protocol, a +daemon that binds its socket and answers, and a `varde-ctl` that +round-trips `status` against it. + +## Decisions + +- **Wire types use strings for hashes/tickets/node ids.** `varde-proto` + stays free of iroh dependencies ("No I/O" taken to also mean no heavy + type imports), and the protocol is consumable from any language without + linking iroh. The daemon parses/validates at its boundary. +- **Envelope shape.** Every reply is `{"ok": bool, "data": {...}}` or + `{"ok": false, "error": {"code", "message"}}`. `data` is internally + tagged with `kind` so clients can dispatch without guessing. Error codes + are a closed enum (`bad_request`, `invalid_argument`, `not_found`, `io`, + `internal`, `unimplemented`). +- **The full request surface is defined now**, including peer-trust and + ticket ops. Unimplemented handlers return a structured `unimplemented` + error instead of the protocol growing per milestone. +- **`varde-daemon` is a lib + thin bin.** Integration tests run real + daemons: the daemon crate's tests spawn the actual binary + (`CARGO_BIN_EXE_varde-daemon`) on ephemeral sockets; the ctl crate's + test runs an in-process daemon (same `run()` entry point) and drives the + real `varde-ctl` binary against it. No mocked transport anywhere. +- **Daemon flag parsing is hand-rolled** (five flags); the spec reserves + clap for `varde-ctl`. Mode (system/user) defaults from euid — root gets + `/var/lib/varde` + `/run/varde/varde.sock`, users get XDG paths — and + `geteuid` is called through a two-line FFI shim instead of pulling in + the `libc` crate for one function. +- **Config precedence** is flags > `VARDE_*` env > TOML file > defaults, + implemented per-field so partial configs compose. An absent or empty + default config file is valid; an explicitly named one must exist. + Unknown keys in the file are rejected (`deny_unknown_fields`) so typos + fail loudly at startup rather than silently using defaults. +- **Defaults encode the traffic posture**: `wan_upload = false`, + upload cap 10 MB/s, discovery on (discovery is the open tier; serving + is still gated on trust, which doesn't exist until milestone 4). +- **Event plumbing exists from day one.** `Subscribe` flips a connection + into a one-way stream fed by a bounded `tokio::broadcast` channel + (laggards drop events rather than buffering unboundedly). Milestones 3/4 + publish into it. + +## Deviations from the spec + +None of substance. `--json` output for `varde-ctl` (a milestone 5 item) +landed early because the printing layer needed a shape anyway. diff --git a/varde-ctl/Cargo.toml b/varde-ctl/Cargo.toml new file mode 100644 index 0000000..11ae5cb --- /dev/null +++ b/varde-ctl/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "varde-ctl" +description = "CLI client for the varde daemon socket API" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true + +[[bin]] +name = "varde-ctl" +path = "src/main.rs" + +[dependencies] +varde-proto = { workspace = true } +anyhow = { workspace = true } +clap = { workspace = true } +serde_json = { workspace = true } +tokio = { workspace = true } + +[dev-dependencies] +varde-daemon = { path = "../varde-daemon" } +tempfile = "3" diff --git a/varde-ctl/src/main.rs b/varde-ctl/src/main.rs new file mode 100644 index 0000000..4d55e90 --- /dev/null +++ b/varde-ctl/src/main.rs @@ -0,0 +1,295 @@ +//! varde-ctl: CLI client for the varde daemon. +//! +//! Doubles as the reference implementation of the socket protocol: every +//! subcommand maps 1:1 onto a [`varde_proto::Request`]. + +use std::path::PathBuf; + +use anyhow::{bail, Context, Result}; +use clap::{Parser, Subcommand, ValueEnum}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::UnixStream; +use varde_proto::{Envelope, MaterializeMode, PinPolicy, Request, ResponseData}; + +#[derive(Parser)] +#[command(name = "varde-ctl", version, about = "Control the varde daemon")] +struct Cli { + /// Socket path (default: /run/varde/varde.sock as root, + /// $XDG_RUNTIME_DIR/varde.sock otherwise; env: VARDE_SOCKET) + #[arg(long, global = true, env = "VARDE_SOCKET")] + socket: Option, + + /// Print raw JSON responses instead of human-readable output + #[arg(long, global = true)] + json: bool, + + #[command(subcommand)] + command: Command, +} + +#[derive(Subcommand)] +enum Command { + /// Import a file or directory into the store + Add { + /// Path to import (made absolute before sending) + path: PathBuf, + /// Recurse into directories + #[arg(short, long)] + recursive: bool, + }, + /// Pin a hash: keep it, fetch it if absent, serve it per policy + Pin { + /// BLAKE3 hash (hex) + hash: String, + /// Serve to any LAN peer, not just trusted ones + #[arg(long)] + open_lan: bool, + }, + /// Remove a pin (data stays until `gc`) + Unpin { + /// BLAKE3 hash (hex) + hash: String, + }, + /// Export content from the store to a path + Materialize { + /// BLAKE3 hash (hex) + hash: String, + /// Destination path (made absolute before sending) + dest: PathBuf, + /// Export mode + #[arg(long, value_enum, default_value_t = CliMode::ReflinkOrCopy)] + mode: CliMode, + }, + /// Show daemon status, or status of one hash + Status { + /// BLAKE3 hash (hex) + hash: Option, + }, + /// List pins + List, + /// Ticket operations (out-of-band sharing) + #[command(subcommand)] + Ticket(TicketCommand), + /// Peer trust operations + #[command(subcommand)] + Peer(PeerCommand), + /// Drop all unpinned blobs + Gc, + /// Stream daemon events to stdout (one JSON object per line) + Subscribe, +} + +#[derive(Subcommand)] +enum TicketCommand { + /// Export a ticket for a hash + Export { + /// BLAKE3 hash (hex) + hash: String, + }, + /// Import a ticket: pin and fetch its content + Import { + /// Ticket string from `ticket export` + ticket: String, + /// Serve to any LAN peer, not just trusted ones + #[arg(long)] + open_lan: bool, + }, +} + +#[derive(Subcommand)] +enum PeerCommand { + /// Trust a peer NodeId + Trust { + /// iroh NodeId (z-base-32) + node_id: String, + }, + /// Revoke trust in a peer NodeId + Untrust { + /// iroh NodeId (z-base-32) + node_id: String, + }, + /// List trusted peers + List, +} + +#[derive(Clone, Copy, ValueEnum)] +enum CliMode { + ReflinkOrCopy, + Copy, +} + +impl From for MaterializeMode { + fn from(m: CliMode) -> Self { + match m { + CliMode::ReflinkOrCopy => MaterializeMode::ReflinkOrCopy, + CliMode::Copy => MaterializeMode::Copy, + } + } +} + +fn default_socket() -> PathBuf { + // Mirrors the daemon's mode detection: root talks to the system + // daemon, everyone else to their user daemon. + if is_root() { + PathBuf::from("/run/varde/varde.sock") + } else if let Some(dir) = std::env::var_os("XDG_RUNTIME_DIR") { + PathBuf::from(dir).join("varde.sock") + } else { + PathBuf::from("/run/varde/varde.sock") + } +} + +fn is_root() -> bool { + // SAFETY: geteuid has no preconditions and cannot fail. + unsafe { geteuid() == 0 } +} + +extern "C" { + fn geteuid() -> u32; +} + +fn absolute(path: PathBuf) -> Result { + let abs = + std::path::absolute(&path).with_context(|| format!("resolving path {}", path.display()))?; + Ok(abs.display().to_string()) +} + +fn to_request(command: Command) -> Result { + Ok(match command { + Command::Add { path, recursive } => Request::Add { + path: absolute(path)?, + recursive, + }, + Command::Pin { hash, open_lan } => Request::Pin { + hash, + policy: PinPolicy { open_lan }, + }, + Command::Unpin { hash } => Request::Unpin { hash }, + Command::Materialize { hash, dest, mode } => Request::Materialize { + hash, + dest: absolute(dest)?, + mode: mode.into(), + }, + Command::Status { hash } => Request::Status { hash }, + Command::List => Request::List {}, + Command::Ticket(TicketCommand::Export { hash }) => Request::TicketExport { hash }, + Command::Ticket(TicketCommand::Import { ticket, open_lan }) => Request::TicketImport { + ticket, + pin_policy: PinPolicy { open_lan }, + }, + Command::Peer(PeerCommand::Trust { node_id }) => Request::PeerTrust { node_id }, + Command::Peer(PeerCommand::Untrust { node_id }) => Request::PeerUntrust { node_id }, + Command::Peer(PeerCommand::List) => Request::PeerList {}, + Command::Gc => Request::Gc {}, + Command::Subscribe => Request::Subscribe {}, + }) +} + +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<()> { + let cli = Cli::parse(); + let socket = cli.socket.clone().unwrap_or_else(default_socket); + let subscribe = matches!(cli.command, Command::Subscribe); + let request = to_request(cli.command)?; + + let stream = UnixStream::connect(&socket) + .await + .with_context(|| format!("connecting to daemon at {}", socket.display()))?; + let (read_half, mut write_half) = stream.into_split(); + let mut lines = BufReader::new(read_half).lines(); + + let mut line = serde_json::to_string(&request)?; + line.push('\n'); + write_half.write_all(line.as_bytes()).await?; + + let reply = lines + .next_line() + .await? + .context("daemon closed the connection without replying")?; + let envelope: Envelope = + serde_json::from_str(&reply).context("daemon sent an unparseable reply")?; + + if !envelope.ok { + let err = envelope.error.context("error reply without error body")?; + bail!("{:?}: {}", err.code, err.message); + } + + if cli.json { + println!("{reply}"); + } else { + print_human(envelope.data.as_ref()); + } + + if subscribe { + // Connection is now an event stream; relay lines until it closes. + while let Some(event) = lines.next_line().await? { + println!("{event}"); + } + } + Ok(()) +} + +fn print_human(data: Option<&ResponseData>) { + let Some(data) = data else { + println!("ok"); + return; + }; + match data { + ResponseData::Added { hash, bytes } => println!("added {hash} ({bytes} bytes)"), + ResponseData::Pinned { hash } => println!("pinned {hash}"), + ResponseData::Done {} => println!("ok"), + ResponseData::Materialized { bytes, reflinked } => { + let how = if *reflinked { "reflinked" } else { "copied" }; + println!("materialized {bytes} bytes ({how})"); + } + ResponseData::Status(s) => { + println!("varde-daemon {} (protocol {})", s.version, s.protocol); + if let Some(node_id) = &s.node_id { + println!("node id: {node_id}"); + } + println!("store: {}", s.store_path); + println!("pins: {}", s.pins); + println!("peers: {} connected", s.connected_peers); + } + ResponseData::HashStatus(h) => { + println!("hash: {}", h.hash); + let total = h + .total_bytes + .map(|t| t.to_string()) + .unwrap_or_else(|| "?".to_string()); + println!("bytes: {}/{total}", h.have_bytes); + println!("complete: {}", h.complete); + println!("pinned: {}", h.pinned); + if !h.active_peers.is_empty() { + println!("active: {}", h.active_peers.join(", ")); + } + } + ResponseData::Pins { pins } => { + if pins.is_empty() { + println!("no pins"); + } + for pin in pins { + let state = if pin.complete { "complete" } else { "partial" }; + let lan = if pin.policy.open_lan { " open-lan" } else { "" }; + println!("{} {state}{lan}", pin.hash); + } + } + ResponseData::Ticket { ticket } => println!("{ticket}"), + ResponseData::Peers { peers } => { + if peers.is_empty() { + println!("no trusted peers"); + } + for peer in peers { + let state = if peer.connected { + "connected" + } else { + "offline" + }; + println!("{} {state}", peer.node_id); + } + } + ResponseData::GcDone { blobs_removed } => { + println!("gc: removed {blobs_removed} blobs"); + } + } +} diff --git a/varde-ctl/tests/roundtrip.rs b/varde-ctl/tests/roundtrip.rs new file mode 100644 index 0000000..bdbb9f3 --- /dev/null +++ b/varde-ctl/tests/roundtrip.rs @@ -0,0 +1,61 @@ +//! Milestone 1: `varde-ctl status` round-trips against a real daemon. + +use std::process::Command; +use std::time::Duration; + +use varde_daemon::config::Config; + +/// Start an in-process daemon on an ephemeral socket, then run the real +/// varde-ctl binary against it. +#[test] +fn ctl_status_round_trips() { + let dir = tempfile::tempdir().expect("tempdir"); + let socket_path = dir.path().join("varde.sock"); + let config = Config { + store_dir: dir.path().join("store"), + socket_path: socket_path.clone(), + max_upload_bytes_per_sec: 0, + max_download_bytes_per_sec: 0, + discovery: false, + wan_upload: false, + }; + + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .expect("tokio runtime"); + let _daemon = rt.spawn(varde_daemon::run(config)); + + // run() binds the socket synchronously at first poll; give it a beat. + let deadline = std::time::Instant::now() + Duration::from_secs(10); + while !socket_path.exists() && std::time::Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(20)); + } + + let output = Command::new(env!("CARGO_BIN_EXE_varde-ctl")) + .args(["--socket", socket_path.to_str().unwrap(), "status"]) + .output() + .expect("running varde-ctl"); + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(output.status.success(), "varde-ctl failed: {stderr}"); + assert!( + stdout.contains("varde-daemon"), + "unexpected output: {stdout}" + ); + + let output = Command::new(env!("CARGO_BIN_EXE_varde-ctl")) + .args([ + "--socket", + socket_path.to_str().unwrap(), + "--json", + "status", + ]) + .output() + .expect("running varde-ctl --json"); + assert!(output.status.success()); + let json: serde_json::Value = + serde_json::from_slice(&output.stdout).expect("--json output is JSON"); + assert_eq!(json["ok"], true); +} diff --git a/varde-daemon/Cargo.toml b/varde-daemon/Cargo.toml new file mode 100644 index 0000000..807ecc8 --- /dev/null +++ b/varde-daemon/Cargo.toml @@ -0,0 +1,26 @@ +[package] +name = "varde-daemon" +description = "varde: content-addressed blob mirror daemon on iroh" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true + +[[bin]] +name = "varde-daemon" +path = "src/main.rs" + +[dependencies] +varde-proto = { workspace = true } +anyhow = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true } +toml = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } + +[dev-dependencies] +tempfile = "3" diff --git a/varde-daemon/src/config.rs b/varde-daemon/src/config.rs new file mode 100644 index 0000000..d189768 --- /dev/null +++ b/varde-daemon/src/config.rs @@ -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, + socket_path: Option, + max_upload_bytes_per_sec: Option, + max_download_bytes_per_sec: Option, + discovery: Option, + wan_upload: Option, +} + +/// Values collected from CLI flags; `None` means "not given". +#[derive(Debug, Default)] +pub struct Overrides { + pub config_path: Option, + pub store_dir: Option, + pub socket_path: Option, +} + +/// 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 { + 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> { + 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> { + 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 { + 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()); + } +} diff --git a/varde-daemon/src/daemon.rs b/varde-daemon/src/daemon.rs new file mode 100644 index 0000000..9be02fe --- /dev/null +++ b/varde-daemon/src/daemon.rs @@ -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, +} + +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 { + 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:?}"), + ), + } + } +} diff --git a/varde-daemon/src/lib.rs b/varde-daemon/src/lib.rs new file mode 100644 index 0000000..e2015aa --- /dev/null +++ b/varde-daemon/src/lib.rs @@ -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 +} diff --git a/varde-daemon/src/main.rs b/varde-daemon/src/main.rs new file mode 100644 index 0000000..214e133 --- /dev/null +++ b/varde-daemon/src/main.rs @@ -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 Config file (default: /etc/varde/config.toml or + $XDG_CONFIG_HOME/varde/config.toml) + --store Store directory override + --socket 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 { + 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"), + } +} diff --git a/varde-daemon/src/server.rs b/varde-daemon/src/server.rs new file mode 100644 index 0000000..441db22 --- /dev/null +++ b/varde-daemon/src/server.rs @@ -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 { + 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) -> 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) -> 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( + 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(()) +} diff --git a/varde-daemon/tests/socket.rs b/varde-daemon/tests/socket.rs new file mode 100644 index 0000000..5165c17 --- /dev/null +++ b/varde-daemon/tests/socket.rs @@ -0,0 +1,60 @@ +//! Milestone 1: the daemon binds its socket and answers the protocol. + +mod support; + +use varde_proto::{ErrorCode, Request, ResponseData, PROTOCOL_VERSION}; + +#[test] +fn status_round_trips_on_empty_daemon() { + let daemon = support::spawn_daemon(); + let mut client = support::Client::connect(&daemon.socket); + + let reply = client.request(&Request::Status { hash: None }); + assert!(reply.ok, "status failed: {:?}", reply.error); + match reply.data { + Some(ResponseData::Status(s)) => { + assert_eq!(s.protocol, PROTOCOL_VERSION); + assert_eq!(s.pins, 0); + assert_eq!(s.store_path, daemon.store.display().to_string()); + } + other => panic!("unexpected status payload: {other:?}"), + } +} + +#[test] +fn list_is_empty_on_fresh_daemon() { + let daemon = support::spawn_daemon(); + let mut client = support::Client::connect(&daemon.socket); + + let reply = client.request(&Request::List {}); + assert!(reply.ok); + match reply.data { + Some(ResponseData::Pins { pins }) => assert!(pins.is_empty()), + other => panic!("unexpected list payload: {other:?}"), + } +} + +#[test] +fn malformed_json_gets_structured_error_and_connection_survives() { + let daemon = support::spawn_daemon(); + let mut client = support::Client::connect(&daemon.socket); + + let reply = client.send_raw("this is not json"); + assert!(!reply.ok); + assert_eq!(reply.error.expect("error body").code, ErrorCode::BadRequest); + + // The same connection keeps working after a bad request. + let reply = client.request(&Request::Status { hash: None }); + assert!(reply.ok); +} + +#[test] +fn multiple_concurrent_clients() { + let daemon = support::spawn_daemon(); + let mut a = support::Client::connect(&daemon.socket); + let mut b = support::Client::connect(&daemon.socket); + + assert!(a.request(&Request::Status { hash: None }).ok); + assert!(b.request(&Request::List {}).ok); + assert!(a.request(&Request::List {}).ok); +} diff --git a/varde-daemon/tests/support/mod.rs b/varde-daemon/tests/support/mod.rs new file mode 100644 index 0000000..54b1645 --- /dev/null +++ b/varde-daemon/tests/support/mod.rs @@ -0,0 +1,132 @@ +//! Shared helpers for integration tests: spawn the real daemon binary on +//! ephemeral paths and speak the JSON-lines protocol to it. + +use std::io::{BufRead, BufReader, Write}; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command}; +use std::time::{Duration, Instant}; + +use varde_proto::{Envelope, Request}; + +/// A running daemon process, killed on drop. +pub struct DaemonProc { + pub child: Child, + pub socket: PathBuf, + pub store: PathBuf, + // Kept alive so the store/socket dirs survive as long as the daemon. + _dir: Option, +} + +impl Drop for DaemonProc { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +/// Spawn the compiled varde-daemon on a fresh tempdir. +pub fn spawn_daemon() -> DaemonProc { + let dir = tempfile::tempdir().expect("tempdir"); + let mut proc = spawn_daemon_at(dir.path()); + proc._dir = Some(dir); + proc +} + +/// Spawn the daemon against an existing directory (for restart tests). +pub fn spawn_daemon_at(dir: &Path) -> DaemonProc { + let socket = dir.join("varde.sock"); + let store = dir.join("store"); + let child = Command::new(env!("CARGO_BIN_EXE_varde-daemon")) + .arg("--user") + .arg("--socket") + .arg(&socket) + .arg("--store") + .arg(&store) + // Isolate from any config file on the developer's machine. + .env("VARDE_CONFIG", dir.join("no-config.toml")) + .env("VARDE_DISCOVERY", "false") + .spawn() + .expect("spawning varde-daemon"); + // An explicitly named config must exist for the daemon to start. + std::fs::write(dir.join("no-config.toml"), "").expect("writing empty config"); + + let proc = DaemonProc { + child, + socket, + store, + _dir: None, + }; + wait_for_socket(&proc.socket); + proc +} + +fn wait_for_socket(path: &Path) { + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline { + if UnixStream::connect(path).is_ok() { + return; + } + std::thread::sleep(Duration::from_millis(20)); + } + panic!("daemon socket {} never appeared", path.display()); +} + +/// One connected API client. +pub struct Client { + reader: BufReader, + writer: UnixStream, +} + +impl Client { + pub fn connect(socket: &Path) -> Client { + let stream = UnixStream::connect(socket).expect("connecting to daemon"); + let reader = BufReader::new(stream.try_clone().expect("cloning stream")); + Client { + reader, + writer: stream, + } + } + + /// Send a request and read the response envelope. + pub fn request(&mut self, req: &Request) -> Envelope { + let mut line = serde_json::to_string(req).expect("serializing request"); + line.push('\n'); + self.writer + .write_all(line.as_bytes()) + .expect("writing request"); + self.read_envelope() + } + + /// Send a raw line (for malformed-input tests). + pub fn send_raw(&mut self, raw: &str) -> Envelope { + self.writer + .write_all(raw.as_bytes()) + .expect("writing raw line"); + self.writer.write_all(b"\n").expect("writing newline"); + self.read_envelope() + } + + fn read_envelope(&mut self) -> Envelope { + let mut reply = String::new(); + self.reader.read_line(&mut reply).expect("reading reply"); + assert!( + !reply.is_empty(), + "daemon closed connection without replying" + ); + serde_json::from_str(&reply).expect("parsing reply envelope") + } + + /// Read one line of the event stream (after Subscribe). + // Each test binary compiles its own copy of this module; not every + // binary uses every helper. + #[allow(dead_code)] + pub fn read_event_line(&mut self) -> Option { + let mut line = String::new(); + match self.reader.read_line(&mut line) { + Ok(0) => None, + Ok(_) => Some(line), + Err(e) => panic!("reading event line: {e}"), + } + } +} diff --git a/varde-proto/Cargo.toml b/varde-proto/Cargo.toml new file mode 100644 index 0000000..b36c868 --- /dev/null +++ b/varde-proto/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "varde-proto" +description = "Wire types for the varde socket API: requests, responses, events. No I/O." +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true + +[dependencies] +serde = { workspace = true } + +[dev-dependencies] +serde_json = { workspace = true } diff --git a/varde-proto/src/lib.rs b/varde-proto/src/lib.rs new file mode 100644 index 0000000..85a04a5 --- /dev/null +++ b/varde-proto/src/lib.rs @@ -0,0 +1,407 @@ +//! Wire types for the varde socket API. +//! +//! The protocol is JSON Lines over a unix stream socket: the client writes +//! one [`Request`] per line, the daemon answers with one [`Envelope`] per +//! line. A [`Request::Subscribe`] switches the connection into event mode, +//! after which the daemon writes one [`Event`] per line until the client +//! disconnects. +//! +//! Hashes, tickets and node ids travel as strings (BLAKE3 hex, iroh ticket +//! base32, z-base-32 NodeId respectively) so this crate stays free of iroh +//! dependencies and the protocol stays trivially consumable from any +//! language. + +use std::collections::BTreeMap; + +use serde::{Deserialize, Serialize}; + +/// Version of the socket protocol described by this crate. +/// +/// Bumped on incompatible changes; reported in [`GlobalStatus`]. +pub const PROTOCOL_VERSION: u32 = 1; + +/// A request from a client to the daemon. One JSON object per line. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum Request { + /// Import a file or directory at `path` into the store. + /// + /// `path` must be absolute (the daemon has no meaningful cwd). + /// Returns [`ResponseData::Added`] with the root hash — for a directory + /// this is the HashSeq root, for a file the blob hash. + Add { + /// Absolute path of the file or directory to import. + path: String, + /// Import directories recursively. Adding a directory with + /// `recursive: false` is an error. + #[serde(default)] + recursive: bool, + }, + /// Standing intent: keep this content, fetch it if absent, serve it to + /// peers according to `policy`. + Pin { + /// BLAKE3 hash (hex) of the blob or HashSeq root to pin. + hash: String, + /// Replication policy for this pin. + #[serde(default)] + policy: PinPolicy, + }, + /// Remove a standing pin. Does not delete data — see [`Request::Gc`]. + Unpin { + /// Hash (hex) of the pin to remove. + hash: String, + }, + /// Export content from the store to a filesystem path. + Materialize { + /// Hash (hex) of the blob or HashSeq root to export. + hash: String, + /// Absolute destination path. + dest: String, + /// How to get bytes out of the store. + #[serde(default)] + mode: MaterializeMode, + }, + /// Query daemon status, globally or for one hash. + Status { + /// If set, report on this hash; otherwise report global status. + #[serde(default, skip_serializing_if = "Option::is_none")] + hash: Option, + }, + /// List all pins with their policies and completeness. + List {}, + /// Produce an iroh blob ticket for out-of-band sharing of `hash`. + TicketExport { + /// Hash (hex) of the content to share. + hash: String, + }, + /// Import an iroh blob ticket: pin the content and fetch it from the + /// provider embedded in the ticket. + TicketImport { + /// The ticket string, as produced by [`Request::TicketExport`]. + ticket: String, + /// Policy for the pin created by the import. + #[serde(default)] + pin_policy: PinPolicy, + }, + /// Trust a peer: content is served to and fetched from trusted peers + /// only (unless a pin is `open_lan`). + PeerTrust { + /// The peer's iroh NodeId (z-base-32). + node_id: String, + }, + /// Revoke trust in a peer. + PeerUntrust { + /// The peer's iroh NodeId (z-base-32). + node_id: String, + }, + /// List trusted peers and their current connectivity. + PeerList {}, + /// Drop all blobs not reachable from a pin. Explicit, never automatic. + Gc {}, + /// Switch this connection to an event stream. The daemon acknowledges + /// with an ok envelope, then writes one [`Event`] per line. + Subscribe {}, +} + +/// Replication policy attached to a pin. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct PinPolicy { + /// Serve this content to *any* LAN peer, not just trusted ones. + /// Defaults to off: trust gates participation. + #[serde(default)] + pub open_lan: bool, +} + +/// How [`Request::Materialize`] gets bytes out of the store. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum MaterializeMode { + /// Try a reflink (`FICLONE`) first, fall back to a plain copy. + #[default] + ReflinkOrCopy, + /// Always copy bytes. + Copy, +} + +/// The envelope wrapped around every daemon reply. +/// +/// `ok: true` replies carry `data`, `ok: false` replies carry `error`. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Envelope { + /// Whether the request succeeded. + pub ok: bool, + /// Payload on success. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub data: Option, + /// Structured error on failure. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +impl Envelope { + /// Build a success envelope carrying `data`. + pub fn ok(data: ResponseData) -> Self { + Envelope { + ok: true, + data: Some(data), + error: None, + } + } + + /// Build an error envelope. + pub fn err(code: ErrorCode, message: impl Into) -> Self { + Envelope { + ok: false, + data: None, + error: Some(ApiError { + code, + message: message.into(), + }), + } + } +} + +/// A structured error. Never a bare string. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ApiError { + /// Machine-readable error class. + pub code: ErrorCode, + /// Human-readable detail. + pub message: String, +} + +/// Machine-readable error classes. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ErrorCode { + /// The request line was not valid JSON or not a known request. + BadRequest, + /// A hash, ticket or node id failed to parse. + InvalidArgument, + /// The referenced content is not in the store. + NotFound, + /// Filesystem-level failure (permissions, missing path, ...). + Io, + /// The store or endpoint failed internally. + Internal, + /// The feature exists in the protocol but this daemon build or + /// milestone does not implement it yet. + Unimplemented, +} + +/// Success payloads, tagged by response kind. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ResponseData { + /// Content was imported. Reply to [`Request::Add`]. + Added { + /// Root hash (hex) of the imported content. + hash: String, + /// Total bytes imported. + bytes: u64, + }, + /// A pin was created or updated. Reply to [`Request::Pin`] and + /// [`Request::TicketImport`]. + Pinned { + /// The pinned root hash (hex). + hash: String, + }, + /// Generic acknowledgement for requests with nothing to report + /// (`Unpin`, `PeerTrust`, `PeerUntrust`, `Subscribe` ack). + Done {}, + /// Content was exported. Reply to [`Request::Materialize`]. + Materialized { + /// Bytes written to the destination. + bytes: u64, + /// Whether at least one file was reflinked rather than copied. + reflinked: bool, + }, + /// Global daemon status. Reply to [`Request::Status`] without a hash. + Status(GlobalStatus), + /// Per-hash status. Reply to [`Request::Status`] with a hash. + HashStatus(HashStatus), + /// All pins. Reply to [`Request::List`]. + Pins { + /// One entry per pin. + pins: Vec, + }, + /// A serialized iroh ticket. Reply to [`Request::TicketExport`]. + Ticket { + /// The ticket string. + ticket: String, + }, + /// Trusted peers. Reply to [`Request::PeerList`]. + Peers { + /// One entry per trusted peer. + peers: Vec, + }, + /// Garbage collection result. Reply to [`Request::Gc`]. + GcDone { + /// Number of blobs removed. + blobs_removed: u64, + }, +} + +/// Global daemon status. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct GlobalStatus { + /// Daemon version (crate version). + pub version: String, + /// Socket protocol version, see [`PROTOCOL_VERSION`]. + pub protocol: u32, + /// This daemon's iroh NodeId (z-base-32), if the endpoint is up. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub node_id: Option, + /// Path of the blob store directory. + pub store_path: String, + /// Number of standing pins. + pub pins: u64, + /// Number of connected trusted peers. + pub connected_peers: u64, +} + +/// Status of one hash. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct HashStatus { + /// The hash (hex) this reports on. + pub hash: String, + /// Bytes present locally. + pub have_bytes: u64, + /// Total size if known (unknown while a fetch has not seen the size). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub total_bytes: Option, + /// Whether the content is fully present and verified. + pub complete: bool, + /// Whether a pin exists for this hash. + pub pinned: bool, + /// Node ids of peers currently transferring this hash with us. + pub active_peers: Vec, +} + +/// One pin as reported by [`ResponseData::Pins`]. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct PinInfo { + /// The pinned root hash (hex). + pub hash: String, + /// The pin's replication policy. + pub policy: PinPolicy, + /// Whether the content is fully present locally. + pub complete: bool, + /// Bytes present locally. + pub have_bytes: u64, + /// Total size if known. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub total_bytes: Option, +} + +/// One trusted peer as reported by [`ResponseData::Peers`]. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct PeerInfo { + /// The peer's iroh NodeId (z-base-32). + pub node_id: String, + /// Whether we currently hold a connection to this peer. + pub connected: bool, +} + +/// Events streamed after [`Request::Subscribe`]. One JSON object per line. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "event", rename_all = "snake_case")] +pub enum Event { + /// Bytes moved for a hash. + TransferProgress { + /// The hash being transferred (hex). + hash: String, + /// Direction of the transfer. + direction: Direction, + /// Bytes transferred so far. + bytes_done: u64, + /// Total bytes if known. + #[serde(default, skip_serializing_if = "Option::is_none")] + bytes_total: Option, + }, + /// A trusted peer became reachable. + PeerJoined { + /// The peer's NodeId (z-base-32). + node_id: String, + }, + /// A trusted peer went away. + PeerLeft { + /// The peer's NodeId (z-base-32). + node_id: String, + }, + /// A pin finished fetching and verified completely. + PinComplete { + /// The pinned root hash (hex). + hash: String, + }, + /// A signed announcement was received on a subscribed discovery topic. + Announcement { + /// The announced HashSeq root (hex). + root: String, + /// The announcing author's ed25519 public key (hex). + author: String, + /// Free-form metadata carried by the announcement. + meta: BTreeMap, + }, +} + +/// Transfer direction for [`Event::TransferProgress`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Direction { + /// We are fetching. + Download, + /// We are serving. + Upload, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn request_wire_shape() { + let req = Request::Add { + path: "/tmp/x".into(), + recursive: true, + }; + let json = serde_json::to_string(&req).unwrap(); + assert_eq!(json, r#"{"type":"add","path":"/tmp/x","recursive":true}"#); + let back: Request = serde_json::from_str(&json).unwrap(); + assert_eq!(back, req); + } + + #[test] + fn status_request_defaults() { + let req: Request = serde_json::from_str(r#"{"type":"status"}"#).unwrap(); + assert_eq!(req, Request::Status { hash: None }); + } + + #[test] + fn envelope_ok_carries_flag() { + let env = Envelope::ok(ResponseData::Done {}); + let json = serde_json::to_string(&env).unwrap(); + assert!(json.starts_with(r#"{"ok":true"#)); + let back: Envelope = serde_json::from_str(&json).unwrap(); + assert!(back.ok && back.error.is_none()); + } + + #[test] + fn envelope_err_is_structured() { + let env = Envelope::err(ErrorCode::NotFound, "no such hash"); + let json = serde_json::to_string(&env).unwrap(); + let back: Envelope = serde_json::from_str(&json).unwrap(); + let err = back.error.unwrap(); + assert_eq!(err.code, ErrorCode::NotFound); + assert!(!back.ok); + } + + #[test] + fn pin_policy_defaults_closed() { + let req: Request = serde_json::from_str(r#"{"type":"pin","hash":"abc"}"#).unwrap(); + match req { + Request::Pin { policy, .. } => assert!(!policy.open_lan), + _ => panic!("wrong variant"), + } + } +}