fix(runtime): harden the queue turn record against a double release and a State panic

export_test.go

@@ -17,3 +17,32 @@ func BurstScopeCount(c *Chat) int {
 	defer c.burstMu.Unlock()
 	return len(c.burstScopes)
 }
+
+// QueueTurnCount reports the number of live queue turn records, for hardening
+// tests proving an idle scope retains no record (ADR 0015 bounded retention).
+func QueueTurnCount(c *Chat) int {
+	assert(c != nil, "QueueTurnCount called on nil runtime")
+	c.queueMu.Lock()
+	defer c.queueMu.Unlock()
+	return len(c.queueTurns)
+}
+
+// QueuePendingCount reports the number of live pending queue waiters, for
+// hardening tests proving a waiter that ended leaves no stale entry.
+func QueuePendingCount(c *Chat) int {
+	assert(c != nil, "QueuePendingCount called on nil runtime")
+	c.queueMu.Lock()
+	defer c.queueMu.Unlock()
+	return len(c.pending)
+}
+
+// QueueDeferredRelease enters scope's queue turn record as the prelude does
+// and returns the deferred release dispatch builds for that delivery, with a
+// no-op admission release, for hardening tests proving a repeated release
+// leaves the scope once.
+func QueueDeferredRelease(c *Chat, scope string) func() {
+	assert(c != nil, "QueueDeferredRelease called on nil runtime")
+	entry := &queueEntry{}
+	c.enterQueueScope(scope, entry)
+	return c.deferredRelease(entry, func() {})
+}

queue_edge_test.go

@@ -462,9 +462,10 @@ func TestQueueDeferredWaiterDrainedAndCancelledByShutdown(t *testing.T) {
 
 type gatedAcquireState struct {
 	*fakeState
-	gateCall int
-	reached  chan struct{}
-	release  chan struct{}
+	gateCall  int
+	reached   chan struct{}
+	release   chan struct{}
+	gatePanic any
 
 	mu    sync.Mutex
 	calls int
@@ -487,6 +488,9 @@ func (s *gatedAcquireState) AcquireLock(ctx context.Context, key string, ttl tim
 	if gated {
 		close(s.reached)
 		<-s.release
+		if s.gatePanic != nil {
+			panic(s.gatePanic)
+		}
 	}
 	return s.fakeState.AcquireLock(ctx, key, ttl)
 }

queue_hardening_test.go

@@ -0,0 +1,133 @@
+package chat_test
+
+import (
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/coder/chat"
+)
+
+func TestQueueIdleScopeRetainsNoTurnRecord(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(0)
+			run := newGatedQueueRun(t, state, tc.mode, "A")
+
+			run.post("A")
+			waitClosed(t, run.blockStarted, "A's handler did not start")
+			run.post("B")
+			eventually(t, 2*time.Second, func() bool { return state.acquireCalls() >= bFirstPoll },
+				"B did not park and poll for the lock")
+			if res := postEventResultFor(run.bot, "fake", mentionEvent("D", "fake:v1:thread-2")); res.err != nil || res.status != http.StatusOK {
+				t.Fatalf("D: status=%d err=%v", res.status, res.err)
+			}
+			eventually(t, 2*time.Second, func() bool { return chat.QueueTurnCount(run.bot) == 1 },
+				"D's scope kept its turn record after D finished")
+
+			close(run.releaseBlock)
+			eventually(t, 4*time.Second, func() bool { return len(run.handledIDs()) == 3 },
+				"B did not run after A")
+			if res := <-run.post("A"); res.err != nil || res.status != http.StatusOK {
+				t.Fatalf("redelivery of A: status=%d err=%v", res.status, res.err)
+			}
+			eventually(t, 2*time.Second, func() bool { return chat.QueueTurnCount(run.bot) == 0 },
+				"an idle scope kept its turn record")
+		})
+	}
+}
+
+func TestQueueDeferredReleaseCalledTwiceLeavesItsScopeOnce(t *testing.T) {
+	t.Parallel()
+
+	var logs syncBuffer
+	bot := newQueueRuntime(t, newFakeState(), newFakeAdapter("fake"), &logs)
+	first := chat.QueueDeferredRelease(bot, "scope-1")
+	second := chat.QueueDeferredRelease(bot, "scope-1")
+
+	first()
+	first()
+	if got := chat.QueueTurnCount(bot); got != 1 {
+		t.Fatalf("turn records = %d after one of two deliveries released twice, want 1", got)
+	}
+	second()
+	second()
+	if got := chat.QueueTurnCount(bot); got != 0 {
+		t.Fatalf("turn records = %d after both deliveries released, want 0", got)
+	}
+}
+
+type recoveredPost struct {
+	postEventResult
+	panicked any
+}
+
+func (r *gatedQueueRun) postRecovering(id string) <-chan recoveredPost {
+	done := make(chan recoveredPost, 1)
+	go func() {
+		var res recoveredPost
+		defer func() {
+			res.panicked = recover()
+			done <- res
+		}()
+		res.postEventResult = postEventResultFor(r.bot, "fake", mentionEvent(id, queueThread))
+	}()
+	return done
+}
+
+func receivePost(t *testing.T, done <-chan recoveredPost, name string) recoveredPost {
+	t.Helper()
+	select {
+	case res := <-done:
+		return res
+	case <-time.After(2 * time.Second):
+		t.Fatalf("%s did not return", name)
+		return recoveredPost{}
+	}
+}
+
+func TestQueueStatePanicWhileWaiterPollsLeavesNoStaleWaiter(t *testing.T) {
+	t.Parallel()
+
+	const (
+		aPreludeAcquire = 1 + iota
+		bPreludeAcquire
+		bFirstPoll
+	)
+	const statePanic = "state AcquireLock panicked"
+	state := newGatedAcquireState(bFirstPoll)
+	state.gatePanic = statePanic
+	run := newGatedQueueRun(t, state, chat.DispatchSync, "A")
+
+	a := run.postRecovering("A")
+	waitClosed(t, run.blockStarted, "A's handler did not start")
+	b := run.postRecovering("B")
+	waitClosed(t, state.reached, "B did not poll for the lock")
+	close(run.releaseBlock)
+	if res := receivePost(t, a, "A"); res.panicked != nil || res.status != http.StatusOK {
+		t.Fatalf("A: status=%d panic=%v", res.status, res.panicked)
+	}
+
+	close(state.release)
+	if res := receivePost(t, b, "B"); res.panicked != statePanic {
+		t.Errorf("B's dispatch panicked with %v, want the State's own panic", res.panicked)
+	}
+	if got := chat.QueuePendingCount(run.bot); got != 0 {
+		t.Fatalf("pending waiters = %d after B's poll panicked, want 0", got)
+	}
+	if res := receivePost(t, run.postRecovering("A"), "the redelivery of A"); res.panicked != nil || res.status != http.StatusOK {
+		t.Fatalf("the redelivery of A after the State panic: status=%d panic=%v", res.status, res.panicked)
+	}
+	if got := chat.QueueTurnCount(run.bot); got != 0 {
+		t.Fatalf("turn records = %d after every delivery finished, want 0", got)
+	}
+}

