fix(runtime): free the admission slot when the deferred prelude panics

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 {

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