package main import ( "context" "sync" "testing" "time" ) func TestMessageQueue_FIFO(t *testing.T) { q := newMessageQueue() q.enqueue("s", "first") q.enqueue("s", "second") q.enqueue("s", "third") want := []string{"first", "second", "third"} for _, w := range want { got, ok := q.dequeue("s") if !ok || got != w { t.Fatalf("dequeue = %q,%v want %q,true", got, ok, w) } } if _, ok := q.dequeue("s"); ok { t.Fatal("dequeue on drained queue should return ok=false") } } func TestMessageQueue_RequeueFront(t *testing.T) { q := newMessageQueue() q.enqueue("s", "a") q.enqueue("s", "b") // Pop "a", then push it back to the front; "a" must come out before "b". a, _ := q.dequeue("s") q.requeueFront("s", a) got, _ := q.dequeue("s") if got != "a" { t.Fatalf("after requeueFront, dequeue = %q want %q", got, "a") } got2, _ := q.dequeue("s") if got2 != "b" { t.Fatalf("next dequeue = %q want %q", got2, "b") } } func TestMessageQueue_IsolatedPerSession(t *testing.T) { q := newMessageQueue() q.enqueue("s1", "one") q.enqueue("s2", "two") if got, _ := q.dequeue("s1"); got != "one" { t.Fatalf("s1 = %q want one", got) } if got, _ := q.dequeue("s2"); got != "two" { t.Fatalf("s2 = %q want two", got) } if q.peek("s1") != 0 || q.peek("s2") != 0 { t.Fatal("both sessions should be drained") } } func TestMessageQueue_Concurrent(t *testing.T) { q := newMessageQueue() const n = maxQueuedPerSession // stay under the cap so every enqueue lands var wg sync.WaitGroup for i := 0; i < n; i++ { wg.Add(1) go func(i int) { defer wg.Done() q.enqueue("s", "m") }(i) } wg.Wait() if q.peek("s") != n { t.Fatalf("peek = %d want %d (all enqueues must be counted)", q.peek("s"), n) } seen := 0 for { if _, ok := q.dequeue("s"); !ok { break } seen++ } if seen != n { t.Fatalf("drained %d want %d", seen, n) } } func TestMessageQueue_CapsOverflow(t *testing.T) { q := newMessageQueue() for i := 0; i < maxQueuedPerSession; i++ { if !q.enqueue("s", "m") { t.Fatalf("enqueue #%d within cap should succeed", i) } } if q.enqueue("s", "overflow") { t.Fatal("enqueue past the cap should return false (dropped)") } if got := q.peek("s"); got != maxQueuedPerSession { t.Fatalf("peek = %d want %d (overflow must not append)", got, maxQueuedPerSession) } } // drainQueued on an empty queue must be a no-op: it returns immediately and // never touches the gate (so the session stays free for the next turn). func TestDrainQueued_NoOpOnEmpty(t *testing.T) { a := &agent{gate: newTurnGate(), queue: newMessageQueue()} a.drainQueued(context.Background(), "s") if !a.gate.acquire("s", 0) { t.Fatal("gate should be free after a no-op drain (drain must not hold it)") } a.gate.release("s") } // With a queued message but the gate held by another turn, drainQueued must // re-queue the message and return WITHOUT running a turn (no store/provider → a // real run would panic). This is the "never stack" property: a busy gate // defers to the holder's own release-drain. func TestDrainQueued_RequeuesWhenBusy(t *testing.T) { prev := drainAcquireWait drainAcquireWait = 10 * time.Millisecond t.Cleanup(func() { drainAcquireWait = prev }) a := &agent{gate: newTurnGate(), queue: newMessageQueue()} if !a.gate.acquire("s", 0) { t.Fatal("precondition: hold the gate") } a.queue.enqueue("s", "queued-msg") done := make(chan struct{}) go func() { a.drainQueued(context.Background(), "s") // must not panic; must requeue close(done) }() select { case <-done: case <-time.After(time.Second): t.Fatal("drainQueued did not return promptly while the gate was busy") } if got := a.queue.peek("s"); got != 1 { t.Fatalf("message should be re-queued while busy; peek = %d want 1", got) } a.gate.release("s") }