diff --git a/crates/webcodex-runner/src/webcodex_runner/projects.rs b/crates/webcodex-runner/src/webcodex_runner/projects.rs index cd7d37b89..353ad0e7b 100644 --- a/crates/webcodex-runner/src/webcodex_runner/projects.rs +++ b/crates/webcodex-runner/src/webcodex_runner/projects.rs @@ -24,6 +24,7 @@ mod lifecycle; mod managed_worktree; mod registration; +pub(crate) use catalog::load_runner_project_metadata; pub(crate) use catalog::{ find_project_shell_context, find_project_shell_context_by_id, load_runner_project_summaries_from_dir, project_root_fingerprint, diff --git a/crates/webcodex-runner/src/webcodex_runner/projects/catalog.rs b/crates/webcodex-runner/src/webcodex_runner/projects/catalog.rs index dfe8e94b6..fca248eb2 100644 --- a/crates/webcodex-runner/src/webcodex_runner/projects/catalog.rs +++ b/crates/webcodex-runner/src/webcodex_runner/projects/catalog.rs @@ -674,6 +674,7 @@ fn load_runner_project_with_incarnation(file: &Path) -> Result, ) -> Vec { let entries = match std::fs::read_dir(dir) { @@ -725,7 +726,10 @@ fn load_runner_project_summaries_from_dir_with_shutdown( } warn_empty_hook_commands(&file, &project); projects.push(runner_project_summary_with_shutdown( - &project, updated_at, true, shutdown, + &project, + updated_at, + include_git, + shutdown, )); } projects.sort_by(|a, b| a.id.cmp(&b.id)); @@ -733,12 +737,29 @@ fn load_runner_project_summaries_from_dir_with_shutdown( } pub(crate) fn load_runner_project_summaries_from_dir(dir: &Path) -> Vec { - load_runner_project_summaries_from_dir_with_shutdown(dir, None) + load_runner_project_summaries_from_dir_with_shutdown(dir, true, None) +} + +/// The validated registration catalog without any Git probe. Absent Git fields +/// mean unknown or pending, never a clean or empty repository. +pub(crate) fn load_runner_project_metadata( + cfg: &RunnerConfig, + shutdown: Option<&AtomicBool>, +) -> Vec { + load_runner_project_catalog(cfg, false, shutdown) } fn load_runner_project_summaries( cfg: &RunnerConfig, shutdown: Option<&AtomicBool>, +) -> Vec { + load_runner_project_catalog(cfg, true, shutdown) +} + +fn load_runner_project_catalog( + cfg: &RunnerConfig, + include_git: bool, + shutdown: Option<&AtomicBool>, ) -> Vec { // Loaded configs always carry a materialized project_registry_dir; a bare // test-built config that cannot derive one reports the error instead of @@ -750,7 +771,7 @@ fn load_runner_project_summaries( return Vec::new(); } }; - load_runner_project_summaries_from_dir_with_shutdown(&dir, shutdown) + load_runner_project_summaries_from_dir_with_shutdown(&dir, include_git, shutdown) } impl RunnerProjectCache { diff --git a/crates/webcodex-runner/src/webcodex_runner/transport.rs b/crates/webcodex-runner/src/webcodex_runner/transport.rs index 61a89d855..f65afd7ec 100644 --- a/crates/webcodex-runner/src/webcodex_runner/transport.rs +++ b/crates/webcodex-runner/src/webcodex_runner/transport.rs @@ -54,7 +54,7 @@ use project_inventory::{ use project_inventory::{ log_project_inventory_degraded, paged_sync_after_registration, polling_projects_for_poll, InitialProjectInventory, InitialProjectScanState, PollingProjectRefresh, ProjectInventorySync, - ProjectScanReceiver, StreamingProjectInventoryCoordinator, + ProjectScanPublication, ProjectScanReceiver, StreamingProjectInventoryCoordinator, }; use registration::{build_register_request_with_provider_status, register, RegisterRecoveryAction}; use reqwest::blocking::Client; diff --git a/crates/webcodex-runner/src/webcodex_runner/transport/project_inventory.rs b/crates/webcodex-runner/src/webcodex_runner/transport/project_inventory.rs index 1c8c4d2f3..110f9389b 100644 --- a/crates/webcodex-runner/src/webcodex_runner/transport/project_inventory.rs +++ b/crates/webcodex-runner/src/webcodex_runner/transport/project_inventory.rs @@ -37,14 +37,24 @@ pub(super) enum InitialProjectInventory { Disabled, } -pub(super) type ProjectScanReceiver = - tokio::sync::watch::Receiver>>>; +/// One stage of the initial scan. The registration catalog is published first +/// with unknown Git fields, then replaced once when Git enrichment completes. +#[derive(Clone)] +pub(super) struct ProjectScanPublication { + pub(super) projects: std::sync::Arc>, + pub(super) complete: bool, + /// Inventory generation the catalog was read under. A later Project + /// mutation makes it stale even if a stream has not consumed it yet. + pub(super) generation: u64, +} + +pub(super) type ProjectScanReceiver = tokio::sync::watch::Receiver>; /// Shared across transport sessions so a mutation fences an in-flight scan even /// after its original stream disconnects. pub(super) struct InitialProjectScanState { pub(super) receiver: Option, - pub(super) invalidated: bool, + pub(super) generation: u64, pub(super) changed: tokio::sync::watch::Sender<()>, } @@ -53,7 +63,7 @@ impl Default for InitialProjectScanState { let (changed, _) = tokio::sync::watch::channel(()); Self { receiver: None, - invalidated: false, + generation: 0, changed, } } @@ -468,6 +478,33 @@ impl StreamingProjectInventoryCoordinator { } self.project_cache.invalidate(); let projects = runtime.project_summaries(&mut self.project_cache, cfg); + self.replace_projects(projects, transport, out_tx, reason_code); + } + + /// A mutation during Git enrichment must not leave the older catalog + /// visible until the worker finishes, and needs no Git probe to correct it. + pub(super) fn refresh_from_project_metadata( + &mut self, + transport: StreamTransport, + cfg: &RunnerConfig, + runtime: &RunnerRuntimeState, + out_tx: &tokio::sync::mpsc::Sender, + reason_code: &str, + ) { + if !self.supported { + return; + } + let projects = runtime.project_metadata_summaries(cfg); + self.replace_projects(projects, transport, out_tx, reason_code); + } + + fn replace_projects( + &mut self, + projects: Vec, + transport: StreamTransport, + out_tx: &tokio::sync::mpsc::Sender, + reason_code: &str, + ) { let projects_count = projects.len(); self.sync = Some(ProjectInventorySync::new(projects)); self.retry_backoff.reset(); diff --git a/crates/webcodex-runner/src/webcodex_runner/transport/runtime.rs b/crates/webcodex-runner/src/webcodex_runner/transport/runtime.rs index c9e83f2c2..bc55109f4 100644 --- a/crates/webcodex-runner/src/webcodex_runner/transport/runtime.rs +++ b/crates/webcodex-runner/src/webcodex_runner/transport/runtime.rs @@ -2,6 +2,12 @@ use super::*; +enum ProjectScanPublishOutcome { + Published, + Stale, + Stopped, +} + impl RunnerRuntimeState { pub(crate) fn new(cfg: &RunnerConfig, path: PathBuf) -> Self { Self::with_shutdown_budget(cfg, path, DEFAULT_SHUTDOWN_BUDGET) @@ -87,23 +93,36 @@ impl RunnerRuntimeState { cache.get_with_shutdown(cfg, Some(shutdown.as_ref())) } - /// Registration and keepalive must not wait for serial Git probes. Reuse - /// an in-flight scan across reconnects, so only one scanner can run per - /// process. The existing shutdown flag, Git tree ownership, and bounded - /// background-thread drain still own cancellation and cleanup. + pub(super) fn project_metadata_summaries( + &self, + cfg: &RunnerConfig, + ) -> Vec { + let shutdown = self.shutdown_flag(); + crate::webcodex_runner::projects::load_runner_project_metadata(cfg, Some(shutdown.as_ref())) + } + + /// Registration, keepalive and the first authorized catalog must not wait + /// for serial Git probes. Reuse an in-flight scan across reconnects, so + /// only one scanner can run per process. The existing shutdown flag, Git + /// tree ownership, and bounded background-thread drain still own + /// cancellation and cleanup. pub(super) fn initial_project_scan( &self, cfg: &RunnerConfig, ) -> Result { + let metadata_runtime = self.clone(); + let metadata_cfg = cfg.clone(); let runtime = self.clone(); let cfg = cfg.clone(); - self.initial_project_scan_with(move || { - runtime.project_summaries(&mut RunnerProjectCache::default(), &cfg) - }) + self.initial_project_scan_with( + move || metadata_runtime.project_metadata_summaries(&metadata_cfg), + move || runtime.project_summaries(&mut RunnerProjectCache::default(), &cfg), + ) } pub(super) fn initial_project_scan_with( &self, + mut scan_metadata: impl FnMut() -> Vec + Send + 'static, mut scan_projects: impl FnMut() -> Vec + Send + 'static, ) -> Result { let mut scan = self @@ -114,12 +133,15 @@ impl RunnerRuntimeState { return Err("project_inventory_scan_stopped".to_string()); } if let Some(receiver) = scan.receiver.as_ref() { - if receiver.borrow().is_none() && receiver.has_changed().is_ok() { + let complete = receiver + .borrow() + .as_ref() + .is_some_and(|publication| publication.complete); + if !complete && receiver.has_changed().is_ok() { return Ok(receiver.clone()); } } self.background_threads.reap_finished(); - scan.invalidated = false; let (sender, receiver) = tokio::sync::watch::channel(None); let runtime = self.clone(); let handle = std::thread::Builder::new() @@ -128,24 +150,20 @@ impl RunnerRuntimeState { if runtime.shutdown_requested() { break; } - let projects = scan_projects(); - let mut scan = runtime - .initial_project_scan - .lock() - .unwrap_or_else(|e| e.into_inner()); - if runtime.shutdown_requested() { - break; + let generation = runtime.project_inventory_generation(); + let metadata = scan_metadata(); + match runtime.publish_project_scan(&sender, generation, metadata, false) { + ProjectScanPublishOutcome::Published => {} + ProjectScanPublishOutcome::Stale => continue, + ProjectScanPublishOutcome::Stopped => break, } - if scan.invalidated { - // Keep the one owned worker, but discard a catalog observed - // before a Project mutation. Reconnects share this receiver. - scan.invalidated = false; - continue; + let projects = scan_projects(); + match runtime.publish_project_scan(&sender, generation, projects, true) { + ProjectScanPublishOutcome::Stale => continue, + ProjectScanPublishOutcome::Published | ProjectScanPublishOutcome::Stopped => { + break + } } - // Serialize publication with invalidation so an obsolete result - // cannot become the pending scan of a later connection. - let _ = sender.send(Some(Arc::new(projects))); - break; }) .map_err(|_| "project_inventory_scan_start_failed".to_string())?; self.register_background_thread(handle); @@ -153,6 +171,66 @@ impl RunnerRuntimeState { Ok(receiver) } + /// Serialize publication with invalidation so an obsolete catalog cannot + /// become the pending scan of a later connection. Reconnects share the + /// receiver, so a stale result restarts the one owned worker instead. + fn publish_project_scan( + &self, + sender: &tokio::sync::watch::Sender>, + generation: u64, + projects: Vec, + complete: bool, + ) -> ProjectScanPublishOutcome { + let scan = self + .initial_project_scan + .lock() + .unwrap_or_else(|e| e.into_inner()); + if self.shutdown_requested() { + return ProjectScanPublishOutcome::Stopped; + } + if scan.generation != generation { + return ProjectScanPublishOutcome::Stale; + } + let _ = sender.send(Some(ProjectScanPublication { + projects: Arc::new(projects), + complete, + generation, + })); + ProjectScanPublishOutcome::Published + } + + fn project_inventory_generation(&self) -> u64 { + self.initial_project_scan + .lock() + .unwrap_or_else(|e| e.into_inner()) + .generation + } + + /// Wait for a catalog this stream has not published. A catalog read before + /// the latest mutation stays unpublished even when it is already buffered. + pub(super) async fn next_project_scan_publication( + &self, + receiver: &mut ProjectScanReceiver, + last: Option<&ProjectScanPublication>, + ) -> Result { + loop { + let current = receiver.borrow_and_update().clone(); + if let Some(publication) = current { + let already_published = + last.is_some_and(|last| Arc::ptr_eq(&last.projects, &publication.projects)); + if !already_published + && publication.generation == self.project_inventory_generation() + { + return Ok(publication); + } + } + receiver + .changed() + .await + .map_err(|_| "project_inventory_scan_failed")?; + } + } + pub(super) fn project_inventory_changes(&self) -> tokio::sync::watch::Receiver<()> { self.initial_project_scan .lock() @@ -166,7 +244,7 @@ impl RunnerRuntimeState { .initial_project_scan .lock() .unwrap_or_else(|e| e.into_inner()); - scan.invalidated = true; + scan.generation = scan.generation.wrapping_add(1); // Runtime ownership preserves completion notifications from dispatches // whose original stream has already disconnected. Watch coalesces bursts. scan.changed.send_replace(()); diff --git a/crates/webcodex-runner/src/webcodex_runner/transport/stream.rs b/crates/webcodex-runner/src/webcodex_runner/transport/stream.rs index 2ef0ab457..018f7a085 100644 --- a/crates/webcodex-runner/src/webcodex_runner/transport/stream.rs +++ b/crates/webcodex-runner/src/webcodex_runner/transport/stream.rs @@ -276,6 +276,7 @@ where #[cfg(test)] InitialProjectInventory::Disabled => (None, None), }; + let mut last_scan_publication: Option = None; let mut project_inventory = if initial_scan.is_some() { StreamingProjectInventoryCoordinator::pending_scan() } else { @@ -294,13 +295,9 @@ where let project_inventory_retry_deadline = project_inventory_retry_at.unwrap_or_else(tokio::time::Instant::now); tokio::select! { - projects = async { + publication = async { let receiver = initial_scan.as_mut().expect("guarded pending scan"); - if receiver.borrow().is_none() { - receiver.changed().await.map_err(|_| "project_inventory_scan_failed")?; - } - let projects = receiver.borrow().clone().ok_or("project_inventory_scan_failed")?; - Ok::<_, &str>(projects) + runtime.next_project_scan_publication(receiver, last_scan_publication.as_ref()).await }, if initial_scan.is_some() => { // A result and a mutation can become ready together. Keep the // receiver pending until the known dirty notification replaces @@ -308,15 +305,28 @@ where if project_inventory_refresh_rx.has_changed().unwrap_or(false) { continue; } - initial_scan = None; - let projects = match projects { - Ok(projects) => projects, + let publication = match publication { + Ok(publication) => publication, + Err(_) if last_scan_publication.is_some() => { + // The catalog is already published; only its Git details + // are unavailable, and absent fields already mean unknown. + initial_scan = None; + last_scan_publication = None; + continue; + } Err(error) => { session_error = Some(error.to_string()); break; } }; - project_inventory = StreamingProjectInventoryCoordinator::new(Some(paged_sync_after_registration((*projects).clone()))); + let projects = (*publication.projects).clone(); + if publication.complete { + initial_scan = None; + last_scan_publication = None; + } else { + last_scan_publication = Some(publication); + } + project_inventory = StreamingProjectInventoryCoordinator::new(Some(paged_sync_after_registration(projects))); project_inventory.queue_pending(transport, &out_tx); } _ = tokio::time::sleep_until(project_inventory_retry_deadline), if project_inventory_retry_at.is_some() => { @@ -325,6 +335,15 @@ where refresh = project_inventory_refresh_rx.changed() => { if refresh.is_ok() { if initial_scan.is_some() { + if last_scan_publication.is_some() { + project_inventory.refresh_from_project_metadata( + transport, + cfg, + runtime, + &out_tx, + "project_inventory_local_project_mutation", + ); + } // Reuse the dirty worker, or replace an already completed // but unconsumed scan. Never run a second synchronous Git // scan on the stream thread while initial discovery waits. diff --git a/crates/webcodex-runner/src/webcodex_runner/transport_tests/initial_project_scan.rs b/crates/webcodex-runner/src/webcodex_runner/transport_tests/initial_project_scan.rs index e08782234..9e455fc3e 100644 --- a/crates/webcodex-runner/src/webcodex_runner/transport_tests/initial_project_scan.rs +++ b/crates/webcodex-runner/src/webcodex_runner/transport_tests/initial_project_scan.rs @@ -55,7 +55,13 @@ async fn initial_project_scan_does_not_block_registration_keepalive_or_paged_pub let expected = (0..65) .map(|index| synthetic_project_summary(index, None)) .collect::>(); - scan_tx.send(Some(Arc::new(expected.clone()))).unwrap(); + scan_tx + .send(Some(ProjectScanPublication { + projects: Arc::new(expected.clone()), + complete: true, + generation: 0, + })) + .unwrap(); let mut received = Vec::new(); let mut generation = None; loop { @@ -132,7 +138,7 @@ async fn initial_project_scan_worker_is_owned_drained_and_closed_after_shutdown( .unwrap() .unwrap(); } - assert!(receiver.borrow().as_ref().unwrap().is_empty()); + assert!(receiver.borrow().as_ref().unwrap().projects.is_empty()); assert_eq!(runtime.background_threads.pending(), 1); runtime.request_shutdown_signal(); assert!(runtime.initial_project_scan(&cfg).is_err()); @@ -244,6 +250,7 @@ struct HeldInitialProjectScan { receiver: ProjectScanReceiver, release: std::sync::mpsc::Sender<()>, scans: Arc, + generation: u64, } impl HeldInitialProjectScan { @@ -252,8 +259,8 @@ impl HeldInitialProjectScan { // thread. Wait for the real dirty fence before releasing the old scan. tokio::time::timeout(Duration::from_secs(3), async { loop { - let invalidated = runtime.initial_project_scan.lock().unwrap().invalidated; - if invalidated { + let generation = runtime.initial_project_scan.lock().unwrap().generation; + if generation != self.generation { break; } tokio::task::yield_now().await; @@ -276,16 +283,21 @@ async fn hold_initial_project_scan( let worker_scans = Arc::clone(&scans); let scan_runtime = runtime.clone(); let scan_cfg = cfg.clone(); + let enrich_runtime = runtime.clone(); + let enrich_cfg = cfg.clone(); + let generation = runtime.initial_project_scan.lock().unwrap().generation; let receiver = runtime - .initial_project_scan_with(move || { - let projects = - scan_runtime.project_summaries(&mut RunnerProjectCache::default(), &scan_cfg); - if worker_scans.fetch_add(1, Ordering::SeqCst) == 0 { - let _ = started_tx.take().unwrap().send(()); - let _ = release_rx.recv_timeout(Duration::from_secs(10)); - } - projects - }) + .initial_project_scan_with( + move || { + let projects = scan_runtime.project_metadata_summaries(&scan_cfg); + if worker_scans.fetch_add(1, Ordering::SeqCst) == 0 { + let _ = started_tx.take().unwrap().send(()); + let _ = release_rx.recv_timeout(Duration::from_secs(10)); + } + projects + }, + move || enrich_runtime.project_metadata_summaries(&enrich_cfg), + ) .unwrap(); tokio::time::timeout(Duration::from_secs(3), started_rx) .await @@ -295,6 +307,7 @@ async fn hold_initial_project_scan( receiver, release, scans, + generation, } } @@ -555,7 +568,7 @@ async fn initial_project_scan_completed_before_dirty_event_is_replaced_before_pu assert!(Instant::now() < deadline, "held scan did not complete"); std::thread::yield_now(); } - assert_eq!(scan.receiver.borrow().as_ref().unwrap().len(), 1); + assert_eq!(scan.receiver.borrow().as_ref().unwrap().projects.len(), 1); std::fs::remove_file(registry.join("project-0000.toml")).unwrap(); runtime.invalidate_project_inventory(); @@ -662,3 +675,351 @@ async fn initial_project_scan_old_stream_mutation_notifies_reconnected_stream() assert!(page.snapshot_sequence > before_page.snapshot_sequence); assert!(!registry.join("project-0000.toml").exists()); } + +struct HeldProjectEnrichment { + release: std::sync::mpsc::Sender<()>, + enrichments: Arc, +} + +/// Holds the first Git enrichment like a slow or unreachable repository, while +/// the metadata stage reads the real registry. +async fn hold_project_enrichment( + runtime: &RunnerRuntimeState, + cfg: &RunnerConfig, +) -> HeldProjectEnrichment { + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let mut started_tx = Some(started_tx); + let (release, release_rx) = std::sync::mpsc::channel(); + let enrichments = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let worker_enrichments = Arc::clone(&enrichments); + let metadata_runtime = runtime.clone(); + let metadata_cfg = cfg.clone(); + let enrich_runtime = runtime.clone(); + let enrich_cfg = cfg.clone(); + runtime + .initial_project_scan_with( + move || metadata_runtime.project_metadata_summaries(&metadata_cfg), + move || { + if worker_enrichments.fetch_add(1, Ordering::SeqCst) == 0 { + let _ = started_tx.take().unwrap().send(()); + let _ = release_rx.recv_timeout(Duration::from_secs(10)); + } + let mut projects = enrich_runtime.project_metadata_summaries(&enrich_cfg); + for project in &mut projects { + project.git_branch = Some("main".to_string()); + } + projects + }, + ) + .unwrap(); + tokio::time::timeout(Duration::from_secs(3), started_rx) + .await + .unwrap() + .unwrap(); + HeldProjectEnrichment { + release, + enrichments, + } +} + +async fn next_inventory_page( + ws: &mut tokio_tungstenite::WebSocketStream, +) -> webcodex_core::runner_protocol::ShellProjectInventoryPage { + tokio::time::timeout(Duration::from_secs(3), async { + loop { + let message = ws.next().await.unwrap().unwrap(); + match RunnerEnvelope::from_slice(message.into_text().unwrap().as_bytes()).unwrap() { + RunnerEnvelope::ProjectInventoryPage { page } => return page, + RunnerEnvelope::Result { payload } => { + assert_eq!(payload.result.exit_code, Some(0), "{payload:?}"); + } + other => panic!("expected inventory page, got {}", other.kind()), + } + } + }) + .await + .expect("inventory page must not wait for the held Git enrichment") +} + +fn project_ids(page: &webcodex_core::runner_protocol::ShellProjectInventoryPage) -> Vec<&str> { + page.projects + .iter() + .map(|project| project.id.as_str()) + .collect() +} + +fn registry_config(temp: &Path, listen: String, count: usize) -> (RunnerConfig, PathBuf) { + let registry = temp.join("project-registry"); + write_synthetic_project_configs(®istry, &temp.join("projects"), count); + let mut cfg = test_runner_config(listen); + cfg.project_registry_dir = Some(registry.clone()); + (cfg, registry) +} + +#[tokio::test] +async fn initial_inventory_publishes_catalog_before_git_enrichment_then_replaces_it() { + let temp = tempfile::tempdir().unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (cfg, _) = registry_config( + temp.path(), + format!("http://{}", listener.local_addr().unwrap()), + 2, + ); + let runtime = test_runtime(&cfg); + let scan = hold_project_enrichment(&runtime, &cfg).await; + let session_runtime = runtime.clone(); + let session = tokio::spawn(async move { + websocket_session_classified( + &cfg, + InitialProjectInventory::Scan, + "inst-test", + &session_runtime, + ) + .await + }); + let (socket, _) = listener.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(socket).await.unwrap(); + read_register(&mut ws).await; + send_registered_ack_only(&mut ws).await; + + let first = next_inventory_page(&mut ws).await; + assert_eq!(project_ids(&first), ["project-0000", "project-0001"]); + assert!(first.complete); + for project in &first.projects { + assert_eq!(project.git_branch, None, "unknown until enrichment"); + assert_eq!(project.git_head, None); + assert_eq!(project.git_dirty, None, "pending must not read as clean"); + } + assert_eq!(scan.enrichments.load(Ordering::SeqCst), 1); + + scan.release.send(()).unwrap(); + let enriched = next_inventory_page(&mut ws).await; + assert_ne!(enriched.generation, first.generation); + assert_eq!(project_ids(&enriched), ["project-0000", "project-0001"]); + assert!(enriched + .projects + .iter() + .all(|project| project.git_branch.as_deref() == Some("main"))); + + runtime.request_shutdown_signal(); + let _ = ws.close(None).await; + let _ = tokio::time::timeout(Duration::from_secs(3), session).await; + assert!(runtime.shutdown().timed_out_phases.is_empty()); +} + +#[tokio::test] +async fn mutation_during_git_enrichment_republishes_catalog_and_discards_stale_result() { + let temp = tempfile::tempdir().unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (cfg, registry) = registry_config( + temp.path(), + format!("http://{}", listener.local_addr().unwrap()), + 2, + ); + let runtime = test_runtime(&cfg); + let scan = hold_project_enrichment(&runtime, &cfg).await; + let session_cfg = cfg.clone(); + let session_runtime = runtime.clone(); + let session = tokio::spawn(async move { + websocket_session_classified( + &session_cfg, + InitialProjectInventory::Scan, + "inst-test", + &session_runtime, + ) + .await + }); + let (socket, _) = listener.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(socket).await.unwrap(); + read_register(&mut ws).await; + send_registered_ack_only(&mut ws).await; + let first = next_inventory_page(&mut ws).await; + assert_eq!(first.total_reported, 2); + + let request = initial_scan_unregister_request(&cfg, &first.projects[0], "unregister-enriching"); + ws.send(WsMessage::Text( + RunnerEnvelope::Request { request } + .to_json() + .unwrap() + .into(), + )) + .await + .unwrap(); + // The held enrichment has not been released: this correction cannot have + // come from the worker. + let corrected = next_inventory_page(&mut ws).await; + assert_eq!(project_ids(&corrected), ["project-0001"]); + assert!(!registry.join("project-0000.toml").exists()); + assert_eq!(scan.enrichments.load(Ordering::SeqCst), 1); + + scan.release.send(()).unwrap(); + let enriched = tokio::time::timeout(Duration::from_secs(3), async { + loop { + let page = next_inventory_page(&mut ws).await; + assert_eq!( + project_ids(&page), + ["project-0001"], + "the result read before the mutation must never be published" + ); + if page.projects[0].git_branch.is_some() { + return page; + } + } + }) + .await + .unwrap(); + assert_eq!(enriched.total_reported, 1); + assert_eq!(scan.enrichments.load(Ordering::SeqCst), 2); + + runtime.request_shutdown_signal(); + let _ = ws.close(None).await; + let _ = tokio::time::timeout(Duration::from_secs(3), session).await; + assert!(runtime.shutdown().timed_out_phases.is_empty()); +} + +#[tokio::test] +async fn reconnect_during_git_enrichment_gets_catalog_and_shares_the_one_worker() { + let temp = tempfile::tempdir().unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (cfg, _) = registry_config( + temp.path(), + format!("http://{}", listener.local_addr().unwrap()), + 1, + ); + let runtime = test_runtime(&cfg); + let scan = hold_project_enrichment(&runtime, &cfg).await; + + for attempt in 0..2 { + let session_cfg = cfg.clone(); + let session_runtime = runtime.clone(); + let session = tokio::spawn(async move { + websocket_session_classified( + &session_cfg, + InitialProjectInventory::Scan, + "inst-test", + &session_runtime, + ) + .await + }); + let (socket, _) = listener.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(socket).await.unwrap(); + read_register(&mut ws).await; + send_registered_ack_only(&mut ws).await; + let page = next_inventory_page(&mut ws).await; + assert_eq!(project_ids(&page), ["project-0000"], "attempt {attempt}"); + assert_eq!(page.projects[0].git_branch, None); + assert_eq!(scan.enrichments.load(Ordering::SeqCst), 1); + assert_eq!(runtime.background_threads.pending(), 1); + if attempt == 0 { + ws.close(None).await.unwrap(); + let _ = tokio::time::timeout(Duration::from_secs(3), session).await; + continue; + } + scan.release.send(()).unwrap(); + let enriched = next_inventory_page(&mut ws).await; + assert_eq!(enriched.projects[0].git_branch.as_deref(), Some("main")); + runtime.request_shutdown_signal(); + let _ = ws.close(None).await; + let _ = tokio::time::timeout(Duration::from_secs(3), session).await; + } + assert!(runtime.shutdown().timed_out_phases.is_empty()); +} + +#[tokio::test] +async fn shutdown_during_git_enrichment_ends_the_session_without_a_scan_error() { + let temp = tempfile::tempdir().unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (cfg, _) = registry_config( + temp.path(), + format!("http://{}", listener.local_addr().unwrap()), + 1, + ); + let runtime = test_runtime(&cfg); + let scan = hold_project_enrichment(&runtime, &cfg).await; + let session_runtime = runtime.clone(); + let session = tokio::spawn(async move { + super::supervisor::supervise_stream_transports( + &cfg, + false, + "inst-test", + &session_runtime, + StreamSupervisorMode::Strict(StreamTransport::WebSocket), + ) + .await + }); + let (socket, _) = listener.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(socket).await.unwrap(); + read_register(&mut ws).await; + send_registered_ack_only(&mut ws).await; + next_inventory_page(&mut ws).await; + + runtime.request_shutdown_signal(); + let goodbye = tokio::time::timeout(Duration::from_secs(3), ws.next()) + .await + .expect("a held Git probe must not delay shutdown") + .unwrap() + .unwrap(); + assert!(matches!( + RunnerEnvelope::from_slice(goodbye.into_text().unwrap().as_bytes()).unwrap(), + RunnerEnvelope::Goodbye { .. } + )); + // Real Git probes observe the shutdown flag; release this fixture's hold + // only after proving the transport did not wait for it. + scan.release.send(()).unwrap(); + ws.close(None).await.unwrap(); + tokio::time::timeout(Duration::from_secs(3), session) + .await + .unwrap() + .unwrap() + .unwrap(); + let report = runtime.shutdown(); + assert!(report.timed_out_phases.is_empty()); + assert_eq!(runtime.background_threads.pending(), 0); +} + +#[test] +fn project_metadata_keeps_authority_facts_and_leaves_git_unknown() { + let temp = tempfile::tempdir().unwrap(); + let (cfg, _) = registry_config(temp.path(), "http://127.0.0.1:1".to_string(), 2); + let repository = temp.path().join("projects").join("project-0000"); + std::fs::create_dir_all(&repository).unwrap(); + let git = |args: &[&str]| { + let status = std::process::Command::new("git") + .args(["-c", "user.name=t", "-c", "user.email=t@example.invalid"]) + .args(args) + .current_dir(&repository) + .status() + .unwrap(); + assert!(status.success(), "git {args:?}"); + }; + git(&["init", "-q", "-b", "trunk"]); + git(&["commit", "-q", "--allow-empty", "-m", "first"]); + let runtime = test_runtime(&cfg); + + let metadata = runtime.project_metadata_summaries(&cfg); + let full = runtime.project_summaries(&mut RunnerProjectCache::default(), &cfg); + + assert_eq!(metadata.len(), 2); + assert_eq!(metadata[0].git_branch, None); + assert_eq!(metadata[0].git_head, None); + assert_eq!(metadata[0].git_dirty, None); + assert_eq!(full[0].git_branch.as_deref(), Some("trunk")); + assert_eq!(full[0].git_dirty, Some(false)); + assert!(metadata[0].root_fingerprint.is_some()); + assert_eq!( + metadata[1].root_fingerprint, None, + "an unresolved root must not be reported as a confirmed identity" + ); + for (metadata, full) in metadata.iter().zip(&full) { + let mut comparable = full.clone(); + comparable.git_branch = None; + comparable.git_head = None; + comparable.git_dirty = None; + comparable.updated_at = metadata.updated_at; + assert_eq!( + serde_json::to_value(metadata).unwrap(), + serde_json::to_value(&comparable).unwrap() + ); + } + runtime.shutdown(); +} diff --git a/docs/RUNNER.md b/docs/RUNNER.md index fcba6f8b7..11f2f9dac 100644 --- a/docs/RUNNER.md +++ b/docs/RUNNER.md @@ -65,13 +65,15 @@ When upgrading an older installation across the 0.4 boundary, upgrade the first- The exact protocol-generation field names, baseline capability list, registration grammar, and compatibility-test matrix are maintainer/wire-contract details and are intentionally omitted from this operations guide. -Streaming Runners register before gathering project Git metadata. The initial -project catalog is scanned in one process-owned background worker, then sent -through the usual acknowledged inventory pages. Registration, keepalive and -shutdown remain responsive while a large or slow catalog is being inspected; -project inventory can still be pending just after the Runner connects. Reconnects -reuse an in-flight scan, and process shutdown cancels its Git probes through the -existing owned-tree cleanup path. +Streaming Runners register before gathering project Git metadata. One +process-owned background worker first publishes the validated project catalog +through the usual acknowledged inventory pages, then replaces it once Git details +are known. Until then a Project's `git_branch`, `git_head` and `git_dirty` are +absent, which means unknown, not clean. Registration, keepalive and shutdown +remain responsive while a large or slow catalog is being inspected. A Project +mutation republishes the catalog immediately and restarts enrichment, so an older +result is never published. Reconnects reuse an in-flight scan, and process +shutdown cancels its Git probes through the existing owned-tree cleanup path. A ChatGPT Host message that the current conversation does not support developer MCPs is not a Runner heartbeat or reconnect result. If ChatGPT cannot dispatch