From a69e651250eece67ef2a29483cafa7b25cc92909 Mon Sep 17 00:00:00 2001 From: Ultradesu Date: Thu, 23 Jul 2026 15:28:03 +0300 Subject: [PATCH] Improved content id search mechanics --- crates/music-dht/src/lib.rs | 6 +- crates/music-dht/src/message.rs | 34 +- crates/music-dht/src/node.rs | 386 +++++++++++++++----- crates/music-dht/src/record.rs | 11 +- crates/music-dht/src/request.rs | 4 +- crates/music-dht/src/service.rs | 64 ++-- crates/music-dht/tests/content_id_lookup.rs | 98 +++++ 7 files changed, 461 insertions(+), 142 deletions(-) create mode 100644 crates/music-dht/tests/content_id_lookup.rs diff --git a/crates/music-dht/src/lib.rs b/crates/music-dht/src/lib.rs index b6fd18f..06010b8 100644 --- a/crates/music-dht/src/lib.rs +++ b/crates/music-dht/src/lib.rs @@ -96,9 +96,9 @@ pub use dht::{StoreDecision, decide_store}; 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, + Hello, MAX_PEER_EXCHANGE_CONTACTS, MAX_RECORDS_PER_BATCH, MAX_RECORDS_PER_RESPONSE, + MusicDhtMessage, PeerExchange, PingRequest, PongResponse, RequestEnvelope, RequestId, + ResponseEnvelope, StoreBatchRequest, StoreBatchResponse, StoreRecordRequest, }; pub use normalization::{normalize_name, tokenize}; pub use record::{ diff --git a/crates/music-dht/src/message.rs b/crates/music-dht/src/message.rs index f3cda0c..4814804 100644 --- a/crates/music-dht/src/message.rs +++ b/crates/music-dht/src/message.rs @@ -9,11 +9,13 @@ use crate::record::{DhtKey, StoredRecord}; use crate::routing::{NodeContact, NodeId}; /// Version of the music-dht protocol. -pub const DHT_PROTOCOL_VERSION: u16 = 3; +pub const DHT_PROTOCOL_VERSION: u16 = 4; /// 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 = 256; +/// Maximum number of entries in a single [`StoreBatchRequest`]. +pub const MAX_RECORDS_PER_BATCH: usize = 64; /// Correlates a response with its request. #[derive(Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] @@ -127,7 +129,8 @@ pub enum FindValueResponse { }, } -/// Asks the receiver to store a replica of a record. +/// One record replica to store, always carried inside a +/// [`StoreBatchRequest`]. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StoreRecordRequest { /// The key the record is published under. @@ -136,11 +139,22 @@ pub struct StoreRecordRequest { pub record: StoredRecord, } -/// Reply to [`StoreRecordRequest`]. +/// Asks the receiver to store replicas of several records at once. +/// +/// Batching keeps a full-library republish at roughly O(peers) requests +/// instead of O(records × keys): each receiver validates every entry +/// individually, so one bad entry never poisons the rest of the batch. #[derive(Debug, Clone, Serialize, Deserialize)] -pub struct StoreRecordResponse { - /// `true` if the record was accepted and stored (or refreshed). - pub stored: bool, +pub struct StoreBatchRequest { + /// At most [`MAX_RECORDS_PER_BATCH`] entries. + pub entries: Vec, +} + +/// Reply to [`StoreBatchRequest`]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StoreBatchResponse { + /// How many entries were accepted and stored (or refreshed). + pub stored: u32, } /// Every message exchanged between music-dht peers. @@ -167,8 +181,8 @@ pub enum MusicDhtMessage { /// Reply to `FindValue`. FindValueResult(ResponseEnvelope), - /// Replication request. - StoreRecord(RequestEnvelope), - /// Reply to `StoreRecord`. - StoreRecordResult(ResponseEnvelope), + /// Batched replication request. + StoreBatch(RequestEnvelope), + /// Reply to `StoreBatch`. + StoreBatchResult(ResponseEnvelope), } diff --git a/crates/music-dht/src/node.rs b/crates/music-dht/src/node.rs index 0820442..e325982 100644 --- a/crates/music-dht/src/node.rs +++ b/crates/music-dht/src/node.rs @@ -19,8 +19,9 @@ 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, + Hello, MAX_PEER_EXCHANGE_CONTACTS, MAX_RECORDS_PER_BATCH, MusicDhtMessage, PeerExchange, + PingRequest, PongResponse, RequestEnvelope, RequestId, ResponseEnvelope, StoreBatchRequest, + StoreBatchResponse, StoreRecordRequest, }; use crate::record::{ACTIVE_RECORD_TTL, DhtKey, LibraryItem, StoredRecord, TOMBSTONE_TTL, now_ms}; use crate::request::{DhtResponse, PendingRequests}; @@ -59,15 +60,22 @@ enum OutboundRequest { Ping, FindNode(FindNodeRequest), FindValue(FindValueRequest), - Store(Box), + StoreBatch(Box), } +/// Upper bound on the estimated payload bytes of one [`StoreBatchRequest`]. +/// Kept far below federation-net's frame limit so envelope overhead and +/// estimation error never push a frame over it. +const MAX_BATCH_BYTES: usize = 128 * 1024; + +/// Items processed per publish wave; bounds the memory used for per-peer +/// batch construction during a full-library republish. +const PUBLISH_WAVE_ITEMS: usize = 256; + /// Result of one iterative lookup. pub(crate) struct LookupOutcome { /// Records found (value lookups only). pub records: Vec, - /// Closest known contacts to the target, best first, at most `K`. - pub closest: Vec, /// Number of distinct peers actually queried. pub queried: usize, /// Number of distinct nodes known to the lookup (seeds + discovered). @@ -330,11 +338,11 @@ impl Node { }); let _ = self.send_message(peer, &response).await; } - MusicDhtMessage::StoreRecord(env) => { - let stored = self.answer_store(env.payload, &peer).await; - let response = MusicDhtMessage::StoreRecordResult(ResponseEnvelope { + MusicDhtMessage::StoreBatch(env) => { + let payload = self.answer_store_batch(env.payload, &peer).await; + let response = MusicDhtMessage::StoreBatchResult(ResponseEnvelope { request_id: env.request_id, - payload: StoreRecordResponse { stored }, + payload, }); let _ = self.send_message(peer, &response).await; } @@ -350,9 +358,9 @@ impl Node { self.pending .complete(&env.request_id, &peer, DhtResponse::FindValue(env.payload)); } - MusicDhtMessage::StoreRecordResult(env) => { + MusicDhtMessage::StoreBatchResult(env) => { self.pending - .complete(&env.request_id, &peer, DhtResponse::Store(env.payload)); + .complete(&env.request_id, &peer, DhtResponse::StoreBatch(env.payload)); } } } @@ -452,6 +460,30 @@ impl Node { } } + /// Validates and stores every entry of a batch individually: one bad + /// entry never poisons the rest. + async fn answer_store_batch( + &self, + request: StoreBatchRequest, + sender: &EndpointId, + ) -> StoreBatchResponse { + if request.entries.len() > MAX_RECORDS_PER_BATCH { + warn!( + from = %sender, + count = request.entries.len(), + "rejected oversized store batch" + ); + return StoreBatchResponse { stored: 0 }; + } + let mut stored = 0u32; + for entry in request.entries { + if self.answer_store(entry, sender).await { + stored += 1; + } + } + StoreBatchResponse { stored } + } + /// Makes sure a connection to the contact exists, dialing its ticket if /// necessary. The Hello exchange runs asynchronously via the event loop. /// @@ -561,7 +593,7 @@ impl Node { request_id, payload, }), - OutboundRequest::Store(payload) => MusicDhtMessage::StoreRecord(RequestEnvelope { + OutboundRequest::StoreBatch(payload) => MusicDhtMessage::StoreBatch(RequestEnvelope { request_id, payload: *payload, }), @@ -601,13 +633,22 @@ impl Node { /// /// 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 + /// and stops as soon as records arrive. How it stops depends on + /// `drain_round`: `false` returns on the **first** records and cancels + /// the in-flight requests (right when any single replica suffices, e.g. + /// a content-id resolution); `true` awaits the rest of the started round + /// first, so up to [`ALPHA`] peers contribute records (right for name + /// searches, where different peers store different records under the + /// same token key). Contacts in dial backoff are skipped. Never + /// broadcasts: at most [`ALPHA`] requests run concurrently and at most /// [`MAX_LOOKUP_REQUESTS`] are sent in total, all hard-bounded by the /// lookup timeout. - pub async fn lookup(&self, target: [u8; 32], find_value: Option) -> LookupOutcome { + pub async fn lookup( + &self, + target: [u8; 32], + find_value: Option, + drain_round: bool, + ) -> LookupOutcome { let started = Instant::now(); let deadline = tokio::time::Instant::now() + self.config.lookup_timeout; let mut candidates: Vec = lock(&self.routing).closest(&target, K); @@ -698,11 +739,16 @@ impl Node { last_seen_ms: now_ms(), }); } - if find_value.is_some() && found_records { + if find_value.is_some() && found_records && !drain_round { // Dropping `in_flight` cancels the outstanding requests. break 'rounds; } } + if find_value.is_some() && !records.is_empty() { + // Drain mode: the round that produced records has completed; + // no further rounds are needed. + break; + } if tokio::time::Instant::now() >= deadline { debug!("lookup deadline reached"); break; @@ -713,12 +759,6 @@ impl Node { } } - // 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); debug!( queried = queried.len(), discovered = known.len(), @@ -728,84 +768,144 @@ impl Node { ); 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 { + /// Publishes item records (active or tombstones) under all their DHT + /// keys to the closest known nodes. + /// + /// Publish targets come straight from the routing table instead of one + /// iterative network lookup per key: hello/gossip and search lookups keep + /// the table fresh, and while the network is small every peer is in it, + /// so the K closest to any key are exact. On larger networks this is the + /// usual Kademlia approximation — a value lookup still converges onto the + /// same neighborhood through `CloserNodes` hops, and every record also + /// lives on its owner. This turns a full-library republish into + /// O(peers) batched requests instead of O(items × keys) lookups. + pub async fn publish_items(&self, items: &[LibraryItem]) -> Result { self.ensure_running()?; - let ttl = if item.deleted { - TOMBSTONE_TTL - } else { - ACTIVE_RECORD_TTL - }; - let record = StoredRecord { - item: item.clone(), - publisher: self.endpoint_id, - expires_at_ms: now_ms() + ttl.as_millis() as u64, - }; - let keys = item.dht_keys(&self.config.network_id); - let mut remote_nodes: HashSet = HashSet::new(); + let mut total = PublishStats::default(); + for wave in items.chunks(PUBLISH_WAVE_ITEMS) { + let stats = self.publish_wave(wave).await?; + total.records += stats.records; + total.keys += stats.keys; + // remote_nodes counts unique nodes per wave; report the widest + // replication seen across waves. + total.remote_nodes = total.remote_nodes.max(stats.remote_nodes); + total.local_replica |= stats.local_replica; + } + Ok(total) + } + + /// Publishes one bounded wave of items: groups all (key, record) pairs + /// by receiving peer and sends them as [`StoreBatchRequest`]s, one peer + /// pipeline at a time per peer, all peers concurrently. + async fn publish_wave(&self, items: &[LibraryItem]) -> Result { + let now = now_ms(); + let contacts: Vec = self + .known_contacts() + .into_iter() + .filter(|contact| !self.dial_backoff_active(&contact.peer_id, now)) + .collect(); + + let mut keys_total = 0usize; let mut local_replica = false; + // Batches under construction, keyed by the index into `contacts`. + let mut per_peer: HashMap> = HashMap::new(); + // K-closest target sets are memoized per key: token keys repeat + // heavily across the items of one artist or release. + let mut targets_memo: HashMap> = HashMap::new(); - for key in &keys { - let outcome = self.lookup(*key.as_bytes(), None).await; - let targets = outcome.closest; + for item in items { + 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 + ttl.as_millis() as u64, + }; + let keys = item.dht_keys(&self.config.network_id); + keys_total += keys.len(); + for key in keys { + let targets = targets_memo + .entry(key) + .or_insert_with(|| closest_contact_indices(&contacts, &key, K)); - // 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"), + // 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(contacts[*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"), + for index in targets.iter() { + per_peer.entry(*index).or_default().push(StoreRecordRequest { + key, + record: record.clone(), + }); } } } + + // One concurrent pipeline per peer; batches within a pipeline are + // sequential so a slow peer only throttles itself. + let sends = per_peer.into_iter().map(|(index, entries)| { + let contact = &contacts[index]; + async move { + let mut accepted_any = false; + for entries in chunk_store_batches(entries) { + let sent = entries.len(); + let request = + OutboundRequest::StoreBatch(Box::new(StoreBatchRequest { entries })); + match self.request(contact, request).await { + Ok(DhtResponse::StoreBatch(StoreBatchResponse { stored })) => { + accepted_any |= stored > 0; + if (stored as usize) < sent { + debug!( + peer = %contact.peer_id, + sent, + stored, + "peer declined part of the store batch" + ); + } + } + Ok(_) => {} + Err(err) => { + debug!(peer = %contact.peer_id, error = %err, "store batch failed"); + } + } + } + (contact.peer_id, accepted_any) + } + }); + let mut remote_nodes: HashSet = HashSet::new(); + for (peer, accepted) in join_all(sends).await { + if accepted { + remote_nodes.insert(peer); + } + } debug!( - item = %item.id, - tombstone = item.deleted, - keys = keys.len(), + items = items.len(), + keys = keys_total, nodes = remote_nodes.len(), - "published item record" + "published item wave" ); Ok(PublishStats { - records: 1, - keys: keys.len(), + records: items.len(), + keys: keys_total, remote_nodes: remote_nodes.len(), local_replica, }) @@ -815,16 +915,7 @@ impl Node { pub async fn republish_all(&self) -> Result { self.ensure_running()?; let items = self.db.local_items_for_republish(now_ms()).await?; - let mut total = PublishStats::default(); - for item in &items { - let stats = self.publish_item(item).await?; - total.records += 1; - total.keys += stats.keys; - // remote_nodes counts unique nodes per record; report the widest - // replication seen across records. - total.remote_nodes = total.remote_nodes.max(stats.remote_nodes); - total.local_replica |= stats.local_replica; - } + let total = self.publish_items(&items).await?; info!( records = total.records, keys = total.keys, @@ -843,6 +934,40 @@ impl Node { } } +/// Indices of the up-to-`count` contacts closest to `key` by XOR distance. +fn closest_contact_indices(contacts: &[NodeContact], key: &DhtKey, count: usize) -> Vec { + let mut indices: Vec = (0..contacts.len()).collect(); + indices.sort_by_key(|index| distance(contacts[*index].node_id.as_bytes(), key.as_bytes())); + indices.truncate(count); + indices +} + +/// Splits store entries into batches respecting both the entry-count and the +/// estimated byte limits, so a batch always fits one transport frame. +fn chunk_store_batches(entries: Vec) -> Vec> { + let mut batches = Vec::new(); + let mut current: Vec = Vec::new(); + let mut current_bytes = 0usize; + for entry in entries { + let bytes = postcard::to_stdvec(&entry) + .map(|encoded| encoded.len()) + .unwrap_or(MAX_BATCH_BYTES); + if !current.is_empty() + && (current.len() >= MAX_RECORDS_PER_BATCH + || current_bytes.saturating_add(bytes) > MAX_BATCH_BYTES) + { + batches.push(std::mem::take(&mut current)); + current_bytes = 0; + } + current_bytes += bytes; + current.push(entry); + } + if !current.is_empty() { + batches.push(current); + } + batches +} + /// `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); @@ -884,6 +1009,79 @@ mod tests { } } + fn store_entry(name: &str) -> StoreRecordRequest { + let owner = test_peer(1); + let normalized = crate::normalization::normalize_name(name); + StoreRecordRequest { + key: DhtKey::exact(&federation_net::NetworkId::from_name("test"), &normalized), + record: StoredRecord { + item: LibraryItem { + id: crate::record::ItemId::from_bytes([7u8; 32]), + owner, + kind: crate::record::ItemKind::Artist, + name: name.to_string(), + normalized_name: normalized, + artist_names: Vec::new(), + featured_artist_names: Vec::new(), + year: None, + release_type: None, + release_title: None, + track_number: None, + disc_number: None, + duration_seconds: None, + content_id: None, + revision: 1, + deleted: false, + updated_at_ms: 0, + }, + publisher: owner, + expires_at_ms: 1000, + }, + } + } + + #[test] + fn store_batches_respect_count_and_byte_limits() { + // Count limit: MAX_RECORDS_PER_BATCH small entries per batch. + let entries: Vec<_> = (0..MAX_RECORDS_PER_BATCH + 1) + .map(|_| store_entry("Massive Attack")) + .collect(); + let batches = chunk_store_batches(entries); + assert_eq!(batches.len(), 2); + assert_eq!(batches[0].len(), MAX_RECORDS_PER_BATCH); + assert_eq!(batches[1].len(), 1); + + // Byte limit: huge entries split well before the count limit. + let huge = "x".repeat(crate::record::MAX_ITEM_NAME_BYTES); + let entries: Vec<_> = (0..MAX_RECORDS_PER_BATCH).map(|_| store_entry(&huge)).collect(); + for batch in chunk_store_batches(entries) { + assert!(!batch.is_empty()); + let bytes: usize = batch + .iter() + .map(|entry| postcard::to_stdvec(entry).expect("encodes").len()) + .sum(); + assert!(bytes <= MAX_BATCH_BYTES); + } + + assert!(chunk_store_batches(Vec::new()).is_empty()); + } + + #[test] + fn closest_indices_sort_by_key_distance() { + let contacts: Vec = (1..=12u8).map(contact).collect(); + let key = DhtKey::from_bytes([0x42u8; 32]); + let indices = closest_contact_indices(&contacts, &key, K); + assert_eq!(indices.len(), K); + for pair in indices.windows(2) { + assert!( + distance(contacts[pair[0]].node_id.as_bytes(), key.as_bytes()) + <= distance(contacts[pair[1]].node_id.as_bytes(), key.as_bytes()) + ); + } + // Fewer contacts than requested: everyone is a target. + assert_eq!(closest_contact_indices(&contacts[..3], &key, K).len(), 3); + } + #[test] fn dial_backoff_doubles_and_is_capped() { assert_eq!(dial_backoff(1), Duration::from_secs(30)); diff --git a/crates/music-dht/src/record.rs b/crates/music-dht/src/record.rs index 5c88b85..76c994b 100644 --- a/crates/music-dht/src/record.rs +++ b/crates/music-dht/src/record.rs @@ -22,10 +22,13 @@ pub const MAX_TOKENS_PER_ITEM: usize = 32; 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. -pub const TOMBSTONE_TTL: Duration = Duration::from_secs(2 * 60 * 60); +/// Maximum lifetime of an active DHT record. Long enough that replicas — and +/// the share links resolved through them — survive the owner going offline +/// for a workday; the owner refreshes the expiry on every republish. +pub const ACTIVE_RECORD_TTL: Duration = Duration::from_secs(12 * 60 * 60); +/// Maximum lifetime of a tombstone. Must exceed [`ACTIVE_RECORD_TTL`] so a +/// stale active replica can never outlive the tombstone that deletes it. +pub const TOMBSTONE_TTL: Duration = Duration::from_secs(24 * 60 * 60); fn fmt_hex(bytes: &[u8; 32], f: &mut fmt::Formatter<'_>) -> fmt::Result { for byte in bytes { diff --git a/crates/music-dht/src/request.rs b/crates/music-dht/src/request.rs index 83a9d0e..29d92da 100644 --- a/crates/music-dht/src/request.rs +++ b/crates/music-dht/src/request.rs @@ -8,7 +8,7 @@ use tokio::sync::oneshot; use crate::error::{MusicDhtError, Result}; use crate::message::{ - FindNodeResponse, FindValueResponse, PongResponse, RequestId, StoreRecordResponse, + FindNodeResponse, FindValueResponse, PongResponse, RequestId, StoreBatchResponse, }; /// Maximum number of simultaneously pending requests. @@ -20,7 +20,7 @@ pub(crate) enum DhtResponse { Pong(PongResponse), FindNode(FindNodeResponse), FindValue(FindValueResponse), - Store(StoreRecordResponse), + StoreBatch(StoreBatchResponse), } struct PendingEntry { diff --git a/crates/music-dht/src/service.rs b/crates/music-dht/src/service.rs index d7a3f58..9c28f58 100644 --- a/crates/music-dht/src/service.rs +++ b/crates/music-dht/src/service.rs @@ -502,11 +502,13 @@ impl MusicDhtService { 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 let Err(err) = self.node.publish_items(&to_publish).await { + tracing::warn!( + items = to_publish.len(), + error = %err, + "failed to publish changed items" + ); + stats.failed += to_publish.len(); } if stats.added + stats.updated + stats.removed > 0 { info!( @@ -558,35 +560,38 @@ impl MusicDhtService { let local_results = self.node.db.search_local(normalized.clone()).await?; - let mut queried_nodes = 0usize; - let mut discovered_nodes = 0usize; + // The exact key plus one key per unique token — 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. + let mut keys = vec![DhtKey::exact(&network_id, &normalized)]; + let mut seen_keys: HashSet = keys.iter().copied().collect(); + for token in &tokens { + let key = DhtKey::token(&network_id, token); + if seen_keys.insert(key) { + keys.push(key); + } + } + // (item id, owner) -> best record seen so far. let mut merged: HashMap<(ItemId, PeerId), StoredRecord> = HashMap::new(); - // Step 1: the exact key — local replicas, then the network. - let exact_key = DhtKey::exact(&network_id, &normalized); - merge_records( - &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_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); + // Local replicas first, then all network lookups concurrently: one + // slow key must not serialize the others. + for key in &keys { merge_records( &mut merged, - self.node.db.dht_records_by_key(key, now_ms()).await?, + self.node.db.dht_records_by_key(*key, now_ms()).await?, ); - let outcome = self.node.lookup(*key.as_bytes(), Some(key)).await; + } + let outcomes = futures::future::join_all( + keys.iter() + .map(|key| self.node.lookup(*key.as_bytes(), Some(*key), true)), + ) + .await; + let mut queried_nodes = 0usize; + let mut discovered_nodes = 0usize; + for outcome in outcomes { queried_nodes += outcome.queried; discovered_nodes = discovered_nodes.max(outcome.discovered); merge_records(&mut merged, outcome.records); @@ -648,7 +653,8 @@ impl MusicDhtService { &mut merged, self.node.db.dht_records_by_key(key, now_ms()).await?, ); - let outcome = self.node.lookup(*key.as_bytes(), Some(key)).await; + // Any single replica resolves a content id, so the first records win. + let outcome = self.node.lookup(*key.as_bytes(), Some(key), false).await; let queried_nodes = outcome.queried; let discovered_nodes = outcome.discovered; merge_records(&mut merged, outcome.records); diff --git a/crates/music-dht/tests/content_id_lookup.rs b/crates/music-dht/tests/content_id_lookup.rs new file mode 100644 index 0000000..234c6eb --- /dev/null +++ b/crates/music-dht/tests/content_id_lookup.rs @@ -0,0 +1,98 @@ +//! Focused integration test: a track published by one node must be +//! resolvable by content id from another node (the share-link flow). + +use std::path::Path; +use std::time::Duration; + +use music_dht::{ + ItemKind, ItemSpec, MusicDhtConfig, MusicDhtEventReceiver, MusicDhtService, NetworkId, +}; + +const TEST_TIMEOUT: Duration = Duration::from_secs(240); + +const CONTENT_ID: &str = "b3:661eb31d76ab7cef89a19bb8e978c0eb357ae62395a73fa5539a0efd35110dd9"; + +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") +} + +#[tokio::test] +async fn share_link_content_id_lookup() { + tokio::time::timeout(TEST_TIMEOUT, async { + let dir_a = tempfile::tempdir().expect("tempdir"); + let dir_b = tempfile::tempdir().expect("tempdir"); + + let (a, _events_a) = start(dir_a.path(), "content-id-test").await; + let ticket_a = a.ticket().await.expect("ticket"); + let (b, _events_b) = start(dir_b.path(), "content-id-test").await; + b.connect(ticket_a).await.expect("b connects to a"); + + // B publishes one track with a content id and many featured artists + // (mirrors the real-world failing share link). + let spec = ItemSpec { + local_key: "track:1".into(), + kind: ItemKind::Track, + name: "Ежемесячные".into(), + artist_names: vec!["Основной Артист".into()], + featured_artist_names: vec![ + "Pyrokinesis".into(), + "Блёв МС".into(), + "Артем Татищевский".into(), + "WormGanger".into(), + "Рудбой".into(), + "АнальгиН-56 школа".into(), + ], + year: Some(2024), + release_type: Some("album".into()), + release_title: Some("Тестовый релиз".into()), + track_number: Some(1), + disc_number: Some(1), + duration_seconds: Some(200.0), + content_id: Some(CONTENT_ID.into()), + }; + let stats = b.sync_library(vec![spec]).await.expect("sync"); + assert_eq!(stats.added, 1, "track is published"); + assert_eq!(stats.failed, 0); + + // A resolves the share link by content id. + let started = std::time::Instant::now(); + let found = loop { + let outcome = a + .search_content_id(CONTENT_ID) + .await + .expect("content id search"); + let hit = outcome + .network_results + .iter() + .find(|item| item.kind == ItemKind::Track); + if let Some(hit) = hit { + break hit.clone(); + } + assert!( + started.elapsed() < Duration::from_secs(45), + "timed out resolving the content id from the other node" + ); + tokio::time::sleep(Duration::from_millis(500)).await; + }; + assert_eq!(found.content_id.as_deref(), Some(CONTENT_ID)); + assert_eq!(found.owner, b.endpoint_id()); + + a.shutdown().await.expect("shutdown a"); + b.shutdown().await.expect("shutdown b"); + }) + .await + .expect("test timed out"); +}