queue_edge_test.go

package chat_test

import (
	"context"
	"log/slog"
	"net/http"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/coder/chat"
)

func newQueueRuntime(t *testing.T, state chat.State, adapter chat.Adapter, logs *syncBuffer, mutate ...func(*chat.RuntimeOptions)) *chat.Chat {
	t.Helper()
	options := chat.RuntimeOptions{
		DedupeTTL:     time.Hour,
		ThreadLockTTL: 40 * time.Millisecond,
		Concurrency:   chat.ConcurrencyQueue,
		Dispatch:      chat.DispatchDeferred,
		MaxDetached:   1024,
		DetachTimeout: 5 * time.Second,
	}
	for _, m := range mutate {
		m(&options)
	}
	bot, err := chat.New(context.Background(),
		chat.WithState(state),
		chat.WithAdapter(adapter),
		chat.WithLogger(slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug}))),
		chat.WithRuntimeOptions(options),
	)
	if err != nil {
		t.Fatalf("new queue runtime: %v", err)
	}
	return bot
}

func TestQueueSingleFollowUpRunsAfterInflightHandler(t *testing.T) {
	t.Parallel()

	state := newFakeState()
	adapter := newFakeAdapter("fake")
	var logs syncBuffer
	bot := newQueueRuntime(t, state, adapter, &logs)

	firstStarted := make(chan struct{})
	releaseFirst := make(chan struct{})
	var mu sync.Mutex
	var handled []string
	bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error {
		mu.Lock()
		first := len(handled) == 0
		handled = append(handled, ev.Event.ID)
		mu.Unlock()
		if first {
			close(firstStarted)
			<-releaseFirst
		}
		return nil
	})

	if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK {
		t.Fatalf("first status = %d", status)
	}
	select {
	case <-firstStarted:
	case <-time.After(time.Second):
		t.Fatal("first handler did not start")
	}

	// One follow-up arrives while the first handler holds the lock. It must
	// queue, not drop.
	followDone := make(chan postEventResult, 1)
	go func() { followDone <- postEventResultFor(bot, "fake", mentionEvent("follow", "fake:v1:thread-1")) }()

	// Give the follow-up time to register as a queued waiter, then release first.
	time.Sleep(30 * time.Millisecond)
	close(releaseFirst)

	select {
	case res := <-followDone:
		if res.err != nil || res.status != http.StatusOK {
			t.Fatalf("follow-up result status=%d err=%v", res.status, res.err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("follow-up dispatch did not return")
	}

	deadline := time.After(2 * time.Second)
	for {
		mu.Lock()
		n := len(handled)
		mu.Unlock()
		if n >= 2 {
			break
		}
		select {
		case <-deadline:
			mu.Lock()
			got := append([]string(nil), handled...)
			mu.Unlock()
			t.Fatalf("follow-up did not run after in-flight handler, handled = %v", got)
		case <-time.After(5 * time.Millisecond):
		}
	}

	time.Sleep(40 * time.Millisecond)
	mu.Lock()
	defer mu.Unlock()
	if len(handled) != 2 {
		t.Fatalf("handled = %v, want exactly first then follow", handled)
	}
	if handled[0] != "first" || handled[1] != "follow" {
		t.Fatalf("handled order = %v, want [first follow]", handled)
	}
	// A lone follow-up was never superseded, so no supersession should be logged.
	if strings.Contains(logs.String(), "chat queue superseded") {
		t.Fatalf("lone follow-up should not be superseded; logs:\n%s", logs.String())
	}
}

func TestQueueSupersededEventSurfacesSupersededByObservation(t *testing.T) {
	t.Parallel()

	state := newFakeState()
	adapter := newFakeAdapter("fake")
	var logs syncBuffer
	bot := newQueueRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) {
		o.ThreadLockTTL = 60 * time.Millisecond
	})

	firstStarted := make(chan struct{})
	releaseFirst := make(chan struct{})
	var mu sync.Mutex
	var handled []string
	bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error {
		mu.Lock()
		first := len(handled) == 0
		handled = append(handled, ev.Event.ID)
		mu.Unlock()
		if first {
			close(firstStarted)
			<-releaseFirst
		}
		return nil
	})

	if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK {
		t.Fatalf("first status = %d", status)
	}
	select {
	case <-firstStarted:
	case <-time.After(time.Second):
		t.Fatal("first handler did not start")
	}

	// "older" queues first, then "newer" supersedes it.
	var wg sync.WaitGroup
	for _, id := range []string{"older", "newer"} {
		wg.Go(func() {
			res := postEventResultFor(bot, "fake", mentionEvent(id, "fake:v1:thread-1"))
			if res.err != nil || res.status != http.StatusOK {
				t.Errorf("follow-up %s status=%d err=%v", id, res.status, res.err)
			}
		})
		time.Sleep(15 * time.Millisecond)
	}

	close(releaseFirst)
	wg.Wait()

	deadline := time.After(2 * time.Second)
	for {
		mu.Lock()
		n := len(handled)
		mu.Unlock()
		if n >= 2 {
			break
		}
		select {
		case <-deadline:
			t.Fatalf("queued winner did not run, handled = %v", handled)
		case <-time.After(5 * time.Millisecond):
		}
	}
	time.Sleep(40 * time.Millisecond)

	mu.Lock()
	defer mu.Unlock()
	if len(handled) != 2 {
		t.Fatalf("handled = %v, want first then the newest follow-up only", handled)
	}
	if handled[1] != "newer" {
		t.Fatalf("queued winner = %q, want newer (most recent)", handled[1])
	}

	out := logs.String()
	if !strings.Contains(out, "chat queue superseded") {
		t.Fatalf("superseded event not surfaced via observation; logs:\n%s", out)
	}
	// The displacing event id must be carried on superseded_by.
	if !strings.Contains(out, "superseded_by=newer") {
		t.Fatalf("supersession did not carry superseded_by=newer; logs:\n%s", out)
	}
}

