diff --git a/crates/music-dht/src/lib.rs b/crates/music-dht/src/lib.rs index 06010b8..c72beee 100644 --- a/crates/music-dht/src/lib.rs +++ b/crates/music-dht/src/lib.rs @@ -20,8 +20,10 @@ //! 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). +//! each key. Publishers pick the targets from their routing table and send +//! batched store requests (one pipeline per peer), so even a large library +//! republishes in seconds; searches discover records 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. diff --git a/crates/music-dht/tests/content_id_lookup.rs b/crates/music-dht/tests/content_id_lookup.rs index 234c6eb..535133f 100644 --- a/crates/music-dht/tests/content_id_lookup.rs +++ b/crates/music-dht/tests/content_id_lookup.rs @@ -96,3 +96,70 @@ async fn share_link_content_id_lookup() { .await .expect("test timed out"); } + +/// Regression test for the original share-link failure: with a large +/// library, every track — including the tail of the publish order — must be +/// resolvable by content id. The old per-key-lookup publisher could not keep +/// a big library alive within the record TTL. +#[tokio::test] +async fn large_library_tail_is_resolvable() { + tokio::time::timeout(TEST_TIMEOUT, async { + const TRACKS: usize = 300; + let content_id_of = |index: usize| format!("b3:{index:064x}"); + + let dir_a = tempfile::tempdir().expect("tempdir"); + let dir_b = tempfile::tempdir().expect("tempdir"); + let (a, _events_a) = start(dir_a.path(), "content-id-bulk-test").await; + let ticket_a = a.ticket().await.expect("ticket"); + let (b, _events_b) = start(dir_b.path(), "content-id-bulk-test").await; + b.connect(ticket_a).await.expect("b connects to a"); + + // Crosses both the publish-wave and the store-batch boundaries. + let library: Vec = (0..TRACKS) + .map(|index| ItemSpec { + local_key: format!("track:{index}"), + kind: ItemKind::Track, + name: format!("Track {index}"), + artist_names: vec![format!("Artist {}", index % 7)], + featured_artist_names: Vec::new(), + year: Some(2024), + release_type: Some("album".into()), + release_title: Some(format!("Album {}", index % 23)), + track_number: Some((index % 12) as i32 + 1), + disc_number: None, + duration_seconds: Some(200.0), + content_id: Some(content_id_of(index)), + }) + .collect(); + let stats = b.sync_library(library).await.expect("sync"); + assert_eq!(stats.added, TRACKS); + assert_eq!(stats.failed, 0); + + // The last item of the publish order must be findable, not only the + // head of the library. + for index in [0, TRACKS / 2, TRACKS - 1] { + let wanted = content_id_of(index); + let started = std::time::Instant::now(); + loop { + let outcome = a.search_content_id(&wanted).await.expect("search"); + if outcome + .network_results + .iter() + .any(|item| item.content_id.as_deref() == Some(wanted.as_str())) + { + break; + } + assert!( + started.elapsed() < Duration::from_secs(45), + "timed out resolving track {index} by content id" + ); + tokio::time::sleep(Duration::from_millis(500)).await; + } + } + + a.shutdown().await.expect("shutdown a"); + b.shutdown().await.expect("shutdown b"); + }) + .await + .expect("test timed out"); +} diff --git a/crates/music-dht/tests/integration.rs b/crates/music-dht/tests/integration.rs index 58893ef..aa1e8e0 100644 --- a/crates/music-dht/tests/integration.rs +++ b/crates/music-dht/tests/integration.rs @@ -155,17 +155,25 @@ async fn library_sync_and_distributed_search() { }) .await; - // A content change is republished with a higher revision. + // A content change is republished with a higher revision. The + // "mezzanine" query also returns the track (its release title is + // indexed), so pick the release out of the results. 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) + items + .iter() + .any(|item| item.kind == ItemKind::Release && item.year == Some(1997)) }) .await; - assert!(results[0].revision >= 2); + let release = results + .iter() + .find(|item| item.kind == ItemKind::Release) + .expect("release in results"); + assert!(release.revision >= 2); // Removing the track from the library tombstones it network-wide. let mut shrunk = changed.clone();