diff --git a/service/submitqueue/gateway/server/queues.yaml b/service/submitqueue/gateway/server/queues.yaml index df7571fe..39d41e28 100644 --- a/service/submitqueue/gateway/server/queues.yaml +++ b/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 diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 2ed845c3..b0fbf086 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/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, }, diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go index b8d18eef..7ff4b7ca 100644 --- a/test/e2e/submitqueue/harness_test.go +++ b/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) { diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index eeb87826..67bf21e5 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/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() {