func TestQueueDropRemainsDefaultRegression(t *testing.T) {
	t.Parallel()

	state := newFakeState()
	adapter := newFakeAdapter("fake")
	var logs syncBuffer
	// Deferred dispatch but Concurrency left as the zero value (ConcurrencyDrop).
	bot, err := chat.New(context.Background(),
		chat.WithState(state),
		chat.WithAdapter(adapter),
		chat.WithLogger(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug}))),
		chat.WithRuntimeOptions(chat.RuntimeOptions{
			DedupeTTL:     time.Hour,
			ThreadLockTTL: time.Hour,
			Dispatch:      chat.DispatchDeferred,
			MaxDetached:   1024,
			DetachTimeout: 5 * time.Second,
		}),
	)
	if err != nil {
		t.Fatalf("new runtime: %v", err)
	}

	firstStarted := make(chan struct{})
	releaseFirst := make(chan struct{})
	var mu sync.Mutex
	var handled []string
	bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error {
		mu.Lock()
		first := len(handled) == 0
		handled = append(handled, ev.Event.ID)
		mu.Unlock()
		if first {
			close(firstStarted)
			<-releaseFirst
		}
		return nil
	})

	if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK {
		t.Fatalf("first status = %d", status)
	}
	select {
	case <-firstStarted:
	case <-time.After(time.Second):
		t.Fatal("first handler did not start")
	}

	// Follow-up under drop: acked and dropped, never queued.
	if status := postEvent(t, bot, "fake", mentionEvent("dropped", "fake:v1:thread-1")); status != http.StatusOK {
		t.Fatalf("dropped follow-up status = %d", status)
	}

	mu.Lock()
	if len(handled) != 1 {
		got := append([]string(nil), handled...)
		mu.Unlock()
		t.Fatalf("drop strategy ran a follow-up: handled = %v", got)
	}
	mu.Unlock()

	out := logs.String()
	if !strings.Contains(out, "chat lock conflict dropped") {
		t.Fatalf("drop did not surface a lock-conflict-dropped observation; logs:\n%s", out)
	}
	if strings.Contains(out, "chat queue superseded") || strings.Contains(out, "chat queue wait abandoned") {
		t.Fatalf("queue observations leaked under the drop default; logs:\n%s", out)
	}

	close(releaseFirst)

	// Let the first handler finish; no second handler should ever run.
	deadline := time.Now().Add(200 * time.Millisecond)
	for time.Now().Before(deadline) {
		mu.Lock()
		n := len(handled)
		mu.Unlock()
		if n > 1 {
			t.Fatalf("dropped follow-up ran after first completed")
		}
		time.Sleep(10 * time.Millisecond)
	}
}

func TestQueueWaitAbandonedSurfacesObservation(t *testing.T) {
	t.Parallel()

	state := newFakeState()
	adapter := newFakeAdapter("fake")
	var logs syncBuffer
	bot := newQueueRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) {
		o.ThreadLockTTL = time.Hour
		// The detached waiter abandons at DetachTimeout; keep it short so the
		// abandonment fires quickly without relying on the request lifetime.
		o.DetachTimeout = 80 * time.Millisecond
	})

	var mu sync.Mutex
	var calls int
	bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error {
		mu.Lock()
		calls++
		mu.Unlock()
		return nil
	})

	// Hold the lock with an outside token that never releases.
	if _, acquired, err := state.AcquireLock(context.Background(), "fake:v1:thread-1", time.Hour); err != nil || !acquired {
		t.Fatalf("seed lock acquired=%v err=%v", acquired, err)
	}

	// Ack is prompt (ack-then-work): the post returns 200 well before the
	// detached waiter abandons at DetachTimeout.
	if status := postEvent(t, bot, "fake", mentionEvent("abandoned", "fake:v1:thread-1")); status != http.StatusOK {
		t.Fatalf("abandoned queue wait status = %d, want 200", status)
	}

	// The abandonment must be observable once the Detached Work Context expires.
	deadline := time.After(2 * time.Second)
	for !strings.Contains(logs.String(), "chat queue wait abandoned") {
		select {
		case <-deadline:
			t.Fatalf("abandonment not surfaced via observation; logs:\n%s", logs.String())
		case <-time.After(5 * time.Millisecond):
		}
	}

	mu.Lock()
	defer mu.Unlock()
	if calls != 0 {
		t.Fatalf("handler ran despite never acquiring the lock, calls = %d", calls)
	}
}

func TestQueueDeferredFollowUpAcksPromptlyWhileInflightWorks(t *testing.T) {
	t.Parallel()

	state := newFakeState()
	adapter := newFakeAdapter("fake")
	var logs syncBuffer
	bot := newQueueRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) {
		o.ThreadLockTTL = time.Hour
		o.DetachTimeout = 5 * time.Second
	})

	firstStarted := make(chan struct{})
	releaseFirst := make(chan struct{})
	defer close(releaseFirst)
	var mu sync.Mutex
	var handled []string
	bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error {
		mu.Lock()
		first := len(handled) == 0
		handled = append(handled, ev.Event.ID)
		mu.Unlock()
		if first {
			close(firstStarted)
			<-releaseFirst // hold the lock indefinitely
		}
		return nil
	})

	if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK {
		t.Fatalf("first status = %d", status)
	}
	select {
	case <-firstStarted:
	case <-time.After(time.Second):
		t.Fatal("first handler did not start")
	}

	// The follow-up's post (ack) must return promptly even though the in-flight
	// handler still holds the lock and never releases it. If the queue wait still
	// blocked the request, this post would not return until releaseFirst (never).
	acked := make(chan int, 1)
	go func() { acked <- postEvent(t, bot, "fake", mentionEvent("follow", "fake:v1:thread-1")) }()
	select {
	case status := <-acked:
		if status != http.StatusOK {
			t.Fatalf("queued follow-up ack status = %d, want 200", status)
		}
	case <-time.After(500 * time.Millisecond):
		t.Fatal("queued follow-up ack blocked on the in-flight handler (not ack-then-work)")
	}

	// The follow-up's handler must NOT have run yet: it is waiting on the detached
	// context for the lock the first handler still holds.
	mu.Lock()
	if len(handled) != 1 {
		got := append([]string(nil), handled...)
		mu.Unlock()
		t.Fatalf("queued follow-up ran before the in-flight handler released, handled = %v", got)
	}
	mu.Unlock()
}

