fix: keep surviving flows on a failed IMDS repair and repair guests concurrently

Clear a tap's flows only once its endpoint is set up, so a repair that fails
after a kernel-module reload no longer takes down a guest that still had
connectivity. Repair up to 8 datapaths at once, since an ovs-vswitchd restart
drops every guest together. Test that teardown holds the ENI lock across
detach and tap cleanup, and count only the test guest's flows in the e2e.

docs/testing/e2e-coverage.md

@@ -27,7 +27,7 @@ The largest suite, 27 top-level tests. Notable entry points:
 - `TestSGPolicyDatapath`, `TestSGReachabilityPolicy` — security groups as an actual datapath, not just API state.
 - `TestWANEgressSMTPBlock` — egress port blocking.
 - `TestInstanceMetadata` — IMDS from inside a guest.
-- **`TestIMDSDatapathSurvivesVswitchdRestart`** restarts `ovs-vswitchd` under a guest with an Elastic IP, once through systemd and once with SIGKILL. After each, the guest must answer over its EIP and fetch its instance ID through IMDSv2 within 60 seconds, and `br-imds` must hold the flows it had before. It is sequential because it takes the whole node's datapath down.
+- **`TestIMDSDatapathSurvivesVswitchdRestart`** restarts `ovs-vswitchd` under a guest with an Elastic IP, once through systemd and once with SIGKILL. After each, the guest must answer over its EIP and fetch its instance ID through IMDSv2 within 60 seconds, and `br-imds` must hold all 7 of the guest's own flows. It is sequential because it takes the whole node's datapath down.
 - `TestStopStart`, `TestEarlyRebootLiveness`, `TestGuestChurnDurability`, `TestSnapshotLifecycle`, `TestSnapshotBackedLaunch`, `TestCreateImage`, `TestVolumeLifecycle`, `TestSpotInstanceLifecycle`, `TestLaunchTemplateBoot`.
 - `TestNegativeErrorPaths` — the `8a`–`8t` AWS error-shape cases (malformed AMI ID, invalid instance type, volume in use, detach-root forbidden, unknown or malformed instance IDs on the lifecycle verbs, and so on).
 - **`TestKVReplication`** — the single-node half of the KV replication policy, and it is not the multi-node assertion with a smaller number in it. On one node one replica per bucket is the correct answer rather than the defect, so what it proves is the opposite: that nothing declared a replica count this deployment cannot place, that `spx admin kv replicas` exits 0 and reports `WANT 1`, and that `--repair` run twice against a cluster with nothing to repair changes no bucket's replica count and leaves every bucket serving. The multi-node suites cannot cover any of that, because on three nodes the unplaceable count is the right one. It lives here rather than in its own suite so it runs in every single-node nightly cell, across all three host and network permutations.

spinifex/domains/network/host/imds_attach.go

