From 2d35386b433b3824214729c16520049d37f6fa73 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 20:53:55 +0000 Subject: [PATCH 1/8] test(runtime): pin the queue turn record's bounded retention TestQueueIdleScopeRetainsNoTurnRecord reads queueTurns through the new QueueTurnCount hook. It fails with the delete in leaveQueueScope removed. --- export_test.go | 9 +++++++++ queue_hardening_test.go | 43 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+) create mode 100644 queue_hardening_test.go diff --git a/export_test.go b/export_test.go index c044582..10adc6e 100644 --- a/export_test.go +++ b/export_test.go @@ -17,3 +17,12 @@ 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) +} diff --git a/queue_hardening_test.go b/queue_hardening_test.go new file mode 100644 index 0000000..d66210a --- /dev/null +++ b/queue_hardening_test.go @@ -0,0 +1,43 @@ +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 bFirstPoll = 3 // after the prelude acquires of A and B + 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") + }) + } +} From 15d9692f8b0ffdf3cdbf756e08585541ec6757e6 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 20:54:31 +0000 Subject: [PATCH 2/8] test(runtime): show a deferred queue release called twice leaves its scope twice dispatch now builds the release in deferredRelease, with no change in behavior, so the QueueDeferredRelease hook gets the same release. TestQueueDeferredReleaseCalledTwiceLeavesItsScopeOnce fails: the second call drops the scope's turn record while another delivery is in flight. --- export_test.go | 10 ++++++++++ queue_hardening_test.go | 20 ++++++++++++++++++++ runtime.go | 14 ++++++++++---- 3 files changed, 40 insertions(+), 4 deletions(-) diff --git a/export_test.go b/export_test.go index 10adc6e..dac5c01 100644 --- a/export_test.go +++ b/export_test.go @@ -26,3 +26,13 @@ func QueueTurnCount(c *Chat) int { defer c.queueMu.Unlock() return len(c.queueTurns) } + +// QueueDeferredRelease counts one queue delivery in scope's turn record, as +// the prelude does, and returns the release dispatch gives that delivery's +// deferred path. +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() {}) +} diff --git a/queue_hardening_test.go b/queue_hardening_test.go index d66210a..489d992 100644 --- a/queue_hardening_test.go +++ b/queue_hardening_test.go @@ -41,3 +41,23 @@ func TestQueueIdleScopeRetainsNoTurnRecord(t *testing.T) { }) } } + +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) + } +} diff --git a/runtime.go b/runtime.go index 2ab7b13..e72b35d 100644 --- a/runtime.go +++ b/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,15 @@ func (c *Chat) dispatch(ctx context.Context, event *Event) error { return c.dispatchSync(ctx, event, seq, entry) } +// deferredRelease returns the release a deferred delivery calls at its outcome: +// it leaves the delivery's queue scope and frees its admission slot. +func (c *Chat) deferredRelease(entry *queueEntry, release func()) func() { + return 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 From a3a3fdefd42eb555934f8f3d3ae0cf9b651213bc Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 20:54:58 +0000 Subject: [PATCH 3/8] fix(runtime): make the deferred queue release idempotent The release that dispatch gives a deferred delivery wraps the idempotent release from admit, but a second call left the queue scope again. It could drop the turn record while another delivery was in flight, or trip the leaveQueueScope assert. sync.OnceFunc makes it idempotent like admit's. --- runtime.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/runtime.go b/runtime.go index e72b35d..9c1195f 100644 --- a/runtime.go +++ b/runtime.go @@ -660,12 +660,13 @@ func (c *Chat) dispatch(ctx context.Context, event *Event) error { } // deferredRelease returns the release a deferred delivery calls at its outcome: -// it leaves the delivery's queue scope and frees its admission slot. +// it leaves the delivery's queue scope and frees its admission slot. Like the +// release admit returns, it is idempotent. func (c *Chat) deferredRelease(entry *queueEntry, release func()) func() { - return func() { + return sync.OnceFunc(func() { c.leaveQueueScope(entry) release() - } + }) } // dispatchSync runs the prelude and the routed handler inline under the request From 53c356c32ac978adb0c215301a96cfd4520fc9d7 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 20:55:37 +0000 Subject: [PATCH 4/8] test(runtime): show a State panic in a queue waiter's poll leaves a stale waiter Under DispatchSync, a panic from State.AcquireLock while a queue waiter polls skips queueForLock's clearPending. The waiter's leaveQueueScope then trips its assert in place of the State's panic, and so does a later duplicate for the scope. TestQueueStatePanicWhileWaiterPollsLeavesNoStaleWaiter recovers each panic as net/http does and fails on both. --- queue_edge_test.go | 10 +++++-- queue_hardening_test.go | 63 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 3 deletions(-) diff --git a/queue_edge_test.go b/queue_edge_test.go index 84422b1..a11ae77 100644 --- a/queue_edge_test.go +++ b/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) } diff --git a/queue_hardening_test.go b/queue_hardening_test.go index 489d992..77c5493 100644 --- a/queue_hardening_test.go +++ b/queue_hardening_test.go @@ -61,3 +61,66 @@ func TestQueueDeferredReleaseCalledTwiceLeavesItsScopeOnce(t *testing.T) { t.Fatalf("turn records = %d after both deliveries released, want 0", got) } } + +type recoveredPost struct { + postEventResult + panicked any +} + +// postRecovering posts id to the queue thread and recovers a panic from +// dispatch, as net/http does for a handler. +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 ( + bFirstPoll = 3 // after the prelude acquires of A and B + 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 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) + } +} From 569d6a90c1fe26235880eb34846214a3968e57af Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 20:56:12 +0000 Subject: [PATCH 5/8] fix(runtime): clear a queue waiter's pending entry when its poll panics queueForLock cleared the waiter's entry only on the abandoned and failed outcomes, so a panic from State.AcquireLock left it behind. One deferred clearPending now covers every exit. On the other exits the entry is already gone or belongs to a newer waiter, and clearPending leaves it. --- runtime.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/runtime.go b/runtime.go index 9c1195f..d4a061b 100644 --- a/runtime.go +++ b/runtime.go @@ -1450,8 +1450,10 @@ func (c *Chat) isPending(scope string, event *Event) bool { // in-flight handler releases. It returns acquireSuperseded if a newer follow-up // displaced this waiter, acquireAbandoned if ctx ended, acquireFailed on an // AcquireLock error, or the lease (acquireHeld) once acquired. label names the -// coordination mode in the wait observations. +// coordination mode in the wait observations. It clears the waiter's pending +// entry on every exit, including a panic from State. 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) } @@ -1472,10 +1474,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 From 984f2ee4d44393ce0b1fc8067ca66171ed317dc5 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 20:56:52 +0000 Subject: [PATCH 6/8] docs(runtime): state that Adapter.ValidateThreadID must not block ConcurrencyQueue's GoDoc relies on this rule, but only the Observer interface stated it. --- types.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/types.go b/types.go index ee81aea..d9cce03 100644 --- a/types.go +++ b/types.go @@ -133,6 +133,8 @@ 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, and ConcurrencyQueue relies on that. ValidateThreadID(ThreadID) (ThreadRef, error) PostMessage(context.Context, ThreadRef, PostableMessage) (*SentMessage, error) BotActor() Actor From bfe6f309aec24306e08379facb930d4133e9f749 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 21:00:42 +0000 Subject: [PATCH 7/8] chore(runtime): trim comments in the queue turn record hardening The poll-call constants follow the iota pattern of queue_edge_test.go, and comments the code already states are gone. The ValidateThreadID GoDoc now names what a block there costs under ConcurrencyQueue. --- export_test.go | 7 ++++--- queue_hardening_test.go | 14 +++++++++----- runtime.go | 6 +----- types.go | 3 ++- 4 files changed, 16 insertions(+), 14 deletions(-) diff --git a/export_test.go b/export_test.go index dac5c01..6b80159 100644 --- a/export_test.go +++ b/export_test.go @@ -27,9 +27,10 @@ func QueueTurnCount(c *Chat) int { return len(c.queueTurns) } -// QueueDeferredRelease counts one queue delivery in scope's turn record, as -// the prelude does, and returns the release dispatch gives that delivery's -// deferred path. +// 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{} diff --git a/queue_hardening_test.go b/queue_hardening_test.go index 77c5493..406bb18 100644 --- a/queue_hardening_test.go +++ b/queue_hardening_test.go @@ -15,7 +15,11 @@ func TestQueueIdleScopeRetainsNoTurnRecord(t *testing.T) { t.Run(tc.name, func(t *testing.T) { t.Parallel() - const bFirstPoll = 3 // after the prelude acquires of A and B + const ( + aPreludeAcquire = 1 + iota + bPreludeAcquire + bFirstPoll + ) state := newGatedAcquireState(0) run := newGatedQueueRun(t, state, tc.mode, "A") @@ -67,8 +71,6 @@ type recoveredPost struct { panicked any } -// postRecovering posts id to the queue thread and recovers a panic from -// dispatch, as net/http does for a handler. func (r *gatedQueueRun) postRecovering(id string) <-chan recoveredPost { done := make(chan recoveredPost, 1) go func() { @@ -97,9 +99,11 @@ func TestQueueStatePanicWhileWaiterPollsLeavesNoStaleWaiter(t *testing.T) { t.Parallel() const ( - bFirstPoll = 3 // after the prelude acquires of A and B - statePanic = "state AcquireLock panicked" + aPreludeAcquire = 1 + iota + bPreludeAcquire + bFirstPoll ) + const statePanic = "state AcquireLock panicked" state := newGatedAcquireState(bFirstPoll) state.gatePanic = statePanic run := newGatedQueueRun(t, state, chat.DispatchSync, "A") diff --git a/runtime.go b/runtime.go index d4a061b..a24809a 100644 --- a/runtime.go +++ b/runtime.go @@ -659,9 +659,6 @@ func (c *Chat) dispatch(ctx context.Context, event *Event) error { return c.dispatchSync(ctx, event, seq, entry) } -// deferredRelease returns the release a deferred delivery calls at its outcome: -// it leaves the delivery's queue scope and frees its admission slot. Like the -// release admit returns, it is idempotent. func (c *Chat) deferredRelease(entry *queueEntry, release func()) func() { return sync.OnceFunc(func() { c.leaveQueueScope(entry) @@ -1450,8 +1447,7 @@ func (c *Chat) isPending(scope string, event *Event) bool { // in-flight handler releases. It returns acquireSuperseded if a newer follow-up // displaced this waiter, acquireAbandoned if ctx ended, acquireFailed on an // AcquireLock error, or the lease (acquireHeld) once acquired. label names the -// coordination mode in the wait observations. It clears the waiter's pending -// entry on every exit, including a panic from State. +// 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 { diff --git a/types.go b/types.go index d9cce03..5035b06 100644 --- a/types.go +++ b/types.go @@ -134,7 +134,8 @@ type Adapter interface { Shutdown(context.Context) error Webhook(DispatchFunc) http.Handler // ValidateThreadID must not block. Dispatch calls it before the lock scope - // is known, and ConcurrencyQueue relies on that. + // 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 From e64f797958536e5eb415b23d19f9440f832add29 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 21:18:44 +0000 Subject: [PATCH 8/8] test(runtime): assert a panicking queue waiter leaves no pending entry TestQueueStatePanicWhileWaiterPollsLeavesNoStaleWaiter checked only the turn record count, so a fix that deleted the leaveQueueScope assert instead of clearing the waiter still passed while the stale waiter stayed in c.pending. Add QueuePendingCount and assert it is 0 after B's panic. --- export_test.go | 9 +++++++++ queue_hardening_test.go | 3 +++ 2 files changed, 12 insertions(+) diff --git a/export_test.go b/export_test.go index 6b80159..93211af 100644 --- a/export_test.go +++ b/export_test.go @@ -27,6 +27,15 @@ func QueueTurnCount(c *Chat) int { 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 diff --git a/queue_hardening_test.go b/queue_hardening_test.go index 406bb18..c17f838 100644 --- a/queue_hardening_test.go +++ b/queue_hardening_test.go @@ -121,6 +121,9 @@ func TestQueueStatePanicWhileWaiterPollsLeavesNoStaleWaiter(t *testing.T) { 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) }