feat(scheduler): detect and report stuck queries (#39)

Adds an operator-facing warning when a distributed query stops making
progress, with enough detail in-line to diagnose without rerunning
with debug logging. Targets spiceai/spiceai#10832 where the production
wedge is rare and high-impact: by the time it is noticed the cluster
needs restarting, so any debug-level telemetry would no longer be
in scope.

Single diagnostic surface, single info-level signal:

  1. A 30s background loop on the scheduler samples per-stage progress
     for every active job. Capture uses a 500ms try_read budget per
     graph so a held write lock is itself recorded as a separate state
     ("Locked") rather than blocking the snapshot.

  2. After four consecutive samples with no per-stage movement, while
     the cluster has live executors and the job is not terminal, the
     loop emits one block of warn lines. The block re-fires every 30s
     while the condition holds and goes silent the moment progress
     resumes.

  3. The block contains:
       - Primary line: query id, elapsed stuck time, pending task count
       - One line per Running stage with unassigned/uncomplete partitions
       - "Lock could not be read" line if the snapshot timed out
       - Alive executor count
       - The event handler currently in flight on the scheduler event
         loop with its elapsed time, if a handler has been running
         longer than 30s
     Lines after the primary are emitted only when their underlying
     check fires, so the block reads as a coherent diagnosis of which
     code path is hung.

  4. To support (3), EventLoop now publishes an EventInFlight slot
     populated around each on_receive call and exposed via
     EventLoop::in_flight_signal(). EventAction gains an event_label
     method (default empty) that the scheduler implements to give each
     QueryStageSchedulerEvent variant a stable label for the warn block.

Volume: zero lines from this instrumentation at default RUST_LOG=info
when the cluster is healthy. When a query is stuck, ~3-6 lines per
30s while the condition holds.

Verified: cargo build, cargo test -p ballista-core -p ballista-scheduler
(83 passed, 1 ignored, includes 4 new tests covering the warn-block
helpers), cargo clippy -D warnings, cargo fmt --check all clean.

Author: phillipleblanc

ballista/core/src/event_loop.rs

@@ -18,14 +18,38 @@
 //! Event loop infrastructure for asynchronous message processing.
 
 use std::sync::Arc;
+use std::sync::Mutex;
 use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Instant;
 
 use async_trait::async_trait;
 use log::{error, info};
 use tokio::sync::mpsc;
 
 use crate::error::{BallistaError, Result};
 
+/// Snapshot of the event handler currently executing in an [`EventLoop`].
+///
+/// Populated by the loop on entry to [`EventAction::on_receive`] and cleared
+/// on return. External diagnostic code (see the scheduler's stuck-query
+/// detector) reads this to attribute a hang to a specific event variant
+/// and learn how long it has been in flight.
+#[derive(Clone, Debug)]
+pub struct EventInFlight {
+    /// Short, stable identifier for the event variant being processed,
+    /// supplied by [`EventAction::event_label`].
+    pub label: &'static str,
+    /// Instant the handler began.
+    pub started_at: Instant,
+}
+
+/// Shared slot tracking which event is currently being processed by the
+/// loop. `Some(_)` while a handler runs, `None` otherwise.
+///
+/// Uses a synchronous mutex — the lock is held only to swap a small Option
+/// at the start and end of each `on_receive`, never across an await.
+pub type EventProgressSignal = Arc<Mutex<Option<EventInFlight>>>;
+
 /// Trait defining actions to be performed in response to events in an event loop.
 #[async_trait]
 pub trait EventAction<E>: Send + Sync {
@@ -45,6 +69,14 @@ pub trait EventAction<E>: Send + Sync {
 
     /// Called when an error occurs during event processing.
     fn on_error(&self, error: BallistaError);
+
+    /// Short, stable label for the given event variant, used by external
+    /// diagnostic code to identify which handler is currently running.
+    /// Default returns an empty string, meaning the event loop will not
+    /// publish a tracking entry for events from this action.
+    fn event_label(&self, _event: &E) -> &'static str {
+        ""
+    }
 }
 
 /// An asynchronous event loop that processes events through a channel.
@@ -57,6 +89,7 @@ pub struct EventLoop<E> {
     stopped: Arc<AtomicBool>,
     action: Arc<dyn EventAction<E>>,
     tx_event: Option<mpsc::Sender<E>>,
+    in_flight: EventProgressSignal,
 }
 
 impl<E: Send + 'static> EventLoop<E> {
@@ -72,9 +105,17 @@ impl<E: Send + 'static> EventLoop<E> {
             stopped: Arc::new(AtomicBool::new(false)),
             action,
             tx_event: None,
+            in_flight: Arc::new(Mutex::new(None)),
         }
     }
 
+    /// Returns a handle to the shared signal that tracks the event
+    /// currently being processed. Used by diagnostic code to attribute
+    /// a stuck condition to a specific handler.
+    pub fn in_flight_signal(&self) -> EventProgressSignal {
+        self.in_flight.clone()
+    }
+
     fn run(&self, mut rx_event: mpsc::Receiver<E>) {
         assert!(
             self.tx_event.is_some(),
@@ -84,11 +125,27 @@ impl<E: Send + 'static> EventLoop<E> {
         let name = self.name.clone();
         let stopped = self.stopped.clone();
         let action = self.action.clone();
+        let in_flight = self.in_flight.clone();
         tokio::spawn(async move {
             info!("Starting the event loop {name}");
             while !stopped.load(Ordering::SeqCst) {
                 if let Some(event) = rx_event.recv().await {
-                    if let Err(e) = action.on_receive(event, &tx_event, &rx_event).await {
+                    let label = action.event_label(&event);
+                    if !label.is_empty()
+                        && let Ok(mut slot) = in_flight.lock()
+                    {
+                        *slot = Some(EventInFlight {
+                            label,
+                            started_at: Instant::now(),
+                        });
+                    }
+                    let result = action.on_receive(event, &tx_event, &rx_event).await;
+                    if !label.is_empty()
+                        && let Ok(mut slot) = in_flight.lock()
+                    {
+                        *slot = None;
+                    }
+                    if let Err(e) = result {
                         error!("Fail to process event due to {e}");
                         action.on_error(e);
                     }