Furumi/frid
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Extend DHT scheme

Browse Source
This commit is contained in:
Ultradesu
2026-07-20 16:28:45 +03:00
parent fdbc0e2ad0
commit a897737978
8 changed files with 234 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/music-dht/README.md
+11 -5
View File
@@ -14,11 +14,13 @@ 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.
`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
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.
## API
@@ -34,8 +36,12 @@ service.sync_library(vec![
kind: ItemKind::Artist,
name: "Massive Attack".into(),
artist_names: vec![],
featured_artist_names: vec![],
year: None,
release_type: None,
release_title: None,
track_number: None,
disc_number: None,
duration_seconds: None,
},
// ...
crates/music-dht/src/database.rs
+40 -13
View File
@@ -210,10 +210,13 @@ impl Database {
"SELECT payload
FROM local_items ORDER BY normalized_name",
)?;
let rows = stmt.query_map([], item_from_row)?;
let rows = stmt.query_map([], |row| row.get::<_, Vec<u8>>(0))?;
let mut result = Vec::new();
for row in rows {
let item = row?;
let payload = row?;
let Ok(item) = postcard::from_bytes::<LibraryItem>(&payload) else {
continue;
};
if include_deleted || !item.deleted {
result.push(item);
}
@@ -458,17 +461,6 @@ impl MusicDhtStorage for Database {
}
}
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::*;
@@ -486,8 +478,12 @@ mod tests {
name: name.to_string(),
normalized_name: crate::normalization::normalize_name(name),
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,
revision,
deleted,
@@ -522,6 +518,37 @@ mod tests {
assert!(none.is_empty());
}
#[tokio::test]
async fn undecodable_local_item_payload_is_skipped() {
let (_dir, db) = open_temp().await;
db.call(|conn| {
conn.execute(
"INSERT INTO local_items
(id, normalized_name, revision, deleted, updated_at_ms, payload)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![
&[1u8; 32],
"legacy",
1_i64,
0_i64,
0_i64,
vec![0xff_u8, 0xff],
],
)?;
Ok(())
})
.await
.expect("insert corrupt row");
assert!(db.list_local_items(true).await.expect("list").is_empty());
assert!(
db.search_local("legacy".into())
.await
.expect("search")
.is_empty()
);
}
#[tokio::test]
async fn local_search_uses_artist_names() {
let (_dir, db) = open_temp().await;
crates/music-dht/src/dht.rs
+22 -2
View File
@@ -83,10 +83,14 @@ pub(crate) fn validate_store(
if tokens.len() > MAX_TOKENS_PER_ITEM {
return Err("too many tokens".into());
}
if item.artist_names.len() > MAX_ARTISTS_PER_ITEM {
if item.artist_names.len() + item.featured_artist_names.len() > MAX_ARTISTS_PER_ITEM {
return Err("too many artists".into());
}
for artist in &item.artist_names {
for artist in item
.artist_names
.iter()
.chain(item.featured_artist_names.iter())
{
if artist.len() > MAX_ITEM_NAME_BYTES {
return Err("artist name too long".into());
}
@@ -94,6 +98,14 @@ pub(crate) fn validate_store(
return Err("artist name has too many tokens".into());
}
}
if let Some(release_title) = &item.release_title {
if release_title.len() > MAX_ITEM_NAME_BYTES {
return Err("release title too long".into());
}
if tokenize(&normalize_name(release_title)).len() > MAX_TOKENS_PER_ITEM {
return Err("release title has too many tokens".into());
}
}
let key_matches = request.key == DhtKey::exact(network_id, &item.normalized_name)
|| item
@@ -138,8 +150,12 @@ mod tests {
name: "Massive Attack".into(),
normalized_name: "massive attack".into(),
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,
revision,
deleted,
@@ -160,8 +176,12 @@ mod tests {
name: "Teardrop".into(),
normalized_name: "teardrop".into(),
artist_names: vec!["Massive Attack".into()],
featured_artist_names: Vec::new(),
year: Some(1998),
release_type: None,
release_title: None,
track_number: None,
disc_number: None,
duration_seconds: Some(330.0),
revision: 1,
deleted: false,
crates/music-dht/src/lib.rs
+6 -1
View File
@@ -14,7 +14,8 @@
//! `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).
//! the name, every main/featured artist name and the track's release title
//! (so a track is found by its artists).
//! * 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).
@@ -47,8 +48,12 @@
//! kind: ItemKind::Track,
//! name: "Teardrop".into(),
//! artist_names: vec!["Massive Attack".into()],
//! featured_artist_names: vec![],
//! year: Some(1998),
//! release_type: None,
//! release_title: Some("Mezzanine".into()),
//! track_number: Some(10),
//! disc_number: Some(1),
//! duration_seconds: Some(330.0),
//! }])
//! .await?;
crates/music-dht/src/message.rs
+1 -1
View File
@@ -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 = 1;
pub const DHT_PROTOCOL_VERSION: u16 = 2;
/// Maximum number of contacts in a single [`PeerExchange`].
pub const MAX_PEER_EXCHANGE_CONTACTS: usize = 32;
/// Maximum number of records in a single [`FindValueResponse`].
crates/music-dht/src/record.rs
+63 -9
View File
@@ -17,7 +17,8 @@ pub type PeerId = EndpointId;
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.
/// 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 lifetime of an active DHT record.
pub const ACTIVE_RECORD_TTL: Duration = Duration::from_secs(30 * 60);
@@ -150,13 +151,28 @@ pub struct LibraryItem {
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).
/// Display names of the item's main artists (empty for artist records).
/// Lets a release or track be found by its artist's name.
pub artist_names: Vec<String>,
/// Display names of featured artists for track records. These names are
/// indexed too, so artist lookups can discover guest appearances without
/// downloading the track first.
#[serde(default)]
pub featured_artist_names: Vec<String>,
/// Release/track year, when known.
pub year: Option<i32>,
/// Release type (album, ep, single, ...) for release records.
/// Release type (album, ep, single, ...) for release records and track
/// release context.
pub release_type: Option<String>,
/// Release title for track records, when known.
#[serde(default)]
pub release_title: Option<String>,
/// Track number inside the release, when known.
#[serde(default)]
pub track_number: Option<i32>,
/// Disc number inside the release, when known.
#[serde(default)]
pub disc_number: Option<i32>,
/// Track duration in seconds for track records.
pub duration_seconds: Option<f64>,
/// Monotonically increasing revision; bumped on every change.
@@ -169,9 +185,9 @@ pub struct LibraryItem {
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.
/// the name plus one key per unique token of the name, every artist name
/// and track release title — so a release/track is also found by searching
/// for its artists or release context.
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();
@@ -183,13 +199,20 @@ impl LibraryItem {
keys
}
/// All tokens this item is searchable by: its own name plus the names of
/// its artists.
/// All tokens this item is searchable by: its own name, its artist names
/// and the release title carried by track records.
pub fn search_tokens(&self) -> Vec<String> {
let mut tokens = tokenize(&self.normalized_name);
for artist in &self.artist_names {
for artist in self
.artist_names
.iter()
.chain(self.featured_artist_names.iter())
{
tokens.extend(tokenize(&normalize_name(artist)));
}
if let Some(release_title) = &self.release_title {
tokens.extend(tokenize(&normalize_name(release_title)));
}
tokens
}
}
@@ -340,8 +363,12 @@ mod tests {
name: "Attack Attack".into(),
normalized_name: "attack attack".into(),
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,
revision: 1,
deleted: false,
@@ -355,6 +382,33 @@ mod tests {
assert_eq!(keys[1], DhtKey::token(&net, "attack"));
}
#[test]
fn search_tokens_cover_featured_artists_and_release_title() {
let key = test_peer(7);
let item = LibraryItem {
id: ItemId::from_bytes([2u8; 32]),
owner: key,
kind: ItemKind::Track,
name: "Karmacoma".into(),
normalized_name: "karmacoma".into(),
artist_names: vec!["Massive Attack".into()],
featured_artist_names: vec!["Tricky".into()],
year: Some(1994),
release_type: Some("album".into()),
release_title: Some("Protection".into()),
track_number: Some(3),
disc_number: Some(1),
duration_seconds: Some(314.0),
revision: 1,
deleted: false,
updated_at_ms: 0,
};
let tokens = item.search_tokens();
assert!(tokens.contains(&"massive".to_string()));
assert!(tokens.contains(&"tricky".to_string()));
assert!(tokens.contains(&"protection".to_string()));
}
#[test]
fn name_validation() {
assert!(validate_name("Massive Attack").is_ok());
crates/music-dht/src/service.rs
+77 -14
View File
@@ -18,14 +18,14 @@ 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,
DhtKey, ItemId, ItemKind, LibraryItem, MAX_ARTISTS_PER_ITEM, MAX_ITEM_NAME_BYTES,
MAX_TOKENS_PER_ITEM, 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";
pub const SCHEMA_NAME: &str = "music-dht-poc-v2";
/// Capacity of the application event channel.
const EVENT_CHANNEL_CAPACITY: usize = 256;
@@ -92,16 +92,70 @@ pub struct ItemSpec {
pub kind: ItemKind,
/// Display name or title.
pub name: String,
/// Display names of the item's artists (empty for artist records).
/// Display names of the item's main artists (empty for artist records).
pub artist_names: Vec<String>,
/// Display names of the item's featured artists (track records only).
pub featured_artist_names: Vec<String>,
/// Release/track year, when known.
pub year: Option<i32>,
/// Release type (album, ep, ...) for releases.
/// Release type (album, ep, ...) for releases and track release context.
pub release_type: Option<String>,
/// Release title for track records, when known.
pub release_title: Option<String>,
/// Track number inside the release, when known.
pub track_number: Option<i32>,
/// Disc number inside the release, when known.
pub disc_number: Option<i32>,
/// Track duration in seconds for tracks.
pub duration_seconds: Option<f64>,
}
fn sanitize_artist_names(
names: Vec<String>,
seen: &mut HashSet<String>,
limit: usize,
) -> Vec<String> {
names
.into_iter()
.filter_map(|name| {
let name = name.trim().to_string();
if name.is_empty() || name.len() > MAX_ITEM_NAME_BYTES {
return None;
}
let normalized = normalize_name(&name);
if normalized.is_empty()
|| tokenize(&normalized).len() > MAX_TOKENS_PER_ITEM
|| !seen.insert(normalized)
{
return None;
}
Some(name)
})
.take(limit)
.collect()
}
fn sanitize_optional_name(value: Option<String>) -> Option<String> {
let value = value?.trim().to_string();
if value.is_empty() || value.len() > MAX_ITEM_NAME_BYTES {
return None;
}
let normalized = normalize_name(&value);
if normalized.is_empty() || tokenize(&normalized).len() > MAX_TOKENS_PER_ITEM {
return None;
}
Some(value)
}
fn sanitize_optional_text(value: Option<String>) -> Option<String> {
let value = value?.trim().to_string();
(!value.is_empty() && value.len() <= MAX_ITEM_NAME_BYTES).then_some(value)
}
fn positive_index(value: Option<i32>) -> Option<i32> {
value.filter(|number| *number > 0)
}
/// Result of one [`MusicDhtService::sync_library`] call.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct SyncStats {
@@ -123,8 +177,12 @@ fn same_content(a: &LibraryItem, b: &LibraryItem) -> bool {
a.kind == b.kind
&& a.name == b.name
&& a.artist_names == b.artist_names
&& a.featured_artist_names == b.featured_artist_names
&& a.year == b.year
&& a.release_type == b.release_type
&& a.release_title == b.release_title
&& a.track_number == b.track_number
&& a.disc_number == b.disc_number
&& a.duration_seconds == b.duration_seconds
}
@@ -361,13 +419,14 @@ impl MusicDhtService {
// 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 seen_artists = HashSet::new();
let artist_names =
sanitize_artist_names(spec.artist_names, &mut seen_artists, MAX_ARTISTS_PER_ITEM);
let featured_artist_names = sanitize_artist_names(
spec.featured_artist_names,
&mut seen_artists,
MAX_ARTISTS_PER_ITEM.saturating_sub(artist_names.len()),
);
let mut item = LibraryItem {
id,
owner,
@@ -375,9 +434,13 @@ impl MusicDhtService {
name,
normalized_name: normalized,
artist_names,
featured_artist_names,
year: spec.year,
release_type: spec.release_type,
duration_seconds: spec.duration_seconds,
release_type: sanitize_optional_text(spec.release_type),
release_title: sanitize_optional_name(spec.release_title),
track_number: positive_index(spec.track_number),
disc_number: positive_index(spec.disc_number),
duration_seconds: spec.duration_seconds.filter(|duration| *duration > 0.0),
revision: 1,
deleted: false,
updated_at_ms: now_ms(),
crates/music-dht/tests/integration.rs
+14 -2
View File
@@ -39,8 +39,12 @@ fn spec(local_key: &str, kind: ItemKind, name: &str, artists: &[&str]) -> ItemSp
kind,
name: name.to_string(),
artist_names: artists.iter().map(|s| s.to_string()).collect(),
featured_artist_names: Vec::new(),
year: Some(1998),
release_type: (kind == ItemKind::Release).then(|| "album".to_string()),
release_title: (kind == ItemKind::Track).then(|| "Mezzanine".to_string()),
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),
}
}
@@ -223,7 +227,12 @@ async fn byte_stream_to_item_owner() {
stream.peer_id
});
node_a
.sync_library(vec![spec("track:1", ItemKind::Track, "Teardrop", &["Massive Attack"])])
.sync_library(vec![spec(
"track:1",
ItemKind::Track,
"Teardrop",
&["Massive Attack"],
)])
.await
.expect("sync");
@@ -241,7 +250,10 @@ async fn byte_stream_to_item_owner() {
assert_eq!(owner, node_a.endpoint_id());
// B streams the bytes from the owner.
let mut stream = node_b.open_stream(owner, BLOB_ALPN).await.expect("open stream");
let mut stream = node_b
.open_stream(owner, BLOB_ALPN)
.await
.expect("open stream");
let mut received = Vec::new();
let mut chunk = [0u8; 1024];
while let Some(n) = stream.recv.read(&mut chunk).await.expect("read") {