runtime.go

@@ -650,10 +650,7 @@ func (c *Chat) dispatch(ctx context.Context, event *Event) error {
 	// dispatchSync returns.
 	entry := &queueEntry{}
 	if c.options.Dispatch == DispatchDeferred {
-		return c.dispatchDeferred(ctx, event, seq, entry, func() {
-			c.leaveQueueScope(entry)
-			release()
-		})
+		return c.dispatchDeferred(ctx, event, seq, entry, c.deferredRelease(entry, release))
 	}
 	// A sync delivery retains no work for Shutdown to drain: its caller (an
 	// HTTP server or a Listener) owns its lifetime.
@@ -662,6 +659,13 @@ func (c *Chat) dispatch(ctx context.Context, event *Event) error {
 	return c.dispatchSync(ctx, event, seq, entry)
 }
 
+func (c *Chat) deferredRelease(entry *queueEntry, release func()) func() {
+	return sync.OnceFunc(func() {
+		c.leaveQueueScope(entry)
+		release()
+	})
+}
+
 // dispatchSync runs the prelude and the routed handler inline under the request
 // context, releasing the Thread Lock when the handler returns. Under the queue
 // strategy a Lock Conflict waits inline (bounded by ctx) for the in-flight
@@ -1445,6 +1449,7 @@ func (c *Chat) isPending(scope string, event *Event) bool {
 // AcquireLock error, or the lease (acquireHeld) once acquired. label names the
 // coordination mode in the wait observations.
 func (c *Chat) queueForLock(ctx context.Context, scope string, event *Event, label string) (LockLease, acquireOutcome) {
+	defer c.clearPending(scope, event)
 	superseded := func() bool {
 		return !c.isPending(scope, event)
 	}
@@ -1465,10 +1470,8 @@ func (c *Chat) queueForLock(ctx context.Context, scope string, event *Event, lab
 		// just stops.
 		c.logger.Debug("chat "+label+" waiter superseded", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID)
 	case acquireAbandoned:
-		c.clearPending(scope, event)
 		c.logger.Info("chat "+label+" wait abandoned", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID, "error", ctx.Err())
 	case acquireFailed:
-		c.clearPending(scope, event)
 		c.logger.Error("chat "+label+" acquire lock failed", "error", err, "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID)
 	}
 	return lease, outcome

types.go

@@ -133,6 +133,9 @@ type Adapter interface {
 	Init(context.Context) error
 	Shutdown(context.Context) error
 	Webhook(DispatchFunc) http.Handler
+	// ValidateThreadID must not block. Dispatch calls it before the lock scope
+	// is known, so under [ConcurrencyQueue] a delivery delayed here can run
+	// after newer deliveries for its scope have finished.
 	ValidateThreadID(ThreadID) (ThreadRef, error)
 	PostMessage(context.Context, ThreadRef, PostableMessage) (*SentMessage, error)
 	BotActor() Actor