From 3ec817986b1b419d4d448b7f80a0ea8b80633e95 Mon Sep 17 00:00:00 2001 From: HarpePluie <320289713+HarpeLm@users.noreply.github.com> Date: Tue, 10 Nov 2026 08:22:54 +0100 Subject: [PATCH] feat(runner): publish the project catalog before Git enrichment The first inventory waited for serial Git probes of every saved project, so one slow repository delayed the whole list after each Runner start. The one owned scan worker now publishes the validated catalog first, with Git fields absent to mean unknown, then replaces it once enrichment completes. Each publication records the inventory generation it was read under, so a result from before a Project mutation is never published, and a mutation during enrichment republishes the catalog immediately. Fixes #1271 --- .../src/webcodex_runner/projects.rs | 1 + .../src/webcodex_runner/projects/catalog.rs | 27 +- .../src/webcodex_runner/transport.rs | 2 +- .../transport/project_inventory.rs | 45 +- .../src/webcodex_runner/transport/runtime.rs | 130 ++++-- .../src/webcodex_runner/transport/stream.rs | 39 +- .../transport_tests/initial_project_scan.rs | 389 +++++++++++++++++- docs/RUNNER.md | 16 +- 8 files changed, 584 insertions(+), 65 deletions(-) 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