Quarantine missing platform log batches
This commit is contained in:
@@ -1227,6 +1227,10 @@ func (client sessionLogBatchClient) IngestLogBatch(ctx context.Context, batch pr
|
||||
batch.Checksum = checksum
|
||||
}
|
||||
response, err := client.client.IngestLogBatch(ctx, batch)
|
||||
if err != nil && logBatchNotFoundError(err) {
|
||||
log.Printf("RUN phase=durable_uploaders.logs status=spool_quarantine stream=%s firstSeq=%d lastSeq=%d reason=platform_not_found error=%s", safeOptional(batch.LogStreamID), batch.FirstSeq, batch.LastSeq, RedactText(err.Error()))
|
||||
return protocol.LogBatchIngestResponse{}, spool.PermanentLogBatchRejection("platform_not_found", err)
|
||||
}
|
||||
if err != nil && (logBatchSequenceGapError(err) || logBatchAcknowledgedRangeConflict(err)) {
|
||||
reason := "platform_sequence_gap"
|
||||
if logBatchAcknowledgedRangeConflict(err) {
|
||||
@@ -1246,6 +1250,11 @@ func (client sessionLogBatchClient) IngestLogBatch(ctx context.Context, batch pr
|
||||
return response, err
|
||||
}
|
||||
|
||||
func logBatchNotFoundError(err error) bool {
|
||||
var httpErr interface{ HTTPStatus() int }
|
||||
return errors.As(err, &httpErr) && httpErr.HTTPStatus() == http.StatusNotFound
|
||||
}
|
||||
|
||||
func logBatchSequenceGapError(err error) bool {
|
||||
var gapError interface{ LogBatchSequenceGap() bool }
|
||||
return errors.As(err, &gapError) && gapError.LogBatchSequenceGap()
|
||||
|
||||
@@ -640,6 +640,16 @@ func TestSessionLogBatchClientQuarantinesSequenceConflict(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionLogBatchClientQuarantinesPlatformNotFound(t *testing.T) {
|
||||
recorder := &recordingDurableLogClient{err: api.PlatformRequestError{Status: http.StatusNotFound, Code: "not_found"}}
|
||||
client := sessionLogBatchClient{client: recorder, runEndpointID: "run-current", sessionToken: "token-current"}
|
||||
_, err := client.IngestLogBatch(context.Background(), protocol.LogBatchIngestRequest{LogStreamID: "run.stale-endpoint.stale-server.session.stdout", FirstSeq: 1, LastSeq: 1, Entries: []protocol.LogEntry{{Seq: 1, Timestamp: workerTestTime(), Line: "line"}}})
|
||||
var permanent spool.PermanentLogBatchError
|
||||
if !errors.As(err, &permanent) || permanent.Reason != "platform_not_found" {
|
||||
t.Fatalf("expected permanent platform_not_found rejection, got %#v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionLogBatchClientQuarantinesSessionMetadataMismatch(t *testing.T) {
|
||||
recorder := &recordingDurableLogClient{err: api.PlatformRequestError{Status: http.StatusBadRequest, Code: "validation_failed", Details: []string{"log session metadata must match stream"}}}
|
||||
client := sessionLogBatchClient{client: recorder, runEndpointID: "run-current", sessionToken: "token-current"}
|
||||
|
||||
Reference in New Issue
Block a user