From b9c61e2249169468062184527ff4cd7eed8b0aff Mon Sep 17 00:00:00 2001 From: Ultradesu Date: Thu, 16 Jul 2026 17:24:43 +0300 Subject: [PATCH] Added music-dht --- Cargo.lock | 35 +- Cargo.toml | 3 +- crates/music-dht/Cargo.toml | 26 + crates/music-dht/README.md | 78 +++ crates/music-dht/src/config.rs | 208 ++++++ crates/music-dht/src/database.rs | 440 ++++++++++++ crates/music-dht/src/dht.rs | 279 ++++++++ crates/music-dht/src/error.rs | 65 ++ crates/music-dht/src/lib.rs | 108 +++ crates/music-dht/src/message.rs | 174 +++++ crates/music-dht/src/node.rs | 924 ++++++++++++++++++++++++++ crates/music-dht/src/normalization.rs | 87 +++ crates/music-dht/src/record.rs | 378 +++++++++++ crates/music-dht/src/request.rs | 192 ++++++ crates/music-dht/src/routing.rs | 338 ++++++++++ crates/music-dht/src/service.rs | 535 +++++++++++++++ crates/music-dht/tests/integration.rs | 189 ++++++ 17 files changed, 4051 insertions(+), 8 deletions(-) create mode 100644 crates/music-dht/Cargo.toml create mode 100644 crates/music-dht/README.md create mode 100644 crates/music-dht/src/config.rs create mode 100644 crates/music-dht/src/database.rs create mode 100644 crates/music-dht/src/dht.rs create mode 100644 crates/music-dht/src/error.rs create mode 100644 crates/music-dht/src/lib.rs create mode 100644 crates/music-dht/src/message.rs create mode 100644 crates/music-dht/src/node.rs create mode 100644 crates/music-dht/src/normalization.rs create mode 100644 crates/music-dht/src/record.rs create mode 100644 crates/music-dht/src/request.rs create mode 100644 crates/music-dht/src/routing.rs create mode 100644 crates/music-dht/src/service.rs create mode 100644 crates/music-dht/tests/integration.rs diff --git a/Cargo.lock b/Cargo.lock index e96c6af..f09d184 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1269,11 +1269,11 @@ dependencies = [ [[package]] name = "hashlink" -version = "0.12.1" +version = "0.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32069d97bb81e38fa67eab65e3393bf804bb85969f2bc06bf13f64aef5aba248" +checksum = "824e001ac4f3012dd16a264bec811403a67ca9deb6c102fc5049b32c4574b35f" dependencies = [ - "hashbrown 0.17.1", + "hashbrown 0.16.1", ] [[package]] @@ -1987,9 +1987,9 @@ checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" [[package]] name = "libsqlite3-sys" -version = "0.38.1" +version = "0.37.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f6c19a05435c21ac299d71b6a9c13db3e3f47c520517d58990a462a1397a61db" +checksum = "b1f111c8c41e7c61a49cd34e44c7619462967221a6443b0ec299e0ac30cfb9b1" dependencies = [ "cc", "pkg-config", @@ -2134,6 +2134,27 @@ dependencies = [ "uuid", ] +[[package]] +name = "music-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", +] + [[package]] name = "n0-error" version = "1.0.0" @@ -3004,9 +3025,9 @@ dependencies = [ [[package]] name = "rusqlite" -version = "0.40.1" +version = "0.39.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "11438310b19e3109b6446c33d1ed5e889428cf2e278407bc7896bc4aaea43323" +checksum = "a0d2b0146dd9661bf67bb107c0bb2a55064d556eeb3fc314151b957f313bcd4e" dependencies = [ "bitflags", "fallible-iterator", diff --git a/Cargo.toml b/Cargo.toml index 596c521..717f6c1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,6 +3,7 @@ resolver = "3" members = [ "crates/federation-net", "crates/artist-dht", + "crates/music-dht", "apps/federation-net-demo", "apps/artist-dht-cli", ] @@ -25,7 +26,7 @@ blake3 = "1" futures = "0.3" uuid = { version = "1", features = ["v7"] } unicode-normalization = "0.1" -rusqlite = { version = "0.40", features = ["bundled"] } +rusqlite = { version = "0.39", features = ["bundled"] } thiserror = "2" tracing = "0.1" clap = { version = "4", features = ["derive"] } diff --git a/crates/music-dht/Cargo.toml b/crates/music-dht/Cargo.toml new file mode 100644 index 0000000..a347c1d --- /dev/null +++ b/crates/music-dht/Cargo.toml @@ -0,0 +1,26 @@ +[package] +name = "music-dht" +version = "0.1.0" +description = "Distributed music library search: 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 } +unicode-normalization = { workspace = true } +rusqlite = { workspace = true } +thiserror = { workspace = true } +tracing = { workspace = true } +rand = { workspace = true } +data-encoding = { workspace = true } + +[dev-dependencies] +anyhow = { workspace = true } +tempfile = { workspace = true } +iroh = { workspace = true } diff --git a/crates/music-dht/README.md b/crates/music-dht/README.md new file mode 100644 index 0000000..dfb3d06 --- /dev/null +++ b/crates/music-dht/README.md @@ -0,0 +1,78 @@ +# music-dht + +A distributed **music library directory** built on +[`federation-net`](../federation-net): peers publish their local library +index — artists, releases and tracks (names and small metadata, **never +files**) — into a Kademlia-style DHT and search each other's libraries. +Every running node is simultaneously a client, a DHT router and a storage +node; there are no dedicated servers of any kind. + +This crate is the grown-up sibling of the minimal +[`artist-dht`](../artist-dht) example: same DHT machinery, richer records and +an application-oriented API. + +## Records + +A [`LibraryItem`] carries: `kind` (artist | release | track), `name`, +`artist_names` (for releases/tracks), `year`, `release_type`, +`duration_seconds`, plus ownership and versioning metadata. Records are +published under one exact key (the normalized name) and one token key per +word of the name **and of every artist name**, so searching for an artist +also returns their releases and tracks. + +## API + +The application does not add or delete records one by one — it declares the +desired state and the service diffs: + +```rust +let (service, events) = MusicDhtService::start(config).await?; +// Publish (and later re-publish) the whole library; matched by local_key. +service.sync_library(vec![ + ItemSpec { + local_key: "artist:1".into(), + kind: ItemKind::Artist, + name: "Massive Attack".into(), + artist_names: vec![], + year: None, + release_type: None, + duration_seconds: None, + }, + // ... +]).await?; +let outcome = service.search_network("teardrop").await?; +``` + +`sync_library` is idempotent: item ids are derived from +`(owner, kind, local_key)`, so unchanged items are skipped, changed ones are +republished with a bumped revision and items that disappeared from the input +are tombstoned network-wide. + +## Peer discovery + +With `.rendezvous(RendezvousConfig::default())` in the config, peers of a +network find each other knowing **only the network id** (a shared rendezvous +record in the public BitTorrent Mainline DHT — see the `federation-net` +README). Tickets (`service.ticket()` / `service.connect(ticket)`) remain as a +manual fallback for isolated networks. + +The network id is a public rendezvous token: anyone who knows it can join +and see the published names. Use a unique, hard-to-guess name for a private +network. + +## Consumers + +[`furumi-fd`](../../../furumi-stack/furumi-fd) uses this crate for its +federation feature: every instance publishes its library index and can search +the libraries of all other instances on the same network. + +## Verification + +```bash +cargo fmt --check +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo test -p music-dht +``` + +The integration test starts three real nodes in one process (they use Iroh's +public relay infrastructure), so it needs network access. diff --git a/crates/music-dht/src/config.rs b/crates/music-dht/src/config.rs new file mode 100644 index 0000000..3473c1c --- /dev/null +++ b/crates/music-dht/src/config.rs @@ -0,0 +1,208 @@ +//! Service configuration. + +use std::path::PathBuf; +use std::time::Duration; + +use federation_net::{NetworkId, RendezvousConfig}; + +use crate::error::{MusicDhtError, 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); +/// Default timeout for on-demand dials during DHT operations. +pub const DEFAULT_DIAL_TIMEOUT: Duration = Duration::from_secs(5); + +/// Configuration for an [`crate::MusicDhtService`]. +/// +/// Use [`MusicDhtConfig::builder`] to construct a validated instance. +#[derive(Debug, Clone)] +pub struct MusicDhtConfig { + /// 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, + /// Timeout for dialing a contact **on demand during DHT operations** + /// (lookups, publishes). Deliberately shorter than `transport_timeout`: + /// a dead contact must not stall a whole lookup, and a peer that needs + /// longer than this to dial will still be reached by the periodic + /// rendezvous/republish machinery. + pub dial_timeout: Duration, + /// Automatic peer discovery over the mainline DHT: peers of the same + /// network find each other knowing nothing but the network id. `None` + /// disables it; peers are then connected via tickets only. + pub rendezvous: Option, +} + +impl MusicDhtConfig { + /// Returns a new [`MusicDhtConfigBuilder`]. + pub fn builder() -> MusicDhtConfigBuilder { + MusicDhtConfigBuilder::default() + } +} + +/// Builder for [`MusicDhtConfig`]. +/// +/// `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 MusicDhtConfigBuilder { + data_dir: Option, + network_id: Option, + republish_interval: Option, + expire_interval: Option, + request_timeout: Option, + lookup_timeout: Option, + transport_timeout: Option, + dial_timeout: Option, + rendezvous: Option, +} + +impl MusicDhtConfigBuilder { + /// Sets the data directory. + pub fn data_dir(mut self, dir: impl Into) -> Self { + self.data_dir = Some(dir.into()); + self + } + + /// Sets the network identifier. + pub fn network_id(mut self, network_id: NetworkId) -> Self { + self.network_id = Some(network_id); + self + } + + /// Sets the republish interval. + pub fn republish_interval(mut self, interval: Duration) -> Self { + self.republish_interval = Some(interval); + self + } + + /// Sets the expired-record sweep interval. + pub fn expire_interval(mut self, interval: Duration) -> Self { + self.expire_interval = Some(interval); + self + } + + /// Sets the timeout of a single DHT request. + pub fn request_timeout(mut self, timeout: Duration) -> Self { + self.request_timeout = Some(timeout); + self + } + + /// Sets the timeout of a whole iterative lookup. + pub fn lookup_timeout(mut self, timeout: Duration) -> Self { + self.lookup_timeout = Some(timeout); + self + } + + /// Sets the timeout for transport operations (dialing, handshakes). + pub fn transport_timeout(mut self, timeout: Duration) -> Self { + self.transport_timeout = Some(timeout); + self + } + + /// Sets the timeout for on-demand dials during DHT operations. + pub fn dial_timeout(mut self, timeout: Duration) -> Self { + self.dial_timeout = Some(timeout); + self + } + + /// Enables automatic peer discovery over the mainline DHT. + pub fn rendezvous(mut self, rendezvous: RendezvousConfig) -> Self { + self.rendezvous = Some(rendezvous); + self + } + + /// Validates and builds the configuration. + pub fn build(self) -> Result { + let data_dir = self + .data_dir + .ok_or_else(|| MusicDhtError::Database("data_dir is required".into()))?; + if data_dir.as_os_str().is_empty() { + return Err(MusicDhtError::Database("data_dir must not be empty".into())); + } + let network_id = self + .network_id + .ok_or_else(|| MusicDhtError::Network("network_id is required".into()))?; + + let config = MusicDhtConfig { + 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), + dial_timeout: self.dial_timeout.unwrap_or(DEFAULT_DIAL_TIMEOUT), + rendezvous: self.rendezvous, + }; + 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), + ("dial_timeout", config.dial_timeout), + ] { + if value.is_zero() { + return Err(MusicDhtError::Database(format!( + "{name} must be greater than zero" + ))); + } + } + Ok(config) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn builder_applies_defaults() { + let config = MusicDhtConfig::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!(MusicDhtConfig::builder().build().is_err()); + assert!( + MusicDhtConfig::builder() + .data_dir("./dir") + .network_id(NetworkId::from_name("test")) + .request_timeout(Duration::ZERO) + .build() + .is_err() + ); + } +} diff --git a/crates/music-dht/src/database.rs b/crates/music-dht/src/database.rs new file mode 100644 index 0000000..9e250d0 --- /dev/null +++ b/crates/music-dht/src/database.rs @@ -0,0 +1,440 @@ +//! 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::{MusicDhtError, Result}; +use crate::message::MAX_RECORDS_PER_RESPONSE; +use crate::normalization::tokenize; +use crate::record::{DhtKey, LibraryItem, StoredRecord, TOMBSTONE_TTL}; +use crate::routing::{NodeContact, NodeId}; + +const SCHEMA: &str = " +CREATE TABLE IF NOT EXISTS local_items ( + id BLOB PRIMARY KEY, + normalized_name TEXT NOT NULL, + revision INTEGER NOT NULL, + deleted INTEGER NOT NULL DEFAULT 0, + updated_at_ms INTEGER NOT NULL, + payload BLOB NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_local_items_normalized_name + ON local_items(normalized_name); + +CREATE TABLE IF NOT EXISTS dht_records ( + dht_key BLOB NOT NULL, + artist_id BLOB NOT NULL, + owner_peer_id TEXT NOT NULL, + payload BLOB NOT NULL, + revision INTEGER NOT NULL, + deleted INTEGER NOT NULL, + expires_at_ms INTEGER NOT NULL, + + PRIMARY KEY (dht_key, artist_id, owner_peer_id) +); + +CREATE INDEX IF NOT EXISTS idx_dht_records_expires_at + ON dht_records(expires_at_ms); + +CREATE TABLE IF NOT EXISTS known_peers ( + peer_id TEXT PRIMARY KEY, + node_id BLOB NOT NULL, + ticket TEXT NOT NULL, + last_seen_ms INTEGER NOT NULL +); +"; + +/// Handle to the local SQLite database. +/// +/// Cheap to clone; all clones share one connection guarded by a mutex that is +/// only ever locked from blocking-pool threads. +#[derive(Clone)] +pub(crate) struct Database { + conn: Arc>, +} + +impl Database { + /// Opens (creating if needed) the database at `path` and applies the + /// schema. + pub async fn open(path: &Path) -> Result { + let path = path.to_path_buf(); + let conn = tokio::task::spawn_blocking(move || -> Result { + let conn = Connection::open(&path).map_err(|err| { + MusicDhtError::Database(format!("failed to open {}: {err}", path.display())) + })?; + conn.execute_batch(SCHEMA) + .map_err(|err| MusicDhtError::Database(format!("failed to apply schema: {err}")))?; + Ok(conn) + }) + .await + .map_err(|err| MusicDhtError::Database(format!("database task panicked: {err}")))??; + Ok(Self { + conn: Arc::new(Mutex::new(conn)), + }) + } + + /// Runs a closure against the connection on the blocking pool. + async fn call(&self, f: F) -> Result + where + F: FnOnce(&Connection) -> rusqlite::Result + Send + 'static, + R: Send + 'static, + { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let guard = conn + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + f(&guard).map_err(|err| MusicDhtError::Database(err.to_string())) + }) + .await + .map_err(|err| MusicDhtError::Database(format!("database task panicked: {err}")))? + } + + /// Inserts or replaces a locally owned item record. + pub async fn upsert_local_item(&self, item: &LibraryItem) -> Result<()> { + let item = item.clone(); + self.call(move |conn| { + let payload = postcard::to_stdvec(&item) + .map_err(|err| rusqlite::Error::ToSqlConversionFailure(Box::new(err)))?; + conn.execute( + "INSERT OR REPLACE INTO local_items + (id, normalized_name, revision, deleted, updated_at_ms, payload) + VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + params![ + item.id.as_bytes().as_slice(), + item.normalized_name, + item.revision as i64, + item.deleted as i64, + item.updated_at_ms as i64, + payload, + ], + )?; + Ok(()) + }) + .await + } + + /// Lists locally owned items. Tombstones are excluded unless + /// `include_deleted` is set. + pub async fn list_local_items(&self, include_deleted: bool) -> Result> { + self.call(move |conn| { + let mut stmt = conn.prepare( + "SELECT payload + FROM local_items ORDER BY normalized_name", + )?; + let rows = stmt.query_map([], item_from_row)?; + let mut result = Vec::new(); + for row in rows { + let item = row?; + if include_deleted || !item.deleted { + result.push(item); + } + } + Ok(result) + }) + .await + } + + /// Returns everything that must be republished: active records plus + /// tombstones that have not outlived [`TOMBSTONE_TTL`] yet. + pub async fn local_items_for_republish(&self, now_ms: u64) -> Result> { + let all = self.list_local_items(true).await?; + let tombstone_ttl = TOMBSTONE_TTL.as_millis() as u64; + Ok(all + .into_iter() + .filter(|item| { + !item.deleted || item.updated_at_ms.saturating_add(tombstone_ttl) > now_ms + }) + .collect()) + } + + /// Searches locally owned active items: exact normalized match, or all + /// query tokens present in the item's token set. + pub async fn search_local(&self, normalized_query: String) -> Result> { + let all = self.list_local_items(false).await?; + let query_tokens = tokenize(&normalized_query); + Ok(all + .into_iter() + .filter(|item| { + if item.normalized_name == normalized_query { + return true; + } + if query_tokens.is_empty() { + return false; + } + let artist_tokens = tokenize(&item.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: StoredRecord) -> Result { + self.call(move |conn| { + let existing: Option<(i64, i64, i64)> = conn + .query_row( + "SELECT revision, deleted, expires_at_ms FROM dht_records + WHERE dht_key = ?1 AND artist_id = ?2 AND owner_peer_id = ?3", + params![ + key.as_bytes().as_slice(), + record.item.id.as_bytes().as_slice(), + record.item.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.item.id.as_bytes().as_slice(), + record.item.owner.to_string(), + payload, + record.item.revision as i64, + record.item.deleted as i64, + record.expires_at_ms as i64, + ], + )?; + Ok(true) + } + } + }) + .await + } + + /// Returns non-expired replicas stored under `key`, including tombstones + /// (they inform other peers about deletions). Capped at + /// [`MAX_RECORDS_PER_RESPONSE`]. + pub async fn dht_records_by_key(&self, key: DhtKey, now_ms: u64) -> Result> { + self.call(move |conn| { + let mut stmt = conn.prepare( + "SELECT payload FROM dht_records + WHERE dht_key = ?1 AND expires_at_ms > ?2 + LIMIT ?3", + )?; + let rows = stmt.query_map( + params![ + key.as_bytes().as_slice(), + now_ms as i64, + MAX_RECORDS_PER_RESPONSE as i64 + ], + |row| row.get::<_, Vec>(0), + )?; + let mut records = Vec::new(); + for row in rows { + let payload = row?; + // A payload we cannot decode is skipped, not fatal. + if let Ok(record) = postcard::from_bytes::(&payload) { + records.push(record); + } + } + Ok(records) + }) + .await + } + + /// Deletes expired replicas. Returns the number of removed rows. + pub async fn delete_expired_records(&self, now_ms: u64) -> Result { + self.call(move |conn| { + conn.execute( + "DELETE FROM dht_records WHERE expires_at_ms <= ?1", + params![now_ms as i64], + ) + }) + .await + } + + /// Inserts or refreshes a known peer contact. + pub async fn upsert_known_peer(&self, contact: &NodeContact) -> Result<()> { + let contact = contact.clone(); + self.call(move |conn| { + conn.execute( + "INSERT OR REPLACE INTO known_peers (peer_id, node_id, ticket, last_seen_ms) + VALUES (?1, ?2, ?3, ?4)", + params![ + contact.peer_id.to_string(), + contact.node_id.as_bytes().as_slice(), + contact.ticket, + contact.last_seen_ms as i64, + ], + )?; + Ok(()) + }) + .await + } + + /// Deletes a persisted peer contact (e.g. after repeated failed dials). + pub async fn delete_known_peer(&self, peer_id: EndpointId) -> Result<()> { + self.call(move |conn| { + conn.execute( + "DELETE FROM known_peers WHERE peer_id = ?1", + params![peer_id.to_string()], + )?; + Ok(()) + }) + .await + } + + /// Loads all persisted peer contacts. + pub async fn load_known_peers(&self) -> Result> { + self.call(|conn| { + let mut stmt = + conn.prepare("SELECT peer_id, node_id, ticket, last_seen_ms FROM known_peers")?; + let rows = stmt.query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, Vec>(1)?, + row.get::<_, String>(2)?, + row.get::<_, i64>(3)?, + )) + })?; + let mut contacts = Vec::new(); + for row in rows { + let (peer_id, node_id, ticket, last_seen_ms) = row?; + let Ok(peer_id) = EndpointId::from_str(&peer_id) else { + continue; + }; + let Ok(node_id) = <[u8; 32]>::try_from(node_id.as_slice()) else { + continue; + }; + contacts.push(NodeContact { + node_id: NodeId::from_bytes(node_id), + peer_id, + ticket, + last_seen_ms: last_seen_ms as u64, + }); + } + Ok(contacts) + }) + .await + } +} + +fn item_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let payload: Vec = row.get(0)?; + postcard::from_bytes(&payload).map_err(|err| { + rusqlite::Error::FromSqlConversionFailure( + 0, + rusqlite::types::Type::Blob, + format!("undecodable local item payload: {err}").into(), + ) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::record::{ItemId, ItemKind, now_ms}; + + fn test_peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + fn item(owner: EndpointId, name: &str, revision: u64, deleted: bool) -> LibraryItem { + LibraryItem { + id: ItemId::derive(&owner, ItemKind::Artist, name), + owner, + kind: ItemKind::Artist, + name: name.to_string(), + normalized_name: crate::normalization::normalize_name(name), + artist_names: Vec::new(), + year: None, + release_type: None, + duration_seconds: None, + 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_item_round_trip() { + let (_dir, db) = open_temp().await; + let owner = test_peer(1); + let item = item(owner, "Massive Attack", 1, false); + db.upsert_local_item(&item).await.expect("upsert"); + let listed = db.list_local_items(true).await.expect("list"); + assert_eq!(listed, vec![item.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 item = item(owner, "Massive Attack", 1, false); + let key = DhtKey::exact(&federation_net::NetworkId::from_name("t"), "massive attack"); + let now = now_ms(); + let record = StoredRecord { + item, + publisher: owner, + expires_at_ms: now + 50, + }; + assert!(db.store_dht_record(key, record).await.expect("store")); + assert_eq!(db.dht_records_by_key(key, now).await.expect("get").len(), 1); + // After expiry the record is filtered out and then swept. + let later = now + 100; + assert!( + db.dht_records_by_key(key, later) + .await + .expect("get") + .is_empty() + ); + assert_eq!(db.delete_expired_records(later).await.expect("sweep"), 1); + } + + #[tokio::test] + async fn known_peers_round_trip() { + let (_dir, db) = open_temp().await; + let peer = test_peer(2); + let contact = NodeContact { + node_id: NodeId::from_endpoint(&peer), + peer_id: peer, + ticket: "fnet-test".into(), + last_seen_ms: 42, + }; + db.upsert_known_peer(&contact).await.expect("upsert"); + let loaded = db.load_known_peers().await.expect("load"); + assert_eq!(loaded, vec![contact]); + } +} diff --git a/crates/music-dht/src/dht.rs b/crates/music-dht/src/dht.rs new file mode 100644 index 0000000..2582da4 --- /dev/null +++ b/crates/music-dht/src/dht.rs @@ -0,0 +1,279 @@ +//! DHT record validation and replacement rules. + +use federation_net::NetworkId; + +use crate::message::StoreRecordRequest; +use crate::normalization::{normalize_name, tokenize}; +use crate::record::{ + ACTIVE_RECORD_TTL, DhtKey, MAX_ITEM_NAME_BYTES, MAX_TOKENS_PER_ITEM, StoredRecord, + 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: &StoredRecord, +) -> StoreDecision { + let Some((revision, deleted, expires_at_ms)) = existing else { + return StoreDecision::Write; + }; + let item = &incoming.item; + if item.revision > revision { + return StoreDecision::Write; + } + if item.revision < revision { + return StoreDecision::Ignore; + } + // Equal revisions. + match (deleted, item.deleted) { + (false, true) => StoreDecision::Write, + (true, false) => StoreDecision::Ignore, + _ => { + if incoming.expires_at_ms > expires_at_ms { + StoreDecision::RefreshExpiry(incoming.expires_at_ms) + } else { + StoreDecision::Ignore + } + } + } +} + +/// Validates an incoming `StoreRecord` request. +/// +/// Checks the size limits, that the record is internally consistent, that the +/// key actually corresponds to the record's name or one of its tokens, and +/// clamps the expiry to the maximum TTL allowed for the record type. Returns +/// the record with a possibly clamped `expires_at_ms`. +pub(crate) fn validate_store( + request: StoreRecordRequest, + network_id: &NetworkId, + now_ms: u64, +) -> Result { + let mut record = request.record; + let item = &record.item; + + if item.name.len() > MAX_ITEM_NAME_BYTES { + return Err("item name too long".into()); + } + if item.normalized_name != normalize_name(&item.name) { + return Err("normalized name does not match the item name".into()); + } + if item.normalized_name.is_empty() { + return Err("item name normalizes to nothing".into()); + } + let tokens = tokenize(&item.normalized_name); + if tokens.len() > MAX_TOKENS_PER_ITEM { + return Err("too many tokens".into()); + } + + let key_matches = request.key == DhtKey::exact(network_id, &item.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 item.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::{ItemId, LibraryItem}; + + fn test_peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + fn record(revision: u64, deleted: bool, expires_at_ms: u64) -> StoredRecord { + let owner = test_peer(1); + StoredRecord { + item: LibraryItem { + id: ItemId::from_bytes([9u8; 32]), + owner, + kind: crate::record::ItemKind::Artist, + name: "Massive Attack".into(), + normalized_name: "massive attack".into(), + artist_names: Vec::new(), + year: None, + release_type: None, + duration_seconds: None, + 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.item.normalized_name = "something else".into(); + assert!( + validate_store( + StoreRecordRequest { + key: DhtKey::exact(&net, "something else"), + record: bad, + }, + &net, + now, + ) + .is_err() + ); + } +} diff --git a/crates/music-dht/src/error.rs b/crates/music-dht/src/error.rs new file mode 100644 index 0000000..8dee393 --- /dev/null +++ b/crates/music-dht/src/error.rs @@ -0,0 +1,65 @@ +//! Error types for the music-dht library. + +/// Convenient result alias used across the library. +pub type Result = std::result::Result; + +/// All errors that can be returned by the public API of this library. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum MusicDhtError { + /// 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 item name is empty or normalizes to nothing. + #[error("invalid item name")] + InvalidItemName, + + /// The item name exceeds the maximum allowed length. + #[error("item name is too long")] + ItemNameTooLong, + + /// No item with the given id exists locally. + #[error("item not found")] + ItemNotFound, + + /// Only locally created items can be deleted. + #[error("cannot delete a remote item")] + CannotDeleteRemoteItem, + + /// A DHT request did not receive a response in time. + #[error("request timed out")] + Timeout, + + /// The lookup exhausted its request budget without finishing. + #[error("lookup budget exhausted")] + LookupBudgetExhausted, + + /// A stored peer ticket could not be parsed. + #[error("invalid peer ticket: {0}")] + InvalidTicket(String), + + /// A peer violated the DHT protocol. + #[error("protocol error: {0}")] + Protocol(String), + + /// The service is shutting down and no longer accepts operations. + #[error("service is shutting down")] + ShuttingDown, +} + +impl From for MusicDhtError { + fn from(err: federation_net::NetworkError) -> Self { + use federation_net::NetworkError; + match err { + NetworkError::Timeout => Self::Timeout, + NetworkError::ShuttingDown => Self::ShuttingDown, + NetworkError::InvalidTicket(msg) => Self::InvalidTicket(msg), + other => Self::Network(other.to_string()), + } + } +} diff --git a/crates/music-dht/src/lib.rs b/crates/music-dht/src/lib.rs new file mode 100644 index 0000000..a3a8413 --- /dev/null +++ b/crates/music-dht/src/lib.rs @@ -0,0 +1,108 @@ +//! # music-dht +//! +//! A distributed music library directory on top of [`federation_net`]: +//! peers publish their local library (artists, releases, tracks — names and +//! small metadata, never files) into a Kademlia-style DHT and search each +//! other's libraries. Every running node is simultaneously a client, a DHT +//! router and a storage node — there are no dedicated bootstrap, index or +//! search servers. This crate is the grown-up sibling of the minimal +//! `artist-dht` example. +//! +//! ## How it works +//! +//! * Every peer derives a stable 256-bit [`NodeId`] from its persistent +//! `federation-net` endpoint id. +//! * [`LibraryItem`] records are published under BLAKE3-derived [`DhtKey`]s: +//! one exact key for the whole normalized name plus one key per token of +//! the name and of every artist name (so a track is found by its artist). +//! * 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 music_dht::{ItemKind, ItemSpec, MusicDhtConfig, MusicDhtService}; +//! use federation_net::{NetworkId, RendezvousConfig}; +//! +//! # async fn run() -> music_dht::Result<()> { +//! let config = MusicDhtConfig::builder() +//! .data_dir("./peer-a") +//! .network_id(NetworkId::from_name("my-music-network")) +//! // Find other peers knowing only the network id. +//! .rendezvous(RendezvousConfig::default()) +//! .build()?; +//! let (service, mut events) = MusicDhtService::start(config).await?; +//! +//! // Publish (and keep in sync) the local library. +//! service +//! .sync_library(vec![ItemSpec { +//! local_key: "track:42".into(), +//! kind: ItemKind::Track, +//! name: "Teardrop".into(), +//! artist_names: vec!["Massive Attack".into()], +//! year: Some(1998), +//! release_type: None, +//! duration_seconds: Some(330.0), +//! }]) +//! .await?; +//! +//! // Search the whole network. +//! let outcome = service.search_network("massive attack").await?; +//! for item in &outcome.network_results { +//! println!("{} {} owned by {}", item.kind, item.name, item.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::{ + DEFAULT_EXPIRE_INTERVAL, DEFAULT_LOOKUP_TIMEOUT, DEFAULT_REPUBLISH_INTERVAL, + DEFAULT_REQUEST_TIMEOUT, DEFAULT_TRANSPORT_TIMEOUT, MusicDhtConfig, MusicDhtConfigBuilder, +}; +pub use error::{MusicDhtError, Result}; +pub use message::{ + DHT_PROTOCOL_VERSION, FindNodeRequest, FindNodeResponse, FindValueRequest, FindValueResponse, + Hello, MAX_PEER_EXCHANGE_CONTACTS, MAX_RECORDS_PER_RESPONSE, MusicDhtMessage, PeerExchange, + PingRequest, PongResponse, RequestEnvelope, RequestId, ResponseEnvelope, StoreRecordRequest, + StoreRecordResponse, +}; +pub use normalization::{normalize_name, tokenize}; +pub use record::{ + ACTIVE_RECORD_TTL, DhtKey, ItemId, ItemKind, LibraryItem, MAX_ARTISTS_PER_ITEM, + MAX_ITEM_NAME_BYTES, MAX_TOKENS_PER_ITEM, PeerId, StoredRecord, 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::{ + ItemSpec, MusicDhtEvent, MusicDhtEventReceiver, MusicDhtService, PublishStats, SCHEMA_NAME, + SearchOutcome, SyncStats, +}; + +// Re-exported types from the transport layer that appear in this API. +pub use federation_net::{EndpointId, NetworkId, PeerTicket, RendezvousConfig}; diff --git a/crates/music-dht/src/message.rs b/crates/music-dht/src/message.rs new file mode 100644 index 0000000..61a8724 --- /dev/null +++ b/crates/music-dht/src/message.rs @@ -0,0 +1,174 @@ +//! The domain protocol carried over `federation-net`. + +use std::fmt; + +use federation_net::EndpointId; +use serde::{Deserialize, Serialize}; + +use crate::record::{DhtKey, StoredRecord}; +use crate::routing::{NodeContact, NodeId}; + +/// Version of the music-dht protocol. +pub const DHT_PROTOCOL_VERSION: u16 = 1; +/// Maximum number of contacts in a single [`PeerExchange`]. +pub const MAX_PEER_EXCHANGE_CONTACTS: usize = 32; +/// Maximum number of records in a single [`FindValueResponse`]. +pub const MAX_RECORDS_PER_RESPONSE: usize = 100; + +/// Correlates a response with its request. +#[derive(Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct RequestId([u8; 16]); + +impl RequestId { + /// Generates a random request id. + pub fn random() -> Self { + Self(rand::random()) + } + + /// Returns the raw bytes. + pub fn as_bytes(&self) -> &[u8; 16] { + &self.0 + } +} + +impl fmt::Debug for RequestId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "RequestId(")?; + for byte in &self.0 { + write!(f, "{byte:02x}")?; + } + write!(f, ")") + } +} + +/// Wraps a request payload with its correlation id. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RequestEnvelope { + /// Correlation id; echoed back in the response. + pub request_id: RequestId, + /// The request itself. + pub payload: T, +} + +/// Wraps a response payload with the correlation id of its request. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ResponseEnvelope { + /// Correlation id of the request being answered. + pub request_id: RequestId, + /// The response itself. + pub payload: T, +} + +/// Introduction sent right after a connection is established. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Hello { + /// DHT identifier of the sender. + pub node_id: NodeId, + /// Transport identifier of the sender (informational; the authenticated + /// id always comes from the connection itself). + pub peer_id: EndpointId, + /// Ticket other peers can use to reach the sender. + pub ticket: String, + /// Protocol version of the sender. + pub protocol_version: u16, +} + +/// A batch of known contacts, shared after [`Hello`]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PeerExchange { + /// Up to [`MAX_PEER_EXCHANGE_CONTACTS`] contacts. + pub peers: Vec, +} + +/// Liveness probe. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PingRequest {} + +/// Reply to [`PingRequest`]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PongResponse { + /// DHT identifier of the responder. + pub node_id: NodeId, +} + +/// Asks for the closest known nodes to `target`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FindNodeRequest { + /// Point of the key space to search around. + pub target: NodeId, +} + +/// Reply to [`FindNodeRequest`]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FindNodeResponse { + /// Up to `K` known nodes closest to the target. + pub nodes: Vec, +} + +/// Asks for records stored under `key`, or the closest nodes to it. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FindValueRequest { + /// The DHT key to look up. + pub key: DhtKey, +} + +/// Reply to [`FindValueRequest`]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum FindValueResponse { + /// The responder stores records under the key. + Records { + /// Up to [`MAX_RECORDS_PER_RESPONSE`] non-expired records. + records: Vec, + }, + /// The responder has nothing stored; here are closer nodes instead. + CloserNodes { + /// Up to `K` known nodes closest to the key. + nodes: Vec, + }, +} + +/// Asks the receiver to store a replica of a record. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StoreRecordRequest { + /// The key the record is published under. + pub key: DhtKey, + /// The record to store. + pub record: StoredRecord, +} + +/// 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 music-dht peers. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[allow(clippy::large_enum_variant)] +pub enum MusicDhtMessage { + /// Introduction after connect. + Hello(Hello), + /// Contact gossip after `Hello`. + PeerExchange(PeerExchange), + + /// Liveness probe. + Ping(RequestEnvelope), + /// Reply to `Ping`. + Pong(ResponseEnvelope), + + /// Node lookup request. + FindNode(RequestEnvelope), + /// Reply to `FindNode`. + FindNodeResult(ResponseEnvelope), + + /// Value lookup request. + FindValue(RequestEnvelope), + /// Reply to `FindValue`. + FindValueResult(ResponseEnvelope), + + /// Replication request. + StoreRecord(RequestEnvelope), + /// Reply to `StoreRecord`. + StoreRecordResult(ResponseEnvelope), +} diff --git a/crates/music-dht/src/node.rs b/crates/music-dht/src/node.rs new file mode 100644 index 0000000..944709c --- /dev/null +++ b/crates/music-dht/src/node.rs @@ -0,0 +1,924 @@ +//! 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::{Duration, Instant}; + +use federation_net::{EndpointId, NetworkEngine, NetworkEvent, NetworkEventReceiver, PeerTicket}; +use futures::future::join_all; +use futures::stream::{FuturesUnordered, StreamExt}; +use tokio::sync::mpsc; +use tokio::time::timeout; +use tracing::{debug, info, warn}; + +use crate::config::MusicDhtConfig; +use crate::database::Database; +use crate::dht::validate_store; +use crate::error::{MusicDhtError, Result}; +use crate::message::{ + DHT_PROTOCOL_VERSION, FindNodeRequest, FindNodeResponse, FindValueRequest, FindValueResponse, + Hello, MAX_PEER_EXCHANGE_CONTACTS, MusicDhtMessage, PeerExchange, PingRequest, PongResponse, + RequestEnvelope, RequestId, ResponseEnvelope, StoreRecordRequest, StoreRecordResponse, +}; +use crate::record::{ACTIVE_RECORD_TTL, DhtKey, LibraryItem, StoredRecord, TOMBSTONE_TTL, now_ms}; +use crate::request::{DhtResponse, PendingRequests}; +use crate::routing::{ALPHA, K, MAX_LOOKUP_REQUESTS, NodeContact, NodeId, RoutingTable, distance}; +use crate::service::{MusicDhtEvent, PublishStats}; + +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(PoisonError::into_inner) +} + +/// Base of the exponential backoff applied to a contact after a failed dial. +const DIAL_BACKOFF_BASE: Duration = Duration::from_secs(30); +/// Upper bound of the dial backoff. +const DIAL_BACKOFF_MAX: Duration = Duration::from_secs(10 * 60); +/// Consecutive failed dials after which a contact is evicted entirely. +const DIAL_FAILURES_BEFORE_EVICT: u32 = 5; + +/// Dial-failure state of one currently unreachable contact. +#[derive(Debug, Clone, Copy)] +struct DialFailure { + consecutive: u32, + last_attempt_ms: u64, +} + +/// How long a contact is skipped after `consecutive` failed dials: +/// 30s, 1m, 2m, 4m, 8m, then capped at 10 minutes. +fn dial_backoff(consecutive: u32) -> Duration { + let exponent = consecutive.saturating_sub(1).min(8); + DIAL_BACKOFF_BASE + .saturating_mul(1u32 << exponent) + .min(DIAL_BACKOFF_MAX) +} + +/// An outbound DHT request, before it is wrapped in an envelope. +enum OutboundRequest { + Ping, + FindNode(FindNodeRequest), + FindValue(FindValueRequest), + Store(Box), +} + +/// Result of one iterative lookup. +pub(crate) struct LookupOutcome { + /// Records found (value lookups only). + pub records: Vec, + /// Closest known contacts to the target, best first, at most `K`. + pub closest: Vec, + /// Number of distinct peers actually queried. + pub queried: usize, + /// Number of distinct nodes known to the lookup (seeds + discovered). + pub discovered: usize, +} + +/// Shared state of one DHT node. +pub(crate) struct Node { + pub engine: NetworkEngine, + pub db: Database, + pub config: MusicDhtConfig, + pub node_id: NodeId, + pub endpoint_id: EndpointId, + routing: Mutex, + pending: PendingRequests, + /// Peers we already introduced ourselves to (per connection). + hello_sent: Mutex>, + /// Peers we already gossiped contacts to (per connection). + exchange_sent: Mutex>, + /// Contacts that recently failed to dial, with their backoff state. + dial_failures: Mutex>, + events: Mutex>>, + /// Set once the post-startup republish has been triggered. + initial_republish_done: AtomicBool, + shutting_down: AtomicBool, +} + +impl Node { + pub fn new( + engine: NetworkEngine, + db: Database, + config: MusicDhtConfig, + events: mpsc::Sender, + ) -> Self { + let endpoint_id = engine.endpoint_id(); + let node_id = NodeId::from_endpoint(&endpoint_id); + Self { + engine, + db, + config, + node_id, + endpoint_id, + routing: Mutex::new(RoutingTable::new(node_id)), + pending: PendingRequests::default(), + hello_sent: Mutex::new(HashSet::new()), + exchange_sent: Mutex::new(HashSet::new()), + dial_failures: Mutex::new(HashMap::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(MusicDhtError::ShuttingDown) + } else { + Ok(()) + } + } + + async fn emit(&self, event: MusicDhtEvent) { + let sender = lock(&self.events).clone(); + if let Some(sender) = sender { + let _ = sender.send(event).await; + } + } + + /// All known DHT contacts. + pub fn known_contacts(&self) -> Vec { + lock(&self.routing).contacts() + } + + /// Seeds the routing table (used at startup with persisted contacts). + pub fn seed_contacts(&self, contacts: Vec) { + let mut routing = lock(&self.routing); + for contact in contacts { + if contact.peer_id != self.endpoint_id { + routing.upsert(contact); + } + } + } + + /// Adds or refreshes a contact learned from the network. + /// + /// The node id is always re-derived from the endpoint id instead of + /// trusting the gossiped value. The first contact ever learned triggers + /// the post-startup republish. + async fn upsert_contact(self: &Arc, peer_id: EndpointId, ticket: String) { + if peer_id == self.endpoint_id { + return; + } + let contact = NodeContact { + node_id: NodeId::from_endpoint(&peer_id), + peer_id, + ticket, + last_seen_ms: now_ms(), + }; + let is_new = lock(&self.routing).upsert(contact.clone()); + if let Err(err) = self.db.upsert_known_peer(&contact).await { + warn!(error = %err, "failed to persist known peer"); + } + if is_new { + info!(peer = %contact.peer_id, node = %contact.node_id, "learned new DHT contact"); + self.emit(MusicDhtEvent::ContactDiscovered { + contact: contact.clone(), + }) + .await; + } + self.maybe_trigger_initial_republish(); + } + + /// Spawns the post-startup republish once at least one contact is known. + pub fn maybe_trigger_initial_republish(self: &Arc) { + if lock(&self.routing).is_empty() || self.is_shutting_down() { + return; + } + if self.initial_republish_done.swap(true, Ordering::SeqCst) { + return; + } + let node = self.clone(); + tokio::spawn(async move { + match node.republish_all().await { + Ok(stats) => info!( + records = stats.records, + keys = stats.keys, + nodes = stats.remote_nodes, + "post-startup republish finished" + ), + Err(err) => warn!(error = %err, "post-startup republish failed"), + } + }); + } + + /// Consumes `federation-net` events until the engine shuts down. + pub async fn run_event_loop( + self: Arc, + mut receiver: NetworkEventReceiver, + ) { + while let Some(event) = receiver.recv().await { + match event { + NetworkEvent::PeerConnected { peer_id, .. } => { + debug!(peer = %peer_id, "peer connected"); + self.clear_dial_failures(&peer_id); + self.emit(MusicDhtEvent::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(MusicDhtEvent::PeerDisconnected { peer_id }).await; + } + NetworkEvent::MessageReceived { peer_id, message } => { + self.on_message(peer_id, message).await; + } + NetworkEvent::ProtocolError { peer_id, error } => { + self.emit(MusicDhtEvent::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: &MusicDhtMessage) -> Result<()> { + self.engine.send(peer, message).await.map_err(Into::into) + } + + async fn send_hello(self: &Arc, peer: EndpointId) { + // Mark before sending so a crossing Hello does not trigger an echo. + if !lock(&self.hello_sent).insert(peer) { + return; + } + let ticket = match self.engine.ticket().await { + Ok(ticket) => ticket.to_string(), + Err(err) => { + warn!(error = %err, "cannot create own ticket for hello"); + lock(&self.hello_sent).remove(&peer); + return; + } + }; + let hello = MusicDhtMessage::Hello(Hello { + node_id: self.node_id, + peer_id: self.endpoint_id, + ticket, + protocol_version: DHT_PROTOCOL_VERSION, + }); + if let Err(err) = self.send_message(peer, &hello).await { + debug!(peer = %peer, error = %err, "failed to send hello"); + lock(&self.hello_sent).remove(&peer); + } + } + + async fn send_peer_exchange(self: &Arc, peer: EndpointId) { + if !lock(&self.exchange_sent).insert(peer) { + return; + } + let mut peers: Vec = self + .known_contacts() + .into_iter() + .filter(|contact| contact.peer_id != peer && contact.peer_id != self.endpoint_id) + .collect(); + // Prefer the most recently seen contacts. + peers.sort_by_key(|contact| std::cmp::Reverse(contact.last_seen_ms)); + peers.truncate(MAX_PEER_EXCHANGE_CONTACTS); + if peers.is_empty() { + return; + } + debug!(peer = %peer, count = peers.len(), "sending peer exchange"); + let message = MusicDhtMessage::PeerExchange(PeerExchange { peers }); + if let Err(err) = self.send_message(peer, &message).await { + debug!(peer = %peer, error = %err, "failed to send peer exchange"); + } + } + + async fn on_message(self: &Arc, peer: EndpointId, message: MusicDhtMessage) { + lock(&self.routing).touch(&peer, now_ms()); + match message { + MusicDhtMessage::Hello(hello) => self.on_hello(peer, hello).await, + MusicDhtMessage::PeerExchange(exchange) => { + self.on_peer_exchange(peer, exchange).await; + } + MusicDhtMessage::Ping(env) => { + let response = MusicDhtMessage::Pong(ResponseEnvelope { + request_id: env.request_id, + payload: PongResponse { + node_id: self.node_id, + }, + }); + let _ = self.send_message(peer, &response).await; + } + MusicDhtMessage::FindNode(env) => { + let nodes = self.closest_for_response(env.payload.target.as_bytes(), &peer); + let response = MusicDhtMessage::FindNodeResult(ResponseEnvelope { + request_id: env.request_id, + payload: FindNodeResponse { nodes }, + }); + let _ = self.send_message(peer, &response).await; + } + MusicDhtMessage::FindValue(env) => { + let payload = self.answer_find_value(&env.payload, &peer).await; + let response = MusicDhtMessage::FindValueResult(ResponseEnvelope { + request_id: env.request_id, + payload, + }); + let _ = self.send_message(peer, &response).await; + } + MusicDhtMessage::StoreRecord(env) => { + let stored = self.answer_store(env.payload, &peer).await; + let response = MusicDhtMessage::StoreRecordResult(ResponseEnvelope { + request_id: env.request_id, + payload: StoreRecordResponse { stored }, + }); + let _ = self.send_message(peer, &response).await; + } + MusicDhtMessage::Pong(env) => { + self.pending + .complete(&env.request_id, &peer, DhtResponse::Pong(env.payload)); + } + MusicDhtMessage::FindNodeResult(env) => { + self.pending + .complete(&env.request_id, &peer, DhtResponse::FindNode(env.payload)); + } + MusicDhtMessage::FindValueResult(env) => { + self.pending + .complete(&env.request_id, &peer, DhtResponse::FindValue(env.payload)); + } + MusicDhtMessage::StoreRecordResult(env) => { + self.pending + .complete(&env.request_id, &peer, DhtResponse::Store(env.payload)); + } + } + } + + async fn on_hello(self: &Arc, peer: EndpointId, hello: Hello) { + if hello.protocol_version != DHT_PROTOCOL_VERSION { + warn!(peer = %peer, version = hello.protocol_version, "unsupported DHT protocol version"); + return; + } + // The authenticated identity comes from the connection; the id fields + // inside the payload must be consistent with it. + if hello.peer_id != peer || hello.node_id != NodeId::from_endpoint(&peer) { + warn!(peer = %peer, "hello with inconsistent identity; ignoring"); + self.emit(MusicDhtEvent::Error { + message: format!("peer {peer} sent a hello with a mismatched identity"), + }) + .await; + return; + } + debug!(peer = %peer, "received hello"); + self.upsert_contact(peer, hello.ticket).await; + // Introduce ourselves if the remote connected first, then gossip. + self.send_hello(peer).await; + self.send_peer_exchange(peer).await; + } + + async fn on_peer_exchange(self: &Arc, peer: EndpointId, exchange: PeerExchange) { + let contacts = sanitize_peer_exchange(self.endpoint_id, peer, exchange.peers); + let accepted = contacts.len(); + for contact in contacts { + self.upsert_contact(contact.peer_id, contact.ticket).await; + } + debug!(peer = %peer, accepted, "processed peer exchange"); + } + + /// Contacts for a FindNode/FindValue response: closest to the target, + /// excluding the requester itself. + fn closest_for_response(&self, target: &[u8; 32], requester: &EndpointId) -> Vec { + lock(&self.routing) + .closest(target, K + 1) + .into_iter() + .filter(|contact| &contact.peer_id != requester) + .take(K) + .collect() + } + + async fn answer_find_value( + &self, + request: &FindValueRequest, + requester: &EndpointId, + ) -> FindValueResponse { + match self.db.dht_records_by_key(request.key, now_ms()).await { + Ok(records) if !records.is_empty() => FindValueResponse::Records { records }, + Ok(_) => FindValueResponse::CloserNodes { + nodes: self.closest_for_response(request.key.as_bytes(), requester), + }, + Err(err) => { + warn!(error = %err, "find-value lookup in the local store failed"); + FindValueResponse::CloserNodes { + nodes: self.closest_for_response(request.key.as_bytes(), requester), + } + } + } + } + + async fn answer_store(&self, request: StoreRecordRequest, sender: &EndpointId) -> bool { + if self.is_shutting_down() { + return false; + } + let key = request.key; + match validate_store(request, &self.config.network_id, now_ms()) { + Ok(record) => { + let artist_id = record.item.id; + let deleted = record.item.deleted; + match self.db.store_dht_record(key, record).await { + Ok(stored) => { + if stored { + info!( + item = %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. + /// + /// On-demand dials are bounded by the (short) `dial_timeout` and feed the + /// dial-failure backoff, so unreachable contacts cannot stall lookups. + async fn ensure_connected(&self, contact: &NodeContact) -> Result { + self.ensure_running()?; + if self.engine.is_connected(contact.peer_id) { + return Ok(contact.peer_id); + } + let ticket: PeerTicket = contact + .ticket + .parse() + .map_err(|err| MusicDhtError::InvalidTicket(format!("{err}")))?; + debug!(peer = %contact.peer_id, "connecting on demand"); + let result = timeout(self.config.dial_timeout, self.engine.connect(ticket)) + .await + .map_err(|_| MusicDhtError::Timeout) + .and_then(|res| res.map_err(Into::into)); + match result { + Ok(peer) => { + self.clear_dial_failures(&peer); + Ok(peer) + } + Err(err) => { + self.note_dial_failure(contact).await; + Err(err) + } + } + } + + /// Forgets the dial-failure history of a peer (it proved reachable). + fn clear_dial_failures(&self, peer: &EndpointId) { + lock(&self.dial_failures).remove(peer); + } + + /// `true` if the contact recently failed to dial and its backoff window + /// has not elapsed yet. Connected peers are never considered backed off. + fn dial_backoff_active(&self, peer: &EndpointId, now_ms: u64) -> bool { + if self.engine.is_connected(*peer) { + return false; + } + match lock(&self.dial_failures).get(peer) { + Some(failure) => { + let backoff = dial_backoff(failure.consecutive).as_millis() as u64; + now_ms < failure.last_attempt_ms.saturating_add(backoff) + } + None => false, + } + } + + /// Records a failed dial; after [`DIAL_FAILURES_BEFORE_EVICT`] failures + /// in a row the contact is dropped from the routing table and the + /// database (gossip re-adds it with a clean slate if it comes back). + async fn note_dial_failure(&self, contact: &NodeContact) { + let consecutive = { + let mut failures = lock(&self.dial_failures); + let failure = failures.entry(contact.peer_id).or_insert(DialFailure { + consecutive: 0, + last_attempt_ms: 0, + }); + failure.consecutive += 1; + failure.last_attempt_ms = now_ms(); + failure.consecutive + }; + if consecutive < DIAL_FAILURES_BEFORE_EVICT { + debug!( + peer = %contact.peer_id, + consecutive, + backoff_s = dial_backoff(consecutive).as_secs(), + "dial failed; backing off" + ); + return; + } + lock(&self.dial_failures).remove(&contact.peer_id); + lock(&self.routing).remove(&contact.peer_id); + if let Err(err) = self.db.delete_known_peer(contact.peer_id).await { + warn!(error = %err, "failed to delete evicted peer from the database"); + } + info!(peer = %contact.peer_id, "evicted unreachable DHT contact"); + } + + /// Sends one request and awaits its response, cleaning up the pending + /// entry on timeout. + async fn request( + &self, + contact: &NodeContact, + request: OutboundRequest, + ) -> Result { + let peer = self.ensure_connected(contact).await?; + let request_id = RequestId::random(); + let receiver = self.pending.register(request_id, peer)?; + // Lookups may cancel this future (early exit); the guard makes sure + // the pending entry never outlives it. + let _cleanup = self.pending.remove_on_drop(request_id); + tracing::trace!(pending = self.pending.len(), peer = %peer, "sending DHT request"); + let message = match request { + OutboundRequest::Ping => MusicDhtMessage::Ping(RequestEnvelope { + request_id, + payload: PingRequest {}, + }), + OutboundRequest::FindNode(payload) => MusicDhtMessage::FindNode(RequestEnvelope { + request_id, + payload, + }), + OutboundRequest::FindValue(payload) => MusicDhtMessage::FindValue(RequestEnvelope { + request_id, + payload, + }), + OutboundRequest::Store(payload) => MusicDhtMessage::StoreRecord(RequestEnvelope { + request_id, + payload: *payload, + }), + }; + self.send_message(peer, &message).await?; + match timeout(self.config.request_timeout, receiver).await { + Ok(Ok(response)) => { + lock(&self.routing).touch(&peer, now_ms()); + Ok(response) + } + Ok(Err(_)) => Err(MusicDhtError::Protocol("response channel closed".into())), + Err(_) => Err(MusicDhtError::Timeout), + } + } + + /// Measures the round-trip time to a known contact and verifies its + /// DHT identity. + pub async fn ping(&self, contact: &NodeContact) -> Result { + let started = Instant::now(); + match self.request(contact, OutboundRequest::Ping).await? { + DhtResponse::Pong(pong) => { + if pong.node_id != NodeId::from_endpoint(&contact.peer_id) { + return Err(MusicDhtError::Protocol( + "pong with a mismatched node id".into(), + )); + } + lock(&self.routing).touch(&contact.peer_id, now_ms()); + Ok(started.elapsed()) + } + _ => Err(MusicDhtError::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 returns as soon as the **first** records arrive — in-flight + /// requests to slower or unreachable peers are cancelled instead of + /// awaited. Contacts in dial backoff are skipped. Never broadcasts: at + /// most [`ALPHA`] requests run concurrently and at most + /// [`MAX_LOOKUP_REQUESTS`] are sent in total, all hard-bounded by the + /// lookup timeout. + pub async fn lookup(&self, target: [u8; 32], find_value: Option) -> LookupOutcome { + let started = Instant::now(); + let deadline = tokio::time::Instant::now() + self.config.lookup_timeout; + let mut candidates: Vec = lock(&self.routing).closest(&target, K); + let mut known: HashSet = + candidates.iter().map(|contact| contact.peer_id).collect(); + let mut queried: HashSet = HashSet::new(); + let mut records: HashMap<(crate::record::ItemId, EndpointId), StoredRecord> = + HashMap::new(); + let mut sent = 0usize; + + debug!(target = %NodeId::from_bytes(target), seeds = candidates.len(), "lookup started"); + + 'rounds: loop { + if tokio::time::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 round_now = now_ms(); + let batch: Vec = candidates + .iter() + .filter(|contact| { + !queried.contains(&contact.peer_id) + && !self.dial_backoff_active(&contact.peer_id, round_now) + }) + .take(ALPHA.min(budget)) + .cloned() + .collect(); + if batch.is_empty() { + break; + } + sent += batch.len(); + for contact in &batch { + queried.insert(contact.peer_id); + } + + // Process responses as they complete: one dead contact must not + // hold back the answers of the live ones. + let mut in_flight: FuturesUnordered<_> = batch + .iter() + .map(|contact| async move { + let request = match find_value { + Some(key) => OutboundRequest::FindValue(FindValueRequest { key }), + None => OutboundRequest::FindNode(FindNodeRequest { + target: NodeId::from_bytes(target), + }), + }; + (contact, self.request(contact, request).await) + }) + .collect(); + while let Ok(next) = tokio::time::timeout_at(deadline, in_flight.next()).await { + let Some((contact, result)) = next else { + break; // The round is complete. + }; + let mut found_records = false; + let nodes = match result { + Ok(DhtResponse::FindNode(response)) => response.nodes, + Ok(DhtResponse::FindValue(FindValueResponse::Records { records: found })) => { + for record in found { + let key = (record.item.id, record.item.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 { + // Dropping `in_flight` cancels the outstanding requests. + break 'rounds; + } + } + if tokio::time::Instant::now() >= deadline { + debug!("lookup deadline reached"); + break; + } + if sent >= MAX_LOOKUP_REQUESTS { + debug!("lookup request budget exhausted"); + break; + } + } + + // The closest set feeds publishes (store targets); contacts that are + // currently backed off would only burn a dial timeout each. + let closest_now = now_ms(); + candidates.retain(|contact| !self.dial_backoff_active(&contact.peer_id, closest_now)); + 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 item 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_item(&self, item: &LibraryItem) -> Result { + self.ensure_running()?; + let ttl = if item.deleted { + TOMBSTONE_TTL + } else { + ACTIVE_RECORD_TTL + }; + let record = StoredRecord { + item: item.clone(), + publisher: self.endpoint_id, + expires_at_ms: now_ms() + ttl.as_millis() as u64, + }; + let keys = item.dht_keys(&self.config.network_id); + let mut remote_nodes: HashSet = HashSet::new(); + let mut local_replica = false; + + for key in &keys { + let outcome = self.lookup(*key.as_bytes(), None).await; + let targets = outcome.closest; + + // The record belongs on this node too if it is among the K + // closest (always true while the network is smaller than K). + let own_distance = distance(self.node_id.as_bytes(), key.as_bytes()); + let self_is_close = targets.len() < K + || targets.last().is_none_or(|farthest| { + own_distance <= distance(farthest.node_id.as_bytes(), key.as_bytes()) + }); + if self_is_close { + match self.db.store_dht_record(*key, record.clone()).await { + Ok(_) => local_replica = true, + Err(err) => warn!(error = %err, "failed to store own replica"), + } + } + + let stores = targets.iter().map(|contact| async { + let result = self + .request( + contact, + OutboundRequest::Store(Box::new(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!( + item = %item.id, + tombstone = item.deleted, + keys = keys.len(), + nodes = remote_nodes.len(), + "published item record" + ); + Ok(PublishStats { + records: 1, + keys: keys.len(), + remote_nodes: remote_nodes.len(), + local_replica, + }) + } + + /// Republishes every local record that is still alive. + pub async fn republish_all(&self) -> Result { + self.ensure_running()?; + let items = self.db.local_items_for_republish(now_ms()).await?; + let mut total = PublishStats::default(); + for item in &items { + let stats = self.publish_item(item).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: &StoredRecord, existing: &StoredRecord) -> bool { + let (c, e) = (&candidate.item, &existing.item); + c.revision > e.revision || (c.revision == e.revision && c.deleted && !e.deleted) +} + +/// Filters an incoming peer-exchange batch: drops our own contact, the +/// sender's contact and duplicate endpoint ids, and enforces the batch cap. +pub(crate) fn sanitize_peer_exchange( + own: EndpointId, + sender: EndpointId, + peers: Vec, +) -> Vec { + let mut seen: HashSet = HashSet::new(); + peers + .into_iter() + .take(MAX_PEER_EXCHANGE_CONTACTS) + .filter(|contact| { + contact.peer_id != own && contact.peer_id != sender && seen.insert(contact.peer_id) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + fn contact(seed: u8) -> NodeContact { + let peer = test_peer(seed); + NodeContact { + node_id: NodeId::from_endpoint(&peer), + peer_id: peer, + ticket: format!("fnet-test-{seed}"), + last_seen_ms: 0, + } + } + + #[test] + fn dial_backoff_doubles_and_is_capped() { + assert_eq!(dial_backoff(1), Duration::from_secs(30)); + assert_eq!(dial_backoff(2), Duration::from_secs(60)); + assert_eq!(dial_backoff(3), Duration::from_secs(120)); + assert_eq!(dial_backoff(5), Duration::from_secs(480)); + // Capped at the maximum from the 6th failure on, even for huge counts. + assert_eq!(dial_backoff(6), DIAL_BACKOFF_MAX); + assert_eq!(dial_backoff(u32::MAX), DIAL_BACKOFF_MAX); + } + + #[test] + fn peer_exchange_drops_duplicates_self_and_sender() { + let own = test_peer(1); + let sender = test_peer(2); + let peers = vec![ + contact(3), + contact(3), // duplicate + contact(1), // ourselves + contact(2), // the sender + contact(4), + ]; + let sanitized = sanitize_peer_exchange(own, sender, peers); + let ids: Vec = sanitized.iter().map(|c| c.peer_id).collect(); + assert_eq!(ids, vec![test_peer(3), test_peer(4)]); + } + + #[test] + fn peer_exchange_is_capped() { + let own = test_peer(1); + let sender = test_peer(2); + let peers: Vec = (10..10 + MAX_PEER_EXCHANGE_CONTACTS as u8 + 8) + .map(contact) + .collect(); + let sanitized = sanitize_peer_exchange(own, sender, peers); + assert_eq!(sanitized.len(), MAX_PEER_EXCHANGE_CONTACTS); + } +} diff --git a/crates/music-dht/src/normalization.rs b/crates/music-dht/src/normalization.rs new file mode 100644 index 0000000..3901b87 --- /dev/null +++ b/crates/music-dht/src/normalization.rs @@ -0,0 +1,87 @@ +//! LibraryItem name normalization and tokenization. + +use unicode_normalization::UnicodeNormalization; + +/// Normalizes an item 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 music_dht::normalize_name; +/// assert_eq!(normalize_name("Massive Attack"), "massive attack"); +/// assert_eq!(normalize_name(" MASSIVE ATTACK "), "massive attack"); +/// assert_eq!(normalize_name("Massive-Attack"), "massive attack"); +/// assert_eq!(normalize_name("Björk"), "björk"); +/// ``` +pub fn normalize_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 music_dht::tokenize; +/// assert_eq!(tokenize("massive attack"), vec!["massive", "attack"]); +/// ``` +pub fn tokenize(normalized: &str) -> Vec { + normalized + .split(' ') + .filter(|token| !token.is_empty()) + .map(str::to_string) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn normalization_is_deterministic() { + let a = normalize_name("Massive Attack"); + let b = normalize_name("Massive Attack"); + assert_eq!(a, b); + assert_eq!(a, "massive attack"); + } + + #[test] + fn normalization_handles_case_and_whitespace() { + assert_eq!(normalize_name(" MASSIVE ATTACK "), "massive attack"); + assert_eq!(normalize_name("Massive-Attack"), "massive attack"); + assert_eq!(normalize_name("Massive___Attack!!!"), "massive attack"); + assert_eq!(normalize_name("Björk"), "björk"); + assert_eq!(normalize_name(" "), ""); + assert_eq!(normalize_name("!!!"), ""); + } + + #[test] + fn normalization_applies_nfkc() { + // U+FF21 FULLWIDTH LATIN CAPITAL LETTER A normalizes to 'A' → 'a'. + assert_eq!(normalize_name("\u{FF21}BBA"), "abba"); + } + + #[test] + fn tokenize_splits_and_skips_empty() { + assert_eq!(tokenize("massive attack"), vec!["massive", "attack"]); + assert_eq!(tokenize(""), Vec::::new()); + assert_eq!(tokenize("solo"), vec!["solo"]); + } +} diff --git a/crates/music-dht/src/record.rs b/crates/music-dht/src/record.rs new file mode 100644 index 0000000..90dd45f --- /dev/null +++ b/crates/music-dht/src/record.rs @@ -0,0 +1,378 @@ +//! LibraryItem 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 crate::error::MusicDhtError; +use crate::normalization::{normalize_name, tokenize}; + +/// Identifier of the peer that owns a record (its `federation-net` endpoint). +pub type PeerId = EndpointId; + +/// Maximum item name length in UTF-8 bytes. +pub const MAX_ITEM_NAME_BYTES: usize = 512; +/// Maximum number of tokens a single item name may produce. +pub const MAX_TOKENS_PER_ITEM: usize = 32; +/// Maximum number of artist names carried by one release/track record. +pub const MAX_ARTISTS_PER_ITEM: usize = 16; +/// 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(()) +} + +/// Kind of a library item. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum ItemKind { + /// A musical artist. + Artist, + /// A release (album, EP, single, ...). + Release, + /// A single track. + Track, +} + +impl ItemKind { + /// Stable lowercase name used in id derivation and APIs. + pub fn as_str(&self) -> &'static str { + match self { + Self::Artist => "artist", + Self::Release => "release", + Self::Track => "track", + } + } +} + +impl fmt::Display for ItemKind { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl FromStr for ItemKind { + type Err = MusicDhtError; + + fn from_str(s: &str) -> Result { + match s { + "artist" => Ok(Self::Artist), + "release" => Ok(Self::Release), + "track" => Ok(Self::Track), + _ => Err(MusicDhtError::InvalidItemName), + } + } +} + +/// Stable identifier of one item record. +/// +/// Derived deterministically from the owning peer, the item kind and an +/// application-chosen local key, so republishing the same library item always +/// yields the same id (letting higher revisions supersede older ones), while +/// two peers publishing the same name still produce distinct records. +#[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] +pub struct ItemId([u8; 32]); + +impl ItemId { + /// Derives an item id: + /// `BLAKE3("music-dht:item:" || owner || kind || ":" || local_key)`. + pub fn derive(owner: &EndpointId, kind: ItemKind, local_key: &str) -> Self { + let mut hasher = blake3::Hasher::new(); + hasher.update(b"music-dht:item:"); + hasher.update(owner.as_bytes()); + hasher.update(kind.as_str().as_bytes()); + hasher.update(b":"); + hasher.update(local_key.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 ItemId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt_hex(&self.0, f) + } +} + +impl fmt::Debug for ItemId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "ItemId(")?; + fmt_hex(&self.0, f)?; + write!(f, ")") + } +} + +impl FromStr for ItemId { + type Err = MusicDhtError; + + fn from_str(s: &str) -> Result { + let bytes = data_encoding::HEXLOWER_PERMISSIVE + .decode(s.trim().as_bytes()) + .map_err(|_| MusicDhtError::ItemNotFound)?; + let bytes: [u8; 32] = bytes.try_into().map_err(|_| MusicDhtError::ItemNotFound)?; + Ok(Self(bytes)) + } +} + +/// One item record as stored by its owner. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct LibraryItem { + /// Stable identifier of this record. + pub id: ItemId, + /// Peer that created (and owns) the record. + pub owner: PeerId, + /// Kind of the item (artist, release or track). + pub kind: ItemKind, + /// Human-readable item name or title. + pub name: String, + /// Normalized form of the name, used for indexing. + pub normalized_name: String, + /// Display names of the item's artists (empty for artist records). + /// Lets a release or track be found by its artist's name. + pub artist_names: Vec, + /// Release/track year, when known. + pub year: Option, + /// Release type (album, ep, single, ...) for release records. + pub release_type: Option, + /// Track duration in seconds for track records. + pub duration_seconds: Option, + /// 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 LibraryItem { + /// Returns the DHT keys this item is published under: the exact key of + /// the name plus one key per unique token of the name **and of every + /// artist name** — so a release or track is also found by searching for + /// its artist. + pub fn dht_keys(&self, network_id: &NetworkId) -> Vec { + let mut keys = vec![DhtKey::exact(network_id, &self.normalized_name)]; + let mut seen = std::collections::HashSet::new(); + for token in self.search_tokens() { + if seen.insert(token.clone()) { + keys.push(DhtKey::token(network_id, &token)); + } + } + keys + } + + /// All tokens this item is searchable by: its own name plus the names of + /// its artists. + pub fn search_tokens(&self) -> Vec { + let mut tokens = tokenize(&self.normalized_name); + for artist in &self.artist_names { + tokens.extend(tokenize(&normalize_name(artist))); + } + tokens + } +} + +/// A replicated DHT entry: an item plus replication metadata. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct StoredRecord { + /// The item record itself. + pub item: LibraryItem, + /// 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 || "item: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"item:exact:"); + hasher.update(normalized_name.as_bytes()); + Self(*hasher.finalize().as_bytes()) + } + + /// Key for single-token lookups: + /// `BLAKE3(NetworkId || "item: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"item: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 item name and returns its normalized form. +pub(crate) fn validate_name(name: &str) -> Result { + if name.len() > MAX_ITEM_NAME_BYTES { + return Err(MusicDhtError::ItemNameTooLong); + } + let normalized = normalize_name(name); + if normalized.is_empty() { + return Err(MusicDhtError::InvalidItemName); + } + if tokenize(&normalized).len() > MAX_TOKENS_PER_ITEM { + return Err(MusicDhtError::ItemNameTooLong); + } + Ok(normalized) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + #[test] + fn item_id_is_stable_per_owner_kind_and_key() { + let key = test_peer(7); + let a = ItemId::derive(&key, ItemKind::Artist, "artist:42"); + let b = ItemId::derive(&key, ItemKind::Artist, "artist:42"); + assert_eq!(a, b); + // Different local key, kind or owner -> different id. + assert_ne!(a, ItemId::derive(&key, ItemKind::Artist, "artist:43")); + assert_ne!(a, ItemId::derive(&key, ItemKind::Release, "artist:42")); + assert_ne!( + a, + ItemId::derive(&test_peer(8), ItemKind::Artist, "artist:42") + ); + } + + #[test] + fn artist_id_hex_round_trip() { + let id = ItemId::from_bytes([0xabu8; 32]); + let parsed: ItemId = 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 item = LibraryItem { + id: ItemId::from_bytes([1u8; 32]), + owner: key, + kind: ItemKind::Artist, + name: "Attack Attack".into(), + normalized_name: "attack attack".into(), + artist_names: Vec::new(), + year: None, + release_type: None, + duration_seconds: None, + revision: 1, + deleted: false, + updated_at_ms: 0, + }; + let net = NetworkId::from_name("test"); + let keys = item.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(MusicDhtError::InvalidItemName) + )); + assert!(matches!( + validate_name(&"x".repeat(MAX_ITEM_NAME_BYTES + 1)), + Err(MusicDhtError::ItemNameTooLong) + )); + let many_tokens = (0..MAX_TOKENS_PER_ITEM + 1) + .map(|i| format!("t{i}")) + .collect::>() + .join(" "); + assert!(matches!( + validate_name(&many_tokens), + Err(MusicDhtError::ItemNameTooLong) + )); + } +} diff --git a/crates/music-dht/src/request.rs b/crates/music-dht/src/request.rs new file mode 100644 index 0000000..83a9d0e --- /dev/null +++ b/crates/music-dht/src/request.rs @@ -0,0 +1,192 @@ +//! 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::{MusicDhtError, Result}; +use crate::message::{ + FindNodeResponse, FindValueResponse, PongResponse, RequestId, StoreRecordResponse, +}; + +/// Maximum number of simultaneously pending requests. +pub const MAX_PENDING_REQUESTS: usize = 1024; + +/// A response payload of any DHT request type. +#[derive(Debug)] +pub(crate) enum DhtResponse { + Pong(PongResponse), + FindNode(FindNodeResponse), + FindValue(FindValueResponse), + Store(StoreRecordResponse), +} + +struct PendingEntry { + /// The peer the response is expected from. + peer: EndpointId, + sender: oneshot::Sender, +} + +/// Correlates responses with awaiting requesters. +/// +/// Entries are removed when the response arrives, and the requester removes +/// its own entry on timeout, so the map cannot grow without bound; a hard cap +/// of [`MAX_PENDING_REQUESTS`] guards against bugs. +#[derive(Default)] +pub(crate) struct PendingRequests { + map: Mutex>, +} + +impl PendingRequests { + /// Registers a new pending request and returns the receiver for its + /// response. + pub fn register( + &self, + request_id: RequestId, + peer: EndpointId, + ) -> Result> { + let mut map = lock(&self.map); + if map.len() >= MAX_PENDING_REQUESTS { + return Err(MusicDhtError::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); + } + + /// Returns a guard that removes the entry when dropped, making a request + /// future safe to cancel (e.g. when a lookup exits early). Removing an + /// already-completed entry is a no-op. + pub fn remove_on_drop(&self, request_id: RequestId) -> PendingCleanup<'_> { + PendingCleanup { + pending: self, + request_id, + } + } + + /// Number of currently pending requests. + pub fn len(&self) -> usize { + lock(&self.map).len() + } +} + +/// Removes a pending entry on drop; see [`PendingRequests::remove_on_drop`]. +pub(crate) struct PendingCleanup<'a> { + pending: &'a PendingRequests, + request_id: RequestId, +} + +impl Drop for PendingCleanup<'_> { + fn drop(&mut self) { + self.pending.remove(&self.request_id); + } +} + +fn lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::routing::NodeId; + + fn test_peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + fn pong() -> DhtResponse { + DhtResponse::Pong(PongResponse { + node_id: NodeId::from_bytes([0u8; 32]), + }) + } + + #[tokio::test] + async fn response_resolves_pending_request() { + let pending = PendingRequests::default(); + let peer = test_peer(1); + let id = RequestId::random(); + let receiver = pending.register(id, peer).expect("register"); + assert!(pending.complete(&id, &peer, pong())); + assert!(receiver.await.is_ok()); + assert_eq!(pending.len(), 0); + } + + #[tokio::test] + async fn response_from_wrong_peer_is_ignored() { + let pending = PendingRequests::default(); + let peer = test_peer(1); + let wrong = test_peer(2); + let id = RequestId::random(); + let _receiver = pending.register(id, peer).expect("register"); + assert!(!pending.complete(&id, &wrong, pong())); + // The entry is still pending for the right peer. + assert_eq!(pending.len(), 1); + } + + #[tokio::test] + async fn entry_is_removed_after_timeout() { + let pending = PendingRequests::default(); + let peer = test_peer(1); + let id = RequestId::random(); + let receiver = pending.register(id, peer).expect("register"); + // Simulate the requester timing out: it removes its own entry. + let result = tokio::time::timeout(std::time::Duration::from_millis(20), receiver).await; + assert!(result.is_err()); + pending.remove(&id); + assert_eq!(pending.len(), 0); + // A late response finds nothing to complete. + assert!(!pending.complete(&id, &peer, pong())); + } + + #[test] + fn pending_map_is_capped() { + let pending = PendingRequests::default(); + let peer = test_peer(1); + let mut receivers = Vec::new(); + for _ in 0..MAX_PENDING_REQUESTS { + receivers.push( + pending + .register(RequestId::random(), peer) + .expect("register"), + ); + } + assert!(pending.register(RequestId::random(), peer).is_err()); + } +} diff --git a/crates/music-dht/src/routing.rs b/crates/music-dht/src/routing.rs new file mode 100644 index 0000000..d871726 --- /dev/null +++ b/crates/music-dht/src/routing.rs @@ -0,0 +1,338 @@ +//! 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("music-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"music-dht:node:"); + hasher.update(endpoint_id.as_bytes()); + Self(*hasher.finalize().as_bytes()) + } + + /// Creates a node id from raw bytes. + pub fn from_bytes(bytes: [u8; 32]) -> Self { + Self(bytes) + } + + /// Returns the raw bytes of this id. + pub fn as_bytes(&self) -> &[u8; 32] { + &self.0 + } +} + +impl fmt::Display for NodeId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + for byte in &self.0 { + write!(f, "{byte:02x}")?; + } + Ok(()) + } +} + +impl fmt::Debug for NodeId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "NodeId({self})") + } +} + +/// XOR distance between two points of the 256-bit key space, compared as +/// unsigned big-endian integers. +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug)] +pub struct Distance([u8; 32]); + +/// Computes the XOR distance between two 256-bit values. +pub fn distance(a: &[u8; 32], b: &[u8; 32]) -> Distance { + let mut out = [0u8; 32]; + for (i, byte) in out.iter_mut().enumerate() { + *byte = a[i] ^ b[i]; + } + Distance(out) +} + +impl Distance { + /// Index of the k-bucket this distance falls into: the position of the + /// highest set bit (0..=255). Returns `None` for a zero distance (self). + pub fn bucket_index(&self) -> Option { + for (i, byte) in self.0.iter().enumerate() { + if *byte != 0 { + return Some(255 - (i * 8 + byte.leading_zeros() as usize)); + } + } + None + } +} + +/// Everything needed to reach another DHT node. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct NodeContact { + /// DHT identifier of the node. + pub node_id: NodeId, + /// Transport identifier of the node. + pub peer_id: EndpointId, + /// `federation-net` ticket used to connect on demand. + pub ticket: String, + /// Unix timestamp (milliseconds) of the last observed activity. + pub last_seen_ms: u64, +} + +/// A simplified Kademlia routing table: 256 k-buckets of up to [`K`] contacts. +pub struct RoutingTable { + own_id: NodeId, + buckets: Vec>, +} + +impl RoutingTable { + /// Creates an empty routing table for the given local node id. + pub fn new(own_id: NodeId) -> Self { + Self { + own_id, + buckets: vec![Vec::new(); 256], + } + } + + /// Returns the local node id. + pub fn own_id(&self) -> NodeId { + self.own_id + } + + /// Inserts or refreshes a contact. Returns `true` if the contact was not + /// known before. + /// + /// A full bucket evicts its least recently seen contact; the PoC skips + /// the classic ping-before-evict procedure. + pub fn upsert(&mut self, contact: NodeContact) -> bool { + let Some(index) = + distance(self.own_id.as_bytes(), contact.node_id.as_bytes()).bucket_index() + else { + // Zero distance: never store ourselves. + return false; + }; + let bucket = &mut self.buckets[index]; + if let Some(existing) = bucket + .iter_mut() + .find(|entry| entry.peer_id == contact.peer_id) + { + existing.node_id = contact.node_id; + existing.ticket = contact.ticket; + existing.last_seen_ms = existing.last_seen_ms.max(contact.last_seen_ms); + return false; + } + if bucket.len() >= K { + // Evict the least recently seen contact. + if let Some((oldest, _)) = bucket + .iter() + .enumerate() + .min_by_key(|(_, entry)| entry.last_seen_ms) + { + bucket.remove(oldest); + } + } + bucket.push(contact); + true + } + + /// Refreshes the `last_seen_ms` of a known peer. Returns `true` if the + /// peer was found. + pub fn touch(&mut self, peer_id: &EndpointId, now_ms: u64) -> bool { + for bucket in &mut self.buckets { + if let Some(entry) = bucket.iter_mut().find(|entry| &entry.peer_id == peer_id) { + entry.last_seen_ms = entry.last_seen_ms.max(now_ms); + return true; + } + } + false + } + + /// Removes a contact by its transport id. Returns `true` if it was known. + pub fn remove(&mut self, peer_id: &EndpointId) -> bool { + for bucket in &mut self.buckets { + if let Some(index) = bucket.iter().position(|entry| &entry.peer_id == peer_id) { + bucket.remove(index); + return true; + } + } + false + } + + /// Looks up a contact by its transport id. + pub fn get(&self, peer_id: &EndpointId) -> Option { + self.buckets + .iter() + .flatten() + .find(|entry| &entry.peer_id == peer_id) + .cloned() + } + + /// Returns up to `count` known contacts closest to `target` by XOR + /// distance. + pub fn closest(&self, target: &[u8; 32], count: usize) -> Vec { + let mut contacts: Vec = self.buckets.iter().flatten().cloned().collect(); + contacts.sort_by_key(|contact| distance(contact.node_id.as_bytes(), target)); + contacts.truncate(count); + contacts + } + + /// Returns all known contacts. + pub fn contacts(&self) -> Vec { + self.buckets.iter().flatten().cloned().collect() + } + + /// Returns the number of known contacts. + pub fn len(&self) -> usize { + self.buckets.iter().map(Vec::len).sum() + } + + /// Returns `true` if no contacts are known. + pub fn is_empty(&self) -> bool { + self.len() == 0 + } +} + +/// Convenience: distance from a node to a DHT key. +pub fn key_distance(node: &NodeId, key: &DhtKey) -> Distance { + distance(node.as_bytes(), key.as_bytes()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_peer(seed: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[seed; 32]).public() + } + + fn contact(seed: u8, last_seen_ms: u64) -> NodeContact { + let peer = test_peer(seed); + NodeContact { + node_id: NodeId::from_endpoint(&peer), + peer_id: peer, + ticket: format!("fnet-test-{seed}"), + last_seen_ms, + } + } + + #[test] + fn node_id_is_deterministic() { + let peer = test_peer(1); + assert_eq!(NodeId::from_endpoint(&peer), NodeId::from_endpoint(&peer)); + assert_ne!( + NodeId::from_endpoint(&peer), + NodeId::from_endpoint(&test_peer(2)) + ); + } + + #[test] + fn xor_distance_properties() { + let a = [0b1010_0000u8; 32]; + let b = [0b0000_0000u8; 32]; + assert_eq!(distance(&a, &a), distance(&b, &b)); + assert_eq!(distance(&a, &b), distance(&b, &a)); + // d(a, a) == 0 and is the smallest possible distance. + assert!(distance(&a, &a) < distance(&a, &b)); + + // Big-endian comparison: a difference in the first byte outweighs + // any difference in later bytes. + let mut c = [0u8; 32]; + c[0] = 1; + let mut d = [0u8; 32]; + d[31] = 0xff; + assert!(distance(&c, &b) > distance(&d, &b)); + } + + #[test] + fn bucket_index_matches_highest_bit() { + let zero = [0u8; 32]; + let mut one = [0u8; 32]; + one[31] = 1; + assert_eq!(distance(&zero, &one).bucket_index(), Some(0)); + let mut top = [0u8; 32]; + top[0] = 0x80; + assert_eq!(distance(&zero, &top).bucket_index(), Some(255)); + assert_eq!(distance(&zero, &zero).bucket_index(), None); + } + + #[test] + fn closest_sorts_by_distance() { + let own = NodeId::from_bytes([0u8; 32]); + let mut table = RoutingTable::new(own); + for seed in 1..=20u8 { + table.upsert(contact(seed, seed as u64)); + } + let target = [0x42u8; 32]; + let closest = table.closest(&target, K); + assert!(closest.len() <= K); + for pair in closest.windows(2) { + assert!( + distance(pair[0].node_id.as_bytes(), &target) + <= distance(pair[1].node_id.as_bytes(), &target) + ); + } + } + + #[test] + fn bucket_is_capped_at_k() { + // All contacts whose distance to `own` shares the same highest bit + // land in one bucket; force that by controlling the node ids. + let own = NodeId::from_bytes([0u8; 32]); + let mut table = RoutingTable::new(own); + for i in 0..(K as u8 + 4) { + let peer = test_peer(i + 1); + let mut id = [0x80u8; 32]; + id[31] = i; + table.upsert(NodeContact { + node_id: NodeId::from_bytes(id), + peer_id: peer, + ticket: String::new(), + last_seen_ms: u64::from(i), + }); + } + assert_eq!(table.len(), K); + // The oldest contacts (smallest last_seen_ms) were evicted. + let contacts = table.contacts(); + assert!(contacts.iter().all(|c| c.last_seen_ms >= 4)); + } + + #[test] + fn remove_deletes_contact() { + let own = NodeId::from_bytes([0u8; 32]); + let mut table = RoutingTable::new(own); + table.upsert(contact(1, 10)); + table.upsert(contact(2, 10)); + assert!(table.remove(&test_peer(1))); + assert!(!table.remove(&test_peer(1))); + assert_eq!(table.len(), 1); + assert!(table.get(&test_peer(1)).is_none()); + } + + #[test] + fn upsert_refreshes_existing_contact() { + let own = NodeId::from_bytes([0u8; 32]); + let mut table = RoutingTable::new(own); + assert!(table.upsert(contact(1, 10))); + assert!(!table.upsert(contact(1, 20))); + assert_eq!(table.len(), 1); + assert_eq!(table.contacts()[0].last_seen_ms, 20); + } +} diff --git a/crates/music-dht/src/service.rs b/crates/music-dht/src/service.rs new file mode 100644 index 0000000..aec69a7 --- /dev/null +++ b/crates/music-dht/src/service.rs @@ -0,0 +1,535 @@ +//! The public service facade: lifecycle, item operations and search. + +use std::collections::{HashMap, HashSet}; +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::MusicDhtConfig; +use crate::database::Database; +use crate::error::{MusicDhtError, Result}; +use crate::node::{Node, record_supersedes}; +use crate::normalization::{normalize_name, tokenize}; +use crate::record::{ + DhtKey, ItemId, ItemKind, LibraryItem, MAX_ARTISTS_PER_ITEM, MAX_ITEM_NAME_BYTES, PeerId, + StoredRecord, now_ms, validate_name, +}; +use crate::routing::{NodeContact, NodeId}; + +/// Fixed schema of the music-dht protocol; peers with a different schema are +/// rejected by `federation-net` during the handshake. +pub const SCHEMA_NAME: &str = "music-dht-poc-v1"; + +/// Capacity of the application event channel. +const EVENT_CHANNEL_CAPACITY: usize = 256; + +/// Events delivered to the application. +#[derive(Debug)] +pub enum MusicDhtEvent { + /// 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 MusicDhtEventReceiver { + rx: mpsc::Receiver, +} + +impl MusicDhtEventReceiver { + /// Receives the next event; `None` after shutdown. + pub async fn recv(&mut self) -> Option { + self.rx.recv().await + } +} + +/// Statistics of one publish or republish operation. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct PublishStats { + /// Number of item 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, +} + +/// Description of one library item to publish, as provided by the +/// application. +#[derive(Debug, Clone, PartialEq)] +pub struct ItemSpec { + /// Application-chosen stable key (e.g. `"track:42"`). The record id is + /// derived from it, so republishing the same key updates the same record. + pub local_key: String, + /// Kind of the item. + pub kind: ItemKind, + /// Display name or title. + pub name: String, + /// Display names of the item's artists (empty for artist records). + pub artist_names: Vec, + /// Release/track year, when known. + pub year: Option, + /// Release type (album, ep, ...) for releases. + pub release_type: Option, + /// Track duration in seconds for tracks. + pub duration_seconds: Option, +} + +/// Result of one [`MusicDhtService::sync_library`] call. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct SyncStats { + /// Items published for the first time (or resurrected). + pub added: usize, + /// Items whose content changed and was republished. + pub updated: usize, + /// Items tombstoned because they left the library. + pub removed: usize, + /// Items already up to date. + pub unchanged: usize, + /// Items skipped or whose publish failed. + pub failed: usize, +} + +/// `true` if the two records describe the same content (revision, deletion +/// state and timestamps are ignored). +fn same_content(a: &LibraryItem, b: &LibraryItem) -> bool { + a.kind == b.kind + && a.name == b.name + && a.artist_names == b.artist_names + && a.year == b.year + && a.release_type == b.release_type + && a.duration_seconds == b.duration_seconds +} + +/// Result of a combined local + network search. +#[derive(Debug)] +pub struct SearchOutcome { + /// Matches from the local `local_items` table. + pub local_results: Vec, + /// Matches found in the DHT (including local replicas), tombstones and + /// duplicates already filtered out. Artists matching every query token + /// come first. + pub network_results: Vec, + /// Number of distinct peers queried during the lookups. + pub queried_nodes: usize, + /// Number of distinct nodes discovered during the lookups. + pub discovered_nodes: usize, + /// Total wall-clock duration of the search. + pub duration: Duration, +} + +/// A distributed item 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 MusicDhtService { + node: Arc, + tasks: std::sync::Mutex>>, +} + +impl MusicDhtService { + /// Starts the service: opens the database, starts the network engine, + /// loads persisted contacts and spawns the maintenance tasks. + pub async fn start(config: MusicDhtConfig) -> Result<(Self, MusicDhtEventReceiver)> { + let mut engine_builder = NetworkConfig::builder() + .data_dir(&config.data_dir) + .network_id(config.network_id) + .schema_id(SchemaId::from_name(SCHEMA_NAME)) + .request_timeout(config.transport_timeout); + if let Some(rendezvous) = config.rendezvous.clone() { + engine_builder = engine_builder.rendezvous(rendezvous); + } + let engine_config = engine_builder + .build() + .map_err(|err| MusicDhtError::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, + "music-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: std::sync::Mutex::new(tasks), + }, + MusicDhtEventReceiver { rx: event_rx }, + )) + } + + /// Returns the transport identifier of this peer. + pub fn endpoint_id(&self) -> EndpointId { + self.node.endpoint_id + } + + /// Returns the DHT identifier of this peer. + pub fn node_id(&self) -> NodeId { + self.node.node_id + } + + /// Creates a shareable ticket for this peer. + pub async fn ticket(&self) -> Result { + self.node.ensure_running()?; + self.node.engine.ticket().await.map_err(Into::into) + } + + /// Connects to another peer by ticket. Contacts are exchanged + /// automatically once the connection is up. + pub async fn connect(&self, ticket: PeerTicket) -> Result { + self.node.ensure_running()?; + let peer = self.node.engine.connect(ticket).await?; + Ok(peer) + } + + /// All DHT contacts currently known to this node. + pub fn known_peers(&self) -> Vec { + self.node.known_contacts() + } + + /// Transport connections that are currently open. + pub fn connected_peers(&self) -> Vec { + self.node.engine.connected_peers() + } + + /// Returns `true` if a transport connection to `peer` is open. + pub fn is_connected(&self, peer: EndpointId) -> bool { + self.node.engine.is_connected(peer) + } + + /// Synchronizes the published library with `specs`: the desired set of + /// items this peer wants to share. + /// + /// Items are matched by their application-chosen `local_key` (the record + /// id is derived from it), so calling this repeatedly is idempotent: + /// new/changed items are (re)published with a bumped revision, items + /// missing from `specs` are tombstoned, unchanged items are left alone. + pub async fn sync_library(&self, specs: Vec) -> Result { + self.node.ensure_running()?; + let owner = self.node.endpoint_id; + let existing: HashMap = self + .node + .db + .list_local_items(true) + .await? + .into_iter() + .map(|item| (item.id, item)) + .collect(); + + let mut stats = SyncStats::default(); + let mut seen: HashSet = HashSet::new(); + let mut to_publish: Vec = Vec::new(); + + for spec in specs { + let name = spec.name.trim().to_string(); + let normalized = match validate_name(&name) { + Ok(normalized) => normalized, + Err(err) => { + tracing::debug!(key = %spec.local_key, error = %err, "skipping item with invalid name"); + stats.failed += 1; + continue; + } + }; + let id = ItemId::derive(&owner, spec.kind, &spec.local_key); + if !seen.insert(id) { + // Duplicate local key in the input; first occurrence wins. + continue; + } + let artist_names: Vec = spec + .artist_names + .into_iter() + .map(|n| n.trim().to_string()) + .filter(|n| !n.is_empty() && n.len() <= MAX_ITEM_NAME_BYTES) + .take(MAX_ARTISTS_PER_ITEM) + .collect(); + let mut item = LibraryItem { + id, + owner, + kind: spec.kind, + name, + normalized_name: normalized, + artist_names, + year: spec.year, + release_type: spec.release_type, + duration_seconds: spec.duration_seconds, + revision: 1, + deleted: false, + updated_at_ms: now_ms(), + }; + match existing.get(&id) { + Some(current) if !current.deleted && same_content(current, &item) => { + stats.unchanged += 1; + } + Some(current) => { + item.revision = current.revision + 1; + self.node.db.upsert_local_item(&item).await?; + if current.deleted { + stats.added += 1; + } else { + stats.updated += 1; + } + to_publish.push(item); + } + None => { + self.node.db.upsert_local_item(&item).await?; + stats.added += 1; + to_publish.push(item); + } + } + } + + // Items that disappeared from the library become tombstones. + for (id, current) in &existing { + if seen.contains(id) || current.deleted { + continue; + } + let mut item = current.clone(); + item.revision += 1; + item.deleted = true; + item.updated_at_ms = now_ms(); + self.node.db.upsert_local_item(&item).await?; + stats.removed += 1; + to_publish.push(item); + } + + for item in &to_publish { + if let Err(err) = self.node.publish_item(item).await { + tracing::warn!(item = %item.id, error = %err, "failed to publish item"); + stats.failed += 1; + } + } + if stats.added + stats.updated + stats.removed > 0 { + info!( + added = stats.added, + updated = stats.updated, + removed = stats.removed, + unchanged = stats.unchanged, + "library sync finished" + ); + } + Ok(stats) + } + + /// Lists all locally owned active items. + pub async fn list_local_items(&self) -> Result> { + self.node.db.list_local_items(false).await + } + + /// Searches only the local database. + pub async fn search_local(&self, query: &str) -> Result> { + let normalized = normalize_name(query); + if normalized.is_empty() { + return Err(MusicDhtError::InvalidItemName); + } + self.node.db.search_local(normalized).await + } + + /// Searches locally and across the DHT. + /// + /// The exact key and every token key are looked up and their results + /// merged, so a query for an artist also returns the artist's releases + /// and tracks. No broadcast is involved: every step is an iterative + /// Kademlia-style lookup. + pub async fn search_network(&self, query: &str) -> Result { + self.node.ensure_running()?; + let started = Instant::now(); + let normalized = normalize_name(query); + if normalized.is_empty() { + return Err(MusicDhtError::InvalidItemName); + } + 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; + // (item id, owner) -> best record seen so far. + let mut merged: HashMap<(ItemId, PeerId), StoredRecord> = HashMap::new(); + fn merge(merged: &mut HashMap<(ItemId, PeerId), StoredRecord>, records: Vec) { + for record in records { + let key = (record.item.id, record.item.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 — always, not only as a fallback. The exact key + // of "massive attack" carries the artist record only; the artist's + // releases and tracks live under the token keys. + for token in &tokens { + let key = DhtKey::token(&network_id, token); + merge( + &mut merged, + self.node.db.dht_records_by_key(key, now_ms()).await?, + ); + let outcome = self.node.lookup(*key.as_bytes(), Some(key)).await; + queried_nodes += outcome.queried; + discovered_nodes = discovered_nodes.max(outcome.discovered); + merge(&mut merged, outcome.records); + } + + // Drop tombstones and expired records, then rank: full-token matches + // first, then alphabetically. + let now = now_ms(); + let mut network_results: Vec = merged + .into_values() + .filter(|record| !record.item.deleted && record.expires_at_ms > now) + .map(|record| record.item) + .collect(); + let matches_all_tokens = |item: &LibraryItem| { + let item_tokens = item.search_tokens(); + tokens + .iter() + .all(|token| item_tokens.iter().any(|t| t == token)) + }; + network_results.sort_by(|a, b| { + matches_all_tokens(b) + .cmp(&matches_all_tokens(a)) + .then_with(|| a.normalized_name.cmp(&b.normalized_name)) + }); + + Ok(SearchOutcome { + local_results, + network_results, + queried_nodes, + discovered_nodes, + duration: started.elapsed(), + }) + } + + /// Pings a known peer: verifies liveness and DHT identity, returns the + /// round-trip time and refreshes the contact. + pub async fn ping(&self, peer: EndpointId) -> Result { + self.node.ensure_running()?; + let contact = self + .node + .known_contacts() + .into_iter() + .find(|contact| contact.peer_id == peer) + .ok_or_else(|| MusicDhtError::Network(format!("unknown peer {peer}")))?; + self.node.ping(&contact).await + } + + /// Republishes all live local records immediately. + pub async fn republish(&self) -> Result { + self.node.republish_all().await + } + + /// Shuts the service down gracefully: stops the maintenance tasks, shuts + /// the network engine down and closes the event channel. Safe to call + /// once from any handle; repeated calls are no-ops. + pub async fn shutdown(&self) -> Result<()> { + info!("music-dht node shutting down"); + self.node.begin_shutdown(); + let tasks: Vec> = { + let mut guard = self + .tasks + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + std::mem::take(&mut *guard) + }; + for task in &tasks { + task.abort(); + } + self.node.engine.clone().shutdown().await?; + for task in tasks { + let _ = task.await; + } + info!("music-dht node shut down"); + Ok(()) + } +} + +async fn republish_timer(node: Arc) { + let mut interval = tokio::time::interval(node.config.republish_interval); + // The first tick fires immediately; skip it, the initial republish is + // triggered by contact discovery instead. + interval.tick().await; + loop { + interval.tick().await; + if node.is_shutting_down() { + break; + } + if node.known_contacts().is_empty() { + continue; + } + if let Err(err) = node.republish_all().await { + tracing::warn!(error = %err, "periodic republish failed"); + } + } +} + +async fn expire_timer(node: Arc) { + let mut interval = tokio::time::interval(node.config.expire_interval); + interval.tick().await; + loop { + interval.tick().await; + if node.is_shutting_down() { + break; + } + node.sweep_expired().await; + } +} diff --git a/crates/music-dht/tests/integration.rs b/crates/music-dht/tests/integration.rs new file mode 100644 index 0000000..6d1cc66 --- /dev/null +++ b/crates/music-dht/tests/integration.rs @@ -0,0 +1,189 @@ +//! Integration test: three DHT nodes in one Tokio runtime synchronizing and +//! searching a small music library (artists, releases, tracks). + +use std::path::Path; +use std::time::Duration; + +use music_dht::{ + EndpointId, ItemKind, ItemSpec, MusicDhtConfig, MusicDhtEventReceiver, MusicDhtService, + NetworkId, +}; + +/// Hard cap on the 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 = (MusicDhtService, MusicDhtEventReceiver); + +async fn start(dir: &Path, network: &str) -> Started { + let config = MusicDhtConfig::builder() + .data_dir(dir) + .network_id(NetworkId::from_name(network)) + .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"); + MusicDhtService::start(config) + .await + .expect("service starts") +} + +fn spec(local_key: &str, kind: ItemKind, name: &str, artists: &[&str]) -> ItemSpec { + ItemSpec { + local_key: local_key.to_string(), + kind, + name: name.to_string(), + artist_names: artists.iter().map(|s| s.to_string()).collect(), + year: Some(1998), + release_type: (kind == ItemKind::Release).then(|| "album".to_string()), + duration_seconds: (kind == ItemKind::Track).then_some(287.0), + } +} + +/// Polls `check` until it returns `Some` or the deadline passes. +async fn wait_for( + what: &str, + deadline: Duration, + mut check: impl AsyncFnMut() -> Option, +) -> T { + let started = std::time::Instant::now(); + loop { + if let Some(value) = check().await { + return value; + } + assert!(started.elapsed() < deadline, "timed out waiting for {what}"); + tokio::time::sleep(Duration::from_millis(300)).await; + } +} + +fn knows_peer(service: &MusicDhtService, peer: EndpointId) -> bool { + service + .known_peers() + .iter() + .any(|contact| contact.peer_id == peer) +} + +/// Searches repeatedly until `accept` returns true for the results, or the +/// deadline passes (the DHT is eventually consistent and a transiently +/// dropped connection can make a single search come up short). +async fn search_until( + what: &str, + service: &MusicDhtService, + query: &str, + accept: impl Fn(&[music_dht::LibraryItem]) -> bool, +) -> Vec { + wait_for(what, Duration::from_secs(45), async || { + let outcome = service.search_network(query).await.expect("search"); + accept(&outcome.network_results).then_some(outcome.network_results) + }) + .await +} + +#[tokio::test] +async fn library_sync_and_distributed_search() { + 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"); + + // Bootstrap: B and C join through A's ticket; C learns about B via + // peer exchange. + let (a, _events_a) = start(dir_a.path(), "test-library").await; + let ticket_a = a.ticket().await.expect("ticket"); + let (b, _events_b) = start(dir_b.path(), "test-library").await; + b.connect(ticket_a.clone()).await.expect("b connects to a"); + let (c, _events_c) = start(dir_c.path(), "test-library").await; + c.connect(ticket_a.clone()).await.expect("c connects to a"); + wait_for( + "C to learn about B via peer exchange", + Duration::from_secs(30), + async || knows_peer(&c, b.endpoint_id()).then_some(()), + ) + .await; + + // B publishes a small library: an artist, their release, one track. + let library = vec![ + spec("artist:1", ItemKind::Artist, "Massive Attack", &[]), + spec( + "release:1", + ItemKind::Release, + "Mezzanine", + &["Massive Attack"], + ), + spec("track:1", ItemKind::Track, "Teardrop", &["Massive Attack"]), + ]; + let stats = b.sync_library(library.clone()).await.expect("sync"); + assert_eq!(stats.added, 3, "all three items are new"); + assert_eq!(stats.failed, 0); + + // A immediately-repeated sync publishes nothing. + let stats = b.sync_library(library.clone()).await.expect("sync"); + assert_eq!(stats.unchanged, 3, "identical sync must be a no-op"); + + // A finds the track by its exact title. + let results = search_until("A to find the track", &a, "teardrop", |items| { + items.len() == 1 + }) + .await; + let track = &results[0]; + assert_eq!(track.kind, ItemKind::Track); + assert_eq!(track.name, "Teardrop"); + assert_eq!(track.artist_names, vec!["Massive Attack".to_string()]); + assert_eq!(track.duration_seconds, Some(287.0)); + assert_eq!(track.owner, b.endpoint_id()); + + // C searches by the artist's name and finds all three kinds: the + // artist itself plus the release and track carrying its name. + search_until("C to find all three kinds", &c, "massive attack", |items| { + let kinds: Vec = items.iter().map(|item| item.kind).collect(); + [ItemKind::Artist, ItemKind::Release, ItemKind::Track] + .iter() + .all(|kind| kinds.contains(kind)) + }) + .await; + + // A content change is republished with a higher revision. + let mut changed = library.clone(); + changed[1].year = Some(1997); + let stats = b.sync_library(changed.clone()).await.expect("sync"); + assert_eq!(stats.updated, 1); + assert_eq!(stats.unchanged, 2); + let results = search_until("A to see the updated release", &a, "mezzanine", |items| { + items.len() == 1 && items[0].year == Some(1997) + }) + .await; + assert!(results[0].revision >= 2); + + // Removing the track from the library tombstones it network-wide. + let mut shrunk = changed.clone(); + shrunk.retain(|spec| spec.local_key != "track:1"); + let stats = b.sync_library(shrunk).await.expect("sync"); + assert_eq!(stats.removed, 1); + search_until("the tombstone to hide the track", &a, "teardrop", |items| { + items.is_empty() + }) + .await; + + // The rest of the library is still reachable — including from + // replicas after the owner shuts down. + b.shutdown().await.expect("shutdown b"); + search_until( + "the release to survive on replicas after the owner left", + &c, + "mezzanine", + |items| items.len() == 1, + ) + .await; + + a.shutdown().await.expect("shutdown a"); + c.shutdown().await.expect("shutdown c"); + }) + .await + .expect("test timed out"); +}