Relay current process output without spool

2026-08-31 23:35:14 +08:00
parent cde99e19fa
commit c8f5213954
9 changed files with 239 additions and 90 deletions
@@ -132,7 +132,7 @@ func TestTypedProcessOutputCaptureResumesAfterRunRestart(t *testing.T) {
 	waitForManagedTailers(t, restarted.managed.(*OSManagedProcessSupervisor))
 }

-func TestManagedProcessOutputAfterCanceledRunContextIsSpooledOnRestart(t *testing.T) {
+func TestManagedProcessOutputAfterCanceledRunContextIsNotSpooledOnRestart(t *testing.T) {
 	root := t.TempDir()
 	assignment := executionAssignment(protocol.RunCapabilityProcessStart)
 	assignment.ExecutionInput.LogSources = []protocol.RuntimeLogSourcePlan{
@@ -199,24 +199,14 @@ func TestManagedProcessOutputAfterCanceledRunContextIsSpooledOnRestart(t *testin
 		WithProcessLogSink(&SpoolLogSink{RunEndpointID: assignment.RunEndpointID, SessionToken: "new-session", Spool: restartedSpool}),
 	)
 	restarted.ResumeManagedProcessLogs(context.Background())
-	waitForSpooledText(t, restartedSpool, "managed stdout triggered after-cancel", "managed stderr triggered after-cancel")
-	restartedManaged := restarted.managed.(*OSManagedProcessSupervisor)
-	resumed := waitForManagedOffsets(t, restartedManaged, scope, identity.StdoutOffset, identity.StderrOffset)
-	if resumed.LogSessionID != identity.LogSessionID || resumed.StdoutOffset <= identity.StdoutOffset || resumed.StderrOffset <= identity.StderrOffset {
-		t.Fatalf("restart did not retain session and commit offsets: before=%+v after=%+v", identity, resumed)
+	time.Sleep(300 * time.Millisecond)
+	if pending := mustPendingLogs(t, restartedSpool); len(pending) != 0 {
+		t.Fatalf("restart replayed output observed before Run restart: %+v", pending)
 	}
-	for _, batch := range mustPendingLogs(t, restartedSpool) {
-		if batch.LogSessionID != identity.LogSessionID || !batch.SessionStartedAt.Equal(identity.StartedAt) {
-			t.Fatalf("spooled batch lost process session metadata: %+v", batch)
-		}
-		if batch.FirstSeq != 1 || len(batch.Entries) == 0 || batch.Entries[0].Seq != 1 {
-			t.Fatalf("fresh generation stream did not start durably at sequence 1: %+v", batch)
-		}
-		for index, entry := range batch.Entries {
-			if entry.Seq != uint64(index+1) {
-				t.Fatalf("fresh generation stream sequence is not contiguous: %+v", batch)
-			}
-		}
+	restartedManaged := restarted.managed.(*OSManagedProcessSupervisor)
+	resumed := restartedManaged.Status(ProcessIdentity{Scope: scope})
+	if resumed.LogSessionID != identity.LogSessionID || resumed.StdoutOffset != identity.StdoutOffset || resumed.StderrOffset != identity.StderrOffset {
+		t.Fatalf("restart changed output offsets without observing new output: before=%+v after=%+v", identity, resumed)
 	}
 	stop := executionAssignment(protocol.RunCapabilityProcessStop)
 	stop.TargetKey = "actions/stop.json"
@@ -293,7 +283,7 @@ func TestImmediateManagedProcessRestartTailsNewGeneration(t *testing.T) {
 	waitForManagedTailers(t, managed)
 }

-func TestManagedProcessRestartRetainsUndrainedRetiredGeneration(t *testing.T) {
+func TestManagedProcessRestartDoesNotRetainUndrainedRetiredGeneration(t *testing.T) {
 	root := t.TempDir()
 	assignment := executionAssignment(protocol.RunCapabilityProcessStart)
 	assignment.ExecutionInput.LogSources = []protocol.RuntimeLogSourcePlan{{Kind: "process.stdout", StreamKey: "game.console.stdout"}, {Kind: "process.stderr", StreamKey: "game.console.stderr"}}
@@ -328,9 +318,9 @@ func TestManagedProcessRestartRetainsUndrainedRetiredGeneration(t *testing.T) {
 		t.Fatalf("process restart reused log session: first=%+v second=%+v", first, second)
 	}
 	oldManaged.mu.Lock()
-	if len(oldManaged.retired) != 1 {
+	if len(oldManaged.retired) != 0 {
 		oldManaged.mu.Unlock()
-		t.Fatalf("expected one durable retired generation, got %+v", oldManaged.retired)
+		t.Fatalf("expected no durable retired generation, got %+v", oldManaged.retired)
 	}
 	oldManaged.stopTailersLocked(first)
 	oldManaged.stopTailersLocked(second)
@@ -342,7 +332,10 @@ func TestManagedProcessRestartRetainsUndrainedRetiredGeneration(t *testing.T) {
 	}
 	restarted := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root), WithProcessLogSink(&SpoolLogSink{RunEndpointID: assignment.RunEndpointID, SessionToken: "restart-session", Spool: logSpool}))
 	restarted.ResumeManagedProcessLogs(context.Background())
-	waitForSpooledText(t, logSpool, "retired-a", "current-b")
+	time.Sleep(300 * time.Millisecond)
+	if pending := mustPendingLogs(t, logSpool); len(pending) != 0 {
+		t.Fatalf("restart replayed retired output: %+v", pending)
+	}
 	restartedManaged := restarted.managed.(*OSManagedProcessSupervisor)
 	waitForManagedTailers(t, restartedManaged)
 	restartedManaged.mu.Lock()
@@ -353,7 +346,7 @@ func TestManagedProcessRestartRetainsUndrainedRetiredGeneration(t *testing.T) {
 	}
 }

-func TestManagedProcessSourceCursorDeduplicatesCommittedLineWithStaleJournalOffset(t *testing.T) {
+func TestManagedProcessRestartDoesNotReplayHistoricalOutput(t *testing.T) {
 	root := t.TempDir()
 	stateDir := filepath.Join(root, "state")
 	outputDir := filepath.Join(stateDir, "process-output")
@@ -361,7 +354,7 @@ func TestManagedProcessSourceCursorDeduplicatesCommittedLineWithStaleJournalOffs
 		t.Fatalf("create process output dir: %v", err)
 	}
 	line := "committed before offset\n"
-	identity := ProcessIdentity{Scope: filepath.Join(root, "instances", "server-execution", "local"), ServerInstanceID: "server-execution", RunEndpointID: "run-execution", JobID: "execution-job", Capability: protocol.RunCapabilityProcessStart, ProfileKey: "local", LogSessionID: "session-stale-offset", PID: 12345, StartedAt: time.Now().UTC(), State: "exited", StdoutLogRef: "stale.stdout.log", StderrLogRef: "stale.stderr.log", StdoutStreamKey: "game.console.stdout", StderrStreamKey: "game.console.stderr"}
+	identity := ProcessIdentity{Scope: filepath.Join(root, "instances", "server-execution", "local"), ServerInstanceID: "server-execution", RunEndpointID: "run-execution", JobID: "execution-job", Capability: protocol.RunCapabilityProcessStart, ProfileKey: "local", LogSessionID: "session-stale-offset", PID: os.Getpid(), StartedAt: time.Now().UTC(), State: "running", StdoutLogRef: "stale.stdout.log", StderrLogRef: "stale.stderr.log", StdoutStreamKey: "game.console.stdout", StderrStreamKey: "game.console.stderr"}
 	if err := os.WriteFile(filepath.Join(outputDir, identity.StdoutLogRef), []byte(line), 0o600); err != nil {
 		t.Fatalf("write stale stdout: %v", err)
 	}
@@ -375,34 +368,21 @@ func TestManagedProcessSourceCursorDeduplicatesCommittedLineWithStaleJournalOffs
 	if err := os.WriteFile(filepath.Join(stateDir, "processes.json"), body, 0o600); err != nil {
 		t.Fatalf("write process journal: %v", err)
 	}
-	logSpool, err := spool.NewLogSpool(filepath.Join(root, "spool"))
-	if err != nil {
-		t.Fatalf("new log spool: %v", err)
-	}
-	assignment := assignmentFromProcessIdentity(identity)
-	sink := &SpoolLogSink{RunEndpointID: identity.RunEndpointID, SessionToken: "old-run-session", Spool: logSpool}
-	if err := sink.AppendWithCursor(context.Background(), assignment, "stdout", strings.TrimSpace(line), ProcessLogCursor{StartOffset: 0, EndOffset: int64(len(line))}); err != nil {
-		t.Fatalf("commit line before offset: %v", err)
-	}
-	streamID := logStreamIDForAssignment(assignment, identity.StdoutStreamKey)
-	if err := logSpool.Ack(protocol.LogBatchIngestResponse{LogStreamID: streamID, AcceptedFrom: 1, AcceptedTo: 1}); err != nil {
-		t.Fatalf("ack committed line: %v", err)
-	}
-	restartedSpool, err := spool.NewLogSpool(filepath.Join(root, "spool"))
-	if err != nil {
-		t.Fatalf("restart log spool: %v", err)
-	}
-	restarted := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root), WithProcessLogSink(&SpoolLogSink{RunEndpointID: identity.RunEndpointID, SessionToken: "new-run-session", Spool: restartedSpool}))
+	sink := &recordingLogSink{}
+	restarted := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root), WithProcessLogSink(sink))
 	restarted.ResumeManagedProcessLogs(context.Background())
 	managed := restarted.managed.(*OSManagedProcessSupervisor)
 	resumed := waitForManagedOffsets(t, managed, identity.Scope, 0, -1)
-	waitForManagedTailers(t, managed)
+	time.Sleep(100 * time.Millisecond)
 	if resumed.StdoutOffset != int64(len(line)) {
-		t.Fatalf("stale journal offset was not advanced: %+v", resumed)
+		t.Fatalf("restart did not move live tail to the current output boundary: %+v", resumed)
 	}
-	if pending := mustPendingLogs(t, restartedSpool); len(pending) != 0 {
-		t.Fatalf("committed source cursor was enqueued twice: %+v", pending)
+	if lines := sink.snapshot(); len(lines) != 0 {
+		t.Fatalf("restart replayed historical output: %+v", lines)
 	}
+	managed.mu.Lock()
+	managed.stopTailersLocked(identity)
+	managed.mu.Unlock()
 }

 func TestManagedProcessSupervisorMigratesLiveLegacySession(t *testing.T) {
@@ -205,11 +205,6 @@ func (supervisor *OSManagedProcessSupervisor) Start(ctx context.Context, command
 	identity.UpdatedAt = identity.StartedAt
 	identity.CommandFingerprint = fingerprintArgs(command.Args)
 	previous, hadPrevious := supervisor.items[key]
-	retiredKey := ""
-	if hadPrevious && supervisor.hasPendingOutput(previous) {
-		retiredKey = managedProcessGenerationKey(previous)
-		supervisor.retired[retiredKey] = previous
-	}
 	supervisor.items[key] = identity
 	if err := supervisor.persistLocked(); err != nil {
 		_ = forceManagedProcessStop(identity)
@@ -219,9 +214,6 @@ func (supervisor *OSManagedProcessSupervisor) Start(ctx context.Context, command
 		} else {
 			delete(supervisor.items, key)
 		}
-		if retiredKey != "" {
-			delete(supervisor.retired, retiredKey)
-		}
 		log.Printf("RUN phase=process.managed status=persist_failed job=%s pid=%d error=%s", safeOptional(identity.JobID), identity.PID, RedactText(err.Error()))
 		return ProcessIdentity{}, err
 	}
@@ -324,6 +316,8 @@ func (supervisor *OSManagedProcessSupervisor) ResumeOutput(output ManagedProcess
 	changed := false
 	for _, item := range supervisor.items {
 		if item.State == "running" && supervisor.isAlive(item) {
+			item.StdoutOffset = supervisor.outputFileSize(item.StdoutLogRef, item.StdoutOffset)
+			item.StderrOffset = supervisor.outputFileSize(item.StderrLogRef, item.StderrOffset)
 			item.State = "running"
 			item.ObservationSeq++
 			item.UpdatedAt = now
@@ -331,17 +325,9 @@ func (supervisor *OSManagedProcessSupervisor) ResumeOutput(output ManagedProcess
 			changed = true
 			supervisor.startTailersLocked(item, output)
 			log.Printf("RUN phase=process.managed status=resume_output pid=%d server=%s scope=%s", item.PID, item.ServerInstanceID, safeOptional(item.Scope))
-		} else if supervisor.hasPendingOutput(item) {
-			supervisor.startDrainTailersLocked(item, output)
-			log.Printf("RUN phase=process.managed status=resume_output_drain pid=%d server=%s scope=%s", item.PID, item.ServerInstanceID, safeOptional(item.Scope))
 		}
 	}
-	for key, item := range supervisor.retired {
-		if supervisor.hasPendingOutput(item) {
-			supervisor.startDrainTailersLocked(item, output)
-			log.Printf("RUN phase=process.managed status=resume_retired_output_drain pid=%d server=%s scope=%s", item.PID, item.ServerInstanceID, safeOptional(item.Scope))
-			continue
-		}
+	for key := range supervisor.retired {
 		delete(supervisor.retired, key)
 		changed = true
 	}
@@ -350,6 +336,17 @@ func (supervisor *OSManagedProcessSupervisor) ResumeOutput(output ManagedProcess
 	}
 }

+func (supervisor *OSManagedProcessSupervisor) outputFileSize(ref string, fallback int64) int64 {
+	if ref == "" {
+		return fallback
+	}
+	info, err := os.Stat(filepath.Join(supervisor.outputRoot, "state", "process-output", filepath.Base(ref)))
+	if err != nil || info.Size() < fallback {
+		return fallback
+	}
+	return info.Size()
+}
+
 func (supervisor *OSManagedProcessSupervisor) Reconcile() {
 	supervisor.mu.Lock()
 	defer supervisor.mu.Unlock()
@@ -361,7 +358,9 @@ func (supervisor *OSManagedProcessSupervisor) Reconcile() {
 			item.UpdatedAt = time.Now().UTC()
 			item.ObservationSeq++
 			supervisor.items[key] = item
-			supervisor.drainTailersLocked(item)
+			// A Run restart must never replay output produced before this
+			// process became observable again.
+			supervisor.stopTailersLocked(item)
 			changed = true
 		}
 	}
@@ -359,6 +359,9 @@ func (worker *Worker) registerUnlocked(ctx context.Context) error {
 		}
 		sink.mu.Unlock()
 	}
+	if sink, ok := worker.executor.logSink.(*LiveLogSink); ok {
+		sink.SetSession(state.RunEndpointID, state.SessionToken)
+	}
 	if hook, ok := worker.executor.artifactHook.(*QueueArtifactHook); ok {
 		hook.RunEndpointID = state.RunEndpointID
 		hook.SessionToken = state.SessionToken
@@ -1438,6 +1441,118 @@ type SpoolLogSink struct {
 	Progress      func(context.Context, string, string) (uint64, error)
 }

+type LiveLogClient interface {
+	RelayLiveLogBatch(context.Context, protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error)
+}
+
+// LiveLogSink is process-local and best-effort. It has no disk spool, no
+// resend backlog, and does not make lifecycle execution wait for the platform.
+type LiveLogSink struct {
+	Client        LiveLogClient
+	RunEndpointID string
+	SessionToken  string
+
+	mu        sync.Mutex
+	sequences map[string]uint64
+	queue     chan protocol.LogBatchIngestRequest
+	closed    bool
+	closeOnce sync.Once
+}
+
+func NewLiveLogSink(client LiveLogClient) *LiveLogSink {
+	sink := &LiveLogSink{Client: client, sequences: map[string]uint64{}, queue: make(chan protocol.LogBatchIngestRequest, 512)}
+	go sink.dispatch()
+	return sink
+}
+
+func (sink *LiveLogSink) SetSession(endpointID string, sessionToken string) {
+	sink.mu.Lock()
+	sink.RunEndpointID = endpointID
+	sink.SessionToken = sessionToken
+	sink.mu.Unlock()
+}
+
+func (sink *LiveLogSink) Append(ctx context.Context, assignment protocol.RunJobAssignment, stream string, line string) error {
+	return sink.append(assignment, stream, line)
+}
+
+func (sink *LiveLogSink) AppendWithCursor(ctx context.Context, assignment protocol.RunJobAssignment, stream string, line string, _ ProcessLogCursor) error {
+	return sink.append(assignment, stream, line)
+}
+
+func (sink *LiveLogSink) append(assignment protocol.RunJobAssignment, stream string, line string) error {
+	if sink == nil {
+		return nil
+	}
+	redactedLine := RedactText(line)
+	redacted := line != redactedLine || strings.HasPrefix(redactedLine, "[redacted protected ")
+	streamKey := declaredProcessStreamKey(assignment, stream)
+	logStreamID := logStreamIDForAssignment(assignment, streamKey)
+	sink.mu.Lock()
+	defer sink.mu.Unlock()
+	if sink.closed || sink.queue == nil {
+		return nil
+	}
+	endpointID := sink.RunEndpointID
+	if endpointID == "" {
+		endpointID = assignment.RunEndpointID
+	}
+	if endpointID == "" || sink.SessionToken == "" || assignment.ServerInstanceID == "" {
+		return nil
+	}
+	sink.sequences[logStreamID]++
+	sequence := sink.sequences[logStreamID]
+	batch := protocol.LogBatchIngestRequest{
+		RunEndpointID:    endpointID,
+		SessionToken:     sink.SessionToken,
+		LogStreamID:      logStreamID,
+		ServerInstanceID: assignment.ServerInstanceID,
+		StreamKey:        streamKey,
+		Source:           "process",
+		LogSessionID:     assignment.LogSessionID,
+		SessionStartedAt: assignment.SessionStartedAt,
+		FirstSeq:         sequence,
+		LastSeq:          sequence,
+		Compression:      "none",
+		Entries:          []protocol.LogEntry{{Seq: sequence, Timestamp: time.Now().UTC(), Level: "info", Line: redactedLine, Redacted: redacted}},
+	}
+	batch.Checksum, _ = checksumForLogEntries(batch.Entries)
+	select {
+	case sink.queue <- batch:
+	default:
+		log.Printf("RUN phase=live_log_relay status=dropped stream=%s sequence=%d reason=transient_queue_full", safeOptional(logStreamID), sequence)
+	}
+	return nil
+}
+
+func (sink *LiveLogSink) dispatch() {
+	for batch := range sink.queue {
+		if sink.Client == nil {
+			continue
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		_, err := sink.Client.RelayLiveLogBatch(ctx, batch)
+		cancel()
+		if err != nil {
+			log.Printf("RUN phase=live_log_relay status=dropped stream=%s sequence=%d error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, RedactText(err.Error()))
+		}
+	}
+}
+
+func (sink *LiveLogSink) Close() {
+	if sink == nil {
+		return
+	}
+	sink.closeOnce.Do(func() {
+		sink.mu.Lock()
+		sink.closed = true
+		if sink.queue != nil {
+			close(sink.queue)
+		}
+		sink.mu.Unlock()
+	})
+}
+
 func (sink *SpoolLogSink) Append(ctx context.Context, assignment protocol.RunJobAssignment, stream string, line string) error {
 	return sink.append(ctx, assignment, stream, line, nil)
 }
@@ -404,6 +404,33 @@ func TestWorkerSpoolHooksUseRegisteredSession(t *testing.T) {
 	}
 }

+func TestLiveLogSinkQueuesCurrentBatchWithoutDurableSpool(t *testing.T) {
+	client := &recordingLiveLogClient{received: make(chan protocol.LogBatchIngestRequest, 1)}
+	sink := NewLiveLogSink(client)
+	sink.SetSession("run-test", "session-token")
+	t.Cleanup(sink.Close)
+	assignment := protocol.RunJobAssignment{
+		JobID:            "execution-job",
+		RunEndpointID:    "run-test",
+		ServerInstanceID: "server-worker",
+		Capability:       protocol.RunCapabilityProcessStart,
+		LogSessionID:     "generation-current",
+		SessionStartedAt: workerTestTime(),
+		ExecutionInput:   protocol.RunJobExecutionInput{LogSources: []protocol.RuntimeLogSourcePlan{{Kind: "process.stdout", StreamKey: "game.console.stdout"}}},
+	}
+	if err := sink.Append(context.Background(), assignment, "stdout", "current output"); err != nil {
+		t.Fatalf("append live log: %v", err)
+	}
+	select {
+	case batch := <-client.received:
+		if batch.LogStreamID != "run.run-test.server-worker.generation-current.game.console.stdout" || batch.FirstSeq != 1 || batch.LastSeq != 1 || batch.Checksum == "" || len(batch.Entries) != 1 || batch.Entries[0].Line != "current output" {
+			t.Fatalf("unexpected live batch: %+v", batch)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("live log was not dispatched")
+	}
+}
+
 func TestSpoolLogSinkUsesRunScopedStreamForAutonomousLifecycle(t *testing.T) {
 	logSpool, err := spool.NewLogSpool(t.TempDir())
 	if err != nil {
@@ -898,6 +925,15 @@ type recordingDurableLogClient struct {
 	err   error
 }

+type recordingLiveLogClient struct {
+	received chan protocol.LogBatchIngestRequest
+}
+
+func (client *recordingLiveLogClient) RelayLiveLogBatch(_ context.Context, batch protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) {
+	client.received <- batch
+	return protocol.LogBatchIngestResponse{Accepted: true, LogStreamID: batch.LogStreamID, AcceptedFrom: batch.FirstSeq, AcceptedTo: batch.LastSeq}, nil
+}
+
 func (client *recordingDurableLogClient) IngestLogBatch(_ context.Context, batch protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) {
 	client.batch = batch
 	if client.err != nil {