actor/queue_test.go

package actor

import (
	"fmt"
	"math/rand"
	"reflect"
	"runtime"
	"sync"
	"testing"
	"time"
)

func queueAwait(t *testing.T, done <-chan struct{}) {
	t.Helper()
	timer := time.NewTimer(10 * time.Second)
	defer timer.Stop()
	select {
	case <-done:
	case <-timer.C:
		t.Fatal("queue operation did not finish")
	}
}

func queuePopInts(t *testing.T, q *queue[int], batch *node[int], want []int) {
	t.Helper()
	for _, expected := range want {
		if batch == nil {
			t.Fatalf("batch ended before message %d", expected)
		}
		retired := batch
		actual, next := q.pop(batch)
		batch = next
		if actual != expected {
			t.Fatalf("message=%d, want %d", actual, expected)
		}
		// These tests have no concurrent producer that could reuse the node
		// between pop and inspection.
		if retired.value != 0 || retired.next != nil {
			t.Fatal("consumed node retains its value or next link")
		}
	}
	if batch != nil {
		t.Fatal("batch contains unexpected messages")
	}
}

func TestQueueEmptyAndCapacityBoundaries(t *testing.T) {
	for _, capacity := range []int{1, 2, 3, 8, 64} {
		for _, count := range []int{capacity - 1, capacity, capacity + 1, capacity*3 + 1} {
			t.Run(fmt.Sprintf("capacity=%d/messages=%d", capacity, count), func(t *testing.T) {
				q := newQueue[int](capacity)
				addresses := make([]*node[int], capacity)
				for i := range q.fixed {
					addresses[i] = &q.fixed[i]
				}
				for range 3 {
					if !q.empty() || q.takeBatch(4, runtime.Gosched) != nil {
						t.Fatal("new queue is not empty")
					}
				}
				want := make([]int, count)
				for i := range want {
					want[i] = i // Includes the zero value and singleton zero case.
					q.push(i)
				}
				fixed, dynamic := 0, 0
				for n := q.head.Load(); n != nil; n = n.next {
					if n.fixed {
						fixed++
					} else {
						dynamic++
					}
				}
				if fixed != min(count, capacity) || dynamic != max(0, count-capacity) {
					t.Fatalf("storage boundary: fixed=%d dynamic=%d", fixed, dynamic)
				}
				batch := q.takeBatch(4, runtime.Gosched)
				// empty describes the published stack, not the detached batch.
				for range 3 {
					if !q.empty() || q.takeBatch(4, runtime.Gosched) != nil {
						t.Fatal("detached messages remained on the published stack")
					}
				}
				queuePopInts(t, q, batch, want)
				for i := range q.fixed {
					if &q.fixed[i] != addresses[i] || !q.fixed[i].fixed || q.fixed[i].value != 0 || q.fixed[i].next != nil {
						t.Fatalf("fixed slot %d moved or retained consumed data", i)
					}
				}
			})
		}
	}
}

func TestQueueNewMessageAtFixedStorageConsumptionStages(t *testing.T) {
	for _, test := range []struct {
		name      string
		initial   int
		consumed  int
		wantFixed bool
	}{
		{name: "fixed_full", initial: 4},
		{name: "half_fixed_consumed_with_dynamic_messages", initial: 7, consumed: 2},
		{name: "all_fixed_consumed_with_dynamic_messages", initial: 7, consumed: 4, wantFixed: true},
	} {
		t.Run(test.name, func(t *testing.T) {
			q := newQueue[int](4)
			for value := 1; value <= test.initial; value++ {
				q.push(value)
			}
			var batch *node[int]
			if test.consumed > 0 {
				batch = q.takeBatch(4, runtime.Gosched)
				for expected := 1; expected <= test.consumed; expected++ {
					if batch == nil {
						t.Fatal("initial batch ended before the requested consumption stage")
					}
					value, next := q.pop(batch)
					batch = next
					if value != expected {
						t.Fatalf("initial message=%d, want %d", value, expected)
					}
				}
				fixed, dynamic := 0, 0
				for n := batch; n != nil; n = n.next {
					if n.fixed {
						fixed++
					} else {
						dynamic++
					}
				}
				if fixed != 4-test.consumed || dynamic != 3 {
					t.Fatalf("remaining batch: fixed=%d dynamic=%d", fixed, dynamic)
				}
			}
			message := test.initial + 1
			q.push(message)
			published := q.head.Load()
			if published == nil || published.value != message || published.fixed != test.wantFixed {
				t.Fatalf("new message node=%+v, want value=%d fixed=%v", published, message, test.wantFixed)
			}
			if test.wantFixed && published != &q.fixed[0] {
				t.Fatal("completed fixed round did not reuse its first slot")
			}
			var remaining []int
			for value := test.consumed + 1; value <= test.initial; value++ {
				remaining = append(remaining, value)
			}
			if batch != nil {
				// Older dynamic messages stay ahead of the new message even
				// when that message has reused the fixed storage.
				queuePopInts(t, q, batch, remaining)
				remaining = nil
			}
			remaining = append(remaining, message)
			queuePopInts(t, q, q.takeBatch(4, runtime.Gosched), remaining)
			if !q.empty() {
				t.Fatal("new message left extra published data")
			}
		})
	}
}

