Make an S3 listing the server cuts short an error #333
+29
-2
@@ -18,6 +18,13 @@ import (
|
|||||||
"github.com/aws/smithy-go/logging"
|
"github.com/aws/smithy-go/logging"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// ErrNoContinuationToken is returned by ListObjects and ListObjectsStream
|
||||||
|
// when the server marks a page of a listing truncated but sends no
|
||||||
|
// continuation token to fetch the rest with, as a server answering V2
|
||||||
|
// listings V1-style does.
|
||||||
|
var ErrNoContinuationToken = errors.New(
|
||||||
|
"server marked the listing truncated but sent no continuation token")
|
||||||
|
|
||||||
// Client wraps the AWS S3 client for vaultik operations.
|
// Client wraps the AWS S3 client for vaultik operations.
|
||||||
// It provides a simplified interface for S3 operations with automatic
|
// It provides a simplified interface for S3 operations with automatic
|
||||||
// prefix handling and connection management. All operations are performed
|
// prefix handling and connection management. All operations are performed
|
||||||
@@ -214,7 +221,7 @@ func (c *Client) ListObjects(ctx context.Context, prefix string) ([]string, erro
|
|||||||
})
|
})
|
||||||
|
|
||||||
for paginator.HasMorePages() {
|
for paginator.HasMorePages() {
|
||||||
page, err := paginator.NextPage(ctx)
|
page, err := nextListPage(ctx, paginator)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -235,6 +242,26 @@ func (c *Client) ListObjects(ctx context.Context, prefix string) ([]string, erro
|
|||||||
return keys, nil
|
return keys, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// nextListPage fetches the next page of a listing. The SDK paginator takes
|
||||||
|
// a page marked truncated without a continuation token for the last page
|
||||||
|
// and stops without an error, so such a page is ErrNoContinuationToken
|
||||||
|
// here instead.
|
||||||
|
func nextListPage(
|
||||||
|
ctx context.Context, paginator *s3.ListObjectsV2Paginator,
|
||||||
|
) (*s3.ListObjectsV2Output, error) {
|
||||||
|
page, err := paginator.NextPage(ctx)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
truncated := aws.ToBool(page.IsTruncated)
|
||||||
|
if truncated && aws.ToString(page.NextContinuationToken) == "" {
|
||||||
|
return nil, ErrNoContinuationToken
|
||||||
|
}
|
||||||
|
|
||||||
|
return page, nil
|
||||||
|
}
|
||||||
|
|
||||||
// HeadObject checks if an object exists in S3.
|
// HeadObject checks if an object exists in S3.
|
||||||
// Returns true if the object exists, false otherwise.
|
// Returns true if the object exists, false otherwise.
|
||||||
// The key is automatically prefixed with the configured prefix.
|
// The key is automatically prefixed with the configured prefix.
|
||||||
@@ -300,7 +327,7 @@ func (c *Client) ListObjectsStream(
|
|||||||
})
|
})
|
||||||
|
|
||||||
for paginator.HasMorePages() {
|
for paginator.HasMorePages() {
|
||||||
page, err := paginator.NextPage(ctx)
|
page, err := nextListPage(ctx, paginator)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
ch <- ObjectInfo{Err: err}
|
ch <- ObjectInfo{Err: err}
|
||||||
|
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"slices"
|
"slices"
|
||||||
@@ -22,13 +23,16 @@ import (
|
|||||||
// s3TestBucket is the bucket created for each in-process S3 server.
|
// s3TestBucket is the bucket created for each in-process S3 server.
|
||||||
const s3TestBucket = "test-bucket"
|
const s3TestBucket = "test-bucket"
|
||||||
|
|
||||||
// Credentials for the tests that build a storer from a config.Config. The
|
// Credentials for the tests' S3 clients. The test servers accept any.
|
||||||
// in-process S3 server accepts any.
|
|
||||||
const (
|
const (
|
||||||
s3TestAccessKeyID = "key"
|
s3TestAccessKeyID = "key"
|
||||||
s3TestSecretAccessKey = "secret"
|
s3TestSecretAccessKey = "secret"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// s3TestRegion is the region of the S3 clients built here with
|
||||||
|
// s3.NewClient. The test servers ignore it.
|
||||||
|
const s3TestRegion = "us-east-1"
|
||||||
|
|
||||||
// newS3Storer builds an s3:// backend backed by a fresh in-process
|
// newS3Storer builds an s3:// backend backed by a fresh in-process
|
||||||
// S3 server. It reuses the same in-memory S3 harness (gofakes3 + s3mem
|
// S3 server. It reuses the same in-memory S3 harness (gofakes3 + s3mem
|
||||||
// over httptest) that internal/s3 and the not-found regression test use,
|
// over httptest) that internal/s3 and the not-found regression test use,
|
||||||
@@ -55,7 +59,7 @@ func newS3Storer(t *testing.T) storage.Storer {
|
|||||||
Bucket: s3TestBucket,
|
Bucket: s3TestBucket,
|
||||||
AccessKeyID: "test",
|
AccessKeyID: "test",
|
||||||
SecretAccessKey: "test",
|
SecretAccessKey: "test",
|
||||||
Region: "us-east-1",
|
Region: s3TestRegion,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("new client: %v", err)
|
t.Fatalf("new client: %v", err)
|
||||||
@@ -93,6 +97,68 @@ func TestS3StorerMissingKeyMapsToErrNotFound(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestS3ListingCutShortIsAnError checks that List and ListStream fail on a
|
||||||
|
// listing page marked truncated that carries no continuation token, as a
|
||||||
|
// server answering V2 listings V1-style sends it. Taken as the last page,
|
||||||
|
// it would hide every later key, and prune would treat the blobs of the
|
||||||
|
// snapshots listed there as unreferenced.
|
||||||
|
func TestS3ListingCutShortIsAnError(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
const (
|
||||||
|
prefix = "metadata/"
|
||||||
|
truncatedPage = `<?xml version="1.0" encoding="UTF-8"?>
|
||||||
|
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
|
||||||
|
<Name>test-bucket</Name>
|
||||||
|
<Prefix>metadata/</Prefix>
|
||||||
|
<KeyCount>1</KeyCount>
|
||||||
|
<MaxKeys>1</MaxKeys>
|
||||||
|
<IsTruncated>true</IsTruncated>
|
||||||
|
<Contents>
|
||||||
|
<Key>metadata/snap/manifest.json.zst</Key>
|
||||||
|
<Size>8</Size>
|
||||||
|
</Contents>
|
||||||
|
</ListBucketResult>`
|
||||||
|
)
|
||||||
|
|
||||||
|
srv := httptest.NewServer(http.HandlerFunc(
|
||||||
|
func(w http.ResponseWriter, _ *http.Request) {
|
||||||
|
_, _ = io.WriteString(w, truncatedPage)
|
||||||
|
}))
|
||||||
|
t.Cleanup(srv.Close)
|
||||||
|
|
||||||
|
client, err := s3.NewClient(context.Background(), s3.Config{
|
||||||
|
Endpoint: srv.URL,
|
||||||
|
Bucket: s3TestBucket,
|
||||||
|
AccessKeyID: s3TestAccessKeyID,
|
||||||
|
SecretAccessKey: s3TestSecretAccessKey,
|
||||||
|
Region: s3TestRegion,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("new client: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
storer := storage.NewS3Storer(client)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
keys, err := storer.List(ctx, prefix)
|
||||||
|
if !errors.Is(err, s3.ErrNoContinuationToken) {
|
||||||
|
t.Errorf("List = %q, %v; want ErrNoContinuationToken", keys, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var streamErr error
|
||||||
|
|
||||||
|
for obj := range storer.ListStream(ctx, prefix) {
|
||||||
|
if obj.Err != nil {
|
||||||
|
streamErr = obj.Err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if !errors.Is(streamErr, s3.ErrNoContinuationToken) {
|
||||||
|
t.Errorf("ListStream error = %v, want ErrNoContinuationToken", streamErr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestS3URLPrefixKeyLayout pins the bucket keys an s3:// URL reads and
|
// TestS3URLPrefixKeyLayout pins the bucket keys an s3:// URL reads and
|
||||||
// writes: the README's remote storage layout, with the prefix joined to
|
// writes: the README's remote storage layout, with the prefix joined to
|
||||||
// each key by one "/". s3://b/p and s3://b/p/ must be the same
|
// each key by one "/". s3://b/p and s3://b/p/ must be the same
|
||||||
|
|||||||
Reference in New Issue
Block a user