From 8ee1db9cf89ea604c6b8a8c3fe089a714c1e321f Mon Sep 17 00:00:00 2001 From: Ultradesu Date: Mon, 20 Jul 2026 18:04:54 +0300 Subject: [PATCH] Extend DHT scheme with content_id --- crates/music-dht/README.md | 9 ++- crates/music-dht/src/database.rs | 1 + crates/music-dht/src/dht.rs | 57 +++++++++++++++- crates/music-dht/src/lib.rs | 7 +- crates/music-dht/src/message.rs | 2 +- crates/music-dht/src/record.rs | 80 ++++++++++++++++++++++ crates/music-dht/src/service.rs | 98 ++++++++++++++++++++++----- crates/music-dht/tests/integration.rs | 3 + 8 files changed, 232 insertions(+), 25 deletions(-) diff --git a/crates/music-dht/README.md b/crates/music-dht/README.md index 63df205..95ae883 100644 --- a/crates/music-dht/README.md +++ b/crates/music-dht/README.md @@ -16,11 +16,13 @@ an application-oriented API. A [`LibraryItem`] carries: `kind` (artist | release | track), `name`, `artist_names` (main artists for releases/tracks), `featured_artist_names` (for track guest appearances), `year`, `release_type`, `release_title`, -`track_number`, `disc_number`, `duration_seconds`, plus ownership and -versioning metadata. Records are published under one exact key (the +`track_number`, `disc_number`, `duration_seconds`, optional track `content_id` +(`b3:<64 hex>`), plus ownership and versioning metadata. Records are +published under one exact key (the normalized name) and one token key per word of the name, every main/featured artist name and the track release title, so searching for an artist also -returns their releases, tracks and guest appearances. +returns their releases, tracks and guest appearances. Track records with a +`content_id` are also published under a content key for exact-audio fallback. ## API @@ -43,6 +45,7 @@ service.sync_library(vec![ track_number: None, disc_number: None, duration_seconds: None, + content_id: None, }, // ... ]).await?; diff --git a/crates/music-dht/src/database.rs b/crates/music-dht/src/database.rs index 8062dcc..222a608 100644 --- a/crates/music-dht/src/database.rs +++ b/crates/music-dht/src/database.rs @@ -485,6 +485,7 @@ mod tests { track_number: None, disc_number: None, duration_seconds: None, + content_id: None, revision, deleted, updated_at_ms: now_ms(), diff --git a/crates/music-dht/src/dht.rs b/crates/music-dht/src/dht.rs index b591392..52cf60d 100644 --- a/crates/music-dht/src/dht.rs +++ b/crates/music-dht/src/dht.rs @@ -5,8 +5,8 @@ use federation_net::NetworkId; use crate::message::StoreRecordRequest; use crate::normalization::{normalize_name, tokenize}; use crate::record::{ - ACTIVE_RECORD_TTL, DhtKey, MAX_ARTISTS_PER_ITEM, MAX_ITEM_NAME_BYTES, MAX_TOKENS_PER_ITEM, - StoredRecord, TOMBSTONE_TTL, + ACTIVE_RECORD_TTL, DhtKey, ItemKind, MAX_ARTISTS_PER_ITEM, MAX_ITEM_NAME_BYTES, + MAX_TOKENS_PER_ITEM, StoredRecord, TOMBSTONE_TTL, normalize_content_id, }; /// Outcome of comparing an incoming record with the stored one. @@ -106,12 +106,24 @@ pub(crate) fn validate_store( return Err("release title has too many tokens".into()); } } + if let Some(content_id) = &item.content_id { + if item.kind != ItemKind::Track { + return Err("content id is only valid for track records".into()); + } + if normalize_content_id(content_id).as_deref() != Some(content_id.as_str()) { + return Err("invalid content id".into()); + } + } let key_matches = request.key == DhtKey::exact(network_id, &item.normalized_name) || item .search_tokens() .iter() - .any(|token| request.key == DhtKey::token(network_id, token)); + .any(|token| request.key == DhtKey::token(network_id, token)) + || item + .content_id + .as_deref() + .is_some_and(|content_id| request.key == DhtKey::content(network_id, content_id)); if !key_matches { return Err("key does not correspond to the record".into()); } @@ -157,6 +169,7 @@ mod tests { track_number: None, disc_number: None, duration_seconds: None, + content_id: None, revision, deleted, updated_at_ms: 0, @@ -183,6 +196,9 @@ mod tests { track_number: None, disc_number: None, duration_seconds: Some(330.0), + content_id: Some( + "b3:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".into(), + ), revision: 1, deleted: false, updated_at_ms: 0, @@ -357,4 +373,39 @@ mod tests { .is_ok() ); } + + #[test] + fn validate_accepts_content_key() { + let net = NetworkId::from_name("test"); + let now = 1_000_000; + let rec = track_with_artist(now + ACTIVE_RECORD_TTL.as_millis() as u64); + let content_id = rec.item.content_id.clone().expect("content id"); + + assert!( + validate_store( + StoreRecordRequest { + key: DhtKey::content(&net, &content_id), + record: rec.clone(), + }, + &net, + now, + ) + .is_ok() + ); + + let mut bad = rec; + bad.item.content_id = + Some("B3:0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF".into()); + assert!( + validate_store( + StoreRecordRequest { + key: DhtKey::content(&net, &content_id), + record: bad, + }, + &net, + now, + ) + .is_err() + ); + } } diff --git a/crates/music-dht/src/lib.rs b/crates/music-dht/src/lib.rs index e68e977..b6fd18f 100644 --- a/crates/music-dht/src/lib.rs +++ b/crates/music-dht/src/lib.rs @@ -16,6 +16,9 @@ //! one exact key for the whole normalized name plus one key per token of //! the name, every main/featured artist name and the track's release title //! (so a track is found by its artists). +//! * Track records may also carry a compact `b3:` content id and are then +//! published under a content key, so applications can find another peer with +//! the exact same audio bytes. //! * 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). @@ -55,6 +58,7 @@ //! track_number: Some(10), //! disc_number: Some(1), //! duration_seconds: Some(330.0), +//! content_id: None, //! }]) //! .await?; //! @@ -99,7 +103,8 @@ pub use message::{ 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, + MAX_CONTENT_ID_BYTES, MAX_ITEM_NAME_BYTES, MAX_TOKENS_PER_ITEM, PeerId, StoredRecord, + TOMBSTONE_TTL, normalize_content_id, }; pub use request::MAX_PENDING_REQUESTS; pub use routing::{ diff --git a/crates/music-dht/src/message.rs b/crates/music-dht/src/message.rs index 1f52c7d..f3cda0c 100644 --- a/crates/music-dht/src/message.rs +++ b/crates/music-dht/src/message.rs @@ -9,7 +9,7 @@ use crate::record::{DhtKey, StoredRecord}; use crate::routing::{NodeContact, NodeId}; /// Version of the music-dht protocol. -pub const DHT_PROTOCOL_VERSION: u16 = 2; +pub const DHT_PROTOCOL_VERSION: u16 = 3; /// Maximum number of contacts in a single [`PeerExchange`]. pub const MAX_PEER_EXCHANGE_CONTACTS: usize = 32; /// Maximum number of records in a single [`FindValueResponse`]. diff --git a/crates/music-dht/src/record.rs b/crates/music-dht/src/record.rs index b019f0f..5c88b85 100644 --- a/crates/music-dht/src/record.rs +++ b/crates/music-dht/src/record.rs @@ -20,6 +20,8 @@ pub const MAX_TOKENS_PER_ITEM: usize = 32; /// Maximum number of artist names carried by one release/track record across /// main and featured roles. pub const MAX_ARTISTS_PER_ITEM: usize = 16; +/// Maximum canonical content id length in UTF-8 bytes. +pub const MAX_CONTENT_ID_BYTES: usize = 96; /// Maximum lifetime of an active DHT record. pub const ACTIVE_RECORD_TTL: Duration = Duration::from_secs(30 * 60); /// Maximum lifetime of a tombstone. @@ -175,6 +177,10 @@ pub struct LibraryItem { pub disc_number: Option, /// Track duration in seconds for track records. pub duration_seconds: Option, + /// Stable audio content id for track records, currently + /// `b3:<64 lowercase hex chars>` (BLAKE3 of the audio file bytes). + #[serde(default)] + pub content_id: Option, /// Monotonically increasing revision; bumped on every change. pub revision: u64, /// `true` if this record is a deletion tombstone. @@ -196,6 +202,11 @@ impl LibraryItem { keys.push(DhtKey::token(network_id, &token)); } } + if self.kind == ItemKind::Track + && let Some(content_id) = &self.content_id + { + keys.push(DhtKey::content(network_id, content_id)); + } keys } @@ -255,6 +266,16 @@ impl DhtKey { Self(*hasher.finalize().as_bytes()) } + /// Key for stable audio content lookups: + /// `BLAKE3(NetworkId || "item:content:" || content_id)`. + pub fn content(network_id: &NetworkId, content_id: &str) -> Self { + let mut hasher = blake3::Hasher::new(); + hasher.update(network_id.as_bytes()); + hasher.update(b"item:content:"); + hasher.update(content_id.as_bytes()); + Self(*hasher.finalize().as_bytes()) + } + /// Creates a key from raw bytes. pub fn from_bytes(bytes: [u8; 32]) -> Self { Self(bytes) @@ -303,6 +324,23 @@ pub(crate) fn validate_name(name: &str) -> Result { Ok(normalized) } +/// Returns a canonical content id accepted by this DHT version. +/// +/// The current format is `b3:<64 lowercase hex chars>`, where the payload is +/// the BLAKE3 hash of the audio file bytes. The short algorithm prefix keeps +/// the DHT key stable if another content-id format is ever added later. +pub fn normalize_content_id(content_id: &str) -> Option { + let value = content_id.trim().to_ascii_lowercase(); + if value.is_empty() || value.len() > MAX_CONTENT_ID_BYTES { + return None; + } + let hash = value.strip_prefix("b3:")?; + if hash.len() != 64 || !hash.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return None; + } + Some(value) +} + #[cfg(test)] mod tests { use super::*; @@ -370,6 +408,7 @@ mod tests { track_number: None, disc_number: None, duration_seconds: None, + content_id: None, revision: 1, deleted: false, updated_at_ms: 0, @@ -399,6 +438,9 @@ mod tests { track_number: Some(3), disc_number: Some(1), duration_seconds: Some(314.0), + content_id: Some( + "b3:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".into(), + ), revision: 1, deleted: false, updated_at_ms: 0, @@ -429,4 +471,42 @@ mod tests { Err(MusicDhtError::ItemNameTooLong) )); } + + #[test] + fn content_id_validation_and_keying() { + let content_id = "B3:0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF"; + let canonical = normalize_content_id(content_id).expect("valid content id"); + assert_eq!( + canonical, + "b3:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + ); + assert!(normalize_content_id("sha256:abc").is_none()); + assert!(normalize_content_id("b3:not-hex").is_none()); + + let owner = test_peer(9); + let item = LibraryItem { + id: ItemId::from_bytes([3u8; 32]), + owner, + kind: ItemKind::Track, + name: "Teardrop".into(), + normalized_name: "teardrop".into(), + artist_names: vec!["Massive Attack".into()], + featured_artist_names: Vec::new(), + year: Some(1998), + release_type: Some("album".into()), + release_title: Some("Mezzanine".into()), + track_number: Some(10), + disc_number: Some(1), + duration_seconds: Some(330.0), + content_id: Some(canonical.clone()), + revision: 1, + deleted: false, + updated_at_ms: 0, + }; + let net = NetworkId::from_name("test"); + assert!( + item.dht_keys(&net) + .contains(&DhtKey::content(&net, &canonical)) + ); + } } diff --git a/crates/music-dht/src/service.rs b/crates/music-dht/src/service.rs index 1503926..f3dd192 100644 --- a/crates/music-dht/src/service.rs +++ b/crates/music-dht/src/service.rs @@ -19,13 +19,13 @@ 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, - MAX_TOKENS_PER_ITEM, PeerId, StoredRecord, now_ms, validate_name, + MAX_TOKENS_PER_ITEM, PeerId, StoredRecord, normalize_content_id, 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-v2"; +pub const SCHEMA_NAME: &str = "music-dht-poc-v3"; /// Capacity of the application event channel. const EVENT_CHANNEL_CAPACITY: usize = 256; @@ -108,6 +108,8 @@ pub struct ItemSpec { pub disc_number: Option, /// Track duration in seconds for tracks. pub duration_seconds: Option, + /// Stable audio content id for tracks (`b3:<64 lowercase hex chars>`). + pub content_id: Option, } fn sanitize_artist_names( @@ -184,6 +186,19 @@ fn same_content(a: &LibraryItem, b: &LibraryItem) -> bool { && a.track_number == b.track_number && a.disc_number == b.disc_number && a.duration_seconds == b.duration_seconds + && a.content_id == b.content_id +} + +fn merge_records(merged: &mut HashMap<(ItemId, PeerId), StoredRecord>, records: Vec) { + for record in records { + let key = (record.item.id, record.item.owner); + match merged.get(&key) { + Some(existing) if !record_supersedes(&record, existing) => {} + _ => { + merged.insert(key, record); + } + } + } } /// Result of a combined local + network search. @@ -427,6 +442,11 @@ impl MusicDhtService { &mut seen_artists, MAX_ARTISTS_PER_ITEM.saturating_sub(artist_names.len()), ); + let content_id = if spec.kind == ItemKind::Track { + spec.content_id.as_deref().and_then(normalize_content_id) + } else { + None + }; let mut item = LibraryItem { id, owner, @@ -441,6 +461,7 @@ impl MusicDhtService { track_number: positive_index(spec.track_number), disc_number: positive_index(spec.disc_number), duration_seconds: spec.duration_seconds.filter(|duration| *duration > 0.0), + content_id, revision: 1, deleted: false, updated_at_ms: now_ms(), @@ -535,21 +556,10 @@ impl MusicDhtService { let mut discovered_nodes = 0usize; // (item id, owner) -> best record seen so far. let mut merged: HashMap<(ItemId, PeerId), StoredRecord> = HashMap::new(); - fn merge(merged: &mut HashMap<(ItemId, PeerId), StoredRecord>, records: Vec) { - for record in records { - let key = (record.item.id, record.item.owner); - match merged.get(&key) { - Some(existing) if !record_supersedes(&record, existing) => {} - _ => { - merged.insert(key, record); - } - } - } - } // Step 1: the exact key — local replicas, then the network. let exact_key = DhtKey::exact(&network_id, &normalized); - merge( + merge_records( &mut merged, self.node.db.dht_records_by_key(exact_key, now_ms()).await?, ); @@ -559,21 +569,21 @@ impl MusicDhtService { .await; queried_nodes += outcome.queried; discovered_nodes = discovered_nodes.max(outcome.discovered); - merge(&mut merged, outcome.records); + merge_records(&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( + merge_records( &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); + merge_records(&mut merged, outcome.records); } // Drop tombstones and expired records, then rank: full-token matches @@ -605,6 +615,60 @@ impl MusicDhtService { }) } + /// Searches locally and across the DHT for the exact same audio content. + /// + /// `content_id` must be the canonical `b3:<64 hex>` identifier generated + /// from the audio file bytes. This lookup is intended for playback + /// fallback: if the originally liked peer is offline, the application can + /// find another peer that published the same track bytes. + pub async fn search_content_id(&self, content_id: &str) -> Result { + self.node.ensure_running()?; + let started = Instant::now(); + let content_id = normalize_content_id(content_id).ok_or(MusicDhtError::InvalidItemName)?; + let network_id = self.node.config.network_id; + + let local_results: Vec = self + .node + .db + .list_local_items(false) + .await? + .into_iter() + .filter(|item| item.content_id.as_deref() == Some(content_id.as_str())) + .collect(); + + let key = DhtKey::content(&network_id, &content_id); + let mut merged: HashMap<(ItemId, PeerId), StoredRecord> = HashMap::new(); + merge_records( + &mut merged, + self.node.db.dht_records_by_key(key, now_ms()).await?, + ); + let outcome = self.node.lookup(*key.as_bytes(), Some(key)).await; + let queried_nodes = outcome.queried; + let discovered_nodes = outcome.discovered; + merge_records(&mut merged, outcome.records); + + let now = now_ms(); + let mut network_results: Vec = merged + .into_values() + .filter(|record| !record.item.deleted && record.expires_at_ms > now) + .filter(|record| record.item.content_id.as_deref() == Some(content_id.as_str())) + .map(|record| record.item) + .collect(); + network_results.sort_by(|a, b| { + a.normalized_name + .cmp(&b.normalized_name) + .then_with(|| a.owner.to_string().cmp(&b.owner.to_string())) + }); + + 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 { diff --git a/crates/music-dht/tests/integration.rs b/crates/music-dht/tests/integration.rs index f696915..58893ef 100644 --- a/crates/music-dht/tests/integration.rs +++ b/crates/music-dht/tests/integration.rs @@ -46,6 +46,9 @@ fn spec(local_key: &str, kind: ItemKind, name: &str, artists: &[&str]) -> ItemSp track_number: (kind == ItemKind::Track).then_some(10), disc_number: (kind == ItemKind::Track).then_some(1), duration_seconds: (kind == ItemKind::Track).then_some(287.0), + content_id: (kind == ItemKind::Track).then(|| { + "b3:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".to_string() + }), } }