refactor: introduce result collector + improve error handling in executeJob - #538
refactor: introduce result collector + improve error handling in executeJob#538evacchi wants to merge 2 commits into
executeJob#538Conversation
github-actions
Bot
added
the
release-note-none
There was a problem hiding this comment.
Pull request overview
This PR refactors job execution output handling by replacing the per-job outputWriters (buffered writers + mutexes) with a new “result collector” abstraction that is intended to both write results (output/error JSONL) and drive executionProgress updates.
Changes:
- Refactors
executeJob/processModelto emit results via aresultCollectorinstead of writing JSONL directly with shared buffered writers. - Centralizes terminal error handling in
executeJob(abort siblings on first error, then determine final outcome from context state). - Updates
executor_test.goto construct/start/flush the new collector in model-processing tests.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| internal/processor/worker/executor.go | Switches execution to a new resultCollector interface and updates cancellation/drain paths to emit ResultItems. |
| internal/processor/worker/executor_test.go | Updates model execution tests to use the new collector instead of outputWriters. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
45923fa to
dc78183
Compare
2b9139c to
eaaaf20
Compare
Signed-off-by: Edoardo Vacchi <evacchi@users.noreply.github.com>
Signed-off-by: Edoardo Vacchi <evacchi@users.noreply.github.com>
2a2250e to
762e139
Compare
| continue | ||
| } | ||
|
|
||
| c.progress.record(ctx, result.isSuccess()) |
There was a problem hiding this comment.
This introduces a new coupling between result draining and progress publication. resultCollector is a single goroutine, so if progress.record() blocks on the status store, the collector stops draining results, collect() eventually backpressures model goroutines, and executeJob() can hang even though output writes are otherwise fine. In the previous design, progress updates were best-effort work done by worker goroutines, so a slow status-store call could stall one updater but not the entire result pipeline.
|
|
||
| func newResultCollector(outputWriter, errorWriter *bufio.Writer, progress *executionProgress, logger logr.Logger, abortFn context.CancelFunc) *resultCollector { | ||
| if abortFn == nil { | ||
| panic("resultCollector: abortFn cannot be nil") |
There was a problem hiding this comment.
nit: I think it is better to avoid panic here. Our usual convention is to return an error (or treat nil as a no-op) rather than panic in production paths.
I noticed executeJob already substitutes a no-op when requestAbortFn is nil, so this may be unreachable today, but if a future caller hits this constructor directly, a panic would take down
the process. Returning an error (or documenting/handling nil explicitly) would make that failure mode safer.
|
this can also be closed as superseded by work in #580 |
Why is this PR needed?
introduce a result collector that both logs errors/status and updates job progress status; in the future we should be able to decouple sending a request from collecting the results for asynchronous processing.
What does this PR do?
it refactors the outputWriter to resultCollector that also handles (by delegating to the
executionProgress) status updates.How was this tested?
Checklist
git commit -s) per DCOmake ci)make test-e2e)Related Issues
closes #537, related #530, #529