refactor(batch): atomic ID counter, ItemError, ctx-aware cancellation

.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 on a best-effort basis so already-read items are never dropped (post-cancel batch sizes are not guaranteed).
+
+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.*

.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 on a best-effort basis so already-read items are not dropped (post-cancel batch sizes are not guaranteed). 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.*

.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).*

.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*

.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*