diff --git a/burst.go b/burst.go index db7edf8..d7111fe 100644 --- a/burst.go +++ b/burst.go @@ -95,10 +95,10 @@ 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) + 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 diff --git a/docs/adr/0015-runtime-coordination.md b/docs/adr/0015-runtime-coordination.md index c9ccdcb..d2dbc22 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 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 new file mode 100644 index 0000000..6ff1aef --- /dev/null +++ b/prelude_panic_test.go @@ -0,0 +1,334 @@ +package chat_test + +import ( + "bytes" + "context" + "encoding/json" + "io" + "log" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "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) +} + +func (s *firstMarkPanicState) lockFree() bool { + s.fakeState.mu.Lock() + defer s.fakeState.mu.Unlock() + return len(s.locked) == 0 +} + +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) + openGate := sync.OnceFunc(func() { close(state.release) }) + t.Cleanup(openGate) + 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") + openGate() + + 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() + + state := &firstMarkPanicState{fakeState: newFakeState()} + observer := &recordingObserver{} + bot := newPanicRuntime(t, panicRuntimeOptions(chat.ConcurrencyQueue, tc.mode), + chat.WithState(state), chat.WithObserver(observer)) + 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") + 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", + }) + 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) + 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") + 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) + } + 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") + }) + } +} + +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", + }) + 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) + } +} diff --git a/runtime.go b/runtime.go index 7fd1ccc..1b8c0a0 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,7 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que c.safeEnd(span, OutcomeDuplicate) return preludeWork{}, true, err } + defer c.abandonEventAcceptance(event.ID, acceptance) finish := func(err error) error { c.finishEventAcceptance(event.ID, acceptance, err) return err @@ -880,7 +883,16 @@ func (c *Chat) prelude(ctx context.Context, event *Event, seq uint64, entry *que if lease == (LockLease{}) { return } - c.releaseLock(ctx, lease, event.ThreadID) + held := lease + lease = LockLease{} + c.releaseLock(ctx, held, event.ThreadID) + } + defer releaseOnResolve() + handLeaseToTail := 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) { @@ -925,9 +937,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 handLeaseToTail(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) { @@ -940,9 +952,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 handLeaseToTail(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 { @@ -961,9 +973,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 handLeaseToTail(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 @@ -1448,8 +1460,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 } @@ -1623,6 +1635,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: