Fix valve invariant violation when contention < capacity (#870)

### Background / Why?

The valve's gate condition in `lockedEnqueue` used `outstandingCounts[valveID] > 1` to decide whether to valve a new arrival. This is a proxy for "there's already a droppable representative in the CoDel queue", but it's wrong after a grant. When a request is granted, `lockedOnGrant` deletes `droppablePerValve[valveID]` (the droppable became undroppable), but `outstandingCounts` remains elevated because the granted request is still outstanding. Subsequent arrivals for the same valve ID see count > 1 and get valved, but there's no droppable representative to ever trigger promotion. As a result, the invariant ("each nonempty valve always has exactly one droppable entry in the CoDel queue") is broken and requests are stranded indefinitely.

This was correct in the Python implementation because:
1. it was a lock, not a semaphore
2. the invariant was slightly different (each nonempty valve has one representative in the codelq, not one DROPPABLE representative). As we noted at the time of coming up with the N-holder invariant, that invariant is actually superior even for the 1-holder (lock) case, it just only mattered in an extreme edge case we likely never saw.

The practical impact: with capacity N and M concurrent requests from the same valve ID where M ≤ N, only the first request is ever granted. The rest are serialized one-at-a-time through release→promote→grant chains, leaving N-1 capacity slots unused. This case probably wouldn't ever happen in production since it requires all requests to be for only one valve ID.

The fix replaces the `outstandingCounts > 1` check with a direct `droppablePerValve[valveID]` existence check, which directly encodes the invariant.

### Testing

new unit tests

AI disclosure: Claude Code assisted with development. Every line of code was either written by or carefully reviewed by me :)

Author: bgwines

go/vt/vttablet/tabletserver/loadshed/selfcontentionaware_codelq.go

@@ -142,7 +142,7 @@ func (q *SelfContentionAwareCoDelQueue) lockedEnqueue(valveID string, priority f
 
 	if valveID != "" {
 		q.outstandingCounts[valveID]++
-		if q.outstandingCounts[valveID] > 1 {
+		if q.droppablePerValve[valveID] != nil {
 			q.pendingRequests[valveID] = append(q.pendingRequests[valveID], req)
 			return req
 		}

go/vt/vttablet/tabletserver/loadshed/snake_nholder_test.go

@@ -398,6 +398,85 @@ func TestSnake_NHolder_ReleaseCallbacks(t *testing.T) {
 	assert.Equal(t, int64(3), count.Load(), "release callback should fire for each holder")
 }
 
+// --- N-holder: valve invariant under sufficient capacity ---
+
+func TestSnake_NHolder_ValveInvariant_AllGranted(t *testing.T) {
+	const capacity = 10
+	const M = 5
+
+	cfg := defaultSnakeConfig()
+	cfg.Capacity = func() int { return capacity }
+	s := NewSnake(cfg)
+
+	unlocks := make([]*SafeUnlock, M)
+	for i := range M {
+		ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
+		u, err := s.Acquire(ctx, "foo")
+		cancel()
+		require.NoError(t, err, "request %d should be granted (capacity=%d, holders=%d)", i, capacity, i)
+		unlocks[i] = u
+	}
+
+	assert.Equal(t, M, s.nGranted(), "all %d requests should be granted concurrently", M)
+
+	for _, u := range unlocks {
+		u.Release()
+	}
+	assert.Equal(t, 0, s.nGranted())
+}
+
+// --- N-holder: valve invariant under exhausted capacity ---
+
+func TestSnake_NHolder_ValveInvariant_CapacityExhausted(t *testing.T) {
+	const capacity = 3
+
+	cfg := defaultSnakeConfig()
+	cfg.Capacity = func() int { return capacity }
+	s := NewSnake(cfg)
+
+	unlocks := make([]*SafeUnlock, capacity)
+	for i := range capacity {
+		ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
+		u, err := s.Acquire(ctx, "foo")
+		cancel()
+		require.NoError(t, err, "request %d should be granted", i)
+		unlocks[i] = u
+	}
+	assert.Equal(t, capacity, s.nGranted())
+
+	blocked := make(chan struct{})
+	granted := make(chan *SafeUnlock, 1)
+	go func() {
+		close(blocked)
+		u, err := s.Acquire(context.Background(), "foo")
+		if err == nil {
+			granted <- u
+		}
+	}()
+	<-blocked
+	time.Sleep(10 * time.Millisecond)
+
+	s.mu.Lock()
+	droppable, hasDroppable := s.q.droppablePerValve["foo"]
+	s.mu.Unlock()
+	assert.True(t, hasDroppable, "nonempty valve must have a droppable representative")
+	assert.NotNil(t, droppable)
+
+	unlocks[0].Release()
+
+	select {
+	case u := <-granted:
+		assert.Equal(t, capacity, s.nGranted())
+		u.Release()
+	case <-time.After(2 * time.Second):
+		t.Fatal("blocked request should have been granted after release")
+	}
+
+	for i := 1; i < capacity; i++ {
+		unlocks[i].Release()
+	}
+}
+
 // --- N-holder: memory cleanup with multi-slot ---
 
 func TestSnake_NHolder_MemoryCleanup(t *testing.T) {