diff --git a/runtime/execution_test.go b/runtime/execution_test.go index d55a2f1..8eec72a 100644 --- a/runtime/execution_test.go +++ b/runtime/execution_test.go @@ -497,6 +497,13 @@ func TestTypedProcessStartStopIsIdempotentAndReconciles(t *testing.T) { if resumed.LogSessionID != first.LogSessionID { t.Fatalf("expected Run restart to retain session, before=%q after=%q", first.LogSessionID, resumed.LogSessionID) } + beforeResumeSeq := resumed.ObservationSeq + beforeResumeUpdatedAt := resumed.UpdatedAt + restarted.ResumeManagedProcessLogs(context.Background()) + resumed = restarted.managed.(*OSManagedProcessSupervisor).Status(ProcessIdentity{Scope: processScope(root, assignment)}) + if resumed.ObservationSeq <= beforeResumeSeq || !resumed.UpdatedAt.After(beforeResumeUpdatedAt) { + t.Fatalf("expected resume to refresh process observation, beforeSeq=%d after=%+v", beforeResumeSeq, resumed) + } stop := executionAssignment(protocol.RunCapabilityProcessStop) stop.TargetKey = "actions/stop.json" stopped := restarted.Execute(stop) diff --git a/runtime/process_supervisor.go b/runtime/process_supervisor.go index 53aa2a7..441baae 100644 --- a/runtime/process_supervisor.go +++ b/runtime/process_supervisor.go @@ -320,8 +320,15 @@ func (supervisor *OSManagedProcessSupervisor) Status(identity ProcessIdentity) P func (supervisor *OSManagedProcessSupervisor) ResumeOutput(output ManagedProcessOutput) { supervisor.mu.Lock() defer supervisor.mu.Unlock() + now := time.Now().UTC() + changed := false for _, item := range supervisor.items { if item.State == "running" && supervisor.isAlive(item) { + item.State = "running" + item.ObservationSeq++ + item.UpdatedAt = now + supervisor.items[item.Scope] = item + 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) { @@ -329,7 +336,6 @@ func (supervisor *OSManagedProcessSupervisor) ResumeOutput(output ManagedProcess log.Printf("RUN phase=process.managed status=resume_output_drain pid=%d server=%s scope=%s", item.PID, item.ServerInstanceID, safeOptional(item.Scope)) } } - changed := false for key, item := range supervisor.retired { if supervisor.hasPendingOutput(item) { supervisor.startDrainTailersLocked(item, output) diff --git a/runtime/worker.go b/runtime/worker.go index 2295573..e9f5d73 100644 --- a/runtime/worker.go +++ b/runtime/worker.go @@ -1185,14 +1185,19 @@ type durableLogClient interface { IngestLogBatch(context.Context, protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) } +type durableLogProgressClient interface { + GetRunLogStreamProgress(context.Context, protocol.RunLogStreamProgressRequest) (protocol.RunLogStreamProgressResponse, error) +} + type durableArtifactClient interface { UploadArtifactChunk(context.Context, protocol.ArtifactChunkUploadRequest) (protocol.ArtifactChunkUploadResponse, error) } type sessionLogBatchClient struct { - client durableLogClient - runEndpointID string - sessionToken string + client durableLogClient + progressClient durableLogProgressClient + runEndpointID string + sessionToken string } type sessionLogStreamProgressClient struct { @@ -1250,6 +1255,26 @@ func (client sessionLogBatchClient) IngestLogBatch(ctx context.Context, batch pr return response, err } +func (client sessionLogBatchClient) LogStreamLatestSeq(ctx context.Context, batch protocol.LogBatchIngestRequest) (uint64, error) { + if client.progressClient == nil { + return 0, fmt.Errorf("platform log progress client is unavailable") + } + response, err := client.progressClient.GetRunLogStreamProgress(ctx, protocol.RunLogStreamProgressRequest{ + RunEndpointID: client.runEndpointID, + SessionToken: client.sessionToken, + ServerInstanceID: batch.ServerInstanceID, + LogStreamID: batch.LogStreamID, + }) + if err != nil { + return 0, err + } + if !response.Accepted || response.LogStreamID != batch.LogStreamID { + return 0, fmt.Errorf("platform log stream progress response is invalid") + } + log.Printf("RUN phase=durable_uploaders.logs status=sequence_recover stream=%s localFirstSeq=%d platformLatestSeq=%d", safeOptional(batch.LogStreamID), batch.FirstSeq, response.LatestSeq) + return response.LatestSeq, nil +} + func logBatchNotFoundError(err error) bool { var httpErr interface{ HTTPStatus() int } return errors.As(err, &httpErr) && httpErr.HTTPStatus() == http.StatusNotFound @@ -1305,6 +1330,7 @@ func (worker *Worker) runDurableUploaders(ctx context.Context, done chan<- struc logSink, hasLogSink := worker.executor.logSink.(*SpoolLogSink) artifactHook, hasArtifactHook := worker.executor.artifactHook.(*QueueArtifactHook) logClient, hasLogClient := worker.client.(durableLogClient) + progressClient, _ := worker.client.(durableLogProgressClient) artifactClient, hasArtifactClient := worker.client.(durableArtifactClient) if (!hasLogSink || !hasLogClient) && (!hasArtifactHook || !hasArtifactClient) { log.Printf("RUN phase=durable_uploaders status=disabled logs=%t artifacts=%t", hasLogSink && hasLogClient, hasArtifactHook && hasArtifactClient) @@ -1328,7 +1354,7 @@ func (worker *Worker) runDurableUploaders(ctx context.Context, done chan<- struc startedAt := time.Now() log.Printf("RUN phase=durable_uploaders.logs status=flush_start timeoutMs=%d", durableUploaderFlushTimeout.Milliseconds()) flushCtx, cancel := context.WithTimeout(ctx, durableUploaderFlushTimeout) - client := sessionLogBatchClient{client: logClient, runEndpointID: state.RunEndpointID, sessionToken: state.SessionToken} + client := sessionLogBatchClient{client: logClient, progressClient: progressClient, runEndpointID: state.RunEndpointID, sessionToken: state.SessionToken} flushed, err := logSink.Spool.Flush(flushCtx, client) if err != nil { log.Printf("RUN phase=durable_uploaders.logs status=flush_failed flushed=%d durationMs=%d error=%s", flushed, time.Since(startedAt).Milliseconds(), RedactText(err.Error())) diff --git a/runtime/worker_test.go b/runtime/worker_test.go index 036987d..9779ae3 100644 --- a/runtime/worker_test.go +++ b/runtime/worker_test.go @@ -636,6 +636,19 @@ func TestSessionLogBatchClientQuarantinesSequenceGap(t *testing.T) { } } +func TestSessionLogBatchClientRecoversLatestSeqThroughProgressEndpoint(t *testing.T) { + workerClient := newFakeWorkerClient() + workerClient.logStreamProgress = 736 + client := sessionLogBatchClient{client: &recordingDurableLogClient{}, progressClient: workerClient, runEndpointID: "run-current", sessionToken: "token-current"} + latestSeq, err := client.LogStreamLatestSeq(context.Background(), protocol.LogBatchIngestRequest{LogStreamID: "run.run-current.server-worker.session.stdout", ServerInstanceID: "server-worker", FirstSeq: 10413}) + if err != nil || latestSeq != 736 { + t.Fatalf("recover latest sequence: latestSeq=%d err=%v", latestSeq, err) + } + if len(workerClient.logProgressRequests) != 1 || workerClient.logProgressRequests[0].RunEndpointID != "run-current" || workerClient.logProgressRequests[0].SessionToken != "token-current" || workerClient.logProgressRequests[0].ServerInstanceID != "server-worker" || workerClient.logProgressRequests[0].LogStreamID != "run.run-current.server-worker.session.stdout" { + t.Fatalf("unexpected progress request: %+v", workerClient.logProgressRequests) + } +} + func TestSessionLogBatchClientQuarantinesSequenceConflict(t *testing.T) { recorder := &recordingDurableLogClient{err: api.PlatformRequestError{Status: http.StatusBadRequest, Code: "validation_failed", Details: []string{"log batch conflicts with acknowledged range"}}} client := sessionLogBatchClient{client: recorder, runEndpointID: "run-current", sessionToken: "token-current"} diff --git a/spool/log_spool.go b/spool/log_spool.go index dd3f98d..bb7d10b 100644 --- a/spool/log_spool.go +++ b/spool/log_spool.go @@ -428,6 +428,10 @@ type LogBatchClient interface { IngestLogBatch(context.Context, protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) } +type LogStreamProgressClient interface { + LogStreamLatestSeq(context.Context, protocol.LogBatchIngestRequest) (uint64, error) +} + type PermanentLogBatchError struct { Reason string Err error @@ -479,6 +483,21 @@ func (spool LogSpool) Flush(ctx context.Context, client LogBatchClient) (int, er if err != nil { var permanent PermanentLogBatchError if errors.As(err, &permanent) { + if permanent.Reason == "platform_sequence_gap" { + latestSeq, progressErr := recoverLogStreamLatestSeq(ctx, client, batch) + if progressErr != nil { + return acknowledged, err + } + rejected, rejectErr := spool.RejectStreamAfter(batch.LogStreamID, latestSeq, permanent.Reason) + if rejectErr != nil { + return acknowledged, rejectErr + } + if resetErr := spool.ResetStreamWatermark(batch.LogStreamID, latestSeq); resetErr != nil { + return acknowledged, resetErr + } + acknowledged += rejected + return acknowledged, nil + } if rejectErr := spool.Reject(batch, permanent.Reason); rejectErr != nil { return acknowledged, rejectErr } @@ -498,6 +517,50 @@ func (spool LogSpool) Flush(ctx context.Context, client LogBatchClient) (int, er return acknowledged, nil } +func recoverLogStreamLatestSeq(ctx context.Context, client LogBatchClient, batch protocol.LogBatchIngestRequest) (uint64, error) { + progressClient, ok := client.(LogStreamProgressClient) + if !ok { + return 0, fmt.Errorf("log stream progress client is required after sequence gap") + } + return progressClient.LogStreamLatestSeq(ctx, batch) +} + +func (spool LogSpool) ResetStreamWatermark(logStreamID string, latestSeq uint64) error { + spool.mu.Lock() + defer spool.mu.Unlock() + watermark := spool.watermarks[logStreamID] + watermark.Allocated = latestSeq + watermark.Acknowledged = latestSeq + spool.watermarks[logStreamID] = watermark + return spool.persistWatermarksLocked() +} + +func (spool LogSpool) RejectStreamAfter(logStreamID string, latestSeq uint64, reason string) (int, error) { + spool.mu.Lock() + defer spool.mu.Unlock() + segments, err := spool.pendingSegmentsLocked() + if err != nil { + return 0, err + } + rejectedDir := filepath.Join(filepath.Dir(spool.dir), "logs-rejected") + rejected := 0 + for _, segment := range segments { + batch := segment.batch.LogBatchIngestRequest + if batch.LogStreamID != logStreamID || batch.FirstSeq <= latestSeq { + continue + } + if err := os.MkdirAll(rejectedDir, 0o755); err != nil { + return rejected, fmt.Errorf("create rejected log spool directory: %w", err) + } + rejectedPath := filepath.Join(rejectedDir, fmt.Sprintf("%s.%s", filepath.Base(segment.path), sanitizeSegmentName(reason))) + if err := os.Rename(segment.path, rejectedPath); err != nil { + return rejected, fmt.Errorf("move rejected log spool segment: %w", err) + } + rejected++ + } + return rejected, nil +} + func (spool LogSpool) Reject(batch protocol.LogBatchIngestRequest, reason string) error { spool.mu.Lock() defer spool.mu.Unlock() diff --git a/spool/log_spool_test.go b/spool/log_spool_test.go index 64414e3..ebe063e 100644 --- a/spool/log_spool_test.go +++ b/spool/log_spool_test.go @@ -157,6 +157,37 @@ func TestLogSpoolFlushPrioritizesNewerProcessSessions(t *testing.T) { } } +func TestLogSpoolSequenceGapResetsWatermarkToPlatformLatest(t *testing.T) { + logSpool, err := NewLogSpool(t.TempDir()) + if err != nil { + t.Fatalf("new log spool: %v", err) + } + if err := logSpool.Enqueue(validSpoolLogBatch(10, 10)); err != nil { + t.Fatalf("enqueue gapped batch: %v", err) + } + if err := logSpool.Enqueue(validSpoolLogBatch(11, 11)); err != nil { + t.Fatalf("enqueue later gapped batch: %v", err) + } + client := &recoveringSequenceGapLogBatchClient{latestSeq: 5} + flushed, err := logSpool.Flush(context.Background(), client) + if err != nil || flushed != 2 { + t.Fatalf("recover sequence gap: flushed=%d err=%v", flushed, err) + } + if len(client.progressBatches) != 1 || client.progressBatches[0].FirstSeq != 10 { + t.Fatalf("expected one progress recovery from gapped batch, got %+v", client.progressBatches) + } + pending, err := logSpool.Pending() + if err != nil || len(pending) != 0 { + t.Fatalf("expected gapped batches rejected, pending=%+v err=%v", pending, err) + } + checksum := func(entries []protocol.LogEntry) (string, error) { return fmt.Sprintf("sha256:%d", len(entries)), nil } + next := validSpoolLogBatch(0, 0) + sequence, appended, err := logSpool.EnqueueNextAggregated(context.Background(), next, nil, nil, checksum) + if err != nil || !appended || sequence != 6 { + t.Fatalf("expected next sequence to resume after platform latest, sequence=%d appended=%t err=%v", sequence, appended, err) + } +} + func TestLogSpoolRestoresPendingAllocationWithoutWatermark(t *testing.T) { root := t.TempDir() first, err := NewLogSpool(root) @@ -298,9 +329,23 @@ func (client *blockingLogBatchClient) IngestLogBatch(_ context.Context, batch pr type permanentRejectLogBatchClient struct{} func (permanentRejectLogBatchClient) IngestLogBatch(context.Context, protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) { + return protocol.LogBatchIngestResponse{}, PermanentLogBatchRejection("platform_not_found", nil) +} + +type recoveringSequenceGapLogBatchClient struct { + latestSeq uint64 + progressBatches []protocol.LogBatchIngestRequest +} + +func (client *recoveringSequenceGapLogBatchClient) IngestLogBatch(context.Context, protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error) { return protocol.LogBatchIngestResponse{}, PermanentLogBatchRejection("platform_sequence_gap", nil) } +func (client *recoveringSequenceGapLogBatchClient) LogStreamLatestSeq(_ context.Context, batch protocol.LogBatchIngestRequest) (uint64, error) { + client.progressBatches = append(client.progressBatches, batch) + return client.latestSeq, nil +} + type recordingLogBatchClient struct { batches []protocol.LogBatchIngestRequest }