fix(nebius): sweep orphaned validation CI resources and make TerminateInstance idempotent

.github/workflows/validation-nebius.yml

@@ -42,10 +42,25 @@ jobs:
                   TEST_PRIVATE_KEY_BASE64: ${{ secrets.TEST_PRIVATE_KEY_BASE64 }}
                   TEST_PUBLIC_KEY_BASE64: ${{ secrets.TEST_PUBLIC_KEY_BASE64 }}
                   VALIDATION_TEST: true
+                  # Stamps this run's resources for the sweep below.
+                  CI_RUN_ID: ${{ github.run_id }}
               run: |
                   cd v1/providers/nebius
                   go test -v -short=false -timeout=30m ./...
 
+            # Deletes any VM/network/subnet/disk this run left behind. Matches only this run's
+            # ci-run-id label, so it never touches another run's or a production resource.
+            - name: Sweep leaked validation/integration resources
+              if: always()
+              env:
+                  NEBIUS_SERVICE_ACCOUNT_JSON: ${{ secrets.NEBIUS_SERVICE_ACCOUNT_JSON }}
+                  NEBIUS_TENANT_ID: ${{ secrets.NEBIUS_TENANT_ID }}
+                  SWEEP_ORPHANS: "true"
+                  CI_RUN_ID: ${{ github.run_id }}
+              run: |
+                  cd v1/providers/nebius
+                  go test -v -run '^TestSweepOrphans$' -count=1 -timeout=20m .
+
             - name: Upload test results
               uses: actions/upload-artifact@v4
               if: always()

internal/validation/suite.go

@@ -2,6 +2,7 @@ package validation
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -14,6 +15,36 @@ type ProviderConfig struct {
 	Location   string
 	StableIDs  []v1.InstanceTypeID
 	Credential v1.CloudCredential
+	// Tags are extra labels (e.g. the CI run ID) stamped on every instance and its
+	// network/subnet/disk so a post-run sweep can delete this run's resources.
+	Tags map[string]string
+}
+
+// registerInstanceCleanup schedules termination via t.Cleanup so it runs even after a t.Fatalf,
+// on a fresh 8-min context. Not-found counts as success; a terminal failure fails the test.
+// Returns markTerminated, which the caller invokes once it has terminated the instance itself to
+// skip the redundant delete. Call this right after a create, before any assertion.
+func registerInstanceCleanup(t *testing.T, client v1.CloudCreateTerminateInstance, cloudID v1.CloudProviderInstanceID) (markTerminated func()) {
+	t.Helper()
+	terminated := false
+	markTerminated = func() { terminated = true }
+	if cloudID == "" {
+		return markTerminated
+	}
+	t.Cleanup(func() {
+		if terminated {
+			return
+		}
+		// Single 8-min attempt on a fresh context (real terminates ~1 min; internal wait caps at
+		// 5 minutes).
+		ctx, cancel := context.WithTimeout(context.Background(), 8*time.Minute)
+		defer cancel()
+		err := client.TerminateInstance(ctx, cloudID)
+		if err != nil && !errors.Is(err, v1.ErrInstanceNotFound) && !errors.Is(err, v1.ErrResourceNotFound) {
+			t.Errorf("LEAKED INSTANCE %s: cleanup terminate failed: %v", cloudID, err)
+		}
+	})
+	return markTerminated
 }
 
 func RunValidationSuite(t *testing.T, config ProviderConfig) {
@@ -123,18 +154,19 @@ func RunInstanceLifecycleValidation(t *testing.T, config ProviderConfig) {
 				break
 			}
 		}
+		attrs.Tags = config.Tags
 		instance, err := v1.ValidateCreateInstance(ctx, client, attrs, selectedType)
