From d0dbfca15ecdfe91f09f4c3d143e4c5cfd33a5e9 Mon Sep 17 00:00:00 2001
From: Vaughn Friesen <vdfriesen@vaughnfriesen.com>
Date: Fri, 29 May 2026 20:38:30 +0800
Subject: [PATCH 1/5] feat(batch)!: inline item ID counter and reusable Batch
 (BREAKING: remove IDBufferSize)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Replace the dedicated ID-generator goroutine and its buffered channel with an
inline counter on Batch, reset on each Go() so a reused Batch numbers from
zero. Only Go (reset, before goroutines start) and the single doReader
goroutine touch it, so a plain counter is race-free — no atomic or lock, and
no 64-bit-alignment concern on 32-bit platforms. Removes
BufferConfig.IDBufferSize and the DefaultIDBufferSize constant (BREAKING) and
eliminates a pre-existing Batch-reuse data race on ID generation.

Finalize shutdown under b.mu — close done and clear running before closing
errs — so once CollectErrors returns or Done() fires the Batch is fully torn
down and reusable without tripping the "Concurrent calls" guard.

Fix a timer leak in waitForItems (time.NewTimer + defer Stop() + Reset()
instead of time.After), and clarify that CollectErrors blocks until the error
channel closes.

Tests: reuse loops including a test pinning the per-run ID reset to 0..n-1,
plus a throughput benchmark. The inline counter benchmarks ~40% faster than
the previous channel-based generator.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 .gitignore                  |   4 +-
 README.md                   |   4 +-
 batch/batch.go              |  73 +++++++++++---------------
 batch/benchmark_test.go     |  65 +++++++++++++++++++++++
 batch/buffer_config_test.go |  11 ----
 batch/constants.go          |   4 --
 batch/helpers.go            |   5 +-
 batch/reuse_test.go         | 100 ++++++++++++++++++++++++++++++++++++
 8 files changed, 202 insertions(+), 64 deletions(-)
 create mode 100644 batch/benchmark_test.go
 create mode 100644 batch/reuse_test.go

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..2aceb60 100644
--- a/README.md
+++ b/README.md
@@ -255,7 +255,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,9 +296,8 @@ go func() {
 Or using helper functions:
 
 ```go
-// Collect all errors
+// Collect all errors (blocks until processing completes)
 errs := batch.CollectErrors(batchProcessor.Go(ctx, source, processor))
-<-batchProcessor.Done()
 
 // Or use the RunBatchAndWait helper
 errs := batch.RunBatchAndWait(ctx, batchProcessor, source, processor)
diff --git a/batch/batch.go b/batch/batch.go
index 6b7cb46..f81b79d 100644
--- a/batch/batch.go
+++ b/batch/batch.go
@@ -22,10 +22,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
@@ -73,7 +69,7 @@ type Batch[T any] struct {
 	src          Source[T]
 	processors   []Processor[T]
 	items        chan *Item[T]
-	ids          chan uint64
+	nextID       uint64
 	done         chan struct{}
 
 	mu      sync.Mutex
@@ -100,7 +96,6 @@ func New[T any](config Config) *Batch[T] {
 //
 //	b := batch.New[any](config).WithBufferConfig(batch.BufferConfig{
 //		ItemBufferSize:  1000,
-//		IDBufferSize:    1000,
 //		ErrorBufferSize: 500,
 //	})
 //
@@ -272,21 +267,17 @@ 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{})
+	// Reset the item ID counter so a reused Batch starts numbering from zero.
+	b.nextID = 0
 
-	go b.doIDGenerator()
 	go b.doReader(ctx)
 	go b.doProcessors(ctx)
 
@@ -323,22 +314,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.
@@ -367,7 +342,10 @@ func (b *Batch[T]) doReader(ctx context.Context) {
 				out = nil
 				continue
 			}
-			id := <-b.ids
+			// Only doReader increments nextID, and Go resets it before
+			// starting this goroutine, so a plain counter is race-free.
+			id := b.nextID
+			b.nextID++
 			b.items <- &Item[T]{
 				ID:   id,
 				Data: data,
@@ -434,10 +412,15 @@ func (b *Batch[T]) doProcessors(ctx context.Context) {
 	}
 
 	wg.Wait()
-	close(b.errs)
-	close(b.done)
+
+	// Finalize shutdown under b.mu, closing done and clearing running before
+	// errs. A CollectErrors caller (which returns once errs closes) therefore
+	// always observes a fully torn-down, reusable Batch — done closed and
+	// running false — and a reusing Go() blocks on b.mu until this completes.
 	b.mu.Lock()
+	close(b.done)
 	b.running = false
+	close(b.errs)
 	b.mu.Unlock()
 }
 
@@ -480,24 +463,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 +504,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/benchmark_test.go b/batch/benchmark_test.go
new file mode 100644
index 0000000..f1e1217
--- /dev/null
+++ b/batch/benchmark_test.go
@@ -0,0 +1,65 @@
+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 so the benchmark runs
+// identically on master, where Batch reuse is not yet supported.
+//
+// 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)
+		IgnoreErrors(bt.Go(ctx, &benchSource{n: itemsPerRun}, proc))
+		<-bt.Done()
+	}
+}
diff --git a/batch/buffer_config_test.go b/batch/buffer_config_test.go
index 65c6060..e8b2f3c 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)
 		}
@@ -62,7 +58,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,
 		}
 
@@ -172,9 +167,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 +175,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/helpers.go b/batch/helpers.go
index e65ada8..bbc467a 100644
--- a/batch/helpers.go
+++ b/batch/helpers.go
@@ -27,12 +27,13 @@ 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()
+//	// CollectErrors blocks until processing is done.
 //	for _, err := range errs {
 //		log.Printf("Error: %v", err)
 //	}
diff --git a/batch/reuse_test.go b/batch/reuse_test.go
new file mode 100644
index 0000000..97338f8
--- /dev/null
+++ b/batch/reuse_test.go
@@ -0,0 +1,100 @@
+package batch_test
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	. "github.com/MasterOfBinary/gobatch/batch"
+)
+
+// TestReuse_AfterCollectErrors verifies the CollectErrors completion
+// contract: once CollectErrors returns, the Batch is fully torn down
+// (Done closed, running cleared) so the same Batch can be reused
+// immediately without hitting the "Concurrent calls" panic.
+//
+// This is a regression test for shutdown ordering in doProcessors: if
+// b.errs is closed before b.done/b.running are finalized, a caller that
+// follows the documented "no need to wait on Done()" guidance and reuses
+// the Batch can intermittently panic. Run with -race to surface it.
+func TestReuse_AfterCollectErrors(t *testing.T) {
+	b := New[any](NewConstantConfig(&ConfigValues{MinItems: 1, MaxItems: 5}))
+
+	for i := 0; i < 200; i++ {
+		src := &testSource{Items: []any{1, 2, 3, 4, 5}}
+		proc := &countProcessor{}
+
+		// CollectErrors blocks until b.errs is closed; per its docs the
+		// caller need not also wait on Done(). Reuse b on the very next
+		// line — this panics if running is still true at that point.
+		errs := CollectErrors(b.Go(context.Background(), src, proc))
+		for range errs {
+		}
+	}
+}
+
+// TestReuse_AfterDone verifies the other synchronization path flagged in
+// review: a caller that waits on <-Done() and then immediately reuses the
+// Batch must not hit the "Concurrent calls" panic. This requires that
+// running is cleared no later than Done() becoming observable. Shutdown
+// closes done and clears running inside the same b.mu section that
+// doProcessors holds across both, so a Go() from a Done() waiter blocks
+// on the mutex until running is false. Run with -race -count to surface
+// any regression.
+func TestReuse_AfterDone(t *testing.T) {
+	b := New[any](NewConstantConfig(&ConfigValues{MinItems: 1, MaxItems: 5}))
+
+	for i := 0; i < 200; i++ {
+		src := &testSource{Items: []any{1, 2, 3, 4, 5}}
+		proc := &countProcessor{}
+
+		errs := b.Go(context.Background(), src, proc)
+		IgnoreErrors(errs)
+
+		// Wake on Done() and reuse the Batch immediately. If running is
+		// still true when Done() is observable, the next Go() panics.
+		<-b.Done()
+	}
+}
+
+// idRecorder is a processor that records the ID of every item it processes.
+type idRecorder struct {
+	mu  sync.Mutex
+	ids []uint64
+}
+
+func (r *idRecorder) Process(_ context.Context, items []*Item[any]) ([]*Item[any], error) {
+	r.mu.Lock()
+	for _, it := range items {
+		r.ids = append(r.ids, it.ID)
+	}
+	r.mu.Unlock()
+	return items, nil
+}
+
+// TestReuse_RestartsIDsFromZero pins the core ID contract: every run of a
+// reused Batch numbers items 0..n-1. This guards against the per-run reset
+// silently regressing (e.g. IDs starting at 1, or not resetting between runs)
+// — a failure mode the existing reuse tests, which only assert "no panic",
+// would not catch.
+func TestReuse_RestartsIDsFromZero(t *testing.T) {
+	// MinItems == MaxItems == 5 with a 5-item source yields exactly one batch
+	// per run, so the recorded IDs are 0..4 in order.
+	b := New[any](NewConstantConfig(&ConfigValues{MinItems: 5, MaxItems: 5}))
+
+	for iter := 0; iter < 5; iter++ {
+		rec := &idRecorder{}
+		IgnoreErrors(b.Go(context.Background(), &testSource{Items: []any{10, 20, 30, 40, 50}}, rec))
+		<-b.Done()
+
+		want := []uint64{0, 1, 2, 3, 4}
+		if len(rec.ids) != len(want) {
+			t.Fatalf("iteration %d: got %d IDs %v, want %v", iter, len(rec.ids), rec.ids, want)
+		}
+		for i, id := range rec.ids {
+			if id != want[i] {
+				t.Fatalf("iteration %d: IDs = %v, want %v", iter, rec.ids, want)
+			}
+		}
+	}
+}

From b0e9325b1e9541ae21a2b731a7eea0d276466690 Mon Sep 17 00:00:00 2001
From: Vaughn Friesen <vdfriesen@vaughnfriesen.com>
Date: Fri, 29 May 2026 22:23:17 +0800
Subject: [PATCH 2/5] refactor(batch)!: single-use Batch; Go returns (<-chan
 error, error)

Replaces the reusable-Batch design with single-use semantics, which removes
a whole class of cross-run concurrency hazards (the shared ID-counter race
and the Done()/Go() reuse race surfaced in review).

- Go now returns (<-chan error, error). Start failures surface via the
  returned error using errors.Is-able sentinels (ErrBatchUsed, ErrNilSource)
  instead of panicking (second/concurrent call) or smuggling the error onto
  the pipeline channel (nil source). On a start error the returned channel is
  non-nil and already closed, so ranging over it is always safe.
- A Batch runs once; a second Go() returns ErrBatchUsed. Create a new Batch
  with New to run again.
- The item-ID counter is now a doReader-local variable, so it needs no
  atomic or lock by construction.
- Shutdown drops the mutex around closing done/errs (nothing reassigns them
  without reuse).
- RunBatchAndWait/ExecuteBatches surface start errors; all call sites,
  examples, README, and docs updated. reuse_test.go removed;
  single_use_test.go added.

BREAKING CHANGE: Batch.Go returns (<-chan error, error) and Batch is
single-use. Update call sites to `errs, err := b.Go(...)` and create a new
Batch per run.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 README.md                            |  22 +++++-
 batch/batch.go                       | 104 ++++++++++++++++-----------
 batch/batch_test.go                  |  32 ++++-----
 batch/benchmark_test.go              |   7 +-
 batch/buffer_config_test.go          |   8 +--
 batch/doc.go                         |   6 +-
 batch/dynamic_config_test.go         |   2 +-
 batch/error_handling_test.go         |  39 +++-------
 batch/errors.go                      |  14 +++-
 batch/example_custom_config_test.go  |   6 +-
 batch/example_dynamic_config_test.go |   6 +-
 batch/example_test.go                |   6 +-
 batch/helpers.go                     |  34 +++++++--
 batch/reuse_test.go                  | 100 --------------------------
 batch/single_use_test.go             |  67 +++++++++++++++++
 doc.go                               |   6 +-
 example_test.go                      |   7 +-
 17 files changed, 253 insertions(+), 213 deletions(-)
 delete mode 100644 batch/reuse_test.go
 create mode 100644 batch/single_use_test.go

diff --git a/README.md b/README.md
index 2aceb60..68b1a4b 100644
--- a/README.md
+++ b/README.md
@@ -164,7 +164,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)
@@ -297,9 +300,14 @@ Or using helper functions:
 
 ```go
 // Collect all errors (blocks until processing completes)
-errs := batch.CollectErrors(batchProcessor.Go(ctx, source, processor))
+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 {
@@ -307,6 +315,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 f81b79d..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 {
@@ -44,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())
 //	}
@@ -56,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
 //
@@ -69,12 +87,11 @@ type Batch[T any] struct {
 	src          Source[T]
 	processors   []Processor[T]
 	items        chan *Item[T]
-	nextID       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,
@@ -104,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")
 	}
 
@@ -203,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.
@@ -213,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 {
@@ -227,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
@@ -275,13 +296,11 @@ func (b *Batch[T]) Go(ctx context.Context, s Source[T], procs ...Processor[T]) <
 	b.items = make(chan *Item[T], itemBuf)
 	b.errs = make(chan error, errBuf)
 	b.done = make(chan struct{})
-	// Reset the item ID counter so a reused Batch starts numbering from zero.
-	b.nextID = 0
 
 	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.
@@ -292,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")
@@ -333,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 {
@@ -342,10 +369,8 @@ func (b *Batch[T]) doReader(ctx context.Context) {
 				out = nil
 				continue
 			}
-			// Only doReader increments nextID, and Go resets it before
-			// starting this goroutine, so a plain counter is race-free.
-			id := b.nextID
-			b.nextID++
+			id := nextID
+			nextID++
 			b.items <- &Item[T]{
 				ID:   id,
 				Data: data,
@@ -413,15 +438,12 @@ func (b *Batch[T]) doProcessors(ctx context.Context) {
 
 	wg.Wait()
 
-	// Finalize shutdown under b.mu, closing done and clearing running before
-	// errs. A CollectErrors caller (which returns once errs closes) therefore
-	// always observes a fully torn-down, reusable Batch — done closed and
-	// running false — and a reusing Go() blocks on b.mu until this completes.
-	b.mu.Lock()
+	// 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.running = false
 	close(b.errs)
-	b.mu.Unlock()
 }
 
 // fixConfig corrects invalid ConfigValues to ensure consistent batch behavior.
diff --git a/batch/batch_test.go b/batch/batch_test.go
index d7a5b1e..54c3d12 100644
--- a/batch/batch_test.go
+++ b/batch/batch_test.go
@@ -22,7 +22,7 @@ func TestBatch_ProcessorChainingAndErrorTracking(t *testing.T) {
 		errProc := &errorPerItemProcessor{FailEvery: 3}
 		countProc := &countProcessor{count: &count}
 
-		errs := batch.Go(context.Background(), src, errProc, countProc)
+		errs, _ := batch.Go(context.Background(), src, errProc, countProc)
 
 		received := 0
 		for err := range errs {
@@ -52,7 +52,7 @@ 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, _ := batch.Go(context.Background(), src, countProc)
 		<-batch.Done()
 
 		var found bool
@@ -75,7 +75,7 @@ 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, _ := batch.Go(context.Background(), src, proc)
 
 		var found bool
 		var unwrappedErr error
@@ -120,7 +120,7 @@ func TestBatch_ProcessorChainingAndErrorTracking(t *testing.T) {
 		ctx, cancel := context.WithCancel(context.Background())
 
 		// Start processing
-		_ = batch.Go(ctx, src, proc)
+		_, _ = batch.Go(ctx, src, proc)
 
 		// Give some time for processing to start
 		time.Sleep(50 * time.Millisecond)
@@ -290,7 +290,7 @@ func TestBatch_ProcessorChainingAndErrorTracking(t *testing.T) {
 					},
 				}
 
-				_ = batch.Go(context.Background(), src, proc)
+				_, _ = batch.Go(context.Background(), src, proc)
 				<-batch.Done()
 
 				got := int(atomic.LoadUint32(&count))
@@ -388,7 +388,7 @@ func TestBatch_ProcessorChainingAndErrorTracking(t *testing.T) {
 			ctx := context.Background()
 
 			// Start processing and wait for completion
-			errs := b.Go(ctx, s, p)
+			errs, _ := b.Go(ctx, s, p)
 			for range errs {
 				// Consume errors
 			}
@@ -460,7 +460,7 @@ func TestBatch_ComplexProcessingPipeline(t *testing.T) {
 		var count uint32
 		counter := &countProcessor{count: &count}
 
-		errs := batch.Go(context.Background(), src, transformer, filter, counter)
+		errs, _ := batch.Go(context.Background(), src, transformer, filter, counter)
 
 		// Drain errors
 		for range errs {
@@ -506,7 +506,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 +544,7 @@ func TestBatch_RobustnessAndEdgeCases(t *testing.T) {
 		proc := &countProcessor{count: &count}
 
 		// Process large batch
-		errs := batch.Go(context.Background(), src, proc)
+		errs, _ := batch.Go(context.Background(), src, proc)
 		<-batch.Done()
 
 		// Drain errors
@@ -567,7 +567,7 @@ func TestBatch_RobustnessAndEdgeCases(t *testing.T) {
 		var count uint32
 		proc := &countProcessor{count: &count}
 
-		errs := batch.Go(context.Background(), src, proc)
+		errs, _ := batch.Go(context.Background(), src, proc)
 		<-batch.Done()
 
 		// Drain errors
@@ -595,7 +595,7 @@ func TestBatch_RobustnessAndEdgeCases(t *testing.T) {
 		var count uint32
 		proc := &countProcessor{count: &count}
 
-		errs := batch.Go(context.Background(), src, proc)
+		errs, _ := batch.Go(context.Background(), src, proc)
 		<-batch.Done()
 
 		// Drain errors
@@ -623,7 +623,7 @@ func TestBatch_RobustnessAndEdgeCases(t *testing.T) {
 		var count uint32
 		proc := &countProcessor{count: &count}
 
-		errs := batch.Go(context.Background(), src, proc)
+		errs, _ := batch.Go(context.Background(), src, proc)
 		<-batch.Done()
 
 		// Drain errors
@@ -647,7 +647,7 @@ func TestBatch_NoProcessors(t *testing.T) {
 		src := &testSource{Items: items}
 
 		// Call Go with source but no processors
-		errs := batch.Go(context.Background(), src)
+		errs, _ := batch.Go(context.Background(), src)
 
 		// Count errors instead of collecting them
 		errorCount := 0
@@ -675,7 +675,7 @@ 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, _ := batch.Go(context.Background(), src, emptyProcessors...)
 
 		// Use a counter instead of collecting errors
 		errorCount := 0
@@ -716,7 +716,7 @@ func TestBatch_NoTimersWithMinItems(t *testing.T) {
 			},
 		}
 
-		errs := batch.Go(context.Background(), src, proc)
+		errs, _ := batch.Go(context.Background(), src, proc)
 		<-batch.Done()
 		for range errs {
 			// Drain errors
@@ -746,7 +746,7 @@ 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, _ := b.Go(context.Background(), src)
 		<-b.Done()
 		for range errs {
 		}
diff --git a/batch/benchmark_test.go b/batch/benchmark_test.go
index f1e1217..12ed0c2 100644
--- a/batch/benchmark_test.go
+++ b/batch/benchmark_test.go
@@ -41,8 +41,8 @@ func (benchPassthrough) Process(_ context.Context, items []*Item[any]) ([]*Item[
 // 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 so the benchmark runs
-// identically on master, where Batch reuse is not yet supported.
+// setup cost. A fresh Batch is created each iteration, which is also the
+// required usage: a Batch is single-use.
 //
 // Compare across branches with:
 //
@@ -59,7 +59,8 @@ func BenchmarkBatchThroughput(b *testing.B) {
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		bt := New[any](cfg)
-		IgnoreErrors(bt.Go(ctx, &benchSource{n: itemsPerRun}, proc))
+		errs, _ := bt.Go(ctx, &benchSource{n: itemsPerRun}, proc)
+		IgnoreErrors(errs)
 		<-bt.Done()
 	}
 }
diff --git a/batch/buffer_config_test.go b/batch/buffer_config_test.go
index e8b2f3c..1828bd7 100644
--- a/batch/buffer_config_test.go
+++ b/batch/buffer_config_test.go
@@ -48,7 +48,7 @@ func TestBatch_WithBufferConfig(t *testing.T) {
 		})
 
 		// Start processing to trigger channel creation
-		errs := b.Go(context.Background(), src)
+		errs, _ := b.Go(context.Background(), src)
 		IgnoreErrors(errs)
 		<-b.Done()
 
@@ -76,7 +76,7 @@ func TestBatch_WithBufferConfig(t *testing.T) {
 		})
 
 		// Start processing - should use default buffer sizes
-		errs := b.Go(context.Background(), src)
+		errs, _ := b.Go(context.Background(), src)
 		IgnoreErrors(errs)
 		<-b.Done()
 
@@ -99,7 +99,7 @@ func TestBatch_WithBufferConfig(t *testing.T) {
 		})
 
 		// Start batch processing
-		errs := b.Go(context.Background(), src)
+		errs, _ := b.Go(context.Background(), src)
 
 		// Should panic when trying to set buffer config after Go
 		defer func() {
@@ -149,7 +149,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)
 		}()
 
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..875cb91 100644
--- a/batch/dynamic_config_test.go
+++ b/batch/dynamic_config_test.go
@@ -53,7 +53,7 @@ func TestBatch_DynamicConfiguration(t *testing.T) {
 		}
 
 		// Start batch processing with initial config
-		errs := batch.Go(context.Background(), src, proc)
+		errs, _ := batch.Go(context.Background(), src, proc)
 
 		// 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..c5e1029 100644
--- a/batch/error_handling_test.go
+++ b/batch/error_handling_test.go
@@ -21,7 +21,7 @@ func TestBatch_ErrorHandling(t *testing.T) {
 			processorErr: procErr,
 		}
 
-		errs := batch.Go(context.Background(), src, proc)
+		errs, _ := batch.Go(context.Background(), src, proc)
 
 		var foundErr bool
 		for err := range errs {
@@ -49,7 +49,7 @@ func TestBatch_ErrorHandling(t *testing.T) {
 
 		proc := &countProcessor{count: new(uint32)}
 
-		errs := batch.Go(context.Background(), src, proc)
+		errs, _ := batch.Go(context.Background(), src, proc)
 
 		var foundErr bool
 		for err := range errs {
@@ -66,32 +66,9 @@ 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)
-		}
-	})
+	// Nil-source handling is covered by TestGo_NilSourceReturnsErrNilSource,
+	// which asserts the start error is reported via Go's return value rather
+	// than the pipeline error channel.
 
 	t.Run("nil processor filtering", func(t *testing.T) {
 		batch := New[any](NewConstantConfig(&ConfigValues{}))
@@ -102,7 +79,7 @@ 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, _ := batch.Go(context.Background(), src, nil, validProc, nil)
 
 		// Count errors instead of collecting them
 		errorCount := 0
@@ -131,7 +108,7 @@ func TestBatch_NilChannelHandling(t *testing.T) {
 		// Create a source that returns a nil output channel
 		nilChannelSource := &nilOutputChannelSource{}
 
-		errs := batch.Go(context.Background(), nilChannelSource)
+		errs, _ := batch.Go(context.Background(), nilChannelSource)
 
 		var foundErr bool
 		var errMsg string
@@ -160,7 +137,7 @@ func TestBatch_NilChannelHandling(t *testing.T) {
 		// Create a source that returns a nil error channel
 		nilChannelSource := &nilErrorChannelSource{}
 
-		errs := batch.Go(context.Background(), nilChannelSource)
+		errs, _ := batch.Go(context.Background(), nilChannelSource)
 
 		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 bbc467a..5262161 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
@@ -32,7 +37,11 @@ func IgnoreErrors(errs <-chan error) {
 //
 // Example usage:
 //
-//	errs := batch.CollectErrors(myBatch.Go(ctx, source, processor))
+//	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)
@@ -61,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
@@ -115,7 +129,13 @@ func ExecuteBatches[T any](ctx context.Context, configs ...*BatchConfig[T]) []er
 				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/reuse_test.go b/batch/reuse_test.go
deleted file mode 100644
index 97338f8..0000000
--- a/batch/reuse_test.go
+++ /dev/null
@@ -1,100 +0,0 @@
-package batch_test
-
-import (
-	"context"
-	"sync"
-	"testing"
-
-	. "github.com/MasterOfBinary/gobatch/batch"
-)
-
-// TestReuse_AfterCollectErrors verifies the CollectErrors completion
-// contract: once CollectErrors returns, the Batch is fully torn down
-// (Done closed, running cleared) so the same Batch can be reused
-// immediately without hitting the "Concurrent calls" panic.
-//
-// This is a regression test for shutdown ordering in doProcessors: if
-// b.errs is closed before b.done/b.running are finalized, a caller that
-// follows the documented "no need to wait on Done()" guidance and reuses
-// the Batch can intermittently panic. Run with -race to surface it.
-func TestReuse_AfterCollectErrors(t *testing.T) {
-	b := New[any](NewConstantConfig(&ConfigValues{MinItems: 1, MaxItems: 5}))
-
-	for i := 0; i < 200; i++ {
-		src := &testSource{Items: []any{1, 2, 3, 4, 5}}
-		proc := &countProcessor{}
-
-		// CollectErrors blocks until b.errs is closed; per its docs the
-		// caller need not also wait on Done(). Reuse b on the very next
-		// line — this panics if running is still true at that point.
-		errs := CollectErrors(b.Go(context.Background(), src, proc))
-		for range errs {
-		}
-	}
-}
-
-// TestReuse_AfterDone verifies the other synchronization path flagged in
-// review: a caller that waits on <-Done() and then immediately reuses the
-// Batch must not hit the "Concurrent calls" panic. This requires that
-// running is cleared no later than Done() becoming observable. Shutdown
-// closes done and clears running inside the same b.mu section that
-// doProcessors holds across both, so a Go() from a Done() waiter blocks
-// on the mutex until running is false. Run with -race -count to surface
-// any regression.
-func TestReuse_AfterDone(t *testing.T) {
-	b := New[any](NewConstantConfig(&ConfigValues{MinItems: 1, MaxItems: 5}))
-
-	for i := 0; i < 200; i++ {
-		src := &testSource{Items: []any{1, 2, 3, 4, 5}}
-		proc := &countProcessor{}
-
-		errs := b.Go(context.Background(), src, proc)
-		IgnoreErrors(errs)
-
-		// Wake on Done() and reuse the Batch immediately. If running is
-		// still true when Done() is observable, the next Go() panics.
-		<-b.Done()
-	}
-}
-
-// idRecorder is a processor that records the ID of every item it processes.
-type idRecorder struct {
-	mu  sync.Mutex
-	ids []uint64
-}
-
-func (r *idRecorder) Process(_ context.Context, items []*Item[any]) ([]*Item[any], error) {
-	r.mu.Lock()
-	for _, it := range items {
-		r.ids = append(r.ids, it.ID)
-	}
-	r.mu.Unlock()
-	return items, nil
-}
-
-// TestReuse_RestartsIDsFromZero pins the core ID contract: every run of a
-// reused Batch numbers items 0..n-1. This guards against the per-run reset
-// silently regressing (e.g. IDs starting at 1, or not resetting between runs)
-// — a failure mode the existing reuse tests, which only assert "no panic",
-// would not catch.
-func TestReuse_RestartsIDsFromZero(t *testing.T) {
-	// MinItems == MaxItems == 5 with a 5-item source yields exactly one batch
-	// per run, so the recorded IDs are 0..4 in order.
-	b := New[any](NewConstantConfig(&ConfigValues{MinItems: 5, MaxItems: 5}))
-
-	for iter := 0; iter < 5; iter++ {
-		rec := &idRecorder{}
-		IgnoreErrors(b.Go(context.Background(), &testSource{Items: []any{10, 20, 30, 40, 50}}, rec))
-		<-b.Done()
-
-		want := []uint64{0, 1, 2, 3, 4}
-		if len(rec.ids) != len(want) {
-			t.Fatalf("iteration %d: got %d IDs %v, want %v", iter, len(rec.ids), rec.ids, want)
-		}
-		for i, id := range rec.ids {
-			if id != want[i] {
-				t.Fatalf("iteration %d: IDs = %v, want %v", iter, rec.ids, want)
-			}
-		}
-	}
-}
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

From 03fb0fa4e38c9769162491e1911dbbf6c50cbdfd Mon Sep 17 00:00:00 2001
From: Vaughn Friesen <vdfriesen@vaughnfriesen.com>
Date: Fri, 29 May 2026 23:09:59 +0800
Subject: [PATCH 3/5] fix(batch): ExecuteBatches surfaces ErrNilSource;
 document nil Config default

ExecuteBatches skipped a BatchConfig whose Source was nil before ever calling
Go, so a missing source produced no work and no error and callers could
advance as if the batch had completed. Drop the nil-source check from the skip
guard so cfg.B.Go reports ErrNilSource, which is then collected. Nil
*BatchConfig / nil Batch entries remain a tolerated silent skip (unchanged,
separately tested contract).

Docs: note in the README that a nil Config (or the zero-value Batch) uses the
default configuration, and add the missing "log" import to the Basic Usage
example.

Adds TestExecuteBatches_NilSourceSurfacesError.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 README.md             |  3 +++
 batch/helpers.go      |  5 ++++-
 batch/helpers_test.go | 30 ++++++++++++++++++++++++++++++
 3 files changed, 37 insertions(+), 1 deletion(-)

diff --git a/README.md b/README.md
index 68b1a4b..56afdbc 100644
--- a/README.md
+++ b/README.md
@@ -124,6 +124,7 @@ package main
 import (
 	"context"
 	"fmt"
+	"log"
 	"time"
 
 	"github.com/MasterOfBinary/gobatch/batch"
@@ -203,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.
diff --git a/batch/helpers.go b/batch/helpers.go
index 5262161..eb8f965 100644
--- a/batch/helpers.go
+++ b/batch/helpers.go
@@ -125,7 +125,10 @@ 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
 			}
 
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 {

From d9e9611e6205add76034dedc4a7a02046b6ebb65 Mon Sep 17 00:00:00 2001
From: Vaughn Friesen <vdfriesen@vaughnfriesen.com>
Date: Fri, 29 May 2026 23:36:33 +0800
Subject: [PATCH 4/5] test(batch): assert Go's start error is nil at call sites

Capture and assert err == nil after Batch.Go in tests instead of discarding
it, so an unexpected start error fails loudly rather than surfacing as a
confusing downstream symptom. Two Go calls inside spawned goroutines are left
discarding the error (t.Fatal is illegal off the test goroutine, and a
downstream assertion already catches a start error there).

Also drop a review-only comment in error_handling_test.go that explained where
nil-source coverage moved; that context belongs on the PR, not in the code.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 batch/batch_test.go          | 75 ++++++++++++++++++++++++++++--------
 batch/benchmark_test.go      |  5 ++-
 batch/buffer_config_test.go  | 15 ++++++--
 batch/dynamic_config_test.go |  5 ++-
 batch/error_handling_test.go | 29 +++++++++-----
 5 files changed, 100 insertions(+), 29 deletions(-)

diff --git a/batch/batch_test.go b/batch/batch_test.go
index 54c3d12..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 {
@@ -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
index 12ed0c2..d73e258 100644
--- a/batch/benchmark_test.go
+++ b/batch/benchmark_test.go
@@ -59,7 +59,10 @@ func BenchmarkBatchThroughput(b *testing.B) {
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		bt := New[any](cfg)
-		errs, _ := bt.Go(ctx, &benchSource{n: itemsPerRun}, proc)
+		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 1828bd7..2ec263a 100644
--- a/batch/buffer_config_test.go
+++ b/batch/buffer_config_test.go
@@ -48,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()
 
@@ -76,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()
 
@@ -99,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() {
diff --git a/batch/dynamic_config_test.go b/batch/dynamic_config_test.go
index 875cb91..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 c5e1029..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,10 +72,6 @@ func TestBatch_ErrorHandling(t *testing.T) {
 		}
 	})
 
-	// Nil-source handling is covered by TestGo_NilSourceReturnsErrNilSource,
-	// which asserts the start error is reported via Go's return value rather
-	// than the pipeline error channel.
-
 	t.Run("nil processor filtering", func(t *testing.T) {
 		batch := New[any](NewConstantConfig(&ConfigValues{}))
 		src := &testSource{Items: []any{1, 2, 3}}
@@ -79,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
@@ -108,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
@@ -137,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

From d49ace643129a27b0e59de81958ea93ee518aca7 Mon Sep 17 00:00:00 2001
From: Vaughn Friesen <vdfriesen@vaughnfriesen.com>
Date: Sat, 30 May 2026 00:10:53 +0800
Subject: [PATCH 5/5] fix(batch): check Go's start error in TestMaxTimeIdle
 (errcheck)

The single-use refactor made Go return (<-chan error, error); this call
site still dropped the error, which golangci-lint's errcheck flags. Capture
and assert the start error and drain the channel, matching the other call
sites.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 batch/maxtime_test.go | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

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)