fix(runtime): keep the newer event when the supersede log panics

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,91 @@ 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]
+	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)
+	})
+}
+
+func (a *weakEventAdapter) collected(id string) bool {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	event, ok := a.events[id]
+	return ok && event.Value() == nil && a.messages[id].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]{},
+		messages:    map[string]weak.Pointer[chat.Message]{},
+	}
+	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()
+}

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. 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):
 

prelude_panic_test.go

@@ -9,6 +9,7 @@ import (
 	"log/slog"
 	"net/http"
 	"net/http/httptest"
+	"slices"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -332,3 +333,230 @@ func TestBurstMemberLoggerPanicStillRunsItsScope(t *testing.T) {
 		t.Fatalf("Shutdown after the logger panicked: %v", err)
 	}
 }
+
+func TestPreludePanicEndsItsDispatchSpan(t *testing.T) {
+	t.Parallel()
+
+	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)
+				}
+			})
+		}
+	}
+}
+
+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)
+		})
+	}
+}