func TestQueueDeferredWaiterDrainedAndCancelledByShutdown(t *testing.T) {
	t.Parallel()

	state := newFakeState()
	adapter := newFakeAdapter("fake")
	var logs syncBuffer
	bot := newQueueRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) {
		o.ThreadLockTTL = time.Hour
		// Long enough that only Shutdown (not the timeout) can end the wait.
		o.DetachTimeout = time.Hour
	})

	var mu sync.Mutex
	var calls int
	bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error {
		mu.Lock()
		calls++
		mu.Unlock()
		return nil
	})

	// Hold the lock with an outside token that never releases, so the queued
	// waiter can only ever be ended by Shutdown.
	if _, acquired, err := state.AcquireLock(context.Background(), "fake:v1:thread-1", time.Hour); err != nil || !acquired {
		t.Fatalf("seed lock acquired=%v err=%v", acquired, err)
	}

	if status := postEvent(t, bot, "fake", mentionEvent("waiter", "fake:v1:thread-1")); status != http.StatusOK {
		t.Fatalf("waiter status = %d", status)
	}

	// Give the detached waiter time to enter its poll loop.
	time.Sleep(40 * time.Millisecond)

	// Shutdown must return (drain completes) even though the lock never frees,
	// because baseCancel cancels the waiter's detached context and the inflight
	// WaitGroup tracks it.
	done := make(chan error, 1)
	go func() { done <- bot.Shutdown(context.Background()) }()
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("shutdown error: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("shutdown did not drain the queued waiter (not cancelled by baseCancel / not tracked by inflight)")
	}

	if !strings.Contains(logs.String(), "chat queue wait abandoned") {
		t.Fatalf("shutdown-cancelled waiter not surfaced as abandoned; logs:\n%s", logs.String())
	}
	mu.Lock()
	defer mu.Unlock()
	if calls != 0 {
		t.Fatalf("handler ran despite the lock never freeing, calls = %d", calls)
	}
}

type gatedAcquireState struct {
	*fakeState
	gateCall  int
	reached   chan struct{}
	release   chan struct{}
	gatePanic any

	mu    sync.Mutex
	calls int
}

func newGatedAcquireState(gateCall int) *gatedAcquireState {
	return &gatedAcquireState{
		fakeState: newFakeState(),
		gateCall:  gateCall,
		reached:   make(chan struct{}),
		release:   make(chan struct{}),
	}
}

func (s *gatedAcquireState) AcquireLock(ctx context.Context, key string, ttl time.Duration) (chat.LockLease, bool, error) {
	s.mu.Lock()
	s.calls++
	gated := s.calls == s.gateCall
	s.mu.Unlock()
	if gated {
		close(s.reached)
		<-s.release
		if s.gatePanic != nil {
			panic(s.gatePanic)
		}
	}
	return s.fakeState.AcquireLock(ctx, key, ttl)
}

func (s *gatedAcquireState) acquireCalls() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.calls
}

func (s *gatedAcquireState) lockHeld(key string) bool {
	s.fakeState.mu.Lock()
	defer s.fakeState.mu.Unlock()
	_, held := s.fakeState.locked[key]
	return held
}

const queueThread = chat.ThreadID("fake:v1:thread-1")

var queueDispatchModes = []struct {
	name string
	mode chat.DispatchMode
}{
	{name: "deferred", mode: chat.DispatchDeferred},
	{name: "sync", mode: chat.DispatchSync},
}

type gatedQueueRun struct {
	bot          *chat.Chat
	obs          *recordingObserver
	logs         *syncBuffer
	blockStarted chan struct{}
	releaseBlock chan struct{}

	mu      sync.Mutex
	handled []string
}

