From d61d7a6bacd1dd948b81849909dda8ea726eef14 Mon Sep 17 00:00:00 2001 From: Ultradesu Date: Tue, 28 Jul 2026 22:20:33 +0100 Subject: [PATCH] Bump protocols. Added proto status --- Cargo.lock | 10 +++--- Cargo.toml | 4 +-- src/federation/capabilities.rs | 66 ++++++++++++++++++++++++++++++++++ src/federation/mod.rs | 8 +++++ src/federation/serve.rs | 2 ++ 5 files changed, 84 insertions(+), 6 deletions(-) create mode 100644 src/federation/capabilities.rs diff --git a/Cargo.lock b/Cargo.lock index f38904a..18969e4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1707,8 +1707,9 @@ dependencies = [ [[package]] name = "federation-net" -version = "0.1.0" -source = "git+https://gt.hexor.cy/ab/frid.git#8de7d1292708fa0b225e5a4a9d5ab4f0676202d3" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15a8707baeccb46b5935138f9cb3df3c988c0730b807a2634d901f26b39250d6" dependencies = [ "blake3", "data-encoding", @@ -3613,8 +3614,9 @@ dependencies = [ [[package]] name = "music-dht" -version = "0.2.0" -source = "git+https://gt.hexor.cy/ab/frid.git#8de7d1292708fa0b225e5a4a9d5ab4f0676202d3" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5f32daa9edf769fb5686e92ae6884e9fda6ea082452e4208309fc14301e26aef" dependencies = [ "async-trait", "blake3", diff --git a/Cargo.toml b/Cargo.toml index b998668..1016552 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "furumusic" -version = "0.9.6" +version = "0.9.7" edition = "2024" description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL" @@ -39,4 +39,4 @@ uuid = "1" librqbit = { version = "8.1.1", features = ["disable-upload"] } # P2P federation: publishes the library into a shared DHT and serves audio / # catalogs to furumi peers (TUI clients) over the frid stack. -music-dht = { git = "https://gt.hexor.cy/ab/frid.git" } +music-dht = "0.3" diff --git a/src/federation/capabilities.rs b/src/federation/capabilities.rs new file mode 100644 index 0000000..4bdbd61 --- /dev/null +++ b/src/federation/capabilities.rs @@ -0,0 +1,66 @@ +//! Informational publication of the protocol versions exposed by this peer. + +use std::time::Duration; + +use anyhow::Result; +use music_dht::StreamAcceptor; +use music_dht::capabilities::{ + CAPABILITIES_PROTOCOL_VERSION, CapabilityManifest, CapabilityMessage, JAM_ID, read_message, + write_message, +}; + +use super::serve::AUDIO_PROTOCOL_VERSION; + +fn local_manifest() -> CapabilityManifest { + CapabilityManifest::frid("furumusic", env!("CARGO_PKG_VERSION")) + // The web server does not expose federation Jam yet. + .without_protocol(JAM_ID) + .with_protocol("audio", AUDIO_PROTOCOL_VERSION) +} + +pub async fn serve(mut acceptor: StreamAcceptor) { + while let Some(stream) = acceptor.accept().await { + tokio::spawn(async move { + if let Err(error) = serve_one(stream).await { + tracing::debug!("capability stream failed: {error:#}"); + } + }); + } +} + +async fn serve_one(mut stream: music_dht::ByteStream) -> Result<()> { + let response = match read_message(&mut stream).await? { + CapabilityMessage::Get { + version: CAPABILITIES_PROTOCOL_VERSION, + } => CapabilityMessage::Manifest { + manifest: local_manifest(), + }, + CapabilityMessage::Get { version } => CapabilityMessage::Error { + message: format!("unsupported capability protocol {version}"), + }, + _ => CapabilityMessage::Error { + message: "expected capability request".to_string(), + }, + }; + write_message(&mut stream, &response).await?; + stream.send.finish()?; + let _ = tokio::time::timeout(Duration::from_secs(2), stream.send.stopped()).await; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn manifest_describes_only_supported_player_protocols() { + let manifest = local_manifest(); + assert_eq!(manifest.application, "furumusic"); + assert_eq!( + manifest.protocols.get("audio"), + Some(&AUDIO_PROTOCOL_VERSION) + ); + assert!(!manifest.protocols.contains_key(JAM_ID)); + manifest.validate().unwrap(); + } +} diff --git a/src/federation/mod.rs b/src/federation/mod.rs index 85c1b61..73e2064 100644 --- a/src/federation/mod.rs +++ b/src/federation/mod.rs @@ -12,6 +12,7 @@ //! `federation_network_id`, `federation_save_on_listen`) and apply on the fly — saving the settings //! starts, stops or re-joins the node without a server restart. +mod capabilities; pub mod client; pub mod devices; mod receive; @@ -24,6 +25,7 @@ use std::sync::{Arc, OnceLock}; use std::time::Duration; use anyhow::{Context, Result}; +use music_dht::capabilities::CAPABILITIES_ALPN; use music_dht::{ ByteStream, ByteStreamConnectionStats, ItemKind, ItemSpec, MusicDhtConfig, MusicDhtService, NetworkId, PeerTicket, PublishStats, RendezvousConfig, SyncStats, @@ -380,6 +382,7 @@ impl Federation { .stream_protocol(AUDIO_ALPN) .stream_protocol(CATALOG_ALPN) .stream_protocol(devices::SYNC_ALPN) + .schema_independent_stream_protocol(CAPABILITIES_ALPN) .build() .map_err(|err| anyhow::anyhow!("invalid federation config: {err}"))?; let (service, mut events) = @@ -447,6 +450,10 @@ impl Federation { device_hub, Arc::clone(&self.transport_stats), )); + let capabilities_acceptor = service + .stream_acceptor(CAPABILITIES_ALPN) + .map_err(|err| anyhow::anyhow!("failed to take the capabilities acceptor: {err}"))?; + let capabilities_task = tokio::spawn(capabilities::serve(capabilities_acceptor)); *guard = Some(Running { service, @@ -458,6 +465,7 @@ impl Federation { catalog_task, device_task, device_sync_task, + capabilities_task, ], }); self.set_error(None); diff --git a/src/federation/serve.rs b/src/federation/serve.rs index 388ed87..b1669c8 100644 --- a/src/federation/serve.rs +++ b/src/federation/serve.rs @@ -21,6 +21,8 @@ use super::{TransportStats, record_stream_transport}; /// ALPN of the peer-to-peer audio streaming protocol. pub const AUDIO_ALPN: &[u8] = b"furumi-fd/audio/1"; +/// Version of the peer-to-peer audio streaming protocol. +pub const AUDIO_PROTOCOL_VERSION: u16 = 1; /// Maximum size of a JSON protocol line (request or response header). const MAX_PROTOCOL_LINE: usize = 4096;