fix(logs): print logs with zero timestamps immediately to prevent indefinite stalling

pkg/client/logmerger.go

@@ -186,6 +186,13 @@ func (m *LogMerger) run() {
 				e.stream.lastSeen = e.entry.Timestamp
 			}
 			if e.entry.Stream == api.LogStreamStdout || e.entry.Stream == api.LogStreamStderr {
+				// Forward log entries with unparseable timestamps immediately. Without a timestamp we can't place
+				// them on the merged timeline, and buffering them would stall the watermark.
+				if e.entry.Timestamp.IsZero() {
+					m.output <- e.entry
+					<-e.stream.semaphore
+					continue
+				}
 				heap.Push(&m.queue, queuedEntry{entry: e.entry, semaphore: e.stream.semaphore})
 			}
 

pkg/client/logmerger_test.go

@@ -216,6 +216,37 @@ func TestLogMerger_ErrorForwarding(t *testing.T) {
 	assert.Equal(t, "ch1-first", string(results[0].Message))
 }
 
+func TestLogMerger_ZeroTimestampForwarding(t *testing.T) {
+	t.Parallel()
+
+	// Regression for https://github.com/psviderski/uncloud/issues/324: zero-timestamp entries sent
+	// before any real timestamp used to pile up in the heap (watermark stuck at zero) until the
+	// per-stream semaphore saturated and the reader blocked.
+	const zeroCount = logMergerMaxInFlightPerStream + 50
+	ch := make(chan api.ServiceLogEntry, zeroCount+10)
+	merger := NewLogMerger([]<-chan api.ServiceLogEntry{ch}, LogMergerOptions{})
+	output := merger.Stream()
+
+	for i := 0; i < zeroCount; i++ {
+		ch <- testEntry(api.LogStreamStdout, time.Time{}, "zero")
+	}
+	t1 := time.Now()
+	ch <- testEntry(api.LogStreamStdout, t1, "real-after")
+	close(ch)
+
+	results := collectEntries(t, output, 0)
+	require.Len(t, results, zeroCount+1)
+
+	for i := 0; i < zeroCount; i++ {
+		assert.True(t, results[i].Timestamp.IsZero(), "entry %d should have zero timestamp", i)
+		assert.Equal(t, "zero", string(results[i].Message))
+	}
+
+	last := results[len(results)-1]
+	assert.Equal(t, "real-after", string(last.Message))
+	assert.Equal(t, t1, last.Timestamp)
+}
+
 func TestLogMerger_OutOfOrderSingleStream(t *testing.T) {
 	t.Parallel()
 
@@ -324,16 +355,6 @@ func TestLogMerger_StalledStreamExcludedFromWatermark(t *testing.T) {
 	})
 	output := merger.Stream()
 
-	//// Collect all output entries in a separate goroutine.
-	//var results []api.ServiceLogEntry
-	//doneCollecting := make(chan struct{})
-	//go func() {
-	//	for entry := range output {
-	//		results = append(results, entry)
-	//	}
-	//	close(doneCollecting)
-	//}()
-
 	t1 := time.Now()
 	t2 := t1.Add(time.Second)
 	t3 := t1.Add(2 * time.Second)