Skip to content

Commit da340ff

Browse files
albertywualbertywu
committed
fix(orchestrator): reconcile score redelivery
1 parent f601ca2 commit da340ff

8 files changed

Lines changed: 321 additions & 33 deletions

File tree

‎service/submitqueue/orchestrator/server/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ go_library(
2323
"//platform/extension/messagequeue/mysql:go_default_library",
2424
"//platform/http:go_default_library",
2525
"//submitqueue/core/changeset:go_default_library",
26+
"//submitqueue/core/errs:go_default_library",
2627
"//submitqueue/core/topickey:go_default_library",
2728
"//submitqueue/entity:go_default_library",
2829
"//submitqueue/extension/buildrunner:go_default_library",

‎service/submitqueue/orchestrator/server/main.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ import (
4343
queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
4444
"github.com/uber/submitqueue/platform/http"
4545
"github.com/uber/submitqueue/submitqueue/core/changeset"
46+
submitqueueerrs "github.com/uber/submitqueue/submitqueue/core/errs"
4647
"github.com/uber/submitqueue/submitqueue/core/topickey"
4748
"github.com/uber/submitqueue/submitqueue/entity"
4849
"github.com/uber/submitqueue/submitqueue/extension/buildrunner"
@@ -218,6 +219,7 @@ func run() error {
218219
// subscriptions are final destinations (there is no further DLQ).
219220
primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry,
220221
errs.NewClassifierProcessor(
222+
submitqueueerrs.Classifier,
221223
genericerrs.Classifier,
222224
// Storage (submitqueue/extension/storage/mysql) and queue (platform/extension/messagequeue/mysql)
223225
// both run on the same MySQL driver, so a single classifier covers

‎submitqueue/core/errs/BUILD.bazel‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
load("@rules_go//go:def.bzl", "go_library", "go_test")
2+
3+
go_library(
4+
name = "go_default_library",
5+
srcs = ["errs.go"],
6+
importpath = "github.com/uber/submitqueue/submitqueue/core/errs",
7+
visibility = ["//visibility:public"],
8+
deps = [
9+
"//platform/errs:go_default_library",
10+
"//submitqueue/extension/storage:go_default_library",
11+
],
12+
)
13+
14+
go_test(
15+
name = "go_default_test",
16+
srcs = ["errs_test.go"],
17+
embed = [":go_default_library"],
18+
deps = [
19+
"//platform/errs:go_default_library",
20+
"//submitqueue/extension/storage:go_default_library",
21+
"@com_github_stretchr_testify//assert:go_default_library",
22+
],
23+
)

‎submitqueue/core/errs/errs.go‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
// Copyright (c) 2025 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// Package errs classifies SubmitQueue domain errors for consumer retry policy.
16+
package errs
17+
18+
import (
19+
platformerrs "github.com/uber/submitqueue/platform/errs"
20+
"github.com/uber/submitqueue/submitqueue/extension/storage"
21+
)
22+
23+
// Classifier recognizes SubmitQueue domain sentinels that have a consistent
24+
// workflow-level retry policy.
25+
var Classifier platformerrs.Classifier = classifier{}
26+
27+
type classifier struct{}
28+
29+
// Classify inspects one error-chain node.
30+
func (classifier) Classify(err error) platformerrs.Verdict {
31+
if err == storage.ErrVersionMismatch {
32+
return platformerrs.InfraRetryable
33+
}
34+
return platformerrs.Unknown
35+
}

‎submitqueue/core/errs/errs_test.go‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
// Copyright (c) 2025 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package errs
16+
17+
import (
18+
"fmt"
19+
"testing"
20+
21+
"github.com/stretchr/testify/assert"
22+
platformerrs "github.com/uber/submitqueue/platform/errs"
23+
"github.com/uber/submitqueue/submitqueue/extension/storage"
24+
)
25+
26+
func TestClassifier(t *testing.T) {
27+
tests := []struct {
28+
name string
29+
err error
30+
want platformerrs.Verdict
31+
}{
32+
{
33+
name: "version mismatch is retryable",
34+
err: storage.ErrVersionMismatch,
35+
want: platformerrs.InfraRetryable,
36+
},
37+
{
38+
name: "other storage sentinel is unknown",
39+
err: storage.ErrNotFound,
40+
want: platformerrs.Unknown,
41+
},
42+
{
43+
name: "wrapped node is left to processor walk",
44+
err: fmt.Errorf("update: %w", storage.ErrVersionMismatch),
45+
want: platformerrs.Unknown,
46+
},
47+
}
48+
49+
for _, tt := range tests {
50+
t.Run(tt.name, func(t *testing.T) {
51+
assert.Equal(t, tt.want, Classifier.Classify(tt.err))
52+
})
53+
}
54+
}
55+
56+
func TestClassifierProcessorMarksWrappedVersionMismatchRetryable(t *testing.T) {
57+
raw := fmt.Errorf("update batch: %w", storage.ErrVersionMismatch)
58+
processed := platformerrs.NewClassifierProcessor(Classifier).Process(raw)
59+
60+
assert.True(t, platformerrs.IsRetryable(processed))
61+
assert.ErrorIs(t, processed, storage.ErrVersionMismatch)
62+
}

‎submitqueue/orchestrator/controller/score/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ go_test(
3232
"//submitqueue/core/topickey:go_default_library",
3333
"//submitqueue/entity:go_default_library",
3434
"//submitqueue/extension/scorer/mock:go_default_library",
35+
"//submitqueue/extension/storage:go_default_library",
3536
"//submitqueue/extension/storage/mock:go_default_library",
3637
"@com_github_stretchr_testify//assert:go_default_library",
3738
"@com_github_stretchr_testify//require:go_default_library",

‎submitqueue/orchestrator/controller/score/score.go‎

Lines changed: 52 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -104,10 +104,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
104104
"partition_key", msg.PartitionKey,
105105
)
106106

107-
// Short-circuit when the batch is in BatchStateCancelling — the cancel
108-
// controller has handed the batch off to speculate, which owns the terminal
109-
// write to Cancelled and the downstream dependent / conclude publishes. We
110-
// must not race it to conclude (conclude requires terminal). Silently ack.
107+
var batchScore float64
108+
// Short-circuit when the batch is in BatchStateCancelling. The cancel
109+
// controller has handed the batch off to speculate, which owns the
110+
// terminal write and downstream fanout.
111111
if batch.State == entity.BatchStateCancelling {
112112
c.metricsScope.Counter("skipped_cancelling").Inc(1)
113113
c.logger.Infow("skipping score for cancelling batch",
@@ -116,12 +116,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
116116
return nil
117117
}
118118

119-
// Short-circuit if the batch is already terminal. Score never writes a
120-
// terminal state, so it owns no recovery here: whichever controller wrote
121-
// the terminal state (speculate.cancelBatch / failOnDependency, or merge)
122-
// already published to conclude, and speculate's terminal self-heal
123-
// republishes conclude on every redelivery of a terminal batch. Silently
124-
// ack — same pattern as build / buildsignal on halted.
119+
// Score owns no terminal-state recovery. The controller that wrote the
120+
// terminal state owns its remaining fanout.
125121
if batch.State.IsTerminal() {
126122
c.metricsScope.Counter("skipped_terminal").Inc(1)
127123
c.logger.Infow("skipping score for terminal batch",
@@ -131,25 +127,54 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
131127
return nil
132128
}
133129

134-
// Score the batch. The scorer resolves the batch's changes itself.
135-
batchScore, err := c.scoreBatch(ctx, batch)
136-
if err != nil {
137-
metrics.NamedCounter(c.metricsScope, opName, "scorer_errors", 1)
138-
return fmt.Errorf("failed to score batch %s: %w", batch.ID, err)
139-
}
130+
switch batch.State {
131+
case entity.BatchStateCreated:
132+
// Score the batch. The scorer resolves the batch's changes itself.
133+
batchScore, err = c.scoreBatch(ctx, batch)
134+
if err != nil {
135+
metrics.NamedCounter(c.metricsScope, opName, "scorer_errors", 1)
136+
return fmt.Errorf("failed to score batch %s: %w", batch.ID, err)
137+
}
138+
139+
newVersion := batch.Version + 1
140+
err = c.store.GetBatchStore().UpdateScoreAndState(
141+
ctx,
142+
batch.ID,
143+
batch.Version,
144+
newVersion,
145+
batchScore,
146+
entity.BatchStateScored,
147+
)
148+
if err != nil {
149+
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
150+
return fmt.Errorf("failed to update score for batch %s: %w", batch.ID, err)
151+
}
152+
153+
batch.Version = newVersion
154+
batch.Score = batchScore
155+
batch.State = entity.BatchStateScored
156+
c.logger.Infow("scored batch",
157+
"batch_id", batch.ID,
158+
"score", batchScore,
159+
)
140160

141-
// Atomically update score and state to "scored" in the database
142-
newVersion := batch.Version + 1
143-
if err := c.store.GetBatchStore().UpdateScoreAndState(ctx, batch.ID, batch.Version, newVersion, batchScore, entity.BatchStateScored); err != nil {
144-
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
145-
return fmt.Errorf("failed to update score for batch %s: %w", batch.ID, err)
146-
}
147-
batch.Version = newVersion
161+
case entity.BatchStateScored:
162+
// The durable transition already happened, but its fanout may be
163+
// incomplete. Preserve the committed score and replay all outputs.
164+
batchScore = batch.Score
165+
166+
case entity.BatchStateSpeculating, entity.BatchStateMerging:
167+
// Normal under at-least-once delivery: a prior score attempt may have
168+
// published to speculate before its acknowledgement was recorded.
169+
// Downstream processing has already advanced the batch, so this stale
170+
// delivery is satisfied and must not regress the batch to Scored.
171+
c.metricsScope.Counter("skipped_downstream").Inc(1)
172+
return nil
148173

149-
c.logger.Infow("scored batch",
150-
"batch_id", batch.ID,
151-
"score", batchScore,
152-
)
174+
default:
175+
c.metricsScope.Counter("unexpected_state").Inc(1)
176+
return fmt.Errorf("unexpected batch state %q for batch %s", batch.State, batch.ID)
177+
}
153178

154179
// Publish request log entries for all requests in the batch
155180
if err := corerequest.PublishBatchLogs(ctx, c.registry, batch.Contains, entity.RequestStatusScored, map[string]string{

0 commit comments

Comments
 (0)