func newGatedQueueRun(t *testing.T, state chat.State, mode chat.DispatchMode, blockID string) *gatedQueueRun {
	t.Helper()
	run := &gatedQueueRun{
		obs:          &recordingObserver{},
		logs:         newSyncBuffer(),
		blockStarted: make(chan struct{}),
		releaseBlock: make(chan struct{}),
	}
	bot, err := chat.New(context.Background(),
		chat.WithState(state),
		chat.WithAdapter(newFakeAdapter("fake")),
		chat.WithLogger(slog.New(slog.NewTextHandler(run.logs, nil))),
		chat.WithObserver(run.obs),
		chat.WithRuntimeOptions(chat.RuntimeOptions{
			DedupeTTL:     time.Hour,
			ThreadLockTTL: time.Hour,
			Concurrency:   chat.ConcurrencyQueue,
			Dispatch:      mode,
			MaxDetached:   1024,
			DetachTimeout: 5 * time.Second,
		}),
	)
	if err != nil {
		t.Fatalf("new runtime: %v", err)
	}
	t.Cleanup(func() {
		// A check that fails before a gate opens leaves work blocked on it, so bound
		// the drain: an unbounded Shutdown hangs and hides the failure.
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = bot.Shutdown(ctx)
	})
	var blockOnce sync.Once
	bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error {
		run.mu.Lock()
		run.handled = append(run.handled, ev.Event.ID)
		run.mu.Unlock()
		if ev.Event.ID == blockID {
			blockOnce.Do(func() { close(run.blockStarted) })
			<-run.releaseBlock
		}
		return nil
	})
	run.bot = bot
	return run
}

func (r *gatedQueueRun) post(id string) <-chan postEventResult {
	done := make(chan postEventResult, 1)
	go func() { done <- postEventResultFor(r.bot, "fake", mentionEvent(id, queueThread)) }()
	return done
}

func (r *gatedQueueRun) handledIDs() []string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return append([]string(nil), r.handled...)
}

func waitClosed(t *testing.T, ch <-chan struct{}, msg string) {
	t.Helper()
	select {
	case <-ch:
	case <-time.After(2 * time.Second):
		t.Fatal(msg)
	}
}

func TestQueueNewerDeliveryThatTakesTheFreeLockSupersedesParkedWaiter(t *testing.T) {
	t.Parallel()

	for _, tc := range queueDispatchModes {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			const (
				aPreludeAcquire = 1 + iota
				bPreludeAcquire
				bFirstPoll
			)
			state := newGatedAcquireState(bFirstPoll)
			run := newGatedQueueRun(t, state, tc.mode, "A")

			results := []<-chan postEventResult{run.post("A")}
			waitClosed(t, run.blockStarted, "A's handler did not start")
			results = append(results, run.post("B"))
			waitClosed(t, state.reached, "B did not poll for the lock")
			close(run.releaseBlock)
			eventually(t, 2*time.Second, func() bool { return !state.lockHeld(string(queueThread)) },
				"A did not release the Thread Lock")
			results = append(results, run.post("C"))
			eventually(t, 2*time.Second, func() bool {
				got := run.handledIDs()
				return len(got) == 2 && got[1] == "C" && !state.lockHeld(string(queueThread))
			}, "C did not take the free lock and run")
			close(state.release)
			eventually(t, 2*time.Second, func() bool { return len(run.obs.terminalOutcomes()) == 3 },
				"B did not reach a terminal outcome")

			for i, done := range results {
				select {
				case res := <-done:
					if res.err != nil || res.status != http.StatusOK {
						t.Fatalf("delivery %d: status=%d err=%v", i, res.status, res.err)
					}
				case <-time.After(2 * time.Second):
					t.Fatalf("delivery %d did not return", i)
				}
			}
			if got := run.handledIDs(); !equalStrings(got, []string{"A", "C"}) {
				t.Fatalf("handled = %v, want [A C]: B was queued before C, so C supersedes it", got)
			}
			var handledCount, ignoredCount int
			for _, outcome := range run.obs.terminalOutcomes() {
				switch outcome {
				case chat.OutcomeHandled:
					handledCount++
				case chat.OutcomeIgnored:
					ignoredCount++
				}
			}
			if handledCount != 2 || ignoredCount != 1 {
				t.Fatalf("outcomes = %v, want A and C handled and B ignored", run.obs.terminalOutcomes())
			}
			if !strings.Contains(run.logs.String(), "superseded_by=C") {
				t.Fatalf("B's supersession by C was not logged; logs:\n%s", run.logs.String())
			}
		})
	}
}

