fix(priority): release popped element references

blocking_memory_test.go

@@ -0,0 +1,53 @@
+package queue
+
+import "testing"
+
+func TestBlockingGetZerosPoppedSlot(t *testing.T) {
+	type payload struct{ id int }
+
+	q := NewBlocking[*payload](nil)
+	if err := q.Offer(&payload{id: 1}); err != nil {
+		t.Fatalf("offer: %v", err)
+	}
+
+	backing := q.elems[:cap(q.elems)]
+	got, err := q.Get()
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if got == nil || got.id != 1 {
+		t.Fatalf("unexpected element: %v", got)
+	}
+	if len(q.elems) != 0 {
+		t.Fatalf("expected empty queue after Get, got size %d", len(q.elems))
+	}
+	if backing[0] != nil {
+		t.Fatal("backing slot still retains the popped payload")
+	}
+}
+
+func TestBlockingClearZerosBackingSlots(t *testing.T) {
+	type payload struct{ id int }
+
+	q := NewBlocking[*payload](nil)
+	for i := 0; i < 3; i++ {
+		if err := q.Offer(&payload{id: i + 1}); err != nil {
+			t.Fatalf("offer: %v", err)
+		}
+	}
+
+	backing := q.elems[:cap(q.elems)]
+	removed := q.Clear()
+	if len(removed) != 3 {
+		t.Fatalf("expected 3 removed elements, got %d", len(removed))
+	}
+	if q.Size() != 0 {
+		t.Fatalf("expected empty queue after clear, got size %d", q.Size())
+	}
+
+	for i, elem := range backing {
+		if elem != nil {
+			t.Fatalf("backing slot %d still retains a payload after Clear", i)
+		}
+	}
+}

blocking_test.go

@@ -6,7 +6,6 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
-	"runtime"
 	"sort"
 	"sync"
 	"testing"
@@ -33,8 +32,6 @@ func TestBlocking(t *testing.T) {
 	t.Run("CondWaitWithCapacity", testBlockingCondWaitWithCapacity)
 	t.Run("MarshalJSON", testBlockingMarshalJSON)
 	t.Run("NewDoesNotAliasCallerSlice", testBlockingNewDoesNotAliasCallerSlice)
-	t.Run("GetReleasesReference", testBlockingGetReleasesReference)
-	t.Run("ClearReleasesReferences", testBlockingClearReleasesReferences)
 	t.Run("OfferBroadcastsToAllWaiters", testBlockingOfferBroadcastsToAllWaiters)
 }
 
@@ -92,114 +89,6 @@ func testBlockingOfferBroadcastsToAllWaiters(t *testing.T) {
 	}
 }
 
-func testBlockingGetReleasesReference(t *testing.T) {
-	t.Parallel()
-
-	type payload struct{ id int }
-
-	var bq *queue.Blocking[*payload]
-
-	finalized := make(chan struct{}, 1)
-
-	// Create the payload in a nested scope so p/got don't keep it alive
-	// after the offer/get cycle.
-	func() {
-		p := &payload{id: 42}
-		runtime.SetFinalizer(p, func(*payload) {
-			finalized <- struct{}{}
-		})
-
-		bq = queue.NewBlocking[*payload](nil)
-
-		if err := bq.Offer(p); err != nil {
-			t.Fatalf("offer: %v", err)
-		}
-
-		got, err := bq.Get()
-		if err != nil {
-			t.Fatalf("get: %v", err)
-		}
-
-		if got != p {
-			t.Fatalf("unexpected element %v", got)
-		}
-	}()
-
-	// Queue is still live. After a successful Get, the queue must not
-	// retain the popped element. Nudge GC until the finalizer fires or
-	// we give up.
-	deadline := time.After(time.Second)
-
-	for {
-		runtime.GC() //nolint:revive // explicit GC needed to drive finalizer
-
-		select {
-		case <-finalized:
-			runtime.KeepAlive(bq)
-			return
-		case <-deadline:
-			runtime.KeepAlive(bq)
-			t.Fatal("popped element not finalized; backing array still holds it")
-		default:
-			time.Sleep(10 * time.Millisecond)
-		}
-	}
-}
-
-func testBlockingClearReleasesReferences(t *testing.T) {
-	t.Parallel()
-
-	type payload struct{ id int }
-
-	var bq *queue.Blocking[*payload]
-
-	finalized := make(chan struct{}, 3)
-
-	func() {
-		items := []*payload{{id: 1}, {id: 2}, {id: 3}}
-
-		for _, it := range items {
-			runtime.SetFinalizer(it, func(*payload) {
-				finalized <- struct{}{}
-			})
-		}
-
-		bq = queue.NewBlocking[*payload](nil)
-
-		for _, it := range items {
-			if err := bq.Offer(it); err != nil {
-				t.Fatalf("offer: %v", err)
-			}
-		}
-
-		cleared := bq.Clear()
-		if len(cleared) != len(items) {
-			t.Fatalf("expected %d cleared, got %d", len(items), len(cleared))
-		}
-	}()
-
-	// All items should be eligible for finalization now — the queue is
-	// empty and the temporary `cleared` slice has gone out of scope.
-	deadline := time.After(time.Second)
-
-	count := 0
-	for count < 3 {
-		runtime.GC() //nolint:revive // explicit GC needed to drive finalizer
-
-		select {
-		case <-finalized:
-			count++
-		case <-deadline:
-			runtime.KeepAlive(bq)
-			t.Fatalf("only %d/3 payloads finalized; Clear kept references", count)
-		default:
-			time.Sleep(10 * time.Millisecond)
-		}
-	}
-
-	runtime.KeepAlive(bq)
-}
-
 func testBlockingNewDoesNotAliasCallerSlice(t *testing.T) {
 	t.Parallel()
 

circular_test.go

@@ -5,9 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"reflect"
-	"runtime"
 	"testing"
-	"time"
 
 	"github.com/adrianbrad/queue"
 )
