diff --git a/spinifex/handlers/dns/reconcile.go b/spinifex/handlers/dns/reconcile.go index 96d4057d7..4a182f366 100644 --- a/spinifex/handlers/dns/reconcile.go +++ b/spinifex/handlers/dns/reconcile.go @@ -129,9 +129,11 @@ func (r *Reconciler) Run(ctx context.Context) { reconciler.Run(ctx, reconciler.Config{ Name: "dns", Sources: r.sources, - Reconcile: func(ctx context.Context) error { + // No revisit deadline: the desired set is a function of the buckets + // being watched, so a change is the only reason to run early. + Reconcile: func(ctx context.Context) (time.Duration, error) { r.reconcileOnce(ctx) - return nil + return 0, nil }, Resync: r.interval, }) diff --git a/spinifex/handlers/rds/backup_test.go b/spinifex/handlers/rds/backup_test.go index 89e56336c..22bf0960a 100644 --- a/spinifex/handlers/rds/backup_test.go +++ b/spinifex/handlers/rds/backup_test.go @@ -798,7 +798,7 @@ func TestReconciler_RunsTheBackupWindowOnItsAccountPass(t *testing.T) { rec.Agent = AgentState{InstanceID: testInstance, EngineHealth: EngineHealthHealthy, LastSeen: &now} seedInstance(t, h.svc, rec) - require.NoError(t, NewReconciler(h.svc, "node-a").reconcileOnce(t.Context())) + require.NoError(t, onePass(t, NewReconciler(h.svc, "node-a"))) assert.Len(t, h.snaps.created, 1) assert.Len(t, h.automatedStamps(t, testDBID), 1) diff --git a/spinifex/handlers/rds/delete_test.go b/spinifex/handlers/rds/delete_test.go index 6bbb986f0..891c0e707 100644 --- a/spinifex/handlers/rds/delete_test.go +++ b/spinifex/handlers/rds/delete_test.go @@ -358,7 +358,7 @@ func TestReconciler_ResumesAnInterruptedDelete(t *testing.T) { seedInstance(t, h.svc, rec) reconciler := NewReconciler(h.svc, "node-a") - require.NoError(t, reconciler.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, reconciler)) assert.Equal(t, []string{testInstance}, h.launcher.terminated) assert.False(t, h.recordExists(t, testDBID)) @@ -374,7 +374,7 @@ func TestReconciler_ResumesAnInterruptedStop(t *testing.T) { seedInstance(t, h.svc, rec) reconciler := NewReconciler(h.svc, "node-a") - require.NoError(t, reconciler.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, reconciler)) assert.Equal(t, []string{"stop:" + testInstance}, h.cmdr.calls) assert.Equal(t, StatusStopped, h.record(t).Status) diff --git a/spinifex/handlers/rds/lifecycle_test.go b/spinifex/handlers/rds/lifecycle_test.go index a1ffd1123..5e73460ff 100644 --- a/spinifex/handlers/rds/lifecycle_test.go +++ b/spinifex/handlers/rds/lifecycle_test.go @@ -507,7 +507,7 @@ func TestReconciler_DoesNotCallAStillRunningVMStopped(t *testing.T) { rec.TransitionStartedAt = &started seedInstance(t, h.svc, rec) - require.NoError(t, NewReconciler(h.svc, "node-a").reconcileOnce(t.Context())) + require.NoError(t, onePass(t, NewReconciler(h.svc, "node-a"))) // Left stopping so the next pass retries; the bound is what ends it. assert.Equal(t, StatusStopping, h.record(t).Status) @@ -676,7 +676,7 @@ func TestReconciler_CompletesARestartOnAHeartbeatFromTheRestartedEngine(t *testi now := time.Now().UTC() seedInstance(t, h.svc, restartingRecord(status, now.Add(-time.Minute), now)) - require.NoError(t, NewReconciler(h.svc, "node-a").reconcileOnce(t.Context())) + require.NoError(t, onePass(t, NewReconciler(h.svc, "node-a"))) rec := h.record(t) assert.Equal(t, StatusAvailable, rec.Status) @@ -695,7 +695,7 @@ func TestReconciler_CompletesARestartOnAPersistedHeartbeatInsideTheFloor(t *test persisted := time.Now().UTC().Add(-HeartbeatStaleAfter - time.Minute) seedInstance(t, h.svc, restartingRecord(StatusRebooting, persisted.Add(-time.Second), persisted)) - require.NoError(t, NewReconciler(h.svc, "node-a").reconcileOnce(t.Context())) + require.NoError(t, onePass(t, NewReconciler(h.svc, "node-a"))) assert.Equal(t, StatusAvailable, h.record(t).Status) } @@ -709,7 +709,7 @@ func TestReconciler_IgnoresAHeartbeatPredatingTheRestart(t *testing.T) { now := time.Now().UTC() seedInstance(t, h.svc, restartingRecord(StatusRebooting, now, now.Add(-time.Minute))) - require.NoError(t, NewReconciler(h.svc, "node-a").reconcileOnce(t.Context())) + require.NoError(t, onePass(t, NewReconciler(h.svc, "node-a"))) assert.Equal(t, StatusRebooting, h.record(t).Status) } @@ -723,7 +723,7 @@ func TestReconciler_MarksFailedWhenARestartOverrunsItsBound(t *testing.T) { started := now.Add(-2 * transitionTimeout) seedInstance(t, h.svc, restartingRecord(StatusStarting, started, started.Add(-time.Minute))) - require.NoError(t, NewReconciler(h.svc, "node-a").reconcileOnce(t.Context())) + require.NoError(t, onePass(t, NewReconciler(h.svc, "node-a"))) rec := h.record(t) assert.Equal(t, StatusFailed, rec.Status) diff --git a/spinifex/handlers/rds/modify_lease_test.go b/spinifex/handlers/rds/modify_lease_test.go index b20f81778..1fb2433af 100644 --- a/spinifex/handlers/rds/modify_lease_test.go +++ b/spinifex/handlers/rds/modify_lease_test.go @@ -34,7 +34,7 @@ func TestModifyDBInstance_HoldsTheLeaseSoAReconcilePassCannotReEnterTheChange(t var sweep sync.Once var swept error h.storage.onModify = func() { - sweep.Do(func() { swept = h.rec.reconcileOnce(t.Context()) }) + sweep.Do(func() { swept = onePass(t, h.rec) }) } in := modifyInput() @@ -57,7 +57,7 @@ func TestReconciler_LeavesAModifyItsHolderIsStillInside(t *testing.T) { rec.ModifyLease = heldLease(modifyLeaseTTL) seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) assert.Empty(t, h.storage.modified) assert.Empty(t, h.cmdr.calls) @@ -76,7 +76,7 @@ func TestReconciler_ResumesAModifyWhoseHolderIsGone(t *testing.T) { rec.ModifyLease = heldLease(-time.Second) seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) assert.Equal(t, int64(50), h.storage.sizes[testDataVolume]) assert.Equal(t, int64(50), h.record(t).AllocatedStorage) @@ -91,7 +91,7 @@ func TestApplyPendingModifications_ReleasesTheLeaseWhateverTheOutcome(t *testing seedInstance(t, h.svc, modifyingRecord( &PendingModifiedValues{AllocatedStorage: aws.Int64(50), RequestedAt: time.Now().UTC()})) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) assert.Nil(t, h.record(t).ModifyLease) }) @@ -101,7 +101,7 @@ func TestApplyPendingModifications_ReleasesTheLeaseWhateverTheOutcome(t *testing seedInstance(t, h.svc, modifyingRecord( &PendingModifiedValues{AllocatedStorage: aws.Int64(50), RequestedAt: time.Now().UTC()})) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) assert.Nil(t, h.record(t).ModifyLease, "a failed attempt must not hold the lease against its own retry") }) } @@ -118,7 +118,7 @@ func TestReconciler_FailsAModifyWhoseHolderOverrunsTheBudget(t *testing.T) { rec.TransitionStartedAt = &started seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) stored := h.record(t) assert.Equal(t, StatusFailed, stored.Status) @@ -156,7 +156,7 @@ func TestWithModifyLease_RenewsAndCancelsTheApplyWhenTakenOver(t *testing.T) { return lease != nil && lease.ExpiresAt.After(initial.ExpiresAt) }, time.Second, 10*time.Millisecond, "a blocking apply must keep extending its lease") - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) assert.Empty(t, h.storage.modified, "a reconcile pass must not claim a renewed lease") takeoverExpiry := time.Now().UTC().Add(time.Second) diff --git a/spinifex/handlers/rds/pending_test.go b/spinifex/handlers/rds/pending_test.go index 469f77840..ab76b49df 100644 --- a/spinifex/handlers/rds/pending_test.go +++ b/spinifex/handlers/rds/pending_test.go @@ -251,7 +251,7 @@ func TestReconciler_ResumesAnInterruptedModify(t *testing.T) { rec := modifyingRecord(&PendingModifiedValues{AllocatedStorage: aws.Int64(50), RequestedAt: time.Now().UTC()}) seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) assert.Equal(t, int64(50), h.storage.sizes[testDataVolume]) stored := h.record(t) @@ -271,7 +271,7 @@ func TestReconciler_FinishesTheFilesystemGrowOnceTheAgentIsBack(t *testing.T) { rec.Agent = AgentState{InstanceID: testInstance, EngineHealth: EngineHealthHealthy, LastSeen: &beat} seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) issued := h.agent.received() require.Len(t, issued, 1) @@ -280,7 +280,7 @@ func TestReconciler_FinishesTheFilesystemGrowOnceTheAgentIsBack(t *testing.T) { // The record moved under the revision that pass read, so the transition to // available is the next pass's rather than a raced write. - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) assert.Equal(t, StatusAvailable, h.record(t).Status) } @@ -295,7 +295,7 @@ func TestReconciler_FailsAModifyThatOverrunsItsBudget(t *testing.T) { rec.Agent = AgentState{} seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) stored := h.record(t) assert.Equal(t, StatusFailed, stored.Status) @@ -313,7 +313,7 @@ func TestReconciler_FailsAModifyItCannotApplyWithinTheBudget(t *testing.T) { rec.TransitionStartedAt = &started seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) stored := h.record(t) assert.Equal(t, StatusFailed, stored.Status) @@ -331,11 +331,11 @@ func TestReconciler_RetriesAFailedModifyInsideTheBudget(t *testing.T) { rec := modifyingRecord(&PendingModifiedValues{AllocatedStorage: aws.Int64(50), RequestedAt: time.Now().UTC()}) seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) assert.Equal(t, StatusModifying, h.record(t).Status) h.storage.modifyErr = nil - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) assert.Equal(t, int64(50), h.record(t).AllocatedStorage) } diff --git a/spinifex/handlers/rds/reconciler.go b/spinifex/handlers/rds/reconciler.go index e5a6d57a3..537bd7fc6 100644 --- a/spinifex/handlers/rds/reconciler.go +++ b/spinifex/handlers/rds/reconciler.go @@ -12,6 +12,8 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/kvlease" + "github.com/mulgadc/spinifex/spinifex/kvstore" + "github.com/mulgadc/spinifex/spinifex/reconciler" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go/jetstream" ) @@ -23,6 +25,11 @@ const ( leaseRefresh = 20 * time.Second reconcileInterval = 15 * time.Second + // The drift backstop, and so also the cadence on which a backup or + // maintenance window is noticed. A window is a half-hour slot, so noticing + // one this late is well inside what it promises. + reconcileResync = 5 * time.Minute + // How long a creating instance may go without a healthy heartbeat before it // is called failed, unless the config overrides it. It has to cover a cold // boot plus initdb on the slowest class, so it is deliberately generous — a @@ -126,22 +133,39 @@ func (r *Reconciler) Run(ctx context.Context) { <-leadershipDone }() - reconcileTicker := time.NewTicker(reconcileInterval) - defer reconcileTicker.Stop() + reconciler.Run(ctx, reconciler.Config{ + Name: "rds", + Sources: []reconciler.Source{reconciler.Dynamic(r.watchBuckets, ">")}, + Reconcile: r.reconcilePass, + Resync: reconcileResync, + }) +} - for { - select { - case <-ctx.Done(): - return - case <-reconcileTicker.C: - if !r.isLeader() { - continue - } - if err := r.reconcileOnce(ctx); err != nil { - slog.ErrorContext(ctx, "rds reconciler: pass failed", "holder", r.holder, "err", err) - } - } +// Every per-account bucket, re-read each resync so a new tenant is watched +// without a restart. The filter is the whole bucket because a DB instance key +// is one subject token, which no prefix wildcard can match. +func (r *Reconciler) watchBuckets(ctx context.Context) ([]*kvstore.Bucket, error) { + js, err := r.svc.js() + if err != nil { + return nil, err } + return AccountWatchBuckets(ctx, js) +} + +// One pass, and when the loop should run again with nothing having changed. +func (r *Reconciler) reconcilePass(ctx context.Context) (time.Duration, error) { + if !r.isLeader() { + // Leadership changes without anything being written, so a follower has + // to keep asking. The pass itself is a lease check and nothing more. + return leaseRefresh, nil + } + revisit, err := r.reconcileOnce(ctx) + if err != nil { + slog.ErrorContext(ctx, "rds reconciler: pass failed", "holder", r.holder, "err", err) + } + // Already logged, so it is not returned again: the loop would log it a + // second time under its own message. + return revisit, nil } // The shared GC backstop's cluster-wide gate. The reconciler's lease is already @@ -166,15 +190,16 @@ func (r *Reconciler) leaderBucket(ctx context.Context) (jetstream.KeyValue, erro // One pass across every tenant. A bucket that cannot be read stops the pass // with an error rather than being skipped silently, so a partial view shows up // in the log instead of looking like a fleet with nothing to do. -func (r *Reconciler) reconcileOnce(ctx context.Context) error { +func (r *Reconciler) reconcileOnce(ctx context.Context) (time.Duration, error) { js, err := r.svc.js() if err != nil { - return err + return 0, err } buckets, err := AccountBucketNames(ctx, js) if err != nil { - return fmt.Errorf("rds reconciler: enumerate account buckets: %w", err) + return 0, fmt.Errorf("rds reconciler: enumerate account buckets: %w", err) } + var revisit time.Duration var failures []error for _, bucket := range buckets { kv, err := js.KeyValue(ctx, bucket) @@ -182,67 +207,140 @@ func (r *Reconciler) reconcileOnce(ctx context.Context) error { failures = append(failures, fmt.Errorf("open %s: %w", bucket, err)) continue } - if err := r.reconcileAccount(ctx, kv, AccountIDFromBucketName(bucket)); err != nil { + due, err := r.reconcileAccount(ctx, kv, AccountIDFromBucketName(bucket)) + revisit = reconciler.Earliest(revisit, due) + if err != nil { failures = append(failures, fmt.Errorf("reconcile %s: %w", bucket, err)) } } - return errors.Join(failures...) + return revisit, errors.Join(failures...) } -func (r *Reconciler) reconcileAccount(ctx context.Context, kv jetstream.KeyValue, accountID string) error { +func (r *Reconciler) reconcileAccount(ctx context.Context, kv jetstream.KeyValue, accountID string) (time.Duration, error) { ids, err := ListDBInstanceIDs(ctx, kv) if err != nil { - return err + return 0, err } + var revisit time.Duration var failures []error for _, id := range ids { - if err := r.reconcileInstance(ctx, kv, accountID, id); err != nil { + due, err := r.reconcileInstance(ctx, kv, accountID, id) + revisit = reconciler.Earliest(revisit, due) + if err != nil { failures = append(failures, awserrors.Errorf(id, "%w", err)) } if err := r.reconcileWindows(ctx, kv, accountID, id); err != nil { failures = append(failures, awserrors.Errorf(id, "%w", err)) } } - if err := r.reconcileSnapshots(ctx, kv, accountID); err != nil { + due, err := r.reconcileSnapshots(ctx, kv, accountID) + revisit = reconciler.Earliest(revisit, due) + if err != nil { failures = append(failures, err) } - return errors.Join(failures...) + return revisit, errors.Join(failures...) } // The reconciler owns every transitional state that no single API call can // finish: the one it drives itself (creating), and the ones whose caller may // have died partway through. A settled instance is left alone. -func (r *Reconciler) reconcileInstance(ctx context.Context, kv jetstream.KeyValue, accountID, id string) error { +func (r *Reconciler) reconcileInstance(ctx context.Context, kv jetstream.KeyValue, accountID, id string) (time.Duration, error) { var rec DBInstanceRecord rev, found, err := getJSONRevision(ctx, kv, DBInstanceKey(id), &rec) if err != nil || !found { - return err + return 0, err } + // Read from the record as it stands, not as a handler below leaves it: a + // handler that moves the status writes to KV, and that write wakes the loop + // on its own. + revisit := r.revisitFor(accountID, &rec) if err := r.reportStalePendingBootstrap(ctx, kv, accountID, &rec); err != nil { - return err + return revisit, err } remediated, err := r.remediateSystemENISG(ctx, kv, rev, &rec) if err != nil || remediated { - return err + return revisit, err } + var handled error switch rec.Status { case StatusCreating: - return r.reconcileCreating(ctx, kv, rev, accountID, &rec) + handled = r.reconcileCreating(ctx, kv, rev, accountID, &rec) case StatusRebooting, StatusStarting: - return r.reconcileRestarting(ctx, kv, rev, accountID, &rec) + handled = r.reconcileRestarting(ctx, kv, rev, accountID, &rec) case StatusModifying: - return r.reconcileModifying(ctx, kv, rev, accountID, &rec) + handled = r.reconcileModifying(ctx, kv, rev, accountID, &rec) case StatusStopping: - return r.reconcileStopping(ctx, kv, rev, accountID, &rec) + handled = r.reconcileStopping(ctx, kv, rev, accountID, &rec) case StatusDeleting: - return r.reconcileDeleting(ctx, kv, rev, accountID, &rec) + handled = r.reconcileDeleting(ctx, kv, rev, accountID, &rec) case StatusBackingUp: - return r.reconcileBackingUp(ctx, kv, rev, accountID, &rec) + handled = r.reconcileBackingUp(ctx, kv, rev, accountID, &rec) case StatusAvailable, StatusFailed: - return r.reconcileHealth(ctx, kv, rev, accountID, &rec) - default: - return nil + handled = r.reconcileHealth(ctx, kv, rev, accountID, &rec) } + return revisit, handled +} + +// When this instance next needs looking at with nothing having changed, which +// is the only thing a watch cannot tell the loop. +// +// A transitional record is waiting on a VM lifecycle state that lives in EC2 and +// is never written to KV, so there is no signal to wait for and it keeps the +// interval the ticker used to provide. A settled one is waiting for its +// heartbeat to go stale — silence, which by definition writes nothing — so it +// asks to be looked at the instant the beat expires, which is sharper than the +// ticker managed. Anything else is genuinely inert until someone writes. +func (r *Reconciler) revisitFor(accountID string, rec *DBInstanceRecord) time.Duration { + switch rec.Status { + case StatusCreating, StatusRebooting, StatusStarting, StatusModifying, + StatusStopping, StatusDeleting, StatusBackingUp: + return reconcileInterval + case StatusFailed: + // Terminal for the control plane, and recovery arrives as a heartbeat, + // which is a write the watch already sees. + return 0 + case StatusAvailable: + return r.settledRevisit(accountID, rec) + default: + return 0 + } +} + +// The deadline for an available instance, which is whichever of its two clocks +// runs out first. +// +// Before the beat expires, that is the expiry itself. After it, the failure +// clock is what decides: the instance is dark but stays available until the +// grace window passes and a later pass agrees it is still dark. Returning +// nothing here was wrong — it left the pass that does the failing with nothing +// to schedule it, so a dead database waited for the resync. +func (r *Reconciler) settledRevisit(accountID string, rec *DBInstanceRecord) time.Duration { + lastSeen, bound := r.lastHeartbeat(accountID, rec) + if lastSeen.IsZero() { + // Never reported at all, so there is no beat to expire and the failure + // clock is the only one running. + return r.failureRevisit(rec) + } + if expiry := time.Until(lastSeen.Add(bound)); expiry > 0 { + return expiry + } + return r.failureRevisit(rec) +} + +// How long until the failure grace window closes. An unstamped clock is stamped +// by the pass that just ran, so the window starts from here. +func (r *Reconciler) failureRevisit(rec *DBInstanceRecord) time.Duration { + grace := r.svc.failureGrace() + if rec.UnhealthySince == nil { + return grace + } + if remaining := time.Until(rec.UnhealthySince.Add(grace)); remaining > 0 { + return remaining + } + // Already due and still not failed, which is the degraded reading: dark agent, + // live VM. Nothing has changed, so retry at the old interval rather than + // immediately, which would spin for as long as the condition holds. + return reconcileInterval } // Moves a system NIC launched before the RDS system security group existed onto @@ -565,11 +663,12 @@ func (r *Reconciler) reconcileWindows(ctx context.Context, kv jetstream.KeyValue // snapshot is the authority: it was tagged with the DB snapshot identifier before // the record was flipped, so its presence says the data exists and its absence // says the create died before cutting it. -func (r *Reconciler) reconcileSnapshots(ctx context.Context, kv jetstream.KeyValue, accountID string) error { +func (r *Reconciler) reconcileSnapshots(ctx context.Context, kv jetstream.KeyValue, accountID string) (time.Duration, error) { ids, err := ListDBSnapshotIDs(ctx, kv) if err != nil { - return fmt.Errorf("list DB snapshots: %w", err) + return 0, fmt.Errorf("list DB snapshots: %w", err) } + var revisit time.Duration var failures []error for _, id := range ids { var rec DBSnapshotRecord @@ -578,15 +677,20 @@ func (r *Reconciler) reconcileSnapshots(ctx context.Context, kv jetstream.KeyVal failures = append(failures, fmt.Errorf("read DB snapshot %s: %w", id, err)) continue } - if !found || rec.Status != SnapshotStatusCreating || - time.Since(rec.CreatedAt) <= snapshotResolveTimeout { + if !found || rec.Status != SnapshotStatusCreating { + continue + } + // Not due yet, so ask to be back when it is rather than leaving a dead + // worker's record to the resync. + if due := time.Until(rec.CreatedAt.Add(snapshotResolveTimeout)); due > 0 { + revisit = reconciler.Earliest(revisit, due) continue } if err := r.resolveCreatingSnapshot(ctx, kv, rev, accountID, &rec); err != nil { failures = append(failures, fmt.Errorf("resolve DB snapshot %s: %w", id, err)) } } - return errors.Join(failures...) + return revisit, errors.Join(failures...) } // Either adopts the EC2 snapshot the dead worker cut, or withdraws the record so diff --git a/spinifex/handlers/rds/reconciler_test.go b/spinifex/handlers/rds/reconciler_test.go index c3b3436a3..1445fb39c 100644 --- a/spinifex/handlers/rds/reconciler_test.go +++ b/spinifex/handlers/rds/reconciler_test.go @@ -15,6 +15,15 @@ import ( "github.com/stretchr/testify/require" ) +// onePass runs one reconcile and keeps only its error, for the tests that assert +// on what converged rather than on when the loop should next run. The tests that +// do care about the deadline call reconcileOnce directly. +func onePass(t *testing.T, r *Reconciler) error { + t.Helper() + _, err := r.reconcileOnce(t.Context()) + return err +} + // fakeInstanceState answers the VM half of the bootstrap check. type fakeInstanceState struct { state string @@ -96,7 +105,7 @@ func TestReconciler_MarksAvailableOnHealthyHeartbeatFromARunningVM(t *testing.T) rec.FormatAuthorized = true seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) status, reason := h.statusOf(t, testDBID) assert.Equal(t, StatusAvailable, status) @@ -126,7 +135,7 @@ func TestReconciler_HoldsCreatingWhenTheVMIsNotRunning(t *testing.T) { h.state.state = "pending" seedInstance(t, h.svc, healthyRecord()) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) status, _ := h.statusOf(t, testDBID) assert.Equal(t, StatusCreating, status) @@ -158,7 +167,7 @@ func TestReconciler_HoldsCreatingUntilTheEngineIsHealthy(t *testing.T) { tc.mutate(&rec) seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) status, _ := h.statusOf(t, testDBID) assert.Equal(t, StatusCreating, status) @@ -181,7 +190,7 @@ func TestReconciler_CompletesACreateOnAPersistedHeartbeatInsideTheFloor(t *testi rec.Agent.LastSeen = &persisted seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) status, _ := h.statusOf(t, testDBID) assert.Equal(t, StatusAvailable, status) @@ -200,7 +209,7 @@ func TestReconciler_MarksFailedWhenTheBootstrapTimesOut(t *testing.T) { rec.CreatedAt = time.Now().UTC().Add(-10 * time.Minute) seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) status, reason := h.statusOf(t, testDBID) assert.Equal(t, StatusFailed, status) @@ -218,7 +227,7 @@ func TestReconciler_LeavesASlowBootstrapAloneInsideTheWindow(t *testing.T) { rec.CreatedAt = time.Now().UTC().Add(-10 * time.Minute) seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) status, _ := h.statusOf(t, testDBID) assert.Equal(t, StatusCreating, status) @@ -235,7 +244,7 @@ func TestReconciler_LeavesSettledInstancesAlone(t *testing.T) { rec.CreatedAt = time.Now().UTC().Add(-24 * time.Hour) seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) got, _ := h.statusOf(t, testDBID) assert.Equal(t, StatusStopped, got) @@ -250,7 +259,7 @@ func TestReconciler_SurfacesAVMStateLookupFailure(t *testing.T) { h.state.err = errors.New("no node answered the describe") seedInstance(t, h.svc, healthyRecord()) - err := h.rec.reconcileOnce(t.Context()) + err := onePass(t, h.rec) require.Error(t, err) assert.Contains(t, err.Error(), "no node answered the describe") @@ -265,7 +274,7 @@ func TestReconciler_FallsBackToTheHeartbeatWhenVMStateIsUnwired(t *testing.T) { h := newReconcileHarness(t, func(d *Deps) { d.InstanceState = nil }) seedInstance(t, h.svc, healthyRecord()) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) status, _ := h.statusOf(t, testDBID) assert.Equal(t, StatusAvailable, status) @@ -304,7 +313,7 @@ func TestReconciler_MovesAnExistingSystemNICOntoTheRDSSystemGroup(t *testing.T) rec.SystemENIID = "eni-sys01" seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) require.Len(t, enis.modified, 1) assert.Equal(t, "eni-sys01", aws.StringValue(enis.modified[0].NetworkInterfaceId)) @@ -316,7 +325,7 @@ func TestReconciler_MovesAnExistingSystemNICOntoTheRDSSystemGroup(t *testing.T) status, _ := h.statusOf(t, testDBID) assert.Equal(t, StatusCreating, status) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) assert.Len(t, enis.modified, 1, "once the group is recorded the sweep is a string compare") status, _ = h.statusOf(t, testDBID) assert.Equal(t, StatusAvailable, status) @@ -332,8 +341,8 @@ func TestReconciler_EnsuresTheSystemGroupOncePerProcess(t *testing.T) { rec.SystemENIID = "eni-sys01" seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) + require.NoError(t, onePass(t, h.rec)) assert.Len(t, enis.sgDescribed, 1) assert.Len(t, enis.sgCreated, 1) @@ -349,13 +358,13 @@ func TestReconciler_RetriesTheSystemNICMoveAfterAFailure(t *testing.T) { rec.SystemENIID = "eni-sys01" seedInstance(t, h.svc, rec) - err := h.rec.reconcileOnce(t.Context()) + err := onePass(t, h.rec) require.Error(t, err) assert.Contains(t, err.Error(), "vpcd is not answering") assert.Empty(t, h.systemSGOf(t, testDBID)) enis.modifyErr = nil - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) assert.Len(t, enis.modified, 1) assert.Equal(t, testSystemSG, h.systemSGOf(t, testDBID)) } diff --git a/spinifex/handlers/rds/recovery_test.go b/spinifex/handlers/rds/recovery_test.go index 8a67d1bec..0cdc3a035 100644 --- a/spinifex/handlers/rds/recovery_test.go +++ b/spinifex/handlers/rds/recovery_test.go @@ -96,7 +96,7 @@ func TestReconciler_MarksADarkInstanceFailed(t *testing.T) { h.state.state = "stopped" seedInstance(t, h.svc, darkRecord()) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) rec := h.recordOf(t, testDBID) assert.Equal(t, StatusFailed, rec.Status) @@ -114,7 +114,7 @@ func TestReconciler_RecordsAnEventForADarkInstance(t *testing.T) { h.state.state = "stopped" seedInstance(t, h.svc, darkRecord()) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) out, err := h.svc.DescribeEvents(t.Context(), &rds.DescribeEventsInput{ SourceType: aws.String(EventSourceTypeDBInstance), @@ -134,7 +134,7 @@ func TestReconciler_LeavesAHealthyInstanceUntouchedWithoutAVMLookup(t *testing.T h := newReconcileHarness(t) seedInstance(t, h.svc, servingRecord()) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) rec := h.recordOf(t, testDBID) assert.Equal(t, StatusAvailable, rec.Status) @@ -171,7 +171,7 @@ func TestReconciler_HoldsAvailableOnHalfTheEvidence(t *testing.T) { tc.mutate(&rec) seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) stored := h.recordOf(t, testDBID) assert.Equal(t, StatusAvailable, stored.Status) @@ -189,7 +189,7 @@ func TestReconciler_StartsTheClockBeforeFailingWithinTheGrace(t *testing.T) { h.state.state = "stopped" seedInstance(t, h.svc, darkRecord()) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) rec := h.recordOf(t, testDBID) assert.Equal(t, StatusAvailable, rec.Status) @@ -210,7 +210,7 @@ func TestReconciler_FailsOnAClockStampedByAnEarlierLeader(t *testing.T) { // A different node holding the lease, with no beats of its own in memory. other := NewReconciler(h.svc, "node-b") - require.NoError(t, other.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, other)) stored := h.recordOf(t, testDBID) assert.Equal(t, StatusFailed, stored.Status) @@ -233,7 +233,7 @@ func TestReconciler_TheClockResetsOnlyOnAHealthyHeartbeat(t *testing.T) { // The VM is back but the agent still is not: not a failure, and not a reason // to forget that the instance has been dark for half an hour. h.state.state = instanceStateRunning - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) stored := h.recordOf(t, testDBID) require.NotNil(t, stored.UnhealthySince) assert.Equal(t, stamped, stored.UnhealthySince.UTC()) @@ -245,7 +245,7 @@ func TestReconciler_TheClockResetsOnlyOnAHealthyHeartbeat(t *testing.T) { recovered.Agent.LastSeen = &now return recovered }()) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) stored = h.recordOf(t, testDBID) assert.Equal(t, StatusAvailable, stored.Status) @@ -265,7 +265,7 @@ func TestReconciler_ClearsFailedOnARecoveredHeartbeat(t *testing.T) { rec.UnhealthySince = &stamped seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) stored := h.recordOf(t, testDBID) assert.Equal(t, StatusAvailable, stored.Status) @@ -293,7 +293,7 @@ func TestReconciler_LeavesAStillDarkFailedInstanceAlone(t *testing.T) { rec.FailureReason = "the original reason" seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) stored := h.recordOf(t, testDBID) assert.Equal(t, StatusFailed, stored.Status) @@ -316,7 +316,7 @@ func TestReconciler_AllowsThePersistFloorOnAPersistedHeartbeat(t *testing.T) { rec.Agent.LastSeen = &persisted seedInstance(t, h.svc, rec) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) stored := h.recordOf(t, testDBID) assert.Equal(t, StatusAvailable, stored.Status) @@ -338,7 +338,7 @@ func TestReconciler_UsesTheTightBoundOnABeatThisNodeHandled(t *testing.T) { seedInstance(t, h.svc, rec) h.svc.liveness[testAccountID+"/"+testDBID] = &agentLiveness{lastSeen: beat, health: EngineHealthHealthy} - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) stored := h.recordOf(t, testDBID) assert.Equal(t, StatusFailed, stored.Status) @@ -356,7 +356,7 @@ func TestReconciler_NeverFailsAnInstanceWhenVMStateIsUnwired(t *testing.T) { }) seedInstance(t, h.svc, darkRecord()) - require.NoError(t, h.rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, h.rec)) stored := h.recordOf(t, testDBID) assert.Equal(t, StatusAvailable, stored.Status) @@ -371,7 +371,7 @@ func TestReconciler_SurfacesAVMLookupFailureWithoutFailingTheInstance(t *testing h.state.err = errors.New("no node answered the describe") seedInstance(t, h.svc, darkRecord()) - err := h.rec.reconcileOnce(t.Context()) + err := onePass(t, h.rec) require.Error(t, err) assert.Contains(t, err.Error(), "no node answered the describe") diff --git a/spinifex/handlers/rds/revisit_test.go b/spinifex/handlers/rds/revisit_test.go new file mode 100644 index 000000000..92ce4d271 --- /dev/null +++ b/spinifex/handlers/rds/revisit_test.go @@ -0,0 +1,182 @@ +package handlers_rds + +//test:in-package — asserts on the deadline reconcileOnce returns, which is +// unexported and is the whole subject of these tests. + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// settledHeartbeatBound is how long after its last beat a settled instance is +// still believed. It is what a settled record's deadline is measured against. +const settledHeartbeatBound = HeartbeatStaleAfter + HeartbeatPersistFloor + +// A transitional instance is waiting on an EC2 lifecycle state that no KV write +// announces, so there is no update for a watch to deliver and the pass has to +// ask to come back. +func TestReconcileOnce_ATransitionalInstanceAsksForTheShortInterval(t *testing.T) { + t.Parallel() + h := newReconcileHarness(t) + h.state.state = "pending" + seedInstance(t, h.svc, healthyRecord()) + + revisit, err := h.rec.reconcileOnce(t.Context()) + require.NoError(t, err) + + status, _ := h.statusOf(t, testDBID) + require.Equal(t, StatusCreating, status, "the VM is not running, so this pass must not settle it") + assert.Equal(t, reconcileInterval, revisit) +} + +// A settled instance is waiting for its agent to fall silent, and silence writes +// nothing. The deadline is the instant the beat expires, which is sharper than +// the ticker that used to notice it up to one interval late. +func TestReconcileOnce_ASettledInstanceAsksForItsHeartbeatDeadline(t *testing.T) { + t.Parallel() + h := newReconcileHarness(t) + rec := healthyRecord() + rec.Status = StatusAvailable + seedInstance(t, h.svc, rec) + + revisit, err := h.rec.reconcileOnce(t.Context()) + require.NoError(t, err) + + status, _ := h.statusOf(t, testDBID) + require.Equal(t, StatusAvailable, status) + assert.Greater(t, revisit, settledHeartbeatBound-time.Minute, + "a beat seconds old must not be treated as nearly stale") + assert.LessOrEqual(t, revisit, settledHeartbeatBound) +} + +// The window between a beat expiring and the instance being called failed is +// where the whole detector lives, and nothing writes to KV during it. Returning +// no deadline here left the pass that does the failing with nothing to schedule +// it, so a dead database waited for the resync — slower than the ticker it +// replaced, and the exact regression this design exists to avoid. +func TestReconcileOnce_ADarkInstanceAsksForItsFailureGrace(t *testing.T) { + t.Parallel() + h := newReconcileHarness(t) + rec := healthyRecord() + rec.Status = StatusAvailable + stale := time.Now().UTC().Add(-2 * settledHeartbeatBound) + rec.Agent.LastSeen = &stale + seedInstance(t, h.svc, rec) + + revisit, err := h.rec.reconcileOnce(t.Context()) + require.NoError(t, err) + + assert.Positive(t, revisit, "an expired beat must not read as nothing left to wait for") + assert.LessOrEqual(t, revisit, h.svc.failureGrace()) +} + +// Partway through the grace window the deadline is what remains of it, not the +// whole window again — otherwise each pass would push the failure further out. +func TestReconcileOnce_AFailureClockAlreadyRunningAsksForWhatIsLeftOfIt(t *testing.T) { + t.Parallel() + h := newReconcileHarness(t) + rec := healthyRecord() + rec.Status = StatusAvailable + stale := time.Now().UTC().Add(-2 * settledHeartbeatBound) + rec.Agent.LastSeen = &stale + started := time.Now().UTC().Add(-h.svc.failureGrace() / 2) + rec.UnhealthySince = &started + seedInstance(t, h.svc, rec) + + revisit, err := h.rec.reconcileOnce(t.Context()) + require.NoError(t, err) + + assert.Positive(t, revisit) + assert.LessOrEqual(t, revisit, h.svc.failureGrace()/2) +} + +// A failed instance has no clock left to run. Recovery is a heartbeat, which is +// a KV write the watch already delivers. +func TestReconcileOnce_AFailedInstanceAsksForNoDeadline(t *testing.T) { + t.Parallel() + h := newReconcileHarness(t) + rec := healthyRecord() + rec.Status = StatusFailed + stale := time.Now().UTC().Add(-2 * settledHeartbeatBound) + rec.Agent.LastSeen = &stale + seedInstance(t, h.svc, rec) + + revisit, err := h.rec.reconcileOnce(t.Context()) + require.NoError(t, err) + + assert.Zero(t, revisit) +} + +// The earliest across the fleet is the one that matters: one instance mid-create +// sets the pace for the pass regardless of how many settled ones sit beside it. +func TestReconcileOnce_TheEarliestDeadlineAcrossTheFleetWins(t *testing.T) { + t.Parallel() + h := newReconcileHarness(t) + h.state.state = "pending" + + settled := healthyRecord() + settled.Status = StatusAvailable + seedInstance(t, h.svc, settled) + + transitional := healthyRecord() + transitional.DBInstanceIdentifier = "db-still-creating" + seedInstance(t, h.svc, transitional) + + revisit, err := h.rec.reconcileOnce(t.Context()) + require.NoError(t, err) + + assert.Equal(t, reconcileInterval, revisit) +} + +// A region with nothing in it stops asking altogether, which is the case the +// whole change is for: the loop falls back to the resync and nothing else. +func TestReconcileOnce_NothingToDoAsksForNoDeadline(t *testing.T) { + t.Parallel() + h := newReconcileHarness(t) + + revisit, err := h.rec.reconcileOnce(t.Context()) + require.NoError(t, err) + + assert.Zero(t, revisit) +} + +// A snapshot record left in creating is settled against EC2 once its resolve +// timeout elapses. Until then the pass asks to be back at that instant rather +// than leaving a dead worker's record to the resync. +func TestReconcileOnce_ACreatingSnapshotAsksForItsResolveDeadline(t *testing.T) { + t.Parallel() + h := newReconcileHarness(t) + kv, err := h.svc.bucket(t.Context(), testAccountID) + require.NoError(t, err) + creating := DBSnapshotRecord{ + DBSnapshotIdentifier: testSnapshotID, + AccountID: testAccountID, + Status: SnapshotStatusCreating, + CreatedAt: time.Now().UTC().Add(-snapshotResolveTimeout + 2*time.Minute), + } + require.NoError(t, putJSON(t.Context(), kv, DBSnapshotKey(testSnapshotID), &creating)) + + revisit, err := h.rec.reconcileOnce(t.Context()) + require.NoError(t, err) + + assert.Greater(t, revisit, time.Minute, "the resolve timeout has two minutes left to run") + assert.LessOrEqual(t, revisit, 2*time.Minute) +} + +// A follower does no work but cannot be left waiting on a watch either: the +// lease turning over is not a KV write it sees. +func TestReconcilePass_AFollowerAsksAgainAtTheLeaseRate(t *testing.T) { + t.Parallel() + h := newReconcileHarness(t) + seedInstance(t, h.svc, healthyRecord()) + + revisit, err := h.rec.reconcilePass(t.Context()) + require.NoError(t, err) + + status, _ := h.statusOf(t, testDBID) + assert.Equal(t, StatusCreating, status, "a follower must not reconcile") + assert.Equal(t, leaseRefresh, revisit) +} diff --git a/spinifex/handlers/rds/snapshot_test.go b/spinifex/handlers/rds/snapshot_test.go index f91731a2f..893d4713a 100644 --- a/spinifex/handlers/rds/snapshot_test.go +++ b/spinifex/handlers/rds/snapshot_test.go @@ -565,7 +565,7 @@ func TestReconciler_AdoptsTheEC2SnapshotOfAnUnfinishedDBSnapshot(t *testing.T) { rdsSnapshotAccountTagKey: testAccountID, }} - require.NoError(t, rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, rec)) stored, found := h.snapshot(t, testSnapshotID) require.True(t, found) @@ -603,7 +603,7 @@ func TestReconciler_DoesNotAdoptAnotherAccountsSnapshot(t *testing.T) { }, } - require.NoError(t, reconciler.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, reconciler)) stored, found := h.snapshot(t, testSnapshotID) require.True(t, found) @@ -627,7 +627,7 @@ func TestReconciler_KeepsCreatingSnapshotWhenEC2LookupIsIncomplete(t *testing.T) } require.NoError(t, putJSON(t.Context(), kv, DBSnapshotKey(testSnapshotID), &stale)) - err = reconciler.reconcileOnce(t.Context()) + err = onePass(t, reconciler) require.Error(t, err) assert.Contains(t, err.Error(), "metadata temporarily unavailable") @@ -661,7 +661,7 @@ func TestReconciler_DoesNotDeleteAConcurrentlyCompletedSnapshot(t *testing.T) { require.NoError(t, updateJSON(t.Context(), kv, DBSnapshotKey(testSnapshotID), rev, ¤t)) } - require.NoError(t, reconciler.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, reconciler)) stored, found := h.snapshot(t, testSnapshotID) require.True(t, found) @@ -693,7 +693,7 @@ func TestReconciler_RejectsAmbiguousEC2Snapshots(t *testing.T) { "snap-duplicate-b": matchingTags, } - err = reconciler.reconcileOnce(t.Context()) + err = onePass(t, reconciler) require.Error(t, err) assert.Contains(t, err.Error(), "multiple EC2 snapshots") @@ -719,7 +719,7 @@ func TestReconciler_WithdrawsADBSnapshotWhoseDataWasNeverCut(t *testing.T) { } require.NoError(t, putJSON(t.Context(), kv, DBSnapshotKey(testSnapshotID), &stale)) - require.NoError(t, rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, rec)) _, found := h.snapshot(t, testSnapshotID) assert.False(t, found) @@ -744,7 +744,7 @@ func TestReconciler_LeavesAFreshCreatingSnapshotAlone(t *testing.T) { } require.NoError(t, putJSON(t.Context(), kv, DBSnapshotKey(testSnapshotID), &fresh)) - require.NoError(t, rec.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, rec)) stored, found := h.snapshot(t, testSnapshotID) require.True(t, found) @@ -770,7 +770,7 @@ func TestReconciler_ReturnsAnInstanceStuckInBackingUp(t *testing.T) { } seedInstance(t, h.svc, rec) - require.NoError(t, reconciler.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, reconciler)) stored := h.instance(t, testDBID) assert.Equal(t, resume, stored.Status) @@ -797,7 +797,7 @@ func TestReconciler_LeavesAnInFlightSnapshotAlone(t *testing.T) { } seedInstance(t, h.svc, rec) - require.NoError(t, reconciler.reconcileOnce(t.Context())) + require.NoError(t, onePass(t, reconciler)) stored := h.instance(t, testDBID) assert.Equal(t, StatusBackingUp, stored.Status) diff --git a/spinifex/reconciler/queue_test.go b/spinifex/reconciler/queue_test.go index 74101a5e1..b33cdf68f 100644 --- a/spinifex/reconciler/queue_test.go +++ b/spinifex/reconciler/queue_test.go @@ -68,8 +68,8 @@ func TestRun_AnUpdateReconcilesTheKeyThatChanged(t *testing.T) { run(t, reconciler.Config{ Name: "one", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, - ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, + ReconcileKey: func(_ context.Context, key string) (time.Duration, error) { keys.record(key); return 0, nil }, Resync: time.Hour, }) p.waitFor(t, 1, "the startup pass") @@ -90,8 +90,8 @@ func TestRun_ABurstReconcilesEachKeyOnce(t *testing.T) { run(t, reconciler.Config{ Name: "burst", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, - ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, + ReconcileKey: func(_ context.Context, key string) (time.Duration, error) { keys.record(key); return 0, nil }, Resync: time.Hour, Debounce: 200 * time.Millisecond, }) @@ -117,8 +117,8 @@ func TestRun_RepeatedWritesToOneKeyReconcileItOnce(t *testing.T) { run(t, reconciler.Config{ Name: "dedup", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, - ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, + ReconcileKey: func(_ context.Context, key string) (time.Duration, error) { keys.record(key); return 0, nil }, Resync: time.Hour, Debounce: 200 * time.Millisecond, }) @@ -143,8 +143,8 @@ func TestRun_AChangeDuringAKeysReconcileIsNotLost(t *testing.T) { run(t, reconciler.Config{ Name: "inflight", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, - ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, + ReconcileKey: func(_ context.Context, key string) (time.Duration, error) { keys.record(key); return 0, nil }, Resync: time.Hour, Debounce: 50 * time.Millisecond, }) @@ -170,8 +170,8 @@ func TestRun_KeyForMapsUpdatesOntoTheCallersUnitOfWork(t *testing.T) { run(t, reconciler.Config{ Name: "keyfor", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, - ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, + ReconcileKey: func(_ context.Context, key string) (time.Duration, error) { keys.record(key); return 0, nil }, KeyFor: func(entry jetstream.KeyValueEntry) ([]string, bool) { return []string{"account-1"}, true }, @@ -199,8 +199,8 @@ func TestRun_AnUnattributableUpdateFallsBackToTheWholeSet(t *testing.T) { run(t, reconciler.Config{ Name: "fallback", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, - ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, + ReconcileKey: func(_ context.Context, key string) (time.Duration, error) { keys.record(key); return 0, nil }, KeyFor: func(entry jetstream.KeyValueEntry) ([]string, bool) { if entry.Operation() == jetstream.KeyValuePut { return []string{entry.Key()}, true @@ -230,8 +230,8 @@ func TestRun_TheResyncStillRunsTheWholeSet(t *testing.T) { run(t, reconciler.Config{ Name: "resync", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, - ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, + ReconcileKey: func(_ context.Context, key string) (time.Duration, error) { keys.record(key); return 0, nil }, Resync: 150 * time.Millisecond, }) @@ -246,10 +246,10 @@ func TestRun_AFailedKeyDoesNotStopTheLoop(t *testing.T) { run(t, reconciler.Config{ Name: "failure", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, - ReconcileKey: func(_ context.Context, key string) error { + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, + ReconcileKey: func(_ context.Context, key string) (time.Duration, error) { keys.record(key) - return assert.AnError + return 0, assert.AnError }, Resync: time.Hour, Debounce: 50 * time.Millisecond, diff --git a/spinifex/reconciler/reconciler.go b/spinifex/reconciler/reconciler.go index 81bc464ce..e11e59b5e 100644 --- a/spinifex/reconciler/reconciler.go +++ b/spinifex/reconciler/reconciler.go @@ -7,6 +7,11 @@ // the key that changed is carried through to it. The whole-set Reconcile stays // required either way: it is the startup pass and the resync backstop, so a key // the queue never learns about still converges. +// +// A pass may also name a deadline to be revisited by. That is for the loops +// waiting on something no KV write announces — a VM reaching running, an agent +// falling silent, a timeout expiring — which a watch cannot see at all, and for +// which the resync alone is too blunt to be a failure detector. package reconciler import ( @@ -56,7 +61,14 @@ type Config struct { // and the resync backstop. A per-key queue only converges if something // behind it covers what the queue never learned: a key deleted while the // watch was down is enqueued by nothing, since nobody observes an absence. - Reconcile func(ctx context.Context) error + // + // The duration it returns is a revisit deadline: run me again within this, + // even if nothing changes. Zero means no deadline, which is the right answer + // for a loop that is purely a function of the buckets it watches. It is for + // the loops that are not — the ones waiting on a VM to reach running, an + // endpoint to answer, or an agent to fall silent, none of which write to KV + // and so none of which a watch can see. + Reconcile func(ctx context.Context) (revisit time.Duration, err error) // ReconcileKey handles one changed key, for a caller whose work is per // resource rather than a recompute of the whole set. Unset means every // change runs Reconcile, which is the whole-set behaviour. @@ -64,7 +76,11 @@ type Config struct { // It is called once per distinct key in a burst, so a caller whose unit of // work is coarser than a key — an account rather than an instance — maps // updates onto that unit with KeyFor rather than deduplicating here. - ReconcileKey func(ctx context.Context, key string) error + // + // Its revisit deadline means the same thing as Reconcile's, and is honoured + // by running a whole-set pass: one key's view cannot say what the others + // need, and only a whole-set pass can re-derive the deadline for all of them. + ReconcileKey func(ctx context.Context, key string) (revisit time.Duration, err error) // KeyFor maps a watch update onto the keys it dirties. Nil means the key // that changed is the key to reconcile. // @@ -157,11 +173,19 @@ func Run(ctx context.Context, cfg Config) { // the two would otherwise be invisible until the next resync, because the // pass that would have seen it ran before the watch existed. w.resync(ctx) - pass(ctx, cfg, "startup") resync := time.NewTicker(cfg.Resync) defer resync.Stop() + // Fires when a pass asked to be revisited sooner than the resync. Created + // already expired and drained, so a loop whose passes never ask keeps the + // resync as its only timer. + revisit := time.NewTimer(0) + <-revisit.C + defer revisit.Stop() + + rearm(revisit, clampRevisit(pass(ctx, cfg, "startup"), cfg.Resync)) + for { select { case <-ctx.Done(): @@ -174,35 +198,81 @@ func Run(ctx context.Context, cfg Config) { // Whatever is queued is covered by the whole-set pass about to run, // so it is dropped rather than reconciled again after it. queue.take() - pass(ctx, cfg, "resync") + rearm(revisit, clampRevisit(pass(ctx, cfg, "resync"), cfg.Resync)) + case <-revisit.C: + // A deadline is always served by a whole-set pass: the caller asked + // to be revisited without anything having changed, so there is no + // key to name, and only a whole-set pass can set the next deadline. + queue.take() + rearm(revisit, clampRevisit(pass(ctx, cfg, "deadline"), cfg.Resync)) case <-changes: if settle(ctx, changes, cfg.Debounce) { - serve(ctx, cfg, queue) + rearm(revisit, clampRevisit(serve(ctx, cfg, queue), cfg.Resync)) } } } } +// clampRevisit bounds a deadline to the resync. A caller may ask to be revisited +// sooner but never later: the resync is the outer bound on staleness, and is not +// something a pass gets to extend. Non-positive means no deadline. +func clampRevisit(d, resync time.Duration) time.Duration { + if d <= 0 { + return 0 + } + return min(d, resync) +} + +// rearm resets a timer that may or may not have fired, and leaves it stopped for +// a zero duration. Draining is conditional because the loop consumes the tick +// itself on the path that a deadline pass runs from. +func rearm(t *time.Timer, d time.Duration) { + if !t.Stop() { + select { + case <-t.C: + default: + } + } + if d > 0 { + t.Reset(d) + } +} + +// Earliest is the soonest of two revisit deadlines, treating a non-positive one +// as "no deadline" rather than as "immediately". Exported because a whole-set +// pass has to fold one of these per resource before it can return just one. +func Earliest(a, b time.Duration) time.Duration { + switch { + case a <= 0: + return max(b, 0) + case b <= 0: + return a + default: + return min(a, b) + } +} + // serve runs the work a settled burst produced: one whole-set pass, or one pass -// per distinct key that changed. -func serve(ctx context.Context, cfg Config, queue *workQueue) { +// per distinct key that changed. It reports the soonest deadline those passes +// asked for. +func serve(ctx context.Context, cfg Config, queue *workQueue) time.Duration { if cfg.ReconcileKey == nil { - pass(ctx, cfg, "change") - return + return pass(ctx, cfg, "change") } whole, keys := queue.take() if whole { // A whole-set pass covers every key that was waiting alongside it, so // the keys it returned are deliberately not reconciled again. - pass(ctx, cfg, "change") - return + return pass(ctx, cfg, "change") } + var revisit time.Duration for _, key := range keys { if ctx.Err() != nil { - return + return revisit } - keyPass(ctx, cfg, key) + revisit = Earliest(revisit, keyPass(ctx, cfg, key)) } + return revisit } // settle waits for a burst to stop arriving, so one multi-key write produces @@ -232,30 +302,38 @@ func settle(ctx context.Context, changes <-chan struct{}, debounce time.Duration } // pass runs one reconcile, logging rather than returning a failure: the loop -// outlives any single pass, and the next resync retries. -func pass(ctx context.Context, cfg Config, trigger string) { +// outlives any single pass, and the next resync retries. A failed pass keeps +// its deadline, so a caller partway through a transition still gets revisited +// on its own schedule rather than dropping back to the resync. +func pass(ctx context.Context, cfg Config, trigger string) time.Duration { start := time.Now() - if err := cfg.Reconcile(ctx); err != nil { + revisit, err := cfg.Reconcile(ctx) + if err != nil { slog.WarnContext(ctx, "reconcile pass failed, retrying next cycle", "loop", cfg.Name, "trigger", trigger, "duration_ms", otelsetup.Millis(time.Since(start)), "err", err) - return + return revisit } slog.DebugContext(ctx, "reconcile pass complete", - "loop", cfg.Name, "trigger", trigger, "duration_ms", otelsetup.Millis(time.Since(start))) + "loop", cfg.Name, "trigger", trigger, "duration_ms", otelsetup.Millis(time.Since(start)), + "revisit_ms", otelsetup.Millis(revisit)) + return revisit } // keyPass runs one per-key reconcile. Its failure is logged rather than // retried here: the resync re-reconciles the whole set, so a key that failed is // covered by the same backstop as a key the queue never saw. -func keyPass(ctx context.Context, cfg Config, key string) { +func keyPass(ctx context.Context, cfg Config, key string) time.Duration { start := time.Now() - if err := cfg.ReconcileKey(ctx, key); err != nil { + revisit, err := cfg.ReconcileKey(ctx, key) + if err != nil { slog.WarnContext(ctx, "reconcile key failed, retrying on the next resync", "loop", cfg.Name, "key", key, "duration_ms", otelsetup.Millis(time.Since(start)), "err", err) - return + return revisit } slog.DebugContext(ctx, "reconcile key complete", - "loop", cfg.Name, "key", key, "duration_ms", otelsetup.Millis(time.Since(start))) + "loop", cfg.Name, "key", key, "duration_ms", otelsetup.Millis(time.Since(start)), + "revisit_ms", otelsetup.Millis(revisit)) + return revisit } // watchSet holds one watcher per bucket currently being watched, keyed by diff --git a/spinifex/reconciler/reconciler_test.go b/spinifex/reconciler/reconciler_test.go index b286365a1..3aedf1598 100644 --- a/spinifex/reconciler/reconciler_test.go +++ b/spinifex/reconciler/reconciler_test.go @@ -108,7 +108,7 @@ func TestRun_ReconcilesOnceAtStartup(t *testing.T) { run(t, reconciler.Config{ Name: "startup", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, Resync: time.Hour, }) @@ -123,7 +123,7 @@ func TestRun_AnUpdateTriggersAPass(t *testing.T) { run(t, reconciler.Config{ Name: "update", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, Resync: time.Hour, }) p.waitFor(t, 1, "the startup pass") @@ -142,7 +142,7 @@ func TestRun_ABurstCollapsesIntoOnePass(t *testing.T) { run(t, reconciler.Config{ Name: "burst", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, Resync: time.Hour, Debounce: 200 * time.Millisecond, }) @@ -163,7 +163,7 @@ func TestRun_ResyncFiresWithNoWatchTraffic(t *testing.T) { run(t, reconciler.Config{ Name: "resync", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, Resync: 150 * time.Millisecond, }) @@ -177,7 +177,7 @@ func TestRun_NoSourcesStillReconcilesOnTheResync(t *testing.T) { run(t, reconciler.Config{ Name: "no-sources", - Reconcile: func(context.Context) error { p.record(); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, Resync: 150 * time.Millisecond, }) @@ -194,12 +194,12 @@ func TestRun_AFailedPassDoesNotStopTheLoop(t *testing.T) { run(t, reconciler.Config{ Name: "failure", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, - Reconcile: func(context.Context) error { + Reconcile: func(context.Context) (time.Duration, error) { p.record() if calls.Add(1) == 1 { - return assert.AnError + return 0, assert.AnError } - return nil + return 0, nil }, Resync: 150 * time.Millisecond, }) @@ -217,7 +217,7 @@ func TestRun_AFilterExcludesOtherKeys(t *testing.T) { run(t, reconciler.Config{ Name: "filter", Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, "lb.*")}, - Reconcile: func(context.Context) error { p.record(); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, Resync: time.Hour, }) p.waitFor(t, 1, "the startup pass") @@ -264,7 +264,7 @@ func TestRun_ABucketAppearingAfterStartupIsPickedUp(t *testing.T) { run(t, reconciler.Config{ Name: "dynamic", Sources: []reconciler.Source{reconciler.Dynamic(set.list, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, Resync: 200 * time.Millisecond, }) p.waitFor(t, 1, "the startup pass") @@ -295,7 +295,7 @@ func TestRun_AnUnwatchableSourceStillResyncs(t *testing.T) { run(t, reconciler.Config{ Name: "unwatchable", Sources: []reconciler.Source{reconciler.Fixed(unwatchable, ">")}, - Reconcile: func(context.Context) error { p.record(); return nil }, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, Resync: 150 * time.Millisecond, }) diff --git a/spinifex/reconciler/revisit_test.go b/spinifex/reconciler/revisit_test.go new file mode 100644 index 000000000..b806d9728 --- /dev/null +++ b/spinifex/reconciler/revisit_test.go @@ -0,0 +1,129 @@ +package reconciler_test + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/reconciler" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestRun_ARevisitDeadlineFiresWithNoWatchTraffic is the reason the deadline +// exists: a loop waiting on something outside KV gets no watch update at all, +// so without this it would run only on the resync. +func TestRun_ARevisitDeadlineFiresWithNoWatchTraffic(t *testing.T) { + store, _ := newBucket(t, "revisit-deadline") + p := newPasses() + + run(t, reconciler.Config{ + Name: "deadline", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 80 * time.Millisecond, nil }, + Resync: time.Hour, + }) + + p.waitFor(t, 4, "repeated deadline passes against a bucket nobody is writing to") +} + +// TestRun_ARevisitDeadlineIsClampedToTheResync holds the one direction a caller +// may not move: the resync is the outer bound on staleness, so a longer deadline +// is ignored rather than allowed to defer past it. +func TestRun_ARevisitDeadlineIsClampedToTheResync(t *testing.T) { + store, _ := newBucket(t, "revisit-clamped") + p := newPasses() + + run(t, reconciler.Config{ + Name: "clamped", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return time.Hour, nil }, + Resync: 150 * time.Millisecond, + }) + + p.waitFor(t, 4, "resync passes despite a pass asking for an hour") +} + +// TestRun_DroppingTheDeadlineStopsTheEarlyPasses covers a loop whose work +// finishes: once the last transitional resource settles it stops asking, and the +// loop must fall back to the resync rather than keep the old deadline armed. +func TestRun_DroppingTheDeadlineStopsTheEarlyPasses(t *testing.T) { + store, _ := newBucket(t, "revisit-dropped") + p := newPasses() + var calls atomic.Int64 + + run(t, reconciler.Config{ + Name: "dropped", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) (time.Duration, error) { + p.record() + if calls.Add(1) == 1 { + return 50 * time.Millisecond, nil + } + return 0, nil + }, + Resync: time.Hour, + }) + + p.waitFor(t, 2, "the pass the first deadline asked for") + p.stillAt(t, 2, "a pass that asks for no deadline must not be revisited") +} + +// TestRun_AFailedPassKeepsItsRevisitDeadline stops a failure demoting a loop to +// the resync. A pass that fails partway through a transition is exactly the one +// that most needs revisiting on its own schedule. +func TestRun_AFailedPassKeepsItsRevisitDeadline(t *testing.T) { + store, _ := newBucket(t, "revisit-failed") + p := newPasses() + + run(t, reconciler.Config{ + Name: "failed", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) (time.Duration, error) { + p.record() + return 80 * time.Millisecond, assert.AnError + }, + Resync: time.Hour, + }) + + p.waitFor(t, 4, "deadline passes after each one failed") +} + +// TestRun_AChangeBeforeTheDeadlineStillTriggersAPass keeps the deadline additive. +// The wait is capped well below the deadline asked for, so a pass arriving at all +// proves the change woke the loop rather than the timer. +func TestRun_AChangeBeforeTheDeadlineStillTriggersAPass(t *testing.T) { + store, _ := newBucket(t, "revisit-change-wins") + p := newPasses() + + run(t, reconciler.Config{ + Name: "change-wins", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return time.Minute, nil }, + Resync: time.Hour, + }) + p.waitFor(t, 1, "the startup pass") + + require.NoError(t, store.Set(t.Context(), "one", &record{Name: "one"})) + p.waitFor(t, 2, "the pass triggered by the write, long before the deadline") +} + +// TestRun_ADeadlinePassRunsTheWholeSet pins which half of a per-key loop a +// deadline serves. Nothing changed, so there is no key to name, and only a +// whole-set pass can work out what the next deadline should be. +func TestRun_ADeadlinePassRunsTheWholeSet(t *testing.T) { + store, _ := newBucket(t, "revisit-whole-set") + p, keys := newPasses(), newKeyLog() + + run(t, reconciler.Config{ + Name: "whole-set", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 80 * time.Millisecond, nil }, + ReconcileKey: func(_ context.Context, key string) (time.Duration, error) { keys.record(key); return 0, nil }, + Resync: time.Hour, + }) + + p.waitFor(t, 4, "deadline passes on the whole-set path") + keys.stillAt(t, 0, "a deadline with nothing changed must not reconcile a key") +} diff --git a/spinifex/services/awsgw/awsgw.go b/spinifex/services/awsgw/awsgw.go index 79ae0d2f2..97c7c8f65 100644 --- a/spinifex/services/awsgw/awsgw.go +++ b/spinifex/services/awsgw/awsgw.go @@ -622,11 +622,13 @@ func runQuotaReconcile(ctx context.Context, quota *handlers_quota.Service, natsC reconciler.Run(ctx, reconciler.Config{ Name: "quota", Sources: []reconciler.Source{reconciler.Fixed(kvstore.NewBucket(js, cfg), daemon.InstanceRecordPrefix+"*")}, - Reconcile: func(ctx context.Context) error { - return underLeader(func() error { return quota.Reconcile(ctx, accounts, list) }) + // No revisit deadline either way: a counter only moves when an instance + // record does, and that is a KV write the watch already sees. + Reconcile: func(ctx context.Context) (time.Duration, error) { + return 0, underLeader(func() error { return quota.Reconcile(ctx, accounts, list) }) }, - ReconcileKey: func(ctx context.Context, accountID string) error { - return underLeader(func() error { return quota.ReconcileAccount(ctx, accountID, list) }) + ReconcileKey: func(ctx context.Context, accountID string) (time.Duration, error) { + return 0, underLeader(func() error { return quota.ReconcileAccount(ctx, accountID, list) }) }, KeyFor: quotaKeyFor, Resync: handlers_quota.ReconcileInterval,