DS 2359 observation loop atomic pipeline output#21237
Conversation
|
|
There was a problem hiding this comment.
Pull request overview
This PR ensures pipeline streams are extracted atomically to prevent partial failures where some streams from a multi-stream pipeline have values while others don't. The change addresses a concurrency issue in the LLO observation loop where BridgeTask's use of overtimeContext could cause waiters to bail early with errors while the executor eventually succeeds.
Changes:
- Modified observation context to block waiters until pipeline execution completes, ensuring all goroutines receive identical results
- Added
removeIncompleteGroupsfunction to enforce all-or-nothing cache writes per pipeline group - Extracted
StreamValueCacheinterface to enable test verification of atomic behavior
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
core/services/llo/observation/observation_context.go |
Removed ctx.Done() handling from waiter path to prevent partial results |
core/services/llo/observation/data_source.go |
Added validation to drop incomplete pipeline groups before caching |
core/services/llo/observation/cache.go |
Extracted StreamValueCache interface for testability |
core/services/llo/observation/observation_context_test.go |
Added tests for partial extraction failures and concurrent atomic output |
core/services/llo/observation/data_source_test.go |
Added comprehensive tests for atomic cache writes and removeIncompleteGroups edge cases |
.changeset/warm-roses-argue.md |
Added changeset with minor formatting issue |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
|
||
| mu.Lock() | ||
| observedValues[streamID] = val | ||
| successfulStreamIDs = append(successfulStreamIDs, streamID) |
There was a problem hiding this comment.
do we also want to also adjust the successful stream count based on the dropped streams from removeIncompleteGroups?
There was a problem hiding this comment.
sure I've added separate field to log out the count of dropped streamIds and the field for successfulStreamIDs is just the count of values post filtering
There was a problem hiding this comment.
is the original successfulStreamIDs even used anywhere then?
| } | ||
| delete(observedValues, sid) | ||
| } | ||
| lggr.Debugw("Discarding incomplete pipeline group", |
There was a problem hiding this comment.
do we know under what specific conditions this occurs and would it be indicative of issues with data freshness or otherwise? im wondering if we should add metrics/upgrade to a warn log to keep an eye on it
There was a problem hiding this comment.
This was occurring more frequently with the race that's fixed in observation_context.go. now this might occur if there are individual task level failures i.e. task didn't complete in time, or bad data, or missing data, etc...
I don't think we need more metrics right now & we can lean into stream values telemetry to figure out DON wide issues
There was a problem hiding this comment.
the skip logic still iterates over individual streams - to preserve atomicity, we should attempt to refresh/observe all the streams within a group if any value is expired?
There was a problem hiding this comment.
very good suggestion makes the observation even more correct - been added
|
✅ No conflicts with other open PRs targeting |
|
|
|
||
| mu.Lock() | ||
| observedValues[streamID] = val | ||
| successfulStreamIDs = append(successfulStreamIDs, streamID) |
There was a problem hiding this comment.
is the original successfulStreamIDs even used anywhere then?
Uh oh!
There was an error while loading. Please reload this page.