func TestQueueFixedRoundReopensOnlyAfterFullConsumption(t *testing.T) {
	q := newQueue[int](3)
	first := &q.fixed[0]
	for value := 1; value <= 7; value++ {
		q.push(value)
	}
	batch := q.takeBatch(4, runtime.Gosched)
	for expected := 1; expected <= 3; expected++ {
		value, next := q.pop(batch)
		batch = next
		if value != expected {
			t.Fatalf("fixed message=%d, want %d", value, expected)
		}
		if expected < 3 {
			if q.reserved.Load() != 3 || q.consumed != expected {
				t.Fatal("partially consumed fixed round was reopened")
			}
			q.push(7 + expected)
			if q.head.Load().fixed {
				t.Fatal("producer reused a fixed slot before the round finished")
			}
		}
	}
	if q.reserved.Load() != 0 || q.consumed != 0 {
		t.Fatal("fully consumed fixed round did not reopen")
	}
	q.push(10)
	if q.head.Load() != first {
		t.Fatal("new round did not reuse the original first fixed node")
	}
	// Fixed storage has reopened while the old batch still holds dynamic 4..7.
	queuePopInts(t, q, batch, []int{4, 5, 6, 7})
	queuePopInts(t, q, q.takeBatch(4, runtime.Gosched), []int{8, 9, 10})
	q.push(11)
	q.push(12)
	queuePopInts(t, q, q.takeBatch(4, runtime.Gosched), []int{11, 12})
	for round := range 32 {
		want := []int{round*3 + 13, round*3 + 14, round*3 + 15}
		for _, value := range want {
			q.push(value)
		}
		queuePopInts(t, q, q.takeBatch(4, runtime.Gosched), want)
		if q.reserved.Load() != 0 || q.consumed != 0 || &q.fixed[0] != first {
			t.Fatalf("fixed round %d did not complete in the same storage", round)
		}
	}
}

func TestQueuePublicationDuringBatchReversalRemainsInNextBatch(t *testing.T) {
	q := newQueue[int](2)
	for value := 1; value <= 6; value++ {
		q.push(value)
	}
	yields := 0
	batch := q.takeBatch(2, func() {
		yields++
		switch yields {
		case 1:
			if !q.empty() {
				t.Error("published stack was not detached before reversal")
			}
			q.push(100)
		case 2:
			q.push(101)
		}
	})
	if yields != 3 || q.empty() {
		t.Fatalf("reversal did not isolate new publications: yields=%d empty=%v", yields, q.empty())
	}
	for expected := 1; expected <= 2; expected++ {
		value, next := q.pop(batch)
		batch = next
		if value != expected {
			t.Fatalf("detached message=%d, want %d", value, expected)
		}
		q.push(101 + expected)
	}
	queuePopInts(t, q, batch, []int{3, 4, 5, 6})
	queuePopInts(t, q, q.takeBatch(2, runtime.Gosched), []int{100, 101, 102, 103})
	if !q.empty() {
		t.Fatal("new publications were lost or duplicated during reversal")
	}
}

