Skip to content

Commit 00c135f

Browse files
albertywualbertywu
committed
fix(orchestrator): reconcile score redelivery
1 parent 8e5b426 commit 00c135f

1 file changed

Lines changed: 83 additions & 45 deletions

File tree

  • submitqueue/orchestrator/controller/score

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

Lines changed: 83 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ package score
1616

1717
import (
1818
"context"
19+
"errors"
1920
"fmt"
2021

2122
"github.com/uber-go/tally"
@@ -104,53 +105,90 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
104105
"partition_key", msg.PartitionKey,
105106
)
106107

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.
111-
if batch.State == entity.BatchStateCancelling {
112-
c.metricsScope.Counter("skipped_cancelling").Inc(1)
113-
c.logger.Infow("skipping score for cancelling batch",
114-
"batch_id", batch.ID,
115-
)
116-
return nil
108+
var batchScore float64
109+
for {
110+
// Short-circuit when the batch is in BatchStateCancelling. The cancel
111+
// controller has handed the batch off to speculate, which owns the
112+
// terminal write and downstream fanout.
113+
if batch.State == entity.BatchStateCancelling {
114+
c.metricsScope.Counter("skipped_cancelling").Inc(1)
115+
c.logger.Infow("skipping score for cancelling batch",
116+
"batch_id", batch.ID,
117+
)
118+
return nil
119+
}
120+
121+
// Score owns no terminal-state recovery. The controller that wrote the
122+
// terminal state owns its remaining fanout.
123+
if batch.State.IsTerminal() {
124+
c.metricsScope.Counter("skipped_terminal").Inc(1)
125+
c.logger.Infow("skipping score for terminal batch",
126+
"batch_id", batch.ID,
127+
"state", string(batch.State),
128+
)
129+
return nil
130+
}
131+
132+
switch batch.State {
133+
case entity.BatchStateCreated:
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+
}
140+
141+
newVersion := batch.Version + 1
142+
err = c.store.GetBatchStore().UpdateScoreAndState(
143+
ctx,
144+
batch.ID,
145+
batch.Version,
146+
newVersion,
147+
batchScore,
148+
entity.BatchStateScored,
149+
)
150+
if errors.Is(err, storage.ErrVersionMismatch) {
151+
c.metricsScope.Counter("version_conflicts").Inc(1)
152+
batch, err = c.store.GetBatchStore().Get(ctx, batch.ID)
153+
if err != nil {
154+
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
155+
return fmt.Errorf("failed to reload batch %s after version conflict: %w", bid.ID, err)
156+
}
157+
continue
158+
}
159+
if err != nil {
160+
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
161+
return fmt.Errorf("failed to update score for batch %s: %w", batch.ID, err)
162+
}
163+
164+
batch.Version = newVersion
165+
batch.Score = batchScore
166+
batch.State = entity.BatchStateScored
167+
c.logger.Infow("scored batch",
168+
"batch_id", batch.ID,
169+
"score", batchScore,
170+
)
171+
172+
case entity.BatchStateScored:
173+
// The durable transition already happened, but its fanout may be
174+
// incomplete. Preserve the committed score and replay all outputs.
175+
batchScore = batch.Score
176+
c.metricsScope.Counter("replayed_scored").Inc(1)
177+
178+
case entity.BatchStateSpeculating, entity.BatchStateMerging:
179+
// A downstream controller consumed the handoff, so this score
180+
// delivery is stale and must not regress the batch to Scored.
181+
c.metricsScope.Counter("skipped_downstream").Inc(1)
182+
return nil
183+
184+
default:
185+
c.metricsScope.Counter("unexpected_state").Inc(1)
186+
return fmt.Errorf("unexpected batch state %q for batch %s", batch.State, batch.ID)
187+
}
188+
189+
break
117190
}
118191

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.
125-
if batch.State.IsTerminal() {
126-
c.metricsScope.Counter("skipped_terminal").Inc(1)
127-
c.logger.Infow("skipping score for terminal batch",
128-
"batch_id", batch.ID,
129-
"state", string(batch.State),
130-
)
131-
return nil
132-
}
133-
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-
}
140-
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
148-
149-
c.logger.Infow("scored batch",
150-
"batch_id", batch.ID,
151-
"score", batchScore,
152-
)
153-
154192
// Publish request log entries for all requests in the batch
155193
if err := corerequest.PublishBatchLogs(ctx, c.registry, batch.Contains, entity.RequestStatusScored, map[string]string{
156194
"batch_id": batch.ID,

0 commit comments

Comments
 (0)