@@ -74,7 +74,7 @@ func imdsDetachSpec(eniID string) IMDSTapDatapath {
 // endpoint, patch, demux/egress flows, and reply routing. Stale flows are cleared
 // once here (all installers share the tap's cookie) before any flow is added.
 //
-// Bridge, cookie clear, endpoint, and patch are connectivity-critical (error bare):
+// Bridge, endpoint, cookie clear, and patch are connectivity-critical (error bare):
 // ListIMDSTaps advertises the tap off the patch and vpcd binds the endpoint, so the
 // endpoint must exist whenever the patch does, else vpcd binds a missing device every
 // reconcile forever. The flows and reply routing only serve, so theirs wrap
@@ -90,10 +90,12 @@ func installIMDSDatapath(ctx context.Context, r Runner, d IMDSTapDatapath) error
 	// The INPUT accept and the remap DNAT are wildcarded over every endpoint and
 	// installed once by vpcd at start. They are not attempted here: the daemon
 	// has no iptables grant and no CAP_NET_ADMIN, so trying aborts every launch.
-	if err := clearIMDSFlowsByCookie(ctx, r, imdsFlowCookie(d.Endpoint)); err != nil {
+	if err := ensureIMDSEndpoint(ctx, r, d); err != nil {
 		return err
 	}
-	if err := ensureIMDSEndpoint(ctx, r, d); err != nil {
+	// Cleared after the endpoint so a failed endpoint step leaves a repaired
+	// guest's surviving flows, and with them its connectivity, in place.
+	if err := clearIMDSFlowsByCookie(ctx, r, imdsFlowCookie(d.Endpoint)); err != nil {
 		return err
 	}
 	if err := installTapPatch(ctx, r, d); err != nil {

spinifex/domains/network/host/imds_drift.go

@@ -19,6 +19,11 @@ import (
 // egress flow, and the two tap<->patch forward flows.
 const IMDSTapFlowCount = 7
 
+// IMDSFlowCookie is the cookie every br-imds flow of eniID's datapath carries.
+func IMDSFlowCookie(eniID string) string {
+	return imdsFlowCookie(IMDSEndpointName(eniID))
+}
+
 // endpointSysctlDir is where the kernel exposes per-device IPv4 settings.
 var endpointSysctlDir = "/proc/sys/net/ipv4/conf"
 

spinifex/domains/network/host/imds_drift_test.go

@@ -173,6 +173,25 @@ func TestRepairIMDSDatapath(t *testing.T) {
 		}
 	})
 
+	// After a kmod reload the flows survive and only the endpoint is lost; a
+	// repair that fails on the endpoint must not take the flows with it.
+	t.Run("endpoint failure keeps the surviving flows", func(t *testing.T) {
+		s := newStubRunner()
+		s.expect("ovs-vsctl list-ports "+IMDSBridge, []byte(d.Tap+"\n"), nil)
+		s.expect("ovs-vsctl --may-exist add-br", nil, nil)
+		s.expect("ovs-vsctl set", nil, nil)
+		s.expect("ip link set "+IMDSBridge, nil, nil)
+		s.expect("ovs-vsctl --may-exist add-port "+IMDSBridge+" "+d.Endpoint, nil, errors.New("ovsdb busy"))
+		s.expect("ovs-ofctl", nil, nil)
+
+		if err := repairIMDSDatapath(context.Background(), s, d); err == nil {
+			t.Fatal("want the endpoint failure to surface")
+		}
+		if s.called("ovs-ofctl del-flows") {
+			t.Errorf("flows cleared before the endpoint was ready; calls: %v", s.calls)
+		}
+	})
+
 	t.Run("port list failure installs nothing", func(t *testing.T) {
 		s := newStubRunner()
 		s.expect("ovs-vsctl list-ports", nil, errors.New("ovsdb unreachable"))

spinifex/runtime/compute/vm/imds_reconcile.go

@@ -15,6 +15,11 @@ import (
 // after ovs-vswitchd restarts. A var so tests can shorten it.
 var IMDSDatapathReconcileInterval = 10 * time.Second
 
+// imdsRepairConcurrency bounds how many datapaths one pass re-installs at once.
+// An ovs-vswitchd restart drops every guest's flows together, so serial repairs
+// would keep the last guest dark for the sum of everyone else's install.
+const imdsRepairConcurrency = 8
+
 // imdsDatapathTarget is a running guest whose primary ENI has an IMDS datapath,
 // copied out of the manager lock with everything a re-install needs.
 type imdsDatapathTarget struct {
@@ -62,9 +67,16 @@ func (m *Manager) reconcileIMDSDatapaths() {
 		slog.Warn("IMDS: datapath state read failed; skipping reconcile pass", "err", err)
 		return
 	}
+	sem := make(chan struct{}, imdsRepairConcurrency)
+	var wg sync.WaitGroup
 	for _, t := range imdsDatapathsToRepair(targets, states) {
-		m.repairIMDSDatapath(t, states[t.eniID])
+		sem <- struct{}{}
+		wg.Go(func() {
+			defer func() { <-sem }()
+			m.repairIMDSDatapath(t, states[t.eniID])
+		})
 	}
+	wg.Wait()
 }
 
 // imdsDatapathTargets returns the running guests attachPrimaryIMDSDatapath

spinifex/runtime/compute/vm/imds_reconcile_test.go

@@ -2,6 +2,10 @@ package vm
 
 import (
 	"errors"
+	"fmt"
+	"slices"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -189,6 +193,96 @@ func TestRepairIMDSDatapath_WaitsForTeardownLock(t *testing.T) {
 	}
 }
 
+// A repair landing between detach and tap cleanup would find the tap still on
+// br-imds and re-create the endpoint and patch of a guest being torn down.
+func TestCleanupTapDevices_HoldsENILockAcrossDetachAndCleanup(t *testing.T) {
+	inDetach, releaseDetach := make(chan struct{}), make(chan struct{})
+	inCleanup, releaseCleanup := make(chan struct{}), make(chan struct{})
+	plumber := &fakeNetworkPlumber{
+		detachHook:  func() { close(inDetach); <-releaseDetach },
+		cleanupHook: func() { close(inCleanup); <-releaseCleanup },
+	}
+	m := NewManagerWithDeps(Deps{NetworkPlumber: plumber})
+	v := imdsTestVM("i-a", "eni-a", StateRunning)
+	m.Insert(v)
+	target, _ := imdsTargetOf(v)
+
+	var wg sync.WaitGroup
+	wg.Go(func() { m.cleanupTapDevices(v) })
+	<-inDetach
+	wg.Go(func() { m.repairIMDSDatapath(target, imdsFlushed) })
+	for _, step := range []struct {
+		name    string
+		release chan struct{}
+		next    chan struct{}
+	}{{"detach", releaseDetach, inCleanup}, {"tap cleanup", releaseCleanup, nil}} {
+		time.Sleep(50 * time.Millisecond)
+		if n := plumber.repairCount(); n != 0 {
+			t.Fatalf("repair ran during %s: %d calls", step.name, n)
+		}
+		close(step.release)
+		if step.next != nil {
+			<-step.next
+		}
+	}
+	wg.Wait()
+
+	want := []string{"detach:eni-a", "cleanup:" + TapDeviceName("eni-a"), "repair:eni-a"}
+	if !slices.Equal(plumber.order, want) {
+		t.Errorf("call order = %v, want %v", plumber.order, want)
+	}
+}
+
+// An ovs-vswitchd restart drifts every guest at once; repairs must overlap so
+// the last guest is not dark for the sum of everyone else's install.
+func TestReconcileIMDSDatapaths_RepairsConcurrentlyUpToLimit(t *testing.T) {
+	const guests = imdsRepairConcurrency + 4
+	var inFlight atomic.Int32
+	var overLimit atomic.Bool
+	full, release := make(chan struct{}), make(chan struct{})
+	var fullOnce sync.Once
+	plumber := &fakeNetworkPlumber{
+		imdsStates: map[string]IMDSDatapathState{},
+		repairHook: func() {
+			n := inFlight.Add(1)
+			if n > imdsRepairConcurrency {
+				overLimit.Store(true)
+			}
+			if n == imdsRepairConcurrency {
+				fullOnce.Do(func() { close(full) })
+			}
+			<-release
+			inFlight.Add(-1)
+		},
+	}
+	m := NewManagerWithDeps(Deps{NetworkPlumber: plumber})
+	for i := range guests {
+		eni := fmt.Sprintf("eni-%d", i)
+		plumber.imdsStates[eni] = imdsFlushed
+		m.Insert(imdsTestVM(fmt.Sprintf("i-%d", i), eni, StateRunning))
+	}
+
+	done := make(chan struct{})
+	go func() {
+		m.reconcileIMDSDatapaths()
+		close(done)
+	}()
+	select {
+	case <-full:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("repairs did not reach %d in flight: %d", imdsRepairConcurrency, inFlight.Load())
+	}
+	time.Sleep(50 * time.Millisecond)
+	if overLimit.Load() {
+		t.Errorf("more than %d repairs ran at once", imdsRepairConcurrency)
+	}
+	close(release)
+	<-done
+	if n := plumber.repairCount(); n != guests {
+		t.Errorf("repairs = %d, want %d", n, guests)
+	}
+}
+
 func TestRepairIMDSDatapath_DetachedTapIsNotAnError(t *testing.T) {
 	plumber := &fakeNetworkPlumber{imdsRepairErr: ErrIMDSTapDetached}
 	m := NewManagerWithDeps(Deps{NetworkPlumber: plumber})

spinifex/runtime/compute/vm/network_helpers_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"reflect"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -13,6 +14,11 @@ import (
 
 // fakeNetworkPlumber records calls so tests can assert per-spec behaviour.
 type fakeNetworkPlumber struct {
+	mu                  sync.Mutex
+	order               []string // cleanup, detach and repair calls, in order
+	detachHook          func()
+	cleanupHook         func()
+	repairHook          func()
 	setupCalls          []TapSpec
 	cleanupCalls        []string
 	imdsAttachCalls     []imdsAttachCall
@@ -53,7 +59,13 @@ func (p *fakeNetworkPlumber) SetupTap(spec TapSpec) error {
 }
 
 func (p *fakeNetworkPlumber) CleanupTap(name string) error {
+	if p.cleanupHook != nil {
+		p.cleanupHook()
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	p.cleanupCalls = append(p.cleanupCalls, name)
+	p.order = append(p.order, "cleanup:"+name)
 	return p.cleanupErr
 }
 
@@ -63,7 +75,13 @@ func (p *fakeNetworkPlumber) AttachIMDSDatapath(eniID, mac, subnetID string) err
 }
 
 func (p *fakeNetworkPlumber) DetachIMDSDatapath(eniID string) error {
+	if p.detachHook != nil {
+		p.detachHook()
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	p.imdsDetachCalls = append(p.imdsDetachCalls, eniID)
+	p.order = append(p.order, "detach:"+eniID)
 	return p.imdsDetachErr
 }
 
@@ -81,10 +99,22 @@ func (p *fakeNetworkPlumber) IMDSDatapathStates(eniIDs []string) (map[string]IMD
 }
 
 func (p *fakeNetworkPlumber) RepairIMDSDatapath(eniID, mac, subnetID string) error {
+	if p.repairHook != nil {
+		p.repairHook()
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	p.imdsRepairCalls = append(p.imdsRepairCalls, imdsAttachCall{eniID: eniID, mac: mac, subnetID: subnetID})
+	p.order = append(p.order, "repair:"+eniID)
 	return p.imdsRepairErr
 }
 
+func (p *fakeNetworkPlumber) repairCount() int {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return len(p.imdsRepairCalls)
+}
+
 func (p *fakeNetworkPlumber) EnsureIMDSDatapathBridge() error {
 	p.ensureBridgeCalls++
 	return p.ensureBridgeErr

tests/e2e/single/vswitchdrestart_test.go

@@ -77,10 +77,10 @@ func runIMDSDatapathSurvivesVswitchdRestart(t *testing.T, fix *Fixture) {
 	}
 	tgt := harness.SSHTarget{User: "ubuntu", Host: eipIP, Port: 22, KeyPath: keyPath}
 
+	cookie := host.IMDSFlowCookie(primaryENI(t, harness.WaitForInstanceState(t, c, instanceID, "running")))
 	require.NoError(t, guestIMDSAnswers(t, tgt, instanceID), "IMDS before any restart")
-	baseline := imdsBridgeFlowCount(t)
-	require.Positive(t, baseline, "br-imds has no flows with a guest running")
-	harness.Detail(t, "eip", eipIP, "unit", unit, "br_imds_flows", baseline)
+	require.Equal(t, host.IMDSTapFlowCount, guestIMDSFlowCount(t, cookie), "guest's br-imds flows before any restart")
+	harness.Detail(t, "eip", eipIP, "unit", unit, "cookie", cookie)
 
 	for _, disruption := range []struct {
 		name string
@@ -100,8 +100,8 @@ func runIMDSDatapathSurvivesVswitchdRestart(t *testing.T, fix *Fixture) {
 			if err := guestIMDSAnswers(t, tgt, instanceID); err != nil {
 				return err
 			}
-			if n := imdsBridgeFlowCount(t); n != baseline {
-				return fmt.Errorf("br-imds has %d flows, had %d before the restart", n, baseline)
+			if n := guestIMDSFlowCount(t, cookie); n != host.IMDSTapFlowCount {
+				return fmt.Errorf("guest has %d br-imds flows, want %d", n, host.IMDSTapFlowCount)
 			}
 			return nil
 		}, vswitchdRecoveryBudget, 2*time.Second)
@@ -127,11 +127,11 @@ func guestIMDSAnswers(t *testing.T, tgt harness.SSHTarget, instanceID string) er
 	return nil
 }
 
-// imdsBridgeFlowCount returns how many flows br-imds holds, or -1 while
-// ovs-vswitchd is not answering.
-func imdsBridgeFlowCount(t *testing.T) int {
+// guestIMDSFlowCount returns how many br-imds flows carry the guest's cookie, or
+// -1 while ovs-vswitchd is not answering. Other guests' teardowns cannot skew it.
+func guestIMDSFlowCount(t *testing.T, cookie string) int {
 	t.Helper()
-	out, err := exec.Command("sudo", "-n", "ovs-ofctl", "--no-stats", "dump-flows", host.IMDSBridge).Output()
+	out, err := exec.Command("sudo", "-n", "ovs-ofctl", "--no-stats", "dump-flows", host.IMDSBridge, "cookie="+cookie+"/-1").Output()
 	if err != nil {
 		return -1
 	}