From d5f4d34e07a156f2c4023d24c112e1b40a3d60fa Mon Sep 17 00:00:00 2001 From: npc0-hue Date: Mon, 31 Aug 2026 05:11:54 +0800 Subject: [PATCH] Prioritize fresh process log sessions --- runtime/worker_test.go | 8 +++++++- spool/log_spool.go | 38 +++++++++++++++++++++++++++++++++++--- spool/log_spool_test.go | 41 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 83 insertions(+), 4 deletions(-) diff --git a/runtime/worker_test.go b/runtime/worker_test.go index dafaaa4..036987d 100644 --- a/runtime/worker_test.go +++ b/runtime/worker_test.go @@ -564,7 +564,13 @@ func TestSpoolLogSinkKeepsSequencesIndependentAndDurable(t *testing.T) { if err != nil { t.Fatalf("pending logs: %v", err) } - if len(pending) != 2 || pending[0].StreamKey != "stderr" || pending[0].FirstSeq != 1 || pending[0].LastSeq != 1 || pending[1].StreamKey != "stdout" || pending[1].FirstSeq != 1 || pending[1].LastSeq != 2 || len(pending[1].Entries) != 2 { + pendingByStream := map[string]protocol.LogBatchIngestRequest{} + for _, batch := range pending { + pendingByStream[batch.StreamKey] = batch + } + stdoutBatch := pendingByStream["stdout"] + stderrBatch := pendingByStream["stderr"] + if len(pending) != 2 || stderrBatch.StreamKey != "stderr" || stderrBatch.FirstSeq != 1 || stderrBatch.LastSeq != 1 || stdoutBatch.StreamKey != "stdout" || stdoutBatch.FirstSeq != 1 || stdoutBatch.LastSeq != 2 || len(stdoutBatch.Entries) != 2 { t.Fatalf("expected independent per-stream sequences, got %+v", pending) } client := &recordingDurableLogClient{} diff --git a/spool/log_spool.go b/spool/log_spool.go index 29c5a63..dd3f98d 100644 --- a/spool/log_spool.go +++ b/spool/log_spool.go @@ -10,6 +10,7 @@ import ( "sort" "strings" "sync" + "time" "browser.local/run/protocol" ) @@ -44,8 +45,9 @@ type durableLogBatch struct { } type pendingLogSegment struct { - path string - batch durableLogBatch + path string + modTime time.Time + batch durableLogBatch } func NewLogSpool(dir string) (LogSpool, error) { @@ -317,11 +319,41 @@ func (spool LogSpool) pendingSegmentsLocked() ([]pendingLogSegment, error) { } return nil, err } - segments = append(segments, pendingLogSegment{path: path, batch: batch}) + var modTime time.Time + if info, statErr := os.Stat(path); statErr == nil { + modTime = info.ModTime() + } + segments = append(segments, pendingLogSegment{path: path, modTime: modTime, batch: batch}) } + sort.SliceStable(segments, func(i, j int) bool { return pendingLogSegmentBefore(segments[i], segments[j]) }) return segments, nil } +func pendingLogSegmentBefore(left pendingLogSegment, right pendingLogSegment) bool { + leftBatch := left.batch.LogBatchIngestRequest + rightBatch := right.batch.LogBatchIngestRequest + if leftBatch.LogStreamID == rightBatch.LogStreamID { + if leftBatch.FirstSeq != rightBatch.FirstSeq { + return leftBatch.FirstSeq < rightBatch.FirstSeq + } + return leftBatch.LastSeq < rightBatch.LastSeq + } + leftTime := pendingLogSegmentPriorityTime(left) + rightTime := pendingLogSegmentPriorityTime(right) + if !leftTime.Equal(rightTime) { + return leftTime.After(rightTime) + } + return left.path < right.path +} + +func pendingLogSegmentPriorityTime(segment pendingLogSegment) time.Time { + batch := segment.batch.LogBatchIngestRequest + if batch.Source == "process" && strings.TrimSpace(batch.LogSessionID) != "" && !batch.SessionStartedAt.IsZero() { + return batch.SessionStartedAt + } + return segment.modTime +} + func (spool LogSpool) Ack(response protocol.LogBatchIngestResponse) error { if response.AcceptedFrom == 0 || response.AcceptedTo < response.AcceptedFrom { return nil diff --git a/spool/log_spool_test.go b/spool/log_spool_test.go index 69bf77c..64414e3 100644 --- a/spool/log_spool_test.go +++ b/spool/log_spool_test.go @@ -125,6 +125,38 @@ func TestLogSpoolQuarantinesPermanentRejectedBatch(t *testing.T) { } } +func TestLogSpoolFlushPrioritizesNewerProcessSessions(t *testing.T) { + logSpool, err := NewLogSpool(t.TempDir()) + if err != nil { + t.Fatalf("new log spool: %v", err) + } + oldBatch := validSpoolLogBatch(1, 1) + oldBatch.LogStreamID = "run.endpoint.server.a-old.stdout" + oldBatch.LogSessionID = "a-old" + oldBatch.SessionStartedAt = time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC) + newBatch := validSpoolLogBatch(1, 1) + newBatch.LogStreamID = "run.endpoint.server.z-new.stdout" + newBatch.LogSessionID = "z-new" + newBatch.SessionStartedAt = time.Date(2026, 7, 3, 13, 0, 0, 0, time.UTC) + newBatchNext := newBatch + newBatchNext.FirstSeq = 2 + newBatchNext.LastSeq = 2 + newBatchNext.Entries = []protocol.LogEntry{{Seq: 2, Timestamp: time.Date(2026, 7, 3, 13, 0, 2, 0, time.UTC), Level: "info", Line: "line 2"}} + for _, batch := range []protocol.LogBatchIngestRequest{oldBatch, newBatch, newBatchNext} { + if err := logSpool.Enqueue(batch); err != nil { + t.Fatalf("enqueue %s:%d: %v", batch.LogStreamID, batch.FirstSeq, err) + } + } + client := &recordingLogBatchClient{} + flushed, err := logSpool.Flush(context.Background(), client) + if err != nil || flushed != 3 { + t.Fatalf("flush: flushed=%d err=%v", flushed, err) + } + if len(client.batches) != 3 || client.batches[0].LogStreamID != newBatch.LogStreamID || client.batches[0].FirstSeq != 1 || client.batches[1].LogStreamID != newBatch.LogStreamID || client.batches[1].FirstSeq != 2 || client.batches[2].LogStreamID != oldBatch.LogStreamID { + t.Fatalf("expected new session first while preserving per-stream order, got %+v", client.batches) + } +} + func TestLogSpoolRestoresPendingAllocationWithoutWatermark(t *testing.T) { root := t.TempDir() first, err := NewLogSpool(root) @@ -269,6 +301,15 @@ func (permanentRejectLogBatchClient) IngestLogBatch(context.Context, protocol.Lo return protocol.LogBatchIngestResponse{}, PermanentLogBatchRejection("platform_sequence_gap", nil) } +type recordingLogBatchClient struct { + batches []protocol.LogBatchIngestRequest +} + +func (client *recordingLogBatchClient) IngestLogBatch(_ context.Context, batch protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) { + client.batches = append(client.batches, batch) + return protocol.LogBatchIngestResponse{Accepted: true, LogStreamID: batch.LogStreamID, AcceptedFrom: batch.FirstSeq, AcceptedTo: batch.LastSeq}, nil +} + func validSpoolLogBatch(firstSeq uint64, lastSeq uint64) protocol.LogBatchIngestRequest { entries := make([]protocol.LogEntry, 0, lastSeq-firstSeq+1) for seq := firstSeq; seq <= lastSeq; seq++ {