Close three kernel gates the benchmark notes flagged.

AGENT_CHAT.md

@@ -1,5 +1,9 @@
 # RedlineDB Agent Chat
 
+## 2026-09-25T18:40Z fix/kernel-p0-gates
+
+Claim: canonical checkout `/home/ubuntu/redlineDB`, branch `fix/kernel-p0-gates` from `origin/main` `76ba2a52c469c8917929418ff1f4fe8f5b6623cf`. Clock eviction can decay a hot clean frame, a free row lock is not granted ahead of a queued waiter, a WAL segment that fills exactly does not grow past its limit, and a morsel arena offset that does not fit in `u32` aborts instead of truncating. No speed claim. No worktree.
+
 ## 2026-09-25T09:00Z docs/parity-record
 
 Claim: canonical checkout `/home/ubuntu/redlineDB`, branch `docs/parity-record` from `origin/main` `d5d1c57bbab97bf43d0b74cdfd0a660aa4bfe6b2`. The Postgres skip list and `docs/beyond-postgres-skips.md` now say the 265-case gate already passes, including the 88 entries that were still marked deferred. No SQL change. No generated README block edit. No worktree.

crates/kernel/src/engine/lock.rs

@@ -81,12 +81,15 @@ impl RowLockManager {
         // Fast path: lock free or already held by us. Touches no FIFO.
         let state = rows.entry(key).or_default();
         match state.owner {
-            None => {
+            // A free row with nobody waiting can be granted immediately.
+            // A free row that already has waiters belongs to the front of
+            // that queue; a newcomer joins the back instead of cutting in.
+            None if state.waiters.is_empty() => {
                 state.owner = Some(tx_id);
                 return Ok(());
             }
             Some(owner) if owner == tx_id => return Ok(()),
-            Some(_) => {}
+            _ => {}
         }
 
         // Slow path: enqueue ourselves onto the per-row FIFO and park
@@ -121,23 +124,22 @@ impl RowLockManager {
             let state = rows.entry(key).or_default();
             match state.owner {
                 None => {
-                    // Confirm the front-of-FIFO targeting: pop ourselves
-                    // off the head if present, then take ownership.
+                    // Only the waiter at the front may take a free lock.
+                    // A spurious wake of someone further back must keep waiting.
                     if state
                         .waiters
                         .front()
                         .is_some_and(|cv| Arc::ptr_eq(cv, &my_cv))
                     {
                         state.waiters.pop_front();
-                    } else {
-                        // Spurious wake; ensure we're still parked.
-                        if !state.waiters.iter().any(|cv| Arc::ptr_eq(cv, &my_cv)) {
-                            state.waiters.push_back(Arc::clone(&my_cv));
-                        }
+                        state.owner = Some(tx_id);
+                        self.record_lock_wait_us(wait_started.elapsed());
+                        return Ok(());
                     }
-                    state.owner = Some(tx_id);
-                    self.record_lock_wait_us(wait_started.elapsed());
-                    return Ok(());
+                    if !state.waiters.iter().any(|cv| Arc::ptr_eq(cv, &my_cv)) {
+                        state.waiters.push_back(Arc::clone(&my_cv));
+                    }
+                    continue;
                 }
                 Some(owner) if owner == tx_id => {
                     // Re-entrant acquire that snuck in — drop our slot.
@@ -300,6 +302,32 @@ mod tests {
         }
     }
 
+    /// A caller that arrives after the holder unlocks must not take the
+    /// row while an earlier waiter is still at the front of the queue.
+    #[test]
+    fn newcomer_does_not_cut_ahead_of_a_queued_waiter() {
+        let mgr = Arc::new(RowLockManager::new(1, Duration::from_secs(5)));
+        let rel_id = RelId(3);
+        let row_id = RowId(9);
+        mgr.lock(rel_id, row_id, TxId(1)).unwrap();
+
+        let mgr_waiter = Arc::clone(&mgr);
+        let order = Arc::new(std::sync::Mutex::new(Vec::new()));
+        let order_waiter = Arc::clone(&order);
+        let waiter = thread::spawn(move || {
+            mgr_waiter.lock(rel_id, row_id, TxId(2)).unwrap();
+            order_waiter.lock().expect("order").push(2);
+            mgr_waiter.unlock(rel_id, row_id, TxId(2));
+        });
+        thread::sleep(Duration::from_millis(100));
+        mgr.unlock(rel_id, row_id, TxId(1));
+        mgr.lock(rel_id, row_id, TxId(3)).unwrap();
+        order.lock().expect("order").push(3);
+        mgr.unlock(rel_id, row_id, TxId(3));
+        waiter.join().unwrap();
+        assert_eq!(*order.lock().expect("order"), vec![2, 3]);
+    }
+
     #[test]
     fn unlock_with_no_waiters_is_a_noop() {
         let mgr = RowLockManager::new(2, Duration::from_secs(1));

crates/kernel/src/storage/buffer.rs

@@ -90,6 +90,10 @@ struct FrameEntry {
     ready: Condvar,
 }
 
+/// Clock hand ceiling. `pin` saturates here; `evict_one` must be allowed
+/// this many decays plus one zero-usage visit.
+const CLOCK_MAX_USAGE: u8 = 5;
+
 #[derive(Debug)]
 pub(crate) struct FrameState {
     pub(crate) page: Option<Page>,
@@ -396,7 +400,7 @@ impl Inner {
                     continue;
                 }
                 state.pin_count += 1;
-                state.usage_count = state.usage_count.saturating_add(1).min(5);
+                state.usage_count = state.usage_count.saturating_add(1).min(CLOCK_MAX_USAGE);
                 drop(state);
                 return Ok(PageGuard { page_id, frame });
             }
@@ -611,7 +615,11 @@ impl Inner {
         }
 
         let start = self.clock_hand.fetch_add(1, Ordering::Relaxed);
-        for idx in 0..frames.len().saturating_mul(2) {
+        // Usage climbs to CLOCK_MAX_USAGE on each pin and only falls here.
+        // Two passes cannot decay a hot frame to zero, so a clean unpinned
+        // page was reported as unevictable.
+        let visits = frames.len().saturating_mul(CLOCK_MAX_USAGE as usize + 1);
+        for idx in 0..visits {
             let (page_id, frame) = &frames[(start + idx) % frames.len()];
             let mut state = frame
                 .state

crates/kernel/tests/storage_tests.rs

@@ -136,6 +136,25 @@ fn buffer_pool_eviction_skips_pinned_pages_and_stays_bounded() {
     drop(pinned);
 }
 
+#[test]
+fn buffer_pool_evicts_a_clean_frame_whose_usage_was_saturated() {
+    let temp = TempDir::new().unwrap();
+    let file =
+        Arc::new(PageFile::create(temp.path().join("data.redline"), TEST_PAGE_SIZE).unwrap());
+    let pool = BufferPool::new(Arc::clone(&file), 1).unwrap();
+    let first = pool.allocate(PageKind::Heap, RelId(1)).unwrap();
+    let first_id = first.page_id();
+    first.mark_dirty(Lsn(1)).unwrap();
+    drop(first);
+    pool.flush_page(first_id, Lsn(1)).unwrap();
+    for _ in 0..4 {
+        drop(pool.pin(first_id).unwrap());
+    }
+    let second = pool.allocate(PageKind::Heap, RelId(1)).unwrap();
+    assert_ne!(second.page_id(), first_id);
+    assert_eq!(pool.resident_pages(), 1);
+}
+
 #[test]
 fn buffer_pool_errors_when_all_pages_are_pinned() {
     let temp = TempDir::new().unwrap();

crates/kernel/tests/wal_tests.rs

@@ -164,6 +164,29 @@ fn wal_reader_scans_across_segment_rollover() {
     assert_eq!(records[2].payload, vec![b'c'; 32]);
 }
 
+#[test]
+fn exact_segment_fill_does_not_grow_past_the_limit() {
+    let temp = TempDir::new().unwrap();
+    // Header is 48 bytes. Payload 16 makes each record 64 bytes.
+    // Two records fill a 128-byte segment exactly; the third rotates.
+    let segment_bytes = 128;
+    let config = WalConfig {
+        segment_bytes,
+        ..WalConfig::default()
+    };
+    let mut manager = WalManager::create(temp.path(), config).unwrap();
+    let payload = vec![b'x'; 16];
+    for _ in 0..3 {
+        manager
+            .append(WalRecordKind::PageDelta, TxId(1), payload.clone())
+            .unwrap();
+    }
+    manager.flush().unwrap();
+    let first = temp.path().join(format!("{:020}.wal", 1_u64));
+    let len = std::fs::metadata(&first).unwrap().len();
+    assert_eq!(len, segment_bytes, "first segment grew to {len}");
+}
+
 #[test]
 fn wal_reader_stops_at_truncated_final_record() {
     let temp = TempDir::new().unwrap();

crates/sql/src/exec/morsel/arena.rs

@@ -44,12 +44,14 @@ impl<'a> BytesArena<'a> {
     /// pushes of bounded size cost O(N) total.
     pub fn push(&mut self, bytes: &[u8]) -> usize {
         let idx = self.len();
-        debug_assert!(
-            self.data.len() + bytes.len() <= u32::MAX as usize,
-            "BytesArena overflow"
-        );
+        let end = self
+            .data
+            .len()
+            .checked_add(bytes.len())
+            .expect("BytesArena length overflow");
+        let end_u32 = u32::try_from(end).expect("BytesArena offset does not fit in u32");
         self.data.extend_from_slice(bytes);
-        self.offsets.push(self.data.len() as u32);
+        self.offsets.push(end_u32);
         idx
     }
 
@@ -87,6 +89,16 @@ mod tests {
         assert_eq!(a.get(4), None);
     }
 
+    #[test]
+    fn an_offset_past_u32_max_does_not_fit() {
+        let past = (u32::MAX as usize).saturating_add(1);
+        assert!(u32::try_from(past).is_err());
+        let bump = Bump::new();
+        let mut arena = BytesArena::new(&bump, 1);
+        arena.push(b"ok");
+        assert_eq!(*arena.offsets.last().unwrap() as usize, arena.data.len());
+    }
+
     #[test]
     fn offsets_are_monotone() {
         let bump = Bump::new();