test(e2e): assert speculation-tree parity on the landing pipeline

test/e2e/submitqueue/harness_test.go

@@ -34,8 +34,24 @@ import (
 	mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb"
 	gatewaypb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb"
 	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
 )
 
+// allBatchStates enumerates every entity.BatchState, active and terminal. Neither
+// entity.ActiveBatchStates nor entity.DependencyBatchStates covers the terminal
+// states (Succeeded, Failed, Cancelled), but batchForRequest needs to find a
+// request's batch regardless of what state it has settled into.
+var allBatchStates = []entity.BatchState{
+	entity.BatchStateCreated,
+	entity.BatchStateScored,
+	entity.BatchStateSpeculating,
+	entity.BatchStateMerging,
+	entity.BatchStateCancelling,
+	entity.BatchStateSucceeded,
+	entity.BatchStateFailed,
+	entity.BatchStateCancelled,
+}
+
 // land submits a request with the default REBASE strategy and returns its sqid.
 // URIs may carry "sq-fake=<token>" markers to steer negative paths (see
 // submitqueue/core/fakemarker); the happy path uses a plain change URI.
@@ -159,3 +175,90 @@ func isTerminalStatus(status entity.RequestStatus) bool {
 		return false
 	}
 }
+
+// batchForRequest finds the batch containing sqid's request within queue. The
+// gateway's Land response Sqid is the request ID verbatim, and the batch
+// controller sets a batch's Contains to exactly the request IDs it contains, so
+// scanning every batch in the queue for one whose Contains includes sqid is the
+// only way to map a request to its batch — there is no direct request->batch
+// index. Fails the test if no such batch is found.
+func (s *E2EIntegrationSuite) batchForRequest(queue, sqid string) entity.Batch {
+	t := s.T()
+	batches, err := s.batchStore.GetByQueueAndStates(s.ctx, queue, allBatchStates)
+	require.NoError(t, err, "failed to list batches for queue %s", queue)
+	for _, b := range batches {
+		for _, id := range b.Contains {
+			if id == sqid {
+				return b
+			}
+		}
+	}
+	require.Fail(t, "no batch found containing request", "sqid=%s queue=%s", sqid, queue)
+	return entity.Batch{}
+}
+
+// findBatchForRequest is the tolerant counterpart to batchForRequest: it reports
+// ok=false instead of failing the test when no batch contains sqid. Used where
+// "no batch was ever created for this request" is a legitimate outcome (e.g. a
+// cancel that raced ahead of batch creation), not a bug.
+func (s *E2EIntegrationSuite) findBatchForRequest(queue, sqid string) (entity.Batch, bool) {
+	t := s.T()
+	batches, err := s.batchStore.GetByQueueAndStates(s.ctx, queue, allBatchStates)
+	require.NoError(t, err, "failed to list batches for queue %s", queue)
+	for _, b := range batches {
+		for _, id := range b.Contains {
+			if id == sqid {
+				return b, true
+			}
+		}
+	}
+	return entity.Batch{}, false
+}
+
+// speculationTree fetches batchID's speculation tree, failing the test if the
+// store errors (including not-found — callers that expect "no tree" as a
+// legitimate outcome should call speculationTreeIfExists instead).
+func (s *E2EIntegrationSuite) speculationTree(batchID string) entity.SpeculationTree {
+	t := s.T()
+	tree, err := s.speculationTreeStore.Get(s.ctx, batchID)
+	require.NoError(t, err, "failed to get speculation tree for batch %s", batchID)
+	return tree
+}
+
+// speculationTreeIfExists is the tolerant counterpart to speculationTree: it
+// reports ok=false (instead of failing the test) when the store reports
+// ErrNotFound, since "no tree" is a legitimate outcome for a batch that was
+// cancelled before speculation ever ran. Any other error still fails the test.
+func (s *E2EIntegrationSuite) speculationTreeIfExists(batchID string) (entity.SpeculationTree, bool) {
+	t := s.T()
+	tree, err := s.speculationTreeStore.Get(s.ctx, batchID)
+	if storage.IsNotFound(err) {
+		return entity.SpeculationTree{}, false
+	}
+	require.NoError(t, err, "failed to get speculation tree for batch %s", batchID)
+	return tree, true
+}
+
+// buildsForBatch collects every build triggered for batchID's speculation
+// paths. The tree is the index: each path's assigned ID keys its path→build
+// mapping row, which names the build row. A missing mapping just means that
+// path never built.
+func (s *E2EIntegrationSuite) buildsForBatch(batchID string) []entity.Build {
+	t := s.T()
+	tree, ok := s.speculationTreeIfExists(batchID)
+	if !ok {
+		return nil
+	}
+	var builds []entity.Build
+	for _, p := range tree.Paths {
+		pb, err := s.pathBuildStore.Get(s.ctx, p.ID)
+		if storage.IsNotFound(err) {
+			continue
+		}
+		require.NoError(t, err, "failed to get path→build mapping for path %s of batch %s", p.ID, batchID)
+		b, err := s.buildStore.Get(s.ctx, pb.BuildID)
+		require.NoError(t, err, "failed to get build %s for path %s of batch %s", pb.BuildID, p.ID, batchID)
+		builds = append(builds, b)
+	}
+	return builds
+}

