fix(stovepipe): Retain request build associations

stovepipe/controller/build/build.go

@@ -20,7 +20,9 @@ package build
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"math"
 
 	"github.com/uber-go/tally"
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
@@ -134,43 +136,79 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		baseURI = request.BaseURI
 	}
 
-	buildID, err := buildRunner.Trigger(ctx, baseURI, request.URI, nil)
-	if err != nil {
-		return fmt.Errorf("failed to trigger build for request %s: %w", request.ID, err)
-	}
-
-	build := entity.Build{
-		ID:        buildID.ID,
-		RequestID: request.ID,
-		Status:    entity.BuildStatusAccepted,
-		Version:   1,
-	}
-	if err := store.GetBuildStore().Create(ctx, build); err != nil {
-		return fmt.Errorf("failed to persist build %s: %w", build.ID, err)
+	if request.BuildID == "" {
+		buildID, err := buildRunner.Trigger(ctx, baseURI, request.URI, nil)
+		if err != nil {
+			return fmt.Errorf("failed to trigger build for request %s: %w", request.ID, err)
+		}
+		build := entity.Build{ID: buildID.ID, RequestID: request.ID, Status: entity.BuildStatusAccepted, Version: 1}
+		if err := store.GetBuildStore().Create(ctx, build); errors.Is(err, storage.ErrAlreadyExists) {
+			retained, err := loader.ByID(ctx, build.ID, store.GetBuildStore().Get, "existing build")
+			if err != nil {
+				return err
+			}
+			if retained.RequestID != request.ID {
+				return fmt.Errorf("build %s belongs to another request", build.ID)
+			}
+		} else if err != nil {
+			return fmt.Errorf("failed to persist build %s: %w", build.ID, err)
+		}
+		request, err = c.associateBuild(ctx, store, request, build.ID)
+		if err != nil {
+			return err
+		}
 	}
-	if err := c.persistBuildTriggeredLog(ctx, store, request, build.ID); err != nil {
+	if err := c.persistBuildTriggeredLog(ctx, store, request, request.BuildID); err != nil {
 		return err
 	}
 
-	signalErr := c.publishBuildSignal(ctx, build.ID, request.Queue)
+	signalErr := c.publishBuildSignal(ctx, request.BuildID, request.Queue)
 	if signalErr != nil {
-		signalErr = fmt.Errorf("failed to publish build signal for %s: %w", build.ID, signalErr)
+		signalErr = fmt.Errorf("failed to publish build signal for %s: %w", request.BuildID, signalErr)
 	}
 
-	analysisErr := c.publishBuildAnalysis(ctx, build.ID, request.Queue)
+	analysisErr := c.publishBuildAnalysis(ctx, request.BuildID, request.Queue)
 	if err := errs.Group(signalErr, analysisErr); err != nil {
 		return err
 	}
 
 	c.logger.Debugw("triggered build",
 		"request_id", request.ID,
-		"build_id", build.ID,
+		"build_id", request.BuildID,
 		"queue", request.Queue,
 		"base_uri", baseURI,
 	)
 	return nil
 }
 
+func (c *Controller) associateBuild(ctx context.Context, store storage.Storage, request entity.Request, buildID string) (entity.Request, error) {
+	for request.BuildID == "" {
+		if err := ctx.Err(); err != nil {
+			return entity.Request{}, err
+		}
+		if request.Version == math.MaxInt32 {
+			return entity.Request{}, fmt.Errorf("request version exhausted")
+		}
+		next := request
+		next.BuildID = buildID
+		newVersion := request.Version + 1
+		err := store.GetRequestStore().Update(ctx, next, request.Version, newVersion)
+		if errors.Is(err, storage.ErrVersionMismatch) {
+			request, err = c.loadRequest(ctx, store, request.ID)
+			if err != nil {
+				return entity.Request{}, err
+			}
+			continue
+		}
+		if err != nil {
+			return entity.Request{}, fmt.Errorf("associate build with request %s: %w", request.ID, err)
+		}
+		next.Version = newVersion
+		return next, nil
+	}
+	return request, nil
+}
+
 func (c *Controller) persistBuildTriggeredLog(ctx context.Context, store storage.Storage, request entity.Request, buildID string) error {
 	log := requestlog.NewRequestEventLog(
 		request,

stovepipe/controller/build/build_analysis_test.go

@@ -45,6 +45,12 @@ func TestBuildAnalysisPipeline(t *testing.T) {
 	var savedBuild entity.Build
 	request := processingRequest(entity.BuildStrategyFull, "")
 	queue := entity.Queue{Name: testQueue, InFlightCount: 1, Version: 1}
+	m.association.DoAndReturn(func(_ context.Context, value entity.Request, old, next int32) error {
+		require.Equal(t, request.Version, old)
+		request = value
+		request.Version = next
+		return nil
+	})
 
 	queueStore := storagemock.NewMockQueueStore(ctrl)
 	issueStore := storagemock.NewMockBuildIssueStore(ctrl)
@@ -76,7 +82,7 @@ func TestBuildAnalysisPipeline(t *testing.T) {
 		request = value
 		request.Version = next
 		return nil
-	})
+	}).AnyTimes()
 
 	queueStore.EXPECT().Get(gomock.Any(), testQueue).DoAndReturn(func(context.Context, string) (entity.Queue, error) {
 		return queue, nil

stovepipe/controller/build/build_test.go

@@ -19,6 +19,7 @@ import (
 	"database/sql"
 	"database/sql/driver"
 	"errors"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -66,6 +67,7 @@ type buildMocks struct {
 	runnerFactory *buildrunnermock.MockFactory
 	runner        *buildrunnermock.MockBuildRunner
 	publisher     *mqmock.MockPublisher
+	association   *gomock.Call
 	metricsScope  tally.TestScope
 }
 
@@ -105,6 +107,9 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildMoc
 	require.NoError(t, err)
 
 	c := NewController(zap.NewNop().Sugar(), scope, staticStorageFactory{store: m.store}, m.materializer, m.runnerFactory, registry, stovepipemq.TopicKeyBuild, "stovepipe-build")
+	m.association = m.reqStore.EXPECT().Update(gomock.Any(), gomock.Cond(func(request entity.Request) bool {
+		return request.BuildID == testBuildID && request.Version == 1
+	}), int32(1), int32(2)).Return(nil).AnyTimes()
 	return c, m
 }
 
@@ -332,14 +337,15 @@ func TestProcess(t *testing.T) {
 			},
 		},
 		{
-			name:    "duplicate runner build id is rejected",
+			name:    "runner build belonging to another request is rejected",
 			wantErr: true,
 			setup: func(m buildMocks) {
 				req := processingRequest(entity.BuildStrategyFull, "")
 				m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(req, nil)
 				m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil)
 				m.runner.EXPECT().Trigger(gomock.Any(), "", testHeadURI, entity.BuildMetadata(nil)).Return(entity.BuildID{ID: testBuildID}, nil)
 				m.buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists)
+				m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{ID: testBuildID, RequestID: "other"}, nil)
 			},
 		},
 		{
@@ -509,3 +515,40 @@ func TestAnalysisPublishDeduplicatesRetriesOfTheSameOccasion(t *testing.T) {
 	assert.Equal(t, ids[0], ids[1])
 	assert.NotEqual(t, publish.IntentID(testBuildID, "finished"), ids[0])
 }
+
+func TestBuildAssociationIsRetriedBeforeAnyDownstreamPublication(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	c, m := newController(t, ctrl)
+	request := processingRequest(entity.BuildStrategyFull, "")
+	m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil)
+	m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil)
+	m.runner.EXPECT().Trigger(gomock.Any(), "", testHeadURI, entity.BuildMetadata(nil)).Return(entity.BuildID{ID: testBuildID}, nil)
+	m.buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
+	m.association.Return(assert.AnError).Times(1)
+
+	require.ErrorIs(t, c.Process(queueContext(), delivery(t, ctrl, buildPayload(t, testID))), assert.AnError)
+}
+
+func TestBuildRetryReusesItsAssociation(t *testing.T) {
+	for _, associated := range []bool{false, true} {
+		t.Run(fmt.Sprint(associated), func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			c, m := newController(t, ctrl)
+			request := processingRequest(entity.BuildStrategyFull, "")
+			if associated {
+				request.BuildID = testBuildID
+			}
+			m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil)
+			m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil)
+			if !associated {
+				m.runner.EXPECT().Trigger(gomock.Any(), "", testHeadURI, entity.BuildMetadata(nil)).Return(entity.BuildID{ID: testBuildID}, nil)
+				m.buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists)
+				m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{ID: testBuildID, RequestID: testID}, nil)
+			}
+			expectBuildTriggered(m)
+			m.publisher.EXPECT().Publish(gomock.Any(), "buildsignal", gomock.Any()).Return(nil)
+			m.publisher.EXPECT().Publish(gomock.Any(), "buildanalysis", gomock.Any()).Return(nil)
+			require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, buildPayload(t, testID))))
+		})
+	}
+}