+		// Register cleanup before the fatal below: create can return a non-nil instance with an
+		// error, and a t.Fatalf would skip a later defer and leak the VM.
+		markTerminated := func() {}
+		if instance != nil {
+			markTerminated = registerInstanceCleanup(t, client, instance.CloudID)
+		}
 		if err != nil {
 			t.Fatalf("ValidateCreateInstance failed: %v", err)
 		}
 		require.NotNil(t, instance)
 
-		defer func() {
-			if instance != nil {
-				_ = client.TerminateInstance(ctx, instance.CloudID)
-			}
-		}()
-
 		t.Run("ValidateListCreatedInstance", func(t *testing.T) {
 			err := v1.ValidateListCreatedInstance(ctx, client, instance)
 			require.NoError(t, err, "ValidateListCreatedInstance should pass")
@@ -165,6 +197,7 @@ func RunInstanceLifecycleValidation(t *testing.T, config ProviderConfig) {
 		t.Run("ValidateTerminateInstance", func(t *testing.T) {
 			err := v1.ValidateTerminateInstance(ctx, client, instance)
 			require.NoError(t, err, "ValidateTerminateInstance should pass")
+			markTerminated() // already terminated; skip redundant delete
 		})
 	})
 }
@@ -312,16 +345,16 @@ func RunFirewallValidation(t *testing.T, config ProviderConfig, opts FirewallVal
 	require.NotEmpty(t, attrs.InstanceType, "Should find available instance type")
 
 	// Create instance for firewall testing
+	attrs.Tags = config.Tags
 	instance, err := v1.ValidateCreateInstance(ctx, client, attrs, selectedType)
+	// Register cleanup before the assertions below so a failed require cannot leak the VM.
+	markTerminated := func() {}
+	if instance != nil {
+		markTerminated = registerInstanceCleanup(t, client, instance.CloudID)
+	}
 	require.NoError(t, err, "ValidateCreateInstance should pass")
 	require.NotNil(t, instance)
 
-	defer func() {
-		if instance != nil {
-			_ = client.TerminateInstance(ctx, instance.CloudID)
-		}
-	}()
-
 	// Wait for instance to be running and SSH accessible
 	t.Run("ValidateSSHAccessible", func(t *testing.T) {
 		err := v1.ValidateInstanceSSHAccessible(ctx, client, instance, ssh.GetTestPrivateKey())
@@ -349,6 +382,7 @@ func RunFirewallValidation(t *testing.T, config ProviderConfig, opts FirewallVal
 	t.Run("ValidateTerminateInstance", func(t *testing.T) {
 		err := v1.ValidateTerminateInstance(ctx, client, instance)
 		require.NoError(t, err, "ValidateTerminateInstance should pass")
+		markTerminated() // already terminated; skip redundant delete
 	})
 }
 

internal/validation/sweep.go

@@ -0,0 +1,85 @@
+package validation
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	v1 "github.com/brevdev/cloud/v1"
+)
+
+// CIRunIDLabel re-exports v1.CIRunIDLabel so the sweeper and create paths share one value.
+const CIRunIDLabel = v1.CIRunIDLabel
+
+// SweepOpts configures SweepOrphanedInstances.
+type SweepOpts struct {
+	// RunID is the CIRunIDLabel label value to match. It is REQUIRED and must be unique per CI run.
+	RunID string
+	// Logf, if set, receives human-readable progress lines. Defaults to fmt.Printf.
+	Logf func(format string, args ...any)
+}
+
+// SweepResult reports what a sweep scanned and did.
+type SweepResult struct {
+	Scanned int
+	Matched []v1.Instance
+	Deleted []v1.CloudProviderInstanceID
+	Failed  map[v1.CloudProviderInstanceID]error
+}
+
+// SweepOrphanedInstances lists every instance visible to client and terminates exactly the ones
+// whose CIRunIDLabel label equals opts.RunID — i.e. the VMs created by this CI run.
+func SweepOrphanedInstances(ctx context.Context, client v1.CloudClient, opts SweepOpts) (SweepResult, error) {
+	res := SweepResult{Failed: map[v1.CloudProviderInstanceID]error{}}
+	logf := opts.Logf
+	if logf == nil {
+		logf = func(format string, args ...any) { fmt.Printf(format+"\n", args...) }
+	}
+
+	// A run ID is mandatory: without it there is nothing safe to match on.
+	if opts.RunID == "" {
+		return res, errors.New("SweepOrphanedInstances: RunID is required")
+	}
+
+	instances, err := client.ListInstances(ctx, v1.ListInstancesArgs{})
+	if err != nil {
+		return res, fmt.Errorf("failed to list instances: %w", err)
+	}
+	res.Scanned = len(instances)
+
+	for i := range instances {
+		inst := instances[i]
+		if inst.Tags[CIRunIDLabel] != opts.RunID {
+			continue
+		}
+		res.Matched = append(res.Matched, inst)
+	}
+
+	logf("[sweep] scanned=%d matched=%d runID=%q", res.Scanned, len(res.Matched), opts.RunID)
+
+	for i := range res.Matched {
+		inst := res.Matched[i]
+		logf("[sweep] terminating cloudID=%s name=%q owner=%s location=%s",
+			inst.CloudID, inst.Name, inst.CloudCredRefID, inst.Location)
+		if err := terminateWithTimeout(ctx, client, inst.CloudID); err != nil {
+			res.Failed[inst.CloudID] = err
+			logf("[sweep] FAILED to terminate cloudID=%s: %v", inst.CloudID, err)
+			continue
+		}
+		res.Deleted = append(res.Deleted, inst.CloudID)
+	}
+	return res, nil
+}
+
+// terminateWithTimeout terminates an instance in a single attempt on an 8-min context, treating a
+// not-found result as success (the instance is already gone).
+func terminateWithTimeout(ctx context.Context, client v1.CloudCreateTerminateInstance, id v1.CloudProviderInstanceID) error {
+	termCtx, cancel := context.WithTimeout(ctx, 8*time.Minute)
+	defer cancel()
+	err := client.TerminateInstance(termCtx, id)
+	if err == nil || errors.Is(err, v1.ErrInstanceNotFound) || errors.Is(err, v1.ErrResourceNotFound) {
+		return nil
+	}
+	return err
+}

internal/validation/sweep_test.go

@@ -0,0 +1,104 @@
+package validation
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	v1 "github.com/brevdev/cloud/v1"
+)
+
+// fakeSweepClient is a CloudClient whose ListInstances returns a canned set and whose
+// TerminateInstance records the IDs it was asked to delete. All other methods come from
+// NotImplCloudClient and return ErrNotImplemented (the sweeper never calls them).
+type fakeSweepClient struct {
+	v1.NotImplCloudClient
+	instances  []v1.Instance
+	terminated []v1.CloudProviderInstanceID
+}
+
+var _ v1.CloudClient = (*fakeSweepClient)(nil)
+
+func (c *fakeSweepClient) ListInstances(context.Context, v1.ListInstancesArgs) ([]v1.Instance, error) {
+	return c.instances, nil
+}
+
+func (c *fakeSweepClient) TerminateInstance(_ context.Context, id v1.CloudProviderInstanceID) error {
+	c.terminated = append(c.terminated, id)
+	return nil
+}
+
+func matchedIDs(res SweepResult) []v1.CloudProviderInstanceID {
+	ids := make([]v1.CloudProviderInstanceID, 0, len(res.Matched))
+	for _, m := range res.Matched {
+		ids = append(ids, m.CloudID)
+	}
+	return ids
+}
+
+func assertSameSet(t *testing.T, got, want []v1.CloudProviderInstanceID) {
+	t.Helper()
+	gm := map[v1.CloudProviderInstanceID]bool{}
+	for _, g := range got {
+		gm[g] = true
+	}
+	wm := map[v1.CloudProviderInstanceID]bool{}
+	for _, w := range want {
+		wm[w] = true
+	}
+	for w := range wm {
+		if !gm[w] {
+			t.Fatalf("missing %q: got %v want %v", w, got, want)
+		}
+	}
+	for g := range gm {
+		if !wm[g] {
+			t.Fatalf("unexpected %q: got %v want %v", g, got, want)
+		}
+	}
+}
+
+// TestSweepOrphanedInstances_OnlyMatchesRunID is the safety-critical test: the sweep terminates
+// exactly the VMs carrying this run's ci-run-id label and nothing else.
+func TestSweepOrphanedInstances_OnlyMatchesRunID(t *testing.T) {
+	old := time.Now().Add(-10 * 24 * time.Hour)
+	fresh := time.Now().Add(-1 * time.Minute)
+
+	instances := []v1.Instance{
+		// This run's VMs — must be deleted, regardless of age.
+		{CloudID: "A-thisrun", CloudCredRefID: "validation-test", CreatedAt: fresh, Tags: v1.Tags{CIRunIDLabel: "run-1"}},
+		{CloudID: "B-thisrun-int", CloudCredRefID: "integration-test-ref", CreatedAt: fresh, Tags: v1.Tags{CIRunIDLabel: "run-1"}},
+		// Another run's VM — must NOT be deleted (different run id).
+		{CloudID: "C-otherrun", CloudCredRefID: "validation-test", CreatedAt: old, Tags: v1.Tags{CIRunIDLabel: "run-2"}},
+		// Legacy CI VM with no run label (predates this feature) — must NOT be deleted (manual cleanup).
+		{CloudID: "D-legacy-nolabel", CloudCredRefID: "validation-test", CreatedAt: old},
+		// Production VM that reused a CI credential ref but has no run label — must NOT be deleted.
+		{CloudID: "E-prod-reused-cred", CloudCredRefID: "validation-test", CreatedAt: old},
+		// Ordinary production VM — must NOT be deleted.
+		{CloudID: "F-prod", CloudCredRefID: "cloud-cred-abc123", CreatedAt: old},
+	}
+
+	fc := &fakeSweepClient{instances: instances}
+	res, err := SweepOrphanedInstances(context.Background(), fc, SweepOpts{RunID: "run-1", Logf: t.Logf})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	want := []v1.CloudProviderInstanceID{"A-thisrun", "B-thisrun-int"}
+	assertSameSet(t, matchedIDs(res), want)
+	assertSameSet(t, res.Deleted, want)
+	assertSameSet(t, fc.terminated, want)
+}
+
+// TestSweepOrphanedInstances_RequiresRunID pins the safety invariant: without a run ID the sweep
+// refuses to run and terminates nothing.
+func TestSweepOrphanedInstances_RequiresRunID(t *testing.T) {
+	fc := &fakeSweepClient{instances: []v1.Instance{
+		{CloudID: "x", CloudCredRefID: "validation-test", Tags: v1.Tags{CIRunIDLabel: "run-1"}},
+	}}
+	if _, err := SweepOrphanedInstances(context.Background(), fc, SweepOpts{RunID: ""}); err == nil {
+		t.Error("expected an error when RunID is empty")
+	}
+	if len(fc.terminated) != 0 {
+		t.Errorf("empty RunID must terminate nothing, terminated: %v", fc.terminated)
+	}
+}

v1/providers/nebius/errors.go

@@ -27,6 +27,10 @@ func (e *NebiusError) Error() string {
 
 // isNotFoundError checks if an error is a "not found" error
 func isNotFoundError(err error) bool {
+	// Also match the v1 sentinels, for errors already mapped by callers like TerminateInstance.
+	if errors.Is(err, v1.ErrInstanceNotFound) || errors.Is(err, v1.ErrResourceNotFound) {
+		return true
+	}
 	// Check for gRPC NotFound status code
 	if status, ok := status.FromError(err); ok {
 		return status.Code() == codes.NotFound