From a97f0a04b34edfa067e8540a7bed9450434fee6d Mon Sep 17 00:00:00 2001 From: Ultradesu Date: Fri, 10 Jul 2026 15:10:03 +0300 Subject: [PATCH] added example --- Cargo.lock | 226 ++++++- Cargo.toml | 15 +- README.md | 9 + apps/artist-dht-cli/Cargo.toml | 16 + apps/artist-dht-cli/src/main.rs | 403 ++++++++++++ crates/artist-dht/Cargo.toml | 27 + crates/artist-dht/README.md | 152 +++++ crates/artist-dht/src/config.rs | 181 ++++++ crates/artist-dht/src/database.rs | 480 +++++++++++++++ crates/artist-dht/src/dht.rs | 274 +++++++++ crates/artist-dht/src/error.rs | 65 ++ crates/artist-dht/src/lib.rs | 92 +++ crates/artist-dht/src/message.rs | 174 ++++++ crates/artist-dht/src/node.rs | 808 +++++++++++++++++++++++++ crates/artist-dht/src/normalization.rs | 93 +++ crates/artist-dht/src/record.rs | 300 +++++++++ crates/artist-dht/src/request.rs | 170 ++++++ crates/artist-dht/src/routing.rs | 315 ++++++++++ crates/artist-dht/src/service.rs | 417 +++++++++++++ crates/artist-dht/tests/integration.rs | 238 ++++++++ crates/federation-net/src/engine.rs | 5 + 21 files changed, 4456 insertions(+), 4 deletions(-) create mode 100644 apps/artist-dht-cli/Cargo.toml create mode 100644 apps/artist-dht-cli/src/main.rs create mode 100644 crates/artist-dht/Cargo.toml create mode 100644 crates/artist-dht/README.md create mode 100644 crates/artist-dht/src/config.rs create mode 100644 crates/artist-dht/src/database.rs create mode 100644 crates/artist-dht/src/dht.rs create mode 100644 crates/artist-dht/src/error.rs create mode 100644 crates/artist-dht/src/lib.rs create mode 100644 crates/artist-dht/src/message.rs create mode 100644 crates/artist-dht/src/node.rs create mode 100644 crates/artist-dht/src/normalization.rs create mode 100644 crates/artist-dht/src/record.rs create mode 100644 crates/artist-dht/src/request.rs create mode 100644 crates/artist-dht/src/routing.rs create mode 100644 crates/artist-dht/src/service.rs create mode 100644 crates/artist-dht/tests/integration.rs diff --git a/Cargo.lock b/Cargo.lock index ac2302a..464d349 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -138,6 +138,41 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" +[[package]] +name = "artist-dht" +version = "0.1.0" +dependencies = [ + "anyhow", + "blake3", + "data-encoding", + "federation-net", + "futures", + "iroh", + "postcard", + "rand 0.9.4", + "rusqlite", + "serde", + "tempfile", + "thiserror 2.0.18", + "tokio", + "tracing", + "unicode-normalization", + "uuid", +] + +[[package]] +name = "artist-dht-cli" +version = "0.1.0" +dependencies = [ + "anyhow", + "artist-dht", + "clap", + "federation-net", + "rustyline", + "tokio", + "tracing-subscriber", +] + [[package]] name = "async-trait" version = "0.1.89" @@ -379,6 +414,15 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" +[[package]] +name = "clipboard-win" +version = "5.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bde03770d3df201d4fb868f2c9c59e66a3e4e2bd06692a0fe701e7103c7e84d4" +dependencies = [ + "error-code", +] + [[package]] name = "cmov" version = "0.5.4" @@ -811,6 +855,12 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" +[[package]] +name = "endian-type" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "869b0adbda23651a9c5c0c3d270aac9fcb52e8622a8f2b17e57802d7791962f2" + [[package]] name = "enum-assoc" version = "1.3.0" @@ -838,6 +888,24 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "error-code" +version = "3.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dea2df4cf52843e0452895c455a1a2cfbb842a1e7329671acf418fdc53ed4c59" + +[[package]] +name = "fallible-iterator" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649" + +[[package]] +name = "fallible-streaming-iterator" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" + [[package]] name = "fastrand" version = "2.4.1" @@ -1136,6 +1204,15 @@ dependencies = [ "byteorder", ] +[[package]] +name = "hashbrown" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +dependencies = [ + "foldhash", +] + [[package]] name = "hashbrown" version = "0.17.1" @@ -1147,6 +1224,15 @@ dependencies = [ "foldhash", ] +[[package]] +name = "hashlink" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32069d97bb81e38fa67eab65e3393bf804bb85969f2bc06bf13f64aef5aba248" +dependencies = [ + "hashbrown 0.17.1", +] + [[package]] name = "heapless" version = "0.7.17" @@ -1244,6 +1330,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "home" +version = "0.5.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc627f471c528ff0c4a49e1d5e60450c8f6461dd6d10ba9dcd3a61d3dff7728d" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "http" version = "1.4.2" @@ -1523,7 +1618,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" dependencies = [ "equivalent", - "hashbrown", + "hashbrown 0.17.1", ] [[package]] @@ -1847,6 +1942,17 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libsqlite3-sys" +version = "0.38.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6c19a05435c21ac299d71b6a9c13db3e3f47c520517d58990a462a1397a61db" +dependencies = [ + "cc", + "pkg-config", + "vcpkg", +] + [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -1893,7 +1999,7 @@ version = "0.18.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b6180140927ee907000b0aa540091f6ea512ead4447c92b8fc35bc72788a5a6" dependencies = [ - "hashbrown", + "hashbrown 0.17.1", ] [[package]] @@ -2122,6 +2228,27 @@ dependencies = [ "wmi", ] +[[package]] +name = "nibble_vec" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77a5d83df9f36fe23f0c3648c6bbb8b0298bb5f1939c8f2704431371f4b84d43" +dependencies = [ + "smallvec", +] + +[[package]] +name = "nix" +version = "0.31.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d" +dependencies = [ + "bitflags", + "cfg-if", + "cfg_aliases", + "libc", +] + [[package]] name = "noq" version = "1.0.1" @@ -2463,6 +2590,12 @@ dependencies = [ "spki", ] +[[package]] +name = "pkg-config" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" + [[package]] name = "plist" version = "1.8.0" @@ -2634,6 +2767,16 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "radix_trie" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b4431027dcd37fc2a73ef740b5f233aa805897935b8bce0195e41bbf9a3289a" +dependencies = [ + "endian-type", + "nibble_vec", +] + [[package]] name = "rand" version = "0.9.4" @@ -2772,6 +2915,31 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rsqlite-vfs" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c51c9ae4df8a7fba42103df5c621fa3c37eccf3a3c650879e90fc48b11cc192c" +dependencies = [ + "hashbrown 0.16.1", + "thiserror 2.0.18", +] + +[[package]] +name = "rusqlite" +version = "0.40.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11438310b19e3109b6446c33d1ed5e889428cf2e278407bc7896bc4aaea43323" +dependencies = [ + "bitflags", + "fallible-iterator", + "fallible-streaming-iterator", + "hashlink", + "libsqlite3-sys", + "smallvec", + "sqlite-wasm-rs", +] + [[package]] name = "rustc-hash" version = "2.1.3" @@ -2881,6 +3049,27 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" +[[package]] +name = "rustyline" +version = "18.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53f6a737db68eb1a8ccff86b584b2fc13eca6a7bb6f78ebc7c529547e3ab9684" +dependencies = [ + "bitflags", + "cfg-if", + "clipboard-win", + "home", + "libc", + "log", + "memchr", + "nix", + "radix_trie", + "unicode-segmentation", + "unicode-width", + "utf8parse", + "windows-sys 0.61.2", +] + [[package]] name = "ryu" version = "1.0.23" @@ -3149,6 +3338,18 @@ dependencies = [ "der", ] +[[package]] +name = "sqlite-wasm-rs" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc3efc0da82635d7e1ced0053bbbfa8c7ab9645d0bf36ceb4f7127bb85315d75" +dependencies = [ + "cc", + "js-sys", + "rsqlite-vfs", + "wasm-bindgen", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -3608,12 +3809,27 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-normalization" +version = "0.1.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fd4f6878c9cb28d874b009da9e8d183b5abc80117c40bbd187a1fde336be6e8" +dependencies = [ + "tinyvec", +] + [[package]] name = "unicode-segmentation" version = "1.13.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8" +[[package]] +name = "unicode-width" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254" + [[package]] name = "unicode-xid" version = "0.2.6" @@ -3678,6 +3894,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "vcpkg" +version = "0.2.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" + [[package]] name = "vergen" version = "9.1.0" diff --git a/Cargo.toml b/Cargo.toml index ea7131a..d50ac87 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,13 +1,19 @@ [workspace] resolver = "3" -members = ["crates/federation-net", "apps/federation-net-demo"] +members = [ + "crates/federation-net", + "crates/artist-dht", + "apps/federation-net-demo", + "apps/artist-dht-cli", +] [workspace.package] edition = "2024" license = "MIT OR Apache-2.0" -rust-version = "1.85" +rust-version = "1.91" [workspace.dependencies] +federation-net = { path = "crates/federation-net" } iroh = "1" iroh-base = "1" iroh-tickets = "1" @@ -15,11 +21,16 @@ tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time" serde = { version = "1", features = ["derive"] } postcard = { version = "1", features = ["alloc"] } blake3 = "1" +futures = "0.3" +uuid = { version = "1", features = ["v7"] } +unicode-normalization = "0.1" +rusqlite = { version = "0.40", features = ["bundled"] } thiserror = "2" tracing = "0.1" clap = { version = "4", features = ["derive"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] } anyhow = "1" +rustyline = "18" data-encoding = "2" rand = "0.9" tempfile = "3" diff --git a/README.md b/README.md index 5034d58..4a686fe 100644 --- a/README.md +++ b/README.md @@ -152,6 +152,15 @@ event channel (`NetworkEventReceiver`): Consume events promptly: the channel is bounded and the engine applies back-pressure instead of buffering without limit. +## Example: distributed search (`artist-dht`) + +The workspace also contains a bigger example built entirely on this library: +[`crates/artist-dht`](crates/artist-dht/README.md) — a Kademlia-style DHT +where every peer stores, routes and searches artist records without any +dedicated servers, plus its interactive CLI +[`apps/artist-dht-cli`](apps/artist-dht-cli). See its README for the +three-peer demo scenario. + ## Verification ```bash diff --git a/apps/artist-dht-cli/Cargo.toml b/apps/artist-dht-cli/Cargo.toml new file mode 100644 index 0000000..0f928be --- /dev/null +++ b/apps/artist-dht-cli/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "artist-dht-cli" +version = "0.1.0" +description = "Interactive CLI for the artist-dht PoC" +edition.workspace = true +license.workspace = true +rust-version.workspace = true + +[dependencies] +artist-dht = { path = "../../crates/artist-dht" } +federation-net = { workspace = true } +tokio = { workspace = true } +clap = { workspace = true } +tracing-subscriber = { workspace = true } +anyhow = { workspace = true } +rustyline = { workspace = true } diff --git a/apps/artist-dht-cli/src/main.rs b/apps/artist-dht-cli/src/main.rs new file mode 100644 index 0000000..7177108 --- /dev/null +++ b/apps/artist-dht-cli/src/main.rs @@ -0,0 +1,403 @@ +//! Interactive CLI for the artist-dht PoC. +//! +//! Starts a DHT node, optionally connects to other peers by ticket and lets +//! the user manage and search artists with slash commands. + +use std::path::PathBuf; +use std::sync::mpsc as std_mpsc; + +use anyhow::Context; +use artist_dht::{ + Artist, ArtistDhtConfig, ArtistDhtEvent, ArtistDhtService, NetworkId, PeerTicket, SearchOutcome, +}; +use clap::Parser; +use rustyline::ExternalPrinter; +use rustyline::error::ReadlineError; + +#[derive(Debug, Parser)] +#[command(name = "artist-dht-cli", about = "Distributed artist directory PoC")] +struct Args { + /// Directory for the peer identity and database. + #[arg(long)] + data_dir: PathBuf, + + /// Name of the network to join (all peers must use the same name). + #[arg(long)] + network_id: String, + + /// Display name of this peer (used only for the prompt). + #[arg(long)] + name: String, + + /// Ticket(s) of peers to connect to on startup; may be repeated. + #[arg(long)] + connect: Vec, + + /// Enable verbose logging. + #[arg(long)] + verbose: bool, +} + +/// A line of user input, or the request to stop. +enum Input { + Line(String), + Quit, +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let args = Args::parse(); + + let filter = if args.verbose { + "artist_dht=debug,federation_net=debug,info" + } else { + "warn" + }; + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(filter)), + ) + .with_writer(std::io::stderr) + .init(); + + let config = ArtistDhtConfig::builder() + .data_dir(&args.data_dir) + .network_id(NetworkId::from_name(&args.network_id)) + .build() + .context("invalid configuration")?; + + let (service, mut events) = ArtistDhtService::start(config) + .await + .context("failed to start the DHT node")?; + + println!("Endpoint ID: {}", service.endpoint_id()); + println!("Node ID: {}", service.node_id()); + match service.ticket().await { + Ok(ticket) => println!("Ticket: {ticket}"), + Err(err) => eprintln!("Could not create a ticket yet: {err}"), + } + + for ticket in &args.connect { + let ticket: PeerTicket = ticket.parse().context("invalid ticket")?; + match service.connect(ticket).await { + Ok(peer) => println!("Connected to: {peer}"), + Err(err) => { + println!("Connection rejected: {err}"); + let _ = service.shutdown().await; + std::process::exit(1); + } + } + } + if args.connect.is_empty() { + println!("Waiting for peers... (share the ticket above)"); + } + println!("Type /help for commands."); + + // Rustyline is blocking, so it runs on its own thread and forwards lines + // through a channel; the external printer keeps async output readable. + let mut editor = rustyline::DefaultEditor::new().context("failed to init the line editor")?; + let mut printer = editor + .create_external_printer() + .context("failed to create the console printer")?; + let prompt = format!("{}> ", args.name); + let (line_tx, line_rx) = std_mpsc::sync_channel::(16); + std::thread::spawn(move || { + loop { + match editor.readline(&prompt) { + Ok(line) => { + let _ = editor.add_history_entry(&line); + if line_tx.send(Input::Line(line)).is_err() { + break; + } + } + Err(ReadlineError::Interrupted) | Err(ReadlineError::Eof) => { + let _ = line_tx.send(Input::Quit); + break; + } + Err(err) => { + eprintln!("input error: {err}"); + let _ = line_tx.send(Input::Quit); + break; + } + } + } + }); + let (async_line_tx, mut async_line_rx) = tokio::sync::mpsc::channel::(16); + std::thread::spawn(move || { + while let Ok(input) = line_rx.recv() { + if async_line_tx.blocking_send(input).is_err() { + break; + } + } + }); + + loop { + tokio::select! { + _ = tokio::signal::ctrl_c() => break, + input = async_line_rx.recv() => { + match input { + Some(Input::Line(line)) => { + if handle_line(&service, line.trim()).await { + break; + } + } + Some(Input::Quit) | None => break, + } + } + event = events.recv() => { + match event { + Some(event) => { + let _ = printer.print(format_event(event)); + } + None => { + println!("Node stopped."); + break; + } + } + } + } + } + + service.shutdown().await.context("shutdown failed")?; + println!("Bye."); + // The rustyline thread keeps stdin busy; exit explicitly after a clean + // shutdown. + std::process::exit(0); +} + +/// Executes one command line. Returns `true` when the user asked to quit. +async fn handle_line(service: &ArtistDhtService, line: &str) -> bool { + let (command, rest) = match line.split_once(' ') { + Some((command, rest)) => (command, rest.trim()), + None => (line, ""), + }; + match command { + "" => {} + "/help" => print_help(), + "/quit" => return true, + "/id" => { + println!("Endpoint ID: {}", service.endpoint_id()); + println!("Node ID: {}", service.node_id()); + } + "/ticket" => match service.ticket().await { + Ok(ticket) => println!("Ticket: {ticket}"), + Err(err) => println!("Could not create a ticket: {err}"), + }, + "/peers" => { + let peers = service.connected_peers(); + if peers.is_empty() { + println!("No connected peers."); + } else { + println!("Connected peers: {}", peers.len()); + for peer in peers { + println!(" {peer}"); + } + } + } + "/routing" => print_routing(service), + "/list" => match service.list_local_artists().await { + Ok(artists) if artists.is_empty() => println!("No local artists."), + Ok(artists) => { + println!("Local artists: {}", artists.len()); + for artist in artists { + println!(" {} {}", short_id(&artist), artist.name); + } + } + Err(err) => println!("Error: {err}"), + }, + "/add" => { + if rest.is_empty() { + println!("Usage: /add "); + } else { + match service.add_artist(rest.to_string()).await { + Ok((artist, stats)) => { + println!("Added artist:"); + println!(" ID: {}", artist.id); + println!(" Name: {}", artist.name); + println!(" Normalized: {}", artist.normalized_name); + println!( + "Published {} keys to {} DHT nodes{}.", + stats.keys, + stats.remote_nodes, + if stats.local_replica { + " (+ local replica)" + } else { + "" + } + ); + } + Err(err) => println!("Error: {err}"), + } + } + } + "/delete" => { + if rest.is_empty() { + println!("Usage: /delete "); + } else { + match service.resolve_local_artist_id(rest).await { + Ok(artist_id) => match service.delete_artist(artist_id).await { + Ok(stats) => { + println!("Deleted artist {artist_id}"); + println!( + "Published tombstone to {} DHT nodes{}.", + stats.remote_nodes, + if stats.local_replica { + " (+ local replica)" + } else { + "" + } + ); + } + Err(err) => println!("Error: {err}"), + }, + Err(_) => println!("No unique local artist matches '{rest}'."), + } + } + } + "/search-local" => { + if rest.is_empty() { + println!("Usage: /search-local "); + } else { + match service.search_local(rest).await { + Ok(artists) if artists.is_empty() => println!("No matches."), + Ok(artists) => { + for artist in artists { + println!(" {} {}", short_id(&artist), artist.name); + } + } + Err(err) => println!("Error: {err}"), + } + } + } + "/search" => { + if rest.is_empty() { + println!("Usage: /search "); + } else { + match service.search_network(rest).await { + Ok(outcome) => print_search(rest, &outcome), + Err(err) => println!("Error: {err}"), + } + } + } + "/republish" => match service.republish().await { + Ok(stats) => println!( + "Republished {} records ({} keys) to up to {} nodes.", + stats.records, stats.keys, stats.remote_nodes + ), + Err(err) => println!("Error: {err}"), + }, + other => println!("Unknown command: {other}. Type /help."), + } + false +} + +fn print_help() { + println!( + "Commands:\n\ + \x20 /help show this help\n\ + \x20 /id show endpoint and node ids\n\ + \x20 /ticket print the connection ticket\n\ + \x20 /peers list open connections\n\ + \x20 /routing list known DHT contacts\n\ + \x20 /list list local artists\n\ + \x20 /add add and publish an artist\n\ + \x20 /delete delete a local artist (id or hex prefix)\n\ + \x20 /search-local search the local database only\n\ + \x20 /search search locally and across the DHT\n\ + \x20 /republish republish local records now\n\ + \x20 /quit shut down" + ); +} + +fn print_routing(service: &ArtistDhtService) { + let mut contacts = service.known_peers(); + if contacts.is_empty() { + println!("No known DHT peers."); + return; + } + contacts.sort_by_key(|contact| std::cmp::Reverse(contact.last_seen_ms)); + println!("Known DHT peers: {}", contacts.len()); + println!( + "{:<15} {:<15} {:<10} Last seen", + "Peer ID", "Node ID", "Connected" + ); + let now = now_ms(); + for contact in contacts { + let connected = if service.is_connected(contact.peer_id) { + "yes" + } else { + "no" + }; + let ago_s = now.saturating_sub(contact.last_seen_ms) / 1000; + println!( + "{:<15} {:<15} {:<10} {}s ago", + shorten(&contact.peer_id.to_string()), + shorten(&contact.node_id.to_string()), + connected, + ago_s + ); + } +} + +fn print_search(query: &str, outcome: &SearchOutcome) { + println!("Search: {}", artist_dht::normalize_artist_name(query)); + println!(); + println!("Local results:"); + if outcome.local_results.is_empty() { + println!(" No matches."); + } else { + for artist in &outcome.local_results { + println!(" {} {}", short_id(artist), artist.name); + } + } + println!(); + println!("DHT results:"); + if outcome.network_results.is_empty() { + println!(" No matches."); + } else { + for (i, artist) in outcome.network_results.iter().enumerate() { + println!(" {}. {}", i + 1, artist.name); + println!(" Artist ID: {}", artist.id); + println!(" Owner: {}", artist.owner); + println!(" Revision: {}", artist.revision); + } + } + println!(); + println!("Lookup:"); + println!(" Queried nodes: {}", outcome.queried_nodes); + println!(" Discovered nodes: {}", outcome.discovered_nodes); + println!(" Duration: {} ms", outcome.duration.as_millis()); +} + +fn format_event(event: ArtistDhtEvent) -> String { + match event { + ArtistDhtEvent::PeerConnected { peer_id } => format!("Peer connected: {peer_id}"), + ArtistDhtEvent::PeerDisconnected { peer_id } => { + format!("Peer disconnected: {peer_id}") + } + ArtistDhtEvent::ContactDiscovered { contact } => { + format!("Discovered DHT contact: {}", contact.peer_id) + } + ArtistDhtEvent::Error { message } => format!("Error: {message}"), + } +} + +fn short_id(artist: &Artist) -> String { + shorten(&artist.id.to_string()) +} + +fn shorten(hex: &str) -> String { + if hex.len() > 12 { + format!("{}...", &hex[..12]) + } else { + hex.to_string() + } +} + +fn now_ms() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or_default() +} diff --git a/crates/artist-dht/Cargo.toml b/crates/artist-dht/Cargo.toml new file mode 100644 index 0000000..7732195 --- /dev/null +++ b/crates/artist-dht/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "artist-dht" +version = "0.1.0" +description = "Distributed artist search PoC: a Kademlia-style DHT on top of federation-net" +edition.workspace = true +license.workspace = true +rust-version.workspace = true + +[dependencies] +federation-net = { workspace = true } +tokio = { workspace = true } +serde = { workspace = true } +postcard = { workspace = true } +blake3 = { workspace = true } +futures = { workspace = true } +uuid = { workspace = true } +unicode-normalization = { workspace = true } +rusqlite = { workspace = true } +thiserror = { workspace = true } +tracing = { workspace = true } +rand = { workspace = true } +data-encoding = { workspace = true } + +[dev-dependencies] +anyhow = { workspace = true } +tempfile = { workspace = true } +iroh = { workspace = true } diff --git a/crates/artist-dht/README.md b/crates/artist-dht/README.md new file mode 100644 index 0000000..3961c80 --- /dev/null +++ b/crates/artist-dht/README.md @@ -0,0 +1,152 @@ +# artist-dht + +A proof-of-concept **distributed artist directory** built on top of +[`federation-net`](../federation-net). Every running process is a full DHT +participant — client, router and storage node at once. There are no +bootstrap servers, index nodes or search servers: the "server" is the set of +running peers itself. + +## What it demonstrates + +```text +/add on Peer B → name normalization → publish index records into the DHT + → replication to regular peers +/search on A/C → iterative Kademlia-style lookup (no broadcast) + → record of Peer B is found, with its owner id +``` + +* Records are published under BLAKE3 keys: one **exact key** for the whole + normalized name and one **token key** per word, so `/search massive` finds + *Massive Attack*. +* Each peer has a stable 256-bit `NodeId` derived from its persistent + `federation-net` endpoint id; records live on the `K = 8` XOR-closest + nodes. +* Peers discover each other through automatic Hello/PeerExchange gossip; + connections to newly learned peers are opened **on demand** from stored + tickets. +* Deletions propagate as **tombstones** (revision-based, they beat active + records of the same or lower revision). Replicas expire by TTL (30 min + active, 2 h tombstones); owners republish every 10 minutes. +* Local state (identity, own artists, replicas, known peers) lives in + `/state.sqlite3` and `/identity.key`. + +Out of scope (by design): fuzzy search, content transfer, CRDTs, consensus, +signatures on DHT records, Sybil protection, accounts, GUI. + +## Bootstrapping limitation + +There is deliberately **no bootstrap server**, so a peer cannot join a +network it has no contact in: you must pass the ticket of *any* already +running peer via `--connect`. That first peer has no special role — once +contacts have spread through peer exchange, it can be switched off. + +## Running the demo (three peers) + +Start Peer A (alice): + +```bash +cargo run -p artist-dht-cli -- \ + --data-dir ./peer-a \ + --network-id demo-artists \ + --name alice +``` + +Copy the printed `Ticket: fnet...`. Start Peer B (bob) with it: + +```bash +cargo run -p artist-dht-cli -- \ + --data-dir ./peer-b \ + --network-id demo-artists \ + --name bob \ + --connect 'fnet...' +``` + +Start Peer C (charlie), connected **only to A** — it will learn about B via +peer exchange: + +```bash +cargo run -p artist-dht-cli -- \ + --data-dir ./peer-c \ + --network-id demo-artists \ + --name charlie \ + --connect 'fnet...' +``` + +`--connect` may be repeated to dial several peers. All three processes must +use the same `--network-id`; a peer from another network is rejected during +the transport handshake. + +### Demo scenario + +1. On **bob**: `/add Massive Attack` and `/add Portishead` — each prints the + artist id and how many DHT nodes stored the replicas. +2. On **alice**: `/search massive attack` — the DHT results list *Massive + Attack* with bob's endpoint id as the owner. +3. On **charlie**: `/search portishead` (and `/search massive`) — same + records found through an iterative lookup, not a broadcast. +4. Stop **bob** (Ctrl+C). Both alice and charlie still find the records from + replicas until the TTL expires. +5. Restart **bob** with the same `--data-dir`: the endpoint id is unchanged, + the local database is intact and records are republished automatically. +6. On **bob**: `/delete ` (a unique hex prefix is enough) — a + tombstone propagates and the other peers stop returning the record. + +### Commands + +```text +/help show help +/id show endpoint and node ids +/ticket print the connection ticket +/peers list open connections +/routing list known DHT contacts (connected / last seen) +/list list local artists +/add add and publish an artist +/delete delete a local artist (id or unique hex prefix) +/search-local search the local database only +/search search locally and across the DHT +/republish republish local records now +/quit shut down gracefully +``` + +## Protocol limits + +```text +K = 8, ALPHA = 3, max lookup requests = 32 +max contacts per PeerExchange = 32 +max records per FindValue response = 100 +max artist name = 512 bytes, max tokens = 32 +max pending requests = 1024 +request timeout = 5 s, lookup timeout = 15 s +``` + +## Library usage + +The CLI is a thin wrapper around the `artist-dht` library: + +```rust +use artist_dht::{ArtistDhtConfig, ArtistDhtService}; +use federation_net::NetworkId; + +let config = ArtistDhtConfig::builder() + .data_dir("./peer-a") + .network_id(NetworkId::from_name("demo-artists")) + .build()?; +let (service, mut events) = ArtistDhtService::start(config).await?; + +let (artist, stats) = service.add_artist("Massive Attack".into()).await?; +let outcome = service.search_network("massive").await?; +service.shutdown().await?; +``` + +## Verification + +```bash +cargo fmt --check +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo test --workspace +``` + +The integration tests start three real DHT nodes in one process and walk the +whole demo scenario (publish, distributed search, replica survival after the +owner leaves, restart, tombstones), so they need network access and take a +couple of minutes. diff --git a/crates/artist-dht/src/config.rs b/crates/artist-dht/src/config.rs new file mode 100644 index 0000000..5050be6 --- /dev/null +++ b/crates/artist-dht/src/config.rs @@ -0,0 +1,181 @@ +//! Service configuration. + +use std::path::PathBuf; +use std::time::Duration; + +use federation_net::NetworkId; + +use crate::error::{ArtistDhtError, Result}; + +/// Default interval between republish rounds. +pub const DEFAULT_REPUBLISH_INTERVAL: Duration = Duration::from_secs(10 * 60); +/// Default interval between expired-record sweeps. +pub const DEFAULT_EXPIRE_INTERVAL: Duration = Duration::from_secs(60); +/// Default timeout of a single DHT request. +pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(5); +/// Default timeout of a whole iterative lookup. +pub const DEFAULT_LOOKUP_TIMEOUT: Duration = Duration::from_secs(15); +/// Default timeout for transport operations (dialing, handshakes, sends). +pub const DEFAULT_TRANSPORT_TIMEOUT: Duration = Duration::from_secs(15); + +/// Configuration for an [`crate::ArtistDhtService`]. +/// +/// Use [`ArtistDhtConfig::builder`] to construct a validated instance. +#[derive(Debug, Clone)] +pub struct ArtistDhtConfig { + /// Directory for the peer identity and the SQLite database. + pub data_dir: PathBuf, + /// Network this peer participates in. + pub network_id: NetworkId, + /// Interval between automatic republish rounds. + pub republish_interval: Duration, + /// Interval between sweeps of expired DHT records. + pub expire_interval: Duration, + /// Timeout of a single DHT request. + pub request_timeout: Duration, + /// Timeout of a whole iterative lookup. + pub lookup_timeout: Duration, + /// Timeout for transport operations: dialing a peer, handshakes and + /// message delivery. Kept separate from `request_timeout` because + /// establishing a connection through relays can take much longer than a + /// request over an existing one. + pub transport_timeout: Duration, +} + +impl ArtistDhtConfig { + /// Returns a new [`ArtistDhtConfigBuilder`]. + pub fn builder() -> ArtistDhtConfigBuilder { + ArtistDhtConfigBuilder::default() + } +} + +/// Builder for [`ArtistDhtConfig`]. +/// +/// `data_dir` and `network_id` are required; the timers default to the +/// production values and are configurable mainly for tests. +#[derive(Debug, Default, Clone)] +pub struct ArtistDhtConfigBuilder { + data_dir: Option, + network_id: Option, + republish_interval: Option, + expire_interval: Option, + request_timeout: Option, + lookup_timeout: Option, + transport_timeout: Option, +} + +impl ArtistDhtConfigBuilder { + /// Sets the data directory. + pub fn data_dir(mut self, dir: impl Into) -> Self { + self.data_dir = Some(dir.into()); + self + } + + /// Sets the network identifier. + pub fn network_id(mut self, network_id: NetworkId) -> Self { + self.network_id = Some(network_id); + self + } + + /// Sets the republish interval. + pub fn republish_interval(mut self, interval: Duration) -> Self { + self.republish_interval = Some(interval); + self + } + + /// Sets the expired-record sweep interval. + pub fn expire_interval(mut self, interval: Duration) -> Self { + self.expire_interval = Some(interval); + self + } + + /// Sets the timeout of a single DHT request. + pub fn request_timeout(mut self, timeout: Duration) -> Self { + self.request_timeout = Some(timeout); + self + } + + /// Sets the timeout of a whole iterative lookup. + pub fn lookup_timeout(mut self, timeout: Duration) -> Self { + self.lookup_timeout = Some(timeout); + self + } + + /// Sets the timeout for transport operations (dialing, handshakes). + pub fn transport_timeout(mut self, timeout: Duration) -> Self { + self.transport_timeout = Some(timeout); + self + } + + /// Validates and builds the configuration. + pub fn build(self) -> Result { + let data_dir = self + .data_dir + .ok_or_else(|| ArtistDhtError::Database("data_dir is required".into()))?; + if data_dir.as_os_str().is_empty() { + return Err(ArtistDhtError::Database( + "data_dir must not be empty".into(), + )); + } + let network_id = self + .network_id + .ok_or_else(|| ArtistDhtError::Network("network_id is required".into()))?; + + let config = ArtistDhtConfig { + data_dir, + network_id, + republish_interval: self + .republish_interval + .unwrap_or(DEFAULT_REPUBLISH_INTERVAL), + expire_interval: self.expire_interval.unwrap_or(DEFAULT_EXPIRE_INTERVAL), + request_timeout: self.request_timeout.unwrap_or(DEFAULT_REQUEST_TIMEOUT), + lookup_timeout: self.lookup_timeout.unwrap_or(DEFAULT_LOOKUP_TIMEOUT), + transport_timeout: self.transport_timeout.unwrap_or(DEFAULT_TRANSPORT_TIMEOUT), + }; + for (name, value) in [ + ("republish_interval", config.republish_interval), + ("expire_interval", config.expire_interval), + ("request_timeout", config.request_timeout), + ("lookup_timeout", config.lookup_timeout), + ("transport_timeout", config.transport_timeout), + ] { + if value.is_zero() { + return Err(ArtistDhtError::Database(format!( + "{name} must be greater than zero" + ))); + } + } + Ok(config) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn builder_applies_defaults() { + let config = ArtistDhtConfig::builder() + .data_dir("./dir") + .network_id(NetworkId::from_name("test")) + .build() + .expect("valid"); + assert_eq!(config.republish_interval, DEFAULT_REPUBLISH_INTERVAL); + assert_eq!(config.expire_interval, DEFAULT_EXPIRE_INTERVAL); + assert_eq!(config.request_timeout, DEFAULT_REQUEST_TIMEOUT); + assert_eq!(config.lookup_timeout, DEFAULT_LOOKUP_TIMEOUT); + } + + #[test] + fn builder_rejects_missing_or_invalid() { + assert!(ArtistDhtConfig::builder().build().is_err()); + assert!( + ArtistDhtConfig::builder() + .data_dir("./dir") + .network_id(NetworkId::from_name("test")) + .request_timeout(Duration::ZERO) + .build() + .is_err() + ); + } +} diff --git a/crates/artist-dht/src/database.rs b/crates/artist-dht/src/database.rs new file mode 100644 index 0000000..fe267f3 --- /dev/null +++ b/crates/artist-dht/src/database.rs @@ -0,0 +1,480 @@ +//! Local SQLite persistence. +//! +//! `rusqlite` is synchronous, so every database call runs on the blocking +//! thread pool via `tokio::task::spawn_blocking`; the async runtime is never +//! blocked on file I/O. + +use std::path::Path; +use std::str::FromStr; +use std::sync::{Arc, Mutex}; + +use federation_net::EndpointId; +use rusqlite::{Connection, OptionalExtension, params}; + +use crate::dht::{StoreDecision, decide_store}; +use crate::error::{ArtistDhtError, Result}; +use crate::message::MAX_RECORDS_PER_RESPONSE; +use crate::normalization::tokenize; +use crate::record::{Artist, ArtistId, DhtKey, StoredArtistRecord, TOMBSTONE_TTL}; +use crate::routing::{NodeContact, NodeId}; + +const SCHEMA: &str = " +CREATE TABLE IF NOT EXISTS local_artists ( + id BLOB PRIMARY KEY, + owner_peer_id TEXT NOT NULL, + name TEXT NOT NULL, + normalized_name TEXT NOT NULL, + revision INTEGER NOT NULL, + deleted INTEGER NOT NULL DEFAULT 0, + updated_at_ms INTEGER NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_local_artists_normalized_name + ON local_artists(normalized_name); + +CREATE TABLE IF NOT EXISTS dht_records ( + dht_key BLOB NOT NULL, + artist_id BLOB NOT NULL, + owner_peer_id TEXT NOT NULL, + payload BLOB NOT NULL, + revision INTEGER NOT NULL, + deleted INTEGER NOT NULL, + expires_at_ms INTEGER NOT NULL, + + PRIMARY KEY (dht_key, artist_id, owner_peer_id) +); + +CREATE INDEX IF NOT EXISTS idx_dht_records_expires_at + ON dht_records(expires_at_ms); + +CREATE TABLE IF NOT EXISTS known_peers ( + peer_id TEXT PRIMARY KEY, + node_id BLOB NOT NULL, + ticket TEXT NOT NULL, + last_seen_ms INTEGER NOT NULL +); +"; + +/// Handle to the local SQLite database. +/// +/// Cheap to clone; all clones share one connection guarded by a mutex that is +/// only ever locked from blocking-pool threads. +#[derive(Clone)] +pub(crate) struct Database { + conn: Arc>, +} + +impl Database { + /// Opens (creating if needed) the database at `path` and applies the + /// schema. + pub async fn open(path: &Path) -> Result { + let path = path.to_path_buf(); + let conn = tokio::task::spawn_blocking(move || -> Result { + let conn = Connection::open(&path).map_err(|err| { + ArtistDhtError::Database(format!("failed to open {}: {err}", path.display())) + })?; + conn.execute_batch(SCHEMA).map_err(|err| { + ArtistDhtError::Database(format!("failed to apply schema: {err}")) + })?; + Ok(conn) + }) + .await + .map_err(|err| ArtistDhtError::Database(format!("database task panicked: {err}")))??; + Ok(Self { + conn: Arc::new(Mutex::new(conn)), + }) + } + + /// Runs a closure against the connection on the blocking pool. + async fn call(&self, f: F) -> Result + where + F: FnOnce(&Connection) -> rusqlite::Result + Send + 'static, + R: Send + 'static, + { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let guard = conn + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + f(&guard).map_err(|err| ArtistDhtError::Database(err.to_string())) + }) + .await + .map_err(|err| ArtistDhtError::Database(format!("database task panicked: {err}")))? + } + + /// Inserts or replaces a locally owned artist record. + pub async fn upsert_local_artist(&self, artist: &Artist) -> Result<()> { + let artist = artist.clone(); + self.call(move |conn| { + conn.execute( + "INSERT OR REPLACE INTO local_artists + (id, owner_peer_id, name, normalized_name, revision, deleted, updated_at_ms) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + params![ + artist.id.as_bytes().as_slice(), + artist.owner.to_string(), + artist.name, + artist.normalized_name, + artist.revision as i64, + artist.deleted as i64, + artist.updated_at_ms as i64, + ], + )?; + Ok(()) + }) + .await + } + + /// Fetches one locally owned artist by id. + pub async fn get_local_artist(&self, id: ArtistId) -> Result> { + self.call(move |conn| { + conn.query_row( + "SELECT id, owner_peer_id, name, normalized_name, revision, deleted, updated_at_ms + FROM local_artists WHERE id = ?1", + params![id.as_bytes().as_slice()], + artist_from_row, + ) + .optional() + }) + .await + } + + /// Finds locally owned artists whose id starts with the given hex prefix. + pub async fn find_local_by_id_prefix(&self, prefix: String) -> Result> { + let prefix = prefix.to_lowercase(); + self.call(move |conn| { + let mut stmt = conn.prepare( + "SELECT id, owner_peer_id, name, normalized_name, revision, deleted, updated_at_ms + FROM local_artists WHERE deleted = 0", + )?; + let rows = stmt.query_map([], artist_from_row)?; + let mut result = Vec::new(); + for row in rows { + let artist = row?; + if artist.id.to_hex().starts_with(&prefix) { + result.push(artist); + } + } + Ok(result) + }) + .await + } + + /// Lists locally owned artists. Tombstones are excluded unless + /// `include_deleted` is set. + pub async fn list_local_artists(&self, include_deleted: bool) -> Result> { + self.call(move |conn| { + let mut stmt = conn.prepare( + "SELECT id, owner_peer_id, name, normalized_name, revision, deleted, updated_at_ms + FROM local_artists ORDER BY normalized_name", + )?; + let rows = stmt.query_map([], artist_from_row)?; + let mut result = Vec::new(); + for row in rows { + let artist = row?; + if include_deleted || !artist.deleted { + result.push(artist); + } + } + Ok(result) + }) + .await + } + + /// Returns everything that must be republished: active records plus + /// tombstones that have not outlived [`TOMBSTONE_TTL`] yet. + pub async fn local_artists_for_republish(&self, now_ms: u64) -> Result> { + let all = self.list_local_artists(true).await?; + let tombstone_ttl = TOMBSTONE_TTL.as_millis() as u64; + Ok(all + .into_iter() + .filter(|artist| { + !artist.deleted || artist.updated_at_ms.saturating_add(tombstone_ttl) > now_ms + }) + .collect()) + } + + /// Searches locally owned active artists: exact normalized match, or all + /// query tokens present in the artist's token set. + pub async fn search_local(&self, normalized_query: String) -> Result> { + let all = self.list_local_artists(false).await?; + let query_tokens = tokenize(&normalized_query); + Ok(all + .into_iter() + .filter(|artist| { + if artist.normalized_name == normalized_query { + return true; + } + if query_tokens.is_empty() { + return false; + } + let artist_tokens = tokenize(&artist.normalized_name); + query_tokens + .iter() + .all(|token| artist_tokens.iter().any(|t| t == token)) + }) + .collect()) + } + + /// Applies a validated incoming record to the replica store, following + /// the revision/tombstone rules. Returns `true` if the record was written + /// or refreshed. + pub async fn store_dht_record(&self, key: DhtKey, record: StoredArtistRecord) -> Result { + self.call(move |conn| { + let existing: Option<(i64, i64, i64)> = conn + .query_row( + "SELECT revision, deleted, expires_at_ms FROM dht_records + WHERE dht_key = ?1 AND artist_id = ?2 AND owner_peer_id = ?3", + params![ + key.as_bytes().as_slice(), + record.artist.id.as_bytes().as_slice(), + record.artist.owner.to_string(), + ], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .optional()?; + let existing = + existing.map(|(rev, del, exp)| (rev as u64, del != 0, exp as u64)); + match decide_store(existing, &record) { + StoreDecision::Ignore => Ok(false), + StoreDecision::Write | StoreDecision::RefreshExpiry(_) => { + let payload = postcard::to_stdvec(&record).map_err(|err| { + rusqlite::Error::ToSqlConversionFailure(Box::new(err)) + })?; + conn.execute( + "INSERT OR REPLACE INTO dht_records + (dht_key, artist_id, owner_peer_id, payload, revision, deleted, expires_at_ms) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + params![ + key.as_bytes().as_slice(), + record.artist.id.as_bytes().as_slice(), + record.artist.owner.to_string(), + payload, + record.artist.revision as i64, + record.artist.deleted as i64, + record.expires_at_ms as i64, + ], + )?; + Ok(true) + } + } + }) + .await + } + + /// Returns non-expired replicas stored under `key`, including tombstones + /// (they inform other peers about deletions). Capped at + /// [`MAX_RECORDS_PER_RESPONSE`]. + pub async fn dht_records_by_key( + &self, + key: DhtKey, + now_ms: u64, + ) -> Result> { + self.call(move |conn| { + let mut stmt = conn.prepare( + "SELECT payload FROM dht_records + WHERE dht_key = ?1 AND expires_at_ms > ?2 + LIMIT ?3", + )?; + let rows = stmt.query_map( + params![ + key.as_bytes().as_slice(), + now_ms as i64, + MAX_RECORDS_PER_RESPONSE as i64 + ], + |row| row.get::<_, Vec>(0), + )?; + let mut records = Vec::new(); + for row in rows { + let payload = row?; + // A payload we cannot decode is skipped, not fatal. + if let Ok(record) = postcard::from_bytes::(&payload) { + records.push(record); + } + } + Ok(records) + }) + .await + } + + /// Deletes expired replicas. Returns the number of removed rows. + pub async fn delete_expired_records(&self, now_ms: u64) -> Result { + self.call(move |conn| { + conn.execute( + "DELETE FROM dht_records WHERE expires_at_ms <= ?1", + params![now_ms as i64], + ) + }) + .await + } + + /// Inserts or refreshes a known peer contact. + pub async fn upsert_known_peer(&self, contact: &NodeContact) -> Result<()> { + let contact = contact.clone(); + self.call(move |conn| { + conn.execute( + "INSERT OR REPLACE INTO known_peers (peer_id, node_id, ticket, last_seen_ms) + VALUES (?1, ?2, ?3, ?4)", + params![ + contact.peer_id.to_string(), + contact.node_id.as_bytes().as_slice(), + contact.ticket, + contact.last_seen_ms as i64, + ], + )?; + Ok(()) + }) + .await + } + + /// Loads all persisted peer contacts. + pub async fn load_known_peers(&self) -> Result> { + self.call(|conn| { + let mut stmt = + conn.prepare("SELECT peer_id, node_id, ticket, last_seen_ms FROM known_peers")?; + let rows = stmt.query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, Vec>(1)?, + row.get::<_, String>(2)?, + row.get::<_, i64>(3)?, + )) + })?; + let mut contacts = Vec::new(); + for row in rows { + let (peer_id, node_id, ticket, last_seen_ms) = row?; + let Ok(peer_id) = EndpointId::from_str(&peer_id) else { + continue; + }; + let Ok(node_id) = <[u8; 32]>::try_from(node_id.as_slice()) else { + continue; + }; + contacts.push(NodeContact { + node_id: NodeId::from_bytes(node_id), + peer_id, + ticket, + last_seen_ms: last_seen_ms as u64, + }); + } + Ok(contacts) + }) + .await + } +} + +fn artist_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let id: Vec = row.get(0)?; + let owner: String = row.get(1)?; + let id = <[u8; 32]>::try_from(id.as_slice()).map_err(|_| { + rusqlite::Error::FromSqlConversionFailure( + 0, + rusqlite::types::Type::Blob, + "artist id must be 32 bytes".into(), + ) + })?; + let owner = EndpointId::from_str(&owner).map_err(|err| { + rusqlite::Error::FromSqlConversionFailure( + 1, + rusqlite::types::Type::Text, + format!("invalid owner peer id: {err}").into(), + ) + })?; + Ok(Artist { + id: ArtistId::from_bytes(id), + owner, + name: row.get(2)?, + normalized_name: row.get(3)?, + revision: row.get::<_, i64>(4)? as u64, + deleted: row.get::<_, i64>(5)? != 0, + updated_at_ms: row.get::<_, i64>(6)? as u64, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::record::now_ms; + + fn test_peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + fn artist(owner: EndpointId, name: &str, revision: u64, deleted: bool) -> Artist { + Artist { + id: ArtistId::derive(&owner, &uuid::Uuid::from_u128(1)), + owner, + name: name.to_string(), + normalized_name: crate::normalization::normalize_artist_name(name), + revision, + deleted, + updated_at_ms: now_ms(), + } + } + + async fn open_temp() -> (tempfile::TempDir, Database) { + let dir = tempfile::tempdir().expect("tempdir"); + let db = Database::open(&dir.path().join("state.sqlite3")) + .await + .expect("open db"); + (dir, db) + } + + #[tokio::test] + async fn local_artist_round_trip() { + let (_dir, db) = open_temp().await; + let owner = test_peer(1); + let artist = artist(owner, "Massive Attack", 1, false); + db.upsert_local_artist(&artist).await.expect("upsert"); + let loaded = db.get_local_artist(artist.id).await.expect("get"); + assert_eq!(loaded, Some(artist.clone())); + let found = db + .search_local("massive attack".into()) + .await + .expect("search"); + assert_eq!(found.len(), 1); + let by_token = db.search_local("attack".into()).await.expect("search"); + assert_eq!(by_token.len(), 1); + let none = db.search_local("portishead".into()).await.expect("search"); + assert!(none.is_empty()); + } + + #[tokio::test] + async fn expired_dht_record_is_not_returned() { + let (_dir, db) = open_temp().await; + let owner = test_peer(1); + let artist = artist(owner, "Massive Attack", 1, false); + let key = DhtKey::exact(&federation_net::NetworkId::from_name("t"), "massive attack"); + let now = now_ms(); + let record = StoredArtistRecord { + artist, + publisher: owner, + expires_at_ms: now + 50, + }; + assert!(db.store_dht_record(key, record).await.expect("store")); + assert_eq!(db.dht_records_by_key(key, now).await.expect("get").len(), 1); + // After expiry the record is filtered out and then swept. + let later = now + 100; + assert!( + db.dht_records_by_key(key, later) + .await + .expect("get") + .is_empty() + ); + assert_eq!(db.delete_expired_records(later).await.expect("sweep"), 1); + } + + #[tokio::test] + async fn known_peers_round_trip() { + let (_dir, db) = open_temp().await; + let peer = test_peer(2); + let contact = NodeContact { + node_id: NodeId::from_endpoint(&peer), + peer_id: peer, + ticket: "fnet-test".into(), + last_seen_ms: 42, + }; + db.upsert_known_peer(&contact).await.expect("upsert"); + let loaded = db.load_known_peers().await.expect("load"); + assert_eq!(loaded, vec![contact]); + } +} diff --git a/crates/artist-dht/src/dht.rs b/crates/artist-dht/src/dht.rs new file mode 100644 index 0000000..c3a9739 --- /dev/null +++ b/crates/artist-dht/src/dht.rs @@ -0,0 +1,274 @@ +//! DHT record validation and replacement rules. + +use federation_net::NetworkId; + +use crate::message::StoreRecordRequest; +use crate::normalization::{normalize_artist_name, tokenize}; +use crate::record::{ + ACTIVE_RECORD_TTL, DhtKey, MAX_ARTIST_NAME_BYTES, MAX_TOKENS_PER_ARTIST, StoredArtistRecord, + TOMBSTONE_TTL, +}; + +/// Outcome of comparing an incoming record with the stored one. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum StoreDecision { + /// No record stored yet, or the incoming one supersedes it: write it. + Write, + /// Same logical record: keep the stored row but extend its expiry. + RefreshExpiry(u64), + /// The incoming record is older or otherwise loses: ignore it. + Ignore, +} + +/// Decides what to do with an incoming record given the stored state +/// `(revision, deleted, expires_at_ms)` for the same +/// `(key, artist_id, owner)`. +/// +/// Rules: +/// * a higher revision always wins; +/// * on equal revisions a tombstone beats an active record; +/// * on equal revisions and equal deletion state the record is the same — +/// only the expiry is refreshed (this is how republish extends TTL); +/// * an older revision never replaces a newer one, in particular an old +/// active record never resurrects a tombstone. +pub(crate) fn decide_store( + existing: Option<(u64, bool, u64)>, + incoming: &StoredArtistRecord, +) -> StoreDecision { + let Some((revision, deleted, expires_at_ms)) = existing else { + return StoreDecision::Write; + }; + let artist = &incoming.artist; + if artist.revision > revision { + return StoreDecision::Write; + } + if artist.revision < revision { + return StoreDecision::Ignore; + } + // Equal revisions. + match (deleted, artist.deleted) { + (false, true) => StoreDecision::Write, + (true, false) => StoreDecision::Ignore, + _ => { + if incoming.expires_at_ms > expires_at_ms { + StoreDecision::RefreshExpiry(incoming.expires_at_ms) + } else { + StoreDecision::Ignore + } + } + } +} + +/// Validates an incoming `StoreRecord` request. +/// +/// Checks the size limits, that the record is internally consistent, that the +/// key actually corresponds to the record's name or one of its tokens, and +/// clamps the expiry to the maximum TTL allowed for the record type. Returns +/// the record with a possibly clamped `expires_at_ms`. +pub(crate) fn validate_store( + request: StoreRecordRequest, + network_id: &NetworkId, + now_ms: u64, +) -> Result { + let mut record = request.record; + let artist = &record.artist; + + if artist.name.len() > MAX_ARTIST_NAME_BYTES { + return Err("artist name too long".into()); + } + if artist.normalized_name != normalize_artist_name(&artist.name) { + return Err("normalized name does not match the artist name".into()); + } + if artist.normalized_name.is_empty() { + return Err("artist name normalizes to nothing".into()); + } + let tokens = tokenize(&artist.normalized_name); + if tokens.len() > MAX_TOKENS_PER_ARTIST { + return Err("too many tokens".into()); + } + + let key_matches = request.key == DhtKey::exact(network_id, &artist.normalized_name) + || tokens + .iter() + .any(|token| request.key == DhtKey::token(network_id, token)); + if !key_matches { + return Err("key does not correspond to the record".into()); + } + + if record.expires_at_ms <= now_ms { + return Err("record is already expired".into()); + } + let max_ttl_ms = if artist.deleted { + TOMBSTONE_TTL.as_millis() as u64 + } else { + ACTIVE_RECORD_TTL.as_millis() as u64 + }; + record.expires_at_ms = record.expires_at_ms.min(now_ms + max_ttl_ms); + + Ok(record) +} + +#[cfg(test)] +mod tests { + use federation_net::EndpointId; + + use super::*; + use crate::record::{Artist, ArtistId}; + + fn test_peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + fn record(revision: u64, deleted: bool, expires_at_ms: u64) -> StoredArtistRecord { + let owner = test_peer(1); + StoredArtistRecord { + artist: Artist { + id: ArtistId::from_bytes([9u8; 32]), + owner, + name: "Massive Attack".into(), + normalized_name: "massive attack".into(), + revision, + deleted, + updated_at_ms: 0, + }, + publisher: owner, + expires_at_ms, + } + } + + #[test] + fn newer_revision_replaces_older() { + let incoming = record(2, false, 1000); + assert_eq!( + decide_store(Some((1, false, 500)), &incoming), + StoreDecision::Write + ); + } + + #[test] + fn older_revision_is_ignored() { + let incoming = record(1, false, 1000); + assert_eq!( + decide_store(Some((2, false, 500)), &incoming), + StoreDecision::Ignore + ); + } + + #[test] + fn tombstone_beats_active_record_of_same_revision() { + let incoming = record(1, true, 1000); + assert_eq!( + decide_store(Some((1, false, 500)), &incoming), + StoreDecision::Write + ); + } + + #[test] + fn active_record_does_not_resurrect_tombstone() { + let incoming = record(1, false, 1000); + assert_eq!( + decide_store(Some((1, true, 500)), &incoming), + StoreDecision::Ignore + ); + // Even an older active record loses to a newer tombstone. + let incoming = record(1, false, 1000); + assert_eq!( + decide_store(Some((2, true, 500)), &incoming), + StoreDecision::Ignore + ); + } + + #[test] + fn republish_refreshes_expiry() { + let incoming = record(1, false, 2000); + assert_eq!( + decide_store(Some((1, false, 500)), &incoming), + StoreDecision::RefreshExpiry(2000) + ); + let stale = record(1, false, 100); + assert_eq!( + decide_store(Some((1, false, 500)), &stale), + StoreDecision::Ignore + ); + } + + #[test] + fn missing_record_is_written() { + let incoming = record(1, false, 1000); + assert_eq!(decide_store(None, &incoming), StoreDecision::Write); + } + + #[test] + fn validate_checks_key_and_clamps_ttl() { + let net = NetworkId::from_name("test"); + let now = 1_000_000; + let rec = record(1, false, now + ACTIVE_RECORD_TTL.as_millis() as u64 * 10); + + // Correct exact key: accepted, expiry clamped to the maximum TTL. + let ok = validate_store( + StoreRecordRequest { + key: DhtKey::exact(&net, "massive attack"), + record: rec.clone(), + }, + &net, + now, + ) + .expect("valid"); + assert_eq!(ok.expires_at_ms, now + ACTIVE_RECORD_TTL.as_millis() as u64); + + // Correct token key: accepted. + assert!( + validate_store( + StoreRecordRequest { + key: DhtKey::token(&net, "massive"), + record: rec.clone(), + }, + &net, + now, + ) + .is_ok() + ); + + // Unrelated key: rejected. + assert!( + validate_store( + StoreRecordRequest { + key: DhtKey::exact(&net, "portishead"), + record: rec.clone(), + }, + &net, + now, + ) + .is_err() + ); + + // Expired record: rejected. + let expired = record(1, false, now - 1); + assert!( + validate_store( + StoreRecordRequest { + key: DhtKey::exact(&net, "massive attack"), + record: expired, + }, + &net, + now, + ) + .is_err() + ); + + // Inconsistent normalization: rejected. + let mut bad = rec; + bad.artist.normalized_name = "something else".into(); + assert!( + validate_store( + StoreRecordRequest { + key: DhtKey::exact(&net, "something else"), + record: bad, + }, + &net, + now, + ) + .is_err() + ); + } +} diff --git a/crates/artist-dht/src/error.rs b/crates/artist-dht/src/error.rs new file mode 100644 index 0000000..e8ccf3f --- /dev/null +++ b/crates/artist-dht/src/error.rs @@ -0,0 +1,65 @@ +//! Error types for the artist-dht library. + +/// Convenient result alias used across the library. +pub type Result = std::result::Result; + +/// All errors that can be returned by the public API of this library. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum ArtistDhtError { + /// A local database operation failed. + #[error("database error: {0}")] + Database(String), + + /// The underlying network layer reported an error. + #[error("network error: {0}")] + Network(String), + + /// The artist name is empty or normalizes to nothing. + #[error("invalid artist name")] + InvalidArtistName, + + /// The artist name exceeds the maximum allowed length. + #[error("artist name is too long")] + ArtistNameTooLong, + + /// No artist with the given id exists locally. + #[error("artist not found")] + ArtistNotFound, + + /// Only locally created artists can be deleted. + #[error("cannot delete a remote artist")] + CannotDeleteRemoteArtist, + + /// A DHT request did not receive a response in time. + #[error("request timed out")] + Timeout, + + /// The lookup exhausted its request budget without finishing. + #[error("lookup budget exhausted")] + LookupBudgetExhausted, + + /// A stored peer ticket could not be parsed. + #[error("invalid peer ticket: {0}")] + InvalidTicket(String), + + /// A peer violated the DHT protocol. + #[error("protocol error: {0}")] + Protocol(String), + + /// The service is shutting down and no longer accepts operations. + #[error("service is shutting down")] + ShuttingDown, +} + +impl From for ArtistDhtError { + fn from(err: federation_net::NetworkError) -> Self { + use federation_net::NetworkError; + match err { + NetworkError::Timeout => Self::Timeout, + NetworkError::ShuttingDown => Self::ShuttingDown, + NetworkError::InvalidTicket(msg) => Self::InvalidTicket(msg), + other => Self::Network(other.to_string()), + } + } +} diff --git a/crates/artist-dht/src/lib.rs b/crates/artist-dht/src/lib.rs new file mode 100644 index 0000000..7d9e816 --- /dev/null +++ b/crates/artist-dht/src/lib.rs @@ -0,0 +1,92 @@ +//! # artist-dht +//! +//! A proof-of-concept distributed artist directory on top of +//! [`federation_net`]. Every running node is simultaneously a client, a DHT +//! router and a storage node — there are no dedicated bootstrap, index or +//! search servers. +//! +//! ## How it works +//! +//! * Every peer derives a stable 256-bit [`NodeId`] from its persistent +//! `federation-net` endpoint id. +//! * Artist records are published under BLAKE3-derived [`DhtKey`]s: one exact +//! key for the whole normalized name plus one key per name token. +//! * Records are replicated to the `K` nodes whose ids are XOR-closest to +//! each key, discovered with an iterative Kademlia-style lookup (never a +//! broadcast). +//! * Peers learn about each other through a Hello/PeerExchange gossip that +//! runs automatically on every new connection; connections to further +//! nodes are opened on demand from stored tickets. +//! * Deletions propagate as tombstones that win over active records of the +//! same or lower revision; replicas expire by TTL and owners republish +//! periodically. +//! +//! ## Example +//! +//! ```no_run +//! use artist_dht::{ArtistDhtConfig, ArtistDhtService}; +//! use federation_net::NetworkId; +//! +//! # async fn run() -> artist_dht::Result<()> { +//! let config = ArtistDhtConfig::builder() +//! .data_dir("./peer-a") +//! .network_id(NetworkId::from_name("demo-artists")) +//! .build()?; +//! let (service, mut events) = ArtistDhtService::start(config).await?; +//! println!("share this ticket: {}", service.ticket().await?); +//! +//! let (artist, stats) = service.add_artist("Massive Attack".into()).await?; +//! println!("published {} under {} keys", artist.name, stats.keys); +//! +//! let outcome = service.search_network("massive").await?; +//! for artist in &outcome.network_results { +//! println!("found {} owned by {}", artist.name, artist.owner); +//! } +//! # while let Some(event) = events.recv().await { drop(event); } +//! # service.shutdown().await +//! # } +//! ``` + +#![warn(missing_docs)] +#![forbid(unsafe_code)] + +mod config; +mod database; +mod dht; +mod error; +mod message; +mod node; +mod normalization; +mod record; +mod request; +mod routing; +mod service; + +pub use config::{ + ArtistDhtConfig, ArtistDhtConfigBuilder, DEFAULT_EXPIRE_INTERVAL, DEFAULT_LOOKUP_TIMEOUT, + DEFAULT_REPUBLISH_INTERVAL, DEFAULT_REQUEST_TIMEOUT, DEFAULT_TRANSPORT_TIMEOUT, +}; +pub use error::{ArtistDhtError, Result}; +pub use message::{ + ArtistDhtMessage, DHT_PROTOCOL_VERSION, FindNodeRequest, FindNodeResponse, FindValueRequest, + FindValueResponse, Hello, MAX_PEER_EXCHANGE_CONTACTS, MAX_RECORDS_PER_RESPONSE, PeerExchange, + PingRequest, PongResponse, RequestEnvelope, RequestId, ResponseEnvelope, StoreRecordRequest, + StoreRecordResponse, +}; +pub use normalization::{normalize_artist_name, tokenize}; +pub use record::{ + ACTIVE_RECORD_TTL, Artist, ArtistId, DhtKey, MAX_ARTIST_NAME_BYTES, MAX_TOKENS_PER_ARTIST, + PeerId, StoredArtistRecord, TOMBSTONE_TTL, +}; +pub use request::MAX_PENDING_REQUESTS; +pub use routing::{ + ALPHA, Distance, K, MAX_LOOKUP_REQUESTS, NodeContact, NodeId, RoutingTable, distance, + key_distance, +}; +pub use service::{ + ArtistDhtEvent, ArtistDhtEventReceiver, ArtistDhtService, PublishStats, SCHEMA_NAME, + SearchOutcome, +}; + +// Re-exported types from the transport layer that appear in this API. +pub use federation_net::{EndpointId, NetworkId, PeerTicket}; diff --git a/crates/artist-dht/src/message.rs b/crates/artist-dht/src/message.rs new file mode 100644 index 0000000..be94cbe --- /dev/null +++ b/crates/artist-dht/src/message.rs @@ -0,0 +1,174 @@ +//! The domain protocol carried over `federation-net`. + +use std::fmt; + +use federation_net::EndpointId; +use serde::{Deserialize, Serialize}; + +use crate::record::{DhtKey, StoredArtistRecord}; +use crate::routing::{NodeContact, NodeId}; + +/// Version of the artist-dht protocol. +pub const DHT_PROTOCOL_VERSION: u16 = 1; +/// Maximum number of contacts in a single [`PeerExchange`]. +pub const MAX_PEER_EXCHANGE_CONTACTS: usize = 32; +/// Maximum number of records in a single [`FindValueResponse`]. +pub const MAX_RECORDS_PER_RESPONSE: usize = 100; + +/// Correlates a response with its request. +#[derive(Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct RequestId([u8; 16]); + +impl RequestId { + /// Generates a random request id. + pub fn random() -> Self { + Self(rand::random()) + } + + /// Returns the raw bytes. + pub fn as_bytes(&self) -> &[u8; 16] { + &self.0 + } +} + +impl fmt::Debug for RequestId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "RequestId(")?; + for byte in &self.0 { + write!(f, "{byte:02x}")?; + } + write!(f, ")") + } +} + +/// Wraps a request payload with its correlation id. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RequestEnvelope { + /// Correlation id; echoed back in the response. + pub request_id: RequestId, + /// The request itself. + pub payload: T, +} + +/// Wraps a response payload with the correlation id of its request. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ResponseEnvelope { + /// Correlation id of the request being answered. + pub request_id: RequestId, + /// The response itself. + pub payload: T, +} + +/// Introduction sent right after a connection is established. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Hello { + /// DHT identifier of the sender. + pub node_id: NodeId, + /// Transport identifier of the sender (informational; the authenticated + /// id always comes from the connection itself). + pub peer_id: EndpointId, + /// Ticket other peers can use to reach the sender. + pub ticket: String, + /// Protocol version of the sender. + pub protocol_version: u16, +} + +/// A batch of known contacts, shared after [`Hello`]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PeerExchange { + /// Up to [`MAX_PEER_EXCHANGE_CONTACTS`] contacts. + pub peers: Vec, +} + +/// Liveness probe. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PingRequest {} + +/// Reply to [`PingRequest`]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PongResponse { + /// DHT identifier of the responder. + pub node_id: NodeId, +} + +/// Asks for the closest known nodes to `target`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FindNodeRequest { + /// Point of the key space to search around. + pub target: NodeId, +} + +/// Reply to [`FindNodeRequest`]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FindNodeResponse { + /// Up to `K` known nodes closest to the target. + pub nodes: Vec, +} + +/// Asks for records stored under `key`, or the closest nodes to it. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FindValueRequest { + /// The DHT key to look up. + pub key: DhtKey, +} + +/// Reply to [`FindValueRequest`]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum FindValueResponse { + /// The responder stores records under the key. + Records { + /// Up to [`MAX_RECORDS_PER_RESPONSE`] non-expired records. + records: Vec, + }, + /// The responder has nothing stored; here are closer nodes instead. + CloserNodes { + /// Up to `K` known nodes closest to the key. + nodes: Vec, + }, +} + +/// Asks the receiver to store a replica of a record. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StoreRecordRequest { + /// The key the record is published under. + pub key: DhtKey, + /// The record to store. + pub record: StoredArtistRecord, +} + +/// Reply to [`StoreRecordRequest`]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StoreRecordResponse { + /// `true` if the record was accepted and stored (or refreshed). + pub stored: bool, +} + +/// Every message exchanged between artist-dht peers. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[allow(clippy::large_enum_variant)] +pub enum ArtistDhtMessage { + /// Introduction after connect. + Hello(Hello), + /// Contact gossip after `Hello`. + PeerExchange(PeerExchange), + + /// Liveness probe. + Ping(RequestEnvelope), + /// Reply to `Ping`. + Pong(ResponseEnvelope), + + /// Node lookup request. + FindNode(RequestEnvelope), + /// Reply to `FindNode`. + FindNodeResult(ResponseEnvelope), + + /// Value lookup request. + FindValue(RequestEnvelope), + /// Reply to `FindValue`. + FindValueResult(ResponseEnvelope), + + /// Replication request. + StoreRecord(RequestEnvelope), + /// Reply to `StoreRecord`. + StoreRecordResult(ResponseEnvelope), +} diff --git a/crates/artist-dht/src/node.rs b/crates/artist-dht/src/node.rs new file mode 100644 index 0000000..bf72cd8 --- /dev/null +++ b/crates/artist-dht/src/node.rs @@ -0,0 +1,808 @@ +//! The DHT node: event handling, peer exchange, iterative lookups and +//! publication. + +use std::collections::{HashMap, HashSet}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; +use std::time::Instant; + +use federation_net::{EndpointId, NetworkEngine, NetworkEvent, NetworkEventReceiver, PeerTicket}; +use futures::future::join_all; +use tokio::sync::mpsc; +use tokio::time::timeout; +use tracing::{debug, info, warn}; + +use crate::config::ArtistDhtConfig; +use crate::database::Database; +use crate::dht::validate_store; +use crate::error::{ArtistDhtError, Result}; +use crate::message::{ + ArtistDhtMessage, DHT_PROTOCOL_VERSION, FindNodeRequest, FindNodeResponse, FindValueRequest, + FindValueResponse, Hello, MAX_PEER_EXCHANGE_CONTACTS, PeerExchange, PingRequest, PongResponse, + RequestEnvelope, RequestId, ResponseEnvelope, StoreRecordRequest, StoreRecordResponse, +}; +use crate::record::{ACTIVE_RECORD_TTL, Artist, DhtKey, StoredArtistRecord, TOMBSTONE_TTL, now_ms}; +use crate::request::{DhtResponse, PendingRequests}; +use crate::routing::{ALPHA, K, MAX_LOOKUP_REQUESTS, NodeContact, NodeId, RoutingTable, distance}; +use crate::service::{ArtistDhtEvent, PublishStats}; + +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(PoisonError::into_inner) +} + +/// An outbound DHT request, before it is wrapped in an envelope. +enum OutboundRequest { + Ping, + FindNode(FindNodeRequest), + FindValue(FindValueRequest), + Store(StoreRecordRequest), +} + +/// Result of one iterative lookup. +pub(crate) struct LookupOutcome { + /// Records found (value lookups only). + pub records: Vec, + /// Closest known contacts to the target, best first, at most `K`. + pub closest: Vec, + /// Number of distinct peers actually queried. + pub queried: usize, + /// Number of distinct nodes known to the lookup (seeds + discovered). + pub discovered: usize, +} + +/// Shared state of one DHT node. +pub(crate) struct Node { + pub engine: NetworkEngine, + pub db: Database, + pub config: ArtistDhtConfig, + pub node_id: NodeId, + pub endpoint_id: EndpointId, + routing: Mutex, + pending: PendingRequests, + /// Peers we already introduced ourselves to (per connection). + hello_sent: Mutex>, + /// Peers we already gossiped contacts to (per connection). + exchange_sent: Mutex>, + events: Mutex>>, + /// Set once the post-startup republish has been triggered. + initial_republish_done: AtomicBool, + shutting_down: AtomicBool, +} + +impl Node { + pub fn new( + engine: NetworkEngine, + db: Database, + config: ArtistDhtConfig, + events: mpsc::Sender, + ) -> Self { + let endpoint_id = engine.endpoint_id(); + let node_id = NodeId::from_endpoint(&endpoint_id); + Self { + engine, + db, + config, + node_id, + endpoint_id, + routing: Mutex::new(RoutingTable::new(node_id)), + pending: PendingRequests::default(), + hello_sent: Mutex::new(HashSet::new()), + exchange_sent: Mutex::new(HashSet::new()), + events: Mutex::new(Some(events)), + initial_republish_done: AtomicBool::new(false), + shutting_down: AtomicBool::new(false), + } + } + + pub fn is_shutting_down(&self) -> bool { + self.shutting_down.load(Ordering::SeqCst) + } + + pub fn begin_shutdown(&self) { + self.shutting_down.store(true, Ordering::SeqCst); + *lock(&self.events) = None; + } + + pub fn ensure_running(&self) -> Result<()> { + if self.is_shutting_down() { + Err(ArtistDhtError::ShuttingDown) + } else { + Ok(()) + } + } + + async fn emit(&self, event: ArtistDhtEvent) { + let sender = lock(&self.events).clone(); + if let Some(sender) = sender { + let _ = sender.send(event).await; + } + } + + /// All known DHT contacts. + pub fn known_contacts(&self) -> Vec { + lock(&self.routing).contacts() + } + + /// Seeds the routing table (used at startup with persisted contacts). + pub fn seed_contacts(&self, contacts: Vec) { + let mut routing = lock(&self.routing); + for contact in contacts { + if contact.peer_id != self.endpoint_id { + routing.upsert(contact); + } + } + } + + /// Adds or refreshes a contact learned from the network. + /// + /// The node id is always re-derived from the endpoint id instead of + /// trusting the gossiped value. The first contact ever learned triggers + /// the post-startup republish. + async fn upsert_contact(self: &Arc, peer_id: EndpointId, ticket: String) { + if peer_id == self.endpoint_id { + return; + } + let contact = NodeContact { + node_id: NodeId::from_endpoint(&peer_id), + peer_id, + ticket, + last_seen_ms: now_ms(), + }; + let is_new = lock(&self.routing).upsert(contact.clone()); + if let Err(err) = self.db.upsert_known_peer(&contact).await { + warn!(error = %err, "failed to persist known peer"); + } + if is_new { + info!(peer = %contact.peer_id, node = %contact.node_id, "learned new DHT contact"); + self.emit(ArtistDhtEvent::ContactDiscovered { + contact: contact.clone(), + }) + .await; + } + self.maybe_trigger_initial_republish(); + } + + /// Spawns the post-startup republish once at least one contact is known. + pub fn maybe_trigger_initial_republish(self: &Arc) { + if lock(&self.routing).is_empty() || self.is_shutting_down() { + return; + } + if self.initial_republish_done.swap(true, Ordering::SeqCst) { + return; + } + let node = self.clone(); + tokio::spawn(async move { + match node.republish_all().await { + Ok(stats) => info!( + records = stats.records, + keys = stats.keys, + nodes = stats.remote_nodes, + "post-startup republish finished" + ), + Err(err) => warn!(error = %err, "post-startup republish failed"), + } + }); + } + + /// Consumes `federation-net` events until the engine shuts down. + pub async fn run_event_loop( + self: Arc, + mut receiver: NetworkEventReceiver, + ) { + while let Some(event) = receiver.recv().await { + match event { + NetworkEvent::PeerConnected { peer_id, .. } => { + debug!(peer = %peer_id, "peer connected"); + self.emit(ArtistDhtEvent::PeerConnected { peer_id }).await; + self.send_hello(peer_id).await; + } + NetworkEvent::PeerDisconnected { peer_id, .. } => { + debug!(peer = %peer_id, "peer disconnected"); + lock(&self.hello_sent).remove(&peer_id); + lock(&self.exchange_sent).remove(&peer_id); + self.emit(ArtistDhtEvent::PeerDisconnected { peer_id }) + .await; + } + NetworkEvent::MessageReceived { peer_id, message } => { + self.on_message(peer_id, message).await; + } + NetworkEvent::ProtocolError { peer_id, error } => { + self.emit(ArtistDhtEvent::Error { + message: match peer_id { + Some(peer) => format!("transport error with {peer}: {error}"), + None => format!("transport error: {error}"), + }, + }) + .await; + } + } + } + debug!("network event loop finished"); + } + + async fn send_message(&self, peer: EndpointId, message: &ArtistDhtMessage) -> Result<()> { + self.engine.send(peer, message).await.map_err(Into::into) + } + + async fn send_hello(self: &Arc, peer: EndpointId) { + // Mark before sending so a crossing Hello does not trigger an echo. + if !lock(&self.hello_sent).insert(peer) { + return; + } + let ticket = match self.engine.ticket().await { + Ok(ticket) => ticket.to_string(), + Err(err) => { + warn!(error = %err, "cannot create own ticket for hello"); + lock(&self.hello_sent).remove(&peer); + return; + } + }; + let hello = ArtistDhtMessage::Hello(Hello { + node_id: self.node_id, + peer_id: self.endpoint_id, + ticket, + protocol_version: DHT_PROTOCOL_VERSION, + }); + if let Err(err) = self.send_message(peer, &hello).await { + debug!(peer = %peer, error = %err, "failed to send hello"); + lock(&self.hello_sent).remove(&peer); + } + } + + async fn send_peer_exchange(self: &Arc, peer: EndpointId) { + if !lock(&self.exchange_sent).insert(peer) { + return; + } + let mut peers: Vec = self + .known_contacts() + .into_iter() + .filter(|contact| contact.peer_id != peer && contact.peer_id != self.endpoint_id) + .collect(); + // Prefer the most recently seen contacts. + peers.sort_by_key(|contact| std::cmp::Reverse(contact.last_seen_ms)); + peers.truncate(MAX_PEER_EXCHANGE_CONTACTS); + if peers.is_empty() { + return; + } + debug!(peer = %peer, count = peers.len(), "sending peer exchange"); + let message = ArtistDhtMessage::PeerExchange(PeerExchange { peers }); + if let Err(err) = self.send_message(peer, &message).await { + debug!(peer = %peer, error = %err, "failed to send peer exchange"); + } + } + + async fn on_message(self: &Arc, peer: EndpointId, message: ArtistDhtMessage) { + lock(&self.routing).touch(&peer, now_ms()); + match message { + ArtistDhtMessage::Hello(hello) => self.on_hello(peer, hello).await, + ArtistDhtMessage::PeerExchange(exchange) => { + self.on_peer_exchange(peer, exchange).await; + } + ArtistDhtMessage::Ping(env) => { + let response = ArtistDhtMessage::Pong(ResponseEnvelope { + request_id: env.request_id, + payload: PongResponse { + node_id: self.node_id, + }, + }); + let _ = self.send_message(peer, &response).await; + } + ArtistDhtMessage::FindNode(env) => { + let nodes = self.closest_for_response(env.payload.target.as_bytes(), &peer); + let response = ArtistDhtMessage::FindNodeResult(ResponseEnvelope { + request_id: env.request_id, + payload: FindNodeResponse { nodes }, + }); + let _ = self.send_message(peer, &response).await; + } + ArtistDhtMessage::FindValue(env) => { + let payload = self.answer_find_value(&env.payload, &peer).await; + let response = ArtistDhtMessage::FindValueResult(ResponseEnvelope { + request_id: env.request_id, + payload, + }); + let _ = self.send_message(peer, &response).await; + } + ArtistDhtMessage::StoreRecord(env) => { + let stored = self.answer_store(env.payload, &peer).await; + let response = ArtistDhtMessage::StoreRecordResult(ResponseEnvelope { + request_id: env.request_id, + payload: StoreRecordResponse { stored }, + }); + let _ = self.send_message(peer, &response).await; + } + ArtistDhtMessage::Pong(env) => { + self.pending + .complete(&env.request_id, &peer, DhtResponse::Pong(env.payload)); + } + ArtistDhtMessage::FindNodeResult(env) => { + self.pending + .complete(&env.request_id, &peer, DhtResponse::FindNode(env.payload)); + } + ArtistDhtMessage::FindValueResult(env) => { + self.pending + .complete(&env.request_id, &peer, DhtResponse::FindValue(env.payload)); + } + ArtistDhtMessage::StoreRecordResult(env) => { + self.pending + .complete(&env.request_id, &peer, DhtResponse::Store(env.payload)); + } + } + } + + async fn on_hello(self: &Arc, peer: EndpointId, hello: Hello) { + if hello.protocol_version != DHT_PROTOCOL_VERSION { + warn!(peer = %peer, version = hello.protocol_version, "unsupported DHT protocol version"); + return; + } + // The authenticated identity comes from the connection; the id fields + // inside the payload must be consistent with it. + if hello.peer_id != peer || hello.node_id != NodeId::from_endpoint(&peer) { + warn!(peer = %peer, "hello with inconsistent identity; ignoring"); + self.emit(ArtistDhtEvent::Error { + message: format!("peer {peer} sent a hello with a mismatched identity"), + }) + .await; + return; + } + debug!(peer = %peer, "received hello"); + self.upsert_contact(peer, hello.ticket).await; + // Introduce ourselves if the remote connected first, then gossip. + self.send_hello(peer).await; + self.send_peer_exchange(peer).await; + } + + async fn on_peer_exchange(self: &Arc, peer: EndpointId, exchange: PeerExchange) { + let contacts = sanitize_peer_exchange(self.endpoint_id, peer, exchange.peers); + let accepted = contacts.len(); + for contact in contacts { + self.upsert_contact(contact.peer_id, contact.ticket).await; + } + debug!(peer = %peer, accepted, "processed peer exchange"); + } + + /// Contacts for a FindNode/FindValue response: closest to the target, + /// excluding the requester itself. + fn closest_for_response(&self, target: &[u8; 32], requester: &EndpointId) -> Vec { + lock(&self.routing) + .closest(target, K + 1) + .into_iter() + .filter(|contact| &contact.peer_id != requester) + .take(K) + .collect() + } + + async fn answer_find_value( + &self, + request: &FindValueRequest, + requester: &EndpointId, + ) -> FindValueResponse { + match self.db.dht_records_by_key(request.key, now_ms()).await { + Ok(records) if !records.is_empty() => FindValueResponse::Records { records }, + Ok(_) => FindValueResponse::CloserNodes { + nodes: self.closest_for_response(request.key.as_bytes(), requester), + }, + Err(err) => { + warn!(error = %err, "find-value lookup in the local store failed"); + FindValueResponse::CloserNodes { + nodes: self.closest_for_response(request.key.as_bytes(), requester), + } + } + } + } + + async fn answer_store(&self, request: StoreRecordRequest, sender: &EndpointId) -> bool { + if self.is_shutting_down() { + return false; + } + let key = request.key; + match validate_store(request, &self.config.network_id, now_ms()) { + Ok(record) => { + let artist_id = record.artist.id; + let deleted = record.artist.deleted; + match self.db.store_dht_record(key, record).await { + Ok(stored) => { + if stored { + info!( + artist = %artist_id, + tombstone = deleted, + from = %sender, + "stored DHT record" + ); + } + stored + } + Err(err) => { + warn!(error = %err, "failed to store DHT record"); + false + } + } + } + Err(reason) => { + warn!(from = %sender, reason = %reason, "rejected DHT store request"); + false + } + } + } + + /// Makes sure a connection to the contact exists, dialing its ticket if + /// necessary. The Hello exchange runs asynchronously via the event loop. + async fn ensure_connected(&self, contact: &NodeContact) -> Result { + self.ensure_running()?; + if self.engine.is_connected(contact.peer_id) { + return Ok(contact.peer_id); + } + let ticket: PeerTicket = contact + .ticket + .parse() + .map_err(|err| ArtistDhtError::InvalidTicket(format!("{err}")))?; + debug!(peer = %contact.peer_id, "connecting on demand"); + let peer = self.engine.connect(ticket).await?; + Ok(peer) + } + + /// Sends one request and awaits its response, cleaning up the pending + /// entry on timeout. + async fn request( + &self, + contact: &NodeContact, + request: OutboundRequest, + ) -> Result { + let peer = self.ensure_connected(contact).await?; + let request_id = RequestId::random(); + let receiver = self.pending.register(request_id, peer)?; + tracing::trace!(pending = self.pending.len(), peer = %peer, "sending DHT request"); + let message = match request { + OutboundRequest::Ping => ArtistDhtMessage::Ping(RequestEnvelope { + request_id, + payload: PingRequest {}, + }), + OutboundRequest::FindNode(payload) => ArtistDhtMessage::FindNode(RequestEnvelope { + request_id, + payload, + }), + OutboundRequest::FindValue(payload) => ArtistDhtMessage::FindValue(RequestEnvelope { + request_id, + payload, + }), + OutboundRequest::Store(payload) => ArtistDhtMessage::StoreRecord(RequestEnvelope { + request_id, + payload, + }), + }; + if let Err(err) = self.send_message(peer, &message).await { + self.pending.remove(&request_id); + return Err(err); + } + match timeout(self.config.request_timeout, receiver).await { + Ok(Ok(response)) => { + lock(&self.routing).touch(&peer, now_ms()); + Ok(response) + } + Ok(Err(_)) => { + self.pending.remove(&request_id); + Err(ArtistDhtError::Protocol("response channel closed".into())) + } + Err(_) => { + self.pending.remove(&request_id); + Err(ArtistDhtError::Timeout) + } + } + } + + /// Measures the round-trip time to a known contact and verifies its + /// DHT identity. + pub async fn ping(&self, contact: &NodeContact) -> Result { + let started = Instant::now(); + match self.request(contact, OutboundRequest::Ping).await? { + DhtResponse::Pong(pong) => { + if pong.node_id != NodeId::from_endpoint(&contact.peer_id) { + return Err(ArtistDhtError::Protocol( + "pong with a mismatched node id".into(), + )); + } + lock(&self.routing).touch(&contact.peer_id, now_ms()); + Ok(started.elapsed()) + } + _ => Err(ArtistDhtError::Protocol( + "unexpected response to ping".into(), + )), + } + } + + /// Iterative Kademlia-style lookup. + /// + /// With `find_value: None` this is a node lookup converging on the + /// closest known nodes to `target`; with `Some(key)` it sends `FindValue` + /// and stops as soon as records are found. Never broadcasts: at most + /// [`ALPHA`] requests run concurrently and at most + /// [`MAX_LOOKUP_REQUESTS`] are sent in total, all bounded by the lookup + /// timeout. + pub async fn lookup(&self, target: [u8; 32], find_value: Option) -> LookupOutcome { + let started = Instant::now(); + let deadline = started + self.config.lookup_timeout; + let mut candidates: Vec = lock(&self.routing).closest(&target, K); + let mut known: HashSet = + candidates.iter().map(|contact| contact.peer_id).collect(); + let mut queried: HashSet = HashSet::new(); + let mut records: HashMap<(crate::record::ArtistId, EndpointId), StoredArtistRecord> = + HashMap::new(); + let mut sent = 0usize; + + debug!(target = %NodeId::from_bytes(target), seeds = candidates.len(), "lookup started"); + + loop { + if Instant::now() >= deadline { + debug!("lookup deadline reached"); + break; + } + candidates.sort_by_key(|contact| distance(contact.node_id.as_bytes(), &target)); + let budget = MAX_LOOKUP_REQUESTS.saturating_sub(sent); + let batch: Vec = candidates + .iter() + .filter(|contact| !queried.contains(&contact.peer_id)) + .take(ALPHA.min(budget)) + .cloned() + .collect(); + if batch.is_empty() { + break; + } + sent += batch.len(); + for contact in &batch { + queried.insert(contact.peer_id); + } + + let futures = batch.iter().map(|contact| { + let request = match find_value { + Some(key) => OutboundRequest::FindValue(FindValueRequest { key }), + None => OutboundRequest::FindNode(FindNodeRequest { + target: NodeId::from_bytes(target), + }), + }; + self.request(contact, request) + }); + let results = join_all(futures).await; + + let mut found_records = false; + for (contact, result) in batch.iter().zip(results) { + let nodes = match result { + Ok(DhtResponse::FindNode(response)) => response.nodes, + Ok(DhtResponse::FindValue(FindValueResponse::Records { records: found })) => { + for record in found { + let key = (record.artist.id, record.artist.owner); + match records.get(&key) { + Some(existing) if !record_supersedes(&record, existing) => {} + _ => { + records.insert(key, record); + } + } + } + found_records = true; + Vec::new() + } + Ok(DhtResponse::FindValue(FindValueResponse::CloserNodes { nodes })) => nodes, + Ok(_) => Vec::new(), + Err(err) => { + debug!(peer = %contact.peer_id, error = %err, "lookup request failed"); + Vec::new() + } + }; + for node in nodes.into_iter().take(K) { + if node.peer_id == self.endpoint_id || !known.insert(node.peer_id) { + continue; + } + // Re-derive the node id instead of trusting gossip. + candidates.push(NodeContact { + node_id: NodeId::from_endpoint(&node.peer_id), + peer_id: node.peer_id, + ticket: node.ticket, + last_seen_ms: now_ms(), + }); + } + } + if find_value.is_some() && found_records { + break; + } + if sent >= MAX_LOOKUP_REQUESTS { + debug!("lookup request budget exhausted"); + break; + } + } + + candidates.sort_by_key(|contact| distance(contact.node_id.as_bytes(), &target)); + candidates.truncate(K); + info!( + queried = queried.len(), + discovered = known.len(), + records = records.len(), + elapsed_ms = started.elapsed().as_millis() as u64, + "lookup finished" + ); + LookupOutcome { + records: records.into_values().collect(), + closest: candidates, + queried: queried.len(), + discovered: known.len(), + } + } + + /// Publishes one artist record (active or tombstone) under all its DHT + /// keys to the closest known nodes. Returns + /// `(keys, remote nodes stored, local replica stored)`. + pub async fn publish_artist(&self, artist: &Artist) -> Result { + self.ensure_running()?; + let ttl = if artist.deleted { + TOMBSTONE_TTL + } else { + ACTIVE_RECORD_TTL + }; + let record = StoredArtistRecord { + artist: artist.clone(), + publisher: self.endpoint_id, + expires_at_ms: now_ms() + ttl.as_millis() as u64, + }; + let keys = artist.dht_keys(&self.config.network_id); + let mut remote_nodes: HashSet = HashSet::new(); + let mut local_replica = false; + + for key in &keys { + let outcome = self.lookup(*key.as_bytes(), None).await; + let targets = outcome.closest; + + // The record belongs on this node too if it is among the K + // closest (always true while the network is smaller than K). + let own_distance = distance(self.node_id.as_bytes(), key.as_bytes()); + let self_is_close = targets.len() < K + || targets.last().is_none_or(|farthest| { + own_distance <= distance(farthest.node_id.as_bytes(), key.as_bytes()) + }); + if self_is_close { + match self.db.store_dht_record(*key, record.clone()).await { + Ok(_) => local_replica = true, + Err(err) => warn!(error = %err, "failed to store own replica"), + } + } + + let stores = targets.iter().map(|contact| async { + let result = self + .request( + contact, + OutboundRequest::Store(StoreRecordRequest { + key: *key, + record: record.clone(), + }), + ) + .await; + (contact.peer_id, result) + }); + for (peer, result) in join_all(stores).await { + match result { + Ok(DhtResponse::Store(StoreRecordResponse { stored: true })) => { + remote_nodes.insert(peer); + } + Ok(DhtResponse::Store(StoreRecordResponse { stored: false })) => { + debug!(peer = %peer, "peer declined to store the record"); + } + Ok(_) => {} + Err(err) => debug!(peer = %peer, error = %err, "store request failed"), + } + } + } + info!( + artist = %artist.id, + tombstone = artist.deleted, + keys = keys.len(), + nodes = remote_nodes.len(), + "published artist record" + ); + Ok(PublishStats { + records: 1, + keys: keys.len(), + remote_nodes: remote_nodes.len(), + local_replica, + }) + } + + /// Republishes every local record that is still alive. + pub async fn republish_all(&self) -> Result { + self.ensure_running()?; + let artists = self.db.local_artists_for_republish(now_ms()).await?; + let mut total = PublishStats::default(); + for artist in &artists { + let stats = self.publish_artist(artist).await?; + total.records += 1; + total.keys += stats.keys; + // remote_nodes counts unique nodes per record; report the widest + // replication seen across records. + total.remote_nodes = total.remote_nodes.max(stats.remote_nodes); + total.local_replica |= stats.local_replica; + } + info!( + records = total.records, + keys = total.keys, + "republish finished" + ); + Ok(total) + } + + /// Drops expired replicas from the local store. + pub async fn sweep_expired(&self) { + match self.db.delete_expired_records(now_ms()).await { + Ok(0) => {} + Ok(count) => info!(count, "removed expired DHT records"), + Err(err) => warn!(error = %err, "failed to sweep expired records"), + } + } +} + +/// `true` if `candidate` should replace `existing` in a search result set. +pub(crate) fn record_supersedes( + candidate: &StoredArtistRecord, + existing: &StoredArtistRecord, +) -> bool { + let (c, e) = (&candidate.artist, &existing.artist); + c.revision > e.revision || (c.revision == e.revision && c.deleted && !e.deleted) +} + +/// Filters an incoming peer-exchange batch: drops our own contact, the +/// sender's contact and duplicate endpoint ids, and enforces the batch cap. +pub(crate) fn sanitize_peer_exchange( + own: EndpointId, + sender: EndpointId, + peers: Vec, +) -> Vec { + let mut seen: HashSet = HashSet::new(); + peers + .into_iter() + .take(MAX_PEER_EXCHANGE_CONTACTS) + .filter(|contact| { + contact.peer_id != own && contact.peer_id != sender && seen.insert(contact.peer_id) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + fn contact(seed: u8) -> NodeContact { + let peer = test_peer(seed); + NodeContact { + node_id: NodeId::from_endpoint(&peer), + peer_id: peer, + ticket: format!("fnet-test-{seed}"), + last_seen_ms: 0, + } + } + + #[test] + fn peer_exchange_drops_duplicates_self_and_sender() { + let own = test_peer(1); + let sender = test_peer(2); + let peers = vec![ + contact(3), + contact(3), // duplicate + contact(1), // ourselves + contact(2), // the sender + contact(4), + ]; + let sanitized = sanitize_peer_exchange(own, sender, peers); + let ids: Vec = sanitized.iter().map(|c| c.peer_id).collect(); + assert_eq!(ids, vec![test_peer(3), test_peer(4)]); + } + + #[test] + fn peer_exchange_is_capped() { + let own = test_peer(1); + let sender = test_peer(2); + let peers: Vec = (10..10 + MAX_PEER_EXCHANGE_CONTACTS as u8 + 8) + .map(contact) + .collect(); + let sanitized = sanitize_peer_exchange(own, sender, peers); + assert_eq!(sanitized.len(), MAX_PEER_EXCHANGE_CONTACTS); + } +} diff --git a/crates/artist-dht/src/normalization.rs b/crates/artist-dht/src/normalization.rs new file mode 100644 index 0000000..e295962 --- /dev/null +++ b/crates/artist-dht/src/normalization.rs @@ -0,0 +1,93 @@ +//! Artist name normalization and tokenization. + +use unicode_normalization::UnicodeNormalization; + +/// Normalizes an artist name for indexing and comparison. +/// +/// The algorithm is: Unicode NFKC normalization, lowercasing, replacing every +/// non-alphanumeric character with a space, collapsing repeated spaces and +/// trimming. The result is deterministic for a given input. +/// +/// ``` +/// use artist_dht::normalize_artist_name; +/// assert_eq!(normalize_artist_name("Massive Attack"), "massive attack"); +/// assert_eq!(normalize_artist_name(" MASSIVE ATTACK "), "massive attack"); +/// assert_eq!(normalize_artist_name("Massive-Attack"), "massive attack"); +/// assert_eq!(normalize_artist_name("Björk"), "björk"); +/// ``` +pub fn normalize_artist_name(input: &str) -> String { + let mut result = String::with_capacity(input.len()); + let mut pending_space = false; + for ch in input.nfkc() { + if ch.is_alphanumeric() { + if pending_space && !result.is_empty() { + result.push(' '); + } + pending_space = false; + for lower in ch.to_lowercase() { + result.push(lower); + } + } else { + pending_space = true; + } + } + result +} + +/// Splits a normalized name into search tokens. +/// +/// Empty tokens are ignored. +/// +/// ``` +/// use artist_dht::tokenize; +/// assert_eq!(tokenize("massive attack"), vec!["massive", "attack"]); +/// ``` +pub fn tokenize(normalized: &str) -> Vec { + normalized + .split(' ') + .filter(|token| !token.is_empty()) + .map(str::to_string) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn normalization_is_deterministic() { + let a = normalize_artist_name("Massive Attack"); + let b = normalize_artist_name("Massive Attack"); + assert_eq!(a, b); + assert_eq!(a, "massive attack"); + } + + #[test] + fn normalization_handles_case_and_whitespace() { + assert_eq!( + normalize_artist_name(" MASSIVE ATTACK "), + "massive attack" + ); + assert_eq!(normalize_artist_name("Massive-Attack"), "massive attack"); + assert_eq!( + normalize_artist_name("Massive___Attack!!!"), + "massive attack" + ); + assert_eq!(normalize_artist_name("Björk"), "björk"); + assert_eq!(normalize_artist_name(" "), ""); + assert_eq!(normalize_artist_name("!!!"), ""); + } + + #[test] + fn normalization_applies_nfkc() { + // U+FF21 FULLWIDTH LATIN CAPITAL LETTER A normalizes to 'A' → 'a'. + assert_eq!(normalize_artist_name("\u{FF21}BBA"), "abba"); + } + + #[test] + fn tokenize_splits_and_skips_empty() { + assert_eq!(tokenize("massive attack"), vec!["massive", "attack"]); + assert_eq!(tokenize(""), Vec::::new()); + assert_eq!(tokenize("solo"), vec!["solo"]); + } +} diff --git a/crates/artist-dht/src/record.rs b/crates/artist-dht/src/record.rs new file mode 100644 index 0000000..939ceeb --- /dev/null +++ b/crates/artist-dht/src/record.rs @@ -0,0 +1,300 @@ +//! Artist records, DHT keys and record lifetime rules. + +use std::fmt; +use std::str::FromStr; +use std::time::Duration; + +use federation_net::{EndpointId, NetworkId}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::error::ArtistDhtError; +use crate::normalization::{normalize_artist_name, tokenize}; + +/// Identifier of the peer that owns a record (its `federation-net` endpoint). +pub type PeerId = EndpointId; + +/// Maximum artist name length in UTF-8 bytes. +pub const MAX_ARTIST_NAME_BYTES: usize = 512; +/// Maximum number of tokens a single artist name may produce. +pub const MAX_TOKENS_PER_ARTIST: usize = 32; +/// Maximum lifetime of an active DHT record. +pub const ACTIVE_RECORD_TTL: Duration = Duration::from_secs(30 * 60); +/// Maximum lifetime of a tombstone. +pub const TOMBSTONE_TTL: Duration = Duration::from_secs(2 * 60 * 60); + +fn fmt_hex(bytes: &[u8; 32], f: &mut fmt::Formatter<'_>) -> fmt::Result { + for byte in bytes { + write!(f, "{byte:02x}")?; + } + Ok(()) +} + +/// Stable identifier of one artist record. +/// +/// Derived deterministically from the owning peer and a fresh UUID, so two +/// peers adding the same name produce two distinct records. +#[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] +pub struct ArtistId([u8; 32]); + +impl ArtistId { + /// Derives an artist id: `BLAKE3("artist-dht:artist:" || owner || uuid)`. + pub fn derive(owner: &EndpointId, uuid: &Uuid) -> Self { + let mut hasher = blake3::Hasher::new(); + hasher.update(b"artist-dht:artist:"); + hasher.update(owner.as_bytes()); + hasher.update(uuid.as_bytes()); + Self(*hasher.finalize().as_bytes()) + } + + /// Creates an id from raw bytes. + pub fn from_bytes(bytes: [u8; 32]) -> Self { + Self(bytes) + } + + /// Returns the raw bytes of this id. + pub fn as_bytes(&self) -> &[u8; 32] { + &self.0 + } + + /// Renders the id as lowercase hex. + pub fn to_hex(&self) -> String { + data_encoding::HEXLOWER.encode(&self.0) + } +} + +impl fmt::Display for ArtistId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt_hex(&self.0, f) + } +} + +impl fmt::Debug for ArtistId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "ArtistId(")?; + fmt_hex(&self.0, f)?; + write!(f, ")") + } +} + +impl FromStr for ArtistId { + type Err = ArtistDhtError; + + fn from_str(s: &str) -> Result { + let bytes = data_encoding::HEXLOWER_PERMISSIVE + .decode(s.trim().as_bytes()) + .map_err(|_| ArtistDhtError::ArtistNotFound)?; + let bytes: [u8; 32] = bytes + .try_into() + .map_err(|_| ArtistDhtError::ArtistNotFound)?; + Ok(Self(bytes)) + } +} + +/// One artist record as stored by its owner. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Artist { + /// Stable identifier of this record. + pub id: ArtistId, + /// Peer that created (and owns) the record. + pub owner: PeerId, + /// Human-readable artist name as entered by the user. + pub name: String, + /// Normalized form of the name, used for indexing. + pub normalized_name: String, + /// Monotonically increasing revision; bumped on every change. + pub revision: u64, + /// `true` if this record is a deletion tombstone. + pub deleted: bool, + /// Unix timestamp (milliseconds) of the last modification. + pub updated_at_ms: u64, +} + +impl Artist { + /// Returns the DHT keys this artist is published under: the exact key + /// plus one key per unique token. + pub fn dht_keys(&self, network_id: &NetworkId) -> Vec { + let mut keys = vec![DhtKey::exact(network_id, &self.normalized_name)]; + let mut seen = std::collections::HashSet::new(); + for token in tokenize(&self.normalized_name) { + if seen.insert(token.clone()) { + keys.push(DhtKey::token(network_id, &token)); + } + } + keys + } +} + +/// A replicated DHT entry: an artist plus replication metadata. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct StoredArtistRecord { + /// The artist record itself. + pub artist: Artist, + /// Peer that pushed this replica. Not cryptographically verified in the + /// PoC: the connection authenticates the direct sender, not the origin + /// of a replicated record. + pub publisher: PeerId, + /// Unix timestamp (milliseconds) after which the replica must be dropped. + pub expires_at_ms: u64, +} + +/// A 256-bit DHT key. Lives in the same key space as [`crate::NodeId`]. +#[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] +pub struct DhtKey([u8; 32]); + +impl DhtKey { + /// Key for exact-name lookups: + /// `BLAKE3(NetworkId || "artist:exact:" || normalized_name)`. + pub fn exact(network_id: &NetworkId, normalized_name: &str) -> Self { + let mut hasher = blake3::Hasher::new(); + hasher.update(network_id.as_bytes()); + hasher.update(b"artist:exact:"); + hasher.update(normalized_name.as_bytes()); + Self(*hasher.finalize().as_bytes()) + } + + /// Key for single-token lookups: + /// `BLAKE3(NetworkId || "artist:token:" || token)`. + pub fn token(network_id: &NetworkId, token: &str) -> Self { + let mut hasher = blake3::Hasher::new(); + hasher.update(network_id.as_bytes()); + hasher.update(b"artist:token:"); + hasher.update(token.as_bytes()); + Self(*hasher.finalize().as_bytes()) + } + + /// Creates a key from raw bytes. + pub fn from_bytes(bytes: [u8; 32]) -> Self { + Self(bytes) + } + + /// Returns the raw bytes of this key. + pub fn as_bytes(&self) -> &[u8; 32] { + &self.0 + } +} + +impl fmt::Display for DhtKey { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt_hex(&self.0, f) + } +} + +impl fmt::Debug for DhtKey { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "DhtKey(")?; + fmt_hex(&self.0, f)?; + write!(f, ")") + } +} + +/// Returns the current Unix time in milliseconds. +pub(crate) fn now_ms() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or_default() +} + +/// Validates a user-supplied artist name and returns its normalized form. +pub(crate) fn validate_name(name: &str) -> Result { + if name.len() > MAX_ARTIST_NAME_BYTES { + return Err(ArtistDhtError::ArtistNameTooLong); + } + let normalized = normalize_artist_name(name); + if normalized.is_empty() { + return Err(ArtistDhtError::InvalidArtistName); + } + if tokenize(&normalized).len() > MAX_TOKENS_PER_ARTIST { + return Err(ArtistDhtError::ArtistNameTooLong); + } + Ok(normalized) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + #[test] + fn artist_id_is_deterministic() { + let key = test_peer(7); + let uuid = Uuid::from_u128(42); + let a = ArtistId::derive(&key, &uuid); + let b = ArtistId::derive(&key, &uuid); + assert_eq!(a, b); + let other = ArtistId::derive(&key, &Uuid::from_u128(43)); + assert_ne!(a, other); + } + + #[test] + fn artist_id_hex_round_trip() { + let id = ArtistId::from_bytes([0xabu8; 32]); + let parsed: ArtistId = id.to_hex().parse().expect("parse"); + assert_eq!(parsed, id); + } + + #[test] + fn exact_key_is_deterministic_and_distinct() { + let net = NetworkId::from_name("test"); + let a = DhtKey::exact(&net, "massive attack"); + let b = DhtKey::exact(&net, "massive attack"); + assert_eq!(a, b); + assert_ne!(a, DhtKey::exact(&net, "portishead")); + // A different network yields different keys for the same name. + let other_net = NetworkId::from_name("other"); + assert_ne!(a, DhtKey::exact(&other_net, "massive attack")); + } + + #[test] + fn token_key_is_deterministic_and_distinct_from_exact() { + let net = NetworkId::from_name("test"); + let token = DhtKey::token(&net, "massive"); + assert_eq!(token, DhtKey::token(&net, "massive")); + assert_ne!(token, DhtKey::exact(&net, "massive")); + } + + #[test] + fn dht_keys_cover_exact_and_unique_tokens() { + let key = test_peer(7); + let artist = Artist { + id: ArtistId::from_bytes([1u8; 32]), + owner: key, + name: "Attack Attack".into(), + normalized_name: "attack attack".into(), + revision: 1, + deleted: false, + updated_at_ms: 0, + }; + let net = NetworkId::from_name("test"); + let keys = artist.dht_keys(&net); + // One exact key + one deduplicated token key. + assert_eq!(keys.len(), 2); + assert_eq!(keys[0], DhtKey::exact(&net, "attack attack")); + assert_eq!(keys[1], DhtKey::token(&net, "attack")); + } + + #[test] + fn name_validation() { + assert!(validate_name("Massive Attack").is_ok()); + assert!(matches!( + validate_name("!!!"), + Err(ArtistDhtError::InvalidArtistName) + )); + assert!(matches!( + validate_name(&"x".repeat(MAX_ARTIST_NAME_BYTES + 1)), + Err(ArtistDhtError::ArtistNameTooLong) + )); + let many_tokens = (0..MAX_TOKENS_PER_ARTIST + 1) + .map(|i| format!("t{i}")) + .collect::>() + .join(" "); + assert!(matches!( + validate_name(&many_tokens), + Err(ArtistDhtError::ArtistNameTooLong) + )); + } +} diff --git a/crates/artist-dht/src/request.rs b/crates/artist-dht/src/request.rs new file mode 100644 index 0000000..7c20346 --- /dev/null +++ b/crates/artist-dht/src/request.rs @@ -0,0 +1,170 @@ +//! Tracking of in-flight DHT requests. + +use std::collections::HashMap; +use std::sync::Mutex; + +use federation_net::EndpointId; +use tokio::sync::oneshot; + +use crate::error::{ArtistDhtError, Result}; +use crate::message::{ + FindNodeResponse, FindValueResponse, PongResponse, RequestId, StoreRecordResponse, +}; + +/// Maximum number of simultaneously pending requests. +pub const MAX_PENDING_REQUESTS: usize = 1024; + +/// A response payload of any DHT request type. +#[derive(Debug)] +pub(crate) enum DhtResponse { + Pong(PongResponse), + FindNode(FindNodeResponse), + FindValue(FindValueResponse), + Store(StoreRecordResponse), +} + +struct PendingEntry { + /// The peer the response is expected from. + peer: EndpointId, + sender: oneshot::Sender, +} + +/// Correlates responses with awaiting requesters. +/// +/// Entries are removed when the response arrives, and the requester removes +/// its own entry on timeout, so the map cannot grow without bound; a hard cap +/// of [`MAX_PENDING_REQUESTS`] guards against bugs. +#[derive(Default)] +pub(crate) struct PendingRequests { + map: Mutex>, +} + +impl PendingRequests { + /// Registers a new pending request and returns the receiver for its + /// response. + pub fn register( + &self, + request_id: RequestId, + peer: EndpointId, + ) -> Result> { + let mut map = lock(&self.map); + if map.len() >= MAX_PENDING_REQUESTS { + return Err(ArtistDhtError::Protocol( + "too many pending requests".to_string(), + )); + } + let (sender, receiver) = oneshot::channel(); + map.insert(request_id, PendingEntry { peer, sender }); + Ok(receiver) + } + + /// Completes a pending request with a response from `from_peer`. + /// + /// The response is delivered only if it comes from the peer the request + /// was sent to; otherwise the entry stays and the stray response is + /// dropped. Returns `true` if a waiting requester was resolved. + pub fn complete( + &self, + request_id: &RequestId, + from_peer: &EndpointId, + response: DhtResponse, + ) -> bool { + let mut map = lock(&self.map); + match map.get(request_id) { + Some(entry) if &entry.peer == from_peer => { + if let Some(entry) = map.remove(request_id) { + // The requester may have timed out already; that is fine. + let _ = entry.sender.send(response); + return true; + } + false + } + _ => false, + } + } + + /// Removes a pending request, e.g. after a timeout. + pub fn remove(&self, request_id: &RequestId) { + lock(&self.map).remove(request_id); + } + + /// Number of currently pending requests. + pub fn len(&self) -> usize { + lock(&self.map).len() + } +} + +fn lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::routing::NodeId; + + fn test_peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + fn pong() -> DhtResponse { + DhtResponse::Pong(PongResponse { + node_id: NodeId::from_bytes([0u8; 32]), + }) + } + + #[tokio::test] + async fn response_resolves_pending_request() { + let pending = PendingRequests::default(); + let peer = test_peer(1); + let id = RequestId::random(); + let receiver = pending.register(id, peer).expect("register"); + assert!(pending.complete(&id, &peer, pong())); + assert!(receiver.await.is_ok()); + assert_eq!(pending.len(), 0); + } + + #[tokio::test] + async fn response_from_wrong_peer_is_ignored() { + let pending = PendingRequests::default(); + let peer = test_peer(1); + let wrong = test_peer(2); + let id = RequestId::random(); + let _receiver = pending.register(id, peer).expect("register"); + assert!(!pending.complete(&id, &wrong, pong())); + // The entry is still pending for the right peer. + assert_eq!(pending.len(), 1); + } + + #[tokio::test] + async fn entry_is_removed_after_timeout() { + let pending = PendingRequests::default(); + let peer = test_peer(1); + let id = RequestId::random(); + let receiver = pending.register(id, peer).expect("register"); + // Simulate the requester timing out: it removes its own entry. + let result = tokio::time::timeout(std::time::Duration::from_millis(20), receiver).await; + assert!(result.is_err()); + pending.remove(&id); + assert_eq!(pending.len(), 0); + // A late response finds nothing to complete. + assert!(!pending.complete(&id, &peer, pong())); + } + + #[test] + fn pending_map_is_capped() { + let pending = PendingRequests::default(); + let peer = test_peer(1); + let mut receivers = Vec::new(); + for _ in 0..MAX_PENDING_REQUESTS { + receivers.push( + pending + .register(RequestId::random(), peer) + .expect("register"), + ); + } + assert!(pending.register(RequestId::random(), peer).is_err()); + } +} diff --git a/crates/artist-dht/src/routing.rs b/crates/artist-dht/src/routing.rs new file mode 100644 index 0000000..b81a0f6 --- /dev/null +++ b/crates/artist-dht/src/routing.rs @@ -0,0 +1,315 @@ +//! Kademlia-style node identifiers, XOR distance and routing table. + +use std::fmt; + +use federation_net::EndpointId; +use serde::{Deserialize, Serialize}; + +use crate::record::DhtKey; + +/// Replication factor: how many closest nodes store a record and how many +/// contacts a single response may carry. +pub const K: usize = 8; +/// Lookup parallelism: how many candidates are queried concurrently. +pub const ALPHA: usize = 3; +/// Hard budget of requests a single iterative lookup may send. +pub const MAX_LOOKUP_REQUESTS: usize = 32; + +/// A 256-bit DHT node identifier, derived from the peer's endpoint id. +#[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] +pub struct NodeId([u8; 32]); + +impl NodeId { + /// Derives the node id: `BLAKE3("artist-dht:node:" || endpoint id)`. + /// + /// The endpoint id is persistent, so the node id is stable across + /// restarts. + pub fn from_endpoint(endpoint_id: &EndpointId) -> Self { + let mut hasher = blake3::Hasher::new(); + hasher.update(b"artist-dht:node:"); + hasher.update(endpoint_id.as_bytes()); + Self(*hasher.finalize().as_bytes()) + } + + /// Creates a node id from raw bytes. + pub fn from_bytes(bytes: [u8; 32]) -> Self { + Self(bytes) + } + + /// Returns the raw bytes of this id. + pub fn as_bytes(&self) -> &[u8; 32] { + &self.0 + } +} + +impl fmt::Display for NodeId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + for byte in &self.0 { + write!(f, "{byte:02x}")?; + } + Ok(()) + } +} + +impl fmt::Debug for NodeId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "NodeId({self})") + } +} + +/// XOR distance between two points of the 256-bit key space, compared as +/// unsigned big-endian integers. +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug)] +pub struct Distance([u8; 32]); + +/// Computes the XOR distance between two 256-bit values. +pub fn distance(a: &[u8; 32], b: &[u8; 32]) -> Distance { + let mut out = [0u8; 32]; + for (i, byte) in out.iter_mut().enumerate() { + *byte = a[i] ^ b[i]; + } + Distance(out) +} + +impl Distance { + /// Index of the k-bucket this distance falls into: the position of the + /// highest set bit (0..=255). Returns `None` for a zero distance (self). + pub fn bucket_index(&self) -> Option { + for (i, byte) in self.0.iter().enumerate() { + if *byte != 0 { + return Some(255 - (i * 8 + byte.leading_zeros() as usize)); + } + } + None + } +} + +/// Everything needed to reach another DHT node. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct NodeContact { + /// DHT identifier of the node. + pub node_id: NodeId, + /// Transport identifier of the node. + pub peer_id: EndpointId, + /// `federation-net` ticket used to connect on demand. + pub ticket: String, + /// Unix timestamp (milliseconds) of the last observed activity. + pub last_seen_ms: u64, +} + +/// A simplified Kademlia routing table: 256 k-buckets of up to [`K`] contacts. +pub struct RoutingTable { + own_id: NodeId, + buckets: Vec>, +} + +impl RoutingTable { + /// Creates an empty routing table for the given local node id. + pub fn new(own_id: NodeId) -> Self { + Self { + own_id, + buckets: vec![Vec::new(); 256], + } + } + + /// Returns the local node id. + pub fn own_id(&self) -> NodeId { + self.own_id + } + + /// Inserts or refreshes a contact. Returns `true` if the contact was not + /// known before. + /// + /// A full bucket evicts its least recently seen contact; the PoC skips + /// the classic ping-before-evict procedure. + pub fn upsert(&mut self, contact: NodeContact) -> bool { + let Some(index) = + distance(self.own_id.as_bytes(), contact.node_id.as_bytes()).bucket_index() + else { + // Zero distance: never store ourselves. + return false; + }; + let bucket = &mut self.buckets[index]; + if let Some(existing) = bucket + .iter_mut() + .find(|entry| entry.peer_id == contact.peer_id) + { + existing.node_id = contact.node_id; + existing.ticket = contact.ticket; + existing.last_seen_ms = existing.last_seen_ms.max(contact.last_seen_ms); + return false; + } + if bucket.len() >= K { + // Evict the least recently seen contact. + if let Some((oldest, _)) = bucket + .iter() + .enumerate() + .min_by_key(|(_, entry)| entry.last_seen_ms) + { + bucket.remove(oldest); + } + } + bucket.push(contact); + true + } + + /// Refreshes the `last_seen_ms` of a known peer. Returns `true` if the + /// peer was found. + pub fn touch(&mut self, peer_id: &EndpointId, now_ms: u64) -> bool { + for bucket in &mut self.buckets { + if let Some(entry) = bucket.iter_mut().find(|entry| &entry.peer_id == peer_id) { + entry.last_seen_ms = entry.last_seen_ms.max(now_ms); + return true; + } + } + false + } + + /// Looks up a contact by its transport id. + pub fn get(&self, peer_id: &EndpointId) -> Option { + self.buckets + .iter() + .flatten() + .find(|entry| &entry.peer_id == peer_id) + .cloned() + } + + /// Returns up to `count` known contacts closest to `target` by XOR + /// distance. + pub fn closest(&self, target: &[u8; 32], count: usize) -> Vec { + let mut contacts: Vec = self.buckets.iter().flatten().cloned().collect(); + contacts.sort_by_key(|contact| distance(contact.node_id.as_bytes(), target)); + contacts.truncate(count); + contacts + } + + /// Returns all known contacts. + pub fn contacts(&self) -> Vec { + self.buckets.iter().flatten().cloned().collect() + } + + /// Returns the number of known contacts. + pub fn len(&self) -> usize { + self.buckets.iter().map(Vec::len).sum() + } + + /// Returns `true` if no contacts are known. + pub fn is_empty(&self) -> bool { + self.len() == 0 + } +} + +/// Convenience: distance from a node to a DHT key. +pub fn key_distance(node: &NodeId, key: &DhtKey) -> Distance { + distance(node.as_bytes(), key.as_bytes()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + fn contact(seed: u8, last_seen_ms: u64) -> NodeContact { + let peer = test_peer(seed); + NodeContact { + node_id: NodeId::from_endpoint(&peer), + peer_id: peer, + ticket: format!("fnet-test-{seed}"), + last_seen_ms, + } + } + + #[test] + fn node_id_is_deterministic() { + let peer = test_peer(1); + assert_eq!(NodeId::from_endpoint(&peer), NodeId::from_endpoint(&peer)); + assert_ne!( + NodeId::from_endpoint(&peer), + NodeId::from_endpoint(&test_peer(2)) + ); + } + + #[test] + fn xor_distance_properties() { + let a = [0b1010_0000u8; 32]; + let b = [0b0000_0000u8; 32]; + assert_eq!(distance(&a, &a), distance(&b, &b)); + assert_eq!(distance(&a, &b), distance(&b, &a)); + // d(a, a) == 0 and is the smallest possible distance. + assert!(distance(&a, &a) < distance(&a, &b)); + + // Big-endian comparison: a difference in the first byte outweighs + // any difference in later bytes. + let mut c = [0u8; 32]; + c[0] = 1; + let mut d = [0u8; 32]; + d[31] = 0xff; + assert!(distance(&c, &b) > distance(&d, &b)); + } + + #[test] + fn bucket_index_matches_highest_bit() { + let zero = [0u8; 32]; + let mut one = [0u8; 32]; + one[31] = 1; + assert_eq!(distance(&zero, &one).bucket_index(), Some(0)); + let mut top = [0u8; 32]; + top[0] = 0x80; + assert_eq!(distance(&zero, &top).bucket_index(), Some(255)); + assert_eq!(distance(&zero, &zero).bucket_index(), None); + } + + #[test] + fn closest_sorts_by_distance() { + let own = NodeId::from_bytes([0u8; 32]); + let mut table = RoutingTable::new(own); + for seed in 1..=20u8 { + table.upsert(contact(seed, seed as u64)); + } + let target = [0x42u8; 32]; + let closest = table.closest(&target, K); + assert!(closest.len() <= K); + for pair in closest.windows(2) { + assert!( + distance(pair[0].node_id.as_bytes(), &target) + <= distance(pair[1].node_id.as_bytes(), &target) + ); + } + } + + #[test] + fn bucket_is_capped_at_k() { + // All contacts whose distance to `own` shares the same highest bit + // land in one bucket; force that by controlling the node ids. + let own = NodeId::from_bytes([0u8; 32]); + let mut table = RoutingTable::new(own); + for i in 0..(K as u8 + 4) { + let peer = test_peer(i + 1); + let mut id = [0x80u8; 32]; + id[31] = i; + table.upsert(NodeContact { + node_id: NodeId::from_bytes(id), + peer_id: peer, + ticket: String::new(), + last_seen_ms: u64::from(i), + }); + } + assert_eq!(table.len(), K); + // The oldest contacts (smallest last_seen_ms) were evicted. + let contacts = table.contacts(); + assert!(contacts.iter().all(|c| c.last_seen_ms >= 4)); + } + + #[test] + fn upsert_refreshes_existing_contact() { + let own = NodeId::from_bytes([0u8; 32]); + let mut table = RoutingTable::new(own); + assert!(table.upsert(contact(1, 10))); + assert!(!table.upsert(contact(1, 20))); + assert_eq!(table.len(), 1); + assert_eq!(table.contacts()[0].last_seen_ms, 20); + } +} diff --git a/crates/artist-dht/src/service.rs b/crates/artist-dht/src/service.rs new file mode 100644 index 0000000..1cf02a1 --- /dev/null +++ b/crates/artist-dht/src/service.rs @@ -0,0 +1,417 @@ +//! The public service facade: lifecycle, artist operations and search. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use federation_net::{EndpointId, NetworkConfig, NetworkEngine, PeerTicket, SchemaId}; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use tracing::info; + +use crate::config::ArtistDhtConfig; +use crate::database::Database; +use crate::error::{ArtistDhtError, Result}; +use crate::node::{Node, record_supersedes}; +use crate::normalization::{normalize_artist_name, tokenize}; +use crate::record::{Artist, ArtistId, DhtKey, PeerId, StoredArtistRecord, now_ms, validate_name}; +use crate::routing::{NodeContact, NodeId}; + +/// Fixed schema of the artist-dht protocol; peers with a different schema are +/// rejected by `federation-net` during the handshake. +pub const SCHEMA_NAME: &str = "artist-dht-poc-v1"; + +/// Capacity of the application event channel. +const EVENT_CHANNEL_CAPACITY: usize = 256; + +/// Events delivered to the application. +#[derive(Debug)] +pub enum ArtistDhtEvent { + /// A transport connection to a peer was established. + PeerConnected { + /// The connected peer. + peer_id: EndpointId, + }, + /// A transport connection to a peer closed. + PeerDisconnected { + /// The disconnected peer. + peer_id: EndpointId, + }, + /// A previously unknown DHT contact was learned. + ContactDiscovered { + /// The new contact. + contact: NodeContact, + }, + /// A non-fatal error occurred. + Error { + /// Human-readable description. + message: String, + }, +} + +/// Receiving side of the service event channel. +#[derive(Debug)] +pub struct ArtistDhtEventReceiver { + rx: mpsc::Receiver, +} + +impl ArtistDhtEventReceiver { + /// Receives the next event; `None` after shutdown. + pub async fn recv(&mut self) -> Option { + self.rx.recv().await + } +} + +/// Statistics of one publish or republish operation. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct PublishStats { + /// Number of artist records published. + pub records: usize, + /// Total number of DHT keys published. + pub keys: usize, + /// Number of distinct remote nodes that accepted at least one replica. + pub remote_nodes: usize, + /// Whether a replica was also stored locally. + pub local_replica: bool, +} + +/// Result of a combined local + network search. +#[derive(Debug)] +pub struct SearchOutcome { + /// Matches from the local `local_artists` table. + pub local_results: Vec, + /// Matches found in the DHT (including local replicas), tombstones and + /// duplicates already filtered out. Artists matching every query token + /// come first. + pub network_results: Vec, + /// Number of distinct peers queried during the lookups. + pub queried_nodes: usize, + /// Number of distinct nodes discovered during the lookups. + pub discovered_nodes: usize, + /// Total wall-clock duration of the search. + pub duration: Duration, +} + +/// A distributed artist directory node. +/// +/// Every instance is simultaneously a client, a DHT router and a storage +/// node; there are no special server roles. See the crate documentation for +/// the protocol description. +pub struct ArtistDhtService { + node: Arc, + tasks: Vec>, +} + +impl ArtistDhtService { + /// Starts the service: opens the database, starts the network engine, + /// loads persisted contacts and spawns the maintenance tasks. + pub async fn start(config: ArtistDhtConfig) -> Result<(Self, ArtistDhtEventReceiver)> { + let engine_config = NetworkConfig::builder() + .data_dir(&config.data_dir) + .network_id(config.network_id) + .schema_id(SchemaId::from_name(SCHEMA_NAME)) + .request_timeout(config.transport_timeout) + .build() + .map_err(|err| ArtistDhtError::Network(err.to_string()))?; + let (engine, net_events) = NetworkEngine::start(engine_config).await?; + let db = Database::open(&config.data_dir.join("state.sqlite3")).await?; + + let (event_tx, event_rx) = mpsc::channel(EVENT_CHANNEL_CAPACITY); + let node = Arc::new(Node::new(engine, db, config.clone(), event_tx)); + info!( + endpoint_id = %node.endpoint_id, + node_id = %node.node_id, + "artist-dht node starting" + ); + + // Contacts persisted by earlier runs seed the routing table; if any + // exist, local records are republished right away. + let persisted = node.db.load_known_peers().await?; + if !persisted.is_empty() { + info!(count = persisted.len(), "loaded persisted DHT contacts"); + node.seed_contacts(persisted); + } + + let tasks = vec![ + tokio::spawn(node.clone().run_event_loop(net_events)), + tokio::spawn(republish_timer(node.clone())), + tokio::spawn(expire_timer(node.clone())), + ]; + node.maybe_trigger_initial_republish(); + + Ok(( + Self { node, tasks }, + ArtistDhtEventReceiver { rx: event_rx }, + )) + } + + /// Returns the transport identifier of this peer. + pub fn endpoint_id(&self) -> EndpointId { + self.node.endpoint_id + } + + /// Returns the DHT identifier of this peer. + pub fn node_id(&self) -> NodeId { + self.node.node_id + } + + /// Creates a shareable ticket for this peer. + pub async fn ticket(&self) -> Result { + self.node.ensure_running()?; + self.node.engine.ticket().await.map_err(Into::into) + } + + /// Connects to another peer by ticket. Contacts are exchanged + /// automatically once the connection is up. + pub async fn connect(&self, ticket: PeerTicket) -> Result { + self.node.ensure_running()?; + let peer = self.node.engine.connect(ticket).await?; + Ok(peer) + } + + /// All DHT contacts currently known to this node. + pub fn known_peers(&self) -> Vec { + self.node.known_contacts() + } + + /// Transport connections that are currently open. + pub fn connected_peers(&self) -> Vec { + self.node.engine.connected_peers() + } + + /// Returns `true` if a transport connection to `peer` is open. + pub fn is_connected(&self, peer: EndpointId) -> bool { + self.node.engine.is_connected(peer) + } + + /// Adds a new artist to the local database and publishes it to the DHT. + pub async fn add_artist(&self, name: String) -> Result<(Artist, PublishStats)> { + self.node.ensure_running()?; + let name = name.trim().to_string(); + let normalized = validate_name(&name)?; + let uuid = uuid::Uuid::now_v7(); + let artist = Artist { + id: ArtistId::derive(&self.node.endpoint_id, &uuid), + owner: self.node.endpoint_id, + name, + normalized_name: normalized, + revision: 1, + deleted: false, + updated_at_ms: now_ms(), + }; + self.node.db.upsert_local_artist(&artist).await?; + info!(artist = %artist.id, name = %artist.name, "added local artist"); + let stats = self.node.publish_artist(&artist).await?; + Ok((artist, stats)) + } + + /// Deletes a locally owned artist: stores a tombstone and publishes it. + pub async fn delete_artist(&self, artist_id: ArtistId) -> Result { + self.node.ensure_running()?; + let Some(mut artist) = self.node.db.get_local_artist(artist_id).await? else { + return Err(ArtistDhtError::ArtistNotFound); + }; + if artist.owner != self.node.endpoint_id { + return Err(ArtistDhtError::CannotDeleteRemoteArtist); + } + if artist.deleted { + return Err(ArtistDhtError::ArtistNotFound); + } + artist.revision += 1; + artist.deleted = true; + artist.updated_at_ms = now_ms(); + self.node.db.upsert_local_artist(&artist).await?; + info!(artist = %artist.id, "deleted local artist; publishing tombstone"); + self.node.publish_artist(&artist).await + } + + /// Resolves a (possibly shortened) hex artist id against local records. + /// + /// Returns [`ArtistDhtError::ArtistNotFound`] unless exactly one active + /// local artist matches the prefix. + pub async fn resolve_local_artist_id(&self, prefix: &str) -> Result { + let prefix = prefix.trim().to_lowercase(); + if prefix.len() < 4 || !prefix.chars().all(|c| c.is_ascii_hexdigit()) { + return Err(ArtistDhtError::ArtistNotFound); + } + let matches = self.node.db.find_local_by_id_prefix(prefix).await?; + match matches.as_slice() { + [artist] => Ok(artist.id), + _ => Err(ArtistDhtError::ArtistNotFound), + } + } + + /// Lists all locally owned active artists. + pub async fn list_local_artists(&self) -> Result> { + self.node.db.list_local_artists(false).await + } + + /// Searches only the local database. + pub async fn search_local(&self, query: &str) -> Result> { + let normalized = normalize_artist_name(query); + if normalized.is_empty() { + return Err(ArtistDhtError::InvalidArtistName); + } + self.node.db.search_local(normalized).await + } + + /// Searches locally and across the DHT. + /// + /// The exact key is looked up first; only if it yields nothing the token + /// keys are tried. No broadcast is involved: every step is an iterative + /// Kademlia-style lookup. + pub async fn search_network(&self, query: &str) -> Result { + self.node.ensure_running()?; + let started = Instant::now(); + let normalized = normalize_artist_name(query); + if normalized.is_empty() { + return Err(ArtistDhtError::InvalidArtistName); + } + let tokens = tokenize(&normalized); + let network_id = self.node.config.network_id; + + let local_results = self.node.db.search_local(normalized.clone()).await?; + + let mut queried_nodes = 0usize; + let mut discovered_nodes = 0usize; + // (artist id, owner) -> best record seen so far. + let mut merged: HashMap<(ArtistId, PeerId), StoredArtistRecord> = HashMap::new(); + fn merge( + merged: &mut HashMap<(ArtistId, PeerId), StoredArtistRecord>, + records: Vec, + ) { + for record in records { + let key = (record.artist.id, record.artist.owner); + match merged.get(&key) { + Some(existing) if !record_supersedes(&record, existing) => {} + _ => { + merged.insert(key, record); + } + } + } + } + + // Step 1: the exact key — local replicas, then the network. + let exact_key = DhtKey::exact(&network_id, &normalized); + merge( + &mut merged, + self.node.db.dht_records_by_key(exact_key, now_ms()).await?, + ); + let outcome = self + .node + .lookup(*exact_key.as_bytes(), Some(exact_key)) + .await; + queried_nodes += outcome.queried; + discovered_nodes = discovered_nodes.max(outcome.discovered); + merge(&mut merged, outcome.records); + + // Step 2: token keys, only when the exact key produced no live match. + let has_live_match = merged.values().any(|record| !record.artist.deleted); + if !has_live_match { + for token in &tokens { + let key = DhtKey::token(&network_id, token); + merge( + &mut merged, + self.node.db.dht_records_by_key(key, now_ms()).await?, + ); + let outcome = self.node.lookup(*key.as_bytes(), Some(key)).await; + queried_nodes += outcome.queried; + discovered_nodes = discovered_nodes.max(outcome.discovered); + merge(&mut merged, outcome.records); + } + } + + // Drop tombstones and expired records, then rank: full-token matches + // first, then alphabetically. + let now = now_ms(); + let mut network_results: Vec = merged + .into_values() + .filter(|record| !record.artist.deleted && record.expires_at_ms > now) + .map(|record| record.artist) + .collect(); + let matches_all_tokens = |artist: &Artist| { + let artist_tokens = tokenize(&artist.normalized_name); + tokens + .iter() + .all(|token| artist_tokens.iter().any(|t| t == token)) + }; + network_results.sort_by(|a, b| { + matches_all_tokens(b) + .cmp(&matches_all_tokens(a)) + .then_with(|| a.normalized_name.cmp(&b.normalized_name)) + }); + + Ok(SearchOutcome { + local_results, + network_results, + queried_nodes, + discovered_nodes, + duration: started.elapsed(), + }) + } + + /// Pings a known peer: verifies liveness and DHT identity, returns the + /// round-trip time and refreshes the contact. + pub async fn ping(&self, peer: EndpointId) -> Result { + self.node.ensure_running()?; + let contact = self + .node + .known_contacts() + .into_iter() + .find(|contact| contact.peer_id == peer) + .ok_or_else(|| ArtistDhtError::Network(format!("unknown peer {peer}")))?; + self.node.ping(&contact).await + } + + /// Republishes all live local records immediately. + pub async fn republish(&self) -> Result { + self.node.republish_all().await + } + + /// Shuts the service down gracefully: stops the maintenance tasks, shuts + /// the network engine down and closes the event channel. + pub async fn shutdown(self) -> Result<()> { + info!("artist-dht node shutting down"); + self.node.begin_shutdown(); + for task in &self.tasks { + task.abort(); + } + self.node.engine.clone().shutdown().await?; + for task in self.tasks { + let _ = task.await; + } + info!("artist-dht node shut down"); + Ok(()) + } +} + +async fn republish_timer(node: Arc) { + let mut interval = tokio::time::interval(node.config.republish_interval); + // The first tick fires immediately; skip it, the initial republish is + // triggered by contact discovery instead. + interval.tick().await; + loop { + interval.tick().await; + if node.is_shutting_down() { + break; + } + if node.known_contacts().is_empty() { + continue; + } + if let Err(err) = node.republish_all().await { + tracing::warn!(error = %err, "periodic republish failed"); + } + } +} + +async fn expire_timer(node: Arc) { + let mut interval = tokio::time::interval(node.config.expire_interval); + interval.tick().await; + loop { + interval.tick().await; + if node.is_shutting_down() { + break; + } + node.sweep_expired().await; + } +} diff --git a/crates/artist-dht/tests/integration.rs b/crates/artist-dht/tests/integration.rs new file mode 100644 index 0000000..d7ad70a --- /dev/null +++ b/crates/artist-dht/tests/integration.rs @@ -0,0 +1,238 @@ +//! Integration tests: three DHT nodes in one Tokio runtime. + +use std::path::Path; +use std::time::Duration; + +use artist_dht::{ + ArtistDhtConfig, ArtistDhtError, ArtistDhtEventReceiver, ArtistDhtService, EndpointId, + NetworkId, +}; + +/// Hard cap on every test so a regression can never hang CI. +const TEST_TIMEOUT: Duration = Duration::from_secs(240); + +/// Serializes the network-facing tests: many concurrent endpoints contend on +/// relay discovery and produce spurious timeouts. +static NET_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + +type Started = (ArtistDhtService, ArtistDhtEventReceiver); + +async fn start(dir: &Path, network: &str) -> Started { + let config = ArtistDhtConfig::builder() + .data_dir(dir) + .network_id(NetworkId::from_name(network)) + // Fast timers so republish/expiry behavior is observable in tests. + .republish_interval(Duration::from_secs(5)) + .expire_interval(Duration::from_secs(2)) + .request_timeout(Duration::from_secs(5)) + .lookup_timeout(Duration::from_secs(10)) + .build() + .expect("valid config"); + ArtistDhtService::start(config) + .await + .expect("service starts") +} + +/// Polls `check` until it returns `Some` or the deadline passes. +async fn wait_for( + what: &str, + deadline: Duration, + mut check: impl AsyncFnMut() -> Option, +) -> T { + let started = std::time::Instant::now(); + loop { + if let Some(value) = check().await { + return value; + } + assert!(started.elapsed() < deadline, "timed out waiting for {what}"); + tokio::time::sleep(Duration::from_millis(300)).await; + } +} + +fn knows_peer(service: &ArtistDhtService, peer: EndpointId) -> bool { + service + .known_peers() + .iter() + .any(|contact| contact.peer_id == peer) +} + +/// The full demo scenario: bootstrap through one peer, peer exchange, +/// publish, distributed search, replica survival, restart and tombstones. +#[tokio::test] +async fn full_distributed_flow() { + let _net = NET_LOCK.lock().await; + tokio::time::timeout(TEST_TIMEOUT, async { + let dir_a = tempfile::tempdir().expect("tempdir"); + let dir_b = tempfile::tempdir().expect("tempdir"); + let dir_c = tempfile::tempdir().expect("tempdir"); + + // Step 1: peer A starts alone and issues a ticket. + let (a, _events_a) = start(dir_a.path(), "test-artists").await; + let ticket_a = a.ticket().await.expect("ticket"); + + // Step 2: peer B joins through A. + let (b, _events_b) = start(dir_b.path(), "test-artists").await; + let peer_a = b.connect(ticket_a.clone()).await.expect("b connects to a"); + assert_eq!(peer_a, a.endpoint_id()); + + // Step 3: peer C joins through A only... + let (c, _events_c) = start(dir_c.path(), "test-artists").await; + c.connect(ticket_a.clone()).await.expect("c connects to a"); + + // ...and learns about B via peer exchange, without connecting to it. + wait_for( + "C to learn about B via peer exchange", + Duration::from_secs(30), + async || knows_peer(&c, b.endpoint_id()).then_some(()), + ) + .await; + + // Step 4: B adds artists; they are published into the DHT. + let (massive, stats) = b + .add_artist("Massive Attack".into()) + .await + .expect("add artist"); + assert_eq!(massive.normalized_name, "massive attack"); + // 1 exact key + 2 token keys. + assert_eq!(stats.keys, 3); + assert!( + stats.remote_nodes >= 1, + "the record must be replicated to at least one other node" + ); + b.add_artist("Portishead".into()).await.expect("add artist"); + + // Step 5: A finds the record through the DHT (exact search). + let outcome = a.search_network("massive attack").await.expect("search"); + assert!(outcome.local_results.is_empty(), "A has no local artists"); + assert_eq!(outcome.network_results.len(), 1); + assert_eq!(outcome.network_results[0].name, "Massive Attack"); + // The result names the owner, peer B. + assert_eq!(outcome.network_results[0].owner, b.endpoint_id()); + + // Step 6: C finds Portishead via a single-token search. + let outcome = c.search_network("portishead").await.expect("search"); + assert_eq!(outcome.network_results.len(), 1); + assert_eq!(outcome.network_results[0].owner, b.endpoint_id()); + + // Token search also finds multi-token names. + let outcome = c.search_network("massive").await.expect("search"); + assert!( + outcome + .network_results + .iter() + .any(|artist| artist.name == "Massive Attack"), + "token search must find 'Massive Attack'" + ); + + // Step 7: stop B; replicas must keep the record findable. + let b_id = b.endpoint_id(); + b.shutdown().await.expect("shutdown b"); + let outcome = a.search_network("massive attack").await.expect("search"); + assert_eq!( + outcome.network_results.len(), + 1, + "the record must survive on replicas after the owner left" + ); + let outcome = c.search_network("massive attack").await.expect("search"); + assert_eq!(outcome.network_results.len(), 1); + + // Step 8: B restarts with the same data dir: same identity, and its + // local database is intact. + let (b, _events_b) = start(dir_b.path(), "test-artists").await; + assert_eq!(b.endpoint_id(), b_id, "endpoint id must persist"); + let local = b.list_local_artists().await.expect("list"); + assert_eq!(local.len(), 2, "local database must persist"); + b.connect(ticket_a).await.expect("b reconnects to a"); + + // Step 9: B deletes Massive Attack; the tombstone propagates and the + // other peers stop returning the record. + let stats = b.delete_artist(massive.id).await.expect("delete"); + assert!(stats.remote_nodes >= 1, "tombstone must reach replicas"); + let outcome = a.search_network("massive attack").await.expect("search"); + assert!( + outcome.network_results.is_empty(), + "A must not return a tombstoned record, got {:?}", + outcome.network_results + ); + let outcome = c.search_network("massive").await.expect("search"); + assert!( + outcome + .network_results + .iter() + .all(|artist| artist.id != massive.id), + "C must not return the tombstoned record" + ); + // Portishead is still there. + let outcome = a.search_network("portishead").await.expect("search"); + assert_eq!(outcome.network_results.len(), 1); + + a.shutdown().await.expect("shutdown a"); + b.shutdown().await.expect("shutdown b"); + c.shutdown().await.expect("shutdown c"); + }) + .await + .expect("test timed out"); +} + +/// A peer from a different network is rejected by the transport handshake. +#[tokio::test] +async fn different_network_is_rejected() { + let _net = NET_LOCK.lock().await; + tokio::time::timeout(TEST_TIMEOUT, async { + let dir_a = tempfile::tempdir().expect("tempdir"); + let dir_b = tempfile::tempdir().expect("tempdir"); + let (a, _events_a) = start(dir_a.path(), "network-one").await; + let (b, _events_b) = start(dir_b.path(), "network-two").await; + + let ticket = a.ticket().await.expect("ticket"); + let err = b.connect(ticket).await.expect_err("must be rejected"); + assert!( + matches!(err, ArtistDhtError::Network(_)), + "expected a network error, got {err:?}" + ); + assert!(b.connected_peers().is_empty()); + + a.shutdown().await.expect("shutdown a"); + b.shutdown().await.expect("shutdown b"); + }) + .await + .expect("test timed out"); +} + +/// A lookup over dead contacts finishes within its budget and timeout +/// instead of hanging. +#[tokio::test] +async fn lookup_terminates_with_dead_contacts() { + let _net = NET_LOCK.lock().await; + tokio::time::timeout(TEST_TIMEOUT, async { + let dir_a = tempfile::tempdir().expect("tempdir"); + let dir_b = tempfile::tempdir().expect("tempdir"); + let (a, _events_a) = start(dir_a.path(), "test-artists").await; + let (b, _events_b) = start(dir_b.path(), "test-artists").await; + + let ticket_a = a.ticket().await.expect("ticket"); + b.connect(ticket_a).await.expect("connect"); + wait_for("A to learn about B", Duration::from_secs(30), async || { + knows_peer(&a, b.endpoint_id()).then_some(()) + }) + .await; + + // Kill B: A still remembers it in the routing table. + b.shutdown().await.expect("shutdown b"); + + let started = std::time::Instant::now(); + let outcome = a.search_network("anything").await.expect("search finishes"); + assert!(outcome.network_results.is_empty()); + // Bounded by the lookup timeout per key (exact + 1 token) with slack + // for connection attempts; the essential property is that it returns. + assert!( + started.elapsed() < Duration::from_secs(60), + "lookup took too long: {:?}", + started.elapsed() + ); + + a.shutdown().await.expect("shutdown a"); + }) + .await + .expect("test timed out"); +} diff --git a/crates/federation-net/src/engine.rs b/crates/federation-net/src/engine.rs index e2f7fa9..1139912 100644 --- a/crates/federation-net/src/engine.rs +++ b/crates/federation-net/src/engine.rs @@ -684,6 +684,11 @@ impl NetworkEngine { lock(&self.shared.peers).keys().copied().collect() } + /// Returns `true` if there is an active connection to `peer`. + pub fn is_connected(&self, peer: EndpointId) -> bool { + lock(&self.shared.peers).contains_key(&peer) + } + /// Closes the connection to `peer` and removes it from the registry. pub async fn disconnect(&self, peer: EndpointId) -> Result<()> { let state = lock(&self.shared.peers)