fix(master): report pool permits at acquire/release, split claiming from running

crates/lance-context-master/src/scheduler.rs

@@ -1188,18 +1188,23 @@ fn spawn_filtered_pool_poller(
     } else {
         "general"
     };
-    let pool_total = pool.available_permits();
+    // Permits in use are split by what holds them: `claiming` while the etcd
+    // claim is outstanding, `running` once a task owns the permit. Both are
+    // updated at the permit boundaries themselves, so a claim that blocks on
+    // etcd for minutes shows as a held permit rather than as idle capacity.
+    let claiming =
+        metrics::gauge!("master_task_pool_in_use", "pool" => pool_label, "state" => "claiming");
+    let running =
+        metrics::gauge!("master_task_pool_in_use", "pool" => pool_label, "state" => "running");
+    claiming.set(0.0);
+    running.set(0.0);
     tokio::spawn(async move {
         loop {
             if report_depth {
                 if let Ok(queued) = state.task_store.queue_depth().await {
                     metrics::gauge!("master_task_queue_depth").set(queued as f64);
                 }
             }
-            // Every pool reports, including the resident-only slot added in
-            // #331 that the queue-depth gauge never covered.
-            metrics::gauge!("master_task_pool_in_use", "pool" => pool_label)
-                .set(pool_total.saturating_sub(pool.available_permits()) as f64);
             while pool.available_permits() > 0 {
                 // Reserve both budgets before taking durable table ownership.
                 let compact_permit = if kinds == TaskKinds::COMPACT {
@@ -1216,6 +1221,7 @@ fn spawn_filtered_pool_poller(
                 let Some(operation) = state.admission.try_admit() else {
                     break;
                 };
+                claiming.increment(1.0);
                 let claim_start = std::time::Instant::now();
                 let claim = if resident_only {
                     state
@@ -1225,6 +1231,7 @@ fn spawn_filtered_pool_poller(
                 } else {
                     state.task_store.claim_next_of_kinds(kinds).await
                 };
+                claiming.decrement(1.0);
                 match claim {
                     Ok(Some(claim)) => {
                         let claim_elapsed = claim_start.elapsed();
@@ -1235,11 +1242,14 @@ fn spawn_filtered_pool_poller(
                             permit_wait: Duration::ZERO,
                         };
                         let st = state.clone();
+                        let running = running.clone();
+                        running.increment(1.0);
                         tokio::spawn(async move {
                             run_task(&st, claim, timing).await;
                             drop(permit);
                             drop(compact_permit);
                             drop(operation);
+                            running.decrement(1.0);
                         });
                     }
                     Ok(None) => break,

docs/operations/maintenance-runbook.md

@@ -27,7 +27,7 @@ restart something because it "has been running a long time", stop and check
 | Question | Look at |
 |---|---|
 | Is WAL piling up anywhere? | `master_wal_pending_generations_max`, `master_wal_pending_generations_total`, `master_stores_pending_over_read_cap`; per table `GET /api/v1/experiments/{name}` → `pending_wal_generations`, `fragment_count` |
-| Are tasks waiting instead of running? | `master_task_schedule_to_bind_seconds{kind}` (enqueue → claim). Rising with `master_task_pool_in_use{pool}` at capacity = **capacity**. Rising with pools idle = **admission is blocking** (§4). |
+| Are tasks waiting instead of running? | `master_task_schedule_to_bind_seconds{kind}` (enqueue → claim). Rising with `master_task_pool_in_use{pool,state="running"}` at capacity = **capacity**. Rising with `state="claiming"` high = **the etcd claim itself is slow or blocked** (§4, admission scan). Rising with both low = **nothing eligible** (§4). |
 | Is something queued at all? | `master_task_queue_depth`; `GET /api/v1/tasks/{id}` for a specific task |
 | Is a running merge actually working? | `GET /api/v1/merge-progress?target=<name>` (§3) |
 | Is compaction/index getting its commit turn? | `master_commit_turn_requests_total{kind,result}`: `served` vs `expired`; `master_commit_turn_wait_seconds`; `master_merge_yield_to_compaction_total` (§5) |
@@ -51,7 +51,7 @@ restart something because it "has been running a long time", stop and check
      a changing `sequence` is progress.
    * `reported` → read `work_report` (§3).
 3. If a task is live and progressing: this is **capacity**, not a fault. Check
-   `master_task_pool_in_use{pool="merge"}` and `{pool="resident_merge"}`; the
+   `master_task_pool_in_use{pool="merge",state="running"}` and `{pool="resident_merge",state="running"}`; the
    executor's memory budget (`ROLLOUT_APPEND_LOCAL_MEMORY_BYTES`,
    `CATCHUP_MERGE_MEMORY_BYTES`); and whether the table is
    oversized-generation bound (`work_report` shows waiting-for-memory phases).