diff --git a/.gitignore b/.gitignore index ae75693..13758d5 100644 --- a/.gitignore +++ b/.gitignore @@ -28,4 +28,6 @@ coverage.txt coverage.out # Gogland project files -.idea \ No newline at end of file +.idea +# Planning/codebase-mapping docs (local only) +.planning/ diff --git a/README.md b/README.md index 28f5fcb..56afdbc 100644 --- a/README.md +++ b/README.md @@ -124,6 +124,7 @@ package main import ( "context" "fmt" + "log" "time" "github.com/MasterOfBinary/gobatch/batch" @@ -164,7 +165,10 @@ func main() { ctx := context.Background() // Start batch processing with processors chained - errs := b.Go(ctx, src, doubleProc, printProc) + errs, err := b.Go(ctx, src, doubleProc, printProc) + if err != nil { + log.Fatal(err) + } // Ignore errors for this simple example batch.IgnoreErrors(errs) @@ -200,6 +204,8 @@ You can choose between: - **`ConstantConfig`** for static, unchanging settings. - **`DynamicConfig`** for runtime-adjustable settings that can be updated while processing. +Passing a `nil` `Config` to `New` (or using the zero-value `&Batch[T]{}`) uses a default configuration, where items are processed immediately as they are read. + Configuration options include: - `MinItems`: Minimum number of items to process in a batch. @@ -255,7 +261,6 @@ You can fine-tune the performance by customizing the internal channel buffer siz // Configure custom buffer sizes batchProcessor := batch.New[int](config).WithBufferConfig(batch.BufferConfig{ ItemBufferSize: 1000, // Buffer for incoming items - IDBufferSize: 1000, // Buffer for ID generation ErrorBufferSize: 500, // Buffer for error reporting }) ``` @@ -297,11 +302,15 @@ go func() { Or using helper functions: ```go -// Collect all errors -errs := batch.CollectErrors(batchProcessor.Go(ctx, source, processor)) -<-batchProcessor.Done() +// Collect all errors (blocks until processing completes) +pipeErrs, err := batchProcessor.Go(ctx, source, processor) +if err != nil { + // Handle start error (e.g. batch.ErrNilSource, batch.ErrBatchUsed) + log.Fatal(err) +} +errs := batch.CollectErrors(pipeErrs) -// Or use the RunBatchAndWait helper +// Or use the RunBatchAndWait helper, which folds a start error into the slice errs := batch.RunBatchAndWait(ctx, batchProcessor, source, processor) for _, err := range errs { @@ -309,6 +318,14 @@ for _, err := range errs { } ``` +### Batch lifecycle + +A `Batch` is **single-use**: call `Go` exactly once per `Batch`. Calling `Go` +again returns `batch.ErrBatchUsed` (along with a closed, drainable error channel) +instead of starting a second run — create a fresh `Batch` with `New` for each +run. `Go` also returns `batch.ErrNilSource` when the source is nil. Both errors +are checkable with `errors.Is`. + ## Documentation See the [pkg.go.dev docs](https://pkg.go.dev/github.com/MasterOfBinary/gobatch) for documentation diff --git a/batch/batch.go b/batch/batch.go index 6b7cb46..6e51a2c 100644 --- a/batch/batch.go +++ b/batch/batch.go @@ -15,6 +15,16 @@ var closedDone = func() chan struct{} { return ch }() +// closedErrs is a pre-closed, empty error channel returned by Go when it rejects +// a call (a nil source or an already-used Batch). Returning a closed channel +// rather than nil keeps a caller that ranges over Go's first return value safe +// even when it ignores the returned error. +var closedErrs = func() chan error { + ch := make(chan error) + close(ch) + return ch +}() + // BufferConfig configures the internal buffer sizes used by Batch. // If not specified, default values are used. type BufferConfig struct { @@ -22,10 +32,6 @@ type BufferConfig struct { // Default: DefaultItemBufferSize ItemBufferSize int - // IDBufferSize is the buffer size for the ID generator channel. - // Default: DefaultIDBufferSize - IDBufferSize int - // ErrorBufferSize is the buffer size for the error channel. // Default: DefaultErrorBufferSize ErrorBufferSize int @@ -48,11 +54,15 @@ type BufferConfig struct { // // Batch runs asynchronously after Go is called. When processing is complete, // either the error channel returned from Go is closed, or the channel returned -// from Done is closed. +// from Done is closed. A Batch is single-use: create a new one with New for +// each run. // // A simple way to wait for completion while handling errors: // -// errs := b.Go(ctx, s, p) +// errs, err := b.Go(ctx, s, p) +// if err != nil { +// log.Fatal(err) +// } // for err := range errs { // log.Print(err.Error()) // } @@ -60,7 +70,11 @@ type BufferConfig struct { // // If errors don't need to be handled, IgnoreErrors can be used: // -// batch.IgnoreErrors(b.Go(ctx, s, p)) +// errs, err := b.Go(ctx, s, p) +// if err != nil { +// log.Fatal(err) +// } +// batch.IgnoreErrors(errs) // <-b.Done() // // Now batch processing is done // @@ -73,12 +87,11 @@ type Batch[T any] struct { src Source[T] processors []Processor[T] items chan *Item[T] - ids chan uint64 done chan struct{} - mu sync.Mutex - running bool - errs chan error + mu sync.Mutex + used bool + errs chan error } // New creates a new Batch using the provided config. If config is nil, @@ -100,7 +113,6 @@ func New[T any](config Config) *Batch[T] { // // b := batch.New[any](config).WithBufferConfig(batch.BufferConfig{ // ItemBufferSize: 1000, -// IDBufferSize: 1000, // ErrorBufferSize: 500, // }) // @@ -109,7 +121,7 @@ func (b *Batch[T]) WithBufferConfig(config BufferConfig) *Batch[T] { b.mu.Lock() defer b.mu.Unlock() - if b.running { + if b.used { panic("batch: WithBufferConfig cannot be called after Go() has started") } @@ -208,8 +220,14 @@ type Processor[T any] interface { // - Items are grouped into batches based on the Config. // - Each batch is processed through the Processors in sequence. // -// Go must only be called once at a time. Calling Go again while a batch is -// already running will cause a panic. +// A Batch is single-use. Go returns the pipeline error channel together with a +// start error: +// - If the Batch has already been used, Go returns ErrBatchUsed. +// - If s is nil, Go returns ErrNilSource. +// +// On a start error the returned channel is non-nil and already closed, so it is +// always safe to range over even if the error is not checked. To run again, +// create a new Batch with New. Use errors.Is to test the returned error. // // Context cancellation: // - Go does not immediately stop processing when the context is canceled. @@ -218,7 +236,10 @@ type Processor[T any] interface { // Example: // // b := batch.New[any](config) -// errs := b.Go(ctx, source, processor) +// errs, err := b.Go(ctx, source, processor) +// if err != nil { +// log.Fatal(err) +// } // // go func() { // for err := range errs { @@ -232,29 +253,24 @@ type Processor[T any] interface { // - The Source must close its channels when reading is complete. // - Processors must check for context cancellation and stop early if needed. // - All items that have already been read will be processed even if the context is canceled. -func (b *Batch[T]) Go(ctx context.Context, s Source[T], procs ...Processor[T]) <-chan error { +func (b *Batch[T]) Go(ctx context.Context, s Source[T], procs ...Processor[T]) (<-chan error, error) { b.mu.Lock() defer b.mu.Unlock() - if b.running { - panic("Concurrent calls to Batch.Go are not allowed") + // A Batch is single-use. Reject a second call before touching any state so an + // already-running or already-finished Batch is never disturbed. + if b.used { + return closedErrs, ErrBatchUsed } - if b.config == nil { - b.config = NewConstantConfig(nil) + if s == nil { + return closedErrs, ErrNilSource } - b.running = true + b.used = true - // Check if source is nil and return error if it is - if s == nil { - b.errs = make(chan error, 1) - b.done = make(chan struct{}) - b.errs <- errors.New("source cannot be nil") - close(b.errs) - close(b.done) - b.running = false - return b.errs + if b.config == nil { + b.config = NewConstantConfig(nil) } b.src = s @@ -272,25 +288,19 @@ func (b *Batch[T]) Go(ctx context.Context, s Source[T], procs ...Processor[T]) < if itemBuf <= 0 { itemBuf = DefaultItemBufferSize } - idBuf := b.bufferConfig.IDBufferSize - if idBuf <= 0 { - idBuf = DefaultIDBufferSize - } errBuf := b.bufferConfig.ErrorBufferSize if errBuf <= 0 { errBuf = DefaultErrorBufferSize } b.items = make(chan *Item[T], itemBuf) - b.ids = make(chan uint64, idBuf) b.errs = make(chan error, errBuf) b.done = make(chan struct{}) - go b.doIDGenerator() go b.doReader(ctx) go b.doProcessors(ctx) - return b.errs + return b.errs, nil } // Done returns a channel that is closed when batch processing is complete. @@ -301,7 +311,11 @@ func (b *Batch[T]) Go(ctx context.Context, s Source[T], procs ...Processor[T]) < // Example: // // b := batch.New[any](config) -// batch.IgnoreErrors(b.Go(ctx, source, processor)) +// errs, err := b.Go(ctx, source, processor) +// if err != nil { +// log.Fatal(err) +// } +// batch.IgnoreErrors(errs) // // <-b.Done() // fmt.Println("Processing complete") @@ -323,22 +337,6 @@ func (b *Batch[T]) Done() <-chan struct{} { return b.done } -// doIDGenerator generates unique IDs for items in the pipeline. -// -// It runs as a background goroutine, incrementing a counter starting from zero -// and sending each ID on the ids channel. It exits when the done channel is closed. -func (b *Batch[T]) doIDGenerator() { - var id uint64 - for { - select { - case b.ids <- id: - id++ - case <-b.done: - return - } - } -} - // doReader reads items from the Source and forwards them to the batch processor. // // It starts the Source.Read goroutine, then listens for data and errors. @@ -358,6 +356,10 @@ func (b *Batch[T]) doReader(ctx context.Context) { return } + // nextID is goroutine-local: doReader is the only place item IDs are + // assigned, and a single-use Batch runs doReader exactly once, so the + // counter needs no synchronization (no atomic, no lock). + var nextID uint64 var outClosed, errsClosed bool for !outClosed || !errsClosed { select { @@ -367,7 +369,8 @@ func (b *Batch[T]) doReader(ctx context.Context) { out = nil continue } - id := <-b.ids + id := nextID + nextID++ b.items <- &Item[T]{ ID: id, Data: data, @@ -434,11 +437,13 @@ func (b *Batch[T]) doProcessors(ctx context.Context) { } wg.Wait() - close(b.errs) + + // Close done before errs. A CollectErrors caller returns once errs closes, + // so closing done first guarantees it also observes Done() as closed. No + // lock is needed: a Batch is single-use, so nothing reassigns done or errs + // while they are being closed here. close(b.done) - b.mu.Lock() - b.running = false - b.mu.Unlock() + close(b.errs) } // fixConfig corrects invalid ConfigValues to ensure consistent batch behavior. @@ -480,24 +485,28 @@ func (b *Batch[T]) waitForItems(_ context.Context, config ConfigValues) []*Item[ var ( reachedMinTime bool batch = make([]*Item[T], 0, config.MinItems) - minTimer <-chan time.Time - maxTimer <-chan time.Time + minTimerCh <-chan time.Time + maxTimerCh <-chan time.Time + minTimer *time.Timer + maxTimer *time.Timer ) // Be careful not to set timers that end right away. Instead, if a - // min or max time is not specified, use a nil channel so the select - // statement ignores it. + // min or max time is not specified, leave the channel nil so the + // select statement ignores it. Timers are stopped on return so a + // timer does not leak when a batch returns before its timer fires. if config.MinTime > 0 { - minTimer = time.After(config.MinTime) + minTimer = time.NewTimer(config.MinTime) + defer minTimer.Stop() + minTimerCh = minTimer.C } else { - minTimer = nil reachedMinTime = true } if config.MaxTime > 0 { - maxTimer = time.After(config.MaxTime) - } else { - maxTimer = nil + maxTimer = time.NewTimer(config.MaxTime) + defer maxTimer.Stop() + maxTimerCh = maxTimer.C } for { @@ -517,20 +526,20 @@ func (b *Batch[T]) waitForItems(_ context.Context, config ConfigValues) []*Item[ return batch } - case <-minTimer: + case <-minTimerCh: reachedMinTime = true if uint64(len(batch)) >= config.MinItems { return batch } // Keep waiting until MinItems is met - case <-maxTimer: + case <-maxTimerCh: if len(batch) > 0 { return batch } // If max timer fires with no items, restart it so we don't wait indefinitely if config.MaxTime > 0 { - maxTimer = time.After(config.MaxTime) + maxTimer.Reset(config.MaxTime) } } } diff --git a/batch/batch_test.go b/batch/batch_test.go index d7a5b1e..467a6b1 100644 --- a/batch/batch_test.go +++ b/batch/batch_test.go @@ -22,7 +22,10 @@ func TestBatch_ProcessorChainingAndErrorTracking(t *testing.T) { errProc := &errorPerItemProcessor{FailEvery: 3} countProc := &countProcessor{count: &count} - errs := batch.Go(context.Background(), src, errProc, countProc) + errs, err := batch.Go(context.Background(), src, errProc, countProc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } received := 0 for err := range errs { @@ -52,7 +55,10 @@ func TestBatch_ProcessorChainingAndErrorTracking(t *testing.T) { src := &testSource{Items: []any{1, 2}, WithErr: srcErr} countProc := &countProcessor{count: new(uint32)} - errs := batch.Go(context.Background(), src, countProc) + errs, err := batch.Go(context.Background(), src, countProc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } <-batch.Done() var found bool @@ -75,7 +81,10 @@ func TestBatch_ProcessorChainingAndErrorTracking(t *testing.T) { src := &testSource{Items: []any{1, 2, 3}} proc := &countProcessor{count: new(uint32), processorErr: procErr} - errs := batch.Go(context.Background(), src, proc) + errs, err := batch.Go(context.Background(), src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } var found bool var unwrappedErr error @@ -120,7 +129,10 @@ func TestBatch_ProcessorChainingAndErrorTracking(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) // Start processing - _ = batch.Go(ctx, src, proc) + _, err := batch.Go(ctx, src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } // Give some time for processing to start time.Sleep(50 * time.Millisecond) @@ -290,7 +302,10 @@ func TestBatch_ProcessorChainingAndErrorTracking(t *testing.T) { }, } - _ = batch.Go(context.Background(), src, proc) + _, err := batch.Go(context.Background(), src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } <-batch.Done() got := int(atomic.LoadUint32(&count)) @@ -388,7 +403,10 @@ func TestBatch_ProcessorChainingAndErrorTracking(t *testing.T) { ctx := context.Background() // Start processing and wait for completion - errs := b.Go(ctx, s, p) + errs, err := b.Go(ctx, s, p) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } for range errs { // Consume errors } @@ -460,7 +478,10 @@ func TestBatch_ComplexProcessingPipeline(t *testing.T) { var count uint32 counter := &countProcessor{count: &count} - errs := batch.Go(context.Background(), src, transformer, filter, counter) + errs, err := batch.Go(context.Background(), src, transformer, filter, counter) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } // Drain errors for range errs { @@ -506,7 +527,7 @@ func TestBatch_ConcurrentProcessing(t *testing.T) { src := &testSource{Items: items} proc := &countProcessor{count: counters[i]} - _ = batch.Go(context.Background(), src, proc) + _, _ = batch.Go(context.Background(), src, proc) <-batch.Done() }() } @@ -544,7 +565,10 @@ func TestBatch_RobustnessAndEdgeCases(t *testing.T) { proc := &countProcessor{count: &count} // Process large batch - errs := batch.Go(context.Background(), src, proc) + errs, err := batch.Go(context.Background(), src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } <-batch.Done() // Drain errors @@ -567,7 +591,10 @@ func TestBatch_RobustnessAndEdgeCases(t *testing.T) { var count uint32 proc := &countProcessor{count: &count} - errs := batch.Go(context.Background(), src, proc) + errs, err := batch.Go(context.Background(), src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } <-batch.Done() // Drain errors @@ -595,7 +622,10 @@ func TestBatch_RobustnessAndEdgeCases(t *testing.T) { var count uint32 proc := &countProcessor{count: &count} - errs := batch.Go(context.Background(), src, proc) + errs, err := batch.Go(context.Background(), src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } <-batch.Done() // Drain errors @@ -623,7 +653,10 @@ func TestBatch_RobustnessAndEdgeCases(t *testing.T) { var count uint32 proc := &countProcessor{count: &count} - errs := batch.Go(context.Background(), src, proc) + errs, err := batch.Go(context.Background(), src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } <-batch.Done() // Drain errors @@ -647,7 +680,10 @@ func TestBatch_NoProcessors(t *testing.T) { src := &testSource{Items: items} // Call Go with source but no processors - errs := batch.Go(context.Background(), src) + errs, err := batch.Go(context.Background(), src) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } // Count errors instead of collecting them errorCount := 0 @@ -675,7 +711,10 @@ func TestBatch_NoProcessors(t *testing.T) { emptyProcessors := make([]Processor[any], 0) // Call Go with source and empty processor slice - errs := batch.Go(context.Background(), src, emptyProcessors...) + errs, err := batch.Go(context.Background(), src, emptyProcessors...) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } // Use a counter instead of collecting errors errorCount := 0 @@ -716,7 +755,10 @@ func TestBatch_NoTimersWithMinItems(t *testing.T) { }, } - errs := batch.Go(context.Background(), src, proc) + errs, err := batch.Go(context.Background(), src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } <-batch.Done() for range errs { // Drain errors @@ -746,7 +788,10 @@ func TestBatch_DoneNonBlocking(t *testing.T) { t.Run("after Go", func(t *testing.T) { b := New[any](NewConstantConfig(nil)) src := &testSource{Items: []any{}} - errs := b.Go(context.Background(), src) + errs, err := b.Go(context.Background(), src) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } <-b.Done() for range errs { } diff --git a/batch/benchmark_test.go b/batch/benchmark_test.go new file mode 100644 index 0000000..d73e258 --- /dev/null +++ b/batch/benchmark_test.go @@ -0,0 +1,69 @@ +package batch_test + +import ( + "context" + "testing" + + . "github.com/MasterOfBinary/gobatch/batch" +) + +// benchSource emits the integers [0, n) as fast as it can, respecting context +// cancellation. It exists to drive a high item rate through the pipeline so the +// per-item ID-assignment path in doReader is exercised. +type benchSource struct{ n int } + +func (s *benchSource) 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 out <- i: + } + } + }() + return out, errs +} + +// benchPassthrough is a no-op processor: it returns the batch unchanged so the +// benchmark measures pipeline/ID overhead rather than processing work. +type benchPassthrough struct{} + +func (benchPassthrough) Process(_ context.Context, items []*Item[any]) ([]*Item[any], error) { + return items, nil +} + +// BenchmarkBatchThroughput measures end-to-end throughput for a fixed number of +// items per run. Every item gets an ID assigned in doReader, so this captures +// the cost of the ID-generation mechanism (the inline counter on this branch +// vs. the dedicated goroutine + buffered channel on master) plus the per-Go +// setup cost. A fresh Batch is created each iteration, which is also the +// required usage: a Batch is single-use. +// +// Compare across branches with: +// +// go test -run '^$' -bench BenchmarkBatchThroughput -benchmem -count=10 ./batch +// +// then feed both outputs to benchstat. +func BenchmarkBatchThroughput(b *testing.B) { + const itemsPerRun = 1000 + cfg := NewConstantConfig(&ConfigValues{MaxItems: 100}) + proc := benchPassthrough{} + ctx := context.Background() + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + bt := New[any](cfg) + errs, err := bt.Go(ctx, &benchSource{n: itemsPerRun}, proc) + if err != nil { + b.Fatal(err) + } + IgnoreErrors(errs) + <-bt.Done() + } +} diff --git a/batch/buffer_config_test.go b/batch/buffer_config_test.go index 65c6060..2ec263a 100644 --- a/batch/buffer_config_test.go +++ b/batch/buffer_config_test.go @@ -17,7 +17,6 @@ func TestBatch_WithBufferConfig(t *testing.T) { t.Run("custom buffer sizes", func(t *testing.T) { customConfig := BufferConfig{ ItemBufferSize: 500, - IDBufferSize: 600, ErrorBufferSize: 200, } @@ -27,9 +26,6 @@ func TestBatch_WithBufferConfig(t *testing.T) { if b.bufferConfig.ItemBufferSize != 500 { t.Errorf("expected ItemBufferSize=500, got %d", b.bufferConfig.ItemBufferSize) } - if b.bufferConfig.IDBufferSize != 600 { - t.Errorf("expected IDBufferSize=600, got %d", b.bufferConfig.IDBufferSize) - } if b.bufferConfig.ErrorBufferSize != 200 { t.Errorf("expected ErrorBufferSize=200, got %d", b.bufferConfig.ErrorBufferSize) } @@ -52,7 +48,10 @@ func TestBatch_WithBufferConfig(t *testing.T) { }) // Start processing to trigger channel creation - errs := b.Go(context.Background(), src) + errs, err := b.Go(context.Background(), src) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } IgnoreErrors(errs) <-b.Done() @@ -62,7 +61,6 @@ func TestBatch_WithBufferConfig(t *testing.T) { t.Run("negative values use defaults", func(t *testing.T) { customConfig := BufferConfig{ ItemBufferSize: -1, - IDBufferSize: -1, ErrorBufferSize: -1, } @@ -81,7 +79,10 @@ func TestBatch_WithBufferConfig(t *testing.T) { }) // Start processing - should use default buffer sizes - errs := b.Go(context.Background(), src) + errs, err := b.Go(context.Background(), src) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } IgnoreErrors(errs) <-b.Done() @@ -104,7 +105,10 @@ func TestBatch_WithBufferConfig(t *testing.T) { }) // Start batch processing - errs := b.Go(context.Background(), src) + errs, err := b.Go(context.Background(), src) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } // Should panic when trying to set buffer config after Go defer func() { @@ -154,7 +158,7 @@ func TestBatch_WithBufferConfig(t *testing.T) { close(errs) return out, errs }) - errs := b.Go(context.Background(), src) + errs, _ := b.Go(context.Background(), src) IgnoreErrors(errs) }() @@ -172,9 +176,6 @@ func TestDefaultConstants(t *testing.T) { if DefaultItemBufferSize < 1 { t.Errorf("DefaultItemBufferSize should be positive, got %d", DefaultItemBufferSize) } - if DefaultIDBufferSize < 1 { - t.Errorf("DefaultIDBufferSize should be positive, got %d", DefaultIDBufferSize) - } if DefaultErrorBufferSize < 1 { t.Errorf("DefaultErrorBufferSize should be positive, got %d", DefaultErrorBufferSize) } @@ -183,9 +184,6 @@ func TestDefaultConstants(t *testing.T) { if DefaultItemBufferSize != 100 { t.Errorf("DefaultItemBufferSize changed from expected 100 to %d", DefaultItemBufferSize) } - if DefaultIDBufferSize != 100 { - t.Errorf("DefaultIDBufferSize changed from expected 100 to %d", DefaultIDBufferSize) - } if DefaultErrorBufferSize != 100 { t.Errorf("DefaultErrorBufferSize changed from expected 100 to %d", DefaultErrorBufferSize) } diff --git a/batch/constants.go b/batch/constants.go index c48a030..1fec626 100644 --- a/batch/constants.go +++ b/batch/constants.go @@ -7,10 +7,6 @@ const ( // This determines how many items can be queued between the reader and processor. DefaultItemBufferSize = 100 - // DefaultIDBufferSize is the default buffer size for the ID generator channel. - // This should match or exceed the item buffer size to avoid blocking. - DefaultIDBufferSize = 100 - // DefaultErrorBufferSize is the default buffer size for the error channel. // This should be large enough to handle bursts of errors without blocking. DefaultErrorBufferSize = 100 diff --git a/batch/doc.go b/batch/doc.go index f906263..990779c 100644 --- a/batch/doc.go +++ b/batch/doc.go @@ -33,7 +33,11 @@ // b := New[int](cfg) // src := &source.Nil[int]{} // proc := &processor.Nil[int]{} -// IgnoreErrors(b.Go(ctx, src, proc)) +// errs, err := b.Go(ctx, src, proc) +// if err != nil { +// log.Fatal(err) +// } +// IgnoreErrors(errs) // <-b.Done() // // The configuration is reloaded before each batch is collected. This allows diff --git a/batch/dynamic_config_test.go b/batch/dynamic_config_test.go index b2ae0d9..0e5821f 100644 --- a/batch/dynamic_config_test.go +++ b/batch/dynamic_config_test.go @@ -53,7 +53,10 @@ func TestBatch_DynamicConfiguration(t *testing.T) { } // Start batch processing with initial config - errs := batch.Go(context.Background(), src, proc) + errs, err := batch.Go(context.Background(), src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } // Wait a bit for some items to be read, but not processed due to MinItems: 50 time.Sleep(100 * time.Millisecond) diff --git a/batch/error_handling_test.go b/batch/error_handling_test.go index c8056ee..aad4206 100644 --- a/batch/error_handling_test.go +++ b/batch/error_handling_test.go @@ -21,7 +21,10 @@ func TestBatch_ErrorHandling(t *testing.T) { processorErr: procErr, } - errs := batch.Go(context.Background(), src, proc) + errs, err := batch.Go(context.Background(), src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } var foundErr bool for err := range errs { @@ -49,7 +52,10 @@ func TestBatch_ErrorHandling(t *testing.T) { proc := &countProcessor{count: new(uint32)} - errs := batch.Go(context.Background(), src, proc) + errs, err := batch.Go(context.Background(), src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } var foundErr bool for err := range errs { @@ -66,33 +72,6 @@ func TestBatch_ErrorHandling(t *testing.T) { } }) - t.Run("nil source handling", func(t *testing.T) { - batch := New[any](NewConstantConfig(&ConfigValues{})) - - // Pass nil source - errs := batch.Go(context.Background(), nil) - - var foundErr bool - var errMsg string - for err := range errs { - if err != nil { - foundErr = true - errMsg = err.Error() - break - } - } - - <-batch.Done() - - if !foundErr { - t.Error("expected error with nil source") - } - - if !strings.Contains(errMsg, "source cannot be nil") { - t.Errorf("expected 'source cannot be nil' error, got: %s", errMsg) - } - }) - t.Run("nil processor filtering", func(t *testing.T) { batch := New[any](NewConstantConfig(&ConfigValues{})) src := &testSource{Items: []any{1, 2, 3}} @@ -102,7 +81,10 @@ func TestBatch_ErrorHandling(t *testing.T) { validProc := &countProcessor{count: &count} // Pass a mix of nil and valid processors - errs := batch.Go(context.Background(), src, nil, validProc, nil) + errs, err := batch.Go(context.Background(), src, nil, validProc, nil) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } // Count errors instead of collecting them errorCount := 0 @@ -131,7 +113,10 @@ func TestBatch_NilChannelHandling(t *testing.T) { // Create a source that returns a nil output channel nilChannelSource := &nilOutputChannelSource{} - errs := batch.Go(context.Background(), nilChannelSource) + errs, err := batch.Go(context.Background(), nilChannelSource) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } var foundErr bool var errMsg string @@ -160,7 +145,10 @@ func TestBatch_NilChannelHandling(t *testing.T) { // Create a source that returns a nil error channel nilChannelSource := &nilErrorChannelSource{} - errs := batch.Go(context.Background(), nilChannelSource) + errs, err := batch.Go(context.Background(), nilChannelSource) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } var foundErr bool var errMsg string diff --git a/batch/errors.go b/batch/errors.go index b5a3d20..77e1dfa 100644 --- a/batch/errors.go +++ b/batch/errors.go @@ -1,6 +1,18 @@ package batch -import "fmt" +import ( + "errors" + "fmt" +) + +// ErrNilSource is returned by Batch.Go when the provided Source is nil. +// Use errors.Is to check for it. +var ErrNilSource = errors.New("batch: source cannot be nil") + +// ErrBatchUsed is returned by Batch.Go when it is called on a Batch that has +// already been used. A Batch is single-use: create a new one with New to run +// again. Use errors.Is to check for it. +var ErrBatchUsed = errors.New("batch: Batch is single-use; create a new Batch with New to run again") // ProcessorError is returned when a processor fails. It wraps the original // error from the processor to maintain the error chain while providing diff --git a/batch/example_custom_config_test.go b/batch/example_custom_config_test.go index 4d51d6e..f6baec6 100644 --- a/batch/example_custom_config_test.go +++ b/batch/example_custom_config_test.go @@ -128,7 +128,11 @@ func Example_customConfig() { defer cancel() fmt.Println("Starting batch processing...") - errs := b.Go(ctx, src, p) + errs, err := b.Go(ctx, src, p) + if err != nil { + fmt.Println(err) + return + } batch.IgnoreErrors(errs) <-b.Done() diff --git a/batch/example_dynamic_config_test.go b/batch/example_dynamic_config_test.go index e923348..afca92a 100644 --- a/batch/example_dynamic_config_test.go +++ b/batch/example_dynamic_config_test.go @@ -46,7 +46,11 @@ func Example_dynamicConfig() { fmt.Println("=== Dynamic Config Example ===") - errs := b.Go(ctx, src, monitor) + errs, err := b.Go(ctx, src, monitor) + if err != nil { + fmt.Println(err) + return + } // Simulate sending data and changing config dynamically go func() { diff --git a/batch/example_test.go b/batch/example_test.go index 3ad4abe..4f2d1c2 100644 --- a/batch/example_test.go +++ b/batch/example_test.go @@ -43,7 +43,11 @@ func Example() { ctx := context.Background() // Start processing with both processors chained - errs := b.Go(ctx, src, doubleProc, printProc) + errs, err := b.Go(ctx, src, doubleProc, printProc) + if err != nil { + fmt.Println(err) + return + } // Ignore errors batch.IgnoreErrors(errs) diff --git a/batch/helpers.go b/batch/helpers.go index e65ada8..eb8f965 100644 --- a/batch/helpers.go +++ b/batch/helpers.go @@ -9,12 +9,17 @@ import ( // It can be used with Batch.Go if errors aren't needed. Ignoring the returned // channel without reading from it can block once the buffer fills. For example: // -// // NOTE: bad - this can cause a deadlock! -// _ = batch.Go(ctx, p, s) +// // NOTE: bad - leaving errs undrained can deadlock once the buffer fills! +// errs, _ := myBatch.Go(ctx, s, p) +// _ = errs // // Instead, IgnoreErrors can be used to safely discard all errors: // -// batch.IgnoreErrors(myBatch.Go(ctx, p, s)) +// errs, err := myBatch.Go(ctx, s, p) +// if err != nil { +// log.Fatal(err) +// } +// batch.IgnoreErrors(errs) func IgnoreErrors(errs <-chan error) { // nil channels always block, so check for nil first to avoid a goroutine // leak @@ -27,12 +32,17 @@ func IgnoreErrors(errs <-chan error) { } // CollectErrors collects all errors from the error channel into a slice. -// This is useful when you need to process all errors after batch processing completes. +// It blocks until the error channel is closed (i.e., until batch processing +// completes), so there is no need to wait on Done afterwards. // // Example usage: // -// errs := batch.CollectErrors(myBatch.Go(ctx, source, processor)) -// <-myBatch.Done() +// pipeErrs, err := myBatch.Go(ctx, source, processor) +// if err != nil { +// log.Fatal(err) +// } +// errs := batch.CollectErrors(pipeErrs) +// // CollectErrors blocks until processing is done. // for _, err := range errs { // log.Printf("Error: %v", err) // } @@ -60,8 +70,13 @@ func CollectErrors(errs <-chan error) []error { // // Handle errors // } func RunBatchAndWait[T any](ctx context.Context, b *Batch[T], s Source[T], procs ...Processor[T]) []error { - // Start the batch processing - errs := b.Go(ctx, s, procs...) + // Start the batch processing. A start error (e.g. ErrNilSource or + // ErrBatchUsed) is surfaced in the returned slice so callers that only + // inspect the slice still see the failure. + errs, err := b.Go(ctx, s, procs...) + if err != nil { + return []error{err} + } // Collect all errors into a slice var collectedErrors []error @@ -110,11 +125,20 @@ func ExecuteBatches[T any](ctx context.Context, configs ...*BatchConfig[T]) []er go func(cfg *BatchConfig[T]) { defer wg.Done() - if cfg == nil || cfg.B == nil || cfg.S == nil { + // Skip entries with nothing to run. A nil source is NOT skipped + // here: cfg.B.Go reports it as ErrNilSource below, so a missing + // source surfaces an error instead of silently dropping the batch. + if cfg == nil || cfg.B == nil { return } - errs := cfg.B.Go(ctx, cfg.S, cfg.P...) + errs, err := cfg.B.Go(ctx, cfg.S, cfg.P...) + if err != nil { + mu.Lock() + allErrs = append(allErrs, err) + mu.Unlock() + return + } for err := range errs { mu.Lock() allErrs = append(allErrs, err) diff --git a/batch/helpers_test.go b/batch/helpers_test.go index 6f1fd2b..2e92f88 100644 --- a/batch/helpers_test.go +++ b/batch/helpers_test.go @@ -203,6 +203,36 @@ func TestExecuteBatches_NilConfig(t *testing.T) { } } +// TestExecuteBatches_NilSourceSurfacesError verifies that a configured batch +// with a nil Source surfaces ErrNilSource instead of being silently skipped, +// so callers cannot mistake unprocessed data for a completed run. Valid configs +// in the same call must still run to completion. +func TestExecuteBatches_NilSourceSurfacesError(t *testing.T) { + valid := New[any](NewConstantConfig(&ConfigValues{})) + var count uint32 + + configs := []*BatchConfig[any]{ + {B: valid, S: &testSource{Items: []any{1, 2, 3}}, P: []Processor[any]{&countProcessor{count: &count}}}, + {B: New[any](NewConstantConfig(&ConfigValues{})), S: nil, P: []Processor[any]{&countProcessor{count: new(uint32)}}}, + } + + errs := ExecuteBatches(context.Background(), configs...) + + if atomic.LoadUint32(&count) != 3 { + t.Errorf("valid batch: expected 3 items processed, got %d", count) + } + + var found bool + for _, err := range errs { + if errors.Is(err, ErrNilSource) { + found = true + } + } + if !found { + t.Fatalf("expected ErrNilSource to be surfaced for the nil-source config, got %v", errs) + } +} + // Helper types for testing type countProcessor struct { diff --git a/batch/maxtime_test.go b/batch/maxtime_test.go index 078d465..caa7279 100644 --- a/batch/maxtime_test.go +++ b/batch/maxtime_test.go @@ -41,7 +41,11 @@ func TestMaxTimeIdle(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - b.Go(ctx, src, proc) + errs, err := b.Go(ctx, src, proc) + if err != nil { + t.Fatalf("Go returned unexpected error: %v", err) + } + batch.IgnoreErrors(errs) // 1. Wait for longer than MaxTime (200ms > 100ms) to trigger the idle expiration bug time.Sleep(200 * time.Millisecond) diff --git a/batch/single_use_test.go b/batch/single_use_test.go new file mode 100644 index 0000000..319e800 --- /dev/null +++ b/batch/single_use_test.go @@ -0,0 +1,67 @@ +package batch_test + +import ( + "context" + "errors" + "testing" + + . "github.com/MasterOfBinary/gobatch/batch" +) + +// TestGo_SecondCallReturnsErrBatchUsed pins the single-use contract: a Batch may +// run exactly once. A second Go() on the same Batch — even after the first run +// has fully completed — returns ErrBatchUsed instead of starting another run. +// Callers that want to run again must create a fresh Batch with New. +func TestGo_SecondCallReturnsErrBatchUsed(t *testing.T) { + b := New[any](NewConstantConfig(&ConfigValues{MinItems: 1, MaxItems: 5})) + + errs, err := b.Go(context.Background(), &testSource{Items: []any{1, 2, 3}}, &countProcessor{}) + if err != nil { + t.Fatalf("first Go returned unexpected error: %v", err) + } + IgnoreErrors(errs) + <-b.Done() + + errs2, err2 := b.Go(context.Background(), &testSource{Items: []any{4, 5, 6}}, &countProcessor{}) + if !errors.Is(err2, ErrBatchUsed) { + t.Fatalf("second Go: got error %v, want ErrBatchUsed", err2) + } + // A rejected Go must still return a non-nil, already-closed channel so a + // caller that ranges over it without checking err does not block forever. + if errs2 == nil { + t.Fatal("second Go returned a nil error channel; want a closed, drainable channel") + } + for range errs2 { + t.Fatal("error channel from a rejected Go should be empty") + } +} + +// TestGo_NilSourceReturnsErrNilSource verifies that a nil Source is reported via +// the returned error rather than being smuggled onto the pipeline error channel. +func TestGo_NilSourceReturnsErrNilSource(t *testing.T) { + b := New[any](NewConstantConfig(&ConfigValues{})) + + errs, err := b.Go(context.Background(), nil) + if !errors.Is(err, ErrNilSource) { + t.Fatalf("Go(nil): got error %v, want ErrNilSource", err) + } + if errs == nil { + t.Fatal("Go(nil) returned a nil error channel; want a closed, drainable channel") + } + for range errs { + t.Fatal("error channel from a rejected Go should be empty") + } +} + +// TestGo_SuccessReturnsNilError verifies the happy path returns a nil start error +// alongside the live pipeline error channel. +func TestGo_SuccessReturnsNilError(t *testing.T) { + b := New[any](NewConstantConfig(&ConfigValues{MinItems: 1, MaxItems: 5})) + + errs, err := b.Go(context.Background(), &testSource{Items: []any{1, 2, 3}}, &countProcessor{}) + if err != nil { + t.Fatalf("Go returned unexpected start error: %v", err) + } + IgnoreErrors(errs) + <-b.Done() +} diff --git a/doc.go b/doc.go index 53ca9f0..74d1ff1 100644 --- a/doc.go +++ b/doc.go @@ -19,7 +19,11 @@ // return v, nil // }} // -// batch.IgnoreErrors(b.Go(context.Background(), src, proc)) +// errs, err := b.Go(context.Background(), src, proc) +// if err != nil { +// log.Fatal(err) +// } +// batch.IgnoreErrors(errs) // <-b.Done() // // Output: diff --git a/example_test.go b/example_test.go index f26fc2e..f7d0974 100644 --- a/example_test.go +++ b/example_test.go @@ -23,7 +23,12 @@ func Example() { return v, nil }} - batch.IgnoreErrors(b.Go(context.Background(), src, proc)) + errs, err := b.Go(context.Background(), src, proc) + if err != nil { + fmt.Println(err) + return + } + batch.IgnoreErrors(errs) <-b.Done() // Output: // hello