diff --git a/submitqueue/client/view.go b/submitqueue/client/view.go index ee1b9384..949be748 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 a58d195a..9821a330 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 diff --git a/submitqueue/entity/batch.go b/submitqueue/entity/batch.go index f9b280cb..a423c729 100644 --- a/submitqueue/entity/batch.go +++ b/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 diff --git a/submitqueue/extension/storage/mysql/batch_store.go b/submitqueue/extension/storage/mysql/batch_store.go index 2ac21a71..25d8876c 100644 --- a/submitqueue/extension/storage/mysql/batch_store.go +++ b/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( diff --git a/submitqueue/extension/storage/mysql/batch_store_test.go b/submitqueue/extension/storage/mysql/batch_store_test.go index 7cd69aef..9e0bd3bd 100644 --- a/submitqueue/extension/storage/mysql/batch_store_test.go +++ b/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,7 +153,7 @@ 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)) }, }, @@ -160,7 +161,7 @@ func TestBatchStore_Create(t *testing.T) { 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)) }, }, diff --git a/submitqueue/extension/storage/mysql/schema/batch.sql b/submitqueue/extension/storage/mysql/schema/batch.sql index 9bf5d187..6a6ef36d 100644 --- a/submitqueue/extension/storage/mysql/schema/batch.sql +++ b/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; diff --git a/submitqueue/orchestrator/controller/conclude/conclude.go b/submitqueue/orchestrator/controller/conclude/conclude.go index 88e6bf2c..ab327718 100644 --- a/submitqueue/orchestrator/controller/conclude/conclude.go +++ b/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 { diff --git a/submitqueue/orchestrator/controller/conclude/conclude_test.go b/submitqueue/orchestrator/controller/conclude/conclude_test.go index 72661566..ce4aa02f 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: 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) diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go index 795c1d99..f433cede 100644 --- a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go +++ b/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, diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go index f4f4565f..6588cd1b 100644 --- a/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go +++ b/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() diff --git a/submitqueue/orchestrator/controller/speculate/finalize.go b/submitqueue/orchestrator/controller/speculate/finalize.go index a74f81f2..0024ba3f 100644 --- a/submitqueue/orchestrator/controller/speculate/finalize.go +++ b/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