From 61171a5ea845e987a0b8cdb145e6b5086936803f Mon Sep 17 00:00:00 2001 From: Bendik Lynghaug Date: Tue, 14 Jul 2026 21:18:25 +0200 Subject: [PATCH] Milestone 3: iroh endpoint, tickets, pin-triggered fetch The daemon binds an iroh endpoint (ed25519 identity at secret.key, 0600), serves its store via Blobs/Router on the standard ALPN, and fetches with the iroh-blobs Downloader rather than the rpc client to keep quic-rpc out of the tree. Relay is disabled unless wan_upload is set: LAN-only, zero WAN upload by default. TicketImport pins first (the pin is the GC root protecting in-flight data), registers the ticket's NodeAddr with the endpoint (the downloader dials by NodeId alone), then fetches in the background; completion emits pin_complete. Tests: two-daemon localhost transfers (blob + directory collection, byte-identical), and kill -9 mid-transfer followed by restart on the same store resuming to completion. Dependencies: iroh 0.35 (endpoint/router, pairs with iroh-blobs 0.35), rand 0.8 (secret key generation, same version iroh uses). Co-Authored-By: Claude Fable 5 --- Cargo.lock | 2 + docs/milestone-3.md | 50 +++++++ varde-daemon/Cargo.toml | 2 + varde-daemon/src/daemon.rs | 63 ++++++++- varde-daemon/src/lib.rs | 1 + varde-daemon/src/main.rs | 4 +- varde-daemon/src/store.rs | 5 + varde-daemon/src/transfer.rs | 188 +++++++++++++++++++++++++ varde-daemon/tests/support/mod.rs | 8 ++ varde-daemon/tests/transfer_ops.rs | 218 +++++++++++++++++++++++++++++ 10 files changed, 532 insertions(+), 9 deletions(-) create mode 100644 docs/milestone-3.md create mode 100644 varde-daemon/src/transfer.rs create mode 100644 varde-daemon/tests/transfer_ops.rs diff --git a/Cargo.lock b/Cargo.lock index 966c447..f09f6d2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4572,9 +4572,11 @@ name = "varde-daemon" version = "0.1.0" dependencies = [ "anyhow", + "iroh", "iroh-blobs", "iroh-io", "proptest", + "rand 0.8.7", "reflink-copy", "serde", "serde_json", diff --git a/docs/milestone-3.md b/docs/milestone-3.md new file mode 100644 index 0000000..4243bc1 --- /dev/null +++ b/docs/milestone-3.md @@ -0,0 +1,50 @@ +# Milestone 3 — Transfer + +The daemon now has an iroh endpoint: it serves its store over +`iroh_blobs::ALPN` via the standard `Blobs`/`Router` stack, and fetches +pinned content with the iroh-blobs `Downloader`. `TicketExport` / +`TicketImport` complete the v1 out-of-band sharing loop; the pin created +by an import triggers the fetch from the ticket-embedded provider. + +## Decisions + +- **Endpoint identity** is an ed25519 key at `/secret.key` + (hex, mode `0600`, created on first start). Node ids surface in the + API as opaque strings in iroh's canonical encoding. +- **Relay disabled unless `wan_upload = true`.** The default posture is + LAN-only with zero WAN upload; with the relay off the endpoint is only + reachable over direct paths and never moves bytes through third-party + infrastructure. Tickets between machines that can't reach each other + directly need `wan_upload` on both ends — documented behavior, not a + bug. +- **Downloader, not the rpc client.** The `Blobs` rpc client + (`add_from_path`/`download`) would pull in the whole quic-rpc feature + surface; the underlying `Downloader` gives queueing, retries and + resume with no extra dependencies. One found-the-hard-way detail: the + downloader dials by NodeId only, so the ticket's `NodeAddr` must be + registered with the endpoint (`add_node_addr`) before queueing, or + every dial fails with "no addressing information". +- **Pin-before-fetch.** `TicketImport` records the pin (and the format + from the ticket) before the fetch starts: the pin is the GC root that + protects in-flight data, and the standing intent survives an + unreachable provider or a crash. Re-issuing the import after a crash + resumes from iroh-blobs' on-disk partial state. +- **Fetches are background tasks.** `TicketImport` replies immediately + with the pinned hash; completion is observable via `Status {hash}` + polling or the `pin_complete` event on a subscribed connection. +- **A bare `Pin` still fetches nothing** — there are no known providers + until LAN discovery lands in milestone 4. The pin is recorded and the + spec's "fetch if absent" activates when provider sources exist. +- **No serving gate yet**: this milestone serves any peer that presents + the hash (tickets are unguessable capability tokens; BLAKE3 hashes of + private content should be treated as secrets). The trust allowlist + gate is milestone 4's first change. + +## Tests + +Two-daemon localhost integration: single blob and directory collection +transfer, byte-identical after materialize on the fetcher. Crash test: +SIGKILL the fetcher ~150 ms into a 64 MB transfer, restart on the same +store, re-import the ticket, content completes and matches (proves +clean restart with partial on-disk state; on a fast machine the kill +may land after completion, which still exercises restart-with-state). diff --git a/varde-daemon/Cargo.toml b/varde-daemon/Cargo.toml index 15711e9..4f68842 100644 --- a/varde-daemon/Cargo.toml +++ b/varde-daemon/Cargo.toml @@ -13,7 +13,9 @@ path = "src/main.rs" [dependencies] varde-proto = { workspace = true } +iroh = { workspace = true } iroh-blobs = { workspace = true } +rand = "0.8" iroh-io = { workspace = true } reflink-copy = { workspace = true } anyhow = { workspace = true } diff --git a/varde-daemon/src/daemon.rs b/varde-daemon/src/daemon.rs index a1f64f9..805da93 100644 --- a/varde-daemon/src/daemon.rs +++ b/varde-daemon/src/daemon.rs @@ -13,6 +13,7 @@ use varde_proto::{ use crate::config::Config; 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. @@ -20,11 +21,13 @@ pub struct Daemon { config: Config, store: BlobStore, meta: Meta, + transfer: Transfer, events: broadcast::Sender, } impl Daemon { - /// Open the blob store and metadata under the configured store dir. + /// 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 { std::fs::create_dir_all(&config.store_dir) .with_context(|| format!("creating store directory {}", config.store_dir.display()))?; @@ -33,14 +36,23 @@ impl Daemon { // 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 { config, store, meta, + transfer, events, }) } + /// Flush the store and close the endpoint. + pub async fn shutdown(&self) { + self.transfer.shutdown().await; + self.store.shutdown().await; + } + pub fn config(&self) -> &Config { &self.config } @@ -106,7 +118,7 @@ impl Daemon { Request::Status { hash: None } => Ok(ResponseData::Status(GlobalStatus { version: env!("CARGO_PKG_VERSION").to_string(), protocol: varde_proto::PROTOCOL_VERSION, - node_id: None, + 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, @@ -143,6 +155,40 @@ impl Daemon { let blobs_removed = self.store.gc(roots).await?; Ok(ResponseData::GcDone { blobs_removed }) } + Request::TicketExport { hash } => { + let parsed = parse_hash(&hash)?; + let (_have, _total, complete) = self.store.presence(parsed).await?; + if !complete { + return Err(OpError::NotFound(format!( + "{} is not fully present; cannot export a ticket", + parsed.to_hex() + ))); + } + let content = HashAndFormat { + hash: parsed, + format: blob_format(self.meta.format_of(&parsed.to_hex())), + }; + let ticket = self.transfer.export_ticket(content).await?; + Ok(ResponseData::Ticket { ticket }) + } + Request::TicketImport { ticket, pin_policy } => { + let (hash, format, provider) = + crate::transfer::parse_ticket(&ticket).map_err(OpError::InvalidArgument)?; + let stored = match format { + BlobFormat::Raw => StoredFormat::Raw, + BlobFormat::HashSeq => StoredFormat::HashSeq, + }; + self.meta.record_format(&hash.to_hex(), stored)?; + // Pin first: the pin is the GC root that keeps the fetch's + // 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.transfer + .spawn_fetch(HashAndFormat { hash, format }, vec![provider]); + Ok(ResponseData::Pinned { + hash: hash.to_hex().to_string(), + }) + } Request::Subscribe {} => Ok(ResponseData::Done {}), other => Err(OpError::Unimplemented(format!( "not implemented yet: {other:?}" @@ -155,19 +201,22 @@ impl Daemon { let mut roots = Vec::new(); for (hash, _record) in self.meta.pins() { let parsed = parse_hash(&hash)?; - let format = match self.meta.format_of(&hash) { - StoredFormat::Raw => BlobFormat::Raw, - StoredFormat::HashSeq => BlobFormat::HashSeq, - }; roots.push(HashAndFormat { hash: parsed, - format, + format: blob_format(self.meta.format_of(&hash)), }); } Ok(roots) } } +fn blob_format(format: StoredFormat) -> BlobFormat { + match format { + StoredFormat::Raw => BlobFormat::Raw, + StoredFormat::HashSeq => BlobFormat::HashSeq, + } +} + fn parse_hash(s: &str) -> Result { Hash::from_str(s).map_err(|e| OpError::InvalidArgument(format!("invalid hash {s:?}: {e}"))) } diff --git a/varde-daemon/src/lib.rs b/varde-daemon/src/lib.rs index cdda363..4e1f577 100644 --- a/varde-daemon/src/lib.rs +++ b/varde-daemon/src/lib.rs @@ -8,6 +8,7 @@ pub mod daemon; pub mod meta; pub mod server; pub mod store; +pub mod transfer; use std::sync::Arc; diff --git a/varde-daemon/src/main.rs b/varde-daemon/src/main.rs index ed6a7a8..f0b2e2f 100644 --- a/varde-daemon/src/main.rs +++ b/varde-daemon/src/main.rs @@ -90,8 +90,8 @@ async fn run(config: Config) -> Result<()> { r = shutdown_signal() => r.map(|signal| info!(signal, "shutting down")), }; - // Flush the store and remove the socket so the next start is clean. - daemon.store().shutdown().await; + // Close the endpoint, flush the store, remove the socket. + daemon.shutdown().await; let _ = std::fs::remove_file(&socket_path); result } diff --git a/varde-daemon/src/store.rs b/varde-daemon/src/store.rs index b928124..aebb2f4 100644 --- a/varde-daemon/src/store.rs +++ b/varde-daemon/src/store.rs @@ -87,6 +87,11 @@ impl BlobStore { &self.store } + /// The local pool the transfer layer shares for non-`Send` blob work. + pub fn pool_handle(&self) -> LocalPoolHandle { + self.pool.clone() + } + /// Import a file or directory. Returns the root hash, total imported /// bytes, and the resulting format (`Raw` for files, `HashSeq` for /// directories). diff --git a/varde-daemon/src/transfer.rs b/varde-daemon/src/transfer.rs new file mode 100644 index 0000000..d32b68a --- /dev/null +++ b/varde-daemon/src/transfer.rs @@ -0,0 +1,188 @@ +//! 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. + +use std::path::Path; + +use anyhow::{Context, Result}; +use iroh::protocol::Router; +use iroh::{Endpoint, NodeAddr, RelayMode, SecretKey}; +use iroh_blobs::downloader::{DownloadRequest, Downloader}; +use iroh_blobs::net_protocol::Blobs; +use iroh_blobs::ticket::BlobTicket; +use iroh_blobs::util::local_pool::LocalPoolHandle; +use iroh_blobs::{BlobFormat, Hash, HashAndFormat}; +use tokio::sync::broadcast; +use tracing::{debug, info, warn}; +use varde_proto::Event; + +use crate::config::Config; +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. +#[derive(Debug)] +pub struct Transfer { + endpoint: Endpoint, + router: Router, + downloader: Downloader, + events: broadcast::Sender, +} + +impl Transfer { + /// Bind the endpoint and start serving the store. + /// + /// With `wan_upload` off (the default) the relay is disabled: the + /// endpoint is only reachable via direct (LAN/localhost) paths and + /// never uploads through third-party infrastructure. + pub async fn start( + config: &Config, + store: &BlobStore, + 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); + 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(); + let router = Router::builder(endpoint.clone()) + .accept(iroh_blobs::ALPN, blobs) + .spawn(); + + Ok(Transfer { + endpoint, + router, + downloader, + events, + }) + } + + /// Our stable node id. + pub fn node_id(&self) -> String { + self.endpoint.node_id().to_string() + } + + /// Produce a ticket for out-of-band sharing of `content`. + pub async fn export_ticket(&self, content: HashAndFormat) -> Result { + let addr = self + .endpoint + .node_addr() + .await + .context("waiting for endpoint address")?; + let 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. + 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". + 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"); + } + } + let request = DownloadRequest::new(content, providers); + let handle = downloader.queue(request).await; + match handle.await { + Ok(_stats) => { + info!(hash = %content.hash.to_hex(), "fetch complete"); + let _ = events.send(Event::PinComplete { + hash: content.hash.to_hex().to_string(), + }); + } + Err(e) => { + warn!(hash = %content.hash.to_hex(), error = %e, "fetch failed"); + } + } + }); + } + + /// Gracefully close the endpoint. + pub async fn shutdown(&self) { + if let Err(e) = self.router.shutdown().await { + warn!(error = %e, "router shutdown"); + } + self.endpoint.close().await; + } +} + +/// 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}"))?; + Ok((ticket.hash(), ticket.format(), ticket.node_addr().clone())) +} + +/// Load the endpoint secret key from `path`, creating it (mode 0600) on +/// first start. +fn load_or_create_secret(path: &Path) -> Result { + match std::fs::read_to_string(path) { + Ok(hex) => parse_secret(hex.trim()) + .with_context(|| format!("parsing secret key {}", path.display())), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + let secret = SecretKey::generate(rand::rngs::OsRng); + let hex = hex_encode(&secret.to_bytes()); + use std::io::Write; + use std::os::unix::fs::OpenOptionsExt; + let mut file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o600) + .open(path) + .with_context(|| format!("creating secret key {}", path.display()))?; + file.write_all(hex.as_bytes())?; + Ok(secret) + } + Err(e) => Err(e).with_context(|| format!("reading secret key {}", path.display())), + } +} + +fn parse_secret(hex: &str) -> Result { + anyhow::ensure!(hex.len() == 64, "expected 64 hex chars, got {}", hex.len()); + let mut bytes = [0u8; 32]; + for (i, byte) in bytes.iter_mut().enumerate() { + *byte = u8::from_str_radix(&hex[2 * i..2 * i + 2], 16) + .map_err(|e| anyhow::anyhow!("bad hex at {}: {e}", 2 * i))?; + } + Ok(SecretKey::from_bytes(&bytes)) +} + +fn hex_encode(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn secret_key_round_trips_through_file() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("secret.key"); + let first = load_or_create_secret(&path).unwrap(); + let second = load_or_create_secret(&path).unwrap(); + assert_eq!(first.to_bytes(), second.to_bytes()); + + use std::os::unix::fs::PermissionsExt; + let mode = std::fs::metadata(&path).unwrap().permissions().mode(); + assert_eq!(mode & 0o777, 0o600, "secret key must be 0600"); + } +} diff --git a/varde-daemon/tests/support/mod.rs b/varde-daemon/tests/support/mod.rs index 6967a87..03b24c7 100644 --- a/varde-daemon/tests/support/mod.rs +++ b/varde-daemon/tests/support/mod.rs @@ -22,6 +22,14 @@ pub struct DaemonProc { _dir: Option, } +impl DaemonProc { + /// SIGKILL the daemon — no cleanup handlers run, simulating a crash. + pub fn kill_dash_nine(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + impl Drop for DaemonProc { fn drop(&mut self) { let _ = self.child.kill(); diff --git a/varde-daemon/tests/transfer_ops.rs b/varde-daemon/tests/transfer_ops.rs new file mode 100644 index 0000000..787b3cf --- /dev/null +++ b/varde-daemon/tests/transfer_ops.rs @@ -0,0 +1,218 @@ +//! Milestone 3: ticket export/import between two real daemons over +//! localhost, and crash-resume behavior. + +mod support; + +use std::path::Path; +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 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); +} + +/// Poll per-hash status until the content is complete or the deadline +/// passes. +fn wait_complete(socket: &Path, hash: &str, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + let mut client = support::Client::connect(socket); + let reply = client.request(&Request::Status { + hash: Some(hash.to_string()), + }); + if reply.ok { + if let Some(ResponseData::HashStatus(s)) = reply.data { + if s.complete { + return true; + } + } + } + std::thread::sleep(Duration::from_millis(100)); + } + false +} + +#[test] +fn ticket_transfers_between_two_daemons() { + 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(); + + // Distinctive, incompressible-ish payload big enough to span many + // chunks. + let payload: Vec = (0..2_000_000u32).flat_map(|i| i.to_le_bytes()).collect(); + let src = dir.path().join("shared.bin"); + std::fs::write(&src, &payload).unwrap(); + + let hash = add_file(&mut provider_client, &src); + // Pin on the provider so the content is a deliberate, GC-proof share. + let reply = provider_client.request(&Request::Pin { + hash: hash.clone(), + policy: PinPolicy::default(), + }); + assert!(reply.ok); + + let ticket = export_ticket(&mut provider_client, &hash); + import_ticket(&mut fetcher_client, &ticket); + + assert!( + wait_complete(&fetcher.socket, &hash, Duration::from_secs(60)), + "fetch did not complete in time" + ); + + // The fetched bytes materialize identically on the second daemon. + let dest = dir.path().join("fetched.bin"); + let reply = fetcher_client.request(&Request::Materialize { + hash: hash.clone(), + dest: dest.display().to_string(), + mode: MaterializeMode::Copy, + }); + assert!(reply.ok, "materialize failed: {:?}", reply.error); + assert_eq!(std::fs::read(&dest).unwrap(), payload); + + // The pin created by the import shows up in list, complete. + let reply = fetcher_client.request(&Request::List {}); + match reply.data { + Some(ResponseData::Pins { pins }) => { + assert_eq!(pins.len(), 1); + assert_eq!(pins[0].hash, hash); + assert!(pins[0].complete); + } + other => panic!("unexpected list reply: {other:?}"), + } +} + +#[test] +fn directory_ticket_transfers_collection() { + 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 tree = dir.path().join("tree"); + std::fs::create_dir_all(tree.join("nested")).unwrap(); + std::fs::write(tree.join("a.txt"), b"hello varde").unwrap(); + std::fs::write(tree.join("nested/b.bin"), vec![3u8; 100_000]).unwrap(); + + let reply = provider_client.request(&Request::Add { + path: tree.display().to_string(), + recursive: true, + }); + let hash = match reply.data { + Some(ResponseData::Added { hash, .. }) => hash, + other => panic!("unexpected add reply: {other:?}"), + }; + let reply = provider_client.request(&Request::Pin { + hash: hash.clone(), + policy: PinPolicy::default(), + }); + assert!(reply.ok); + + let ticket = export_ticket(&mut provider_client, &hash); + import_ticket(&mut fetcher_client, &ticket); + assert!( + wait_complete(&fetcher.socket, &hash, Duration::from_secs(60)), + "fetch did not complete in time" + ); + + let dest = dir.path().join("out"); + let reply = fetcher_client.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.join("a.txt")).unwrap(), b"hello varde"); + assert_eq!( + std::fs::read(dest.join("nested/b.bin")).unwrap(), + vec![3u8; 100_000] + ); +} + +/// The daemon must survive `kill -9` mid-transfer and resume cleanly on +/// restart: iroh-blobs keeps partial state on disk, and re-issuing the +/// import after restart completes the content. +#[test] +fn kill_dash_nine_mid_transfer_then_resume() { + let provider = support::spawn_daemon(); + let mut provider_client = support::Client::connect(&provider.socket); + let dir = tempfile::tempdir().unwrap(); + + // Big enough that the transfer takes long enough to interrupt. + let payload: Vec = (0..16_000_000u32).flat_map(|i| i.to_le_bytes()).collect(); + let src = dir.path().join("big.bin"); + std::fs::write(&src, &payload).unwrap(); + let hash = add_file(&mut provider_client, &src); + let reply = provider_client.request(&Request::Pin { + hash: hash.clone(), + policy: PinPolicy::default(), + }); + assert!(reply.ok); + let ticket = export_ticket(&mut provider_client, &hash); + + // Fetcher gets its own persistent dir that survives the kill. + let fetcher_dir = tempfile::tempdir().unwrap(); + let mut fetcher = support::spawn_daemon_at(fetcher_dir.path()); + let mut fetcher_client = support::Client::connect(&fetcher.socket); + import_ticket(&mut fetcher_client, &ticket); + + // SIGKILL shortly after the fetch starts. If the machine is fast + // enough that the transfer already finished, the test still proves a + // clean restart with intact state. + std::thread::sleep(Duration::from_millis(150)); + fetcher.kill_dash_nine(); + + // Restart on the same store; the socket file is stale but the daemon + // replaces it. Re-issue the import to resume. + let fetcher = support::spawn_daemon_at(fetcher_dir.path()); + let mut fetcher_client = support::Client::connect(&fetcher.socket); + import_ticket(&mut fetcher_client, &ticket); + + assert!( + wait_complete(&fetcher.socket, &hash, Duration::from_secs(120)), + "fetch did not complete after restart" + ); + let dest = dir.path().join("resumed.bin"); + let reply = fetcher_client.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, + "resumed content differs" + ); +}