diff --git a/export_test.go b/export_test.go index c044582..93211af 100644 --- a/export_test.go +++ b/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() {}) +} 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 new file mode 100644 index 0000000..c17f838 --- /dev/null +++ b/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) + } +} diff --git a/runtime.go b/runtime.go index 2ab7b13..a24809a 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,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 diff --git a/types.go b/types.go index ee81aea..5035b06 100644 --- a/types.go +++ b/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