Recover managed process log sessions

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()

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
 }