fix(execution): recover from panics inside FanOutWithConcurrency goroutines (closes #669)

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)

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")
+}