diff --git a/.repository-projection.json b/.repository-projection.json index 90f503bff..6059c438a 100644 --- a/.repository-projection.json +++ b/.repository-projection.json @@ -3,11 +3,11 @@ "projection": "deixic-code", "projectionSchemaVersion": 1, "sourceRepository": "dx-corp/mono", - "sourceSha": "028183748c33b32f658596ed0f96ef1c60424961", + "sourceSha": "43eb0edd715c489c97e0fb1fe22f52a44dbcfd95", "destinationRepository": "dx-corp/code", - "priorProjectedBase": "a2b3e4a7f7f38bf2b195b8402bf5b804450a0486", + "priorProjectedBase": "44c5b0b9c35bc88ea604bae376b1a1b9c4d0d7a5", "definitionDigest": "82936441c776e3e8edb5d215a75007ec9714a233f489d460075d79d5ef5ba32f", "toolDigest": "45502ff0478e541d02f34cc39ac332935f8d3c0030fa221afe5bcc3b5e51f88e", - "contentDigest": "e6e93ea0769f4bd63e5aea305914b79869813d5138b1161347463c2055779d3d", + "contentDigest": "fd67fb0c9b781bf5e60fd477ccb05c531bb038c0c4bcb52ccf50947d8c5bcd1e", "publicationEligible": true } diff --git a/packages/local-host-rs/src/hosted_runner.rs b/packages/local-host-rs/src/hosted_runner.rs index 040cce809..0186890ad 100644 --- a/packages/local-host-rs/src/hosted_runner.rs +++ b/packages/local-host-rs/src/hosted_runner.rs @@ -89,6 +89,10 @@ pub(super) const HOSTED_RUNNER_DRAIN_FINALIZATION_PENDING_STATUS: &str = const DEFAULT_HEARTBEAT_INTERVAL_MS: u64 = 15_000; const CONNECTION_IDLE_MS: i64 = (DEFAULT_HEARTBEAT_INTERVAL_MS as i64) * 3; const MAINTENANCE_PUMP_INTERVAL: Duration = Duration::from_millis(100); +// Connected transports notify the pump after enqueueing every event. Keep a +// slower safety drain for missed notifications without waking 10 times a +// second throughout an otherwise idle hosted session. +const NOTIFIED_MAINTENANCE_PUMP_INTERVAL: Duration = Duration::from_secs(5); const THREAD_PERSISTENCE_RECOVERY_RETRY_DELAY: Duration = Duration::from_millis(100); const MAX_EVENTS: usize = 1024; // Response retries are short-lived transport retries; retain enough completed @@ -579,6 +583,12 @@ pub trait HostedRunnerHeadlessMessageExecutor: Send + Sync { Ok(None) } + /// Pending response acknowledgements and failed ledger writes need the + /// short maintenance cadence even when transport events are notified. + fn has_pending_maintenance_work(&self) -> bool { + false + } + /// Report whether a runtime that was previously connected has lost its /// transport. Hosted mode owns one child generation, so the event pump /// treats this as terminal while leaving `drain` available for export. @@ -1135,6 +1145,24 @@ impl HostedRunnerHeadlessMessageExecutor for AgentSupervisorHostedRunnerMessageE Ok(supervisor.event_notification()) } + fn has_pending_maintenance_work(&self) -> bool { + !self + .queued_responses + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_empty() + || !self + .queued_unkeyed_responses + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_empty() + || !self + .memory_completed_responses + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_empty() + } + fn disconnected_after_ready(&self) -> Result { let supervisor = self .supervisor @@ -1982,40 +2010,50 @@ async fn pump_agent_events( cancelled: CancellationToken, maintenance_interval: Duration, ) { - let mut interval = tokio::time::interval(maintenance_interval); + let mut first_tick = true; loop { - let should_pump = tokio::select! { - () = cancelled.cancelled() => break, - _ = interval.tick() => true, - result = wait_for_event_notification(&shared) => { - if let Err(error) = result { - shared.publish_runtime_error("event_pump_failed", error); - break; - } - true + let notification = match shared.message_executor.event_notification() { + Ok(notification) => notification, + Err(error) => { + shared.publish_runtime_error("event_pump_failed", error); + break; } }; + let can_use_safety_tick = notification.is_some() + && !shared + .thread_persistence_retry_pending + .load(Ordering::Acquire) + && !shared.message_executor.has_pending_maintenance_work(); + let tick_interval = if first_tick { + Duration::ZERO + } else if can_use_safety_tick { + NOTIFIED_MAINTENANCE_PUMP_INTERVAL.max(maintenance_interval) + } else { + maintenance_interval + }; + first_tick = false; + tokio::select! { + () = cancelled.cancelled() => break, + () = tokio::time::sleep(tick_interval) => {}, + () = async { + if let Some(notification) = notification { + notification.notified().await; + } else { + std::future::pending::<()>().await; + } + } => {}, + } let lifecycle = shared.mutation_lifecycle.clone(); let _lifecycle = tokio::select! { () = cancelled.cancelled() => break, lifecycle = lifecycle.lock() => lifecycle, }; - if should_pump && matches!(pump_tick(&shared), PumpTick::Stop) { + if matches!(pump_tick(&shared), PumpTick::Stop) { break; } } } -async fn wait_for_event_notification(shared: &SharedRunner) -> Result<(), HostedRunnerError> { - let notification = shared.message_executor.event_notification()?; - if let Some(notification) = notification { - notification.notified().await; - Ok(()) - } else { - std::future::pending().await - } -} - #[derive(Debug, PartialEq, Eq)] enum PumpTick { Continue, diff --git a/packages/local-host-rs/src/hosted_runner/tests.rs b/packages/local-host-rs/src/hosted_runner/tests.rs index 1092417bf..aecd14978 100644 --- a/packages/local-host-rs/src/hosted_runner/tests.rs +++ b/packages/local-host-rs/src/hosted_runner/tests.rs @@ -2592,6 +2592,9 @@ impl HostedRunnerHeadlessMessageExecutor for PumpOnlyRuntimeExecutor { struct NotifyingPumpRuntimeExecutor { queued: Arc>>, notification: Arc, + drain_count: AtomicUsize, + expose_notification: bool, + pending_maintenance: AtomicBool, } impl Default for NotifyingPumpRuntimeExecutor { @@ -2599,6 +2602,9 @@ impl Default for NotifyingPumpRuntimeExecutor { Self { queued: Arc::new(Mutex::new(Vec::new())), notification: Arc::new(Notify::new()), + drain_count: AtomicUsize::new(0), + expose_notification: true, + pending_maintenance: AtomicBool::new(false), } } } @@ -2631,6 +2637,7 @@ impl HostedRunnerHeadlessMessageExecutor for NotifyingPumpRuntimeExecutor { } fn drain(&self) -> Result { + self.drain_count.fetch_add(1, Ordering::Relaxed); Ok(HostedRunnerDrainResult { messages: std::mem::take(&mut *self.queued.lock().expect("queued ready event")), consumed_response_keys: Vec::new(), @@ -2639,7 +2646,13 @@ impl HostedRunnerHeadlessMessageExecutor for NotifyingPumpRuntimeExecutor { } fn event_notification(&self) -> Result>, HostedRunnerError> { - Ok(Some(Arc::clone(&self.notification))) + Ok(self + .expose_notification + .then(|| Arc::clone(&self.notification))) + } + + fn has_pending_maintenance_work(&self) -> bool { + self.pending_maintenance.load(Ordering::Relaxed) } } @@ -7776,6 +7789,160 @@ async fn transport_notification_publishes_ready_before_maintenance_tick() { handle.shutdown().await; } +#[tokio::test(start_paused = true)] +async fn notified_event_pump_has_a_bounded_idle_drain_rate() { + let workspace = tempdir().expect("workspace"); + let executor = Arc::new(NotifyingPumpRuntimeExecutor::default()); + let shared = SharedRunner::new_with_message_executor_and_restore( + test_config(workspace.path().to_path_buf()), + executor.clone(), + None, + ); + let cancelled = CancellationToken::new(); + let pump = tokio::spawn(pump_agent_events( + shared, + cancelled.clone(), + MAINTENANCE_PUMP_INTERVAL, + )); + tokio::task::yield_now().await; + let initial_drains = executor.drain_count.load(Ordering::Relaxed); + assert_eq!(initial_drains, 1, "the pump must drain on startup"); + + // Advance in small steps so Tokio cannot collapse a missed interval into + // one callback. This measures the work a real idle minute would schedule. + for _ in 0..600 { + tokio::time::advance(Duration::from_millis(100)).await; + tokio::task::yield_now().await; + } + let idle_drains = executor.drain_count.load(Ordering::Relaxed) - initial_drains; + assert!( + idle_drains <= 12, + "notified transport drained {idle_drains} times in one idle minute" + ); + + executor.queue_ready_event(); + tokio::task::yield_now().await; + assert_eq!( + executor.drain_count.load(Ordering::Relaxed), + initial_drains + idle_drains + 1, + "a real event must wake the pump without waiting for maintenance" + ); + cancelled.cancel(); + pump.await.expect("pump task"); +} + +#[tokio::test(start_paused = true)] +async fn executor_without_notification_keeps_short_maintenance_fallback() { + let workspace = tempdir().expect("workspace"); + let executor = Arc::new(NotifyingPumpRuntimeExecutor { + expose_notification: false, + ..Default::default() + }); + let shared = SharedRunner::new_with_message_executor_and_restore( + test_config(workspace.path().to_path_buf()), + executor.clone(), + None, + ); + let cancelled = CancellationToken::new(); + let pump = tokio::spawn(pump_agent_events( + shared, + cancelled.clone(), + MAINTENANCE_PUMP_INTERVAL, + )); + tokio::task::yield_now().await; + let initial_drains = executor.drain_count.load(Ordering::Relaxed); + assert_eq!(initial_drains, 1); + for _ in 0..10 { + tokio::time::advance(Duration::from_millis(100)).await; + tokio::task::yield_now().await; + } + assert_eq!( + executor.drain_count.load(Ordering::Relaxed), + initial_drains + 10 + ); + cancelled.cancel(); + pump.await.expect("pump task"); +} + +#[tokio::test(start_paused = true)] +async fn notified_event_pump_keeps_short_retry_while_work_is_pending() { + let workspace = tempdir().expect("workspace"); + let executor = Arc::new(NotifyingPumpRuntimeExecutor::default()); + executor.pending_maintenance.store(true, Ordering::Relaxed); + let shared = SharedRunner::new_with_message_executor_and_restore( + test_config(workspace.path().to_path_buf()), + executor.clone(), + None, + ); + let cancelled = CancellationToken::new(); + let pump = tokio::spawn(pump_agent_events( + shared, + cancelled.clone(), + MAINTENANCE_PUMP_INTERVAL, + )); + tokio::task::yield_now().await; + let initial_drains = executor.drain_count.load(Ordering::Relaxed); + assert_eq!(initial_drains, 1); + for _ in 0..10 { + tokio::time::advance(Duration::from_millis(100)).await; + tokio::task::yield_now().await; + } + assert_eq!( + executor.drain_count.load(Ordering::Relaxed), + initial_drains + 10, + "pending acknowledgements and persistence retries need the 100 ms cadence" + ); + cancelled.cancel(); + pump.await.expect("pump task"); +} + +#[cfg(target_os = "linux")] +#[tokio::test] +#[ignore = "manual one-minute process CPU sample; use --ignored --nocapture"] +async fn notified_event_pump_idle_cpu_probe() { + fn process_cpu() -> Duration { + let mut clock = std::mem::MaybeUninit::::uninit(); + // SAFETY: clock points to writable timespec storage and the clock ID + // is a Linux process CPU clock. + assert_eq!( + unsafe { libc::clock_gettime(libc::CLOCK_PROCESS_CPUTIME_ID, clock.as_mut_ptr()) }, + 0, + "read process CPU clock" + ); + // SAFETY: a successful clock_gettime initialized the whole timespec. + let clock = unsafe { clock.assume_init() }; + Duration::new(clock.tv_sec as u64, clock.tv_nsec as u32) + } + + let workspace = tempdir().expect("workspace"); + let executor = Arc::new(NotifyingPumpRuntimeExecutor::default()); + let shared = SharedRunner::new_with_message_executor_and_restore( + test_config(workspace.path().to_path_buf()), + executor.clone(), + None, + ); + let cancelled = CancellationToken::new(); + let pump = tokio::spawn(pump_agent_events( + shared, + cancelled.clone(), + MAINTENANCE_PUMP_INTERVAL, + )); + tokio::task::yield_now().await; + let before_drains = executor.drain_count.load(Ordering::Relaxed); + let before_cpu = process_cpu(); + tokio::time::sleep(Duration::from_mins(1)).await; + let cpu = process_cpu() + .checked_sub(before_cpu) + .expect("process CPU clock is monotonic"); + let drains = executor.drain_count.load(Ordering::Relaxed) - before_drains; + println!( + "idle_60s_process_cpu_ms={} drain_calls={drains}", + cpu.as_millis() + ); + cancelled.cancel(); + pump.await.expect("pump task"); +} + #[tokio::test] async fn event_pump_publishes_agent_events_without_a_follow_up_http_request() { let workspace = tempdir().expect("workspace");