diff --git a/src/federation/devices.rs b/src/federation/devices.rs index 2dab5a3..125aa4f 100644 --- a/src/federation/devices.rs +++ b/src/federation/devices.rs @@ -1081,50 +1081,53 @@ pub async fn sync_loop( transport_stats: Arc, ) { let mut interval = tokio::time::interval(DEVICE_SYNC_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut polls = tokio::task::JoinSet::new(); + let mut active = std::collections::HashMap::new(); loop { - interval.tick().await; - if let Err(err) = sync_once_all( - &pool, - Arc::clone(&service), - Arc::clone(&hub), - Arc::clone(&transport_stats), - ) - .await - { - tracing::debug!("web fed device sync tick failed: {err:#}"); + tokio::select! { + _ = interval.tick() => { + let users: Vec = match sqlx::query_scalar( + "SELECT DISTINCT user_id FROM furumusic__fed_device WHERE trusted_at_ms IS NOT NULL AND revoked_at_ms IS NULL" + ).fetch_all(&pool).await { + Ok(users) => users, + Err(error) => { tracing::warn!("device poll listing failed: {error:#}"); continue; } + }; + for user_id in users { + let devices = match active_remote_devices(&pool, user_id).await { + Ok(devices) => devices, + Err(error) => { let _ = set_last_error(&pool, user_id, Some(&format!("{error:#}"))).await; continue; } + }; + for device in devices { + let key = (user_id, device.device_id.clone()); + if device.endpoint_ticket.trim().is_empty() || active.values().any(|id| id == &key) { continue; } + let pool = pool.clone(); + let service = Arc::clone(&service); + let hub = Arc::clone(&hub); + let stats = Arc::clone(&transport_stats); + let handle = polls.spawn(async move { + tokio::time::timeout(Duration::from_secs(30), sync_device(&pool, service, hub, stats, user_id, &device)) + .await.context("device sync exchange timed out")? + }); + active.insert(handle.id(), key); + } + } + } + Some(completed) = polls.join_next_with_id(), if !polls.is_empty() => { + let (task, result) = match completed { + Ok((task, result)) => (task, result), + Err(error) => (error.id(), Err(anyhow::Error::from(error))), + }; + if let Some((user_id, device_id)) = active.remove(&task) + && let Err(error) = result { + tracing::debug!(device = %device_id, "web fed device sync failed: {error:#}"); + let _ = set_last_error(&pool, user_id, Some(&format!("{}: {error:#}", short_id(&device_id)))).await; + } + } } } } -pub async fn sync_once_all( - pool: &sqlx::PgPool, - service: Arc, - hub: Arc, - transport_stats: Arc, -) -> Result<()> { - let rows = sqlx::query( - "SELECT DISTINCT user_id FROM furumusic__fed_device - WHERE trusted_at_ms IS NOT NULL AND revoked_at_ms IS NULL", - ) - .fetch_all(pool) - .await?; - for row in rows { - let user_id: i64 = row.get("user_id"); - if let Err(err) = sync_once( - pool, - Arc::clone(&service), - Arc::clone(&hub), - Arc::clone(&transport_stats), - user_id, - ) - .await - { - set_last_error(pool, user_id, Some(&format!("{err:#}"))).await?; - } - } - Ok(()) -} - pub async fn sync_once( pool: &sqlx::PgPool, service: Arc, @@ -5040,3 +5043,6 @@ fn base64url_decode(value: &str) -> Result> { } Ok(out) } + +#[cfg(test)] +mod interop_tests; diff --git a/src/federation/devices/interop_tests.rs b/src/federation/devices/interop_tests.rs new file mode 100644 index 0000000..7c90357 --- /dev/null +++ b/src/federation/devices/interop_tests.rs @@ -0,0 +1,137 @@ +//! Cross-binary protocol contract; the peer is the TUI's production adapter. +use super::*; +use tokio::io::AsyncWriteExt; + +#[tokio::test] +#[ignore = "run furumi_tui/scripts/test_device_interop.py"] +async fn localhost_tui_peer() { + tokio::time::timeout(Duration::from_secs(30), async { + let dir = + std::path::PathBuf::from(std::env::var("FURUMI_INTEROP_DIR").expect("interop runner")); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + std::fs::write( + dir.join("web-address"), + listener.local_addr().unwrap().to_string(), + ) + .unwrap(); + let (mut stream, _) = listener.accept().await.unwrap(); + let hub = PlayerDeviceHub::default(); + let web_id = "web-interop"; + let mut engine = Engine::new( + web_id.into(), + PlaybackConfig::passive(), + Default::default(), + 0, + ); + for phase in 0..3 { + let hello: WireMessage = + serde_json::from_slice(&read_line(&mut stream).await.unwrap()).unwrap(); + let WireMessage::Hello { + profile, + playback: Some(snapshot), + .. + } = hello + else { + panic!("expected TUI hello") + }; + assert_eq!(profile.protocol_version, PROTOCOL_VERSION); + assert_eq!(snapshot.device_id, profile.device_id); + let announcement = snapshot + .coordination + .as_ref() + .expect("versioned playback envelope"); + assert!(engine.observe(&profile.device_id, announcement, phase * 1000)); + if phase == 0 { + assert_eq!(engine.owner(), Some(profile.device_id.as_str())); + assert!( + !engine.tick(10_000), + "passive web server must not seize playback" + ); + assert_eq!(snapshot.state.position_secs, 42.5); + assert_eq!(snapshot.state.volume, 73); + assert!(snapshot.state.shuffle); + // Exercise the production browser hub projection without a + // PostgreSQL library or an audio device. + hub.apply_fed_playback_state_json( + 1, + &profile.device_id, + &profile.name, + true, + serde_json::json!({"tracks": [], "index": 0, "track": null, + "position_seconds": 42.5, "duration_seconds": 100.0, + "paused": false, "shuffle": true, "repeat_mode": "all", + "volume": 0.73, "updated_at_ms": now_ms()}), + ) + .unwrap(); + assert_eq!( + hub.active_device_id_for_commands(1), + Some(format!("fed:{}", profile.device_id)) + ); + } else { + assert_eq!( + engine.owner(), + Some(if phase == 1 { + web_id + } else { + profile.device_id.as_str() + }) + ); + } + let mut state = snapshot.state; + let command = if phase < 2 { + let owner = if phase == 0 { + web_id + } else { + profile.device_id.as_str() + }; + assert!(engine.transfer(owner, (phase + 1) * 1000)); + PlaybackCommand::ActiveChanged { + active_device_id: owner.into(), + active_device_name: "interop".into(), + state: state.clone(), + } + } else { + state.paused = true; + state.position_secs = 87.0; + PlaybackCommand::SetState { + state: state.clone(), + seek: true, + } + }; + engine.set_output(true, engine.is_owner()); + engine.heartbeat((phase + 1) * 1000); + let response = WireMessage::SyncResponse { + accepted: true, + error: None, + devices: vec![], + vector: BTreeMap::new(), + snapshot: SyncSnapshot::default(), + playback: Some(PlaybackSnapshot { + device_id: web_id.into(), + device_name: "WEB".into(), + active: engine.is_owner(), + updated_at_ms: now_ms(), + state, + coordination: Some(engine.announcement()), + }), + ops: vec![SyncOpWire { + op_id: format!("{web_id}:{}", phase + 1), + origin_device_id: web_id.into(), + seq: (phase + 1) as i64, + hlc_ms: now_ms(), + payload: SyncOpPayload::PlaybackCommand { + target_device_id: profile.device_id, + command, + authority: engine.stamp(), + }, + }], + }; + let mut bytes = serde_json::to_vec(&response).unwrap(); + bytes.push(b'\n'); + stream.write_all(&bytes).await.unwrap(); + } + assert_eq!(read_line(&mut stream).await.unwrap(), b"ok"); + }) + .await + .expect("TUI/web exchange timed out"); +} diff --git a/src/main.rs b/src/main.rs index b485a42..d779a5e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -26,7 +26,7 @@ use cot::auth::PasswordVerificationResult; use cot::cli::CliMetadata; use cot::common_types::Password; use cot::config::{ - DatabaseConfig, MiddlewareConfig, ProjectConfig, SameSite, SessionMiddlewareConfig, + DatabaseConfig, Expiry, MiddlewareConfig, ProjectConfig, SameSite, SessionMiddlewareConfig, SessionStoreConfig, SessionStoreTypeConfig, }; use cot::db::Database; @@ -522,6 +522,9 @@ impl Project for FuruProject { MiddlewareConfig::builder() .session( SessionMiddlewareConfig::builder() + .expiry(Expiry::OnInactivity(std::time::Duration::from_secs( + 365 * 24 * 60 * 60, + ))) .secure(false) .same_site(SameSite::Lax) .store( diff --git a/templates/player/scripts.html b/templates/player/scripts.html index 036498a..cd112ca 100644 --- a/templates/player/scripts.html +++ b/templates/player/scripts.html @@ -1,5 +1,23 @@