test(e2e): cover gateway List API

Author: albertywu

service/submitqueue/gateway/server/queues.yaml

@@ -6,6 +6,7 @@
 queues:
   - name: test-queue
   - name: e2e-test-queue
+  - name: e2e-list-queue
   - name: e2e-cancel-queue
   # Routes to an analyzer that always errors (conflictfake.FailAlways) so e2e can
   # exercise the conflict-analysis error path. See newQueueRegistry in the

service/submitqueue/orchestrator/server/main.go

@@ -941,6 +941,7 @@ func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.
 		byQueue: map[string]queueExtensions{
 			"test-queue":               testQueue,
 			"e2e-test-queue":           e2eQueue,
+			"e2e-list-queue":           e2eQueue,
 			"e2e-conflict-error-queue": conflictErrQueue,
 			"file-overlap-queue":       fileOverlapQueue,
 		},

test/e2e/submitqueue/harness_test.go

@@ -51,6 +51,65 @@ func (s *E2EIntegrationSuite) land(queue string, uris ...string) string {
 	return resp.Sqid
 }
 
+// list calls the gateway List RPC and fails the test on unexpected errors.
+func (s *E2EIntegrationSuite) list(req *gatewaypb.ListRequest) *gatewaypb.ListResponse {
+	t := s.T()
+	resp, err := s.gatewayClient.List(s.ctx, req)
+	require.NoError(t, err, "List failed for queue %s", req.Queue)
+	return resp
+}
+
+// awaitListContains polls List until all expected sqids are visible in one
+// response page. The gateway updates request summaries synchronously for Land
+// and asynchronously from the log topic for later statuses, so callers use the
+// same bounded polling style as Status.
+func (s *E2EIntegrationSuite) awaitListContains(req *gatewaypb.ListRequest, want ...string) *gatewaypb.ListResponse {
+	t := s.T()
+	var resp *gatewaypb.ListResponse
+	require.Eventually(t, func() bool {
+		var err error
+		resp, err = s.gatewayClient.List(s.ctx, req)
+		if err != nil {
+			s.log.Logf("List(%s) not ready yet: %v", req.Queue, err)
+			return false
+		}
+		got := summarySQIDs(resp.Requests)
+		s.log.Logf("List(%s) = %v (want %v)", req.Queue, got, want)
+		return containsAll(got, want)
+	}, persistTimeout, persistPollInterval,
+		"List(%s) should contain sqids %v", req.Queue, want)
+	return resp
+}
+
+func summarySQIDs(summaries []*gatewaypb.RequestSummary) []string {
+	out := make([]string, len(summaries))
+	for i, summary := range summaries {
+		out[i] = summary.Sqid
+	}
+	return out
+}
+
+func containsAll(got []string, want []string) bool {
+	seen := make(map[string]struct{}, len(got))
+	for _, sqid := range got {
+		seen[sqid] = struct{}{}
+	}
+	for _, sqid := range want {
+		if _, ok := seen[sqid]; !ok {
+			return false
+		}
+	}
+	return true
+}
+
+func mapFromStrings(values []string) map[string]struct{} {
+	out := make(map[string]struct{}, len(values))
+	for _, value := range values {
+		out[value] = struct{}{}
+	}
+	return out
+}
+
 // currentStatus reads the request's current customer-facing status via the
 // Status RPC. A transport error is returned so callers can keep polling.
 func (s *E2EIntegrationSuite) currentStatus(sqid string) (entity.RequestStatus, error) {

test/e2e/submitqueue/suite_test.go

@@ -218,6 +218,129 @@ func (s *E2EIntegrationSuite) TestLand_HappyPath_ReachesLanded() {
 		"operating store should show request %s in terminal state landed", sqid)
 }
 
