Never crash: the work gate counts held bytes exactly past usize::MAX (#2576)

Review follow-up: a saturating usize count lost reservations. With a
usize::MAX permit and an urgent 101-byte one, dropping the large one first
reset the count to 0 while the 101 bytes were still held; and on a gate whose
limit is usize::MAX the saturated sum never exceeded it, so a second holder
didn't wait. The held count is now a u128 (exact for any number of usize
permits); usage() reports it saturated to usize. The test covers both release
orders and the usize::MAX limit.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PtjHGt5gnCHXgLt4NbGorf

crates/engine/src/media.rs

@@ -2177,25 +2177,50 @@ mod tests {
 
     /// Issue #2576: work weighed with saturating maths can ask for `usize::MAX` bytes. Adding to
     /// that overflowed (a panic with the gate's mutex held in a debug build, a wrapped count that
-    /// let everything in for a release one); the count now saturates, and later work waits.
+    /// let everything in for a release one). The count is now exact past `usize::MAX`: later work
+    /// waits, whichever permit is released first, and even on a gate whose limit is `usize::MAX`.
     #[test]
-    fn work_gate_saturates_instead_of_overflowing() {
+    fn work_gate_counts_past_usize_max_instead_of_overflowing() {
         use crate::memory::WorkGate;
+        // a holder of one byte, on its own thread; still waiting 50 ms later
+        let waiting = |g: &Arc<WorkGate>| {
+            let g2 = g.clone();
+            let t = std::thread::spawn(move || {
+                let _b = g2.acquire(1);
+                g2.usage().0
+            });
+            std::thread::sleep(std::time::Duration::from_millis(50));
+            assert!(!t.is_finished(), "the next holder waits");
+            t
+        };
+
+        // the huge permit released first: the urgent one is still counted
         let g = Arc::new(WorkGate::new(100));
         let huge = g.acquire(usize::MAX);
-        let urgent = g.acquire_urgent(1);
-        assert_eq!(g.usage().0, usize::MAX);
+        let urgent = g.acquire_urgent(101);
+        assert_eq!(g.usage().0, usize::MAX, "reported saturated");
+        drop(huge);
+        assert_eq!(g.usage().0, 101);
+        let t = waiting(&g);
         drop(urgent);
-        let g2 = g.clone();
-        let t = std::thread::spawn(move || {
-            let _b = g2.acquire(1);
-            g2.usage().0
-        });
-        std::thread::sleep(std::time::Duration::from_millis(50));
-        assert!(!t.is_finished(), "the second holder waits while the first holds everything");
+        assert_eq!(t.join().unwrap(), 1);
+        assert_eq!(g.usage().0, 0);
+
+        // the urgent permit released first
+        let huge = g.acquire(usize::MAX);
+        drop(g.acquire_urgent(1));
+        assert_eq!(g.usage().0, usize::MAX);
+        let t = waiting(&g);
         drop(huge);
         assert_eq!(t.join().unwrap(), 1);
         assert_eq!(g.usage().0, 0);
+
+        // a gate without a practical limit is still full when it holds usize::MAX
+        let g = Arc::new(WorkGate::new(usize::MAX));
+        let huge = g.acquire(usize::MAX);
+        let t = waiting(&g);
+        drop(huge);
+        assert_eq!(t.join().unwrap(), 1);
     }
 
     #[test]

crates/engine/src/memory.rs

@@ -99,7 +99,10 @@ pub fn cache_share(budget: usize) -> usize {
 /// A weighted semaphore over bytes of working memory: [`WorkGate::acquire`] waits while other
 /// holders already use the limit (one holder may always proceed, whatever its size).
 pub struct WorkGate {
-    state: Mutex<(usize, usize)>,
+    /// Bytes held and the limit. Held is a `u128`: callers weigh work with saturating maths, so a
+    /// permit may hold `usize::MAX`, and the exact total of several such permits still fits, so
+    /// releasing one in any order leaves the others counted (issue #2576).
+    state: Mutex<(u128, usize)>,
     cv: Condvar,
 }
 
@@ -114,7 +117,7 @@ impl WorkGate {
         WorkGate { state: Mutex::new((0, limit)), cv: Condvar::new() }
     }
 
-    fn lock(&self) -> std::sync::MutexGuard<'_, (usize, usize)> {
+    fn lock(&self) -> std::sync::MutexGuard<'_, (u128, usize)> {
         self.state.lock().unwrap_or_else(|e| e.into_inner())
     }
 
@@ -123,27 +126,27 @@ impl WorkGate {
         self.cv.notify_all();
     }
 
-    /// Bytes held now and the limit.
+    /// Bytes held now (at most `usize::MAX`) and the limit.
     pub fn usage(&self) -> (usize, usize) {
-        *self.lock()
+        let g = self.lock();
+        (usize::try_from(g.0).unwrap_or(usize::MAX), g.1)
     }
 
-    /// Hold `bytes`, waiting until they fit (or nothing else is held). Callers weigh work with
-    /// saturating maths, so `bytes` may be `usize::MAX`: the count saturates there, never overflows
-    /// (issue #2576).
+    /// Hold `bytes`, waiting until they fit (or nothing else is held). Even `usize::MAX` bytes on
+    /// a gate whose limit is `usize::MAX` make the next holder wait.
     pub fn acquire(&self, bytes: usize) -> Permit<'_> {
         let mut g = self.lock();
-        while g.0 > 0 && g.0.saturating_add(bytes) > g.1 {
+        while g.0 > 0 && g.0.saturating_add(bytes as u128) > g.1 as u128 {
             g = self.cv.wait(g).unwrap_or_else(|e| e.into_inner());
         }
-        g.0 = g.0.saturating_add(bytes);
+        g.0 = g.0.saturating_add(bytes as u128);
         Permit { gate: self, bytes }
     }
 
     /// Hold `bytes` without waiting (interactive work): counted, so background work waits for it.
     pub fn acquire_urgent(&self, bytes: usize) -> Permit<'_> {
         let mut g = self.lock();
-        g.0 = g.0.saturating_add(bytes);
+        g.0 = g.0.saturating_add(bytes as u128);
         Permit { gate: self, bytes }
     }
 }
@@ -151,7 +154,7 @@ impl WorkGate {
 impl Drop for Permit<'_> {
     fn drop(&mut self) {
         let mut g = self.gate.lock();
-        g.0 = g.0.saturating_sub(self.bytes);
+        g.0 = g.0.saturating_sub(self.bytes as u128);
         drop(g);
         self.gate.cv.notify_all();
     }