fix: requeue an incomplete reconcile pass instead of waiting a drift interval (#840)

spinifex/network/external/dhcp/acquire_budget_internal_test.go

@@ -28,3 +28,72 @@ func TestAcquireLadderFitsInsideClientReadTimeout(t *testing.T) {
 	assert.Greater(t, defaultRPCTimeout, defaultAcquireBudget,
 		"the RPC must outlast the acquire, or a slow DORA reports a timeout instead of its own error")
 }
+
+// newBackoffTestManager builds a Manager for schedule selection only; it never
+// DORAs, so a bare fake client and store suffice.
+func newBackoffTestManager(t *testing.T, cfg ManagerConfig) *Manager {
+	t.Helper()
+	cfg.Client = &Fake{}
+	cfg.Store = &Store{}
+	m, err := NewManager(cfg)
+	assert.NoError(t, err)
+	return m
+}
+
+// A failed gw-lrp acquire costs the whole VPC its external gateway until a
+// reconcile pass returns, and nothing external times that DORA — the allocator
+// calls the manager in-process. A failed ENI or EIP acquire blocks an AWS client
+// that must hear back inside its read timeout. The purposes must not share a
+// budget.
+func TestAcquireBackoffIsPerPurpose(t *testing.T) {
+	m := newBackoffTestManager(t, ManagerConfig{})
+
+	gwSchedule, gwBudget := m.acquireBackoffFor(PurposeGatewayLRP)
+	assert.Equal(t, gwLRPAcquireSchedule, gwSchedule)
+	assert.Equal(t, gwLRPAcquireBudget, gwBudget)
+
+	for _, purpose := range []string{PurposeEIP, PurposeENIPublic, PurposeNATGWExternal, ""} {
+		schedule, budget := m.acquireBackoffFor(purpose)
+		assert.Equal(t, defaultAcquireSchedule, schedule,
+			"purpose %q must keep the default ladder: its caller is an AWS client that will retry untokened", purpose)
+		assert.Equal(t, defaultAcquireBudget, budget,
+			"purpose %q must keep the default budget", purpose)
+	}
+}
+
+// The gw-lrp window has to be wider than the default, or the split buys nothing:
+// the ladder was shortened for the EIP path alone and took gw-lrp down with it.
+func TestGatewayLRPAcquireWindowExceedsDefault(t *testing.T) {
+	ladder := func(schedule []time.Duration) time.Duration {
+		total := time.Duration(0)
+		for _, d := range schedule {
+			total += d + acquireAttemptJitter
+		}
+		return total
+	}
+	assert.Greater(t, ladder(gwLRPAcquireSchedule), ladder(defaultAcquireSchedule),
+		"the gw-lrp ladder must outlast the default one")
+	assert.Greater(t, gwLRPAcquireBudget, defaultAcquireBudget,
+		"the gw-lrp budget must outlast the default one")
+	assert.GreaterOrEqual(t, gwLRPAcquireBudget, ladder(gwLRPAcquireSchedule),
+		"the budget must leave room for the full ladder plus jitter, or later attempts never run")
+}
+
+// Both pairs stay configurable, so an operator on a slow upstream can widen
+// either without a rebuild.
+func TestAcquireBackoffConfigOverrides(t *testing.T) {
+	m := newBackoffTestManager(t, ManagerConfig{
+		AcquireSchedule:           []time.Duration{time.Second},
+		AcquireBudget:             2 * time.Second,
+		GatewayLRPAcquireSchedule: []time.Duration{3 * time.Second},
+		GatewayLRPAcquireBudget:   4 * time.Second,
+	})
+
+	gwSchedule, gwBudget := m.acquireBackoffFor(PurposeGatewayLRP)
+	assert.Equal(t, []time.Duration{3 * time.Second}, gwSchedule)
+	assert.Equal(t, 4*time.Second, gwBudget)
+
+	schedule, budget := m.acquireBackoffFor(PurposeEIP)
+	assert.Equal(t, []time.Duration{time.Second}, schedule)
+	assert.Equal(t, 2*time.Second, budget)
+}

spinifex/network/external/dhcp/manager.go

@@ -38,6 +38,19 @@ var defaultAcquireSchedule = []time.Duration{4 * time.Second, 8 * time.Second, 1
 // the daemon room to answer inside a 60s client read timeout, per the schedule.
 const defaultAcquireBudget = 45 * time.Second
 
+// gwLRPAcquireSchedule is the per-attempt ladder for purpose=gw-lrp. Longer than
+// the default because no client is timing this DORA: AttachInternetGateway
+// publishes vpc.igw-attach fire-and-forget and the allocator calls the manager
+// in-process, so the read-timeout ceiling that bounds the EIP path does not
+// apply. The two purposes split on that constraint, not only on cost of failure.
+var gwLRPAcquireSchedule = []time.Duration{4 * time.Second, 8 * time.Second, 16 * time.Second, 32 * time.Second}
+
+// gwLRPAcquireBudget caps the wallclock for a gw-lrp DORA loop, with room over
+// the ladder for jitter. A lost gw-lrp lease costs the whole VPC its external
+// gateway until a reconcile pass returns, so it is worth waiting out a stall an
+// ENI or EIP acquire would rightly abandon.
+const gwLRPAcquireBudget = 90 * time.Second
+
 // acquireAttemptJitter is the ± window applied to each per-attempt
 // timeout so concurrent acquires across vpcds don't synchronise.
 const acquireAttemptJitter = time.Second
@@ -51,6 +64,10 @@ type ManagerConfig struct {
 	Now             func() time.Time
 	AcquireSchedule []time.Duration
 	AcquireBudget   time.Duration
+	// GatewayLRPAcquireSchedule/Budget override the outer DORA backoff for
+	// purpose=gw-lrp only; zero values fall back to the gw-lrp defaults.
+	GatewayLRPAcquireSchedule []time.Duration
+	GatewayLRPAcquireBudget   time.Duration
 	// IfaceIPs lists the IPs bound to a named interface; tests override.
 	// Used to detect MAC-keyed upstream routers on interface-MAC pools.
 	IfaceIPs func(iface string) ([]net.IP, error)
@@ -68,6 +85,8 @@ type Manager struct {
 	now             func() time.Time
 	acquireSchedule []time.Duration
 	acquireBudget   time.Duration
+	gwLRPSchedule   []time.Duration
+	gwLRPBudget     time.Duration
 	ifaceIPs        func(iface string) ([]net.IP, error)
 	nodeName        string
 
@@ -121,6 +140,14 @@ func NewManager(cfg ManagerConfig) (*Manager, error) {
 	if budget <= 0 {
 		budget = defaultAcquireBudget
 	}
+	gwSchedule := cfg.GatewayLRPAcquireSchedule
+	if len(gwSchedule) == 0 {
+		gwSchedule = gwLRPAcquireSchedule
+	}
+	gwBudget := cfg.GatewayLRPAcquireBudget
+	if gwBudget <= 0 {
+		gwBudget = gwLRPAcquireBudget
+	}
 	ifaceIPs := cfg.IfaceIPs
 	if ifaceIPs == nil {
 		ifaceIPs = defaultIfaceIPs
@@ -136,6 +163,8 @@ func NewManager(cfg ManagerConfig) (*Manager, error) {
 		now:             now,
 		acquireSchedule: schedule,
 		acquireBudget:   budget,
+		gwLRPSchedule:   gwSchedule,
+		gwLRPBudget:     gwBudget,
 		ifaceIPs:        ifaceIPs,
 		nodeName:        nodeName,
 		loops:           map[string]*leaseLoop{},
@@ -408,7 +437,7 @@ func (m *Manager) run(ctx context.Context, self *leaseLoop, e Entry, reaffirm bo
 }
 
 func (m *Manager) doAcquire(ctx context.Context, e *Entry) error {
-	lease, err := m.acquireWithBackoff(ctx, AcquireRequest{
+	lease, err := m.acquireWithBackoff(ctx, e.Purpose, AcquireRequest{
 		Bridge:      e.Lease.Bridge,
 		ClientID:    e.Lease.ClientID,
 		Hostname:    e.Lease.Hostname,
@@ -456,16 +485,27 @@ func (m *Manager) applyLease(ctx context.Context, e *Entry, fresh *Lease) error
 	return nil
 }
 
+// acquireBackoffFor returns the per-attempt ladder and wallclock budget for
+// purpose. gw-lrp gets its own, longer pair; every other purpose shares the
+// default one.
+func (m *Manager) acquireBackoffFor(purpose string) ([]time.Duration, time.Duration) {
+	if purpose == PurposeGatewayLRP {
+		return m.gwLRPSchedule, m.gwLRPBudget
+	}
+	return m.acquireSchedule, m.acquireBudget
+}
+
 // acquireWithBackoff drives client.Acquire with per-attempt timeouts from
-// the schedule, capped by AcquireBudget. Jitter prevents synchronised wakes.
+// purpose's schedule, capped by its budget. Jitter prevents synchronised wakes.
 // ctx.Err() short-circuits; returns last error when all attempts fail.
-func (m *Manager) acquireWithBackoff(ctx context.Context, req AcquireRequest) (*Lease, error) {
-	if len(m.acquireSchedule) == 0 {
+func (m *Manager) acquireWithBackoff(ctx context.Context, purpose string, req AcquireRequest) (*Lease, error) {
+	schedule, budget := m.acquireBackoffFor(purpose)
+	if len(schedule) == 0 {
 		return m.client.Acquire(ctx, req)
 	}
-	deadline := m.now().Add(m.acquireBudget)
+	deadline := m.now().Add(budget)
 	var lastErr error
-	for i, base := range m.acquireSchedule {
+	for i, base := range schedule {
 		if ctx.Err() != nil {
 			return nil, ctx.Err()
 		}
@@ -493,13 +533,13 @@ func (m *Manager) acquireWithBackoff(ctx context.Context, req AcquireRequest) (*
 		}
 		lastErr = err
 		slog.Warn("dhcp manager: acquire attempt failed",
-			"client_id", req.ClientID, "attempt", i+1, "of", len(m.acquireSchedule),
+			"client_id", req.ClientID, "purpose", purpose, "attempt", i+1, "of", len(schedule),
 			"timeout_ms", otelsetup.Millis(attempt), "err", err)
 	}
 	if lastErr == nil {
 		lastErr = errors.New("acquire budget exhausted before first attempt")
 	}
-	return nil, fmt.Errorf("acquire after %d attempts: %w", len(m.acquireSchedule), lastErr)
+	return nil, fmt.Errorf("acquire after %d attempts: %w", len(schedule), lastErr)
 }
 
 func (m *Manager) doRenew(ctx context.Context, e *Entry) error {
@@ -625,7 +665,7 @@ func (m *Manager) acquireLocked(ctx context.Context, req acquireWireRequest) (*E
 	if vendorClass == "" {
 		vendorClass = m.leaseVendorClass()
 	}
-	lease, err := m.acquireWithBackoff(ctx, AcquireRequest{
+	lease, err := m.acquireWithBackoff(ctx, req.Purpose, AcquireRequest{
 		Bridge:      req.Bridge,
 		ClientID:    req.ClientID,
 		Hostname:    hostname,

spinifex/network/reconcile/apply.go

@@ -102,7 +102,7 @@ func eniIDFromPort(lspName string) string {
 
 // applyVPCs ensures every intent VPC has a LogicalRouter. Stray OVN-only
 // routers are left alone.
-func (r *reconciler) applyVPCs(ctx context.Context, intent IntentState, actual ActualState) {
+func (r *reconciler) applyVPCs(ctx context.Context, intent IntentState, actual ActualState, res *passResult) {
 	for vpcID, spec := range intent.VPCs {
 		routerName := topology.VPCRouter(vpcID)
 		if _, ok := actual.Routers[routerName]; !ok {
@@ -115,6 +115,7 @@ func (r *reconciler) applyVPCs(ctx context.Context, intent IntentState, actual A
 			}
 			if _, _, err := r.ovn.EnsureLogicalRouter(ctx, lr); err != nil {
 				slog.Error("reconcile/apply: ensure VPC router failed", "vpc_id", vpcID, "err", err)
+				res.fail(classVPC, vpcID, err)
 				continue
 			}
 			actual.Routers[routerName] = struct{}{}
@@ -125,7 +126,7 @@ func (r *reconciler) applyVPCs(ctx context.Context, intent IntentState, actual A
 
 // applySubnets ensures every intent subnet has a LogicalSwitch, LRP, router LSP,
 // and DHCPOptions row. Each step is idempotent.
-func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actual ActualState) {
+func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actual ActualState, res *passResult) {
 	for subnetID, spec := range intent.Subnets {
 		switchName := topology.SubnetSwitch(subnetID)
 		routerName := topology.VPCRouter(spec.VPCID)
@@ -135,6 +136,7 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua
 		gwIP, prefixBits, err := topology.SubnetGatewayCIDR(spec.CIDR)
 		if err != nil {
 			slog.Error("reconcile/apply: subnet gateway calc failed", "subnet_id", subnetID, "err", err)
+			res.fail(classSubnet, subnetID, err)
 			continue
 		}
 		gwCIDRString := fmt.Sprintf("%s/%d", gwIP, prefixBits)
@@ -150,6 +152,7 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua
 			}
 			if _, _, err := r.ovn.EnsureLogicalSwitch(ctx, ls); err != nil {
 				slog.Error("reconcile/apply: ensure subnet switch failed", "subnet_id", subnetID, "err", err)
+				res.fail(classSubnet, subnetID, err)
 				continue
 			}
 			actual.Switches[switchName] = struct{}{}
@@ -167,6 +170,7 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua
 			}
 			if err := r.ovn.CreateLogicalRouterPort(ctx, routerName, lrp); err != nil {
 				slog.Error("reconcile/apply: create subnet LRP failed", "subnet_id", subnetID, "err", err)
+				res.fail(classSubnet, subnetID, err)
 				continue
 			}
 			actual.RouterPorts[routerPortName] = struct{}{}
@@ -187,6 +191,7 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua
 			}
 			if err := r.ovn.CreateLogicalSwitchPort(ctx, switchName, lsp); err != nil {
 				slog.Error("reconcile/apply: create subnet router-LSP failed", "subnet_id", subnetID, "err", err)
+				res.fail(classSubnet, subnetID, err)
 				continue
 			}
 		}
@@ -205,6 +210,7 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua
 			}
 			if _, dErr := r.ovn.CreateDHCPOptions(ctx, opts); dErr != nil {
 				slog.Warn("reconcile/apply: create DHCP options failed (non-fatal)", "subnet_id", subnetID, "err", dErr)
+				res.fail(classSubnet, subnetID, dErr)
 			}
 		case !maps.Equal(existing.Options, want):
 			// Converge, don't leave create-time values in place. Toggling
@@ -212,6 +218,7 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua
 			// the wide figure on an encrypted path blackholes large segments.
 			if dErr := r.ovn.UpdateDHCPOptionsOptions(ctx, existing.UUID, want); dErr != nil {
 				slog.Warn("reconcile/apply: update DHCP options failed (non-fatal)", "subnet_id", subnetID, "err", dErr)
+				res.fail(classSubnet, subnetID, dErr)
 			} else {
 				slog.Info("reconcile/apply: converged DHCP options", "subnet_id", subnetID, "mtu", want["mtu"])
 			}
@@ -225,15 +232,17 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua
 // applySGs ensures every intent SG has a port group; when pruneOrphans is true,
 // deletes sg_* PGs with no matching intent SG. Startup passes false to avoid
 // deleting in-flight resources before peer subscribers have converged.
-func (r *reconciler) applySGs(ctx context.Context, intent IntentState, actual ActualState, pruneOrphans bool) {
+func (r *reconciler) applySGs(ctx context.Context, intent IntentState, actual ActualState, pruneOrphans bool, res *passResult) {
 	for groupID, spec := range intent.SGs {
 		if err := r.topology.EnsureSGPortGroup(ctx, groupID); err != nil {
 			slog.Error("reconcile/apply: EnsureSGPortGroup failed", "sg", groupID, "err", err)
+			res.fail(classSG, groupID, err)
 			continue
 		}
 		actual.PortGroups[topology.SecurityGroupPortGroup(groupID)] = struct{}{}
 		if err := r.sg.EnsureSG(ctx, spec); err != nil {
 			slog.Error("reconcile/apply: EnsureSG failed", "sg", groupID, "err", err)
+			res.fail(classSG, groupID, err)
 		}
 	}
 
@@ -254,6 +263,7 @@ func (r *reconciler) applySGs(ctx context.Context, intent IntentState, actual Ac
 		}
 		if err := r.topology.DeleteSGPortGroupByName(ctx, pgName); err != nil {
 			slog.Warn("reconcile/apply: orphan DeleteSGPortGroupByName failed", "pg", pgName, "err", err)
+			res.fail(classSG, pgName, err)
 			continue
 		}
 		delete(actual.PortGroups, pgName)
@@ -265,7 +275,7 @@ func (r *reconciler) applySGs(ctx context.Context, intent IntentState, actual Ac
 // SGIDs. Existing ports use diff-based UpdatePortGroupMemberships to avoid gaps.
 // When pruneOrphans is true, ENI LSPs with no matching intent ENI are torn down;
 // startup passes false so in-flight ports survive until subscribers converge.
-func (r *reconciler) applyPorts(ctx context.Context, intent IntentState, actual ActualState, pruneOrphans bool) {
+func (r *reconciler) applyPorts(ctx context.Context, intent IntentState, actual ActualState, pruneOrphans bool, res *passResult) {
 	for portID, spec := range intent.Ports {
 		portName := topology.Port(portID)
 		switchName := topology.SubnetSwitch(spec.SubnetID)
@@ -297,6 +307,7 @@ func (r *reconciler) applyPorts(ctx context.Context, intent IntentState, actual
 			}
 			if err := r.ovn.CreateLogicalSwitchPortInGroups(ctx, switchName, lsp, desiredPGs); err != nil {
 				slog.Error("reconcile/apply: create ENI port failed", "port", portName, "err", err)
+				res.fail(classPort, portName, err)
 			}
 			continue
 		}
@@ -304,6 +315,7 @@ func (r *reconciler) applyPorts(ctx context.Context, intent IntentState, actual
 		currentPGs, err := r.ovn.ListPortGroupsForPort(ctx, portName)
 		if err != nil {
 			slog.Warn("reconcile/apply: list port groups for port failed", "port", portName, "err", err)
+			res.fail(classPort, portName, err)
 			continue
 		}
 		addPGs, removePGs := diffSets(desiredPGs, currentPGs)
@@ -312,23 +324,25 @@ func (r *reconciler) applyPorts(ctx context.Context, intent IntentState, actual
 		}
 		if err := r.ovn.UpdatePortGroupMemberships(ctx, portName, addPGs, removePGs); err != nil {
 			slog.Warn("reconcile/apply: update port group memberships failed", "port", portName, "err", err)
+			res.fail(classPort, portName, err)
 		}
 	}
 
 	if !pruneOrphans {
 		return
 	}
-	r.pruneOrphanPorts(ctx, intent)
+	r.pruneOrphanPorts(ctx, intent, res)
 }
 
 // pruneOrphanPorts deletes guest LSPs whose spinifex:eni_id has no matching
 // intent ENI, closing the create-only gap that leaks ports across instance
 // terminate and host reinstall. DeletePort clears PG memberships then removes
 // the LSP (composed cascade).
-func (r *reconciler) pruneOrphanPorts(ctx context.Context, intent IntentState) {
+func (r *reconciler) pruneOrphanPorts(ctx context.Context, intent IntentState, res *passResult) {
 	lsps, err := r.ovn.ListLogicalSwitchPorts(ctx)
 	if err != nil {
 		slog.Warn("reconcile/apply: list LSPs for orphan prune failed", "err", err)
+		res.fail(classPort, "orphan-prune", err)
 		return
 	}
 	for i := range lsps {
@@ -342,6 +356,7 @@ func (r *reconciler) pruneOrphanPorts(ctx context.Context, intent IntentState) {
 		spec := topology.PortSpec{PortID: eniID, SubnetID: lsps[i].ExternalIDs["spinifex:subnet_id"]}
 		if err := r.topology.DeletePort(ctx, spec); err != nil {
 			slog.Warn("reconcile/apply: orphan ENI DeletePort failed", "port", lsps[i].Name, "err", err)
+			res.fail(classPort, lsps[i].Name, err)
 			continue
 		}
 		slog.Info("reconcile/apply: removed orphan ENI port", "port", lsps[i].Name, "eni_id", eniID)
@@ -352,14 +367,15 @@ func (r *reconciler) pruneOrphanPorts(ctx context.Context, intent IntentState) {
 // existing IGWs. AttachIGW is idempotent and must run even when the gateway
 // switch port already exists: its already-attached path re-ensures host state
 // (routed-NAT ingress routes) that survives in OVN but not across reboots.
-func (r *reconciler) applyIGWs(ctx context.Context, intent IntentState, actual ActualState) {
+func (r *reconciler) applyIGWs(ctx context.Context, intent IntentState, actual ActualState, res *passResult) {
 	for vpcID, spec := range intent.IGWs {
 		if err := r.igw.AttachIGW(ctx, spec); err != nil {
 			slog.Error("reconcile/apply: AttachIGW failed", "vpc_id", vpcID, "err", err)
+			res.fail(classIGW, vpcID, err)
 			continue
 		}
 		actual.ExternalSwch[vpcID] = struct{}{}
-		r.rebindGatewayChassis(ctx, vpcID, eipProbeIP(intent, vpcID))
+		r.rebindGatewayChassis(ctx, vpcID, eipProbeIP(intent, vpcID), res)
 	}
 }
 
@@ -377,19 +393,23 @@ func eipProbeIP(intent IntentState, vpcID string) string {
 }
 
 // rebindGatewayChassis re-asserts chassis priority tuples on the gateway LRP.
-func (r *reconciler) rebindGatewayChassis(ctx context.Context, vpcID, eipIP string) {
+func (r *reconciler) rebindGatewayChassis(ctx context.Context, vpcID, eipIP string, res *passResult) {
 	if len(r.chassis) == 0 {
 		return
 	}
 	gwPortName := topology.GatewayRouterPort(vpcID)
 	lrp, err := r.ovn.GetLogicalRouterPort(ctx, gwPortName)
 	if err != nil {
+		slog.Warn("reconcile/apply: gateway LRP read failed; skipping chassis rebind and datapath gate",
+			"vpc_id", vpcID, "port", gwPortName, "err", err)
+		res.fail(classIGW, vpcID, err)
 		return
 	}
 	for i, chassis := range r.chassis {
 		priority := max(20-(i*5), 1)
 		if err := r.ovn.SetGatewayChassis(ctx, gwPortName, chassis, priority); err != nil {
 			slog.Warn("reconcile/apply: SetGatewayChassis failed", "vpc_id", vpcID, "chassis", chassis, "err", err)
+			res.fail(classIGW, vpcID, err)
 		}
 	}
 	r.ensureGatewayClaimed(ctx, topology.GatewayChassisRedirectPort(vpcID))
@@ -550,10 +570,11 @@ func (r *reconciler) ensureGatewayClaimed(ctx context.Context, crPortName string
 // proves the gateway-chassis flow exists, not the gatewayLRP->guest hop, so a guest
 // whose port has not converged (e.g. just after a host reboot) stays dark while
 // every other signal is green.
-func (r *reconciler) applyEIPs(ctx context.Context, intent IntentState, _ ActualState) {
+func (r *reconciler) applyEIPs(ctx context.Context, intent IntentState, _ ActualState, res *passResult) {
 	for _, spec := range r.floatingIPSpecs(intent) {
 		if err := r.nat.AddEIP(ctx, spec); err != nil {
 			slog.Error("reconcile/apply: AddEIP failed", "external_ip", spec.ExternalIP, "logical_ip", spec.LogicalIP, "err", err)
+			res.fail(classEIP, spec.ExternalIP, err)
 		}
 		r.ensureGuestPortDatapath(ctx, spec)
 	}
@@ -607,19 +628,21 @@ func (r *reconciler) floatingIPSpecs(intent IntentState) []policy.EIPSpec {
 // and union its ports into the live set so a mid-pass launch counts as live; skip the
 // prune entirely if the re-read fails rather than risk a false sweep against a snapshot
 // known to be stale.
-func (r *reconciler) pruneOrphanEIPs(ctx context.Context, intent IntentState) {
+func (r *reconciler) pruneOrphanEIPs(ctx context.Context, intent IntentState, res *passResult) {
 	live := make(map[string]struct{}, len(intent.Ports)+len(intent.EIPs))
 	addLivePorts(live, intent)
 	if r.reloadIntent != nil {
 		fresh, err := r.reloadIntent(ctx)
 		if err != nil {
 			slog.Warn("reconcile/apply: fresh intent re-read failed; skipping orphan EIP prune", "err", err)
+			res.fail(classEIP, "orphan-prune", err)
 			return
 		}
 		addLivePorts(live, fresh)
 	}
 	if pruned, err := r.nat.PruneOrphanEIPs(ctx, live); err != nil {
 		slog.Warn("reconcile/apply: orphan EIP prune failed", "err", err)
+		res.fail(classEIP, "orphan-prune", err)
 	} else if pruned > 0 {
 		slog.Info("reconcile/apply: pruned orphan dnat_and_snat rows", "count", pruned)
 	}
@@ -644,14 +667,14 @@ func addLivePorts(live map[string]struct{}, intent IntentState) {
 // drop gate. Public IPs come from two disjoint sources: auto-assigned and ELB
 // addresses recorded on the ENI (intent.Ports) and user EIPs in the EIP bucket
 // (intent.EIPs). Both need the same /32 reroute above the gate.
-func (r *reconciler) applyPublicInstanceEgress(ctx context.Context, intent IntentState, _ ActualState) {
+func (r *reconciler) applyPublicInstanceEgress(ctx context.Context, intent IntentState, _ ActualState, res *passResult) {
 	for _, p := range intent.Ports {
 		if p.PublicIP.IsValid() {
-			r.ensureEIPEgressExemption(ctx, intent, p.VPCID, p.SubnetID, p.PrivateIP.String())
+			r.ensureEIPEgressExemption(ctx, intent, p.VPCID, p.SubnetID, p.PrivateIP.String(), res)
 		}
 	}
 	for _, e := range intent.EIPs {
-		r.ensureEIPEgressExemption(ctx, intent, e.VPCID, subnetIDForIP(intent.Subnets, e.LogicalIP), e.LogicalIP)
+		r.ensureEIPEgressExemption(ctx, intent, e.VPCID, subnetIDForIP(intent.Subnets, e.LogicalIP), e.LogicalIP, res)
 	}
 }
 
@@ -663,7 +686,7 @@ func (r *reconciler) applyPublicInstanceEgress(ctx context.Context, intent Inten
 // the datapath; the instance's dnat_and_snat supplies SNAT. Scoped to subnets that
 // actually carry a drop gate: routed subnets egress via their priority-1000 reroute
 // and need no exemption, so the gate presence bounds the blast radius.
-func (r *reconciler) ensureEIPEgressExemption(ctx context.Context, intent IntentState, vpcID, subnetID, instanceIP string) {
+func (r *reconciler) ensureEIPEgressExemption(ctx context.Context, intent IntentState, vpcID, subnetID, instanceIP string, res *passResult) {
 	if subnetID == "" {
 		slog.Warn("reconcile/apply: public instance maps to no subnet; skipping egress exemption",
 			"vpc_id", vpcID, "instance_ip", instanceIP)
@@ -675,6 +698,7 @@ func (r *reconciler) ensureEIPEgressExemption(ctx context.Context, intent Intent
 	if err := r.igw.EnsureEIPInstanceEgress(ctx, vpcID, subnetID, instanceIP); err != nil {
 		slog.Error("reconcile/apply: EnsureEIPInstanceEgress failed",
 			"vpc_id", vpcID, "subnet_id", subnetID, "instance_ip", instanceIP, "err", err)
+		res.fail(classPublicEgress, instanceIP, err)
 	}
 }
 
@@ -762,43 +786,47 @@ func (r *reconciler) ensureGuestPortDatapath(ctx context.Context, spec policy.EI
 }
 
 // applyNATGWs runs every intent NAT gateway through NATManager.AddNATGateway.
-func (r *reconciler) applyNATGWs(ctx context.Context, intent IntentState, _ ActualState) {
+func (r *reconciler) applyNATGWs(ctx context.Context, intent IntentState, _ ActualState, res *passResult) {
 	for _, spec := range intent.NATGWs {
 		if err := r.nat.AddNATGateway(ctx, spec); err != nil {
 			slog.Warn("reconcile/apply: AddNATGateway failed (likely already exists)",
 				"natgw_id", spec.NATGatewayID, "subnet_cidr", spec.SubnetCIDR, "err", err)
+			res.fail(classNATGW, spec.NATGatewayID, err)
 		}
 	}
 }
 
 // applyIGWRoutes installs per-subnet egress reroute policies from intent.
 // Closes the bootstrap race: events fire before subscribers attach; KV retains the route.
-func (r *reconciler) applyIGWRoutes(ctx context.Context, intent IntentState, _ ActualState) {
+func (r *reconciler) applyIGWRoutes(ctx context.Context, intent IntentState, _ ActualState, res *passResult) {
 	for _, spec := range intent.IGWRoutes {
 		if err := r.igw.EnsureSubnetEgress(ctx, spec.VPCID, spec.SubnetID, spec.DestCIDR); err != nil {
 			slog.Error("reconcile/apply: EnsureSubnetEgress failed",
 				"vpc_id", spec.VPCID, "subnet_id", spec.SubnetID, "cidr", spec.DestCIDR.String(), "err", err)
+			res.fail(classIGWRoute, spec.SubnetID, err)
 		}
 	}
 }
 
 // applyNATGWRoutes is the NATGW priority sibling of applyIGWRoutes.
-func (r *reconciler) applyNATGWRoutes(ctx context.Context, intent IntentState, _ ActualState) {
+func (r *reconciler) applyNATGWRoutes(ctx context.Context, intent IntentState, _ ActualState, res *passResult) {
 	for _, spec := range intent.NATGWRoutes {
 		if err := r.igw.EnsureNATGatewaySubnetEgress(ctx, spec.VPCID, spec.SubnetID, spec.DestCIDR); err != nil {
 			slog.Error("reconcile/apply: EnsureNATGatewaySubnetEgress failed",
 				"vpc_id", spec.VPCID, "subnet_id", spec.SubnetID, "cidr", spec.DestCIDR.String(), "err", err)
+			res.fail(classNATGWRoute, spec.SubnetID, err)
 		}
 	}
 }
 
 // applyDropGates installs DROP policies for subnets with an attached IGW but
 // no 0.0.0.0/0 route, preventing unintended egress via the VPC default route.
-func (r *reconciler) applyDropGates(ctx context.Context, intent IntentState, _ ActualState) {
+func (r *reconciler) applyDropGates(ctx context.Context, intent IntentState, _ ActualState, res *passResult) {
 	for _, spec := range intent.DropGates {
 		if err := r.igw.EnsureSubnetEgressDrop(ctx, spec.VPCID, spec.SubnetID, spec.DestCIDR); err != nil {
 			slog.Error("reconcile/apply: EnsureSubnetEgressDrop failed",
 				"vpc_id", spec.VPCID, "subnet_id", spec.SubnetID, "cidr", spec.DestCIDR.String(), "err", err)
+			res.fail(classDropGate, spec.SubnetID, err)
 		}
 	}
 }

spinifex/network/reconcile/apply_test.go

@@ -596,7 +596,7 @@ func TestReconcile_PruneOrphanEIPs_SweepsAbsentOwners(t *testing.T) {
 	}
 
 	intent := freshIntent(t) // Ports has eni-a only
-	r.pruneOrphanEIPs(ctx, intent)
+	r.pruneOrphanEIPs(ctx, intent, &passResult{})
 
 	if findNATByExternal(m, "dnat_and_snat", "192.168.1.10") == nil {
 		t.Errorf("live EIP row must survive the prune")
@@ -647,7 +647,7 @@ func TestReconcile_PruneOrphanEIPs_SparesMidPassLaunch(t *testing.T) {
 		return fresh, nil
 	}
 
-	r.pruneOrphanEIPs(ctx, snapshot)
+	r.pruneOrphanEIPs(ctx, snapshot, &passResult{})
 
 	if findNATByExternal(m, "dnat_and_snat", "192.168.1.20") == nil {
 		t.Errorf("mid-pass launch row must survive the prune via the fresh re-read")
@@ -676,7 +676,7 @@ func TestReconcile_PruneOrphanEIPs_SkipsOnReloadError(t *testing.T) {
 	r.reloadIntent = func(context.Context) (IntentState, error) {
 		return IntentState{}, errors.New("kv unavailable")
 	}
-	r.pruneOrphanEIPs(ctx, freshIntent(t))
+	r.pruneOrphanEIPs(ctx, freshIntent(t), &passResult{})
 
 	if findNATByExternal(m, "dnat_and_snat", "192.168.1.11") == nil {
 		t.Errorf("prune must be skipped when the fresh re-read fails")
@@ -797,3 +797,121 @@ func TestApplySubnets_ConvergesDriftedDHCPOptions(t *testing.T) {
 		t.Errorf("DHCP options row was replaced (%s -> %s), want an in-place update", before.UUID, after.UUID)
 	}
 }
+
+// stubIGW forces AttachIGW to fail while delegating everything else to a real
+// manager, so a pass fails exactly one resource and the rest still applies.
+type stubIGW struct {
+	external.IGWManager
+
+	attachErr error
+}
+
+func (s *stubIGW) AttachIGW(ctx context.Context, spec external.IGWSpec) error {
+	if s.attachErr != nil {
+		return s.attachErr
+	}
+	return s.IGWManager.AttachIGW(ctx, spec)
+}
+
+// igwIntent is freshIntent plus an IGW for vpc-a, the resource whose failed
+// attach a pass used to swallow.
+func igwIntent(t *testing.T) IntentState {
+	t.Helper()
+	intent := freshIntent(t)
+	intent.IGWs["vpc-a"] = external.IGWSpec{VPCID: "vpc-a", InternetGatewayID: "igw-a"}
+	return intent
+}
+
+// A failed AttachIGW must surface as ErrPassIncomplete rather than a nil return:
+// a swallowed failure is indistinguishable from a converged pass, so nothing
+// retries it until the next drift tick.
+func TestReconcile_FailedAttachIGWReportsPassIncomplete(t *testing.T) {
+	rec, m := newTestReconciler(t)
+	ctx := context.Background()
+	attachErr := errors.New("allocate gateway LRP IP: dhcp gw-lrp acquire: context deadline exceeded")
+	rec.igw = &stubIGW{IGWManager: rec.igw, attachErr: attachErr}
+	// Without a chassis the rebind returns on its first line, which would make
+	// the assertion below pass whether or not the failure short-circuits it.
+	rec.chassis = []string{"chassis-1"}
+
+	err := rec.Reconcile(ctx, igwIntent(t))
+	if !errors.Is(err, ErrPassIncomplete) {
+		t.Fatalf("Reconcile = %v, want ErrPassIncomplete — a swallowed AttachIGW failure "+
+			"leaves the drift loop unable to tell a converged pass from a broken one", err)
+	}
+	// The rest of the pass still applies: only the IGW is unconverged.
+	if _, ok := m.Ports[topology.Port("eni-a")]; !ok {
+		t.Errorf("ENI port missing: a failed IGW attach must not abort the pass")
+	}
+	if _, ok := m.RouterPorts[topology.GatewayRouterPort("vpc-a")]; ok {
+		t.Errorf("gateway LRP present after a failed attach")
+	}
+	if m.SetGatewayChassisCalls != 0 {
+		t.Errorf("SetGatewayChassis called %d times after a failed attach, want 0 — "+
+			"rebinding chassis onto a gateway LRP that was never created", m.SetGatewayChassisCalls)
+	}
+}
+
+// The converged path must stay clean: a pass with the same intent and a working
+// AttachIGW returns nil, so the drift loop resets to DriftInterval.
+func TestReconcile_AttachedIGWReportsConverged(t *testing.T) {
+	rec, m := newTestReconciler(t)
+	ctx := context.Background()
+	rec.chassis = []string{"chassis-1"}
+
+	if err := rec.Reconcile(ctx, igwIntent(t)); err != nil {
+		t.Fatalf("Reconcile = %v, want nil", err)
+	}
+	if _, ok := m.RouterPorts[topology.GatewayRouterPort("vpc-a")]; !ok {
+		t.Errorf("gateway LRP missing after a successful attach")
+	}
+	if m.SetGatewayChassisCalls == 0 {
+		t.Errorf("SetGatewayChassis never called after a successful attach: the rebind " +
+			"must run, or a rebooted chassis never reclaims the gateway")
+	}
+}
+
+// stubNAT forces AddEIP to fail, delegating everything else.
+type stubNAT struct {
+	policy.NATManager
+
+	addErr error
+}
+
+func (s *stubNAT) AddEIP(ctx context.Context, spec policy.EIPSpec) error {
+	if s.addErr != nil {
+		return s.addErr
+	}
+	return s.NATManager.AddEIP(ctx, spec)
+}
+
+// The incomplete signal is a per-pass contract, not an IGW special case: a stage
+// other than applyIGWs failing must report itself the same way.
+func TestReconcile_FailedAddEIPReportsPassIncomplete(t *testing.T) {
+	rec, _ := newTestReconciler(t)
+	ctx := context.Background()
+	rec.nat = &stubNAT{NATManager: rec.nat, addErr: errors.New("ovsdb unreachable")}
+
+	intent := freshIntent(t)
+	intent.EIPs["10.0.1.10"] = policy.EIPSpec{
+		VPCID: "vpc-a", ExternalIP: "192.168.1.10", LogicalIP: "10.0.1.10",
+		PortName: topology.Port("eni-a"),
+	}
+	if err := rec.Reconcile(ctx, intent); !errors.Is(err, ErrPassIncomplete) {
+		t.Fatalf("Reconcile = %v, want ErrPassIncomplete for a failed AddEIP", err)
+	}
+}
+
+// The convergence summary counts failures per class, so an operator reading one
+// line knows which stage did not converge and how widely.
+func TestPassResult_SummaryCountsPerClass(t *testing.T) {
+	res := &passResult{}
+	res.fail(classIGW, "vpc-a", errors.New("x"))
+	res.fail(classEIP, "192.168.1.10", errors.New("y"))
+	res.fail(classEIP, "192.168.1.11", errors.New("z"))
+
+	want := []any{"unconverged", 3, classEIP, 2, classIGW, 1}
+	if got := res.summaryKV(); !slices.Equal(got, want) {
+		t.Errorf("summaryKV() = %v, want %v", got, want)
+	}
+}

spinifex/network/reconcile/drift.go

@@ -2,9 +2,11 @@ package reconcile
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"time"
 
+	"github.com/mulgadc/spinifex/spinifex/otelsetup"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -13,42 +15,93 @@ import (
 // integration tests can shrink it.
 var DriftInterval = 5 * time.Minute
 
+// driftBackoffBase is the requeue gap after the first incomplete pass. Var so
+// tests can shrink it.
+var driftBackoffBase = 5 * time.Second
+
+// driftBackoffFactor grows the requeue gap per consecutive incomplete pass, up
+// to DriftInterval. Without a requeue the repair latency for any transient apply
+// failure is a full DriftInterval, so a 57-second DHCP stall costs five minutes
+// of a VPC having no external gateway.
+const driftBackoffFactor = 3
+
 // DriftLoop runs Reconcile every DriftInterval, gated on AcquireLeader so
-// only one vpcd scans at a time. Returns when ctx is cancelled.
-func DriftLoop(ctx context.Context, rec Reconciler, nc *nats.Conn, localAZ, holder string) {
+// only one vpcd scans at a time. An incomplete pass requeues on a short
+// backoff instead. Returns when ctx is cancelled.
+//
+// startup is the outcome of the bootstrap ReconcileApplyOnly pass, or nil when
+// this node did not run one. Seeding from it matters because the gateway LRP's
+// first DORA happens in that pass: without the seed a resource that failed
+// before the loop started waits a full DriftInterval for its first retry.
+func DriftLoop(ctx context.Context, rec Reconciler, nc *nats.Conn, localAZ, holder string, startup error) {
 	js, err := jetstream.New(nc)
 	if err != nil {
 		slog.Error("reconcile/drift: JetStream context unavailable, drift loop disabled", "err", err)
 		return
 	}
 
-	ticker := time.NewTicker(DriftInterval)
-	defer ticker.Stop()
+	backoff := nextDriftBackoff(0, startup)
+	timer := time.NewTimer(driftWait(backoff))
+	defer timer.Stop()
 
 	for {
 		select {
 		case <-ctx.Done():
 			return
-		case <-ticker.C:
-			runDriftCycle(ctx, rec, nc, js, localAZ, holder)
+		case <-timer.C:
+			previous := backoff
+			backoff = nextDriftBackoff(backoff, runDriftCycle(ctx, rec, nc, js, localAZ, holder))
+			wait := driftWait(backoff)
+			if backoff != previous {
+				slog.Info("reconcile/drift: requeue interval changed", "wait_ms", otelsetup.Millis(wait))
+			}
+			timer.Reset(wait)
 		}
 	}
 }
 
+// nextDriftBackoff returns the requeue backoff after a pass: zero once a pass
+// converges, otherwise the previous backoff tripled from driftBackoffBase and
+// held at DriftInterval. Tracked separately from the wait it produces, so a
+// backoff sitting at the cap is not mistaken for a converged loop and reset.
+func nextDriftBackoff(current time.Duration, err error) time.Duration {
+	if !errors.Is(err, ErrPassIncomplete) {
+		return 0
+	}
+	next := driftBackoffBase
+	if current > 0 {
+		next = current * driftBackoffFactor
+	}
+	return min(next, DriftInterval)
+}
+
+// driftWait is the gap before the next pass for a backoff; no backoff means a
+// full DriftInterval.
+func driftWait(backoff time.Duration) time.Duration {
+	if backoff <= 0 {
+		return DriftInterval
+	}
+	return backoff
+}
+
 // runDriftCycle is one tick body, split out so tests can drive it directly.
-func runDriftCycle(ctx context.Context, rec Reconciler, nc *nats.Conn, js jetstream.JetStream, localAZ, holder string) {
+// Returns the reconcile outcome so the caller can size the next wait; a cycle
+// this node did not win the election for returns nil.
+func runDriftCycle(ctx context.Context, rec Reconciler, nc *nats.Conn, js jetstream.JetStream, localAZ, holder string) error {
 	release, elected := AcquireLeader(ctx, nc, KVBucketVPCDReconcile, holder)
 	if !elected {
-		return
+		return nil
 	}
 	defer release()
 
 	intent, err := LoadIntentFromKV(ctx, js, localAZ)
 	if err != nil {
 		slog.Warn("reconcile/drift: load intent failed", "err", err)
-		return
+		return err
 	}
 	if err := rec.Reconcile(ctx, intent); err != nil {
 		slog.Warn("reconcile/drift: reconcile failed", "err", err)
+		return err
 	}
+	return nil
 }

spinifex/network/reconcile/drift_test.go

@@ -0,0 +1,246 @@
+//test:in-package — the backoff ladder is unexported state (driftBackoffBase,
+//nextDriftBackoff, driftWait) and the tests drive runDriftCycle directly to
+//pin what DriftLoop does with its return value.
+
+package reconcile
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+)
+
+// stubReconciler returns a canned outcome per pass, so the requeue schedule can
+// be driven without a live OVN. Safe for concurrent use: DriftLoop calls it from
+// its own goroutine while the test polls the count.
+type stubReconciler struct {
+	mu       sync.Mutex
+	outcomes []error
+	calls    int
+}
+
+var _ Reconciler = (*stubReconciler)(nil)
+
+func (s *stubReconciler) Reconcile(context.Context, IntentState) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	err := s.outcomes[min(s.calls, len(s.outcomes)-1)]
+	s.calls++
+	return err
+}
+
+func (s *stubReconciler) ReconcileApplyOnly(ctx context.Context, intent IntentState) error {
+	return s.Reconcile(ctx, intent)
+}
+
+func (s *stubReconciler) callCount() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.calls
+}
+
+func incompleteErr() error {
+	return fmt.Errorf("reconcile: 1 resource(s) unconverged: %w", ErrPassIncomplete)
+}
+
+// shrinkDriftTiming compresses the loop's timings so a test observes several
+// requeues in milliseconds. The interval stays long so that a loop which
+// wrongly falls back to it registers as silence rather than as a fast requeue.
+func shrinkDriftTiming(t *testing.T, interval, base time.Duration) {
+	t.Helper()
+	oldInterval, oldBase := DriftInterval, driftBackoffBase
+	DriftInterval, driftBackoffBase = interval, base
+	t.Cleanup(func() { DriftInterval, driftBackoffBase = oldInterval, oldBase })
+}
+
+// waitForCalls polls until rec has been called want times or within elapses.
+func waitForCalls(t *testing.T, rec *stubReconciler, want int, within time.Duration) int {
+	t.Helper()
+	deadline := time.Now().Add(within)
+	for time.Now().Before(deadline) {
+		if got := rec.callCount(); got >= want {
+			return got
+		}
+		time.Sleep(time.Millisecond)
+	}
+	return rec.callCount()
+}
+
+// The whole point of the change: an incomplete pass must come back in
+// milliseconds-scaled backoff rather than a full DriftInterval. Driving the real
+// DriftLoop is what pins timer.Reset to the computed wait — a loop that resets
+// to DriftInterval instead still satisfies every pure-function test below.
+func TestDriftLoop_IncompletePassRequeuesFast(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	shrinkDriftTiming(t, 10*time.Second, time.Millisecond)
+
+	rec := &stubReconciler{outcomes: []error{incompleteErr()}}
+	go DriftLoop(t.Context(), rec, nc, "us-east-1a", "node-1", incompleteErr())
+
+	if got := waitForCalls(t, rec, 4, 5*time.Second); got < 4 {
+		t.Fatalf("reconcile called %d times, want >= 4 — an incomplete pass is waiting "+
+			"a full DriftInterval instead of requeueing on the backoff", got)
+	}
+}
+
+// The startup ReconcileApplyOnly pass is where the gateway LRP's first DORA
+// happens. Its outcome must seed the loop, or the resource it failed to converge
+// waits a full DriftInterval for its first retry — the original bug, on boot.
+func TestDriftLoop_SeedsBackoffFromStartupOutcome(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	shrinkDriftTiming(t, 10*time.Second, time.Millisecond)
+
+	// Converged from here on: only the seed can bring the first pass forward.
+	rec := &stubReconciler{outcomes: []error{nil}}
+	go DriftLoop(t.Context(), rec, nc, "us-east-1a", "node-1", incompleteErr())
+
+	if got := waitForCalls(t, rec, 1, 5*time.Second); got < 1 {
+		t.Fatalf("reconcile never ran: a failed startup pass must seed the backoff "+
+			"rather than leave the loop armed at DriftInterval (calls=%d)", got)
+	}
+}
+
+// The converse: a clean startup must not pull the first pass forward, or every
+// vpcd reconciles immediately on boot regardless of need.
+func TestDriftLoop_ConvergedStartupWaitsFullInterval(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	shrinkDriftTiming(t, 10*time.Second, time.Millisecond)
+
+	rec := &stubReconciler{outcomes: []error{incompleteErr()}}
+	go DriftLoop(t.Context(), rec, nc, "us-east-1a", "node-1", nil)
+
+	time.Sleep(300 * time.Millisecond)
+	if got := rec.callCount(); got != 0 {
+		t.Errorf("reconcile ran %d times within 300ms of a converged startup, want 0 "+
+			"— the first pass must wait DriftInterval (10s here)", got)
+	}
+}
+
+// A converged pass must clear the backoff so the loop drops back to its routine
+// cadence, rather than staying on the fast requeue forever.
+func TestDriftLoop_ConvergedPassReturnsToDriftInterval(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	shrinkDriftTiming(t, 10*time.Second, time.Millisecond)
+
+	// Pass 1 incomplete (requeues in ~3ms), pass 2 converged (must requeue at 10s).
+	rec := &stubReconciler{outcomes: []error{incompleteErr(), nil}}
+	go DriftLoop(t.Context(), rec, nc, "us-east-1a", "node-1", incompleteErr())
+
+	if got := waitForCalls(t, rec, 2, 5*time.Second); got < 2 {
+		t.Fatalf("reconcile called %d times, want 2 before the loop settles", got)
+	}
+	time.Sleep(300 * time.Millisecond)
+	if got := rec.callCount(); got != 2 {
+		t.Errorf("reconcile called %d times, want exactly 2 — a converged pass must "+
+			"reset the backoff to DriftInterval, not stay on the fast requeue", got)
+	}
+}
+
+// runDriftCycle's return is what sizes the next wait, so a pass this node did not
+// win must not be reported as converged-or-broken; it reports nil and the caller
+// leaves the cadence alone.
+func TestRunDriftCycle_ReportsReconcileOutcome(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+
+	rec := &stubReconciler{outcomes: []error{incompleteErr()}}
+	err := runDriftCycle(t.Context(), rec, nc, js, "us-east-1a", "node-1")
+	if !errors.Is(err, ErrPassIncomplete) {
+		t.Fatalf("runDriftCycle = %v, want ErrPassIncomplete propagated from the pass", err)
+	}
+
+	converged := &stubReconciler{outcomes: []error{nil}}
+	if err := runDriftCycle(t.Context(), converged, nc, js, "us-east-1a", "node-1"); err != nil {
+		t.Errorf("runDriftCycle = %v, want nil for a converged pass", err)
+	}
+}
+
+// A cycle this node loses the election for must not run a pass at all.
+func TestRunDriftCycle_SkipsWhenNotElected(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+
+	release, ok := AcquireLeader(t.Context(), nc, KVBucketVPCDReconcile, "other-node")
+	if !ok {
+		t.Fatal("failed to take the lock the test needs held")
+	}
+	defer release()
+
+	rec := &stubReconciler{outcomes: []error{incompleteErr()}}
+	if err := runDriftCycle(t.Context(), rec, nc, js, "us-east-1a", "node-1"); err != nil {
+		t.Errorf("runDriftCycle = %v, want nil when not elected", err)
+	}
+	if got := rec.callCount(); got != 0 {
+		t.Errorf("reconcile ran %d times without winning the election, want 0", got)
+	}
+}
+
+// The ladder itself, as pure functions: exact gaps are asserted here because the
+// loop tests above run on compressed timings and can only prove ordering.
+func TestDrift_IncompletePassBacksOffThenResets(t *testing.T) {
+	incomplete := incompleteErr()
+	outcomes := []error{incomplete, incomplete, incomplete, incomplete, incomplete, incomplete, nil}
+
+	var backoff time.Duration
+	var waits []time.Duration
+	for _, outcome := range outcomes {
+		backoff = nextDriftBackoff(backoff, outcome)
+		waits = append(waits, driftWait(backoff))
+	}
+
+	want := []time.Duration{
+		5 * time.Second, 15 * time.Second, 45 * time.Second,
+		135 * time.Second, DriftInterval, DriftInterval, // capped, then held
+		DriftInterval, // converged: back to the routine interval
+	}
+	for i := range want {
+		if waits[i] != want[i] {
+			t.Fatalf("wait after pass %d = %v, want %v (full sequence %v)", i+1, waits[i], want[i], waits)
+		}
+	}
+	// The cap and the reset produce the same wait, so pin the state behind it:
+	// a converged pass must clear the backoff, not sit at the cap.
+	if backoff != 0 {
+		t.Errorf("backoff after a converged pass = %v, want 0", backoff)
+	}
+}
+
+// A pass that converged, or failed for a reason other than non-convergence
+// (e.g. a scan failure), must not shorten the interval — only an incomplete
+// pass has a known-transient resource worth retrying early.
+func TestDrift_NonIncompleteOutcomesKeepDriftInterval(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+	}{
+		{"converged", nil},
+		{"scan failure", errors.New("scan actual OVN state: ovsdb unreachable")},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Start from a backoff already in force, so a reset is observable.
+			if got := nextDriftBackoff(45*time.Second, tt.err); got != 0 {
+				t.Errorf("nextDriftBackoff(45s, %v) = %v, want 0", tt.err, got)
+			}
+			if got := driftWait(0); got != DriftInterval {
+				t.Errorf("driftWait(0) = %v, want %v", got, DriftInterval)
+			}
+		})
+	}
+}
+
+// The backoff must never exceed DriftInterval: a permanently broken resource
+// degrades to today's behaviour instead of hammering the reconcile loop.
+func TestDrift_BackoffCapsAtDriftInterval(t *testing.T) {
+	incomplete := incompleteErr()
+	backoff := DriftInterval
+	for range 3 {
+		backoff = nextDriftBackoff(backoff, incomplete)
+		if backoff > DriftInterval {
+			t.Fatalf("backoff = %v, want <= DriftInterval (%v)", backoff, DriftInterval)
+		}
+	}
+}

spinifex/network/reconcile/lock.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"github.com/mulgadc/spinifex/spinifex/otelsetup"
 	"log/slog"
+	"sync"
 	"time"
 
 	"github.com/nats-io/nats.go"
@@ -15,13 +16,22 @@ import (
 const (
 	KVBucketVPCDReconcile = "spinifex-vpcd-reconcile"
 	reconcileLeaderKey    = "leader"
-	reconcileLeaderTTL    = 60 * time.Second
 
 	// reconcileReleaseTimeout bounds the lock delete on the way out, which runs
 	// on a detached context and so cannot inherit the caller's deadline.
 	reconcileReleaseTimeout = 5 * time.Second
 )
 
+// Leader-key lifetime. Vars (not consts) so tests can shrink them.
+//
+// reconcileLeaderRenew is the gap between refreshes: a pass can outlive the TTL
+// on its own — one stalled gw-lrp DORA runs ~64s — so the key is renewed while
+// the pass runs rather than left to expire underneath it.
+var (
+	reconcileLeaderTTL   = 60 * time.Second
+	reconcileLeaderRenew = 20 * time.Second
+)
+
 // Bounded wait for JetStream quorum on cold multi-node start. Vars (not
 // consts) so tests can shrink them.
 var (
@@ -72,20 +82,80 @@ func AcquireLeader(ctx context.Context, nc *nats.Conn, bucket, holder string) (f
 		}
 	}
 
-	if _, err := kv.Create(ctx, reconcileLeaderKey, []byte(holder)); err != nil {
+	rev, err := kv.Create(ctx, reconcileLeaderKey, []byte(holder))
+	if err != nil {
 		slog.Info("reconcile/lock: another holder is leader, skipping reconcile", "holder", holder, "bucket", bucket, "err", err)
 		return nil, false
 	}
 
 	slog.Info("reconcile/lock: elected", "holder", holder, "bucket", bucket)
+	lease := &leaderLease{kv: kv, rev: rev}
+	renewCtx, stopRenew := context.WithCancel(ctx)
+	go lease.renew(renewCtx, holder, bucket)
+
 	return func() {
+		stopRenew()
+		rev, held := lease.revision()
+		if !held {
+			// Renewal lost the key, so another node may already hold it. An
+			// unguarded delete here would drop that node's lock, not ours.
+			slog.Warn("reconcile/lock: lock already lost before release; not deleting", "holder", holder, "bucket", bucket)
+			return
+		}
 		// Release outlives ctx: shutdown is the common reason to release, and
 		// skipping the delete would park the lock for the full TTL and stall
 		// every other node's reconcile.
 		releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), reconcileReleaseTimeout)
 		defer cancel()
-		if err := kv.Delete(releaseCtx, reconcileLeaderKey); err != nil {
+		if err := kv.Delete(releaseCtx, reconcileLeaderKey, jetstream.LastRevision(rev)); err != nil {
 			slog.Warn("reconcile/lock: failed to release lock (TTL will reap)", "holder", holder, "bucket", bucket, "err", err)
 		}
 	}, true
 }
+
+// leaderLease tracks the revision of a held leader key. Renewal and the final
+// delete are both CAS-guarded on it, so a pass that outlives the TTL can never
+// delete the key of the node that took over from it.
+type leaderLease struct {
+	mu   sync.Mutex
+	kv   jetstream.KeyValue
+	rev  uint64
+	lost bool
+}
+
+// revision returns the current revision and whether the lease is still held.
+func (l *leaderLease) revision() (uint64, bool) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.rev, !l.lost
+}
+
+// renew refreshes the leader key until ctx is cancelled or a refresh loses the
+// CAS, which means the key expired and another node claimed it.
+func (l *leaderLease) renew(ctx context.Context, holder, bucket string) {
+	ticker := time.NewTicker(reconcileLeaderRenew)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			rev, held := l.revision()
+			if !held {
+				return
+			}
+			next, err := l.kv.Update(ctx, reconcileLeaderKey, []byte(holder), rev)
+			if err != nil {
+				l.mu.Lock()
+				l.lost = true
+				l.mu.Unlock()
+				slog.Warn("reconcile/lock: leader key renewal failed; another node may take over mid-pass",
+					"holder", holder, "bucket", bucket, "err", err)
+				return
+			}
+			l.mu.Lock()
+			l.rev = next
+			l.mu.Unlock()
+		}
+	}
+}

spinifex/network/reconcile/lock_test.go

@@ -3,6 +3,7 @@ package reconcile
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/stretchr/testify/assert"
@@ -51,3 +52,54 @@ func TestAcquireLeader_ReleaseSurvivesCancelledContext(t *testing.T) {
 	require.True(t, ok, "lock must be free immediately after release, not held until the TTL reaps it")
 	release2()
 }
+
+// shrinkLeaderTiming compresses the leader key's lifetime so a test can outlast
+// the TTL in a second rather than a minute.
+func shrinkLeaderTiming(t *testing.T, ttl, renew time.Duration) {
+	t.Helper()
+	oldTTL, oldRenew := reconcileLeaderTTL, reconcileLeaderRenew
+	reconcileLeaderTTL, reconcileLeaderRenew = ttl, renew
+	t.Cleanup(func() { reconcileLeaderTTL, reconcileLeaderRenew = oldTTL, oldRenew })
+}
+
+// A pass can outlive the TTL on its own — one stalled gw-lrp DORA runs ~64s
+// against a 60s TTL — so the key must be renewed while the pass runs. Without
+// this a peer elects itself mid-pass and two reconcilers race the same OVN NB.
+func TestAcquireLeader_RenewsKeyBeyondTTL(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	shrinkLeaderTiming(t, time.Second, 100*time.Millisecond)
+
+	release, ok := AcquireLeader(t.Context(), nc, KVBucketVPCDReconcile, "node-1")
+	require.True(t, ok)
+	defer release()
+
+	// Well past the TTL: an unrenewed key would have expired by now.
+	time.Sleep(2500 * time.Millisecond)
+
+	loserRelease, ok := AcquireLeader(t.Context(), nc, KVBucketVPCDReconcile, "node-2")
+	assert.False(t, ok, "leader key expired mid-pass: a peer can now reconcile concurrently")
+	assert.Nil(t, loserRelease)
+}
+
+// If the key was lost anyway, the stale holder's release must not delete the
+// successor's key — an unguarded delete hands the lock to a third node while
+// the successor believes it still holds it.
+func TestAcquireLeader_ReleaseDoesNotDeleteSuccessorKey(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+
+	release, ok := AcquireLeader(t.Context(), nc, KVBucketVPCDReconcile, "node-1")
+	require.True(t, ok)
+
+	// Stand in for a successor that claimed the key after it expired: the write
+	// bumps the revision, so node-1's release is now CAS-stale.
+	kv, err := js.KeyValue(t.Context(), KVBucketVPCDReconcile)
+	require.NoError(t, err)
+	_, err = kv.Put(t.Context(), reconcileLeaderKey, []byte("node-2"))
+	require.NoError(t, err)
+
+	release()
+
+	entry, err := kv.Get(t.Context(), reconcileLeaderKey)
+	require.NoError(t, err, "stale release deleted the successor's leader key")
+	assert.Equal(t, "node-2", string(entry.Value()), "successor's key must survive a stale release")
+}

spinifex/network/reconcile/reconcile.go

@@ -8,6 +8,8 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"maps"
+	"slices"
 	"sync"
 	"time"
 
@@ -17,9 +19,18 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/network/topology"
 )
 
+// ErrPassIncomplete marks a pass that applied what it could but left at least
+// one resource unconverged. A pass keeps applying past a per-resource failure,
+// so without this the caller cannot tell a clean pass from one that dropped ten
+// resources on the floor. DriftLoop tests for it with errors.Is and requeues on
+// a short backoff instead of waiting a full DriftInterval.
+var ErrPassIncomplete = errors.New("reconcile: pass incomplete")
+
 // Reconciler converges OVN NB DB to a declared IntentState. Implementations
 // are idempotent: a second call with the same IntentState is a no-op.
 type Reconciler interface {
+	// Reconcile returns a scan failure as-is, and wraps ErrPassIncomplete when
+	// the scan succeeded but some resources did not converge.
 	Reconcile(ctx context.Context, intent IntentState) error
 	// ReconcileApplyOnly skips orphan-pruning. Startup uses this to avoid
 	// racing peer subscribers that haven't processed in-flight create events
@@ -186,8 +197,58 @@ func New(cfg Config) (Reconciler, error) {
 	}, nil
 }
 
+// passFailure is one resource an apply stage could not converge.
+type passFailure struct {
+	class string
+	id    string
+	err   error
+}
+
+// passResult accumulates per-resource apply failures so a pass that logged and
+// continued past them still reports itself as incomplete to the drift loop.
+type passResult struct {
+	failures []passFailure
+}
+
+// fail records that id in class did not converge.
+func (p *passResult) fail(class, id string, err error) {
+	p.failures = append(p.failures, passFailure{class: class, id: id, err: err})
+}
+
+// summaryKV renders per-class failure counts as slog key/values, class-sorted so
+// the summary line is stable across passes.
+func (p *passResult) summaryKV() []any {
+	counts := make(map[string]int, len(p.failures))
+	for _, f := range p.failures {
+		counts[f.class]++
+	}
+	kv := make([]any, 0, 2+2*len(counts))
+	kv = append(kv, "unconverged", len(p.failures))
+	for _, class := range slices.Sorted(maps.Keys(counts)) {
+		kv = append(kv, class, counts[class])
+	}
+	return kv
+}
+
+// Failure classes for the per-pass convergence summary, one per apply stage so
+// the summary names the stage that did not converge.
+const (
+	classVPC          = "vpc"
+	classSubnet       = "subnet"
+	classSG           = "sg"
+	classPort         = "port"
+	classIGW          = "igw"
+	classEIP          = "eip"
+	classNATGW        = "natgw"
+	classIGWRoute     = "igw_route"
+	classNATGWRoute   = "natgw_route"
+	classDropGate     = "drop_gate"
+	classPublicEgress = "public_egress"
+)
+
 // Reconcile diffs intent vs. actual OVN state and applies in topological order.
-// Per-stage errors are logged; only a scan failure is returned.
+// Per-stage errors are logged and the pass continues; a scan failure is returned
+// as-is and an unconverged resource surfaces as ErrPassIncomplete.
 func (r *reconciler) Reconcile(ctx context.Context, intent IntentState) error {
 	return r.reconcile(ctx, intent, true)
 }
@@ -217,21 +278,26 @@ func (r *reconciler) reconcile(ctx context.Context, intent IntentState, pruneOrp
 		"intent_natgw_routes", len(intent.NATGWRoutes),
 	)
 
-	r.applyVPCs(ctx, intent, actual)
-	r.applySubnets(ctx, intent, actual)
-	r.applySGs(ctx, intent, actual, pruneOrphans)
-	r.applyPorts(ctx, intent, actual, pruneOrphans)
-	r.applyIGWs(ctx, intent, actual)
-	r.applyEIPs(ctx, intent, actual)
+	res := &passResult{}
+	r.applyVPCs(ctx, intent, actual, res)
+	r.applySubnets(ctx, intent, actual, res)
+	r.applySGs(ctx, intent, actual, pruneOrphans, res)
+	r.applyPorts(ctx, intent, actual, pruneOrphans, res)
+	r.applyIGWs(ctx, intent, actual, res)
+	r.applyEIPs(ctx, intent, actual, res)
 	if pruneOrphans {
-		r.pruneOrphanEIPs(ctx, intent)
+		r.pruneOrphanEIPs(ctx, intent, res)
 	}
-	r.applyNATGWs(ctx, intent, actual)
-	r.applyIGWRoutes(ctx, intent, actual)
-	r.applyNATGWRoutes(ctx, intent, actual)
-	r.applyDropGates(ctx, intent, actual)
-	r.applyPublicInstanceEgress(ctx, intent, actual)
+	r.applyNATGWs(ctx, intent, actual, res)
+	r.applyIGWRoutes(ctx, intent, actual, res)
+	r.applyNATGWRoutes(ctx, intent, actual, res)
+	r.applyDropGates(ctx, intent, actual, res)
+	r.applyPublicInstanceEgress(ctx, intent, actual, res)
 
-	slog.Info("reconcile: complete")
-	return nil
+	if len(res.failures) == 0 {
+		slog.Info("reconcile: complete", "converged", true)
+		return nil
+	}
+	slog.Warn("reconcile: pass incomplete", res.summaryKV()...)
+	return fmt.Errorf("reconcile: %d resource(s) unconverged: %w", len(res.failures), ErrPassIncomplete)
 }

spinifex/vpcd/vpcd.go

@@ -700,12 +700,15 @@ func launchService(cfg *Config) error {
 	// Startup reconcile (leader-gated, apply-only). Orphan pruning is skipped because intent may be stale:
 	// a peer's vpc.create-sg could be mid-flight and a prune would sweep those port groups as orphans.
 	// Drift loop uses full Reconcile.
+	// startupErr seeds the drift loop's backoff so a resource the bootstrap pass
+	// could not converge is retried on the short requeue, not a full interval.
+	var startupErr error
 	if isLeader {
 		intent, intentErr := reconcile.LoadIntentFromKV(ctx, js, cfg.AZ)
 		if intentErr != nil {
 			slog.Warn("vpcd: startup intent load failed", "err", intentErr)
-		} else if err := rec.ReconcileApplyOnly(ctx, intent); err != nil {
-			slog.Warn("vpcd: startup reconcile failed", "err", err)
+		} else if startupErr = rec.ReconcileApplyOnly(ctx, intent); startupErr != nil {
+			slog.Warn("vpcd: startup reconcile failed", "err", startupErr)
 		}
 		releaseLeader()
 	}
@@ -715,7 +718,7 @@ func launchService(cfg *Config) error {
 	defer loopCancel()
 	loopDone := make(chan struct{})
 	go func() {
-		reconcile.DriftLoop(loopCtx, rec, nc, cfg.AZ, holder)
+		reconcile.DriftLoop(loopCtx, rec, nc, cfg.AZ, holder, startupErr)
 		close(loopDone)
 	}()