Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions service/submitqueue/gateway/server/queues.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions service/submitqueue/orchestrator/server/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
},
Expand Down
59 changes: 59 additions & 0 deletions test/e2e/submitqueue/harness_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
123 changes: 123 additions & 0 deletions test/e2e/submitqueue/suite_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down
Loading