parser: pass on a split line that never ends once its first part is old

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

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) {

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.

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)
 		}