test/e2e/submitqueue/suite_test.go

@@ -55,6 +55,11 @@ type E2EIntegrationSuite struct {
 	queueDB            *sql.DB                 // Queue database
 	requestLog         storage.RequestLogStore // White-box view of the request_log status timeline (app DB)
 	requestStore       storage.RequestStore    // White-box view of the internal RequestState (app DB)
+
+	batchStore           storage.BatchStore                // White-box view of batches (app DB) — used to map a request's sqid to its batch
+	speculationTreeStore storage.SpeculationTreeStore      // White-box view of a batch's speculation tree (app DB)
+	buildStore           storage.BuildStore                // White-box view of builds triggered per batch (app DB)
+	pathBuildStore       storage.SpeculationPathBuildStore // White-box view of the path→build mapping (app DB)
 }
 
 func TestE2EIntegration(t *testing.T) {
@@ -115,6 +120,10 @@ func (s *E2EIntegrationSuite) SetupSuite() {
 	// status history) and the operating store (point-in-time RequestState).
 	s.requestLog = storagemysql.NewRequestLogStore(s.db, tally.NoopScope)
 	s.requestStore = storagemysql.NewRequestStore(s.db, tally.NoopScope)
+	s.batchStore = storagemysql.NewBatchStore(s.db, tally.NoopScope)
+	s.speculationTreeStore = storagemysql.NewSpeculationTreeStore(s.db, tally.NoopScope)
+	s.buildStore = storagemysql.NewBuildStore(s.db, tally.NoopScope)
+	s.pathBuildStore = storagemysql.NewSpeculationPathBuildStore(s.db, tally.NoopScope)
 
 	// Connect to Gateway gRPC service
 	var gatewayConn *grpc.ClientConn
@@ -216,6 +225,45 @@ func (s *E2EIntegrationSuite) TestLand_HappyPath_ReachesLanded() {
 	// terminal check, not a sequence.
 	assert.Equal(s.T(), entity.RequestStateLanded, s.terminalState(sqid),
 		"operating store should show request %s in terminal state landed", sqid)
+
+	// White-box (speculation tree parity): the batch this request landed in went
+	// through the tree-driven speculation pipeline (speculate -> prioritize ->
+	// build -> speculate reconcile) and its persisted state proves parity with
+	// the old flow — one build per batch, driven by exactly one speculation
+	// path. By the time the request has reached "landed", speculate's reconcile
+	// step has already stamped the tree path's BuildID and flipped its Status to
+	// passed (that write strictly precedes the merge publish, and the build row
+	// precedes that), so a single synchronous read suffices — no polling needed
+	// for path/build status. The tree's Version can still legitimately creep up
+	// afterward (late buildsignal polls), so only a lower bound is asserted.
+	t := s.T()
+	batch := s.batchForRequest("e2e-test-queue", sqid)
+	tree := s.speculationTree(batch.ID)
+
+	require.Len(t, tree.Paths, 1, "batch %s should have exactly one speculation path (chain enumerator)", batch.ID)
+	path := tree.Paths[0]
+
+	assert.Equal(t, batch.ID, path.Path.Head, "path Head should be the batch itself")
+	// batch.Dependencies and path.Path.Base each round-trip through JSON
+	// independently (one via the batch table's dependencies column, the other
+	// via the speculation_tree table's paths column) and empirically end up
+	// with different nil-vs-empty-slice representations for "no dependencies"
+	// even though the batch has none either way; normalize both to non-nil
+	// before comparing so this asserts the same ordered contents, not which
+	// side happens to be nil.
+	wantBase := append([]string{}, batch.Dependencies...)
+	gotBase := append([]string{}, path.Path.Base...)
+	assert.Equal(t, wantBase, gotBase, "path Base should equal the batch's Dependencies in order")
+	assert.Equal(t, entity.SpeculationPathStatusPassed, path.Status, "the batch's single path should have passed")
+	assert.NotEmpty(t, path.BuildID, "a passed path should have a BuildID stamped by reconcile")
+	assert.Greater(t, path.Score, float32(0), "the path scorer should have produced a positive score")
+	assert.Greater(t, tree.Version, int32(1), "the tree version should have advanced past its initial Create at version 1")
+
+	builds := s.buildsForBatch(batch.ID)
+	if assert.Len(t, builds, 1, "batch %s should have exactly one build (one build per batch is parity's core claim)", batch.ID) {
+		assert.Equal(t, path.ID, builds[0].SpeculationPathID, "the build should reference the tree's single path by ID")
+		assert.Equal(t, entity.BuildStatusSucceeded, builds[0].Status, "the build should have succeeded")
+	}
 }
 
 // TestCancelRequest_InvalidSqid verifies the gateway rejects an empty sqid
@@ -243,6 +291,13 @@ func (s *E2EIntegrationSuite) TestCancelRequest_InvalidSqid() {
 // pipeline-pause lever (e.g. a runway "park" marker that withholds the
 // merge-conflict-check signal so the request is caught pre-batch) — that is the
 // next incremental, per-stage addition on top of this harness.
+//
+// Because the terminal outcome is one of three possibilities, the white-box
+// speculation-tree check appended at the end of this test is branch-tolerant
+// rather than a single fixed expectation: it only asserts per-path "cancelled"
+// status when the request actually landed on Cancelled, and otherwise asserts
+// only the invariant that holds regardless of which terminal status was
+// reached — no build is left non-terminal.
 func (s *E2EIntegrationSuite) TestCancel_RecordsIntent() {
 	t := s.T()
 
@@ -259,4 +314,39 @@ func (s *E2EIntegrationSuite) TestCancel_RecordsIntent() {
 		entity.RequestStatusAccepted,
 		entity.RequestStatusCancelling,
 	)
+
+	// White-box (speculation-tree tolerant check): cancellation here is
+	// best-effort and races the pipeline (see comment above), so the request may
+	// end up Landed, Error, or Cancelled. Whatever the outcome, no build should be
+	// left non-terminal once the request itself has reached a terminal status; if
+	// the outcome specifically was Cancelled and a tree exists, every path in it
+	// should also show Cancelled.
+	final := s.awaitTerminal(sqid)
+	s.log.Logf("Cancel race settled: sqid=%s final status=%q", sqid, final)
+
+	batch, ok := s.findBatchForRequest("e2e-cancel-queue", sqid)
+	if !ok {
+		s.log.Logf("no batch was ever created for %s (cancel raced ahead of batch creation)", sqid)
+		return
+	}
+
+	tree, ok := s.speculationTreeIfExists(batch.ID)
+	if !ok {
+		s.log.Logf("no speculation tree for batch %s (cancelled before speculation ran)", batch.ID)
+		return
+	}
+
+	if final == entity.RequestStatusCancelled {
+		s.log.Logf("request %s reached Cancelled; asserting every path in batch %s's tree is cancelled", sqid, batch.ID)
+		for _, p := range tree.Paths {
+			assert.Equal(t, entity.SpeculationPathStatusCancelled, p.Status,
+				"batch %s path %+v should be cancelled", batch.ID, p.Path)
+		}
+	} else {
+		s.log.Logf("request %s reached terminal status %q (not Cancelled); skipping the per-path cancelled check", sqid, final)
+	}
+
+	for _, b := range s.buildsForBatch(batch.ID) {
+		assert.True(t, b.Status.IsTerminal(), "build %s for batch %s should not be left non-terminal (status=%s)", b.ID, batch.ID, b.Status)
+	}
 }