diff --git a/batch/batch.go b/batch/batch.go index 6e51a2c..0bd61c2 100644 --- a/batch/batch.go +++ b/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 diff --git a/batch/batch_test.go b/batch/batch_test.go index 467a6b1..c79317b 100644 --- a/batch/batch_test.go +++ b/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 ", 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) } diff --git a/batch/errors.go b/batch/errors.go index 77e1dfa..958fca2 100644 --- a/batch/errors.go +++ b/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 diff --git a/batch/hardening_internal_test.go b/batch/hardening_internal_test.go new file mode 100644 index 0000000..0d43169 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/batch/hardening_test.go b/batch/hardening_test.go new file mode 100644 index 0000000..e9bcf30 --- /dev/null +++ b/batch/hardening_test.go @@ -0,0 +1,580 @@ +package batch_test + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + "testing" + "time" + + . "github.com/MasterOfBinary/gobatch/batch" +) + +// manyErrorsSource emits N errors (and no data items). It is used to fill the +// error buffer so the pipeline wedges when nobody drains b.errs. +type manyErrorsSource struct { + N int +} + +func (s *manyErrorsSource) Read(ctx context.Context) (<-chan any, <-chan error) { + out := make(chan any) + errs := make(chan error) + go func() { + defer close(out) + defer close(errs) + for i := 0; i < s.N; i++ { + select { + case <-ctx.Done(): + return + case errs <- fmt.Errorf("err %d", i): + } + } + }() + return out, errs +} + +// TestCancelUnblocksWedgedReader verifies that cancelling the context unblocks a +// pipeline that has wedged because the error buffer filled and nobody is +// draining it. Before the fix the internal sends to b.errs were unguarded +// blocking sends, so cancel() did not free the reader and Done() never closed. +// +// The test is bounded by a timeout so the RED state fails fast instead of +// hanging the suite. +func TestCancelUnblocksWedgedReader(t *testing.T) { + // Tiny error buffer so it fills almost immediately, and a source that emits + // far more errors than the buffer can hold. + b := New[any](NewConstantConfig(&ConfigValues{})). + WithBufferConfig(BufferConfig{ErrorBufferSize: 1}) + + src := &manyErrorsSource{N: 1000} + + ctx, cancel := context.WithCancel(context.Background()) + + // Start processing but deliberately never drain the returned error channel. + _, err := b.Go(ctx, src) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } + + // Give the reader time to fill the 1-slot error buffer and wedge. + time.Sleep(50 * time.Millisecond) + + // Cancelling must break the pipeline out of the blocked send. + cancel() + + select { + case <-b.Done(): + // Pipeline completed after cancellation - correct. + case <-time.After(2 * time.Second): + t.Fatal("deadlock: Done() did not close within 2s after cancel; " + + "the wedged reader was not unblocked by context cancellation") + } +} + +// TestCancelUnblocksWedgedProcessor verifies the same guarantee for the +// per-batch processor goroutine: if a processor returns errors that cannot be +// delivered because the error buffer is full and nobody is draining, cancelling +// the context must let the goroutine exit so the pipeline can complete. +func TestCancelUnblocksWedgedProcessor(t *testing.T) { + b := New[any](NewConstantConfig(&ConfigValues{MinItems: 1})). + WithBufferConfig(BufferConfig{ErrorBufferSize: 1}) + + // A source that emits many data items so many batches are produced, each of + // which fails and tries to send a ProcessorError. + items := make([]any, 1000) + for i := range items { + items[i] = i + } + src := &sliceSource{items: items} + + procErr := errors.New("always fails") + proc := &alwaysErrProcessor{err: procErr} + + ctx, cancel := context.WithCancel(context.Background()) + + _, err := b.Go(ctx, src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } + + time.Sleep(50 * time.Millisecond) + cancel() + + select { + case <-b.Done(): + case <-time.After(2 * time.Second): + t.Fatal("deadlock: Done() did not close within 2s after cancel; " + + "the wedged processor goroutine was not unblocked by context cancellation") + } +} + +// TestCancelUnblocksWedgedItemErrorSend verifies the same guarantee for the +// per-item error forwarding loop: if items come back from the processors with +// their Error field set and the sends wedge on a full, undrained error buffer, +// cancelling the context must let the goroutine exit. +func TestCancelUnblocksWedgedItemErrorSend(t *testing.T) { + b := New[any](NewConstantConfig(&ConfigValues{MinItems: 1})). + WithBufferConfig(BufferConfig{ErrorBufferSize: 1}) + + items := make([]any, 1000) + for i := range items { + items[i] = i + } + src := &sliceSource{items: items} + + itemErr := errors.New("item failed") + proc := &markItemsErrProcessor{err: itemErr} + + ctx, cancel := context.WithCancel(context.Background()) + + _, err := b.Go(ctx, src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } + + time.Sleep(50 * time.Millisecond) + cancel() + + select { + case <-b.Done(): + case <-time.After(2 * time.Second): + t.Fatal("deadlock: Done() did not close within 2s after cancel; " + + "the wedged item-error send was not unblocked by context cancellation") + } +} + +// markItemsErrProcessor sets Error on every item and reports no stage error. +type markItemsErrProcessor struct { + err error +} + +func (p *markItemsErrProcessor) Process(ctx context.Context, items []*Item[any]) ([]*Item[any], error) { + for _, item := range items { + item.Error = p.err + } + return items, nil +} + +// alwaysErrProcessor returns a processor-wide error for every batch. +type alwaysErrProcessor struct { + err error +} + +func (p *alwaysErrProcessor) Process(ctx context.Context, items []*Item[any]) ([]*Item[any], error) { + return items, p.err +} + +// panicProcessor panics inside Process, simulating a buggy user processor. +type panicProcessor struct { + msg string +} + +func (p *panicProcessor) Process(ctx context.Context, items []*Item[any]) ([]*Item[any], error) { + panic(p.msg) +} + +// TestPanicInProcessorIsRecovered verifies that a panic in a user Processor does +// not crash the host process. Before the fix, the per-batch goroutine called +// proc.Process with no recover, so a panic took down the whole process. After +// the fix the panic is recovered, surfaced as a ProcessorError on the error +// channel, and the pipeline completes cleanly (errs and done close). +func TestPanicInProcessorIsRecovered(t *testing.T) { + b := New[any](NewConstantConfig(&ConfigValues{MinItems: 1})) + src := &testSource{Items: []any{1, 2, 3}} + proc := &panicProcessor{msg: "boom in processor"} + + errs, err := b.Go(context.Background(), src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } + + var sawPanicErr bool + for err := range errs { + var pe *ProcessorError + if errors.As(err, &pe) { + // The recovered panic must be surfaced and carry the panic value. + if containsStr(err.Error(), "processor panic") && + containsStr(err.Error(), "boom in processor") { + sawPanicErr = true + } + } + } + + // The pipeline must complete rather than crash or hang. + select { + case <-b.Done(): + case <-time.After(2 * time.Second): + t.Fatal("Done() did not close within 2s after a panicking processor") + } + + if !sawPanicErr { + t.Fatal("expected a ProcessorError describing the recovered panic") + } +} + +// TestMaxTimeMultipleIdleCyclesThenLateItem exercises several idle MaxTime +// cycles (the batch stays empty past MaxTime more than once) and then delivers +// a late item, asserting it is still processed. This locks in the re-arm +// behavior so the timer refactor (single *time.Timer with Reset instead of a +// fresh time.After each idle fire) cannot regress it. +func TestMaxTimeMultipleIdleCyclesThenLateItem(t *testing.T) { + cfg := NewConstantConfig(&ConfigValues{ + MinItems: 10, // high, so MinItems alone never triggers + MaxTime: 50 * time.Millisecond, // fires repeatedly while idle + }) + + b := New[any](cfg) + + input := make(chan any) + src := &chanSource{in: input} + + var processed int32 + proc := &countingProc{n: &processed} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + errs, err := b.Go(ctx, src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } + go func() { + for range errs { + } + }() + + // Let several idle MaxTime cycles elapse (4 x 50ms = 200ms+). + time.Sleep(220 * time.Millisecond) + + // Now deliver a late item; it must still be picked up and processed. + input <- 42 + + // Give the late item time to be batched (within one MaxTime cycle) and run. + deadline := time.After(2 * time.Second) + for atomic.LoadInt32(&processed) == 0 { + select { + case <-deadline: + t.Fatal("late item was not processed after multiple idle MaxTime cycles") + case <-time.After(10 * time.Millisecond): + } + } + + close(input) + select { + case <-b.Done(): + case <-time.After(2 * time.Second): + t.Fatal("Done() did not close within 2s") + } + + if got := atomic.LoadInt32(&processed); got != 1 { + t.Errorf("expected exactly 1 item processed, got %d", got) + } +} + +// chanSource adapts a caller-owned input channel into a Source, respecting +// context cancellation. +type chanSource struct { + in chan any +} + +func (s *chanSource) Read(ctx context.Context) (<-chan any, <-chan error) { + out := make(chan any) + errs := make(chan error) + go func() { + defer close(out) + defer close(errs) + for { + select { + case <-ctx.Done(): + return + case v, ok := <-s.in: + if !ok { + return + } + select { + case <-ctx.Done(): + return + case out <- v: + } + } + } + }() + return out, errs +} + +// countingProc atomically counts the items it processes. +type countingProc struct { + n *int32 +} + +func (p *countingProc) Process(ctx context.Context, items []*Item[any]) ([]*Item[any], error) { + atomic.AddInt32(p.n, int32(len(items))) + return items, nil +} + +// TestDoneRace verifies there is no data race between Done() reading b.done and +// Go() writing it. Done() previously read b.done without holding b.mu, while Go +// writes it under the lock. Run with -race to surface the report (RED) before +// the fix; after the fix it must be clean. +func TestDoneRace(t *testing.T) { + b := New[any](NewConstantConfig(&ConfigValues{})) + src := &testSource{Items: []any{1, 2, 3}} + + stop := make(chan struct{}) + done := make(chan struct{}) + + // Hammer Done() concurrently with Go(). + go func() { + defer close(done) + for { + select { + case <-stop: + return + default: + _ = b.Done() + } + } + }() + + // Let the reader goroutine spin up and start calling Done(). + time.Sleep(5 * time.Millisecond) + + errs, err := b.Go(context.Background(), src) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } + go func() { + for range errs { + } + }() + + <-b.Done() + close(stop) + <-done +} + +// errAfterCancelProcessor waits for the context to be canceled, then marks +// every item with err. It forces all downstream error sends to happen with an +// already-canceled context. +type errAfterCancelProcessor struct { + err error +} + +func (p *errAfterCancelProcessor) Process(ctx context.Context, items []*Item[any]) ([]*Item[any], error) { + <-ctx.Done() + for _, item := range items { + item.Error = p.err + } + return items, nil +} + +// TestCanceledContextStillDeliversErrorsWithBufferRoom verifies that a +// canceled context does not preempt error sends that would succeed +// immediately. Every item error produced after cancellation must still reach +// the error channel when the buffer has room: cancellation may only escape a +// send that would actually block. +func TestCanceledContextStillDeliversErrorsWithBufferRoom(t *testing.T) { + const n = 50 // well under the default error buffer of 100 + + b := New[any](NewConstantConfig(&ConfigValues{MinItems: n, MaxItems: n})) + + items := make([]any, n) + for i := range items { + items[i] = i + } + src := &sliceSource{items: items} + + itemErr := errors.New("item failed after cancel") + proc := &errAfterCancelProcessor{err: itemErr} + + ctx, cancel := context.WithCancel(context.Background()) + + errs, err := b.Go(ctx, src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } + + // Let the batch assemble and enter the processor, then cancel. The + // processor only marks errors after cancellation, so every send happens + // on a canceled context. + time.Sleep(50 * time.Millisecond) + cancel() + + var got int + for err := range errs { + if containsStr(err.Error(), "item failed after cancel") { + got++ + } + } + if got != n { + t.Errorf("expected all %d item errors delivered after cancel, got %d", n, got) + } +} + +// TestBlockedErrorSendDeliversToSlowConsumer verifies that an error send that +// finds the buffer full blocks and then delivers once the consumer drains a +// slot, rather than dropping — with a live (uncanceled) context, every error +// must arrive no matter how slow the consumer is. +func TestBlockedErrorSendDeliversToSlowConsumer(t *testing.T) { + const n = 3 + + b := New[any](NewConstantConfig(&ConfigValues{MinItems: n, MaxItems: n})). + WithBufferConfig(BufferConfig{ErrorBufferSize: 1}) + + items := make([]any, n) + for i := range items { + items[i] = i + } + src := &sliceSource{items: items} + + itemErr := errors.New("slow consumer error") + proc := &markItemsErrProcessor{err: itemErr} + + errs, err := b.Go(context.Background(), src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } + + // Drain slowly: the 1-slot buffer fills on the first send, so later sends + // must block until a receive frees the slot, then still deliver. + var got int + for err := range errs { + if containsStr(err.Error(), "slow consumer error") { + got++ + } + time.Sleep(50 * time.Millisecond) + } + if got != n { + t.Errorf("expected all %d errors delivered to a slow consumer, got %d", n, got) + } +} + +// nilPanicProcessor panics with a nil value. Under the module's Go 1.18 +// semantics recover() returns nil for panic(nil). +type nilPanicProcessor struct{} + +func (nilPanicProcessor) Process(context.Context, []*Item[any]) ([]*Item[any], error) { + panic(nil) +} + +// TestPanicNilIsReported verifies that panic(nil) in a user processor is still +// surfaced as a ProcessorError. recover() returns nil for panic(nil) under the +// module's Go 1.18 semantics, so the recovery path must not key off the +// recovered value being non-nil. +func TestPanicNilIsReported(t *testing.T) { + b := New[any](NewConstantConfig(&ConfigValues{MinItems: 1})) + src := &testSource{Items: []any{1}} + + errs, err := b.Go(context.Background(), src, nilPanicProcessor{}) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } + + var sawPanicErr bool + for err := range errs { + if containsStr(err.Error(), "processor panic") { + sawPanicErr = true + } + } + if !sawPanicErr { + t.Fatal("expected a ProcessorError for panic(nil); got none") + } + + select { + case <-b.Done(): + case <-time.After(2 * time.Second): + t.Fatal("Done() did not close within 2s after panic(nil)") + } +} + +// errPanicProcessor panics with an error value. +type errPanicProcessor struct { + err error +} + +func (p errPanicProcessor) Process(context.Context, []*Item[any]) ([]*Item[any], error) { + panic(p.err) +} + +// TestPanicErrorPreservesIdentity verifies that an error-valued panic keeps +// its identity through ProcessorError, so errors.Is reaches the original +// panic value. +func TestPanicErrorPreservesIdentity(t *testing.T) { + sentinel := errors.New("sentinel panic error") + + b := New[any](NewConstantConfig(&ConfigValues{MinItems: 1})) + src := &testSource{Items: []any{1}} + + errs, err := b.Go(context.Background(), src, errPanicProcessor{err: sentinel}) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } + + var sawSentinel bool + for err := range errs { + if errors.Is(err, sentinel) { + sawSentinel = true + } + } + if !sawSentinel { + t.Fatal("errors.Is could not reach the error-valued panic payload through ProcessorError") + } +} + +// nilChannelSource models a broken Source that returns nil channels from Read. +type nilChannelSource struct{} + +func (nilChannelSource) Read(context.Context) (<-chan any, <-chan error) { + return nil, nil +} + +// TestNilChannelSourceReportsError verifies that a Source returning nil +// channels surfaces an error on the error channel and the pipeline still +// completes instead of hanging. +func TestNilChannelSourceReportsError(t *testing.T) { + b := New[any](NewConstantConfig(&ConfigValues{})) + + errs, err := b.Go(context.Background(), nilChannelSource{}) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } + + var sawInvalidSource bool + for err := range errs { + if containsStr(err.Error(), "invalid source implementation") { + sawInvalidSource = true + } + } + if !sawInvalidSource { + t.Fatal("expected an error reporting the nil source channels") + } + + select { + case <-b.Done(): + case <-time.After(2 * time.Second): + t.Fatal("Done() did not close within 2s for a nil-channel source") + } +} + +// TestSourceErrorMessage covers SourceError.Error formatting. +func TestSourceErrorMessage(t *testing.T) { + err := &SourceError{Err: errors.New("boom")} + const want = "source error: boom" + if got := err.Error(); got != want { + t.Errorf("SourceError.Error() = %q, want %q", got, want) + } +} + +// containsStr is a tiny substring helper to avoid importing strings here. +func containsStr(haystack, needle string) bool { + if len(needle) == 0 { + return true + } + for i := 0; i+len(needle) <= len(haystack); i++ { + if haystack[i:i+len(needle)] == needle { + return true + } + } + return false +}