func TestQueueDuplicateLockTakerKeepsParkedWaiter(t *testing.T) {
	t.Parallel()

	for _, tc := range queueDispatchModes {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			const (
				aPreludeAcquire = 1 + iota
				bPreludeAcquire
				bFirstPoll
			)
			state := newGatedAcquireState(bFirstPoll)
			run := newGatedQueueRun(t, state, tc.mode, "A")

			run.post("A")
			waitClosed(t, run.blockStarted, "A's handler did not start")
			run.post("B")
			waitClosed(t, state.reached, "B did not poll for the lock")
			close(run.releaseBlock)
			eventually(t, 2*time.Second, func() bool { return !state.lockHeld(string(queueThread)) },
				"A did not release the Thread Lock")
			run.post("A")
			eventually(t, 2*time.Second, func() bool {
				outcomes := run.obs.terminalOutcomes()
				return len(outcomes) == 2 && outcomes[1] == chat.OutcomeDuplicate && !state.lockHeld(string(queueThread))
			}, "the redelivery of A did not take the free lock and end as a duplicate")
			close(state.release)
			eventually(t, 2*time.Second, func() bool { return len(run.obs.terminalOutcomes()) == 3 },
				"B did not reach a terminal outcome")

			if got := run.handledIDs(); !equalStrings(got, []string{"A", "B"}) {
				t.Fatalf("handled = %v, want [A B]: a duplicate must not supersede the waiter", got)
			}
			if strings.Contains(run.logs.String(), "superseded_by") {
				t.Fatalf("unexpected supersession; logs:\n%s", run.logs.String())
			}
		})
	}
}

func TestQueueOlderLockTakerKeepsNewerWaiter(t *testing.T) {
	t.Parallel()

	for _, tc := range queueDispatchModes {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			const (
				aPreludeAcquire = 1 + iota
				cPreludeAcquire
				bPreludeAcquire
				bFirstPoll
			)
			state := newGatedAcquireState(cPreludeAcquire)
			run := newGatedQueueRun(t, state, tc.mode, "A")

			run.post("A")
			waitClosed(t, run.blockStarted, "A's handler did not start")
			run.post("C")
			waitClosed(t, state.reached, "C's prelude did not reach the lock")
			run.post("B")
			eventually(t, 2*time.Second, func() bool { return state.acquireCalls() >= bFirstPoll },
				"B did not park and poll for the lock")
			close(run.releaseBlock)
			eventually(t, 2*time.Second, func() bool { return !state.lockHeld(string(queueThread)) },
				"A did not release the Thread Lock")
			close(state.release)
			eventually(t, 4*time.Second, func() bool { return len(run.obs.terminalOutcomes()) == 3 },
				"the deliveries did not finish")

			if got := run.handledIDs(); !equalStrings(got, []string{"A", "C", "B"}) {
				t.Fatalf("handled = %v, want [A C B]: the older C must not supersede the newer B", got)
			}
		})
	}
}

type gatedMarkState struct {
	*gatedAcquireState
	gateID      string
	markReached chan struct{}
	markRelease chan struct{}
}

func newGatedMarkState(gateID string) *gatedMarkState {
	return &gatedMarkState{
		gatedAcquireState: newGatedAcquireState(0),
		gateID:            gateID,
		markReached:       make(chan struct{}),
		markRelease:       make(chan struct{}),
	}
}

func (s *gatedMarkState) MarkEvent(ctx context.Context, id string, ttl time.Duration) (bool, error) {
	if id == s.gateID {
		close(s.markReached)
		<-s.markRelease
	}
	return s.gatedAcquireState.MarkEvent(ctx, id, ttl)
}

