Reseed: zero-converge drained accounting keys

DiegoTavares · DiegoTavares · commit dd78d32b3886 · 2026-06-02T14:09:55.000-07:00
The booked-counter recompute built its Redis reseed solely from a
SUM(proc) snapshot, which only returns keys that still have procs. A
key whose booked counter drifted stale-high (e.g. a lost decrement)
and whose procs then drained to zero vanished from the snapshot
entirely, so no resetting HSET was never emitted. The stale value
persisted indefinitely for a subscription/folder/job this wedged the
key, falsely failing the burst/cap check in the booking Lua with no
path to recovery.

Solution: Overlay the snapshot on a zero-baseline of every enumerable key.
Before folding in the proc sums, seed int_cores=0/int_gpus=0 for every
acct:sub / acct:folder / acct:job / acct:point key that exists for a
scheduler-managed show, enumerated by reusing the same limit-table
queries the limit reseed already runs (subscription, folder_resource,
job_resource non-FINISHED, point). A baseline key absent from the
snapshot has no procs and emits a resetting 0; a key with procs emits
its true sum. This is the same key universe and same order of op count
the limit reseed already writes each cycle, and it lets recompute
converge a drained key back to truth. Bootstrap reseed benefits too,
since it shares reseed_redis_once.

acct:layer is intentionally not zero-baselined: layers have no limit
table to enumerate from, and the booking Lua never reads the layer
counter (HINCRBY-only), so residual layer drift is cosmetic. Orphaned
keys for FINISHED jobs are likewise left in place (never read again);
reclaiming that memory is tracked as future work.
diff --git a/docs/_docs/developer-guide/redis-accounting.md b/docs/_docs/developer-guide/redis-accounting.md
@@ -310,6 +310,31 @@ The result is dual-written to:
 The dual write happens under the same SUM query for consistency: both PG and
 Redis end up showing the same snapshot.
 
