From 76143de3fd97e57ebb85060a5d1886f1feaa0dfa Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:30:10 +0000 Subject: [PATCH 1/9] test(runtime): reproduce the prelude panic leaks of #139 The three tests serve the runtime behind a real httptest.Server, so net/http recovers each panic. On main each one fails: a redelivery of the panicked event waits until its client gives up, the next delivery for the thread never gets the Thread Lock, and a burst member whose join log panicked never runs. --- prelude_panic_test.go | 252 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 252 insertions(+) create mode 100644 prelude_panic_test.go diff --git a/prelude_panic_test.go b/prelude_panic_test.go new file mode 100644 index 0000000..4167193 --- /dev/null +++ b/prelude_panic_test.go @@ -0,0 +1,252 @@ +package chat_test + +import ( + "bytes" + "context" + "encoding/json" + "io" + "log" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/coder/chat" +) + +const preludePanic = "panicked in the prelude" + +// panicServer serves a runtime's fake webhook on a real httptest.Server, so a +// panic in dispatch reaches net/http, which recovers it and aborts the request. +type panicServer struct { + srv *httptest.Server + client *http.Client + log syncBuffer +} + +type panicPost struct { + status int + err error +} + +func newPanicServer(t *testing.T, bot *chat.Chat) *panicServer { + t.Helper() + handler, err := bot.Webhook("fake") + if err != nil { + t.Fatalf("webhook: %v", err) + } + s := &panicServer{srv: httptest.NewUnstartedServer(handler)} + s.srv.Config.ErrorLog = log.New(&s.log, "", 0) + s.srv.Start() + t.Cleanup(s.srv.Close) + s.client = s.srv.Client() + s.client.Timeout = 2 * time.Second + return s +} + +func (s *panicServer) post(ev chat.Event) panicPost { + body, err := json.Marshal(ev) + if err != nil { + return panicPost{err: err} + } + resp, err := s.client.Post(s.srv.URL, "application/json", bytes.NewReader(body)) + if err != nil { + return panicPost{err: err} + } + resp.Body.Close() + return panicPost{status: resp.StatusCode} +} + +func (s *panicServer) postAsync(ev chat.Event) <-chan panicPost { + done := make(chan panicPost, 1) + go func() { done <- s.post(ev) }() + return done +} + +func (s *panicServer) requireAborted(t *testing.T, res panicPost, name string) { + t.Helper() + if res.err == nil { + t.Fatalf("%s: status = %d, want the panic to abort the request", name, res.status) + } + if !strings.Contains(s.log.String(), preludePanic) { + t.Fatalf("%s: net/http did not recover the panic (%v), server log: %q", name, res.err, s.log.String()) + } +} + +func newPanicRuntime(t *testing.T, options chat.RuntimeOptions, opts ...chat.Option) *chat.Chat { + t.Helper() + opts = append([]chat.Option{ + chat.WithAdapter(newFakeAdapter("fake")), + chat.WithLogger(slog.New(slog.NewTextHandler(io.Discard, nil))), + chat.WithRuntimeOptions(options), + }, opts...) + bot, err := chat.New(context.Background(), opts...) + if err != nil { + t.Fatalf("new runtime: %v", err) + } + return bot +} + +func panicRuntimeOptions(concurrency chat.ConcurrencyStrategy, mode chat.DispatchMode) chat.RuntimeOptions { + return chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + Concurrency: concurrency, + Dispatch: mode, + MaxDetached: 1024, + DetachTimeout: 5 * time.Second, + } +} + +type firstMarkPanicState struct { + *fakeState + panicked atomic.Bool +} + +func (s *firstMarkPanicState) MarkEvent(ctx context.Context, id string, ttl time.Duration) (bool, error) { + if s.panicked.CompareAndSwap(false, true) { + panic(preludePanic) + } + return s.fakeState.MarkEvent(ctx, id, ttl) +} + +// firstRecordPanicHandler panics on the first record whose message is msg. +type firstRecordPanicHandler struct { + slog.Handler + msg string + panicked *atomic.Bool +} + +func (h firstRecordPanicHandler) Handle(ctx context.Context, r slog.Record) error { + if r.Message == h.msg && h.panicked.CompareAndSwap(false, true) { + panic(preludePanic) + } + return h.Handler.Handle(ctx, r) +} + +func TestPreludePanicFinishesItsEventAcceptance(t *testing.T) { + t.Parallel() + + for _, tc := range queueDispatchModes { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + state := newGatedAcquireState(1) + state.gatePanic = preludePanic + observer := &recordingObserver{} + bot := newPanicRuntime(t, panicRuntimeOptions(chat.ConcurrencyDrop, tc.mode), + chat.WithState(state), chat.WithObserver(observer)) + var handled atomic.Int32 + bot.OnNewMention(func(context.Context, *chat.MessageEvent) error { + handled.Add(1) + return nil + }) + srv := newPanicServer(t, bot) + event := mentionEvent("A", queueThread) + + first := srv.postAsync(event) + waitClosed(t, state.reached, "A's prelude did not reach the Thread Lock") + duplicate := srv.postAsync(event) + eventually(t, 2*time.Second, func() bool { + observer.mu.Lock() + defer observer.mu.Unlock() + return observer.opened == 2 + }, "the duplicate of A did not start its dispatch") + close(state.release) + + srv.requireAborted(t, <-first, "A") + if res := <-duplicate; res.err == nil && res.status == http.StatusOK && state.acquireCalls() == 1 { + t.Fatal("the duplicate of A was acknowledged, but the panicked delivery never accepted A") + } + if res := srv.post(event); res.err != nil || res.status != http.StatusOK { + t.Fatalf("the redelivery of A: status=%d err=%v, want 200", res.status, res.err) + } + eventually(t, 2*time.Second, func() bool { return handled.Load() == 1 }, + "the redelivery of A did not run its handler") + }) + } +} + +func TestPreludePanicReleasesItsThreadLock(t *testing.T) { + t.Parallel() + + for _, tc := range queueDispatchModes { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + bot := newPanicRuntime(t, panicRuntimeOptions(chat.ConcurrencyQueue, tc.mode), + chat.WithState(&firstMarkPanicState{fakeState: newFakeState()})) + var handled atomic.Int32 + bot.OnNewMention(func(_ context.Context, ev *chat.MessageEvent) error { + if ev.Event.ID == "B" { + handled.Add(1) + } + return nil + }) + srv := newPanicServer(t, bot) + + srv.requireAborted(t, srv.post(mentionEvent("A", queueThread)), "A") + if res := srv.post(mentionEvent("B", queueThread)); res.err != nil || res.status != http.StatusOK { + t.Fatalf("B: status=%d err=%v, want 200", res.status, res.err) + } + eventually(t, 2*time.Second, func() bool { return handled.Load() == 1 }, + "B did not run: A's MarkEvent panic kept the Thread Lock, whose TTL is an hour") + }) + } +} + +func TestBurstMemberLoggerPanicStillRunsItsScope(t *testing.T) { + t.Parallel() + + logger := slog.New(firstRecordPanicHandler{ + Handler: slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelDebug}), + msg: "chat burst member joined", + panicked: &atomic.Bool{}, + }) + options := panicRuntimeOptions(chat.ConcurrencyBurst, chat.DispatchDeferred) + options.MaxDetached = 1 + options.BurstWindow = time.Millisecond + options.ThreadLockTTL = 200 * time.Millisecond + bot := newPanicRuntime(t, options, chat.WithState(newFakeState()), chat.WithLogger(logger)) + started := make(chan string, 2) + releaseA := make(chan struct{}) + bot.OnNewMention(func(_ context.Context, ev *chat.MessageEvent) error { + started <- ev.Event.ID + if ev.Event.ID == "A" { + <-releaseA + } + return nil + }) + srv := newPanicServer(t, bot) + waitStarted := func(want, msg string) { + t.Helper() + select { + case got := <-started: + if got != want { + t.Fatalf("handler started for %s, want %s", got, want) + } + case <-time.After(2 * time.Second): + t.Fatal(msg) + } + } + + srv.requireAborted(t, srv.post(mentionEvent("A", queueThread)), "A") + waitStarted("A", "A's burst member did not run after the logger panicked") + if res := srv.post(mentionEvent("B", queueThread)); res.err != nil || res.status == http.StatusOK { + t.Fatalf("B: status=%d err=%v, want an admission rejection while A holds the only slot", res.status, res.err) + } + close(releaseA) + eventually(t, 2*time.Second, func() bool { + return srv.post(mentionEvent("B", queueThread)).status == http.StatusOK + }, "B was not admitted after A finished") + waitStarted("B", "B's burst member did not run") + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := bot.Shutdown(ctx); err != nil { + t.Fatalf("Shutdown after the logger panicked: %v", err) + } +} From a855cd7ae84198452a6cdedee17ebf7471a0d35c Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:31:07 +0000 Subject: [PATCH 2/9] fix(runtime): finish the acceptance entry when the prelude panics The prelude defers finishEventAcceptance, which now returns early once the entry is done. A panic from State or the logger therefore frees the entry: duplicates that wait on it fail, and a redelivery runs. --- runtime.go | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/runtime.go b/runtime.go index 1afed1d..a2047e6 100644 --- a/runtime.go +++ b/runtime.go @@ -279,6 +279,8 @@ type eventAcceptance struct { err error } +var errPreludePanicked = errors.New("chat: the event's primary delivery panicked before it accepted the event") + func New(ctx context.Context, opts ...Option) (*Chat, error) { cfg := config{ logger: slog.Default(), @@ -804,6 +806,10 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que c.safeEnd(span, OutcomeDuplicate) return preludeWork{}, true, err } + // Every return below has finished the entry already. Only a panic from + // State or the logger reaches this with the entry open: its duplicates + // fail, and a redelivery becomes the primary. + defer c.finishEventAcceptance(event.ID, acceptance, errPreludePanicked) finish := func(err error) error { c.finishEventAcceptance(event.ID, acceptance, err) return err @@ -1599,6 +1605,11 @@ func (c *Chat) beginEventAcceptance(eventID string) (*eventAcceptance, bool) { func (c *Chat) finishEventAcceptance(eventID string, acceptance *eventAcceptance, err error) { c.acceptancesMu.Lock() defer c.acceptancesMu.Unlock() + select { + case <-acceptance.done: + return + default: + } if c.eventAcceptances[eventID] == acceptance { delete(c.eventAcceptances, eventID) } From 42e3a83184f9bce93f1af60ee8566bd0fa561204 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:45:06 +0000 Subject: [PATCH 3/9] test(runtime): pin the lease release on a supersede log panic A logger panic in the queue supersede log leaves the Thread Lock with the prelude, like a MarkEvent panic. The lock test also checks that a delivery which resolves in the prelude releases its lease once. --- prelude_panic_test.go | 78 ++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 77 insertions(+), 1 deletion(-) diff --git a/prelude_panic_test.go b/prelude_panic_test.go index 4167193..efb48da 100644 --- a/prelude_panic_test.go +++ b/prelude_panic_test.go @@ -113,6 +113,12 @@ func (s *firstMarkPanicState) MarkEvent(ctx context.Context, id string, ttl time return s.fakeState.MarkEvent(ctx, id, ttl) } +func (s *firstMarkPanicState) lockFree() bool { + s.fakeState.mu.Lock() + defer s.fakeState.mu.Unlock() + return len(s.locked) == 0 +} + // firstRecordPanicHandler panics on the first record whose message is msg. type firstRecordPanicHandler struct { slog.Handler @@ -177,8 +183,10 @@ func TestPreludePanicReleasesItsThreadLock(t *testing.T) { t.Run(tc.name, func(t *testing.T) { t.Parallel() + state := &firstMarkPanicState{fakeState: newFakeState()} + observer := &recordingObserver{} bot := newPanicRuntime(t, panicRuntimeOptions(chat.ConcurrencyQueue, tc.mode), - chat.WithState(&firstMarkPanicState{fakeState: newFakeState()})) + chat.WithState(state), chat.WithObserver(observer)) var handled atomic.Int32 bot.OnNewMention(func(_ context.Context, ev *chat.MessageEvent) error { if ev.Event.ID == "B" { @@ -194,6 +202,74 @@ func TestPreludePanicReleasesItsThreadLock(t *testing.T) { } eventually(t, 2*time.Second, func() bool { return handled.Load() == 1 }, "B did not run: A's MarkEvent panic kept the Thread Lock, whose TTL is an hour") + eventually(t, 2*time.Second, state.lockFree, "B did not release the Thread Lock") + if res := srv.post(mentionEvent("B", queueThread)); res.err != nil || res.status != http.StatusOK { + t.Fatalf("the redelivery of B: status=%d err=%v, want 200", res.status, res.err) + } + if observer.hasEvent(chat.ObsLockReleaseFailed) { + t.Fatal("a Thread Lock release failed: the prelude released a lease twice") + } + }) + } +} + +func TestQueueSupersedeLoggerPanicReleasesItsThreadLock(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) + logger := slog.New(firstRecordPanicHandler{ + Handler: slog.NewTextHandler(io.Discard, nil), + msg: "chat queue superseded", + panicked: &atomic.Bool{}, + }) + bot := newPanicRuntime(t, panicRuntimeOptions(chat.ConcurrencyQueue, tc.mode), + chat.WithState(state), chat.WithLogger(logger)) + aStarted := make(chan struct{}) + releaseA := make(chan struct{}) + var dHandled atomic.Bool + bot.OnNewMention(func(_ context.Context, ev *chat.MessageEvent) error { + switch ev.Event.ID { + case "A": + close(aStarted) + <-releaseA + case "B": + t.Error("B ran, but C superseded it") + case "D": + dHandled.Store(true) + } + return nil + }) + srv := newPanicServer(t, bot) + + a := srv.postAsync(mentionEvent("A", queueThread)) + waitClosed(t, aStarted, "A's handler did not start") + b := srv.postAsync(mentionEvent("B", queueThread)) + waitClosed(t, state.reached, "B did not poll for the lock") + close(releaseA) + if res := <-a; res.err != nil || res.status != http.StatusOK { + t.Fatalf("A: status=%d err=%v, want 200", res.status, res.err) + } + eventually(t, 2*time.Second, func() bool { return !state.lockHeld(string(queueThread)) }, + "A did not release the Thread Lock") + srv.requireAborted(t, srv.post(mentionEvent("C", queueThread)), "C") + close(state.release) + if res := <-b; res.err != nil || res.status != http.StatusOK { + t.Fatalf("B: status=%d err=%v, want 200 for a superseded waiter", res.status, res.err) + } + if res := srv.post(mentionEvent("D", queueThread)); res.err != nil || res.status != http.StatusOK { + t.Fatalf("D: status=%d err=%v, want 200", res.status, res.err) + } + eventually(t, 2*time.Second, dHandled.Load, + "D did not run: C's supersede log panicked while C held the Thread Lock") }) } } From d9aa9a09d9def49a7f00e34a17220e1228fa3547 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:45:32 +0000 Subject: [PATCH 4/9] fix(runtime): release the Thread Lock when the prelude panics The prelude defers releaseOnResolve, which now clears the lease it frees. Routed work takes the lease out first, so the defer frees only a lease a panic left behind. takeQueueTurn deletes the superseded waiter before it logs, so a logger panic there cannot leave that waiter pending behind a newer turn. --- runtime.go | 27 +++++++++++++++++++-------- 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/runtime.go b/runtime.go index a2047e6..1421db9 100644 --- a/runtime.go +++ b/runtime.go @@ -882,11 +882,22 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que // releaseOnResolve releases the lease when the event resolves here; an event // whose strategy defers lock coordination to the tail holds no lease yet. + // The deferred call frees a lease that a panic left with the prelude. + // handOff gives the lease to the dispatch tail with the routed work. releaseOnResolve := func() { if lease == (LockLease{}) { return } - c.releaseLock(ctx, lease, event.ThreadID) + held := lease + lease = LockLease{} + c.releaseLock(ctx, held, event.ThreadID) + } + defer releaseOnResolve() + handOff := func(work preludeWork, resolved bool, err error) (preludeWork, bool, error) { + if !resolved { + lease = LockLease{} + } + return work, resolved, err } // resolveError releases the lease and closes the span for a failed prelude. resolveError := func(err error) (preludeWork, bool, error) { @@ -931,9 +942,9 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que return resolveIgnored("no-command-handler", "chat ignored command with no handler", slog.LevelInfo) } cmdEvent := &CommandEvent{Event: event, Thread: thread, Command: event.Command} - return c.routedWork(ctx, span, event, lease, scope, seq, conflicted, "command", func(ctx context.Context) error { + return handOff(c.routedWork(ctx, span, event, lease, scope, seq, conflicted, "command", func(ctx context.Context) error { return handler(ctx, cmdEvent) - }, acceptEvent, releaseOnResolve) + }, acceptEvent, releaseOnResolve)) case event.Interaction != nil: if isSelfActor(event.Interaction.Actor, bot) { @@ -946,9 +957,9 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que return resolveIgnored("no-interaction-handler", "chat ignored interaction with no handler", slog.LevelInfo) } intEvent := &InteractionEvent{Event: event, Thread: thread, Interaction: event.Interaction} - return c.routedWork(ctx, span, event, lease, scope, seq, conflicted, "interaction", func(ctx context.Context) error { + return handOff(c.routedWork(ctx, span, event, lease, scope, seq, conflicted, "interaction", func(ctx context.Context) error { return handler(ctx, intEvent) - }, acceptEvent, releaseOnResolve) + }, acceptEvent, releaseOnResolve)) } if event.Message == nil { @@ -967,9 +978,9 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que } msgEvent := &MessageEvent{Event: event, Thread: thread, Message: event.Message} - return c.routedWork(ctx, span, event, lease, scope, seq, conflicted, route, func(ctx context.Context) error { + return handOff(c.routedWork(ctx, span, event, lease, scope, seq, conflicted, route, func(ctx context.Context) error { return handler(ctx, msgEvent) - }, acceptEvent, releaseOnResolve) + }, acceptEvent, releaseOnResolve)) } // routedWork accepts a routed event (deduping), applies the Concurrency @@ -1437,8 +1448,8 @@ func (c *Chat) takeQueueTurn(scope string, event *Event, seq uint64) bool { return true } 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) + c.displaceLocked(prev, event, "queue") return true } From 27fc1429a458c4c4b7f625d6450a61cf4befd1ed Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:46:09 +0000 Subject: [PATCH 5/9] fix(runtime): start the burst runner when the join log panics joinBurstBatch now starts the runner it promised in a defer. The runner still starts after the join log, and a logger panic no longer strands the member, its admission slot, and every later member of the scope. --- burst.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/burst.go b/burst.go index db7edf8..40ba4e8 100644 --- a/burst.go +++ b/burst.go @@ -95,10 +95,12 @@ func (c *Chat) joinBurstBatch(work preludeWork) { c.inflight.Add(1) } c.burstMu.Unlock() - c.logger.Debug("chat burst member joined", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID, "route", work.route) if startRunner { - go c.runBurstScope(scope) + // Deferred so the runner starts after the join log, and still starts + // when the logger panics: it owns this member and every later one. + defer func() { go c.runBurstScope(scope) }() } + c.logger.Debug("chat burst member joined", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID, "route", work.route) } // runBurstScope is the per-scope burst runner: it dispatches sealed batches in From 3cd1fddeae7b42db8a974545f2832c00f53ae36a Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:46:52 +0000 Subject: [PATCH 6/9] docs(adr): record the prelude panic release rule in ADR 0015 --- docs/adr/0015-runtime-coordination.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/adr/0015-runtime-coordination.md b/docs/adr/0015-runtime-coordination.md index c9ccdcb..f3bee20 100644 --- a/docs/adr/0015-runtime-coordination.md +++ b/docs/adr/0015-runtime-coordination.md @@ -39,6 +39,7 @@ This qualifies two accepted ADRs, flagged per the domain docs' ADR-conflict rule 1. **Bounded retention.** Everything an admitted deferred delivery retains (goroutine, payload, closure — running tails, parked waiters, slot-waiters, future batch members) counts against the cap, and capacity is never released while that work remains retained. 2. **Honest rejection.** A rejected delivery is never acknowledged as handled, never marked in **Event Identity** dedupe, and always observable: a new observation (`admission_rejected`, carrying adapter and tenant) and a new terminal dispatch outcome (`admission-rejected`). 3. **No silent loss.** Every admitted delivery reaches an observable terminal outcome — including under **Runtime Shutdown**. +4. **Panic release.** When `State` or the app's `slog` handler panics during the prelude, the runtime releases every resource the delivery took and does not recover the panic. Per-strategy invariant under `DispatchDeferred` (one line each): From 716ffa43439cae314f5f7b5e3d2521f5061bea15 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:47:29 +0000 Subject: [PATCH 7/9] test(runtime): keep the panic handler's flag in the handler --- prelude_panic_test.go | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/prelude_panic_test.go b/prelude_panic_test.go index efb48da..654c278 100644 --- a/prelude_panic_test.go +++ b/prelude_panic_test.go @@ -123,10 +123,10 @@ func (s *firstMarkPanicState) lockFree() bool { type firstRecordPanicHandler struct { slog.Handler msg string - panicked *atomic.Bool + panicked atomic.Bool } -func (h firstRecordPanicHandler) Handle(ctx context.Context, r slog.Record) error { +func (h *firstRecordPanicHandler) Handle(ctx context.Context, r slog.Record) error { if r.Message == h.msg && h.panicked.CompareAndSwap(false, true) { panic(preludePanic) } @@ -226,10 +226,9 @@ func TestQueueSupersedeLoggerPanicReleasesItsThreadLock(t *testing.T) { bFirstPoll ) state := newGatedAcquireState(bFirstPoll) - logger := slog.New(firstRecordPanicHandler{ - Handler: slog.NewTextHandler(io.Discard, nil), - msg: "chat queue superseded", - panicked: &atomic.Bool{}, + logger := slog.New(&firstRecordPanicHandler{ + Handler: slog.NewTextHandler(io.Discard, nil), + msg: "chat queue superseded", }) bot := newPanicRuntime(t, panicRuntimeOptions(chat.ConcurrencyQueue, tc.mode), chat.WithState(state), chat.WithLogger(logger)) @@ -277,10 +276,9 @@ func TestQueueSupersedeLoggerPanicReleasesItsThreadLock(t *testing.T) { func TestBurstMemberLoggerPanicStillRunsItsScope(t *testing.T) { t.Parallel() - logger := slog.New(firstRecordPanicHandler{ - Handler: slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelDebug}), - msg: "chat burst member joined", - panicked: &atomic.Bool{}, + logger := slog.New(&firstRecordPanicHandler{ + Handler: slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelDebug}), + msg: "chat burst member joined", }) options := panicRuntimeOptions(chat.ConcurrencyBurst, chat.DispatchDeferred) options.MaxDetached = 1 From e804b8c39856b8a288b8e4452cdcdb991bfa03e9 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 07:50:47 +0000 Subject: [PATCH 8/9] refactor(runtime): name the acceptance abandon and the lease handoff Comment review: the deferred finish read as if every delivery ended with the panic error. abandonEventAcceptance owns the open-entry check, so finishEventAcceptance stays strict, and handLeaseToTail names the handoff. The ADR rule names locks, slots, and records instead of every resource. --- burst.go | 2 -- docs/adr/0015-runtime-coordination.md | 2 +- prelude_panic_test.go | 1 - runtime.go | 32 ++++++++++++++------------- 4 files changed, 18 insertions(+), 19 deletions(-) diff --git a/burst.go b/burst.go index 40ba4e8..d7111fe 100644 --- a/burst.go +++ b/burst.go @@ -96,8 +96,6 @@ func (c *Chat) joinBurstBatch(work preludeWork) { } c.burstMu.Unlock() if startRunner { - // Deferred so the runner starts after the join log, and still starts - // when the logger panics: it owns this member and every later one. defer func() { go c.runBurstScope(scope) }() } c.logger.Debug("chat burst member joined", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID, "route", work.route) diff --git a/docs/adr/0015-runtime-coordination.md b/docs/adr/0015-runtime-coordination.md index f3bee20..d2dbc22 100644 --- a/docs/adr/0015-runtime-coordination.md +++ b/docs/adr/0015-runtime-coordination.md @@ -39,7 +39,7 @@ This qualifies two accepted ADRs, flagged per the domain docs' ADR-conflict rule 1. **Bounded retention.** Everything an admitted deferred delivery retains (goroutine, payload, closure — running tails, parked waiters, slot-waiters, future batch members) counts against the cap, and capacity is never released while that work remains retained. 2. **Honest rejection.** A rejected delivery is never acknowledged as handled, never marked in **Event Identity** dedupe, and always observable: a new observation (`admission_rejected`, carrying adapter and tenant) and a new terminal dispatch outcome (`admission-rejected`). 3. **No silent loss.** Every admitted delivery reaches an observable terminal outcome — including under **Runtime Shutdown**. -4. **Panic release.** When `State` or the app's `slog` handler panics during the prelude, the runtime releases every resource the delivery took and does not recover the panic. +4. **Panic release.** When `State` or the app's `slog` handler panics during the prelude, the runtime releases every lock, slot, and record the delivery holds and does not recover the panic. Per-strategy invariant under `DispatchDeferred` (one line each): diff --git a/prelude_panic_test.go b/prelude_panic_test.go index 654c278..715c4a4 100644 --- a/prelude_panic_test.go +++ b/prelude_panic_test.go @@ -119,7 +119,6 @@ func (s *firstMarkPanicState) lockFree() bool { return len(s.locked) == 0 } -// firstRecordPanicHandler panics on the first record whose message is msg. type firstRecordPanicHandler struct { slog.Handler msg string diff --git a/runtime.go b/runtime.go index 1421db9..a8132da 100644 --- a/runtime.go +++ b/runtime.go @@ -806,10 +806,7 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que c.safeEnd(span, OutcomeDuplicate) return preludeWork{}, true, err } - // Every return below has finished the entry already. Only a panic from - // State or the logger reaches this with the entry open: its duplicates - // fail, and a redelivery becomes the primary. - defer c.finishEventAcceptance(event.ID, acceptance, errPreludePanicked) + defer c.abandonEventAcceptance(event.ID, acceptance) finish := func(err error) error { c.finishEventAcceptance(event.ID, acceptance, err) return err @@ -882,8 +879,6 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que // releaseOnResolve releases the lease when the event resolves here; an event // whose strategy defers lock coordination to the tail holds no lease yet. - // The deferred call frees a lease that a panic left with the prelude. - // handOff gives the lease to the dispatch tail with the routed work. releaseOnResolve := func() { if lease == (LockLease{}) { return @@ -893,7 +888,7 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que c.releaseLock(ctx, held, event.ThreadID) } defer releaseOnResolve() - handOff := func(work preludeWork, resolved bool, err error) (preludeWork, bool, error) { + handLeaseToTail := func(work preludeWork, resolved bool, err error) (preludeWork, bool, error) { if !resolved { lease = LockLease{} } @@ -942,7 +937,7 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que return resolveIgnored("no-command-handler", "chat ignored command with no handler", slog.LevelInfo) } cmdEvent := &CommandEvent{Event: event, Thread: thread, Command: event.Command} - return handOff(c.routedWork(ctx, span, event, lease, scope, seq, conflicted, "command", func(ctx context.Context) error { + return handLeaseToTail(c.routedWork(ctx, span, event, lease, scope, seq, conflicted, "command", func(ctx context.Context) error { return handler(ctx, cmdEvent) }, acceptEvent, releaseOnResolve)) @@ -957,7 +952,7 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que return resolveIgnored("no-interaction-handler", "chat ignored interaction with no handler", slog.LevelInfo) } intEvent := &InteractionEvent{Event: event, Thread: thread, Interaction: event.Interaction} - return handOff(c.routedWork(ctx, span, event, lease, scope, seq, conflicted, "interaction", func(ctx context.Context) error { + return handLeaseToTail(c.routedWork(ctx, span, event, lease, scope, seq, conflicted, "interaction", func(ctx context.Context) error { return handler(ctx, intEvent) }, acceptEvent, releaseOnResolve)) } @@ -978,7 +973,7 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que } msgEvent := &MessageEvent{Event: event, Thread: thread, Message: event.Message} - return handOff(c.routedWork(ctx, span, event, lease, scope, seq, conflicted, route, func(ctx context.Context) error { + return handLeaseToTail(c.routedWork(ctx, span, event, lease, scope, seq, conflicted, route, func(ctx context.Context) error { return handler(ctx, msgEvent) }, acceptEvent, releaseOnResolve)) } @@ -1616,11 +1611,6 @@ func (c *Chat) beginEventAcceptance(eventID string) (*eventAcceptance, bool) { func (c *Chat) finishEventAcceptance(eventID string, acceptance *eventAcceptance, err error) { c.acceptancesMu.Lock() defer c.acceptancesMu.Unlock() - select { - case <-acceptance.done: - return - default: - } if c.eventAcceptances[eventID] == acceptance { delete(c.eventAcceptances, eventID) } @@ -1628,6 +1618,18 @@ func (c *Chat) finishEventAcceptance(eventID string, acceptance *eventAcceptance close(acceptance.done) } +// abandonEventAcceptance finishes an entry the prelude left open, which only a +// panic does, so its duplicates fail and a redelivery becomes the primary. +// Only the primary's goroutine closes done, so the check needs no lock. +func (c *Chat) abandonEventAcceptance(eventID string, acceptance *eventAcceptance) { + select { + case <-acceptance.done: + return + default: + } + c.finishEventAcceptance(eventID, acceptance, errPreludePanicked) +} + func waitEventAcceptance(ctx context.Context, acceptance *eventAcceptance) error { select { case <-acceptance.done: From 17a07ad72b134eb8c4e88da023955ccbe9dc9683 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 08:45:55 +0000 Subject: [PATCH 9/9] test(runtime): open the panic tests' gates before the server closes A failure before a gate opened left a request blocked on it, and srv.Close waited on that request until the go test timeout. --- prelude_panic_test.go | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/prelude_panic_test.go b/prelude_panic_test.go index 715c4a4..6ff1aef 100644 --- a/prelude_panic_test.go +++ b/prelude_panic_test.go @@ -10,6 +10,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync" "sync/atomic" "testing" "time" @@ -150,6 +151,8 @@ func TestPreludePanicFinishesItsEventAcceptance(t *testing.T) { return nil }) srv := newPanicServer(t, bot) + openGate := sync.OnceFunc(func() { close(state.release) }) + t.Cleanup(openGate) event := mentionEvent("A", queueThread) first := srv.postAsync(event) @@ -160,7 +163,7 @@ func TestPreludePanicFinishesItsEventAcceptance(t *testing.T) { defer observer.mu.Unlock() return observer.opened == 2 }, "the duplicate of A did not start its dispatch") - close(state.release) + openGate() srv.requireAborted(t, <-first, "A") if res := <-duplicate; res.err == nil && res.status == http.StatusOK && state.acquireCalls() == 1 { @@ -247,19 +250,25 @@ func TestQueueSupersedeLoggerPanicReleasesItsThreadLock(t *testing.T) { return nil }) srv := newPanicServer(t, bot) + finishA := sync.OnceFunc(func() { close(releaseA) }) + openGate := sync.OnceFunc(func() { close(state.release) }) + t.Cleanup(func() { + finishA() + openGate() + }) a := srv.postAsync(mentionEvent("A", queueThread)) waitClosed(t, aStarted, "A's handler did not start") b := srv.postAsync(mentionEvent("B", queueThread)) waitClosed(t, state.reached, "B did not poll for the lock") - close(releaseA) + finishA() if res := <-a; res.err != nil || res.status != http.StatusOK { t.Fatalf("A: status=%d err=%v, want 200", res.status, res.err) } eventually(t, 2*time.Second, func() bool { return !state.lockHeld(string(queueThread)) }, "A did not release the Thread Lock") srv.requireAborted(t, srv.post(mentionEvent("C", queueThread)), "C") - close(state.release) + openGate() if res := <-b; res.err != nil || res.status != http.StatusOK { t.Fatalf("B: status=%d err=%v, want 200 for a superseded waiter", res.status, res.err) }