fix(orchestrator): fix score reconciliation bugs

service/submitqueue/orchestrator/server/BUILD.bazel

@@ -23,6 +23,7 @@ go_library(
         "//platform/extension/messagequeue/mysql:go_default_library",
         "//platform/http:go_default_library",
         "//submitqueue/core/changeset:go_default_library",
+        "//submitqueue/core/errs:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/buildrunner:go_default_library",

service/submitqueue/orchestrator/server/main.go

@@ -43,6 +43,7 @@ import (
 	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
 	"github.com/uber/submitqueue/platform/http"
 	"github.com/uber/submitqueue/submitqueue/core/changeset"
+	submitqueueerrs "github.com/uber/submitqueue/submitqueue/core/errs"
 	"github.com/uber/submitqueue/submitqueue/core/topickey"
 	"github.com/uber/submitqueue/submitqueue/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/buildrunner"
@@ -218,6 +219,7 @@ func run() error {
 	// subscriptions are final destinations (there is no further DLQ).
 	primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry,
 		errs.NewClassifierProcessor(
+			submitqueueerrs.Classifier,
 			genericerrs.Classifier,
 			// Storage (submitqueue/extension/storage/mysql) and queue (platform/extension/messagequeue/mysql)
 			// both run on the same MySQL driver, so a single classifier covers

submitqueue/core/errs/BUILD.bazel

@@ -0,0 +1,23 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["errs.go"],
+    importpath = "github.com/uber/submitqueue/submitqueue/core/errs",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//platform/errs:go_default_library",
+        "//submitqueue/extension/storage:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["errs_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/errs:go_default_library",
+        "//submitqueue/extension/storage:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+    ],
+)

submitqueue/core/errs/errs.go

@@ -0,0 +1,35 @@
+// 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 errs classifies SubmitQueue domain errors for consumer retry policy.
+package errs
+
+import (
+	platformerrs "github.com/uber/submitqueue/platform/errs"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
+)
+
+// Classifier recognizes SubmitQueue domain sentinels that have a consistent
+// workflow-level retry policy.
+var Classifier platformerrs.Classifier = classifier{}
+
+type classifier struct{}
+
+// Classify inspects one error-chain node.
+func (classifier) Classify(err error) platformerrs.Verdict {
+	if err == storage.ErrVersionMismatch {
+		return platformerrs.InfraRetryable
+	}
+	return platformerrs.Unknown
+}

submitqueue/core/errs/errs_test.go

@@ -0,0 +1,62 @@
+// 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 errs
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	platformerrs "github.com/uber/submitqueue/platform/errs"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
+)
+
+func TestClassifier(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want platformerrs.Verdict
+	}{
+		{
+			name: "version mismatch is retryable",
+			err:  storage.ErrVersionMismatch,
+			want: platformerrs.InfraRetryable,
+		},
+		{
+			name: "other storage sentinel is unknown",
+			err:  storage.ErrNotFound,
+			want: platformerrs.Unknown,
+		},
+		{
+			name: "wrapped node is left to processor walk",
+			err:  fmt.Errorf("update: %w", storage.ErrVersionMismatch),
+			want: platformerrs.Unknown,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, Classifier.Classify(tt.err))
+		})
+	}
+}
+
+func TestClassifierProcessorMarksWrappedVersionMismatchRetryable(t *testing.T) {
+	raw := fmt.Errorf("update batch: %w", storage.ErrVersionMismatch)
+	processed := platformerrs.NewClassifierProcessor(Classifier).Process(raw)
+
+	assert.True(t, platformerrs.IsRetryable(processed))
+	assert.ErrorIs(t, processed, storage.ErrVersionMismatch)
+}

submitqueue/orchestrator/controller/score/BUILD.bazel

@@ -32,6 +32,7 @@ go_test(
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/scorer/mock:go_default_library",
+        "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/extension/storage/mock:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_stretchr_testify//require:go_default_library",

submitqueue/orchestrator/controller/score/score.go

@@ -104,10 +104,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		"partition_key", msg.PartitionKey,
 	)
 
