refactor(batch): atomic ID counter, ItemError, ctx-aware cancellation - #63
refactor(batch): atomic ID counter, ItemError, ctx-aware cancellation#63MasterOfBinary wants to merge 6 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.
bug_002: add ItemError to error-handling guidance in CLAUDE.md and AGENTS.md (was missed when other docs were updated for the *ItemError breaking change). bug_003: soften the post-cancel drain wording in CHANGELOG, the waitForItems godoc, and the planning docs. The drain is best-effort (probabilistic select), not a one-at-a-time guarantee; no items are dropped but post-cancel batch sizes are not guaranteed. Doc/comment-only; no logic changes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #63 +/- ##
==========================================
- 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:
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces several significant changes to the GoBatch library, including a new ItemError type for per-item failures, an atomic counter for ID generation that replaces a dedicated goroutine, and improved context cancellation logic in waitForItems to ensure buffered items are drained without dropping data. It also fixes potential timer leaks and a race condition during shutdown. The feedback highlights an inconsistency in the ItemError implementation, which uses pointer receivers for the error and Unwrap interfaces, whereas the existing ProcessorError and SourceError types use value receivers.
| 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 | ||
| } |
There was a problem hiding this comment.
The ItemError type correctly implements the error and Unwrap interfaces. Note that while ProcessorError and SourceError use value receivers for these methods, ItemError uses pointer receivers. While this is inconsistent with the other error types in the package, it is technically correct since ItemError instances are always sent as pointers to the error channel in doProcessors.
|
Superseded by #64. Per maintainer direction the breaking changes in this PR (atomic ID counter, |
1 participant
Summary
Ships the cancellation-and-IDs refactor plus the doc corrections from its ultrareview.
Refactor (existing commits):
BufferConfig.IDBufferSizeandDefaultIDBufferSize).ItemErrortype — per-item failures now surface as*ItemError(carryingItemID) instead of*ProcessorError(BREAKING).waitForItemsnow honorsctx.Done()— previously the context was ignored, delaying shutdown by up toMaxTime.Ultrareview fixes (this branch's commit):
bug_002— addedItemErrorto the error-handling guidance inCLAUDE.md/AGENTS.md(missed when the other docs were updated).bug_003— softened the post-cancel drain wording (CHANGELOG,waitForItemsgodoc, planning docs). The drain is best-effort/probabilistic, not a one-at-a-time guarantee; no items are dropped but post-cancel batch sizes are not guaranteed. Docs now match behavior (no logic change).Test plan
go build ./... && go vet ./...cleangofmtclean🤖 Generated with Claude Code