From e200884c4adf054d07d743177cc95e46adfbd986 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Wed, 19 Aug 2026 17:45:35 -0700 Subject: [PATCH 1/2] fix(orchestrator): carry a failed batch's reason to the request error log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? When a batch failed, every request in it terminated as `error` with an empty message: mergesignal and speculate recorded only the terminal batch state, and conclude terminated the request with an empty reason — so a "did not land: error" carried no diagnostic at all. The reason is known at the failure site, but the requests are terminated later, in conclude. ### What? Carry the failure reason to conclude on the fan-out message's metadata rather than persisting it as batch state. mergesignal stamps the merge result's reason (or "merge failed" when it is empty) and speculate stamps "no speculation path could pass"; conclude reads it from the message and passes it to TerminateRequest, which already writes it to the request's terminal log. The landed and cancelled paths carry no reason, exactly as before. This mirrors the dead-letter path, which already sources its reason from the delivery context. Message metadata is durably persisted and redelivered by the queue backend, so the reason survives at-least-once retries — and the batch entity stays purely structural, with no new field and no schema migration. A `publish.MessageWithMetadata` helper and a shared `topickey.MetadataKeyFailureReason` constant carry the wiring. ## Test Plan ✅ `bazel test //submitqueue/orchestrator/...` (14/14). Updated tests assert the reason rides the conclude message metadata (mergesignal, speculate finalize) and reaches the request's terminal log (conclude). Broad build and CI gates (gazelle / tidy / mocks / fmt) clean. --- platform/publish/publish.go | 9 +++- submitqueue/core/topickey/topickey.go | 6 +++ .../controller/conclude/BUILD.bazel | 1 + .../controller/conclude/conclude.go | 6 ++- .../controller/conclude/conclude_test.go | 51 +++++++++++++++++++ .../controller/mergesignal/mergesignal.go | 33 +++++++++--- .../mergesignal/mergesignal_test.go | 28 ++++++++-- .../controller/speculate/finalize.go | 9 +++- .../controller/speculate/run_test.go | 5 ++ .../controller/speculate/speculate.go | 9 +++- 10 files changed, 141 insertions(+), 16 deletions(-) diff --git a/platform/publish/publish.go b/platform/publish/publish.go index 5ab6a90e2..4f280c315 100644 --- a/platform/publish/publish.go +++ b/platform/publish/publish.go @@ -47,6 +47,13 @@ import ( // which is what makes redelivery safe, while a new cause about the same entity // can never be swallowed by an older row. func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string) error { + return MessageWithMetadata(ctx, registry, key, msgID, payload, partitionKey, nil) +} + +// MessageWithMetadata is Message with side-band message metadata (headers/attributes) +// attached to the delivery. Use it to carry diagnostic context that is not part of +// the payload — the backend persists and redelivers metadata alongside the message. +func MessageWithMetadata(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string, metadata map[string]string) error { q, ok := registry.Queue(key) if !ok { return fmt.Errorf("no queue registered for topic key %s", key) @@ -56,7 +63,7 @@ func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer. return fmt.Errorf("no topic name registered for topic key %s", key) } - msg := entityqueue.NewMessage(msgID, payload, partitionKey, nil) + msg := entityqueue.NewMessage(msgID, payload, partitionKey, metadata) return q.Publisher().Publish(ctx, topicName, msg) } diff --git a/submitqueue/core/topickey/topickey.go b/submitqueue/core/topickey/topickey.go index 1950ed262..fa31076fa 100644 --- a/submitqueue/core/topickey/topickey.go +++ b/submitqueue/core/topickey/topickey.go @@ -51,3 +51,9 @@ const ( // TopicKeyLog is the pipeline stage where per-request logs are written. TopicKeyLog TopicKey = "log" ) + +// MetadataKeyFailureReason is the conclude message's metadata attribute carrying +// a failed batch's human-readable reason. Set by the failure sites (merge and +// speculate) on the conclude publish and read by conclude to stamp the request's +// terminal log; absent on the landed and cancelled paths. +const MetadataKeyFailureReason = "failure_reason" diff --git a/submitqueue/orchestrator/controller/conclude/BUILD.bazel b/submitqueue/orchestrator/controller/conclude/BUILD.bazel index 4cff6b197..91992843d 100644 --- a/submitqueue/orchestrator/controller/conclude/BUILD.bazel +++ b/submitqueue/orchestrator/controller/conclude/BUILD.bazel @@ -9,6 +9,7 @@ go_library( "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", "//submitqueue/core/request:go_default_library", + "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", diff --git a/submitqueue/orchestrator/controller/conclude/conclude.go b/submitqueue/orchestrator/controller/conclude/conclude.go index 88e6bf2ce..8f7c28d38 100644 --- a/submitqueue/orchestrator/controller/conclude/conclude.go +++ b/submitqueue/orchestrator/controller/conclude/conclude.go @@ -22,6 +22,7 @@ import ( "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" corerequest "github.com/uber/submitqueue/submitqueue/core/request" + "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" "go.uber.org/zap" @@ -123,8 +124,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er // the batch but missing from the store is a hard error — the whole batch is // retried (and eventually dead-lettered) rather than silently skipped. We // translate the result into per-outcome logs and metrics. + // The failure reason rides the conclude message, not the batch: the failing + // stage stamps it here and it is empty on the landed and cancelled paths. + failureReason := msg.Metadata[topickey.MetadataKeyFailureReason] for _, requestID := range batch.Contains { - res, err := corerequest.TerminateRequest(ctx, store, c.registry, requestID, requestState, "", map[string]string{ + res, err := corerequest.TerminateRequest(ctx, store, c.registry, requestID, requestState, failureReason, map[string]string{ "batch_id": batch.ID, }) if err != nil { diff --git a/submitqueue/orchestrator/controller/conclude/conclude_test.go b/submitqueue/orchestrator/controller/conclude/conclude_test.go index 72661566e..f39910da0 100644 --- a/submitqueue/orchestrator/controller/conclude/conclude_test.go +++ b/submitqueue/orchestrator/controller/conclude/conclude_test.go @@ -450,6 +450,57 @@ func TestController_Process(t *testing.T) { } } +// TestController_Process_FailedBatchCarriesReasonToRequestLog is the propagation +// this change exists for: the failure reason carried on the conclude message +// reaches the request's terminal error log, instead of the empty message the +// request used to carry. +func TestController_Process_FailedBatchCarriesReasonToRequestLog(t *testing.T) { + ctrl := gomock.NewController(t) + + const reason = "merge failed: conflict in pkg/a/foo.go" + batch := entity.Batch{ + ID: "test-queue/batch/9", + Queue: "test-queue", + Contains: []string{"test-queue/9"}, + State: entity.BatchStateFailed, + Version: 2, + } + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + + request := entity.Request{ID: "test-queue/9", Queue: "test-queue", Version: 1, State: entity.RequestStateProcessing} + requestStore := storagemock.NewMockRequestStore(ctrl) + requestStore.EXPECT().Get(gomock.Any(), "test-queue/9").Return(request, nil) + requestStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(nil) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + + controller, pub := newTestController(t, ctrl, store, false) + + var logged entity.RequestLog + pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + log, err := entity.RequestLogFromBytes(msg.Payload) + require.NoError(t, err) + logged = log + return nil + }, + ) + + msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, map[string]string{topickey.MetadataKeyFailureReason: reason}) + delivery := consumermock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(msg).AnyTimes() + delivery.EXPECT().Attempt().Return(1).AnyTimes() + + require.NoError(t, controller.Process(context.Background(), delivery)) + + assert.Equal(t, entity.RequestStatusError, logged.Status) + assert.Equal(t, reason, logged.LastError) +} + func TestController_Process_StorageFailure(t *testing.T) { ctrl := gomock.NewController(t) diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go index 795c1d994..7611dd837 100644 --- a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go +++ b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go @@ -122,6 +122,18 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return nil } + // A merge failure's reason travels to conclude on the fan-out message, not on + // the batch, so it reaches the request's terminal log without becoming durable + // batch state. Empty on the merged path. Computed before the idempotency check + // so a redelivered failed batch re-fans-out with its reason intact. + var failureReason string + if result.Outcome != runwaypb.Outcome_SUCCEEDED { + failureReason = result.Reason + if failureReason == "" { + failureReason = "merge failed" + } + } + // Idempotency: a previous delivery already transitioned this batch to a // terminal state. Repair the membership record (a prior attempt may have // CAS'd without completing the record move), re-fan-out in case that @@ -132,7 +144,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er metrics.NamedCounter(c.metricsScope, opName, "state_update_errors", 1) return err } - return c.fanout(ctx, batch.ID, batch.Queue) + return c.fanout(ctx, batch.ID, batch.Queue, failureReason) } var newState entity.BatchState @@ -157,7 +169,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return err } - return c.fanout(ctx, batch.ID, batch.Queue) + return c.fanout(ctx, batch.ID, batch.Queue, failureReason) } // fanout publishes the batch ID to conclude (so requests are updated) and to @@ -170,12 +182,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er // scoped the same way because speculate publishes there too when a batch goes // terminal on its own; the two mean the same thing but are decided at // different moments, so neither may swallow the other. -func (c *Controller) fanout(ctx context.Context, batchID, queue string) error { - if err := c.publish(ctx, topickey.TopicKeyConclude, publish.IntentID(batchID, "conclude", "merged"), batchID, queue); err != nil { +func (c *Controller) fanout(ctx context.Context, batchID, queue, failureReason string) error { + var concludeMeta map[string]string + if failureReason != "" { + concludeMeta = map[string]string{topickey.MetadataKeyFailureReason: failureReason} + } + if err := c.publish(ctx, topickey.TopicKeyConclude, publish.IntentID(batchID, "conclude", "merged"), batchID, queue, concludeMeta); err != nil { metrics.NamedCounter(c.metricsScope, "process", "publish_conclude_errors", 1) return fmt.Errorf("failed to publish to conclude: %w", err) } - if err := c.publish(ctx, topickey.TopicKeySpeculate, publish.IntentID(batchID, "merged"), batchID, queue); err != nil { + if err := c.publish(ctx, topickey.TopicKeySpeculate, publish.IntentID(batchID, "merged"), batchID, queue, nil); err != nil { metrics.NamedCounter(c.metricsScope, "process", "publish_speculate_errors", 1) return fmt.Errorf("failed to publish to speculate: %w", err) } @@ -183,14 +199,15 @@ func (c *Controller) fanout(ctx context.Context, batchID, queue string) error { } // publish publishes a batch ID to the given topic key under msgID, stamped -// with and partitioned by the batch's queue. -func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, msgID, batchID, queue string) error { +// with and partitioned by the batch's queue. metadata rides the message as +// side-band headers (nil for none). +func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, msgID, batchID, queue string, metadata map[string]string) error { payload, err := entity.BatchID{ID: batchID, Queue: queue}.ToBytes() if err != nil { return fmt.Errorf("failed to serialize batch ID: %w", err) } - if err := publish.Message(ctx, c.registry, key, msgID, payload, queue); err != nil { + if err := publish.MessageWithMetadata(ctx, c.registry, key, msgID, payload, queue, metadata); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go index f4f4565fd..73b78c3e3 100644 --- a/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go +++ b/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go @@ -218,21 +218,41 @@ func TestProcess_NotMergedMarksBatchFailed(t *testing.T) { Version: 3, } batchStore.EXPECT().Get(gomock.Any(), testBatchID).Return(batch, nil) + // The failed transition records only the terminal state; the reason travels + // to conclude on the message, not on the batch. batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateFailed), int32(3), int32(4)).Return(nil) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - var got []string - c := newController(t, store, recordingRegistry(t, ctrl, &got)) + byTopic := map[string]entityqueue.Message{} + pub := queuemock.NewMockPublisher(ctrl) + pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, topic string, msg entityqueue.Message) error { + byTopic[topic] = msg + return nil + }, + ).AnyTimes() + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(pub).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: q}, + {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: q}, + }) + require.NoError(t, err) res := runwaymq.MergeResult{Id: testBatchID, Outcome: runwaypb.Outcome_FAILED, Reason: "conflict in foo.go"} msg := entityqueue.NewMessage(testBatchID, resultPayload(t, res), testQueue, nil) // Not-merged is an expected terminal outcome, so Process acks (no error). - require.NoError(t, c.Process(context.Background(), newDelivery(ctrl, msg))) + require.NoError(t, newController(t, store, registry).Process(context.Background(), newDelivery(ctrl, msg))) - assert.ElementsMatch(t, []string{"conclude", "speculate"}, got) + require.Contains(t, byTopic, "conclude") + require.Contains(t, byTopic, "speculate") + // The merge reason rides the conclude message so conclude can stamp it on + // the request's terminal log; the speculate wake-up carries none. + assert.Equal(t, "conflict in foo.go", byTopic["conclude"].Metadata[topickey.MetadataKeyFailureReason]) + assert.Empty(t, byTopic["speculate"].Metadata[topickey.MetadataKeyFailureReason]) } func TestProcess_CancellingShortCircuit(t *testing.T) { diff --git a/submitqueue/orchestrator/controller/speculate/finalize.go b/submitqueue/orchestrator/controller/speculate/finalize.go index a74f81f20..8096e0863 100644 --- a/submitqueue/orchestrator/controller/speculate/finalize.go +++ b/submitqueue/orchestrator/controller/speculate/finalize.go @@ -346,12 +346,19 @@ func (c *Controller) applyOutcome(ctx context.Context, store storage.Storage, ba } case outcomeFail, outcomeCancel: + // A failed batch carries its reason to conclude on the message, so the + // requests' terminal log records why speculation could not land it. + // Cancellation carries none — the cancel path owns that reason. + var concludeMeta map[string]string + if decision == outcomeFail { + concludeMeta = map[string]string{topickey.MetadataKeyFailureReason: "no speculation path could pass; every candidate build failed"} + } // Named for the run that decided it, so a redelivery re-deriving the // same outcome does not conclude the batch twice, and so it stays // distinct from the conclude mergesignal sends for a merged batch. // A conclude that goes missing is recovered by fanout, which is // deliberately un-deduplicated. - if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, publish.IntentID(batch.ID, "conclude", "speculate"), batch.ID, batch.Queue, batch.Queue); err != nil { + if err := c.publishBatchIDWithMetadata(ctx, topickey.TopicKeyConclude, publish.IntentID(batch.ID, "conclude", "speculate"), batch.ID, batch.Queue, batch.Queue, concludeMeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return true, fmt.Errorf("failed to publish batch %s to conclude: %w", batch.ID, err) } diff --git a/submitqueue/orchestrator/controller/speculate/run_test.go b/submitqueue/orchestrator/controller/speculate/run_test.go index a2ad82983..306373640 100644 --- a/submitqueue/orchestrator/controller/speculate/run_test.go +++ b/submitqueue/orchestrator/controller/speculate/run_test.go @@ -784,6 +784,11 @@ func TestRun_FailedHeadConcludesAfterTheStateWrite(t *testing.T) { assert.Equal(t, []string{"conclude"}, h.published) assert.Empty(t, publishedBeforeWrite, "conclude rejects a non-terminal batch, so it must not be published before the write") + // The failure reason rides the conclude message so conclude can stamp it on + // the failed requests' terminal log, without persisting it as batch state. + require.Len(t, h.messages, 1) + assert.Equal(t, "no speculation path could pass; every candidate build failed", + h.messages[0].Metadata[topickey.MetadataKeyFailureReason]) } // A failure resolves a dependency, which can fail everything stacked on it. The diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go index 17f29f8c3..ec7c90425 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate.go +++ b/submitqueue/orchestrator/controller/speculate/speculate.go @@ -224,11 +224,18 @@ func (c *Controller) fanout(ctx context.Context, batchID, queue string) error { // want the queue, so a queue's batches are processed in order, but the build // dispatch partitions by batch so heads dispatch in parallel. func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, msgID, batchID, queue, partitionKey string) error { + return c.publishBatchIDWithMetadata(ctx, key, msgID, batchID, queue, partitionKey, nil) +} + +// publishBatchIDWithMetadata is publishBatchID with side-band message metadata +// attached to the delivery (nil for none). Used to carry a failed batch's reason +// to conclude without persisting it as batch state. +func (c *Controller) publishBatchIDWithMetadata(ctx context.Context, key consumer.TopicKey, msgID, batchID, queue, partitionKey string, metadata map[string]string) error { payload, err := entity.BatchID{ID: batchID, Queue: queue}.ToBytes() if err != nil { return fmt.Errorf("failed to serialize batch ID: %w", err) } - return publish.Message(ctx, c.registry, key, msgID, payload, partitionKey) + return publish.MessageWithMetadata(ctx, c.registry, key, msgID, payload, partitionKey, metadata) } // attributed records what a failure was about and counts it by subject type. From c9a98b54409067369f95f5508d5cb35896ffa799 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Tue, 18 Aug 2026 21:38:51 -0700 Subject: [PATCH 2/2] fix(client): show a failed request's reason in the run summary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The final "N request(s) did not land" line printed only id=status, so a failure read as "demo-queue/630=error" with no reason — even once the orchestrator began recording one on the request's terminal log. Append the request's last error to each failed entry when it has one, so a watched or scripted run reports why a change did not land, not only that it did not. --- submitqueue/client/view.go | 6 +++++- submitqueue/client/view_test.go | 6 ++++++ 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/submitqueue/client/view.go b/submitqueue/client/view.go index ee1b93843..949be748f 100644 --- a/submitqueue/client/view.go +++ b/submitqueue/client/view.go @@ -272,7 +272,11 @@ func summarize(rows []*Row) error { var failed []string for _, rw := range rows { if rw.Status != string(entity.RequestStatusLanded) { - failed = append(failed, fmt.Sprintf("%s=%s", rw.SQID, rw.Status)) + entry := fmt.Sprintf("%s=%s", rw.SQID, rw.Status) + if rw.Note != "" { + entry += ": " + rw.Note + } + failed = append(failed, entry) } } if len(failed) > 0 { diff --git a/submitqueue/client/view_test.go b/submitqueue/client/view_test.go index a58d195a9..9821a3308 100644 --- a/submitqueue/client/view_test.go +++ b/submitqueue/client/view_test.go @@ -504,6 +504,12 @@ func TestOutcome(t *testing.T) { func TestSummarize(t *testing.T) { assert.NoError(t, summarize([]*Row{{SQID: "q/1", Status: "landed"}})) assert.Error(t, summarize([]*Row{{SQID: "q/1", Status: "landed"}, {SQID: "q/2", Status: "error"}})) + + // A failure reason, when the request carries one, is part of the summary so a + // scripted run reports why rather than only that. + err := summarize([]*Row{{SQID: "q/2", Status: "error", Note: "merge failed: conflict in foo.go"}}) + require.Error(t, err) + assert.Contains(t, err.Error(), "q/2=error: merge failed: conflict in foo.go") } // TestRowLineAlignment is the column contract: on every row the stage begins at