fix: fail closed on incomplete instance inventory during account teardown (#1200)

Account teardown listed instances account-wide through a path that returns
whatever nodes answered. A silent or erroring node, a node answering twice,
or a failed instance-bucket query gave a short list with no error, which
teardown counted as drained.

Add DescribeAccountInstancesComplete, which requires a clean reply from
every configured node by node ID plus both instance buckets, and otherwise
returns a retryable ServiceUnavailable. Thread the configured node set into
the teardown engine from the admin API and the CLI. The engine already
aborts on a listing error, so the account stays TERMINATING and storage
teardown does not run.

cmd/spinifex/cmd/admin_account_delete.go

@@ -5,7 +5,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"maps"
 	"os"
+	"slices"
 	"strings"
 	"time"
 
@@ -90,7 +92,10 @@ func runAccountDelete(cmd *cobra.Command, args []string) error {
 		node.Predastore.SecretKey,
 	)
 
-	engine, err := accountteardown.NewClusterEngine(ctx, nc, len(cfg.Nodes), svc, buckets)
+	// The configured node set, in the namespace daemons reply under, is what
+	// the instance inventory must hear from in full before it reads as empty.
+	nodeIDs := slices.Collect(maps.Keys(cfg.Nodes))
+	engine, err := accountteardown.NewClusterEngine(ctx, nc, len(cfg.Nodes), nodeIDs, svc, buckets)
 	if err != nil {
 		return err
 	}

spinifex/accountteardown/cluster.go

@@ -21,6 +21,7 @@ func NewClusterEngine(
 	ctx context.Context,
 	nc *nats.Conn,
 	expectedNodes int,
+	nodeIDs []string,
 	svc handlers_iam.IAMService,
 	buckets BucketStore,
 ) (*Engine, error) {
@@ -64,7 +65,7 @@ func NewClusterEngine(
 	// live spot request launches a replacement for anything already terminated.
 	reapers = append(reapers, ELBv2Reapers(nc)...)
 	reapers = append(reapers, EC2ExtraReapers(nc)...)
-	reapers = append(reapers, EC2Reapers(nc, expectedNodes)...)
+	reapers = append(reapers, EC2Reapers(nc, expectedNodes, nodeIDs)...)
 
 	reapers = append(reapers, S3Reapers(buckets)...)
 	reapers = append(reapers, ACMReapers(nc)...)

spinifex/accountteardown/cluster_test.go

@@ -23,13 +23,13 @@ func TestNewClusterEngineCoversEveryStage(t *testing.T) {
 	svc, err := handlers_iam.NewIAMServiceImpl(t.Context(), nc, masterKey)
 	require.NoError(t, err)
 
-	engine, err := NewClusterEngine(t.Context(), nc, 1, svc, newFakeBuckets())
+	engine, err := NewClusterEngine(t.Context(), nc, 1, []string{"node-1"}, svc, newFakeBuckets())
 	require.NoError(t, err)
 	require.NotNil(t, engine.Accounts)
 
 	// A teardown with no object store would report every stage drained and
 	// still delete the account record, leaving the tenant's objects behind.
-	_, err = NewClusterEngine(t.Context(), nc, 1, svc, nil)
+	_, err = NewClusterEngine(t.Context(), nc, 1, []string{"node-1"}, svc, nil)
 	require.Error(t, err)
 
 	for _, stage := range Stages() {
@@ -51,7 +51,7 @@ func TestClusterEnginePrechecks(t *testing.T) {
 	svc, err := handlers_iam.NewIAMServiceImpl(t.Context(), nc, masterKey)
 	require.NoError(t, err)
 
-	engine, err := NewClusterEngine(t.Context(), nc, 1, svc, newFakeBuckets())
+	engine, err := NewClusterEngine(t.Context(), nc, 1, []string{"node-1"}, svc, newFakeBuckets())
 	require.NoError(t, err)
 
 	// Account ids are sequential and the first is the protected super admin.

spinifex/accountteardown/reapers_ec2.go

@@ -31,9 +31,9 @@ import (
 // They drive the same NATS-backed operations the gateway serves a customer
 // request with, which is what keeps OVN teardown, IPAM release and quota
 // accounting correct without teardown having to reimplement any of it.
-func EC2Reapers(nc *nats.Conn, expectedNodes int) []Reaper {
+func EC2Reapers(nc *nats.Conn, expectedNodes int, nodeIDs []string) []Reaper {
 	return []Reaper{
-		&instanceReaper{nc: nc, expectedNodes: expectedNodes},
+		&instanceReaper{nc: nc, nodeIDs: nodeIDs},
 
 		// Snapshots before volumes: a volume with a live snapshot refuses to
 		// delete, because snapshot-backed clones read from its chunk prefix.
@@ -67,17 +67,22 @@ func EC2Reapers(nc *nats.Conn, expectedNodes int) []Reaper {
 var terminalInstanceStates = map[string]bool{"terminated": true, "shutting-down": true}
 
 type instanceReaper struct {
-	nc            *nats.Conn
-	expectedNodes int
+	nc *nats.Conn
+	// nodeIDs is the configured node set; all must answer before an empty
+	// listing counts as drained.
+	nodeIDs []string
+
+	// describeOpts tunes the inventory fan-out; production leaves it empty.
+	describeOpts []gateway_ec2_instance.DescribeOption
 }
 
 func (r *instanceReaper) Kind() string { return "instance" }
 func (r *instanceReaper) Stage() Stage { return StageCompute }
 
 func (r *instanceReaper) List(ctx context.Context, accountID string) ([]Resource, error) {
-	// Checked: a partial answer here would read as "no instances left" and let
-	// storage deletion start while a guest still holds a volume.
-	out, err := gateway_ec2_instance.DescribeInstancesChecked(ctx, &ec2.DescribeInstancesInput{}, r.nc, r.expectedNodes, nil, accountID)
+	// Complete or nothing: a partial answer would read as "no instances left"
+	// and let storage deletion start while a guest still holds a volume.
+	out, err := gateway_ec2_instance.DescribeAccountInstancesComplete(ctx, r.nc, r.nodeIDs, accountID, r.describeOpts...)
 	if err != nil {
 		return nil, err
 	}

spinifex/accountteardown/reapers_ec2_inventory_test.go

@@ -0,0 +1,194 @@
+package accountteardown
+
+//test:in-package — the instance reaper is unexported, and whether its listing
+// can be mistaken for an empty account is the safety property under test.
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	gateway_ec2_instance "github.com/mulgadc/spinifex/spinifex/gateway/ec2/instance"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var inventoryNodes = []string{"node-1", "node-2"}
+
+// testInventoryReaper builds the instance reaper for a two-node cluster, with
+// a short fan-out deadline so a silent node costs the test little.
+func testInventoryReaper(nc *nats.Conn) *instanceReaper {
+	return &instanceReaper{nc: nc, nodeIDs: inventoryNodes, describeOpts: []gateway_ec2_instance.DescribeOption{
+		gateway_ec2_instance.WithFanoutTimeout(300 * time.Millisecond),
+	}}
+}
+
+// answerAsNode replies on subject the way a daemon does, carrying the node ID
+// header the fan-out attributes frames by.
+func answerAsNode(t *testing.T, nc *nats.Conn, subject, nodeID string, payload []byte) {
+	t.Helper()
+	sub, err := nc.Subscribe(subject, func(msg *nats.Msg) {
+		reply := nats.NewMsg(msg.Reply)
+		reply.Data = payload
+		if nodeID != "" {
+			reply.Header.Set(utils.NodeIDHeader, nodeID)
+		}
+		_ = msg.RespondMsg(reply)
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+}
+
+func describeOutput(t *testing.T, instanceIDs ...string) []byte {
+	t.Helper()
+	out := ec2.DescribeInstancesOutput{Reservations: []*ec2.Reservation{}}
+	if len(instanceIDs) > 0 {
+		reservation := &ec2.Reservation{ReservationId: aws.String("r-" + instanceIDs[0])}
+		for _, id := range instanceIDs {
+			reservation.Instances = append(reservation.Instances, &ec2.Instance{
+				InstanceId: aws.String(id),
+				State:      &ec2.InstanceState{Name: aws.String("running")},
+			})
+		}
+		out.Reservations = append(out.Reservations, reservation)
+	}
+	payload, err := json.Marshal(out)
+	require.NoError(t, err)
+	return payload
+}
+
+// emptyInstanceBuckets answers the stopped and terminated bucket queries with
+// nothing, so only the running-instance fan-out decides completeness.
+func emptyInstanceBuckets(t *testing.T, nc *nats.Conn) {
+	t.Helper()
+	answerAsNode(t, nc, "ec2.DescribeStoppedInstances", "", describeOutput(t))
+	answerAsNode(t, nc, "ec2.DescribeTerminatedInstances", "", describeOutput(t))
+}
+
+// A node that never answers is the timeout case: its running guests are absent
+// from the aggregate, so an empty answer from the rest is not evidence the
+// account has no instances.
+func TestInstanceReaperFailsClosedWhenANodeDoesNotAnswer(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	emptyInstanceBuckets(t, nc)
+	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t))
+	require.NoError(t, nc.Flush())
+
+	found, err := testInventoryReaper(nc).List(testCtx(t), testAccountID)
+
+	require.Error(t, err, "a listing missing a node must not read as an empty account, got %v", found)
+	code, _, ok := awserrors.ResolveErrorDetail(err)
+	require.True(t, ok)
+	assert.Equal(t, awserrors.ErrorServiceUnavailable, code)
+}
+
+// The partial list is the dangerous shape: the reaper would terminate what it
+// can see and then report the stage drained on the next, equally partial pass.
+func TestInstanceReaperFailsClosedOnAPartialList(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	emptyInstanceBuckets(t, nc)
+	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t, "i-on-node-1"))
+	require.NoError(t, nc.Flush())
+
+	found, err := testInventoryReaper(nc).List(testCtx(t), testAccountID)
+
+	require.Error(t, err, "a partial listing must not be returned as the inventory, got %v", found)
+}
+
+// An error frame from a node is no better than silence: it says nothing about
+// what that node is running.
+func TestInstanceReaperFailsClosedWhenANodeAnswersWithAnError(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	emptyInstanceBuckets(t, nc)
+	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t))
+	answerAsNode(t, nc, "ec2.DescribeInstances", "node-2", utils.GenerateErrorPayload(awserrors.ErrorServerInternal))
+	require.NoError(t, nc.Flush())
+
+	_, err := testInventoryReaper(nc).List(testCtx(t), testAccountID)
+
+	require.Error(t, err)
+}
+
+// Completeness is judged against the configured node set, not a count of
+// replies: two frames from one node while another is silent is still missing
+// a node.
+func TestInstanceReaperFailsClosedWhenOneNodeAnswersTwice(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	emptyInstanceBuckets(t, nc)
+	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t))
+	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t))
+	require.NoError(t, nc.Flush())
+
+	_, err := testInventoryReaper(nc).List(testCtx(t), testAccountID)
+
+	require.Error(t, err)
+}
+
+// Stopped instances live in a bucket, not on a node. A failed bucket query
+// hides them as surely as a silent node hides running ones.
+func TestInstanceReaperFailsClosedWhenTheStoppedBucketFails(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	answerAsNode(t, nc, "ec2.DescribeStoppedInstances", "", utils.GenerateErrorPayload(awserrors.ErrorServerInternal))
+	answerAsNode(t, nc, "ec2.DescribeTerminatedInstances", "", describeOutput(t))
+	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t))
+	answerAsNode(t, nc, "ec2.DescribeInstances", "node-2", describeOutput(t))
+	require.NoError(t, nc.Flush())
+
+	_, err := testInventoryReaper(nc).List(testCtx(t), testAccountID)
+
+	require.Error(t, err)
+}
+
+// The healthy cluster still lists: every configured node answered, so the
+// union is the account's inventory.
+func TestInstanceReaperListsEveryNodeWhenAllAnswer(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	emptyInstanceBuckets(t, nc)
+	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t, "i-on-node-1"))
+	answerAsNode(t, nc, "ec2.DescribeInstances", "node-2", describeOutput(t, "i-on-node-2"))
+	require.NoError(t, nc.Flush())
+
+	started := time.Now()
+	found, err := testInventoryReaper(nc).List(testCtx(t), testAccountID)
+	require.NoError(t, err)
+
+	var ids []string
+	for _, resource := range found {
+		ids = append(ids, resource.ID)
+	}
+	assert.ElementsMatch(t, []string{"i-on-node-1", "i-on-node-2"}, ids)
+	assert.Less(t, time.Since(started), 2*time.Second, "a fully answered fan-out must not wait out its deadline")
+}
+
+// End to end through the engine: with a node silent, the compute stage must
+// not read as drained and the account record must survive.
+func TestTeardownDoesNotReportComputeDrainedWhileANodeIsSilent(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	emptyInstanceBuckets(t, nc)
+	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t))
+	require.NoError(t, nc.Flush())
+
+	accounts := newFakeAccounts()
+	volumes := &fakeReaper{kind: "volume", stage: StageStorage,
+		resources: []Resource{{Kind: "volume", ID: "vol-held-by-node-2"}}}
+	engine := quietEngine(accounts, testInventoryReaper(nc), volumes)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	result, err := engine.Teardown(ctx, request())
+
+	require.Error(t, err, "teardown must not succeed on an incomplete instance inventory")
+	assert.ErrorContains(t, err, "list instance")
+	assert.False(t, accounts.deleted, "the account record must survive an incomplete inventory")
+	for _, stage := range result.Stages {
+		assert.NotEqual(t, StageCompute, stage.Stage, "the compute stage must not be reported finished")
+	}
+	assert.Equal(t, 0, volumes.deletesAttempted, "storage must not be touched before compute is proven drained")
+}