-	// Short-circuit when the batch is in BatchStateCancelling — the cancel
-	// controller has handed the batch off to speculate, which owns the terminal
-	// write to Cancelled and the downstream dependent / conclude publishes. We
-	// must not race it to conclude (conclude requires terminal). Silently ack.
+	var batchScore float64
+	// Short-circuit when the batch is in BatchStateCancelling. The cancel
+	// controller has handed the batch off to speculate, which owns the
+	// terminal write and downstream fanout.
 	if batch.State == entity.BatchStateCancelling {
 		c.metricsScope.Counter("skipped_cancelling").Inc(1)
 		c.logger.Infow("skipping score for cancelling batch",
@@ -116,12 +116,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		return nil
 	}
 
-	// Short-circuit if the batch is already terminal. Score never writes a
-	// terminal state, so it owns no recovery here: whichever controller wrote
-	// the terminal state (speculate.cancelBatch / failOnDependency, or merge)
-	// already published to conclude, and speculate's terminal self-heal
-	// republishes conclude on every redelivery of a terminal batch. Silently
-	// ack — same pattern as build / buildsignal on halted.
+	// Score owns no terminal-state recovery. The controller that wrote the
+	// terminal state owns its remaining fanout.
 	if batch.State.IsTerminal() {
 		c.metricsScope.Counter("skipped_terminal").Inc(1)
 		c.logger.Infow("skipping score for terminal batch",
@@ -131,25 +127,54 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		return nil
 	}
 
-	// Score the batch. The scorer resolves the batch's changes itself.
-	batchScore, err := c.scoreBatch(ctx, batch)
-	if err != nil {
-		metrics.NamedCounter(c.metricsScope, opName, "scorer_errors", 1)
-		return fmt.Errorf("failed to score batch %s: %w", batch.ID, err)
-	}
+	switch batch.State {
+	case entity.BatchStateCreated:
+		// Score the batch. The scorer resolves the batch's changes itself.
+		batchScore, err = c.scoreBatch(ctx, batch)
+		if err != nil {
+			metrics.NamedCounter(c.metricsScope, opName, "scorer_errors", 1)
+			return fmt.Errorf("failed to score batch %s: %w", batch.ID, err)
+		}
+
+		newVersion := batch.Version + 1
+		err = c.store.GetBatchStore().UpdateScoreAndState(
+			ctx,
+			batch.ID,
+			batch.Version,
+			newVersion,
+			batchScore,
+			entity.BatchStateScored,
+		)
+		if err != nil {
+			metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
+			return fmt.Errorf("failed to update score for batch %s: %w", batch.ID, err)
+		}
+
+		batch.Version = newVersion
+		batch.Score = batchScore
+		batch.State = entity.BatchStateScored
+		c.logger.Infow("scored batch",
+			"batch_id", batch.ID,
+			"score", batchScore,
+		)
 
-	// Atomically update score and state to "scored" in the database
-	newVersion := batch.Version + 1
-	if err := c.store.GetBatchStore().UpdateScoreAndState(ctx, batch.ID, batch.Version, newVersion, batchScore, entity.BatchStateScored); err != nil {
-		metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
-		return fmt.Errorf("failed to update score for batch %s: %w", batch.ID, err)
-	}
-	batch.Version = newVersion
+	case entity.BatchStateScored:
+		// The durable transition already happened, but its fanout may be
+		// incomplete. Preserve the committed score and replay all outputs.
+		batchScore = batch.Score
+
+	case entity.BatchStateSpeculating, entity.BatchStateMerging:
+		// Normal under at-least-once delivery: a prior score attempt may have
+		// published to speculate before its acknowledgement was recorded.
+		// Downstream processing has already advanced the batch, so this stale
+		// delivery is satisfied and must not regress the batch to Scored.
+		c.metricsScope.Counter("skipped_downstream").Inc(1)
+		return nil
 
-	c.logger.Infow("scored batch",
-		"batch_id", batch.ID,
-		"score", batchScore,
-	)
+	default:
+		c.metricsScope.Counter("unexpected_state").Inc(1)
+		return fmt.Errorf("unexpected batch state %q for batch %s", batch.State, batch.ID)
+	}
 
 	// Publish request log entries for all requests in the batch
 	if err := corerequest.PublishBatchLogs(ctx, c.registry, batch.Contains, entity.RequestStatusScored, map[string]string{