diff --git a/lib/autostandby/controller.go b/lib/autostandby/controller.go index 3c4407f5..dff84cc2 100644 --- a/lib/autostandby/controller.go +++ b/lib/autostandby/controller.go @@ -118,6 +118,29 @@ type controllerState struct { standbyExecuting bool } +type runtimePersistenceErrorMode uint8 + +const ( + runtimePersistencePropagate runtimePersistenceErrorMode = iota + runtimePersistenceBestEffort +) + +type runtimePersistenceLock struct { + mu sync.Mutex + refs int +} + +// runtimePersistence preserves controller mutation order while metadata writes +// run without holding the controller mutex. +type runtimePersistence struct { + id string + runtime *Runtime + generation uint64 + errorMode runtimePersistenceErrorMode + operation string + lock *runtimePersistenceLock +} + // Controller decides when eligible instances should transition to standby. type Controller struct { store InstanceStore @@ -136,11 +159,14 @@ type Controller struct { standbySlots chan struct{} standbyWG sync.WaitGroup - mu sync.RWMutex - states map[string]*controllerState - standbyInFlight int - observerConnected bool - lastObserverErr error + mu sync.RWMutex + states map[string]*controllerState + runtimeGenerations map[string]uint64 + runtimePersistLocks map[string]*runtimePersistenceLock + nextRuntimeGeneration uint64 + standbyInFlight int + observerConnected bool + lastObserverErr error } // NewController creates a new event-driven auto-standby controller. @@ -184,6 +210,8 @@ func NewController(store InstanceStore, source ConnectionSource, opts Controller streamReady: make(chan ConnectionStream, 4), standbySlots: make(chan struct{}, maxConcurrentStandbys), states: make(map[string]*controllerState), + runtimeGenerations: make(map[string]uint64), + runtimePersistLocks: make(map[string]*runtimePersistenceLock), } c.metrics = newMetrics(opts.Meter, opts.Tracer, c) return c @@ -537,16 +565,19 @@ func (c *Controller) periodicSnapshotSync(ctx context.Context) error { func (c *Controller) seedInstanceState(ctx context.Context, inst Instance, conns []Connection, now time.Time) error { c.mu.Lock() - defer c.mu.Unlock() - - return c.refreshInstanceLocked(ctx, inst, conns, now) + persistence, err := c.refreshInstanceLocked(inst, conns, now) + c.mu.Unlock() + if err != nil { + return err + } + return c.persistRuntime(ctx, persistence) } func (c *Controller) handleInstanceEvent(ctx context.Context, event InstanceEvent) error { if event.Action == InstanceEventDelete { c.mu.Lock() - defer c.mu.Unlock() c.removeStateLocked(event.InstanceID) + c.mu.Unlock() return nil } if event.Instance == nil { @@ -559,11 +590,15 @@ func (c *Controller) handleInstanceEvent(ctx context.Context, event InstanceEven } c.mu.Lock() - defer c.mu.Unlock() - return c.refreshInstanceLocked(ctx, *event.Instance, conns, c.now().UTC()) + persistence, err := c.refreshInstanceLocked(*event.Instance, conns, c.now().UTC()) + c.mu.Unlock() + if err != nil { + return err + } + return c.persistRuntime(ctx, persistence) } -func (c *Controller) refreshInstanceLocked(ctx context.Context, inst Instance, conns []Connection, now time.Time) error { +func (c *Controller) refreshInstanceLocked(inst Instance, conns []Connection, now time.Time) (runtimePersistence, error) { state := c.ensureStateLocked(inst.ID) state.instance = cloneInstance(inst) @@ -571,21 +606,21 @@ func (c *Controller) refreshInstanceLocked(ctx context.Context, inst Instance, c hadRuntime := inst.Runtime != nil || state.idleSince != nil || state.lastInboundAt != nil c.clearStateLocked(state) if hadRuntime { - return c.persistRuntime(ctx, inst.ID, nil) + return c.prepareRuntimePersistenceLocked(inst.ID, nil, runtimePersistencePropagate, "refresh disabled instance"), nil } - return nil + return runtimePersistence{}, nil } compiled, err := compilePolicy(inst.AutoStandby) if err != nil { - return err + return runtimePersistence{}, err } state.compiledPolicy = compiled state.idleTimeout = compiled.idleTimeout activeSet, err := matchingConnections(inst, compiled, conns) if err != nil { - return err + return runtimePersistence{}, err } // Cancel any queued standby attempt only once the refresh is guaranteed to // re-establish a countdown or reconcile below; an erroring refresh above @@ -603,11 +638,12 @@ func (c *Controller) refreshInstanceLocked(ctx context.Context, inst Instance, c } c.cancelTimerLocked(state) c.armReconcileLocked(inst.ID, state) - return c.persistRuntime(ctx, inst.ID, &Runtime{ + return c.prepareRuntimePersistenceLocked(inst.ID, &Runtime{ LastInboundActivityAt: cloneTimePtr(state.lastInboundAt), - }) + }, runtimePersistencePropagate, "refresh active instance"), nil } + var persistence runtimePersistence if runtime != nil && runtime.IdleSince != nil { state.idleSince = cloneTimePtr(runtime.IdleSince) state.lastInboundAt = cloneTimePtr(runtime.LastInboundActivityAt) @@ -618,19 +654,13 @@ func (c *Controller) refreshInstanceLocked(ctx context.Context, inst Instance, c } else { state.lastInboundAt = nil } - runtime = &Runtime{ + persistence = c.prepareRuntimePersistenceLocked(inst.ID, &Runtime{ IdleSince: cloneTimePtr(state.idleSince), LastInboundActivityAt: cloneTimePtr(state.lastInboundAt), - } - // Persist failures must not strand the instance without a countdown; - // the runtime only matters for recovery across controller restarts. - if err := c.persistRuntime(ctx, inst.ID, runtime); err != nil { - c.recordControllerError("persist_runtime") - c.log.Warn("auto-standby failed to persist runtime during refresh", "instance_id", inst.ID, "error", err) - } + }, runtimePersistenceBestEffort, "refresh idle instance") } c.armTimerLocked(inst.ID, state, now) - return nil + return persistence, nil } func (c *Controller) handleConnectionEvent(ctx context.Context, event ConnectionEvent) { @@ -649,8 +679,7 @@ func (c *Controller) handleConnectionEvent(ctx context.Context, event Connection c.recordConntrackEvent(string(event.Type), "received") c.mu.Lock() - defer c.mu.Unlock() - + persistences := make([]runtimePersistence, 0, 1) for id, state := range c.states { if state.compiledPolicy == nil { continue @@ -676,13 +705,10 @@ func (c *Controller) handleConnectionEvent(ctx context.Context, event Connection state.standbyRequested = false c.cancelReconcileLocked(state) c.armTimerLocked(id, state, idleSince) - if err := c.persistRuntime(ctx, id, &Runtime{ + persistences = append(persistences, c.prepareRuntimePersistenceLocked(id, &Runtime{ IdleSince: cloneTimePtr(state.idleSince), LastInboundActivityAt: cloneTimePtr(state.lastInboundAt), - }); err != nil { - c.recordControllerError("persist_runtime") - c.log.Warn("auto-standby failed to persist runtime when idle countdown started", "instance_id", id, "error", err) - } + }, runtimePersistenceBestEffort, "start idle countdown")) c.log.Info("auto-standby idle countdown started", "instance_id", id, "idle_timeout", state.idleTimeout) continue } @@ -695,12 +721,9 @@ func (c *Controller) handleConnectionEvent(ctx context.Context, event Connection state.standbyRequested = false c.cancelTimerLocked(state) c.armReconcileLocked(id, state) - if err := c.persistRuntime(ctx, id, &Runtime{ + persistences = append(persistences, c.prepareRuntimePersistenceLocked(id, &Runtime{ LastInboundActivityAt: cloneTimePtr(state.lastInboundAt), - }); err != nil { - c.recordControllerError("persist_runtime") - c.log.Warn("auto-standby failed to persist runtime after inbound activity", "instance_id", id, "error", err) - } + }, runtimePersistenceBestEffort, "record inbound activity")) c.log.Info("auto-standby inbound activity observed", "instance_id", id, "active_inbound_connections", len(state.activeInbound)) case ConnectionEventDestroy: if _, ok := state.activeInbound[key]; !ok { @@ -716,16 +739,18 @@ func (c *Controller) handleConnectionEvent(ctx context.Context, event Connection state.standbyRequested = false c.cancelReconcileLocked(state) c.armTimerLocked(id, state, idleSince) - if err := c.persistRuntime(ctx, id, &Runtime{ + persistences = append(persistences, c.prepareRuntimePersistenceLocked(id, &Runtime{ IdleSince: cloneTimePtr(state.idleSince), LastInboundActivityAt: cloneTimePtr(state.lastInboundAt), - }); err != nil { - c.recordControllerError("persist_runtime") - c.log.Warn("auto-standby failed to persist runtime when idle countdown started", "instance_id", id, "error", err) - } + }, runtimePersistenceBestEffort, "restart idle countdown")) c.log.Info("auto-standby idle countdown started", "instance_id", id, "idle_timeout", state.idleTimeout) } } + c.mu.Unlock() + + for _, persistence := range persistences { + _ = c.persistRuntime(ctx, persistence) + } } // confirmIdleBeforeStandby re-reads the conntrack table and reports whether the @@ -738,12 +763,11 @@ func (c *Controller) confirmIdleBeforeStandby(ctx context.Context, id string) bo conns, listErr := c.source.ListConnections(ctx) c.mu.Lock() - defer c.mu.Unlock() - state := c.states[id] // Activity can land between the timer firing and this check, and it already // owns idleSince and the reconcile loop; the paths below must not clobber it. if state == nil || state.compiledPolicy == nil || len(state.activeInbound) > 0 { + c.mu.Unlock() return false } @@ -756,18 +780,19 @@ func (c *Controller) confirmIdleBeforeStandby(ctx context.Context, id string) bo idleSince := c.now().UTC() state.idleSince = &idleSince c.armTimerLocked(id, state, idleSince) - if persistErr := c.persistRuntime(ctx, id, &Runtime{ + persistence := c.prepareRuntimePersistenceLocked(id, &Runtime{ IdleSince: cloneTimePtr(state.idleSince), LastInboundActivityAt: cloneTimePtr(state.lastInboundAt), - }); persistErr != nil { - c.recordControllerError("persist_runtime") - c.log.Warn("auto-standby failed to persist runtime after unconfirmed standby", "instance_id", id, "error", persistErr) - } + }, runtimePersistenceBestEffort, "handle unconfirmed standby") + c.mu.Unlock() + + _ = c.persistRuntime(ctx, persistence) c.recordControllerError("standby_confirm") c.log.Warn("auto-standby could not confirm idle before standby", "instance_id", id, "error", err) return false } if len(activeSet) == 0 { + c.mu.Unlock() return true } @@ -777,12 +802,12 @@ func (c *Controller) confirmIdleBeforeStandby(ctx context.Context, id string) bo state.lastInboundAt = &now c.cancelTimerLocked(state) c.armReconcileLocked(id, state) - if err := c.persistRuntime(ctx, id, &Runtime{ + persistence := c.prepareRuntimePersistenceLocked(id, &Runtime{ LastInboundActivityAt: cloneTimePtr(state.lastInboundAt), - }); err != nil { - c.recordControllerError("persist_runtime") - c.log.Warn("auto-standby failed to persist runtime after standby confirmation found connections", "instance_id", id, "error", err) - } + }, runtimePersistenceBestEffort, "record standby confirmation activity") + c.mu.Unlock() + + _ = c.persistRuntime(ctx, persistence) c.log.Info("auto-standby skipped standby, conntrack still reports inbound connections", "instance_id", id, "active_inbound_connections", len(activeSet)) return false } @@ -902,32 +927,31 @@ func (c *Controller) executeStandby(ctx context.Context, id string, instanceName c.recordControllerError("standby") c.mu.Lock() - defer c.mu.Unlock() if errors.Is(err, ErrInstanceNotFound) { c.log.Info("auto-standby target instance no longer exists, dropping state", "instance_id", id, "instance_name", instanceName) c.removeStateLocked(id) + c.mu.Unlock() return } c.log.Warn("auto-standby standby attempt failed", "instance_id", id, "instance_name", instanceName, "error", err) + var persistence runtimePersistence if state := c.states[id]; state != nil { state.standbyRequested = false // Inbound activity that arrived during the attempt owns the state // now; the reconcile/destroy flow restarts the countdown once the // connections drain. - if len(state.activeInbound) > 0 { - return - } - idleSince := c.now().UTC() - state.idleSince = &idleSince - c.armTimerLocked(id, state, idleSince) - if persistErr := c.persistRuntime(ctx, id, &Runtime{ - IdleSince: cloneTimePtr(state.idleSince), - LastInboundActivityAt: cloneTimePtr(state.lastInboundAt), - }); persistErr != nil { - c.recordControllerError("persist_runtime") - c.log.Warn("auto-standby failed to persist runtime after standby failure", "instance_id", id, "error", persistErr) + if len(state.activeInbound) == 0 { + idleSince := c.now().UTC() + state.idleSince = &idleSince + c.armTimerLocked(id, state, idleSince) + persistence = c.prepareRuntimePersistenceLocked(id, &Runtime{ + IdleSince: cloneTimePtr(state.idleSince), + LastInboundActivityAt: cloneTimePtr(state.lastInboundAt), + }, runtimePersistenceBestEffort, "recover from standby failure") } } + c.mu.Unlock() + _ = c.persistRuntime(ctx, persistence) return } @@ -935,14 +959,13 @@ func (c *Controller) executeStandby(ctx context.Context, id string, instanceName c.log.Info("instance entered standby due to inbound inactivity", "instance_id", id, "instance_name", instanceName, "idle_timeout", idleTimeout) c.mu.Lock() - defer c.mu.Unlock() + var persistence runtimePersistence if state := c.states[id]; state != nil { c.clearStateLocked(state) - if err := c.persistRuntime(ctx, id, nil); err != nil { - c.recordControllerError("persist_runtime") - c.log.Warn("auto-standby failed to clear runtime after standby", "instance_id", id, "error", err) - } + persistence = c.prepareRuntimePersistenceLocked(id, nil, runtimePersistenceBestEffort, "clear runtime after standby") } + c.mu.Unlock() + _ = c.persistRuntime(ctx, persistence) } func (c *Controller) handleActiveReconcile(ctx context.Context, id string) { @@ -962,10 +985,9 @@ func (c *Controller) handleActiveReconcile(ctx context.Context, id string) { now := c.now().UTC() c.mu.Lock() - defer c.mu.Unlock() - state := c.states[id] if state == nil || state.compiledPolicy == nil { + c.mu.Unlock() return } @@ -976,6 +998,7 @@ func (c *Controller) handleActiveReconcile(ctx context.Context, id string) { if len(state.activeInbound) > 0 { c.armReconcileLocked(id, state) } + c.mu.Unlock() return } @@ -983,6 +1006,7 @@ func (c *Controller) handleActiveReconcile(ctx context.Context, id string) { if len(activeSet) > 0 { state.standbyRequested = false c.armReconcileLocked(id, state) + c.mu.Unlock() return } @@ -990,14 +1014,15 @@ func (c *Controller) handleActiveReconcile(ctx context.Context, id string) { state.standbyRequested = false c.cancelReconcileLocked(state) c.armTimerLocked(id, state, now) - if err := c.persistRuntime(ctx, id, &Runtime{ + persistence := c.prepareRuntimePersistenceLocked(id, &Runtime{ IdleSince: cloneTimePtr(state.idleSince), LastInboundActivityAt: cloneTimePtr(state.lastInboundAt), - }); err != nil { - c.recordControllerError("persist_runtime") - c.log.Warn("auto-standby failed to persist runtime after active connection reconcile drained", "instance_id", id, "error", err) - } - c.log.Info("auto-standby idle countdown started after active connection reconcile", "instance_id", id, "idle_timeout", state.idleTimeout) + }, runtimePersistenceBestEffort, "finish active connection reconcile") + idleTimeout := state.idleTimeout + c.mu.Unlock() + + _ = c.persistRuntime(ctx, persistence) + c.log.Info("auto-standby idle countdown started after active connection reconcile", "instance_id", id, "idle_timeout", idleTimeout) } func (c *Controller) reconnectStream(ctx context.Context) { @@ -1048,6 +1073,7 @@ func (c *Controller) removeStateLocked(id string) { c.cancelReconcileLocked(state) } delete(c.states, id) + delete(c.runtimeGenerations, id) } func (c *Controller) clearStateLocked(state *controllerState) { @@ -1129,8 +1155,61 @@ func (c *Controller) stopAllTimers() { } } -func (c *Controller) persistRuntime(ctx context.Context, id string, runtime *Runtime) error { - return c.store.SetRuntime(ctx, id, runtime) +func (c *Controller) prepareRuntimePersistenceLocked(id string, runtime *Runtime, errorMode runtimePersistenceErrorMode, operation string) runtimePersistence { + c.nextRuntimeGeneration++ + generation := c.nextRuntimeGeneration + c.runtimeGenerations[id] = generation + + lock := c.runtimePersistLocks[id] + if lock == nil { + lock = &runtimePersistenceLock{} + c.runtimePersistLocks[id] = lock + } + lock.refs++ + + return runtimePersistence{ + id: id, + runtime: cloneRuntime(runtime), + generation: generation, + errorMode: errorMode, + operation: operation, + lock: lock, + } +} + +func (c *Controller) persistRuntime(ctx context.Context, persistence runtimePersistence) error { + if persistence.generation == 0 { + return nil + } + + persistence.lock.mu.Lock() + defer c.releaseRuntimePersistenceLock(persistence) + defer persistence.lock.mu.Unlock() + + c.mu.RLock() + generation := c.runtimeGenerations[persistence.id] + c.mu.RUnlock() + if generation != persistence.generation { + return nil + } + + err := c.store.SetRuntime(ctx, persistence.id, persistence.runtime) + if err != nil && persistence.errorMode == runtimePersistenceBestEffort { + c.recordControllerError("persist_runtime") + c.log.Warn("auto-standby failed to persist runtime", "instance_id", persistence.id, "operation", persistence.operation, "error", err) + return nil + } + return err +} + +func (c *Controller) releaseRuntimePersistenceLock(persistence runtimePersistence) { + c.mu.Lock() + defer c.mu.Unlock() + + persistence.lock.refs-- + if persistence.lock.refs == 0 && c.runtimePersistLocks[persistence.id] == persistence.lock { + delete(c.runtimePersistLocks, persistence.id) + } } func (c *Controller) setObserverConnected(connected bool) { diff --git a/lib/autostandby/controller_test.go b/lib/autostandby/controller_test.go index 10d8b7d7..b8ead05c 100644 --- a/lib/autostandby/controller_test.go +++ b/lib/autostandby/controller_test.go @@ -14,16 +14,19 @@ import ( ) type fakeInstanceStore struct { - mu sync.Mutex - instances []Instance - standbyIDs []string - persistedRuntime map[string]*Runtime - events chan InstanceEvent - standbyErr error - listErr error - setRuntimeErr error - standbyStarted chan string - standbyRelease chan struct{} + mu sync.Mutex + instances []Instance + standbyIDs []string + persistedRuntime map[string]*Runtime + events chan InstanceEvent + standbyErr error + listErr error + setRuntimeErr error + setRuntimeStarted chan string + setRuntimeRelease chan struct{} + setRuntimeReleaseByID map[string]chan struct{} + standbyStarted chan string + standbyRelease chan struct{} } func newFakeInstanceStore(instances []Instance) *fakeInstanceStore { @@ -67,6 +70,17 @@ func (f *fakeInstanceStore) standbyCalls() []string { } func (f *fakeInstanceStore) SetRuntime(_ context.Context, id string, runtime *Runtime) error { + if f.setRuntimeStarted != nil { + f.setRuntimeStarted <- id + } + release := f.setRuntimeRelease + if f.setRuntimeReleaseByID != nil { + release = f.setRuntimeReleaseByID[id] + } + if release != nil { + <-release + } + f.mu.Lock() defer f.mu.Unlock() if f.setRuntimeErr != nil { @@ -1348,6 +1362,162 @@ func TestRefreshPersistFailureStillArmsIdleTimer(t *testing.T) { controller.mu.RUnlock() } +func TestRuntimePersistenceDoesNotBlockOtherInstances(t *testing.T) { + t.Parallel() + + now := time.Date(2026, 4, 6, 10, 55, 0, 0, time.UTC) + first := Instance{ + ID: "inst-persist-first", + Name: "inst-persist-first", + State: StateRunning, + NetworkEnabled: true, + IP: "192.168.100.110", + AutoStandby: &Policy{Enabled: true, IdleTimeout: "1m"}, + } + second := first + second.ID = "inst-persist-second" + second.Name = second.ID + second.IP = "192.168.100.111" + conns := []Connection{ + { + OriginalSourceIP: mustAddr("1.2.3.4"), + OriginalSourcePort: 50010, + OriginalDestinationIP: mustAddr(first.IP), + OriginalDestinationPort: 8080, + TCPState: TCPStateEstablished, + }, + { + OriginalSourceIP: mustAddr("1.2.3.4"), + OriginalSourcePort: 50011, + OriginalDestinationIP: mustAddr(second.IP), + OriginalDestinationPort: 8080, + TCPState: TCPStateEstablished, + }, + } + firstRelease := make(chan struct{}) + store := newFakeInstanceStore([]Instance{first, second}) + store.setRuntimeStarted = make(chan string, 2) + store.setRuntimeReleaseByID = map[string]chan struct{}{first.ID: firstRelease} + controller := NewController(store, &fakeConnectionSource{}, ControllerOptions{ + Now: func() time.Time { return now }, + }) + + firstDone := make(chan error, 1) + go func() { + firstDone <- controller.seedInstanceState(context.Background(), first, conns, now) + }() + require.Equal(t, first.ID, <-store.setRuntimeStarted) + + secondDone := make(chan error, 1) + go func() { + secondDone <- controller.seedInstanceState(context.Background(), second, conns, now) + }() + + select { + case id := <-store.setRuntimeStarted: + require.Equal(t, second.ID, id) + case <-time.After(time.Second): + close(firstRelease) + require.FailNow(t, "runtime persistence waited on another instance") + } + select { + case err := <-secondDone: + require.NoError(t, err) + case <-time.After(time.Second): + close(firstRelease) + require.FailNow(t, "runtime persistence did not complete for another instance") + } + + close(firstRelease) + require.NoError(t, <-firstDone) +} + +func TestRuntimePersistenceKeepsLatestGeneration(t *testing.T) { + t.Parallel() + + store := newFakeInstanceStore(nil) + controller := NewController(store, &fakeConnectionSource{}, ControllerOptions{}) + firstTime := time.Date(2026, 4, 6, 10, 55, 0, 0, time.UTC) + secondTime := firstTime.Add(time.Second) + + controller.mu.Lock() + first := controller.prepareRuntimePersistenceLocked("inst-persist-order", &Runtime{IdleSince: &firstTime}, runtimePersistencePropagate, "test first generation") + second := controller.prepareRuntimePersistenceLocked("inst-persist-order", &Runtime{IdleSince: &secondTime}, runtimePersistencePropagate, "test second generation") + controller.mu.Unlock() + + require.NoError(t, controller.persistRuntime(context.Background(), first)) + assert.NotContains(t, store.persistedRuntime, "inst-persist-order") + require.NoError(t, controller.persistRuntime(context.Background(), second)) + require.NotNil(t, store.persistedRuntime["inst-persist-order"]) + assert.Equal(t, secondTime, *store.persistedRuntime["inst-persist-order"].IdleSince) +} + +func TestRuntimePersistenceSkipsDeletedInstance(t *testing.T) { + t.Parallel() + + store := newFakeInstanceStore(nil) + controller := NewController(store, &fakeConnectionSource{}, ControllerOptions{}) + now := time.Date(2026, 4, 6, 10, 55, 0, 0, time.UTC) + + controller.mu.Lock() + persistence := controller.prepareRuntimePersistenceLocked("inst-persist-deleted", &Runtime{IdleSince: &now}, runtimePersistencePropagate, "test deleted instance") + controller.removeStateLocked("inst-persist-deleted") + controller.mu.Unlock() + + require.NoError(t, controller.persistRuntime(context.Background(), persistence)) + assert.NotContains(t, store.persistedRuntime, "inst-persist-deleted") +} + +func TestHoldStandbyDoesNotWaitForRuntimePersistence(t *testing.T) { + t.Parallel() + + now := time.Date(2026, 4, 6, 10, 55, 0, 0, time.UTC) + inst := Instance{ + ID: "inst-hold-during-persist", + Name: "inst-hold-during-persist", + State: StateRunning, + NetworkEnabled: true, + IP: "192.168.100.100", + AutoStandby: &Policy{Enabled: true, IdleTimeout: "1m"}, + } + conn := Connection{ + OriginalSourceIP: mustAddr("1.2.3.4"), + OriginalSourcePort: 50010, + OriginalDestinationIP: mustAddr(inst.IP), + OriginalDestinationPort: 8080, + TCPState: TCPStateEstablished, + } + store := newFakeInstanceStore([]Instance{inst}) + store.setRuntimeStarted = make(chan string, 1) + store.setRuntimeRelease = make(chan struct{}) + controller := NewController(store, &fakeConnectionSource{connections: []Connection{conn}}, ControllerOptions{ + Now: func() time.Time { return now }, + }) + + resyncDone := make(chan error, 1) + go func() { + resyncDone <- controller.startupResync(context.Background()) + }() + require.Equal(t, inst.ID, <-store.setRuntimeStarted) + + holdDone := make(chan error, 1) + go func() { + _, err := controller.HoldStandby(context.Background(), inst) + holdDone <- err + }() + + select { + case err := <-holdDone: + require.NoError(t, err) + case <-time.After(time.Second): + close(store.setRuntimeRelease) + require.FailNow(t, "hold waited for runtime persistence") + } + + close(store.setRuntimeRelease) + require.NoError(t, <-resyncDone) +} + func TestHoldStandbyExtendsArmedCountdown(t *testing.T) { t.Parallel()