spinifex/accountteardown/reapers_ec2_test.go

@@ -53,6 +53,10 @@ func newFakeCluster(t *testing.T) *fakeCluster {
 	return cluster
 }
 
+// fakeClusterNode is the one node the fake cluster answers as. Replies carry
+// its ID because an inventory fan-out only counts attributable frames.
+const fakeClusterNode = "node-1"
+
 func (c *fakeCluster) serve(msg *nats.Msg) {
 	c.mu.Lock()
 	c.calls = append(c.calls, msg.Subject)
@@ -61,18 +65,24 @@ func (c *fakeCluster) serve(msg *nats.Msg) {
 	reply, hasReply := c.replies[msg.Subject]
 	c.mu.Unlock()
 
+	respond := func(data []byte) {
+		out := nats.NewMsg(msg.Reply)
+		out.Data = data
+		out.Header.Set(utils.NodeIDHeader, fakeClusterNode)
+		_ = msg.RespondMsg(out)
+	}
 	switch {
 	case isError:
-		_ = msg.Respond(utils.GenerateErrorPayload(code))
+		respond(utils.GenerateErrorPayload(code))
 	case hasReply:
 		payload, err := json.Marshal(reply)
 		if err != nil {
-			_ = msg.Respond(utils.GenerateErrorPayload("InternalError"))
+			respond(utils.GenerateErrorPayload("InternalError"))
 			return
 		}
-		_ = msg.Respond(payload)
+		respond(payload)
 	default:
-		_ = msg.Respond([]byte(`{}`))
+		respond([]byte(`{}`))
 	}
 }
 
@@ -112,7 +122,7 @@ func testCtx(t *testing.T) context.Context {
 // The stage graph is what keeps teardown converging, so it is asserted rather
 // than left to the order the constructor happens to list.
 func TestEC2ReapersAreOrderedByDependency(t *testing.T) {
-	reapers := EC2Reapers(nil, 1)
+	reapers := EC2Reapers(nil, 1, []string{"node-1"})
 
 	var kinds []string
 	for _, reaper := range reapers {
@@ -157,7 +167,7 @@ func TestInstanceReaperSkipsTerminalStates(t *testing.T) {
 		}}},
 	})
 
