fix: stop the detach seal, EIP teardown and key-pair races failing the nightly (#1003)

* fix: stop the detach seal, EIP teardown and key-pair races failing the nightly

Four independent faults the full-suite nightly surfaced, each of which had
been masked while the heavy cells were skipped.

unmountVolume SIGKILLed nbdkit the moment the request arrived. Removing the
guest device has already started the plugin's own Close, which drains to the
backend and leaves a seal receipt, so the kill landed 61ms into that flush and
threw it away — the receipt path never fired once across a whole run. What was
left was 74 orphaned WAL files for the fallback seal to replay, which ran past
the caller's 180s deadline and returned DetachVolume ServerInternal with the
volume still attached. It now terminates gracefully first, bounded by the
plugin's own drain budget, and only escalates if that does not take.

DeleteEIP returned early when the dnat_and_snat row was already absent, before
the neighbour flush and the host unbind. An absent row is not a reassignment —
nobody owns the address — so the /32 route and proxy-ARP entry were orphaned
and a disassociate left the address still answering from the host. The
reassignment cases keep their early return.

EnsureKeyPair was not atomic: a failure after CreateKeyPair left the pair on
the cluster with no cleanup registered, so every later caller failed with
InvalidKeyPair.Duplicate in a test that never ran. The describe that follows a
create also had no tolerance for the propagation window on a multi-node
cluster; it now retries, and still reports a duplicate the moment it sees one.

An admission refusal could not name its cause. The static vCPU/memory budget
and the live MemAvailable clamp both surface as InsufficientInstanceCapacity,
and they have different fixes; the subscription transition now records which
gate closed.

* fix: identify the default-deny by priority, not by being a drop

DenyIsNotEthertypeScoped bucketed every drop ACL by direction, so the last one
seen won each direction. The platform WAN egress block added at priority 1049
is also a drop and is legitimately ip4-scoped, so it displaced the default-deny
in the map and the assertion fired on its qualifier instead:

  from-lport deny on sg_c94870c0d68288dc6 is IPv4-scoped:
  "inport == @sg_c94870c0d68288dc6 && ip4 && ip4.dst != {10.0.0.0/8, ...}"

Select the two default-denies by their own priorities. What the test asserts is
unchanged — the default-deny still must carry no ethertype qualifier, for the
reason denyIngressACL states: OVN allows on no-match, so an ip4-scoped deny
lets every other ethertype through.

* fix: drain every EIP on an ENI at teardown, not just the first

findByENI returned the first matching record and stopped, so an ENI holding
more than one EIP had all but one left marked associated to an interface being
deleted — with the host route still installed and no record left naming the ENI
to find them. Teardown now clears them all.

* ci: serialise e2e-nightly runs so two do not share the hypervisors

Only iso-single and iso-multi were guarded, by the iso-qemu-rack group. Every
other job could overlap a second run of the workflow, with both sets of cells
building clusters on the same five hosts. Queue rather than cancel: a dispatch
is usually a verification run whose result someone is waiting on.

.github/workflows/e2e-nightly.yml

@@ -16,6 +16,14 @@ permissions:
   # context exposes the id itself.
   actions: read
 
+# Every cell builds its cluster on the same five hypervisors, so two runs in
+# flight at once compete for their disks and their libvirt guests. Queue the
+# second rather than cancelling it: a dispatch is usually a verification run
+# and discarding it loses the result someone is waiting on.
+concurrency:
+  group: e2e-nightly
+  cancel-in-progress: false
+
 # Scheduled runs are loaded from the default branch (main), so the schedule
 # trigger must stay in sync on main for cron to fire.
 env:

spinifex/daemon/daemon.go

@@ -2832,6 +2832,22 @@ func (rm *ResourceManager) canAllocate(instanceType *ec2.InstanceTypeInfo, count
 // admission never advertises more instances than there are GPU slots to
 // back them.
 func (rm *ResourceManager) canAllocateLocked(instanceType *ec2.InstanceTypeInfo, count int) int {
+	n, _ := rm.admitLocked(instanceType, count)
+	return n
+}
+
+// admit is admitLocked under the read lock.
+func (rm *ResourceManager) admit(instanceType *ec2.InstanceTypeInfo, count int) (int, string) {
+	rm.mu.RLock()
+	defer rm.mu.RUnlock()
+	return rm.admitLocked(instanceType, count)
+}
+
+// admitLocked returns the admission count and which gate bound it. A refusal
+// that names its gate is the difference between "the node is full" and "the
+// host is short of free memory right now", which have different fixes and are
+// otherwise indistinguishable from InsufficientInstanceCapacity.
+func (rm *ResourceManager) admitLocked(instanceType *ec2.InstanceTypeInfo, count int) (int, string) {
 	instanceTypeName := ""
 	if instanceType.InstanceType != nil {
 		instanceTypeName = *instanceType.InstanceType
@@ -2846,7 +2862,7 @@ func (rm *ResourceManager) canAllocateLocked(instanceType *ec2.InstanceTypeInfo,
 		}
 	}
 
-	n := canAllocateCount(
+	budget := canAllocateCount(
 		rm.hostVCPU-rm.reservedVCPU-rm.reservedCRVCPU, rm.allocatedVCPU,
 		rm.hostMemGB-rm.reservedMem-rm.reservedCRMem, rm.allocatedMem,
 		instanceTypeVCPUs(instanceType),
@@ -2854,7 +2870,11 @@ func (rm *ResourceManager) canAllocateLocked(instanceType *ec2.InstanceTypeInfo,
 		count,
 		availGPU, requiresGPU,
 	)
-	return rm.liveMemGate(n, instanceType)
+	n := rm.liveMemGate(budget, instanceType)
+	if n < budget {
+		return n, "live-memory"
+	}
+	return n, "budget"
 }
 
 // liveMemGate clamps n by current MemAvailable, catching overcommit that the
@@ -2970,7 +2990,8 @@ func (rm *ResourceManager) updateInstanceSubscriptions() {
 	defer rm.subsMu.Unlock()
 
 	for typeName, typeInfo := range rm.instanceTypes {
-		canFit := rm.canAllocate(typeInfo, 1) >= 1
+		fits, gate := rm.admit(typeInfo, 1)
+		canFit := fits >= 1
 
 		subjectRoot := "ec2.RunInstances"
 		handler := rm.handler
@@ -3002,7 +3023,7 @@ func (rm *ResourceManager) updateInstanceSubscriptions() {
 				slog.Error("Failed to unsubscribe from instance type topic", "topic", queueTopic, "err", err)
 			}
 			delete(rm.instanceSubs, queueTopic)
-			slog.Info("Unsubscribed from instance type (capacity full)", "topic", queueTopic)
+			slog.Info("Unsubscribed from instance type (capacity full)", "topic", queueTopic, "gate", gate)
 		}
 
 		if rm.nodeID != "" {

spinifex/handlers/ec2/eip/service_impl.go

@@ -337,50 +337,64 @@ func (s *EIPServiceImpl) DisassociateByENI(ctx context.Context, accountID, eniID
 	if eniID == "" {
 		return false, nil
 	}
-	record, key, revision, err := s.findByENI(ctx, accountID, eniID)
+	matches, err := s.findByENI(ctx, accountID, eniID)
 	if err != nil {
 		return false, err
 	}
-	if record == nil {
+	if len(matches) == 0 {
 		return false, nil
 	}
 
-	// The MAC comes off the record rather than a fresh ENI lookup: this runs as
-	// part of tearing that ENI down, so the interface may already be gone.
-	s.publishNATEvent("vpc.delete-nat", record.VpcId, record.PublicIp, record.PrivateIp, eniID, record.MacAddress)
+	for _, m := range matches {
+		record := m.record
 
-	publicIP, allocationID := record.PublicIp, record.AllocationId
-	clearAssociation(record)
+		// The MAC comes off the record rather than a fresh ENI lookup: this runs
+		// as part of tearing that ENI down, so the interface may already be gone.
+		s.publishNATEvent("vpc.delete-nat", record.VpcId, record.PublicIp, record.PrivateIp, eniID, record.MacAddress)
 
-	data, err := json.Marshal(record)
-	if err != nil {
-		return false, fmt.Errorf("marshal EIP record: %w", err)
+		publicIP, allocationID := record.PublicIp, record.AllocationId
+		clearAssociation(record)
+
+		data, err := json.Marshal(record)
+		if err != nil {
+			return false, fmt.Errorf("marshal EIP record: %w", err)
+		}
+		if _, err := s.eipKV.Update(ctx, m.key, data, m.revision); err != nil {
+			return false, errors.New(awserrors.ErrorServerInternal)
+		}
+
+		slog.InfoContext(ctx, "DisassociateByENI completed",
+			"eniId", eniID, "publicIp", publicIP, "allocationId", allocationID, "accountID", accountID)
 	}
-	if _, err := s.eipKV.Update(ctx, key, data, revision); err != nil {
-		return false, errors.New(awserrors.ErrorServerInternal)
-	}
-
-	slog.InfoContext(ctx, "DisassociateByENI completed",
-		"eniId", eniID, "publicIp", publicIP, "allocationId", allocationID, "accountID", accountID)
 	return true, nil
 }
 
-// findByENI returns the associated EIP record holding eniID, or a nil record
-// when the account has none. Unlike findByAssociationID an absent match is not
-// an error: teardown asks this of every ENI, and most carry no EIP.
-func (s *EIPServiceImpl) findByENI(ctx context.Context, accountID, eniID string) (*EIPRecord, string, uint64, error) {
+// eipMatch is a decoded EIP record with the CAS handle needed to write it back.
+type eipMatch struct {
+	record   *EIPRecord
+	key      string
+	revision uint64
+}
+
+// findByENI returns every associated EIP record holding eniID, empty when the
+// account has none. Unlike findByAssociationID an absent match is not an error:
+// teardown asks this of every ENI, and most carry no EIP. All matches are
+// returned, not the first: an ENI can hold more than one EIP, and stopping at
+// one strands the rest associated to an interface that is being deleted.
+func (s *EIPServiceImpl) findByENI(ctx context.Context, accountID, eniID string) ([]eipMatch, error) {
 	prefix := accountID + "."
 	keys, err := s.eipKV.Keys(ctx)
 	if err != nil {
 		if errors.Is(err, jetstream.ErrNoKeysFound) {
-			return nil, "", 0, nil
+			return nil, nil
 		}
-		return nil, "", 0, errors.New(awserrors.ErrorServerInternal)
+		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 
+	var matches []eipMatch
 	for _, k := range keys {
 		if err := ctx.Err(); err != nil {
-			return nil, "", 0, err
+			return nil, err
 		}
 		if k == utils.VersionKey || !strings.HasPrefix(k, prefix) {
 			continue
@@ -393,7 +407,7 @@ func (s *EIPServiceImpl) findByENI(ctx context.Context, accountID, eniID string)
 			// An unreadable record must not read as "this ENI has no EIP":
 			// the caller deletes the interface on that answer, and the
 			// association would be stranded with nothing left to find it.
-			return nil, "", 0, fmt.Errorf("eipKV.Get(%s): %w", k, err)
+			return nil, fmt.Errorf("eipKV.Get(%s): %w", k, err)
 		}
 		var record EIPRecord
 		if err := json.Unmarshal(entry.Value(), &record); err != nil {
@@ -401,10 +415,10 @@ func (s *EIPServiceImpl) findByENI(ctx context.Context, accountID, eniID string)
 			continue
 		}
 		if record.ENIId == eniID {
-			return &record, k, entry.Revision(), nil
+			matches = append(matches, eipMatch{record: &record, key: k, revision: entry.Revision()})
 		}
 	}
-	return nil, "", 0, nil
+	return matches, nil
 }
 
 // describeAddressesValidFilters defines the set of filter names accepted by DescribeAddresses.

spinifex/handlers/ec2/eip/service_impl_test.go

@@ -776,3 +776,38 @@ func TestEIP_PublishNATEvent_PortNameHasPortPrefix(t *testing.T) {
 	assert.Equal(t, "10.0.0.5", got.LogicalIP)
 	assert.Equal(t, "198.51.100.10", got.ExternalIP)
 }
+
+// TestEIP_DisassociateByENI_ClearsEveryEIPOnTheInterface pins that teardown
+// drains all of an ENI's EIPs. Stopping at the first left the rest marked
+// associated to an interface being deleted, with their host routes installed
+// and nothing remaining that could find them.
+func TestEIP_DisassociateByENI_ClearsEveryEIPOnTheInterface(t *testing.T) {
+	svc, _, _ := setupTestEIP(t)
+
+	var allocIDs []string
+	for range 3 {
+		out, err := svc.AllocateAddress(context.Background(), &ec2.AllocateAddressInput{}, testAccountID)
+		require.NoError(t, err)
+		allocIDs = append(allocIDs, *out.AllocationId)
+		associateToENI(t, svc, *out.AllocationId, "eni-shared")
+	}
+
+	found, err := svc.DisassociateByENI(t.Context(), testAccountID, "eni-shared")
+	require.NoError(t, err)
+	assert.True(t, found)
+
+	for _, allocID := range allocIDs {
+		entry, err := svc.eipKV.Get(t.Context(), testAccountID+"."+allocID)
+		require.NoError(t, err)
+		var record EIPRecord
+		require.NoError(t, json.Unmarshal(entry.Value(), &record))
+		assert.Equal(t, "allocated", record.State, "%s still associated", allocID)
+		assert.Empty(t, record.ENIId, "%s still names the ENI", allocID)
+		assert.Empty(t, record.AssociationId, "%s kept its association", allocID)
+	}
+
+	// One pass drains them all, so the retry teardown always makes finds none.
+	found, err = svc.DisassociateByENI(t.Context(), testAccountID, "eni-shared")
+	require.NoError(t, err)
+	assert.False(t, found)
+}

spinifex/network/policy/nat.go

@@ -391,6 +391,11 @@ func (m *natManager) DeleteEIP(ctx context.Context, vpcID, externalIP, logicalIP
 	// Deleting on a pair that recycled identically would tear down the new owner's
 	// rule and ARP entry; the stamped logical port is the discriminator that
 	// survives identical-pair reuse.
+	//
+	// A row that is simply absent is not a reassignment. Nobody owns the IP, so
+	// the OVN delete has nothing to do but the host plumbing below is orphaned
+	// and is exactly what needs removing.
+	rowGone := false
 	if logicalIP != "" || portName != "" {
 		existing, err := m.ovn.FindNATByExternalIP(ctx, "dnat_and_snat", externalIP)
 		switch {
@@ -398,7 +403,7 @@ func (m *natManager) DeleteEIP(ctx context.Context, vpcID, externalIP, logicalIP
 			slog.Warn("policy: DeleteEIP ownership lookup failed, proceeding with delete",
 				"external_ip", externalIP, "logical_ip", logicalIP, "err", err)
 		case existing == nil:
-			return nil
+			rowGone = true
 		case logicalIP != "" && existing.LogicalIP != logicalIP:
 			slog.Info("policy: DeleteEIP skip — external IP reassigned to a different logical IP (stale delete)",
 				"external_ip", externalIP, "stale_logical_ip", logicalIP, "current_logical_ip", existing.LogicalIP)
@@ -411,9 +416,11 @@ func (m *natManager) DeleteEIP(ctx context.Context, vpcID, externalIP, logicalIP
 			return nil
 		}
 	}
-	if err := m.ovn.DeleteNATByExternalIP(ctx, router, "dnat_and_snat", externalIP); err != nil {
-		if !errors.Is(err, ovn.ErrNATNotFound) {
-			return fmt.Errorf("delete dnat_and_snat %s on %s: %w", externalIP, router, err)
+	if !rowGone {
+		if err := m.ovn.DeleteNATByExternalIP(ctx, router, "dnat_and_snat", externalIP); err != nil {
+			if !errors.Is(err, ovn.ErrNATNotFound) {
+				return fmt.Errorf("delete dnat_and_snat %s on %s: %w", externalIP, router, err)
+			}
 		}
 	}
 	// Flush host ARP for the released IP so the next owner isn't shadowed. Best-effort.

spinifex/network/policy/nat_binder_test.go

@@ -136,6 +136,43 @@ func TestNATManager_DeleteEIP_Routed_UnbindsHostEIP(t *testing.T) {
 	assert.Equal(t, []string{"192.168.1.200"}, b.unbinds)
 }
 
+// An absent dnat_and_snat row is not a reassignment: nobody owns the address,
+// so the orphaned host plumbing is precisely what the delete has to remove.
+// Returning early on it left the /32 route and proxy-ARP entry behind, which
+// is what a disassociate looked like from the host.
+func TestNATManager_DeleteEIP_Routed_UnbindsWhenRowAlreadyGone(t *testing.T) {
+	m := mock.New()
+	seedRouter(t, m, "vpc-1")
+	seedGatewayPortIP(t, m, "vpc-1", "100.127.0.10", nil)
+	b := &recordedBinder{}
+	mgr, err := NewNATManager(m, NATModeRouted, WithHostEIPBinder(b.hooks()))
+	require.NoError(t, err)
+
+	// No AddEIP: the row never existed, standing in for one already removed.
+	require.NoError(t, mgr.DeleteEIP(context.Background(), "vpc-1", "192.168.1.200", "10.0.1.5", "eni-1"))
+	assert.Equal(t, []string{"192.168.1.200"}, b.unbinds)
+}
+
+// The reassignment cases keep their early return: a different owner holds the
+// address, and its host plumbing must survive a late delete for the old one.
+func TestNATManager_DeleteEIP_Routed_ReassignedRowKeepsHostPlumbing(t *testing.T) {
+	m := mock.New()
+	seedRouter(t, m, "vpc-1")
+	seedGatewayPortIP(t, m, "vpc-1", "100.127.0.10", nil)
+	b := &recordedBinder{}
+	mgr, err := NewNATManager(m, NATModeRouted, WithHostEIPBinder(b.hooks()))
+	require.NoError(t, err)
+
+	require.NoError(t, mgr.AddEIP(context.Background(), EIPSpec{
+		VPCID: "vpc-1", ExternalIP: "192.168.1.200", LogicalIP: "10.0.1.9",
+	}))
+	b.unbinds = nil
+
+	// Stale delete naming the previous logical IP.
+	require.NoError(t, mgr.DeleteEIP(context.Background(), "vpc-1", "192.168.1.200", "10.0.1.5", ""))
+	assert.Empty(t, b.unbinds, "a reassigned EIP must keep its host plumbing")
+}
+
 func TestNATManager_AddEIP_NonRoutedNeverBinds(t *testing.T) {
 	for _, mode := range []NATMode{NATModeDistributed, NATModeCentralized} {
 		m := mock.New()

spinifex/services/viperblockd/provider_handlers.go

@@ -34,6 +34,11 @@ import (
 // persists and the byte units the ebsprovider wire contract uses.
 const bytesPerGiB = 1024 * 1024 * 1024
 
+// pluginSealGrace bounds the wait for an nbdkit that is already sealing to
+// finish and exit. The plugin bounds its own drain at 20s, so this only has to
+// cover that plus writing the receipt.
+const pluginSealGrace = 25 * time.Second
+
 // providerObjectStoreFactory builds the objectstore.ObjectStore DeleteVolume
 // and DeleteSnapshot use to remove S3 object prefixes. Tests override this to
 // inject objectstore.NewMemoryObjectStore(), keeping the unit tests free of
@@ -1613,6 +1618,12 @@ func unmountVolume(ctx context.Context, cfg *Config, volumeName string) (types.E
 			matched.VB.Detach()
 		}
 
+		// Removing the guest device starts the plugin's own Close, which
+		// drains to the backend and leaves a receipt. SIGKILLing into that
+		// discards the drain and leaves the WAL for the fallback seal below to
+		// replay, which costs far more than the caller's deadline allows.
+		utils.TerminateProcess(matched.PID, pluginSealGrace)
+
 		// The seal below rewrites the directory nbdkit writes, so a kill that
 		// did not take makes it a concurrent write to that directory. Fail the
 		// unmount instead: the entry stays mounted and a retry re-attempts

spinifex/utils/process.go

@@ -79,6 +79,25 @@ func ForceKillProcess(pid int, timeout time.Duration) error {
 	return WaitForProcessExit(pid, timeout)
 }
 
+// TerminateProcess SIGTERMs a process and waits up to grace for it to exit,
+// reporting whether it did. It never escalates: callers that must establish
+// the process is gone follow it with ForceKillProcess, which is where the
+// SIGKILL and the exit confirmation belong.
+func TerminateProcess(pid int, grace time.Duration) bool {
+	if pid <= 0 {
+		return false
+	}
+	process, err := os.FindProcess(pid)
+	if err != nil {
+		return false
+	}
+	// Already gone is the outcome asked for, so report it as one.
+	if err := process.Signal(syscall.SIGTERM); err != nil {
+		return errors.Is(err, os.ErrProcessDone) || errors.Is(err, syscall.ESRCH)
+	}
+	return WaitForProcessExit(pid, grace) == nil
+}
+
 // killProcessPollInterval and killProcessGracePeriod drive KillProcess's
 // liveness poll: check every pollInterval instead of blocking for the full
 // gracePeriod, so a process that dies quickly is detected almost immediately.

spinifex/utils/process_terminate_test.go

@@ -0,0 +1,76 @@
+package utils_test
+
+import (
+	"bufio"
+	"os/exec"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/mulgadc/spinifex/spinifex/utils"
+)
+
+// The unmount path calls this on an nbdkit that is already sealing, so the
+// grace has to be spent only when the process needs it: one that exits on
+// SIGTERM must be reaped in milliseconds, not after the full budget.
+func TestTerminateProcessReturnsAsSoonAsTheProcessExits(t *testing.T) {
+	cmd := exec.Command("sleep", "60") // dies immediately on default SIGTERM
+	require.NoError(t, cmd.Start())
+	pid := cmd.Process.Pid
+	done := make(chan struct{})
+	go func() { _ = cmd.Wait(); close(done) }()
+
+	start := time.Now()
+	exited := utils.TerminateProcess(pid, 10*time.Second)
+	elapsed := time.Since(start)
+
+	assert.True(t, exited, "a process that dies on SIGTERM must be reported as exited")
+	assert.Less(t, elapsed, time.Second, "must not wait out the grace period")
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("process did not exit after TerminateProcess returned")
+	}
+}
+
+// A plugin wedged mid-drain must not hold the unmount open past the grace:
+// the caller escalates to ForceKillProcess, and can only do that if this
+// returns.
+func TestTerminateProcessGivesUpAtTheGraceAndDoesNotKill(t *testing.T) {
+	cmd := exec.Command("sh", "-c", "trap '' TERM; echo ready; while true; do sleep 1; done")
+	stdout, err := cmd.StdoutPipe()
+	require.NoError(t, err)
+	require.NoError(t, cmd.Start())
+	pid := cmd.Process.Pid
+	t.Cleanup(func() {
+		_ = cmd.Process.Kill()
+		_ = cmd.Wait()
+	})
+
+	scanner := bufio.NewScanner(stdout)
+	require.True(t, scanner.Scan(), "expected ready line from test process")
+	require.Equal(t, "ready", scanner.Text())
+
+	const grace = 200 * time.Millisecond
+	start := time.Now()
+	exited := utils.TerminateProcess(pid, grace)
+	elapsed := time.Since(start)
+
+	assert.False(t, exited)
+	assert.GreaterOrEqual(t, elapsed, grace, "must wait out the grace before giving up")
+	assert.Less(t, elapsed, grace+2*time.Second, "must be bounded by the grace, not hang")
+	assert.True(t, utils.ProcessAlive(pid), "escalation belongs to the caller, not here")
+}
+
+func TestTerminateProcessAlreadyExitedIsSuccess(t *testing.T) {
+	cmd := exec.Command("true")
+	require.NoError(t, cmd.Start())
+	pid := cmd.Process.Pid
+	require.NoError(t, cmd.Wait())
+
+	assert.True(t, utils.TerminateProcess(pid, time.Second))
+	assert.False(t, utils.TerminateProcess(0, time.Second), "an invalid pid is not an exit")
+}

tests/e2e/harness/fixtures.go

@@ -364,11 +364,22 @@ func EnsureKeyPair(t *testing.T, fx *Fixture) (string, string) {
 		if err != nil {
 			return "", nil, fmt.Errorf("CreateKeyPair %s: %w", name, err)
 		}
+		// Past this point the pair exists on the cluster but no cleanup is
+		// registered yet, so anything that fails has to remove it. A pair left
+		// behind fails every later caller with InvalidKeyPair.Duplicate, which
+		// reads as a different bug in a test that never ran.
+		abandon := func(cause error) (string, func() error, error) {
+			_, derr := fx.EC2.DeleteKeyPair(&ec2.DeleteKeyPairInput{KeyName: aws.String(name)})
+			if derr != nil {
+				return "", nil, fmt.Errorf("%w (and could not remove the pair it created: %v)", cause, derr)
+			}
+			return "", nil, cause
+		}
 		if err := os.WriteFile(pemPath, []byte(aws.StringValue(out.KeyMaterial)), 0o600); err != nil {
-			return "", nil, fmt.Errorf("write pem %s: %w", pemPath, err)
+			return abandon(fmt.Errorf("write pem %s: %w", pemPath, err))
 		}
 		if err := assertOneKeyPair(fx, name); err != nil {
-			return "", nil, err
+			return abandon(err)
 		}
 		fx.tagRunResources(aws.StringValue(out.KeyPairId))
 		return aws.StringValue(out.KeyName), func() error {
@@ -385,6 +396,13 @@ func EnsureKeyPair(t *testing.T, fx *Fixture) (string, string) {
 	return id, pemPath
 }
 
+// How long a just-created key pair is given to become visible to a describe,
+// and how often that is retried.
+const (
+	keyPairVisibleWithin = 30 * time.Second
+	keyPairVisiblePoll   = 500 * time.Millisecond
+)
+
 // assertOneKeyPair fails when the cluster holds more than one key pair under
 // name, which it should not be able to.
 //
@@ -394,14 +412,34 @@ func EnsureKeyPair(t *testing.T, fx *Fixture) (string, string) {
 // rejecting the key at the SSH gate five minutes later, once per guest. Cheap
 // to check here, and it names the cause outright.
 func assertOneKeyPair(fx *Fixture, name string) error {
-	out, err := fx.EC2.DescribeKeyPairs(&ec2.DescribeKeyPairsInput{
-		KeyNames: []*string{aws.String(name)},
-	})
-	if err != nil {
-		return fmt.Errorf("DescribeKeyPairs %s: %w", name, err)
-	}
-	if len(out.KeyPairs) == 1 {
-		return nil
+	// The create this follows has already succeeded, so nothing here can
+	// legitimately see zero. On a multi-node cluster the describe can land on a
+	// node that has not caught up yet, which is a propagation window rather
+	// than a result. Two is the answer this exists to catch, and two is
+	// reported the moment it is seen.
+	var out *ec2.DescribeKeyPairsOutput
+	deadline := time.Now().Add(keyPairVisibleWithin)
+	for {
+		var err error
+		out, err = fx.EC2.DescribeKeyPairs(&ec2.DescribeKeyPairsInput{
+			KeyNames: []*string{aws.String(name)},
+		})
+		switch {
+		case err == nil && len(out.KeyPairs) == 1:
+			return nil
+		case err == nil && len(out.KeyPairs) > 1:
+			// The defect. Report it without waiting.
+		case err != nil && !ErrorCodeIs(err, "InvalidKeyPair.NotFound"):
+			return fmt.Errorf("DescribeKeyPairs %s: %w", name, err)
+		default:
+			// Not visible yet, or reported as absent.
+			if time.Now().Before(deadline) {
+				time.Sleep(keyPairVisiblePoll)
+				continue
+			}
+			return fmt.Errorf("key pair %s was created but is still not visible after %s", name, keyPairVisibleWithin)
+		}
+		break
 	}
 
 	got := make([]string, 0, len(out.KeyPairs))

tests/e2e/single/datapath_test.go

@@ -363,11 +363,19 @@ func runSGPolicyDatapath(t *testing.T, fix *Fixture) {
 			harness.Step(t, "8e-5b %s denies carry no ethertype qualifier", targetPG)
 			rows := portGroupACLs(t, targetPG)
 
+			// Match the default-denies by their priorities, not by "any drop".
+			// The platform egress block is also a drop and is legitimately
+			// ip4-scoped, so bucketing every drop by direction lets it stand in
+			// for the rule this is about and fail on its qualifier.
+			denyPriority := map[string]int{
+				"to-lport":   aclPriorityDefaultDenyIngress,
+				"from-lport": aclPriorityDefaultDenyEgress,
+			}
 			denies := map[string]aclRow{}
 			arps := map[string]aclRow{}
 			for _, r := range rows {
 				switch {
-				case r.Action == "drop":
+				case r.Action == "drop" && r.Priority == denyPriority[r.Direction]:
 					denies[r.Direction] = r
 				case strings.Contains(r.Match, "arp"):
 					arps[r.Direction] = r
@@ -625,6 +633,14 @@ func runSGEInstance(t *testing.T, fix *Fixture, subnetID, sgID, userData string)
 }
 
 // aclRow is one NB ACL as OVN holds it, read back rather than rebuilt.
+// The two default-deny priorities, mirroring policy.ACLPriorityDefaultDeny*.
+// Duplicated rather than imported so this suite reads what OVN holds without
+// depending on the package that put it there.
+const (
+	aclPriorityDefaultDenyIngress = 900
+	aclPriorityDefaultDenyEgress  = 800
+)
+
 type aclRow struct {
 	Priority  int
 	Direction string