Fix flow-state merge freeze on repeated PROGRESS snapshots - #174
Open
rowanadair-r7 wants to merge 1 commit into
Open
rowanadair-r7 wants to merge 1 commit into
rowanadair-r7 wants to merge 1 commit into
Conversation
mergeRecords() only replaced the accumulated QueryStats when the incoming record was a genuine completion (running -> terminal). Two consecutive still-running snapshots never updated it, so a flow whose query never reaches a terminal state gets permanently pinned at the very first progress snapshot it ever received, even though the client keeps sending fresher progress. Each FlowStats message is already the complete cumulative status of the flow so far (see responder.FlowContext.getStats() upstream), so it's always safe to adopt the latest snapshot wholesale while the flow is still running.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
mergeRecords()inservices/launcher/flows.goonly replaces the accumulatedQueryStatswhen the incoming stats record represents a genuine transitionfrom running to a terminal state:
Two consecutive still-running (
PROGRESS) snapshots never update it. If aflow's query never reaches a terminal state (e.g. it's stuck on a slow
upload), the collection context is permanently pinned at the very first
progress snapshot it ever received - even though the client keeps sending
fresher progress for as long as the query runs. Since
UpdateFlowStatsderives
TotalUploadedBytes,TotalCollectedRows,TotalLogs,State, etc.purely from
QueryStats, the whole flow view gets stuck showing stale/emptyprogress and never reflects what the client has actually done.
Fix
Each
FlowStatsmessage the client sends already contains the complete,cumulative status of every query in the flow so far (built by iterating the
full
respondersslice - seeresponder.FlowContext.getStats()upstream),never a partial delta. So it's always safe to adopt the latest snapshot
wholesale as long as the accumulator hasn't reached a terminal state yet:
This still freezes correctly once the accumulator reaches a terminal state
(so a stray/out-of-order message arriving after completion doesn't reopen
it), it just stops discarding fresher in-progress snapshots.
Testing
Added
TestFlowStatsProgressNotFrozeniningestion/ingestor_test.go,which sends two consecutive
PROGRESSsnapshots viaIngestor.HandleFlowStatswith increasing row/byte counts and assertsGetFlowDetailsreflects the second, fresher snapshot rather than stayingpinned at the first. Verified it fails against the old logic and passes with
the fix. Also ran the full existing suite (
go test -p 1 ./...) - noregressions;
TestLauncherandTestListDirectoryare unaffected sincethey only exercise a single running->terminal transition, where old and new
logic agree.