fix(db): Make idle broker polls reader-only and notify-driven (eliminates poll heartbeat UPDATEs on idle)

crates/preloop-runner-server/src/bootstrap.rs

@@ -20,8 +20,9 @@ pub struct ServerConfig {
     pub record_flows: Option<PathBuf>,
     /// TLS mode (default: no TLS).
     pub tls: TlsMode,
-    /// Shared counter published with the number of jobs still queued after
-    /// each claim. Supply one to let a co-hosted runner pool scale to demand.
+    /// Shared counter with the number of jobs still queued, refreshed by the
+    /// 5s state sampler. Supply one to let a co-hosted runner pool scale to
+    /// demand.
     pub queue_depth: Option<Arc<std::sync::atomic::AtomicUsize>>,
     /// Shared list, refreshed after each claim, of the `runs-on` labels of
     /// the job at the front of the dispatch queue. Supply one to let a
@@ -483,6 +484,7 @@ pub async fn reap_once(shared: &Arc<SharedState>) {
     // Notify if cancellations or starvation failures occurred
     if cancellation_count > 0 || !starved.is_empty() || !expired_fork_approvals.is_empty() {
         shared.state.message_notify.notify_waiters();
+        shared.state.sampler_notify.notify_waiters();
     }
 
     // Surface fork-PR runs failed closed by the expired approval window.
@@ -595,16 +597,14 @@ pub async fn reap_once(shared: &Arc<SharedState>) {
         Ok(outcome) => {
             // Post-commit: refresh the node-local mirrors the runner
             // supervisor and the pool read, then wake them if the sweep
-            // changed what is schedulable.
-            shared
-                .state
-                .queue_depth
-                .store(outcome.queue_depth, std::sync::atomic::Ordering::Release);
+            // changed what is schedulable. The ready-queue depth itself now
+            // comes from the 5s sampler snapshot.
             if let Ok(mut guard) = shared.state.next_job_runs_on.write() {
                 *guard = outcome.next_runs_on;
             }
             if outcome.promoted > 0 || outcome.failed > 0 {
                 shared.state.message_notify.notify_waiters();
+                shared.state.sampler_notify.notify_waiters();
             }
         }
         Err(error) => warn!(?error, "environment gate promotion sweep failed"),
@@ -1261,54 +1261,104 @@ async fn run_state_sampler(
     );
     heartbeat.beat("state_sampler");
     let mut interval = tokio::time::interval(Duration::from_secs(5));
+    // A tick missed during a coalescing nap delays rather than bursting:
+    // publishing twice back-to-back reports the same counters twice.
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
     // Immediate sample then every 5s.
     interval.tick().await;
+    // When the last tick ran: the burst-reactive early tick below must not
+    // turn a submit storm back into per-operation counting.
+    let mut last_tick = std::time::Instant::now();
     loop {
+        let mut early = false;
+        let mut shutdown = false;
         tokio::select! {
-            _ = interval.tick() => {
-                heartbeat.beat("state_sampler");
-                publish_snapshot(&shared, &store_backend, false).await;
-                // Record pool/queue gauges into OTel instruments so `/metrics`
-                // has a single exposition source (the SDK renderer).
-                 {
-                     let s = shared.state.status_snapshot.read();
-                     shared.state.observability.metrics().pool.record(
-                         s.service.uptime_seconds,
-                         s.pool.desired as u64,
-                         s.pool.preparing,
-                         s.pool.idle as u64,
-                         s.pool.busy as u64,
-                         s.jobs.ready as u64,
-                         s.jobs.claimable as u64,
-                         s.jobs.unclaimable as u64,
-                         s.jobs.dependency_blocked as u64,
-                     );
-                     // Host memory reality on the same cadence: a /proc scan
-                     // is microseconds next to everything else on this tick.
-                     // Recorded even when nothing else changed so OOM
-                     // proximity is a continuous signal, not a sampled one.
-                     shared
-                         .state
-                         .observability
-                         .metrics()
-                         .host
-                         .record(&preloop_observability::vm_telemetry::sample_host());
-                 }
-            }
-            _ = shared.shutdown.cancelled() => {
-                // Publish one last snapshot with the shutdown flag set so
-                // /api/v1/status reports `overall: shutting_down` and
-                // `shutdown_requested: true` while /healthz//readyz already
-                // 503 — without this the flag would only land on the next 5s
-                // tick that never comes.
-                heartbeat.beat("state_sampler");
-                publish_snapshot(&shared, &store_backend, true).await;
-                break;
+            _ = interval.tick() => {}
+            // A submit/claim/complete/cancel wakes the sampler early so the
+            // pool sees a burst in ~ms instead of at the next 5s tick. This
+            // is the sampler's own channel: runner waiters on
+            // `message_notify` are woken with per-job `notify_one` permits
+            // that a parked sampler here would otherwise steal.
+            _ = shared.state.sampler_notify.notified() => { early = true; }
+            _ = shared.shutdown.cancelled() => { shutdown = true; }
+        }
+        if shutdown {
+            // Publish one last snapshot with the shutdown flag set so
+            // /api/v1/status reports `overall: shutting_down` and
+            // `shutdown_requested: true` while /healthz//readyz already
+            // 503 — without this the flag would only land on the next 5s
+            // tick that never comes.
+            heartbeat.beat("state_sampler");
+            publish_snapshot(&shared, &store_backend, true).await;
+            break;
+        }
+        if early
+            && let Some(nap) = Duration::from_secs(1).checked_sub(last_tick.elapsed())
+            && !nap.is_zero()
+        {
+            // Inside the 1s floor: don't drop the event — nap off the rest
+            // of the floor, then sample below. A `continue` here meant a
+            // change during the floor was invisible until the next 5s tick.
+            // Wakes during the nap need no flag: the sample that follows
+            // counts everything already.
+            tokio::select! {
+                _ = tokio::time::sleep(nap) => {}
+                _ = shared.shutdown.cancelled() => {
+                    heartbeat.beat("state_sampler");
+                    publish_snapshot(&shared, &store_backend, true).await;
+                    break;
+                }
             }
         }
+        last_tick = std::time::Instant::now();
+        heartbeat.beat("state_sampler");
+        sample_state_once(&shared, &store_backend).await;
     }
 }
 
+/// One sampler tick: publish the snapshot, then refresh the pool/queue
+/// gauges and host-memory metrics from it. The loop calls this on every
+/// tick/early-wake; tests call it directly (`test_sample_state_once`) since
+/// no sampler task runs under `app()`.
+pub(crate) async fn sample_state_once(
+    shared: &Arc<SharedState>,
+    store_backend: &preloop_observability::status::StoreBackend,
+) {
+    publish_snapshot(shared, store_backend, false).await;
+    // Record pool/queue gauges into OTel instruments so `/metrics` has a
+    // single exposition source (the SDK renderer).
+    let s = shared.state.status_snapshot.read();
+    // The co-hosted runner pool scales off this atomic; it used to be
+    // refreshed by every submit/claim/complete (a full count(*) each). The
+    // sampler's grouped count is the same number at a fixed cadence, plus a
+    // burst-reactive early tick in the loop above.
+    shared
+        .state
+        .queue_depth
+        .store(s.jobs.ready as usize, std::sync::atomic::Ordering::Release);
+    shared.state.pool_status.set_queue_depth(s.jobs.ready);
+    shared.state.observability.metrics().pool.record(
+        s.service.uptime_seconds,
+        s.pool.desired as u64,
+        s.pool.preparing,
+        s.pool.idle as u64,
+        s.pool.busy as u64,
+        s.jobs.ready as u64,
+        s.jobs.claimable as u64,
+        s.jobs.unclaimable as u64,
+        s.jobs.dependency_blocked as u64,
+    );
+    // Host memory reality on the same cadence: a /proc scan is microseconds
+    // next to everything else on this tick. Recorded even when nothing else
+    // changed so OOM proximity is a continuous signal, not a sampled one.
+    shared
+        .state
+        .observability
+        .metrics()
+        .host
+        .record(&preloop_observability::vm_telemetry::sample_host());
+}
+
 fn is_routine_unix_disconnect(error: &(dyn std::error::Error + 'static)) -> bool {
     error
         .downcast_ref::<hyper::Error>()
@@ -1400,19 +1450,11 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> {
     }
     let queue = state.backend.queue_stats().await.unwrap_or_default();
     if let Some(queue_depth) = config.queue_depth.clone() {
+        // A co-hosted pool shares this atomic to scale to demand. The state
+        // sampler refreshes it every 5s (immediate first tick), and the
+        // constructor already seeded it from the recovered store — no
+        // per-operation re-arm needed.
         state.queue_depth = queue_depth;
-        // The pool shares this same atomic and only forks a runner while it
-        // is non-zero; a freshly restarted server has no runners yet, so
-        // nothing will refresh it from a broker poll. Re-arm it with the
-        // ready-queue size recovered from the store, or every job queued
-        // before the restart sits forever with the pool asleep.
-        let queue_len = queue.ready;
-        state
-            .queue_depth
-            .store(queue_len, std::sync::atomic::Ordering::Release);
-        state
-            .pool_status
-            .set_queue_depth(state.queue_depth.load(std::sync::atomic::Ordering::Acquire) as u32);
     }
     if let Some(next_job_runs_on) = config.next_job_runs_on.clone() {
         state.next_job_runs_on = next_job_runs_on;

crates/preloop-runner-server/src/broker.rs

@@ -288,7 +288,10 @@ pub async fn next_message_broker_ref(
     let wait_seconds = params
         .get("waitSeconds")
         .and_then(|value| value.parse::<u64>().ok())
-        .unwrap_or(50);
+        .unwrap_or(50)
+        // A client-chosen window longer than the liveness floor parks a
+        // healthy runner past its own reaping.
+        .min(crate::state::max_poll_window_secs());
     let runner_busy = params
         .get("status")
         .is_some_and(|status| status.eq_ignore_ascii_case("busy"));
@@ -322,6 +325,13 @@ pub async fn next_message_broker_ref(
     let deadline = tokio::time::Instant::now() + Duration::from_secs(wait_seconds);
 
     loop {
+        // Register the waiter *before* probing: a `notify_waiters` landing
+        // between the probe and a later `notified()` registration is lost
+        // forever (it stores no permit), stalling this poll until the
+        // window ends even though work was queued.
+        let notified = shared.state.message_notify.notified();
+        tokio::pin!(notified);
+        notified.as_mut().enable();
         let outcome = shared
             .state
             .backend
@@ -350,13 +360,8 @@ pub async fn next_message_broker_ref(
                     queued,
                     request,
                     runner_id,
-                    queue_depth,
                     next_runs_on,
                 } = *claimed;
-                shared
-                    .state
-                    .queue_depth
-                    .store(queue_depth, std::sync::atomic::Ordering::Release);
                 *shared.state.next_job_runs_on.write().unwrap() = next_runs_on;
                 record_claim_queue_wait(&shared, &Some(queued.clone()));
                 let run_id = queued.run_id;
@@ -393,10 +398,7 @@ pub async fn next_message_broker_ref(
             };
             return Ok((status, Json(body)).into_response());
         }
-        if tokio::time::timeout_at(deadline, shared.state.message_notify.notified())
-            .await
-            .is_err()
-        {
+        if tokio::time::timeout_at(deadline, notified).await.is_err() {
             let status = if runner_busy {
                 StatusCode::ACCEPTED
             } else {
@@ -704,7 +706,10 @@ pub async fn next_message_broker_ref_root(
     let wait = params
         .get("waitSeconds")
         .and_then(|value| value.parse::<u64>().ok())
-        .unwrap_or(50);
+        .unwrap_or(50)
+        // A client-chosen window longer than the liveness floor parks a
+        // healthy runner past its own reaping.
+        .min(crate::state::max_poll_window_secs());
     // The runner may report completion before its worker process has fully
     // exited. GitHub keeps polling with status=Busy during that drain window;
     // never dispatch a successor until the runner reports Online again.
@@ -715,6 +720,13 @@ pub async fn next_message_broker_ref_root(
     let mut deadline = std::time::Instant::now() + Duration::from_secs(wait);
 
     loop {
+        // Register the waiter *before* probing: a `notify_waiters` landing
+        // between the probe and a later `notified()` registration is lost
+        // forever (it stores no permit), stalling this poll until the
+        // window ends even though work was queued.
+        let notified = shared.state.message_notify.notified();
+        tokio::pin!(notified);
+        notified.as_mut().enable();
         // `drained`: Busy poll but nothing is still running — the session's
         // job has finished and the status=Busy report is stale.
         let mut drained = false;
@@ -746,13 +758,8 @@ pub async fn next_message_broker_ref_root(
                     queued,
                     request,
                     runner_id,
-                    queue_depth,
                     next_runs_on,
                 } = *claimed;
-                shared
-                    .state
-                    .queue_depth
-                    .store(queue_depth, std::sync::atomic::Ordering::Release);
                 *shared.state.next_job_runs_on.write().unwrap() = next_runs_on;
                 record_claim_queue_wait(&shared, &Some(queued));
                 Some(broker_job_ref_root(&request, runner_id))
@@ -777,10 +784,12 @@ pub async fn next_message_broker_ref_root(
         if wait == 0 || std::time::Instant::now() >= deadline {
             return Ok(Json(serde_json::Value::Null).into_response());
         }
-        // Wake promptly on cancel/enqueue rather than fixed 250ms sleep.
-        let remaining = deadline.saturating_duration_since(std::time::Instant::now());
-        let slice = remaining.min(Duration::from_secs(3));
-        let _ = tokio::time::timeout(slice, shared.state.message_notify.notified()).await;
+        // One wait per window, like `next_message_broker_ref` and the azdo
+        // path: the poll above is a cheap reader probe now, and
+        // `message_notify` fires on every enqueue, cancellation and
+        // promotion. The waiter was registered before the probe, so a wake
+        // that landed mid-poll is still seen.
+        let _ = tokio::time::timeout_at(deadline.into(), notified).await;
     }
 }
 
@@ -1314,6 +1323,7 @@ async fn fail_unclaimable_request(shared: &Arc<SharedState>, request_id: i64) {
     // Let a long-polling runner pick up a successor immediately rather than
     // waiting out its poll window behind a job that will never run.
     shared.state.message_notify.notify_waiters();
+    shared.state.sampler_notify.notify_waiters();
 }
 
 /// Reconcile job claims whose runner session did not survive a restart.
@@ -1695,6 +1705,7 @@ pub async fn broker_complete_job(
     // Wake long-polling runners so a queued successor job is delivered promptly
     // after cancel/complete (concurrency release path).
     shared.state.message_notify.notify_waiters();
+    shared.state.sampler_notify.notify_waiters();
     Ok(StatusCode::NO_CONTENT)
 }
 

crates/preloop-runner-server/src/control/backend.rs

@@ -43,8 +43,6 @@ pub(crate) struct PromoteOutcome {
     pub(crate) promoted: usize,
     /// Jobs failed closed by a denied or expired environment gate.
     pub(crate) failed: usize,
-    /// Global ready-queue depth after the pass.
-    pub(crate) queue_depth: usize,
     /// `runs-on` labels of the ready-queue front after the pass.
     pub(crate) next_runs_on: Vec<String>,
 }