Skip to content

fix(cdn): prevent log streaming stall in sendStepLog #7644

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fsamin wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix(cdn): prevent log streaming stall in sendStepLog #7644

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 15 additions & 3 deletions engine/cdn/item_logs_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,9 +168,12 @@ func (s *Service) sendStepLog(ctx context.Context, wsClient websocket.Client, ws

log.Debug(ctx, "getItemLogsStreamHandler> iterate over %d lines to send for client %s", len(lines), wsClient.UUID())
oldNextLineToSend := data.scoreNextLineToSend
var linesSent int
for i := range lines {
if data.scoreNextLineToSend > 0 && data.scoreNextLineToSend != lines[i].Number {
break
// Skip lines that don't match expected sequence instead of breaking,
// to avoid stalling the stream when lines arrive out of order.
continue
}

if err := wsClient.Send(WSLine{
Expand All @@ -181,15 +184,24 @@ func (s *Service) sendStepLog(ctx context.Context, wsClient websocket.Client, ws
}); err != nil {
return err
}
linesSent++
if data.scoreNextLineToSend < 0 {
data.scoreNextLineToSend = lines[i].Number + 1
} else {
data.scoreNextLineToSend++
}
}
wsClientData.itemUnitsData[mapIndex] = data
// If all the lines were sent, we can trigger another update, if only one line was send do not trigger an update wait for next event from broker
if len(lines) > 1 && (oldNextLineToSend > 0 || int(data.scoreNextLineToSend-oldNextLineToSend) == len(lines)) {
// Trigger another update if there are potentially more lines to read.
// Also trigger if lines were available but none matched (gap detection),
// so we retry on next tick rather than stalling forever.
if linesSent > 1 && (oldNextLineToSend > 0 || int(data.scoreNextLineToSend-oldNextLineToSend) == linesSent) {
wsClientData.TriggerUpdate()
} else if len(lines) > 0 && linesSent == 0 {
// Lines existed in buffer but none matched scoreNextLineToSend.
// Jump to the first available line to recover from gaps.
data.scoreNextLineToSend = lines[0].Number
wsClientData.itemUnitsData[mapIndex] = data
wsClientData.TriggerUpdate()
}
return nil
Expand Down