added example

This commit is contained in:
Ultradesu
2026-07-10 15:10:03 +03:00
parent 057436adb6
commit a97f0a04b3
21 changed files with 4456 additions and 4 deletions
Generated
+224 -2
View File
@@ -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"
+13 -2
View File
@@ -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"
+9
View File
@@ -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
+16
View File
@@ -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 }
+403
View File
@@ -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<String>,
/// 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::<Input>(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::<Input>(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 <ARTIST_NAME>");
} 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 <ARTIST_ID>");
} 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 <QUERY>");
} 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 <QUERY>");
} 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 <ARTIST_NAME> add and publish an artist\n\
\x20 /delete <ARTIST_ID> delete a local artist (id or hex prefix)\n\
\x20 /search-local <QUERY> search the local database only\n\
\x20 /search <QUERY> 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()
}
+27
View File
@@ -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 }
+152
View File
@@ -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
`<data_dir>/state.sqlite3` and `<data_dir>/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 <ARTIST_ID>` (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 <ARTIST_NAME> add and publish an artist
/delete <ARTIST_ID> delete a local artist (id or unique hex prefix)
/search-local <QUERY> search the local database only
/search <QUERY> 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.
+181
View File
@@ -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<PathBuf>,
network_id: Option<NetworkId>,
republish_interval: Option<Duration>,
expire_interval: Option<Duration>,
request_timeout: Option<Duration>,
lookup_timeout: Option<Duration>,
transport_timeout: Option<Duration>,
}
impl ArtistDhtConfigBuilder {
/// Sets the data directory.
pub fn data_dir(mut self, dir: impl Into<PathBuf>) -> 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<ArtistDhtConfig> {
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()
);
}
}
+480
View File
@@ -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<Mutex<Connection>>,
}
impl Database {
/// Opens (creating if needed) the database at `path` and applies the
/// schema.
pub async fn open(path: &Path) -> Result<Self> {
let path = path.to_path_buf();
let conn = tokio::task::spawn_blocking(move || -> Result<Connection> {
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<F, R>(&self, f: F) -> Result<R>
where
F: FnOnce(&Connection) -> rusqlite::Result<R> + 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<Option<Artist>> {
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<Vec<Artist>> {
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<Vec<Artist>> {
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<Vec<Artist>> {
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<Vec<Artist>> {
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<bool> {
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<Vec<StoredArtistRecord>> {
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<u8>>(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::<StoredArtistRecord>(&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<usize> {
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<Vec<NodeContact>> {
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<u8>>(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<Artist> {
let id: Vec<u8> = 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]);
}
}
+274
View File
@@ -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<StoredArtistRecord, String> {
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()
);
}
}
+65
View File
@@ -0,0 +1,65 @@
//! Error types for the artist-dht library.
/// Convenient result alias used across the library.
pub type Result<T, E = ArtistDhtError> = std::result::Result<T, E>;
/// 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<federation_net::NetworkError> 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()),
}
}
}
+92
View File
@@ -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};
+174
View File
@@ -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<T> {
/// 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<T> {
/// 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<NodeContact>,
}
/// 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<NodeContact>,
}
/// 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<StoredArtistRecord>,
},
/// The responder has nothing stored; here are closer nodes instead.
CloserNodes {
/// Up to `K` known nodes closest to the key.
nodes: Vec<NodeContact>,
},
}
/// 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<PingRequest>),
/// Reply to `Ping`.
Pong(ResponseEnvelope<PongResponse>),
/// Node lookup request.
FindNode(RequestEnvelope<FindNodeRequest>),
/// Reply to `FindNode`.
FindNodeResult(ResponseEnvelope<FindNodeResponse>),
/// Value lookup request.
FindValue(RequestEnvelope<FindValueRequest>),
/// Reply to `FindValue`.
FindValueResult(ResponseEnvelope<FindValueResponse>),
/// Replication request.
StoreRecord(RequestEnvelope<StoreRecordRequest>),
/// Reply to `StoreRecord`.
StoreRecordResult(ResponseEnvelope<StoreRecordResponse>),
}
+808
View File
@@ -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<T>(mutex: &Mutex<T>) -> 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<StoredArtistRecord>,
/// Closest known contacts to the target, best first, at most `K`.
pub closest: Vec<NodeContact>,
/// 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<ArtistDhtMessage>,
pub db: Database,
pub config: ArtistDhtConfig,
pub node_id: NodeId,
pub endpoint_id: EndpointId,
routing: Mutex<RoutingTable>,
pending: PendingRequests,
/// Peers we already introduced ourselves to (per connection).
hello_sent: Mutex<HashSet<EndpointId>>,
/// Peers we already gossiped contacts to (per connection).
exchange_sent: Mutex<HashSet<EndpointId>>,
events: Mutex<Option<mpsc::Sender<ArtistDhtEvent>>>,
/// Set once the post-startup republish has been triggered.
initial_republish_done: AtomicBool,
shutting_down: AtomicBool,
}
impl Node {
pub fn new(
engine: NetworkEngine<ArtistDhtMessage>,
db: Database,
config: ArtistDhtConfig,
events: mpsc::Sender<ArtistDhtEvent>,
) -> 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<NodeContact> {
lock(&self.routing).contacts()
}
/// Seeds the routing table (used at startup with persisted contacts).
pub fn seed_contacts(&self, contacts: Vec<NodeContact>) {
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<Self>, 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<Self>) {
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<Self>,
mut receiver: NetworkEventReceiver<ArtistDhtMessage>,
) {
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<Self>, 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<Self>, peer: EndpointId) {
if !lock(&self.exchange_sent).insert(peer) {
return;
}
let mut peers: Vec<NodeContact> = 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<Self>, 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<Self>, 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<Self>, 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<NodeContact> {
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<EndpointId> {
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<DhtResponse> {
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<std::time::Duration> {
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<DhtKey>) -> LookupOutcome {
let started = Instant::now();
let deadline = started + self.config.lookup_timeout;
let mut candidates: Vec<NodeContact> = lock(&self.routing).closest(&target, K);
let mut known: HashSet<EndpointId> =
candidates.iter().map(|contact| contact.peer_id).collect();
let mut queried: HashSet<EndpointId> = 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<NodeContact> = 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<PublishStats> {
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<EndpointId> = 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<PublishStats> {
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<NodeContact>,
) -> Vec<NodeContact> {
let mut seen: HashSet<EndpointId> = 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<EndpointId> = 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<NodeContact> = (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);
}
}
+93
View File
@@ -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<String> {
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::<String>::new());
assert_eq!(tokenize("solo"), vec!["solo"]);
}
}
+300
View File
@@ -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<Self, Self::Err> {
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<DhtKey> {
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<String, ArtistDhtError> {
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::<Vec<_>>()
.join(" ");
assert!(matches!(
validate_name(&many_tokens),
Err(ArtistDhtError::ArtistNameTooLong)
));
}
}
+170
View File
@@ -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<DhtResponse>,
}
/// 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<HashMap<RequestId, PendingEntry>>,
}
impl PendingRequests {
/// Registers a new pending request and returns the receiver for its
/// response.
pub fn register(
&self,
request_id: RequestId,
peer: EndpointId,
) -> Result<oneshot::Receiver<DhtResponse>> {
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<T>(mutex: &Mutex<T>) -> 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());
}
}
+315
View File
@@ -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<usize> {
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<Vec<NodeContact>>,
}
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<NodeContact> {
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<NodeContact> {
let mut contacts: Vec<NodeContact> = 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<NodeContact> {
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);
}
}
+417
View File
@@ -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<ArtistDhtEvent>,
}
impl ArtistDhtEventReceiver {
/// Receives the next event; `None` after shutdown.
pub async fn recv(&mut self) -> Option<ArtistDhtEvent> {
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<Artist>,
/// 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<Artist>,
/// 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<Node>,
tasks: Vec<JoinHandle<()>>,
}
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<PeerTicket> {
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<EndpointId> {
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<NodeContact> {
self.node.known_contacts()
}
/// Transport connections that are currently open.
pub fn connected_peers(&self) -> Vec<EndpointId> {
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<PublishStats> {
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<ArtistId> {
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<Vec<Artist>> {
self.node.db.list_local_artists(false).await
}
/// Searches only the local database.
pub async fn search_local(&self, query: &str) -> Result<Vec<Artist>> {
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<SearchOutcome> {
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<StoredArtistRecord>,
) {
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<Artist> = 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<Duration> {
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<PublishStats> {
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<Node>) {
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<Node>) {
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;
}
}
+238
View File
@@ -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<T>(
what: &str,
deadline: Duration,
mut check: impl AsyncFnMut() -> Option<T>,
) -> 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");
}
+5
View File
@@ -684,6 +684,11 @@ impl<M: Message> NetworkEngine<M> {
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)