From f8b1c27b4162585d94f664d6d989efc5b895de09 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Wed, 5 Aug 2026 14:15:31 -0700 Subject: [PATCH] recover panics in firehose and queue processing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Defense-in-depth: the firehose consumer and the task queue workers each run in a long-lived goroutine (under an errgroup), so a panic in any one event handler or task processor propagates up, crashes that goroutine, and — because errgroup cancels the group context on any return — tears down event processing or the whole queue for the node. A single bad record (malformed CBOR, a nil-deref in a codec) should not be able to take the node down. Add a shared log.Recover helper (and log.RecoverVoid) that wraps a call, recovers any panic, logs it at error level with the captured stack, and returns it as a *PanicError. Wire it in at the two per-unit boundaries: - queue_processor.go: wrap each processTask call in runQueueWorker so a panicking task is logged and dropped (its existing lease-expiry + max_tries retry semantics still apply) rather than killing the worker. - firehose.go: wrap the handleCreateUpdate call in the create/update branch of the event loop so a panicking record is logged and the loop moves to the next op rather than killing the firehose consumer. Co-Authored-By: Claude Opus 4.8 --- pkg/atproto/firehose.go | 10 ++++- pkg/log/recover.go | 75 ++++++++++++++++++++++++++++++++++ pkg/log/recover_test.go | 72 ++++++++++++++++++++++++++++++++ pkg/statedb/queue_processor.go | 11 ++++- 4 files changed, 165 insertions(+), 3 deletions(-) create mode 100644 pkg/log/recover.go create mode 100644 pkg/log/recover_test.go diff --git a/pkg/atproto/firehose.go b/pkg/atproto/firehose.go index 9cf62bf27..08b29dfcf 100644 --- a/pkg/atproto/firehose.go +++ b/pkg/atproto/firehose.go @@ -665,7 +665,15 @@ func (atsync *ATProtoSynchronizer) handleIndexedOps(ctx context.Context, evt *in break } - err = atsync.handleCreateUpdate(ctx, evt.Repo, rkey, recCBOR, op.Cid.String(), collection, ek == repomgr.EvtKindUpdateRecord, false) + // Defense-in-depth: handleCreateUpdate decodes arbitrary CBOR + // and dispatches across every record type, so a malformed record + // or a bug in any one handler can panic. Recover it so one bad + // event can't crash the firehose consumer goroutine (which would + // tear down event processing for the whole node). The error is + // logged and the loop moves on to the next op. + err = log.Recover(ctx, func() error { + return atsync.handleCreateUpdate(ctx, evt.Repo, rkey, recCBOR, op.Cid.String(), collection, ek == repomgr.EvtKindUpdateRecord, false) + }) if err != nil { log.Error(ctx, "failed to handle create update", "err", err) continue diff --git a/pkg/log/recover.go b/pkg/log/recover.go new file mode 100644 index 000000000..c29437c44 --- /dev/null +++ b/pkg/log/recover.go @@ -0,0 +1,75 @@ +package log + +import ( + "context" + "fmt" + "runtime/debug" +) + +// Recover wraps a function call so that a panic is recovered, logged at error +// level with the provided context, and returned as an error rather than +// propagating and crashing the calling goroutine. +// +// This is defense-in-depth for long-lived goroutines (the firehose event loop, +// the task queue workers) so that one bad event or task can't take down the +// whole node. Callers that already return an error should prefer wrapping +// their body in this to convert panics into logged errors; callers that don't +// care about the returned error can ignore it. +// +// The recovered value is returned as a PanicError so callers can wrap/join it +// like any error. The goroutine's stack is captured and logged to aid triage. +func Recover(ctx context.Context, fn func() error) (err error) { + defer func() { + r := recover() + if r == nil { + return + } + stack := debug.Stack() + Error(ctx, "recovered panic", "panic", r, "stack", string(stack)) + err = &PanicError{Recovered: r, Stack: stack} + }() + return fn() +} + +// RecoverVoid is like Recover but for functions that do not return an error. +// It recovers and logs a panic without propagating it, so a panicking side +// effect can't crash the calling goroutine. Use it when there is nothing +// meaningful to do with a returned error anyway. +func RecoverVoid(ctx context.Context, fn func()) { + defer func() { + r := recover() + if r == nil { + return + } + stack := debug.Stack() + Error(ctx, "recovered panic", "panic", r, "stack", string(stack)) + }() + fn() +} + +// PanicError wraps a value recovered from a panic. It implements the error +// interface so it can flow through normal error-handling paths. +type PanicError struct { + Recovered any + Stack []byte +} + +func (e *PanicError) Error() string { + return "panic recovered: " + asString(e.Recovered) +} + +// Unwrap is intentionally a no-op; a panic has no sentinel error to unwrap +// into, but defining it keeps linters quiet about the field. +func (e *PanicError) Unwrap() error { return nil } + +// asString stringifies the recovered value, tolerating non-string panic +// values (e.g. runtime errors, ints). +func asString(v any) string { + if s, ok := v.(string); ok { + return s + } + if e, ok := v.(error); ok { + return e.Error() + } + return fmt.Sprintf("%v", v) +} diff --git a/pkg/log/recover_test.go b/pkg/log/recover_test.go new file mode 100644 index 000000000..5be6400f1 --- /dev/null +++ b/pkg/log/recover_test.go @@ -0,0 +1,72 @@ +package log + +import ( + "context" + "errors" + "testing" +) + +func TestRecoverReturnsErrorFromPanic(t *testing.T) { + ctx := context.Background() + err := Recover(ctx, func() error { + panic("boom") + }) + if err == nil { + t.Fatal("expected an error from a panicked call, got nil") + } + var pe *PanicError + if !errors.As(err, &pe) { + t.Fatalf("expected a *PanicError, got %T: %v", err, err) + } + if pe.Recovered != "boom" { + t.Fatalf("expected Recovered to be %q, got %v", "boom", pe.Recovered) + } + if len(pe.Stack) == 0 { + t.Fatal("expected a non-empty captured stack") + } + if got := err.Error(); got != "panic recovered: boom" { + t.Fatalf("unexpected Error(): %q", got) + } +} + +func TestRecoverPassesThroughNormalError(t *testing.T) { + ctx := context.Background() + want := errors.New("ordinary failure") + err := Recover(ctx, func() error { + return want + }) + if !errors.Is(err, want) { + t.Fatalf("expected the wrapped normal error to be returned, got %v", err) + } +} + +func TestRecoverReturnsNilOnSuccess(t *testing.T) { + ctx := context.Background() + err := Recover(ctx, func() error { + return nil + }) + if err != nil { + t.Fatalf("expected nil on a successful call, got %v", err) + } +} + +func TestRecoverHandlesNonStringPanic(t *testing.T) { + ctx := context.Background() + err := Recover(ctx, func() error { + panic(42) + }) + if err == nil { + t.Fatal("expected an error from an int panic, got nil") + } + if got := err.Error(); got != "panic recovered: 42" { + t.Fatalf("expected stringified int panic, got %q", got) + } +} + +func TestRecoverVoidDoesNotPropagatePanic(t *testing.T) { + ctx := context.Background() + // If RecoverVoid fails to recover, the panic crashes the test process. + RecoverVoid(ctx, func() { + panic("must not escape") + }) +} diff --git a/pkg/statedb/queue_processor.go b/pkg/statedb/queue_processor.go index 1756f1d49..5810693d6 100644 --- a/pkg/statedb/queue_processor.go +++ b/pkg/statedb/queue_processor.go @@ -140,8 +140,15 @@ func (state *StatefulDB) runQueueWorker(ctx context.Context, workerID string, ta return err } if task != nil { - if err := state.processTask(ctx, task); err != nil { - log.Error(ctx, "failed to process task", "err", err, "worker", workerID) + // Defense-in-depth: a panic inside a task handler (e.g. a + // nil-deref in a codec, or a malformed record) must not crash + // this worker goroutine — errgroup would cancel the whole queue. + // Recover it into a logged error so the task fails and its retry + // semantics (lease expiry + max_tries) still apply. + if err := log.Recover(ctx, func() error { + return state.processTask(ctx, task) + }); err != nil { + log.Error(ctx, "failed to process task", "err", err, "worker", workerID, "taskId", fmt.Sprintf("%d", task.ID)) } continue } -- 2.51.2