diff --git a/submitqueue/orchestrator/controller/cancel/cancel.go b/submitqueue/orchestrator/controller/cancel/cancel.go index 11ebbb064..330e6a068 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel.go +++ b/submitqueue/orchestrator/controller/cancel/cancel.go @@ -147,6 +147,12 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er "partition_key", msg.PartitionKey, ) + // The cancellation may be terminal even though its final log was never published. + if request.State == entity.RequestStateCancelled { + return corerequest.ResendTerminalRequestLog(ctx, c.registry, request) + } + + // Other terminal outcomes belong to their winning writer and are left untouched. 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..a995fcb08 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) diff --git a/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal.go b/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal.go index 97ff06fe0..267aa00ca 100644 --- a/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal.go +++ b/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal.go @@ -114,7 +114,11 @@ 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. + // The failed conflict check may be terminal even though its final log was never published. + if request.State == entity.RequestStateError { + return corerequest.ResendTerminalRequestLog(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 +169,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..9947651e2 100644 --- a/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal_test.go +++ b/submitqueue/orchestrator/controller/landconflictsignal/landconflictsignal_test.go @@ -146,7 +146,7 @@ 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).Return(request, nil).Times(2) // 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) @@ -197,10 +197,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).Times(2) 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..808073c35 100644 --- a/submitqueue/orchestrator/controller/validate/validate.go +++ b/submitqueue/orchestrator/controller/validate/validate.go @@ -133,10 +133,12 @@ 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. + // The rejection may be terminal even though its final log was never published. + if request.State == entity.RequestStateError { + return corerequest.ResendTerminalRequestLog(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..07f543b7e 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) 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..8b16368f2 --- /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" +) + +// ResendTerminalRequestLog resends the request's terminal status without another state write. +// The log identifies the resend; original diagnostics may be unavailable. +// For example, validation can store Error and crash before publishing its status log. +// Redelivery must resend the status without another request-state write. +func ResendTerminalRequestLog(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, "terminal status resent for consistency; original details may not have been retained", 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..72e976efc 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 = "terminal status resent for consistency; original details may not have been retained", 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.