+// TestList_ReturnsFilteredPagedSummaries verifies the customer-facing List RPC
+// against the full stack. Land writes the initial summary through the gateway,
+// later status updates arrive through the request-log topic, and List reads the
+// gateway-owned summary read model through the public RPC surface.
+func (s *E2EIntegrationSuite) TestList_ReturnsFilteredPagedSummaries() {
+	t := s.T()
+
+	beforeWindow := s.land("e2e-list-queue", "github://uber/e2e-list/pull/100/abcdef0123456789abcdef0123456789abcdef00")
+	startTimeMs := time.Now().UnixMilli() + 1
+	windowTimer := time.NewTimer(time.Until(time.UnixMilli(startTimeMs)))
+	<-windowTimer.C
+	firstURI := "github://uber/e2e-list/pull/101/abcdef0123456789abcdef0123456789abcdef01"
+	secondURI := "github://uber/e2e-list/pull/102/abcdef0123456789abcdef0123456789abcdef02"
+	thirdURI := "github://uber/e2e-list/pull/103/abcdef0123456789abcdef0123456789abcdef03"
+	otherURI := "github://uber/e2e-list-other/pull/201/abcdef0123456789abcdef0123456789abcdef03"
+
+	first := s.land("e2e-list-queue", firstURI)
+	second := s.land("e2e-list-queue", secondURI)
+	third := s.land("e2e-list-queue", thirdURI)
+	otherQueue := s.land("e2e-cancel-queue", otherURI)
+	endTimeMs := time.Now().Add(time.Minute).UnixMilli()
+
+	resp := s.awaitListContains(&gatewaypb.ListRequest{
+		Queue:       "e2e-list-queue",
+		StartTimeMs: startTimeMs,
+		EndTimeMs:   endTimeMs,
+		PageSize:    10,
+	}, first, second, third)
+
+	assert.NotContains(t, summarySQIDs(resp.Requests), otherQueue,
+		"List should not return requests from a different queue")
+	assert.NotContains(t, summarySQIDs(resp.Requests), beforeWindow,
+		"List should not return requests admitted before the time window")
+
+	bySQID := make(map[string]*gatewaypb.RequestSummary, len(resp.Requests))
+	for _, summary := range resp.Requests {
+		bySQID[summary.Sqid] = summary
+	}
+
+	firstSummary := bySQID[first]
+	require.NotNil(t, firstSummary, "List response should include %s", first)
+	assert.Equal(t, "e2e-list-queue", firstSummary.Queue)
+	assert.Equal(t, []string{firstURI}, firstSummary.ChangeUris)
+	assert.NotEmpty(t, firstSummary.Status)
+	assert.GreaterOrEqual(t, firstSummary.StartedAtMs, startTimeMs)
+	assert.Less(t, firstSummary.StartedAtMs, endTimeMs)
+	assert.GreaterOrEqual(t, firstSummary.UpdatedAtMs, firstSummary.StartedAtMs)
+
+	s.awaitStatus(first, entity.RequestStatusLanded)
+	s.awaitStatus(second, entity.RequestStatusLanded)
+	s.awaitStatus(third, entity.RequestStatusLanded)
+
+	landedResp := s.awaitListContains(&gatewaypb.ListRequest{
+		Queue:       "e2e-list-queue",
+		StartTimeMs: startTimeMs,
+		EndTimeMs:   endTimeMs,
+		Statuses:    []string{string(entity.RequestStatusLanded)},
+		PageSize:    10,
+	}, first, second, third)
+	for _, summary := range landedResp.Requests {
+		if summary.Sqid != first && summary.Sqid != second && summary.Sqid != third {
+			continue
+		}
+		assert.Equal(t, string(entity.RequestStatusLanded), summary.Status, "landed summary %s should report landed", summary.Sqid)
+		assert.Greater(t, summary.CompletedAtMs, int64(0), "landed summary %s should have completion time", summary.Sqid)
+	}
+
+	var pagedSQIDs []string
+	var pageToken string
+	for {
+		page := s.list(&gatewaypb.ListRequest{
+			Queue:       "e2e-list-queue",
+			StartTimeMs: startTimeMs,
+			EndTimeMs:   endTimeMs,
+			PageSize:    1,
+			PageToken:   pageToken,
+			Sort:        gatewaypb.ListSort_ADMITTED_DESC,
+		})
+		require.LessOrEqual(t, len(page.Requests), 1)
+		if len(page.Requests) == 1 {
+			pagedSQIDs = append(pagedSQIDs, page.Requests[0].Sqid)
+		}
+		if page.NextPageToken == "" {
+			break
+		}
+		pageToken = page.NextPageToken
+	}
+	assert.ElementsMatch(t, []string{first, second, third}, pagedSQIDs)
+	assert.Len(t, mapFromStrings(pagedSQIDs), len(pagedSQIDs), "descending pagination should not return duplicates")
+}
+
+// TestList_ShowsCancelIntent verifies the List summary read model observes the
+// gateway-synchronous cancelling log written by Cancel. Terminal cancellation is
+// intentionally not asserted here because the e2e stack does not yet have a
+// deterministic pipeline pause; the terminal outcome can race with landing.
+func (s *E2EIntegrationSuite) TestList_ShowsCancelIntent() {
+	t := s.T()
+
+	startTimeMs := time.Now().Add(-time.Second).UnixMilli()
+	sqid := s.land("e2e-cancel-queue", "github://uber/e2e-list-cancel/pull/301/abcdef0123456789abcdef0123456789abcdef04")
+	endTimeMs := time.Now().Add(time.Minute).UnixMilli()
+
+	_, err := s.gatewayClient.Cancel(s.ctx, &gatewaypb.CancelRequest{Sqid: sqid, Reason: "e2e list cancel test"})
+	require.NoError(t, err, "Cancel failed")
+
+	resp := s.list(&gatewaypb.ListRequest{
+		Queue:       "e2e-cancel-queue",
+		StartTimeMs: startTimeMs,
+		EndTimeMs:   endTimeMs,
+		Statuses:    []string{string(entity.RequestStatusCancelling)},
+		PageSize:    10,
+	})
+	var summary *gatewaypb.RequestSummary
+	for _, candidate := range resp.Requests {
+		if candidate.Sqid == sqid {
+			summary = candidate
+			break
+		}
+	}
+	require.NotNil(t, summary, "List should include %s with cancelling status immediately after Cancel", sqid)
+	assert.Equal(t, string(entity.RequestStatusCancelling), summary.Status)
+}
+
 // TestCancelRequest_InvalidSqid verifies the gateway rejects an empty sqid
 // synchronously before publishing anything to the cancel queue.
 func (s *E2EIntegrationSuite) TestCancelRequest_InvalidSqid() {