feat(stovepipe): Mark the terminal outcome build

Summary:
**What**:
- Add Build.IsTerminalOutcome and persist it when buildsignal observes a terminal build status.
- Enforce at most one selected build per request with a generated nullable request key and unique MySQL index.

**Why**:
- Keep the selected terminal build as operational state on Build while preventing duplicate builds from both becoming selected.

Test Plan:
- [x] `go test ./stovepipe/extension/storage/mysql ./stovepipe/controller/buildsignal`

Revert Plan:
- Revert this PR to remove the marker and its unique index if the Build-based selection model is not adopted.

Jira Issues: None

Author: prathshenoy

stovepipe/controller/buildsignal/buildsignal.go

@@ -370,6 +370,7 @@ func (c *Controller) reconcile(ctx context.Context, store storage.Storage, build
 	newVersion := build.Version + 1
 	updated := build
 	updated.Status = status
+	updated.IsTerminalOutcome = status.IsTerminal()
 	if err := store.GetBuildStore().Update(ctx, updated, build.Version, newVersion); err != nil {
 		return "", fmt.Errorf("failed to persist status for build %s: %w", build.ID, err)
 	}

stovepipe/controller/buildsignal/buildsignal_test.go

@@ -326,7 +326,9 @@ func TestProcess(t *testing.T) {
 				m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil)
 				m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil)
 				m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil)
-				m.buildStore.EXPECT().Update(gomock.Any(), build(entity.BuildStatusSucceeded, 2), int32(2), int32(3)).Return(nil)
+				updated := build(entity.BuildStatusSucceeded, 2)
+				updated.IsTerminalOutcome = true
+				m.buildStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil)
 				logCall := expectFinish(m, entity.RequestStateSucceeded)
 				m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil).After(logCall)
 			},
@@ -338,7 +340,9 @@ func TestProcess(t *testing.T) {
 				m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil)
 				m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil)
 				m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusFailed, nil, nil)
-				m.buildStore.EXPECT().Update(gomock.Any(), build(entity.BuildStatusFailed, 2), int32(2), int32(3)).Return(nil)
+				updated := build(entity.BuildStatusFailed, 2)
+				updated.IsTerminalOutcome = true
+				m.buildStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil)
 				logCall := expectFinish(m, entity.RequestStateFailed)
 				m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil).After(logCall)
 			},
@@ -350,7 +354,9 @@ func TestProcess(t *testing.T) {
 				m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil)
 				m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil)
 				m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusCancelled, nil, nil)
-				m.buildStore.EXPECT().Update(gomock.Any(), build(entity.BuildStatusCancelled, 2), int32(2), int32(3)).Return(nil)
+				updated := build(entity.BuildStatusCancelled, 2)
+				updated.IsTerminalOutcome = true
+				m.buildStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil)
 				logCall := expectFinish(m, entity.RequestStateCancelled)
 				m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil).After(logCall)
 			},

stovepipe/entity/build.go

@@ -54,9 +54,9 @@ func (s BuildStatus) IsTerminal() bool {
 }
 
 // Build represents a single build triggered for a Request's commit. All
