fix(client): bound the list window at the call time

`List` never set `ReceivedBeforeMs`, and the gateway rejects any list where `received_at_or_after_ms >= received_before_ms` (`gateway/controller/list.go:85`). With `Since` unset the client sent `(0, 0)` and with `-since 1h` it sent `(now-1h, 0)` — both invalid, so every `list` call, and every `watch` that did not name its requests with `-sqid`, failed with InvalidArgument against a real gateway.

Both bounds are now taken from a single `now` before the first page. Fixing them up front is required rather than tidy: the continuation token pins both bounds (`list.go:111`), so a bound recomputed per page would be rejected from the second page on.

Author: behinddwalls

submitqueue/client/query.go

@@ -29,8 +29,8 @@ type ListQuery struct {
 	// own queue.
 	Queue string
 
-	// Since bounds the window to requests received within it. Zero reads from
-	// the beginning of retained history.
+	// Since bounds the window to requests received within it, ending at the
+	// time of the call. Zero reads from the beginning of retained history.
 	Since time.Duration
 
 	// Limit caps how many requests are returned across all pages. Zero means
@@ -52,9 +52,18 @@ func (c *Client) List(ctx context.Context, q ListQuery) ([]*pb.RequestSummary, e
 		return nil, fmt.Errorf("queue must not be empty")
 	}
 
-	req := &pb.ListRequest{Queue: q.Queue, PageSize: int32(q.PageSize)}
+	// Both bounds are fixed before the first page. The gateway requires
+	// received_at_or_after_ms < received_before_ms, so an unset upper bound
+	// rejects every call, and its continuation token pins both bounds, so one
+	// recomputed per page would be rejected from the second page on.
+	now := time.Now()
+	req := &pb.ListRequest{
+		Queue:            q.Queue,
+		PageSize:         int32(q.PageSize),
+		ReceivedBeforeMs: now.UnixMilli(),
+	}
 	if q.Since > 0 {
-		req.ReceivedAtOrAfterMs = time.Now().Add(-q.Since).UnixMilli()
+		req.ReceivedAtOrAfterMs = now.Add(-q.Since).UnixMilli()
 	}
 
 	var out []*pb.RequestSummary

submitqueue/client/query_test.go

@@ -108,6 +108,42 @@ func TestListWithoutSinceLeavesTheWindowOpen(t *testing.T) {
 		"no window means all retained history, not a bound of zero-time")
 }
 
+func TestListClosesTheWindowAtTheCallTime(t *testing.T) {
+	// The gateway rejects a list whose lower bound is not strictly below its
+	// upper one, so leaving the upper bound unset fails every call.
+	gw := &pagingGateway{pages: [][]string{{"q/1"}}}
+	sq, stop := dial(t, gw)
+	defer stop()
+
+	before := time.Now().UnixMilli()
+	_, err := sq.List(context.Background(), ListQuery{Queue: "q"})
+	require.NoError(t, err)
+	after := time.Now().UnixMilli()
+
+	got := gw.lastRequest.GetReceivedBeforeMs()
+	assert.GreaterOrEqual(t, got, before)
+	assert.LessOrEqual(t, got, after)
+	assert.Less(t, gw.lastRequest.GetReceivedAtOrAfterMs(), got,
+		"the gateway requires received_at_or_after_ms < received_before_ms")
+}
+
+func TestListKeepsTheWindowFixedAcrossPages(t *testing.T) {
+	// The continuation token pins both bounds, so a bound recomputed per page
+	// would be rejected from the second page on.
+	gw := &pagingGateway{pages: [][]string{{"q/1"}, {"q/2"}, {"q/3"}}}
+	sq, stop := dial(t, gw)
+	defer stop()
+
+	_, err := sq.List(context.Background(), ListQuery{Queue: "q", Since: time.Hour})
+	require.NoError(t, err)
+
+	require.Len(t, gw.requests, 3)
+	for _, req := range gw.requests[1:] {
+		assert.Equal(t, gw.requests[0].GetReceivedAtOrAfterMs(), req.GetReceivedAtOrAfterMs())
+		assert.Equal(t, gw.requests[0].GetReceivedBeforeMs(), req.GetReceivedBeforeMs())
+	}
+}
+
 func TestRowsFromSummaries(t *testing.T) {
 	received := time.Now().Add(-5 * time.Minute)
 	rows := RowsFromSummaries([]*pb.RequestSummary{
@@ -143,10 +179,12 @@ type pagingGateway struct {
 	alwaysToken bool
 
 	calls       int
+	requests    []*pb.ListRequest
 	lastRequest *pb.ListRequest
 }
 
 func (g *pagingGateway) List(_ context.Context, req *pb.ListRequest) (*pb.ListResponse, error) {
+	g.requests = append(g.requests, req)
 	g.lastRequest = req
 	page := g.calls
 	g.calls++