From 8818097cfc76dd9b0c086d9ec8f201caa91303eb Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 22:15:41 +0000 Subject: [PATCH 1/3] test(runtime): pin the admission release after a State panic in the deferred prelude Under DispatchDeferred, a State panic in the prelude runs on the net/http goroutine. The test serves the runtime behind a real httptest.Server, panics State.AcquireLock under the queue strategy and State.MarkEvent under the burst strategy, and requires that no queue turn record stays behind and that Shutdown drains before a 10s deadline. It fails on main: the panicked delivery keeps its admission slot, so Shutdown returns context deadline exceeded. --- dispatch_deferred_test.go | 77 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/dispatch_deferred_test.go b/dispatch_deferred_test.go index 346c9ff..e693021 100644 --- a/dispatch_deferred_test.go +++ b/dispatch_deferred_test.go @@ -3,9 +3,12 @@ package chat_test import ( "bytes" "context" + "encoding/json" "errors" + "log" "log/slog" "net/http" + "net/http/httptest" "strings" "sync" "testing" @@ -831,6 +834,80 @@ func TestDeferredHandlerCancelledOnLeaseLoss(t *testing.T) { } } +type markPanicState struct { + *fakeState + panicValue any +} + +func (s markPanicState) MarkEvent(context.Context, string, time.Duration) (bool, error) { + panic(s.panicValue) +} + +func TestDeferredPreludeStatePanicReleasesItsAdmission(t *testing.T) { + t.Parallel() + + const statePanic = "state panicked in the prelude" + acquirePanics := newGatedAcquireState(1) + acquirePanics.gatePanic = statePanic + close(acquirePanics.release) + cases := []struct { + name string + state chat.State + options func(*chat.RuntimeOptions) + }{ + { + name: "queue AcquireLock", + state: acquirePanics, + options: func(o *chat.RuntimeOptions) { o.Concurrency = chat.ConcurrencyQueue }, + }, + { + name: "burst MarkEvent", + state: markPanicState{fakeState: newFakeState(), panicValue: statePanic}, + options: func(o *chat.RuntimeOptions) { + o.Concurrency = chat.ConcurrencyBurst + o.BurstWindow = time.Millisecond + }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + bot := newDeferredRuntime(t, tc.state, newFakeAdapter("fake"), tc.options) + handler, err := bot.Webhook("fake") + if err != nil { + t.Fatalf("webhook: %v", err) + } + var serverLog syncBuffer + srv := httptest.NewUnstartedServer(handler) + srv.Config.ErrorLog = log.New(&serverLog, "", 0) + srv.Start() + t.Cleanup(srv.Close) + + body, err := json.Marshal(mentionEvent("A", "fake:v1:thread-1")) + if err != nil { + t.Fatalf("marshal event: %v", err) + } + resp, err := srv.Client().Post(srv.URL, "application/json", bytes.NewReader(body)) + if err == nil { + resp.Body.Close() + t.Fatalf("status = %d, want the State panic to abort the request", resp.StatusCode) + } + if !strings.Contains(serverLog.String(), statePanic) { + t.Fatalf("net/http did not recover the State panic, server log: %q", serverLog.String()) + } + if got := chat.QueueTurnCount(bot); got != 0 { + t.Errorf("turn records = %d after the prelude panicked, want 0", got) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := bot.Shutdown(ctx); err != nil { + t.Fatalf("Shutdown after the prelude panicked: %v", err) + } + }) + } +} + // syncBuffer is a goroutine-safe bytes.Buffer for capturing logs from detached // tails. type syncBuffer struct { From d21eb1378813d2d08f35390c58d9ec413dd9c633 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 22:17:58 +0000 Subject: [PATCH 2/3] fix(runtime): free the admission slot when the deferred prelude panics dispatchDeferred freed the slot only when the prelude returned. A State panic unwound past it to net/http, which recovers the panic, so the delivery kept its slot and its queue turn record, and Shutdown timed out. A deferred call now runs the release on every exit that hands no work to a tail, a panic included. The panic still propagates to net/http. The release is the sync.OnceFunc from #134, so no exit can release twice. --- runtime.go | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/runtime.go b/runtime.go index a24809a..18f64c3 100644 --- a/runtime.go +++ b/runtime.go @@ -709,15 +709,22 @@ func (c *Chat) dispatchSync(ctx context.Context, event *Event, seq uint64, entry // event, launches the detached tail and returns so the adapter can acknowledge // the platform (ack-then-work). release frees the delivery's admission slot. func (c *Chat) dispatchDeferred(ctx context.Context, event *Event, seq uint64, entry *queueEntry, release func()) error { + // A delivery that hands no work to a tail retains nothing past the + // prelude, so every such exit frees its admission slot: an error, a + // resolved event (duplicate, dropped, ignored, unrouted), or a State + // panic, which net/http recovers while the runtime keeps running. + handedOff := false + defer func() { + if !handedOff { + release() + } + }() work, resolved, err := c.prelude(ctx, event, seq, entry) if err != nil || resolved { - // The delivery retains nothing past the prelude: an errored or - // resolved (duplicate, dropped, ignored, unrouted) prelude frees its - // admission slot immediately. - release() return err } work.releaseAdmission = release + handedOff = true if work.burst { // A burst member parks in its scope's collection window instead of // owning a detached tail; its admission slot travels with it and the From dfe6d9f4a018038dea760f439d3c6005530090a5 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Thu, 29 Oct 2026 22:19:13 +0000 Subject: [PATCH 3/3] style(runtime): drop the comment that restates the dispatchDeferred release --- runtime.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/runtime.go b/runtime.go index 18f64c3..1afed1d 100644 --- a/runtime.go +++ b/runtime.go @@ -709,10 +709,6 @@ func (c *Chat) dispatchSync(ctx context.Context, event *Event, seq uint64, entry // event, launches the detached tail and returns so the adapter can acknowledge // the platform (ack-then-work). release frees the delivery's admission slot. func (c *Chat) dispatchDeferred(ctx context.Context, event *Event, seq uint64, entry *queueEntry, release func()) error { - // A delivery that hands no work to a tail retains nothing past the - // prelude, so every such exit frees its admission slot: an error, a - // resolved event (duplicate, dropped, ignored, unrouted), or a State - // panic, which net/http recovers while the runtime keeps running. handedOff := false defer func() { if !handedOff {