-// fields except Status and Version are immutable after creation — build is
-// the sole creator (via BuildStore.Create), and buildsignal is the sole
-// writer of Status/Version afterward.
+// fields except Status, IsTerminalOutcome, and Version are immutable after
+// creation — build is the sole creator (via BuildStore.Create), and
+// buildsignal is the sole writer afterward.
 type Build struct {
 	// ID is the build's own key: the runner-assigned id returned by
 	// Trigger (e.g. a Buildkite build number). Opaque; never parsed or
@@ -67,6 +67,9 @@ type Build struct {
 	RequestID string `json:"request_id"`
 	// Status is the build's lifecycle state.
 	Status BuildStatus `json:"status"`
+	// IsTerminalOutcome reports whether this build established its request's
+	// terminal state. At most one build for a request may be true.
+	IsTerminalOutcome bool `json:"is_terminal_outcome"`
 	// Version is used for optimistic locking. Versioning starts at 1 and
 	// is incremented for each change to the object.
 	Version int32 `json:"version"`

stovepipe/extension/storage/mysql/build_store.go

@@ -46,13 +46,15 @@ func (b *buildStore) Create(ctx context.Context, build entity.Build) (retErr err
 	defer func() { op.Complete(retErr) }()
 
 	_, err := b.db.ExecContext(ctx,
-		`INSERT INTO build (queue, id, request_id, status, version)
-		 VALUES (?, ?, ?, ?, ?)`,
+		`INSERT INTO build (queue, id, request_id, status, version, is_terminal_outcome, terminal_outcome_request_id)
+		 VALUES (?, ?, ?, ?, ?, ?, ?)`,
 		b.queue,
 		build.ID,
 		build.RequestID,
 		build.Status,
 		build.Version,
+		build.IsTerminalOutcome,
+		nil,
 	)
 	if err != nil {
 		if isDuplicateEntry(err) {
@@ -71,13 +73,14 @@ func (b *buildStore) Get(ctx context.Context, id string) (ret entity.Build, retE
 
 	var build entity.Build
 	err := b.db.QueryRowContext(ctx,
-		`SELECT id, request_id, status, version
+		`SELECT id, request_id, status, is_terminal_outcome, version
 		 FROM build WHERE queue = ? AND id = ?`,
 		b.queue, id,
 	).Scan(
 		&build.ID,
 		&build.RequestID,
 		&build.Status,
+		&build.IsTerminalOutcome,
 		&build.Version,
 	)
 
@@ -101,9 +104,11 @@ func (b *buildStore) Update(ctx context.Context, build entity.Build, oldVersion,
 
 	result, err := b.db.ExecContext(ctx,
 		`UPDATE build
-		 SET status = ?, version = ?
+		 SET status = ?, is_terminal_outcome = ?, terminal_outcome_request_id = ?, version = ?
 		 WHERE queue = ? AND id = ? AND version = ?`,
 		build.Status,
+		build.IsTerminalOutcome,
+		terminalOutcomeRequestID(build),
 		newVersion,
 		b.queue,
 		build.ID,
@@ -133,3 +138,10 @@ func (b *buildStore) Update(ctx context.Context, build entity.Build, oldVersion,
 
 	return nil
 }
+
+func terminalOutcomeRequestID(build entity.Build) any {
+	if build.IsTerminalOutcome {
+		return build.RequestID
+	}
+	return nil
+}

stovepipe/extension/storage/mysql/build_store_test.go

@@ -58,15 +58,15 @@ func TestBuildStore_Create(t *testing.T) {
 			name: "success",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO build").
-					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version).
+					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version, build.IsTerminalOutcome, nil).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
 		{
 			name: "duplicate id returns ErrAlreadyExists",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO build").
-					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version).
+					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version, build.IsTerminalOutcome, nil).
 					WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry})
 			},
 			wantErr:   true,
@@ -76,7 +76,7 @@ func TestBuildStore_Create(t *testing.T) {
 			name: "other exec error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO build").
-					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version).
+					WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version, build.IsTerminalOutcome, nil).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -124,9 +124,9 @@ func TestBuildStore_Get(t *testing.T) {
 			name: "found",
 			id:   want.ID,
 			setup: func(mock sqlmock.Sqlmock) {
-				rows := sqlmock.NewRows([]string{"id", "request_id", "status", "version"}).
-					AddRow(want.ID, want.RequestID, string(want.Status), want.Version)
-				mock.ExpectQuery("SELECT id, request_id, status, version").
+				rows := sqlmock.NewRows([]string{"id", "request_id", "status", "is_terminal_outcome", "version"}).
+					AddRow(want.ID, want.RequestID, string(want.Status), want.IsTerminalOutcome, want.Version)
+				mock.ExpectQuery("SELECT id, request_id, status, is_terminal_outcome, version").
 					WithArgs("monorepo/main", want.ID).
 					WillReturnRows(rows)
 			},
@@ -136,7 +136,7 @@ func TestBuildStore_Get(t *testing.T) {
 			name: "not found",
 			id:   "missing",
 			setup: func(mock sqlmock.Sqlmock) {
-				mock.ExpectQuery("SELECT id, request_id, status, version").
+				mock.ExpectQuery("SELECT id, request_id, status, is_terminal_outcome, version").
 					WithArgs("monorepo/main", "missing").
 					WillReturnError(sql.ErrNoRows)
 			},
@@ -147,7 +147,7 @@ func TestBuildStore_Get(t *testing.T) {
 			name: "query error",
 			id:   "bad",
 			setup: func(mock sqlmock.Sqlmock) {
-				mock.ExpectQuery("SELECT id, request_id, status, version").
+				mock.ExpectQuery("SELECT id, request_id, status, is_terminal_outcome, version").
 					WithArgs("monorepo/main", "bad").
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
@@ -191,15 +191,15 @@ func TestBuildStore_Update(t *testing.T) {
 			name: "success",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE build").
-					WithArgs(build.Status, newVersion, "monorepo/main", build.ID, oldVersion).
+					WithArgs(build.Status, build.IsTerminalOutcome, nil, newVersion, "monorepo/main", build.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
 		{
 			name: "version mismatch",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE build").
-					WithArgs(build.Status, newVersion, "monorepo/main", build.ID, oldVersion).
+					WithArgs(build.Status, build.IsTerminalOutcome, nil, newVersion, "monorepo/main", build.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 0))
 			},
 			wantErr:   true,
@@ -209,7 +209,7 @@ func TestBuildStore_Update(t *testing.T) {
 			name: "exec error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE build").
-					WithArgs(build.Status, newVersion, "monorepo/main", build.ID, oldVersion).
+					WithArgs(build.Status, build.IsTerminalOutcome, nil, newVersion, "monorepo/main", build.ID, oldVersion).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -218,7 +218,7 @@ func TestBuildStore_Update(t *testing.T) {
 			name: "rows affected error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE build").
-					WithArgs(build.Status, newVersion, "monorepo/main", build.ID, oldVersion).
+					WithArgs(build.Status, build.IsTerminalOutcome, nil, newVersion, "monorepo/main", build.ID, oldVersion).
 					WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error")))
 			},
 			wantErr: true,

stovepipe/extension/storage/mysql/schema/build.sql

@@ -6,5 +6,8 @@ CREATE TABLE IF NOT EXISTS build (
     request_id  VARCHAR(255) NOT NULL,
     status      VARCHAR(64)  NOT NULL,
     version     INT          NOT NULL,
-    PRIMARY KEY (queue, id)
+	 is_terminal_outcome BOOLEAN NOT NULL DEFAULT FALSE,
+	 terminal_outcome_request_id VARCHAR(255) NULL,
+	 PRIMARY KEY (queue, id),
+	 UNIQUE KEY terminal_outcome_request (queue, terminal_outcome_request_id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;