func TestQueuePausedReservationDoesNotBlockPublishedMessages(t *testing.T) {
	q := newQueue[int](1)
	reserved, publish, done := make(chan struct{}), make(chan struct{}), make(chan struct{})
	releaseProducer := sync.OnceFunc(func() { close(publish) })
	t.Cleanup(releaseProducer)
	go func() {
		n := q.reserveNode()
		close(reserved)
		<-publish
		n.value = 1
		q.publish(n)
		close(done)
	}()
	queueAwait(t, reserved)
	q.push(2)
	q.push(3)
	queuePopInts(t, q, q.takeBatch(4, runtime.Gosched), []int{2, 3})
	if q.reserved.Load() != 1 || q.consumed != 0 || q.fixed[0].value != 0 {
		t.Fatal("unpublished fixed reservation was consumed or reused")
	}
	releaseProducer()
	queueAwait(t, done)
	queuePopInts(t, q, q.takeBatch(4, runtime.Gosched), []int{1})
	q.push(4)
	if q.head.Load() != &q.fixed[0] {
		t.Fatal("completed paused reservation did not allow fixed reuse")
	}
	queuePopInts(t, q, q.takeBatch(4, runtime.Gosched), []int{4})
}

func TestQueueStaleHeadAfterDetachFailsCASAndRepublishes(t *testing.T) {
	q := newQueue[int](2)
	q.push(1)
	paused := q.reserveNode()
	paused.value = 2
	expected := q.head.Load()
	paused.next = expected
	queuePopInts(t, q, q.takeBatch(4, runtime.Gosched), []int{1})
	if q.head.CompareAndSwap(expected, paused) {
		t.Fatal("publication CAS accepted a detached old head")
	}
	q.push(3)
	// Real publish must replace paused.next, which still points at the retired
	// head, with the latest published stack after the failed stale attempt.
	q.publish(paused)
	queuePopInts(t, q, q.takeBatch(4, runtime.Gosched), []int{3, 2})
	if !q.empty() || q.reserved.Load() != 0 {
		t.Fatal("retry lost a message or held the completed fixed round")
	}
}

func TestQueueStaleHeadAddressReuseCanSuccessfullyPublish(t *testing.T) {
	q := newQueue[int](1)
	q.push(1)
	paused := q.reserveNode()
	paused.value = 3
	expected := q.head.Load()
	paused.next = expected
	queuePopInts(t, q, q.takeBatch(4, runtime.Gosched), []int{1})
	q.push(2)
	if q.head.Load() != expected {
		t.Fatal("test did not reuse the old fixed head address")
	}
	if !q.head.CompareAndSwap(expected, paused) {
		t.Fatal("stale pointer CAS should match the current reused head")
	}
	q.push(4)
	queuePopInts(t, q, q.takeBatch(4, runtime.Gosched), []int{2, 3, 4})
	if !q.empty() || paused.value != 0 || paused.next != nil {
		t.Fatal("address reuse corrupted FIFO or retained the dynamic node")
	}
}

func TestQueueReferencePayloadsAndRetiredNodesAreCleared(t *testing.T) {
	type payload struct {
		id      int
		pointer *int
		bytes   []byte
		values  map[string]int
	}
	want := []payload{{}}
	for id := 1; id <= 3; id++ {
		value := id
		want = append(want, payload{id, &value, []byte{byte(id)}, map[string]int{"id": id}})
	}
	q := newQueue[payload](2)
	for _, message := range want {
		q.push(message)
	}
	var nodes []*node[payload]
	for n := q.head.Load(); n != nil; n = n.next {
		nodes = append(nodes, n)
	}
	batch := q.takeBatch(4, runtime.Gosched)
	for _, expected := range want {
		if batch == nil {
			t.Fatal("reference payload batch ended early")
		}
		actual, next := q.pop(batch)
		batch = next
		if !reflect.DeepEqual(actual, expected) || actual.pointer != expected.pointer {
			t.Fatalf("payload=%+v, want %+v", actual, expected)
		}
		if actual.id != 0 {
			if &actual.bytes[0] != &expected.bytes[0] {
				t.Fatal("slice payload was not passed as a Go value")
			}
			actual.values["consumer"] = actual.id
			if expected.values["consumer"] != actual.id {
				t.Fatal("map payload was not passed as a Go value")
			}
		}
	}
	if batch != nil || !q.empty() {
		t.Fatal("reference payload messages were duplicated")
	}
	for _, n := range nodes {
		if !reflect.DeepEqual(n.value, payload{}) || n.next != nil {
			t.Fatal("retired fixed or dynamic node retains payload references or a link")
		}
	}
}

