feat: give a reconcile pass its own revisit deadline, and convert the RDS loop (#948)

* feat: give a reconcile pass its own revisit deadline, and convert the RDS loop

The reconciler assumed every loop was a pure function of the buckets it
watches. Quota and DNS are; RDS, EKS and ECS are not. They also wait on an
EC2 VM reaching running, an endpoint answering, and an agent falling silent,
none of which write to KV, so a watch never sees them.

Converting RDS on the watch alone would have traded traffic for detection
latency: an agent going silent writes nothing, so a dead database would have
been noticed on the 5 minute resync rather than within 15s of its heartbeat
expiring.

Reconcile and ReconcileKey now return a revisit deadline, clamped to the
resync. The RDS reconciler returns the earliest across the fleet: the old
interval for a transitional record, the instant the heartbeat expires for a
settled one, the resolve timeout for a snapshot stuck in creating, and
nothing at all for a region with no databases.

* fix: keep a deadline while a dark instance runs down its failure grace

revisitFor measured a settled instance against its heartbeat expiry, and
returned that difference directly. Once the beat expired the difference went
non-positive, which the loop reads as no deadline at all.

That is the window the detector lives in. Between the beat expiring and the
grace closing nothing writes to KV, so there was nothing left to schedule the
pass that does the failing, and a dead database would have waited for the
resync — slower than the ticker this replaced.

The deadline is now the beat expiry while it is still ahead, and whatever
remains of the failure grace once it is not. A failed instance asks for
nothing: it is terminal, and recovery arrives as a heartbeat, which is a write
the watch already delivers. Already due but still not failed is the degraded
reading, dark agent over a live VM, so it retries at the old interval rather
than immediately.

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,
 	})

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)

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)

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)

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)

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)
 }
 

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

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))
 }

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")
 

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)
+}

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, &current))
 	}
 
-	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)

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,

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

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,
 	})
 

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")
+}

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,