amadan
release: PageLister, an optional bounded listing
List, ListMod and ListSizes all return every key under a prefix. That is right for a sweep that wants the whole picture and wrong for a queue drain that wants the oldest few: the drain caps how many items it handles per app per tick, and without a bounded listing it would enumerate a flooded app's entire queue every thirty seconds regardless of that cap — paying for the listing it had deliberately declined to act on. Optional and type-asserted, like ModLister and SizeLister beside it, not a Store method. Store has eleven implementations — the two real stores, the prefix wrapper, fleetapi and seven test fakes — and adding a method would break all of them for one caller. The S3 test runs against an httptest fake, not a live bucket and not an env gate. An earlier draft of the plan said to gate it "like every other S3Store test", which was wrong: those tests already stand up an in-process fake, and gating this one would have left the only path that matters unexercised in CI. It asserts start-after and max-keys are actually ON THE WIRE, because a drain that quietly enumerated everything would look correct from the outside while the cap did nothing. DirStore leans on ListMod already being key-sorted, so after and limit are a filter and a slice rather than a second walk.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2 files changed,
+190
−0
internal/release/pagelister_test.go+119 −0internal/release/store.go+71 −0
diff --git a/internal/release/pagelister_test.go b/internal/release/pagelister_test.go| new file mode 100644 |
| index 000000000..14ed6704b |
| --- /dev/null |
| +++ b/internal/release/pagelister_test.go |
| @@ -0,0 +1,119 @@ |
| +package release |
| + |
| +import ( |
| + "context" |
| + "fmt" |
| + "net/http" |
| + "net/http/httptest" |
| + "testing" |
| +) |
| + |
| +func TestDirStorePageListerOrdersAndBounds(t *testing.T) { |
| + ctx := context.Background() |
| + d := NewDirStore(t.TempDir()) |
| + for _, k := range []string{"q/c", "q/a", "q/e", "q/b", "q/d", "other/z"} { |
| + if _, err := d.Put(ctx, k, []byte("x"), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + |
| + // Sorted, bounded, and the prefix is respected. |
| + got, err := d.ListPage(ctx, "q/", "", 3) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + want := []string{"q/a", "q/b", "q/c"} |
| + if fmt.Sprint(got) != fmt.Sprint(want) { |
| + t.Fatalf("first page = %v, want %v", got, want) |
| + } |
| + |
| + // startAfter is EXCLUSIVE, so paging is "the last key I saw". |
| + got, err = d.ListPage(ctx, "q/", "q/c", 3) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if want := []string{"q/d", "q/e"}; fmt.Sprint(got) != fmt.Sprint(want) { |
| + t.Fatalf("second page = %v, want %v", got, want) |
| + } |
| + |
| + // Paging to the end terminates rather than repeating. |
| + got, err = d.ListPage(ctx, "q/", "q/e", 3) |
| + if err != nil || len(got) != 0 { |
| + t.Fatalf("past the end = %v, %v", got, err) |
| + } |
| + |
| + // limit <= 0 means no bound. |
| + got, err = d.ListPage(ctx, "q/", "", 0) |
| + if err != nil || len(got) != 5 { |
| + t.Fatalf("unbounded = %v, %v", got, err) |
| + } |
| +} |
| + |
| +// The S3 half, against an in-process fake rather than a live bucket. |
| +// |
| +// The point of this test is that ListPage puts start-after and max-keys ON THE |
| +// WIRE. Nothing else in the package does — the other listings paginate through |
| +// everything — and a drain that silently enumerated a flooded app's whole |
| +// queue every 30 seconds would still look correct from the outside. |
| +func TestS3StorePageListerBoundsTheRequest(t *testing.T) { |
| + var gotStartAfter, gotMaxKeys, gotPrefix string |
| + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + q := r.URL.Query() |
| + gotStartAfter, gotMaxKeys, gotPrefix = q.Get("start-after"), q.Get("max-keys"), q.Get("prefix") |
| + w.Header().Set("Content-Type", "application/xml") |
| + fmt.Fprint(w, `<?xml version="1.0" encoding="UTF-8"?> |
| +<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"> |
| + <Name>b</Name><KeyCount>2</KeyCount><IsTruncated>true</IsTruncated> |
| + <Contents><Key>q/d</Key><Size>1</Size></Contents> |
| + <Contents><Key>q/e</Key><Size>1</Size></Contents> |
| +</ListBucketResult>`) |
| + })) |
| + defer srv.Close() |
| + |
| + rgwEnv(t, srv.URL) |
| + s, err := NewS3Store(context.Background(), "b") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + keys, err := s.ListPage(context.Background(), "q/", "q/c", 2) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if fmt.Sprint(keys) != fmt.Sprint([]string{"q/d", "q/e"}) { |
| + t.Fatalf("keys = %v", keys) |
| + } |
| + if gotStartAfter != "q/c" { |
| + t.Fatalf("start-after = %q, want the caller's cursor on the wire", gotStartAfter) |
| + } |
| + if gotMaxKeys != "2" { |
| + t.Fatalf("max-keys = %q, want the bound on the wire — otherwise the cap is a lie", gotMaxKeys) |
| + } |
| + if gotPrefix != "q/" { |
| + t.Fatalf("prefix = %q", gotPrefix) |
| + } |
| +} |
| + |
| +// An empty cursor sends no start-after at all, rather than an empty one. |
| +func TestS3StorePageListerOmitsEmptyCursor(t *testing.T) { |
| + var sawStartAfter bool |
| + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + _, sawStartAfter = r.URL.Query()["start-after"] |
| + w.Header().Set("Content-Type", "application/xml") |
| + fmt.Fprint(w, `<?xml version="1.0" encoding="UTF-8"?> |
| +<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>b</Name><KeyCount>0</KeyCount></ListBucketResult>`) |
| + })) |
| + defer srv.Close() |
| + |
| + rgwEnv(t, srv.URL) |
| + s, err := NewS3Store(context.Background(), "b") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := s.ListPage(context.Background(), "q/", "", 10); err != nil { |
| + t.Fatal(err) |
| + } |
| + if sawStartAfter { |
| + t.Fatal("an empty cursor was sent as start-after=") |
| + } |
| +} |
diff --git a/internal/release/store.go b/internal/release/store.go| index 3cdb23a1c..8ee6ca5c9 100644 |
| --- a/internal/release/store.go |
| +++ b/internal/release/store.go |
| @@ -188,6 +188,32 @@ var ( |
| _ SizeLister = (*S3Store)(nil) |
| ) |
| +// PageLister is the optional BOUNDED listing. List, ListMod and ListSizes all |
| +// return every key under a prefix, which is right for a sweep that wants the |
| +// whole picture and wrong for a queue drain that wants the oldest few. |
| +// |
| +// The scan queue is the caller. It caps how many items it handles per app per |
| +// tick, and without a bounded listing it would enumerate a flooded app's whole |
| +// queue every 30 seconds regardless of that cap — paying for the listing it |
| +// deliberately declined to act on. |
| +// |
| +// Optional and type-asserted, like ModLister and SizeLister above, NOT a Store |
| +// method: Store has eleven implementations across the real stores, the prefix |
| +// wrapper, fleetapi and seven test fakes, and a new method would break all of |
| +// them for one caller's benefit. Callers degrade to List. |
| +// |
| +// startAfter is EXCLUSIVE and keys come back sorted, so paging is |
| +// "startAfter = the last key I saw". limit <= 0 means no bound. |
| +type PageLister interface { |
| + ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) |
| +} |
| + |
| +// Both real stores answer ListPage; only fakes fall back. |
| +var ( |
| + _ PageLister = (*DirStore)(nil) |
| + _ PageLister = (*S3Store)(nil) |
| +) |
| + |
| // ---- DirStore --------------------------------------------------------------- |
| // DirStore keeps objects as files under a root directory — the test double |
| @@ -286,6 +312,26 @@ func (d *DirStore) List(ctx context.Context, prefix string) ([]string, error) { |
| return out, err |
| } |
| +// ListPage is the bounded listing. It leans on ListMod being key-sorted, so |
| +// "after" and "limit" are a filter and a slice rather than a second walk. |
| +func (d *DirStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) { |
| + kms, err := d.ListMod(ctx, prefix) |
| + if err != nil { |
| + return nil, err |
| + } |
| + out := make([]string, 0, len(kms)) |
| + for _, km := range kms { |
| + if km.Key <= startAfter { |
| + continue |
| + } |
| + out = append(out, km.Key) |
| + if limit > 0 && len(out) == limit { |
| + break |
| + } |
| + } |
| + return out, nil |
| +} |
| + |
| // ListMod is the listing with each file's mtime alongside, key-sorted. |
| func (d *DirStore) ListMod(_ context.Context, prefix string) ([]KeyMod, error) { |
| d.mu.Lock() |
| @@ -550,6 +596,31 @@ func (s *S3Store) List(ctx context.Context, prefix string) ([]string, error) { |
| return out, nil |
| } |
| +// ListPage is the bounded listing, and unlike its siblings it does NOT |
| +// paginate: one request, at most limit keys, which is the whole point. |
| +// |
| +// S3 returns keys in lexicographic order and StartAfter is exclusive, so this |
| +// maps onto the API directly. |
| +func (s *S3Store) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) { |
| + in := &awss3.ListObjectsV2Input{Bucket: &s.Bucket, Prefix: &prefix} |
| + if startAfter != "" { |
| + in.StartAfter = &startAfter |
| + } |
| + if limit > 0 { |
| + n := int32(limit) |
| + in.MaxKeys = &n |
| + } |
| + out, err := s.Client.ListObjectsV2(ctx, in) |
| + if err != nil { |
| + return nil, err |
| + } |
| + keys := make([]string, 0, len(out.Contents)) |
| + for _, o := range out.Contents { |
| + keys = append(keys, aws.ToString(o.Key)) |
| + } |
| + return keys, nil |
| +} |
| + |
| // ListMod is the listing keeping the LastModified S3 already sends with |
| // every entry. |
| func (s *S3Store) ListMod(ctx context.Context, prefix string) ([]KeyMod, error) { |