fix(blocking): wake producers waiting on capacity after Reset (#49)

* test(blocking): add failing test for Reset not waking OfferWait

Asserts that OfferWait returns after Reset frees capacity. Fails on
current main because Reset only broadcasts notEmptyCond, leaving
producers stuck on notFullCond.Wait().

Refs #39

* fix(blocking): broadcast notFullCond on Reset

Reset can free capacity slots when it restores the initial (smaller)
element set. Without broadcasting notFullCond, any goroutine blocked
in OfferWait stays stuck until another consumer runs.

Fixes #39

Author: adrianbrad

blocking.go

@@ -107,6 +107,7 @@ func (bq *Blocking[T]) Reset() {
 	copy(bq.elems, bq.initialElems)
 
 	bq.notEmptyCond.Broadcast()
+	bq.notFullCond.Broadcast()
 }
 
 // ===================================Removal==================================

blocking_test.go

@@ -311,6 +311,46 @@ func testBlockingReset(t *testing.T) {
 			t.Fatalf("expected elem to be %d, got %d", 5, e)
 		}
 	})
+
+	t.Run("UnblocksProducersWaitingOnCapacity", func(t *testing.T) {
+		t.Parallel()
+
+		// Start with fewer elements than capacity so Reset frees a slot.
+		blockingQueue := queue.NewBlocking(
+			[]int{1, 2},
+			queue.WithCapacity(3),
+		)
+
+		if err := blockingQueue.Offer(3); err != nil {
+			t.Fatalf("unexpected err filling queue: %v", err)
+		}
+
+		offerReturned := make(chan struct{})
+
+		go func() {
+			blockingQueue.OfferWait(4)
+			close(offerReturned)
+		}()
+
+		// Give the producer a moment to enter notFullCond.Wait().
+		time.Sleep(10 * time.Millisecond)
+
+		select {
+		case <-offerReturned:
+			t.Fatal("OfferWait returned before Reset; queue was full")
+		default:
+		}
+
+		// Reset shrinks back to the initial two elements, leaving a free slot.
+		// Without broadcasting notFullCond, the producer never wakes.
+		blockingQueue.Reset()
+
+		select {
+		case <-offerReturned:
+		case <-time.After(time.Second):
+			t.Fatal("OfferWait was not unblocked by Reset")
+		}
+	})
 }
 
 func testBlockingOfferWait(t *testing.T) {