Filter stale batches from live terminal logs

platform/api/log_ingest_handlers_test.go

@@ -146,12 +146,48 @@ func TestLogEventsSSELiveOnlyStartsAfterSnapshotTail(t *testing.T) {
 		t.Fatalf("ingest during stream snapshot: %v", err)
 	}
 	next := validLogBatchRequest(t, hello.SessionToken, 2, 2)
+	retimestampLogBatchRequest(t, &next, time.Now().UTC().Add(time.Second))
 	if _, err := core.IngestLogBatch(next.ToDomain()); err != nil {
 		t.Fatalf("ingest next live batch: %v", err)
 	}
 	assertSSEEvent(t, reader, "log", `"seq":2`)
 }
 
+func TestLogEventsSSESkipsBufferedBackfillAfterOpen(t *testing.T) {
+	router := newTestRouter()
+	hello := createLogIngestAPIFixtures(t, router)
+	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequest(t, hello.SessionToken, 1, 1)), http.StatusOK)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	request := httptest.NewRequest(http.MethodGet, "/api/v1/server-instances/server-1/logs/events", nil).WithContext(ctx)
+	streamWriter, streamReader := newSSEPipeResponseWriter()
+	done := make(chan struct{})
+	go func() {
+		router.ServeHTTP(streamWriter, request)
+		_ = streamWriter.Close()
+		close(done)
+	}()
+	t.Cleanup(func() {
+		cancel()
+		_ = streamReader.Close()
+		<-done
+	})
+	if status := <-streamWriter.status; status != http.StatusOK {
+		t.Fatalf("unexpected SSE status: %d", status)
+	}
+	reader := bufio.NewReader(streamReader)
+	assertSSEEvent(t, reader, "session", `"logSessionId":"session-current"`)
+	assertSSEEvent(t, reader, "stream", `"id":"log-1"`)
+	assertSSEEvent(t, reader, "ready", `"streamCount":1`)
+
+	stale := validLogBatchRequest(t, hello.SessionToken, 2, 2)
+	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", stale), http.StatusOK)
+	fresh := validLogBatchRequest(t, hello.SessionToken, 3, 3)
+	retimestampLogBatchRequest(t, &fresh, time.Now().UTC().Add(time.Second))
+	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", fresh), http.StatusOK)
+	assertSSEEvent(t, reader, "log", `"seq":3`)
+}
+
 func TestLogEventsSSEExcludesOlderSupervisedSessions(t *testing.T) {
 	router := newTestRouter()
 	hello := createLogIngestAPIFixtures(t, router)
@@ -240,6 +276,7 @@ func TestLogEventsSSEClearsStoppedSessionAndRestoresRunningSessionWithoutReconne
 	nextLive := validLogBatchRequestForStream(t, hello.SessionToken, next.LogStreamID, "stdout", 2, 2, 21)
 	nextLive.LogSessionID = "session-next"
 	nextLive.SessionStartedAt = nextStartedAt
+	retimestampLogBatchRequest(t, &nextLive, time.Now().UTC().Add(time.Second))
 	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", nextLive), http.StatusOK)
 	assertSSEEvent(t, reader, "log", `"streamId":"run.run-local.server-1.session-next.stdout"`)
 }
@@ -298,6 +335,7 @@ func TestLogEventsSSEOrdersSessionSwitchAndAdditionalStreamWithoutDuplicates(t *
 	stderr := validLogBatchRequestForStream(t, hello.SessionToken, "run.run-local.server-1.session-next.stderr", "stderr", 1, 1, 21)
 	stderr.LogSessionID = "session-next"
 	stderr.SessionStartedAt = nextStartedAt
+	retimestampLogBatchRequest(t, &stderr, time.Now().UTC().Add(time.Second))
 	if _, err := core.IngestLogBatch(stderr.ToDomain()); err != nil {
 		t.Fatalf("ingest next session stderr: %v", err)
 	}
@@ -311,6 +349,7 @@ func TestLogEventsSSEOrdersSessionSwitchAndAdditionalStreamWithoutDuplicates(t *
 	nextLive := validLogBatchRequestForStream(t, hello.SessionToken, next.LogStreamID, "stdout", 2, 2, 22)
 	nextLive.LogSessionID = "session-next"
 	nextLive.SessionStartedAt = nextStartedAt
+	retimestampLogBatchRequest(t, &nextLive, time.Now().UTC().Add(time.Second))
 	if _, err := core.IngestLogBatch(nextLive.ToDomain()); err != nil {
 		t.Fatalf("ingest next session live append: %v", err)
 	}
@@ -452,3 +491,17 @@ func validLogBatchRequestForStream(t *testing.T, sessionToken string, streamID s
 		Entries:          entries,
 	}
 }
+
+func retimestampLogBatchRequest(t *testing.T, request *dto.LogBatchIngestRequest, first time.Time) {
+	t.Helper()
+	domainEntries := make([]domain.LogEntry, 0, len(request.Entries))
+	for index := range request.Entries {
+		request.Entries[index].Timestamp = first.Add(time.Duration(index) * time.Millisecond).UTC()
+		domainEntries = append(domainEntries, domain.LogEntry{Seq: request.Entries[index].Seq, Timestamp: request.Entries[index].Timestamp, Level: request.Entries[index].Level, Line: request.Entries[index].Line, Fields: request.Entries[index].Fields, Redacted: request.Entries[index].Redacted})
+	}
+	checksum, err := validator.LogEntriesChecksum(domainEntries)
+	if err != nil {
+		t.Fatalf("checksum retimestamped entries: %v", err)
+	}
+	request.Checksum = checksum
+}