diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index db382fbe5..a3f2c0159 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -29,7 +29,7 @@ permissions: env: GIT_BRANCH: ${{ github.ref_name }} - SUITES: ${{ inputs.suites || 'e2e-single e2e-multinode e2e-cert e2e-lb e2e-iam' }} + SUITES: ${{ inputs.suites || 'e2e-single e2e-multinode e2e-cert e2e-lb e2e-iam e2e-partialblock' }} TEST_RUN: ${{ inputs.test_run || '' }} # OTLP export to the central Elastic sink (wattle). Repo variable overrides; # empty string disables telemetry entirely. @@ -78,11 +78,12 @@ jobs: # here, before this job provisions a tofu env or pays the Spinifex install. # # The dispatch check gates on exactly the suites this job's Run step - # allow-lists (single|iam|cert|eks|ecs|storagegrowth). Gating on a suite the - # job won't run — lb, which only executes in the multinode job — would - # provision a tofu VM and pay the Spinifex install only to run nothing and - # tear down. Push triggers keep running via the null-check on inputs.suites. - if: needs.integration.result == 'success' && (github.event_name != 'workflow_dispatch' || contains(github.event.inputs.suites, 'e2e-single') || contains(github.event.inputs.suites, 'e2e-iam') || contains(github.event.inputs.suites, 'e2e-cert') || contains(github.event.inputs.suites, 'e2e-eks') || contains(github.event.inputs.suites, 'e2e-ecs') || contains(github.event.inputs.suites, 'e2e-storagegrowth')) + # allow-lists (single|iam|cert|eks|ecs|storagegrowth|partialblock). Gating + # on a suite the job won't run — lb, which only executes in the multinode + # job — would provision a tofu VM and pay the Spinifex install only to run + # nothing and tear down. Push triggers keep running via the null-check on + # inputs.suites. + if: needs.integration.result == 'success' && (github.event_name != 'workflow_dispatch' || contains(github.event.inputs.suites, 'e2e-single') || contains(github.event.inputs.suites, 'e2e-iam') || contains(github.event.inputs.suites, 'e2e-cert') || contains(github.event.inputs.suites, 'e2e-eks') || contains(github.event.inputs.suites, 'e2e-ecs') || contains(github.event.inputs.suites, 'e2e-storagegrowth') || contains(github.event.inputs.suites, 'e2e-partialblock')) runs-on: [self-hosted, ci-single] timeout-minutes: 50 env: @@ -318,7 +319,13 @@ jobs: # Allow-list, not a deny-list: a suite runs here only if named. Deny-lists # silently fan a newly added suite out to every job, which is how lb ended up # running twice per e2e run. - SUITES_RUN=$(echo "${SUITES}" | sed 's/e2e-//g' | tr ' ' '\n' | grep -xE 'single|iam|cert|eks|ecs|storagegrowth' | tr '\n' ' ') + # + # partialblock gates merges because it earned it, not because it is + # new: on one node, same binary and workload, it fails against the + # pre-fix engine (1758 of 8192 halves reverted a generation) and + # passes against the fixed one (0 of 8192). A suite that has never + # been shown to go red for the reason it claims does not belong here. + SUITES_RUN=$(echo "${SUITES}" | sed 's/e2e-//g' | tr ' ' '\n' | grep -xE 'single|iam|cert|eks|ecs|storagegrowth|partialblock' | tr '\n' ' ') if [ -z "$(echo "${SUITES_RUN}" | tr -d ' ')" ]; then echo "no in-VM suites requested (SUITES=${SUITES}); skipping" echo "rc=0" >> "$GITHUB_OUTPUT" diff --git a/spinifex/handlers/ec2/volume/service_impl.go b/spinifex/handlers/ec2/volume/service_impl.go index 9f9792daa..62d38088c 100644 --- a/spinifex/handlers/ec2/volume/service_impl.go +++ b/spinifex/handlers/ec2/volume/service_impl.go @@ -17,6 +17,7 @@ import ( "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/predastore/pkg/masterkey" "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/filterutil" @@ -202,29 +203,9 @@ func (s *VolumeServiceImpl) CreateVolume(ctx context.Context, input *ec2.CreateV return nil, errors.New(awserrors.ErrorServerInternal) } - // GCEnabled: default false unless explicitly set to true, matching the - // nbdkit plugin and viperblockd resolution of the same config field. - gcEnabled := s.config.Viperblock.GCEnabled != nil && *s.config.Viperblock.GCEnabled + vbconfig := s.buildVBConfig(volumeID, volumeSizeBytes, volumeConfig, mkey, snapshotID, sourceVolumeName) - vbconfig := viperblock.VB{ - VolumeName: volumeID, - VolumeSize: volumeSizeBytes, - BaseDir: s.config.WalDir, - Cache: viperblock.Cache{Config: viperblock.CacheConfig{Size: 0}}, - VolumeConfig: volumeConfig, - MasterKey: mkey, - EncryptionEnabled: mkey != nil, - GCEnabled: gcEnabled, - } - - // If created from a snapshot, set the snapshot fields so viperblock's - // LoadState will call OpenFromSnapshot to load the base block map. - if snapshotID != "" { - vbconfig.SnapshotID = snapshotID - vbconfig.SourceVolumeName = sourceVolumeName - } - - vb, err := viperblock.New(&vbconfig, "s3", cfg) + vb, err := viperblock.New(vbconfig, "s3", cfg) if err != nil { slog.ErrorContext(ctx, "CreateVolume failed to create viperblock instance", "err", err) return nil, errors.New(awserrors.ErrorServerInternal) @@ -269,6 +250,36 @@ func (s *VolumeServiceImpl) CreateVolume(ctx context.Context, input *ec2.CreateV return vol, nil } +// buildVBConfig assembles the viperblock.VB config CreateVolume hands to +// viperblock.New. Split out from CreateVolume so the GC-enablement wiring can +// be asserted directly against the returned config, without needing a live +// viperblock/S3 backend to observe it: GCEnabled is not part of +// viperblock.VBState, so it never round-trips through a saved config.json and +// can't be checked by reading volume state back. +func (s *VolumeServiceImpl) buildVBConfig(volumeID string, volumeSizeBytes uint64, volumeConfig viperblock.VolumeConfig, mkey *masterkey.Key, snapshotID, sourceVolumeName string) *viperblock.VB { + vbconfig := &viperblock.VB{ + VolumeName: volumeID, + VolumeSize: volumeSizeBytes, + BaseDir: s.config.WalDir, + Cache: viperblock.Cache{Config: viperblock.CacheConfig{Size: 0}}, + VolumeConfig: volumeConfig, + MasterKey: mkey, + EncryptionEnabled: mkey != nil, + // GCEnabled: default false unless explicitly set to true, matching the + // nbdkit plugin and viperblockd resolution of the same config field. + GCEnabled: s.config.Viperblock.GCEnabled != nil && *s.config.Viperblock.GCEnabled, + } + + // If created from a snapshot, set the snapshot fields so viperblock's + // LoadState will call OpenFromSnapshot to load the base block map. + if snapshotID != "" { + vbconfig.SnapshotID = snapshotID + vbconfig.SourceVolumeName = sourceVolumeName + } + + return vbconfig +} + // describeVolumesValidFilters defines the set of filter names accepted by DescribeVolumes. var describeVolumesValidFilters = map[string]bool{ "volume-id": true, diff --git a/spinifex/handlers/ec2/volume/service_impl_test.go b/spinifex/handlers/ec2/volume/service_impl_test.go index d803f5e0b..e66c91636 100644 --- a/spinifex/handlers/ec2/volume/service_impl_test.go +++ b/spinifex/handlers/ec2/volume/service_impl_test.go @@ -233,6 +233,40 @@ func TestCreateVolume_PassesValidation(t *testing.T) { } } +// TestCreateVolume_BuildVBConfig_GCEnabled asserts that CreateVolume's +// viperblock config carries GCEnabled through from spinifex.toml's +// [viperblock] gc_enabled key (config.Config.Viperblock.GCEnabled). This is +// the config-level seam: GCEnabled isn't part of viperblock.VBState, so it +// never persists to config.json and can't be observed by reading a created +// volume back — the only place it's checkable is the config CreateVolume +// hands to viperblock.New, before construction ever reaches the backend. +// A regression that unwires GC in buildVBConfig, or a future refactor of +// CreateVolume that stops routing through buildVBConfig, fails this test. +func TestCreateVolume_BuildVBConfig_GCEnabled(t *testing.T) { + tests := []struct { + name string + gcEnabled *bool + want bool + }{ + {name: "NilDefaultsToDisabled", gcEnabled: nil, want: false}, + {name: "ExplicitFalse", gcEnabled: aws.Bool(false), want: false}, + {name: "ExplicitTrue", gcEnabled: aws.Bool(true), want: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + svc := newTestVolumeService("ap-southeast-2a") + svc.config.Viperblock.GCEnabled = tt.gcEnabled + + vbconfig := svc.buildVBConfig("vol-gc-test", 10*1024*1024*1024, + viperblock.VolumeConfig{}, nil, "", "") + + assert.Equal(t, tt.want, vbconfig.GCEnabled, + "buildVBConfig GCEnabled must follow config.Viperblock.GCEnabled") + }) + } +} + func TestDeleteVolume_Validation(t *testing.T) { tests := []struct { name string diff --git a/tests/e2e/Makefile b/tests/e2e/Makefile index e02fd58dd..23c99cb77 100644 --- a/tests/e2e/Makefile +++ b/tests/e2e/Makefile @@ -18,9 +18,9 @@ SUITE_TAG ?= $(notdir $(patsubst %/...,%,$(firstword $(PKG)))) # named suite into BIN_DIR/.test so CI can ship binaries (not source + # go toolchain) to the remote VM. BIN_DIR ?= $(CURDIR)/_bin -SUITES_BUILD ?= cert lb eks ecs single iam ecr bedrock gpu natuplink multinode storagegrowth +SUITES_BUILD ?= cert lb eks ecs single iam ecr bedrock gpu natuplink multinode storagegrowth partialblock -.PHONY: e2e e2e-cert e2e-lb e2e-eks e2e-ecs e2e-single e2e-iam e2e-ecr e2e-bedrock e2e-multinode e2e-bm e2e-reboot e2e-tofu e2e-gpu e2e-natuplink e2e-storagegrowth e2e-all clean e2e-build +.PHONY: e2e e2e-cert e2e-lb e2e-eks e2e-ecs e2e-single e2e-iam e2e-ecr e2e-bedrock e2e-multinode e2e-bm e2e-reboot e2e-tofu e2e-gpu e2e-natuplink e2e-storagegrowth e2e-partialblock e2e-all clean e2e-build RUN_FLAG = $(if $(RUN),-run '$(RUN)',) @@ -56,6 +56,7 @@ e2e-tofu: ; $(MAKE) e2e PKG=./tofu/... e2e-gpu: ; $(MAKE) e2e PKG=./gpu/... e2e-natuplink: ; $(MAKE) e2e PKG=./natuplink/... e2e-storagegrowth: ; $(MAKE) e2e PKG=./storagegrowth/... +e2e-partialblock: ; $(MAKE) e2e PKG=./partialblock/... e2e-all: ; $(MAKE) e2e e2e-build: diff --git a/tests/e2e/gpu/main_test.go b/tests/e2e/gpu/main_test.go index 02361efb2..3f6245853 100644 --- a/tests/e2e/gpu/main_test.go +++ b/tests/e2e/gpu/main_test.go @@ -3,6 +3,7 @@ package gpu import ( + "fmt" "os" "sync" "testing" @@ -23,7 +24,12 @@ func TestMain(m *testing.M) { code := m.Run() if pkgFix != nil { if pkgFix.Harness != nil { - pkgFix.Harness.Close() + // A leaked resource fails the run: the suite may have passed, but it + // left state on the node that the next run will trip over. + if err := pkgFix.Harness.Close(); err != nil { + fmt.Fprintf(os.Stderr, "e2e teardown: %v\n", err) + code = 1 + } } if pkgFix.TmpDir != "" { _ = os.RemoveAll(pkgFix.TmpDir) diff --git a/tests/e2e/harness/ec2helpers.go b/tests/e2e/harness/ec2helpers.go index 67ec1827e..0d7727064 100644 --- a/tests/e2e/harness/ec2helpers.go +++ b/tests/e2e/harness/ec2helpers.go @@ -4,6 +4,7 @@ package harness import ( "bytes" + "fmt" "os/exec" "strconv" "strings" @@ -13,6 +14,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/service/ec2" + "github.com/aws/aws-sdk-go/service/ec2/ec2iface" ) // SSHTarget is the minimal addressing tuple used by LsblkRootGiB. Unlike @@ -261,33 +263,132 @@ func DetachVolumeWait(t *testing.T, c *AWSClient, volID string) { WaitForVolumeState(t, c, volID, "available", WithPoll(500*time.Millisecond)) } -// RegisterVolumeTeardown best-effort force-detaches then deletes volID at test -// end. Non-fatal — it polls for "available" within a bounded window and never -// calls t.Fatal so a cleanup hiccup doesn't mask the test's real result. +// RegisterVolumeTeardown force-detaches then deletes volID at test end. It does +// not call t.Fatal, so a cleanup hiccup can't mask the test's real result, but a +// teardown that actually leaves the volume behind fails the test — a silent leak +// is worse than a noisy one. func RegisterVolumeTeardown(t *testing.T, c *AWSClient, volID string) { t.Helper() t.Cleanup(func() { - _, _ = c.EC2.DetachVolume(&ec2.DetachVolumeInput{ - VolumeId: aws.String(volID), - Force: aws.Bool(true), - }) - deadline := time.Now().Add(90 * time.Second) - for time.Now().Before(deadline) { - out, err := c.EC2.DescribeVolumes(&ec2.DescribeVolumesInput{ - VolumeIds: []*string{aws.String(volID)}, - }) - if err != nil || len(out.Volumes) == 0 { - return - } - if aws.StringValue(out.Volumes[0].State) == "available" { - break - } - time.Sleep(2 * time.Second) + if err := teardownVolume(c.EC2, volID); err != nil { + t.Errorf("LEAKED volume %s: %v", volID, err) } - _, _ = c.EC2.DeleteVolume(&ec2.DeleteVolumeInput{VolumeId: aws.String(volID)}) }) } +// Teardown windows. Detach and terminate are both asynchronous, and a volume is +// only released once the guest holding it is down, so these have to outlast a +// slow QEMU shutdown rather than a single API round trip. +const ( + volumeTeardownTimeout = 90 * time.Second + instanceTeardownTimeout = 3 * time.Minute + teardownPollInterval = 2 * time.Second +) + +// teardownVolume force-detaches volID if it is still attached, waits for it to +// reach "available", then deletes it. It returns an error only when the volume +// is still present afterwards — that is, only when it has genuinely leaked. +// +// Deletion has to be self-sufficient rather than rely on teardown ordering. +// Fixture cleanups drain LIFO, so a suite that creates its instance before its +// volume tears the volume down first, while it is still attached: DeleteVolume +// answers VolumeInUse and the volume is stranded. Registration order cannot +// express that dependency, so the teardown resolves it itself and works for any +// creation order. +func teardownVolume(ec2c ec2iface.EC2API, volID string) error { + forced := false + deadline := time.Now().Add(volumeTeardownTimeout) + for { + out, err := ec2c.DescribeVolumes(&ec2.DescribeVolumesInput{ + VolumeIds: []*string{aws.String(volID)}, + }) + if err != nil { + if volumeGone(err) { + return nil + } + return fmt.Errorf("describe volume %s: %w", volID, err) + } + if len(out.Volumes) == 0 { + return nil + } + + switch state := aws.StringValue(out.Volumes[0].State); state { + case ec2.VolumeStateDeleting, ec2.VolumeStateDeleted: + return nil + case ec2.VolumeStateAvailable: + if _, err := ec2c.DeleteVolume(&ec2.DeleteVolumeInput{ + VolumeId: aws.String(volID), + }); err != nil && !volumeGone(err) { + return fmt.Errorf("delete volume %s: %w", volID, err) + } + return nil + default: + // Force, because by teardown the attaching guest is usually already + // terminating and there is nothing left to acknowledge a graceful + // detach. Issued once, then we wait for the state to settle. + if !forced { + _, _ = ec2c.DetachVolume(&ec2.DetachVolumeInput{ + VolumeId: aws.String(volID), + Force: aws.Bool(true), + }) + forced = true + continue + } + if time.Now().After(deadline) { + return fmt.Errorf("volume %s still %s after %s, not deleted", volID, state, volumeTeardownTimeout) + } + } + time.Sleep(teardownPollInterval) + } +} + +// teardownInstance terminates instID and waits until it reports "terminated". +// The wait is the point: TerminateInstances returns as soon as the request is +// accepted, so a volume teardown running straight afterwards would still find +// its volume attached to a guest that has not gone away yet. +func teardownInstance(ec2c ec2iface.EC2API, instID string) error { + if _, err := ec2c.TerminateInstances(&ec2.TerminateInstancesInput{ + InstanceIds: []*string{aws.String(instID)}, + }); err != nil && !instanceGone(err) { + return fmt.Errorf("terminate instance %s: %w", instID, err) + } + + deadline := time.Now().Add(instanceTeardownTimeout) + for { + out, err := ec2c.DescribeInstances(&ec2.DescribeInstancesInput{ + InstanceIds: []*string{aws.String(instID)}, + }) + if err != nil { + if instanceGone(err) { + return nil + } + return fmt.Errorf("describe instance %s: %w", instID, err) + } + if len(out.Reservations) == 0 || len(out.Reservations[0].Instances) == 0 { + return nil + } + state := aws.StringValue(out.Reservations[0].Instances[0].State.Name) + if state == ec2.InstanceStateNameTerminated { + return nil + } + if time.Now().After(deadline) { + return fmt.Errorf("instance %s still %s after %s", instID, state, instanceTeardownTimeout) + } + time.Sleep(teardownPollInterval) + } +} + +// volumeGone and instanceGone report whether err is the API's way of saying the +// resource no longer exists, which is success for a teardown. +func volumeGone(err error) bool { return errCodeIs(err, "InvalidVolume.NotFound") } + +func instanceGone(err error) bool { return errCodeIs(err, "InvalidInstanceID.NotFound") } + +func errCodeIs(err error, code string) bool { + var ae awserr.Error + return asErr(err, &ae) && ae.Code() == code +} + // asErr is a thin errors.As that lets the helper stay testify-free. func asErr(err error, target *awserr.Error) bool { if a, ok := err.(awserr.Error); ok { diff --git a/tests/e2e/harness/fixtures.go b/tests/e2e/harness/fixtures.go index 256ce5bce..066a810ae 100644 --- a/tests/e2e/harness/fixtures.go +++ b/tests/e2e/harness/fixtures.go @@ -15,6 +15,7 @@ import ( "fmt" "os" "path/filepath" + "strings" "sync" "testing" "time" @@ -68,8 +69,12 @@ type Fixture struct { // processCleanups holds teardown callbacks for process-mode fixtures. // Close() runs them LIFO. Unused (and untouched) when parent != nil. + // + // leaks accumulates teardown failures for Close() to return; test-mode + // fixtures report them on the bound *testing.T instead. processMu sync.Mutex processCleanups []func() + leaks []string } // Compile-time interface checks. @@ -132,10 +137,12 @@ func newFixture(t *testing.T, ec2c ec2iface.EC2API, elbc elbv2iface.ELBV2API) *F } } -// Close drains the process-mode cleanup chain LIFO. No-op for test-mode -// fixtures (parent != nil) since those route teardown through t.Cleanup. +// Close drains the process-mode cleanup chain LIFO and returns a non-nil error +// naming every resource it could not tear down, so a caller in TestMain can +// fail the run. No-op returning nil for test-mode fixtures (parent != nil) +// since those route teardown through t.Cleanup. // Safe to call multiple times; subsequent calls drain an empty slice. -func (f *Fixture) Close() { +func (f *Fixture) Close() error { f.processMu.Lock() cleanups := f.processCleanups f.processCleanups = nil @@ -143,18 +150,54 @@ func (f *Fixture) Close() { for i := len(cleanups) - 1; i >= 0; i-- { cleanups[i]() } + + f.processMu.Lock() + leaks := f.leaks + f.leaks = nil + f.processMu.Unlock() + if len(leaks) == 0 { + return nil + } + return fmt.Errorf("%d resource(s) left behind:\n %s", len(leaks), strings.Join(leaks, "\n ")) } // registerCleanup routes a teardown callback to the appropriate sink based // on construction mode. Test mode: t.Cleanup on the bound parent. Process // mode: append to the LIFO drained by Close(). +// +// Each callback is isolated from the rest: a panic in one teardown would +// otherwise abort the whole chain and strand every resource behind it. func (f *Fixture) registerCleanup(fn func()) { + isolated := func() { + defer func() { + if r := recover(); r != nil { + f.reportLeak("cleanup panicked: %v", r) + } + }() + fn() + } if f.parent != nil { - f.parent.Cleanup(fn) + f.parent.Cleanup(isolated) return } f.processMu.Lock() - f.processCleanups = append(f.processCleanups, fn) + f.processCleanups = append(f.processCleanups, isolated) + f.processMu.Unlock() +} + +// reportLeak records a teardown failure. A cleanup that fails has left a real +// resource behind on the node, so this must not be a log line nobody reads: +// test-mode fixtures fail the bound test, process-mode fixtures accumulate the +// failures for Close() to return. +func (f *Fixture) reportLeak(format string, args ...any) { + msg := fmt.Sprintf(format, args...) + if f.parent != nil { + f.parent.Errorf("harness.Fixture: LEAKED resource, teardown failed: %s", msg) + return + } + fmt.Fprintf(os.Stderr, "harness.Fixture: LEAKED resource, teardown failed: %s\n", msg) + f.processMu.Lock() + f.leaks = append(f.leaks, msg) f.processMu.Unlock() } @@ -222,7 +265,7 @@ func (f *Fixture) ensureOnce(t *testing.T, key string, create func() (string, fu if !dup && cleanup != nil { f.registerCleanup(func() { if cerr := cleanup(); cerr != nil { - f.logf("fixture cleanup %s: %v", key, cerr) + f.reportLeak("%s (%s): %v", key, id, cerr) } }) } @@ -631,10 +674,7 @@ func EnsureInstance(t *testing.T, fx *Fixture, spec InstanceSpec) string { } return instID, func() error { - _, derr := fx.EC2.TerminateInstances(&ec2.TerminateInstancesInput{ - InstanceIds: []*string{aws.String(instID)}, - }) - return derr + return teardownInstance(fx.EC2, instID) }, nil }) if err != nil { @@ -683,10 +723,7 @@ func EnsureVolume(t *testing.T, fx *Fixture, az string, sizeGiB int64) string { } return volID, func() error { - _, derr := fx.EC2.DeleteVolume(&ec2.DeleteVolumeInput{ - VolumeId: aws.String(volID), - }) - return derr + return teardownVolume(fx.EC2, volID) }, nil }) if err != nil { diff --git a/tests/e2e/harness/fixtures_test.go b/tests/e2e/harness/fixtures_test.go index 9e970c8b0..c0cbf6977 100644 --- a/tests/e2e/harness/fixtures_test.go +++ b/tests/e2e/harness/fixtures_test.go @@ -5,12 +5,14 @@ package harness import ( "errors" "path/filepath" + "strings" "sync" "sync/atomic" "testing" "time" "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/service/ec2" "github.com/aws/aws-sdk-go/service/ec2/ec2iface" "github.com/aws/aws-sdk-go/service/elbv2/elbv2iface" @@ -28,6 +30,56 @@ type fakeEC2 struct { // holdCreate, when non-nil, blocks CreateKeyPair until closed. Drives // the concurrent-callers test deterministically. holdCreate chan struct{} + + // Volume teardown state. volState is what DescribeVolumes reports; + // DetachVolume flips it to "available", modelling a node that releases the + // volume once the attachment is forced off. deleteVolumeErr, when set, is + // what DeleteVolume returns — used to drive the leak-reporting path. + volMu sync.Mutex + volState string + detachVolume atomic.Int64 + deleteVolume atomic.Int64 + deleteVolumeErr error +} + +func (f *fakeEC2) CreateVolume(*ec2.CreateVolumeInput) (*ec2.Volume, error) { + return &ec2.Volume{VolumeId: aws.String("vol-fake")}, nil +} + +func (f *fakeEC2) DescribeVolumes(*ec2.DescribeVolumesInput) (*ec2.DescribeVolumesOutput, error) { + f.volMu.Lock() + defer f.volMu.Unlock() + return &ec2.DescribeVolumesOutput{ + Volumes: []*ec2.Volume{{VolumeId: aws.String("vol-fake"), State: aws.String(f.volState)}}, + }, nil +} + +func (f *fakeEC2) DetachVolume(*ec2.DetachVolumeInput) (*ec2.VolumeAttachment, error) { + f.detachVolume.Add(1) + f.volMu.Lock() + defer f.volMu.Unlock() + f.volState = ec2.VolumeStateAvailable + return &ec2.VolumeAttachment{}, nil +} + +func (f *fakeEC2) DeleteVolume(*ec2.DeleteVolumeInput) (*ec2.DeleteVolumeOutput, error) { + f.deleteVolume.Add(1) + if f.deleteVolumeErr != nil { + return nil, f.deleteVolumeErr + } + // The refusal the real API gives, and the whole reason teardown has to + // detach first: an attached volume cannot be deleted. + f.volMu.Lock() + defer f.volMu.Unlock() + if f.volState != ec2.VolumeStateAvailable { + return nil, awserr.New("VolumeInUse", + "The specified Amazon EBS volume is attached to an instance.", nil) + } + return &ec2.DeleteVolumeOutput{}, nil +} + +func (f *fakeEC2) CreateTags(*ec2.CreateTagsInput) (*ec2.CreateTagsOutput, error) { + return &ec2.CreateTagsOutput{}, nil } func (f *fakeEC2) CreateKeyPair(in *ec2.CreateKeyPairInput) (*ec2.CreateKeyPairOutput, error) { @@ -151,7 +203,9 @@ func TestProcessFixture_CloseRunsCleanupsLIFO(t *testing.T) { fx.RegisterCleanup(func() { order = append(order, 2) }) fx.RegisterCleanup(func() { order = append(order, 3) }) - fx.Close() + if err := fx.Close(); err != nil { + t.Fatalf("Close err = %v, want nil", err) + } want := []int{3, 2, 1} if len(order) != len(want) { t.Fatalf("cleanup count = %d, want %d", len(order), len(want)) @@ -163,12 +217,116 @@ func TestProcessFixture_CloseRunsCleanupsLIFO(t *testing.T) { } // Second Close is a no-op — no duplicate firings, no panic. - fx.Close() + if err := fx.Close(); err != nil { + t.Fatalf("second Close err = %v, want nil", err) + } if len(order) != len(want) { t.Fatalf("second Close re-fired callbacks: order=%v", order) } } +// newProcessFakeFixture builds a process-mode fixture (parent == nil) over the +// fake EC2, which is the mode whose teardown failures surface through Close(). +func newProcessFakeFixture() (*Fixture, *fakeEC2) { + ec2c := &fakeEC2{} + return &Fixture{ + EC2: ec2c, + ELBv2: &fakeELB{}, + scratch: "test", + memo: map[string]string{}, + cleanups: map[string]struct{}{}, + }, ec2c +} + +// TestTeardownVolume_DetachesBeforeDeleting covers the ordering gap: fixture +// cleanups drain LIFO, so a volume created after its instance is torn down +// while still attached. Teardown must release it rather than let DeleteVolume +// fail with VolumeInUse and strand it. +func TestTeardownVolume_DetachesBeforeDeleting(t *testing.T) { + ec2c := &fakeEC2{volState: ec2.VolumeStateInUse} + + if err := teardownVolume(ec2c, "vol-fake"); err != nil { + t.Fatalf("teardownVolume err = %v, want nil", err) + } + if got := ec2c.detachVolume.Load(); got != 1 { + t.Fatalf("DetachVolume calls = %d, want 1 (attached volume was not released)", got) + } + if got := ec2c.deleteVolume.Load(); got != 1 { + t.Fatalf("DeleteVolume calls = %d, want 1", got) + } +} + +// TestTeardownVolume_AvailableSkipsDetach verifies the common case costs no +// extra API calls: an already-available volume is deleted directly. +func TestTeardownVolume_AvailableSkipsDetach(t *testing.T) { + ec2c := &fakeEC2{volState: ec2.VolumeStateAvailable} + + if err := teardownVolume(ec2c, "vol-fake"); err != nil { + t.Fatalf("teardownVolume err = %v, want nil", err) + } + if got := ec2c.detachVolume.Load(); got != 0 { + t.Fatalf("DetachVolume calls = %d, want 0 for an available volume", got) + } +} + +// TestEnsureVolume_CleanupReleasesAttachedVolume is the regression test for the +// leak itself: the volume is created available, then attached, and teardown has +// to cope. This is exactly the shape of a suite that calls EnsureInstance before +// EnsureVolume, since the LIFO chain tears the volume down first. +func TestEnsureVolume_CleanupReleasesAttachedVolume(t *testing.T) { + fx, ec2c := newProcessFakeFixture() + ec2c.volState = ec2.VolumeStateAvailable + + _ = EnsureVolume(t, fx, "ap-southeast-2a", 10) + ec2c.volState = ec2.VolumeStateInUse + + if err := fx.Close(); err != nil { + t.Fatalf("Close err = %v, want nil (attached volume was not released and is leaked)", err) + } + if got := ec2c.detachVolume.Load(); got != 1 { + t.Fatalf("DetachVolume calls = %d, want 1", got) + } +} + +// TestFixtureClose_ReportsLeak verifies a teardown that fails is reported as a +// leak rather than logged and forgotten. Before this, a failing cleanup left a +// real volume on the node and the run still went green. +func TestFixtureClose_ReportsLeak(t *testing.T) { + fx, ec2c := newProcessFakeFixture() + ec2c.volState = ec2.VolumeStateAvailable + ec2c.deleteVolumeErr = errors.New("boom") + + _ = EnsureVolume(t, fx, "ap-southeast-2a", 10) + + err := fx.Close() + if err == nil { + t.Fatalf("Close err = nil, want a leak report for the failed volume delete") + } + if !strings.Contains(err.Error(), "vol-fake") { + t.Fatalf("leak report %q does not name the leaked volume", err) + } +} + +// TestFixtureClose_PanicDoesNotStrandRest verifies one bad teardown cannot +// abort the chain. An aborted chain strands every resource registered before +// the panic, which is how a whole instance has previously been left running. +func TestFixtureClose_PanicDoesNotStrandRest(t *testing.T) { + fx, _ := newProcessFakeFixture() + + var ran []int + fx.RegisterCleanup(func() { ran = append(ran, 1) }) + fx.RegisterCleanup(func() { panic("teardown blew up") }) + fx.RegisterCleanup(func() { ran = append(ran, 3) }) + + err := fx.Close() + if len(ran) != 2 { + t.Fatalf("cleanups run = %v, want both non-panicking callbacks", ran) + } + if err == nil || !strings.Contains(err.Error(), "panicked") { + t.Fatalf("Close err = %v, want a report naming the panic", err) + } +} + // TestEnsureKeyPair_CleanupRunsOnce verifies cleanup fires DeleteKeyPair // exactly once even after multiple Ensure calls memoize to the same ID. func TestEnsureKeyPair_CleanupRunsOnce(t *testing.T) { diff --git a/tests/e2e/harness/guest_data.go b/tests/e2e/harness/guest_data.go index 4bebae1f5..cc42af5b0 100644 --- a/tests/e2e/harness/guest_data.go +++ b/tests/e2e/harness/guest_data.go @@ -22,19 +22,35 @@ import ( // guestSentinelFile is the file written into a freshly formatted data volume. const guestSentinelFile = "e2e-sentinel.bin" -// guestExecTimeout bounds a single guest command (mkfs/dd are sub-second on the -// small payloads used here; the ceiling only guards against a hung SSH). +// guestExecTimeout bounds a single guest command by default (mkfs/dd are +// sub-second on the small payloads most callers use; the ceiling only guards +// against a hung SSH). Callers driving a larger payload — enough data that +// sustaining it within this ceiling would require an unrealistic transfer +// rate — must use GuestExecTimeout with a budget sized to their own payload +// instead of assuming this default is high enough. const guestExecTimeout = 2 * time.Minute // sha256RE matches a bare sha256 digest so the checksum can be lifted out of // command output that may also carry the file path (`␠␠`). var sha256RE = regexp.MustCompile(`\b[0-9a-f]{64}\b`) -// GuestExec runs cmd over SSH against tgt and returns combined stdout+stderr -// plus the run error. It never calls t.Fatal, so callers can branch on an -// expected non-zero exit. +// GuestExec runs cmd over SSH against tgt with the default guestExecTimeout +// budget and returns combined stdout+stderr plus the run error. It never +// calls t.Fatal, so callers can branch on an expected non-zero exit. func GuestExec(tgt SSHTarget, cmd string) (string, error) { - ctx, cancel := context.WithTimeout(context.Background(), guestExecTimeout) + return GuestExecTimeout(tgt, cmd, guestExecTimeout) +} + +// GuestExecTimeout is GuestExec with a caller-supplied timeout, for commands +// whose payload size makes the default guestExecTimeout unrealistic (e.g. a +// multi-gigabyte write against a network-backed volume). If the command is +// still running when timeout elapses, the returned error is wrapped with +// context.DeadlineExceeded so callers can distinguish "the command ran and +// failed" from "the command never got the chance to finish" via errors.Is — +// a distinction that matters wherever the two would otherwise look like the +// same red test for different reasons. +func GuestExecTimeout(tgt SSHTarget, cmd string, timeout time.Duration) (string, error) { + ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() args := []string{ "-o", "StrictHostKeyChecking=no", @@ -48,6 +64,9 @@ func GuestExec(tgt SSHTarget, cmd string) (string, error) { cmd, } out, err := exec.CommandContext(ctx, "ssh", args...).CombinedOutput() + if err != nil && ctx.Err() == context.DeadlineExceeded { + return string(out), fmt.Errorf("guest command exceeded its %s timeout (still running when the deadline hit): %w: %w", timeout, context.DeadlineExceeded, err) + } return string(out), err } diff --git a/tests/e2e/iam/instance_iam_profile_test.go b/tests/e2e/iam/instance_iam_profile_test.go index 1ac98955e..13d951371 100644 --- a/tests/e2e/iam/instance_iam_profile_test.go +++ b/tests/e2e/iam/instance_iam_profile_test.go @@ -35,6 +35,28 @@ const ( // Self-bootstrapping: it creates its own role + two profiles under the // single-iamprofile- namespace, asserts the EC2 surface, then cleans up so // subsequent tests see no residue. +// +// Nothing here was ported down to the integration tier: its role/profile +// create+attach setup (CreateRole, AttachRolePolicy, CreateInstanceProfile x2, +// AddRoleToInstanceProfile) is scaffolding, not an assertion under test — the +// same IAM CRUD, including the AdministratorAccess-doesn't-pre-seed wrinkle, +// is already asserted by tests/integration's TestIAMRolesAndProfiles. The +// association lifecycle itself (Associate/AlreadyAssociated/Replace/ +// Disassociate/DeleteInstanceProfile-while-bound) is implemented in +// InstanceServiceImpl (handlers/ec2/instance/service_impl.go) against a real +// *vm.VM's mutable IamInstanceProfileArn/AssociationId fields — daemon-side +// state a live guest actually carries. A static stub could only stand in for +// it by modelling the whole instance lifecycle, which is why this stays live +// rather than moving down. The pure gateway-side half of that same code path +// (profile resolution, PassRole enforcement, ID enrichment, NATS error +// mapping in gateway/ec2/instance/IamInstanceProfileAssociation.go) is +// already exhaustively unit-tested in +// IamInstanceProfileAssociation_test.go, and RunInstances' iam:PassRole gate +// specifically is covered by tests/integration's +// TestRunInstances_DeniedWithoutPassRole. The `require.Equal(t, "running", …)` +// singleton guard below is a precondition check, not a wait — nothing here +// blocks on a state transition that a static stub couldn't already satisfy, +// so there is no incidental wait to drop. func runIAMInstanceProfileAssociation(t *testing.T, fix *Fixture) { harness.Phase(t, "Single — EC2 IAM Instance Profile Association") diff --git a/tests/e2e/iam/main_test.go b/tests/e2e/iam/main_test.go index e4cd64d93..cbef65da3 100644 --- a/tests/e2e/iam/main_test.go +++ b/tests/e2e/iam/main_test.go @@ -10,6 +10,7 @@ package iam import ( + "fmt" "os" "sync" "testing" @@ -34,7 +35,12 @@ func TestMain(m *testing.M) { code := m.Run() if pkgFix != nil { if pkgFix.Harness != nil { - pkgFix.Harness.Close() + // A leaked resource fails the run: the suite may have passed, but it + // left state on the node that the next run will trip over. + if err := pkgFix.Harness.Close(); err != nil { + fmt.Fprintf(os.Stderr, "e2e teardown: %v\n", err) + code = 1 + } } if pkgFix.TmpDir != "" { _ = os.RemoveAll(pkgFix.TmpDir) diff --git a/tests/e2e/multinode/main_test.go b/tests/e2e/multinode/main_test.go index b0ae19621..a1b8f09fb 100644 --- a/tests/e2e/multinode/main_test.go +++ b/tests/e2e/multinode/main_test.go @@ -32,7 +32,12 @@ func TestMain(m *testing.M) { code := m.Run() if pkgFix != nil { if pkgFix.Harness != nil { - pkgFix.Harness.Close() + // A leaked resource fails the run: the suite may have passed, but it + // left state on the node that the next run will trip over. + if err := pkgFix.Harness.Close(); err != nil { + fmt.Fprintf(os.Stderr, "e2e teardown: %v\n", err) + code = 1 + } } if pkgFix.TmpDir != "" { _ = os.RemoveAll(pkgFix.TmpDir) diff --git a/tests/e2e/natuplink/natuplink_test.go b/tests/e2e/natuplink/natuplink_test.go index 99b4204c7..f5dc93337 100644 --- a/tests/e2e/natuplink/natuplink_test.go +++ b/tests/e2e/natuplink/natuplink_test.go @@ -90,7 +90,11 @@ func TestNATUplink(t *testing.T) { awsCli := harness.NewAWSClient(t, env) hfx, err := harness.NewProcessFixture(awsCli) require.NoError(t, err, "harness fixture init") - t.Cleanup(hfx.Close) + t.Cleanup(func() { + if err := hfx.Close(); err != nil { + t.Errorf("e2e teardown: %v", err) + } + }) fix := &fixture{ env: env, diff --git a/tests/e2e/partialblock/main_test.go b/tests/e2e/partialblock/main_test.go new file mode 100644 index 000000000..f316b16d2 --- /dev/null +++ b/tests/e2e/partialblock/main_test.go @@ -0,0 +1,155 @@ +//go:build e2e + +// Package partialblock proves that concurrent sub-block writes to the same +// block survive a cold reread when they travel through NBD from a real guest. +// +// The claim is deliberately narrow. The cheap, deterministic gate for the +// underlying read-modify-write race is viperblock's own unit reproducer, which +// runs against a real predastore in a fraction of a second. This suite adds +// exactly one thing that unit test cannot: evidence that the race is reachable +// end to end, through the guest block layer and the NBD transport, rather than +// only through the engine's Go API. +// +// It is its own package rather than a case inside single/ because single/ +// shares one memoized instance across the whole package, and this suite needs +// a raw, unformatted volume it can write with O_DIRECT and then delete. A +// filesystem is precisely what prevents the workload from reaching the code +// under test: it coalesces writes into aligned full-block requests, and the +// read-modify-write path only runs for a partial block. +package partialblock + +import ( + "bufio" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + + "github.com/mulgadc/spinifex/tests/e2e/harness" +) + +var ( + pkgFixOnce sync.Once + pkgFix *Fixture + pkgFixErr error +) + +func TestMain(m *testing.M) { + code := m.Run() + if pkgFix != nil { + if pkgFix.Harness != nil { + // A leaked resource fails the run: the suite may have passed, but it + // left state on the node that the next run will trip over. + if err := pkgFix.Harness.Close(); err != nil { + fmt.Fprintf(os.Stderr, "e2e teardown: %v\n", err) + code = 1 + } + } + if pkgFix.TmpDir != "" { + _ = os.RemoveAll(pkgFix.TmpDir) + } + } + os.Exit(code) +} + +// Fixture carries per-process state shared across this package's tests. +type Fixture struct { + Env *harness.Env + AWS *harness.AWSClient + Harness *harness.Fixture + TmpDir string + + // PoolMode gates the suite: EIP allocation is what makes guest SSH + // reachable (harness.InstancePublicSSHHost refuses to fall back to + // qemu-hostfwd), and guest SSH is the only channel this suite has for + // driving the write workload and reading its verdict back. + PoolMode bool +} + +func (f *Fixture) ArtifactDir(t *testing.T) string { + t.Helper() + return harness.ArtifactDir(t, f.Env) +} + +// requirePartialBlockFixture returns the package-scoped Fixture singleton, +// building it on first call. Skips when SPINIFEX_E2E is unset, the mode is not +// single-node, or the env has no EIP pool. +func requirePartialBlockFixture(t *testing.T) *Fixture { + t.Helper() + pkgFixOnce.Do(func() { + if os.Getenv("SPINIFEX_E2E") == "" { + return + } + env := harness.LoadEnv(t) + if env.Mode != harness.ModeSingle { + pkgFixErr = nil + return + } + awsCli := harness.NewAWSClient(t, env) + h, err := harness.NewProcessFixture(awsCli) + if err != nil { + pkgFixErr = err + return + } + tmpDir, err := os.MkdirTemp("", "partialblock-pkgfix-*") + if err != nil { + pkgFixErr = err + return + } + harness.EnsureDefaultSGOpen(t, awsCli) + pkgFix = &Fixture{ + Env: env, + AWS: awsCli, + Harness: h, + TmpDir: tmpDir, + PoolMode: detectPoolMode(env), + } + }) + if pkgFixErr != nil { + t.Fatalf("partialblock fixture init: %v", pkgFixErr) + } + if pkgFix == nil { + t.Skip("SPINIFEX_E2E unset or SPINIFEX_MODE != single") + } + if !pkgFix.PoolMode { + t.Skip("partialblock suite requires external_mode=pool — a nat env has no EIP and cannot reach guest SSH, the only channel this suite has to drive and verify the write workload") + } + return pkgFix +} + +// detectPoolMode reads external_mode from spinifex.toml. Copied rather than +// imported for the same reason storagegrowth copies it: harness/** is +// deliberately never touched by these standalone-workload packages, and the +// predicate is small and package-local everywhere it already exists. +func detectPoolMode(env *harness.Env) bool { + cfg := os.ExpandEnv("$HOME/spinifex/config/spinifex.toml") + if env.ConfigDir != "" { + cfg = filepath.Join(env.ConfigDir, "spinifex.toml") + } + f, err := os.Open(cfg) + if err != nil { + return false + } + defer f.Close() + + scanner := bufio.NewScanner(f) + inNetwork := false + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if strings.HasPrefix(line, "[") { + inNetwork = line == "[network]" + continue + } + if !inNetwork || !strings.HasPrefix(line, "external_mode") { + continue + } + // external_mode = "pool" — quoted value; only "pool" enables this suite. + if _, rhs, ok := strings.Cut(line, "="); ok { + val := strings.Trim(strings.TrimSpace(rhs), "\"'") + return val == "pool" + } + } + return false +} diff --git a/tests/e2e/partialblock/partial_block_durability_test.go b/tests/e2e/partialblock/partial_block_durability_test.go new file mode 100644 index 000000000..dc6d9da0b --- /dev/null +++ b/tests/e2e/partialblock/partial_block_durability_test.go @@ -0,0 +1,334 @@ +//go:build e2e + +package partialblock + +import ( + "fmt" + "os" + "regexp" + "strconv" + "testing" + "time" + + "github.com/mulgadc/spinifex/tests/e2e/harness" +) + +const ( + // partialBlockDevice is the guest-visible attach point requested for the + // working-set volume. + partialBlockDevice = "/dev/sdp" + + // partialBlockVolumeSizeGiB is the smallest volume that comfortably holds + // the working region. The region itself is a few MiB at the head of the + // device; the rest is never touched. + partialBlockVolumeSizeGiB = 1 + + // blockSize mirrors viperblock's block granularity. The read-modify-write + // path under test runs only when a write covers part of one of these, so + // this value has to match the engine's or the workload silently degrades + // into full-block overwrites that exercise nothing. + blockSize = 4096 + // halfSize splits a block into two disjoint sub-block ranges. Two writers + // each own one half, so every write is partial by construction and the two + // writers collide on the same block without ever overlapping each other's + // bytes — a lost update is therefore unambiguous, not a write-write tie. + halfSize = blockSize / 2 + + // defaultBlocks and defaultRounds size the workload. Both are modest on + // purpose: the predecessor suite moved a 2048 MiB working set, which after + // RS(2,1) left roughly 3 GiB of backend behind per run and twice drove the + // node under predastore's nearfull watermark, wedging every guest on it. + // Collision probability here comes from two writers hammering the same + // blocks in the same order, not from volume. + defaultBlocks = 2048 + defaultRounds = 6 + + // partialBlockIOBaseTimeout is the fixed part of the IO budget: SSH setup + // and interpreter startup, independent of how much data moves. + partialBlockIOBaseTimeout = 60 * time.Second + // partialBlockIOPerWrite budgets one O_DIRECT sub-block write. Deliberately + // generous: each one is a synchronous round trip to a network-backed + // volume, and a tight bound risks the failure mode this suite must avoid — + // a timeout that looks like a red durability test but proves nothing about + // whether any write was lost. + partialBlockIOPerWrite = 3 * time.Millisecond +) + +// wroteRE captures the per-writer byte count each child reports, so the test +// can assert both writers moved exactly the bytes they were asked to rather +// than trusting an exit status alone. +var wroteRE = regexp.MustCompile(`WROTE half=(\d+) bytes=(\d+)`) + +// checkedRE captures the verifier's tally. +var checkedRE = regexp.MustCompile(`CHECKED (\d+) BAD (\d+)`) + +// partialBlockDriver is the guest-side workload. It is Python rather than a +// shell pipeline because the workload needs three things a shell cannot give: +// O_DIRECT with a correctly aligned buffer, writes at sub-block offsets that +// reach the device instead of being merged in the page cache, and a real exit +// status per writer. +// +// That last point is why the predecessor suite could not fail. It ended its +// write phase with `... & wait; sync; sha256sum`, and the status observed was +// sha256sum's; bare `wait` returns 0 unconditionally, so every writer error was +// invisible and a backend ENOSPC presented as a pass. +const partialBlockDriver = ` +import mmap, os, struct, sys + +BLOCK = 4096 +HALF = 2048 +REC = 32 +MAGIC = 0x5042444B + +def fill(buf, blk, half, rnd): + # Stamp the whole half, not just a header, so a splice that reverts only + # part of the range is still caught. + rec = struct.pack(" 0 || writeEnd < blockSize); +// a fully covered block takes a full-overwrite branch that reads nothing and +// can lose nothing. Two writers that both read generation N, each splice their +// own half, and let the higher sequence number win wholesale is the shape of +// the lost update this asserts against — so one half reading back at an +// earlier round is the signature. +// +// The device is raw and unformatted on purpose. A filesystem defeats this +// workload: it coalesces writes into aligned full-block requests, which is how +// the predecessor suite ended up issuing zero sub-4096 and zero unaligned +// requests and passing on a build that still carried the bug. +func TestPartialBlockConcurrentWriteDurability(t *testing.T) { + fix := requirePartialBlockFixture(t) + harness.Phase(t, "Partial-Block Concurrent-Write Durability") + + blocks := envInt(t, "SPINIFEX_PARTIALBLOCK_BLOCKS", defaultBlocks) + rounds := envInt(t, "SPINIFEX_PARTIALBLOCK_ROUNDS", defaultRounds) + writesPerWriter := blocks * rounds + wantBytes := writesPerWriter * halfSize + + instanceID, tgt := ensurePartialBlockInstance(t, fix) + az := harness.DiscoverDefaultAZ(t, fix.Harness) + volID := harness.EnsureVolume(t, fix.Harness, az, partialBlockVolumeSizeGiB) + + harness.Detail(t, "blocks", blocks, "rounds", rounds, "region_mib", blocks*blockSize/(1<<20), + "instance", instanceID, "volume", volID) + + before := harness.GuestDiskSet(t, tgt) + harness.AttachVolumeWait(t, fix.AWS, volID, instanceID, partialBlockDevice) + dev := harness.WaitForNewGuestDisk(t, tgt, before, 90*time.Second) + + // Detach before the fixture's own cleanup deletes the volume. DeleteVolume + // against a still-attached volume is what leaks volumes on teardown, and + // the delete itself is left to the fixture so the volume is released + // exactly once. + t.Cleanup(func() { harness.DetachVolumeWait(t, fix.AWS, volID) }) + + installDriver(t, tgt) + + budget := partialBlockIOBaseTimeout + time.Duration(writesPerWriter)*partialBlockIOPerWrite + harness.Step(t, "writing %d rounds x %d blocks x 2 halves O_DIRECT to /dev/%s (budget %s)", rounds, blocks, dev, budget) + out, err := harness.GuestExecTimeout(tgt, + fmt.Sprintf("sudo python3 /tmp/partialblock.py write /dev/%s %d %d", dev, blocks, rounds), budget) + if err != nil { + t.Fatalf("write phase failed (this is a writer error or timeout, NOT evidence of a lost update): %v\n%s", err, out) + } + + // Assert both writers moved exactly the bytes asked of them. Without this a + // writer that failed early could leave the device holding a consistent but + // stale round, and the verify phase would agree with itself. + assertWriterBytes(t, out, wantBytes) + + harness.Step(t, "dropping page cache") + if out, err := harness.GuestExec(tgt, "sudo sync && echo 3 | sudo tee /proc/sys/vm/drop_caches >/dev/null"); err != nil { + t.Fatalf("drop_caches: %v\n%s", err, out) + } + + harness.Step(t, "cold O_DIRECT reread, asserting every half carries round %d", rounds) + out, err = harness.GuestExecTimeout(tgt, + fmt.Sprintf("sudo python3 /tmp/partialblock.py verify /dev/%s %d %d", dev, blocks, rounds), budget) + if err != nil { + // A non-zero verify exit is the durability failure itself, so the + // output carries the per-block detail and must be surfaced whole. + t.Fatalf("cold reread found lost updates — a partial write was acked and did not survive:\n%s", out) + } + + checked, bad := parseChecked(t, out) + if bad != 0 { + t.Errorf("verify reported %d bad halves out of %d checked", bad, checked) + } + if want := blocks * 2 * 2; checked != want { + t.Errorf("verify checked %d records, want %d — the verifier did not cover the whole region", checked, want) + } + harness.Detail(t, "records_checked", checked, "records_bad", bad) +} + +// ensurePartialBlockInstance launches (or returns the memoized) guest and waits +// for SSH. Guest size is deliberately not a parameter: the memory-pressure +// trigger model the predecessor suite encoded was retracted — the fault +// reproduces with a single sequential writer on a guest with memory to spare — +// so the smallest catalog entry is the right choice here. +func ensurePartialBlockInstance(t *testing.T, fix *Fixture) (instanceID string, tgt harness.SSHTarget) { + t.Helper() + instType, arch := harness.DiscoverNanoInstanceType(t, fix.Harness) + ami := harness.DiscoverUbuntuAMI(t, fix.Harness, arch) + keyName, keyPath := harness.EnsureKeyPair(t, fix.Harness, fix.ArtifactDir(t)) + vpc := harness.EnsureDefaultVPC(t, fix.Harness) + harness.AuthorizeSSHIngress(t, fix.AWS, vpc.SGID) + + instanceID = harness.EnsureInstance(t, fix.Harness, harness.InstanceSpec{ + AMIID: ami, + InstanceType: instType, + KeyName: keyName, + SubnetID: vpc.SubnetID, + SGID: vpc.SGID, + }) + inst := harness.WaitForInstanceState(t, fix.AWS, instanceID, "running") + host, port := harness.InstancePublicSSHHost(t, inst) + + harness.Step(t, "waiting for guest SSH on %s:%d (instance_type=%s)", host, port, instType) + if !harness.TryGuestSSHReady(host, port, "ubuntu", keyPath, 5*time.Minute) { + t.Fatalf("guest %s SSH %s:%d not ready after 5m", instanceID, host, port) + } + return instanceID, harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath} +} + +// installDriver writes the guest-side workload to /tmp and confirms python3 can +// parse it, so a syntax error surfaces here rather than as an opaque non-zero +// exit in the middle of the write phase. +func installDriver(t *testing.T, tgt harness.SSHTarget) { + t.Helper() + script := fmt.Sprintf("cat > /tmp/partialblock.py <<'PBEOF'\n%s\nPBEOF\npython3 -c 'import py_compile,sys; py_compile.compile(\"/tmp/partialblock.py\", doraise=True)'", partialBlockDriver) + if out, err := harness.GuestExec(tgt, script); err != nil { + t.Fatalf("installDriver: %v\n%s", err, out) + } +} + +// assertWriterBytes checks that both halves reported completion and that each +// moved exactly wantBytes. +func assertWriterBytes(t *testing.T, out string, wantBytes int) { + t.Helper() + seen := map[int]int{} + for _, m := range wroteRE.FindAllStringSubmatch(out, -1) { + half, _ := strconv.Atoi(m[1]) + got, _ := strconv.Atoi(m[2]) + seen[half] = got + } + for half := 0; half < 2; half++ { + got, ok := seen[half] + if !ok { + t.Fatalf("writer half=%d reported no byte count — it did not run to completion:\n%s", half, out) + } + if got != wantBytes { + t.Fatalf("writer half=%d wrote %d bytes, want %d — short write path, not a durability result:\n%s", half, got, wantBytes, out) + } + } +} + +// parseChecked pulls the verifier's tally out of its output. +func parseChecked(t *testing.T, out string) (checked, bad int) { + t.Helper() + m := checkedRE.FindStringSubmatch(out) + if m == nil { + t.Fatalf("verify produced no CHECKED line — cannot tell whether it examined anything:\n%s", out) + } + checked, _ = strconv.Atoi(m[1]) + bad, _ = strconv.Atoi(m[2]) + return checked, bad +} + +// envInt reads a positive integer override, failing outright on a malformed +// value rather than silently falling back to the default. +func envInt(t *testing.T, key string, def int) int { + t.Helper() + v := os.Getenv(key) + if v == "" { + return def + } + n, err := strconv.Atoi(v) + if err != nil || n <= 0 { + t.Fatalf("%s=%q is not a positive integer", key, v) + } + return n +} diff --git a/tests/e2e/single/main_test.go b/tests/e2e/single/main_test.go index 504e230e7..a9c335dc3 100644 --- a/tests/e2e/single/main_test.go +++ b/tests/e2e/single/main_test.go @@ -10,6 +10,7 @@ package single import ( "bufio" + "fmt" "os" "path/filepath" "strings" @@ -36,7 +37,12 @@ func TestMain(m *testing.M) { code := m.Run() if pkgFix != nil { if pkgFix.Harness != nil { - pkgFix.Harness.Close() + // A leaked resource fails the run: the suite may have passed, but it + // left state on the node that the next run will trip over. + if err := pkgFix.Harness.Close(); err != nil { + fmt.Fprintf(os.Stderr, "e2e teardown: %v\n", err) + code = 1 + } } if pkgFix.TmpDir != "" { _ = os.RemoveAll(pkgFix.TmpDir) diff --git a/tests/e2e/storagegrowth/image_pull_test.go b/tests/e2e/storagegrowth/image_pull_test.go index 52c6a0874..01985f37a 100644 --- a/tests/e2e/storagegrowth/image_pull_test.go +++ b/tests/e2e/storagegrowth/image_pull_test.go @@ -1184,13 +1184,8 @@ func captureImagePullSample(t *testing.T, tgt harness.SSHTarget, ssh *harness.Pe s.GuestUsedBytes = v } - duCtx, duCancel := context.WithTimeout(context.Background(), 30*time.Second) - duOut, duErr := ssh.Run(duCtx, wanHost, fmt.Sprintf("sudo du -s --block-size=1 %s 2>/dev/null | cut -f1", predastoreBaseDir)) - duCancel() - if duErr != nil { - t.Logf("storagegrowth: host du via PeerSSH on %s: %v", wanHost, duErr) - } else if v, perr := parseLastIntField(string(duOut)); perr != nil { - t.Logf("storagegrowth: parse host du output %q: %v", duOut, perr) + if v, err := hostPredastoreAllocatedBytes(ssh, wanHost); err != nil { + t.Logf("storagegrowth: host du via PeerSSH on %s: %v", wanHost, err) } else { s.HostAllocatedBytes = v } @@ -1316,6 +1311,32 @@ func parseLastIntField(out string) (int64, error) { return strconv.ParseInt(fields[len(fields)-1], 10, 64) } +// hostPredastoreAllocatedBytes reads predastoreBaseDir's total allocated +// bytes on the node at wanHost, over PeerSSH. This is the one measurement +// that actually works against a live deployment: it needs no S3 credential +// at all, only SSH access to the node, unlike a per-volume S3 listing of the +// backend bucket (see assertNPassBackendByteDelta in storage_growth_test.go +// for why that path is unusable here). +// +// The number returned is NODE-WIDE, not attributable to any single volume -- +// du has no concept of which volume a chunk object belongs to, it only sees +// bytes on disk. Callers that need a per-workload signal must take two +// readings and use the delta, and must be able to argue nothing else was +// writing to the node in between. +func hostPredastoreAllocatedBytes(ssh *harness.PeerSSH, wanHost string) (int64, error) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + out, err := ssh.Run(ctx, wanHost, fmt.Sprintf("sudo du -s --block-size=1 %s 2>/dev/null | cut -f1", predastoreBaseDir)) + if err != nil { + return 0, fmt.Errorf("du %s via PeerSSH: %w", predastoreBaseDir, err) + } + v, perr := parseLastIntField(string(out)) + if perr != nil { + return 0, fmt.Errorf("parse du output %q: %w", out, perr) + } + return v, nil +} + // writeImagePullResult persists the workload record at the path named by // npassResultPathEnv (SPINIFEX_STORAGEGROWTH_RESULT_PATH) — the same env var // npass uses, per the shared result contract. See writeNPassResult for why an diff --git a/tests/e2e/storagegrowth/main_test.go b/tests/e2e/storagegrowth/main_test.go index 3fe3054f2..097a56ecb 100644 --- a/tests/e2e/storagegrowth/main_test.go +++ b/tests/e2e/storagegrowth/main_test.go @@ -15,21 +15,40 @@ // // That is a different fault from the *overwrite-orphan* leak, where a key is // rewritten in place and the superseded extent is never tombstoned, so -// compaction can never find it. Keeping the two apart is the whole point, and -// it is why this package does not measure. The two leaks are distinguished -// precisely by predastore's own live-versus-dead split: unreferenced-live bytes -// inflate live totals, overwrite-orphan bytes inflate dead totals. A host-side -// `du` reports one number that mixes them, which cannot separate the target from -// the confound — and this workload drives both at once, because the drain each -// pass forces also rewrites the checkpoint map in place. So the backend is -// measured out of band by a tool that reads predastore's accounting directly. -// That tool does not exist in a standalone spinifex checkout, and depending on -// it here would break this suite's CI. +// compaction can never find it. Keeping the two apart matters for attributing +// blame within a leak, and that attribution is why this package's own +// in-process check stops at a coarse pass/fail gate rather than a full +// breakdown. The two leaks are distinguished precisely by predastore's own +// live-versus-dead split: unreferenced-live bytes inflate live totals, +// overwrite-orphan bytes inflate dead totals. A host-side `du` reports one +// number that mixes them, which cannot separate the target from the confound +// — and this workload drives both at once, because the drain each pass +// forces also rewrites the checkpoint map in place. Splitting that mix +// requires a tool that reads predastore's live-versus-dead accounting +// directly; that tool does not exist in a standalone spinifex checkout, and +// depending on it here would break this suite's CI. // -// The division of labour: this package drives the guest and reports what it did -// (see npassWorkload); the measurement step reads that record for the measured -// denominator and supplies the backend numbers. Run standalone, the workload -// still asserts the guest-side invariants it can see on its own. +// What this package's own tests DO assert on, in-process, is the coarser +// signal a node-wide `du` delta across the workload can still catch: the +// combined live+dead footprint growing well beyond the RS-erasure floor +// (see assertNPassBackendByteDelta in storage_growth_test.go). A per-volume +// S3 listing of the volume's own chunk prefix — what this suite originally +// used, and what the external tool below also reads — turned out to be +// unusable against a live deployment: the internal chunk bucket is owned by +// a system account distinct from the tenant-scoped identity this harness +// authenticates as, and predastore's same-account bucket-ownership check has +// no cross-account bucket-policy escape hatch, so every such listing fails +// closed with AccessDenied regardless of IAM grants. `du` needs no S3 +// credential at all, only SSH to the node, which is why it is what this +// package can actually run itself. +// +// The division of labour: this package drives the guest, runs its own +// coarse node-wide gate, and reports what it did (see npassWorkload); an +// external measurement step reads that record for the measured denominator +// and supplies the fine-grained live-versus-dead backend breakdown this +// package cannot produce on its own. Run standalone, the workload still +// asserts every guest-side invariant, and the coarse gate, that it can see +// on its own. // // This is its own package rather than living in single/ or multinode/: // single/ shares one singleton instance across the whole package that a @@ -46,6 +65,7 @@ package storagegrowth import ( "bufio" + "fmt" "os" "path/filepath" "strings" @@ -66,7 +86,12 @@ func TestMain(m *testing.M) { code := m.Run() if pkgFix != nil { if pkgFix.Harness != nil { - pkgFix.Harness.Close() + // A leaked resource fails the run: the suite may have passed, but it + // left state on the node that the next run will trip over. + if err := pkgFix.Harness.Close(); err != nil { + fmt.Fprintf(os.Stderr, "e2e teardown: %v\n", err) + code = 1 + } } if pkgFix.TmpDir != "" { _ = os.RemoveAll(pkgFix.TmpDir) diff --git a/tests/e2e/storagegrowth/storage_growth_test.go b/tests/e2e/storagegrowth/storage_growth_test.go index 2f324585f..6158d26ab 100644 --- a/tests/e2e/storagegrowth/storage_growth_test.go +++ b/tests/e2e/storagegrowth/storage_growth_test.go @@ -54,18 +54,63 @@ const ( // test PASS. A handoff written there would exist only for runs that failed // — precisely inverted from what a measurement step needs. npassResultPathEnv = "SPINIFEX_STORAGEGROWTH_RESULT_PATH" + + // npassBackendByteAccountingTolerance is the ceiling ratio of node-wide + // predastore-allocated-byte growth to guest-written logical bytes a + // healthy N-pass run may show. + // + // The floor is viperblock's own encoding overhead, not slack for a leak: + // chunks are written under Reed-Solomon 2+1 erasure coding, which splits + // each object into two data shards plus one parity shard of the same + // size -- 1.5x the logical payload, not 3x replication -- and every + // block is additionally framed with a 16-byte AES-GCM tag. + // + // That 1.5x floor is a precondition, not a universal constant. It holds + // because this suite provisions RS(2,1): cmd/spinifex/cmd/templates/ + // predastore.toml's [rs] block defaults to data=2, parity=1. The same + // template documents RS(3,2) as the production recommendation, whose + // floor is 5/3 = ~1.667x and would leave only ~12% headroom under the + // ceiling below instead of the ~25% intended. Recompute this value if the + // suite is ever pointed at a different RS profile rather than assuming + // the band still holds. + // + // A healthy, fully-GC'd volume settles at ~1.5x and does not go + // materially above it. The tolerance widens that floor by 25% (to 1.875x) to + // absorb two effects that are real but not leaks: every chunk pays a + // fixed header regardless of how many blocks it packs (pulling a + // sparse volume's ratio up), and the tail chunk of any write run is + // under-packed by construction (a chunk flushes on full or on drain, + // so the last one of a run is real, unavoidable overhead). Both shrink + // as a proportion of the total as the workload's footprint grows, so a + // large workload should sit close to the bare floor and a small one + // gets the extra headroom it actually needs. + // + // This does not attempt to catch the slow, bounded residual that a + // future partial-chunk compaction pass would close -- only the + // unbounded-growth shape the storage-growth incident produced, where + // a leak multiplies with every overwrite pass instead of settling. + // + // Applying this floor to a node-wide `du` delta (rather than a per-volume + // S3 listing, see assertNPassBackendByteDelta) is only sound because the + // delta is measured tightly around this workload's own passes on a + // dedicated, single-volume, serial environment -- nothing else on the + // node is expected to write to predastore's store in that window. Any + // concurrent writer (another suite, another volume, a background GC/ + // compaction pass) would inflate or shrink the delta for reasons that + // have nothing to do with this workload, and the ratio below would no + // longer mean what it claims to. + npassBackendByteAccountingTolerance = 1.875 ) // npassWorkload records what one invocation of the N-pass workload did. It is -// the whole contract with whatever measures the backing store afterwards. +// the whole contract with whatever measures the backing store afterwards, +// beyond this suite's own coarse in-process gate (assertNPassBackendByteDelta). // -// This suite deliberately reports rather than measures. The guest-side -// quantities below are the ones only a guest can know; the backend's byte -// cost is measured out of band by a tool that can read predastore's own -// live-versus-dead accounting, which is a distinction no host-side `du` can -// make and which is the entire point of the exercise. Coupling this test to -// that tool would also break a standalone spinifex checkout, where it does -// not exist. +// The guest-side quantities below are the ones only a guest can know; a +// live-versus-dead breakdown of the backend's byte cost -- a distinction no +// host-side `du` can make -- is measured out of band by a tool that can read +// predastore's own accounting directly. Coupling this test to that tool would +// also break a standalone spinifex checkout, where it does not exist. // // GuestUsedBytes is measured, never inferred. It is the filesystem's own // used-bytes count after the write, so it includes ext4 metadata as well as @@ -132,12 +177,81 @@ func TestNPassOverwrite(t *testing.T) { harness.Detail(t, "passes", passes, "extent_mib", extentMiB, "instance", instanceID, "volume", volID, "retain_volume", retain) + // Baseline predastore's node-wide allocated bytes immediately before the + // passes start, and read it again immediately after they finish, so the + // delta below brackets this workload's own writes as tightly as + // possible. See assertNPassBackendByteDelta for why this must be a + // delta rather than an absolute snapshot, and why the window must stay + // tight. + ssh := harness.NewPeerSSH() + before, err := hostPredastoreAllocatedBytes(ssh, fix.Env.WANHost) + require.NoErrorf(t, err, "backend byte accounting: baseline node du before passes") + w := runNPassOverwrite(t, fix, tgt, instanceID, volID, passes, extentMiB) w.Retained = retain writeNPassResult(t, w) harness.Detail(t, "passes_completed", w.Passes, "guest_used_bytes", w.GuestUsedBytes, "guest_used_per_pass", fmt.Sprint(w.GuestUsedPerPass)) + + after, err := hostPredastoreAllocatedBytes(ssh, fix.Env.WANHost) + require.NoErrorf(t, err, "backend byte accounting: node du after passes") + + // This is the fail-first demonstration for the storage-growth leak: if + // superseded chunks from the earlier passes are never reclaimed, the + // backend footprint grows by N times the final pass's guest-used bytes + // rather than settling at the erasure floor, and the delta below catches + // that shape without needing vbrefscan's out-of-band scoring pass. It + // complements, not replaces, the RETAIN_VOLUME external-measurement + // path: that path attributes *which* chunks are garbage on a volume kept + // alive for vbrefscan; this one is a pass/fail gate that runs on every + // invocation, retained or not. + assertNPassBackendByteDelta(t, before, after, w.GuestUsedBytes) +} + +// assertNPassBackendByteDelta asserts that predastore's node-wide allocated +// bytes grew, across the N-pass workload, by no more than +// npassBackendByteAccountingTolerance times guestWrittenBytes. +// +// This measures via `du` on the node (hostPredastoreAllocatedBytes) rather +// than listing the volume's own S3 chunk prefix. A per-volume S3 listing is +// what this suite originally used and is what vbrefscan itself reads, but it +// is not usable here: the internal bucket EBS volume chunks live under is +// owned by a system account distinct from the tenant-scoped identity this +// harness authenticates as, predastore enforces a same-account +// bucket-ownership check independent of and on top of any IAM policy grant, +// and cross-account bucket-policy or ACL grants are not implemented -- every +// listing attempt fails with AccessDenied regardless of what IAM policy the +// caller holds. There is also no usable filesystem fallback: predastore's +// distributed backend stores objects in badger plus shard nodes, not a tree +// keyed by object name, so there is nothing under predastoreBaseDir to +// attribute to a single volume by path. +// +// `du` over predastoreBaseDir is therefore the finest-grained measurement +// actually available, and it is NODE-WIDE: it cannot distinguish this +// volume's bytes from any other write happening on the same node. The before/ +// after delta (rather than an absolute reading) is what makes the number +// mean anything at all, and even the delta is only valid because this +// workload runs serially against a single volume on a dedicated environment +// -- nothing else is expected to write to predastore's store in the window +// between the two samples. This approach would be invalid under concurrency: +// a second suite or a background compaction pass writing to the same node in +// that window would be silently attributed to this workload's ratio. +func assertNPassBackendByteDelta(t *testing.T, before, after, guestWrittenBytes int64) { + t.Helper() + require.Positivef(t, guestWrittenBytes, + "backend byte accounting: guestWrittenBytes must be positive, got %d -- the denominator every ratio divides by is unusable", + guestWrittenBytes) + + delta := after - before + ratio := float64(delta) / float64(guestWrittenBytes) + harness.Detail(t, "backend_allocated_bytes_before", before, "backend_allocated_bytes_after", after, + "backend_allocated_bytes_delta", delta, "guest_written_bytes", guestWrittenBytes, + "ratio", fmt.Sprintf("%.3f", ratio)) + + require.LessOrEqualf(t, ratio, npassBackendByteAccountingTolerance, + "backend byte accounting: node predastore allocated bytes grew %d across the N-pass workload, %.2fx guest-written %d bytes -- exceeds the %.3fx RS-erasure-floor tolerance; backend growth looks unbounded, not settled at the floor", + delta, ratio, guestWrittenBytes, npassBackendByteAccountingTolerance) } // runNPassOverwrite drives n complete format+write+fsync+detach cycles against diff --git a/tests/integration/ecr_registry_test.go b/tests/integration/ecr_registry_test.go new file mode 100644 index 000000000..6dcc7aaae --- /dev/null +++ b/tests/integration/ecr_registry_test.go @@ -0,0 +1,173 @@ +//go:build integration + +package integration + +import ( + "fmt" + "net/http" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// This file covers the ECR data plane — the OCI Distribution v2 surface the +// live ecr/ suite drove through docker, crane and skopeo. Those tests need no +// guest: they need a registry, an object store and an HTTP client, all of which +// the in-process gateway already provides on an ephemeral port. Running them +// here means a registry regression fails in the front gate rather than after a +// guest boot, and removes the last reason the ecr suite had to sit behind full +// QEMU provisioning. +// +// What is deliberately NOT moved down: the capacity/load leg, which exists to +// put real parallel pressure on a real predastore. A memory-backed object store +// cannot make that claim, so it stays live. + +// pushBlob uploads one blob through the two-step Distribution upload flow and +// returns its digest. Split out from seedOCIRepo (which pushes a fixed +// single-blob fixture) because these tests push several blobs and then read the +// exact bytes back. +func pushBlob(t *testing.T, gw *Gateway, repo, bearer string, data []byte) string { + t.Helper() + digest := ociDigest(data) + + status, hdr, body := ecrOCIRequest(t, gw, http.MethodPost, "/v2/"+repo+"/blobs/uploads/", bearer, nil) + require.Equal(t, http.StatusAccepted, status, "start blob upload: %s", body) + loc := hdr.Get("Location") + require.NotEmpty(t, loc, "upload response missing Location header") + + status, _, body = ecrOCIRequest(t, gw, http.MethodPut, loc+"?digest="+digest, bearer, data) + require.Equal(t, http.StatusCreated, status, "finish blob upload: %s", body) + return digest +} + +// pushManifest publishes a manifest referencing cfgDigest and layerDigests at +// tag, returning the digest the registry assigned it. +func pushManifest(t *testing.T, gw *Gateway, repo, bearer, tag, cfgDigest string, layerDigests ...string) string { + t.Helper() + + layers := "" + for i, d := range layerDigests { + if i > 0 { + layers += "," + } + layers += fmt.Sprintf(`{"digest":"%s"}`, d) + } + manifest := fmt.Appendf(nil, + `{"schemaVersion":2,"mediaType":"%s","config":{"digest":"%s"},"layers":[%s]}`, + ociManifestMediaType, cfgDigest, layers) + + status, hdr, body := ecrOCIRequest(t, gw, http.MethodPut, "/v2/"+repo+"/manifests/"+tag, bearer, manifest) + require.Equal(t, http.StatusCreated, status, "push manifest: %s", body) + digest := hdr.Get("Docker-Content-Digest") + require.NotEmpty(t, digest, "manifest push did not return Docker-Content-Digest") + return digest +} + +// TestECRRegistryPushPullRoundTrip is the integration-tier equivalent of the +// live docker/crane push-pull tests: push a config blob and a layer blob, push +// a manifest referencing both, then read all three back and assert the bytes +// are byte-identical to what was pushed. +// +// Asserting the returned bytes rather than only the status codes is the point. +// A registry that stores a blob under the wrong digest, truncates it, or serves +// a different blob still answers 200 to every request in this sequence; only +// comparing content catches it. +func TestECRRegistryPushPullRoundTrip(t *testing.T) { + gw := StartGateway(t) + StartECRDaemonLite(t, gw) + + repo := uniqueName("roundtrip") + _, err := gw.ECRClient(t).CreateRepository(&ecr.CreateRepositoryInput{RepositoryName: aws.String(repo)}) + require.NoError(t, err, "create-repository") + bearer := ecrGetLoginPassword(t, gw.ECRClient(t)) + + cfg := []byte(`{"architecture":"amd64","os":"linux"}`) + layer := []byte("integration-tier layer payload, deliberately not empty") + cfgDigest := pushBlob(t, gw, repo, bearer, cfg) + layerDigest := pushBlob(t, gw, repo, bearer, layer) + manifestDigest := pushManifest(t, gw, repo, bearer, "v1", cfgDigest, layerDigest) + + // Pull each blob back and compare content, not just status. + for name, want := range map[string][]byte{cfgDigest: cfg, layerDigest: layer} { + status, _, got := ecrOCIRequest(t, gw, http.MethodGet, "/v2/"+repo+"/blobs/"+name, bearer, nil) + require.Equal(t, http.StatusOK, status, "pull blob %s", name) + assert.Equal(t, want, got, "blob %s read back with different content than was pushed", name) + } + + // The manifest must be retrievable by tag and by digest, and both must + // agree — a registry that resolves a tag to a stale manifest still serves + // 200 for each request taken on its own. + statusTag, _, byTag := ecrOCIRequest(t, gw, http.MethodGet, "/v2/"+repo+"/manifests/v1", bearer, nil) + require.Equal(t, http.StatusOK, statusTag, "pull manifest by tag: %s", byTag) + statusDigest, _, byDigest := ecrOCIRequest(t, gw, http.MethodGet, "/v2/"+repo+"/manifests/"+manifestDigest, bearer, nil) + require.Equal(t, http.StatusOK, statusDigest, "pull manifest by digest: %s", byDigest) + assert.Equal(t, byTag, byDigest, "tag v1 and digest %s resolve to different manifests", manifestDigest) + assert.Equal(t, manifestDigest, ociDigest(byTag), "served manifest does not hash to the digest the registry assigned it") + + // The control plane must see what the data plane stored; the live suite + // asserted this to catch a push that lands in the object store without a + // metadata record. + out, err := gw.ECRClient(t).DescribeImages(&ecr.DescribeImagesInput{RepositoryName: aws.String(repo)}) + require.NoError(t, err, "describe-images") + var tags []string + for _, d := range out.ImageDetails { + tags = append(tags, aws.StringValueSlice(d.ImageTags)...) + } + assert.Contains(t, tags, "v1", "pushed tag not visible via DescribeImages") +} + +// TestECRRegistryPushToUncreatedRepoRejected asserts the registry does not +// auto-create a repository on push. Auto-creation is the permissive default in +// some registries, and it would silently defeat repository-level authorization: +// a principal denied on every existing repo could simply push to a new name. +func TestECRRegistryPushToUncreatedRepoRejected(t *testing.T) { + gw := StartGateway(t) + StartECRDaemonLite(t, gw) + + // A valid token for the registry, but a repository that was never created. + bearer := ecrGetLoginPassword(t, gw.ECRClient(t)) + repo := uniqueName("never-created") + + status, _, body := ecrOCIRequest(t, gw, http.MethodPost, "/v2/"+repo+"/blobs/uploads/", bearer, nil) + assert.NotEqual(t, http.StatusAccepted, status, + "registry accepted an upload for a repository that was never created (auto-create would bypass per-repository authorization): %s", body) + + // And it must not have been created as a side effect of the attempt. + _, err := gw.ECRClient(t).DescribeRepositories(&ecr.DescribeRepositoriesInput{ + RepositoryNames: []*string{aws.String(repo)}, + }) + assert.Error(t, err, "repository %s exists after a rejected push", repo) +} + +// TestECRRegistryRetagWithinRegistry covers what the live skopeo-copy test +// covered: re-publishing an existing manifest under a second tag. Both tags +// must resolve to the same digest, and the original must keep working — a +// retag that rewrites rather than aliases would break the first tag. +func TestECRRegistryRetagWithinRegistry(t *testing.T) { + gw := StartGateway(t) + StartECRDaemonLite(t, gw) + + repo := uniqueName("retag") + _, err := gw.ECRClient(t).CreateRepository(&ecr.CreateRepositoryInput{RepositoryName: aws.String(repo)}) + require.NoError(t, err, "create-repository") + bearer := ecrGetLoginPassword(t, gw.ECRClient(t)) + + cfg := []byte(`{"architecture":"amd64","os":"linux","note":"retag"}`) + cfgDigest := pushBlob(t, gw, repo, bearer, cfg) + v1Digest := pushManifest(t, gw, repo, bearer, "v1", cfgDigest) + + // Re-publish the identical manifest bytes under a second tag, which is what + // a copy within the same registry reduces to. + _, _, manifestBytes := ecrOCIRequest(t, gw, http.MethodGet, "/v2/"+repo+"/manifests/v1", bearer, nil) + status, hdr, body := ecrOCIRequest(t, gw, http.MethodPut, "/v2/"+repo+"/manifests/v2", bearer, manifestBytes) + require.Equal(t, http.StatusCreated, status, "push retagged manifest: %s", body) + assert.Equal(t, v1Digest, hdr.Get("Docker-Content-Digest"), "identical manifest content produced a different digest under the new tag") + + // The original tag must still resolve, and to the same manifest. + statusV1, _, stillV1 := ecrOCIRequest(t, gw, http.MethodGet, "/v2/"+repo+"/manifests/v1", bearer, nil) + require.Equal(t, http.StatusOK, statusV1, "original tag stopped resolving after retag: %s", stillV1) + assert.Equal(t, manifestBytes, stillV1, "original tag resolves to different content after retag") +}