stovepipe/entity/request.go

@@ -115,6 +115,8 @@ type Request struct {
 
 	// State is the current state of the request in the pipeline.
 	State RequestState `json:"state"`
+	// BuildID identifies the associated CI build and is empty before it is triggered.
+	BuildID string `json:"build_id"`
 	// TerminalBuildID is the build that established this request's terminal
 	// state. It is empty when no build established the terminal state.
 	TerminalBuildID string `json:"terminal_build_id"`

stovepipe/extension/storage/mysql/request_store.go

@@ -55,14 +55,15 @@ func (r *requestStore) Create(ctx context.Context, request entity.Request) (retE
 	}
 
 	_, err := r.db.ExecContext(ctx,
-		`INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, version)
-		 VALUES (?, ?, ?, ?, ?, ?, ?)`,
+		`INSERT INTO request (id, queue, uri, state, build_strategy, base_uri, build_id, version)
+		 VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
 		request.ID,
 		request.Queue,
 		request.URI,
 		request.State,
 		request.BuildStrategy,
 		request.BaseURI,
+		request.BuildID,
 		request.Version,
 	)
 	if err != nil {
@@ -82,7 +83,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request,
 
 	var req entity.Request
 	err := r.db.QueryRowContext(ctx,
-		`SELECT id, queue, uri, state, build_strategy, base_uri, COALESCE(terminal_build_id, ''), version
+		`SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE(terminal_build_id, ''), version
 		 FROM request WHERE queue = ? AND id = ?`,
 		r.queue, id,
 	).Scan(
@@ -92,6 +93,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request,
 		&req.State,
 		&req.BuildStrategy,
 		&req.BaseURI,
+		&req.BuildID,
 		&req.TerminalBuildID,
 		&req.Version,
 	)
@@ -107,7 +109,7 @@ func (r *requestStore) Get(ctx context.Context, id string) (ret entity.Request,
 }
 
 // Update persists the mutable fields of request (URI, state, build strategy,
-// base URI, terminal build ID) if the oldVersion, writing newVersion. Returns
+// base URI, associated build ID, terminal build ID) if the oldVersion, writing newVersion. Returns
 // ErrVersionMismatch if the stored version does not match (including when the
 // request does not exist). This is a pure conditional write; the caller owns
 // version arithmetic.
@@ -121,12 +123,13 @@ func (r *requestStore) Update(ctx context.Context, request entity.Request, oldVe
 
 	result, err := r.db.ExecContext(ctx,
 		`UPDATE request
-		 SET uri = ?, state = ?, build_strategy = ?, base_uri = ?, terminal_build_id = NULLIF(?, ''), version = ?
+		 SET uri = ?, state = ?, build_strategy = ?, base_uri = ?, build_id = ?, terminal_build_id = NULLIF(?, ''), version = ?
 		 WHERE queue = ? AND id = ? AND version = ?`,
 		request.URI,
 		request.State,
 		request.BuildStrategy,
 		request.BaseURI,
+		request.BuildID,
 		request.TerminalBuildID,
 		newVersion,
 		request.Queue,

stovepipe/extension/storage/mysql/request_store_test.go

@@ -62,15 +62,15 @@ func TestRequestStore_Create(t *testing.T) {
 			name: "success",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO request").
-					WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.Version).
+					WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.Version).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
 		{
 			name: "duplicate id returns ErrAlreadyExists",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO request").
-					WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.Version).
+					WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.Version).
 					WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry})
 			},
 			wantErr:   true,
