diff --git a/internal/execution/fanout.go b/internal/execution/fanout.go index 3d3763dad..9982cb976 100644 --- a/internal/execution/fanout.go +++ b/internal/execution/fanout.go @@ -3,9 +3,13 @@ package execution import ( "context" + "fmt" "os" + "runtime" "strconv" "sync" + + "github.com/LeanerCloud/CUDly/pkg/logging" ) // ConcurrencyFromEnv reads the CUDLY_MAX_ACCOUNT_PARALLELISM env var and @@ -69,6 +73,25 @@ func FanOutWithConcurrency[T any]( go func(idx int, accountID string) { defer wg.Done() defer func() { <-sem }() // release slot + // recover() guard so a panic inside fn (nil deref, type + // assertion failure, slice OOB, etc.) doesn't crash the + // whole Lambda process and strand the purchase execution + // at 'approved' with no way to debug post-mortem (issue + // #669). Surface the panic as an Err on the result slot + // so the parent aggregator records it on the execution + // row exactly like a regular fn-returned error, and log + // the goroutine stack at Error level for diagnosis. + defer func() { + if r := recover(); r != nil { + buf := make([]byte, 4096) + n := runtime.Stack(buf, false) + logging.Errorf("fan-out goroutine panic (account=%s): %v\n%s", accountID, r, buf[:n]) + results[idx] = Result[T]{ + AccountID: accountID, + Err: fmt.Errorf("panic during fan-out (account=%s): %v", accountID, r), + } + } + }() val, err := fn(ctx, accountID) results[idx] = Result[T]{AccountID: accountID, Value: val, Err: err} }(i, id) diff --git a/internal/execution/fanout_test.go b/internal/execution/fanout_test.go index 993473957..485888797 100644 --- a/internal/execution/fanout_test.go +++ b/internal/execution/fanout_test.go @@ -81,3 +81,37 @@ func TestPartition(t *testing.T) { require.Len(t, failures, 1) assert.Equal(t, "b", failures[0].AccountID) } + +// TestFanOut_PanicInFn asserts that a panic inside the per-item function is +// caught by the goroutine's deferred recover, converted to an Err on that +// item's result slot, and does NOT propagate up and crash the whole process +// (which would strand the surrounding purchase execution at 'approved' and +// terminate the Lambda invocation abnormally — see #669). +func TestFanOut_PanicInFn(t *testing.T) { + ids := []string{"a", "panic-me", "c"} + results := FanOut(context.Background(), ids, func(ctx context.Context, id string) (string, error) { + if id == "panic-me" { + panic("synthetic panic for test") + } + return "ok:" + id, nil + }) + + require.Len(t, results, 3) + // Map by AccountID since FanOut order is non-deterministic. + byID := make(map[string]Result[string], len(results)) + for _, r := range results { + byID[r.AccountID] = r + } + + // Non-panicking items still succeed. + assert.NoError(t, byID["a"].Err) + assert.Equal(t, "ok:a", byID["a"].Value) + assert.NoError(t, byID["c"].Err) + assert.Equal(t, "ok:c", byID["c"].Value) + + // The panicking item is surfaced as an Err containing the panic value. + require.Error(t, byID["panic-me"].Err) + assert.Contains(t, byID["panic-me"].Err.Error(), "panic during fan-out") + assert.Contains(t, byID["panic-me"].Err.Error(), "synthetic panic for test") + assert.Contains(t, byID["panic-me"].Err.Error(), "panic-me") +}