From 224b83f1b5f869d2bd3f034542640ad68a5ca852 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:34:35 +0000 Subject: [PATCH 1/5] test(concurrent): cover a slot freed while Shutdown cancels the tails Shutdown cancels detached tails one at a time. A waiter can win the slot that a cancelled handler frees before its own context is cancelled, and then run its handler. The new test widens that window and fails on main. --- concurrent_test.go | 68 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 68 insertions(+) diff --git a/concurrent_test.go b/concurrent_test.go index 0fa3440..3aaf79d 100644 --- a/concurrent_test.go +++ b/concurrent_test.go @@ -4,8 +4,10 @@ import ( "context" "log/slog" "net/http" + "runtime" "strings" "sync" + "sync/atomic" "testing" "time" @@ -284,3 +286,69 @@ func TestConcurrentConstructionValidation(t *testing.T) { t.Fatal("expected a negative max concurrent bound to fail") } } + +// TestConcurrentSlotFreedDuringShutdownIsAbandoned covers a slot that frees +// while Shutdown cancels the detached tails. Shutdown cancels the tails one at +// a time, so the first handler can return and free the only slot before the +// waiter's context is cancelled. The waiter must still abandon its wait. The +// child contexts of the first handler's context widen that window, because +// Shutdown cancels them after it wakes the first handler. The two tails cancel +// in random order, so the test repeats the scenario. +func TestConcurrentSlotFreedDuringShutdownIsAbandoned(t *testing.T) { + t.Parallel() + if runtime.GOMAXPROCS(0) < 2 { + t.Skip("the first handler must run while Shutdown cancels the tails") + } + + for i := range 20 { + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + bot := newConcurrentRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.MaxConcurrent = 1 + o.DetachTimeout = time.Hour + }) + + children := make([]context.CancelFunc, 0, 50000) + firstStarted := make(chan struct{}) + var waiterRan atomic.Bool + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + if ev.Event.ID != "first" { + waiterRan.Store(true) + return nil + } + for range cap(children) { + _, cancel := context.WithCancel(ctx) + children = append(children, cancel) + } + close(firstStarted) + <-ctx.Done() + return ctx.Err() + }) + + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + select { + case <-firstStarted: + case <-time.After(5 * time.Second): + t.Fatal("first handler did not start") + } + if status := postEvent(t, bot, "fake", mentionEvent("waiter", "fake:v1:thread-2")); status != http.StatusOK { + t.Fatalf("waiter status = %d", status) + } + if err := bot.Shutdown(context.Background()); err != nil { + t.Fatalf("shutdown error: %v", err) + } + for _, cancel := range children { + cancel() + } + + if waiterRan.Load() { + t.Fatalf("iteration %d: waiter ran its handler during Shutdown; logs:\n%s", i, logs.String()) + } + if !strings.Contains(logs.String(), "chat concurrent slot wait abandoned") { + t.Fatalf("iteration %d: abandoned slot wait not surfaced via observation; logs:\n%s", i, logs.String()) + } + } +} From 55cf4a280ccd1aeb53e2919d966bf7b5fc942e83 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:36:55 +0000 Subject: [PATCH 2/5] test(concurrent): shrink the slot-freed-during-Shutdown scenario Ten rounds with 20000 child contexts still fail on main in 150 of 150 runs at -cpu=2, 4, and 96, and in 20 of 20 runs with -race. The test now takes about a fifth of the time. The commit also removes the 40 ms sleep from TestConcurrentSlotWaiterDrainedByShutdown. --- concurrent_test.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/concurrent_test.go b/concurrent_test.go index 3aaf79d..73f368e 100644 --- a/concurrent_test.go +++ b/concurrent_test.go @@ -206,7 +206,6 @@ func TestConcurrentSlotWaiterDrainedByShutdown(t *testing.T) { if status := postEvent(t, bot, "fake", mentionEvent("waiter", "fake:v1:thread-2")); status != http.StatusOK { t.Fatalf("waiter status = %d", status) } - time.Sleep(40 * time.Millisecond) done := make(chan error, 1) go func() { done <- bot.Shutdown(context.Background()) }() @@ -300,7 +299,7 @@ func TestConcurrentSlotFreedDuringShutdownIsAbandoned(t *testing.T) { t.Skip("the first handler must run while Shutdown cancels the tails") } - for i := range 20 { + for i := range 10 { state := newFakeState() adapter := newFakeAdapter("fake") var logs syncBuffer @@ -309,7 +308,7 @@ func TestConcurrentSlotFreedDuringShutdownIsAbandoned(t *testing.T) { o.DetachTimeout = time.Hour }) - children := make([]context.CancelFunc, 0, 50000) + children := make([]context.CancelFunc, 0, 20000) firstStarted := make(chan struct{}) var waiterRan atomic.Bool bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { From 4742d79d31d2e53551512fd657bef5e458f0dcbe Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:37:02 +0000 Subject: [PATCH 3/5] fix(runtime): abandon a concurrency slot won after Shutdown cancelled the tails Shutdown cancels detached tails one at a time. A tail whose handler returns first frees its slot, and a parked waiter can win that slot before its own context is cancelled. The waiter then ran its handler during Shutdown instead of abandoning the wait. A detached tail that wins a slot after the base context is cancelled now releases the slot and takes the abandoned path, with the same log and outcome. Sync dispatch is unchanged. --- runtime.go | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/runtime.go b/runtime.go index 1afed1d..424d32d 100644 --- a/runtime.go +++ b/runtime.go @@ -677,7 +677,7 @@ func (c *Chat) dispatchSync(ctx context.Context, event *Event, seq uint64, entry return err } if work.noLock { - if !c.acquireConcurrencySlot(ctx, event) { + if !c.acquireConcurrencySlot(ctx, event, false) { c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) return nil } @@ -1086,7 +1086,7 @@ func (c *Chat) startDetachedTail(work preludeWork) { defer tailCancel() if work.noLock { - if !c.acquireConcurrencySlot(tailCtx, work.event) { + if !c.acquireConcurrencySlot(tailCtx, work.event, true) { c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) return } @@ -1203,16 +1203,24 @@ func (c *Chat) waitDebounceQuietPeriod(ctx context.Context, work preludeWork) bo } // acquireConcurrencySlot blocks until a MaxConcurrent slot frees, bounded by -// ctx. It reports false when the wait was abandoned. -func (c *Chat) acquireConcurrencySlot(ctx context.Context, event *Event) bool { +// ctx. It reports false when the wait was abandoned. A detached tail also +// abandons a slot it wins after Shutdown cancelled the base context: Shutdown +// cancels the tails one at a time, so a slot freed by a cancelled tail can +// reach a waiter whose own context is not cancelled yet. +func (c *Chat) acquireConcurrencySlot(ctx context.Context, event *Event, detached bool) bool { assert(c.concurrencySlots != nil, "concurrency slots are only used under the concurrent strategy") + var err error select { case c.concurrencySlots <- struct{}{}: - return true + if err = c.baseCtx.Err(); !detached || err == nil { + return true + } + c.releaseConcurrencySlot() case <-ctx.Done(): - c.logger.Info("chat concurrent slot wait abandoned", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID, "error", ctx.Err()) - return false + err = ctx.Err() } + c.logger.Info("chat concurrent slot wait abandoned", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID, "error", err) + return false } func (c *Chat) releaseConcurrencySlot() { From 29a5e9f5c4ea71e1efec0cedd8d2d1914eef060a Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:38:51 +0000 Subject: [PATCH 4/5] refactor(runtime): move the shutdown slot check into acquireDetachedConcurrencySlot Replace the detached flag with a wrapper for detached tails, so the sync path keeps the original function unchanged. --- concurrent_test.go | 12 +++++------- runtime.go | 35 ++++++++++++++++++++++------------- 2 files changed, 27 insertions(+), 20 deletions(-) diff --git a/concurrent_test.go b/concurrent_test.go index 73f368e..0b06db1 100644 --- a/concurrent_test.go +++ b/concurrent_test.go @@ -286,13 +286,9 @@ func TestConcurrentConstructionValidation(t *testing.T) { } } -// TestConcurrentSlotFreedDuringShutdownIsAbandoned covers a slot that frees -// while Shutdown cancels the detached tails. Shutdown cancels the tails one at -// a time, so the first handler can return and free the only slot before the -// waiter's context is cancelled. The waiter must still abandon its wait. The -// child contexts of the first handler's context widen that window, because -// Shutdown cancels them after it wakes the first handler. The two tails cancel -// in random order, so the test repeats the scenario. +// TestConcurrentSlotFreedDuringShutdownIsAbandoned pins that a waiter whose +// slot frees while Shutdown cancels the tails abandons the wait. The tails +// cancel in random order, so the test repeats the scenario. func TestConcurrentSlotFreedDuringShutdownIsAbandoned(t *testing.T) { t.Parallel() if runtime.GOMAXPROCS(0) < 2 { @@ -316,6 +312,8 @@ func TestConcurrentSlotFreedDuringShutdownIsAbandoned(t *testing.T) { waiterRan.Store(true) return nil } + // Shutdown cancels these after it wakes this handler, so the + // waiter's context stays live while this handler frees the slot. for range cap(children) { _, cancel := context.WithCancel(ctx) children = append(children, cancel) diff --git a/runtime.go b/runtime.go index 424d32d..7fd1ccc 100644 --- a/runtime.go +++ b/runtime.go @@ -677,7 +677,7 @@ func (c *Chat) dispatchSync(ctx context.Context, event *Event, seq uint64, entry return err } if work.noLock { - if !c.acquireConcurrencySlot(ctx, event, false) { + if !c.acquireConcurrencySlot(ctx, event) { c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) return nil } @@ -1086,7 +1086,7 @@ func (c *Chat) startDetachedTail(work preludeWork) { defer tailCancel() if work.noLock { - if !c.acquireConcurrencySlot(tailCtx, work.event, true) { + if !c.acquireDetachedConcurrencySlot(tailCtx, work.event) { c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) return } @@ -1203,22 +1203,31 @@ func (c *Chat) waitDebounceQuietPeriod(ctx context.Context, work preludeWork) bo } // acquireConcurrencySlot blocks until a MaxConcurrent slot frees, bounded by -// ctx. It reports false when the wait was abandoned. A detached tail also -// abandons a slot it wins after Shutdown cancelled the base context: Shutdown -// cancels the tails one at a time, so a slot freed by a cancelled tail can -// reach a waiter whose own context is not cancelled yet. -func (c *Chat) acquireConcurrencySlot(ctx context.Context, event *Event, detached bool) bool { +// ctx. It reports false when the wait was abandoned. +func (c *Chat) acquireConcurrencySlot(ctx context.Context, event *Event) bool { assert(c.concurrencySlots != nil, "concurrency slots are only used under the concurrent strategy") - var err error select { case c.concurrencySlots <- struct{}{}: - if err = c.baseCtx.Err(); !detached || err == nil { - return true - } - c.releaseConcurrencySlot() + return true case <-ctx.Done(): - err = ctx.Err() + c.logger.Info("chat concurrent slot wait abandoned", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID, "error", ctx.Err()) + return false + } +} + +// acquireDetachedConcurrencySlot is acquireConcurrencySlot for a detached +// tail. Shutdown cancels the tails one at a time, so a tail can win a slot +// that a cancelled tail freed before its own context is cancelled. Such a +// tail releases the slot and abandons the wait. +func (c *Chat) acquireDetachedConcurrencySlot(ctx context.Context, event *Event) bool { + if !c.acquireConcurrencySlot(ctx, event) { + return false + } + err := c.baseCtx.Err() + if err == nil { + return true } + c.releaseConcurrencySlot() c.logger.Info("chat concurrent slot wait abandoned", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID, "error", err) return false } From 911ba61eafd33914f230740be72e70c1a6dbd8e3 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:51:11 +0000 Subject: [PATCH 5/5] test(concurrent): restore the 40 ms sleep so the old test parks the waiter first Without the sleep the waiter reaches the slot after Shutdown in almost every run, so TestConcurrentSlotWaiterDrainedByShutdown rarely checks a parked waiter. With the runtime fix the parked order no longer flakes. --- concurrent_test.go | 1 + 1 file changed, 1 insertion(+) diff --git a/concurrent_test.go b/concurrent_test.go index 0b06db1..f6c8dc1 100644 --- a/concurrent_test.go +++ b/concurrent_test.go @@ -206,6 +206,7 @@ func TestConcurrentSlotWaiterDrainedByShutdown(t *testing.T) { if status := postEvent(t, bot, "fake", mentionEvent("waiter", "fake:v1:thread-2")); status != http.StatusOK { t.Fatalf("waiter status = %d", status) } + time.Sleep(40 * time.Millisecond) done := make(chan error, 1) go func() { done <- bot.Shutdown(context.Background()) }()