func TestQueueConcurrentProducersAndContinuousConsumer(t *testing.T) {
	const producers, messages, chunk = 6, 400, 32
	type payload struct {
		producer, sequence, id int
		token                  *int
		check                  [3]uint64
	}
	makePayload := func(producer, sequence, id int) payload {
		return payload{producer, sequence, id, &id, [3]uint64{uint64(id), uint64(id) ^ 0x9e3779b9, uint64(id)*17 + 3}}
	}
	for _, capacity := range []int{1, 3, 8} {
		t.Run(fmt.Sprintf("capacity=%d", capacity), func(t *testing.T) {
			q := newQueue[payload](capacity)
			prefill := capacity + 3
			for sequence := range prefill {
				q.push(makePayload(-1, sequence, producers*messages+sequence))
			}
			held := q.takeBatch(4, runtime.Gosched)
			if !q.empty() || held == nil {
				t.Fatal("prefilled batch was not detached")
			}
			stop := make(chan struct{})
			stopWorkers := sync.OnceFunc(func() { close(stop) })
			t.Cleanup(stopWorkers)
			var ack [producers]chan struct{}
			var ready, senders sync.WaitGroup
			ready.Add(producers)
			senders.Add(producers)
			for producer := range producers {
				ack[producer] = make(chan struct{}, 1)
				go func() {
					defer senders.Done()
					reportReady := sync.OnceFunc(ready.Done)
					defer reportReady()
					for sequence := range messages {
						select {
						case <-stop:
							return
						default:
						}
						q.push(makePayload(producer, sequence, producer*messages+sequence))
						if (sequence+1)%chunk == 0 {
							if sequence+1 == chunk {
								reportReady()
							}
							// The next chunk cannot be produced until the single
							// consumer has processed this chunk's last message.
							select {
							case <-ack[producer]:
							case <-stop:
								return
							}
						}
					}
				}()
			}
			readyDone, sendersDone := make(chan struct{}), make(chan struct{})
			go func() { ready.Wait(); close(readyDone) }()
			go func() { senders.Wait(); close(sendersDone) }()
			queueAwait(t, readyDone)
			consumedDone := make(chan struct{})
			fixedPops, consumed := 0, 0
			total := producers*messages + prefill
			go func() {
				defer close(consumedDone)
				defer stopWorkers()
				last := make([]int, producers+1)
				for i := range last {
					last[i] = -1
				}
				seen := make([]bool, total)
				batch := held
				for consumed < total {
					select {
					case <-stop:
						return
					default:
					}
					if batch == nil {
						batch = q.takeBatch(4, runtime.Gosched)
						if batch == nil {
							runtime.Gosched()
							continue
						}
					}
					fixed := batch.fixed
					message, next := q.pop(batch)
					batch = next
					if fixed {
						fixedPops++
					}
					producer := message.producer
					expectedID := producer*messages + message.sequence
					if producer == -1 {
						producer = producers
						expectedID = producers*messages + message.sequence
					}
					if producer < 0 || producer > producers || message.id < 0 || message.id >= total {
						t.Errorf("corrupt payload: %+v", message)
						return
					}
					if message.sequence != last[producer]+1 || message.id != expectedID || seen[message.id] {
						t.Errorf("FIFO/uniqueness: message=%+v previous=%d", message, last[producer])
					}
					if message.token == nil || *message.token != message.id || message.check != makePayload(message.producer, message.sequence, message.id).check {
						t.Errorf("incomplete published payload: %+v", message)
					}
					last[producer] = message.sequence
					seen[message.id] = true
					consumed++
					if producer < producers && (message.sequence+1)%chunk == 0 {
						select {
						case ack[producer] <- struct{}{}:
						case <-stop:
							return
						}
					}
					// Yield while producers have subsequent chunks to publish,
					// including after fixed nodes have become reusable.
					runtime.Gosched()
				}
				for id, present := range seen {
					if !present {
						t.Errorf("missing message %d", id)
					}
				}
				if batch != nil || !q.empty() {
					t.Error("consumer finished with extra queued messages")
				}
			}()
			queueAwait(t, consumedDone)
			queueAwait(t, sendersDone)
			if consumed != total {
				t.Fatalf("consumer stopped after %d messages, want %d", consumed, total)
			}
			if fixedPops < capacity*2 {
				t.Fatalf("test did not exercise multiple fixed rounds: %d fixed pops", fixedPops)
			}
			if !q.empty() {
				t.Fatal("published stack not empty after all producers and consumer finished")
			}
			for i := range q.fixed {
				if q.fixed[i].value != (payload{}) || q.fixed[i].next != nil {
					t.Fatalf("fixed slot %d retained a consumed payload or link", i)
				}
			}
		})
	}
}

