Added music-dht

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 }

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.

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<RendezvousConfig>,
+}
+
+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<PathBuf>,
+    network_id: Option<NetworkId>,
+    republish_interval: Option<Duration>,
+    expire_interval: Option<Duration>,
+    request_timeout: Option<Duration>,
+    lookup_timeout: Option<Duration>,
+    transport_timeout: Option<Duration>,
+    dial_timeout: Option<Duration>,
+    rendezvous: Option<RendezvousConfig>,
+}
+
+impl MusicDhtConfigBuilder {
+    /// Sets the data directory.
+    pub fn data_dir(mut self, dir: impl Into<PathBuf>) -> Self {
+        self.data_dir = Some(dir.into());
+        self
+    }
+
+    /// Sets the network identifier.
+    pub fn network_id(mut self, network_id: NetworkId) -> Self {
+        self.network_id = Some(network_id);
+        self
+    }
+
+    /// Sets the republish interval.
+    pub fn republish_interval(mut self, interval: Duration) -> Self {
+        self.republish_interval = Some(interval);
+        self
+    }
+
+    /// Sets the expired-record sweep interval.
+    pub fn expire_interval(mut self, interval: Duration) -> Self {
+        self.expire_interval = Some(interval);
+        self
+    }
+
+    /// Sets the timeout of a single DHT request.
+    pub fn request_timeout(mut self, timeout: Duration) -> Self {
+        self.request_timeout = Some(timeout);
+        self
+    }
+
+    /// Sets the timeout of a whole iterative lookup.
+    pub fn lookup_timeout(mut self, timeout: Duration) -> Self {
+        self.lookup_timeout = Some(timeout);
+        self
+    }
+
+    /// Sets the timeout for transport operations (dialing, handshakes).
+    pub fn transport_timeout(mut self, timeout: Duration) -> Self {
+        self.transport_timeout = Some(timeout);
+        self
+    }
+
+    /// 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<MusicDhtConfig> {
+        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()
+        );
+    }
+}

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<Mutex<Connection>>,
+}
+
+impl Database {
+    /// Opens (creating if needed) the database at `path` and applies the
+    /// schema.
+    pub async fn open(path: &Path) -> Result<Self> {
+        let path = path.to_path_buf();
+        let conn = tokio::task::spawn_blocking(move || -> Result<Connection> {
+            let conn = Connection::open(&path).map_err(|err| {
+                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<F, R>(&self, f: F) -> Result<R>
+    where
+        F: FnOnce(&Connection) -> rusqlite::Result<R> + Send + 'static,
+        R: Send + 'static,
+    {
+        let conn = self.conn.clone();
+        tokio::task::spawn_blocking(move || {
+            let guard = conn
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            f(&guard).map_err(|err| 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<Vec<LibraryItem>> {
+        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<Vec<LibraryItem>> {
+        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<Vec<LibraryItem>> {
+        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<bool> {
+        self.call(move |conn| {
+            let existing: Option<(i64, i64, i64)> = conn
+                .query_row(
+                    "SELECT revision, deleted, expires_at_ms FROM dht_records
+                     WHERE dht_key = ?1 AND artist_id = ?2 AND owner_peer_id = ?3",
+                    params![
+                        key.as_bytes().as_slice(),
+                        record.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<Vec<StoredRecord>> {
+        self.call(move |conn| {
+            let mut stmt = conn.prepare(
+                "SELECT payload FROM dht_records
+                 WHERE dht_key = ?1 AND expires_at_ms > ?2
+                 LIMIT ?3",
+            )?;
+            let rows = stmt.query_map(
+                params![
+                    key.as_bytes().as_slice(),
+                    now_ms as i64,
+                    MAX_RECORDS_PER_RESPONSE as i64
+                ],
+                |row| row.get::<_, Vec<u8>>(0),
+            )?;
+            let mut records = Vec::new();
+            for row in rows {
+                let payload = row?;
+                // A payload we cannot decode is skipped, not fatal.
+                if let Ok(record) = postcard::from_bytes::<StoredRecord>(&payload) {
+                    records.push(record);
+                }
+            }
+            Ok(records)
+        })
+        .await
+    }
+
+    /// Deletes expired replicas. Returns the number of removed rows.
+    pub async fn delete_expired_records(&self, now_ms: u64) -> Result<usize> {
+        self.call(move |conn| {
+            conn.execute(
+                "DELETE FROM dht_records WHERE expires_at_ms <= ?1",
+                params![now_ms as i64],
+            )
+        })
+        .await
+    }
+
+    /// Inserts or refreshes a known peer contact.
+    pub async fn upsert_known_peer(&self, contact: &NodeContact) -> Result<()> {
+        let contact = contact.clone();
+        self.call(move |conn| {
+            conn.execute(
+                "INSERT OR REPLACE INTO known_peers (peer_id, node_id, ticket, last_seen_ms)
+                 VALUES (?1, ?2, ?3, ?4)",
+                params![
+                    contact.peer_id.to_string(),
+                    contact.node_id.as_bytes().as_slice(),
+                    contact.ticket,
+                    contact.last_seen_ms as i64,
+                ],
+            )?;
+            Ok(())
+        })
+        .await
+    }
+
+    /// 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<Vec<NodeContact>> {
+        self.call(|conn| {
+            let mut stmt =
+                conn.prepare("SELECT peer_id, node_id, ticket, last_seen_ms FROM known_peers")?;
+            let rows = stmt.query_map([], |row| {
+                Ok((
+                    row.get::<_, String>(0)?,
+                    row.get::<_, Vec<u8>>(1)?,
+                    row.get::<_, String>(2)?,
+                    row.get::<_, i64>(3)?,
+                ))
+            })?;
+            let mut contacts = Vec::new();
+            for row in rows {
+                let (peer_id, node_id, ticket, last_seen_ms) = row?;
+                let Ok(peer_id) = EndpointId::from_str(&peer_id) else {
+                    continue;
+                };
+                let Ok(node_id) = <[u8; 32]>::try_from(node_id.as_slice()) else {
+                    continue;
+                };
+                contacts.push(NodeContact {
+                    node_id: NodeId::from_bytes(node_id),
+                    peer_id,
+                    ticket,
+                    last_seen_ms: last_seen_ms as u64,
+                });
+            }
+            Ok(contacts)
+        })
+        .await
+    }
+}
+
+fn item_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<LibraryItem> {
+    let payload: Vec<u8> = 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]);
+    }
+}

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<StoredRecord, String> {
+    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()
+        );
+    }
+}

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<T, E = MusicDhtError> = std::result::Result<T, E>;
+
+/// All errors that can be returned by the public API of this library.
+#[derive(Debug, thiserror::Error)]
+#[non_exhaustive]
+pub enum 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<federation_net::NetworkError> 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()),
+        }
+    }
+}

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};

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<T> {
+    /// Correlation id; echoed back in the response.
+    pub request_id: RequestId,
+    /// The request itself.
+    pub payload: T,
+}
+
+/// Wraps a response payload with the correlation id of its request.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ResponseEnvelope<T> {
+    /// Correlation id of the request being answered.
+    pub request_id: RequestId,
+    /// The response itself.
+    pub payload: T,
+}
+
+/// Introduction sent right after a connection is established.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Hello {
+    /// DHT identifier of the sender.
+    pub node_id: NodeId,
+    /// Transport identifier of the sender (informational; the authenticated
+    /// id always comes from the connection itself).
+    pub peer_id: EndpointId,
+    /// Ticket other peers can use to reach the sender.
+    pub ticket: String,
+    /// Protocol version of the sender.
+    pub protocol_version: u16,
+}
+
+/// A batch of known contacts, shared after [`Hello`].
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PeerExchange {
+    /// Up to [`MAX_PEER_EXCHANGE_CONTACTS`] contacts.
+    pub peers: Vec<NodeContact>,
+}
+
+/// Liveness probe.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PingRequest {}
+
+/// Reply to [`PingRequest`].
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PongResponse {
+    /// DHT identifier of the responder.
+    pub node_id: NodeId,
+}
+
+/// Asks for the closest known nodes to `target`.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct FindNodeRequest {
+    /// Point of the key space to search around.
+    pub target: NodeId,
+}
+
+/// Reply to [`FindNodeRequest`].
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct FindNodeResponse {
+    /// Up to `K` known nodes closest to the target.
+    pub nodes: Vec<NodeContact>,
+}
+
+/// Asks for records stored under `key`, or the closest nodes to it.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct FindValueRequest {
+    /// The DHT key to look up.
+    pub key: DhtKey,
+}
+
+/// Reply to [`FindValueRequest`].
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum FindValueResponse {
+    /// The responder stores records under the key.
+    Records {
+        /// Up to [`MAX_RECORDS_PER_RESPONSE`] non-expired records.
+        records: Vec<StoredRecord>,
+    },
+    /// The responder has nothing stored; here are closer nodes instead.
+    CloserNodes {
+        /// Up to `K` known nodes closest to the key.
+        nodes: Vec<NodeContact>,
+    },
+}
+
+/// Asks the receiver to store a replica of a record.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct StoreRecordRequest {
+    /// The key the record is published under.
+    pub key: DhtKey,
+    /// The record to store.
+    pub record: 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<PingRequest>),
+    /// Reply to `Ping`.
+    Pong(ResponseEnvelope<PongResponse>),
+
+    /// Node lookup request.
+    FindNode(RequestEnvelope<FindNodeRequest>),
+    /// Reply to `FindNode`.
+    FindNodeResult(ResponseEnvelope<FindNodeResponse>),
+
+    /// Value lookup request.
+    FindValue(RequestEnvelope<FindValueRequest>),
+    /// Reply to `FindValue`.
+    FindValueResult(ResponseEnvelope<FindValueResponse>),
+
+    /// Replication request.
+    StoreRecord(RequestEnvelope<StoreRecordRequest>),
+    /// Reply to `StoreRecord`.
+    StoreRecordResult(ResponseEnvelope<StoreRecordResponse>),
+}

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<T>(mutex: &Mutex<T>) -> 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<StoreRecordRequest>),
+}
+
+/// Result of one iterative lookup.
+pub(crate) struct LookupOutcome {
+    /// Records found (value lookups only).
+    pub records: Vec<StoredRecord>,
+    /// Closest known contacts to the target, best first, at most `K`.
+    pub closest: Vec<NodeContact>,
+    /// Number of distinct peers actually queried.
+    pub queried: usize,
+    /// Number of distinct nodes known to the lookup (seeds + discovered).
+    pub discovered: usize,
+}
+
+/// Shared state of one DHT node.
+pub(crate) struct Node {
+    pub engine: NetworkEngine<MusicDhtMessage>,
+    pub db: Database,
+    pub config: MusicDhtConfig,
+    pub node_id: NodeId,
+    pub endpoint_id: EndpointId,
+    routing: Mutex<RoutingTable>,
+    pending: PendingRequests,
+    /// Peers we already introduced ourselves to (per connection).
+    hello_sent: Mutex<HashSet<EndpointId>>,
+    /// Peers we already gossiped contacts to (per connection).
+    exchange_sent: Mutex<HashSet<EndpointId>>,
+    /// Contacts that recently failed to dial, with their backoff state.
+    dial_failures: Mutex<HashMap<EndpointId, DialFailure>>,
+    events: Mutex<Option<mpsc::Sender<MusicDhtEvent>>>,
+    /// Set once the post-startup republish has been triggered.
+    initial_republish_done: AtomicBool,
+    shutting_down: AtomicBool,
+}
+
+impl Node {
+    pub fn new(
+        engine: NetworkEngine<MusicDhtMessage>,
+        db: Database,
+        config: MusicDhtConfig,
+        events: mpsc::Sender<MusicDhtEvent>,
+    ) -> 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<NodeContact> {
+        lock(&self.routing).contacts()
+    }
+
+    /// Seeds the routing table (used at startup with persisted contacts).
+    pub fn seed_contacts(&self, contacts: Vec<NodeContact>) {
+        let mut routing = lock(&self.routing);
+        for contact in contacts {
+            if contact.peer_id != self.endpoint_id {
+                routing.upsert(contact);
+            }
+        }
+    }
+
+    /// Adds or refreshes a contact learned from the network.
+    ///
+    /// The node id is always re-derived from the endpoint id instead of
+    /// trusting the gossiped value. The first contact ever learned triggers
+    /// the post-startup republish.
+    async fn upsert_contact(self: &Arc<Self>, peer_id: EndpointId, ticket: String) {
+        if peer_id == self.endpoint_id {
+            return;
+        }
+        let contact = NodeContact {
+            node_id: NodeId::from_endpoint(&peer_id),
+            peer_id,
+            ticket,
+            last_seen_ms: now_ms(),
+        };
+        let is_new = lock(&self.routing).upsert(contact.clone());
+        if let Err(err) = self.db.upsert_known_peer(&contact).await {
+            warn!(error = %err, "failed to persist known peer");
+        }
+        if is_new {
+            info!(peer = %contact.peer_id, node = %contact.node_id, "learned new DHT contact");
+            self.emit(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<Self>) {
+        if lock(&self.routing).is_empty() || self.is_shutting_down() {
+            return;
+        }
+        if self.initial_republish_done.swap(true, Ordering::SeqCst) {
+            return;
+        }
+        let node = self.clone();
+        tokio::spawn(async move {
+            match node.republish_all().await {
+                Ok(stats) => info!(
+                    records = stats.records,
+                    keys = stats.keys,
+                    nodes = stats.remote_nodes,
+                    "post-startup republish finished"
+                ),
+                Err(err) => warn!(error = %err, "post-startup republish failed"),
+            }
+        });
+    }
+
+    /// Consumes `federation-net` events until the engine shuts down.
+    pub async fn run_event_loop(
+        self: Arc<Self>,
+        mut receiver: NetworkEventReceiver<MusicDhtMessage>,
+    ) {
+        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<Self>, peer: EndpointId) {
+        // Mark before sending so a crossing Hello does not trigger an echo.
+        if !lock(&self.hello_sent).insert(peer) {
+            return;
+        }
+        let ticket = match self.engine.ticket().await {
+            Ok(ticket) => ticket.to_string(),
+            Err(err) => {
+                warn!(error = %err, "cannot create own ticket for hello");
+                lock(&self.hello_sent).remove(&peer);
+                return;
+            }
+        };
+        let hello = 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<Self>, peer: EndpointId) {
+        if !lock(&self.exchange_sent).insert(peer) {
+            return;
+        }
+        let mut peers: Vec<NodeContact> = self
+            .known_contacts()
+            .into_iter()
+            .filter(|contact| contact.peer_id != peer && contact.peer_id != self.endpoint_id)
+            .collect();
+        // Prefer the most recently seen contacts.
+        peers.sort_by_key(|contact| std::cmp::Reverse(contact.last_seen_ms));
+        peers.truncate(MAX_PEER_EXCHANGE_CONTACTS);
+        if peers.is_empty() {
+            return;
+        }
+        debug!(peer = %peer, count = peers.len(), "sending peer exchange");
+        let message = 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<Self>, 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<Self>, peer: EndpointId, hello: Hello) {
+        if hello.protocol_version != DHT_PROTOCOL_VERSION {
+            warn!(peer = %peer, version = hello.protocol_version, "unsupported DHT protocol version");
+            return;
+        }
+        // The authenticated identity comes from the connection; the id fields
+        // inside the payload must be consistent with it.
+        if hello.peer_id != peer || hello.node_id != NodeId::from_endpoint(&peer) {
+            warn!(peer = %peer, "hello with inconsistent identity; ignoring");
+            self.emit(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<Self>, peer: EndpointId, exchange: PeerExchange) {
+        let contacts = sanitize_peer_exchange(self.endpoint_id, peer, exchange.peers);
+        let accepted = contacts.len();
+        for contact in contacts {
+            self.upsert_contact(contact.peer_id, contact.ticket).await;
+        }
+        debug!(peer = %peer, accepted, "processed peer exchange");
+    }
+
+    /// Contacts for a FindNode/FindValue response: closest to the target,
+    /// excluding the requester itself.
+    fn closest_for_response(&self, target: &[u8; 32], requester: &EndpointId) -> Vec<NodeContact> {
+        lock(&self.routing)
+            .closest(target, K + 1)
+            .into_iter()
+            .filter(|contact| &contact.peer_id != requester)
+            .take(K)
+            .collect()
+    }
+
+    async fn answer_find_value(
+        &self,
+        request: &FindValueRequest,
+        requester: &EndpointId,
+    ) -> FindValueResponse {
+        match self.db.dht_records_by_key(request.key, now_ms()).await {
+            Ok(records) if !records.is_empty() => FindValueResponse::Records { records },
+            Ok(_) => FindValueResponse::CloserNodes {
+                nodes: self.closest_for_response(request.key.as_bytes(), requester),
+            },
+            Err(err) => {
+                warn!(error = %err, "find-value lookup in the local store failed");
+                FindValueResponse::CloserNodes {
+                    nodes: self.closest_for_response(request.key.as_bytes(), requester),
+                }
+            }
+        }
+    }
+
+    async fn answer_store(&self, request: StoreRecordRequest, sender: &EndpointId) -> bool {
+        if self.is_shutting_down() {
+            return false;
+        }
+        let key = request.key;
+        match validate_store(request, &self.config.network_id, now_ms()) {
+            Ok(record) => {
+                let artist_id = record.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<EndpointId> {
+        self.ensure_running()?;
+        if self.engine.is_connected(contact.peer_id) {
+            return Ok(contact.peer_id);
+        }
+        let ticket: PeerTicket = contact
+            .ticket
+            .parse()
+            .map_err(|err| 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<DhtResponse> {
+        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<std::time::Duration> {
+        let started = Instant::now();
+        match self.request(contact, OutboundRequest::Ping).await? {
+            DhtResponse::Pong(pong) => {
+                if pong.node_id != NodeId::from_endpoint(&contact.peer_id) {
+                    return Err(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<DhtKey>) -> LookupOutcome {
+        let started = Instant::now();
+        let deadline = tokio::time::Instant::now() + self.config.lookup_timeout;
+        let mut candidates: Vec<NodeContact> = lock(&self.routing).closest(&target, K);
+        let mut known: HashSet<EndpointId> =
+            candidates.iter().map(|contact| contact.peer_id).collect();
+        let mut queried: HashSet<EndpointId> = HashSet::new();
+        let mut records: HashMap<(crate::record::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<NodeContact> = 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<PublishStats> {
+        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<EndpointId> = HashSet::new();
+        let mut local_replica = false;
+
+        for key in &keys {
+            let outcome = self.lookup(*key.as_bytes(), None).await;
+            let targets = outcome.closest;
+
+            // The record belongs on this node too if it is among the K
+            // closest (always true while the network is smaller than K).
+            let own_distance = distance(self.node_id.as_bytes(), key.as_bytes());
+            let self_is_close = targets.len() < K
+                || targets.last().is_none_or(|farthest| {
+                    own_distance <= distance(farthest.node_id.as_bytes(), key.as_bytes())
+                });
+            if self_is_close {
+                match self.db.store_dht_record(*key, record.clone()).await {
+                    Ok(_) => local_replica = true,
+                    Err(err) => warn!(error = %err, "failed to store own replica"),
+                }
+            }
+
+            let stores = targets.iter().map(|contact| async {
+                let result = self
+                    .request(
+                        contact,
+                        OutboundRequest::Store(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<PublishStats> {
+        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<NodeContact>,
+) -> Vec<NodeContact> {
+    let mut seen: HashSet<EndpointId> = HashSet::new();
+    peers
+        .into_iter()
+        .take(MAX_PEER_EXCHANGE_CONTACTS)
+        .filter(|contact| {
+            contact.peer_id != own && contact.peer_id != sender && seen.insert(contact.peer_id)
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn test_peer(seed: u8) -> EndpointId {
+        iroh::SecretKey::from_bytes(&[seed; 32]).public()
+    }
+
+    fn contact(seed: u8) -> NodeContact {
+        let peer = test_peer(seed);
+        NodeContact {
+            node_id: NodeId::from_endpoint(&peer),
+            peer_id: peer,
+            ticket: format!("fnet-test-{seed}"),
+            last_seen_ms: 0,
+        }
+    }
+
+    #[test]
+    fn 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<EndpointId> = sanitized.iter().map(|c| c.peer_id).collect();
+        assert_eq!(ids, vec![test_peer(3), test_peer(4)]);
+    }
+
+    #[test]
+    fn peer_exchange_is_capped() {
+        let own = test_peer(1);
+        let sender = test_peer(2);
+        let peers: Vec<NodeContact> = (10..10 + MAX_PEER_EXCHANGE_CONTACTS as u8 + 8)
+            .map(contact)
+            .collect();
+        let sanitized = sanitize_peer_exchange(own, sender, peers);
+        assert_eq!(sanitized.len(), MAX_PEER_EXCHANGE_CONTACTS);
+    }
+}

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<String> {
+    normalized
+        .split(' ')
+        .filter(|token| !token.is_empty())
+        .map(str::to_string)
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn normalization_is_deterministic() {
+        let a = normalize_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::<String>::new());
+        assert_eq!(tokenize("solo"), vec!["solo"]);
+    }
+}

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<Self, Self::Err> {
+        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<Self, Self::Err> {
+        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<String>,
+    /// Release/track year, when known.
+    pub year: Option<i32>,
+    /// Release type (album, ep, single, ...) for release records.
+    pub release_type: Option<String>,
+    /// Track duration in seconds for track records.
+    pub duration_seconds: Option<f64>,
+    /// 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<DhtKey> {
+        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<String> {
+        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<String, MusicDhtError> {
+    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::<Vec<_>>()
+            .join(" ");
+        assert!(matches!(
+            validate_name(&many_tokens),
+            Err(MusicDhtError::ItemNameTooLong)
+        ));
+    }
+}

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<DhtResponse>,
+}
+
+/// Correlates responses with awaiting requesters.
+///
+/// Entries are removed when the response arrives, and the requester removes
+/// its own entry on timeout, so the map cannot grow without bound; a hard cap
+/// of [`MAX_PENDING_REQUESTS`] guards against bugs.
+#[derive(Default)]
+pub(crate) struct PendingRequests {
+    map: Mutex<HashMap<RequestId, PendingEntry>>,
+}
+
+impl PendingRequests {
+    /// Registers a new pending request and returns the receiver for its
+    /// response.
+    pub fn register(
+        &self,
+        request_id: RequestId,
+        peer: EndpointId,
+    ) -> Result<oneshot::Receiver<DhtResponse>> {
+        let mut map = lock(&self.map);
+        if map.len() >= MAX_PENDING_REQUESTS {
+            return Err(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<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
+    mutex
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::routing::NodeId;
+
+    fn test_peer(seed: u8) -> EndpointId {
+        iroh::SecretKey::from_bytes(&[seed; 32]).public()
+    }
+
+    fn pong() -> DhtResponse {
+        DhtResponse::Pong(PongResponse {
+            node_id: NodeId::from_bytes([0u8; 32]),
+        })
+    }
+
+    #[tokio::test]
+    async fn response_resolves_pending_request() {
+        let pending = PendingRequests::default();
+        let peer = test_peer(1);
+        let id = RequestId::random();
+        let receiver = pending.register(id, peer).expect("register");
+        assert!(pending.complete(&id, &peer, pong()));
+        assert!(receiver.await.is_ok());
+        assert_eq!(pending.len(), 0);
+    }
+
+    #[tokio::test]
+    async fn response_from_wrong_peer_is_ignored() {
+        let pending = PendingRequests::default();
+        let peer = test_peer(1);
+        let wrong = test_peer(2);
+        let id = RequestId::random();
+        let _receiver = pending.register(id, peer).expect("register");
+        assert!(!pending.complete(&id, &wrong, pong()));
+        // The entry is still pending for the right peer.
+        assert_eq!(pending.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn entry_is_removed_after_timeout() {
+        let pending = PendingRequests::default();
+        let peer = test_peer(1);
+        let id = RequestId::random();
+        let receiver = pending.register(id, peer).expect("register");
+        // Simulate the requester timing out: it removes its own entry.
+        let result = tokio::time::timeout(std::time::Duration::from_millis(20), receiver).await;
+        assert!(result.is_err());
+        pending.remove(&id);
+        assert_eq!(pending.len(), 0);
+        // A late response finds nothing to complete.
+        assert!(!pending.complete(&id, &peer, pong()));
+    }
+
+    #[test]
+    fn pending_map_is_capped() {
+        let pending = PendingRequests::default();
+        let peer = test_peer(1);
+        let mut receivers = Vec::new();
+        for _ in 0..MAX_PENDING_REQUESTS {
+            receivers.push(
+                pending
+                    .register(RequestId::random(), peer)
+                    .expect("register"),
+            );
+        }
+        assert!(pending.register(RequestId::random(), peer).is_err());
+    }
+}

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<usize> {
+        for (i, byte) in self.0.iter().enumerate() {
+            if *byte != 0 {
+                return Some(255 - (i * 8 + byte.leading_zeros() as usize));
+            }
+        }
+        None
+    }
+}
+
+/// Everything needed to reach another DHT node.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct NodeContact {
+    /// DHT identifier of the node.
+    pub node_id: NodeId,
+    /// Transport identifier of the node.
+    pub peer_id: EndpointId,
+    /// `federation-net` ticket used to connect on demand.
+    pub ticket: String,
+    /// Unix timestamp (milliseconds) of the last observed activity.
+    pub last_seen_ms: u64,
+}
+
+/// A simplified Kademlia routing table: 256 k-buckets of up to [`K`] contacts.
+pub struct RoutingTable {
+    own_id: NodeId,
+    buckets: Vec<Vec<NodeContact>>,
+}
+
+impl RoutingTable {
+    /// Creates an empty routing table for the given local node id.
+    pub fn new(own_id: NodeId) -> Self {
+        Self {
+            own_id,
+            buckets: vec![Vec::new(); 256],
+        }
+    }
+
+    /// Returns the local node id.
+    pub fn own_id(&self) -> NodeId {
+        self.own_id
+    }
+
+    /// Inserts or refreshes a contact. Returns `true` if the contact was not
+    /// known before.
+    ///
+    /// A full bucket evicts its least recently seen contact; the PoC skips
+    /// the classic ping-before-evict procedure.
+    pub fn upsert(&mut self, contact: NodeContact) -> bool {
+        let Some(index) =
+            distance(self.own_id.as_bytes(), contact.node_id.as_bytes()).bucket_index()
+        else {
+            // Zero distance: never store ourselves.
+            return false;
+        };
+        let bucket = &mut self.buckets[index];
+        if let Some(existing) = bucket
+            .iter_mut()
+            .find(|entry| entry.peer_id == contact.peer_id)
+        {
+            existing.node_id = contact.node_id;
+            existing.ticket = contact.ticket;
+            existing.last_seen_ms = existing.last_seen_ms.max(contact.last_seen_ms);
+            return false;
+        }
+        if bucket.len() >= K {
+            // Evict the least recently seen contact.
+            if let Some((oldest, _)) = bucket
+                .iter()
+                .enumerate()
+                .min_by_key(|(_, entry)| entry.last_seen_ms)
+            {
+                bucket.remove(oldest);
+            }
+        }
+        bucket.push(contact);
+        true
+    }
+
+    /// Refreshes the `last_seen_ms` of a known peer. Returns `true` if the
+    /// peer was found.
+    pub fn touch(&mut self, peer_id: &EndpointId, now_ms: u64) -> bool {
+        for bucket in &mut self.buckets {
+            if let Some(entry) = bucket.iter_mut().find(|entry| &entry.peer_id == peer_id) {
+                entry.last_seen_ms = entry.last_seen_ms.max(now_ms);
+                return true;
+            }
+        }
+        false
+    }
+
+    /// 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<NodeContact> {
+        self.buckets
+            .iter()
+            .flatten()
+            .find(|entry| &entry.peer_id == peer_id)
+            .cloned()
+    }
+
+    /// Returns up to `count` known contacts closest to `target` by XOR
+    /// distance.
+    pub fn closest(&self, target: &[u8; 32], count: usize) -> Vec<NodeContact> {
+        let mut contacts: Vec<NodeContact> = self.buckets.iter().flatten().cloned().collect();
+        contacts.sort_by_key(|contact| distance(contact.node_id.as_bytes(), target));
+        contacts.truncate(count);
+        contacts
+    }
+
+    /// Returns all known contacts.
+    pub fn contacts(&self) -> Vec<NodeContact> {
+        self.buckets.iter().flatten().cloned().collect()
+    }
+
+    /// Returns the number of known contacts.
+    pub fn len(&self) -> usize {
+        self.buckets.iter().map(Vec::len).sum()
+    }
+
+    /// Returns `true` if no contacts are known.
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+}
+
+/// Convenience: distance from a node to a DHT key.
+pub fn key_distance(node: &NodeId, key: &DhtKey) -> Distance {
+    distance(node.as_bytes(), key.as_bytes())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn test_peer(seed: u8) -> EndpointId {
+        iroh::SecretKey::from_bytes(&[seed; 32]).public()
+    }
+
+    fn contact(seed: u8, last_seen_ms: u64) -> NodeContact {
+        let peer = test_peer(seed);
+        NodeContact {
+            node_id: NodeId::from_endpoint(&peer),
+            peer_id: peer,
+            ticket: format!("fnet-test-{seed}"),
+            last_seen_ms,
+        }
+    }
+
+    #[test]
+    fn node_id_is_deterministic() {
+        let peer = test_peer(1);
+        assert_eq!(NodeId::from_endpoint(&peer), NodeId::from_endpoint(&peer));
+        assert_ne!(
+            NodeId::from_endpoint(&peer),
+            NodeId::from_endpoint(&test_peer(2))
+        );
+    }
+
+    #[test]
+    fn xor_distance_properties() {
+        let a = [0b1010_0000u8; 32];
+        let b = [0b0000_0000u8; 32];
+        assert_eq!(distance(&a, &a), distance(&b, &b));
+        assert_eq!(distance(&a, &b), distance(&b, &a));
+        // d(a, a) == 0 and is the smallest possible distance.
+        assert!(distance(&a, &a) < distance(&a, &b));
+
+        // Big-endian comparison: a difference in the first byte outweighs
+        // any difference in later bytes.
+        let mut c = [0u8; 32];
+        c[0] = 1;
+        let mut d = [0u8; 32];
+        d[31] = 0xff;
+        assert!(distance(&c, &b) > distance(&d, &b));
+    }
+
+    #[test]
+    fn bucket_index_matches_highest_bit() {
+        let zero = [0u8; 32];
+        let mut one = [0u8; 32];
+        one[31] = 1;
+        assert_eq!(distance(&zero, &one).bucket_index(), Some(0));
+        let mut top = [0u8; 32];
+        top[0] = 0x80;
+        assert_eq!(distance(&zero, &top).bucket_index(), Some(255));
+        assert_eq!(distance(&zero, &zero).bucket_index(), None);
+    }
+
+    #[test]
+    fn closest_sorts_by_distance() {
+        let own = NodeId::from_bytes([0u8; 32]);
+        let mut table = RoutingTable::new(own);
+        for seed in 1..=20u8 {
+            table.upsert(contact(seed, seed as u64));
+        }
+        let target = [0x42u8; 32];
+        let closest = table.closest(&target, K);
+        assert!(closest.len() <= K);
+        for pair in closest.windows(2) {
+            assert!(
+                distance(pair[0].node_id.as_bytes(), &target)
+                    <= distance(pair[1].node_id.as_bytes(), &target)
+            );
+        }
+    }
+
+    #[test]
+    fn bucket_is_capped_at_k() {
+        // All contacts whose distance to `own` shares the same highest bit
+        // land in one bucket; force that by controlling the node ids.
+        let own = NodeId::from_bytes([0u8; 32]);
+        let mut table = RoutingTable::new(own);
+        for i in 0..(K as u8 + 4) {
+            let peer = test_peer(i + 1);
+            let mut id = [0x80u8; 32];
+            id[31] = i;
+            table.upsert(NodeContact {
+                node_id: NodeId::from_bytes(id),
+                peer_id: peer,
+                ticket: String::new(),
+                last_seen_ms: u64::from(i),
+            });
+        }
+        assert_eq!(table.len(), K);
+        // The oldest contacts (smallest last_seen_ms) were evicted.
+        let contacts = table.contacts();
+        assert!(contacts.iter().all(|c| c.last_seen_ms >= 4));
+    }
+
+    #[test]
+    fn 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);
+    }
+}

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<MusicDhtEvent>,
+}
+
+impl MusicDhtEventReceiver {
+    /// Receives the next event; `None` after shutdown.
+    pub async fn recv(&mut self) -> Option<MusicDhtEvent> {
+        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<String>,
+    /// Release/track year, when known.
+    pub year: Option<i32>,
+    /// Release type (album, ep, ...) for releases.
+    pub release_type: Option<String>,
+    /// Track duration in seconds for tracks.
+    pub duration_seconds: Option<f64>,
+}
+
+/// 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<LibraryItem>,
+    /// 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<LibraryItem>,
+    /// 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<Node>,
+    tasks: std::sync::Mutex<Vec<JoinHandle<()>>>,
+}
+
+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<PeerTicket> {
+        self.node.ensure_running()?;
+        self.node.engine.ticket().await.map_err(Into::into)
+    }
+
+    /// Connects to another peer by ticket. Contacts are exchanged
+    /// automatically once the connection is up.
+    pub async fn connect(&self, ticket: PeerTicket) -> Result<EndpointId> {
+        self.node.ensure_running()?;
+        let peer = self.node.engine.connect(ticket).await?;
+        Ok(peer)
+    }
+
+    /// All DHT contacts currently known to this node.
+    pub fn known_peers(&self) -> Vec<NodeContact> {
+        self.node.known_contacts()
+    }
+
+    /// Transport connections that are currently open.
+    pub fn connected_peers(&self) -> Vec<EndpointId> {
+        self.node.engine.connected_peers()
+    }
+
+    /// Returns `true` if a transport connection to `peer` is open.
+    pub fn is_connected(&self, peer: EndpointId) -> bool {
+        self.node.engine.is_connected(peer)
+    }
+
+    /// 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<ItemSpec>) -> Result<SyncStats> {
+        self.node.ensure_running()?;
+        let owner = self.node.endpoint_id;
+        let existing: HashMap<ItemId, LibraryItem> = 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<ItemId> = HashSet::new();
+        let mut to_publish: Vec<LibraryItem> = 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<String> = 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<Vec<LibraryItem>> {
+        self.node.db.list_local_items(false).await
+    }
+
+    /// Searches only the local database.
+    pub async fn search_local(&self, query: &str) -> Result<Vec<LibraryItem>> {
+        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<SearchOutcome> {
+        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<StoredRecord>) {
+            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<LibraryItem> = 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<Duration> {
+        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<PublishStats> {
+        self.node.republish_all().await
+    }
+
+    /// Shuts the service down gracefully: stops the maintenance tasks, shuts
+    /// the network engine down and closes the event channel. 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<JoinHandle<()>> = {
+            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<Node>) {
+    let mut interval = tokio::time::interval(node.config.republish_interval);
+    // The first tick fires immediately; skip it, the initial republish is
+    // triggered by contact discovery instead.
+    interval.tick().await;
+    loop {
+        interval.tick().await;
+        if node.is_shutting_down() {
+            break;
+        }
+        if node.known_contacts().is_empty() {
+            continue;
+        }
+        if let Err(err) = node.republish_all().await {
+            tracing::warn!(error = %err, "periodic republish failed");
+        }
+    }
+}
+
+async fn expire_timer(node: Arc<Node>) {
+    let mut interval = tokio::time::interval(node.config.expire_interval);
+    interval.tick().await;
+    loop {
+        interval.tick().await;
+        if node.is_shutting_down() {
+            break;
+        }
+        node.sweep_expired().await;
+    }
+}

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<T>(
+    what: &str,
+    deadline: Duration,
+    mut check: impl AsyncFnMut() -> Option<T>,
+) -> T {
+    let started = std::time::Instant::now();
+    loop {
+        if let Some(value) = check().await {
+            return value;
+        }
+        assert!(started.elapsed() < deadline, "timed out waiting for {what}");
+        tokio::time::sleep(Duration::from_millis(300)).await;
+    }
+}
+
+fn knows_peer(service: &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<music_dht::LibraryItem> {
+    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<ItemKind> = 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");
+}