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 { diff --git a/runtime.go b/runtime.go index a24809a..1afed1d 100644 --- a/runtime.go +++ b/runtime.go @@ -709,15 +709,18 @@ 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 { + 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