fix(daemons): race in-flight work against cancellation on shutdown (0.87.4)

Two high findings sharing one root cause: long-running daemon work that
ignored the shutdown signal until it finished naturally.

**Cron tick (crawler).** `CronContext::run_tick` ran the metadata-pass
body inside `AssertUnwindSafe(...).catch_unwind().await` with no
cancellation. A fresh-catalog walk (many minutes) wedged
`DaemonHandle::shutdown`, `Supervisors::reload_crawler` (under the
supervisor lock), and SIGTERM responsiveness for the whole pass.

Wrap the body in `tokio::select! { biased; _ = self.cancel.cancelled() => ...; r = body => ... }`.
On cancel the body future drops, which cascades to dropping
`metadata.run()` — exactly what gives Chromium/lease teardown a chance
via the BrowserManager idle reaper. The advisory unlock + conn drop
still run unconditionally.

**Analysis dispatch.** `process_lease` wrapped the vision call only in
`tokio::time::timeout(self.job_timeout, …)` — default 600s. Toggling
analysis off in the dashboard (under the supervisor lock) or stopping
the container blocked on whichever call was in flight. Same fix:
`tokio::select!` against `self.cancel.cancelled()`. On cancel,
`jobs::release` returns the lease to the queue without burning an
attempt — next tick picks it up.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/src/analysis/daemon.rs

@@ -255,9 +255,30 @@ impl WorkerContext {
 
         let started = std::time::Instant::now();
         let dispatch = AssertUnwindSafe(self.dispatcher.dispatch(page_id)).catch_unwind();
-        let outcome = tokio::time::timeout(self.job_timeout, dispatch).await;
+        // Race the vision call against shutdown. Without this, shutdown
+        // (SIGTERM / `Supervisors::reload_analysis` toggling analysis off)
+        // blocks for up to `job_timeout` (default 600s) on whichever call
+        // happens to be in flight. On cancel we drop the dispatch future
+        // and `jobs::release` the lease — next tick picks it up without
+        // burning an attempt.
+        let cancel_result = tokio::select! {
+            biased;
+            _ = self.cancel.cancelled() => None,
+            o = tokio::time::timeout(self.job_timeout, dispatch) => Some(o),
+        };
         let elapsed_ms = started.elapsed().as_millis() as i64;
         heartbeat.abort();
+        let Some(outcome) = cancel_result else {
+            tracing::info!(
+                worker = self.id,
+                lease_id = %lease.id,
+                "analysis worker: cancelled mid-dispatch — releasing lease"
+            );
+            let _ = jobs::release(&self.pool, lease.id).await;
+            // Don't publish Failed (the page didn't actually fail) and
+            // don't write a duration row — neither outcome is true.
+            return;
+        };
 
         // Flatten timeout / panic / dispatch error into one Result + message.
         let result: Result<(), String> = match outcome {
@@ -413,6 +434,32 @@ pub mod test_support {
         }
     }
 
+    /// A dispatcher that sleeps for `delay` before returning `Ok`. Lets a
+    /// test fire shutdown WHILE a dispatch is in flight, to exercise the
+    /// cancellation race in `process_lease`.
+    pub struct SlowDispatcher {
+        pub calls: AtomicUsize,
+        pub delay: Duration,
+    }
+
+    impl SlowDispatcher {
+        pub fn new(delay: Duration) -> Arc<Self> {
+            Arc::new(Self { calls: AtomicUsize::new(0), delay })
+        }
+        pub fn call_count(&self) -> usize {
+            self.calls.load(Ordering::Acquire)
+        }
+    }
+
+    #[async_trait]
+    impl AnalyzeDispatcher for SlowDispatcher {
+        async fn dispatch(&self, _page_id: Uuid) -> anyhow::Result<()> {
+            self.calls.fetch_add(1, Ordering::AcqRel);
+            tokio::time::sleep(self.delay).await;
+            Ok(())
+        }
+    }
+
     #[async_trait]
     impl AnalyzeDispatcher for CountingDispatcher {
         async fn dispatch(&self, _page_id: Uuid) -> anyhow::Result<()> {

backend/src/crawler/daemon.rs

@@ -280,7 +280,8 @@ impl CronContext {
         // no future tick would ever run — workers would keep going but
         // no new metadata work would be scheduled until daemon restart.
         // The advisory unlock below runs unconditionally so a panicked
-        // tick doesn't leave the lock held for another replica.
+        // (or cancelled) tick doesn't leave the lock held for another
+        // replica.
         let metadata = &self.metadata;
         let pool = &self.pool;
         let retention_days = self.retention_days;
@@ -312,8 +313,23 @@ impl CronContext {
                 tracing::warn!(?e, "cron: persist last_metadata_tick_at failed");
             }
         };
-        if let Err(_panic) = AssertUnwindSafe(body).catch_unwind().await {
-            tracing::error!("cron: tick body panicked — continuing");
+        // Race the tick body against shutdown. Without this, a long
+        // metadata pass on a fresh catalog (minutes) wedges
+        // `DaemonHandle::shutdown`, `Supervisors::reload_crawler` (under
+        // the supervisor lock), and SIGTERM responsiveness for the
+        // entire pass. On cancel, dropping `body` cascades to dropping
+        // `metadata.run()`, which is what gives Chromium/lease cleanup
+        // their chance via the Browse Manager's idle reaper.
+        tokio::select! {
+            biased;
+            _ = self.cancel.cancelled() => {
+                tracing::info!("cron: cancelled mid-tick — abandoning to release advisory lock");
+            }
+            r = AssertUnwindSafe(body).catch_unwind() => {
+                if r.is_err() {
+                    tracing::error!("cron: tick body panicked — continuing");
+                }
+            }
         }
 
         let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
@@ -662,6 +678,35 @@ pub mod test_support {
         }
     }
 
+    /// `MetadataPass` that sleeps for `delay` before returning. Lets a
+    /// test trigger shutdown WHILE a tick body is in flight, exercising
+    /// the cancellation race in `CronContext::run_tick`.
+    pub struct SlowMetadataPass {
+        pub delay: std::time::Duration,
+        pub started: AtomicUsize,
+    }
+
+    impl SlowMetadataPass {
+        pub fn new(delay: std::time::Duration) -> Arc<Self> {
+            Arc::new(Self {
+                delay,
+                started: AtomicUsize::new(0),
+            })
+        }
+        pub fn start_count(&self) -> usize {
+            self.started.load(Ordering::Acquire)
+        }
+    }
+
+    #[async_trait]
+    impl MetadataPass for SlowMetadataPass {
+        async fn run(&self) -> anyhow::Result<pipeline::MetadataStats> {
+            self.started.fetch_add(1, Ordering::AcqRel);
+            tokio::time::sleep(self.delay).await;
+            Ok(pipeline::MetadataStats::default())
+        }
+    }
+
     pub type DispatchFn = Arc<
         dyn Fn(JobPayload) -> futures_util::future::BoxFuture<'static, anyhow::Result<SyncOutcome>>
             + Send