diff --git a/crates/music-dht/src/database.rs b/crates/music-dht/src/database.rs index 222a608..4903da2 100644 --- a/crates/music-dht/src/database.rs +++ b/crates/music-dht/src/database.rs @@ -113,6 +113,13 @@ pub trait MusicDhtStorage: std::fmt::Debug + Send + Sync { /// Returns non-expired replicas stored under `key`, including tombstones. async fn dht_records_by_key(&self, key: DhtKey, now_ms: u64) -> Result>; + /// Counts non-expired DHT replica records stored by this node. + async fn dht_record_count(&self, _now_ms: u64) -> Result { + Err(MusicDhtError::Database( + "DHT record counts are not supported by this storage backend".to_string(), + )) + } + /// Deletes expired replicas. Returns the number of removed rows. async fn delete_expired_records(&self, now_ms: u64) -> Result; @@ -341,6 +348,19 @@ impl Database { .await } + /// Counts non-expired DHT replica records stored by this node. + pub async fn dht_record_count(&self, now_ms: u64) -> Result { + self.call(move |conn| { + let count: i64 = conn.query_row( + "SELECT COUNT(*) FROM dht_records WHERE expires_at_ms > ?1", + params![now_ms as i64], + |row| row.get(0), + )?; + Ok(count.max(0) as usize) + }) + .await + } + /// Deletes expired replicas. Returns the number of removed rows. pub async fn delete_expired_records(&self, now_ms: u64) -> Result { self.call(move |conn| { @@ -444,6 +464,10 @@ impl MusicDhtStorage for Database { Database::dht_records_by_key(self, key, now_ms).await } + async fn dht_record_count(&self, now_ms: u64) -> Result { + Database::dht_record_count(self, now_ms).await + } + async fn delete_expired_records(&self, now_ms: u64) -> Result { Database::delete_expired_records(self, now_ms).await } @@ -580,8 +604,10 @@ mod tests { }; assert!(db.store_dht_record(key, record).await.expect("store")); assert_eq!(db.dht_records_by_key(key, now).await.expect("get").len(), 1); + assert_eq!(db.dht_record_count(now).await.expect("count"), 1); // After expiry the record is filtered out and then swept. let later = now + 100; + assert_eq!(db.dht_record_count(later).await.expect("count"), 0); assert!( db.dht_records_by_key(key, later) .await diff --git a/crates/music-dht/src/node.rs b/crates/music-dht/src/node.rs index 405a881..0820442 100644 --- a/crates/music-dht/src/node.rs +++ b/crates/music-dht/src/node.rs @@ -430,7 +430,7 @@ impl Node { match self.db.store_dht_record(key, record).await { Ok(stored) => { if stored { - info!( + debug!( item = %artist_id, tombstone = deleted, from = %sender, @@ -719,7 +719,7 @@ impl Node { 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); - info!( + debug!( queried = queried.len(), discovered = known.len(), records = records.len(), @@ -796,7 +796,7 @@ impl Node { } } } - info!( + debug!( item = %item.id, tombstone = item.deleted, keys = keys.len(), diff --git a/crates/music-dht/src/service.rs b/crates/music-dht/src/service.rs index f3dd192..d7a3f58 100644 --- a/crates/music-dht/src/service.rs +++ b/crates/music-dht/src/service.rs @@ -525,6 +525,12 @@ impl MusicDhtService { self.node.db.list_local_items(false).await } + /// Counts non-expired DHT replica records stored by this node. + pub async fn dht_record_count(&self) -> Result { + self.node.ensure_running()?; + self.node.db.dht_record_count(now_ms()).await + } + /// Searches only the local database. pub async fn search_local(&self, query: &str) -> Result> { let normalized = normalize_name(query);