admin343/run
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Prioritize fresh process log sessions

Browse Source
This commit is contained in:
npc0-hue
2026-08-31 05:11:54 +08:00
parent 90de21ad9f
commit d5f4d34e07
3 changed files with 83 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spool/log_spool_test.go
+41
View File
@@ -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++ {