diff --git a/go/vt/vttablet/tabletserver/loadshed/codelq.go b/go/vt/vttablet/tabletserver/loadshed/codelq.go
index 99d70a4300c..820075767f9 100644
--- a/go/vt/vttablet/tabletserver/loadshed/codelq.go
+++ b/go/vt/vttablet/tabletserver/loadshed/codelq.go
@@ -70,7 +70,7 @@ import (
         | * timer: armed            |
         *---------------------------*
               |  ^            ^
-    timer     |  |            | dequeue/release w/ sojourn < target
+    timer     |  |            | dequeue w/ sojourn < target
     fires,    |  |            | or queue emptied (sets dropping=false)
     NOT       |  | timer fires,
     healthy   |  | healthy (dropping=false)
@@ -88,9 +88,9 @@ import (
 
     Health condition (checked each easing timer fire):
       healthy := dropping=false
-      dropping is unset by lockedOnGrant() when a granted request's queue-wait
-      sojourn < target, and by lockedPeek/lockedPopElem/lockedRemove/
-      lockedOnGrant when droppableLen reaches 0; reset each timer fire. Note:
+      dropping is unset by lockedDequeue() when a dequeued request's queue-wait
+      sojourn < target, and by lockedRemove/lockedDequeue when
+      droppableLen reaches 0; reset each timer fire. Note:
       while easing, each re-arm transiently re-marks dropping=true; the next
       fire re-evaluates health.
 
@@ -150,7 +150,6 @@ type (
 	// the mutex, which is defined in the files for the higher-level structure
 	CoDelQueue[T any] struct {
 		queue        *list.List
-		firstWaiting *list.Element
 		dropping     bool
 		dropNextNs   int64
 		count        int
@@ -164,7 +163,6 @@ type (
 		nowNs             func() int64
 		scheduleDropTimer func(delayNs int64)
 		stopDropTimer     func()
-		onPeekCleanup     func(*Request[T])
 	}
 )
 
@@ -172,7 +170,7 @@ func (e *DroppedRequestError) Error() string {
 	return "request dropped by CoDel queue"
 }
 
-func newCoDelQueue[T any](cfg CoDelConfig, nowNs func() int64, scheduleDropTimer func(delayNs int64), stopDropTimer func(), onPeekCleanup func(*Request[T])) *CoDelQueue[T] {
+func newCoDelQueue[T any](cfg CoDelConfig, nowNs func() int64, scheduleDropTimer func(delayNs int64), stopDropTimer func()) *CoDelQueue[T] {
 	q := &CoDelQueue[T]{
 		queue:             list.New(),
 		count:             1,
@@ -180,7 +178,6 @@ func newCoDelQueue[T any](cfg CoDelConfig, nowNs func() int64, scheduleDropTimer
 		nowNs:             nowNs,
 		scheduleDropTimer: scheduleDropTimer,
 		stopDropTimer:     stopDropTimer,
-		onPeekCleanup:     onPeekCleanup,
 	}
 	q.droppable.init()
 	return q
@@ -200,10 +197,6 @@ func (q *CoDelQueue[T]) lockedEnqueue(req *Request[T]) {
 	req.codelqEnqueuedAtNs = now
 	req.codelqElem = q.queue.PushBack(req)
 
-	if q.firstWaiting == nil {
-		q.firstWaiting = req.codelqElem
-	}
-
 	if req.isDroppable() {
 		q.droppableLen++
 		q.droppable.insert(req)
@@ -219,71 +212,30 @@ func (q *CoDelQueue[T]) lockedEnqueue(req *Request[T]) {
 	}
 }
 
-// lockedFirstWaiting returns the first not-yet-granted request in the queue.
-func (q *CoDelQueue[T]) lockedFirstWaiting() *Request[T] {
-	if q.firstWaiting == nil {
+// lockedPeek returns the first waiting request in the queue.
+func (q *CoDelQueue[T]) lockedPeek() *Request[T] {
+	first := q.queue.Front()
+	if first == nil {
 		return nil
 	}
-	return q.firstWaiting.Value.(*Request[T])
+	return first.Value.(*Request[T])
 }
 
-// lockedPeek returns the head request without removing it. As a side effect,
-// cleans up done-and-not-granted requests at the head (requests whose result
-// channel has an error). Empty queue transitions to healthy.
-func (q *CoDelQueue[T]) lockedPeek() *Request[T] {
-	for q.queue.Len() > 0 {
-		front := q.queue.Front()
-		req := front.Value.(*Request[T])
-		if req.signaledValue == nil {
+func (q *CoDelQueue[T]) lockedFind(match func(T) bool) *Request[T] {
+	for e := q.queue.Front(); e != nil; e = e.Next() {
+		req := e.Value.(*Request[T])
+		if match(req.value) {
 			return req
 		}
-		q.lockedAdvanceFirstWaiting(front)
-		q.queue.Remove(front)
-		req.codelqElem = nil
-		if req.isDroppable() {
-			q.droppableLen--
-			q.droppable.remove(req)
-		}
-		if q.onPeekCleanup != nil {
-			q.onPeekCleanup(req)
-		}
 	}
-	// Empty queue means the underlying resource is available.
-	q.dropping = false
-	return nil
+	return q.lockedPeek()
 }
 
-// lockedPopElem removes the given element from the queue, signals the request's
-// result channel, and updates bookkeeping. Use with care: this bypasses the
-// health-state transitions in peek/dequeue, so callers are responsible for
-// updating dropping state if appropriate.
-func (q *CoDelQueue[T]) lockedPopElem(elem *list.Element, err error) *Request[T] {
-	req := elem.Value.(*Request[T])
-	q.lockedAdvanceFirstWaiting(elem)
-	q.queue.Remove(elem)
-	req.codelqElem = nil
-
-	if req.signaledValue == nil {
-		req.signal(err)
-	}
-
-	if req.isDroppable() {
-		q.droppableLen--
-		q.droppable.remove(req)
-		if q.droppableLen == 0 && q.dropping {
-			q.dropping = false
-		}
-	}
-
-	return req
-}
-
-// lockedRemove removes a specific request from the queue without signaling it.
+// lockedRemove removes a specific request from the queue.
 func (q *CoDelQueue[T]) lockedRemove(r *Request[T]) {
 	if r.codelqElem == nil {
 		return
 	}
-	q.lockedAdvanceFirstWaiting(r.codelqElem)
 	q.queue.Remove(r.codelqElem)
 	r.codelqElem = nil
 
@@ -296,44 +248,19 @@ func (q *CoDelQueue[T]) lockedRemove(r *Request[T]) {
 	}
 }
 
-func (q *CoDelQueue[T]) lockedOnGrant(r *Request[T]) {
-	// CoDel health check, measured at grant: if this request's queue-wait
+func (q *CoDelQueue[T]) lockedDequeue(r *Request[T]) {
+	// CoDel health check, measured at dequeue: if this request's queue-wait
 	// (now - enqueue) was under target, the system is healthy — leave the
 	// dropping state. Separate from the droppableLen==0 clear below.
 	if q.nowNs()-r.codelqEnqueuedAtNs < q.cfg.TargetNs() {
 		q.dropping = false
 	}
-	if r.isDroppable() {
-		q.droppable.remove(r)
-		q.droppableLen--
-		if q.droppableLen == 0 {
-			q.dropping = false
-		}
-	}
-	q.lockedAdvanceFirstWaiting(r.codelqElem)
-	q.queue.Remove(r.codelqElem)
-	r.codelqElem = nil
-}
-
-// lockedAdvanceFirstWaiting advances the firstWaiting pointer past elem if
-// elem is the current firstWaiting. elem is a queue entry that is no longer
-// waiting — either because it was granted, removed, or dropped.
-func (q *CoDelQueue[T]) lockedAdvanceFirstWaiting(elem *list.Element) {
-	if q.firstWaiting != elem {
-		return
-	}
-	for e := elem.Next(); e != nil; e = e.Next() {
-		if e.Value.(*Request[T]).signaledValue == nil {
-			q.firstWaiting = e
-			return
-		}
-	}
-	q.firstWaiting = nil
+	q.lockedRemove(r)
 }
 
 // lockedFindLowestPriorityDroppable finds the lowest-priority droppable
 // element in the queue — the oldest one at the lowest priority present — or nil
-// if none exists. O(1) via the droppable priority index (see droppableIndex[T]).
+// if none exists. O(1) via the droppable priority index (see droppableIndex).
 func (q *CoDelQueue[T]) lockedFindLowestPriorityDroppable() *list.Element {
 	req := q.droppable.min()
 	if req == nil {
@@ -343,7 +270,7 @@ func (q *CoDelQueue[T]) lockedFindLowestPriorityDroppable() *list.Element {
 }
 
 // lockedRunTimer runs the CoDel drop logic. It is invoked both by the backstop
-// timer and synchronously from the release/dequeue path, so shedding is driven
+// timer and synchronously from the dequeue path, so shedding is driven
 // as slots free rather than waiting for the (possibly late) timer to fire.
 func (q *CoDelQueue[T]) lockedRunTimer(dropFn func() bool) {
 	q.lockedRunTimerLimited(dropFn, -1)
@@ -370,8 +297,8 @@ func (q *CoDelQueue[T]) lockedRunTimerLimited(dropFn func() bool, maxDrops int)
 // lockedAdvance runs the clock-driven core of the CoDel control law: it sheds
 // every request that is due (now >= dropNextNs) while dropping, ramping count,
 // then eases count back down while healthy. Every action is gated on the fresh
-// `now`, so calling it is idempotent and safe outside the timer — the release
-// (dequeue) path invokes it to shed stale requests in real time rather than
+// `now`, so calling it is idempotent and safe outside the timer — the dequeue
+// path invokes it to shed stale requests in real time rather than
 // waiting on the possibly-late backstop timer. It does NOT arm/disarm the timer.
 func (q *CoDelQueue[T]) lockedAdvance(now int64, dropFn func() bool) {
 	q.lockedAdvanceLimited(now, dropFn, -1)
@@ -406,8 +333,7 @@ func (q *CoDelQueue[T]) lockedAdvanceLimited(now int64, dropFn func() bool, maxD
 		// If a request arrived before the end of the interval then it is a
 		// candidate for dropping.
 		if q.droppableLen > 0 {
-			q.lockedPeek() // cleanup
-			if first := q.lockedFirstWaiting(); first != nil && first.codelqEnqueuedAtNs < q.dropNextNs {
+			if first := q.lockedPeek(); first != nil && first.codelqEnqueuedAtNs < q.dropNextNs {
 				q.dropping = true
 			}
 		}
diff --git a/go/vt/vttablet/tabletserver/loadshed/codelq_test.go b/go/vt/vttablet/tabletserver/loadshed/codelq_test.go
index 8f5dc996d61..12d461caeeb 100644
--- a/go/vt/vttablet/tabletserver/loadshed/codelq_test.go
+++ b/go/vt/vttablet/tabletserver/loadshed/codelq_test.go
@@ -26,9 +26,6 @@ import (
 
 type testRequest = Request[struct{}]
 type testCoDelQueue = CoDelQueue[struct{}]
-type testValvedCoDelQueue = ValvedCoDelQueue[struct{}]
-type testSnake = Snake[struct{}]
-type testSafeUnlock = SafeUnlock[struct{}]
 
 func defaultTestConfig() CoDelConfig {
 	return CoDelConfig{
@@ -87,7 +84,7 @@ func (r *testDropTimerRecorder) reset() {
 
 func newTestQueue(cfg CoDelConfig, clock *testClock) (*testCoDelQueue, *testDropTimerRecorder) {
 	rec := &testDropTimerRecorder{}
-	q := newCoDelQueue[struct{}](cfg, clock.nowFunc, rec.schedule, rec.stop, nil)
+	q := newCoDelQueue[struct{}](cfg, clock.nowFunc, rec.schedule, rec.stop)
 	return q, rec
 }
 
@@ -145,14 +142,13 @@ func TestCoDelQueue_Enqueue_UndroppableNoSchedule(t *testing.T) {
 	assert.Equal(t, 0, q.droppableLen)
 }
 
-// testDequeue grants the oldest waiting request.
+// testDequeue removes the oldest waiting request.
 func testDequeue(q *testCoDelQueue) *testRequest {
-	req := q.lockedFirstWaiting()
+	req := q.lockedPeek()
 	if req == nil {
 		return nil
 	}
-	q.lockedOnGrant(req)
-	req.signal(grantSentinel)
+	q.lockedDequeue(req)
 	return req
 }
 
@@ -175,7 +171,7 @@ func TestCoDelQueue_FirstWaiting_FIFO(t *testing.T) {
 	assert.Same(t, r3, d3)
 }
 
-func TestCoDelQueue_OnGrant_DecrementsDroppableLen(t *testing.T) {
+func TestCoDelQueue_Dequeue_DecrementsDroppableLen(t *testing.T) {
 	clock := newTestClock()
 	q, _ := newTestQueue(defaultTestConfig(), clock)
 
@@ -187,7 +183,7 @@ func TestCoDelQueue_OnGrant_DecrementsDroppableLen(t *testing.T) {
 	assert.Equal(t, 1, q.droppableLen)
 }
 
-func TestCoDelQueue_Grant_ExitsDroppingOnTarget(t *testing.T) {
+func TestCoDelQueue_Dequeue_ExitsDroppingOnTarget(t *testing.T) {
 	clock := newTestClock()
 	cfg := defaultTestConfig()
 	cfg.TargetNs = func() int64 { return 1_000_000 }
@@ -209,21 +205,8 @@ func TestCoDelQueue_FirstWaiting_Empty(t *testing.T) {
 	clock := newTestClock()
 	q, _ := newTestQueue(defaultTestConfig(), clock)
 
-	req := q.lockedFirstWaiting()
-	assert.Nil(t, req)
-}
-
-// --- Peek tests ---
-
-func TestCoDelQueue_Peek_Empty(t *testing.T) {
-	clock := newTestClock()
-	q, _ := newTestQueue(defaultTestConfig(), clock)
-
-	q.dropping = true
 	req := q.lockedPeek()
-
 	assert.Nil(t, req)
-	assert.False(t, q.dropping)
 }
 
 func TestCoDelQueue_Peek_ReturnsHead(t *testing.T) {
@@ -238,33 +221,16 @@ func TestCoDelQueue_Peek_ReturnsHead(t *testing.T) {
 	assert.Equal(t, 2, q.lockedLen())
 }
 
-func TestCoDelQueue_Peek_CleansHeadCancelled(t *testing.T) {
-	clock := newTestClock()
-	q, _ := newTestQueue(defaultTestConfig(), clock)
-
-	r1 := testEnqueue(q, 0)
-	r2 := testEnqueue(q, 0)
-
-	r1.signal(&DroppedRequestError{})
-	q.lockedOnGrant(r1)
-
-	peeked := q.lockedPeek()
-	assert.Same(t, r2, peeked)
-	assert.Equal(t, 1, q.lockedLen())
-}
-
-func TestCoDelQueue_OnGrant_EvictsFromListImmediately(t *testing.T) {
+func TestCoDelQueue_Dequeue_EvictsFromListImmediately(t *testing.T) {
 	clock := newTestClock()
 	q, _ := newTestQueue(defaultTestConfig(), clock)
 
 	r1 := testEnqueue(q, 0)
 	require.NotNil(t, r1.codelqElem)
 
-	r1.signal(grantSentinel)
-	q.lockedOnGrant(r1)
+	q.lockedDequeue(r1)
 
 	assert.Nil(t, r1.codelqElem)
-	assert.Nil(t, q.lockedPeek())
 	assert.Equal(t, 0, q.lockedLen())
 }
 
@@ -280,12 +246,10 @@ func TestCoDelQueue_FindLowestPriorityDroppable_Basic(t *testing.T) {
 
 	elem := q.lockedFindLowestPriorityDroppable()
 	require.NotNil(t, elem)
-	dropped := q.lockedPopElem(elem, &DroppedRequestError{})
+	dropped := elem.Value.(*testRequest)
+	q.lockedRemove(dropped)
 	assert.Equal(t, float64(1), dropped.priority)
 	assert.Equal(t, 2, q.lockedLen())
-
-	err := <-dropped.signalChan
-	assert.IsType(t, &DroppedRequestError{}, err)
 }
 
 func TestCoDelQueue_FindLowestPriorityDroppable_ZeroInstantPick(t *testing.T) {
@@ -311,25 +275,10 @@ func TestCoDelQueue_DropSkipsUndroppable(t *testing.T) {
 	elem := q.lockedFindLowestPriorityDroppable()
 	require.NotNil(t, elem)
 	assert.Same(t, droppable, elem.Value.(*testRequest))
-	q.lockedPopElem(elem, &DroppedRequestError{})
+	q.lockedRemove(droppable)
 	assert.Equal(t, 1, q.lockedLen())
 }
 
-func TestCoDelQueue_DropSkipsDone(t *testing.T) {
-	clock := newTestClock()
-	q, _ := newTestQueue(defaultTestConfig(), clock)
-
-	r1 := testEnqueue(q, 0)
-	r2 := testEnqueue(q, 5)
-
-	r1.signal(grantSentinel)
-	q.lockedOnGrant(r1)
-
-	elem := q.lockedFindLowestPriorityDroppable()
-	require.NotNil(t, elem)
-	assert.Same(t, r2, elem.Value.(*testRequest))
-}
-
 func TestCoDelQueue_DropAllUndroppable_ReturnsNil(t *testing.T) {
 	clock := newTestClock()
 	q, _ := newTestQueue(defaultTestConfig(), clock)
@@ -421,7 +370,7 @@ func TestCoDelQueue_RunScheduledDrop_EntersDropping(t *testing.T) {
 		if elem == nil {
 			return false
 		}
-		q.lockedPopElem(elem, &DroppedRequestError{})
+		q.lockedRemove(elem.Value.(*testRequest))
 		return true
 	}
 	rec.reset()
@@ -443,7 +392,7 @@ func TestCoDelQueue_RunScheduledDrop_NothingDroppable(t *testing.T) {
 		if elem == nil {
 			return false
 		}
-		q.lockedPopElem(elem, &DroppedRequestError{})
+		q.lockedRemove(elem.Value.(*testRequest))
 		return true
 	}
 	q.lockedRunTimer(dropFn)
@@ -472,34 +421,33 @@ func TestCoDelQueue_Remove_AlreadyDone(t *testing.T) {
 
 	r1 := testEnqueue(q, 0)
 
-	r1.signal(grantSentinel)
-	q.lockedOnGrant(r1)
+	q.lockedDequeue(r1)
 
 	q.lockedRemove(r1)
 	assert.Equal(t, 0, q.lockedLen())
 }
 
-// --- OnGrant tests ---
+// --- Dequeue tests ---
 
-func TestCoDelQueue_OnGrant(t *testing.T) {
+func TestCoDelQueue_Dequeue(t *testing.T) {
 	clock := newTestClock()
 	q, _ := newTestQueue(defaultTestConfig(), clock)
 
 	r1 := testEnqueue(q, 0)
 	assert.Equal(t, 1, q.droppableLen)
 
-	q.lockedOnGrant(r1)
+	q.lockedDequeue(r1)
 	assert.Equal(t, 0, q.droppableLen)
 }
 
-func TestCoDelQueue_OnGrant_AlreadyNotDroppable(t *testing.T) {
+func TestCoDelQueue_Dequeue_AlreadyNotDroppable(t *testing.T) {
 	clock := newTestClock()
 	q, _ := newTestQueue(defaultTestConfig(), clock)
 
 	r1 := testEnqueue(q, PriorityUndroppable)
 	assert.Equal(t, 0, q.droppableLen)
 
-	q.lockedOnGrant(r1)
+	q.lockedDequeue(r1)
 	assert.Equal(t, 0, q.droppableLen)
 }
 
@@ -532,7 +480,7 @@ func TestCoDelQueue_FastMoving_NoDrop(t *testing.T) {
 	assert.Equal(t, enqueued, dequeued, "fast-moving queue should not drop")
 }
 
-func TestCoDelQueue_Grant_TransitionsToEasing(t *testing.T) {
+func TestCoDelQueue_Dequeue_TransitionsToEasing(t *testing.T) {
 	clock := newTestClock()
 	cfg := CoDelConfig{
 		IntervalNs:     func() int64 { return 1_000_000 },
@@ -552,8 +500,8 @@ func TestCoDelQueue_Grant_TransitionsToEasing(t *testing.T) {
 	q.count = 4
 	q.dropNextNs = clock.now + cfg.IntervalNs()
 
-	// Grant r1 with a fast sojourn.
-	q.lockedOnGrant(r1)
+	// Dequeue r1 with a fast sojourn.
+	q.lockedDequeue(r1)
 
 	assert.False(t, q.dropping, "should exit dropping state")
 	assert.Equal(t, 4, q.count, "count preserved for easing — timer will halve it when it fires")
@@ -561,37 +509,36 @@ func TestCoDelQueue_Grant_TransitionsToEasing(t *testing.T) {
 
 // --- Sojourn measurement tests ---
 //
-// Sojourn is always measured at grant (dispatch): pure queue-wait time, not
-// including resource hold time. Completion (Release) never records sojourn.
+// Sojourn is always measured at dequeue: pure queue-wait time.
 
-func TestCoDelQueue_Sojourn_FastGrantClearsDropping(t *testing.T) {
+func TestCoDelQueue_Sojourn_FastDequeueClearsDropping(t *testing.T) {
 	clock := newTestClock()
 	q, _ := newTestQueue(defaultTestConfig(), clock) // TargetNs = 50ms
 
 	clock.now = 0
 	r := testEnqueue(q, 0)
-	testEnqueue(q, 0) // second droppable keeps droppableLen > 0 after the grant
+	testEnqueue(q, 0) // second droppable keeps droppableLen > 0 after dequeue
 	q.dropping = true
 
-	// Grant after a short queue-wait (< target) clears dropping at grant.
+	// Dequeue after a short queue-wait (< target) clears dropping.
 	// droppableLen stays > 0, so the clear must come from the sojourn check.
 	clock.now = 10 * 1_000_000 // 10ms < 50ms target
-	q.lockedOnGrant(r)
-	assert.False(t, q.dropping, "fast queue-wait clears dropping at grant")
+	q.lockedDequeue(r)
+	assert.False(t, q.dropping, "fast queue-wait clears dropping at dequeue")
 }
 
-func TestCoDelQueue_Sojourn_SlowGrantKeepsDropping(t *testing.T) {
+func TestCoDelQueue_Sojourn_SlowDequeueKeepsDropping(t *testing.T) {
 	clock := newTestClock()
 	q, _ := newTestQueue(defaultTestConfig(), clock) // TargetNs = 50ms
 
 	clock.now = 0
 	r := testEnqueue(q, 0)
-	testEnqueue(q, 0) // second droppable keeps droppableLen > 0 after the grant
+	testEnqueue(q, 0) // second droppable keeps droppableLen > 0 after dequeue
 	q.dropping = true
 
-	// Grant after a long queue-wait (> target) must NOT clear dropping.
+	// Dequeue after a long queue-wait (> target) must NOT clear dropping.
 	clock.now = 100 * 1_000_000 // 100ms > 50ms target
-	q.lockedOnGrant(r)
+	q.lockedDequeue(r)
 	assert.True(t, q.dropping, "slow queue-wait keeps dropping")
 }
 
@@ -739,7 +686,7 @@ func TestCoDelQueue_Easing_DroppableLen_ReentersDroppingWithCurrentCount(t *test
 		if elem == nil {
 			return false
 		}
-		q.lockedPopElem(elem, &DroppedRequestError{})
+		q.lockedRemove(elem.Value.(*testRequest))
 		return true
 	}
 
@@ -751,7 +698,7 @@ func TestCoDelQueue_Easing_DroppableLen_ReentersDroppingWithCurrentCount(t *test
 	assert.True(t, rec.scheduled, "timer should re-arm for continued dropping")
 }
 
-func TestCoDelQueue_Easing_GrantDoesNotResetCount(t *testing.T) {
+func TestCoDelQueue_Easing_DequeueDoesNotResetCount(t *testing.T) {
 	clock := newTestClock()
 	cfg := defaultTestConfig()
 	cfg.TargetNs = func() int64 { return 1_000_000 }
@@ -763,8 +710,7 @@ func TestCoDelQueue_Easing_GrantDoesNotResetCount(t *testing.T) {
 
 	clock.now = 0
 	req := testEnqueue(q, 0)
-	q.lockedOnGrant(req)
-	req.signal(grantSentinel)
+	q.lockedDequeue(req)
 
 	assert.False(t, q.dropping, "should exit dropping")
 	assert.Equal(t, 10, q.count, "count should NOT be reset on transition to healthy")
@@ -778,7 +724,7 @@ func TestCoDelQueue_Easing_DroppingToHealthy_TimerStillFires(t *testing.T) {
 	q, rec := newTestQueue(cfg, clock)
 
 	// Easing with a high count and no droppable entries, armed and due. dropping
-	// is false: a prior grant met target (or the queue drained), so this interval
+	// is false: a prior dequeue met target (or the queue drained), so this interval
 	// is presumed healthy and only decays count.
 	clock.now = 1_000_000_000
 	q.dropping = false
@@ -802,7 +748,7 @@ func TestCoDelQueue_Easing_FullSequence(t *testing.T) {
 	q, rec := newTestQueue(cfg, clock)
 
 	// Easing from count=16, no droppable entries. dropping is false (presumed
-	// healthy: a grant met target or the queue drained), so each fire only
+	// healthy: a dequeue met target or the queue drained), so each fire only
 	// decays count. dropNextNs is seeded to now so the first fire is on time;
 	// each iteration then advances by exactly the scheduled delay.
 	q.dropping = false
@@ -867,7 +813,7 @@ func TestCoDelQueue_SlowMoving_Drops(t *testing.T) {
 		if elem == nil {
 			return false
 		}
-		q.lockedPopElem(elem, &DroppedRequestError{})
+		q.lockedRemove(elem.Value.(*testRequest))
 		return true
 	}
 	q.lockedRunTimer(dropFn)
@@ -889,3 +835,68 @@ func TestCoDelQueue_SlowStart_EnqueueArms(t *testing.T) {
 	assert.True(t, rec.scheduled, "slow-start: droppable enqueue arms the timer")
 	assert.Equal(t, int64(6_000_000_000), q.dropNextNs, "slow-start: first enqueue seeds dropNextNs = now + interval")
 }
+
+func TestSnakeQueue_DequeueRemovesRequest(t *testing.T) {
+	s := NewSnake[string](SnakeConfig{CoDel: defaultTestConfig()})
+
+	req, dropped := s.Enqueue("value", "", 0)
+	require.Empty(t, dropped)
+	dequeued, ok, dropped := s.Dequeue()
+	require.True(t, ok)
+	require.Equal(t, "value", dequeued)
+	require.Empty(t, dropped)
+	require.Equal(t, 0, s.q.lockedLen())
+	require.NotNil(t, req.signaledValue)
+}
+
+func TestSnakeQueue_CancelRemovesRequest(t *testing.T) {
+	s := NewSnake[string](SnakeConfig{CoDel: defaultTestConfig()})
+	req, dropped := s.Enqueue("value", "", 0)
+	require.Empty(t, dropped)
+
+	cancelled, dropped := s.Cancel(req)
+	require.True(t, cancelled)
+	require.Empty(t, dropped)
+	require.Equal(t, 0, s.q.lockedLen())
+	cancelled, dropped = s.Cancel(req)
+	require.False(t, cancelled)
+	require.Empty(t, dropped)
+}
+
+func TestSnakeQueue_CancelRemovesValveWaiter(t *testing.T) {
+	s := NewSnake[string](SnakeConfig{CoDel: defaultTestConfig()})
+	first, dropped := s.Enqueue("first", "valve", 0)
+	require.Empty(t, dropped)
+	second, dropped := s.Enqueue("second", "valve", 0)
+	require.Empty(t, dropped)
+
+	cancelled, dropped := s.Cancel(second)
+	require.True(t, cancelled)
+	require.Empty(t, dropped)
+
+	dequeued, ok, dropped := s.Dequeue()
+	require.True(t, ok)
+	require.Equal(t, "first", dequeued)
+	require.Empty(t, dropped)
+	dequeued, ok, dropped = s.Dequeue()
+	require.False(t, ok)
+	require.Empty(t, dequeued)
+	require.Empty(t, dropped)
+	require.NotNil(t, first.signaledValue)
+}
+
+func TestSnakeQueue_DisabledDoesNotDrop(t *testing.T) {
+	config := SnakeConfig{
+		CoDel:               defaultTestConfig(),
+		LoadsheddingAllowed: func() bool { return false },
+	}
+	s := NewSnake[struct{}](config)
+	for range 6 {
+		_, dropped := s.Enqueue(struct{}{}, "", 0)
+		require.Empty(t, dropped)
+	}
+	s.q.codelq.dropNextNs = 1
+
+	_, _, dropped := s.Dequeue()
+	require.Empty(t, dropped)
+}
diff --git a/go/vt/vttablet/tabletserver/loadshed/dequeue_shed_test.go b/go/vt/vttablet/tabletserver/loadshed/dequeue_shed_test.go
index 59b5906db36..4f5e67e590c 100644
--- a/go/vt/vttablet/tabletserver/loadshed/dequeue_shed_test.go
+++ b/go/vt/vttablet/tabletserver/loadshed/dequeue_shed_test.go
@@ -30,14 +30,14 @@ func dropAllFn(q *testCoDelQueue) func() bool {
 		if elem == nil {
 			return false
 		}
-		q.lockedPopElem(elem, &DroppedRequestError{})
+		q.lockedRemove(elem.Value.(*testRequest))
 		return true
 	}
 }
 
 // TestCoDelQueue_DequeueSheds_AfterEpisodeTornDown reproduces the bug where the
 // dequeue path could not re-establish a dropping episode on its own: once
-// lockedOnGrant cleared `dropping` (a grant whose sojourn was under target), only
+// lockedDequeue cleared `dropping` (a dequeue whose sojourn was under target), only
 // the backstop timer re-armed it. With the timer effectively off (large
 // MinDropDelay), the dequeue path must run the full CoDel logic and shed stale
 // waiters without a timer fire.
@@ -56,8 +56,8 @@ func TestCoDelQueue_DequeueSheds_AfterEpisodeTornDown(t *testing.T) {
 	}
 	assert.True(t, q.dropping, "first droppable enqueue should arm an episode")
 
-	// Simulate the episode teardown that the release path triggers: a grant whose
-	// sojourn is under target clears `dropping` in lockedOnGrant. We reproduce the
+	// Simulate the episode teardown that dequeue triggers: a request whose
+	// sojourn is under target clears `dropping` in lockedDequeue. We reproduce the
 	// cleared state directly (this is the state the dequeue path must recover from).
 	q.dropping = false
 
@@ -65,7 +65,7 @@ func TestCoDelQueue_DequeueSheds_AfterEpisodeTornDown(t *testing.T) {
 	// drops are due. The backstop timer never fires (MinDropDelay=1s).
 	clock.advance(5_000_000_000) // 5s
 
-	// Drive the dequeue path repeatedly (as releases would). It must re-establish
+	// Drive the dequeue path repeatedly. It must re-establish
 	// the episode and shed the stale backlog with no timer fire.
 	before := q.droppableLen
 	for i := 0; i < backlog+2; i++ {
@@ -79,12 +79,7 @@ func TestCoDelQueue_DequeueSheds_AfterEpisodeTornDown(t *testing.T) {
 }
 
 // TestValved_Drop_DefersSignalOutsideLock asserts the deferral contract: a batch
-// drop pass MARKS each dropped request (signaledValue set, so under-lock readers
-// see it) but does NOT send on its channel — the sends are collected in
-// pendingSignals and delivered only when the caller drains them (mirroring the
-// send-after-unlock in Snake). This keeps the goready storm out of the critical
-// section.
-func TestValved_Drop_DefersSignalOutsideLock(t *testing.T) {
+func TestValved_DropReturnsPendingRequests(t *testing.T) {
 	clock := newTestClock()
 	sq, _ := newValvedQueue(clock)
 	// Fast target/interval so drops are due immediately once armed.
@@ -94,9 +89,8 @@ func TestValved_Drop_DefersSignalOutsideLock(t *testing.T) {
 	// A backlog of distinct-valve droppable requests (distinct valves so each is
 	// its own droppable representative and all are eligible to shed).
 	const backlog = 5
-	reqs := make([]*testRequest, backlog)
-	for i := range reqs {
-		reqs[i] = sq.lockedEnqueue(string(rune('a'+i)), 0)
+	for i := range backlog {
+		sq.lockedEnqueue(string(rune('a'+i)), 0)
 	}
 	require.True(t, sq.codelq.dropping, "first droppable enqueue arms an episode")
 
@@ -108,33 +102,12 @@ func TestValved_Drop_DefersSignalOutsideLock(t *testing.T) {
 
 	sq.lockedRunTimer()
 
-	// Every shed request is MARKED under the lock...
-	dropped := 0
-	for _, r := range reqs {
-		if r.signaledValue != nil {
-			dropped++
-			// ...but NOT yet sent: its buffered channel is still empty.
-			assert.Empty(t, r.signalChan, "drop must not send on the channel under the lock")
-		}
-	}
-	require.Positive(t, dropped, "the pass should have shed some requests")
-
-	// The marked requests are queued for deferred delivery.
-	pending := sq.lockedTakePendingSignals()
-	assert.Len(t, pending, dropped, "every marked drop is queued for deferred send")
-	assert.Nil(t, sq.lockedTakePendingSignals(), "taking again yields nothing (ownership transferred)")
-
-	// Delivering the deferred signals sends exactly the drop error to each waiter.
-	for _, r := range pending {
-		r.sendSignal()
-		select {
-		case v := <-r.signalChan:
-			_, isDrop := v.(*DroppedRequestError)
-			assert.True(t, isDrop, "deferred send delivers the drop rejection")
-		default:
-			t.Fatal("sendSignal did not deliver on the channel")
-		}
+	pending := sq.lockedTakePendingDrops()
+	require.NotEmpty(t, pending, "the pass should have shed some requests")
+	for _, req := range pending {
+		assert.NotNil(t, req.signaledValue)
 	}
+	assert.Nil(t, sq.lockedTakePendingDrops(), "taking again yields nothing (ownership transferred)")
 }
 
 func TestValved_DisabledDropAdvancesCoDelWithoutDropping(t *testing.T) {
diff --git a/go/vt/vttablet/tabletserver/loadshed/request.go b/go/vt/vttablet/tabletserver/loadshed/request.go
index f1dcb480655..706a267ff33 100644
--- a/go/vt/vttablet/tabletserver/loadshed/request.go
+++ b/go/vt/vttablet/tabletserver/loadshed/request.go
@@ -24,24 +24,21 @@ import (
 
 type (
 	// Request represents an entry in the CoDel queue. Named a 'request' since
-	// it may be rejected(dropped) or granted. Each request owns a signalChan
-	// that receives nil on grant or a *DroppedRequestError on drop. The
-	// signaledValue allows non-consuming inspection of signal state (used by
-	// lockedPeek to avoid channel pop/push-back): nil means unsignaled, and any
-	// non-nil value means the request has completed.
+	// it may be dropped or dequeued. signaledValue allows the queue to inspect
+	// terminal state without removing anything from the queue.
 	Request[T any] struct {
 		priority           float64
 		codelqEnqueuedAtNs int64
-		signalChan         chan error
-		signaledValue      error
 		codelqElem         *list.Element
 		valveID            string
+		signaledValue      error
+		value              T
 
 		// bucketElem locates this request in the droppableIndex while it is a
 		// droppable queue entry: it is the request's node in its priority
 		// bucket's FIFO list, enabling O(1) removal. bucketIdx is the bucket that
 		// node lives in (0..maxPriorityBucket, or overflowBucket). bucketElem is
-		// nil when the request is not indexed (undroppable, granted, or removed).
+		// nil when the request is not indexed (undroppable, dequeued, or removed).
 		bucketElem *list.Element
 		bucketIdx  int
 	}
@@ -49,16 +46,15 @@ type (
 
 // PriorityUndroppable is a sentinel priority indicating a request that must
 // never be dropped by CoDel. We use negative infinity so it's distinguishable
-// from any real priority value. Callers may pass it to Acquire to force a
-// request undroppable (e.g. health-check queries against system schemas).
+// from any real priority value (e.g. health-check queries against system
+// schemas).
 var PriorityUndroppable = math.Inf(-1)
 
-var grantSentinel = errors.New("granted") //nolint:staticcheck // not an error; sentinel for non-consuming signal state inspection
+var grantSentinel = errors.New("granted") //nolint:staticcheck // sentinel for request state
 
 func newRequest[T any](priority float64) *Request[T] {
 	return &Request[T]{
-		priority:   priority,
-		signalChan: make(chan error, 1),
+		priority: priority,
 	}
 }
 
@@ -66,33 +62,9 @@ func (r *Request[T]) isDroppable() bool {
 	return r.priority != PriorityUndroppable
 }
 
-// Pass grantSentinel on grant and *DroppedRequestError on drop. Must be called
-// exactly once per request. signal marks and sends in one step; use it only
-// where the caller holds no lock across the send, or the send is single-shot
-// (grant, cancel). Batch paths mark under the lock and send after (see
-// markSignaled / sendSignal).
 func (r *Request[T]) signal(val error) {
-	r.markSignaled(val)
-	r.sendSignal()
-}
-
-// markSignaled records the request's terminal outcome without touching the
-// channel. It is safe (and intended) to call under the queue mutex: all
-// under-lock readers key off signaledValue, so it must be set synchronously.
-// Returns true if this call performed the nil->set transition, so the caller can
-// enqueue the deferred send exactly once. Panics on a second distinct signal.
-func (r *Request[T]) markSignaled(val error) bool {
 	if r.signaledValue != nil {
 		panic("loadshed: signal called more than once")
 	}
 	r.signaledValue = val
-	return true
-}
-
-// sendSignal delivers the previously-marked value on the request's channel. It
-// performs the goready of the parked Acquire goroutine, so batch drop paths call
-// it AFTER releasing the queue mutex to keep the wakeup storm out of the critical
-// section. Must be called exactly once, after markSignaled.
-func (r *Request[T]) sendSignal() {
-	r.signalChan <- r.signaledValue
 }
diff --git a/go/vt/vttablet/tabletserver/loadshed/snake.go b/go/vt/vttablet/tabletserver/loadshed/snake.go
index 4d64bb21874..a9b0d6a28d2 100644
--- a/go/vt/vttablet/tabletserver/loadshed/snake.go
+++ b/go/vt/vttablet/tabletserver/loadshed/snake.go
@@ -17,10 +17,7 @@ limitations under the License.
 package loadshed
 
 import (
-	"context"
-	"log"
 	"strconv"
-	"sync"
 	"sync/atomic"
 	"time"
 
@@ -31,22 +28,15 @@ type (
 	// SnakeConfig configures a Snake. Functions are used to allow dynamic runtime
 	// tuning.
 	SnakeConfig struct {
-		Name                string
 		CoDel               CoDelConfig
-		Capacity            func() int
 		LoadsheddingAllowed func() bool
-		AcquireError        func() error
-		ReleaseCBs          []func(error)
+		DropTimerFired      func()
 	}
 
-	// Snake is a CoDel-based load-shedding gate with dynamic capacity. Up to
-	// Capacity() concurrent holders are allowed. Acquire requests are either
-	// granted or dropped, each within a timely manner.
+	// Snake is a CoDel-based load-shedding queue. It decides which waiting
+	// request may proceed; the caller owns execution capacity and handoff.
 	Snake[T any] struct {
-		mu sync.Mutex
-
 		q              *ValvedCoDelQueue[T]
-		holders        map[*Request[T]]struct{}
 		dropTimer      *time.Timer
 		dropTimerArmed bool
 		// dropTimerExpectedNs is the clock time the drop timer was scheduled to
@@ -58,12 +48,12 @@ type (
 		shedCount atomic.Int64
 		// shedByPriority breaks shedCount down by the shed request's priority label
 		// (the caller's original query priority: "0" most important .. "100" least,
-		// "overflow"), so operators can see whether the gate is correctly shedding
+		// "overflow"), so operators can see whether the queue is correctly shedding
 		// low-priority traffic first rather than eating high-priority requests. Nil
 		// until PublishStats registers it (tests and the benchmark build a Snake
 		// without it); the shed path nil-checks. Its sum equals shedCount.
 		shedByPriority *stats.CountersWithMultiLabels
-		// acquireByPriority counts every Acquire, labeled by the same caller
+		// acquireByPriority counts every enqueue, labeled by the same caller
 		// priority as shedByPriority, so shed rate per priority class can be
 		// computed exactly (shedByPriority / acquireByPriority) rather than from
 		// assumed offered-load weights. Nil until PublishStats registers it.
@@ -72,23 +62,13 @@ type (
 		sojourn      *stats.Histogram
 		queueLen     *stats.Histogram
 		droppableLen *stats.Histogram
-		holderCount  *stats.Histogram
 		interval     *stats.Histogram
 		dropCount    *stats.Histogram
 		timerLag     *stats.Histogram
 		valveDepth   *stats.Histogram
 
 		droppingNanos   atomic.Int64
-		droppingSinceNs int64
-	}
-
-	// SafeUnlock is a handle for releasing a slot. Only the goroutine that
-	// acquired the slot should call Release. Release is idempotent.
-	SafeUnlock[T any] struct {
-		s    *Snake[T]
-		req  *Request[T]
-		once sync.Once
-		err  error
+		droppingSinceNs atomic.Int64
 	}
 )
 
@@ -98,16 +78,14 @@ func defaultClock() int64 {
 	return time.Since(epoch).Nanoseconds()
 }
 
-// NewSnake creates a new CoDel-based load-shedding gate.
+// NewSnake creates a new CoDel-based load-shedding queue.
 func NewSnake[T any](cfg SnakeConfig) *Snake[T] {
 	s := &Snake[T]{
 		cfg:          cfg,
 		clockFunc:    defaultClock,
-		holders:      make(map[*Request[T]]struct{}),
 		sojourn:      stats.NewHistogram("", "", loadshedBucketCutoffs),
 		queueLen:     stats.NewHistogram("", "", lengthBucketCutoffs),
 		droppableLen: stats.NewHistogram("", "", lengthBucketCutoffs),
-		holderCount:  stats.NewHistogram("", "", holderBucketCutoffs),
 		interval:     stats.NewHistogram("", "", intervalBucketCutoffs),
 		dropCount:    stats.NewHistogram("", "", lengthBucketCutoffs),
 		timerLag:     stats.NewHistogram("", "", loadshedBucketCutoffs),
@@ -126,222 +104,84 @@ func (s *Snake[T]) lockedObserveValveDepth(valveID string) {
 	s.valveDepth.Add(int64(s.q.lockedValveDepth(valveID)))
 }
 
-func (s *Snake[T]) capacity() int {
-	if s.cfg.Capacity == nil {
-		return 1
-	}
-	return max(s.cfg.Capacity(), 1)
-}
-
-func (s *Snake[T]) hasCapacity() bool {
-	return len(s.holders) < s.capacity()
-}
-
-// Acquire acquires a slot. It blocks until a slot is granted, the request
-// is dropped by CoDel, or the context is cancelled. The returned SafeUnlock
-// must be released via defer unlock.Release().
-//
-// An empty valveID is valid: such requests bypass the per-valve fairness
-// layer but still pass through the CoDel gate. Callers should pass the valve
-// ID through unconditionally rather than gating Acquire on a non-empty ID,
-// which would silently exclude all unkeyed traffic from load shedding.
-//
-// The priority ordering convention is that lower-valued priorities indicate
-// less important requests. Lower values are shed first.
-func (s *Snake[T]) Acquire(ctx context.Context, valveID string, priority float64) (*SafeUnlock[T], error) {
-	priority = s.priority(priority)
-
+func (s *Snake[T]) Enqueue(value T, valveID string, priority float64) (*Request[T], []T) {
 	if s.acquireByPriority != nil {
 		s.acquireByPriority.Add([]string{shedPriorityLabel(priority)}, 1)
 	}
 
-	s.mu.Lock()
 	req := s.q.lockedEnqueue(valveID, priority)
+	req.value = value
 	if valveID != "" {
 		s.lockedObserveValveDepth(valveID)
 	}
-
-	if s.hasCapacity() && req.codelqElem != nil {
-		s.lockedGrant(req)
-		s.lockedObserveLengths()
-		s.mu.Unlock()
-		return &SafeUnlock[T]{s: s, req: req}, nil
-	}
-
-	// Enqueue-advance: a non-granted arrival drives the CoDel control law itself,
-	// so shedding tracks load even when releases are sparse or the backstop timer
-	// fires late. lockedRunTimer only MARKS drops; take the pending rejections and
-	// send them after unlocking s.mu so the goready storm stays off the lock.
-	pending := s.lockedEnqueueAdvance()
+	dropped := s.lockedEnqueueAdvance()
 	s.lockedObserveLengths()
 	s.lockedObserveDropping()
-	s.mu.Unlock()
-	for _, p := range pending {
-		p.sendSignal()
-	}
-
-	select {
-	case val := <-req.signalChan:
-		if val != grantSentinel {
-			return nil, s.acquireError(req.priority)
-		}
-		return &SafeUnlock[T]{s: s, req: req}, nil
-
-	case <-ctx.Done():
-		// Race: Go's select picks randomly when both signalChan and
-		// ctx.Done() are ready simultaneously. The grant may have already
-		// been sent (or be in-flight) by the time we land here. The inner
-		// select resolves this:
-		//   - If signalChan has a grant: we own the slot, release it.
-		//   - If signalChan is empty: the grant might still be in-flight
-		//     (releaser unlocked the mutex but hasn't sent the signal yet).
-		//     We re-acquire the mutex and check holders — if we're already
-		//     granted, release; otherwise cancel from the queue.
-		select {
-		case val := <-req.signalChan:
-			if val == grantSentinel {
-				s.releaseOnCancel(req)
-			}
-		default:
-			s.mu.Lock()
-			if _, granted := s.holders[req]; granted {
-				s.mu.Unlock()
-				s.releaseOnCancel(req)
-			} else {
-				s.q.lockedCancel(req)
-				s.lockedObserveLengths()
-				s.lockedObserveDropping()
-				s.mu.Unlock()
-			}
-		}
-		return nil, ctx.Err()
-	}
+	return req, s.droppedValues(dropped)
 }
 
-// IsHealthy reports whether the CoDel queue is healthy.
-func (s *Snake[T]) IsHealthy() bool {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	return s.q.lockedIsHealthy()
+func (s *Snake[T]) Dequeue() (T, bool, []T) {
+	return s.dequeue(nil)
 }
 
-// SnakeStats is a point-in-time snapshot of Snake's internal state.
-type SnakeStats struct {
-	QueueLen        int
-	DroppableLen    int
-	HolderCount     int
-	Dropping        bool
-	DropCount       int
-	CurrentInterval int64 // ns
+func (s *Snake[T]) DequeueMatching(match func(T) bool) (T, bool, []T) {
+	return s.dequeue(match)
 }
 
-// Stats returns a point-in-time snapshot of Snake's internal state.
-func (s *Snake[T]) Stats() SnakeStats {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	return SnakeStats{
-		QueueLen:        s.q.codelq.lockedLen(),
-		DroppableLen:    s.q.codelq.droppableLen,
-		HolderCount:     len(s.holders),
-		Dropping:        s.q.codelq.dropping,
-		DropCount:       s.q.codelq.count,
-		CurrentInterval: s.q.codelq.lockedCurrentInterval(),
+func (s *Snake[T]) dequeue(match func(T) bool) (T, bool, []T) {
+	var pending []*Request[T]
+	if s.q.lockedNeedsAdvance() {
+		pending = s.lockedEnqueueAdvance()
 	}
-}
-
-// Release releases the slot. exc is an optional error that caused the release
-// (passed to release callbacks). Release is idempotent.
-func (u *SafeUnlock[T]) Release(exc ...error) error {
-	u.once.Do(func() {
-		var excValue error
-		if len(exc) > 0 {
-			excValue = exc[0]
-		}
-		u.err = u.s.release(u.req, excValue)
-	})
-	return u.err
-}
-
-func (s *Snake[T]) release(req *Request[T], excValue error) error {
-	s.mu.Lock()
-	// Release is idempotent and can race a context-cancel release for the same
-	// req. The loser sees the req already gone from holders and no-ops.
-	if _, ok := s.holders[req]; !ok {
-		s.mu.Unlock()
-		return nil
+	req := s.q.lockedPeek()
+	if match != nil {
+		req = s.q.lockedFind(match)
 	}
-	delete(s.holders, req)
-	s.lockedObserveHolderCount()
-	s.lockedReleaseAndShed(req)
-	s.lockedTryGrantOne()
-	s.lockedObserveLengths()
-	s.lockedObserveDropping()
-	pending := s.q.lockedTakePendingSignals()
-	s.mu.Unlock()
-
-	// Deliver drop rejections after releasing s.mu so the goready storm does not
-	// serialize grants/arrivals behind the batch.
-	for _, r := range pending {
-		r.sendSignal()
+	var value T
+	ok := false
+	if req != nil {
+		s.q.lockedDequeue(req)
+		req.signal(grantSentinel)
+		now := s.clockFunc()
+		s.lockedAccrueDropping(now)
+		s.sojourn.Add(now - req.codelqEnqueuedAtNs)
+		value = req.value
+		ok = true
+		var zero T
+		req.value = zero
 	}
-	s.runReleaseCBs(excValue)
-	return nil
-}
-
-func (s *Snake[T]) releaseOnCancel(req *Request[T]) {
-	s.mu.Lock()
-	delete(s.holders, req)
-	s.lockedObserveHolderCount()
-	s.lockedReleaseAndShed(req)
-	s.lockedTryGrantOne()
 	s.lockedObserveLengths()
 	s.lockedObserveDropping()
-	pending := s.q.lockedTakePendingSignals()
-	s.mu.Unlock()
-	for _, r := range pending {
-		r.sendSignal()
-	}
+	return value, ok, s.droppedValues(pending)
 }
 
-// lockedReleaseAndShed releases the request and advances CoDel before granting
-// the next waiter when there is active shedding work.
-func (s *Snake[T]) lockedReleaseAndShed(req *Request[T]) {
-	s.q.lockedRelease(req)
-	if s.q.lockedNeedsAdvance() {
-		s.q.lockedRunTimerIf(s.loadsheddingAllowed)
+func (s *Snake[T]) Cancel(req *Request[T]) (bool, []T) {
+	if req.signaledValue != nil {
+		return false, nil
 	}
+	s.q.lockedCancel(req)
+	var zero T
+	req.value = zero
+	dropped := s.q.lockedTakePendingDrops()
+	s.lockedObserveLengths()
+	s.lockedObserveDropping()
+	return true, s.droppedValues(dropped)
 }
 
-// lockedEnqueueAdvance runs the CoDel control-law advance on every non-granted
-// enqueue so an arrival can drive shedding, not just the release path and the
-// backstop timer — the drop cadence then tracks load even when releases are
-// sparse or the timer fires late. Must hold s.mu. lockedRunTimer only MARKS
-// drops; the pending rejections are returned so the caller sends them AFTER
-// releasing s.mu (draining the goready storm off the lock).
+// lockedEnqueueAdvance runs the CoDel control-law advance on every enqueue so
+// an arrival can drive shedding, not just the dequeue path and the backstop
+// timer. The pending drops are returned so the caller can signal them after
+// releasing the parent mutex.
 func (s *Snake[T]) lockedEnqueueAdvance() []*Request[T] {
 	s.q.lockedRunTimerIf(s.loadsheddingAllowed)
 	s.interval.Add(s.q.lockedCurrentInterval())
 	s.dropCount.Add(int64(s.q.lockedCount()))
-	return s.q.lockedTakePendingSignals()
-}
-
-func (s *Snake[T]) lockedGrant(req *Request[T]) {
-	s.holders[req] = struct{}{}
-	s.lockedObserveHolderCount()
-	s.q.lockedOnGrant(req)
-	now := s.clockFunc()
-	s.lockedAccrueDropping(now)
-	s.sojourn.Add(now - req.codelqEnqueuedAtNs)
-	req.signal(grantSentinel)
-}
-
-func (s *Snake[T]) lockedObserveHolderCount() {
-	s.holderCount.Add(int64(len(s.holders)))
+	return s.q.lockedTakePendingDrops()
 }
 
 func (s *Snake[T]) lockedObserveDropping() {
 	dropping := !s.q.lockedIsHealthy()
-	if dropping == (s.droppingSinceNs != 0) {
+	if dropping == (s.droppingSinceNs.Load() != 0) {
 		return
 	}
 	s.lockedAccrueDropping(s.clockFunc())
@@ -350,55 +190,32 @@ func (s *Snake[T]) lockedObserveDropping() {
 func (s *Snake[T]) lockedAccrueDropping(now int64) {
 	dropping := !s.q.lockedIsHealthy()
 	switch {
-	case dropping && s.droppingSinceNs == 0:
-		s.droppingSinceNs = now
-	case !dropping && s.droppingSinceNs != 0:
-		s.droppingNanos.Add(now - s.droppingSinceNs)
-		s.droppingSinceNs = 0
+	case dropping && s.droppingSinceNs.Load() == 0:
+		s.droppingSinceNs.Store(now)
+	case !dropping && s.droppingSinceNs.Load() != 0:
+		s.droppingNanos.Add(now - s.droppingSinceNs.Swap(0))
 	}
 }
 
-func (s *Snake[T]) lockedTryGrantOne() {
-	if !s.hasCapacity() {
-		return
-	}
-	next := s.q.lockedFirstWaiting()
-	if next != nil {
-		s.lockedGrant(next)
-	}
-}
-
-// runReleaseCBs executes release callbacks outside the mutex.
-func (s *Snake[T]) runReleaseCBs(excValue error) {
-	for _, cb := range s.cfg.ReleaseCBs {
-		func() {
-			defer func() {
-				if r := recover(); r != nil {
-					log.Printf("loadshed: panic in release callback for %s: %v", s.cfg.Name, r)
-				}
-			}()
-			cb(excValue)
-		}()
-	}
-}
-
-func (s *Snake[T]) priority(priority float64) float64 {
-	return priority
-}
-
 func (s *Snake[T]) loadsheddingAllowed() bool {
 	return s.cfg.LoadsheddingAllowed == nil || s.cfg.LoadsheddingAllowed()
 }
 
-func (s *Snake[T]) acquireError(priority float64) error {
-	s.shedCount.Add(1)
-	if s.shedByPriority != nil {
-		s.shedByPriority.Add([]string{shedPriorityLabel(priority)}, 1)
+func (s *Snake[T]) droppedValues(requests []*Request[T]) []T {
+	if len(requests) == 0 {
+		return nil
 	}
-	if s.cfg.AcquireError != nil {
-		return s.cfg.AcquireError()
+	values := make([]T, len(requests))
+	for i, req := range requests {
+		s.shedCount.Add(1)
+		if s.shedByPriority != nil {
+			s.shedByPriority.Add([]string{shedPriorityLabel(req.priority)}, 1)
+		}
+		values[i] = req.value
+		var zero T
+		req.value = zero
 	}
-	return &DroppedRequestError{}
+	return values
 }
 
 // shedPriorityLabel maps a request's internal Snake priority to its shed-metric
@@ -415,22 +232,20 @@ func shedPriorityLabel(priority float64) string {
 }
 
 // ShedCount returns the cumulative number of requests this Snake has shed.
-// Context cancellations are not counted — only gate-driven drops.
+// Context cancellations are not counted — only queue-driven drops.
 func (s *Snake[T]) ShedCount() int64 {
 	return s.shedCount.Load()
 }
 
 func (s *Snake[T]) DroppingNanos() int64 {
-	s.mu.Lock()
-	defer s.mu.Unlock()
 	total := s.droppingNanos.Load()
-	if s.droppingSinceNs != 0 {
-		total += s.clockFunc() - s.droppingSinceNs
+	if since := s.droppingSinceNs.Load(); since != 0 {
+		total += s.clockFunc() - since
 	}
 	return total
 }
 
-// --- timer management (must be called with s.mu held) ---
+// --- timer management (must be called with the parent mutex held) ---
 
 func (s *Snake[T]) lockedScheduleDropTimer(delayNs int64) {
 	if s.dropTimerArmed {
@@ -438,8 +253,9 @@ func (s *Snake[T]) lockedScheduleDropTimer(delayNs int64) {
 	}
 	s.dropTimerArmed = true
 	s.dropTimerExpectedNs = s.clockFunc() + delayNs
-	delay := time.Duration(delayNs) * time.Nanosecond
-	s.dropTimer = time.AfterFunc(delay, s.runDropTimer)
+	if s.cfg.DropTimerFired != nil {
+		s.dropTimer = time.AfterFunc(time.Duration(delayNs)*time.Nanosecond, s.cfg.DropTimerFired)
+	}
 }
 
 func (s *Snake[T]) lockedStopDropTimer() {
@@ -447,14 +263,14 @@ func (s *Snake[T]) lockedStopDropTimer() {
 		return
 	}
 	s.dropTimerArmed = false
-	s.dropTimer.Stop()
+	if s.dropTimer != nil {
+		s.dropTimer.Stop()
+	}
 }
 
-func (s *Snake[T]) runDropTimer() {
-	s.mu.Lock()
+func (s *Snake[T]) LockedDropTimerFired() []T {
 	if !s.dropTimerArmed {
-		s.mu.Unlock()
-		return
+		return nil
 	}
 	s.dropTimerArmed = false
 	// Record how late this fire is versus when it was scheduled. Under CPU
@@ -470,11 +286,6 @@ func (s *Snake[T]) runDropTimer() {
 	s.dropCount.Add(int64(s.q.lockedCount()))
 	s.lockedObserveLengths()
 	s.lockedObserveDropping()
-	pending := s.q.lockedTakePendingSignals()
-	s.mu.Unlock()
-	// Deliver drop rejections after releasing s.mu so the goready storm does not
-	// serialize grants/arrivals behind the batch.
-	for _, r := range pending {
-		r.sendSignal()
-	}
+	dropped := s.q.lockedTakePendingDrops()
+	return s.droppedValues(dropped)
 }
diff --git a/go/vt/vttablet/tabletserver/loadshed/snake_stats.go b/go/vt/vttablet/tabletserver/loadshed/snake_stats.go
index aceb8316e7a..549a97a9a7e 100644
--- a/go/vt/vttablet/tabletserver/loadshed/snake_stats.go
+++ b/go/vt/vttablet/tabletserver/loadshed/snake_stats.go
@@ -49,14 +49,6 @@ var intervalBucketCutoffs = loadshedBucketCutoffs
 
 var lengthBucketCutoffs = []int64{1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096}
 
-// holderBucketCutoffs is a finer-grained set for slot-holder counts, which are
-// bounded by the pool size (typically <= 64) and where the useful resolution is
-// at the low end — the powers-of-two lengthBucketCutoffs collapse the entire
-// operating range into ~6 buckets. Unit steps through the single digits, then
-// tightening steps across the 32-64 range under test, with a little headroom
-// above 64 to catch misconfiguration.
-var holderBucketCutoffs = []int64{1, 2, 3, 4, 5, 6, 8, 10, 12, 16, 20, 24, 32, 40, 48, 56, 64, 96, 128}
-
 func durationNanos(ds ...time.Duration) []int64 {
 	out := make([]int64, len(ds))
 	for i, d := range ds {
@@ -84,10 +76,9 @@ func PublishStats[T any](exporter statsExporter, prefix string, s *Snake[T]) {
 	exporter.NewCounterFunc(prefix+"DroppingNanosTotal", "Cumulative nanoseconds Snake CoDel spent in the dropping state; rate() yields the fraction of time shedding", func() int64 {
 		return s.DroppingNanos()
 	})
-	s.sojourn = exporter.NewHistogram(prefix+"SojournNs", "Distribution of Snake sojourn (time-to-grant: queue wait before slot grant), in nanoseconds", loadshedBucketCutoffs)
+	s.sojourn = exporter.NewHistogram(prefix+"SojournNs", "Distribution of Snake queue wait before dequeue, in nanoseconds", loadshedBucketCutoffs)
 	s.queueLen = exporter.NewHistogram(prefix+"QueueLenObserved", "Distribution of Snake CoDel queue length, sampled at each change", lengthBucketCutoffs)
 	s.droppableLen = exporter.NewHistogram(prefix+"DroppableLenObserved", "Distribution of Snake CoDel droppable queue length, sampled at each change", lengthBucketCutoffs)
-	s.holderCount = exporter.NewHistogram(prefix+"HolderCountObserved", "Distribution of Snake slot holders, sampled at each change", holderBucketCutoffs)
 	s.interval = exporter.NewHistogram(prefix+"IntervalObservedNs", "Distribution of Snake CoDel control interval in nanoseconds, sampled at each timer fire", intervalBucketCutoffs)
 	s.dropCount = exporter.NewHistogram(prefix+"DropCountObserved", "Distribution of Snake CoDel drop count (control-law state), sampled at each timer fire", lengthBucketCutoffs)
 	s.timerLag = exporter.NewHistogram(prefix+"DropTimerLagNs", "Distribution of how late the Snake CoDel drop timer fired versus its scheduled time, in nanoseconds; high values mean shedding decisions are delayed under CPU contention", loadshedBucketCutoffs)
diff --git a/go/vt/vttablet/tabletserver/loadshed/valved_codelq.go b/go/vt/vttablet/tabletserver/loadshed/valved_codelq.go
index 722cb83bfd6..9b3acfee729 100644
--- a/go/vt/vttablet/tabletserver/loadshed/valved_codelq.go
+++ b/go/vt/vttablet/tabletserver/loadshed/valved_codelq.go
@@ -68,8 +68,8 @@ type (
 	// directly into the CoDel queue if there are no entries there for the
 	// valve ID. Otherwise, it is inserted into the valve.
 	//
-	// When the droppable slot for a valve ID is freed (grant, drop, cancel,
-	// or release), the next pending request is promoted into the CoDel queue.
+	// When the droppable slot for a valve ID is freed (dequeue, drop, or
+	// cancel), the next pending request is promoted into the CoDel queue.
 	// All promotion runs under the parent mutex, so there is no race between
 	// removal and promotion.
 	//
@@ -88,43 +88,27 @@ type (
 		// be entries with the same valve ID in the CoDel queue.
 		valves map[string][]*Request[T]
 
-		// outstandingCounts tracks the total number of outstanding requests per
-		// valve ID (in CoDel queue + in valve).
-		outstandingCounts map[string]int
-
 		// droppablePerValve tracks which request is the current droppable
 		// representative in the CoDel queue for each valve ID. Maintains the
 		// invariant that each nonempty valve always has exactly one droppable
 		// entry in the CoDel queue.
 		droppablePerValve map[string]*Request[T]
 
-		// pendingSignals collects requests that lockedDrop marked (signaledValue
-		// set) but whose channel send is deferred until the queue mutex is
-		// released. Draining the goready storm outside the lock keeps grants and
-		// arrivals from serializing behind a large batch drop. The caller takes
-		// this slice before unlocking and sends each afterward (see
-		// lockedTakePendingSignals).
-		pendingSignals []*Request[T]
+		// pendingDrops collects requests removed by the drop path. The caller
+		// takes this slice before unlocking and signals each waiter afterward.
+		pendingDrops []*Request[T]
 	}
 )
 
 func newValvedCoDelQueue[T any](cfg CoDelConfig, nowNs func() int64, scheduleDropTimer func(delayNs int64), stopDropTimer func()) *ValvedCoDelQueue[T] {
 	q := &ValvedCoDelQueue[T]{
 		valves:            make(map[string][]*Request[T]),
-		outstandingCounts: make(map[string]int),
 		droppablePerValve: make(map[string]*Request[T]),
 	}
-	q.codelq = newCoDelQueue(cfg, nowNs, scheduleDropTimer, stopDropTimer, q.onPeekCleanup)
+	q.codelq = newCoDelQueue[T](cfg, nowNs, scheduleDropTimer, stopDropTimer)
 	return q
 }
 
-// onPeekCleanup is called by the CoDel queue when lockedPeek defensively
-// removes a done-with-error request from the list head. Decrements the
-// outstanding count for the request's valve ID.
-func (q *ValvedCoDelQueue[T]) onPeekCleanup(req *Request[T]) {
-	q.decrementOutstanding(req.valveID)
-}
-
 func (q *ValvedCoDelQueue[T]) lockedCurrentInterval() int64 {
 	return q.codelq.lockedCurrentInterval()
 }
@@ -151,27 +135,15 @@ func (q *ValvedCoDelQueue[T]) lockedIsHealthy() bool {
 	return q.codelq.lockedIsHealthy()
 }
 
-// lockedNeedsAdvance reports whether the dequeue path has any CoDel work to do:
-// an episode is active or armed (dropping, or dropNextNs seeded), the count is
-// still easing down, or a droppable backlog exists that a head-sojourn trigger
-// could arm on. When false, lockedDequeue is a guaranteed no-op, so the release
-// path can skip the call — and its clock read — entirely on the healthy fast
-// path.
 func (q *ValvedCoDelQueue[T]) lockedNeedsAdvance() bool {
 	return q.codelq.dropping || q.codelq.dropNextNs != 0 || q.codelq.droppableLen > 0
 }
 
-// lockedPeek returns the head of the CoDel queue without removing it.
-func (q *ValvedCoDelQueue[T]) lockedPeek() *Request[T] {
-	return q.codelq.lockedPeek()
-}
-
 func (q *ValvedCoDelQueue[T]) lockedEnqueue(valveID string, priority float64) *Request[T] {
 	req := newRequest[T](priority)
 	req.valveID = valveID
 
 	if valveID != "" {
-		q.outstandingCounts[valveID]++
 		if q.droppablePerValve[valveID] != nil {
 			q.valves[valveID] = append(q.valves[valveID], req)
 			return req
@@ -182,59 +154,32 @@ func (q *ValvedCoDelQueue[T]) lockedEnqueue(valveID string, priority float64) *R
 	return req
 }
 
-// lockedRelease updates valve accounting for a granted request and promotes a
-// pending request when needed.
-func (q *ValvedCoDelQueue[T]) lockedRelease(req *Request[T]) {
-	q.decrementOutstanding(req.valveID)
-	if req.valveID != "" {
-		// Promote if no droppable entry exists. This handles the case where
-		// the valve was empty at grant time (nothing to promote), but new
-		// requests arrived between grant and release — they're stranded in
-		// the valve with no droppable representative to trigger promotion.
-		if _, has := q.droppablePerValve[req.valveID]; !has {
-			q.lockedPromote(req.valveID)
-		}
-	}
-}
-
 func (q *ValvedCoDelQueue[T]) lockedDrop(req *Request[T]) {
 	q.codelq.lockedRemove(req)
-	// Mark the rejection under the lock (so under-lock readers see signaledValue
-	// immediately) but defer the channel send: it goreadys the parked Acquire
-	// goroutine, and doing that in a batch under s.mu is what serializes grants
-	// behind the drop storm. The caller drains pendingSignals after unlocking.
-	if req.signaledValue == nil {
-		req.markSignaled(&DroppedRequestError{})
-		q.pendingSignals = append(q.pendingSignals, req)
-	}
+	req.signal(&DroppedRequestError{})
+	q.pendingDrops = append(q.pendingDrops, req)
 	q.lockedPromoteOnEvict(req)
 }
 
-// lockedTakePendingSignals hands off the requests marked-but-not-yet-sent by the
-// drop path, clearing the queue's reference. The caller must send each
-// (req.sendSignal()) AFTER releasing the queue mutex. Ownership transfers fully:
-// the buffer is set to nil (not truncated in place) so a concurrent drop pass
-// that re-appends under the lock cannot corrupt the slice the caller is draining
-// after unlocking.
-func (q *ValvedCoDelQueue[T]) lockedTakePendingSignals() []*Request[T] {
-	pending := q.pendingSignals
-	q.pendingSignals = nil
-	return pending
+// lockedTakePendingDrops hands off the requests removed by the drop path,
+// clearing the queue's reference.
+func (q *ValvedCoDelQueue[T]) lockedTakePendingDrops() []*Request[T] {
+	dropped := q.pendingDrops
+	q.pendingDrops = nil
+	return dropped
 }
 
 // lockedCancel cancels a request. If it's in the CoDel queue, it removes it
-// and promotes the next from the valve. If it's pending in the valve, it
-// signals it in place and lets clearDone handle removal during the next
-// promotion — this avoids an O(N) scan of the valve.
+// and promotes the next from the valve. If it's pending in the valve, it marks
+// it in place and lets clearDone handle removal during the next promotion,
+// avoiding an O(N) scan of the valve.
 func (q *ValvedCoDelQueue[T]) lockedCancel(req *Request[T]) {
 	if req.codelqElem != nil {
 		q.codelq.lockedRemove(req)
+		req.signal(&DroppedRequestError{})
 		q.lockedPromoteOnEvict(req)
 		return
 	}
-	// The request may already have been signaled if it was promoted into
-	// the CoDel queue and dropped between the caller's default-branch
-	// (signalChan empty) and mutex acquisition.
 	if req.signaledValue == nil {
 		req.signal(&DroppedRequestError{})
 	}
@@ -261,7 +206,7 @@ func (q *ValvedCoDelQueue[T]) lockedDropFn() func() bool {
 
 // lockedRunTimer runs the CoDel drop logic, finding and dropping the
 // lowest-priority request and triggering valve promotion. It is driven both by
-// the backstop timer and synchronously from the release/dequeue path, so
+// the backstop timer and synchronously from the dequeue path, so
 // shedding tracks target as slots free rather than waiting for the timer.
 func (q *ValvedCoDelQueue[T]) lockedRunTimer() {
 	q.lockedRunTimerIf(func() bool { return true })
@@ -291,16 +236,20 @@ func (q *ValvedCoDelQueue[T]) lockedDropOne() bool {
 	return true
 }
 
-func (q *ValvedCoDelQueue[T]) lockedOnGrant(r *Request[T]) {
-	q.codelq.lockedOnGrant(r)
+func (q *ValvedCoDelQueue[T]) lockedDequeue(r *Request[T]) {
+	q.codelq.lockedDequeue(r)
 	if r.valveID != "" {
 		delete(q.droppablePerValve, r.valveID)
 		q.lockedPromote(r.valveID)
 	}
 }
 
-func (q *ValvedCoDelQueue[T]) lockedFirstWaiting() *Request[T] {
-	return q.codelq.lockedFirstWaiting()
+func (q *ValvedCoDelQueue[T]) lockedPeek() *Request[T] {
+	return q.codelq.lockedPeek()
+}
+
+func (q *ValvedCoDelQueue[T]) lockedFind(match func(T) bool) *Request[T] {
+	return q.codelq.lockedFind(match)
 }
 
 // --- private helpers ---
@@ -312,14 +261,12 @@ func (q *ValvedCoDelQueue[T]) lockedEnqueueToCoDel(req *Request[T], valveID stri
 	q.codelq.lockedEnqueue(req)
 }
 
-// lockedPromoteOnEvict handles involuntary removal of the active request
-// (drop or cancel). Decrements outstanding, then promotes.
+// lockedPromoteOnEvict handles involuntary removal of the active request.
 func (q *ValvedCoDelQueue[T]) lockedPromoteOnEvict(req *Request[T]) {
 	valveID := req.valveID
 	if valveID == "" {
 		return
 	}
-	q.decrementOutstanding(valveID)
 	delete(q.droppablePerValve, valveID)
 	q.lockedPromote(valveID)
 }
@@ -348,9 +295,7 @@ func (q *ValvedCoDelQueue[T]) lockedPromote(valveID string) {
 	q.lockedEnqueueToCoDel(next, valveID)
 }
 
-// clearDone removes done (cancelled) requests from the head of the valve.
-// Their outstanding counts are decremented since the CoDel queue never learned
-// about them.
+// clearDone removes cancelled requests from the head of the valve.
 func (q *ValvedCoDelQueue[T]) clearDone(valveID string) {
 	pending, ok := q.valves[valveID]
 	if !ok {
@@ -360,7 +305,6 @@ func (q *ValvedCoDelQueue[T]) clearDone(valveID string) {
 	for len(pending) > 0 && pending[0].signaledValue != nil {
 		pending[0] = nil
 		pending = pending[1:]
-		q.decrementOutstanding(valveID)
 	}
 
 	if len(pending) == 0 {
@@ -369,13 +313,3 @@ func (q *ValvedCoDelQueue[T]) clearDone(valveID string) {
 		q.valves[valveID] = pending
 	}
 }
-
-func (q *ValvedCoDelQueue[T]) decrementOutstanding(valveID string) {
-	if valveID == "" {
-		return
-	}
-	q.outstandingCounts[valveID]--
-	if q.outstandingCounts[valveID] <= 0 {
-		delete(q.outstandingCounts, valveID)
-	}
-}
diff --git a/go/vt/vttablet/tabletserver/loadshed/valved_codelq_test.go b/go/vt/vttablet/tabletserver/loadshed/valved_codelq_test.go
index d5be3e1487a..c5dc61d2aad 100644
--- a/go/vt/vttablet/tabletserver/loadshed/valved_codelq_test.go
+++ b/go/vt/vttablet/tabletserver/loadshed/valved_codelq_test.go
@@ -23,24 +23,20 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+type testValvedCoDelQueue = ValvedCoDelQueue[struct{}]
+
 func newValvedQueue(clock *testClock) (*testValvedCoDelQueue, *testDropTimerRecorder) {
 	rec := &testDropTimerRecorder{}
 	q := newValvedCoDelQueue[struct{}](defaultTestConfig(), clock.nowFunc, rec.schedule, rec.stop)
 	return q, rec
 }
 
-// testValvedDequeue simulates the grant+complete lifecycle on the
-// ValvedCoDelQueue: gets the first waiting request, marks it not droppable
-// (which triggers eager valve promotion), signals it, and completes it
-// (removing from queue).
 func testValvedDequeue(sq *testValvedCoDelQueue) *testRequest {
-	req := sq.lockedFirstWaiting()
+	req := sq.lockedPeek()
 	if req == nil {
 		return nil
 	}
-	sq.lockedOnGrant(req)
-	req.signal(grantSentinel)
-	sq.lockedRelease(req)
+	sq.lockedDequeue(req)
 	return req
 }
 
@@ -55,7 +51,6 @@ func TestValved_FirstRequest_DirectEntry(t *testing.T) {
 	assert.NotNil(t, req)
 	assert.Equal(t, 1, sq.lockedLen())
 	assert.NotNil(t, req.codelqElem, "should be in the CoDel queue (has list element)")
-	assert.Equal(t, 1, sq.outstandingCounts["id1"])
 }
 
 func TestValved_EmptyValveID_AlwaysDirect(t *testing.T) {
@@ -82,7 +77,6 @@ func TestValved_SecondRequest_Valved(t *testing.T) {
 	assert.NotNil(t, r1.codelqElem, "first enters CoDel queue")
 	assert.Nil(t, r2.codelqElem, "second should be in valve (no list element)")
 	assert.Equal(t, 1, sq.lockedLen(), "only 1 in CoDel queue")
-	assert.Equal(t, 2, sq.outstandingCounts["id1"])
 	require.Len(t, sq.valves["id1"], 1)
 	assert.Same(t, r2, sq.valves["id1"][0])
 }
@@ -115,7 +109,6 @@ func TestValved_FourParallel_SameID(t *testing.T) {
 
 	assert.Equal(t, 1, sq.lockedLen())
 	assert.Len(t, sq.valves["id1"], 3)
-	assert.Equal(t, 4, sq.outstandingCounts["id1"])
 }
 
 // --- Promotion tests ---
@@ -135,7 +128,6 @@ func TestValved_Promotion_OnDequeue(t *testing.T) {
 	assert.NotNil(t, r2.codelqElem, "r2 promoted to CoDel queue after dequeue")
 	assert.Equal(t, 1, sq.lockedLen())
 	assert.Empty(t, sq.valves["id1"])
-	assert.Equal(t, 1, sq.outstandingCounts["id1"])
 }
 
 func TestValved_Promotion_OnDrop(t *testing.T) {
@@ -162,7 +154,6 @@ func TestValved_Promotion_OnCancel(t *testing.T) {
 
 	assert.NotNil(t, r2.codelqElem, "r2 promoted after r1 cancelled")
 	assert.Equal(t, 1, sq.lockedLen())
-	assert.Equal(t, 1, sq.outstandingCounts["id1"])
 }
 
 // --- Cancel tests ---
@@ -180,11 +171,8 @@ func TestValved_CancelInValve(t *testing.T) {
 
 	assert.NotNil(t, r1.codelqElem, "r1 still in CoDel queue")
 	assert.Equal(t, 1, sq.lockedLen())
-	// r3 is signaled in place but not removed from the slice until promotion
 	assert.Len(t, sq.valves["id1"], 3)
-	// outstanding count is not decremented until clearDone runs during promotion
-	assert.Equal(t, 4, sq.outstandingCounts["id1"])
-	assert.NotNil(t, r3.signaledValue, "r3 should be signaled")
+	assert.NotNil(t, r3.signaledValue)
 }
 
 func TestValved_ClearDone_InValve(t *testing.T) {
@@ -195,8 +183,7 @@ func TestValved_ClearDone_InValve(t *testing.T) {
 	r2 := sq.lockedEnqueue("id1", 0)
 	r3 := sq.lockedEnqueue("id1", 0)
 
-	// mark r2 as done (cancelled while in valve)
-	r2.signal(&DroppedRequestError{})
+	sq.lockedCancel(r2)
 
 	// dequeue r1 → promote should skip r2 (done) and promote r3
 	testValvedDequeue(sq)
@@ -224,7 +211,6 @@ func TestValved_CancelInMiddle_EventualPromotion(t *testing.T) {
 	// Dequeue r2 → clearDone finds r3 (now at head), skips it, promotes r4
 	testValvedDequeue(sq)
 	assert.NotNil(t, r4.codelqElem, "r4 promoted (r3 skipped)")
-	assert.Equal(t, 1, sq.outstandingCounts["id1"])
 }
 
 func TestValved_CancelMultipleConsecutiveAtHead(t *testing.T) {
@@ -246,13 +232,10 @@ func TestValved_CancelMultipleConsecutiveAtHead(t *testing.T) {
 	assert.NotNil(t, r4.codelqElem, "r4 promoted (r2 and r3 skipped)")
 	assert.Nil(t, r2.codelqElem, "r2 never entered CoDel queue")
 	assert.Nil(t, r3.codelqElem, "r3 never entered CoDel queue")
-	// outstanding: started at 5, decremented 1 (r1 dequeue) + 2 (r2, r3 clearDone) = 2 remaining
-	assert.Equal(t, 2, sq.outstandingCounts["id1"])
 
 	// Dequeue r4 → promotes r5
 	testValvedDequeue(sq)
 	assert.NotNil(t, r5.codelqElem, "r5 promoted")
-	assert.Equal(t, 1, sq.outstandingCounts["id1"])
 }
 
 func TestValved_AllValveEntriesCancelled(t *testing.T) {
@@ -275,25 +258,20 @@ func TestValved_AllValveEntriesCancelled(t *testing.T) {
 	assert.Equal(t, 0, sq.lockedLen(), "CoDel queue empty")
 	_, exists := sq.valves["id1"]
 	assert.False(t, exists, "valve map entry should be cleaned up")
-	_, exists = sq.outstandingCounts["id1"]
-	assert.False(t, exists, "outstanding count should be cleaned up")
 }
 
-func TestValved_InflatedOutstandingGatesNewArrivals(t *testing.T) {
+func TestValved_CancelledWaiterDoesNotBypassValve(t *testing.T) {
 	clock := newTestClock()
 	sq, _ := newValvedQueue(clock)
 
 	sq.lockedEnqueue("id1", 0)       // r1: active in CoDel
 	r2 := sq.lockedEnqueue("id1", 0) // r2: valve[0]
 
-	// Cancel r2 — outstanding stays inflated (2) until promotion
 	sq.lockedCancel(r2)
-	assert.Equal(t, 2, sq.outstandingCounts["id1"])
 
-	// New arrival for same valve ID should still be valved (conservative)
+	// A new arrival for the same valve ID remains valved behind r1.
 	r3 := sq.lockedEnqueue("id1", 0)
-	assert.Nil(t, r3.codelqElem, "r3 should be valved due to inflated outstanding count")
-	assert.Equal(t, 3, sq.outstandingCounts["id1"])
+	assert.Nil(t, r3.codelqElem, "r3 should be valved")
 	assert.Equal(t, 1, sq.lockedLen(), "still only r1 in CoDel queue")
 }
 
@@ -316,7 +294,6 @@ func TestValved_CancelAllThenNewArrival(t *testing.T) {
 	// Fresh arrival for same valve ID should go directly to CoDel (no stale state)
 	r4 := sq.lockedEnqueue("id1", 0)
 	assert.NotNil(t, r4.codelqElem, "r4 goes directly to CoDel after full cleanup")
-	assert.Equal(t, 1, sq.outstandingCounts["id1"])
 }
 
 func TestValved_CancelInterleavedWithPromotions(t *testing.T) {
@@ -345,7 +322,6 @@ func TestValved_CancelInterleavedWithPromotions(t *testing.T) {
 	// Dequeue r4 → clearDone hits r5 (cancelled at head), skips it, promotes r6
 	testValvedDequeue(sq)
 	assert.NotNil(t, r6.codelqElem, "r6 promoted (r5 skipped)")
-	assert.Equal(t, 1, sq.outstandingCounts["id1"])
 }
 
 func TestValved_MassCancel_OverloadScenario(t *testing.T) {
@@ -359,7 +335,6 @@ func TestValved_MassCancel_OverloadScenario(t *testing.T) {
 	for i := range 50 {
 		requests[i] = sq.lockedEnqueue("id1", 0)
 	}
-	assert.Equal(t, 51, sq.outstandingCounts["id1"])
 	assert.Equal(t, 1, sq.lockedLen())
 
 	// Cancel all but the last 5 (simulating context timeouts in overload)
@@ -371,7 +346,6 @@ func TestValved_MassCancel_OverloadScenario(t *testing.T) {
 	// cancelled entries at the head and promote the first live one
 	testValvedDequeue(sq)
 	assert.NotNil(t, requests[45].codelqElem, "first surviving request promoted")
-	assert.Equal(t, 5, sq.outstandingCounts["id1"])
 
 	// Drain remaining 5
 	for i := 45; i < 50; i++ {
@@ -382,9 +356,7 @@ func TestValved_MassCancel_OverloadScenario(t *testing.T) {
 		}
 	}
 
-	_, exists := sq.outstandingCounts["id1"]
-	assert.False(t, exists, "all outstanding cleaned up")
-	_, exists = sq.valves["id1"]
+	_, exists := sq.valves["id1"]
 	assert.False(t, exists, "valve map cleaned up")
 }
 
@@ -402,41 +374,9 @@ func TestValved_CancelFromValve_DoesNotAffectOtherValveIDs(t *testing.T) {
 	sq.lockedCancel(r1v)
 
 	// id2's valve should be completely unaffected
-	assert.Equal(t, 2, sq.outstandingCounts["id2"])
 	assert.Len(t, sq.valves["id2"], 1)
 	assert.Same(t, r2v, sq.valves["id2"][0])
-	assert.Nil(t, r2v.signaledValue, "id2 valve entry should not be signaled")
-}
-
-// --- Outstanding count tests ---
-
-func TestValved_OutstandingCount_Lifecycle(t *testing.T) {
-	clock := newTestClock()
-	sq, _ := newValvedQueue(clock)
-
-	sq.lockedEnqueue("id1", 0)
-	assert.Equal(t, 1, sq.outstandingCounts["id1"])
-
-	sq.lockedEnqueue("id1", 0)
-	assert.Equal(t, 2, sq.outstandingCounts["id1"])
-
-	testValvedDequeue(sq) // removes first, promotes second
-	assert.Equal(t, 1, sq.outstandingCounts["id1"])
-
-	testValvedDequeue(sq) // removes second
-	assert.Equal(t, 0, sq.outstandingCounts["id1"])
-}
-
-func TestValved_OutstandingCount_SurvivesCancel(t *testing.T) {
-	clock := newTestClock()
-	sq, _ := newValvedQueue(clock)
-
-	r1 := sq.lockedEnqueue("id1", 0)
-	sq.lockedEnqueue("id1", 0)
-	assert.Equal(t, 2, sq.outstandingCounts["id1"])
-
-	sq.lockedCancel(r1)
-	assert.Equal(t, 1, sq.outstandingCounts["id1"])
+	assert.Nil(t, r2v.signaledValue)
 }
 
 func TestValved_EmptyValve_MapCleanup(t *testing.T) {
@@ -453,51 +393,6 @@ func TestValved_EmptyValve_MapCleanup(t *testing.T) {
 	assert.False(t, exists, "empty valve should be removed from map")
 }
 
-// --- Peek cleanup tests ---
-
-// TestValved_PeekCleanup_DecrementsOutstandingCount proves that when
-// lockedPeek defensively removes a done-with-error request from the CoDel
-// queue head, outstanding counts are decremented correctly.
-func TestValved_PeekCleanup_DecrementsOutstandingCount(t *testing.T) {
-	clock := newTestClock()
-	sq, _ := newValvedQueue(clock)
-
-	r1 := sq.lockedEnqueue("id1", 0)
-	sq.lockedEnqueue("id1", 0)
-
-	assert.Equal(t, 2, sq.outstandingCounts["id1"])
-
-	// Simulate the "impossible" state: signal r1 with error without calling
-	// lockedRemove. This leaves elem non-nil, so lockedPeek will find it as
-	// isDone() with non-nil outcome and clean it up.
-	r1.signal(&DroppedRequestError{})
-
-	// lockedPeek should remove r1 and decrement outstanding count
-	result := sq.lockedPeek()
-	assert.Nil(t, result, "r2 is in valve not CoDel queue, so peek returns nil after r1 cleanup")
-	assert.Equal(t, 1, sq.outstandingCounts["id1"], "outstanding should be decremented for cleaned-up request")
-}
-
-// TestValved_PeekCleanup_DecrementsDroppableLen proves that when
-// lockedPeek removes a done droppable request, droppableLen is decremented.
-func TestValved_PeekCleanup_DecrementsDroppableLen(t *testing.T) {
-	clock := newTestClock()
-	sq, _ := newValvedQueue(clock)
-
-	r1 := sq.lockedEnqueue("id1", 0)
-	r2 := sq.lockedEnqueue("id2", 0)
-
-	assert.Equal(t, 2, sq.codelq.droppableLen)
-
-	// Signal r1 without removing it from the list
-	r1.signal(&DroppedRequestError{})
-
-	// lockedPeek should clean up r1 and decrement droppableLen
-	result := sq.lockedPeek()
-	assert.Same(t, r2, result)
-	assert.Equal(t, 1, sq.codelq.droppableLen, "droppableLen should be decremented for cleaned-up request")
-}
-
 // --- FIFO within contention ---
 
 func TestValved_FIFO_WithinContention(t *testing.T) {