fix: drain legacy process log batches

This commit is contained in:
npc0-hue
2026-08-21 23:27:52 +08:00
parent 0a605495af
commit 0cde7cdc66
3 changed files with 31 additions and 2 deletions
+24
View File
@@ -126,6 +126,30 @@ func TestCoreServicePersistsAndEnforcesImmutableProcessLogSessionMetadata(t *tes
}
}
func TestCoreServiceAcceptsLegacyProcessStreamStartWithoutSessionID(t *testing.T) {
svc, sessionToken := newRegisteredLogIngestService(t)
startedAt := time.Date(2026, 7, 3, 12, 30, 0, 0, time.UTC)
streamID := "run.run-local.server-1.stdout"
if _, err := svc.CreateLogStream(domain.LogStream{
ID: streamID,
ServerInstanceID: "server-1",
Source: domain.LogStreamSourceProcess,
StreamKey: "stdout",
SessionStartedAt: startedAt,
StorageBackend: domain.LogStorageBackendLocalSegments,
RetentionPolicy: "default",
}); err != nil {
t.Fatalf("create legacy process stream: %v", err)
}
batch := validLogBatch(t, sessionToken, 1, 1)
batch.LogStreamID = streamID
batch.SessionStartedAt = startedAt
if _, err := svc.IngestLogBatch(batch); err != nil {
t.Fatalf("ingest legacy process batch: %v", err)
}
}
func TestCoreServiceSerializesConsistentSessionStreamCreation(t *testing.T) {
svc, _ := newRegisteredLogIngestService(t)
startedAt := time.Date(2026, 7, 3, 12, 30, 0, 0, time.UTC)
+4 -1
View File
@@ -29,7 +29,10 @@ func ValidateLogBatchIngest(batch domain.LogBatchIngest) error {
}
hasSessionID := strings.TrimSpace(batch.LogSessionID) != ""
hasSessionStart := !batch.SessionStartedAt.IsZero()
if hasSessionID != hasSessionStart {
// Older process streams persisted their start time before Run assigned a
// generation id. Keep those batches ingestible so a durable Run spool can
// drain without blocking newer generation-scoped console output.
if hasSessionID && !hasSessionStart {
violations = append(violations, "logSessionId and sessionStartedAt must be provided together")
}
if (hasSessionID || hasSessionStart) && batch.Source != domain.LogStreamSourceProcess {
+3 -1
View File
@@ -1651,7 +1651,9 @@ func ValidateLogStream(stream domain.LogStream) error {
}
hasSessionID := strings.TrimSpace(stream.LogSessionID) != ""
hasSessionStart := !stream.SessionStartedAt.IsZero()
if hasSessionID != hasSessionStart {
// Keep legacy process streams with only a persisted start time valid while
// requiring the complete pair for new generation-scoped streams.
if hasSessionID && !hasSessionStart {
violations = append(violations, "logSessionId and sessionStartedAt must be provided together")
}
if (hasSessionID || hasSessionStart) && stream.Source != domain.LogStreamSourceProcess {