bug(runtime): ConcurrencyQueue runs an older queued event after a newer one that took the free lock #121

ConcurrencyQueue can run an older queued event after a newer one. The newer event barges past the parked waiter: it takes the free Thread Lock in its prelude and never supersedes the waiter, so the waiter still runs afterwards. The bot then answers messages out of order.

The queue GoDoc says the strategy "dispatches the most recent superseded event for the scope". ADR 0015 §3 decides that "most-recent-wins follows the local dispatch admission sequence". This breaks both.

Expected and actual

Three messages A, B, C arrive for one thread, in that order, on one runtime instance.

• Expected: handlers run for A, then C. B is superseded by the newer C (or, at the least, B never runs after C).
• Actual: handlers run for A, then C, then B.

Steps, in the code's words (runtime.go)

1. A arrives. Its prelude takes the free Thread Lock (c.state.AcquireLock), and its handler runs.
2. B arrives while A holds the lock. B's prelude gets a Lock Conflict, and routedWork registers B as the scope's pending waiter (registerPending). B's tail polls in pollForLock, every queuePollInterval() (ThreadLockTTL/20, at most 1 s).
3. A's handler returns, and runLockedTail releases the lock.
4. C arrives before B's next poll. C's prelude finds the lock free and takes it. C is not conflicted, so it never calls registerPending, and B stays the pending waiter. C's handler runs.
5. C releases the lock. B's next poll takes it, and queueForLock re-checks isPending. B is still the waiter, so B's handler runs after C's.

The window is the time between A's release and B's next poll: up to 1 s with the default 2-minute ThreadLockTTL. A chatty thread under DispatchDeferred hits it whenever a message arrives just after a long turn ends.

Reproduction

This test fails 3 out of 3 runs at 8c246b1 with handled = [A C B]. Put it in the repository root (it uses the chat_test fakes) and run go test -run TestReplayQueueNewerDeliveryBargesPastParkedWaiter -count=3 -v ..

package chat_test

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

	"github.com/coder/chat"
)

func TestReplayQueueNewerDeliveryBargesPastParkedWaiter(t *testing.T) {
	state := newFakeState()
	adapter := newFakeAdapter("fake")
	bot, err := chat.New(context.Background(),
		chat.WithState(state),
		chat.WithAdapter(adapter),
		chat.WithLogger(slog.New(slog.NewTextHandler(newSyncBuffer(), nil))),
		chat.WithRuntimeOptions(chat.RuntimeOptions{
			DedupeTTL: time.Hour,
			// ThreadLockTTL/20 clamps to the 1s maximum poll interval.
			ThreadLockTTL: time.Hour,
			Concurrency:   chat.ConcurrencyQueue,
			Dispatch:      chat.DispatchDeferred,
			MaxDetached:   1024,
			DetachTimeout: 10 * time.Second,
		}),
	)
	if err != nil {
		t.Fatalf("new runtime: %v", err)
	}
	t.Cleanup(func() { _ = bot.Shutdown(context.Background()) })

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

	// Steps 2-3: A arrives, takes the Thread Lock in its prelude, and runs.
	if status := postEvent(t, bot, "fake", mentionEvent("A", "fake:v1:thread-1")); status != http.StatusOK {
		t.Fatalf("A status = %d", status)
	}
	<-firstStarted
	// B arrives while A runs: AcquireLock conflicts, B registers as the
	// scope's pending waiter, and its tail starts polling (1s ticker).
	if status := postEvent(t, bot, "fake", mentionEvent("B", "fake:v1:thread-1")); status != http.StatusOK {
		t.Fatalf("B status = %d", status)
	}
	time.Sleep(20 * time.Millisecond)
	// A's handler returns and its tail releases the lock.
	close(releaseFirst)
	<-done
	time.Sleep(20 * time.Millisecond)
	// C arrives before B's next poll: C's prelude AcquireLock succeeds.
	if status := postEvent(t, bot, "fake", mentionEvent("C", "fake:v1:thread-1")); status != http.StatusOK {
		t.Fatalf("C status = %d", status)
	}
	// Wait for every handler that will run.
	deadline := time.After(5 * time.Second)
	for {
		mu.Lock()
		n := len(handled)
		mu.Unlock()
		if n >= 3 {
			break
		}
		select {
		case <-deadline:
			mu.Lock()
			t.Logf("handled = %v (B was superseded, as most-recent-wins requires)", handled)
			mu.Unlock()
			return
		case <-time.After(10 * time.Millisecond):
		}
	}
	mu.Lock()
	defer mu.Unlock()
	t.Logf("handled = %v", handled)
	if len(handled) == 3 && handled[1] == "C" && handled[2] == "B" {
		t.Fatalf("stale dispatch: older waiter B ran after newer C: %v", handled)
	}
}

I replayed it under DispatchDeferred only. DispatchSync with the queue strategy goes through the same prelude and queueForLock, so I expect the same order there. I did not replay that.

How it was found

A TLA+ model of the per-scope coordination for drop, queue, and debounce (TLC 1.7.4, at most 3 deliveries, one instance) checks the invariant "a waiter that was parked when a newer delivery started its turn never runs afterwards". Under queue, TLC finds the shortest path below. Under debounce the invariant holds, because every debounce delivery goes through registerPending. The model is not in the repository yet.

stateDiagram-v2
  s0: A not sent, B not sent, C not sent
  s1: A arriving, B not sent, C not sent
  s2: A handler runs, B not sent, C not sent
  s3: A handler runs, B arriving, C not sent
  s4: A handled, B arriving, C not sent
  s5: A handled, B waits for lock, C not sent
  s6: A handled, B waits for lock, C arriving
  s7: A handled, B waits for lock, C handler runs
  s8: A handled, B waits for lock, C handled
  s9: A handled, B has lock, C handled
  [*] --> s0
  s0 --> s1: A arrives
  s1 --> s2: A takes the free lock
  s2 --> s3: B arrives
  s3 --> s3: B finds the lock held
  s3 --> s4: A releases the lock
  s4 --> s5: B becomes the waiter
  s5 --> s6: C arrives
  s6 --> s7: C takes the free lock
  s7 --> s8: C releases the lock
  s8 --> s9: B gets the lock
  s9 --> s9: B commits

Fix directions (not decided)

1. In the prelude under queue, when the scope already has a pending waiter, treat the new delivery like a Lock Conflict: register it with registerPending (which supersedes the older waiter) instead of letting it take the free lock.
2. Or, when a delivery takes the lock in its prelude, supersede the scope's older pending waiter (the same displacement and observation as registerPending), so that the waiter exits.

Either way, the expected result of the test above is handled = [A C], with B observable as superseded. Both directions touch the queueMu / AcquireLock ordering that the existing queue and debounce tests guard, so the new step needs its own race check.

Generated with xum • Model: anthropic:claude-opus-5-5 • Thinking: xhigh