func TestQueueDeliveryStalledBeforeRegisteringNeverRunsAfterNewerLockTaker(t *testing.T) {
	t.Parallel()

	for _, tc := range queueDispatchModes {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			state := newGatedMarkState("X")
			run := newGatedQueueRun(t, state, tc.mode, "A")

			run.post("A")
			waitClosed(t, run.blockStarted, "A's handler did not start")
			run.post("X")
			waitClosed(t, state.markReached, "X did not reach its dedupe mark after the Lock Conflict")
			close(run.releaseBlock)
			eventually(t, 2*time.Second, func() bool { return !state.lockHeld(string(queueThread)) },
				"A did not release the Thread Lock")
			run.post("C")
			eventually(t, 2*time.Second, func() bool {
				got := run.handledIDs()
				return len(got) == 2 && got[1] == "C" && !state.lockHeld(string(queueThread))
			}, "C did not take the free lock and run")
			close(state.markRelease)
			eventually(t, 4*time.Second, func() bool { return len(run.obs.terminalOutcomes()) == 3 },
				"X did not reach a terminal outcome")

			if got := run.handledIDs(); !equalStrings(got, []string{"A", "C"}) {
				t.Fatalf("handled = %v, want [A C]: X was admitted before C, so C's turn supersedes it", got)
			}
		})
	}
}

func TestQueueDeliveryStalledInItsLockCallNeverRunsAfterNewerLockTaker(t *testing.T) {
	t.Parallel()

	for _, tc := range queueDispatchModes {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			const xPreludeAcquire = 1
			state := newGatedAcquireState(xPreludeAcquire)
			run := newGatedQueueRun(t, state, tc.mode, "")

			run.post("X")
			waitClosed(t, state.reached, "X's prelude did not reach the lock")
			run.post("C")
			eventually(t, 2*time.Second, func() bool {
				return equalStrings(run.handledIDs(), []string{"C"}) && !state.lockHeld(string(queueThread))
			}, "C did not take the free lock and run")
			close(state.release)
			eventually(t, 2*time.Second, func() bool { return len(run.obs.terminalOutcomes()) == 2 },
				"X did not reach a terminal outcome")

			if got := run.handledIDs(); !equalStrings(got, []string{"C"}) {
				t.Fatalf("handled = %v, want [C]: X was admitted before C, so C's turn supersedes it", got)
			}
			if state.lockHeld(string(queueThread)) {
				t.Fatal("X kept the Thread Lock after it ended as superseded")
			}
		})
	}
}

func TestQueueDeliveryStalledBeforeRegisteringNeverRunsAfterNewerWaiter(t *testing.T) {
	t.Parallel()

	for _, tc := range queueDispatchModes {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			const cFirstPoll = 4 // after the prelude acquires of A, X, and C
			state := newGatedMarkState("X")
			run := newGatedQueueRun(t, state, tc.mode, "A")

			run.post("A")
			waitClosed(t, run.blockStarted, "A's handler did not start")
			run.post("X")
			waitClosed(t, state.markReached, "X did not reach its dedupe mark after the Lock Conflict")
			run.post("C")
			eventually(t, 2*time.Second, func() bool { return state.acquireCalls() >= cFirstPoll },
				"C did not park and poll for the lock")
			close(run.releaseBlock)
			eventually(t, 2*time.Second, func() bool {
				return equalStrings(run.handledIDs(), []string{"A", "C"}) && !state.lockHeld(string(queueThread))
			}, "C did not take its turn after A")
			close(state.markRelease)
			eventually(t, 4*time.Second, func() bool { return len(run.obs.terminalOutcomes()) == 3 },
				"X did not reach a terminal outcome")

			if got := run.handledIDs(); !equalStrings(got, []string{"A", "C"}) {
				t.Fatalf("handled = %v, want [A C]: X was admitted before C, so C's turn supersedes it", got)
			}
		})
	}
}