diff --git a/submitqueue/orchestrator/controller/cancel/cancel.go b/submitqueue/orchestrator/controller/cancel/cancel.go index 11ebbb064..c867f575e 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel.go +++ b/submitqueue/orchestrator/controller/cancel/cancel.go @@ -147,6 +147,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er "partition_key", msg.PartitionKey, ) + if request.State == entity.RequestStateCancelled { + return corerequest.PublishTerminalRequestLog(ctx, c.registry, request) + } + if entity.IsRequestStateTerminal(request.State) { metrics.NamedCounter(c.metricsScope, opName, "already_terminal", 1, metrics.TagsFromContext(ctx)...) return nil diff --git a/submitqueue/orchestrator/controller/cancel/cancel_test.go b/submitqueue/orchestrator/controller/cancel/cancel_test.go index 0b944db64..f2f4cadac 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel_test.go +++ b/submitqueue/orchestrator/controller/cancel/cancel_test.go @@ -161,7 +161,7 @@ func TestProcess_AlreadyTerminal_NoOp(t *testing.T) { reqStore := storagemock.NewMockRequestStore(ctrl) reqStore.EXPECT().Get(gomock.Any(), "1").Return(entity.Request{ - ID: "1", Queue: "q", State: entity.RequestStateCancelled, Version: 5, + ID: "1", Queue: "q", State: entity.RequestStateLanded, Version: 5, }, nil) store := orchstoragemock.NewMockStorage(ctrl) @@ -194,10 +194,17 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) { ctrl := gomock.NewController(t) registry, pub := newRegistry(t, ctrl) + var terminalLogs []entity.RequestLog var publishedTopics []string pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( - func(_ context.Context, topic string, _ entityqueue.Message) error { + func(_ context.Context, topic string, msg entityqueue.Message) error { publishedTopics = append(publishedTopics, topic) + entry, err := sqmq.UnmarshalRequestLog(msg.Payload) + require.NoError(t, err) + terminalLogs = append(terminalLogs, entry) + if len(terminalLogs) == 1 { + return fmt.Errorf("publish unavailable") + } return nil }).AnyTimes() @@ -206,11 +213,17 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) { cancelling := entity.Request{ID: "1", Queue: "q", State: entity.RequestStateCancelling, Version: 3} // Two-step transition: mark Cancelling, then — after TerminateRequest re-reads // the request — the terminal Cancelled CAS. + var terminal entity.Request gomock.InOrder( reqStore.EXPECT().Get(gomock.Any(), "1").Return(started, nil), reqStore.EXPECT().Update(gomock.Any(), requestWithState(started, entity.RequestStateCancelling), int32(2), int32(3)).Return(nil), reqStore.EXPECT().Get(gomock.Any(), "1").Return(cancelling, nil), - reqStore.EXPECT().Update(gomock.Any(), requestWithState(cancelling, entity.RequestStateCancelled), int32(3), int32(4)).Return(nil), + reqStore.EXPECT().Update(gomock.Any(), requestWithState(cancelling, entity.RequestStateCancelled), int32(3), int32(4)).DoAndReturn(func(_ context.Context, snapshot entity.Request, _, newVersion int32) error { + terminal = snapshot + terminal.Version = newVersion + return nil + }), + reqStore.EXPECT().Get(gomock.Any(), "1").DoAndReturn(func(context.Context, string) (entity.Request, error) { return terminal, nil }), ) batchStore := storagemock.NewMockBatchStore(ctrl) @@ -223,10 +236,18 @@ func TestProcess_CancelsUnbatchedRequest(t *testing.T) { controller := newController(t, store, registry) err := controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "1", "user changed mind"), "1")) - require.NoError(t, err) + require.Error(t, err) + require.NoError(t, controller.Process(context.Background(), newDelivery(t, ctrl, cancelPayload(t, "1", "changed reason"), "1"))) + require.Len(t, terminalLogs, 2) + assert.Equal(t, terminalLogs[0].Status, terminalLogs[1].Status) + assert.Equal(t, terminalLogs[0].RequestVersion, terminalLogs[1].RequestVersion) + assert.Equal(t, "original terminal details were not retained", terminalLogs[1].LastError) + assert.Equal(t, "user changed mind", terminalLogs[0].Metadata["reason"]) + assert.Empty(t, terminalLogs[1].Metadata) + assert.Equal(t, int32(4), terminalLogs[1].RequestVersion) // Only the request log entry should have been published. - assert.Equal(t, []string{"log"}, publishedTopics) + assert.Equal(t, []string{"log", "log"}, publishedTopics) } // TestProcess_AlreadyCancelling_SkipsMarkCancelling exercises the idempotent diff --git a/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal.go b/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal.go index 97ff06fe0..5b1b6a447 100644 --- a/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal.go +++ b/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal.go @@ -114,7 +114,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er "partition_key", msg.PartitionKey, ) - // Short-circuit halted requests: the cancel path owns driving them terminal. + if request.State == entity.RequestStateError { + return corerequest.PublishTerminalRequestLog(ctx, c.registry, request) + } + if entity.IsRequestStateHalted(request.State) { metrics.NamedCounter(c.metricsScope, opName, "skipped_halted", 1, metrics.TagsFromContext(ctx)...) c.logger.Infow("skipping landconflict signal for halted request", @@ -165,39 +168,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return nil // Success - message will be acked } -// failRequest drives the request to terminal RequestStateError and records the -// conflict reason on the request log. A not-landable verdict is an expected -// terminal outcome of the check, so the request is concluded here directly. -// -// Idempotent under at-least-once delivery: a redelivery whose request is already -// in Error skips the state CAS but still publishes the log (so a prior attempt -// that flipped the state but failed before logging is repaired); a request that -// reached a different terminal state (e.g. a racing cancel) is left untouched. func (c *Controller) failRequest(ctx context.Context, store orchstorage.Storage, request entity.Request, reason string) error { - switch { - case request.State == entity.RequestStateError: - // Idempotent retry: a prior delivery already wrote Error. Fall through to - // the log publish. - case entity.IsRequestStateTerminal(request.State): - c.logger.Warnw("request already in different terminal state, skipping fail", - "request_id", request.ID, - "state", string(request.State), - ) - return nil - default: - newVersion := request.Version + 1 - request.State = entity.RequestStateError - if err := store.GetRequestStore().Update(ctx, request, request.Version, newVersion); err != nil { - return fmt.Errorf("failed to update request %s state to error: %w", request.ID, err) - } - request.Version = newVersion - } - - logEntry := entity.NewRequestStatusLog(request.Queue, request.ID, entity.RequestStatusError, request.Version, reason, nil) - if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID, ""); err != nil { - return fmt.Errorf("failed to publish request log for %s: %w", request.ID, err) - } - return nil + _, err := corerequest.TerminateRequest(ctx, store, c.registry, request.ID, entity.RequestStateError, reason, nil) + return err } // publishRequestID publishes a request ID to the given topic key, stamped with diff --git a/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal_test.go b/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal_test.go index ef2571383..3d546eb63 100644 --- a/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal_test.go +++ b/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal_test.go @@ -146,15 +146,18 @@ func TestProcess_NotLandableMarksRequestError(t *testing.T) { reqStore := storagemock.NewMockRequestStore(ctrl) request := entity.Request{ID: testRequestID, Queue: testQueue, State: entity.RequestStateStarted, Version: 1} - reqStore.EXPECT().Get(gomock.Any(), testRequestID).Return(request, nil) + reqStore.EXPECT().Get(gomock.Any(), testRequestID).DoAndReturn(func(context.Context, string) (entity.Request, error) { return request, nil }).Times(3) // The request is driven to terminal Error inline (version 1 -> 2). - reqStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(nil) + reqStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).DoAndReturn(func(_ context.Context, snapshot entity.Request, _, newVersion int32) error { + request = snapshot + request.Version = newVersion + return nil + }) store := orchstoragemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() - // One publish is expected — the terminal log entry to the log topic. A publish - // to the batch topic would be a bug (gomock fails on the unexpected 2nd call). + var terminalLogs []entity.RequestLog var gotTopic string var gotPayload []byte pub := queuemock.NewMockPublisher(ctrl) @@ -162,9 +165,15 @@ func TestProcess_NotLandableMarksRequestError(t *testing.T) { func(_ context.Context, topic string, msg entityqueue.Message) error { gotTopic = topic gotPayload = msg.Payload + entry, err := sqmq.UnmarshalRequestLog(msg.Payload) + require.NoError(t, err) + terminalLogs = append(terminalLogs, entry) + if len(terminalLogs) == 1 { + return fmt.Errorf("publish unavailable") + } return nil }, - ) + ).Times(2) q := queuemock.NewMockQueue(ctrl) q.EXPECT().Publisher().Return(pub).AnyTimes() registry, err := consumer.NewTopicRegistry( @@ -180,16 +189,21 @@ func TestProcess_NotLandableMarksRequestError(t *testing.T) { res := runwaymq.MergeResult{Id: testRequestID, Outcome: runwaypb.Outcome_FAILED, Reason: "conflict in foo.go"} msg := entityqueue.NewMessage(testRequestID, resultPayload(t, res), testQueue, nil) - // Not-landable is an expected terminal outcome, so Process acks (no error). + require.Error(t, controller.Process(context.Background(), newDelivery(ctrl, msg))) + res.Outcome, res.Reason = runwaypb.Outcome_SUCCEEDED, "different retry result" + msg = entityqueue.NewMessage(testRequestID, resultPayload(t, res), testQueue, nil) require.NoError(t, controller.Process(context.Background(), newDelivery(ctrl, msg))) + require.Len(t, terminalLogs, 2) + assert.Equal(t, terminalLogs[0].Status, terminalLogs[1].Status) + assert.Equal(t, terminalLogs[0].RequestVersion, terminalLogs[1].RequestVersion) + assert.Equal(t, "original terminal details were not retained", terminalLogs[1].LastError) - // The single publish is the terminal log entry carrying the conflict reason. assert.Equal(t, "log", gotTopic) logEntry, err := sqmq.UnmarshalRequestLog(gotPayload) require.NoError(t, err) assert.Equal(t, entity.RequestStatusError, logEntry.Status) assert.Equal(t, int32(2), logEntry.RequestVersion) - assert.Equal(t, "conflict in foo.go", logEntry.LastError) + assert.Equal(t, "conflict in foo.go", terminalLogs[0].LastError) } func TestFailRequest_UpdateFailureLeavesRequestUnchanged(t *testing.T) { @@ -197,10 +211,11 @@ func TestFailRequest_UpdateFailureLeavesRequestUnchanged(t *testing.T) { request := entity.Request{ID: testRequestID, Queue: testQueue, State: entity.RequestStateStarted, Version: 1} reqStore := storagemock.NewMockRequestStore(ctrl) + reqStore.EXPECT().Get(gomock.Any(), testRequestID).Return(request, nil) reqStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(fmt.Errorf("db down")) store := orchstoragemock.NewMockStorage(ctrl) - store.EXPECT().GetRequestStore().Return(reqStore) + store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, staticStorageFactory{store: store}, consumer.TopicRegistry{}, runwaymq.TopicKeyMergeConflictCheckSignal, "orchestrator-landconflictsignal") diff --git a/submitqueue/orchestrator/controller/validate/validate.go b/submitqueue/orchestrator/controller/validate/validate.go index 99d1ea363..ab4b4d676 100644 --- a/submitqueue/orchestrator/controller/validate/validate.go +++ b/submitqueue/orchestrator/controller/validate/validate.go @@ -133,10 +133,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er "partition_key", msg.PartitionKey, ) - // Short-circuit if the request has been halted — either it already reached a - // terminal state, or the cancel controller has recorded a cancellation intent - // (RequestStateCancelling). Without this guard we would still publish to batch - // and spawn a batch for a request that should never proceed. + if request.State == entity.RequestStateError { + return corerequest.PublishTerminalRequestLog(ctx, c.registry, request) + } + + // Cancellation intent halts forward progress even before the terminal outcome. if entity.IsRequestStateHalted(request.State) { coremetrics.NamedCounter(c.metricsScope, "process", "skipped_halted", 1, coremetrics.TagsFromContext(ctx)...) c.logger.Infow("skipping validate for halted request", diff --git a/submitqueue/orchestrator/controller/validate/validate_test.go b/submitqueue/orchestrator/controller/validate/validate_test.go index 4c2e96edb..4b238ac8f 100644 --- a/submitqueue/orchestrator/controller/validate/validate_test.go +++ b/submitqueue/orchestrator/controller/validate/validate_test.go @@ -614,7 +614,7 @@ func TestController_Process_TerminalShortCircuit(t *testing.T) { for _, state := range []entity.RequestState{ entity.RequestStateCancelled, entity.RequestStateLanded, - entity.RequestStateError, + entity.RequestStateCancelling, } { t.Run(string(state), func(t *testing.T) { ctrl := gomock.NewController(t) @@ -843,10 +843,18 @@ func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *te State: entity.RequestStateStarted, Version: 1, } - store, mockReqStore := newMockStorage(ctrl, request) + mockReqStore := storagemock.NewMockRequestStore(ctrl) + mockReqStore.EXPECT().Get(gomock.Any(), request.ID).DoAndReturn(func(context.Context, string) (entity.Request, error) { return request, nil }).Times(3) + store := orchstoragemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes() store.EXPECT().GetChangeStore().Return(newMockChangeStore(ctrl)).AnyTimes() - mockReqStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(nil) + mockReqStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).DoAndReturn(func(_ context.Context, snapshot entity.Request, _, newVersion int32) error { + request = snapshot + request.Version = newVersion + return nil + }) + var terminalLogs []entity.RequestLog mockPub := queuemock.NewMockPublisher(ctrl) // Fail only the terminal log. The "validating" entry this stage now emits on // the way in goes to the same topic, so failing every publish would abort @@ -855,7 +863,10 @@ func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *te func(_ context.Context, _ string, msg entityqueue.Message) error { entry, err := sqmq.UnmarshalRequestLog(msg.Payload) if err == nil && entry.Status == entity.RequestStatusError { - return fmt.Errorf("publish boom") + terminalLogs = append(terminalLogs, entry) + if len(terminalLogs) == 1 { + return fmt.Errorf("publish boom") + } } return nil }, @@ -871,7 +882,7 @@ func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *te require.NoError(t, err) cpFactory := changeprovidermock.NewMockFactory(ctrl) - cpFactory.EXPECT().For(gomock.Any()).Return(&mockChangeProvider{}, nil).AnyTimes() + cpFactory.EXPECT().For(gomock.Any()).Return(&mockChangeProvider{}, nil) mockValidator := validatormock.NewMockValidator(ctrl) mockValidator.EXPECT().Validate(gomock.Any(), gomock.Any()).Return(fmt.Errorf("%w: some validation error", validator.ErrRejected)) @@ -890,4 +901,11 @@ func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *te err = controller.Process(context.Background(), delivery) require.Error(t, err) assert.ErrorContains(t, err, "failed to terminate rejected request") + require.NoError(t, controller.Process(context.Background(), delivery)) + require.Len(t, terminalLogs, 2) + assert.Equal(t, terminalLogs[0].Status, terminalLogs[1].Status) + assert.Equal(t, terminalLogs[0].RequestVersion, terminalLogs[1].RequestVersion) + assert.Equal(t, "original terminal details were not retained", terminalLogs[1].LastError) + assert.Contains(t, terminalLogs[0].LastError, "some validation error") + assert.Equal(t, int32(2), terminalLogs[1].RequestVersion) } diff --git a/submitqueue/orchestrator/core/request/BUILD.bazel b/submitqueue/orchestrator/core/request/BUILD.bazel index ae3aa7e02..c05271af8 100644 --- a/submitqueue/orchestrator/core/request/BUILD.bazel +++ b/submitqueue/orchestrator/core/request/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "log.go", + "terminal_log.go", "terminate.go", ], importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/core/request", diff --git a/submitqueue/orchestrator/core/request/terminal_log.go b/submitqueue/orchestrator/core/request/terminal_log.go new file mode 100644 index 000000000..7c8ea2215 --- /dev/null +++ b/submitqueue/orchestrator/core/request/terminal_log.go @@ -0,0 +1,39 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package request + +import ( + "context" + "fmt" + + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/submitqueue/entity" +) + +const terminalDetailsNotRetained = "original terminal details were not retained" + +// PublishTerminalRequestLog repairs a terminal request's status without another state write. +// Original diagnostics are not retained; the repair explicitly reports their absence. +func PublishTerminalRequestLog(ctx context.Context, registry consumer.TopicRegistry, request entity.Request) error { + status, err := terminalStateToStatus(request.State) + if err != nil { + return err + } + logEntry := entity.NewRequestStatusLog(request.Queue, request.ID, status, request.Version, terminalDetailsNotRetained, nil) + if err := PublishLog(ctx, registry, logEntry, request.ID, ""); err != nil { + return fmt.Errorf("failed to publish request log for %s: %w", request.ID, err) + } + return nil +} diff --git a/submitqueue/orchestrator/core/request/terminate.go b/submitqueue/orchestrator/core/request/terminate.go index 73f559ef7..67c727ef4 100644 --- a/submitqueue/orchestrator/core/request/terminate.go +++ b/submitqueue/orchestrator/core/request/terminate.go @@ -86,7 +86,7 @@ type TerminationResult struct { // - the request is not found: nothing is done (TerminationOutcomeNotFound). // // The caller owns all logging and metrics — TerminationResult carries the state context needed to do so. -// lastError and metadata are attached to the published RequestLog for diagnosis. +// lastError and metadata describe the first transition; same-state repair reports unavailable original details. // // A storage.ErrVersionMismatch from the conditional write is returned as-is (it is // intrinsically retryable) so the caller's next attempt re-reads and re-evaluates. @@ -123,6 +123,7 @@ func TerminateRequest( // Idempotent retry: a prior attempt already wrote the terminal state. // Skip the CAS and fall through to re-publish the terminal log. outcome = TerminationOutcomeAlreadyInTargetState + lastError, metadata = terminalDetailsNotRetained, nil case entity.IsRequestStateTerminal(request.State): // Divergent terminal state — a concurrent path reached terminal first // and owns the terminal log entry for the state it actually wrote. diff --git a/submitqueue/orchestrator/core/request/terminate_test.go b/submitqueue/orchestrator/core/request/terminate_test.go index 87b483189..c71892661 100644 --- a/submitqueue/orchestrator/core/request/terminate_test.go +++ b/submitqueue/orchestrator/core/request/terminate_test.go @@ -125,6 +125,8 @@ func TestTerminateRequest(t *testing.T) { }, "already in target terminal state republishes log": { targetState: entity.RequestStateError, + lastError: "not the original reason", + metadata: map[string]string{"source": "retry"}, mockFunc: func(rs *storagemock.MockRequestStore) { already := entity.Request{ID: requestID, Queue: "q", State: entity.RequestStateError, Version: 5} rs.EXPECT().Get(gomock.Any(), requestID).Return(already, nil) @@ -137,6 +139,8 @@ func TestTerminateRequest(t *testing.T) { wantLog: func(t *testing.T, log entity.RequestLog) { assert.Equal(t, entity.RequestStatusError, log.Status) assert.Equal(t, int32(5), log.RequestVersion) + assert.Equal(t, "original terminal details were not retained", log.LastError) + assert.Empty(t, log.Metadata) }, }, "already in target terminal state returns republish error": {