diff --git a/go.mod b/go.mod index 4b1e947b7..c4a9145bd 100644 --- a/go.mod +++ b/go.mod @@ -23,7 +23,7 @@ require ( github.com/miekg/dns v1.1.73 github.com/mulgadc/bluebottle v1.18.1-0.20260827045410-e7d5cf022748 github.com/mulgadc/northstar v1.18.0 - github.com/mulgadc/predastore v1.18.1-0.20260828045402-1dba12d04866 + github.com/mulgadc/predastore v1.18.1-0.20260831064140-ff6a1dc42441 github.com/mulgadc/viperblock v1.18.0 github.com/nats-io/nats-server/v2 v2.14.5 github.com/nats-io/nats.go v1.53.1 diff --git a/go.sum b/go.sum index 1d026a1f2..f3658133d 100644 --- a/go.sum +++ b/go.sum @@ -1331,8 +1331,8 @@ github.com/mulgadc/bluebottle v1.18.1-0.20260827045410-e7d5cf022748 h1:kvoYwPtGX github.com/mulgadc/bluebottle v1.18.1-0.20260827045410-e7d5cf022748/go.mod h1:mvvV4d2w+kGLXxO1ymptNlxqUlNJKB2oAhk6PkpVLj8= github.com/mulgadc/northstar v1.18.0 h1:ksRjIKE6qtrw/ZoYaJh8JRBh+xdnuDw8FsmFLFCiXbg= github.com/mulgadc/northstar v1.18.0/go.mod h1:z6iol77R1vsIQyIBc1JQboUPhSJ6eG4bipzfbZZbLEM= -github.com/mulgadc/predastore v1.18.1-0.20260828045402-1dba12d04866 h1:xR8X8FR2vHxjT9y47zT4/ka40d3uLD+qCzwi4huiAzQ= -github.com/mulgadc/predastore v1.18.1-0.20260828045402-1dba12d04866/go.mod h1:Az85JvrDqSk5pIF0A3Y06GwCmx53tktRSXEUHZIlsZ0= +github.com/mulgadc/predastore v1.18.1-0.20260831064140-ff6a1dc42441 h1:5t+iuKpnRWXkKOmh5lLBoziBaWD+H71edWO/O4ceMp8= +github.com/mulgadc/predastore v1.18.1-0.20260831064140-ff6a1dc42441/go.mod h1:ioGP2kHWYpW/opboAqlqQ5C2Jv+a1/z4txtenw7K7Ws= github.com/mulgadc/viperblock v1.18.0 h1:u6yu5lPfZ+RSmfzlg1OXI3whlRBnGp8t685xUzzxlns= github.com/mulgadc/viperblock v1.18.0/go.mod h1:OjqMFZba0TJl9zcW2oKmz2/MrApIxRIFNcafiv5vTkM= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= diff --git a/scripts/diff-coverage.sh b/scripts/diff-coverage.sh index 7b1746a14..aa5dfaff2 100755 --- a/scripts/diff-coverage.sh +++ b/scripts/diff-coverage.sh @@ -70,7 +70,10 @@ trap 'rm -rf "$TMPDIR"' EXIT # -C enables copy detection for the same reason (file split = partial copy). # cmd/installer/ui is excluded: interactive TUI rendering has no meaningful unit # test, and the installer is validated by the ISO hardware run instead. -git diff "${BASE_REF}" HEAD --unified=0 -M -C --diff-filter=AM -- '*.go' ':!*_test.go' ':!spinifex/migrate/*' ':!cmd/installer/ui/*' | awk ' +# spinifex/testutil is excluded because it is fakes and harnesses used only from +# _test.go files. Coverage is recorded per package under test, so a fake another +# package's tests drive reads as zero-covered however hard it is exercised. +git diff "${BASE_REF}" HEAD --unified=0 -M -C --diff-filter=AM -- '*.go' ':!*_test.go' ':!spinifex/migrate/*' ':!cmd/installer/ui/*' ':!spinifex/testutil/*' | awk ' /^\+\+\+ / { file = substr($2, 3) next diff --git a/spinifex/handlers/ec2/key/service_impl.go b/spinifex/handlers/ec2/key/service_impl.go index e4ec64f1a..5d7699448 100644 --- a/spinifex/handlers/ec2/key/service_impl.go +++ b/spinifex/handlers/ec2/key/service_impl.go @@ -388,7 +388,7 @@ func (s *KeyServiceImpl) findKeyPairIdFromKeyName(ctx context.Context, accountID prefix := fmt.Sprintf("keys/%s/", accountID) // List all objects with the keys prefix - result, err := s.store.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + objects, _, err := objectstore.ListAll(ctx, s.store, &s3.ListObjectsV2Input{ Bucket: aws.String(s.bucketName), Prefix: aws.String(prefix), }) @@ -398,8 +398,8 @@ func (s *KeyServiceImpl) findKeyPairIdFromKeyName(ctx context.Context, accountID } // Check each .json metadata file - for _, obj := range result.Contents { - if obj.Key == nil { + for _, obj := range objects { + if obj == nil || obj.Key == nil { continue } @@ -611,7 +611,7 @@ func (s *KeyServiceImpl) DescribeKeyPairs(ctx context.Context, input *ec2.Descri prefix := fmt.Sprintf("keys/%s/", accountID) // List all objects with the keys prefix - result, err := s.store.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + objects, _, err := objectstore.ListAll(ctx, s.store, &s3.ListObjectsV2Input{ Bucket: aws.String(s.bucketName), Prefix: aws.String(prefix), }) @@ -631,8 +631,8 @@ func (s *KeyServiceImpl) DescribeKeyPairs(ctx context.Context, input *ec2.Descri var keyPairs []*ec2.KeyPairInfo // Check each .json metadata file - for _, obj := range result.Contents { - if obj.Key == nil { + for _, obj := range objects { + if obj == nil || obj.Key == nil { continue } diff --git a/spinifex/handlers/ec2/snapshot/delete_snapshot_pagination_test.go b/spinifex/handlers/ec2/snapshot/delete_snapshot_pagination_test.go new file mode 100644 index 000000000..cd53831c7 --- /dev/null +++ b/spinifex/handlers/ec2/snapshot/delete_snapshot_pagination_test.go @@ -0,0 +1,83 @@ +package handlers_ec2_snapshot_test + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/mulgadc/spinifex/spinifex/config" + "github.com/mulgadc/spinifex/spinifex/ebsmetadata" + "github.com/mulgadc/spinifex/spinifex/ebsprovider" + handlers_ec2_snapshot "github.com/mulgadc/spinifex/spinifex/handlers/ec2/snapshot" + "github.com/mulgadc/spinifex/spinifex/testutil/pagedstore" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testAccountID = "111122223333" + +// TestDeleteSnapshot_RemovesEveryObjectAcrossMultiplePages proves DeleteSnapshot +// follows ListObjectsV2's continuation token to exhaustion. Before this fix it +// read only the first page: above the page size it deleted a partial set of +// objects, reported success, and left the rest orphaned under the snapshot +// prefix. +func TestDeleteSnapshot_RemovesEveryObjectAcrossMultiplePages(t *testing.T) { + const pageSize = 3 + store := pagedstore.New(pageSize) + cfg := &config.Config{ + Predastore: config.PredastoreConfig{ + Bucket: "test-bucket", + AccessKey: "test-owner-123", + }, + } + svc := handlers_ec2_snapshot.NewSnapshotServiceImplWithStore(cfg, store, nil) + provider := ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{}) + svc.SetEBSProvider(provider) + + _, err := provider.CreateVolume(context.Background(), ebsprovider.CreateVolumeRequest{ + Versioned: ebsprovider.NewVersioned(), VolumeID: "vol-1", + CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 10 * 1024 * 1024 * 1024}, + AvailabilityZone: "us-east-1a", + }) + require.NoError(t, err) + require.NoError(t, ebsmetadata.NewStore(store, "test-bucket").PutVolume(context.Background(), ebsmetadata.Volume{ + VolumeID: "vol-1", CapacityGiB: 10, AvailabilityZone: "us-east-1a", + })) + + snap, err := svc.CreateSnapshot(context.Background(), &ec2.CreateSnapshotInput{ + VolumeId: aws.String("vol-1"), + }, testAccountID) + require.NoError(t, err) + snapshotID := *snap.SnapshotId + + // Inflate the snapshot's object prefix well past one page: metadata.json + // (written by CreateSnapshot) plus 10 synthetic block objects. + const extraObjects = 10 + for i := range extraObjects { + _, err := store.PutObject(context.Background(), &s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(fmt.Sprintf("%s/block-%03d", snapshotID, i)), + Body: strings.NewReader("x"), + }) + require.NoError(t, err) + } + + store.Calls = 0 + _, err = svc.DeleteSnapshot(context.Background(), &ec2.DeleteSnapshotInput{ + SnapshotId: snap.SnapshotId, + }, testAccountID) + require.NoError(t, err) + + wantPages := (extraObjects + 1 + pageSize - 1) / pageSize + assert.GreaterOrEqual(t, store.Calls, wantPages, "the listing must span more than one page for this test to prove anything") + + remaining, err := store.MemoryObjectStore.ListObjectsV2(context.Background(), &s3.ListObjectsV2Input{ + Bucket: aws.String("test-bucket"), Prefix: aws.String(snapshotID + "/"), + }) + require.NoError(t, err) + assert.Empty(t, remaining.Contents, "every object under the snapshot prefix must be deleted, not just the first page") +} diff --git a/spinifex/handlers/ec2/snapshot/service_impl.go b/spinifex/handlers/ec2/snapshot/service_impl.go index 5d36f7cb4..766ceccab 100644 --- a/spinifex/handlers/ec2/snapshot/service_impl.go +++ b/spinifex/handlers/ec2/snapshot/service_impl.go @@ -508,7 +508,7 @@ func (s *SnapshotServiceImpl) describeSnapshots(ctx context.Context, input *ec2. return nil, errors.New(awserrors.ErrorInvalidParameterValue) } - listResult, err := s.store.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + _, commonPrefixes, err := objectstore.ListAll(ctx, s.store, &s3.ListObjectsV2Input{ Bucket: aws.String(s.config.Predastore.Bucket), Prefix: aws.String("snap-"), Delimiter: aws.String("/"), @@ -526,8 +526,8 @@ func (s *SnapshotServiceImpl) describeSnapshots(ctx context.Context, input *ec2. } var wanted []string - for _, prefix := range listResult.CommonPrefixes { - if prefix.Prefix == nil { + for _, prefix := range commonPrefixes { + if prefix == nil || prefix.Prefix == nil { continue } @@ -734,7 +734,7 @@ func (s *SnapshotServiceImpl) DeleteSnapshot(ctx context.Context, input *ec2.Del return nil, errors.New(awserrors.ErrorServerInternal) } - listResult, err := s.store.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + objects, _, err := objectstore.ListAll(ctx, s.store, &s3.ListObjectsV2Input{ Bucket: aws.String(s.config.Predastore.Bucket), Prefix: aws.String(snapshotID + "/"), }) @@ -743,8 +743,8 @@ func (s *SnapshotServiceImpl) DeleteSnapshot(ctx context.Context, input *ec2.Del return nil, errors.New(awserrors.ErrorServerInternal) } - for _, obj := range listResult.Contents { - if obj.Key == nil { + for _, obj := range objects { + if obj == nil || obj.Key == nil { continue } _, err := s.store.DeleteObject(ctx, &s3.DeleteObjectInput{ diff --git a/spinifex/handlers/ec2/tags/service_impl.go b/spinifex/handlers/ec2/tags/service_impl.go index 5878f0120..d9e01cb37 100644 --- a/spinifex/handlers/ec2/tags/service_impl.go +++ b/spinifex/handlers/ec2/tags/service_impl.go @@ -230,7 +230,7 @@ func (s *TagsServiceImpl) DescribeTags(ctx context.Context, input *ec2.DescribeT var tags []*ec2.TagDescription // List all tag files from S3 scoped to this account - listResult, err := s.store.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + objects, _, err := objectstore.ListAll(ctx, s.store, &s3.ListObjectsV2Input{ Bucket: aws.String(s.config.Predastore.Bucket), Prefix: aws.String(getTagsPrefix(accountID)), }) @@ -240,8 +240,8 @@ func (s *TagsServiceImpl) DescribeTags(ctx context.Context, input *ec2.DescribeT } // Process each tag file - for _, obj := range listResult.Contents { - if obj.Key == nil { + for _, obj := range objects { + if obj == nil || obj.Key == nil { continue } diff --git a/spinifex/handlers/eks/restore_snapshot.go b/spinifex/handlers/eks/restore_snapshot.go index dd6a25f1d..c292519d6 100644 --- a/spinifex/handlers/eks/restore_snapshot.go +++ b/spinifex/handlers/eks/restore_snapshot.go @@ -82,7 +82,7 @@ func resolveLatestSnapshot(ctx context.Context, store objectstore.ObjectStore, a return "", errors.New("eks: no snapshot store configured; pass --snapshot explicitly") } prefix := accountID + "/" + clusterName + "/" - out, err := store.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + objects, _, err := objectstore.ListAll(ctx, store, &s3.ListObjectsV2Input{ Bucket: aws.String(eksBackupsBucket), Prefix: aws.String(prefix), }) @@ -91,7 +91,10 @@ func resolveLatestSnapshot(ctx context.Context, store objectstore.ObjectStore, a } var bestFrequent, bestAny etcdSnapshotKey haveFrequent, haveAny := false, false - for _, obj := range out.Contents { + for _, obj := range objects { + if obj == nil { + continue + } base := strings.TrimPrefix(aws.StringValue(obj.Key), prefix) parsed, ok := parseEtcdSnapshotKey(base) if !ok { diff --git a/spinifex/handlers/eks/restore_snapshot_test.go b/spinifex/handlers/eks/restore_snapshot_test.go index e12be3179..772ec4036 100644 --- a/spinifex/handlers/eks/restore_snapshot_test.go +++ b/spinifex/handlers/eks/restore_snapshot_test.go @@ -3,12 +3,14 @@ package handlers_eks import ( "context" "errors" + "fmt" "strings" "testing" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/s3" "github.com/mulgadc/spinifex/spinifex/objectstore" + "github.com/mulgadc/spinifex/spinifex/testutil/pagedstore" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -84,6 +86,25 @@ func TestResolveLatestSnapshot_NoSnapshotsFound(t *testing.T) { require.Error(t, err) } +// TestResolveLatestSnapshot_MultiPageListingFindsNewest proves the newest +// snapshot is still found when the listing spans more than one page. The +// etcd timestamp format sorts lexicographically, so the newest key sorts +// last and lands on the final page — before this fix, a listing that never +// followed the continuation token would silently pick an older "newest". +func TestResolveLatestSnapshot_MultiPageListingFindsNewest(t *testing.T) { + store := pagedstore.New(2) + const count = 5 + for i := 1; i <= count; i++ { + seedSnapshot(t, store, testAccountID, "alpha", fmt.Sprintf("etcd-frequent-202607%02dT000000Z.snap", i)) + } + + store.Calls = 0 + got, err := resolveLatestSnapshot(context.Background(), store, testAccountID, "alpha") + require.NoError(t, err) + assert.Equal(t, "etcd-frequent-20260705T000000Z.snap", got, "the true newest snapshot, on the final page, must win") + assert.Greater(t, store.Calls, 1, "the listing must span more than one page for this test to prove anything") +} + // restoreSnapshotFixtureMeta builds a single-CP ClusterMeta with everything // RestoreSnapshot needs: a persisted launch template and NLB target-group ARNs. func restoreSnapshotFixtureMeta(name string) *ClusterMeta { diff --git a/spinifex/objectstore/export_test.go b/spinifex/objectstore/export_test.go new file mode 100644 index 000000000..134ec80fc --- /dev/null +++ b/spinifex/objectstore/export_test.go @@ -0,0 +1,8 @@ +//test:in-package — an export_test.go is in-package by definition; that is the +//whole of what it does. The tests using this live in objectstore_test. + +package objectstore + +// MaxListPages exposes maxListPages for the external test package, which +// needs it to assert ListAll gives up after exactly the documented budget. +const MaxListPages = maxListPages diff --git a/spinifex/objectstore/list.go b/spinifex/objectstore/list.go new file mode 100644 index 000000000..83ee9bd45 --- /dev/null +++ b/spinifex/objectstore/list.go @@ -0,0 +1,58 @@ +package objectstore + +import ( + "context" + "fmt" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/s3" +) + +// maxListPages bounds a single listing. It exists only to turn a store that +// keeps claiming truncation into an error rather than a hang; at the default +// page size that is ten million objects, far past any real prefix. +const maxListPages = 10000 + +// ListAll drains a ListObjectsV2 listing to exhaustion, following +// continuation tokens until the store reports IsTruncated false. It returns +// every object and every common prefix seen across all pages, combined, so a +// delimited listing's CommonPrefixes accumulate the same way Contents does. +// +// input is never mutated; each page is requested from a shallow copy with +// only ContinuationToken replaced. +// +// Reading only the first page returns a prefix of the truth with no error, +// which for a caller deleting or selecting by this listing is worse than a +// failure: a partial deletion or an off-by-a-page "latest" pick both look +// like success without this. +func ListAll(ctx context.Context, store ObjectStore, input *s3.ListObjectsV2Input) ([]*s3.Object, []*s3.CommonPrefix, error) { + var contents []*s3.Object + var commonPrefixes []*s3.CommonPrefix + var token *string + + prefix := aws.StringValue(input.Prefix) + + for range maxListPages { + page := *input + page.ContinuationToken = token + + result, err := store.ListObjectsV2(ctx, &page) + if err != nil { + return nil, nil, err + } + + contents = append(contents, result.Contents...) + commonPrefixes = append(commonPrefixes, result.CommonPrefixes...) + + if !aws.BoolValue(result.IsTruncated) { + return contents, commonPrefixes, nil + } + // Truncated with nothing to resume from. Returning what arrived is the + // silent short answer this whole loop exists to prevent. + if aws.StringValue(result.NextContinuationToken) == "" { + return nil, nil, fmt.Errorf("listing %s reported truncation with no continuation token", prefix) + } + token = result.NextContinuationToken + } + return nil, nil, fmt.Errorf("listing %s did not finish within %d pages", prefix, maxListPages) +} diff --git a/spinifex/objectstore/list_test.go b/spinifex/objectstore/list_test.go new file mode 100644 index 000000000..aeeeb785e --- /dev/null +++ b/spinifex/objectstore/list_test.go @@ -0,0 +1,179 @@ +package objectstore_test + +import ( + "context" + "strconv" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/mulgadc/spinifex/spinifex/objectstore" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// pagingStore replays a fixed sequence of ListObjectsV2 pages, honouring +// continuation tokens the way a real paginating store would: each call +// returns the next page and, while pages remain, a token pointing at it. +// +// This is deliberately separate from testutil/pagedstore, which slices a +// real MemoryObjectStore's filtered, sorted contents into pages of a fixed +// size for consumer-side integration tests. These tests are about ListAll's +// own pagination contract: exact page content (including CommonPrefixes, +// which pagedstore does not paginate at all) and the malformed-truncation +// shapes below, neither of which a real store's paging can produce on +// demand. GetObject/PutObject/etc are unused by ListAll, so they come from +// an embedded MemoryObjectStore rather than being hand-stubbed. +type pagingStore struct { + *objectstore.MemoryObjectStore + + pages []*s3.ListObjectsV2Output + calls int +} + +func newPagingStore(pages []*s3.ListObjectsV2Output) *pagingStore { + return &pagingStore{MemoryObjectStore: objectstore.NewMemoryObjectStore(), pages: pages} +} + +func (p *pagingStore) ListObjectsV2(_ context.Context, input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) { + wantToken := "" + if p.calls > 0 { + wantToken = tokenFor(p.calls) + } + if aws.StringValue(input.ContinuationToken) != wantToken { + panic("pagingStore: unexpected continuation token") + } + out := p.pages[p.calls] + p.calls++ + if p.calls < len(p.pages) { + out.IsTruncated = aws.Bool(true) + out.NextContinuationToken = aws.String(tokenFor(p.calls)) + } else { + out.IsTruncated = aws.Bool(false) + out.NextContinuationToken = nil + } + return out, nil +} + +func tokenFor(page int) string { + return "page-" + strconv.Itoa(page) +} + +var _ objectstore.ObjectStore = (*pagingStore)(nil) + +// truncatedNoTokenStore always reports truncation but never supplies a +// continuation token to resume from — the shape a broken or misbehaving +// store could produce. +type truncatedNoTokenStore struct{ *pagingStore } + +func (t *truncatedNoTokenStore) ListObjectsV2(context.Context, *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) { + return &s3.ListObjectsV2Output{IsTruncated: aws.Bool(true)}, nil +} + +// neverEndingStore always claims truncation with a fresh token, modelling a +// server that never finishes a listing. +type neverEndingStore struct{ *pagingStore } + +func (n *neverEndingStore) ListObjectsV2(context.Context, *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) { + n.calls++ + return &s3.ListObjectsV2Output{ + IsTruncated: aws.Bool(true), + NextContinuationToken: aws.String("next"), + }, nil +} + +func TestListAll_SinglePage(t *testing.T) { + store := newPagingStore([]*s3.ListObjectsV2Output{ + {Contents: []*s3.Object{{Key: aws.String("a")}, {Key: aws.String("b")}}}, + }) + + contents, prefixes, err := objectstore.ListAll(context.Background(), store, &s3.ListObjectsV2Input{ + Bucket: aws.String("bucket"), Prefix: aws.String("p/"), + }) + require.NoError(t, err) + assert.Empty(t, prefixes) + require.Len(t, contents, 2) + assert.Equal(t, "a", aws.StringValue(contents[0].Key)) + assert.Equal(t, "b", aws.StringValue(contents[1].Key)) + assert.Equal(t, 1, store.calls) +} + +// A listing larger than one page must follow the continuation token to +// exhaustion rather than stopping after the first page. +func TestListAll_MultiPageFollowsContinuationToken(t *testing.T) { + store := newPagingStore([]*s3.ListObjectsV2Output{ + {Contents: []*s3.Object{{Key: aws.String("a")}, {Key: aws.String("b")}}}, + {Contents: []*s3.Object{{Key: aws.String("c")}}}, + {Contents: []*s3.Object{{Key: aws.String("d")}, {Key: aws.String("e")}}}, + }) + + contents, _, err := objectstore.ListAll(context.Background(), store, &s3.ListObjectsV2Input{ + Bucket: aws.String("bucket"), Prefix: aws.String("p/"), + }) + require.NoError(t, err) + require.Len(t, contents, 5) + var keys []string + for _, obj := range contents { + keys = append(keys, aws.StringValue(obj.Key)) + } + assert.Equal(t, []string{"a", "b", "c", "d", "e"}, keys) + assert.Equal(t, 3, store.calls) +} + +// A delimited listing accumulates CommonPrefixes across pages the same way +// Contents does. +func TestListAll_MultiPageCommonPrefixes(t *testing.T) { + store := newPagingStore([]*s3.ListObjectsV2Output{ + {CommonPrefixes: []*s3.CommonPrefix{{Prefix: aws.String("snap-1/")}}}, + {CommonPrefixes: []*s3.CommonPrefix{{Prefix: aws.String("snap-2/")}}}, + }) + + _, prefixes, err := objectstore.ListAll(context.Background(), store, &s3.ListObjectsV2Input{ + Bucket: aws.String("bucket"), Prefix: aws.String("snap-"), Delimiter: aws.String("/"), + }) + require.NoError(t, err) + require.Len(t, prefixes, 2) + assert.Equal(t, "snap-1/", aws.StringValue(prefixes[0].Prefix)) + assert.Equal(t, "snap-2/", aws.StringValue(prefixes[1].Prefix)) +} + +func TestListAll_ListError(t *testing.T) { + store := &erroringStore{pagingStore: newPagingStore(nil), err: assert.AnError} + _, _, err := objectstore.ListAll(context.Background(), store, &s3.ListObjectsV2Input{ + Bucket: aws.String("bucket"), Prefix: aws.String("p/"), + }) + require.Error(t, err) +} + +type erroringStore struct { + *pagingStore + + err error +} + +func (e *erroringStore) ListObjectsV2(context.Context, *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) { + return nil, e.err +} + +// Truncation reported with no continuation token to resume from must fail +// rather than silently returning the partial page it already read. +func TestListAll_TruncatedWithNoTokenErrors(t *testing.T) { + store := &truncatedNoTokenStore{pagingStore: newPagingStore(nil)} + _, _, err := objectstore.ListAll(context.Background(), store, &s3.ListObjectsV2Input{ + Bucket: aws.String("bucket"), Prefix: aws.String("p/"), + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "reported truncation with no continuation token") +} + +// A store that never stops claiming truncation must fail once the page +// budget is exhausted rather than looping forever. +func TestListAll_NeverEndingTruncationErrors(t *testing.T) { + store := &neverEndingStore{pagingStore: newPagingStore(nil)} + _, _, err := objectstore.ListAll(context.Background(), store, &s3.ListObjectsV2Input{ + Bucket: aws.String("bucket"), Prefix: aws.String("p/"), + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "did not finish within") + assert.Equal(t, objectstore.MaxListPages, store.calls) +} diff --git a/spinifex/testutil/ebsfake/ebsfake.go b/spinifex/testutil/ebsfake/ebsfake.go index 55e096b40..cfe8345d4 100644 --- a/spinifex/testutil/ebsfake/ebsfake.go +++ b/spinifex/testutil/ebsfake/ebsfake.go @@ -35,13 +35,16 @@ func (p *Provider) DeleteVolume(ctx context.Context, req ebsprovider.DeleteVolum return err } for _, prefix := range []string{req.VolumeID + "-efi/", req.VolumeID + "/"} { - listed, err := p.store.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + objects, _, err := objectstore.ListAll(ctx, p.store, &s3.ListObjectsV2Input{ Bucket: aws.String(p.bucket), Prefix: aws.String(prefix), }) if err != nil { return err } - for _, obj := range listed.Contents { + for _, obj := range objects { + if obj == nil { + continue + } if _, err := p.store.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(p.bucket), Key: obj.Key}); err != nil { return err } diff --git a/spinifex/testutil/pagedstore/pagedstore.go b/spinifex/testutil/pagedstore/pagedstore.go new file mode 100644 index 000000000..baa3f9ae4 --- /dev/null +++ b/spinifex/testutil/pagedstore/pagedstore.go @@ -0,0 +1,78 @@ +// Package pagedstore provides an objectstore.ObjectStore fake that serves +// ListObjectsV2 across multiple deterministic pages, honouring +// ContinuationToken and IsTruncated the way a real paginating S3-compatible +// store does. It exists so tests can prove a caller follows continuation +// tokens to exhaustion instead of stopping after the first page. +package pagedstore + +import ( + "context" + "sort" + "strconv" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/mulgadc/spinifex/spinifex/objectstore" +) + +// Store wraps a MemoryObjectStore and splits ListObjectsV2's answer into +// pages of at most PageSize objects, sorted by key so pagination is +// deterministic across runs (MemoryObjectStore itself iterates a map). +type Store struct { + *objectstore.MemoryObjectStore + + PageSize int + + // Calls counts ListObjectsV2 invocations, so tests can assert every page + // was actually walked rather than the loop stopping early by luck. + Calls int +} + +// New returns a paging store with the given per-page object limit. +func New(pageSize int) *Store { + return &Store{MemoryObjectStore: objectstore.NewMemoryObjectStore(), PageSize: pageSize} +} + +var _ objectstore.ObjectStore = (*Store)(nil) + +// ListObjectsV2 delegates to MemoryObjectStore for prefix/delimiter matching, +// then slices the sorted result into one page per call, resuming from the +// offset encoded in ContinuationToken. +func (s *Store) ListObjectsV2(ctx context.Context, input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) { + s.Calls++ + + full, err := s.MemoryObjectStore.ListObjectsV2(ctx, input) + if err != nil { + return nil, err + } + + sort.Slice(full.Contents, func(i, j int) bool { + return aws.StringValue(full.Contents[i].Key) < aws.StringValue(full.Contents[j].Key) + }) + + start := 0 + if tok := aws.StringValue(input.ContinuationToken); tok != "" { + start, err = strconv.Atoi(tok) + if err != nil { + return nil, err + } + } + end := min(start+s.PageSize, len(full.Contents)) + if start > end { + start = end + } + + page := &s3.ListObjectsV2Output{ + Contents: full.Contents[start:end], + CommonPrefixes: full.CommonPrefixes, + Name: input.Bucket, + KeyCount: aws.Int64(int64(end - start)), + } + if end < len(full.Contents) { + page.IsTruncated = aws.Bool(true) + page.NextContinuationToken = aws.String(strconv.Itoa(end)) + } else { + page.IsTruncated = aws.Bool(false) + } + return page, nil +}