From 10fa97a9158f6385bbbe76527180417871bc7d05 Mon Sep 17 00:00:00 2001 From: Ultradesu Date: Tue, 28 Jul 2026 22:00:11 +0100 Subject: [PATCH] Added stat version --- ARCHITECTURE.md | 16 ++ Cargo.lock | 130 +--------- Cargo.toml | 3 +- README.md | 2 - apps/federation-net-demo/Cargo.toml | 17 -- apps/federation-net-demo/src/main.rs | 250 ------------------ crates/federation-net/Cargo.toml | 2 +- crates/federation-net/src/config.rs | 53 ++++ crates/federation-net/src/engine.rs | 32 ++- crates/federation-net/src/lib.rs | 1 + crates/federation-net/src/rendezvous.rs | 9 +- crates/federation-net/tests/integration.rs | 59 +++++ crates/music-dht/Cargo.toml | 2 +- crates/music-dht/src/capabilities.rs | 227 ++++++++++++++++ crates/music-dht/src/catalog.rs | 2 + crates/music-dht/src/config.rs | 12 + crates/music-dht/src/jam.rs | 286 +++++++++++++++++++++ crates/music-dht/src/lib.rs | 2 + crates/music-dht/src/service.rs | 5 +- 19 files changed, 703 insertions(+), 407 deletions(-) delete mode 100644 apps/federation-net-demo/Cargo.toml delete mode 100644 apps/federation-net-demo/src/main.rs create mode 100644 crates/music-dht/src/capabilities.rs create mode 100644 crates/music-dht/src/jam.rs diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index a175c23..f176a7d 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -51,6 +51,8 @@ applications from exchanging domain payloads. Changing serialized messages incompatibly requires a new schema ID. Changing the transport handshake incompatibly requires a protocol-version/ALPN plan. +The music DHT currently derives its schema id from `music-dht-v5`; the former +proof-of-concept name is no longer part of the production protocol identity. ### Discovery is not authorization @@ -67,6 +69,12 @@ Tickets provide explicit discovery and carry enough connection information to dial a peer directly. They remain useful when public rendezvous is disabled or unavailable. +Jam control uses its own auxiliary ALPN and an opaque `frid://j/...` +capability. It reuses the portable playback state and command vocabulary from +personal-device sync, but never carries trusted membership, likes, playlists, +or listening history. Possession of the runtime capability authorizes control +of that host player only. + ### Two data paths Typed messages and raw byte streams serve different workloads: @@ -143,6 +151,14 @@ stream capability and defines shared wire models for Furumi catalog and device sync protocols. Audio transfer, rich catalog exchange, and synchronization can therefore use dedicated ALPNs while sharing identity and connectivity. +Capability discovery is the narrow exception to schema isolation. The bounded, +self-versioned `furumi/capabilities/1` stream still validates the federation +transport version and network id, but remains reachable across application +schema upgrades. A peer publishes the versions it actually supports; consumers +retain the highest versions observed during the current run and may recommend a +manual application update. Capability data is informational only and never +authorizes a peer or triggers an automatic update. + ## Failure model Frid assumes normal distributed-system failures: diff --git a/Cargo.lock b/Cargo.lock index efc30ab..fb85783 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -73,56 +73,6 @@ dependencies = [ "libc", ] -[[package]] -name = "anstream" -version = "1.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" -dependencies = [ - "anstyle", - "anstyle-parse", - "anstyle-query", - "anstyle-wincon", - "colorchoice", - "is_terminal_polyfill", - "utf8parse", -] - -[[package]] -name = "anstyle" -version = "1.0.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" - -[[package]] -name = "anstyle-parse" -version = "1.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" -dependencies = [ - "utf8parse", -] - -[[package]] -name = "anstyle-query" -version = "1.1.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" -dependencies = [ - "windows-sys 0.61.2", -] - -[[package]] -name = "anstyle-wincon" -version = "3.0.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" -dependencies = [ - "anstyle", - "once_cell_polyfill", - "windows-sys 0.61.2", -] - [[package]] name = "anyhow" version = "1.0.104" @@ -351,46 +301,6 @@ dependencies = [ "inout", ] -[[package]] -name = "clap" -version = "4.6.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dd059f9da4f5c36b3787f65d38ccaab1cc315f07b01f89abc8359ee6a8205011" -dependencies = [ - "clap_builder", - "clap_derive", -] - -[[package]] -name = "clap_builder" -version = "4.6.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f09628afdcc538b57f3c6341e9c8e9970f18e4a481690a64974d7023bd33548b" -dependencies = [ - "anstream", - "anstyle", - "clap_lex", - "strsim", -] - -[[package]] -name = "clap_derive" -version = "4.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" -dependencies = [ - "heck", - "proc-macro2", - "quote", - "syn 2.0.119", -] - -[[package]] -name = "clap_lex" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" - [[package]] name = "cmov" version = "0.5.4" @@ -406,12 +316,6 @@ dependencies = [ "thiserror 2.0.19", ] -[[package]] -name = "colorchoice" -version = "1.0.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" - [[package]] name = "combine" version = "4.6.7" @@ -897,7 +801,7 @@ checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" [[package]] name = "federation-net" -version = "0.1.0" +version = "0.2.0" dependencies = [ "anyhow", "blake3", @@ -916,18 +820,6 @@ dependencies = [ "tracing", ] -[[package]] -name = "federation-net-demo" -version = "0.1.0" -dependencies = [ - "anyhow", - "clap", - "federation-net", - "serde", - "tokio", - "tracing-subscriber", -] - [[package]] name = "fiat-crypto" version = "0.3.0" @@ -1821,12 +1713,6 @@ dependencies = [ "serde", ] -[[package]] -name = "is_terminal_polyfill" -version = "1.70.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" - [[package]] name = "itoa" version = "1.0.18" @@ -2081,7 +1967,7 @@ dependencies = [ [[package]] name = "music-dht" -version = "0.2.0" +version = "0.3.0" dependencies = [ "anyhow", "async-trait", @@ -2490,12 +2376,6 @@ dependencies = [ "portable-atomic", ] -[[package]] -name = "once_cell_polyfill" -version = "1.70.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" - [[package]] name = "opaque-debug" version = "0.3.1" @@ -3868,12 +3748,6 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" -[[package]] -name = "utf8parse" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" - [[package]] name = "uuid" version = "1.24.0" diff --git a/Cargo.toml b/Cargo.toml index c9d1b55..3db68c4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,6 @@ resolver = "3" members = [ "crates/federation-net", "crates/music-dht", - "apps/federation-net-demo", ] [workspace.package] @@ -13,7 +12,7 @@ rust-version = "1.97" repository = "https://gt.hexor.cy/ab/frid" [workspace.dependencies] -federation-net = { path = "crates/federation-net", version = "0.1.0" } +federation-net = { path = "crates/federation-net", version = "0.2.0" } iroh = "1" iroh-base = "1" iroh-tickets = "1" diff --git a/README.md b/README.md index 75633e3..37b0bef 100644 --- a/README.md +++ b/README.md @@ -41,8 +41,6 @@ failure handling, and compatibility rules. crates/ federation-net/ generic iroh transport and rendezvous music-dht/ distributed music catalog and content discovery -apps/ - federation-net-demo/ small interactive transport example ``` ## Using the libraries diff --git a/apps/federation-net-demo/Cargo.toml b/apps/federation-net-demo/Cargo.toml deleted file mode 100644 index 27d7051..0000000 --- a/apps/federation-net-demo/Cargo.toml +++ /dev/null @@ -1,17 +0,0 @@ -[package] -name = "federation-net-demo" -version = "0.1.0" -description = "CLI demo for the federation-net P2P engine" -publish = false -repository.workspace = true -edition.workspace = true -license.workspace = true -rust-version.workspace = true - -[dependencies] -federation-net = { path = "../../crates/federation-net" } -tokio = { workspace = true } -serde = { workspace = true } -clap = { workspace = true } -tracing-subscriber = { workspace = true } -anyhow = { workspace = true } diff --git a/apps/federation-net-demo/src/main.rs b/apps/federation-net-demo/src/main.rs deleted file mode 100644 index c7e2a51..0000000 --- a/apps/federation-net-demo/src/main.rs +++ /dev/null @@ -1,250 +0,0 @@ -//! CLI demo for the federation-net P2P engine. -//! -//! Starts a peer, prints a shareable ticket and exchanges chat messages with -//! connected peers. Run two instances (optionally on different machines) and -//! pass the first peer's ticket to the second via `--connect`. - -use std::io::Write; -use std::path::PathBuf; -use std::time::{SystemTime, UNIX_EPOCH}; - -use anyhow::Context; -use clap::Parser; -use federation_net::{ - ConnectionDirection, NetworkConfig, NetworkEngine, NetworkEvent, NetworkId, PeerTicket, - SchemaId, -}; -use tokio::io::{AsyncBufReadExt, BufReader}; - -/// Domain message type of the demo application. -/// -/// The library knows nothing about this type; it is defined entirely here. -#[derive(Debug, serde::Serialize, serde::Deserialize)] -enum DemoMessage { - Text { sender: String, body: String }, - Ping { nonce: u64 }, -} - -/// Schema name of the demo message format. -const DEMO_SCHEMA: &str = "demo-message-v1"; - -#[derive(Debug, Parser)] -#[command( - name = "federation-net-demo", - about = "P2P chat demo for federation-net" -)] -struct Args { - /// Directory for the persistent peer identity. - #[arg(long)] - data_dir: PathBuf, - - /// Human-readable name of the network to join. - #[arg(long)] - network_id: String, - - /// Display name used as the sender of chat messages. - #[arg(long)] - name: String, - - /// Ticket of a peer to connect to on startup. - #[arg(long)] - connect: Option, - - /// Enable verbose logging. - #[arg(long)] - verbose: bool, -} - -fn prompt(name: &str) { - print!("{name}> "); - let _ = std::io::stdout().flush(); -} - -#[tokio::main] -async fn main() -> anyhow::Result<()> { - let args = Args::parse(); - - let filter = if args.verbose { - "federation_net=debug,federation_net_demo=debug,info" - } else { - "warn" - }; - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(filter)), - ) - .with_writer(std::io::stderr) - .init(); - - let config = NetworkConfig::builder() - .data_dir(&args.data_dir) - .network_id(NetworkId::from_name(&args.network_id)) - .schema_id(SchemaId::from_name(DEMO_SCHEMA)) - .build() - .context("invalid configuration")?; - - let (engine, mut events) = NetworkEngine::::start(config) - .await - .context("failed to start the network engine")?; - - println!("Endpoint ID: {}", engine.endpoint_id()); - println!("Network ID: {}", engine.network_id()); - println!("Schema ID: {}", engine.schema_id()); - match engine.ticket().await { - Ok(ticket) => println!("Ticket: {ticket}"), - Err(err) => eprintln!("Could not create a ticket yet: {err}"), - } - - if let Some(ticket) = &args.connect { - let ticket: PeerTicket = ticket.parse().context("invalid ticket")?; - match engine.connect(ticket).await { - Ok(peer_id) => { - println!("Connected to: {peer_id}"); - println!("Type a message and press Enter."); - } - Err(err) => { - println!("Connection rejected: {err}"); - let _ = engine.shutdown().await; - std::process::exit(1); - } - } - } else { - println!("Waiting for peers..."); - } - - let name = args.name.clone(); - let mut stdin = BufReader::new(tokio::io::stdin()).lines(); - prompt(&name); - - loop { - tokio::select! { - _ = tokio::signal::ctrl_c() => { - println!(); - break; - } - line = stdin.next_line() => { - match line { - Ok(Some(line)) => { - if handle_line(&engine, &name, line.trim()).await { - break; - } - prompt(&name); - } - Ok(None) => break, // stdin closed - Err(err) => { - eprintln!("failed to read stdin: {err}"); - break; - } - } - } - event = events.recv() => { - match event { - Some(event) => { - println!(); - print_event(&name, event); - prompt(&name); - } - None => { - println!("Engine stopped."); - break; - } - } - } - } - } - - engine.shutdown().await.context("shutdown failed")?; - println!("Bye."); - // The blocking thread behind `tokio::io::stdin()` keeps the runtime alive - // until stdin closes; the engine is already shut down, so exit directly. - let _ = std::io::stdout().flush(); - std::process::exit(0); -} - -/// Handles one line of user input. Returns `true` when the user asked to quit. -async fn handle_line(engine: &NetworkEngine, name: &str, line: &str) -> bool { - match line { - "" => false, - "/quit" => true, - "/peers" => { - let peers = engine.connected_peers(); - if peers.is_empty() { - println!("No connected peers."); - } else { - for peer in peers { - println!("{peer}"); - } - } - false - } - "/ticket" => { - match engine.ticket().await { - Ok(ticket) => println!("Ticket: {ticket}"), - Err(err) => println!("Could not create a ticket: {err}"), - } - false - } - "/ping" => { - let nonce = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or_default(); - broadcast(engine, DemoMessage::Ping { nonce }).await; - false - } - body => { - broadcast( - engine, - DemoMessage::Text { - sender: name.to_string(), - body: body.to_string(), - }, - ) - .await; - false - } - } -} - -/// Sends a message to every connected peer. -async fn broadcast(engine: &NetworkEngine, message: DemoMessage) { - let peers = engine.connected_peers(); - if peers.is_empty() { - println!("No connected peers."); - return; - } - for peer in peers { - if let Err(err) = engine.send(peer, &message).await { - println!("failed to send to {peer}: {err}"); - } - } -} - -fn print_event(name: &str, event: NetworkEvent) { - match event { - NetworkEvent::PeerConnected { peer_id, direction } => { - let direction = match direction { - ConnectionDirection::Incoming => "incoming", - ConnectionDirection::Outgoing => "outgoing", - }; - println!("Peer connected ({direction}): {peer_id}"); - } - NetworkEvent::PeerDisconnected { peer_id, reason } => match reason { - Some(reason) => println!("Peer disconnected: {peer_id} ({reason})"), - None => println!("Peer disconnected: {peer_id}"), - }, - NetworkEvent::MessageReceived { peer_id, message } => match message { - DemoMessage::Text { body, .. } => { - println!("{name} received from {peer_id}: {body}"); - } - DemoMessage::Ping { nonce } => { - println!("{name} received ping from {peer_id} (nonce {nonce})"); - } - }, - NetworkEvent::ProtocolError { peer_id, error } => match peer_id { - Some(peer_id) => println!("Protocol error with {peer_id}: {error}"), - None => println!("Protocol error: {error}"), - }, - } -} diff --git a/crates/federation-net/Cargo.toml b/crates/federation-net/Cargo.toml index b76badf..3275363 100644 --- a/crates/federation-net/Cargo.toml +++ b/crates/federation-net/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "federation-net" -version = "0.1.0" +version = "0.2.0" description = "Generic peer-to-peer networking engine built on Iroh" readme = "README.md" documentation = "https://docs.rs/federation-net" diff --git a/crates/federation-net/src/config.rs b/crates/federation-net/src/config.rs index 6c1436d..477f744 100644 --- a/crates/federation-net/src/config.rs +++ b/crates/federation-net/src/config.rs @@ -41,6 +41,9 @@ pub struct NetworkConfig { /// Auxiliary ALPN protocols on which this peer accepts raw byte streams /// (see [`crate::NetworkEngine::stream_acceptor`]). pub stream_protocols: Vec>, + /// Auxiliary stream protocols whose handshake is independent of the + /// application schema. Intended for compatibility/capability discovery. + pub schema_independent_stream_protocols: Vec>, } impl NetworkConfig { @@ -65,6 +68,7 @@ pub struct NetworkConfigBuilder { max_concurrent_streams_per_peer: Option, rendezvous: Option, stream_protocols: Vec>, + schema_independent_stream_protocols: Vec>, } impl NetworkConfigBuilder { @@ -124,6 +128,17 @@ impl NetworkConfigBuilder { self } + /// Declares an auxiliary ALPN protocol that validates the network and + /// federation transport version, but deliberately does not require the + /// application schema id to match. + /// + /// This is suitable for bounded, self-versioned capability discovery + /// protocols that must remain reachable across schema upgrades. + pub fn schema_independent_stream_protocol(mut self, alpn: impl Into>) -> Self { + self.schema_independent_stream_protocols.push(alpn.into()); + self + } + /// Validates the configuration and builds a [`NetworkConfig`]. pub fn build(self) -> Result { let data_dir = self @@ -210,6 +225,27 @@ impl NetworkConfigBuilder { ))); } } + for (index, alpn) in self.schema_independent_stream_protocols.iter().enumerate() { + if alpn.is_empty() { + return Err(NetworkError::InvalidConfig( + "schema-independent stream protocol ALPN must not be empty".into(), + )); + } + if alpn.as_slice() == ALPN { + return Err(NetworkError::InvalidConfig( + "schema-independent stream protocol ALPN must differ from the engine ALPN" + .into(), + )); + } + if self.schema_independent_stream_protocols[..index].contains(alpn) + || self.stream_protocols.contains(alpn) + { + return Err(NetworkError::InvalidConfig(format!( + "duplicate stream protocol ALPN: {}", + String::from_utf8_lossy(alpn) + ))); + } + } Ok(NetworkConfig { data_dir, @@ -221,6 +257,7 @@ impl NetworkConfigBuilder { max_concurrent_streams_per_peer, rendezvous: self.rendezvous, stream_protocols: self.stream_protocols, + schema_independent_stream_protocols: self.schema_independent_stream_protocols, }) } } @@ -326,5 +363,21 @@ mod tests { .build() .is_err() ); + assert!( + base_builder() + .stream_protocol("dup/1") + .schema_independent_stream_protocol("dup/1") + .build() + .is_err() + ); + + let config = base_builder() + .schema_independent_stream_protocol("furumi/capabilities/1") + .build() + .expect("valid schema-independent stream protocol"); + assert_eq!( + config.schema_independent_stream_protocols, + vec![b"furumi/capabilities/1".to_vec()] + ); } } diff --git a/crates/federation-net/src/engine.rs b/crates/federation-net/src/engine.rs index db71ea0..b759163 100644 --- a/crates/federation-net/src/engine.rs +++ b/crates/federation-net/src/engine.rs @@ -383,6 +383,28 @@ impl Shared { } } + fn validate_stream_handshake( + &self, + alpn: &[u8], + handshake: &Handshake, + ) -> Option { + if handshake.protocol_version != PROTOCOL_VERSION { + Some(HandshakeErrorCode::UnsupportedProtocolVersion) + } else if handshake.network_id != self.config.network_id { + Some(HandshakeErrorCode::NetworkMismatch) + } else if !self + .config + .schema_independent_stream_protocols + .iter() + .any(|protocol| protocol.as_slice() == alpn) + && handshake.schema_id != self.config.schema_id + { + Some(HandshakeErrorCode::SchemaMismatch) + } else { + None + } + } + /// Drives an established connection: accepts message streams until the /// connection closes, then cleans up the registry entry. async fn connection_loop( @@ -627,7 +649,7 @@ impl Shared { .and_then(|res| res); let verdict = match &handshake { - Ok(handshake) => self.validate_handshake(handshake), + Ok(handshake) => self.validate_stream_handshake(alpn, handshake), Err(_) => Some(HandshakeErrorCode::InvalidHandshake), }; if let Some(code) = verdict { @@ -948,6 +970,7 @@ impl NetworkEngine { let stream_acceptors = config .stream_protocols .iter() + .chain(config.schema_independent_stream_protocols.iter()) .map(|alpn| { let (sender, receiver) = mpsc::channel(STREAM_ACCEPT_QUEUE); ( @@ -982,6 +1005,13 @@ impl NetworkEngine { }; router_builder = router_builder.accept(alpn.clone(), handler); } + for alpn in &shared.config.schema_independent_stream_protocols { + let handler = StreamProtocol:: { + shared: Arc::downgrade(&shared), + alpn: alpn.clone(), + }; + router_builder = router_builder.accept(alpn.clone(), handler); + } let router = router_builder.spawn(); *lock(&shared.router) = Some(router); if let Some(rendezvous) = shared.config.rendezvous.clone() { diff --git a/crates/federation-net/src/lib.rs b/crates/federation-net/src/lib.rs index 8d5426b..fdc77db 100644 --- a/crates/federation-net/src/lib.rs +++ b/crates/federation-net/src/lib.rs @@ -72,6 +72,7 @@ pub use engine::{ pub use error::{NetworkError, Result}; pub use event::{ConnectionDirection, NetworkEvent, NetworkEventReceiver}; pub use protocol::{ALPN, NetworkId, PROTOCOL_VERSION, SchemaId}; +pub use rendezvous::RENDEZVOUS_RECORD_VERSION; pub use rendezvous::{DEFAULT_RENDEZVOUS_ENTRY_TTL, DEFAULT_RENDEZVOUS_INTERVAL, RendezvousConfig}; pub use ticket::{PeerTicket, TICKET_VERSION}; diff --git a/crates/federation-net/src/rendezvous.rs b/crates/federation-net/src/rendezvous.rs index 42446ec..55dfa0a 100644 --- a/crates/federation-net/src/rendezvous.rs +++ b/crates/federation-net/src/rendezvous.rs @@ -38,7 +38,8 @@ pub const DEFAULT_RENDEZVOUS_INTERVAL: Duration = Duration::from_secs(60); pub const DEFAULT_RENDEZVOUS_ENTRY_TTL: Duration = Duration::from_secs(30 * 60); /// Version of the rendezvous record wire format. -const RECORD_VERSION: u16 = 1; +/// Version of the Mainline-DHT rendezvous record envelope. +pub const RENDEZVOUS_RECORD_VERSION: u16 = 1; /// Domain separation context for deriving the record signing key. const KEY_DERIVATION_CONTEXT: &str = "federation-net:rendezvous:v1"; /// BEP44 caps mutable values at 1000 bytes; stay safely below. @@ -104,7 +105,7 @@ pub(crate) fn now_ms() -> u64 { fn decode_record(bytes: &[u8]) -> Option> { let record: RendezvousRecord = postcard::from_bytes(bytes).ok()?; - (record.version == RECORD_VERSION).then_some(record.entries) + (record.version == RENDEZVOUS_RECORD_VERSION).then_some(record.entries) } /// Merges entries from every record instance seen this round with our own, @@ -144,7 +145,7 @@ fn merge_entries( fn encode_record_capped(mut entries: Vec) -> Result> { loop { let record = RendezvousRecord { - version: RECORD_VERSION, + version: RENDEZVOUS_RECORD_VERSION, entries, }; let encoded = postcard::to_stdvec(&record).map_err(|err| { @@ -279,7 +280,7 @@ mod tests { fn malformed_and_wrong_version_records_are_ignored() { assert!(decode_record(b"garbage").is_none()); let record = RendezvousRecord { - version: RECORD_VERSION + 1, + version: RENDEZVOUS_RECORD_VERSION + 1, entries: vec![], }; let encoded = postcard::to_stdvec(&record).expect("encode"); diff --git a/crates/federation-net/tests/integration.rs b/crates/federation-net/tests/integration.rs index e59ed10..ae838e9 100644 --- a/crates/federation-net/tests/integration.rs +++ b/crates/federation-net/tests/integration.rs @@ -12,6 +12,7 @@ use federation_net::{ const TEST_TIMEOUT: Duration = Duration::from_secs(120); /// Timeout used when waiting for a single event. const EVENT_TIMEOUT: Duration = Duration::from_secs(30); +const CAPABILITY_ALPN: &[u8] = b"test/capabilities/1"; #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] enum TestMessage { @@ -207,6 +208,64 @@ async fn schema_mismatch_is_rejected() { .expect("test timed out"); } +#[tokio::test] +async fn schema_independent_stream_crosses_schema_versions() { + let _net = NET_LOCK.lock().await; + tokio::time::timeout(TEST_TIMEOUT, async { + let dir_a = tempfile::tempdir().expect("tempdir"); + let dir_b = tempfile::tempdir().expect("tempdir"); + let config_a = NetworkConfig::builder() + .data_dir(dir_a.path()) + .network_id(NetworkId::from_name("same-network")) + .schema_id(SchemaId::from_name("schema-one")) + .schema_independent_stream_protocol(CAPABILITY_ALPN) + .build() + .expect("config a"); + let config_b = NetworkConfig::builder() + .data_dir(dir_b.path()) + .network_id(NetworkId::from_name("same-network")) + .schema_id(SchemaId::from_name("schema-two")) + .schema_independent_stream_protocol(CAPABILITY_ALPN) + .build() + .expect("config b"); + let (engine_a, _events_a) = Engine::start(config_a).await.expect("engine a"); + let (engine_b, _events_b) = Engine::start(config_b).await.expect("engine b"); + let mut acceptor = engine_a + .stream_acceptor(CAPABILITY_ALPN) + .expect("capability acceptor"); + let address = engine_a.ticket().await.expect("ticket").endpoint_addr; + + let accepting = tokio::spawn(async move { + let mut stream = acceptor.accept().await.expect("incoming stream"); + let value = stream.recv.read_to_end(16).await.expect("read request"); + assert_eq!(value, b"versions?"); + stream.send.write_all(b"v2").await.expect("write response"); + stream.send.finish().expect("finish response"); + let _ = stream.send.stopped().await; + }); + let mut stream = engine_b + .open_stream(address, CAPABILITY_ALPN) + .await + .expect("schema-independent stream"); + stream + .send + .write_all(b"versions?") + .await + .expect("write request"); + stream.send.finish().expect("finish request"); + assert_eq!( + stream.recv.read_to_end(16).await.expect("read response"), + b"v2" + ); + accepting.await.expect("accept task"); + + engine_a.shutdown().await.expect("shutdown a"); + engine_b.shutdown().await.expect("shutdown b"); + }) + .await + .expect("test timed out"); +} + #[tokio::test] async fn shutdown_disconnects_peers() { let _net = NET_LOCK.lock().await; diff --git a/crates/music-dht/Cargo.toml b/crates/music-dht/Cargo.toml index 51bee03..1448d7a 100644 --- a/crates/music-dht/Cargo.toml +++ b/crates/music-dht/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "music-dht" -version = "0.2.0" +version = "0.3.0" description = "Distributed music library search: a Kademlia-style DHT on top of federation-net" readme = "README.md" documentation = "https://docs.rs/music-dht" diff --git a/crates/music-dht/src/capabilities.rs b/crates/music-dht/src/capabilities.rs new file mode 100644 index 0000000..f37d0e5 --- /dev/null +++ b/crates/music-dht/src/capabilities.rs @@ -0,0 +1,227 @@ +//! Runtime publication of application and protocol versions. +//! +//! Capability manifests are informational. They let applications explain +//! interoperability problems and suggest an update, but do not authorize a +//! peer or trigger any update action. + +use std::collections::BTreeMap; + +use federation_net::ByteStream; +use serde::{Deserialize, Serialize}; +use tokio::io::{AsyncRead, AsyncReadExt}; + +use crate::error::{MusicDhtError, Result}; + +/// Auxiliary ALPN used to query a peer's version manifest. +pub const CAPABILITIES_ALPN: &[u8] = b"furumi/capabilities/1"; +/// Capability protocol and manifest envelope version. +pub const CAPABILITIES_PROTOCOL_VERSION: u16 = 1; +/// Maximum accepted capability JSON line. +pub const MAX_CAPABILITIES_LINE: usize = 64 * 1024; +/// Maximum protocol entries accepted from one peer. +pub const MAX_PROTOCOL_ENTRIES: usize = 64; + +/// Stable protocol identifier for federation-net. +pub const FEDERATION_NET_ID: &str = "federation_net"; +/// Stable protocol identifier for endpoint tickets. +pub const TICKET_ID: &str = "ticket"; +/// Stable protocol identifier for Mainline-DHT rendezvous records. +pub const RENDEZVOUS_ID: &str = "rendezvous"; +/// Stable protocol identifier for music-dht. +pub const MUSIC_DHT_ID: &str = "music_dht"; +/// Stable protocol identifier for rich catalog streams. +pub const CATALOG_ID: &str = "catalog"; +/// Stable protocol identifier for personal-device synchronization. +pub const DEVICE_SYNC_ID: &str = "device_sync"; +/// Stable protocol identifier for Jam playback control. +pub const JAM_ID: &str = "jam"; + +/// Application and protocol versions published by one peer. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct CapabilityManifest { + /// Manifest schema version. + pub manifest_version: u16, + /// Application family, for example `furumi`. + pub application: String, + /// User-visible application release. + pub application_version: String, + /// Supported/current protocol version by stable identifier. + pub protocols: BTreeMap, +} + +impl CapabilityManifest { + /// Creates a manifest containing every protocol owned by Frid. + pub fn frid(application: impl Into, application_version: impl Into) -> Self { + let mut protocols = BTreeMap::new(); + protocols.insert( + FEDERATION_NET_ID.to_string(), + federation_net::PROTOCOL_VERSION, + ); + protocols.insert(TICKET_ID.to_string(), federation_net::TICKET_VERSION); + protocols.insert( + RENDEZVOUS_ID.to_string(), + federation_net::RENDEZVOUS_RECORD_VERSION, + ); + protocols.insert(MUSIC_DHT_ID.to_string(), crate::DHT_PROTOCOL_VERSION); + protocols.insert( + CATALOG_ID.to_string(), + crate::catalog::CATALOG_PROTOCOL_VERSION, + ); + protocols.insert( + DEVICE_SYNC_ID.to_string(), + crate::device_sync::DEVICE_SYNC_PROTOCOL_VERSION, + ); + protocols.insert(JAM_ID.to_string(), crate::jam::JAM_PROTOCOL_VERSION); + Self { + manifest_version: CAPABILITIES_PROTOCOL_VERSION, + application: application.into(), + application_version: application_version.into(), + protocols, + } + } + + /// Adds an application-owned protocol to the manifest. + pub fn with_protocol(mut self, id: impl Into, version: u16) -> Self { + self.protocols.insert(id.into(), version); + self + } + + /// Removes a shared protocol that this particular application does not + /// expose, while retaining the canonical versions for the others. + pub fn without_protocol(mut self, id: &str) -> Self { + self.protocols.remove(id); + self + } + + /// Validates bounds and identifiers received from a peer. + pub fn validate(&self) -> Result<()> { + if self.manifest_version != CAPABILITIES_PROTOCOL_VERSION { + return Err(protocol_error(format!( + "unsupported capability manifest version {}", + self.manifest_version + ))); + } + if self.application.trim().is_empty() + || self.application.len() > 64 + || self.application_version.len() > 64 + || self.protocols.len() > MAX_PROTOCOL_ENTRIES + { + return Err(protocol_error("invalid capability manifest bounds")); + } + for (id, version) in &self.protocols { + if id.is_empty() + || id.len() > 64 + || !id + .bytes() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_') + || *version == 0 + { + return Err(protocol_error("invalid capability protocol entry")); + } + } + Ok(()) + } +} + +/// One request/response message on [`CAPABILITIES_ALPN`]. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum CapabilityMessage { + /// Requests the peer's current manifest. + Get { + /// Request protocol version. + version: u16, + }, + /// Returns a manifest. + Manifest { + /// Published version manifest. + manifest: CapabilityManifest, + }, + /// Refuses a malformed or unsupported request. + Error { + /// Human-readable diagnostic. + message: String, + }, +} + +/// Writes one bounded JSON-lines capability message. +pub async fn write_message(stream: &mut ByteStream, message: &CapabilityMessage) -> Result<()> { + let mut bytes = serde_json::to_vec(message).map_err(protocol_error)?; + if bytes.len() > MAX_CAPABILITIES_LINE { + return Err(protocol_error("capability message is too large")); + } + bytes.push(b'\n'); + stream.send.write_all(&bytes).await.map_err(network_error)?; + Ok(()) +} + +/// Reads one bounded JSON-lines capability message. +pub async fn read_message(stream: &mut ByteStream) -> Result { + read_message_from(&mut stream.recv).await +} + +/// Reads one bounded JSON-lines capability message from an async reader. +pub async fn read_message_from(reader: &mut R) -> Result { + let mut bytes = Vec::new(); + let mut byte = [0_u8; 1]; + loop { + let read = reader.read(&mut byte).await.map_err(network_error)?; + if read == 0 || byte[0] == b'\n' { + break; + } + bytes.push(byte[0]); + if bytes.len() > MAX_CAPABILITIES_LINE { + return Err(protocol_error("capability message is too large")); + } + } + if bytes.is_empty() { + return Err(protocol_error("capability message is empty")); + } + let message: CapabilityMessage = serde_json::from_slice(&bytes).map_err(protocol_error)?; + if let CapabilityMessage::Manifest { manifest } = &message { + manifest.validate()?; + } + Ok(message) +} + +fn protocol_error(error: impl std::fmt::Display) -> MusicDhtError { + MusicDhtError::Protocol(error.to_string()) +} + +fn network_error(error: impl std::fmt::Display) -> MusicDhtError { + MusicDhtError::Network(error.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn frid_manifest_contains_every_shared_protocol() { + let manifest = CapabilityManifest::frid("furumi", "1.2.3"); + for id in [ + FEDERATION_NET_ID, + TICKET_ID, + RENDEZVOUS_ID, + MUSIC_DHT_ID, + CATALOG_ID, + DEVICE_SYNC_ID, + JAM_ID, + ] { + assert!(manifest.protocols.contains_key(id), "missing {id}"); + } + manifest.validate().unwrap(); + } + + #[test] + fn application_protocol_is_additive() { + let manifest = CapabilityManifest::frid("furumi", "1.2.3").with_protocol("audio", 1); + assert_eq!(manifest.protocols.get("audio"), Some(&1)); + } + + #[test] + fn invalid_protocol_identifier_is_rejected() { + let manifest = CapabilityManifest::frid("furumi", "1.2.3").with_protocol("Audio Stream", 1); + assert!(manifest.validate().is_err()); + } +} diff --git a/crates/music-dht/src/catalog.rs b/crates/music-dht/src/catalog.rs index 76401d2..0f191cd 100644 --- a/crates/music-dht/src/catalog.rs +++ b/crates/music-dht/src/catalog.rs @@ -8,6 +8,8 @@ use serde::{Deserialize, Serialize}; /// ALPN of the Furumi catalog stream protocol. pub const CATALOG_ALPN: &[u8] = b"furumi-fd/catalog/1"; +/// Version of the Furumi catalog stream protocol. +pub const CATALOG_PROTOCOL_VERSION: u16 = 1; /// Request sent by a catalog client. #[derive(Debug, Clone, Default, Serialize, Deserialize)] diff --git a/crates/music-dht/src/config.rs b/crates/music-dht/src/config.rs index 12240b8..066e5ae 100644 --- a/crates/music-dht/src/config.rs +++ b/crates/music-dht/src/config.rs @@ -55,6 +55,9 @@ pub struct MusicDhtConfig { /// Auxiliary ALPN protocols on which this peer accepts raw byte streams /// (see [`crate::MusicDhtService::stream_acceptor`]). pub stream_protocols: Vec>, + /// Auxiliary stream protocols that remain reachable across DHT schema + /// upgrades, such as bounded capability discovery. + pub schema_independent_stream_protocols: Vec>, } impl MusicDhtConfig { @@ -80,6 +83,7 @@ pub struct MusicDhtConfigBuilder { dial_timeout: Option, rendezvous: Option, stream_protocols: Vec>, + schema_independent_stream_protocols: Vec>, } impl MusicDhtConfigBuilder { @@ -146,6 +150,13 @@ impl MusicDhtConfigBuilder { self } + /// Declares a self-versioned auxiliary protocol that does not require the + /// local and remote DHT schema ids to match. + pub fn schema_independent_stream_protocol(mut self, alpn: impl Into>) -> Self { + self.schema_independent_stream_protocols.push(alpn.into()); + self + } + /// Validates and builds the configuration. pub fn build(self) -> Result { let data_dir = self @@ -171,6 +182,7 @@ impl MusicDhtConfigBuilder { dial_timeout: self.dial_timeout.unwrap_or(DEFAULT_DIAL_TIMEOUT), rendezvous: self.rendezvous, stream_protocols: self.stream_protocols, + schema_independent_stream_protocols: self.schema_independent_stream_protocols, }; for (name, value) in [ ("republish_interval", config.republish_interval), diff --git a/crates/music-dht/src/jam.rs b/crates/music-dht/src/jam.rs new file mode 100644 index 0000000..346716e --- /dev/null +++ b/crates/music-dht/src/jam.rs @@ -0,0 +1,286 @@ +//! Ephemeral shared playback control between independent Furumi peers. +//! +//! Jam is separate from personal-device sync. Possession of a [`JamInvite`] +//! capability authorizes playback control for one host process, but never +//! grants likes, playlists, history, or trusted-device membership. + +use serde::{Deserialize, Serialize}; +use tokio::io::{AsyncRead, AsyncReadExt}; + +use crate::device_sync::{PlaybackCommand, PlaybackSnapshot}; +use crate::error::{MusicDhtError, Result}; + +/// Dedicated stream protocol for Jam control version 1. +pub const JAM_ALPN_V1: &[u8] = b"furumi/jam/1"; +/// Current Jam stream protocol. +pub const JAM_ALPN: &[u8] = JAM_ALPN_V1; +/// Current Jam wire version. +pub const JAM_PROTOCOL_VERSION: u16 = 1; +/// Maximum accepted JSON message. +pub const MAX_JAM_LINE: usize = 8 * 1024 * 1024; +/// Recommended timeout for a host with no participant polls. +pub const DEFAULT_JAM_IDLE_TTL_MS: i64 = 30 * 60 * 1_000; +/// Maximum commands accepted in one participant poll. +pub const MAX_JAM_COMMANDS_PER_POLL: usize = 128; + +/// Long-lived host capability encoded as `frid://j/`. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct JamInvite { + /// Invite schema version. + pub v: u16, + /// Host transport ticket. + #[serde(rename = "t")] + pub ticket: String, + /// Runtime Jam session id. + #[serde(rename = "j")] + pub jam_id: String, + /// Runtime capability secret. + #[serde(rename = "s")] + pub secret: String, + /// Host player/device id. + #[serde(rename = "d")] + pub host_device_id: String, + /// Host display name. + #[serde(rename = "n")] + pub host_name: String, +} + +impl JamInvite { + /// Encodes this invite as an opaque `frid://j/...` capability. + pub fn to_uri(&self) -> Result { + validate_invite(self)?; + let bytes = serde_json::to_vec(self).map_err(protocol_err)?; + Ok(format!("frid://j/{}", base64url_encode(&bytes))) + } + + /// Parses and validates an opaque Jam capability. + pub fn from_uri(uri: &str) -> Result { + let encoded = uri + .trim() + .strip_prefix("frid://j/") + .ok_or_else(|| MusicDhtError::Protocol("expected frid://j invite".to_string()))?; + let bytes = base64url_decode(encoded)?; + let invite: Self = serde_json::from_slice(&bytes).map_err(protocol_err)?; + validate_invite(&invite)?; + Ok(invite) + } +} + +/// Display identity scoped to one runtime Jam. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct JamParticipant { + /// Participant runtime id. + pub participant_id: String, + /// User-visible name. + pub name: String, + /// Last successful host exchange in Unix milliseconds. + pub last_seen_ms: i64, +} + +/// Deduplicated playback command submitted by a participant. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct JamCommand { + /// Unique command id generated by the participant. + pub command_id: String, + /// Participant runtime id. + pub participant_id: String, + /// Command payload shared with personal-device and web control. + pub command: PlaybackCommand, + /// Client timestamp for diagnostics. + pub sent_at_ms: i64, +} + +/// Top-level JSON-lines Jam message. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum JamWireMessage { + /// Participant poll: submits commands and requests host state. + Poll { + /// Protocol version. + version: u16, + /// Jam id from the capability. + jam_id: String, + /// Capability secret. + secret: String, + /// Participant display identity. + participant: JamParticipant, + /// Commands not yet acknowledged by the host. + #[serde(default)] + commands: Vec, + }, + /// Host response to a poll. + Snapshot { + /// Whether the capability was accepted. + accepted: bool, + /// Optional refusal reason. + #[serde(default)] + error: Option, + /// IDs of commands accepted by this host runtime. + #[serde(default)] + acknowledged_command_ids: Vec, + /// Current host playback state. + #[serde(default)] + playback: Option, + /// Currently visible participants. + #[serde(default)] + participants: Vec, + /// Host response time in Unix milliseconds. + host_time_ms: i64, + }, + /// Explicit best-effort participant departure. + Leave { + /// Protocol version. + version: u16, + /// Jam id from the capability. + jam_id: String, + /// Capability secret. + secret: String, + /// Participant runtime id. + participant_id: String, + }, +} + +/// Reads a bounded Jam message from an async reader. +pub async fn read_message_from(reader: &mut R) -> Result { + let mut line = Vec::new(); + let mut byte = [0_u8; 1]; + loop { + let read = reader.read(&mut byte).await.map_err(network_err)?; + if read == 0 || byte[0] == b'\n' { + break; + } + line.push(byte[0]); + if line.len() > MAX_JAM_LINE { + return Err(MusicDhtError::Protocol( + "Jam protocol line is too large".to_string(), + )); + } + } + if line.is_empty() { + return Err(MusicDhtError::Protocol( + "Jam protocol message is empty".to_string(), + )); + } + serde_json::from_slice(&line).map_err(protocol_err) +} + +fn validate_invite(invite: &JamInvite) -> Result<()> { + if invite.v != JAM_PROTOCOL_VERSION { + return Err(MusicDhtError::Protocol(format!( + "unsupported Jam invite version {}", + invite.v + ))); + } + if invite.ticket.trim().is_empty() + || invite.jam_id.trim().is_empty() + || invite.secret.len() < 16 + || invite.host_device_id.trim().is_empty() + || invite.host_name.trim().is_empty() + { + return Err(MusicDhtError::Protocol("incomplete Jam invite".to_string())); + } + Ok(()) +} + +fn base64url_encode(bytes: &[u8]) -> String { + const TABLE: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"; + let mut out = String::new(); + let mut i = 0; + while i < bytes.len() { + let b0 = bytes[i]; + let b1 = bytes.get(i + 1).copied().unwrap_or(0); + let b2 = bytes.get(i + 2).copied().unwrap_or(0); + out.push(TABLE[(b0 >> 2) as usize] as char); + out.push(TABLE[(((b0 & 3) << 4) | (b1 >> 4)) as usize] as char); + if i + 1 < bytes.len() { + out.push(TABLE[(((b1 & 15) << 2) | (b2 >> 6)) as usize] as char); + } + if i + 2 < bytes.len() { + out.push(TABLE[(b2 & 63) as usize] as char); + } + i += 3; + } + out +} + +fn base64url_decode(value: &str) -> Result> { + fn decode(byte: u8) -> Option { + match byte { + b'A'..=b'Z' => Some(byte - b'A'), + b'a'..=b'z' => Some(byte - b'a' + 26), + b'0'..=b'9' => Some(byte - b'0' + 52), + b'-' => Some(62), + b'_' => Some(63), + _ => None, + } + } + let bytes = value.as_bytes(); + if bytes.len() % 4 == 1 { + return Err(invalid_base64()); + } + let mut out = Vec::with_capacity(bytes.len() * 3 / 4); + let mut i = 0; + while i < bytes.len() { + let a = decode(bytes[i]).ok_or_else(invalid_base64)?; + let b = decode(*bytes.get(i + 1).ok_or_else(invalid_base64)?).ok_or_else(invalid_base64)?; + let c = bytes.get(i + 2).and_then(|byte| decode(*byte)); + let d = bytes.get(i + 3).and_then(|byte| decode(*byte)); + out.push((a << 2) | (b >> 4)); + if let Some(c) = c { + out.push((b << 4) | (c >> 2)); + if let Some(d) = d { + out.push((c << 6) | d); + } + } + i += 4; + } + Ok(out) +} + +fn invalid_base64() -> MusicDhtError { + MusicDhtError::Protocol("invalid base64url Jam invite".to_string()) +} + +fn protocol_err(err: impl std::fmt::Display) -> MusicDhtError { + MusicDhtError::Protocol(err.to_string()) +} + +fn network_err(err: impl std::fmt::Display) -> MusicDhtError { + MusicDhtError::Network(err.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn invite() -> JamInvite { + JamInvite { + v: JAM_PROTOCOL_VERSION, + ticket: "endpoint-ticket".to_string(), + jam_id: "jam_123".to_string(), + secret: "0123456789abcdef0123456789abcdef".to_string(), + host_device_id: "dev_host".to_string(), + host_name: "Living room".to_string(), + } + } + + #[test] + fn jam_invite_round_trips() { + let invite = invite(); + let uri = invite.to_uri().unwrap(); + assert!(uri.starts_with("frid://j/")); + assert_eq!(JamInvite::from_uri(&uri).unwrap(), invite); + } + + #[test] + fn personal_device_invites_are_not_jam_capabilities() { + assert!(JamInvite::from_uri("frid://i/abcd").is_err()); + } + + #[test] + fn weak_capability_is_rejected() { + let mut invite = invite(); + invite.secret = "short".to_string(); + assert!(invite.to_uri().is_err()); + } +} diff --git a/crates/music-dht/src/lib.rs b/crates/music-dht/src/lib.rs index 595e80e..64a464f 100644 --- a/crates/music-dht/src/lib.rs +++ b/crates/music-dht/src/lib.rs @@ -76,12 +76,14 @@ #![warn(missing_docs)] #![forbid(unsafe_code)] +pub mod capabilities; pub mod catalog; mod config; mod database; pub mod device_sync; mod dht; mod error; +pub mod jam; mod message; mod node; mod normalization; diff --git a/crates/music-dht/src/service.rs b/crates/music-dht/src/service.rs index a257d44..09d9aae 100644 --- a/crates/music-dht/src/service.rs +++ b/crates/music-dht/src/service.rs @@ -28,7 +28,7 @@ use crate::routing::{NodeContact, NodeId}; /// /// The historical value is retained because changing it would partition /// existing deployments. -pub const SCHEMA_NAME: &str = "music-dht-poc-v3"; +pub const SCHEMA_NAME: &str = "music-dht-v5"; /// Capacity of the application event channel. const EVENT_CHANNEL_CAPACITY: usize = 256; @@ -285,6 +285,9 @@ impl MusicDhtService { for alpn in &config.stream_protocols { engine_builder = engine_builder.stream_protocol(alpn.clone()); } + for alpn in &config.schema_independent_stream_protocols { + engine_builder = engine_builder.schema_independent_stream_protocol(alpn.clone()); + } let engine_config = engine_builder .build() .map_err(|err| MusicDhtError::Network(err.to_string()))?;