+#### Zero-convergence for drained keys
+
+`SUM(proc)` only returns keys that **still have procs**. A key whose booked
+counter drifted stale-high (e.g. a lost decrement) and whose procs then drained
+to zero would vanish from the snapshot entirely - so a snapshot-only reseed
+could never reset it, and the stale value would wedge the key forever (for a
+subscription/folder/job this means falsely failing the burst/cap check in the
+booking Lua with no path to recovery).
+
+To close this, the Redis reseed overlays the snapshot on a **zero-baseline** of
+every enumerable key. Before folding in the proc sums, it seeds `int_cores=0`/
+`int_gpus=0` for every `acct:sub`/`acct:folder`/`acct:job`/`acct:point` key that
+exists for a scheduler-managed show - enumerated by reusing the same limit-table
+queries the limit reseed runs (`subscription`, `folder_resource`,
+`job_resource` non-`FINISHED`, `point`). A key present in the baseline but
+absent from the snapshot has no procs, so it emits a resetting `0`; a key with
+procs emits its true sum. This is the same key universe, and same order of
+magnitude of ops, the limit reseed already writes each cycle.
+
+`acct:layer` is intentionally **not** zero-baselined: layers have no limit table
+to enumerate from, and the booking Lua never reads the layer counter (it is
+`HINCRBY`-only), so residual layer drift is cosmetic. Orphaned `acct:job`/
+`acct:layer` keys for `FINISHED` jobs are likewise left in place (never read
+again); reclaiming that memory is tracked as future work.
+
 ### Limit reseed (every 5 min)
 
 For every scheduler-managed show, the scheduler reads limit fields (`size`,
diff --git a/rust/crates/scheduler/src/accounting/dao.rs b/rust/crates/scheduler/src/accounting/dao.rs
@@ -41,6 +41,22 @@ pub struct BookedSnapshotRow {
     pub gpus: i64,
 }
 
+/// The full universe of enumerable accounting keys for scheduler-managed shows,
+/// used by the booked-counter recompute to seed zero baselines. A key present here
+/// but absent from the `SUM(proc)` snapshot has no procs, so its booked counter must
+/// be reset to 0 - otherwise a counter that drifted stale-high and then drained to
+/// zero procs would never converge (the snapshot only returns keys that still have
+/// procs). Layer keys are intentionally absent: layers have no limit table to
+/// enumerate from, and the booking Lua never reads the layer counter (it is
+/// `HINCRBY`-only), so residual layer drift is cosmetic. See `recompute.rs`.
+#[derive(Debug, Clone, Default)]
+pub struct BaselineKeys {
+    pub subs: Vec<(Uuid, Uuid)>,
+    pub folders: Vec<Uuid>,
+    pub jobs: Vec<Uuid>,
+    pub points: Vec<(Uuid, Uuid)>,
+}
+
 #[derive(Debug, Clone)]
 pub struct SubscriptionLimitsRow {
     pub show_id: Uuid,
@@ -189,6 +205,27 @@ impl AccountingDao {
             .collect())
     }
 
+    /// Enumerate the full set of sub/folder/job/point keys for scheduler-managed
+    /// shows, by projecting the key tuples out of the four limit queries. Reusing the
+    /// limit queries (rather than dedicated key-only SELECTs) guarantees this key
+    /// universe is grain-consistent with both the limit reseed and the booked snapshot
+    /// (same `b_scheduler_managed` / `str_state <> 'FINISHED'` filters). Layer keys are
+    /// not enumerable from a limit table, so they are absent here by design.
+    pub async fn query_booked_baseline_keys(&self) -> Result<BaselineKeys> {
+        let (subs, folders, jobs, points) = tokio::try_join!(
+            self.query_subscription_limits(),
+            self.query_folder_limits(),
+            self.query_job_limits(),
+            self.query_point_limits(),
+        )?;
+        Ok(BaselineKeys {
+            subs: subs.iter().map(|r| (r.show_id, r.alloc_id)).collect(),
+            folders: folders.iter().map(|r| r.folder_id).collect(),
+            jobs: jobs.iter().map(|r| r.job_id).collect(),
+            points: points.iter().map(|r| (r.dept_id, r.show_id)).collect(),
+        })
+    }
+
     pub async fn query_subscription_limits(&self) -> Result<Vec<SubscriptionLimitsRow>> {
         #[derive(sqlx::FromRow)]
         struct Row {
diff --git a/rust/crates/scheduler/src/accounting/recompute.rs b/rust/crates/scheduler/src/accounting/recompute.rs
@@ -34,7 +34,7 @@ use miette::{IntoDiagnostic, Result, WrapErr};
 use tokio::time;
 use tracing::{debug, error, info, warn};
 
-use crate::accounting::dao::BookedSnapshotRow;
+use crate::accounting::dao::{BaselineKeys, BookedSnapshotRow};
 use crate::accounting::redis_client::ReseedOp;
 use crate::accounting::AccountingService;
 use crate::config::CONFIG;
@@ -96,12 +96,21 @@ async fn run_once(service: &AccountingService, pg_dao: &Arc<ResourceAccountingDa
 /// CAS-guarded reseed of the booked-counter fields (`int_cores`/`int_gpus`) on the
 /// five `acct:*` hashes from a fresh `SUM(proc)` snapshot. Used by both the recompute
 /// loop and the bootstrap. On CAS-budget exhaustion: warn and return Ok (per §2.4).
+///
+/// The snapshot is overlaid on a zero-baseline of every enumerable sub/folder/job/point
+/// key (see `query_booked_baseline_keys`), so a key whose counter drifted stale-high and
+/// then drained to zero procs is reset to 0 rather than being left untouched - the
+/// `SUM(proc)` snapshot alone only returns keys that still have procs. Both the snapshot
+/// and the baseline are re-fetched per CAS attempt, matching `limit_reseed::reseed_once`.
 pub async fn reseed_redis_once(service: &AccountingService) -> Result<()> {
     let max_retries = CONFIG.accounting.cas_max_retries;
     for attempt in 0..=max_retries {
         let seq_before = service.redis().get_seq().await.into_diagnostic()?;
-        let rows = service.dao().query_booked_snapshot().await?;
-        let ops = booked_ops_from_snapshot(&rows);
+        let (rows, baseline) = tokio::try_join!(
+            service.dao().query_booked_snapshot(),
+            service.dao().query_booked_baseline_keys(),
+        )?;
+        let ops = booked_ops_from_snapshot(&rows, &baseline);
         debug!(
             "Recompute reseed attempt {}/{}: {} rows -> {} ops at seq={}",
             attempt + 1,
@@ -147,7 +156,16 @@ pub async fn reseed_redis_once(service: &AccountingService) -> Result<()> {
 /// `HSET` (overwrite) rather than `HINCRBY`, emitting one op per row would let later
 /// rows clobber earlier ones - the final value would be whichever row sorted last,
 /// not the sum across rows. Aggregate first, emit once per unique key.
-fn booked_ops_from_snapshot(rows: &[BookedSnapshotRow]) -> Vec<ReseedOp> {
+///
+/// `baseline` seeds a zero entry for every enumerable sub/folder/job/point key before
+/// the proc sums are folded in. A key in the baseline but absent from `rows` has no
+/// procs, so it emits `int_cores=0`/`int_gpus=0` - this is what lets recompute converge
+/// a counter that drifted stale-high and then drained to zero procs (without it, such a
+/// key would simply be missing from the snapshot and never corrected). Layers are not
+/// in the baseline (no limit table to enumerate them and the booking Lua never reads the
+/// layer counter), so the layer map stays purely proc-driven - residual layer drift is
+/// cosmetic by design.
+fn booked_ops_from_snapshot(rows: &[BookedSnapshotRow], baseline: &BaselineKeys) -> Vec<ReseedOp> {
     use std::collections::HashMap;
 
     let mut sub_totals: HashMap<(uuid::Uuid, uuid::Uuid), (i64, i64)> = HashMap::new();
@@ -156,6 +174,21 @@ fn booked_ops_from_snapshot(rows: &[BookedSnapshotRow]) -> Vec<ReseedOp> {
     let mut layer_totals: HashMap<uuid::Uuid, (i64, i64)> = HashMap::new();
     let mut point_totals: HashMap<(uuid::Uuid, uuid::Uuid), (i64, i64)> = HashMap::new();
 
+    // Zero-baseline first: every enumerable key gets a (0, 0) entry so keys with no
+    // procs still emit a resetting HSET. The proc fold below adds on top of these.
+    for &k in &baseline.subs {
+        sub_totals.entry(k).or_default();
+    }
+    for &k in &baseline.folders {
+        folder_totals.entry(k).or_default();
+    }
+    for &k in &baseline.jobs {
+        job_totals.entry(k).or_default();
+    }
+    for &k in &baseline.points {
+        point_totals.entry(k).or_default();
+    }
+
     for r in rows {
         let s = sub_totals.entry((r.show_id, r.alloc_id)).or_default();
         s.0 += r.cores;
@@ -253,9 +286,13 @@ mod tests {
         ops.iter().filter(|o| o.key == key).count()
     }
 
+    fn empty_baseline() -> BaselineKeys {
+        BaselineKeys::default()
+    }
+
     #[test]
     fn snapshot_single_row_expands_to_ten_ops_in_cores() {
-        let ops = booked_ops_from_snapshot(&[fixture_row()]);
+        let ops = booked_ops_from_snapshot(&[fixture_row()], &empty_baseline());
         // 5 unique keys × 2 fields (int_cores, int_gpus).
         assert_eq!(ops.len(), 10);
         let cores_ops: Vec<_> = ops.iter().filter(|o| o.field == "int_cores").collect();
@@ -270,7 +307,7 @@ mod tests {
 
     #[test]
     fn snapshot_keys_match_publisher_format() {
-        let ops = booked_ops_from_snapshot(&[fixture_row()]);
+        let ops = booked_ops_from_snapshot(&[fixture_row()], &empty_baseline());
         let keys: std::collections::HashSet<&str> = ops.iter().map(|o| o.key.as_str()).collect();
         assert!(keys.contains(
             "acct:sub:00000000-0000-0000-0000-000000000000:00000000-0000-0000-0000-000000000000"
@@ -313,7 +350,7 @@ mod tests {
             gpus: 2,
         };
 
-        let ops = booked_ops_from_snapshot(&[row_a, row_b]);
+        let ops = booked_ops_from_snapshot(&[row_a, row_b], &empty_baseline());
 
         // Centicores summed (3500), then /100 -> 35 cores.
         let sub_key = format!("acct:sub:{}:{}", show, alloc);
@@ -367,7 +404,7 @@ mod tests {
             },
         ];
 
-        let ops = booked_ops_from_snapshot(&rows);
+        let ops = booked_ops_from_snapshot(&rows, &empty_baseline());
 
         let job_key = format!("acct:job:{}", job);
         assert_eq!(find_op(&ops, &job_key, "int_cores").value, 17);
@@ -383,4 +420,109 @@ mod tests {
         assert_eq!(find_op(&ops, &sub_a, "int_cores").value, 10);
         assert_eq!(find_op(&ops, &sub_b, "int_cores").value, 7);
     }
+
+    /// A baseline key with no matching proc row (its procs drained to zero) must emit a
+    /// resetting `int_cores=0`/`int_gpus=0` pair so recompute can converge a stale-high
+    /// counter back to truth. This is the core of the zero-convergence fix.
+    #[test]
+    fn baseline_key_absent_from_snapshot_emits_zero_pair() {
+        let show = Uuid::new_v4();
+        let alloc = Uuid::new_v4();
+        let folder = Uuid::new_v4();
+        let job = Uuid::new_v4();
+        let dept = Uuid::new_v4();
+        let baseline = BaselineKeys {
+            subs: vec![(show, alloc)],
+            folders: vec![folder],
+            jobs: vec![job],
+            points: vec![(dept, show)],
+        };
+
+        // No proc rows at all: every baseline key drained to zero.
+        let ops = booked_ops_from_snapshot(&[], &baseline);
+
+        // 4 enumerable keys × 2 fields; layers have no baseline so none appear.
+        assert_eq!(ops.len(), 8);
+        for key in [
+            format!("acct:sub:{}:{}", show, alloc),
+            format!("acct:folder:{}", folder),
+            format!("acct:job:{}", job),
+            format!("acct:point:{}:{}", dept, show),
+        ] {
+            assert_eq!(find_op(&ops, &key, "int_cores").value, 0);
+            assert_eq!(find_op(&ops, &key, "int_gpus").value, 0);
+        }
+    }
+
+    /// A baseline key that also appears in the snapshot is not double-counted: it emits
+    /// one pair carrying the proc sum, not the seeded zero plus the sum.
+    #[test]
+    fn baseline_key_present_in_snapshot_uses_proc_sum_once() {
+        let row = fixture_row(); // all-nil keys, 4200 centicores -> 42 cores, 3 gpus.
+        let baseline = BaselineKeys {
+            subs: vec![(Uuid::nil(), Uuid::nil())],
+            folders: vec![Uuid::nil()],
+            jobs: vec![Uuid::nil()],
+            points: vec![(Uuid::nil(), Uuid::nil())],
+        };
+
+        let ops = booked_ops_from_snapshot(&[row], &baseline);
+
+        // Still one pair per key (no zero/sum duplication): 5 keys × 2 fields.
+        assert_eq!(ops.len(), 10);
+        let job_key = "acct:job:00000000-0000-0000-0000-000000000000";
+        assert_eq!(count_ops_for_key(&ops, job_key), 2);
+        assert_eq!(find_op(&ops, job_key, "int_cores").value, 42);
+        assert_eq!(find_op(&ops, job_key, "int_gpus").value, 3);
+    }
+
+    /// Mixed: one baseline job has procs (keep its sum), another drained to zero (reset).
+    /// The layer that exists only in the snapshot is still emitted from proc data.
+    #[test]
+    fn baseline_resets_drained_key_while_keeping_active_one() {
+        let show = Uuid::new_v4();
+        let alloc = Uuid::new_v4();
+        let folder = Uuid::new_v4();
+        let dept = Uuid::new_v4();
+        let active_job = Uuid::new_v4();
+        let drained_job = Uuid::new_v4();
+        let active_layer = Uuid::new_v4();
+
+        let baseline = BaselineKeys {
+            subs: vec![(show, alloc)],
+            folders: vec![folder],
+            jobs: vec![active_job, drained_job],
+            points: vec![(dept, show)],
+        };
+        let rows = [BookedSnapshotRow {
+            show_id: show,
+            alloc_id: alloc,
+            folder_id: folder,
+            job_id: active_job,
+            layer_id: active_layer,
+            dept_id: dept,
+            cores: 500, // 5 cores
+            gpus: 1,
+        }];
+
+        let ops = booked_ops_from_snapshot(&rows, &baseline);
+
+        let active_key = format!("acct:job:{}", active_job);
+        assert_eq!(find_op(&ops, &active_key, "int_cores").value, 5);
+        assert_eq!(find_op(&ops, &active_key, "int_gpus").value, 1);
+
+        let drained_key = format!("acct:job:{}", drained_job);
+        assert_eq!(find_op(&ops, &drained_key, "int_cores").value, 0);
+        assert_eq!(find_op(&ops, &drained_key, "int_gpus").value, 0);
+
+        // Layer is proc-driven only; the active layer is present, no zero-baseline layers.
+        let layer_key = format!("acct:layer:{}", active_layer);
+        assert_eq!(find_op(&ops, &layer_key, "int_cores").value, 5);
+        assert_eq!(
+            ops.iter()
+                .filter(|o| o.key.starts_with("acct:layer:"))
+                .count(),
+            2
+        );
+    }
 }
diff --git a/rust/crates/scheduler/src/pipeline/matcher.rs b/rust/crates/scheduler/src/pipeline/matcher.rs
@@ -266,9 +266,7 @@ impl MatchingService {
                     tags,
                     cores: cores_requested,
                     memory: layer.mem_min,
-                    validation: move |host| {
-                        Self::validate_match(host, os.as_deref(), threadable)
-                    },
+                    validation: move |host| Self::validate_match(host, os.as_deref(), threadable),
                 })
                 .await
                 .expect("Host Cache actor is unresponsive");
