diff --git a/Cargo.lock b/Cargo.lock index f09f6d2..835dba5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,20 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "acto" +version = "0.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a026259da4f1a13b4af60cda453c392de64c58c12d239c560923e0382f42f2b9" +dependencies = [ + "parking_lot", + "pin-project-lite", + "rustc_version", + "smol_str", + "tokio", + "tracing", +] + [[package]] name = "aead" version = "0.5.2" @@ -1764,6 +1778,7 @@ dependencies = [ "strum", "stun-rs", "surge-ping", + "swarm-discovery", "thiserror 2.0.18", "time", "tokio", @@ -3761,6 +3776,12 @@ dependencies = [ "serde", ] +[[package]] +name = "smol_str" +version = "0.1.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fad6c857cbab2627dcf01ec85a623ca4e7dcb5691cbaa3d7fb7653671f0d09c9" + [[package]] name = "snafu" version = "0.8.9" @@ -3949,6 +3970,21 @@ dependencies = [ "tracing", ] +[[package]] +name = "swarm-discovery" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3a95032b94c1dc318f55e0b130e3d2176cda022310a65c3df0092764ea69562" +dependencies = [ + "acto", + "anyhow", + "hickory-proto", + "rand 0.8.7", + "socket2 0.5.10", + "tokio", + "tracing", +] + [[package]] name = "syn" version = "1.0.109" @@ -4572,9 +4608,13 @@ name = "varde-daemon" version = "0.1.0" dependencies = [ "anyhow", + "async-channel", + "bytes", + "futures-lite", "iroh", "iroh-blobs", "iroh-io", + "n0-future", "proptest", "rand 0.8.7", "reflink-copy", diff --git a/docs/milestone-4.md b/docs/milestone-4.md new file mode 100644 index 0000000..b672be6 --- /dev/null +++ b/docs/milestone-4.md @@ -0,0 +1,55 @@ +# Milestone 4 — LAN discovery, trust, rate limiting, events + +## What landed + +- **LAN discovery** via iroh's `MdnsDiscovery` (the documented `_iroh` + mDNS service — identifiable on the network, not anonymous noise), + enabled by the `discovery` config flag (default on). Sightings stream + to the daemon, which tracks peer presence and reacts based on trust. +- **Trust gate.** The provider is now our own `ProtocolHandler`: + connections from trusted NodeIds get the full store; everyone else + gets a filtered view that answers "not found" for anything outside the + openly-served set. Verified by the forged-ticket test: an untrusted + peer that knows a private hash gets nothing until trusted. +- **Openly-served set** = `open_lan` pins + ticket-exported hashes, plus + their HashSeq children, kept in memory and recomputed on pin changes + and fetch completion. +- **Auto-sync.** When a trusted peer appears on the LAN, every + incomplete pin is queued for download from them (`pin` also fetches + immediately if a trusted peer is present — "fetch if absent" is now + real). Verified end-to-end over actual mdns between two daemons in the + test suite; the test skips gracefully where multicast is unavailable. +- **Token-bucket rate limiters** wrap the store itself (`ShapedStore` + implements the full iroh-blobs `Store` trait): provider data reads pay + into the upload bucket, downloader batch writes pay into the download + bucket. Upload cap 0 disables serving outright (connections closed at + accept). Enforcement is proven by wall-clock in the test suite. +- **Event stream** is live: `transfer_progress` (both directions), + `peer_joined`, `pin_complete`. + +## Decisions and trade-offs + +- **Ticket export grants standing serve-consent for that hash** + (persisted in `meta.json` as `exported`). Handing out a ticket *is* + deliberate publication, and without this either every ticket import + would require the exporter to pre-trust the importer, or tickets + would silently stop working the moment the trust gate exists. This is + the documented consent ladder: trusted peers see everything, + open_lan/exported content is public, everything else is private. +- **Trust is evaluated at connection accept.** Revoking (or granting) + trust applies to *new* connections; an existing connection keeps its + view until it closes. Cheap, predictable, and the window is bounded by + QUIC idle timeouts. Request-level re-checks would need a fork of + iroh-blobs' provider internals (`ResponseWriter` is not constructible + from outside the crate). +- **Gating by filtered store view** (get → None) means an untrusted + peer probing a private hash can't distinguish "refused" from "absent" + — deliberately, that's the less chatty posture. +- **"Connected" for peers means "seen via LAN discovery in the last two + minutes"** — presence, not an open QUIC connection. Good enough for + `peer list` and the status count; honest about what it measures. +- **`fetch only from trusted peers`**: auto-sync only ever dials trusted + peers. Ticket import dials the ticket's provider — importing the + ticket is the explicit consent for that single fetch. +- Download progress events report per-child offsets for collections; + good enough for progress bars, not an exact byte ledger. diff --git a/varde-daemon/Cargo.toml b/varde-daemon/Cargo.toml index 4f68842..beed125 100644 --- a/varde-daemon/Cargo.toml +++ b/varde-daemon/Cargo.toml @@ -13,9 +13,15 @@ path = "src/main.rs" [dependencies] varde-proto = { workspace = true } -iroh = { workspace = true } +iroh = { workspace = true, features = ["discovery-local-network"] } iroh-blobs = { workspace = true } rand = "0.8" +# The next three exist to interoperate with iroh-blobs trait signatures +# and channels; all are already in the dependency tree via iroh. +n0-future = "0.1" +async-channel = "2" +futures-lite = "2" +bytes = "1" iroh-io = { workspace = true } reflink-copy = { workspace = true } anyhow = { workspace = true } @@ -30,3 +36,4 @@ tracing-subscriber = { workspace = true } [dev-dependencies] tempfile = "3" proptest = "1" +tokio = { workspace = true, features = ["test-util"] } diff --git a/varde-daemon/src/daemon.rs b/varde-daemon/src/daemon.rs index 805da93..6780f83 100644 --- a/varde-daemon/src/daemon.rs +++ b/varde-daemon/src/daemon.rs @@ -1,13 +1,17 @@ //! Request dispatch and daemon state. +use std::collections::{BTreeMap, BTreeSet}; use std::path::PathBuf; use std::str::FromStr; +use std::sync::{Arc, Mutex, RwLock}; +use std::time::{Duration, Instant}; use anyhow::Context; use iroh_blobs::{BlobFormat, Hash, HashAndFormat}; use tokio::sync::broadcast; +use tracing::{debug, info, warn}; use varde_proto::{ - Envelope, ErrorCode, Event, GlobalStatus, HashStatus, PinInfo, Request, ResponseData, + Envelope, ErrorCode, Event, GlobalStatus, HashStatus, PeerInfo, PinInfo, Request, ResponseData, }; use crate::config::Config; @@ -15,36 +19,97 @@ use crate::meta::{Meta, StoredFormat}; use crate::store::{BlobStore, OpError}; use crate::transfer::Transfer; -/// Shared daemon state. Handlers grow with the milestones; anything not -/// yet wired returns [`ErrorCode::Unimplemented`] rather than pretending. +/// A LAN peer is considered present if discovery has seen it within this +/// window (mdns re-announces well below this). +const PEER_PRESENCE_WINDOW: Duration = Duration::from_secs(120); + +/// Suppress duplicate PeerJoined events within this window. +const PEER_REJOIN_WINDOW: Duration = Duration::from_secs(60); + +/// Shared daemon state. pub struct Daemon { config: Config, store: BlobStore, - meta: Meta, + meta: Arc, transfer: Transfer, events: broadcast::Sender, + /// Hashes servable to untrusted peers (open_lan pins, exported + /// tickets, and their hashseq children). Read by the provider filter + /// on every untrusted request. + open_set: Arc>>, + /// LAN peers seen via discovery: node id -> last sighting. + lan_peers: Mutex>, } impl Daemon { /// Open the blob store and metadata under the configured store dir, - /// and bring up the iroh endpoint. - pub async fn open(config: Config) -> anyhow::Result { + /// bring up the iroh endpoint, and start the discovery/sync loops. + pub async fn open(config: Config) -> anyhow::Result> { std::fs::create_dir_all(&config.store_dir) .with_context(|| format!("creating store directory {}", config.store_dir.display()))?; let store = BlobStore::open(&config.store_dir).await?; - let meta = Meta::load(&config.store_dir)?; + let meta = Arc::new(Meta::load(&config.store_dir)?); // Capacity bounds memory if a subscriber stalls; laggards get a // Lagged error, not unbounded buffering. let (events, _) = broadcast::channel(1024); - let transfer = - Transfer::start(&config, &store, store.pool_handle(), events.clone()).await?; - Ok(Daemon { + + let open_set: Arc>> = Arc::default(); + let filter_set = open_set.clone(); + let open_filter: crate::shaped::ServeFilter = + Arc::new(move |hash| filter_set.read().expect("open set lock").contains(hash)); + + let transfer = Transfer::start( + &config, + &store, + meta.clone(), + open_filter, + store.pool_handle(), + events.clone(), + ) + .await?; + + let daemon = Arc::new(Daemon { config, store, meta, transfer, events, - }) + open_set, + lan_peers: Mutex::new(BTreeMap::new()), + }); + daemon.recompute_open_set().await; + + // Completed fetches can turn hashseq roots expandable; refresh + // the open set whenever a pin finishes. + let weak = Arc::downgrade(&daemon); + let mut pin_events = daemon.events.subscribe(); + tokio::spawn(async move { + loop { + match pin_events.recv().await { + Ok(Event::PinComplete { .. }) => { + let Some(daemon) = weak.upgrade() else { break }; + daemon.recompute_open_set().await; + } + Ok(_) => {} + Err(broadcast::error::RecvError::Lagged(_)) => {} + Err(broadcast::error::RecvError::Closed) => break, + } + } + }); + + // Discovery loop: track LAN peer sightings, emit join events for + // trusted peers and sync any incomplete pins from them. + if let Some(mut discovered) = daemon.transfer.take_discovery() { + let weak = Arc::downgrade(&daemon); + tokio::spawn(async move { + while let Some(node_id) = discovered.recv().await { + let Some(daemon) = weak.upgrade() else { break }; + daemon.on_peer_seen(node_id).await; + } + }); + } + + Ok(daemon) } /// Flush the store and close the endpoint. @@ -61,15 +126,103 @@ impl Daemon { &self.store } - pub fn meta(&self) -> &Meta { - &self.meta - } - /// New receiver for the daemon event stream. pub fn subscribe_events(&self) -> broadcast::Receiver { self.events.subscribe() } + /// Rebuild the set of hashes servable to untrusted peers. + async fn recompute_open_set(&self) { + let mut open = BTreeSet::new(); + let mut roots: Vec = self + .meta + .pins() + .into_iter() + .filter(|(_, record)| record.policy.open_lan) + .map(|(hash, _)| hash) + .collect(); + roots.extend(self.meta.exported()); + + for hex in roots { + let Ok(hash) = Hash::from_str(&hex) else { + continue; + }; + open.insert(hash); + if self.meta.format_of(&hex) == StoredFormat::HashSeq { + match self.store.hashseq_children(hash).await { + Ok(children) => open.extend(children), + Err(e) => warn!(hash = %hex, error = %e, "expanding open hashseq"), + } + } + } + debug!(hashes = open.len(), "open set recomputed"); + *self.open_set.write().expect("open set lock") = open; + } + + /// Handle a discovery sighting of `node_id`. + async fn on_peer_seen(&self, node_id: iroh::NodeId) { + let id = node_id.to_string(); + let rejoined = { + let mut peers = self.lan_peers.lock().expect("lan peers lock"); + let now = Instant::now(); + let fresh = match peers.get(&id) { + Some(last) => now.duration_since(*last) > PEER_REJOIN_WINDOW, + None => true, + }; + peers.insert(id.clone(), now); + fresh + }; + if !self.meta.is_trusted(&id) { + return; + } + if rejoined { + info!(peer = %id, "trusted peer present on LAN"); + let _ = self.events.send(Event::PeerJoined { node_id: id }); + } + self.sync_pins_from(node_id).await; + } + + /// Queue fetches for every incomplete pin from a trusted LAN peer. + /// The downloader dedupes by content, so repeated sightings are + /// cheap; discovery resolves the peer's addresses. + async fn sync_pins_from(&self, node_id: iroh::NodeId) { + for (hex, _record) in self.meta.pins() { + let Ok(hash) = Hash::from_str(&hex) else { + continue; + }; + match self.store.presence(hash).await { + Ok((_, _, true)) => {} // complete, nothing to do + Ok(_) => { + debug!(hash = %hex, peer = %node_id, "syncing incomplete pin"); + self.transfer.spawn_fetch( + HashAndFormat { + hash, + format: blob_format(self.meta.format_of(&hex)), + }, + vec![iroh::NodeAddr::new(node_id)], + ); + } + Err(e) => warn!(hash = %hex, error = %e, "checking pin presence"), + } + } + } + + /// Trusted peers seen on the LAN recently. + fn present_trusted_peers(&self) -> Vec { + let peers = self.lan_peers.lock().expect("lan peers lock"); + let now = Instant::now(); + self.meta + .trusted_peers() + .into_iter() + .map(|node_id| { + let connected = peers + .get(&node_id) + .is_some_and(|last| now.duration_since(*last) < PEER_PRESENCE_WINDOW); + PeerInfo { node_id, connected } + }) + .collect() + } + /// Handle one API request. Infallible at this level: all failures are /// mapped to structured error envelopes. pub async fn handle(&self, request: Request) -> Envelope { @@ -93,8 +246,26 @@ impl Daemon { Request::Pin { hash, policy } => { let parsed = parse_hash(&hash)?; self.meta.pin(&parsed.to_hex(), policy)?; - // Fetching absent pinned content arrives with the - // transfer milestone. + self.recompute_open_set().await; + // Fetch if absent: try every trusted peer currently + // present on the LAN. + if let Ok((_, _, false)) = self.store.presence(parsed).await { + let present: Vec = self + .present_trusted_peers() + .into_iter() + .filter(|p| p.connected) + .filter_map(|p| p.node_id.parse().ok().map(iroh::NodeAddr::new)) + .collect(); + if !present.is_empty() { + self.transfer.spawn_fetch( + HashAndFormat { + hash: parsed, + format: blob_format(self.meta.format_of(&parsed.to_hex())), + }, + present, + ); + } + } Ok(ResponseData::Pinned { hash: parsed.to_hex().to_string(), }) @@ -104,6 +275,7 @@ impl Daemon { if !self.meta.unpin(&parsed.to_hex())? { return Err(OpError::NotFound(format!("no pin for {}", parsed.to_hex()))); } + self.recompute_open_set().await; Ok(ResponseData::Done {}) } Request::Materialize { hash, dest, mode } => { @@ -115,14 +287,21 @@ impl Daemon { .await?; Ok(ResponseData::Materialized { bytes, reflinked }) } - Request::Status { hash: None } => Ok(ResponseData::Status(GlobalStatus { - version: env!("CARGO_PKG_VERSION").to_string(), - protocol: varde_proto::PROTOCOL_VERSION, - node_id: Some(self.transfer.node_id()), - store_path: self.config.store_dir.display().to_string(), - pins: self.meta.pins().len() as u64, - connected_peers: 0, - })), + Request::Status { hash: None } => { + let connected = self + .present_trusted_peers() + .iter() + .filter(|p| p.connected) + .count() as u64; + Ok(ResponseData::Status(GlobalStatus { + version: env!("CARGO_PKG_VERSION").to_string(), + protocol: varde_proto::PROTOCOL_VERSION, + node_id: Some(self.transfer.node_id()), + store_path: self.config.store_dir.display().to_string(), + pins: self.meta.pins().len() as u64, + connected_peers: connected, + })) + } Request::Status { hash: Some(hash) } => { let parsed = parse_hash(&hash)?; let (have_bytes, total_bytes, complete) = self.store.presence(parsed).await?; @@ -169,6 +348,10 @@ impl Daemon { format: blob_format(self.meta.format_of(&parsed.to_hex())), }; let ticket = self.transfer.export_ticket(content).await?; + // Exporting a ticket is deliberate publication: the + // holder must be able to fetch without being trusted. + self.meta.record_exported(&parsed.to_hex())?; + self.recompute_open_set().await; Ok(ResponseData::Ticket { ticket }) } Request::TicketImport { ticket, pin_policy } => { @@ -183,16 +366,44 @@ impl Daemon { // data alive, and it is the standing intent even if the // provider is unreachable right now. self.meta.pin(&hash.to_hex(), pin_policy)?; + self.recompute_open_set().await; self.transfer .spawn_fetch(HashAndFormat { hash, format }, vec![provider]); Ok(ResponseData::Pinned { hash: hash.to_hex().to_string(), }) } + Request::PeerTrust { node_id } => { + let parsed: iroh::NodeId = node_id + .parse() + .map_err(|e| OpError::InvalidArgument(format!("invalid node id: {e}")))?; + self.meta.trust_peer(&parsed.to_string())?; + info!(peer = %parsed, "peer trusted"); + // If the peer is already visible on the LAN, sync now. + let present = { + let peers = self.lan_peers.lock().expect("lan peers lock"); + peers.get(&parsed.to_string()).is_some_and(|last| { + Instant::now().duration_since(*last) < PEER_PRESENCE_WINDOW + }) + }; + if present { + self.sync_pins_from(parsed).await; + } + Ok(ResponseData::Done {}) + } + Request::PeerUntrust { node_id } => { + let parsed: iroh::NodeId = node_id + .parse() + .map_err(|e| OpError::InvalidArgument(format!("invalid node id: {e}")))?; + if !self.meta.untrust_peer(&parsed.to_string())? { + return Err(OpError::NotFound(format!("{parsed} was not trusted"))); + } + Ok(ResponseData::Done {}) + } + Request::PeerList {} => Ok(ResponseData::Peers { + peers: self.present_trusted_peers(), + }), Request::Subscribe {} => Ok(ResponseData::Done {}), - other => Err(OpError::Unimplemented(format!( - "not implemented yet: {other:?}" - ))), } } diff --git a/varde-daemon/src/lib.rs b/varde-daemon/src/lib.rs index 4e1f577..5a80090 100644 --- a/varde-daemon/src/lib.rs +++ b/varde-daemon/src/lib.rs @@ -7,11 +7,10 @@ pub mod config; pub mod daemon; pub mod meta; pub mod server; +pub mod shaped; pub mod store; pub mod transfer; -use std::sync::Arc; - use anyhow::Result; /// Run a daemon with the given config until the future is dropped or the @@ -19,7 +18,7 @@ use anyhow::Result; /// once this future has been polled the socket path exists. pub async fn run(config: config::Config) -> Result<()> { let socket_path = config.socket_path.clone(); - let daemon = Arc::new(daemon::Daemon::open(config).await?); + let daemon = daemon::Daemon::open(config).await?; let listener = server::bind_socket(&socket_path)?; server::serve(listener, daemon).await } diff --git a/varde-daemon/src/main.rs b/varde-daemon/src/main.rs index f0b2e2f..c0275e3 100644 --- a/varde-daemon/src/main.rs +++ b/varde-daemon/src/main.rs @@ -1,7 +1,6 @@ //! varde-daemon: content-addressed blob mirror daemon. use std::path::PathBuf; -use std::sync::Arc; use anyhow::{bail, Context, Result}; use tracing::info; @@ -82,7 +81,7 @@ fn main() -> Result<()> { async fn run(config: Config) -> Result<()> { let socket_path = config.socket_path.clone(); - let daemon = Arc::new(daemon::Daemon::open(config).await?); + let daemon = daemon::Daemon::open(config).await?; let listener = server::bind_socket(&socket_path)?; let result = tokio::select! { diff --git a/varde-daemon/src/meta.rs b/varde-daemon/src/meta.rs index 9f55189..a857b22 100644 --- a/varde-daemon/src/meta.rs +++ b/varde-daemon/src/meta.rs @@ -40,6 +40,10 @@ struct MetaState { /// Trusted peer NodeIds (z-base-32). #[serde(default)] trusted_peers: BTreeSet, + /// Hashes (hex) the user shared via ticket export: standing consent + /// to serve them to anyone, like an `open_lan` pin. + #[serde(default)] + exported: BTreeSet, } /// Handle to the metadata file. Cheap to share behind an `Arc`. @@ -125,6 +129,23 @@ impl Meta { .unwrap_or(StoredFormat::Raw) } + /// Record standing consent to serve `hash` to anyone (set by ticket + /// export — sharing a ticket is deliberate publication). + pub fn record_exported(&self, hash: &str) -> Result<()> { + self.mutate(|s| { + s.exported.insert(hash.to_string()); + }) + } + + /// Hashes with standing serve-to-anyone consent. + pub fn exported(&self) -> BTreeSet { + self.state + .lock() + .expect("meta lock poisoned") + .exported + .clone() + } + /// Add a trusted peer. Returns false if it was already trusted. pub fn trust_peer(&self, node_id: &str) -> Result { self.mutate(|s| s.trusted_peers.insert(node_id.to_string())) diff --git a/varde-daemon/src/shaped.rs b/varde-daemon/src/shaped.rs new file mode 100644 index 0000000..4fa4f15 --- /dev/null +++ b/varde-daemon/src/shaped.rs @@ -0,0 +1,438 @@ +//! Traffic shaping and per-hash gating for the blob store. +//! +//! [`ShapedStore`] wraps the iroh-blobs fs store and implements the full +//! `Store` trait, adding two token buckets: reads of entry data (the +//! bytes a provider sends to peers) draw from the upload bucket, batch +//! writes of downloaded data draw from the download bucket. An optional +//! filter hides hashes from `get`, which is how untrusted peers are +//! limited to openly-served content — the provider simply answers "not +//! found" for everything else. + +use std::future::Future; +use std::io; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use iroh_blobs::store::bao_tree::io::fsm::{BaoContentItem, Outboard}; +use iroh_blobs::store::fs::Store as FsStore; +use iroh_blobs::store::{ + BaoBatchWriter, ConsistencyCheckProgress, DbIter, EntryStatus, ExportMode, ExportProgressCb, + ImportMode, ImportProgress, Map, MapEntry, MapEntryMut, MapMut, ReadableStore, Store, +}; +use iroh_blobs::util::progress::{BoxedProgressSender, IdGenerator, ProgressSender}; +use iroh_blobs::util::Tag; +use iroh_blobs::{BlobFormat, Hash, HashAndFormat, TempTag}; +use iroh_io::AsyncSliceReader; + +/// An async token bucket. Rate 0 means unlimited. +#[derive(Debug)] +pub struct TokenBucket { + rate: u64, + burst: f64, + state: std::sync::Mutex, +} + +#[derive(Debug)] +struct BucketState { + tokens: f64, + last_refill: Instant, +} + +impl TokenBucket { + pub fn new(rate_bytes_per_sec: u64) -> Arc { + // Burst of one second's worth, floored so tiny rates still make + // progress chunk by chunk. + let burst = (rate_bytes_per_sec as f64).max(64.0 * 1024.0); + Arc::new(TokenBucket { + rate: rate_bytes_per_sec, + burst, + state: std::sync::Mutex::new(BucketState { + tokens: burst, + last_refill: Instant::now(), + }), + }) + } + + /// Take `n` tokens, sleeping until the bucket has refilled enough. + /// Debt-based: the acquire always succeeds immediately in bookkeeping + /// terms, and the caller sleeps off any deficit — this keeps large + /// writes simple while converging on the configured rate. + pub async fn acquire(&self, n: usize) { + if self.rate == 0 { + return; + } + let wait = { + let mut state = self.state.lock().expect("bucket lock poisoned"); + let now = Instant::now(); + let elapsed = now.duration_since(state.last_refill).as_secs_f64(); + state.last_refill = now; + state.tokens = (state.tokens + elapsed * self.rate as f64).min(self.burst); + state.tokens -= n as f64; + if state.tokens >= 0.0 { + None + } else { + Some(Duration::from_secs_f64(-state.tokens / self.rate as f64)) + } + }; + if let Some(wait) = wait { + tokio::time::sleep(wait).await; + } + } +} + +/// Predicate deciding whether a hash may be served on this store view. +pub type ServeFilter = Arc bool + Send + Sync>; + +/// The fs store wrapped with rate limiting and an optional serve filter. +#[derive(Clone)] +pub struct ShapedStore { + inner: FsStore, + upload: Arc, + download: Arc, + filter: Option, +} + +impl std::fmt::Debug for ShapedStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ShapedStore") + .field("inner", &self.inner) + .field("upload", &self.upload) + .field("download", &self.download) + .field("filtered", &self.filter.is_some()) + .finish() + } +} + +impl ShapedStore { + pub fn new( + inner: FsStore, + upload: Arc, + download: Arc, + ) -> ShapedStore { + ShapedStore { + inner, + upload, + download, + filter: None, + } + } + + /// A view of the same store that only serves hashes passing `filter`. + pub fn filtered(&self, filter: ServeFilter) -> ShapedStore { + ShapedStore { + inner: self.inner.clone(), + upload: self.upload.clone(), + download: self.download.clone(), + filter: Some(filter), + } + } +} + +/// An entry whose data reads draw from the upload bucket. +#[derive(Debug, Clone)] +pub struct ShapedEntry { + inner: ::Entry, + upload: Arc, +} + +impl MapEntry for ShapedEntry { + fn hash(&self) -> Hash { + self.inner.hash() + } + + fn size(&self) -> iroh_blobs::store::BaoBlobSize { + self.inner.size() + } + + fn is_complete(&self) -> bool { + self.inner.is_complete() + } + + async fn outboard(&self) -> io::Result { + // The fs entry's inherent (synchronous) methods shadow the trait + // methods here and below. + self.inner.outboard() + } + + async fn data_reader(&self) -> io::Result { + let inner = self.inner.data_reader(); + Ok(ThrottledReader { + inner, + bucket: self.upload.clone(), + }) + } +} + +/// AsyncSliceReader that pays for bytes read from the upload bucket. +#[derive(Debug)] +struct ThrottledReader { + inner: R, + bucket: Arc, +} + +impl AsyncSliceReader for ThrottledReader { + async fn read_at(&mut self, offset: u64, len: usize) -> io::Result { + let bytes = self.inner.read_at(offset, len).await?; + self.bucket.acquire(bytes.len()).await; + Ok(bytes) + } + + async fn size(&mut self) -> io::Result { + self.inner.size().await + } +} + +impl Map for ShapedStore { + type Entry = ShapedEntry; + + async fn get(&self, hash: &Hash) -> io::Result> { + if let Some(filter) = &self.filter { + if !filter(hash) { + return Ok(None); + } + } + Ok(self.inner.get(hash).await?.map(|inner| ShapedEntry { + inner, + upload: self.upload.clone(), + })) + } +} + +/// A mutable entry whose batch writes draw from the download bucket. +#[derive(Debug, Clone)] +pub struct ShapedEntryMut { + inner: ::EntryMut, + upload: Arc, + download: Arc, +} + +impl MapEntry for ShapedEntryMut { + fn hash(&self) -> Hash { + self.inner.hash() + } + + fn size(&self) -> iroh_blobs::store::BaoBlobSize { + self.inner.size() + } + + fn is_complete(&self) -> bool { + self.inner.is_complete() + } + + async fn outboard(&self) -> io::Result { + // The fs entry's inherent (synchronous) methods shadow the trait + // methods here and below. + self.inner.outboard() + } + + async fn data_reader(&self) -> io::Result { + let inner = self.inner.data_reader(); + Ok(ThrottledReader { + inner, + bucket: self.upload.clone(), + }) + } +} + +impl MapEntryMut for ShapedEntryMut { + async fn batch_writer(&self) -> io::Result { + let inner = self.inner.batch_writer().await?; + Ok(ThrottledBatchWriter { + inner, + bucket: self.download.clone(), + }) + } +} + +/// BaoBatchWriter that pays for leaf bytes written from the download +/// bucket. +struct ThrottledBatchWriter { + inner: W, + bucket: Arc, +} + +impl BaoBatchWriter for ThrottledBatchWriter { + async fn write_batch(&mut self, size: u64, batch: Vec) -> io::Result<()> { + let bytes: usize = batch + .iter() + .map(|item| match item { + BaoContentItem::Leaf(leaf) => leaf.data.len(), + // Parent nodes are two hashes of overhead. + BaoContentItem::Parent(_) => 64, + }) + .sum(); + self.bucket.acquire(bytes).await; + self.inner.write_batch(size, batch).await + } + + async fn sync(&mut self) -> io::Result<()> { + self.inner.sync().await + } +} + +impl MapMut for ShapedStore { + type EntryMut = ShapedEntryMut; + + async fn get_mut(&self, hash: &Hash) -> io::Result> { + Ok(self.inner.get_mut(hash).await?.map(|inner| ShapedEntryMut { + inner, + upload: self.upload.clone(), + download: self.download.clone(), + })) + } + + async fn get_or_create(&self, hash: Hash, size: u64) -> io::Result { + let inner = self.inner.get_or_create(hash, size).await?; + Ok(ShapedEntryMut { + inner, + upload: self.upload.clone(), + download: self.download.clone(), + }) + } + + async fn entry_status(&self, hash: &Hash) -> io::Result { + self.inner.entry_status(hash).await + } + + fn entry_status_sync(&self, hash: &Hash) -> io::Result { + self.inner.entry_status_sync(hash) + } + + async fn insert_complete(&self, entry: ShapedEntryMut) -> io::Result<()> { + self.inner.insert_complete(entry.inner).await + } +} + +impl ReadableStore for ShapedStore { + async fn blobs(&self) -> io::Result> { + self.inner.blobs().await + } + + async fn tags( + &self, + from: Option, + to: Option, + ) -> io::Result> { + self.inner.tags(from, to).await + } + + fn temp_tags(&self) -> Box + Send + Sync + 'static> { + self.inner.temp_tags() + } + + async fn consistency_check( + &self, + repair: bool, + tx: BoxedProgressSender, + ) -> io::Result<()> { + self.inner.consistency_check(repair, tx).await + } + + async fn partial_blobs(&self) -> io::Result> { + self.inner.partial_blobs().await + } + + async fn export( + &self, + hash: Hash, + target: PathBuf, + mode: ExportMode, + progress: ExportProgressCb, + ) -> io::Result<()> { + self.inner.export(hash, target, mode, progress).await + } +} + +impl Store for ShapedStore { + async fn import_file( + &self, + data: PathBuf, + mode: ImportMode, + format: BlobFormat, + progress: impl ProgressSender + IdGenerator, + ) -> io::Result<(TempTag, u64)> { + self.inner.import_file(data, mode, format, progress).await + } + + async fn import_bytes(&self, bytes: bytes::Bytes, format: BlobFormat) -> io::Result { + self.inner.import_bytes(bytes, format).await + } + + async fn import_stream( + &self, + data: impl futures_lite::Stream> + Send + Unpin + 'static, + format: BlobFormat, + progress: impl ProgressSender + IdGenerator, + ) -> io::Result<(TempTag, u64)> { + self.inner.import_stream(data, format, progress).await + } + + async fn set_tag(&self, name: Tag, hash: HashAndFormat) -> io::Result<()> { + self.inner.set_tag(name, hash).await + } + + async fn rename_tag(&self, from: Tag, to: Tag) -> io::Result<()> { + self.inner.rename_tag(from, to).await + } + + async fn delete_tags(&self, from: Option, to: Option) -> io::Result<()> { + self.inner.delete_tags(from, to).await + } + + async fn create_tag(&self, hash: HashAndFormat) -> io::Result { + self.inner.create_tag(hash).await + } + + fn temp_tag(&self, value: HashAndFormat) -> TempTag { + self.inner.temp_tag(value) + } + + async fn gc_run(&self, config: iroh_blobs::store::GcConfig, protected_cb: G) + where + G: Fn() -> Gut, + Gut: Future> + Send, + { + self.inner.gc_run(config, protected_cb).await + } + + async fn delete(&self, hashes: Vec) -> io::Result<()> { + self.inner.delete(hashes).await + } + + async fn shutdown(&self) { + self.inner.shutdown().await + } + + async fn sync(&self) -> io::Result<()> { + self.inner.sync().await + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test(start_paused = true)] + async fn bucket_enforces_rate() { + let bucket = TokenBucket::new(1_000_000); // 1 MB/s, 1 MB burst + let start = tokio::time::Instant::now(); + // First megabyte is burst, the next three must take ~3 seconds. + for _ in 0..4 { + bucket.acquire(1_000_000).await; + } + let elapsed = start.elapsed(); + assert!( + elapsed >= Duration::from_secs(2), + "4 MB at 1 MB/s finished too fast: {elapsed:?}" + ); + } + + #[tokio::test] + async fn zero_rate_is_unlimited() { + let bucket = TokenBucket::new(0); + let start = Instant::now(); + bucket.acquire(usize::MAX / 2).await; + assert!(start.elapsed() < Duration::from_millis(50)); + } +} diff --git a/varde-daemon/src/store.rs b/varde-daemon/src/store.rs index aebb2f4..c64af6c 100644 --- a/varde-daemon/src/store.rs +++ b/varde-daemon/src/store.rs @@ -313,6 +313,25 @@ impl BlobStore { } } + /// The child hashes of a complete HashSeq root (empty for absent or + /// partial roots). + pub async fn hashseq_children(&self, root: Hash) -> Result, OpError> { + self.on_pool(move |this| async move { + let Some(entry) = this.store.get(&root).await? else { + return Ok(Vec::new()); + }; + if !entry.is_complete() { + return Ok(Vec::new()); + } + let mut reader = entry.data_reader(); + let bytes = reader.read_to_end().await?; + let seq = HashSeq::try_from(bytes) + .map_err(|e| OpError::Internal(anyhow::anyhow!("invalid hashseq: {e}")))?; + Ok(seq.iter().collect()) + }) + .await + } + /// Drop every blob not reachable from `roots` (mark and sweep). /// In-flight imports are protected by their temp tags; tags stored in /// the blob database are honored too. diff --git a/varde-daemon/src/transfer.rs b/varde-daemon/src/transfer.rs index d32b68a..90608e3 100644 --- a/varde-daemon/src/transfer.rs +++ b/varde-daemon/src/transfer.rs @@ -1,35 +1,106 @@ //! iroh endpoint, blob provider and downloader. //! -//! This module is also the seam where transport-level traffic shaping -//! lives: the rate limiters (milestone 4) and, later, scavenger -//! congestion control would slot in here without touching the daemon -//! logic above. +//! This module is the transport seam: rate limiting lives in +//! [`crate::shaped::ShapedStore`] wired up here, and a scavenger +//! congestion controller would later slot in at the same place without +//! touching the daemon logic above. use std::path::Path; +use std::sync::Arc; use anyhow::{Context, Result}; -use iroh::protocol::Router; -use iroh::{Endpoint, NodeAddr, RelayMode, SecretKey}; +use iroh::discovery::mdns::MdnsDiscovery; +use iroh::endpoint::Connection; +use iroh::protocol::{ProtocolHandler, Router}; +use iroh::{Endpoint, NodeAddr, NodeId, RelayMode, SecretKey}; use iroh_blobs::downloader::{DownloadRequest, Downloader}; -use iroh_blobs::net_protocol::Blobs; -use iroh_blobs::ticket::BlobTicket; +use iroh_blobs::get::db::DownloadProgress; +use iroh_blobs::provider::{handle_connection, CustomEventSender, EventSender}; use iroh_blobs::util::local_pool::LocalPoolHandle; +use iroh_blobs::util::progress::AsyncChannelProgressSender; use iroh_blobs::{BlobFormat, Hash, HashAndFormat}; -use tokio::sync::broadcast; +use n0_future::boxed::BoxFuture; +use n0_future::StreamExt; +use tokio::sync::{broadcast, mpsc}; use tracing::{debug, info, warn}; -use varde_proto::Event; +use varde_proto::{Direction, Event}; use crate::config::Config; +use crate::meta::Meta; +use crate::shaped::{ServeFilter, ShapedStore, TokenBucket}; use crate::store::BlobStore; /// The network side of the daemon: one iroh endpoint serving the blob -/// store, plus a downloader for fetching pinned content from providers. +/// store (trust-gated, rate-limited), plus a downloader for fetching +/// pinned content from providers. #[derive(Debug)] pub struct Transfer { endpoint: Endpoint, router: Router, downloader: Downloader, events: broadcast::Sender, + discovery_rx: std::sync::Mutex>>, +} + +/// Serves blob requests with trust gating: trusted peers see the whole +/// store, untrusted peers see only the openly-served subset (and get +/// "not found" for the rest). With serving disabled (upload cap 0) every +/// incoming connection is closed immediately. +#[derive(Debug, Clone)] +struct GatedProvider { + full_view: ShapedStore, + open_view: ShapedStore, + meta: Arc, + events: EventSender, + pool: LocalPoolHandle, + serving_enabled: bool, +} + +impl ProtocolHandler for GatedProvider { + fn accept(&self, connection: Connection) -> BoxFuture> { + let this = self.clone(); + Box::pin(async move { + let remote = connection.remote_node_id()?; + if !this.serving_enabled { + debug!(%remote, "serving disabled, closing connection"); + connection.close(0u32.into(), b"serving disabled"); + return Ok(()); + } + let trusted = this.meta.is_trusted(&remote.to_string()); + debug!(%remote, trusted, "incoming blobs connection"); + if trusted { + handle_connection(connection, this.full_view, this.events, this.pool).await; + } else { + handle_connection(connection, this.open_view, this.events, this.pool).await; + } + Ok(()) + }) + } +} + +/// Forwards provider transfer events into the daemon event stream. +#[derive(Debug)] +struct UploadEvents(broadcast::Sender); + +impl CustomEventSender for UploadEvents { + fn send(&self, event: iroh_blobs::provider::Event) -> BoxFuture<()> { + self.try_send(event); + Box::pin(async {}) + } + + fn try_send(&self, event: iroh_blobs::provider::Event) { + if let iroh_blobs::provider::Event::TransferProgress { + hash, end_offset, .. + } = event + { + let _ = self.0.send(Event::TransferProgress { + hash: hash.to_hex().to_string(), + direction: Direction::Upload, + bytes_done: end_offset, + bytes_total: None, + }); + } + } } impl Transfer { @@ -41,23 +112,57 @@ impl Transfer { pub async fn start( config: &Config, store: &BlobStore, + meta: Arc, + open_filter: ServeFilter, pool: LocalPoolHandle, events: broadcast::Sender, ) -> Result { let secret = load_or_create_secret(&config.store_dir.join("secret.key"))?; - let mut builder = Endpoint::builder().secret_key(secret); + let mut builder = Endpoint::builder().secret_key(secret.clone()); if !config.wan_upload { builder = builder.relay_mode(RelayMode::Disabled); } - let endpoint = builder.bind().await.context("binding iroh endpoint")?; - info!(node_id = %endpoint.node_id(), "endpoint up"); - let blobs = Blobs::builder(store.inner().clone()) - .local_pool(pool) - .build(&endpoint); - let downloader = blobs.downloader().clone(); + // LAN discovery: advertise under iroh's mDNS-style local + // discovery service (identifiable `_iroh` mdns records, not + // anonymous noise). Discovered node ids stream to the daemon, + // which decides what to do based on trust. + let (discovery_tx, discovery_rx) = mpsc::channel(64); + if config.discovery { + let mdns = + MdnsDiscovery::new(secret.public()).context("starting mdns local discovery")?; + if let Some(mut stream) = iroh::discovery::Discovery::subscribe(&mdns) { + tokio::spawn(async move { + while let Some(item) = stream.next().await { + if discovery_tx.send(item.node_id()).await.is_err() { + break; + } + } + }); + } + builder = builder.discovery(Box::new(mdns)); + } + + let endpoint = builder.bind().await.context("binding iroh endpoint")?; + info!(node_id = %endpoint.node_id(), discovery = config.discovery, "endpoint up"); + + let upload_bucket = TokenBucket::new(config.max_upload_bytes_per_sec); + let download_bucket = TokenBucket::new(config.max_download_bytes_per_sec); + let shaped = ShapedStore::new(store.inner().clone(), upload_bucket, download_bucket); + + let provider_events = EventSender::new(Some(Arc::new(UploadEvents(events.clone())))); + let provider = GatedProvider { + full_view: shaped.clone(), + open_view: shaped.filtered(open_filter), + meta, + events: provider_events, + pool: pool.clone(), + serving_enabled: config.max_upload_bytes_per_sec > 0, + }; + + let downloader = Downloader::new(shaped, endpoint.clone(), pool); let router = Router::builder(endpoint.clone()) - .accept(iroh_blobs::ALPN, blobs) + .accept(iroh_blobs::ALPN, provider) .spawn(); Ok(Transfer { @@ -65,6 +170,7 @@ impl Transfer { router, downloader, events, + discovery_rx: std::sync::Mutex::new(Some(discovery_rx)), }) } @@ -73,6 +179,13 @@ impl Transfer { self.endpoint.node_id().to_string() } + /// Take the stream of locally discovered node ids. Yields each + /// sighting (mdns re-announces periodically); the daemon dedupes. + /// Callable once; returns None afterwards or with discovery off. + pub fn take_discovery(&self) -> Option> { + self.discovery_rx.lock().expect("discovery lock").take() + } + /// Produce a ticket for out-of-band sharing of `content`. pub async fn export_ticket(&self, content: HashAndFormat) -> Result { let addr = self @@ -80,27 +193,38 @@ impl Transfer { .node_addr() .await .context("waiting for endpoint address")?; - let ticket = BlobTicket::new(addr, content.hash, content.format)?; + let ticket = iroh_blobs::ticket::BlobTicket::new(addr, content.hash, content.format)?; Ok(ticket.to_string()) } /// Queue a background fetch of `content` from `providers`. Emits - /// [`Event::PinComplete`] when the content is fully verified locally. + /// progress events and [`Event::PinComplete`] when fully verified. pub fn spawn_fetch(&self, content: HashAndFormat, providers: Vec) { let downloader = self.downloader.clone(); let events = self.events.clone(); let endpoint = self.endpoint.clone(); tokio::spawn(async move { // The downloader dials by NodeId; the endpoint must be taught - // the provider's direct addresses first or the dial fails - // with "no addressing information". + // any direct addresses we know (ticket-embedded ones), or the + // dial fails with "no addressing information". Id-only + // provider entries resolve via discovery instead. for provider in &providers { - debug!(node = %provider.node_id, addrs = ?provider.direct_addresses, "provider"); - if let Err(e) = endpoint.add_node_addr(provider.clone()) { - warn!(node = %provider.node_id, error = %e, "adding provider address"); + if !provider.direct_addresses.is_empty() || provider.relay_url.is_some() { + if let Err(e) = endpoint.add_node_addr(provider.clone()) { + debug!(node = %provider.node_id, error = %e, "adding provider address"); + } } } - let request = DownloadRequest::new(content, providers); + + let (progress_tx, progress_rx) = async_channel::bounded(64); + tokio::spawn(forward_download_progress( + content.hash, + progress_rx, + events.clone(), + )); + + let request = DownloadRequest::new(content, providers) + .progress_sender(AsyncChannelProgressSender::new(progress_tx)); let handle = downloader.queue(request).await; match handle.await { Ok(_stats) => { @@ -125,9 +249,36 @@ impl Transfer { } } +/// Translate iroh-blobs download progress into daemon transfer events. +async fn forward_download_progress( + root: Hash, + rx: async_channel::Receiver, + events: broadcast::Sender, +) { + let mut current_total = None; + while let Ok(progress) = rx.recv().await { + match progress { + DownloadProgress::Found { size, .. } => { + current_total = Some(size); + } + DownloadProgress::Progress { offset, .. } => { + let _ = events.send(Event::TransferProgress { + hash: root.to_hex().to_string(), + direction: Direction::Download, + bytes_done: offset, + bytes_total: current_total, + }); + } + DownloadProgress::AllDone(_) | DownloadProgress::Abort(_) => break, + _ => {} + } + } +} + /// Parse a ticket string into (root, format, provider address). pub fn parse_ticket(ticket: &str) -> Result<(Hash, BlobFormat, NodeAddr), String> { - let ticket: BlobTicket = ticket.parse().map_err(|e| format!("invalid ticket: {e}"))?; + let ticket: iroh_blobs::ticket::BlobTicket = + ticket.parse().map_err(|e| format!("invalid ticket: {e}"))?; Ok((ticket.hash(), ticket.format(), ticket.node_addr().clone())) } diff --git a/varde-daemon/tests/lan_trust.rs b/varde-daemon/tests/lan_trust.rs new file mode 100644 index 0000000..1dcb776 --- /dev/null +++ b/varde-daemon/tests/lan_trust.rs @@ -0,0 +1,287 @@ +//! Milestone 4: trust gating, rate limiting, event streaming, and +//! mdns-based auto-sync between trusted daemons. + +mod support; + +use std::path::Path; +use std::str::FromStr; +use std::time::{Duration, Instant}; + +use varde_proto::{MaterializeMode, PinPolicy, Request, ResponseData}; + +fn add_file(client: &mut support::Client, path: &Path) -> String { + let reply = client.request(&Request::Add { + path: path.display().to_string(), + recursive: false, + }); + assert!(reply.ok, "add failed: {:?}", reply.error); + match reply.data { + Some(ResponseData::Added { hash, .. }) => hash, + other => panic!("unexpected add reply: {other:?}"), + } +} + +fn pin(client: &mut support::Client, hash: &str, open_lan: bool) { + let reply = client.request(&Request::Pin { + hash: hash.to_string(), + policy: PinPolicy { open_lan }, + }); + assert!(reply.ok, "pin failed: {:?}", reply.error); +} + +fn export_ticket(client: &mut support::Client, hash: &str) -> String { + let reply = client.request(&Request::TicketExport { + hash: hash.to_string(), + }); + assert!(reply.ok, "ticket export failed: {:?}", reply.error); + match reply.data { + Some(ResponseData::Ticket { ticket }) => ticket, + other => panic!("unexpected ticket reply: {other:?}"), + } +} + +fn import_ticket(client: &mut support::Client, ticket: &str) { + let reply = client.request(&Request::TicketImport { + ticket: ticket.to_string(), + pin_policy: PinPolicy::default(), + }); + assert!(reply.ok, "ticket import failed: {:?}", reply.error); +} + +fn node_id(client: &mut support::Client) -> String { + let reply = client.request(&Request::Status { hash: None }); + match reply.data { + Some(ResponseData::Status(s)) => s.node_id.expect("daemon has a node id"), + other => panic!("unexpected status reply: {other:?}"), + } +} + +fn is_complete(socket: &Path, hash: &str) -> bool { + let mut client = support::Client::connect(socket); + let reply = client.request(&Request::Status { + hash: Some(hash.to_string()), + }); + matches!( + reply.data, + Some(ResponseData::HashStatus(s)) if s.complete + ) +} + +fn wait_complete(socket: &Path, hash: &str, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if is_complete(socket, hash) { + return true; + } + std::thread::sleep(Duration::from_millis(100)); + } + false +} + +/// Re-target a ticket at another hash on the same provider: what an +/// attacker who learned a hash out of band would hand to their daemon. +fn forge_ticket(real_ticket: &str, target_hash: &str) -> String { + let ticket = iroh_blobs::ticket::BlobTicket::from_str(real_ticket).unwrap(); + let hash = iroh_blobs::Hash::from_str(target_hash).unwrap(); + iroh_blobs::ticket::BlobTicket::new( + ticket.node_addr().clone(), + hash, + iroh_blobs::BlobFormat::Raw, + ) + .unwrap() + .to_string() +} + +#[test] +fn trust_gates_serving_per_hash() { + let provider = support::spawn_daemon(); + let fetcher_dir = tempfile::tempdir().unwrap(); + let fetcher = support::spawn_daemon_at(fetcher_dir.path()); + let mut provider_client = support::Client::connect(&provider.socket); + let mut fetcher_client = support::Client::connect(&fetcher.socket); + let dir = tempfile::tempdir().unwrap(); + + // `shared` is deliberately published via ticket; `private` is pinned + // but never shared. + std::fs::write(dir.path().join("shared.bin"), vec![1u8; 300_000]).unwrap(); + std::fs::write(dir.path().join("private.bin"), vec![2u8; 300_000]).unwrap(); + let shared = add_file(&mut provider_client, &dir.path().join("shared.bin")); + let private = add_file(&mut provider_client, &dir.path().join("private.bin")); + pin(&mut provider_client, &shared, false); + pin(&mut provider_client, &private, false); + + let ticket = export_ticket(&mut provider_client, &shared); + + // The exported content transfers to the untrusted fetcher. + import_ticket(&mut fetcher_client, &ticket); + assert!( + wait_complete(&fetcher.socket, &shared, Duration::from_secs(60)), + "exported content must be fetchable without trust" + ); + + // A forged ticket for the never-shared hash must not: the provider + // answers "not found" to untrusted peers. + let forged = forge_ticket(&ticket, &private); + import_ticket(&mut fetcher_client, &forged); + std::thread::sleep(Duration::from_secs(5)); + assert!( + !is_complete(&fetcher.socket, &private), + "unshared content leaked to an untrusted peer" + ); + + // Once the provider trusts the fetcher, the same request succeeds. + // Trust is evaluated when a connection is accepted, so restart the + // fetcher (same store, same identity) to shed the connection that + // was established while untrusted. + let fetcher_id = node_id(&mut fetcher_client); + let reply = provider_client.request(&Request::PeerTrust { + node_id: fetcher_id, + }); + assert!(reply.ok, "peer trust failed: {:?}", reply.error); + drop(fetcher); + let fetcher = support::spawn_daemon_at(fetcher_dir.path()); + let mut fetcher_client = support::Client::connect(&fetcher.socket); + import_ticket(&mut fetcher_client, &forged); + assert!( + wait_complete(&fetcher.socket, &private, Duration::from_secs(60)), + "trusted peer should be served everything" + ); +} + +#[test] +fn upload_rate_limit_is_enforced() { + let dir = tempfile::tempdir().unwrap(); + // 256 KiB/s upload cap on the provider. + let provider = support::spawn_daemon_with_env(dir.path(), &[("VARDE_MAX_UPLOAD", "262144")]); + let fetcher = support::spawn_daemon(); + let mut provider_client = support::Client::connect(&provider.socket); + let mut fetcher_client = support::Client::connect(&fetcher.socket); + + let payload = vec![7u8; 1_000_000]; + let src = dir.path().join("capped.bin"); + std::fs::write(&src, &payload).unwrap(); + let hash = add_file(&mut provider_client, &src); + pin(&mut provider_client, &hash, false); + let ticket = export_ticket(&mut provider_client, &hash); + + let start = Instant::now(); + import_ticket(&mut fetcher_client, &ticket); + assert!( + wait_complete(&fetcher.socket, &hash, Duration::from_secs(120)), + "capped fetch never completed" + ); + let elapsed = start.elapsed(); + // 1 MB at 256 KiB/s is ~4 s minus one 256 KiB burst: ~3 s. Anything + // under 1.5 s means the limiter did nothing (uncapped localhost + // moves this in well under 200 ms). + assert!( + elapsed >= Duration::from_millis(1500), + "1 MB fetch finished in {elapsed:?} despite 256 KiB/s upload cap" + ); +} + +#[test] +fn subscribe_streams_progress_and_completion() { + let provider = support::spawn_daemon(); + let fetcher = support::spawn_daemon(); + let mut provider_client = support::Client::connect(&provider.socket); + let mut fetcher_client = support::Client::connect(&fetcher.socket); + let dir = tempfile::tempdir().unwrap(); + + let src = dir.path().join("watched.bin"); + std::fs::write(&src, vec![5u8; 2_000_000]).unwrap(); + let hash = add_file(&mut provider_client, &src); + pin(&mut provider_client, &hash, false); + let ticket = export_ticket(&mut provider_client, &hash); + + // Subscribe on the fetcher before starting the transfer. + let mut events = support::Client::connect(&fetcher.socket); + let ack = events.request(&Request::Subscribe {}); + assert!(ack.ok); + events.set_read_timeout(Duration::from_secs(60)); + + import_ticket(&mut fetcher_client, &ticket); + + let mut saw_progress = false; + let mut saw_complete = false; + while let Some(line) = events.read_event_line() { + let event: serde_json::Value = serde_json::from_str(&line).expect("event is JSON"); + match event["event"].as_str() { + Some("transfer_progress") => { + assert_eq!(event["hash"].as_str(), Some(hash.as_str())); + assert_eq!(event["direction"].as_str(), Some("download")); + saw_progress = true; + } + Some("pin_complete") => { + assert_eq!(event["hash"].as_str(), Some(hash.as_str())); + saw_complete = true; + break; + } + _ => {} + } + } + assert!(saw_progress, "no transfer_progress event seen"); + assert!(saw_complete, "no pin_complete event seen"); +} + +/// Two trusted daemons with LAN discovery on: a pin that exists on one +/// side auto-syncs to the other with no ticket exchange. Requires +/// working multicast; skips (with a note) where mdns can't see the +/// sibling process. +#[test] +fn overlapping_pins_auto_sync_via_lan_discovery() { + let dir_a = tempfile::tempdir().unwrap(); + let dir_b = tempfile::tempdir().unwrap(); + let a = support::spawn_daemon_with_env(dir_a.path(), &[("VARDE_DISCOVERY", "true")]); + let b = support::spawn_daemon_with_env(dir_b.path(), &[("VARDE_DISCOVERY", "true")]); + let mut client_a = support::Client::connect(&a.socket); + let mut client_b = support::Client::connect(&b.socket); + + // Mutual trust: A serves B, B auto-fetches from A. + let id_a = node_id(&mut client_a); + let id_b = node_id(&mut client_b); + assert!(client_a.request(&Request::PeerTrust { node_id: id_b }).ok); + assert!(client_b.request(&Request::PeerTrust { node_id: id_a }).ok); + + // Content pinned on A only. + let dir = tempfile::tempdir().unwrap(); + let src = dir.path().join("synced.bin"); + let payload = vec![9u8; 500_000]; + std::fs::write(&src, &payload).unwrap(); + let hash = add_file(&mut client_a, &src); + pin(&mut client_a, &hash, false); + + // B pins the same hash without having the bytes. + pin(&mut client_b, &hash, false); + + // Wait for B to see A via discovery at all. + let deadline = Instant::now() + Duration::from_secs(30); + let mut discovered = false; + while Instant::now() < deadline { + let reply = client_b.request(&Request::PeerList {}); + if let Some(ResponseData::Peers { peers }) = reply.data { + if peers.iter().any(|p| p.connected) { + discovered = true; + break; + } + } + std::thread::sleep(Duration::from_millis(250)); + } + if !discovered { + eprintln!("SKIP: mdns discovery saw no peers (multicast unavailable in this environment)"); + return; + } + + assert!( + wait_complete(&b.socket, &hash, Duration::from_secs(60)), + "pinned content did not auto-sync from trusted LAN peer" + ); + let dest = dir.path().join("synced-out.bin"); + let reply = client_b.request(&Request::Materialize { + hash, + dest: dest.display().to_string(), + mode: MaterializeMode::Copy, + }); + assert!(reply.ok, "materialize failed: {:?}", reply.error); + assert_eq!(std::fs::read(&dest).unwrap(), payload); +} diff --git a/varde-daemon/tests/support/mod.rs b/varde-daemon/tests/support/mod.rs index 03b24c7..4cffdf0 100644 --- a/varde-daemon/tests/support/mod.rs +++ b/varde-daemon/tests/support/mod.rs @@ -47,9 +47,15 @@ pub fn spawn_daemon() -> DaemonProc { /// Spawn the daemon against an existing directory (for restart tests). pub fn spawn_daemon_at(dir: &Path) -> DaemonProc { + spawn_daemon_with_env(dir, &[]) +} + +/// Spawn the daemon with extra environment variables (config overrides). +pub fn spawn_daemon_with_env(dir: &Path, envs: &[(&str, &str)]) -> DaemonProc { let socket = dir.join("varde.sock"); let store = dir.join("store"); - let child = Command::new(env!("CARGO_BIN_EXE_varde-daemon")) + let mut command = Command::new(env!("CARGO_BIN_EXE_varde-daemon")); + command .arg("--user") .arg("--socket") .arg(&socket) @@ -57,9 +63,11 @@ pub fn spawn_daemon_at(dir: &Path) -> DaemonProc { .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"); + .env("VARDE_DISCOVERY", "false"); + for (key, value) in envs { + command.env(key, value); + } + let child = command.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"); @@ -129,6 +137,13 @@ impl Client { serde_json::from_str(&reply).expect("parsing reply envelope") } + /// Bound how long reads may block (applies to the shared socket). + pub fn set_read_timeout(&mut self, timeout: Duration) { + self.writer + .set_read_timeout(Some(timeout)) + .expect("setting read timeout"); + } + /// Read one line of the event stream (after Subscribe). pub fn read_event_line(&mut self) -> Option { let mut line = String::new();