@@ -80,7 +80,7 @@ func TestRequestStore_Create(t *testing.T) {
 			name: "other exec error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("INSERT INTO request").
-					WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.Version).
+					WithArgs(request.ID, request.Queue, request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.Version).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -116,6 +116,7 @@ func TestRequestStore_Get(t *testing.T) {
 		State:           entity.RequestStateProcessing,
 		BuildStrategy:   entity.BuildStrategyFull,
 		BaseURI:         "",
+		BuildID:         "bk-1001",
 		TerminalBuildID: "bk-1001",
 		Version:         2,
 	}
@@ -132,9 +133,9 @@ func TestRequestStore_Get(t *testing.T) {
 			name: "found",
 			id:   want.ID,
 			setup: func(mock sqlmock.Sqlmock) {
-				rows := sqlmock.NewRows([]string{"id", "queue", "uri", "state", "build_strategy", "base_uri", "terminal_build_id", "version"}).
-					AddRow(want.ID, want.Queue, want.URI, string(want.State), string(want.BuildStrategy), want.BaseURI, want.TerminalBuildID, want.Version)
-				mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, COALESCE\\(terminal_build_id, ''\\), version").
+				rows := sqlmock.NewRows([]string{"id", "queue", "uri", "state", "build_strategy", "base_uri", "build_id", "terminal_build_id", "version"}).
+					AddRow(want.ID, want.Queue, want.URI, string(want.State), string(want.BuildStrategy), want.BaseURI, want.BuildID, want.TerminalBuildID, want.Version)
+				mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE\\(terminal_build_id, ''\\), version").
 					WithArgs("monorepo/main", want.ID).
 					WillReturnRows(rows)
 			},
@@ -144,7 +145,7 @@ func TestRequestStore_Get(t *testing.T) {
 			name: "not found",
 			id:   "missing",
 			setup: func(mock sqlmock.Sqlmock) {
-				mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, COALESCE\\(terminal_build_id, ''\\), version").
+				mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE\\(terminal_build_id, ''\\), version").
 					WithArgs("monorepo/main", "missing").
 					WillReturnError(sql.ErrNoRows)
 			},
@@ -155,7 +156,7 @@ func TestRequestStore_Get(t *testing.T) {
 			name: "query error",
 			id:   "bad",
 			setup: func(mock sqlmock.Sqlmock) {
-				mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, COALESCE\\(terminal_build_id, ''\\), version").
+				mock.ExpectQuery("SELECT id, queue, uri, state, build_strategy, base_uri, build_id, COALESCE\\(terminal_build_id, ''\\), version").
 					WithArgs("monorepo/main", "bad").
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
@@ -193,6 +194,7 @@ func TestRequestStore_Update(t *testing.T) {
 		State:           entity.RequestStateProcessing,
 		BuildStrategy:   entity.BuildStrategyFull,
 		BaseURI:         "",
+		BuildID:         "bk-1001",
 		TerminalBuildID: "bk-1001",
 	}
 	const oldVersion, newVersion = int32(1), int32(2)
@@ -207,15 +209,15 @@ func TestRequestStore_Update(t *testing.T) {
 			name: "success",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE request").
-					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion).
+					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 1))
 			},
 		},
 		{
 			name: "version mismatch",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE request").
-					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion).
+					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion).
 					WillReturnResult(sqlmock.NewResult(0, 0))
 			},
 			wantErr:   true,
@@ -225,7 +227,7 @@ func TestRequestStore_Update(t *testing.T) {
 			name: "exec error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE request").
-					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion).
+					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion).
 					WillReturnError(fmt.Errorf("connection reset"))
 			},
 			wantErr: true,
@@ -234,7 +236,7 @@ func TestRequestStore_Update(t *testing.T) {
 			name: "rows affected error",
 			setup: func(mock sqlmock.Sqlmock) {
 				mock.ExpectExec("UPDATE request").
-					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion).
+					WithArgs(request.URI, request.State, request.BuildStrategy, request.BaseURI, request.BuildID, request.TerminalBuildID, newVersion, request.Queue, request.ID, oldVersion).
 					WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error")))
 			},
 			wantErr: true,

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

@@ -8,6 +8,7 @@ CREATE TABLE IF NOT EXISTS request (
     state             VARCHAR(64)  NOT NULL,
     build_strategy    VARCHAR(64)  NOT NULL DEFAULT '',
     base_uri          VARCHAR(255) NOT NULL DEFAULT '',
+    build_id          VARCHAR(255) NOT NULL DEFAULT '',
     version           INT          NOT NULL,
     terminal_build_id VARCHAR(255) NULL,
     PRIMARY KEY (queue, id)