diff --git a/.planning/codebase/ARCHITECTURE.md b/.planning/codebase/ARCHITECTURE.md new file mode 100644 index 0000000..b56128f --- /dev/null +++ b/.planning/codebase/ARCHITECTURE.md @@ -0,0 +1,38 @@ +# Architecture + +GoBatch follows a **pipeline-based batch processing architecture** with five distinct layers: + +1. **Source Layer** (`source/` package): Data ingestion abstraction. Implementations include Channel, Error, and Nil sources that return item and error channels. + +2. **Batching/Orchestration Layer** (`batch/batch.go`): Core Batch type coordinates two goroutines - doReader (assigns each item a unique ID via an atomic counter and forwards it on the items channel) and doProcessors (collects items into batches and chains them through Processors). Cancellation flows from `ctx` into `waitForItems`, which returns any partial batch immediately and then drains remaining buffered items one at a time so already-read items are never dropped. + +3. **Configuration Layer** (`batch/config.go`): Config interface with ConstantConfig and DynamicConfig implementations control batching timing (MinTime/MaxTime) and sizing (MinItems/MaxItems) with priority: `MaxTime = MaxItems > EOF > MinTime > MinItems`. + +4. **Processor Layer** (`processor/` package): Processing abstraction for batch transformations. Built-in processors: Transform (modify data), Filter (remove items), Channel (write to output), Error (simulate failures), Nil (passthrough). + +5. **Helper/Utility Layer** (`batch/helpers.go`, `batch/errors.go`): Convenience functions (IgnoreErrors, CollectErrors, RunBatchAndWait, ExecuteBatches) and error types (SourceError, ProcessorError, ItemError). ItemError carries the failing `ItemID` so callers can correlate per-item failures with specific items, while ProcessorError remains for processor-wide failures. + +## Data Flow + +Source.Read() → doReader assigns IDs via `atomic.AddUint64` and forwards items → items accumulate in `b.items` → waitForItems determines batch ready based on Config (or returns early on `ctx.Done()`) → doProcessors batches spawn goroutines → each batch flows through Processor chain sequentially → errors wrapped (`SourceError` / `ProcessorError` / `ItemError`) and sent on error channel → Done signals completion. + +## Key Abstractions + +- **Source[T]**: Must return (items, errors) channels, close both, respect context +- **Processor[T]**: Takes batch, returns modified batch + error, respects context +- **Config**: Get() called per batch, supports dynamic runtime updates +- **Item[T]**: ID (immutable), Data (mutable), Error (settable) + +## Entry Points + +- `Batch.Go()`: Starts pipeline, returns error channel +- `Batch.Done()`: Returns channel that closes on completion +- Helpers provide simplified patterns (RunBatchAndWait combines Go/errors/Done) + +## Cancellation Contract + +`waitForItems` honors `ctx.Done()` for early shutdown, but it ultimately blocks until `b.items` closes. `b.items` is closed by `doReader` only after the source closes both of its channels, so cancellation only takes effect once the source observes ctx. **Sources that ignore ctx will block the pipeline indefinitely after cancel.** Source authors must propagate ctx into their `Read` goroutine. + +--- + +*Architecture analysis: 2026-04-10. Updated 2026-04-29 to reflect the atomic-counter ID generator and ctx-aware waitForItems.* diff --git a/.planning/codebase/CONCERNS.md b/.planning/codebase/CONCERNS.md new file mode 100644 index 0000000..a711476 --- /dev/null +++ b/.planning/codebase/CONCERNS.md @@ -0,0 +1,53 @@ +# Concerns + +## High Priority + +### Panic-based Error Handling +- **Location:** `batch/batch.go` — `WithBufferConfig()` and `Go()` +- **Issue:** Uses `panic()` for invalid configuration and double-start detection instead of returning errors +- **Risk:** Callers cannot gracefully handle these failures; panics propagate up and crash the program +- **Suggestion:** Return errors instead of panicking, or document panic behavior clearly + +## Medium Priority + +### ~~Unused Context Parameter in waitForItems~~ — RESOLVED (2026-04-29) +- **Location:** `batch/batch.go` — `waitForItems()` method +- **Resolution:** ctx is now wired into the select. On `ctx.Done()` `waitForItems` returns any partial batch immediately, then drains remaining buffered items one at a time so already-read items are not dropped. Subject to the cancellation contract documented in ARCHITECTURE.md (the source must close its channels in response to ctx — sources that ignore ctx will still block). + +### Silent Configuration Adjustment +- **Location:** `batch/config.go` — `fixConfig()` function +- **Issue:** Silently adjusts invalid configuration values (e.g., negative times, zero items) without warning +- **Risk:** Users may not realize their configuration was modified; debugging unexpected behavior is harder +- **Suggestion:** Log warnings when configuration values are adjusted, or return validation errors + +### MaxTime Timer Edge Cases — PARTIALLY ADDRESSED (2026-04-29) +- **Location:** `batch/batch.go` — timer handling in waitForItems +- **Resolved part:** Timer leaks are fixed. `time.After` was replaced with `time.NewTimer` plus `defer Stop()` so timers cannot leak when a batch returns before its timer fires. +- **Remaining issue:** `waitForItems` still resets `maxTimer` indefinitely when it fires with an empty batch (`maxTimer.Reset(config.MaxTime)`). Under a pathological source that produces no items, the loop will spin every `MaxTime` interval. The cancellation path closes one escape hatch, but an explicit idle-timeout policy is still worth considering. + +### Lock Contention in ExecuteBatches +- **Location:** `batch/helpers.go` — `ExecuteBatches()` function +- **Issue:** Uses shared mutex to collect errors from multiple concurrent batches +- **Risk:** Under high concurrency with many errors, lock contention could impact performance +- **Suggestion:** Use channel-based error collection or sync.Map for concurrent append + +## Low Priority + +### Processor Contract Documentation — PARTIALLY ADDRESSED (2026-04-29) +- **Issue:** The exact contract for Processor implementations (when to check item.Error, when to skip vs. process errored items) is implicit +- **Resolved part:** Per-item errors are now reported as `*ItemError` (with the failing `ItemID`) rather than a generic `*ProcessorError`. This makes the conceptual distinction between processor-wide and item-specific failures explicit at the type level. +- **Remaining issue:** The Processor godoc still does not formally specify whether processors should skip items with `Error != nil` or process them. Custom processors will continue to handle this inconsistently until the interface contract is documented. + +### ~~ID Overflow Risk~~ — IMPLEMENTATION SIMPLIFIED (2026-04-29) +- **Location:** `batch/batch.go` — `doReader()` (was `doIDGenerator()`) +- **Status:** The dedicated `doIDGenerator` goroutine and channel are gone. IDs are now produced inline in `doReader` via `atomic.AddUint64(&b.nextID, 1) - 1`. The uint64 overflow property is unchanged — IDs still wrap to 0 after 2^64 items, which remains an acceptable risk in practice. + +## Version Stability + +- Project is explicitly v0 (pre-1.0) — breaking changes expected on master branch +- No tagged releases or semantic versioning in use +- API recently migrated to generics (commit `7a85eca`) + +--- + +*Concerns analysis: 2026-04-10. Updated 2026-04-29 to reflect the cancellation-and-IDs refactor; the High-priority panic concern remains open.* diff --git a/.planning/codebase/CONVENTIONS.md b/.planning/codebase/CONVENTIONS.md new file mode 100644 index 0000000..dd8a026 --- /dev/null +++ b/.planning/codebase/CONVENTIONS.md @@ -0,0 +1,80 @@ +# Conventions + +## Code Style + +- Standard `gofmt` formatting — no custom linter configuration +- Idiomatic Go patterns throughout +- CI runs `go vet ./...` and `golangci-lint run --timeout=3m` + +## Naming + +### Types +- PascalCase for exported types: `Batch[T]`, `Source[T]`, `Processor[T]`, `Item[T]`, `Config`, `ConstantConfig`, `DynamicConfig` +- Error types: `SourceError`, `ProcessorError`, `ItemError` +- Generics use brackets: `[T any]` + +### Functions +- PascalCase for exported: `New[T]()`, `Go()`, `Done()`, `IgnoreErrors()`, `CollectErrors()` +- camelCase for unexported: `doReader()`, `doProcessors()`, `waitForItems()`, `fixConfig()` + +### Files +- Lowercase, single-word names: `batch.go`, `config.go`, `errors.go`, `helpers.go` +- Implementation files named after primary type: `transform.go`, `filter.go`, `channel.go` +- Test files: `{name}_test.go` +- Package docs: `doc.go` in each package + +## Patterns + +### Constructor Pattern +- `New[T]()` constructors for processor and source types +- Accept functional dependencies (functions, channels) as constructor params +- Example: `processor.NewTransform[T](func(data T) (T, error))` in `processor/transform.go` + +### Interface Design +- Small, focused interfaces: `Source[T]` (Read), `Processor[T]` (Process), `Config` (Get) +- Generic type parameters on interfaces for type safety +- Interfaces defined in `batch/batch.go` alongside the core Batch type + +### Concurrency +- `sync.Mutex` for protecting shared state (Batch.mu) +- `sync.WaitGroup` for goroutine coordination +- `sync/atomic` for the per-batch ID counter (`Batch.nextID`) +- Channel-based communication between pipeline stages +- `context.Context` for cancellation propagation; `waitForItems` watches `ctx.Done()` +- Goroutines spawned in `Go()`: doReader, doProcessors + +### Configuration +- `Config` interface with `Get()` method returning `ConfigValues` +- `ConstantConfig`: immutable, set at creation +- `DynamicConfig`: runtime-updatable via `Update()` with mutex protection +- `BufferConfig`: controls internal channel buffer sizes + +## Error Handling + +### Custom Error Types +- `SourceError` wraps errors returned by `Source.Read()` — defined in `batch/errors.go` +- `ProcessorError` wraps processor-wide errors returned as the second value of `Processor.Process()` — defined in `batch/errors.go` +- `ItemError` wraps per-item failures (`item.Error`) and carries the failing `ItemID` — defined in `batch/errors.go` +- All three implement `Unwrap()` for `errors.As()` inspection + +### Error Propagation +- Errors sent on dedicated error channel returned by `Batch.Go()` +- Item-level errors tracked via `Item.Error` field +- Processors should check and respect existing item errors +- Processing continues despite individual item errors + +### Error Utilities +- `IgnoreErrors()`: drains error channel in background goroutine +- `CollectErrors()`: collects all errors into slice after completion +- `RunBatchAndWait()`: combines Go/errors/Done into single call + +## Documentation Style + +- Godoc comments on all exported types, functions, methods, constants +- Package-level `doc.go` in each package with overview and examples +- Interface documentation includes usage examples +- Method docs explain parameters, return values, and behavior + +--- + +*Conventions analysis: 2026-04-10. Updated 2026-04-29 for the cancellation-and-IDs refactor (atomic counter, ItemError type, ctx-aware waitForItems).* diff --git a/.planning/codebase/INTEGRATIONS.md b/.planning/codebase/INTEGRATIONS.md new file mode 100644 index 0000000..7260cbf --- /dev/null +++ b/.planning/codebase/INTEGRATIONS.md @@ -0,0 +1,90 @@ +# External Integrations + +**Analysis Date:** 2026-04-10 + +## APIs & External Services + +**Not Applicable** +- GoBatch is a library, not an application with external service integrations +- Library users define their own external integrations via Source and Processor interfaces + +## Data Storage + +**Databases:** +- Not built-in - Library users implement custom Source interface for database connectivity + +**File Storage:** +- Not built-in - Library users implement custom Source interface for file system access + +**Caching:** +- Not built-in - Library users implement custom Processor interface for cache operations + +## Authentication & Identity + +**Auth Provider:** +- Not applicable - GoBatch is a library without built-in authentication + +## Monitoring & Observability + +**Error Tracking:** +- Errors returned through error channels +- Error wrapping provided: `SourceError`, `ProcessorError`, and `ItemError` (with `ItemID`) types in `batch/errors.go` +- Users can implement custom error handling via channels + +**Logs:** +- Logging not built-in - Library users implement custom logging via Processor interface +- Error chain maintained via `errors.Unwrap()` for debugging + +## CI/CD & Deployment + +**Hosting:** +- GitHub repository: `github.com/MasterOfBinary/gobatch` +- Package registry: pkg.go.dev (automatic via Go module) + +**CI Pipeline:** +- GitHub Actions (`.github/workflows/go.yml`) +- Triggers: Pushes to master/travis-test, pull requests to master +- Matrix testing: Go versions 1.25.x and 1.26.x + +**Coverage Integration:** +- Codecov + - Service: `codecov/codecov-action@v5` + - Auth: `CODECOV_TOKEN` secret + - Purpose: Track test coverage metrics + +## Environment Configuration + +**Required env vars:** +- `CODECOV_TOKEN` - For coverage uploads (CI only, secret) + +**Secrets location:** +- GitHub Secrets (repository-level) +- Not stored in codebase + +## Webhooks & Callbacks + +**Incoming:** +- Not applicable - Library without HTTP endpoints + +**Outgoing:** +- Codecov webhook (automatic coverage uploads via CI) + +## Extensibility Points + +**User-Implemented Integrations:** +- Custom `Source[T]` implementations: `source/doc.go` + - Database sources + - API sources + - File system sources + - Message queue sources + +- Custom `Processor[T]` implementations: `processor/doc.go` + - Database writes + - API calls + - File writes + - Cache operations + - External service notifications + +--- + +*Integration audit: 2026-04-10* diff --git a/.planning/codebase/STACK.md b/.planning/codebase/STACK.md new file mode 100644 index 0000000..57776fb --- /dev/null +++ b/.planning/codebase/STACK.md @@ -0,0 +1,95 @@ +# Technology Stack + +**Analysis Date:** 2026-04-10 + +## Languages + +**Primary:** +- Go 1.18+ - Core library implementation + +**Supported:** +- Go 1.25, 1.26 - Tested via CI/CD pipeline (GitHub Actions) + +## Runtime + +**Environment:** +- Go compiler and runtime (no external runtime dependencies) + +**Package Manager:** +- Go modules (go mod) +- Lockfile: Not applicable (no external dependencies) + +## Frameworks + +**Core:** +- None - Pure Go standard library implementation + +**Testing:** +- Go's built-in `testing` package - Used for all unit tests +- Custom test helpers in `batch/testhelpers_test.go` - Test utilities + +**Build/Dev:** +- `gofmt` - Code formatting (enforced in CI) +- `golangci-lint` v1.64.8 - Linting and code quality +- `go vet` - Basic static analysis + +## Key Dependencies + +**Zero External Dependencies:** +- GoBatch has no external production dependencies +- All functionality built on Go standard library only +- Standard library packages used: + - `context` - Context management + - `errors` - Error handling and wrapping + - `sync` - Synchronization primitives (Mutex) + - `time` - Timing and duration handling + - `fmt` - String formatting + +## Configuration + +**Environment:** +- No environment configuration required for the library itself +- CI/CD environment: Ubuntu latest (via GitHub Actions) + +**Build:** +- Standard Go build commands +- No build configuration files (go.mod/go.sum only) + +## Platform Requirements + +**Development:** +- Go 1.18 or later +- gofmt (included with Go) +- golangci-lint (for linting, installed via CI workflow) +- Unix-like environment for bash scripts + +**Production:** +- Go 1.18 or later +- No external runtime dependencies +- Can be compiled for any platform supported by Go + +## Testing Infrastructure + +**Test Execution:** +- Run all tests with race detection and coverage: `go test -race -coverprofile=coverage.txt -covermode=atomic ./...` +- Coverage tracking via Codecov (secrets-based authentication) +- Tests located alongside source files: `*_test.go` pattern + +**Linting:** +- Format check: `gofmt -l $(git ls-files '*.go')` +- Lint: `golangci-lint run --timeout=3m` + +## CI/CD Platform + +**GitHub Actions:** +- Workflow file: `.github/workflows/go.yml` +- Triggers: Push to master/travis-test branches, pull requests to master +- Jobs: + - Code formatting verification + - Tests with race detection and coverage + - Linter checks + - Coverage upload to Codecov + +--- + +*Stack analysis: 2026-04-10* diff --git a/.planning/codebase/STRUCTURE.md b/.planning/codebase/STRUCTURE.md new file mode 100644 index 0000000..6a3f869 --- /dev/null +++ b/.planning/codebase/STRUCTURE.md @@ -0,0 +1,128 @@ +# Structure + +## Directory Layout + +``` +/Users/vaughn/dev/gobatch/ +├── batch/ # Core batching engine +│ ├── batch.go # Batch type, Source/Processor interfaces +│ ├── config.go # Config interface, ConstantConfig, DynamicConfig +│ ├── errors.go # SourceError, ProcessorError types +│ ├── helpers.go # IgnoreErrors, CollectErrors, RunBatchAndWait, ExecuteBatches +│ ├── constants.go # Default buffer sizes +│ ├── doc.go # Package documentation +│ └── *_test.go # Tests (50+ test files covering all functionality) +├── processor/ # Processor implementations +│ ├── transform.go # Transform processor (modify data) +│ ├── filter.go # Filter processor (remove items) +│ ├── channel.go # Channel processor (write to output) +│ ├── error.go # Error processor (simulate failures) +│ ├── nil.go # Nil processor (passthrough) +│ ├── doc.go # Package documentation +│ └── *_test.go # Tests +├── source/ # Source implementations +│ ├── channel.go # Channel source +│ ├── error.go # Error source +│ ├── nil.go # Nil source +│ ├── doc.go # Package documentation +│ └── *_test.go # Tests +├── doc.go # Root package documentation +├── example_test.go # Top-level usage examples +└── .planning/ + └── codebase/ # Documentation (this directory) +``` + +## Key File Locations + +### Entry Points +- `batch/batch.go`: Main Batch type, Go() and Done() methods +- `example_test.go`: Complete usage example showing source → processor pipeline + +### Configuration +- `batch/config.go`: Config interface (Get method), ConstantConfig (static), DynamicConfig (runtime-updatable) + +### Core Logic +- `batch/batch.go` — `Go()` starts the pipeline (assigns IDs via an atomic counter, spawns `doReader` and `doProcessors`) +- `batch/batch.go` — `waitForItems` implements batching strategy and honors `ctx.Done()` for cancellation drain +- `batch/batch.go` — `doProcessors` runs the processor chain and emits `*ItemError` for per-item failures, `*ProcessorError` for processor-wide failures + +### Interfaces +- `batch/batch.go` — `Source[T]`: `Read(ctx)` returns items and errors channels; both must be closed when the source observes ctx done +- `batch/batch.go` — `Processor[T]`: `Process(ctx, items)` returns modified items + error + +### Error Handling +- `batch/errors.go`: `SourceError`, `ProcessorError`, `ItemError` (with `ItemID`), all implementing `Unwrap()` for error chaining + +### Utilities +- `batch/helpers.go`: IgnoreErrors, CollectErrors, RunBatchAndWait, ExecuteBatches, BatchConfig +- `batch/constants.go`: Default buffer sizes + +## Naming Conventions + +### Files +- Source implementations: `source/{channel,error,nil}.go` +- Processor implementations: `processor/{transform,filter,channel,error,nil}.go` +- Tests: `{package}/{type}_test.go` or `example_test.go` for example-based tests + +### Directories +- Package name matches directory: `batch/`, `processor/`, `source/` +- Each package has `doc.go` for godoc package documentation + +### Functions +- Capitalized exported functions: `New[T]()`, `Go()`, `Done()`, `IgnoreErrors()`, `CollectErrors()` +- Capitalized exported methods: `Process()`, `Read()`, `Get()`, `Update()` +- Unexported goroutines: `doReader()`, `doProcessors()` + +### Types +- Capitalized: `Batch[T]`, `Source[T]`, `Processor[T]`, `Item[T]`, `Config`, `ConstantConfig`, `DynamicConfig` +- Error types: `SourceError`, `ProcessorError`, `ItemError` + +### Interfaces +- Named ending with capitalized letter: `Source[T]`, `Processor[T]`, `Config` +- Methods are capitalized: `Read()`, `Process()`, `Get()` + +## Where to Add New Code + +### New Processor Implementation +- Location: `processor/{name}.go` +- Must implement: `Process(ctx context.Context, items []*batch.Item[T]) ([]*batch.Item[T], error)` +- Follow Transform/Filter pattern: check item.Error, modify items, set item.Error or return error +- Add tests: `processor/{name}_test.go` +- Document: Add godoc comment to type and Process method + +### New Source Implementation +- Location: `source/{name}.go` +- Must implement: `Read(ctx context.Context) (<-chan T, <-chan error)` +- Must close both channels when done +- Respect context cancellation in select statements +- Add tests: `source/{name}_test.go` +- Document: Add godoc comment to type and Read method + +### New Helper Function +- Location: `batch/helpers.go` +- Signature should follow existing patterns (accept Batch, Source, Processor, return []error) +- Add tests: `batch/{name}_test.go` +- Document with godoc including example usage + +### Tests +- Co-located next to implementation (same directory) +- Naming: `{file}_test.go` or `example_{type}_test.go` for examples +- Table-driven tests preferred for multiple scenarios + +## Special Directories + +### batch/ +- Purpose: Core batching engine and orchestration +- Key invariant: Batch type must be thread-safe with mutex protection for state changes + +### processor/ +- Purpose: Built-in data transformation processors +- Key invariant: All processors must respect context cancellation + +### source/ +- Purpose: Built-in data source implementations +- Key invariant: All sources must close both channels they return + +--- + +*Structure analysis: 2026-04-10. Updated 2026-04-29 for the cancellation-and-IDs refactor (no more `doIDGenerator` goroutine, atomic ID counter, `ItemError` type).* diff --git a/.planning/codebase/TESTING.md b/.planning/codebase/TESTING.md new file mode 100644 index 0000000..4fff111 --- /dev/null +++ b/.planning/codebase/TESTING.md @@ -0,0 +1,67 @@ +# Testing + +## Framework + +- Go's built-in `testing` package — no external test frameworks +- Race detection enabled: `go test -race ./...` +- Coverage profiling: `go test -coverprofile=coverage.txt -covermode=atomic ./...` + +## Test Organization + +- Tests co-located with source files in same package +- Naming: `{file}_test.go` (e.g., `batch_test.go`, `config_test.go`, `transform_test.go`) +- Example tests: `example_test.go` at root and `example_{type}_test.go` in packages +- Test helpers: `testhelpers_test.go` in `batch/` package + +## Test Helpers + +Defined in `batch/testhelpers_test.go`: +- `testSource`: configurable test source that emits items on a channel +- `countProcessor`: counts items processed (for verification) +- `errorPerItemProcessor`: sets error on each item (for error path testing) +- Helper functions for creating test configurations and batch setups + +## Test Patterns + +### Table-Driven Tests +- Used for testing multiple scenarios with `t.Run()` subtests +- Common in config tests (`batch/config_test.go`) and processor tests + +### Concurrency Testing +- Race detection (`-race` flag) catches data races +- Tests verify goroutine cleanup and channel closure +- Context cancellation tests in `batch/cancellation_test.go` exercise the ctx-aware shutdown path: `TestCancellation_BufferedItemsDrained` verifies that already-buffered items are still processed after cancel, and `TestCancellation_SourceErrorNoRace` loops 50× to flush out the previous send-on-closed-channel race between `doReader` and `doProcessors`. + +### Error Path Testing +- Dedicated error sources (`source.NewError[T]()`) and processors (`processor.NewError[T]()`) +- Tests verify error wrapping with `errors.As()` for `SourceError`, `ProcessorError`, and `ItemError` +- Item-level error propagation tested via `TestCancellation_ItemErrorType` and `TestCancellation_ProcessorErrorDistinct` + +### Example Tests +- `example_test.go`: end-to-end pipeline examples using `Example` functions +- Serve as both documentation and regression tests +- Verified by `go test` (output checked against `// Output:` comments) + +## Coverage + +- CI runs with `-covermode=atomic` for accurate coverage with goroutines +- Coverage uploaded to Codecov via GitHub Actions (`.github/workflows/ci.yml`) + +## Test Commands + +```bash +# Run all tests with race detection and coverage +go test -race -coverprofile=coverage.txt -covermode=atomic ./... + +# Run specific package tests +go test -race ./batch/... +go test -race ./processor/... +go test -race ./source/... + +# Run specific test +go test -race -run TestBatchGo ./batch/... +``` + +--- + +*Testing analysis: 2026-04-10. Updated 2026-04-29 for the new `cancellation_test.go` and `ItemError` coverage.* diff --git a/CHANGELOG.md b/CHANGELOG.md index 5167109..d0f220b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,45 @@ Note: This project is in early development. The API may change without warning i ## [Unreleased] +### Added + +- New `ItemError` type for per-item errors. It carries the failing item's `ItemID` alongside the wrapped error so callers can correlate errors with specific items. + +### Changed + +- **BREAKING:** Per-item errors (where `item.Error != nil` after a processor returns) are now reported as `*ItemError` instead of `*ProcessorError`. Processor-wide errors (those returned as the second value of `Processor.Process`) are still reported as `*ProcessorError`. +- **BREAKING:** Removed `BufferConfig.IDBufferSize` and the `DefaultIDBufferSize` constant. IDs are now generated with an atomic counter, so the dedicated ID-generator goroutine and channel — and the buffer that backed them — are gone. +- Context cancellation now drains items 1-by-1 once any partial batch has been returned. Downstream consumers that rely on receiving full `MinItems`-sized batches will see smaller batches during shutdown. +- `CollectErrors` documentation clarified to note that it blocks until the error channel is closed; callers no longer need to wait on `Done()` afterward. + +### Fixed + +- `waitForItems` now honors `ctx.Done()`. Previously the context parameter was ignored, which could delay shutdown by up to `MaxTime` and cause the pipeline to keep waiting for batch thresholds long after cancellation. +- Replaced `time.After` with `time.NewTimer` plus `defer Stop()` in `waitForItems` so timers do not leak when a batch returns before its timer fires. +- Closed a "send on closed channel" race window during cancellation: previously, if `waitForItems` had returned and `doProcessors` had closed `b.errs` while `doReader` was still alive, the source's next `SourceError` send would panic. The new cancellation path waits for `b.items` to close before letting `doProcessors` exit, which only happens after `doReader` is fully drained. + +### Migration + +- Any `errors.As(err, &procErr)` switch that used to catch per-item errors needs an additional case for `*ItemError`: + +```go +var ( + srcErr *batch.SourceError + procErr *batch.ProcessorError + itemErr *batch.ItemError +) +switch { +case errors.As(err, &srcErr): + // source-level failure +case errors.As(err, &procErr): + // processor-wide failure +case errors.As(err, &itemErr): + // per-item failure — itemErr.ItemID identifies the item +} +``` + +- Drop `IDBufferSize` from any `BufferConfig` struct literal; remove references to `DefaultIDBufferSize`. + ## [0.5.0] - 2026-02-15 This release introduces a hard switch to generics across the public API. diff --git a/README.md b/README.md index 28f5fcb..4af0da2 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 }) ``` @@ -282,11 +281,14 @@ go func() { for err := range errs { var srcErr *batch.SourceError var procErr *batch.ProcessorError + var itemErr *batch.ItemError switch { case errors.As(err, &srcErr): log.Printf("Source error: %v", srcErr.Unwrap()) case errors.As(err, &procErr): log.Printf("Processor error: %v", procErr.Unwrap()) + case errors.As(err, &itemErr): + log.Printf("Item %d error: %v", itemErr.ItemID, itemErr.Unwrap()) default: log.Printf("Error: %v", err) } @@ -297,9 +299,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..63ca239 100644 --- a/batch/batch.go +++ b/batch/batch.go @@ -4,6 +4,7 @@ import ( "context" "errors" "sync" + "sync/atomic" "time" ) @@ -22,10 +23,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 @@ -33,8 +30,8 @@ type BufferConfig struct { // Batch provides batch processing given a Source and one or more Processors. // Data is read from the Source and processed through each Processor in sequence. -// Any errors are wrapped in either a SourceError or a ProcessorError, so the caller -// can determine where the errors came from. +// Any errors are wrapped in a SourceError, ProcessorError, or ItemError so the +// caller can determine where the errors came from. // // To create a new Batch, call New. Creating one using &Batch[T]{} will also work. // @@ -65,15 +62,16 @@ type BufferConfig struct { // // Now batch processing is done // // Errors returned on the error channel may be wrapped. Source errors will be -// of type SourceError, processor errors will be of type ProcessorError, and -// Batch errors (internal errors) will be plain. +// of type SourceError, processor errors will be of type ProcessorError, +// per-item errors will be of type ItemError, and Batch errors (internal +// errors) will be plain. type Batch[T any] struct { config Config bufferConfig BufferConfig src Source[T] processors []Processor[T] items chan *Item[T] - ids chan uint64 + nextID uint64 done chan struct{} mu sync.Mutex @@ -100,7 +98,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 +269,16 @@ 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) + atomic.StoreUint64(&b.nextID, 0) b.errs = make(chan error, errBuf) b.done = make(chan struct{}) - go b.doIDGenerator() go b.doReader(ctx) go b.doProcessors(ctx) @@ -323,22 +315,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 +343,7 @@ func (b *Batch[T]) doReader(ctx context.Context) { out = nil continue } - id := <-b.ids + id := atomic.AddUint64(&b.nextID, 1) - 1 b.items <- &Item[T]{ ID: id, Data: data, @@ -427,13 +403,17 @@ func (b *Batch[T]) doProcessors(ctx context.Context) { for _, item := range items { if item.Error != nil { - b.errs <- &ProcessorError{Err: item.Error} + b.errs <- &ItemError{ItemID: item.ID, Err: item.Error} } } }(batch) } wg.Wait() + // Invariant: doReader closes b.items only after both source channels + // close, so reaching here means no more sends to b.errs are possible. + // Keep this ordering — closing b.errs while doReader is still running + // would panic on its next SourceError. close(b.errs) close(b.done) b.mu.Lock() @@ -476,32 +456,60 @@ func fixConfig(c ConfigValues) ConfigValues { // - MinItems: If reached, waits until MinTime is also satisfied. // // The method returns the collected batch of items. -func (b *Batch[T]) waitForItems(_ context.Context, config ConfigValues) []*Item[T] { +// +// Cancellation contract: when ctx is cancelled, waitForItems returns any +// partial batch immediately and then drains remaining buffered items one +// at a time so already-read items are not dropped. Eventually the loop +// terminates when b.items closes, which only happens once doReader has +// observed both source channels close. That means cancellation only +// completes when the Source itself observes ctx and closes its channels; +// Sources that ignore ctx will block waitForItems indefinitely after +// cancel. Source authors must propagate ctx into their Read goroutine. +func (b *Batch[T]) waitForItems(ctx context.Context, config ConfigValues) []*Item[T] { var ( reachedMinTime bool + cancelled 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 + ctxDone = ctx.Done() ) - // 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. + // Only create timers when the duration is positive. A nil channel + // is never selected, so the corresponding case is effectively disabled. 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 { select { + case <-ctxDone: + if len(batch) > 0 { + // Return collected items immediately for processing. + return batch + } + // No items yet. Returning here would let doProcessors close + // b.errs while doReader is still sending SourceErrors on it + // (panic: send on closed channel). Wait for b.items to close + // instead — that signals doReader has fully drained. Disable + // timers and ctx so subsequent iterations only watch b.items. + cancelled = true + ctxDone = nil + minTimerCh = nil + maxTimerCh = nil + case item, ok := <-b.items: if !ok { // Source is exhausted, return whatever was collected @@ -510,6 +518,12 @@ func (b *Batch[T]) waitForItems(_ context.Context, config ConfigValues) []*Item[ batch = append(batch, item) + // After cancellation, return each item immediately instead + // of waiting for batch thresholds. + if cancelled { + return batch + } + if uint64(len(batch)) >= config.MinItems && reachedMinTime { return batch } @@ -517,20 +531,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..d8ddbe9 100644 --- a/batch/batch_test.go +++ b/batch/batch_test.go @@ -26,8 +26,8 @@ func TestBatch_ProcessorChainingAndErrorTracking(t *testing.T) { received := 0 for err := range errs { - var processorError *ProcessorError - if !errors.As(err, &processorError) { + var itemError *ItemError + if !errors.As(err, &itemError) { t.Errorf("unexpected error type: %v", err) } received++ 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/cancellation_test.go b/batch/cancellation_test.go new file mode 100644 index 0000000..a71a95c --- /dev/null +++ b/batch/cancellation_test.go @@ -0,0 +1,247 @@ +package batch_test + +import ( + "context" + "errors" + "fmt" + "sort" + "sync" + "testing" + "time" + + . "github.com/MasterOfBinary/gobatch/batch" +) + +// fnSource implements Source[any] by calling a function. +type fnSource struct { + fn func(ctx context.Context) (<-chan any, <-chan error) +} + +func (s *fnSource) Read(ctx context.Context) (<-chan any, <-chan error) { + return s.fn(ctx) +} + +// TestCancellation_BufferedItemsDrained verifies that items already read from +// the source and sitting in the internal buffer are still processed after +// context cancellation, honoring Go()'s contract that already-read items are +// never dropped. +// +// The source pushes every item synchronously and then blocks on ctx.Done, +// so it cannot lose items in response to cancel. MinItems is set above the +// total so waitForItems can never satisfy its threshold — only the new +// ctxDone branch can deliver the accumulated batch. +func TestCancellation_BufferedItemsDrained(t *testing.T) { + const totalItems = 20 + + // delivered is closed by the source goroutine immediately after every + // item has been received by doReader. Because `out` is unbuffered, each + // `out <- i` only returns once doReader has taken the item, so closing + // `delivered` after the loop guarantees all totalItems items are in the + // pipeline (in b.items or in transit inside doReader) — no wall-clock + // sleep required. + delivered := make(chan struct{}) + + src := &fnSource{fn: func(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 < totalItems; i++ { + out <- i + } + close(delivered) + <-ctx.Done() + }() + return out, errs + }} + + var ( + mu sync.Mutex + processedIDs = make(map[uint64]bool) + ) + + proc := &testProcessor{ + processFn: func(_ context.Context, items []*Item[any]) ([]*Item[any], error) { + mu.Lock() + for _, item := range items { + processedIDs[item.ID] = true + } + mu.Unlock() + return items, nil + }, + } + + config := NewConstantConfig(&ConfigValues{ + MinItems: totalItems + 5, + MaxTime: time.Hour, + }) + + b := New[any](config) + ctx, cancel := context.WithCancel(context.Background()) + errCh := b.Go(ctx, src, proc) + + // Wait until every item has been handed off to doReader, then cancel. + <-delivered + cancel() + + for range errCh { + } + <-b.Done() + + mu.Lock() + count := len(processedIDs) + mu.Unlock() + + if count != totalItems { + t.Fatalf("expected all %d buffered items to be processed after cancel, got %d", totalItems, count) + } + for i := uint64(0); i < totalItems; i++ { + if !processedIDs[i] { + t.Errorf("item ID %d was buffered but not processed (dropped)", i) + } + } +} + +// TestCancellation_SourceErrorNoRace verifies that a source emitting an error +// at cancellation time does not race with doProcessors closing the error +// channel. This test should be run with -race. +func TestCancellation_SourceErrorNoRace(t *testing.T) { + // Run many iterations to maximize chance of exposing a race + for i := 0; i < 50; i++ { + src := &fnSource{fn: func(ctx context.Context) (<-chan any, <-chan error) { + out := make(chan any) + errs := make(chan error, 1) + go func() { + defer close(out) + defer close(errs) + + for j := 0; j < 5; j++ { + select { + case out <- j: + case <-ctx.Done(): + errs <- fmt.Errorf("source cancelled: %w", ctx.Err()) + return + } + time.Sleep(5 * time.Millisecond) + } + + <-ctx.Done() + errs <- fmt.Errorf("source cancelled: %w", ctx.Err()) + }() + return out, errs + }} + + proc := &testProcessor{ + processFn: func(ctx context.Context, items []*Item[any]) ([]*Item[any], error) { + return items, nil + }, + } + + b := New[any](NewConstantConfig(&ConfigValues{ + MinItems: 2, + MaxTime: 100 * time.Millisecond, + })) + + ctx, cancel := context.WithCancel(context.Background()) + + errCh := b.Go(ctx, src, proc) + + time.Sleep(15 * time.Millisecond) + cancel() + + // Drain errors — if there's a send-on-closed-channel this panics + for range errCh { + } + <-b.Done() + } +} + +// TestCancellation_ItemErrorType verifies that per-item errors are reported as +// ItemError (not ProcessorError) and include the correct item ID. +func TestCancellation_ItemErrorType(t *testing.T) { + src := &testSource{Items: []any{1, 2, 3, 4, 5}} + + proc := &testProcessor{ + processFn: func(ctx context.Context, items []*Item[any]) ([]*Item[any], error) { + for _, item := range items { + item.Error = fmt.Errorf("bad item %d", item.Data) + } + return items, nil + }, + } + + b := New[any](NewConstantConfig(&ConfigValues{})) + errCh := b.Go(context.Background(), src, proc) + + var itemErrors []*ItemError + for err := range errCh { + var ie *ItemError + if errors.As(err, &ie) { + itemErrors = append(itemErrors, ie) + } else { + t.Errorf("expected ItemError, got %T: %v", err, err) + } + } + <-b.Done() + + if len(itemErrors) != 5 { + t.Fatalf("expected 5 ItemErrors, got %d", len(itemErrors)) + } + + // Per-item batches dispatch on separate goroutines, so error arrival + // order is non-deterministic. Sort by ItemID before checking the set. + sort.Slice(itemErrors, func(i, j int) bool { + return itemErrors[i].ItemID < itemErrors[j].ItemID + }) + for i, ie := range itemErrors { + if ie.ItemID != uint64(i) { + t.Errorf("position %d: expected ItemID=%d, got %d", i, i, ie.ItemID) + } + } +} + +// TestCancellation_ProcessorErrorDistinct verifies that processor-wide errors +// are still reported as ProcessorError, distinct from ItemError. +func TestCancellation_ProcessorErrorDistinct(t *testing.T) { + src := &testSource{Items: []any{1, 2, 3}} + + procErr := errors.New("processor failed") + proc := &testProcessor{ + processFn: func(ctx context.Context, items []*Item[any]) ([]*Item[any], error) { + items[0].Error = errors.New("item bad") + return items, procErr + }, + } + + b := New[any](NewConstantConfig(&ConfigValues{MinItems: 3})) + errCh := b.Go(context.Background(), src, proc) + + var ( + procErrors int + itemErrs int + ) + for err := range errCh { + var pe *ProcessorError + var ie *ItemError + switch { + case errors.As(err, &pe): + procErrors++ + if !errors.Is(err, procErr) { + t.Errorf("ProcessorError should wrap original: got %v", err) + } + case errors.As(err, &ie): + itemErrs++ + default: + t.Errorf("unexpected error type %T: %v", err, err) + } + } + <-b.Done() + + if procErrors == 0 { + t.Error("expected at least one ProcessorError") + } + if itemErrs == 0 { + t.Error("expected at least one ItemError") + } +} 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/errors.go b/batch/errors.go index b5a3d20..488397f 100644 --- a/batch/errors.go +++ b/batch/errors.go @@ -21,6 +21,29 @@ func (e ProcessorError) Unwrap() error { return e.Err } +// ItemError is returned when an individual item has an error after processing. +// Unlike ProcessorError, which indicates a processor-wide failure, ItemError +// represents a failure specific to a single item. The ItemID field identifies +// which item failed, making it easier to debug issues in large batches. +type ItemError struct { + // ItemID is the unique identifier of the item that failed. + ItemID uint64 + + // Err is the underlying error set on the item. + Err error +} + +// Error implements the error interface, returning a formatted error message +// that includes the item ID and the wrapped error. +func (e *ItemError) Error() string { + return fmt.Sprintf("item %d error: %v", e.ItemID, e.Err) +} + +// Unwrap returns the underlying error for compatibility with errors.Is and errors.As. +func (e *ItemError) Unwrap() error { + return e.Err +} + // SourceError is returned when a source fails. It wraps the original // error from the source to maintain the error chain while providing // context about the source of the error. diff --git a/batch/example_error_handling_test.go b/batch/example_error_handling_test.go index 0f169ee..9cb991c 100644 --- a/batch/example_error_handling_test.go +++ b/batch/example_error_handling_test.go @@ -152,12 +152,15 @@ func Example_errorHandling() { for i, err := range errs { var srcErr *batch.SourceError var procErr *batch.ProcessorError + var itemErr *batch.ItemError switch { case errors.As(err, &srcErr): fmt.Printf("%d. Source error: %v\n", i+1, srcErr.Unwrap()) case errors.As(err, &procErr): fmt.Printf("%d. Processor error: %v\n", i+1, procErr.Unwrap()) + case errors.As(err, &itemErr): + fmt.Printf("%d. Item %d error: %v\n", i+1, itemErr.ItemID, itemErr.Unwrap()) default: fmt.Printf("%d. Other error: %v\n", i+1, err) } @@ -191,8 +194,8 @@ func Example_errorHandling() { // 1. Source error: source error at item 5 // 2. Processor error: processor failed on batch 2 // 3. Source error: source error at item 10 - // 4. Processor error: value 12 exceeds maximum 10 - // 5. Processor error: value 15 exceeds maximum 10 - // 6. Processor error: value 20 exceeds maximum 10 - // 7. Processor error: value 25 exceeds maximum 10 + // 4. Item 9 error: value 12 exceeds maximum 10 + // 5. Item 10 error: value 15 exceeds maximum 10 + // 6. Item 11 error: value 20 exceeds maximum 10 + // 7. Item 12 error: value 25 exceeds maximum 10 } diff --git a/batch/example_simple_processor_test.go b/batch/example_simple_processor_test.go index d03ae46..3ab9082 100644 --- a/batch/example_simple_processor_test.go +++ b/batch/example_simple_processor_test.go @@ -65,5 +65,5 @@ func Example_simpleProcessor() { // Batch: [7 8 9] // Batch: [10] // Errors: 1 - // Last error: processor error: value 5 not allowed + // Last error: item 4 error: value 5 not allowed } 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) // }