// runQueueModel uses slices for the two observable FIFO domains: the published
// queue and a separately held, possibly partially consumed batch. Reservations
// have no position in either FIFO until explicitly published.
func runQueueModel(t *testing.T, capacity int, operations []byte) {
	t.Helper()
	if len(operations) > 4096 {
		operations = operations[:4096]
	}
	type reservation struct {
		node  *node[int]
		value int
	}
	q := newQueue[int](capacity)
	fixed := &q.fixed[0]
	var pending []reservation
	var published, detached []int
	var batch *node[int]
	nextID := 0
	nextMessage := func() int { value := nextID; nextID++; return value }
	publish := func(index int) {
		r := pending[index]
		q.publish(r.node)
		published = append(published, r.value)
		pending = append(pending[:index], pending[index+1:]...)
	}
	take := func(inject bool) {
		if batch != nil {
			return
		}
		detached, published = published, nil
		injected := false
		batch = q.takeBatch(2, func() {
			if inject && !injected {
				value := nextMessage()
				q.push(value)
				published = append(published, value)
				injected = true
			}
		})
		if (batch == nil) != (len(detached) == 0) {
			t.Fatalf("model take: batch present=%v detached=%v", batch != nil, detached)
		}
	}
	pop := func() {
		if batch == nil {
			if len(detached) != 0 {
				t.Fatalf("model retained detached messages without a batch: %v", detached)
			}
			return
		}
		if len(detached) == 0 {
			t.Fatal("queue has a batch absent from the reference model")
		}
		retired := batch
		actual, next := q.pop(batch)
		batch = next
		if actual != detached[0] {
			t.Fatalf("model pop: got %d, want %d", actual, detached[0])
		}
		detached = detached[1:]
		if retired.value != 0 || retired.next != nil || (batch == nil) != (len(detached) == 0) {
			t.Fatal("model pop: node not cleared or detached batch length differs")
		}
	}
	for step, operation := range operations {
		switch operation % 7 {
		case 0:
			value := nextMessage()
			n := q.reserveNode()
			n.value = value
			pending = append(pending, reservation{n, value})
		case 1:
			if len(pending) != 0 {
				publish(int(operation/7) % len(pending))
			}
		case 2:
			value := nextMessage()
			q.push(value)
			published = append(published, value)
		case 3:
			take(false)
		case 4, 5:
			pop()
		case 6:
			take(true)
		}
		if q.empty() != (len(published) == 0) {
			t.Fatalf("step %d: empty disagrees with published FIFO (detached=%v reserved=%d)", step, detached, len(pending))
		}
		if &q.fixed[0] != fixed || len(q.fixed) != capacity {
			t.Fatalf("step %d: fixed storage changed", step)
		}
	}
	// Publishing pending reservations in reverse order deliberately separates
	// reservation order from FIFO order, including when a batch is held.
	for len(pending) > 0 {
		publish(len(pending) - 1)
	}
	for batch != nil {
		pop()
	}
	for len(published) > 0 {
		take(false)
		for batch != nil {
			pop()
		}
	}
	if !q.empty() || q.takeBatch(2, runtime.Gosched) != nil {
		t.Fatal("model flush left published messages")
	}
	for i := range q.fixed {
		if q.fixed[i].value != 0 || q.fixed[i].next != nil {
			t.Fatalf("model flush retained data in fixed slot %d", i)
		}
	}
}

func TestQueueMatchesIndependentFIFOModel(t *testing.T) {
	for _, capacity := range []int{1, 2, 3, 8, 64} {
		for _, seed := range []int64{1, 17, 2026, 8675309} {
			t.Run(fmt.Sprintf("capacity=%d/seed=%d", capacity, seed), func(t *testing.T) {
				operations := make([]byte, 2048)
				_, _ = rand.New(rand.NewSource(seed)).Read(operations)
				runQueueModel(t, capacity, operations)
			})
		}
	}
}