uber · behinddwalls · Aug 20, 2026 · Aug 19, 2026
@@ -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)
 }
 

@@ -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 {

@@ -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

@@ -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"
@@ -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",

@@ -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 {

@@ -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)
 

@@ -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,27 +182,32 @@ 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)
 	}
 	return nil
 }
 
 // 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)
 	}
 

@@ -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) {

@@ -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)
 		}

@@ -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

@@ -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.