Cancellation contract, atomic IDs, and ItemError refactor - #62
Cancellation contract, atomic IDs, and ItemError refactor#62MasterOfBinary wants to merge 5 commits into
Conversation
ARCHITECTURE/CONCERNS/CONVENTIONS/STRUCTURE/TESTING/INTEGRATIONS were written before the doIDGenerator goroutine was removed and the ItemError type was added. Update them in place rather than deleting: - ARCHITECTURE: drop the third goroutine; describe the atomic counter and ctx-aware waitForItems; add a Cancellation Contract section. - CONCERNS: mark the medium 'unused ctx' item resolved; mark the low ID-overflow item 'implementation simplified' (uint64 wrap still exists); split MaxTime timer concern into resolved (leak fixed via NewTimer/Stop) and remaining (idle-timeout policy still TBD); split Processor Contract into resolved (ItemError makes the per-item vs processor-wide distinction explicit) and remaining (godoc still doesn't formally specify item.Error handling). - CONVENTIONS/STRUCTURE: drop doIDGenerator references; add ItemError to the error type list; note the atomic ID counter. - TESTING: reference cancellation_test.go and ItemError coverage. Leaves the High-priority panic concern explicitly open.
Replace doIDGenerator goroutine + ids channel with an atomic uint64
counter on Batch. doReader assigns IDs inline via
atomic.AddUint64(&b.nextID, 1) - 1. Drops BufferConfig.IDBufferSize and
DefaultIDBufferSize.
Introduce ItemError{ItemID, Err} for per-item failures. doProcessors
now emits *ItemError for items where item.Error is set, while
*ProcessorError remains for processor-wide failures.
Wire ctx into waitForItems: on ctx.Done() return any partial batch
immediately, then drain remaining buffered items one at a time so
already-read items are not dropped. Closes a send-on-closed-channel
race where doProcessors could close b.errs while doReader was still
alive. Replaces time.After with time.NewTimer + defer Stop() to avoid
timer leaks. Adds godoc documenting the cancellation contract: sources
must propagate ctx for cancellation to terminate.
BREAKING: BufferConfig.IDBufferSize removed; per-item errors moved
from *ProcessorError to *ItemError. See CHANGELOG for migration.
New batch/cancellation_test.go with four tests covering the ctx-aware shutdown path: - TestCancellation_BufferedItemsDrained: items already read from the source must not be dropped when ctx is cancelled. Uses a 'delivered' channel signalled by the source after every item has been handed off to doReader (rather than a wall-clock sleep) so the test is deterministic — no flake risk in CI. - TestCancellation_SourceErrorNoRace: 50 iterations under -race to exercise the previously broken send-on-closed-channel path between doReader's SourceError emit and doProcessors closing b.errs. - TestCancellation_ItemErrorType: verifies per-item failures are wrapped as *ItemError and that ItemID identifies the right item. - TestCancellation_ProcessorErrorDistinct: verifies that *ProcessorError (processor-wide) and *ItemError (per-item) remain distinguishable via errors.As.
Add an Unreleased section covering: - Added: ItemError type for per-item errors, with ItemID to correlate failures to specific items. - Changed (BREAKING): per-item errors now emitted as *ItemError instead of *ProcessorError; BufferConfig.IDBufferSize and DefaultIDBufferSize removed; cancellation drains items 1-by-1 so partial batches are returned during shutdown; CollectErrors doc clarification. - Fixed: ctx now honored in waitForItems; time.After leak fixed via NewTimer+Stop; send-on-closed-channel race window during cancellation closed. - Migration: example errors.As switch including *ItemError; reminder to drop IDBufferSize from any BufferConfig literal.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #62 +/- ##
==========================================
- Coverage 96.79% 96.31% -0.48%
==========================================
Files 12 12
Lines 374 380 +6
==========================================
+ Hits 362 366 +4
- Misses 9 11 +2
Partials 3 3 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Code Review
This pull request introduces significant improvements to the GoBatch pipeline, primarily focusing on robust context cancellation and refined error handling. Key changes include making waitForItems context-aware to ensure proper drainage of buffered items during shutdown, replacing the dedicated ID-generator goroutine with an atomic counter for better performance and simplicity, and introducing a new ItemError type to explicitly distinguish per-item failures from processor-wide errors. Documentation and examples have been updated to reflect these changes, and new tests were added to verify cancellation behavior and error reporting. I have no feedback to provide as the changes are well-implemented and documented.
|
Superseded by #64; breaking changes reverted per maintainer direction. |
Summary
This PR refactors the core batch pipeline around three related improvements:
ItemErrortype — per-item errors (whereitem.Error != nilafter a processor returns) are now reported as*ItemErrorinstead of*ProcessorError.ItemErrorcarries the failing item'sItemIDso callers can correlate errors with specific items. Processor-wide errors (returned as the second value ofProcessor.Process) remain*ProcessorError.BufferConfig.IDBufferSize,DefaultIDBufferSize) are removed. Item IDs are now generated withsync/atomic, eliminating a goroutine, simplifying startup, and removing a class of coordination bugs.waitForItemsnow honorsctx.Done()(previously ignored, delaying shutdown by up toMaxTime). Cancellation drains items one-by-one so partial batches are returned during shutdown. A send-on-closed-channel race window is closed:doProcessorsnow waits forb.itemsto drain before exiting, preventing panics whendoReaderis still alive asb.errscloses.time.Afterreplaced withtime.NewTimer+defer Stop()to prevent timer leaks.Full details — including a migration guide and
errors.Asswitch example — are in the[Unreleased]section of CHANGELOG.md.Breaking changes
*ItemErrorreplaces*ProcessorErrorfor per-item errors — update anyerrors.Asswitches (see CHANGELOG migration guide).BufferConfig.IDBufferSizeandDefaultIDBufferSizeremoved — drop from anyBufferConfigstruct literals.MinItems) may now be returned during shutdown.Test plan
go build ./...— passesgo vet ./...— passesgo test -race -coverprofile=coverage.txt -covermode=atomic ./...— passes (batch: 97.9%, processor: 98.2%, source: 95.8%)batch/covers the race window, context-awareness, and partial-batch drain behaviourOut of scope
The High-priority panic-handling cleanup flagged in CONCERNS.md (unrecovered panics from goroutines crashing the process) is intentionally excluded from this PR and will be addressed in a follow-up MR.