From 1088a1bf6783e11ca96bef644f3f5a7874985e78 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 09:45:46 +0000 Subject: [PATCH 01/11] test(runtime): show a panicked prelude leaves its dispatch span open A MarkEvent panic in the prelude aborts the request, but the delivery's dispatch span never ends, so tracing shows no terminal outcome for it. --- prelude_panic_test.go | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/prelude_panic_test.go b/prelude_panic_test.go index 6ff1aef..389a70c 100644 --- a/prelude_panic_test.go +++ b/prelude_panic_test.go @@ -332,3 +332,26 @@ func TestBurstMemberLoggerPanicStillRunsItsScope(t *testing.T) { t.Fatalf("Shutdown after the logger panicked: %v", err) } } + +func TestPreludePanicEndsItsDispatchSpan(t *testing.T) { + t.Parallel() + + for _, tc := range queueDispatchModes { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + observer := &recordingObserver{} + bot := newPanicRuntime(t, panicRuntimeOptions(chat.ConcurrencyQueue, tc.mode), + chat.WithState(&firstMarkPanicState{fakeState: newFakeState()}), chat.WithObserver(observer)) + bot.OnNewMention(func(context.Context, *chat.MessageEvent) error { return nil }) + srv := newPanicServer(t, bot) + + srv.requireAborted(t, srv.post(mentionEvent("A", queueThread)), "A") + eventually(t, 2*time.Second, func() bool { return len(observer.terminalOutcomes()) == 1 }, + "A's dispatch span did not end after its MarkEvent panicked") + if got := observer.terminalOutcomes(); got[0] != chat.OutcomeError { + t.Fatalf("A's dispatch span ended %q, want %q", got[0], chat.OutcomeError) + } + }) + } +} From f36b0190bf91585cdf9cba9dfe0630360045c241 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 09:46:14 +0000 Subject: [PATCH 02/11] test(runtime): cover a span left open by a panic after acceptance A logger panic on the unrouted-message line comes after MarkEvent, so a fix keyed to the unfinished acceptance entry would miss it. --- prelude_panic_test.go | 55 ++++++++++++++++++++++++++++++------------- 1 file changed, 38 insertions(+), 17 deletions(-) diff --git a/prelude_panic_test.go b/prelude_panic_test.go index 389a70c..9f49a34 100644 --- a/prelude_panic_test.go +++ b/prelude_panic_test.go @@ -336,22 +336,43 @@ func TestBurstMemberLoggerPanicStillRunsItsScope(t *testing.T) { func TestPreludePanicEndsItsDispatchSpan(t *testing.T) { t.Parallel() - for _, tc := range queueDispatchModes { - t.Run(tc.name, func(t *testing.T) { - t.Parallel() - - observer := &recordingObserver{} - bot := newPanicRuntime(t, panicRuntimeOptions(chat.ConcurrencyQueue, tc.mode), - chat.WithState(&firstMarkPanicState{fakeState: newFakeState()}), chat.WithObserver(observer)) - bot.OnNewMention(func(context.Context, *chat.MessageEvent) error { return nil }) - srv := newPanicServer(t, bot) - - srv.requireAborted(t, srv.post(mentionEvent("A", queueThread)), "A") - eventually(t, 2*time.Second, func() bool { return len(observer.terminalOutcomes()) == 1 }, - "A's dispatch span did not end after its MarkEvent panicked") - if got := observer.terminalOutcomes(); got[0] != chat.OutcomeError { - t.Fatalf("A's dispatch span ended %q, want %q", got[0], chat.OutcomeError) - } - }) + cases := []struct { + name string + opts func() []chat.Option + }{ + { + name: "MarkEvent", + opts: func() []chat.Option { + return []chat.Option{chat.WithState(&firstMarkPanicState{fakeState: newFakeState()})} + }, + }, + { + name: "ignored log", + opts: func() []chat.Option { + return []chat.Option{chat.WithState(newFakeState()), chat.WithLogger(slog.New(&firstRecordPanicHandler{ + Handler: slog.NewTextHandler(io.Discard, nil), + msg: "chat ignored unrouted message", + }))} + }, + }, + } + for _, tc := range cases { + for _, mode := range queueDispatchModes { + t.Run(tc.name+"/"+mode.name, func(t *testing.T) { + t.Parallel() + + observer := &recordingObserver{} + bot := newPanicRuntime(t, panicRuntimeOptions(chat.ConcurrencyQueue, mode.mode), + append(tc.opts(), chat.WithObserver(observer))...) + srv := newPanicServer(t, bot) + + srv.requireAborted(t, srv.post(mentionEvent("A", queueThread)), "A") + eventually(t, 2*time.Second, func() bool { return len(observer.terminalOutcomes()) == 1 }, + "A's dispatch span did not end after its prelude panicked") + if got := observer.terminalOutcomes(); got[0] != chat.OutcomeError { + t.Fatalf("A's dispatch span ended %q, want %q", got[0], chat.OutcomeError) + } + }) + } } } From c326c768043ceeac7f945f503f0f5154e29bc6e9 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 09:47:05 +0000 Subject: [PATCH 03/11] fix(runtime): end the dispatch span of a prelude that panicked The prelude owns its span until it ends it on a resolved outcome or returns it in work. A panic does neither, so a deferred check on the zero results ends the span with the error outcome after the other releases run. The panic still reaches net/http. ADR 0015 invariant 4 gains the rule. --- docs/adr/0015-runtime-coordination.md | 2 +- runtime.go | 9 ++++++++- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/docs/adr/0015-runtime-coordination.md b/docs/adr/0015-runtime-coordination.md index d2dbc22..5287759 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 lock, slot, and record the delivery holds 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, ends its dispatch span with the `error` outcome, and does not recover the panic. Per-strategy invariant under `DispatchDeferred` (one line each): diff --git a/runtime.go b/runtime.go index 1b8c0a0..5af5c69 100644 --- a/runtime.go +++ b/runtime.go @@ -795,9 +795,16 @@ type preludeWork struct { // Under the queue strategy a Lock Conflict on a routed event does not block: the // event is registered as pending (no lease) and returned with needsLock=true so // the tail acquires the lock, keeping ack prompt under DispatchDeferred. -func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *queueEntry) (preludeWork, bool, error) { +func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *queueEntry) (work preludeWork, resolved bool, err error) { spanCtx, span := c.safeDispatch(ctx, AdapterAttr(event.Adapter), TenantAttr(event.Tenant)) ctx = spanCtx + // A panic leaves the results zero: the prelude neither ended the span on a + // resolved outcome nor handed it to the tail in work. + defer func() { + if !resolved && work.span == nil { + c.safeEnd(span, OutcomeError) + } + }() acceptance, primary := c.beginEventAcceptance(event.ID) if !primary { From 1054920244d2a5ab4d46e8b5dc66dd7347bae85d Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 09:47:59 +0000 Subject: [PATCH 04/11] test(runtime): show a supersede-log panic loses the newer event The supersede log runs on the displacing delivery's own goroutine before that delivery's run starts. When the app's slog handler panics there, the newer event never runs. Under debounce the displaced A runs instead of B. Under queue, C never runs after it takes the free lock, and a waiter displaced through registerPending is never superseded. --- prelude_panic_test.go | 184 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 184 insertions(+) diff --git a/prelude_panic_test.go b/prelude_panic_test.go index 9f49a34..57ec52e 100644 --- a/prelude_panic_test.go +++ b/prelude_panic_test.go @@ -9,6 +9,7 @@ import ( "log/slog" "net/http" "net/http/httptest" + "slices" "strings" "sync" "sync/atomic" @@ -376,3 +377,186 @@ func TestPreludePanicEndsItsDispatchSpan(t *testing.T) { } } } + +func TestDebounceSupersedeLoggerPanicStillRunsTheNewestEvent(t *testing.T) { + t.Parallel() + + const aFirstPoll = 1 + state := newGatedAcquireState(aFirstPoll) + logger := slog.New(&firstRecordPanicHandler{ + Handler: slog.NewTextHandler(io.Discard, nil), + msg: "chat debounce superseded", + }) + observer := &recordingObserver{} + options := panicRuntimeOptions(chat.ConcurrencyDebounce, chat.DispatchDeferred) + options.DebounceInterval = time.Millisecond + bot := newPanicRuntime(t, options, chat.WithState(state), chat.WithLogger(logger), chat.WithObserver(observer)) + handled := make(chan string, 2) + bot.OnNewMention(func(_ context.Context, ev *chat.MessageEvent) error { + handled <- ev.Event.ID + return nil + }) + srv := newPanicServer(t, bot) + openGate := sync.OnceFunc(func() { close(state.release) }) + t.Cleanup(openGate) + + if res := srv.post(mentionEvent("A", queueThread)); res.err != nil || res.status != http.StatusOK { + t.Fatalf("A: status=%d err=%v, want 200", res.status, res.err) + } + waitClosed(t, state.reached, "A did not poll for the lock") + srv.requireAborted(t, srv.post(mentionEvent("B", queueThread)), "B") + openGate() + select { + case id := <-handled: + if id != "B" { + t.Fatalf("%s ran, want B: B superseded A", id) + } + case <-time.After(2 * time.Second): + t.Fatal("B did not run: its supersede log panicked") + } + requireOutcomes(t, observer, chat.OutcomeHandled, chat.OutcomeIgnored) +} + +func requireOutcomes(t *testing.T, observer *recordingObserver, want ...chat.DispatchOutcome) { + t.Helper() + eventually(t, 2*time.Second, func() bool { return len(observer.terminalOutcomes()) >= len(want) }, + "a delivery did not reach a terminal outcome") + got := observer.terminalOutcomes() + slices.Sort(got) + slices.Sort(want) + if !slices.Equal(got, want) { + t.Fatalf("outcomes = %v, want %v", got, want) + } +} + +func TestQueueTurnSupersedeLoggerPanicStillRunsTheNewerEvent(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", + }) + observer := &recordingObserver{} + bot := newPanicRuntime(t, panicRuntimeOptions(chat.ConcurrencyQueue, tc.mode), + chat.WithState(state), chat.WithLogger(logger), chat.WithObserver(observer)) + aStarted := make(chan struct{}) + releaseA := make(chan struct{}) + var cHandled 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 "C": + cHandled.Store(true) + } + 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") + 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") + eventually(t, 2*time.Second, cHandled.Load, "C did not run: its supersede log panicked") + 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) + } + requireOutcomes(t, observer, chat.OutcomeHandled, chat.OutcomeHandled, chat.OutcomeIgnored) + }) + } +} + +func TestQueueWaiterSupersedeLoggerPanicStillRunsTheNewerEvent(t *testing.T) { + t.Parallel() + + for _, tc := range queueDispatchModes { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + const ( + aPreludeAcquire = 1 + iota + wPreludeAcquire + wFirstPoll + ) + state := newGatedAcquireState(wFirstPoll) + logger := slog.New(&firstRecordPanicHandler{ + Handler: slog.NewTextHandler(io.Discard, nil), + msg: "chat queue superseded", + }) + observer := &recordingObserver{} + options := panicRuntimeOptions(chat.ConcurrencyQueue, tc.mode) + options.ThreadLockTTL = 5 * time.Second + bot := newPanicRuntime(t, options, chat.WithState(state), chat.WithLogger(logger), chat.WithObserver(observer)) + aStarted := make(chan struct{}) + releaseA := make(chan struct{}) + var bHandled atomic.Bool + bot.OnNewMention(func(_ context.Context, ev *chat.MessageEvent) error { + switch ev.Event.ID { + case "A": + close(aStarted) + <-releaseA + case "W": + t.Error("W ran, but B superseded it") + case "B": + bHandled.Store(true) + } + return nil + }) + srv := newPanicServer(t, bot) + finishA := sync.OnceFunc(func() { close(releaseA) }) + openGate := sync.OnceFunc(func() { close(state.release) }) + t.Cleanup(func() { + openGate() + finishA() + }) + + a := srv.postAsync(mentionEvent("A", queueThread)) + waitClosed(t, aStarted, "A's handler did not start") + w := srv.postAsync(mentionEvent("W", queueThread)) + waitClosed(t, state.reached, "W did not poll for the lock") + b := srv.postAsync(mentionEvent("B", queueThread)) + openGate() + eventually(t, 2*time.Second, func() bool { + got := observer.terminalOutcomes() + return len(got) == 1 && got[0] == chat.OutcomeIgnored + }, "W did not end superseded: B's supersede log panicked before the handoff") + 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, bHandled.Load, "B did not run: its supersede log panicked") + srv.requireAborted(t, <-b, "B") + if res := <-w; res.err != nil || res.status != http.StatusOK { + t.Fatalf("W: status=%d err=%v, want 200 for a superseded waiter", res.status, res.err) + } + requireOutcomes(t, observer, chat.OutcomeHandled, chat.OutcomeHandled, chat.OutcomeIgnored) + }) + } +} From 95f591896562d92d476beb7218b5300b6a56cccd Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 09:49:15 +0000 Subject: [PATCH 05/11] fix(runtime): log a supersession only after the newer run is secured registerPending and takeQueueTurn now finish the handoff under queueMu without calling the app logger, and return the displaced waiter's event. The dispatcher writes the same supersede line once the newer delivery's run is secured: under DispatchDeferred right after its tail starts, and under DispatchSync after its handler returns. A panicking slog handler on that line still aborts the request, but the newer event runs and the displaced waiter ends superseded. --- runtime.go | 69 ++++++++++++++++++++++++++++++++---------------------- 1 file changed, 41 insertions(+), 28 deletions(-) diff --git a/runtime.go b/runtime.go index 5af5c69..386e843 100644 --- a/runtime.go +++ b/runtime.go @@ -678,6 +678,7 @@ func (c *Chat) dispatchSync(ctx context.Context, event *Event, seq uint64, entry if err != nil || resolved { return err } + defer c.logSupersession(work) if work.noLock { if !c.acquireConcurrencySlot(ctx, event) { c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) @@ -733,6 +734,7 @@ func (c *Chat) dispatchDeferred(ctx context.Context, event *Event, seq uint64, e return nil } c.startDetachedTail(work) + c.logSupersession(work) return nil } @@ -758,6 +760,10 @@ type preludeWork struct { // debounce quiet-period wait exits promptly on it instead of parking through // its full interval. displaced <-chan struct{} + // supersedes is the pending waiter's event this delivery displaced. The + // dispatcher logs the supersession only once this delivery's run is + // secured, so a panicking slog handler cannot lose it. + supersedes *Event // debounce is true when the tail must hold the event through the // DebounceInterval quiet period before waiting for the lock. debounce bool @@ -1038,31 +1044,35 @@ func (c *Chat) routedWork( return work, false, nil case ConcurrencyQueue: + work.waitLabel = "queue" if conflicted { work.needsLock = true - work.waitLabel = "queue" - displaced, registered := c.registerPending(scope, event, seq, work.waitLabel) + displaced, supersedes, registered := c.registerPending(scope, event, seq) if !registered { return resolveStale(work.waitLabel) } work.displaced = displaced + work.supersedes = supersedes return work, false, nil } - if !c.takeQueueTurn(scope, event, seq) { + supersedes, took := c.takeQueueTurn(scope, event, seq) + if !took { releaseOnResolve() return resolveStale("queue") } + work.supersedes = supersedes return work, false, nil case ConcurrencyDebounce: work.needsLock = true work.debounce = true work.waitLabel = "debounce" - displaced, registered := c.registerPending(scope, event, seq, work.waitLabel) + displaced, supersedes, registered := c.registerPending(scope, event, seq) if !registered { return resolveStale(work.waitLabel) } work.displaced = displaced + work.supersedes = supersedes return work, false, nil case ConcurrencyConcurrent: @@ -1421,55 +1431,58 @@ func (c *Chat) leaveQueueScope(entry *queueEntry) { // its prelude (validation, dedupe, routing) never displaces a waiter admitted // after it, and instead reports registered=false so the caller resolves it as // superseded. Under queue it also reports registered=false when the scope's -// latest turn was admitted after it. Recording supersession here, where the -// displacing event's id is known, lets the record carry superseded_by. label -// names the coordination mode ("queue" or "debounce") so the observation names -// the strategy that displaced the waiter. The returned channel closes when -// this registration is itself superseded. -func (c *Chat) registerPending(scope string, event *Event, seq uint64, label string) (displaced <-chan struct{}, registered bool) { +// latest turn was admitted after it. supersedes is the displaced waiter's +// event, for the caller's supersession record. The returned channel closes +// when this registration is itself superseded. +func (c *Chat) registerPending(scope string, event *Event, seq uint64) (displaced <-chan struct{}, supersedes *Event, registered bool) { c.queueMu.Lock() defer c.queueMu.Unlock() if turn := c.queueTurns[scope]; turn != nil && turn.won > seq { - return nil, false + return nil, nil, false } if prev := c.pending[scope]; prev != nil { if prev.seq > seq { - return nil, false + return nil, nil, false } - c.displaceLocked(prev, event, label) + close(prev.displaced) + supersedes = prev.event } waiter := &pendingWaiter{event: event, seq: seq, displaced: make(chan struct{})} c.pending[scope] = waiter - return waiter.displaced, true -} - -func (c *Chat) displaceLocked(prev *pendingWaiter, by *Event, label string) { - c.logger.Info("chat "+label+" superseded", "adapter", prev.event.Adapter, - "event_id", prev.event.ID, "thread_id", prev.event.ThreadID, "superseded_by", by.ID) - close(prev.displaced) + return waiter.displaced, supersedes, true } // takeQueueTurn gives event, which took the free lock in its prelude, its -// scope's turn: it supersedes an older pending waiter and makes every delivery -// admitted before event stale. It reports false, and takes no turn, when a -// newer delivery already took one. -func (c *Chat) takeQueueTurn(scope string, event *Event, seq uint64) bool { +// scope's turn: it supersedes an older pending waiter, whose event it returns, +// and makes every delivery admitted before event stale. It reports took=false, +// and takes no turn, when a newer delivery already took one. +func (c *Chat) takeQueueTurn(scope string, event *Event, seq uint64) (supersedes *Event, took bool) { c.queueMu.Lock() defer c.queueMu.Unlock() turn := c.queueTurns[scope] assert(turn != nil, "a queue delivery takes a turn in a scope it entered") if turn.won > seq { - return false + return nil, false } turn.won = seq prev := c.pending[scope] if prev == nil || prev.seq > seq { - return true + return nil, true } assert(prev.seq != seq, "a delivery that holds its prelude lease is never its scope's pending waiter") delete(c.pending, scope) - c.displaceLocked(prev, event, "queue") - return true + close(prev.displaced) + return prev.event, true +} + +// logSupersession records that work's delivery displaced an earlier waiter. +func (c *Chat) logSupersession(work preludeWork) { + prev := work.supersedes + if prev == nil { + return + } + c.logger.Info("chat "+work.waitLabel+" superseded", "adapter", prev.Adapter, + "event_id", prev.ID, "thread_id", prev.ThreadID, "superseded_by", work.event.ID) } // isPending reports whether event is still its scope's most-recent pending From 49616bca210e3237c1b9049dd76fd08f3132a1a7 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 09:50:48 +0000 Subject: [PATCH 06/11] chore(runtime): trim comments Signed-off-by: Thomas Kosiewski --- runtime.go | 19 ++++++------------- 1 file changed, 6 insertions(+), 13 deletions(-) diff --git a/runtime.go b/runtime.go index 386e843..50aaf87 100644 --- a/runtime.go +++ b/runtime.go @@ -759,10 +759,7 @@ type preludeWork struct { // displaced closes when a newer event supersedes this pending waiter; the // debounce quiet-period wait exits promptly on it instead of parking through // its full interval. - displaced <-chan struct{} - // supersedes is the pending waiter's event this delivery displaced. The - // dispatcher logs the supersession only once this delivery's run is - // secured, so a panicking slog handler cannot lose it. + displaced <-chan struct{} supersedes *Event // debounce is true when the tail must hold the event through the // DebounceInterval quiet period before waiting for the lock. @@ -804,8 +801,6 @@ type preludeWork struct { func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *queueEntry) (work preludeWork, resolved bool, err error) { spanCtx, span := c.safeDispatch(ctx, AdapterAttr(event.Adapter), TenantAttr(event.Tenant)) ctx = spanCtx - // A panic leaves the results zero: the prelude neither ended the span on a - // resolved outcome nor handed it to the tail in work. defer func() { if !resolved && work.span == nil { c.safeEnd(span, OutcomeError) @@ -1431,9 +1426,8 @@ func (c *Chat) leaveQueueScope(entry *queueEntry) { // its prelude (validation, dedupe, routing) never displaces a waiter admitted // after it, and instead reports registered=false so the caller resolves it as // superseded. Under queue it also reports registered=false when the scope's -// latest turn was admitted after it. supersedes is the displaced waiter's -// event, for the caller's supersession record. The returned channel closes -// when this registration is itself superseded. +// latest turn was admitted after it. The returned channel closes when this +// registration is itself superseded. func (c *Chat) registerPending(scope string, event *Event, seq uint64) (displaced <-chan struct{}, supersedes *Event, registered bool) { c.queueMu.Lock() defer c.queueMu.Unlock() @@ -1453,9 +1447,9 @@ func (c *Chat) registerPending(scope string, event *Event, seq uint64) (displace } // takeQueueTurn gives event, which took the free lock in its prelude, its -// scope's turn: it supersedes an older pending waiter, whose event it returns, -// and makes every delivery admitted before event stale. It reports took=false, -// and takes no turn, when a newer delivery already took one. +// scope's turn: it supersedes an older pending waiter and makes every delivery +// admitted before event stale. It reports took=false, and takes no turn, when a +// newer delivery already took one. func (c *Chat) takeQueueTurn(scope string, event *Event, seq uint64) (supersedes *Event, took bool) { c.queueMu.Lock() defer c.queueMu.Unlock() @@ -1475,7 +1469,6 @@ func (c *Chat) takeQueueTurn(scope string, event *Event, seq uint64) (supersedes return prev.event, true } -// logSupersession records that work's delivery displaced an earlier waiter. func (c *Chat) logSupersession(work preludeWork) { prev := work.supersedes if prev == nil { From 73dd13dc99c68d7b5d59d45074d065d7f6f0ca2e Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 09:52:23 +0000 Subject: [PATCH 07/11] refactor(runtime): mark a panicked prelude with an explicit flag The prelude now opens the span, runs acceptAndRoute, and ends the span with the error outcome when acceptAndRoute did not return. That replaces the check on zero named results, which needed a comment to read. --- runtime.go | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/runtime.go b/runtime.go index 50aaf87..75a2657 100644 --- a/runtime.go +++ b/runtime.go @@ -798,15 +798,20 @@ type preludeWork struct { // Under the queue strategy a Lock Conflict on a routed event does not block: the // event is registered as pending (no lease) and returned with needsLock=true so // the tail acquires the lock, keeping ack prompt under DispatchDeferred. -func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *queueEntry) (work preludeWork, resolved bool, err error) { +func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *queueEntry) (preludeWork, bool, error) { spanCtx, span := c.safeDispatch(ctx, AdapterAttr(event.Adapter), TenantAttr(event.Tenant)) - ctx = spanCtx + panicked := true defer func() { - if !resolved && work.span == nil { + if panicked { c.safeEnd(span, OutcomeError) } }() + work, resolved, err := c.acceptAndRoute(spanCtx, span, event, seq, entry) + panicked = false + return work, resolved, err +} +func (c *Chat) acceptAndRoute(ctx context.Context, span DispatchSpan, event *Event, seq uint64, entry *queueEntry) (preludeWork, bool, error) { acceptance, primary := c.beginEventAcceptance(event.ID) if !primary { // The primary owns the terminal observation for the shared Event Identity. From 2bc81a0c7ecc0d54acb956ff206f0683dfccc8c4 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 10:05:32 +0000 Subject: [PATCH 08/11] test(runtime): show the newer tail keeps the displaced event reachable Codex security review on #145: the newer delivery's tail copies work.supersedes, a pointer to the displaced Event. After the displaced tail frees its admission slot, its payload stays reachable until the newer tail ends, so it no longer counts against MaxDetached. --- debounce_test.go | 84 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 84 insertions(+) diff --git a/debounce_test.go b/debounce_test.go index 215c3b0..436ef8b 100644 --- a/debounce_test.go +++ b/debounce_test.go @@ -4,11 +4,13 @@ import ( "context" "log/slog" "net/http" + "runtime" "strings" "sync" "sync/atomic" "testing" "time" + "weak" "github.com/coder/chat" ) @@ -590,3 +592,85 @@ func TestDebounceDeliveryDelayedInValidationCannotDisplaceNewerWaiter(t *testing t.Fatalf("handled = %v, want only [newer]: a delivery delayed in validation displaced the final event", handled) } } + +type weakEventAdapter struct { + *fakeAdapter + mu sync.Mutex + events map[string]weak.Pointer[chat.Event] +} + +func (a *weakEventAdapter) Webhook(dispatch chat.DispatchFunc) http.Handler { + return a.fakeAdapter.Webhook(func(ctx context.Context, ev *chat.Event) error { + a.mu.Lock() + a.events[ev.ID] = weak.Make(ev) + a.mu.Unlock() + return dispatch(ctx, ev) + }) +} + +func (a *weakEventAdapter) collected(id string) bool { + a.mu.Lock() + defer a.mu.Unlock() + ref, ok := a.events[id] + return ok && ref.Value() == nil +} + +func TestDebounceNewerTailDoesNotRetainTheDisplacedEvent(t *testing.T) { + t.Parallel() + + const aFirstPoll = 1 + state := newGatedAcquireState(aFirstPoll) + adapter := &weakEventAdapter{fakeAdapter: newFakeAdapter("fake"), events: map[string]weak.Pointer[chat.Event]{}} + observer := &recordingObserver{} + bot, err := chat.New(context.Background(), + chat.WithState(state), + chat.WithAdapter(adapter), + chat.WithObserver(observer), + chat.WithRuntimeOptions(chat.RuntimeOptions{ + DedupeTTL: time.Hour, + ThreadLockTTL: time.Hour, + Concurrency: chat.ConcurrencyDebounce, + DebounceInterval: time.Millisecond, + Dispatch: chat.DispatchDeferred, + MaxDetached: 1024, + DetachTimeout: 5 * time.Second, + }), + ) + if err != nil { + t.Fatalf("new runtime: %v", err) + } + bStarted := make(chan struct{}) + releaseB := make(chan struct{}) + bot.OnNewMention(func(_ context.Context, ev *chat.MessageEvent) error { + if ev.Event.ID == "B" { + close(bStarted) + <-releaseB + } + return nil + }) + finishB := sync.OnceFunc(func() { close(releaseB) }) + openGate := sync.OnceFunc(func() { close(state.release) }) + t.Cleanup(func() { + openGate() + finishB() + }) + + if status := postEvent(t, bot, "fake", mentionEvent("A", queueThread)); status != http.StatusOK { + t.Fatalf("A status = %d", status) + } + waitClosed(t, state.reached, "A did not poll for the lock") + if status := postEvent(t, bot, "fake", mentionEvent("B", queueThread)); status != http.StatusOK { + t.Fatalf("B status = %d", status) + } + openGate() + waitClosed(t, bStarted, "B's handler did not start") + eventually(t, 2*time.Second, func() bool { + got := observer.terminalOutcomes() + return len(got) == 1 && got[0] == chat.OutcomeIgnored + }, "A's tail did not end superseded") + eventually(t, 5*time.Second, func() bool { + runtime.GC() + return adapter.collected("A") + }, "B's tail kept the displaced A reachable after A's tail freed its admission slot") + finishB() +} From 21994acfa163f5d849133877e087bb9432fc9e58 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 10:06:18 +0000 Subject: [PATCH 09/11] test(runtime): also require the displaced message to be collectable A fix that copies the displaced Event by value would free the Event but keep its Message and Raw payload reachable. --- debounce_test.go | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/debounce_test.go b/debounce_test.go index 436ef8b..44ad7cc 100644 --- a/debounce_test.go +++ b/debounce_test.go @@ -595,14 +595,16 @@ func TestDebounceDeliveryDelayedInValidationCannotDisplaceNewerWaiter(t *testing type weakEventAdapter struct { *fakeAdapter - mu sync.Mutex - events map[string]weak.Pointer[chat.Event] + mu sync.Mutex + events map[string]weak.Pointer[chat.Event] + messages map[string]weak.Pointer[chat.Message] } func (a *weakEventAdapter) Webhook(dispatch chat.DispatchFunc) http.Handler { return a.fakeAdapter.Webhook(func(ctx context.Context, ev *chat.Event) error { a.mu.Lock() a.events[ev.ID] = weak.Make(ev) + a.messages[ev.ID] = weak.Make(ev.Message) a.mu.Unlock() return dispatch(ctx, ev) }) @@ -611,8 +613,8 @@ func (a *weakEventAdapter) Webhook(dispatch chat.DispatchFunc) http.Handler { func (a *weakEventAdapter) collected(id string) bool { a.mu.Lock() defer a.mu.Unlock() - ref, ok := a.events[id] - return ok && ref.Value() == nil + event, ok := a.events[id] + return ok && event.Value() == nil && a.messages[id].Value() == nil } func TestDebounceNewerTailDoesNotRetainTheDisplacedEvent(t *testing.T) { @@ -620,7 +622,11 @@ func TestDebounceNewerTailDoesNotRetainTheDisplacedEvent(t *testing.T) { const aFirstPoll = 1 state := newGatedAcquireState(aFirstPoll) - adapter := &weakEventAdapter{fakeAdapter: newFakeAdapter("fake"), events: map[string]weak.Pointer[chat.Event]{}} + adapter := &weakEventAdapter{ + fakeAdapter: newFakeAdapter("fake"), + events: map[string]weak.Pointer[chat.Event]{}, + messages: map[string]weak.Pointer[chat.Message]{}, + } observer := &recordingObserver{} bot, err := chat.New(context.Background(), chat.WithState(state), From 47f376216d29f4c1dde3640f157c84af2dacfaad Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 10:06:22 +0000 Subject: [PATCH 10/11] fix(runtime): keep only the log fields of a displaced waiter work.supersedes now holds a supersession with the displaced event's adapter, ID, and thread ID instead of the *Event. The newer tail no longer keeps the displaced payload reachable after the displaced tail frees its admission slot, so ADR 0015 invariant 1 holds again. --- runtime.go | 30 +++++++++++++++++++++--------- 1 file changed, 21 insertions(+), 9 deletions(-) diff --git a/runtime.go b/runtime.go index 75a2657..24074d6 100644 --- a/runtime.go +++ b/runtime.go @@ -760,7 +760,7 @@ type preludeWork struct { // debounce quiet-period wait exits promptly on it instead of parking through // its full interval. displaced <-chan struct{} - supersedes *Event + supersedes *supersession // debounce is true when the tail must hold the event through the // DebounceInterval quiet period before waiting for the lock. debounce bool @@ -1433,7 +1433,7 @@ func (c *Chat) leaveQueueScope(entry *queueEntry) { // superseded. Under queue it also reports registered=false when the scope's // latest turn was admitted after it. The returned channel closes when this // registration is itself superseded. -func (c *Chat) registerPending(scope string, event *Event, seq uint64) (displaced <-chan struct{}, supersedes *Event, registered bool) { +func (c *Chat) registerPending(scope string, event *Event, seq uint64) (displaced <-chan struct{}, supersedes *supersession, registered bool) { c.queueMu.Lock() defer c.queueMu.Unlock() if turn := c.queueTurns[scope]; turn != nil && turn.won > seq { @@ -1444,7 +1444,7 @@ func (c *Chat) registerPending(scope string, event *Event, seq uint64) (displace return nil, nil, false } close(prev.displaced) - supersedes = prev.event + supersedes = newSupersession(prev.event) } waiter := &pendingWaiter{event: event, seq: seq, displaced: make(chan struct{})} c.pending[scope] = waiter @@ -1455,7 +1455,7 @@ func (c *Chat) registerPending(scope string, event *Event, seq uint64) (displace // scope's turn: it supersedes an older pending waiter and makes every delivery // admitted before event stale. It reports took=false, and takes no turn, when a // newer delivery already took one. -func (c *Chat) takeQueueTurn(scope string, event *Event, seq uint64) (supersedes *Event, took bool) { +func (c *Chat) takeQueueTurn(scope string, event *Event, seq uint64) (supersedes *supersession, took bool) { c.queueMu.Lock() defer c.queueMu.Unlock() turn := c.queueTurns[scope] @@ -1471,16 +1471,28 @@ func (c *Chat) takeQueueTurn(scope string, event *Event, seq uint64) (supersedes assert(prev.seq != seq, "a delivery that holds its prelude lease is never its scope's pending waiter") delete(c.pending, scope) close(prev.displaced) - return prev.event, true + return newSupersession(prev.event), true +} + +// supersession copies the log fields of a displaced waiter's event, so the +// newer delivery's tail does not keep that event reachable. +type supersession struct { + adapter string + eventID string + threadID ThreadID +} + +func newSupersession(displaced *Event) *supersession { + return &supersession{adapter: displaced.Adapter, eventID: displaced.ID, threadID: displaced.ThreadID} } func (c *Chat) logSupersession(work preludeWork) { - prev := work.supersedes - if prev == nil { + s := work.supersedes + if s == nil { return } - c.logger.Info("chat "+work.waitLabel+" superseded", "adapter", prev.Adapter, - "event_id", prev.ID, "thread_id", prev.ThreadID, "superseded_by", work.event.ID) + c.logger.Info("chat "+work.waitLabel+" superseded", "adapter", s.adapter, + "event_id", s.eventID, "thread_id", s.threadID, "superseded_by", work.event.ID) } // isPending reports whether event is still its scope's most-recent pending From 07da16fc05c2ffe466036035444bdbf43b53a708 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 10:11:33 +0000 Subject: [PATCH 11/11] docs(runtime): correct the supersede comments and extend ADR 0015 rule 4 The waiter comments said registerPending writes the superseded_by record. The displacing delivery's dispatcher writes it now. ADR 0015 invariant 4 gains the triage rule that a slog handler panic cannot drop an event the Concurrency Strategy keeps. --- docs/adr/0015-runtime-coordination.md | 2 +- runtime.go | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/adr/0015-runtime-coordination.md b/docs/adr/0015-runtime-coordination.md index 5287759..1f1d9d1 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 lock, slot, and record the delivery holds, ends its dispatch span with the `error` outcome, 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, ends its dispatch span with the `error` outcome, and does not recover the panic. A panic in the app's `slog` handler also cannot drop an event that the **Concurrency Strategy** keeps. Per-strategy invariant under `DispatchDeferred` (one line each): diff --git a/runtime.go b/runtime.go index 24074d6..ef2a533 100644 --- a/runtime.go +++ b/runtime.go @@ -1222,7 +1222,7 @@ func (c *Chat) waitDebounceQuietPeriod(ctx context.Context, work preludeWork) bo c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) return false case <-work.displaced: - // registerPending already emitted the superseded_by record. + // The displacing delivery writes the superseded_by record. c.logger.Debug("chat debounce waiter superseded", "adapter", work.event.Adapter, "event_id", work.event.ID, "thread_id", work.event.ThreadID) c.safeEnd(work.span, OutcomeIgnored, RouteAttr(work.route)) return false @@ -1528,7 +1528,7 @@ func (c *Chat) queueForLock(ctx context.Context, scope string, event *Event, lab return LockLease{}, acquireSuperseded } case acquireSuperseded: - // registerPending already emitted the superseded_by record; this waiter + // The displacing delivery writes the superseded_by record. This waiter // just stops. c.logger.Debug("chat "+label+" waiter superseded", "adapter", event.Adapter, "event_id", event.ID, "thread_id", event.ThreadID) case acquireAbandoned: