From 3c90fda52c1676a35e62b042c7d2887af4e2ea78 Mon Sep 17 00:00:00 2001 From: Vidyansh Date: Fri, 23 Oct 2026 17:36:34 +0530 Subject: [PATCH] parser: pass on a split line that never ends once its first part is old A line the runtime splits that keeps growing without a newline (a progress bar) stayed pending for good, since each new part reset its expiry, and while any line is pending no offset of its source can be acknowledged. Such a line is now passed on once its first part is 10 multiline timeouts old; its later parts start a new line. Also documents that LogEntry.Offset must only increase within a source. --- decoder_test.go | 11 +++++++++-- offset_test.go | 32 ++++++++++++++++++++++++++++++++ parser.go | 6 ++++-- partial.go | 16 +++++++++++++--- 4 files changed, 58 insertions(+), 7 deletions(-) diff --git a/decoder_test.go b/decoder_test.go index e5ac045..d6eb4aa 100644 --- a/decoder_test.go +++ b/decoder_test.go @@ -101,9 +101,16 @@ func TestPartialLines(t *testing.T) { p.add(LogEntry{Content: "Enter password: ", Stream: "stdout"}, true, t0) p.add(LogEntry{Content: "still ", Stream: "stderr"}, true, t0) p.add(LogEntry{Content: "growing ", Stream: "stderr"}, true, t0.Add(time.Second)) - assert.Equal(t, []LogEntry{{Content: "Enter password: ", Stream: "stdout"}}, p.expired(t0.Add(time.Millisecond))) + assert.Equal(t, []LogEntry{{Content: "Enter password: ", Stream: "stdout"}}, p.expired(t0.Add(time.Millisecond), t0)) assert.Len(t, p, 1) - assert.Empty(t, p.expired(t0.Add(time.Millisecond))) + assert.Empty(t, p.expired(t0.Add(time.Millisecond), t0)) + + // A line that keeps growing (a progress bar written without a newline) comes + // out once its first part is too old, though its latest part is recent. + p.add(LogEntry{Content: "more ", Stream: "stderr"}, true, t0.Add(2*time.Second)) + assert.Empty(t, p.expired(t0.Add(time.Millisecond), t0)) + assert.Equal(t, []LogEntry{{Content: "still growing more ", Stream: "stderr"}}, p.expired(t0.Add(time.Millisecond), t0.Add(time.Millisecond))) + assert.Empty(t, p) } // Lines read from a CRI log file reach the callback with the runtime's time and diff --git a/offset_test.go b/offset_test.go index cb074f1..10dc7a9 100644 --- a/offset_test.go +++ b/offset_test.go @@ -108,6 +108,38 @@ func TestPendingPartialAcknowledgesLinesPassedOnMeanwhile(t *testing.T) { assert.Equal(t, offsetMsg{"never finished", 25}, got[1]) } +// A split line that never ends but keeps growing (a progress bar written without a +// newline, split by the runtime) must not hold acknowledgements back for the whole +// source: once its first part is old enough it is passed on, and its message +// acknowledges the other stream's lines held back meanwhile. +func TestEndlessSplitLineDoesNotBlockAcknowledgements(t *testing.T) { + ch, msgs := offsetParser(t, CriDecoder{}) + send := func(content string, offset int64) { + ch <- LogEntry{Timestamp: time.Now(), Content: content, Offset: offset} + } + send("2026-10-19T10:00:00Z stdout P progress 1% ", 10) + send("2026-10-19T10:00:00Z stderr F an error meanwhile", 20) + offset := int64(20) + deadline := time.Now().Add(2 * time.Second) + acked := func() bool { + for _, m := range msgs() { + if m.offset >= 20 { + return true + } + } + return false + } + // Parts keep arriving well within the multiline timeout (20ms). + for !acked() && time.Now().Before(deadline) { + offset += 10 + send("2026-10-19T10:00:00Z stdout P # ", offset) + time.Sleep(5 * time.Millisecond) + } + got := msgs() + assert.True(t, acked(), "no message acknowledged the stderr line while stdout kept growing: %v", got) + assert.Equal(t, offsetMsg{"an error meanwhile", 0}, got[0], "held while the stdout line is pending") +} + // A split line whose last part never comes is passed on after the timeout with the // offset of the last part read. func TestDanglingPartialCarriesItsOffset(t *testing.T) { diff --git a/parser.go b/parser.go index 01c277b..518bced 100644 --- a/parser.go +++ b/parser.go @@ -40,7 +40,9 @@ type LogEntry struct { // Offset is where the line ends in its source (a file position, for example), // 0 when the source does not track one. The parser hands each message the // Offset of the last line it consumed, so a reader that records it resumes - // after the whole message, never inside it. + // after the whole message, never inside it. Offsets must only increase within + // a source: one that restarts (a truncated or rotated file) needs to keep them + // increasing, e.g. by carrying a file generation in the high bits. Offset int64 } @@ -165,7 +167,7 @@ func NewParser(ch <-chan LogEntry, decoder Decoder, onMsgCallback OnMsgCallbackF case <-ctx.Done(): return case now := <-flush.C: - expired := partials.expired(now.Add(-multilineCollectorTimeout)) + expired := partials.expired(now.Add(-multilineCollectorTimeout), now.Add(-partialLineMaxAge*multilineCollectorTimeout)) for i, entry := range expired { // Two streams' split lines interleave in the source, so only the // last of them can acknowledge. diff --git a/partial.go b/partial.go index cac1a68..594e55f 100644 --- a/partial.go +++ b/partial.go @@ -12,8 +12,13 @@ import ( // time and its last part's Offset, and stops growing at multilineCollectorLimit. type partialLines map[string]*pendingLine +// partialLineMaxAge is how many multiline timeouts a split line may keep growing +// before it is passed on as it is (see expired). +const partialLineMaxAge = 10 + type pendingLine struct { entry LogEntry + first time.Time // when its first part arrived at time.Time // when its latest part arrived } @@ -23,7 +28,7 @@ func (p partialLines) add(e LogEntry, partial bool, now time.Time) (LogEntry, bo pending := p[e.Stream] if pending == nil { if partial { - p[e.Stream] = &pendingLine{entry: e, at: now} + p[e.Stream] = &pendingLine{entry: e, first: now, at: now} } return e, !partial } @@ -42,10 +47,15 @@ func (p partialLines) add(e LogEntry, partial bool, now time.Time) (LogEntry, bo // expired removes and returns, oldest first, the lines whose latest part arrived // before cutoff. Such a line's last part may never come (an unterminated last line, // a prompt), and it must not wait forever or glue onto the stream's next line. -func (p partialLines) expired(cutoff time.Time) []LogEntry { +// +// It also removes the lines whose first part arrived before firstCutoff, though +// parts still arrive: a line that never ends (a progress bar written without a +// newline) would otherwise stay pending for good, and while any line is pending no +// offset of its source can be acknowledged. Its later parts start a new line. +func (p partialLines) expired(cutoff, firstCutoff time.Time) []LogEntry { var lines []*pendingLine for stream, pending := range p { - if pending.at.Before(cutoff) { + if pending.at.Before(cutoff) || pending.first.Before(firstCutoff) { lines = append(lines, pending) delete(p, stream) }