Improved content id search mechanics

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.

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<ItemSpec> = (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");
+}

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();