@@ -27,56 +25,6 @@ func TestCircular(t *testing.T) {
 	t.Run("Iterator", testCircularIterator)
 	t.Run("MarshalJSON", testCircularMarshalJSON)
 	t.Run("NonPositiveCapacity", testCircularNonPositiveCapacity)
-	t.Run("ResetReleasesBeyondInitial", testCircularResetReleasesBeyondInitial)
-}
-
-func testCircularResetReleasesBeyondInitial(t *testing.T) {
-	t.Parallel()
-
-	type payload struct{ id int }
-
-	var cq *queue.Circular[*payload]
-
-	finalized := make(chan struct{}, 3)
-
-	func() {
-		// Capacity 5, 2 initial elems, then offer 3 more. Reset must
-		// release the 3 extras; on unfixed code, slots 2..4 still hold
-		// them via the backing array.
-		cq = queue.NewCircular([]*payload{{id: 1}, {id: 2}}, 5)
-
-		for i := 3; i <= 5; i++ {
-			p := &payload{id: i}
-			runtime.SetFinalizer(p, func(*payload) {
-				finalized <- struct{}{}
-			})
-
-			if err := cq.Offer(p); err != nil {
-				t.Fatalf("offer: %v", err)
-			}
-		}
-
-		cq.Reset()
-	}()
-
-	deadline := time.After(time.Second)
-
-	count := 0
-	for count < 3 {
-		runtime.GC() //nolint:revive // explicit GC needed to drive finalizer
-
-		select {
-		case <-finalized:
-			count++
-		case <-deadline:
-			runtime.KeepAlive(cq)
-			t.Fatalf("only %d/3 extra payloads finalized after Reset", count)
-		default:
-			time.Sleep(10 * time.Millisecond)
-		}
-	}
-
-	runtime.KeepAlive(cq)
 }
 
 func testCircularNonPositiveCapacity(t *testing.T) {

example_blocking_test.go

@@ -2,7 +2,6 @@ package queue_test
 
 import (
 	"fmt"
-	"time"
 
 	"github.com/adrianbrad/queue"
 )
@@ -27,25 +26,15 @@ func ExampleBlocking() {
 	empty = blockingQueue.IsEmpty()
 	fmt.Println("Empty after clear:", empty)
 
-	var (
-		elem  int
-		after time.Duration
-	)
-
+	var elem int
 	done := make(chan struct{})
 
-	start := time.Now()
-
-	// this function waits for a new element to be available in the queue.
+	// This function waits for a new element to be available in the queue.
 	go func() {
 		defer close(done)
-
 		elem = blockingQueue.GetWait()
-		after = time.Since(start)
 	}()
 
-	time.Sleep(time.Millisecond)
-
 	// insert a new element into the queue.
 	if err := blockingQueue.Offer(4); err != nil {
 		fmt.Println("Offer err:", err)
@@ -62,11 +51,7 @@ func ExampleBlocking() {
 
 	<-done
 
-	fmt.Printf(
-		"Elem %d received after %s",
-		elem,
-		after.Round(time.Millisecond),
-	)
+	fmt.Println("Elem received:", elem)
 
 	// Output:
 	// Contains 3: true
@@ -75,5 +60,5 @@ func ExampleBlocking() {
 	// Clear: [1 2 3]
 	// Empty after clear: true
 	// Peeked elem: 4
-	// Elem 4 received after 1ms
+	// Elem received: 4
 }

priority.go

@@ -46,9 +46,12 @@
 func (h *priorityHeap[T]) Pop() any {
 	n := len(h.elems)
 
-	elem := (h.elems)[n-1]
+	elem := h.elems[n-1]
+	// Clear the removed slot so pointer elements are not retained by the backing array.
+	var zero T
+	h.elems[n-1] = zero
 
-	h.elems = (h.elems)[0 : n-1]
+	h.elems = h.elems[:n-1]
 
 	return elem
 }
@@ -115,7 +118,7 @@
 	// provided, the elements are sorted and trimmed to fit the capacity.
 	if options.capacity != nil && *options.capacity < elementsHeap.Len() {
 		sort.Slice(elementsHeap.elems, func(i, j int) bool {
 			return lessFunc((elementsHeap.elems)[i], (elementsHeap.elems)[j])
 		})
 
 		elementsHeap.elems = (elementsHeap.elems)[:*options.capacity]
@@ -273,7 +276,7 @@
 }
 
 // MarshalJSON serializes the Priority queue to JSON in priority order.
 func (pq *Priority[T]) MarshalJSON() ([]byte, error) {
 	pq.lock.RLock()
 
 	output := make([]T, len(pq.elements.elems))

priority_memory_test.go

@@ -0,0 +1,26 @@
+package queue
+
+import "testing"
+
+func TestPriorityGetZerosPoppedSlot(t *testing.T) {
+	type payload struct{ id int }
+
+	q := NewPriority([]*payload{{id: 1}}, func(a, b *payload) bool {
+		return a.id < b.id
+	})
+	backing := q.elements.elems[:cap(q.elements.elems)]
+
+	got, err := q.Get()
+	if err != nil {
+		t.Fatalf("get: %v", err)
+	}
+	if got == nil || got.id != 1 {
+		t.Fatalf("unexpected element: %v", got)
+	}
+	if len(q.elements.elems) != 0 {
+		t.Fatalf("expected empty queue after Get, got size %d", len(q.elements.elems))
+	}
+	if backing[0] != nil {
+		t.Fatal("backing slot still retains the popped payload")
+	}
+}