fix(orchestrator): carry a failed batch's reason to the request error log

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 {

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

submitqueue/entity/batch.go

@@ -170,6 +170,11 @@ type Batch struct {
 	// The state of the batch lifecycle this batch is in. Updateable field with Version for optimistic locking.
 	State BatchState
 
+	// FailureReason is a human-readable explanation of why the batch failed.
+	// Empty unless State is BatchStateFailed, and empty even then when no reason
+	// was recorded — it is never a fabricated placeholder.
+	FailureReason string
+
 	// Version is the version of the object. It is used for optimistic locking.
 	// Versioning starts at 1 and is incremented for each change to the object.
 	Version int32

submitqueue/extension/storage/mysql/batch_store.go

@@ -52,9 +52,9 @@ func (s *batchStore) Get(ctx context.Context, id string) (ret entity.Batch, retE
 	var dependenciesJSON []byte
 
 	err := s.db.QueryRowContext(ctx,
-		"SELECT id, queue, contains, dependencies, state, version FROM batch WHERE queue = ? AND id = ?",
+		"SELECT id, queue, contains, dependencies, state, failure_reason, version FROM batch WHERE queue = ? AND id = ?",
 		s.queue, id,
-	).Scan(&batch.ID, &batch.Queue, &containsJSON, &dependenciesJSON, &batch.State, &batch.Version)
+	).Scan(&batch.ID, &batch.Queue, &containsJSON, &dependenciesJSON, &batch.State, &batch.FailureReason, &batch.Version)
 
 	if errors.Is(err, sql.ErrNoRows) {
 		return entity.Batch{}, storage.WrapNotFound(err)
@@ -94,8 +94,8 @@ func (s *batchStore) Create(ctx context.Context, batch entity.Batch) (retErr err
 	}
 
 	_, err = s.db.ExecContext(ctx,
-		"INSERT INTO batch (id, queue, contains, dependencies, state, version) VALUES (?, ?, ?, ?, ?, ?)",
-		batch.ID, batch.Queue, containsJSON, dependenciesJSON, batch.State, batch.Version,
+		"INSERT INTO batch (id, queue, contains, dependencies, state, failure_reason, version) VALUES (?, ?, ?, ?, ?, ?, ?)",
+		batch.ID, batch.Queue, containsJSON, dependenciesJSON, batch.State, batch.FailureReason, batch.Version,
 	)
 	if err != nil {
 		var mysqlErr *mysql.MySQLError
@@ -130,8 +130,8 @@ func (s *batchStore) Update(ctx context.Context, batch entity.Batch, oldVersion,
 	}
 
 	result, err := s.db.ExecContext(ctx,
-		"UPDATE batch SET contains = ?, dependencies = ?, state = ?, version = ? WHERE queue = ? AND id = ? AND version = ?",
-		containsJSON, dependenciesJSON, batch.State, newVersion, batch.Queue, batch.ID, oldVersion,
+		"UPDATE batch SET contains = ?, dependencies = ?, state = ?, failure_reason = ?, version = ? WHERE queue = ? AND id = ? AND version = ?",
+		containsJSON, dependenciesJSON, batch.State, batch.FailureReason, newVersion, batch.Queue, batch.ID, oldVersion,
 	)
 	if err != nil {
 		return fmt.Errorf(

submitqueue/extension/storage/mysql/batch_store_test.go

@@ -43,12 +43,13 @@ func setupBatchStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, storage.BatchS
 
 func TestBatchStore_Get(t *testing.T) {
 	want := entity.Batch{
-		ID:           "monorepo/batch/1",
-		Queue:        "monorepo",
-		Contains:     []string{"monorepo/1", "monorepo/2"},
-		Dependencies: []string{"monorepo/batch/0"},
-		State:        entity.BatchStateCreated,
-		Version:      1,
+		ID:            "monorepo/batch/1",
+		Queue:         "monorepo",
+		Contains:      []string{"monorepo/1", "monorepo/2"},
+		Dependencies:  []string{"monorepo/batch/0"},
+		State:         entity.BatchStateFailed,
+		FailureReason: "merge failed: conflict in pkg/a/foo.go",
+		Version:       1,
 	}
 	containsJSON, err := json.Marshal(want.Contains)
 	require.NoError(t, err)
@@ -67,9 +68,9 @@ func TestBatchStore_Get(t *testing.T) {
 			name: "found",
 			id:   want.ID,
 			setup: func(mock sqlmock.Sqlmock) {
-				rows := sqlmock.NewRows([]string{"id", "queue", "contains", "dependencies", "state", "version"}).
-					AddRow(want.ID, want.Queue, containsJSON, dependenciesJSON, string(want.State), want.Version)
-				mock.ExpectQuery("SELECT id, queue, contains, dependencies, state, version FROM batch").
+				rows := sqlmock.NewRows([]string{"id", "queue", "contains", "dependencies", "state", "failure_reason", "version"}).
+					AddRow(want.ID, want.Queue, containsJSON, dependenciesJSON, string(want.State), want.FailureReason, want.Version)
+				mock.ExpectQuery("SELECT id, queue, contains, dependencies, state, failure_reason, version FROM batch").
 					WithArgs("monorepo", want.ID).
 					WillReturnRows(rows)
 			},
@@ -79,7 +80,7 @@ func TestBatchStore_Get(t *testing.T) {
 			name: "not found",
 			id:   "missing",
 			setup: func(mock sqlmock.Sqlmock) {
-				mock.ExpectQuery("SELECT id, queue, contains, dependencies, state, version FROM batch").
+				mock.ExpectQuery("SELECT id, queue, contains, dependencies, state, failure_reason, version FROM batch").
 					WithArgs("monorepo", "missing").
 					WillReturnError(sql.ErrNoRows)
 			},
@@ -90,7 +91,7 @@ func TestBatchStore_Get(t *testing.T) {
 			name: "query error",
 			id:   "bad",
 			setup: func(mock sqlmock.Sqlmock) {
-				mock.ExpectQuery("SELECT id, queue, contains, dependencies, state, version FROM batch").
+				mock.ExpectQuery("SELECT id, queue, contains, dependencies, state, failure_reason, version FROM batch").
 					WithArgs("monorepo", "bad").
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
@@ -100,9 +101,9 @@ func TestBatchStore_Get(t *testing.T) {
 			name: "malformed contains JSON",
 			id:   "malformed",
 			setup: func(mock sqlmock.Sqlmock) {
-				rows := sqlmock.NewRows([]string{"id", "queue", "contains", "dependencies", "state", "version"}).
-					AddRow(want.ID, want.Queue, []byte("not json"), dependenciesJSON, string(want.State), want.Version)
-				mock.ExpectQuery("SELECT id, queue, contains, dependencies, state, version FROM batch").
+				rows := sqlmock.NewRows([]string{"id", "queue", "contains", "dependencies", "state", "failure_reason", "version"}).
+					AddRow(want.ID, want.Queue, []byte("not json"), dependenciesJSON, string(want.State), want.FailureReason, want.Version)
+				mock.ExpectQuery("SELECT id, queue, contains, dependencies, state, failure_reason, version FROM batch").
 					WithArgs("monorepo", "malformed").
 					WillReturnRows(rows)
 			},
@@ -152,15 +153,15 @@ func TestBatchStore_Create(t *testing.T) {
 			name: "success",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO batch").
-					WithArgs(batch.ID, batch.Queue, sqlmock.AnyArg(), sqlmock.AnyArg(), batch.State, batch.Version).
+					WithArgs(batch.ID, batch.Queue, sqlmock.AnyArg(), sqlmock.AnyArg(), batch.State, batch.FailureReason, batch.Version).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
 		{
 			name: "duplicate id returns ErrAlreadyExists",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO batch").
-					WithArgs(batch.ID, batch.Queue, sqlmock.AnyArg(), sqlmock.AnyArg(), batch.State, batch.Version).
+					WithArgs(batch.ID, batch.Queue, sqlmock.AnyArg(), sqlmock.AnyArg(), batch.State, batch.FailureReason, batch.Version).
 					WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry})
 			},
 			wantErr:   true,
@@ -170,7 +171,7 @@ func TestBatchStore_Create(t *testing.T) {
 			name: "other exec error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO batch").
-					WithArgs(batch.ID, batch.Queue, sqlmock.AnyArg(), sqlmock.AnyArg(), batch.State, batch.Version).
+					WithArgs(batch.ID, batch.Queue, sqlmock.AnyArg(), sqlmock.AnyArg(), batch.State, batch.FailureReason, batch.Version).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -225,7 +226,7 @@ func TestBatchStore_Update(t *testing.T) {
 			batch: batch,
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE batch").
-					WithArgs(containsJSON, dependenciesJSON, batch.State, newVersion, batch.Queue, batch.ID, oldVersion).
+					WithArgs(containsJSON, dependenciesJSON, batch.State, batch.FailureReason, newVersion, batch.Queue, batch.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
@@ -234,7 +235,7 @@ func TestBatchStore_Update(t *testing.T) {
 			batch: batch,
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE batch").
-					WithArgs(containsJSON, dependenciesJSON, batch.State, newVersion, batch.Queue, batch.ID, oldVersion).
+					WithArgs(containsJSON, dependenciesJSON, batch.State, batch.FailureReason, newVersion, batch.Queue, batch.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 0))
 			},
 			wantErr:   true,
@@ -245,7 +246,7 @@ func TestBatchStore_Update(t *testing.T) {
 			batch: batch,
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE batch").
-					WithArgs(containsJSON, dependenciesJSON, batch.State, newVersion, batch.Queue, batch.ID, oldVersion).
+					WithArgs(containsJSON, dependenciesJSON, batch.State, batch.FailureReason, newVersion, batch.Queue, batch.ID, oldVersion).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -255,7 +256,7 @@ func TestBatchStore_Update(t *testing.T) {
 			batch: batch,
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE batch").
-					WithArgs(containsJSON, dependenciesJSON, batch.State, newVersion, batch.Queue, batch.ID, oldVersion).
+					WithArgs(containsJSON, dependenciesJSON, batch.State, batch.FailureReason, newVersion, batch.Queue, batch.ID, oldVersion).
 					WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error")))
 			},
 			wantErr: true,
@@ -270,7 +271,7 @@ func TestBatchStore_Update(t *testing.T) {
 			},
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE batch").
-					WithArgs([]byte("null"), []byte("null"), batch.State, newVersion, batch.Queue, batch.ID, oldVersion).
+					WithArgs([]byte("null"), []byte("null"), batch.State, batch.FailureReason, newVersion, batch.Queue, batch.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
@@ -286,7 +287,7 @@ func TestBatchStore_Update(t *testing.T) {
 			},
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE batch").
-					WithArgs([]byte("[]"), []byte("[]"), batch.State, newVersion, batch.Queue, batch.ID, oldVersion).
+					WithArgs([]byte("[]"), []byte("[]"), batch.State, batch.FailureReason, newVersion, batch.Queue, batch.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},

submitqueue/extension/storage/mysql/schema/batch.sql

@@ -4,6 +4,7 @@ CREATE TABLE IF NOT EXISTS batch (
     contains JSON NOT NULL,
     dependencies JSON NOT NULL,
     state VARCHAR(255) NOT NULL,
+    failure_reason TEXT NOT NULL,
     version INT NOT NULL,
     PRIMARY KEY (queue, id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

submitqueue/orchestrator/controller/conclude/conclude.go

@@ -124,7 +124,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	// retried (and eventually dead-lettered) rather than silently skipped. We
 	// translate the result into per-outcome logs and metrics.
 	for _, requestID := range batch.Contains {
-		res, err := corerequest.TerminateRequest(ctx, store, c.registry, requestID, requestState, "", map[string]string{
+		// FailureReason is empty unless the batch failed, so this carries the
+		// reason on the error path and nothing on the landed or cancelled ones.
+		res, err := corerequest.TerminateRequest(ctx, store, c.registry, requestID, requestState, batch.FailureReason, map[string]string{
 			"batch_id": batch.ID,
 		})
 		if err != nil {

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: a batch's recorded failure reason 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,
+		FailureReason: reason,
+		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, nil)
+	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)
 

submitqueue/orchestrator/controller/mergesignal/mergesignal.go

@@ -145,6 +145,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	} else {
 		metrics.NamedCounter(c.metricsScope, opName, "not_merged", 1)
 		newState = entity.BatchStateFailed
+		batch.FailureReason = result.Reason
+		if batch.FailureReason == "" {
+			batch.FailureReason = "merge failed"
+		}
 		c.logger.Warnw("batch merge failed",
 			"batch_id", batch.ID,
 			"reason", result.Reason,

submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go

@@ -218,7 +218,11 @@ func TestProcess_NotMergedMarksBatchFailed(t *testing.T) {
 		Version:      3,
 	}
 	batchStore.EXPECT().Get(gomock.Any(), testBatchID).Return(batch, nil)
-	batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateFailed), int32(3), int32(4)).Return(nil)
+	// The merge result's reason is recorded on the batch, so conclude can carry
+	// it onto the request's terminal log instead of an empty error.
+	failed := batchWithState(batch, entity.BatchStateFailed)
+	failed.FailureReason = "conflict in foo.go"
+	batchStore.EXPECT().Update(gomock.Any(), failed, int32(3), int32(4)).Return(nil)
 
 	store := storagemock.NewMockStorage(ctrl)
 	store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()

submitqueue/orchestrator/controller/speculate/finalize.go

@@ -305,6 +305,9 @@ func (c *Controller) applyOutcome(ctx context.Context, store storage.Storage, ba
 
 	case outcomeFail, outcomeCancel:
 		state, _ = decision.terminalState()
+		if decision == outcomeFail {
+			batch.FailureReason = "no speculation path could pass; every candidate build failed"
+		}
 
 	default:
 		// outcomeWait: nothing to enact. Listed explicitly so an unknown or