diff --git a/crates/solstone-linux/src/desktop_component.rs b/crates/solstone-linux/src/desktop_component.rs index 637018b..f436953 100644 --- a/crates/solstone-linux/src/desktop_component.rs +++ b/crates/solstone-linux/src/desktop_component.rs @@ -69,15 +69,13 @@ impl DesktopComponent { self.disabled.load(Ordering::Acquire) } /// Acquire this singleton before capture, registration, export, recovery, or any other side - /// effect. The wire-up lode must call this from `observer::lifecycle`'s Setup closure. + /// effect. pub fn acquire_singleton( &self, bus: &impl BusNameRequester, mut log: impl FnMut(&str), ) -> bool { match bus.request_name( - // Soak scaffolding (delete at cutover): the Python observer owns the production name - // while both implementations run side by side. std::env::var("SOLSTONE_LINUX_BUS_NAME") .unwrap_or_else(|_| OBSERVER_BUS_NAME.to_owned()) .as_str(), diff --git a/crates/solstone-linux/src/doctor.rs b/crates/solstone-linux/src/doctor.rs index 44ae7c2..ffac29f 100644 --- a/crates/solstone-linux/src/doctor.rs +++ b/crates/solstone-linux/src/doctor.rs @@ -748,11 +748,19 @@ mod tests { // AC: sync health covers connected, warning, and failing doctor surfaces. #[test] fn sync_health_warn_ok_fail_matrix() { - use crate::sync_health::{ErrorType, SyncFacts}; + use crate::{ + private_link::LinkFactState, + sync_health::{ErrorType, SyncFacts}, + }; let connected = SyncFacts { pending_confirmed: Some(0), last_successful_contact: Some(100.0), last_successful_sync: Some(100.0), + link: Some(LinkFactState { + carrier_proven: true, + observer_registered: true, + ..LinkFactState::default() + }), ..Default::default() }; let offline = SyncFacts { diff --git a/crates/solstone-linux/src/linked_authority_policy_tests.rs b/crates/solstone-linux/src/linked_authority_policy_tests.rs index c7fab14..01f1ce3 100644 --- a/crates/solstone-linux/src/linked_authority_policy_tests.rs +++ b/crates/solstone-linux/src/linked_authority_policy_tests.rs @@ -71,22 +71,22 @@ fn marked_items(lines: &[&str], marker: &str) -> Vec { continue; } depth = line.matches('{').count() as i64 - line.matches('}').count() as i64; - pending = false; + pending = depth == 0 && !line.contains(';'); } } marked } +fn constructs_origin_relative_url(line: &str) -> bool { + let journal_route = line.contains("/app/") || line.contains("/api/"); + journal_route + && (line.contains("format!(") || line.contains(".join(") || line.contains("push_str(")) +} + #[test] fn active_production_paths_have_no_legacy_direct_authority() { let root = source_root(); - let forbidden = [ - ".bearer_auth(", - "localhost:5015", - "Command::new(\"sol\")", - "format!(\"{}/app", - "format!(\"{}/api", - ]; + let forbidden = [".bearer_auth(", "localhost:5015", "Command::new(\"sol\")"]; for path in rust_sources(&root) { if path @@ -100,6 +100,17 @@ fn active_production_paths_have_no_legacy_direct_authority() { let l3_marked = marked_items(&lines, L3_MARKER); let test_only = marked_items(&lines, "#[cfg(test)]"); for (index, line) in lines.iter().enumerate() { + assert!( + !constructs_origin_relative_url(line) + || path + .file_name() + .is_some_and(|name| name == "private_link.rs") + || l3_marked[index] + || test_only[index], + "{}:{} constructs an origin-relative Journal URL", + path.display(), + index + 1 + ); for needle in forbidden { assert!( !line.contains(needle) || l3_marked[index] || test_only[index], diff --git a/crates/solstone-linux/src/observer.rs b/crates/solstone-linux/src/observer.rs index 6d5f593..ec72f07 100644 --- a/crates/solstone-linux/src/observer.rs +++ b/crates/solstone-linux/src/observer.rs @@ -674,7 +674,7 @@ where } #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::*; use crate::chunking::{AudioLeg, LegBlock, StereoAccumulator}; use std::{ @@ -981,6 +981,18 @@ mod tests { f.observer.initialize().unwrap() } + pub(crate) fn drive_real_observer_ticks() { + let mut fixture = fixture(true); + initialize(&mut fixture); + let published_after_initialize = fixture.states.0.borrow().len(); + fixture.observer.tick().unwrap(); + fixture.observer.tick().unwrap(); + assert_eq!( + fixture.states.0.borrow().len(), + published_after_initialize + 2 + ); + } + // tests/test_observer.py::test_compute_rms_mic_left_sys_right // Already covered by chunking::tests::rms_and_hit_gate_match_chunk_contract; this pins consumption. #[test] diff --git a/crates/solstone-linux/src/private_link.rs b/crates/solstone-linux/src/private_link.rs index e9c9cd8..8da0bf8 100644 --- a/crates/solstone-linux/src/private_link.rs +++ b/crates/solstone-linux/src/private_link.rs @@ -1116,6 +1116,17 @@ impl PrivateLinkOwner { self.session.shutdown().await } + #[cfg(test)] + fn loopback_addr(&self) -> std::net::SocketAddr { + format!( + "{}:{}", + self.capability.inner.origin.host_str().unwrap(), + self.capability.inner.origin.port().unwrap() + ) + .parse() + .unwrap() + } + #[cfg(test)] async fn register(&self, body: &serde_json::Value) -> LinkOutcome { let Ok(url) = confine_path(&self.capability.inner.origin, "/app/observer/register") else { @@ -1186,17 +1197,28 @@ pub(crate) async fn start_private_link_owner_with_lock( state_lock: PrivateStateLock, credential: Credential, expected_name: &str, + hostname: String, + platform: String, + version: String, facts: LinkFacts, ) -> Result { let config_root = state_lock.root().to_path_buf(); let session = start_private_link_session_inner( &config_root, - Some(state_lock), credential, expected_name, - Arc::new(NoWriteFault), - SessionTestCapture::default(), - Some(facts), + SessionStartOptions { + state_lock: Some(state_lock), + persistence_fault: Arc::new(NoWriteFault), + #[cfg(test)] + test_capture: SessionTestCapture::default(), + shared_facts: Some(facts), + registration_metadata: Some(RegistrationMetadata { + hostname, + platform, + version, + }), + }, ) .await?; finish_owner_start(session).await @@ -1434,37 +1456,59 @@ pub(crate) async fn start_private_link_session( ) -> Result { start_private_link_session_inner( config_root, - None, credential, expected_name, - Arc::new(NoWriteFault), - SessionTestCapture::default(), - None, + SessionStartOptions::default(), ) .await } +#[cfg(test)] #[derive(Default)] struct SessionTestCapture { - #[cfg(test)] capability: Option>>>, } +struct RegistrationMetadata { + hostname: String, + platform: String, + version: String, +} + +struct SessionStartOptions { + state_lock: Option, + persistence_fault: Arc, + #[cfg(test)] + test_capture: SessionTestCapture, + shared_facts: Option, + registration_metadata: Option, +} + +impl Default for SessionStartOptions { + fn default() -> Self { + Self { + state_lock: None, + persistence_fault: Arc::new(NoWriteFault), + #[cfg(test)] + test_capture: SessionTestCapture::default(), + shared_facts: None, + registration_metadata: None, + } + } +} + async fn start_private_link_session_inner( config_root: &Path, - state_lock: Option, credential: Credential, expected_name: &str, - persistence_fault: Arc, - _test_capture: SessionTestCapture, - shared_facts: Option, + options: SessionStartOptions, ) -> Result { - let state_lock = match state_lock { + let state_lock = match options.state_lock { Some(state_lock) => state_lock, None => PrivateStateLock::acquire(config_root)?, }; let config_root = state_lock.root().to_path_buf(); - let facts = shared_facts.unwrap_or_default(); + let facts = options.shared_facts.unwrap_or_default(); let paths = private_config_paths(&config_root); let sanitized = match sanitize_link_authority(&paths) { Ok(config) => config, @@ -1491,7 +1535,7 @@ async fn start_private_link_session_inner( let (token_persistence, hook) = TokenPersistence::new( config_root.clone(), credential.clone(), - persistence_fault, + options.persistence_fault, transport_unavailable.clone(), facts.clone(), ); @@ -1544,7 +1588,7 @@ async fn start_private_link_session_inner( .bootstrap_url() .ok_or(PrivateStateError::BootstrapFailed)?; #[cfg(test)] - if let Some(capture) = _test_capture.capability { + if let Some(capture) = options.test_capture.capability { let capability = Url::parse(&bootstrap_url) .ok() .and_then(|url| { @@ -1590,6 +1634,14 @@ async fn start_private_link_session_inner( } Err(error) => return Err(error), } + let registration_metadata = + options + .registration_metadata + .unwrap_or_else(|| RegistrationMetadata { + hostname: expected_name.clone(), + platform: "linux".to_owned(), + version: env!("CARGO_PKG_VERSION").to_owned(), + }); let registration = Arc::new(RegistrationCoordinator { client: client.clone(), origin: origin.clone(), @@ -1597,9 +1649,9 @@ async fn start_private_link_session_inner( config_root: config_root.clone(), credential_instance_id: credential_instance_id.clone(), name: Mutex::new(expected_name.clone()), - hostname: expected_name.clone(), - platform: "linux".to_owned(), - version: env!("CARGO_PKG_VERSION").to_owned(), + hostname: registration_metadata.hostname, + platform: registration_metadata.platform, + version: registration_metadata.version, single_flight: tokio::sync::Mutex::new(()), }); Ok(PrivateLinkSession { @@ -2139,6 +2191,26 @@ mod tests { (recorded, completion_order) } + #[tokio::test] + async fn real_transport_dial_waits_for_admission_guard() { + let peer = PrivateLinkPeer::start().await; + let opener = Arc::new(PrivateLinkOpener::new( + TransportClient::new(peer.credential(), None).unwrap(), + Arc::new(AtomicBool::new(false)), + LinkFacts::default(), + )); + let guard = opener.admission.lock().await; + let dial = tokio::spawn({ + let opener = opener.clone(); + async move { opener.dial_carrier().await } + }); + tokio::task::yield_now().await; + assert!(!dial.is_finished()); + drop(guard); + assert!(dial.await.unwrap().is_ok()); + peer.shutdown().await; + } + impl Pairer for FakePairer { fn pair<'a>( &'a self, @@ -2715,14 +2787,14 @@ mod tests { let captured = Arc::new(Mutex::new(None)); let session = start_private_link_session_inner( temp.path(), - None, peer.credential(), "stream", - Arc::new(NoWriteFault), - SessionTestCapture { - capability: Some(captured.clone()), + SessionStartOptions { + test_capture: SessionTestCapture { + capability: Some(captured.clone()), + }, + ..SessionStartOptions::default() }, - None, ) .await .unwrap(); @@ -3376,15 +3448,15 @@ mod tests { let peer = PrivateLinkPeer::start().await; let session = start_private_link_session_inner( temp.path(), - None, peer.credential(), "stream", - Arc::new(RecordingFault { - stages: Arc::new(Mutex::new(Vec::new())), - fail: Some(stage), - }), - SessionTestCapture::default(), - None, + SessionStartOptions { + persistence_fault: Arc::new(RecordingFault { + stages: Arc::new(Mutex::new(Vec::new())), + fail: Some(stage), + }), + ..SessionStartOptions::default() + }, ) .await .unwrap(); @@ -3415,6 +3487,69 @@ mod tests { assert_token_failure(DurableWriteStage::Fsync).await; } + #[tokio::test] + async fn owner_shutdown_joins_bridge_and_closes_listener() { + let temp = tempfile::tempdir().unwrap(); + let peer = PrivateLinkPeer::start().await; + peer.enqueue_response( + 200, + serde_json::json!({ + "key": "K", + "name": "stream", + "prefix": "prefix", + "ingest_url": "/app/observer/ingest", + "protocol_version": 2 + }) + .to_string(), + ); + let owner = start_private_link_owner(temp.path(), peer.credential(), "stream") + .await + .unwrap(); + let address = owner.loopback_addr(); + assert!(tokio::net::TcpStream::connect(address).await.is_ok()); + owner.shutdown().await.unwrap(); + assert!(tokio::net::TcpStream::connect(address).await.is_err()); + peer.shutdown().await; + } + + #[tokio::test] + async fn owner_shutdown_closes_active_bridge_stream() { + let temp = tempfile::tempdir().unwrap(); + let peer = PrivateLinkPeer::start().await; + peer.enqueue_response( + 200, + serde_json::json!({ + "key": "K", + "name": "stream", + "prefix": "prefix", + "ingest_url": "/app/observer/ingest", + "protocol_version": 2 + }) + .to_string(), + ); + peer.enqueue_response(200, br#"{"items":[],"total":0}"#.to_vec()); + let response_gate = Arc::new(AtomicBool::new(false)); + peer.gate_queued_response_nonblocking(1, response_gate); + let owner = start_private_link_owner(temp.path(), peer.credential(), "stream") + .await + .unwrap(); + let request = tokio::spawn({ + let capability = owner.capability(); + async move { capability.list_day("20260101").await } + }); + peer.wait_for_requests(2).await; + assert!(!request.is_finished()); + tokio::time::timeout(Duration::from_secs(1), owner.shutdown()) + .await + .expect("shutdown must close active bridge streams") + .unwrap(); + assert!(matches!( + request.await.unwrap(), + LinkOutcome::TransportUnavailable + )); + peer.shutdown().await; + } + #[tokio::test] async fn executed_session_surfaces_do_not_disclose_secrets() { let peer = PrivateLinkPeer::start().await; @@ -3431,14 +3566,14 @@ mod tests { let capability = Arc::new(Mutex::new(None)); let session = start_private_link_session_inner( temp.path(), - None, paired, "stream", - Arc::new(NoWriteFault), - SessionTestCapture { - capability: Some(capability.clone()), + SessionStartOptions { + test_capture: SessionTestCapture { + capability: Some(capability.clone()), + }, + ..SessionStartOptions::default() }, - None, ) .await .unwrap(); diff --git a/crates/solstone-linux/src/run.rs b/crates/solstone-linux/src/run.rs index 6017064..205e465 100644 --- a/crates/solstone-linux/src/run.rs +++ b/crates/solstone-linux/src/run.rs @@ -445,12 +445,20 @@ async fn start_linked_owner( return Err(error); } }; - let owner = - start_private_link_owner_with_lock(state_lock, credential, &stream, upload.link_facts()) - .await - .inspect_err(|_| { - upload.publish_link_fact(crate::private_link::LinkFact::TransportUnavailable); - })?; + let (hostname, platform, version) = upload.registration_metadata(); + let owner = start_private_link_owner_with_lock( + state_lock, + credential, + &stream, + hostname, + platform, + version, + upload.link_facts(), + ) + .await + .inspect_err(|_| { + upload.publish_link_fact(crate::private_link::LinkFact::TransportUnavailable); + })?; upload.install_capability(owner.capability()); Ok(owner) } @@ -665,6 +673,7 @@ mod tests { publish_observer_registration, }, private_link_test_peer::PrivateLinkPeer, + test_support::{MockServer, OpportunisticDefaultListenerTrap}, }; use std::{cell::RefCell, rc::Rc, sync::atomic::AtomicUsize}; @@ -1052,9 +1061,14 @@ mod tests { temp: &tempfile::TempDir, transport_enabled: bool, ) -> (Result, LinkFactState) { + let legacy_origin = MockServer::new(Vec::new()).await; + let default_listener = OpportunisticDefaultListenerTrap::bind(); + let pending = temp.path().join("pending.segment"); + std::fs::write(&pending, b"pending").unwrap(); let config = Config { config_dir: temp.path().to_path_buf(), stream: "stream".to_owned(), + server_url: legacy_origin.url.clone(), ..Config::default() }; let upload = Arc::new(UploadClient::new( @@ -1073,38 +1087,34 @@ mod tests { lock, transport_enabled, )); - assert_capture_ticks_advance(); + assert_real_observer_ticks_advance(); let result = start.await.unwrap(); + assert!(pending.exists()); + assert!(legacy_origin.requests().is_empty()); + default_listener.assert_zero_connections(); (result, upload.link_fact_state().unwrap()) } - fn assert_capture_ticks_advance() { + fn assert_real_observer_ticks_advance() { let notifier = RecordingNotifier::default(); - let mut ticks = 0_usize; - for _ in 0..2 { - tick_once(¬ifier, || { - ticks += 1; - Ok(()) - }) - .unwrap(); - } - assert_eq!(ticks, 2); - assert_eq!(notifier.watchdogs.load(Ordering::Acquire), 2); + tick_once(¬ifier, || { + crate::observer::tests::drive_real_observer_ticks(); + Ok(()) + }) + .unwrap(); + assert_eq!(notifier.watchdogs.load(Ordering::Acquire), 1); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn missing_credentials_capture_without_transport() { + async fn missing_credentials_observer_ticks_without_transport() { let temp = tempfile::tempdir().unwrap(); - let pending = temp.path().join("pending.segment"); - std::fs::write(&pending, b"pending").unwrap(); let (result, facts) = drive_real_link_start(&temp, true).await; assert!(result.is_err()); assert!(facts.pairing_required); - assert!(pending.exists()); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn malformed_credentials_capture_without_transport() { + async fn malformed_credentials_observer_ticks_without_transport() { let temp = tempfile::tempdir().unwrap(); std::fs::write(temp.path().join(CREDENTIALS_FILENAME), b"{").unwrap(); let (result, facts) = drive_real_link_start(&temp, true).await; @@ -1113,7 +1123,7 @@ mod tests { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn mismatched_or_corrupt_observer_capture_then_register() { + async fn mismatched_or_corrupt_observer_ticks_then_register() { let temp = tempfile::tempdir().unwrap(); let peer = PrivateLinkPeer::start().await; persist_credential(temp.path(), &peer.credential()).unwrap(); @@ -1129,23 +1139,17 @@ mod tests { let (result, facts) = drive_real_link_start(&temp, true).await; let owner = result.unwrap(); assert!(facts.observer_registered); + let registration: serde_json::Value = + serde_json::from_slice(&peer.requests()[0].body).unwrap(); + assert_eq!(registration["hostname"], "host"); + assert_eq!(registration["label"], "stream"); + assert_ne!(registration["hostname"], registration["label"]); owner.shutdown().await.unwrap(); peer.shutdown().await; } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn unavailable_carrier_capture_without_transport_wait() { - let temp = tempfile::tempdir().unwrap(); - let peer = PrivateLinkPeer::start().await; - persist_credential(temp.path(), &peer.credential()).unwrap(); - peer.shutdown().await; - let (result, facts) = drive_real_link_start(&temp, true).await; - assert!(result.is_err()); - assert!(facts.transport_unavailable); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn failed_bootstrap_capture_without_transport_wait() { + async fn unavailable_peer_observer_ticks_without_transport_wait() { let temp = tempfile::tempdir().unwrap(); let peer = PrivateLinkPeer::start().await; persist_credential(temp.path(), &peer.credential()).unwrap(); @@ -1156,7 +1160,7 @@ mod tests { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn failed_initial_registration_capture_without_transport_wait() { + async fn failed_initial_registration_observer_ticks_without_transport_wait() { let temp = tempfile::tempdir().unwrap(); let peer = PrivateLinkPeer::start().await; persist_credential(temp.path(), &peer.credential()).unwrap(); @@ -1181,7 +1185,7 @@ mod tests { }); peer.wait_for_requests(1).await; assert!(!owner.is_finished()); - assert_capture_ticks_advance(); + assert_real_observer_ticks_advance(); gate.store(true, Ordering::Release); peer.notify_response_gates(); let owner = owner.await.unwrap().unwrap(); @@ -1205,7 +1209,7 @@ mod tests { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn multithreaded_capture_advances_while_real_link_registers() { + async fn multithreaded_observer_ticks_advance_while_real_link_registers() { assert_real_initial_registration_succeeds().await; } @@ -1323,7 +1327,7 @@ mod tests { } #[tokio::test] - async fn runtime_lock_failure_does_not_mutate_capture_config_or_private_state() { + async fn prepare_run_config_lock_failure_does_not_mutate_config_or_private_state() { let temp = tempfile::tempdir().unwrap(); let lock = PrivateStateLock::acquire(temp.path()).unwrap(); let before = std::fs::read_dir(temp.path()) @@ -1346,7 +1350,7 @@ mod tests { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn sanitation_failure_keeps_capture_advancing_and_exposes_fact() { + async fn sanitation_failure_keeps_observer_ticks_advancing_and_exposes_fact() { let temp = tempfile::tempdir().unwrap(); let (result, facts) = drive_real_link_start(&temp, false).await; assert!(result.is_err()); @@ -1389,7 +1393,7 @@ mod tests { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn large_backpressured_upload_does_not_stop_capture_progress() { + async fn large_backpressured_upload_does_not_stop_observer_ticks() { let temp = tempfile::tempdir().unwrap(); let peer = PrivateLinkPeer::start().await; peer.enqueue_response(200, br#"{"status":"ok","segment":"large"}"#.to_vec()); @@ -1401,7 +1405,7 @@ mod tests { let upload = tokio::spawn(async move { client.upload_segment("20260101", "large", &[media]).await }); peer.wait_for_requests(1).await; - assert_capture_ticks_advance(); + assert_real_observer_ticks_advance(); assert!(!upload.is_finished()); gate.store(true, Ordering::Release); peer.notify_response_gates(); @@ -1411,7 +1415,7 @@ mod tests { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn chunked_rejection_does_not_stop_capture_progress() { + async fn chunked_rejection_does_not_stop_observer_ticks() { let temp = tempfile::tempdir().unwrap(); let peer = PrivateLinkPeer::start().await; let (session, _client) = linked_upload_fixture(&temp, &peer).await; @@ -1425,7 +1429,7 @@ mod tests { let capability = session.capability("/app/observer/ingest".to_owned()); async move { capability.ingest(form).await } }); - assert_capture_ticks_advance(); + assert_real_observer_ticks_advance(); match request.await.unwrap() { crate::private_link::LinkOutcome::LocalRejected { status } => { assert_eq!(status, reqwest::StatusCode::BAD_REQUEST); diff --git a/crates/solstone-linux/src/sync.rs b/crates/solstone-linux/src/sync.rs index 108391b..12d255a 100644 --- a/crates/solstone-linux/src/sync.rs +++ b/crates/solstone-linux/src/sync.rs @@ -115,15 +115,16 @@ impl SyncService { let notify = Arc::new(Notify::new()); let pending_trigger = Arc::new(AtomicBool::new(false)); let running = Arc::new(AtomicBool::new(true)); + let link_facts = client.link_facts(); let mut facts = load_facts(&config.state_dir()); facts.in_progress = false; facts.progress.clear(); + facts.link = Some(link_facts.snapshot()); if let Err(error) = save_facts(&config.state_dir(), &facts) { tracing::warn!(%error, "Failed to save sync health"); } let facts = Arc::new(Mutex::new(facts)); let recent_error_count = Arc::new(AtomicU8::new(0)); - let link_facts = client.link_facts(); let mut worker = SyncWorker::new( config.clone(), Arc::clone(&client), @@ -239,6 +240,7 @@ struct SyncWorker { running: Arc, facts: Arc>, recent_error_count: Arc, + link_facts: LinkFacts, synced_days: HashSet, consecutive_failures: u32, last_error_type: Option, @@ -267,6 +269,7 @@ impl SyncWorker { ) -> Self { let synced_days = load_synced_days(&config.state_dir()); let last_recovery_generation = client.recovery_generation(); + let link_facts = client.link_facts(); Self { config, client, @@ -276,6 +279,7 @@ impl SyncWorker { running: control.running, facts, recent_error_count, + link_facts, synced_days, consecutive_failures: 0, last_error_type: None, @@ -768,7 +772,9 @@ impl SyncWorker { } fn save_health(&self) { - if let Err(error) = save_facts(&self.config.state_dir(), &self.facts.lock().unwrap()) { + let mut facts = self.facts.lock().unwrap(); + facts.link = Some(self.link_facts.snapshot()); + if let Err(error) = save_facts(&self.config.state_dir(), &facts) { tracing::warn!(%error, "Failed to save sync health"); } } @@ -1014,9 +1020,11 @@ fn remove_if_empty(path: &Path) { mod tests { use super::*; use crate::{ - private_link::{ObserverState, publish_observer_registration, start_private_link_session}, + private_link::{ + LinkFactState, ObserverState, publish_observer_registration, start_private_link_session, + }, private_link_test_peer::PrivateLinkPeer, - test_support::{MockServer, MutableClock, wait_for_requests}, + test_support::{LinkedMockServer, MockServer, MutableClock, wait_for_requests}, upload::ListingFile, }; use serde_json::{Value, json}; @@ -1111,11 +1119,10 @@ mod tests { temp: &tempfile::TempDir, responses: Vec<(u16, Value)>, retention: i64, - ) -> (MockServer, SyncWorker) { - let server = MockServer::new(responses).await; + ) -> (LinkedMockServer, SyncWorker) { + let server = LinkedMockServer::new(responses).await; let config = Config { - server_url: server.url.clone(), - key: "K".to_owned(), + stream: "desktop".to_owned(), sync_retry_delays: vec![0], cache_retention_days: retention, base_dir: temp.path().to_path_buf(), @@ -1126,8 +1133,9 @@ mod tests { wall: 1_800_000_000.0, mono: 100.0, }); - let client = Arc::new(crate::upload::capability_less_client_for_test( + let client = Arc::new(UploadClient::new( &config, + server.capability(), "host", "linux", "test", @@ -1210,9 +1218,25 @@ mod tests { }) } - fn upload_hits(server: &MockServer) -> usize { + trait RequestLog { + fn logged_requests(&self) -> Vec; + } + + impl RequestLog for MockServer { + fn logged_requests(&self) -> Vec { + self.requests() + } + } + + impl RequestLog for LinkedMockServer { + fn logged_requests(&self) -> Vec { + self.requests() + } + } + + fn upload_hits(server: &impl RequestLog) -> usize { server - .requests() + .logged_requests() .iter() .filter(|request| request.uri == "/app/observer/ingest") .count() @@ -1919,7 +1943,7 @@ mod tests { ..Config::default() }; save_synced_days(&config.state_dir(), &HashSet::from(["20260101".to_owned()])).unwrap(); - let client = Arc::new(crate::upload::capability_less_client_for_test( + let client = Arc::new(crate::upload::linked_fixture_client_for_test( &config, "host", "linux", @@ -1960,6 +1984,11 @@ mod tests { pending_confirmed: Some(0), last_successful_sync: Some(1_800_000_000.0), last_successful_contact: Some(1_800_000_000.0), + link: Some(LinkFactState { + carrier_proven: true, + observer_registered: true, + ..LinkFactState::default() + }), ..SyncFacts::default() }, "on — connected", @@ -2429,7 +2458,7 @@ mod tests { async fn open_probe_worker( temp: &tempfile::TempDir, responses: Vec<(u16, Value)>, - ) -> (MockServer, SyncWorker) { + ) -> (LinkedMockServer, SyncWorker) { let (server, mut worker) = test_worker(temp, responses, -1).await; worker.circuit_open = true; worker.circuit_open_since = 0.0; @@ -2639,7 +2668,7 @@ mod tests { config_dir: temp.path().join("config"), ..Config::default() }; - let client = Arc::new(crate::upload::capability_less_client_for_test( + let client = Arc::new(crate::upload::linked_fixture_client_for_test( &config, "host", "linux", @@ -2770,7 +2799,7 @@ mod tests { responses: Vec<(u16, Value)>, retention: i64, synced: bool, - ) -> (MockServer, SyncWorker, PathBuf) { + ) -> (LinkedMockServer, SyncWorker, PathBuf) { let segment = create_segment(temp, name, b"screen"); let (server, mut worker) = test_worker(temp, responses, retention).await; if synced { @@ -3283,7 +3312,7 @@ mod tests { config_dir: temp.path().join("config"), ..Config::default() }; - let client = Arc::new(crate::upload::capability_less_client_for_test( + let client = Arc::new(crate::upload::linked_fixture_client_for_test( &config, "host", "linux", @@ -3376,7 +3405,7 @@ mod tests { config_dir: temp.path().join("config"), ..Config::default() }; - let client = Arc::new(crate::upload::capability_less_client_for_test( + let client = Arc::new(crate::upload::linked_fixture_client_for_test( &config, "host", "linux", @@ -3412,7 +3441,7 @@ mod tests { config_dir: temp.path().join("config"), ..Config::default() }; - let client = Arc::new(crate::upload::capability_less_client_for_test( + let client = Arc::new(crate::upload::linked_fixture_client_for_test( &config, "host", "linux", @@ -3458,7 +3487,7 @@ mod tests { ..Config::default() }; save_synced_days(&config.state_dir(), &HashSet::from(["20260101".to_owned()])).unwrap(); - let client = Arc::new(crate::upload::capability_less_client_for_test( + let client = Arc::new(crate::upload::linked_fixture_client_for_test( &config, "host", "linux", @@ -3518,7 +3547,7 @@ mod tests { config_dir: temp.path().join("config"), ..Config::default() }; - let client = Arc::new(crate::upload::capability_less_client_for_test( + let client = Arc::new(crate::upload::linked_fixture_client_for_test( &config, "host", "linux", @@ -3566,7 +3595,7 @@ mod tests { config_dir: temp.path().join("config"), ..Config::default() }; - let client = Arc::new(crate::upload::capability_less_client_for_test( + let client = Arc::new(crate::upload::linked_fixture_client_for_test( &config, "host", "linux", @@ -3626,7 +3655,7 @@ mod tests { } assert!(server.requests().is_empty()); notify.notify_one(); - wait_for_requests(&server, 1).await; + server.wait_for_requests(1).await; running.store(false, Ordering::Release); notify.notify_one(); task.await.unwrap(); @@ -3661,7 +3690,7 @@ mod tests { ..Config::default() }; save_synced_days(&config.state_dir(), &HashSet::from(["20260101".to_owned()])).unwrap(); - let client = Arc::new(crate::upload::capability_less_client_for_test( + let client = Arc::new(crate::upload::linked_fixture_client_for_test( &config, "host", "linux", @@ -3770,7 +3799,7 @@ mod tests { }, ) .unwrap(); - let client = Arc::new(crate::upload::capability_less_client_for_test( + let client = Arc::new(crate::upload::linked_fixture_client_for_test( &config, "host", "linux", diff --git a/crates/solstone-linux/src/sync_health.rs b/crates/solstone-linux/src/sync_health.rs index ccf38f4..8553607 100644 --- a/crates/solstone-linux/src/sync_health.rs +++ b/crates/solstone-linux/src/sync_health.rs @@ -305,7 +305,7 @@ pub fn derive_health(facts: &SyncFacts, now: f64, stale_threshold: f64) -> SyncH // Some(401); losing that code would conservatively repaint the failure Revoked. HealthState::Unknown } else if facts.pending_confirmed == Some(0) - && facts.link.as_ref().is_none_or(|link| { + && facts.link.as_ref().is_some_and(|link| { link.carrier_proven && link.observer_registered && !link.transport_unavailable @@ -374,6 +374,22 @@ fn optional_int(data: &Map, key: &str) -> Option { data.get(key).and_then(Value::as_i64) } +fn load_link_facts(data: &Map) -> Option { + let link = data.get("link")?.as_object()?; + let boolean = |key| link.get(key).and_then(Value::as_bool).unwrap_or(false); + Some(LinkFactState { + pairing_required: boolean("pairing_required"), + private_state_invalid: boolean("private_state_invalid"), + config_sanitation_failed: boolean("config_sanitation_failed"), + listener_ready: boolean("listener_ready"), + carrier_proven: boolean("carrier_proven"), + observer_registered: boolean("observer_registered"), + transport_unavailable: boolean("transport_unavailable"), + terminal_revocation: boolean("terminal_revocation"), + token_persistence_failure: boolean("token_persistence_failure"), + }) +} + pub fn load_facts(state_dir: &Path) -> SyncFacts { let Ok(text) = fs::read_to_string(sync_health_path(state_dir)) else { return SyncFacts::default(); @@ -396,7 +412,7 @@ pub fn load_facts(state_dir: &Path) -> SyncFacts { .and_then(Value::as_str) .unwrap_or_default() .to_owned(), - link: None, + link: load_link_facts(&data), } } @@ -413,6 +429,17 @@ pub fn save_facts(state_dir: &Path, facts: &SyncFacts) -> io::Result<()> { "pending_confirmed": facts.pending_confirmed, "in_progress": facts.in_progress, "progress": facts.progress, + "link": facts.link.as_ref().map(|link| json!({ + "pairing_required": link.pairing_required, + "private_state_invalid": link.private_state_invalid, + "config_sanitation_failed": link.config_sanitation_failed, + "listener_ready": link.listener_ready, + "carrier_proven": link.carrier_proven, + "observer_registered": link.observer_registered, + "transport_unavailable": link.transport_unavailable, + "terminal_revocation": link.terminal_revocation, + "token_persistence_failure": link.token_persistence_failure, + })), })) .map_err(io::Error::other)?; text.push('\n'); @@ -546,11 +573,16 @@ mod tests { assert!(health.tooltip.contains("last contact")); } - // tests/test_sync_health.py::test_pending_confirmed_zero_is_only_connected_gate + // Connected requires both zero pending custody and positive linked-transport evidence. #[test] - fn pending_confirmed_zero_is_only_connected_gate() { + fn pending_confirmed_zero_and_link_evidence_gate_connected() { let connected = SyncFacts { pending_confirmed: Some(0), + link: Some(LinkFactState { + carrier_proven: true, + observer_registered: true, + ..LinkFactState::default() + }), ..SyncFacts::default() }; assert_eq!( diff --git a/crates/solstone-linux/src/test_support.rs b/crates/solstone-linux/src/test_support.rs index a3444fc..47f08c6 100644 --- a/crates/solstone-linux/src/test_support.rs +++ b/crates/solstone-linux/src/test_support.rs @@ -1,7 +1,14 @@ // SPDX-License-Identifier: AGPL-3.0-only // Copyright (c) 2026 sol pbc -use crate::observer::Clock; +use crate::{ + observer::Clock, + private_link::{ + ObserverState, PrivateLinkCapability, PrivateLinkSession, publish_observer_registration, + start_private_link_session, + }, + private_link_test_peer::PrivateLinkPeer, +}; use http_body_util::{BodyExt, Full}; use hyper::{ Request, Response, @@ -12,13 +19,13 @@ use hyper::{ use hyper_util::rt::TokioIo; use serde_json::Value; use std::{ - collections::VecDeque, + collections::{HashMap, VecDeque}, convert::Infallible, io, net::TcpListener as StdTcpListener, pin::Pin, sync::{ - Arc, Mutex, + Arc, Mutex, OnceLock, atomic::{AtomicU64, Ordering}, }, task::{Context, Poll}, @@ -122,6 +129,108 @@ pub(crate) struct MockServer { pub(crate) url: String, received: Arc>>, task: JoinHandle<()>, + linked: LinkedMockServer, +} + +static LINKED_FIXTURES: OnceLock>> = OnceLock::new(); + +pub(crate) fn linked_fixture_capability(origin: &str) -> Option { + LINKED_FIXTURES + .get_or_init(|| Mutex::new(HashMap::new())) + .lock() + .unwrap() + .get(origin) + .cloned() +} + +pub(crate) struct LinkedMockServer { + peer: PrivateLinkPeer, + session: PrivateLinkSession, +} + +impl LinkedMockServer { + pub(crate) async fn new(responses: Vec<(u16, Value)>) -> Self { + Self::new_raw( + responses + .into_iter() + .map(|(status, body)| (status, body.to_string().into_bytes())) + .collect(), + ) + .await + } + + pub(crate) async fn new_raw(responses: Vec<(u16, Vec)>) -> Self { + let peer = PrivateLinkPeer::start().await; + for (status, body) in responses { + peer.enqueue_response(status, body); + } + let session = start_private_link_session( + &tempfile::tempdir().unwrap().keep(), + peer.credential(), + "desktop", + ) + .await + .unwrap(); + publish_observer_registration( + &session, + &ObserverState { + credential_instance_id: peer.credential().instance_id, + key: "K".to_owned(), + prefix: "prefix".to_owned(), + name: "desktop".to_owned(), + ingest_url: "/app/observer/ingest".to_owned(), + protocol_version: 2, + }, + ) + .unwrap(); + Self { peer, session } + } + + pub(crate) fn capability(&self) -> PrivateLinkCapability { + self.session.capability("/app/observer/ingest".to_owned()) + } + + pub(crate) fn requests(&self) -> Vec { + self.peer + .requests() + .into_iter() + .map(|request| { + let mut headers = hyper::HeaderMap::new(); + for (name, value) in request.headers { + if let (Ok(name), Ok(value)) = ( + hyper::header::HeaderName::from_bytes(name.as_bytes()), + hyper::header::HeaderValue::from_str(&value), + ) { + headers.append(name, value); + } + } + Received { + method: request.method, + uri: request.path, + headers, + body: request.body, + } + }) + .collect() + } + + pub(crate) fn request_count(&self, uri_substring: &str) -> usize { + self.requests() + .iter() + .filter(|request| request.uri.contains(uri_substring)) + .count() + } + + pub(crate) async fn wait_for_requests(&self, count: usize) { + self.peer.wait_for_requests(count).await; + } + + pub(crate) fn gate_responses(&self, count: usize, gate: Arc) { + for _ in 0..count { + self.peer + .enqueue_gated_response(200, b"{}".to_vec(), gate.clone()); + } + } } pub(crate) enum Action { @@ -144,8 +253,24 @@ impl MockServer { } pub(crate) async fn new_actions(responses: Vec) -> Self { + let linked_responses = responses + .iter() + .map(|action| match action { + Action::Response(status, body) => (*status, body.to_string().into_bytes()), + Action::Raw(status, body) => (*status, body.as_bytes().to_vec()), + Action::OwnedRaw(status, body) => (*status, body.as_bytes().to_vec()), + Action::Disconnect => (503, Vec::new()), + Action::Stream(status, _) => (*status, Vec::new()), + }) + .collect(); + let linked = LinkedMockServer::new_raw(linked_responses).await; let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let url = format!("http://{}", listener.local_addr().unwrap()); + LINKED_FIXTURES + .get_or_init(|| Mutex::new(HashMap::new())) + .lock() + .unwrap() + .insert(url.clone(), linked.capability()); let received = Arc::new(Mutex::new(Vec::new())); let queue = Arc::new(Mutex::new(VecDeque::from(responses))); let task_received = Arc::clone(&received); @@ -232,25 +357,35 @@ impl MockServer { url, received, task, + linked, } } pub(crate) fn requests(&self) -> Vec { - self.received.lock().unwrap().clone() + let linked = self.linked.requests(); + if linked.is_empty() { + self.received.lock().unwrap().clone() + } else { + linked + } } pub(crate) fn request_count(&self, uri_substring: &str) -> usize { - self.received - .lock() - .unwrap() + self.requests() .iter() .filter(|request| request.uri.contains(uri_substring)) .count() } pub(crate) async fn gated() -> (Self, Arc) { + let linked = LinkedMockServer::new_raw(Vec::new()).await; let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let url = format!("http://{}", listener.local_addr().unwrap()); + LINKED_FIXTURES + .get_or_init(|| Mutex::new(HashMap::new())) + .lock() + .unwrap() + .insert(url.clone(), linked.capability()); let received = Arc::new(Mutex::new(Vec::new())); let gate = Arc::new(Notify::new()); let task_received = Arc::clone(&received); @@ -285,11 +420,13 @@ impl MockServer { }); } }); + linked.gate_responses(32, gate.clone()); ( Self { url, received, task, + linked, }, gate, ) @@ -298,6 +435,9 @@ impl MockServer { impl Drop for MockServer { fn drop(&mut self) { + if let Some(fixtures) = LINKED_FIXTURES.get() { + fixtures.lock().unwrap().remove(&self.url); + } self.task.abort(); } } diff --git a/crates/solstone-linux/src/tray_model.rs b/crates/solstone-linux/src/tray_model.rs index 76786d2..f6f47fe 100644 --- a/crates/solstone-linux/src/tray_model.rs +++ b/crates/solstone-linux/src/tray_model.rs @@ -152,7 +152,21 @@ pub fn build(snapshot: &StateSnapshot, interval: i64, now: f64, health: &SyncHea #[cfg(test)] mod tests { use super::*; - use crate::sync_health::{ErrorType, HealthState, SURFACE_BY_STATE, SyncFacts, derive_health}; + use crate::{ + private_link::LinkFactState, + sync_health::{ErrorType, HealthState, SURFACE_BY_STATE, SyncFacts, derive_health}, + }; + fn connected_facts() -> SyncFacts { + SyncFacts { + pending_confirmed: Some(0), + link: Some(LinkFactState { + carrier_proven: true, + observer_registered: true, + ..LinkFactState::default() + }), + ..SyncFacts::default() + } + } fn snapshot() -> StateSnapshot { StateSnapshot { mode: Mode::Screencast, @@ -166,14 +180,7 @@ mod tests { } } fn connected_health() -> SyncHealth { - derive_health( - &SyncFacts { - pending_confirmed: Some(0), - ..Default::default() - }, - 100.0, - 600.0, - ) + derive_health(&connected_facts(), 100.0, 600.0) } #[test] fn countdowns_are_computed_at_read_time() { @@ -240,13 +247,7 @@ mod tests { #[test] fn recording_and_idle_headers_cover_complete_health_axis_from_surfaces() { let cases = [ - ( - SyncFacts { - pending_confirmed: Some(0), - ..Default::default() - }, - HealthState::Connected, - ), + (connected_facts(), HealthState::Connected), ( SyncFacts { in_progress: true, @@ -382,10 +383,7 @@ mod tests { #[test] fn sync_labels_follow_resolved_health_surfaces() { let facts = [ - SyncFacts { - pending_confirmed: Some(0), - ..Default::default() - }, + connected_facts(), SyncFacts { in_progress: true, progress: "3/10 segments".into(), @@ -408,10 +406,7 @@ mod tests { fn icon_ladder_covers_four_statuses_by_seven_health_states() { let cases = [ ( - SyncFacts { - pending_confirmed: Some(0), - ..Default::default() - }, + connected_facts(), ["recording", "idle", "paused", "stopped"], ), ( diff --git a/crates/solstone-linux/src/upload.rs b/crates/solstone-linux/src/upload.rs index 9ed9775..af767e1 100644 --- a/crates/solstone-linux/src/upload.rs +++ b/crates/solstone-linux/src/upload.rs @@ -1,8 +1,6 @@ // SPDX-License-Identifier: AGPL-3.0-only // Copyright (c) 2026 sol pbc -#[cfg(test)] -use crate::config::{ConfigPaths, save_identity}; use crate::{ config::Config, event_sender::{EventSender, SILENT_QUEUE_MAX}, @@ -12,8 +10,6 @@ use crate::{ }, sync_health::ErrorType, }; -#[cfg(test)] -use reqwest::Client; use reqwest::{StatusCode, multipart}; use serde::{Deserialize, Deserializer, de::DeserializeOwned}; #[cfg(test)] @@ -29,12 +25,6 @@ use std::{ }; use tokio_util::sync::CancellationToken; -#[cfg(test)] -const UPLOAD_TIMEOUT: Duration = Duration::from_secs(300); -#[cfg(test)] -const EVENT_TIMEOUT: Duration = Duration::from_secs(30); -#[cfg(test)] -const STREAM_TYPE: &str = "desktop"; #[cfg(test)] const OBSERVER_PROTOCOL_VERSION_HEADER: &str = "X-Solstone-Protocol-Version"; const DEFAULT_RETRY_DELAYS: [i64; 4] = [5, 30, 120, 300]; @@ -102,32 +92,13 @@ pub(crate) struct Inner { fallback_link_facts: crate::private_link::LinkFacts, #[cfg(test)] expose_link_facts: AtomicBool, - #[cfg(test)] - url: String, - #[cfg(test)] - key: Mutex, - #[cfg(test)] - stream: Mutex, revoked: AtomicBool, - #[cfg(test)] - client: Client, cancellation: CancellationToken, - #[cfg(test)] hostname: String, - #[cfg(not(test))] - _hostname: String, - #[cfg(test)] platform: String, - #[cfg(not(test))] - _platform: String, - #[cfg(test)] version: String, - #[cfg(not(test))] - _version: String, retry_delays: Vec, immediate_attempts: usize, - #[cfg(test)] - paths: ConfigPaths, clock: Arc, recovery_lock: tokio::sync::Mutex<()>, last_recovery_attempt: Mutex>, @@ -143,20 +114,13 @@ struct TellingState { rejection_warned: bool, } -#[cfg(test)] -enum RegisterAttempt { - Registered { key: String, name: String }, - GuardRefused { reason_code: Option }, - Failed, -} - pub struct UploadClient { inner: Arc, event_sender: EventSender, } impl UploadClient { - /// Create an observer HTTP client. + /// Create the linked upload and event client. /// /// # Panics /// @@ -200,37 +164,15 @@ impl UploadClient { fallback_link_facts: crate::private_link::LinkFacts::default(), #[cfg(test)] expose_link_facts: AtomicBool::new(true), - #[cfg(test)] - url: config.server_url.trim_end_matches('/').to_owned(), - #[cfg(test)] - key: Mutex::new(config.key.clone()), - #[cfg(test)] - stream: Mutex::new(config.stream.clone()), revoked: AtomicBool::new(false), - #[cfg(test)] - client: Client::new(), cancellation: CancellationToken::new(), - #[cfg(test)] hostname: hostname.into(), - #[cfg(not(test))] - _hostname: hostname.into(), - #[cfg(test)] platform: platform.into(), - #[cfg(not(test))] - _platform: platform.into(), - #[cfg(test)] version: version.into(), - #[cfg(not(test))] - _version: version.into(), retry_delays, immediate_attempts: config .sync_max_retries .clamp(1, MAX_IMMEDIATE_ATTEMPTS as i64) as usize, - #[cfg(test)] - paths: ConfigPaths { - base_dir: Some(config.base_dir.clone()), - config_dir: Some(config.config_dir.clone()), - }, clock, recovery_lock: tokio::sync::Mutex::new(()), last_recovery_attempt: Mutex::new(None), @@ -253,11 +195,6 @@ impl UploadClient { if let Some(capability) = self.inner.capability() { return capability.is_registered(); } - #[cfg(test)] - { - return !self.inner.key.lock().unwrap().is_empty(); - } - #[cfg(not(test))] false } @@ -282,13 +219,24 @@ impl UploadClient { } pub(crate) fn link_facts(&self) -> crate::private_link::LinkFacts { - self.inner.fallback_link_facts.clone() + self.inner.capability().map_or_else( + || self.inner.fallback_link_facts.clone(), + |capability| capability.facts(), + ) } pub(crate) fn publish_link_fact(&self, fact: crate::private_link::LinkFact) { self.inner.publish_link_fact(fact); } + pub(crate) fn registration_metadata(&self) -> (String, String, String) { + ( + self.inner.hostname.clone(), + self.inner.platform.clone(), + self.inner.version.clone(), + ) + } + pub fn request_stop(&self) { self.inner.cancellation.cancel(); } @@ -353,57 +301,7 @@ impl UploadClient { | RepairOutcome::InvalidRegistration => false, }; } - #[cfg(not(test))] - return false; - #[cfg(test)] - { - if self.inner.url.is_empty() { - return false; - } - let _registration = self.inner.recovery_lock.lock().await; - if self.is_registered() { - return true; - } - let attempts = 3.min(self.inner.retry_delays.len()); - for attempt in 0..attempts { - match self.inner.register_once().await { - RegisterAttempt::Registered { key, name } => { - if let Err(error) = save_identity(&self.inner.paths, &key, &name) { - // Named deviation: Python propagates the persistence error; Rust keeps all - // three identity stores unchanged so a later call can retry. - tracing::error!(%error, "Failed to persist registration"); - return false; - } - *self.inner.key.lock().unwrap() = key.clone(); - *self.inner.stream.lock().unwrap() = name.clone(); - config.key = key; - config.stream = name.clone(); - tracing::info!(name, "Registered"); - return true; - } - RegisterAttempt::GuardRefused { reason_code } => { - // One-shot CLI setup deliberately does not consume the daemon telling window. - tracing::error!( - status = 403, - reason_code = reason_code.as_deref(), - key_prefix = key_prefix(&self.inner.key.lock().unwrap()), - recovery_generation = - self.inner.recovery_generation.load(Ordering::Acquire), - "Journal refused local identity repair" - ); - return false; - } - RegisterAttempt::Failed => { - tracing::warn!(attempt = attempt + 1, "Registration attempt failed") - } - } - if attempt + 1 < attempts { - tokio::time::sleep(retry_delay(&self.inner.retry_delays, attempt)).await; - } - } - tracing::error!(attempts, "Registration failed after all attempts"); - false - } + false } pub async fn upload_segment( @@ -415,17 +313,16 @@ impl UploadClient { if self.is_revoked() { return UploadResult::failure(Some(ErrorType::Auth), None); } - #[cfg(test)] - let key = self.inner.key.lock().unwrap().clone(); - #[cfg(test)] - if self.inner.capability().is_none() && (key.is_empty() || self.inner.url.is_empty()) { + if self + .inner + .capability() + .is_some_and(|capability| !capability.is_registered()) + { return UploadResult::failure(Some(ErrorType::Client), None); } if !files.iter().any(|path| path.exists()) { return UploadResult::failure(None, None); } - #[cfg(test)] - let url = format!("{}/app/observer/ingest", self.inner.url); let mut last_error = None; let mut last_status = None; for attempt in 0..self.inner.immediate_attempts { @@ -487,60 +384,7 @@ impl UploadClient { } } } else { - #[cfg(not(test))] return UploadResult::failure(Some(ErrorType::Transient), None); - #[cfg(test)] - { - let response = self - .inner - .client - .post(&url) - .bearer_auth(&key) - .multipart(form) - .timeout(UPLOAD_TIMEOUT) - .send() - .await; - match response { - Ok(response) if response.status() == StatusCode::OK => { - match response.json::().await { - Ok(body) => { - return parse_upload_body(body); - } - Err(error) => { - tracing::warn!( - attempt = attempt + 1, - %error, - "Upload attempt returned malformed JSON" - ); - last_error = Some(ErrorType::Transient); - last_status = Some(StatusCode::OK.as_u16()); - } - } - } - Ok(response) => { - let status = response.status(); - let error_type = Self::classify_error(Some(status.as_u16()), false); - last_error = Some(error_type); - last_status = Some(status.as_u16()); - if status == StatusCode::FORBIDDEN { - self.inner.revoked.store(true, Ordering::Release); - } - if error_type != ErrorType::Transient { - tracing::error!(%status, ?error_type, "Upload rejected"); - return UploadResult::failure( - Some(error_type), - Some(status.as_u16()), - ); - } - tracing::warn!(attempt = attempt + 1, %status, "Upload attempt failed"); - } - Err(error) => { - tracing::warn!(attempt = attempt + 1, %error, "Upload attempt failed"); - last_error = Some(ErrorType::Transient); - last_status = None; - } - } - } } if attempt + 1 < self.inner.immediate_attempts { tokio::select! { @@ -558,12 +402,6 @@ impl UploadClient { if self.is_revoked() { return query_failure(ErrorType::Auth, None); } - #[cfg(test)] - let key = self.inner.key.lock().unwrap().clone(); - #[cfg(test)] - if self.inner.capability().is_none() && (key.is_empty() || self.inner.url.is_empty()) { - return query_failure(ErrorType::Client, None); - } if let Some(capability) = self.inner.capability() { return match capability.list_day(day).await { LinkOutcome::Success { status, body } if status == StatusCode::OK => { @@ -596,41 +434,7 @@ impl UploadClient { LinkOutcome::TransportUnavailable => query_failure(ErrorType::Transient, None), }; } - #[cfg(not(test))] - return query_failure(ErrorType::Transient, None); - #[cfg(test)] - { - let url = format!("{}/app/observer/ingest/segments/{day}", self.inner.url); - let response = self - .inner - .client - .get(url) - .bearer_auth(key) - .header(OBSERVER_PROTOCOL_VERSION_HEADER, "2") - .timeout(EVENT_TIMEOUT) - .send() - .await; - let Ok(response) = response else { - return query_failure(ErrorType::Transient, None); - }; - let status = response.status(); - if status != StatusCode::OK { - let error_type = Self::classify_error(Some(status.as_u16()), false); - if status == StatusCode::FORBIDDEN { - self.inner.revoked.store(true, Ordering::Release); - } - tracing::warn!(%status, ?error_type, "Segments query failed"); - return query_failure(error_type, Some(status.as_u16())); - } - let body = match response.json::().await { - Ok(body) => body, - Err(error) => { - tracing::debug!(%error, "Segments query returned malformed JSON"); - return query_failure(ErrorType::Transient, None); - } - }; - parse_listing(body, status.as_u16()) - } + query_failure(ErrorType::Transient, None) } pub async fn relay_event(&self, tract: &str, event: &str, fields: Map) -> bool { @@ -688,6 +492,27 @@ pub(crate) fn capability_less_client_for_test( client } +#[cfg(test)] +pub(crate) fn linked_fixture_client_for_test( + config: &Config, + hostname: impl Into, + platform: impl Into, + version: impl Into, + clock: Arc, +) -> UploadClient { + let capability = crate::test_support::linked_fixture_capability(&config.server_url) + .expect("linked fixture registered for configured test origin"); + UploadClient::with_silent_capacity( + config, + Some(capability), + hostname, + platform, + version, + clock, + SILENT_QUEUE_MAX, + ) +} + impl Inner { fn publish_link_fact(&self, fact: crate::private_link::LinkFact) { self.capability() @@ -713,9 +538,6 @@ impl Inner { self.dropped_events.fetch_add(1, Ordering::AcqRel); } let now = self.clock.monotonic_seconds(); - #[cfg(test)] - let current_key = self.key.lock().unwrap().clone(); - #[cfg(not(test))] let current_key = String::new(); self.tell_first_rejection(route, ¤t_key, now); if self.recovery_on_cooldown(now) { @@ -760,57 +582,6 @@ impl Inner { } } - #[cfg(test)] - async fn register_once(&self) -> RegisterAttempt { - let stream = self.stream.lock().unwrap().clone(); - let mut descriptor = json!({ - "platform": self.platform, - "hostname": self.hostname, - "stream_type": STREAM_TYPE, - "version": self.version, - }); - if !stream.is_empty() { - descriptor["label"] = Value::String(stream); - } - let request = self - .client - .post(format!("{}/app/observer/register", self.url)) - .json(&descriptor) - .timeout(EVENT_TIMEOUT) - .send(); - let response = tokio::select! { - response = request => response, - () = self.cancellation.cancelled() => return RegisterAttempt::Failed, - }; - let Ok(response) = response else { - return RegisterAttempt::Failed; - }; - if response.status() == StatusCode::FORBIDDEN { - let reason_code = response.json::().await.ok().and_then(|body| { - body.get("reason_code") - .and_then(Value::as_str) - .map(str::to_owned) - }); - return RegisterAttempt::GuardRefused { reason_code }; - } - if response.status() != StatusCode::OK { - return RegisterAttempt::Failed; - } - let Ok(body) = response.json::().await else { - return RegisterAttempt::Failed; - }; - let (Some(key), Some(name)) = ( - body.get("key").and_then(Value::as_str), - body.get("name").and_then(Value::as_str), - ) else { - return RegisterAttempt::Failed; - }; - RegisterAttempt::Registered { - key: key.to_owned(), - name: name.to_owned(), - } - } - fn recovery_on_cooldown(&self, now: f64) -> bool { self.last_recovery_attempt .lock() @@ -899,15 +670,6 @@ impl Inner { if self.revoked.load(Ordering::Acquire) { return false; } - #[cfg(test)] - let key = self.key.lock().unwrap().clone(); - #[cfg(test)] - if self.capability().is_none() && (key.is_empty() || self.url.is_empty()) { - return false; - } - #[cfg(test)] - let mut payload = fields; - #[cfg(not(test))] let payload = fields; if let Some(capability) = self.capability() { return match capability @@ -932,31 +694,7 @@ impl Inner { LinkOutcome::TransportUnavailable | LinkOutcome::LocalRejected { .. } => false, }; } - #[cfg(not(test))] - return false; - #[cfg(test)] - { - payload.insert("tract".into(), Value::String(tract.into())); - payload.insert("event".into(), Value::String(event.into())); - let response = self - .client - .post(format!("{}/app/observer/ingest/event", self.url)) - .bearer_auth(key) - .json(&payload) - .timeout(EVENT_TIMEOUT) - .send() - .await; - match response { - Ok(response) if response.status() == StatusCode::OK => true, - Ok(response) => { - if response.status() == StatusCode::FORBIDDEN { - self.revoked.store(true, Ordering::Release); - } - false - } - Err(_) => false, - } - } + false } } @@ -1152,7 +890,6 @@ pub(crate) fn contract_parse_listing(body: Value, status_code: u16) -> QueryResu mod tests { use super::*; use crate::{ - config::{ConfigPaths, load_config}, private_link::{ LinkOutcome, ObserverState, publish_observer_registration, start_private_link_session, }, @@ -1190,7 +927,7 @@ mod tests { } fn client(config: &Config) -> UploadClient { - crate::upload::capability_less_client_for_test( + crate::upload::linked_fixture_client_for_test( config, "host-a", "linux", @@ -1408,11 +1145,16 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn large_backpressured_upload_allows_small_concurrent_request() { - let (temp, legacy, peer, session, client) = - linked_client(200, json!({"status":"ok","segment":"large"})).await; - let response_gate = Arc::new(AtomicBool::new(false)); - peer.gate_next_response_nonblocking(response_gate.clone()); - peer.enqueue_response(200, json!({"items":[],"total":0}).to_string()); + let (temp, legacy, peer, session, client) = linked_client( + 200, + json!({"status":"ok","segment":"large","items":[],"total":0}), + ) + .await; + peer.hold_request_credit(); + peer.enqueue_response( + 200, + json!({"status":"ok","segment":"large","items":[],"total":0}).to_string(), + ); let media = write_file(&temp, "screen.webm", &vec![0x57; 17 * 1024 * 1024]); let client = Arc::new(client); let upload_client = Arc::clone(&client); @@ -1421,17 +1163,19 @@ mod tests { .upload_segment("20260101", "large", &[media]) .await }); - peer.wait_for_requests(1).await; + peer.wait_for_request_staged_at_least(spl_core::mux::INITIAL_WINDOW) + .await; + assert_eq!(peer.max_request_staged(), spl_core::mux::INITIAL_WINDOW); assert!(!upload.is_finished()); let listing_client = Arc::clone(&client); let listing = tokio::spawn(async move { listing_client.get_server_segments("20260101").await }); - peer.wait_for_requests(2).await; + peer.wait_for_requests(1).await; assert!(listing.await.unwrap().segments.is_some()); assert!(!upload.is_finished()); - response_gate.store(true, Ordering::Release); - peer.notify_response_gates(); + peer.release_request_credit(); assert!(upload.await.unwrap().success); + peer.wait_for_requests(2).await; assert_eq!(peer.requests().len(), 2); assert!(legacy.requests().is_empty()); drop(client); @@ -1483,34 +1227,6 @@ mod tests { peer.shutdown().await; } - // tests/test_upload.py::test_ensure_registered_posts_descriptor_and_persists - #[tokio::test] - async fn ensure_registered_posts_descriptor_and_persists() { - let server = - MockServer::new(vec![(200, json!({"key":"K123456789", "name":"fedora"}))]).await; - let temp = TempDir::new().unwrap(); - let mut config = config(&server, &temp); - config.key.clear(); - config.stream = "host-a".into(); - let client = client(&config); - assert!(client.ensure_registered(&mut config).await); - let request = &server.requests()[0]; - assert_eq!(request.uri, "/app/observer/register"); - assert!(request.headers.get("authorization").is_none()); - let descriptor: Value = serde_json::from_slice(&request.body).unwrap(); - assert_eq!(descriptor["label"], "host-a"); - assert_eq!(descriptor["hostname"], "host-a"); - assert_eq!(descriptor["platform"], "linux"); - assert_eq!(descriptor["version"], "0.1.0"); - assert_eq!(descriptor["stream_type"], "desktop"); - let loaded = load_config(ConfigPaths { - base_dir: Some(config.base_dir.clone()), - config_dir: Some(config.config_dir.clone()), - }); - assert_eq!(loaded.config.key, ""); - assert_eq!(loaded.config.stream, "fedora"); - } - #[tokio::test] async fn registration_upload_listing_event_and_status_share_one_carrier() { let default_trap = OpportunisticDefaultListenerTrap::bind(); @@ -1957,22 +1673,6 @@ mod tests { } } - // AC 10: already-green regression pin; failed registration keeps keyless relay offline. - #[tokio::test] - async fn failed_registration_keeps_empty_key_and_relay_offline() { - let server = MockServer::new(vec![(500, json!({}))]).await; - let temp = TempDir::new().unwrap(); - let mut config = config(&server, &temp); - config.key.clear(); - config.sync_retry_delays = vec![0]; - let client = client(&config); - assert!(!client.ensure_registered(&mut config).await); - let before = server.requests().len(); - assert!(!client.relay_event("observe", "status", Map::new()).await); - assert_eq!(server.requests().len(), before); - assert!(!client.is_registered()); - } - // AC 15: twenty clean relays emit no rejection warning. #[tokio::test] async fn clean_relays_emit_no_rejection_warning() { @@ -2046,12 +1746,7 @@ mod tests { assert_eq!(request.method, "POST"); assert_eq!(request.uri, "/app/observer/ingest"); assert_eq!(request.headers["authorization"], "Bearer K"); - assert!( - request - .headers - .get(OBSERVER_PROTOCOL_VERSION_HEADER) - .is_none() - ); + assert_eq!(request.headers[OBSERVER_PROTOCOL_VERSION_HEADER], "2"); let body = String::from_utf8_lossy(&request.body); for expected in [ "name=\"day\"", @@ -2240,9 +1935,9 @@ mod tests { assert_eq!(terminal_result.error_type, Some(ErrorType::Transient)); } - // AC: connection-refused upload is classified Transient end to end + // AC: a client without a linked capability is classified Transient. #[tokio::test] - async fn connection_refused_upload_is_transient() { + async fn capability_less_upload_is_transient() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); drop(listener); @@ -2254,7 +1949,15 @@ mod tests { ..Config::default() }; let media = write_file(&temp, "audio.flac", b"audio"); - let result = client(&config).upload_segment("d", "s", &[media]).await; + let result = capability_less_client_for_test( + &config, + "host-a", + "linux", + "0.1.0", + Arc::new(MutableClock::new(0.0, 0.0)), + ) + .upload_segment("d", "s", &[media]) + .await; assert_eq!(result.error_type, Some(ErrorType::Transient)); } @@ -2316,7 +2019,7 @@ mod tests { let server = MockServer::new(vec![(200, body)]).await; let temp = TempDir::new().unwrap(); let result = client(&config(&server, &temp)) - .get_server_segments("day") + .get_server_segments("20260101") .await; assert!(!result.legacy && !result.truncated); let entry = &result.segments.unwrap()[0]; @@ -2342,7 +2045,7 @@ mod tests { .await; let temp = TempDir::new().unwrap(); let result = client(&config(&server, &temp)) - .get_server_segments("d") + .get_server_segments("20260101") .await; let entries = result.segments.unwrap(); assert_eq!(entries.len(), 2); @@ -2359,7 +2062,7 @@ mod tests { let server = MockServer::new_actions(vec![Action::Raw(200, "not-json")]).await; let temp = TempDir::new().unwrap(); let result = client(&config(&server, &temp)) - .get_server_segments("d") + .get_server_segments("20260101") .await; assert_eq!(result.segments, None); assert_eq!(result.error_type, Some(ErrorType::Transient)); @@ -2373,7 +2076,7 @@ mod tests { let temp = TempDir::new().unwrap(); assert!( client(&config(&server, &temp)) - .get_server_segments("day") + .get_server_segments("20260101") .await .truncated ); @@ -2385,7 +2088,7 @@ mod tests { let server = MockServer::new(vec![(404, json!({}))]).await; let temp = TempDir::new().unwrap(); let result = client(&config(&server, &temp)) - .get_server_segments("day") + .get_server_segments("20260101") .await; assert_eq!(result.error_type, Some(ErrorType::Incompatible)); assert_eq!(result.status_code, Some(404)); @@ -2407,14 +2110,30 @@ mod tests { // AC: unregistered upload is Client, not Auth, and makes no request #[tokio::test] async fn unregistered_upload_is_client_without_request() { - let server = MockServer::new(vec![]).await; + let peer = PrivateLinkPeer::start().await; let temp = TempDir::new().unwrap(); - let mut config = config(&server, &temp); - config.key.clear(); + let config = Config { + config_dir: temp.path().join("config"), + ..Config::default() + }; + let session = start_private_link_session(&config.config_dir, peer.credential(), "desktop") + .await + .unwrap(); + let client = UploadClient::new( + &config, + session.capability("/app/observer/ingest".to_owned()), + "host", + "linux", + "test", + Arc::new(MutableClock::new(0.0, 0.0)), + ); let media = write_file(&temp, "a.flac", b"a"); - let result = client(&config).upload_segment("d", "s", &[media]).await; + let result = client.upload_segment("d", "s", &[media]).await; assert_eq!(result.error_type, Some(ErrorType::Client)); - assert!(server.requests().is_empty()); + assert!(peer.requests().is_empty()); + drop(client); + session.shutdown().await.unwrap(); + peer.shutdown().await; } // AC: pure classification covers protocol statuses and timeout-shaped missing statuses @@ -2471,42 +2190,19 @@ mod tests { ..Config::default() }; let media = write_file(&temp, "a.flac", b"a"); - let result = client(&transport_config) - .upload_segment("d", "s", &[media]) - .await; + let result = capability_less_client_for_test( + &transport_config, + "host-a", + "linux", + "0.1.0", + Arc::new(MutableClock::new(0.0, 0.0)), + ) + .upload_segment("d", "s", &[media]) + .await; assert_eq!(result.error_type, Some(ErrorType::Transient)); assert_eq!(result.status_code, None); } - // AC: empty retry delays fall back and registration survives 500, network error, then 200 - #[tokio::test] - async fn registration_fallback_retries_and_persists() { - let server = MockServer::new_actions(vec![ - Action::Response(500, json!({})), - Action::Disconnect, - Action::Response(200, json!({"key":"new-key", "name":"fedora"})), - ]) - .await; - let temp = TempDir::new().unwrap(); - let mut config = config(&server, &temp); - config.key.clear(); - config.stream = "host-a".into(); - config.sync_retry_delays = vec![0, 0, 0]; - let client = client(&config); - assert!(client.ensure_registered(&mut config).await); - assert_eq!(server.requests().len(), 3); - assert_eq!(config.key, "new-key"); - assert_eq!(config.stream, "fedora"); - - let empty_server = MockServer::new(vec![(200, json!({"key":"k", "name":"n"}))]).await; - let mut empty = self::config(&empty_server, &temp); - empty.key.clear(); - empty.sync_retry_delays.clear(); - let empty_client = self::client(&empty); - assert!(empty_client.ensure_registered(&mut empty).await); - assert_eq!(empty_server.requests().len(), 1); - } - // AC: empty retry delays use the complete local fallback #[tokio::test] async fn empty_retry_delays_use_full_fallback() { @@ -2518,46 +2214,6 @@ mod tests { assert_eq!(client.inner.retry_delays, vec![5, 30, 120, 300]); } - // AC: malformed registration JSON continues to the next attempt - #[tokio::test] - async fn malformed_registration_json_retries() { - let server = MockServer::new_actions(vec![ - Action::Raw(200, "not-json"), - Action::Response(200, json!({"key":"key", "name":"fedora"})), - ]) - .await; - let temp = TempDir::new().unwrap(); - let mut config = config(&server, &temp); - config.key.clear(); - config.sync_retry_delays = vec![0, 0]; - let client = client(&config); - assert!(client.ensure_registered(&mut config).await); - assert_eq!(server.requests().len(), 2); - } - - // AC: failed registration persistence does not latch in-memory registration - #[tokio::test] - async fn persistence_failed_repair_keeps_old_generation_and_reports_failure() { - let server = MockServer::new(vec![ - (200, json!({"key":"key", "name":"fedora"})), - (200, json!({"key":"key", "name":"fedora"})), - ]) - .await; - let temp = TempDir::new().unwrap(); - let blocked = temp.path().join("blocked"); - std::fs::write(&blocked, b"not a directory").unwrap(); - let mut config = config(&server, &temp); - config.key.clear(); - config.config_dir = blocked.clone(); - let client = client(&config); - assert!(!client.ensure_registered(&mut config).await); - assert!(!client.is_registered()); - std::fs::remove_file(&blocked).unwrap(); - assert!(client.ensure_registered(&mut config).await); - assert!(client.is_registered()); - assert_eq!(server.requests().len(), 2); - } - // AC: cancellation raised during backoff interrupts the active wait #[tokio::test] async fn cancellation_during_backoff_interrupts_wait() { @@ -2766,9 +2422,10 @@ mod tests { async fn submit_status_is_nonblocking_while_relay_is_blocked() { let (server, gate) = MockServer::gated().await; let temp = TempDir::new().unwrap(); + let capability = crate::test_support::linked_fixture_capability(&server.url).unwrap(); let mut client = UploadClient::with_silent_capacity( &config(&server, &temp), - None, + Some(capability), "host", "linux", "v", @@ -2810,9 +2467,10 @@ mod tests { async fn silent_overflow_rejects_incoming_and_stop_is_bounded() { let (server, gate) = MockServer::gated().await; let temp = TempDir::new().unwrap(); + let capability = crate::test_support::linked_fixture_capability(&server.url).unwrap(); let mut client = UploadClient::with_silent_capacity( &config(&server, &temp), - None, + Some(capability), "host", "linux", "v",