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

Remove terminal log replay

Browse Source
This commit is contained in:
npc0-hue
2026-08-25 22:25:09 +08:00
parent 725b71b6b6
commit 7ebe3bb3dd
9 changed files with 40 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
platform/api/log_ingest_handlers_test.go
+16 -34
View File
@@ -81,17 +81,17 @@ func TestLogIngestAPIWorkflow(t *testing.T) {
}
}
func TestLogEventsSSEReplaysHistory(t *testing.T) {
func TestLogEventsSSEDoesNotReplayHistory(t *testing.T) {
router := newTestRouter()
hello := createLogIngestAPIFixtures(t, router)
batch := validLogBatchRequest(t, hello.SessionToken, 1, 2)
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", batch), http.StatusOK)
recorder := performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events?historyLimit=2")
recorder := performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events")
assertStatus(t, recorder, http.StatusOK)
body := recorder.Body.String()
if !strings.Contains(recorder.Header().Get("Content-Type"), "text/event-stream") || !strings.Contains(body, "event: stream") || !strings.Contains(body, "event: log") || !strings.Contains(body, "event: ready") || !strings.Contains(body, `"seq":1`) || !strings.Contains(body, `"seq":2`) {
t.Fatalf("expected stream, history log, and ready SSE events, headers=%v body=%s", recorder.Header(), body)
if !strings.Contains(recorder.Header().Get("Content-Type"), "text/event-stream") || !strings.Contains(body, "event: stream") || !strings.Contains(body, "event: ready") || strings.Contains(body, "event: log") || strings.Contains(body, `"seq":1`) || strings.Contains(body, `"seq":2`) {
t.Fatalf("expected live-only SSE events without history replay, headers=%v body=%s", recorder.Header(), body)
}
}
@@ -122,7 +122,7 @@ func TestLogEventsSSELiveOnlyStartsAfterSnapshotTail(t *testing.T) {
hookResult <- err
}}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
request := httptest.NewRequest(http.MethodGet, "/api/v1/server-instances/server-1/logs/events?historyLimit=0", nil).WithContext(ctx)
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() {
@@ -152,25 +152,6 @@ func TestLogEventsSSELiveOnlyStartsAfterSnapshotTail(t *testing.T) {
assertSSEEvent(t, reader, "log", `"seq":2`)
}
func TestLogEventsSSEUsesServerWideNewestHistory(t *testing.T) {
router := newTestRouter()
hello := createLogIngestAPIFixtures(t, router)
postJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams", dto.LogStreamCreateRequest{
ID: "log-2", ServerInstanceID: "server-1", Source: domain.LogStreamSourceProcess, StreamKey: "stderr",
LogSessionID: "session-current", SessionStartedAt: time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC),
StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default",
})
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequestForStream(t, hello.SessionToken, "log-1", "stdout", 1, 2, 0)), http.StatusOK)
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequestForStream(t, hello.SessionToken, "log-2", "stderr", 1, 2, 10)), http.StatusOK)
recorder := performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events?historyLimit=2")
assertStatus(t, recorder, http.StatusOK)
body := recorder.Body.String()
if strings.Count(body, "event: log") != 2 || !strings.Contains(body, `"streamId":"log-2"`) || strings.Contains(body, `"streamId":"log-1"`) {
t.Fatalf("expected server-wide newest history across streams, body=%s", body)
}
}
func TestLogEventsSSEExcludesOlderSupervisedSessions(t *testing.T) {
router := newTestRouter()
hello := createLogIngestAPIFixtures(t, router)
@@ -182,7 +163,7 @@ func TestLogEventsSSEExcludesOlderSupervisedSessions(t *testing.T) {
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", oldBatch), http.StatusOK)
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequest(t, hello.SessionToken, 1, 1)), http.StatusOK)
recorder := performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events?historyLimit=10")
recorder := performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events")
body := recorder.Body.String()
if !strings.Contains(body, `"logSessionId":"session-current"`) || strings.Contains(body, `"streamId":"log-old"`) {
t.Fatalf("expected only current session in live SSE, body=%s", body)
@@ -196,7 +177,7 @@ func TestLogEventsSSEInitialSnapshotRequiresRunningOnlineProcessFact(t *testing.
stopped := dto.RunLifecycleReportRequest{RunEndpointID: "run-local", SessionToken: hello.SessionToken, ServerInstanceID: "server-1", Capability: domain.LifecycleCapabilityStatus, State: domain.JobStateSucceeded, Progress: dto.JobProgressBody{Percent: 100}, ExecutionResult: dto.RunJobExecutionResultBody{Kind: "process", ProcessState: "stopped", ExitClassification: "requested-stop"}}
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/lifecycle/report", stopped), http.StatusOK)
assertEmptyInitialLogSession(t, performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events?historyLimit=10"))
assertEmptyInitialLogSession(t, performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events"))
running := stopped
running.ManagedProcessID = "log-session:session-missing"
@@ -204,9 +185,9 @@ func TestLogEventsSSEInitialSnapshotRequiresRunningOnlineProcessFact(t *testing.
running.ObservedAt = time.Date(2026, 7, 3, 13, 0, 0, 0, time.UTC)
running.ExecutionResult = dto.RunJobExecutionResultBody{Kind: "process", ProcessState: "running"}
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/lifecycle/report", running), http.StatusOK)
assertEmptyInitialLogSession(t, performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events?historyLimit=10"))
assertEmptyInitialLogSession(t, performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events"))
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/control/heartbeat", dto.RunControlHeartbeatRequest{RunEndpointID: "run-local", SessionToken: hello.SessionToken, Version: "0.1.0", Status: domain.RunEndpointStatusOffline, CapabilityFingerprint: "cap-logs", Capacity: dto.RunCapacityResponse{MaxJobs: 1}}), http.StatusOK)
assertEmptyInitialLogSession(t, performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events?historyLimit=10"))
assertEmptyInitialLogSession(t, performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events"))
}
func TestLogEventsSSEClearsStoppedSessionAndRestoresRunningSessionWithoutReconnect(t *testing.T) {
@@ -215,7 +196,7 @@ func TestLogEventsSSEClearsStoppedSessionAndRestoresRunningSessionWithoutReconne
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?historyLimit=10", nil).WithContext(ctx)
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() {
@@ -234,7 +215,6 @@ func TestLogEventsSSEClearsStoppedSessionAndRestoresRunningSessionWithoutReconne
reader := bufio.NewReader(streamReader)
assertSSEEvent(t, reader, "session", `"logSessionId":"session-current"`)
assertSSEEvent(t, reader, "stream", `"id":"log-1"`)
assertSSEEvent(t, reader, "log", `"seq":1`)
assertSSEEvent(t, reader, "ready", `"streamCount":1`)
statusReport := dto.RunLifecycleReportRequest{RunEndpointID: "run-local", SessionToken: hello.SessionToken, ServerInstanceID: "server-1", Capability: domain.LifecycleCapabilityStatus, State: domain.JobStateSucceeded, Progress: dto.JobProgressBody{Percent: 100}, ExecutionResult: dto.RunJobExecutionResultBody{Kind: "process", ProcessState: "stopped", ExitClassification: "requested-stop"}}
@@ -257,10 +237,14 @@ func TestLogEventsSSEClearsStoppedSessionAndRestoresRunningSessionWithoutReconne
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/lifecycle/report", statusReport), http.StatusOK)
assertSSEEvent(t, reader, "session", `"logSessionId":"session-next"`)
assertSSEEvent(t, reader, "stream", `"id":"run.run-local.server-1.session-next.stdout"`)
nextLive := validLogBatchRequestForStream(t, hello.SessionToken, next.LogStreamID, "stdout", 2, 2, 21)
nextLive.LogSessionID = "session-next"
nextLive.SessionStartedAt = nextStartedAt
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"`)
}
func TestLogEventsSSEOrdersReplaySessionSwitchAndAdditionalStreamWithoutDuplicates(t *testing.T) {
func TestLogEventsSSEOrdersSessionSwitchAndAdditionalStreamWithoutDuplicates(t *testing.T) {
core := service.NewCoreService(repo.NewMemoryStore())
if err := core.SeedLocalPlatformAdmin(); err != nil {
t.Fatalf("seed platform admin: %v", err)
@@ -274,7 +258,7 @@ func TestLogEventsSSEOrdersReplaySessionSwitchAndAdditionalStreamWithoutDuplicat
hookResult <- err
}}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
request := httptest.NewRequest(http.MethodGet, "/api/v1/server-instances/server-1/logs/events?historyLimit=10", nil).WithContext(ctx)
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() {
@@ -293,7 +277,6 @@ func TestLogEventsSSEOrdersReplaySessionSwitchAndAdditionalStreamWithoutDuplicat
reader := bufio.NewReader(streamReader)
assertSSEEvent(t, reader, "session", `"logSessionId":"session-current"`)
assertSSEEvent(t, reader, "stream", `"id":"log-1"`)
assertSSEEvent(t, reader, "log", `"seq":1`)
assertSSEEvent(t, reader, "ready", `"streamCount":1`)
if err := <-hookResult; err != nil {
t.Fatalf("ingest during stream snapshot: %v", err)
@@ -311,7 +294,6 @@ func TestLogEventsSSEOrdersReplaySessionSwitchAndAdditionalStreamWithoutDuplicat
}
assertSSEEvent(t, reader, "session", `"logSessionId":"session-next"`)
assertSSEEvent(t, reader, "stream", `"id":"run.run-local.server-1.session-next.stdout"`)
assertSSEEvent(t, reader, "log", `"seq":1`)
stderr := validLogBatchRequestForStream(t, hello.SessionToken, "run.run-local.server-1.session-next.stderr", "stderr", 1, 1, 21)
stderr.LogSessionID = "session-next"