From d11de761392bf9faa40345b523cc4ebe2474343d Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 11:15:36 +0000 Subject: [PATCH 1/3] test(runtime): reproduce queue barging past a parked waiter (#121) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A newer delivery that finds the Thread Lock free takes it in its prelude. The older delivery that was parked for the lock still runs afterwards, so the handler sees A, C, B. The test holds B's first lock poll with a gated State, lets C run, then releases B. It fails under both dispatch modes. --- _Generated with [`xum`](https://github.com/coder/xum) • Model: `anthropic:claude-opus-5-5` • Thinking: `xhigh`_ --- queue_edge_test.go | 190 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 190 insertions(+) diff --git a/queue_edge_test.go b/queue_edge_test.go index 0f39e07..3555614 100644 --- a/queue_edge_test.go +++ b/queue_edge_test.go @@ -459,3 +459,193 @@ func TestQueueDeferredWaiterDrainedAndCancelledByShutdown(t *testing.T) { t.Fatalf("handler ran despite the lock never freeing, calls = %d", calls) } } + +type gatedAcquireState struct { + *fakeState + gateCall int + reached chan struct{} + release chan struct{} + + 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 + } + 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() { _ = bot.Shutdown(context.Background()) }) + 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()) + } + }) + } +} From f4568acd90c9f6e01f4d4ff4e99c3775f7a723b1 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 11:43:44 +0000 Subject: [PATCH 2/3] fix(runtime): supersede an older queue waiter when a newer event takes the free lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under ConcurrencyQueue, a routed event that takes the free Thread Lock in its prelude now supersedes the scope's pending waiter when that waiter was admitted earlier. The waiter stops at its next poll or at its re-check after AcquireLock, ends as ignored, and the log records superseded_by. Before, the waiter still ran after the newer event, so the bot answered an older message last. The call sits after acceptEvent, so an event that resolves in its prelude (a duplicate, an ignored or unrouted event) supersedes no one. A waiter admitted after the lock holder stays and runs next. Two guard tests pin both rules. registerPending and the new path share displaceLocked, which writes the skip record. Closes #121 --- _Generated with [`xum`](https://github.com/coder/xum) • Model: `anthropic:claude-opus-5-5` • Thinking: `xhigh`_ --- queue_edge_test.go | 78 ++++++++++++++++++++++++++++++++++++++++++++++ runtime.go | 34 ++++++++++++++++---- 2 files changed, 106 insertions(+), 6 deletions(-) diff --git a/queue_edge_test.go b/queue_edge_test.go index 3555614..0975181 100644 --- a/queue_edge_test.go +++ b/queue_edge_test.go @@ -649,3 +649,81 @@ func TestQueueNewerDeliveryThatTakesTheFreeLockSupersedesParkedWaiter(t *testing }) } } + +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) + } + }) + } +} diff --git a/runtime.go b/runtime.go index 70f728f..fba2665 100644 --- a/runtime.go +++ b/runtime.go @@ -18,7 +18,8 @@ const ( // ConcurrencyDrop acknowledges and drops a Lock Conflict. This is the default. ConcurrencyDrop ConcurrencyStrategy = iota // ConcurrencyQueue waits for the in-flight handler, then dispatches the most - // recent superseded event for the scope. + // recent superseded event for the scope. A newer event that finds the lock + // free runs at once and supersedes the waiting event. ConcurrencyQueue // ConcurrencyDebounce coalesces rapid follow-ups: each new routed event for a // scope supersedes the previous waiter and only the final event in a @@ -986,10 +987,12 @@ func (c *Chat) routedWork( } switch c.options.Concurrency { - case ConcurrencyDrop, ConcurrencyQueue: + case ConcurrencyDrop: + assert(!conflicted, "drop resolves a Lock Conflict at the conflict site") + return work, false, nil + + case ConcurrencyQueue: if conflicted { - // Only queue reaches routing with a conflict (drop resolves at the - // conflict site): the event waits as its scope's pending waiter. work.needsLock = true work.waitLabel = "queue" displaced, registered := c.registerPending(scope, event, seq, work.waitLabel) @@ -997,7 +1000,9 @@ func (c *Chat) routedWork( return resolveStale(work.waitLabel) } work.displaced = displaced + return work, false, nil } + c.supersedeOlderWaiter(scope, event, seq) return work, false, nil case ConcurrencyDebounce: @@ -1309,14 +1314,31 @@ func (c *Chat) registerPending(scope string, event *Event, seq uint64, label str if prev.seq > seq { return nil, false } - c.logger.Info("chat "+label+" superseded", "adapter", prev.event.Adapter, "event_id", prev.event.ID, "thread_id", prev.event.ThreadID, "superseded_by", event.ID) - close(prev.displaced) + c.displaceLocked(prev, event, label) } waiter := &pendingWaiter{event: event, seq: seq, displaced: make(chan struct{})} c.pending[scope] = waiter return waiter.displaced, true } +func (c *Chat) displaceLocked(prev *pendingWaiter, by *Event, label string) { + c.logger.Info("chat "+label+" superseded", "adapter", prev.event.Adapter, + "event_id", prev.event.ID, "thread_id", prev.event.ThreadID, "superseded_by", by.ID) + close(prev.displaced) +} + +func (c *Chat) supersedeOlderWaiter(scope string, event *Event, seq uint64) { + c.queueMu.Lock() + defer c.queueMu.Unlock() + prev := c.pending[scope] + if prev == nil || prev.seq > seq { + return + } + assert(prev.seq != seq, "a delivery that holds its prelude lease is never its scope's pending waiter") + c.displaceLocked(prev, event, "queue") + delete(c.pending, scope) +} + // isPending reports whether event is still its scope's most-recent pending // waiter. func (c *Chat) isPending(scope string, event *Event) bool { From 8d4b4d9eeb2ccc16218737276d82acd8aa227e85 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 12:44:46 +0000 Subject: [PATCH 3/3] test(runtime): bound the gated queue tests' cleanup shutdown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A check that fails before the test opens a gate left B's lock poll or A's handler blocked, so the unbounded Shutdown in cleanup waited forever. The test hit the binary timeout and its own failure message never printed. A 5s shutdown context lets the failure report. --- _Generated with [`xum`](https://github.com/coder/xum) • Model: `anthropic:claude-opus-5-5` • Thinking: `xhigh`_ --- queue_edge_test.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/queue_edge_test.go b/queue_edge_test.go index 0975181..bdee078 100644 --- a/queue_edge_test.go +++ b/queue_edge_test.go @@ -550,7 +550,13 @@ func newGatedQueueRun(t *testing.T, state chat.State, mode chat.DispatchMode, bl if err != nil { t.Fatalf("new runtime: %v", err) } - t.Cleanup(func() { _ = bot.Shutdown(context.Background()) }) + 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()