-	reaper := &instanceReaper{nc: cluster.nc, expectedNodes: 1}
+	reaper := &instanceReaper{nc: cluster.nc, nodeIDs: []string{fakeClusterNode}}
 	found, err := reaper.List(testCtx(t), "000000000042")
 	require.NoError(t, err)
 
@@ -170,7 +180,7 @@ func TestInstanceReaperSkipsTerminalStates(t *testing.T) {
 // makes a stuck host the case --force exists for.
 func TestInstanceReaperTerminates(t *testing.T) {
 	cluster := newFakeCluster(t)
-	reaper := &instanceReaper{nc: cluster.nc, expectedNodes: 1}
+	reaper := &instanceReaper{nc: cluster.nc, nodeIDs: []string{fakeClusterNode}}
 
 	require.NoError(t, reaper.Delete(testCtx(t), "000000000042",
 		Resource{Kind: "instance", ID: "i-0abc0123456789def"}, false))
@@ -679,7 +689,7 @@ func TestNetworkInterfaceReaperTreatsAMissingInterfaceAsDeleted(t *testing.T) {
 // subnet delete is refused while an interface is still in it.
 func TestEC2ReapersRemoveInterfacesBeforeSubnets(t *testing.T) {
 	var eni, subnet = -1, -1
-	for i, reaper := range EC2Reapers(nil, 1) {
+	for i, reaper := range EC2Reapers(nil, 1, []string{"node-1"}) {
 		switch reaper.Kind() {
 		case "network-interface":
 			eni = i

spinifex/accountteardown/reapers_ecs_test.go

@@ -178,7 +178,7 @@ func TestECSClusterNameFromARN(t *testing.T) {
 // stopped first, which is what puts ECS ahead of it in the compute stage.
 func TestECSReapersRunBeforeTheInstanceReaper(t *testing.T) {
 	reapers := ECSReapers(nil)
-	reapers = append(reapers, EC2Reapers(nil, 1)...)
+	reapers = append(reapers, EC2Reapers(nil, 1, []string{"node-1"})...)
 
 	ecsIndex := indexOfKind(reapers, "ecs-cluster")
 	instanceIndex := indexOfKind(reapers, "instance")
@@ -197,7 +197,7 @@ func TestECSReapersRunBeforeTheInstanceReaper(t *testing.T) {
 // the registration order above survives it.
 func TestSortReapersKeepsECSAheadOfInstances(t *testing.T) {
 	reapers := ECSReapers(nil)
-	reapers = append(reapers, EC2Reapers(nil, 1)...)
+	reapers = append(reapers, EC2Reapers(nil, 1, []string{"node-1"})...)
 	SortReapers(reapers)
 
 	assert.Less(t, indexOfKind(reapers, "ecs-cluster"), indexOfKind(reapers, "instance"))

spinifex/accountteardown/reapers_s3_test.go

@@ -277,7 +277,7 @@ func TestS3ReaperRunsInTheStorageStage(t *testing.T) {
 // The engine sorts reapers by stage and the sort is stable, so a bucket reaper
 // registered after the EC2 storage reapers stays after them.
 func TestSortReapersKeepsBucketsAfterVolumes(t *testing.T) {
-	reapers := append(EC2Reapers(nil, 1), S3Reapers(newFakeBuckets())...)
+	reapers := append(EC2Reapers(nil, 1, []string{"node-1"}), S3Reapers(newFakeBuckets())...)
 	SortReapers(reapers)
 
 	storage := make([]string, 0, len(reapers))

spinifex/gateway/admin_deleteaccount.go

@@ -111,7 +111,7 @@ func (gw *GatewayConfig) adminDeleteAccount(ctx context.Context, body []byte) (a
 		return nil, errors.New(awserrors.ErrorInternalError)
 	}
 
-	engine, err := accountteardown.NewClusterEngine(ctx, gw.NATSConn, gw.ExpectedNodes, gw.IAMService, gw.BucketStore)
+	engine, err := accountteardown.NewClusterEngine(ctx, gw.NATSConn, gw.ExpectedNodes, gw.NodeIDs, gw.IAMService, gw.BucketStore)
 	if err != nil {
 		slog.Error("DeleteAccount: failed to build teardown engine", "err", err)
 		return nil, errors.New(awserrors.ErrorServiceUnavailable)

spinifex/gateway/admin_deleteaccount_job_test.go

@@ -17,6 +17,7 @@ import (
 	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
 	"github.com/mulgadc/spinifex/spinifex/objectstore"
 	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
@@ -44,8 +45,13 @@ func newDeleteAccountGateway(t *testing.T) *GatewayConfig {
 	require.NoError(t, err)
 
 	for _, subject := range teardownSubjects {
+		// Replies carry a node ID: the instance inventory only counts a frame
+		// it can attribute to a configured node.
 		sub, err := nc.Subscribe(subject, func(msg *nats.Msg) {
-			_ = msg.Respond([]byte(`{}`))
+			reply := nats.NewMsg(msg.Reply)
+			reply.Data = []byte(`{}`)
+			reply.Header.Set(utils.NodeIDHeader, "node-1")
+			_ = msg.RespondMsg(reply)
 		})
 		require.NoError(t, err)
 		t.Cleanup(func() { _ = sub.Unsubscribe() })
@@ -57,6 +63,7 @@ func newDeleteAccountGateway(t *testing.T) *GatewayConfig {
 		IAMService:     svc,
 		NATSConn:       nc,
 		ExpectedNodes:  1,
+		NodeIDs:        []string{"node-1"},
 		BucketStore:    emptyBucketStore{},
 	}
 }

spinifex/gateway/ec2/instance/DescribeInstances.go

@@ -177,6 +177,34 @@ func DescribeInstancesForReconcile(ctx context.Context, input *ec2.DescribeInsta
 	return reservations, complete, err
 }
 
+// DescribeAccountInstancesComplete lists the account's instances, or returns a
+// retryable ServiceUnavailable unless every node in nodeIDs and both instance
+// buckets answered cleanly. An empty nodeIDs is an unknown set, so never complete.
+func DescribeAccountInstancesComplete(ctx context.Context, natsConn *nats.Conn, nodeIDs []string, accountID string, opts ...DescribeOption) (*ec2.DescribeInstancesOutput, error) {
+	if len(nodeIDs) == 0 {
+		slog.ErrorContext(ctx, "DescribeInstances: complete listing refused, configured node set unknown")
+		return nil, incompleteListingError()
+	}
+	reservations, complete, _, err := gatherInstances(ctx, &ec2.DescribeInstancesInput{}, natsConn, 0, nodeIDs, accountID, opts...)
+	if err != nil {
+		return nil, err
+	}
+	if !complete {
+		slog.WarnContext(ctx, "DescribeInstances: complete listing refused, fan-out or bucket query incomplete",
+			"configured_nodes", len(nodeIDs))
+		return nil, incompleteListingError()
+	}
+	return &ec2.DescribeInstancesOutput{Reservations: reservations}, nil
+}
+
+func incompleteListingError() error {
+	return errors.Join(
+		awserrors.Errorf(awserrors.ErrorServiceUnavailable,
+			"cannot list instances: not every node and instance store answered"),
+		awserrors.RetryAfter(awserrors.ErrorServiceUnavailable, describeIncompleteRetryAfter),
+	)
+}
+
 // gatherInstances runs the running-instance fan-out plus the stopped/terminated
 // KV bucket queries and aggregates every reservation. client4xx carries
 // the first deterministic 4xx for the lenient caller to surface when nothing

spinifex/gateway/ec2/instance/DescribeInstances_test.go

@@ -956,3 +956,115 @@ func TestAcquireAbsenceProofSlot_DoesNotBlockWhenExhausted(t *testing.T) {
 		t.Fatal("acquireAbsenceProofSlot blocked; it must refuse immediately so the backlog cannot grow")
 	}
 }
+
+// --- Complete account-wide listing ---
+
+func assertIncompleteListing(t *testing.T, output *ec2.DescribeInstancesOutput, err error) {
+	t.Helper()
+	assert.Nil(t, output, "an incomplete listing must not return the partial reservations")
+	require.Error(t, err)
+	code, _, ok := awserrors.ResolveErrorDetail(err)
+	require.True(t, ok, "error must carry a registered awserrors code")
+	assert.Equal(t, awserrors.ErrorServiceUnavailable, code)
+	assert.False(t, awserrors.IsTerminal(err), "an incomplete listing is retryable")
+}
+
+func TestDescribeAccountInstancesComplete_MissingNodeRefuses(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeEmptyInstanceBuckets(t, nc)
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", mustMarshalDescribeOutput(t, &ec2.DescribeInstancesOutput{
+		Reservations: []*ec2.Reservation{{ReservationId: aws.String("r-1"), Instances: []*ec2.Instance{{InstanceId: aws.String("i-on-node-1")}}}},
+	}))
+	nc.Flush()
+
+	output, err := DescribeAccountInstancesComplete(context.Background(), nc, []string{"node-1", "node-2"}, "123456789012",
+		WithFanoutTimeout(300*time.Millisecond))
+
+	assertIncompleteListing(t, output, err)
+}
+
+func TestDescribeAccountInstancesComplete_AllNodesAnswer(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeEmptyInstanceBuckets(t, nc)
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", mustMarshalDescribeOutput(t, &ec2.DescribeInstancesOutput{
+		Reservations: []*ec2.Reservation{{ReservationId: aws.String("r-1"), Instances: []*ec2.Instance{{InstanceId: aws.String("i-on-node-1")}}}},
+	}))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-2", []byte(`{}`))
+	nc.Flush()
+
+	output, err := DescribeAccountInstancesComplete(context.Background(), nc, []string{"node-1", "node-2"}, "123456789012",
+		WithFanoutTimeout(2*time.Second))
+
+	require.NoError(t, err)
+	require.Len(t, output.Reservations, 1)
+	assert.Equal(t, "i-on-node-1", *output.Reservations[0].Instances[0].InstanceId)
+}
+
+// Without the configured node set there is nothing to judge completeness
+// against, and counting whoever answered is the failure being prevented.
+func TestDescribeAccountInstancesComplete_UnknownNodeSetRefuses(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeEmptyInstanceBuckets(t, nc)
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	nc.Flush()
+
+	output, err := DescribeAccountInstancesComplete(context.Background(), nc, nil, "123456789012",
+		WithFanoutTimeout(300*time.Millisecond))
+
+	assertIncompleteListing(t, output, err)
+}
+
+// A frame with no node ID cannot be attributed, so it might be standing in
+// for the node that went quiet.
+func TestDescribeAccountInstancesComplete_UnattributedFrameRefuses(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeEmptyInstanceBuckets(t, nc)
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "", []byte(`{}`))
+	nc.Flush()
+
+	output, err := DescribeAccountInstancesComplete(context.Background(), nc, []string{"node-1", "node-2"}, "123456789012",
+		WithFanoutTimeout(300*time.Millisecond))
+
+	assertIncompleteListing(t, output, err)
+}
+
+func TestDescribeAccountInstancesComplete_TerminatedBucketFailureRefuses(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAsNode(t, nc, "ec2.DescribeStoppedInstances", "", []byte(`{}`))
+	subscribeAsNode(t, nc, "ec2.DescribeTerminatedInstances", "", utils.GenerateErrorPayload(awserrors.ErrorServerInternal))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	nc.Flush()
+
+	output, err := DescribeAccountInstancesComplete(context.Background(), nc, []string{"node-1"}, "123456789012",
+		WithFanoutTimeout(2*time.Second))
+
+	assertIncompleteListing(t, output, err)
+}
+
+// Pins the lenient by-ID shape: an instance on a silent node comes back as an
+// empty success, indistinguishable from absence. Anything that must prove an
+// instance gone has to use DescribeInstancesChecked with the node set instead.
+func TestDescribeInstances_ExplicitIDOnSilentNode_IsAnEmptySuccess(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeEmptyInstanceBuckets(t, nc)
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{"Reservations":[]}`))
+	nc.Flush()
+
+	input := &ec2.DescribeInstancesInput{InstanceIds: []*string{aws.String("i-on-the-silent-node")}}
+	output, err := DescribeInstances(context.Background(), input, nc, 2, "123456789012",
+		WithFanoutTimeout(300*time.Millisecond))
+
+	require.NoError(t, err)
+	assert.Empty(t, output.Reservations)
+
+	_, err = DescribeInstancesChecked(context.Background(), input, nc, 0, []string{"node-1", "node-2"}, "123456789012",
+		WithFanoutTimeout(300*time.Millisecond))
+	assertIncompleteSweepUnavailable(t, err, "i-on-the-silent-node")
+}