fix(batch): panic recovery, deadlock-safe error sends, bounded prealloc, Done() race

batch/batch.go

@@ -3,6 +3,7 @@ package batch
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 )
@@ -331,10 +332,37 @@ func (b *Batch[T]) Go(ctx context.Context, s Source[T], procs ...Processor[T]) (
 //		fmt.Println("Timed out waiting for processing to finish")
 //	}
 func (b *Batch[T]) Done() <-chan struct{} {
-	if b.done == nil {
+	// Guard the read of b.done with b.mu: Go assigns b.done while holding the
+	// lock, so reading it unlocked is a data race.
+	b.mu.Lock()
+	done := b.done
+	b.mu.Unlock()
+
+	if done == nil {
 		return closedDone
 	}
-	return b.done
+	return done
+}
+
+// sendErr forwards err to the error channel without risking a permanent block:
+// if the buffer is full and nobody is draining it, a canceled context frees the
+// sender. It reports whether the error was delivered.
+//
+// The non-blocking attempt comes first so that a canceled context never
+// preempts a send that would succeed immediately — cancellation only escapes
+// a send that would actually block.
+func (b *Batch[T]) sendErr(ctx context.Context, err error) bool {
+	select {
+	case b.errs <- err:
+		return true
+	default:
+	}
+	select {
+	case b.errs <- err:
+		return true
+	case <-ctx.Done():
+		return false
+	}
 }
 
 // doReader reads items from the Source and forwards them to the batch processor.
@@ -349,9 +377,11 @@ func (b *Batch[T]) doReader(ctx context.Context) {
 	// Get channels from source
 	out, errs := b.src.Read(ctx)
 
-	// Handle nil channels from source - just report an error and finish
+	// Handle nil channels from source - just report an error and finish.
+	// The send to b.errs is context-aware so a cancelled context cannot wedge
+	// the reader if the error buffer is full and nobody is draining it.
 	if out == nil || errs == nil {
-		b.errs <- errors.New("invalid source implementation: returned nil channel(s)")
+		b.sendErr(ctx, errors.New("invalid source implementation: returned nil channel(s)"))
 		close(b.items)
 		return
 	}
@@ -382,7 +412,10 @@ func (b *Batch[T]) doReader(ctx context.Context) {
 				errs = nil
 				continue
 			}
-			b.errs <- &SourceError{Err: err}
+			if !b.sendErr(ctx, &SourceError{Err: err}) {
+				close(b.items)
+				return
+			}
 		}
 	}
 
@@ -415,24 +448,33 @@ func (b *Batch[T]) doProcessors(ctx context.Context) {
 		wg.Add(1)
 		go func(items []*Item[T]) {
 			defer wg.Done()
-			for _, proc := range b.processors {
-				// Skip nil processors (although they should have been filtered out in Go)
-				if proc == nil {
-					continue
+			// Recover from panics in user processors so a single buggy
+			// Process call cannot crash the host process. Declared after
+			// wg.Done so (deferred-LIFO) recover runs first and wg.Done still
+			// fires, letting the pipeline complete. The panic is surfaced as a
+			// ProcessorError via a context-aware send.
+			//
+			// A completion flag detects the panic instead of recover()'s
+			// return value: under this module's Go 1.18 semantics recover()
+			// returns nil for panic(nil), which would otherwise be swallowed.
+			panicked := true
+			defer func() {
+				if !panicked {
+					return
 				}
-
+				r := recover()
 				var err error
-				items, err = proc.Process(ctx, items)
-				if err != nil {
-					b.errs <- &ProcessorError{Err: err}
-				}
-			}
-
-			for _, item := range items {
-				if item.Error != nil {
-					b.errs <- &ProcessorError{Err: item.Error}
+				// Preserve error identity for error-valued panics so
+				// errors.Is/errors.As reach the original through ProcessorError.
+				if e, ok := r.(error); ok {
+					err = fmt.Errorf("processor panic: %w", e)
+				} else {
+					err = fmt.Errorf("processor panic: %v", r)
 				}
-			}
+				b.sendErr(ctx, &ProcessorError{Err: err})
+			}()
+			b.processBatch(ctx, items)
+			panicked = false
 		}(batch)
 	}
 
@@ -446,6 +488,59 @@ func (b *Batch[T]) doProcessors(ctx context.Context) {
 	close(b.errs)
 }
 
+// processBatch runs one batch through the processor chain and forwards
+// per-item errors to the error channel. It returns early only when an error
+// send fails, which sendErr guarantees can happen solely in the wedged state
+// (error buffer full and context canceled) — continuing then would produce
+// more errors that cannot be delivered.
+func (b *Batch[T]) processBatch(ctx context.Context, items []*Item[T]) {
+	// Nil processors were filtered out in Go, so every proc is callable.
+	for _, proc := range b.processors {
+		var err error
+		items, err = proc.Process(ctx, items)
+		if err != nil {
+			if !b.sendErr(ctx, &ProcessorError{Err: err}) {
+				return
+			}
+		}
+	}
+
+	for _, item := range items {
+		if item.Error != nil {
+			if !b.sendErr(ctx, &ProcessorError{Err: item.Error}) {
+				return
+			}
+		}
+	}
+}
+
+// maxPreallocCap bounds the capacity that waitForItems will pre-allocate for a
+// batch slice. Without this bound, a very large MinItems (reachable, for
+// example, via DynamicConfig.UpdateBatchSize(huge, 0)) would be passed directly
+// to make, triggering a multi-terabyte allocation that crashes the process.
+// The slice still grows as needed via append; this only caps the initial hint.
+const maxPreallocCap = 4096
+
+// clampPreallocCap returns a sane, bounded capacity to use when pre-allocating
+// a batch slice for the given config values. It never returns more than
+// maxPreallocCap, and if maxItems is set it is treated as a hard upper bound on
+// the batch size (so the pre-allocation is not larger than the batch can grow).
+//
+// This guards against pathological configurations where minItems is enormous
+// while maxItems is unset, which would otherwise attempt an unbounded
+// allocation.
+func clampPreallocCap(minItems, maxItems uint64) int {
+	c := minItems
+	// If a maximum batch size is set, never pre-allocate beyond it.
+	if maxItems > 0 && maxItems < c {
+		c = maxItems
+	}
+	if c > maxPreallocCap {
+		c = maxPreallocCap
+	}
+	return int(c)
+}
+
 // fixConfig corrects invalid ConfigValues to ensure consistent batch behavior.
 //
 // It applies the following adjustments:
@@ -484,7 +579,7 @@ func fixConfig(c ConfigValues) ConfigValues {
 func (b *Batch[T]) waitForItems(_ context.Context, config ConfigValues) []*Item[T] {
 	var (
 		reachedMinTime bool
-		batch          = make([]*Item[T], 0, config.MinItems)
+		batch          = make([]*Item[T], 0, clampPreallocCap(config.MinItems, config.MaxItems))
 		minTimerCh     <-chan time.Time
 		maxTimerCh     <-chan time.Time
 		minTimer       *time.Timer

batch/batch_test.go

@@ -3,6 +3,7 @@ package batch_test
 import (
 	"context"
 	"errors"
+	"fmt"
 	"math/rand"
 	"sync"
 	"sync/atomic"
@@ -27,21 +28,53 @@ func TestBatch_ProcessorChainingAndErrorTracking(t *testing.T) {
 			t.Fatalf("Go returned unexpected error: %v", err)
 		}
 
-		received := 0
+		// errorPerItemProcessor fails items whose *per-batch local index* is a
+		// multiple of FailEvery (i%3 == 0), NOT their global ID. Which items
+		// fail therefore depends entirely on the batch boundaries, not on a
+		// fixed set of global indices.
+		//
+		// With MinItems=5, no MaxItems and no timers, batching is deterministic:
+		// the reader emits items in strict ID order and doProcessors collects
+		// one batch at a time, cutting the first batch the moment it reaches 5
+		// items. So the source of 9 items splits as:
+		//
+		//	batch 1: IDs [0 1 2 3 4]  -> local idx 0 and 3 fail -> IDs 0, 3
+		//	batch 2: IDs [5 6 7 8]    -> local idx 0 and 3 fail -> IDs 5, 8
+		//
+		// i.e. exactly 4 item errors, on IDs {0, 3, 5, 8}. (The error message
+		// from errorPerItemProcessor is "fail item <ID>", so we recover the ID
+		// from each error to assert which items failed, not merely how many.)
+		failedIDs := make(map[uint64]bool)
 		for err := range errs {
 			var processorError *ProcessorError
 			if !errors.As(err, &processorError) {
 				t.Errorf("unexpected error type: %v", err)
+				continue
 			}
-			received++
+			var id uint64
+			if _, scanErr := fmt.Sscanf(err.Error(), "processor error: fail item %d", &id); scanErr != nil {
+				t.Errorf("could not parse item ID from error %q: %v", err.Error(), scanErr)
+				continue
+			}
+			failedIDs[id] = true
+		}
+
+		wantFailed := map[uint64]bool{0: true, 3: true, 5: true, 8: true}
+		if len(failedIDs) != len(wantFailed) {
+			t.Errorf("expected %d item errors, got %d (IDs %v)", len(wantFailed), len(failedIDs), failedIDs)
 		}
-		// There are 9 items, items at indexes 0, 3, 6 (values 1, 4, 7) will fail (FailEvery=3)
-		// but it appears there is 1 more error that occurs during processing
-		if received != 4 {
-			t.Errorf("expected 4 item errors, got %d", received)
+		for id := range wantFailed {
+			if !failedIDs[id] {
+				t.Errorf("expected item ID %d to carry an error, but it did not (got %v)", id, failedIDs)
+			}
+		}
+		for id := range failedIDs {
+			if !wantFailed[id] {
+				t.Errorf("item ID %d carried an unexpected error (got %v)", id, failedIDs)
+			}
 		}
 
-		// All 9 items should be processed with the fix
+		// All 9 items should still be processed despite per-item errors.
 		if atomic.LoadUint32(&count) != 9 {
 			t.Errorf("expected 9 items processed, got %d", count)
 		}

batch/errors.go

@@ -17,6 +17,17 @@ var ErrBatchUsed = errors.New("batch: Batch is single-use; create a new Batch wi
 // ProcessorError is returned when a processor fails. It wraps the original
 // error from the processor to maintain the error chain while providing
 // context about the source of the error.
+//
+// The engine always wraps with the pointer form, &ProcessorError{}. When
+// checking errors with errors.As, callers must therefore use the pointer
+// target form:
+//
+//	if errors.As(err, new(*ProcessorError)) {
+//		// err came from a processor
+//	}
+//
+// The value-target form, errors.As(err, new(ProcessorError)), does not match,
+// because *ProcessorError is not assignable to ProcessorError.
 type ProcessorError struct {
 	// Err is the underlying error that occurred in the processor.
 	Err error
@@ -36,6 +47,16 @@ func (e ProcessorError) Unwrap() error {
 // SourceError is returned when a source fails. It wraps the original
 // error from the source to maintain the error chain while providing
 // context about the source of the error.
+//
+// The engine always wraps with the pointer form, &SourceError{}. When checking
+// errors with errors.As, callers must therefore use the pointer target form:
+//
+//	if errors.As(err, new(*SourceError)) {
+//		// err came from the source
+//	}
+//
+// The value-target form, errors.As(err, new(SourceError)), does not match,
+// because *SourceError is not assignable to SourceError.
 type SourceError struct {
 	// Err is the underlying error that occurred in the source.
 	Err error

batch/hardening_internal_test.go

@@ -0,0 +1,91 @@
+package batch
+
+import "testing"
+
+// TestClampPreallocCap verifies that the pre-allocation capacity helper returns
+// a sane, bounded value. A huge MinItems (reachable via DynamicConfig with
+// MaxItems==0) must never be used directly as a slice capacity, since that
+// would trigger a multi-TB allocation and crash the process.
+func TestClampPreallocCap(t *testing.T) {
+	tests := []struct {
+		name     string
+		minItems uint64
+		maxItems uint64
+		want     int
+	}{
+		{
+			name:     "small min, no max - exact",
+			minItems: 8,
+			maxItems: 0,
+			want:     8,
+		},
+		{
+			name:     "zero min, no max - zero",
+			minItems: 0,
+			maxItems: 0,
+			want:     0,
+		},
+		{
+			name:     "min at the cap boundary - exact",
+			minItems: maxPreallocCap,
+			maxItems: 0,
+			want:     maxPreallocCap,
+		},
+		{
+			name:     "huge min, no max - clamped to cap",
+			minItems: 1 << 40, // ~1 trillion: a real make() of this would OOM
+			maxItems: 0,
+			want:     maxPreallocCap,
+		},
+		{
+			name:     "max uint64 min, no max - clamped to cap",
+			minItems: ^uint64(0),
+			maxItems: 0,
+			want:     maxPreallocCap,
+		},
+		{
+			name:     "maxItems caps below minItems",
+			minItems: 1000,
+			maxItems: 16,
+			want:     16,
+		},
+		{
+			name:     "maxItems caps a huge minItems",
+			minItems: 1 << 40,
+			maxItems: 32,
+			want:     32,
+		},
+		{
+			// maxItems only ever caps downward; it must not inflate the
+			// pre-allocation. With a small minItems the batch starts small and
+			// grows via append, so we pre-allocate just minItems here.
+			name:     "huge maxItems does not inflate prealloc - uses min",
+			minItems: 10,
+			maxItems: 1 << 40,
+			want:     10,
+		},
+		{
+			name:     "both small, max above min - uses min",
+			minItems: 4,
+			maxItems: 64,
+			want:     4,
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			got := clampPreallocCap(tt.minItems, tt.maxItems)
+			if got != tt.want {
+				t.Errorf("clampPreallocCap(%d, %d) = %d, want %d",
+					tt.minItems, tt.maxItems, got, tt.want)
+			}
+			if got < 0 {
+				t.Errorf("clampPreallocCap returned negative capacity %d", got)
+			}
+			if got > maxPreallocCap {
+				t.Errorf("clampPreallocCap returned %d, which exceeds maxPreallocCap %d", got, maxPreallocCap)
+			}
+		})
+	}
+}