feat: wake the EKS cluster reconciler on the signals it already receives (#952)

The loop polled every 30s for two things it was already being told about: the
bootstrap artifacts are keys in the account bucket a watch can see land, and the
control plane's health arrives over core NATS into a field the reconciler read
only on the next tick.

It now runs on reconciler.Run over a watch of the account bucket, plus a new
Config.Trigger fed by the state-report subscription, and returns a revisit
deadline for what elapsed time alone decides: the create timeout, and the instant
the newest report goes stale. A degraded cluster keeps the old interval rather
than restating the three recovery ladders' grace and backoff arithmetic in a
second place.

The trigger only fires on a report that changes health. The control plane
publishes on its own timer regardless, so forwarding every report would have been
the same 30s tick under another name.

spinifex/handlers/eks/cluster_reconciler.go

@@ -6,13 +6,17 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"maps"
 	"net/http"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/kvlease"
+	"github.com/mulgadc/spinifex/spinifex/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/otelsetup"
+	"github.com/mulgadc/spinifex/spinifex/reconciler"
 
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -21,7 +25,11 @@ import (
 const (
 	defaultReconcileLeaseRefresh = 20 * time.Second
 	defaultReconcileInterval     = 30 * time.Second
-	defaultHealthzTimeout        = 5 * time.Second
+	// reconcileResync is the drift backstop for a loop whose signals now arrive as
+	// watch updates and state reports. It also bounds every deadline below it, so
+	// the 15-minute create timeout is served by the resync rather than by a timer.
+	reconcileResync       = 5 * time.Minute
+	defaultHealthzTimeout = 5 * time.Second
 	// defaultCreateTimeout bounds how long a cluster may sit in CREATING before
 	// being marked FAILED. Measured from meta.CreatedAt so a VM that never boots
 	// reaches a terminal state even across daemon restarts.
@@ -148,6 +156,17 @@ type ClusterReconciler struct {
 	stateSubject    string
 	stateStaleAfter time.Duration
 	latest          atomic.Pointer[ServerStateReport]
+	// wake carries a report that changed the cluster's health into the reconcile
+	// loop, which watches KV and would otherwise never see it: the report arrives
+	// over core NATS and writes nothing.
+	wake chan struct{}
+
+	// terminal records why the loop is ending — a terminal cluster status or a
+	// lost lease — and stopRun cancels it. reconciler.Run has no exit condition
+	// of its own, so ending it is cancelling it, and this is what Run reports.
+	terminalMu sync.Mutex
+	terminal   error
+	stopRun    context.CancelFunc
 
 	// Add-on delivery status source. When addonStatusSub is non-nil the
 	// reconciler subscribes to the per-cluster add-on status subject and CASes
@@ -334,17 +353,20 @@ func NewClusterReconciler(leaderKV, acctKV jetstream.KeyValue, accountID, cluste
 		return nil, errors.New("eks: NewClusterReconciler empty holderID")
 	}
 	r := &ClusterReconciler{
-		leaderKV:           leaderKV,
-		acctKV:             acctKV,
-		accountID:          accountID,
-		clusterName:        clusterName,
-		holderID:           holderID,
-		healthURL:          healthURL,
-		leaseRefresh:       defaultReconcileLeaseRefresh,
-		interval:           defaultReconcileInterval,
-		healthTimeout:      defaultHealthzTimeout,
-		createTimeout:      defaultCreateTimeout,
-		stateStaleAfter:    defaultStateStaleAfter,
+		leaderKV:        leaderKV,
+		acctKV:          acctKV,
+		accountID:       accountID,
+		clusterName:     clusterName,
+		holderID:        holderID,
+		healthURL:       healthURL,
+		leaseRefresh:    defaultReconcileLeaseRefresh,
+		interval:        defaultReconcileInterval,
+		healthTimeout:   defaultHealthzTimeout,
+		createTimeout:   defaultCreateTimeout,
+		stateStaleAfter: defaultStateStaleAfter,
+		// One slot: it means "a report changed something", and a second one
+		// arriving before the pass runs says nothing the first did not.
+		wake:               make(chan struct{}, 1),
 		restartGrace:       defaultCPRestartGrace,
 		restartBackoff:     defaultCPRestartBackoff,
 		maxRestartAttempts: defaultMaxCPRestartAttempts,
@@ -402,6 +424,10 @@ func (r *ClusterReconciler) Run(ctx context.Context) error {
 	if r.lease == nil {
 		return errors.New("ClusterReconciler: Run called without AcquireLease")
 	}
+	runCtx, stop := context.WithCancel(ctx)
+	defer stop()
+	r.stopRun = stop
+
 	if r.stateSub != nil && r.stateSubject != "" {
 		sub, err := r.stateSub.Subscribe(r.stateSubject, func(m *nats.Msg) {
 			report, perr := unmarshalServerStateReport(m.Data)
@@ -410,7 +436,7 @@ func (r *ClusterReconciler) Run(ctx context.Context) error {
 					"cluster", r.clusterName, "subject", m.Subject, "err", perr)
 				return
 			}
-			r.latest.Store(&report)
+			r.storeReport(&report)
 		})
 		if err != nil {
 			return fmt.Errorf("subscribe state report %s: %w", r.stateSubject, err)
@@ -434,34 +460,95 @@ func (r *ClusterReconciler) Run(ctx context.Context) error {
 		defer func() { _ = sub.Unsubscribe() }()
 	}
 
-	reconcileT := time.NewTicker(r.interval)
-	defer reconcileT.Stop()
-
-	lost := r.lease.Lost()
-	if err := r.reconcileOnce(ctx); err != nil {
-		if terminalReconcileErr(err) {
-			return err
-		}
-		slog.Warn("ClusterReconciler: initial reconcile failed",
-			"cluster", r.clusterName, "err", err)
-	}
-
-	for {
+	// The lease turning over is not a KV write this loop watches, so losing it has
+	// to end the loop from outside it.
+	go func() {
 		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case <-lost:
-			return ErrReconcilerLeaseLost
-		case <-reconcileT.C:
-			if err := r.reconcileOnce(ctx); err != nil {
-				if terminalReconcileErr(err) {
-					return err
-				}
-				slog.Warn("ClusterReconciler: reconcile failed",
-					"cluster", r.clusterName, "err", err)
-			}
+		case <-runCtx.Done():
+		case <-r.lease.Lost():
+			r.endRun(ErrReconcilerLeaseLost)
 		}
+	}()
+
+	// The whole account bucket, and that is forced rather than lazy: a cluster key
+	// is "clusters/<name>/meta", "/" is not the NATS subject separator, so the key
+	// is one token and no prefix wildcard selects a single cluster. The cost is
+	// waking on a sibling cluster's writes, which real activity bounds.
+	bucket := kvstore.NewOpenBucket(nil, r.acctKV, kvstore.Config{Name: r.acctKV.Bucket()})
+	reconciler.Run(runCtx, reconciler.Config{
+		Name:      "eks/" + r.clusterName,
+		Sources:   []reconciler.Source{reconciler.Fixed(bucket, ">")},
+		Reconcile: r.reconcilePass,
+		Trigger:   r.wake,
+		Resync:    reconcileResync,
+	})
+
+	if err := r.terminalErr(); err != nil {
+		return err
 	}
+	return ctx.Err()
+}
+
+// reconcilePass runs one reconcile and says when the loop should next run with
+// nothing arriving. A terminal outcome ends the loop rather than being returned:
+// the loop has no notion of finishing, so finishing it is cancelling it.
+func (r *ClusterReconciler) reconcilePass(ctx context.Context) (time.Duration, error) {
+	revisit, err := r.reconcileOnce(ctx)
+	switch {
+	case err == nil:
+		return revisit, nil
+	case terminalReconcileErr(err):
+		r.endRun(err)
+		return 0, nil
+	default:
+		// Retried at the old cadence rather than left to the resync: a pass that
+		// failed has learnt nothing, so waiting five minutes to try again would be
+		// slower than the ticker this replaced.
+		return r.interval, err
+	}
+}
+
+// endRun records the first reason the loop ended and cancels it. First wins: a
+// lost lease and a terminal status can land together, and which one is reported
+// should not depend on the race.
+func (r *ClusterReconciler) endRun(reason error) {
+	r.terminalMu.Lock()
+	if r.terminal == nil {
+		r.terminal = reason
+	}
+	r.terminalMu.Unlock()
+	if r.stopRun != nil {
+		r.stopRun()
+	}
+}
+
+func (r *ClusterReconciler) terminalErr() error {
+	r.terminalMu.Lock()
+	defer r.terminalMu.Unlock()
+	return r.terminal
+}
+
+// storeReport keeps the newest control-plane report and wakes the loop only when
+// it says something about health the last one did not. The control plane
+// publishes on its own timer whether or not anything changed, so waking on every
+// report would be the old tick again under another name.
+func (r *ClusterReconciler) storeReport(report *ServerStateReport) {
+	prev := r.latest.Swap(report)
+	if prev != nil && sameHealth(prev, report) {
+		return
+	}
+	select {
+	case r.wake <- struct{}{}:
+	default:
+	}
+}
+
+// sameHealth reports whether two reports would produce the same observe result.
+// Freshness is deliberately excluded: a report that only repeats itself pushes
+// the staleness deadline out, which the next pass reads for itself.
+func sameHealth(a, b *ServerStateReport) bool {
+	return a.Healthz == b.Healthz && a.Reason == b.Reason &&
+		a.NodeCount == b.NodeCount && maps.Equal(a.NodegroupReady, b.NodegroupReady)
 }
 
 func terminalReconcileErr(err error) bool {
@@ -470,40 +557,45 @@ func terminalReconcileErr(err error) bool {
 		errors.Is(err, ErrClusterNotFound)
 }
 
-func (r *ClusterReconciler) reconcileOnce(ctx context.Context) error {
+// reconcileOnce runs one pass and reports how long the loop may wait with
+// nothing arriving. The signals that do arrive — a bootstrap artifact landing, a
+// status change, a control plane changing its mind about its own health — are
+// watched or triggered, so the deadline covers only what elapsed time decides.
+func (r *ClusterReconciler) reconcileOnce(ctx context.Context) (time.Duration, error) {
 	meta, err := GetClusterMeta(ctx, r.acctKV, r.clusterName)
 	if err != nil {
-		return err
+		return 0, err
 	}
 	switch meta.Status {
 	case ClusterStatusCreating:
 		ready, reason := r.bootstrapReady(ctx, meta)
 		if !ready {
-			// Info, not Debug: this is the per-tick reason a CREATING cluster has
+			// Info, not Debug: this is the per-pass reason a CREATING cluster has
 			// not yet flipped ACTIVE. At Debug it is invisible at the default
 			// level, so a cluster that stalls to the create-timeout FAILED gives
 			// no diagnosable cause. Logged only during the CREATING window.
 			slog.Info("ClusterReconciler: bootstrap not ready",
 				"cluster", r.clusterName, "reason", reason)
-			return r.failIfCreateTimedOut(ctx, meta, "bootstrap not ready: "+reason)
+			return r.createRevisit(meta), r.failIfCreateTimedOut(ctx, meta, "bootstrap not ready: "+reason)
 		}
 		issue, nodeCount, nodegroupReady := r.observe(ctx)
 		if issue != "" {
 			slog.Info("ClusterReconciler: health not ready",
 				"cluster", r.clusterName, "issue", issue)
-			return r.failIfCreateTimedOut(ctx, meta, "healthz not ready: "+issue)
+			return r.createRevisit(meta), r.failIfCreateTimedOut(ctx, meta, "healthz not ready: "+issue)
 		}
 		if err := SetClusterStatus(ctx, r.acctKV, r.clusterName, ClusterStatusActive); err != nil {
-			return err
+			return 0, err
 		}
 		if err := SetClusterHealthState(ctx, r.acctKV, r.clusterName, "", nodeCount, nodegroupReady); err != nil {
 			if errors.Is(err, ErrClusterNotFound) {
-				return ErrClusterNotFound
+				return 0, ErrClusterNotFound
 			}
-			return fmt.Errorf("record cluster health: %w", err)
+			return 0, fmt.Errorf("record cluster health: %w", err)
 		}
 		slog.Info("ClusterReconciler: cluster transitioned to ACTIVE",
 			"cluster", r.clusterName, "nodes", nodeCount)
+		return r.activeRevisit(""), nil
 	case ClusterStatusActive:
 		issue, nodeCount, nodegroupReady := r.observe(ctx)
 		if issue != "" {
@@ -512,19 +604,68 @@ func (r *ClusterReconciler) reconcileOnce(ctx context.Context) error {
 		}
 		if err := SetClusterHealthState(ctx, r.acctKV, r.clusterName, issue, nodeCount, nodegroupReady); err != nil {
 			if errors.Is(err, ErrClusterNotFound) {
-				return ErrClusterNotFound
+				return 0, ErrClusterNotFound
 			}
-			return fmt.Errorf("record cluster health: %w", err)
+			return 0, fmt.Errorf("record cluster health: %w", err)
 		}
 		r.maybeRecoverControlPlane(ctx, meta, issue)
 		r.maybeReformEtcdQuorum(ctx, meta, issue)
 		r.maybeReplaceControlPlaneMember(ctx, meta, issue)
+		return r.activeRevisit(issue), nil
 	case ClusterStatusDeleting:
-		return ErrReconcilerClusterDeleting
+		return 0, ErrReconcilerClusterDeleting
 	case ClusterStatusFailed:
-		return ErrReconcilerClusterFailed
+		return 0, ErrReconcilerClusterFailed
 	}
-	return nil
+	return 0, nil
+}
+
+// createRevisit is what remains of the create timeout. Everything else a
+// CREATING cluster waits for is an event now: the bootstrap artifacts are keys
+// in the watched bucket, and the first healthy report is a trigger. Only the
+// timeout expiring announces itself to nobody.
+func (r *ClusterReconciler) createRevisit(meta *ClusterMeta) time.Duration {
+	if r.createTimeout <= 0 || meta.CreatedAt.IsZero() {
+		return 0
+	}
+	if remaining := time.Until(meta.CreatedAt.Add(r.createTimeout)); remaining > 0 {
+		return remaining
+	}
+	// Already due and still not failed, so the pass that fails it could not. Retry
+	// at the old cadence rather than immediately, which would spin.
+	return r.interval
+}
+
+// activeRevisit is the deadline for a cluster that is already up.
+func (r *ClusterReconciler) activeRevisit(issue string) time.Duration {
+	if issue != "" {
+		// A degraded cluster is on the recovery ladders' clocks — three graces,
+		// three backoffs and three attempt caps that interact — and queries its
+		// members' VM state every pass anyway. Keeping the old interval leaves
+		// that timing exactly as it is rather than restating it here.
+		return r.interval
+	}
+	return r.reportRevisit()
+}
+
+// reportRevisit is the instant the newest control-plane report goes stale, which
+// is what a healthy cluster is waiting for: silence is published by nobody, so
+// nothing else will wake the loop to notice it.
+func (r *ClusterReconciler) reportRevisit() time.Duration {
+	if r.stateSub == nil || r.stateSubject == "" {
+		// Health comes from an HTTP probe, which answers only when asked.
+		return r.interval
+	}
+	report := r.latest.Load()
+	if report == nil {
+		return r.interval
+	}
+	if fresh := time.Until(time.Unix(report.TS, 0).Add(r.stateStaleAfter)); fresh > 0 {
+		return fresh
+	}
+	// Already stale, so there is no expiry left to wait for and the next pass is
+	// what notices. Returning nothing here would leave that pass unscheduled.
+	return r.interval
 }
 
 // maybeRecoverControlPlane attempts a bounded in-place restart of a wedged

spinifex/handlers/eks/cluster_reconciler_recovery_test.go

@@ -189,7 +189,7 @@ func TestActiveBranchRestartsWedgedCP(t *testing.T) {
 	r.degradedSince = time.Now().Add(-10 * time.Minute)
 	r.latest.Store(&ServerStateReport{Healthz: "ok", NodeCount: 2, TS: time.Now().Add(-5 * time.Minute).Unix()})
 
-	require.NoError(t, r.reconcileOnce(context.Background()))
+	require.NoError(t, onePass(t, r))
 
 	meta, err := GetClusterMeta(t.Context(), acctKV, "alpha")
 	require.NoError(t, err)

spinifex/handlers/eks/cluster_reconciler_replace_test.go

@@ -326,7 +326,7 @@ func TestActiveBranchReplacesLostCPMember(t *testing.T) {
 	r.replacingSince = time.Now().Add(-10 * time.Minute)
 	r.latest.Store(freshReport("ok", 3))
 
-	require.NoError(t, r.reconcileOnce(context.Background()))
+	require.NoError(t, onePass(t, r))
 
 	require.Equal(t, 1, prov.calls, "wedged member replaced through the ACTIVE arm")
 	got, err := GetClusterMeta(t.Context(), acctKV, "alpha")

spinifex/handlers/eks/cluster_reconciler_state_test.go

@@ -43,7 +43,7 @@ func TestClusterReconciler_CreatingTransitionsToActiveOnStateReport(t *testing.T
 	seedBootstrapState(t, acctKV)
 	r.latest.Store(freshReport("ok", 3))
 
-	require.NoError(t, r.reconcileOnce(context.Background()))
+	require.NoError(t, onePass(t, r))
 
 	meta, err := GetClusterMeta(t.Context(), acctKV, "alpha")
 	require.NoError(t, err)
@@ -58,7 +58,7 @@ func TestClusterReconciler_CreatingStaysWithoutStateReport(t *testing.T) {
 	seedBootstrapState(t, acctKV)
 	// No report stored.
 
-	require.NoError(t, r.reconcileOnce(context.Background()))
+	require.NoError(t, onePass(t, r))
 
 	meta, err := GetClusterMeta(t.Context(), acctKV, "alpha")
 	require.NoError(t, err)
@@ -71,7 +71,7 @@ func TestClusterReconciler_CreatingStaysOnUnhealthyReport(t *testing.T) {
 	seedBootstrapState(t, acctKV)
 	r.latest.Store(freshReport("fail", 0))
 
-	require.NoError(t, r.reconcileOnce(context.Background()))
+	require.NoError(t, onePass(t, r))
 
 	meta, err := GetClusterMeta(t.Context(), acctKV, "alpha")
 	require.NoError(t, err)
@@ -84,7 +84,7 @@ func TestClusterReconciler_ActiveRecordsNodeCountAndClearsIssue(t *testing.T) {
 	require.NoError(t, SetClusterHealthState(t.Context(), acctKV, "alpha", "stale", 0, nil))
 	r.latest.Store(freshReport("ok", 4))
 
-	require.NoError(t, r.reconcileOnce(context.Background()))
+	require.NoError(t, onePass(t, r))
 
 	meta, err := GetClusterMeta(t.Context(), acctKV, "alpha")
 	require.NoError(t, err)
@@ -98,7 +98,7 @@ func TestClusterReconciler_ActiveFlagsStaleReport(t *testing.T) {
 	require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusActive))
 	r.latest.Store(&ServerStateReport{Healthz: "ok", NodeCount: 2, TS: time.Now().Add(-5 * time.Minute).Unix()})
 
-	require.NoError(t, r.reconcileOnce(context.Background()))
+	require.NoError(t, onePass(t, r))
 
 	meta, err := GetClusterMeta(t.Context(), acctKV, "alpha")
 	require.NoError(t, err)
@@ -114,7 +114,7 @@ func TestClusterReconciler_ActiveUnhealthyReportSurfacesReason(t *testing.T) {
 		Reason: "readyz:[etcd]; etcd:unreachable; disk:ok",
 	})
 
-	require.NoError(t, r.reconcileOnce(context.Background()))
+	require.NoError(t, onePass(t, r))
 
 	meta, err := GetClusterMeta(t.Context(), acctKV, "alpha")
 	require.NoError(t, err)
@@ -128,7 +128,7 @@ func TestClusterReconciler_ActiveUnhealthyReportWithoutReasonStaysTerse(t *testi
 	require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusActive))
 	r.latest.Store(freshReport("fail", 0))
 
-	require.NoError(t, r.reconcileOnce(context.Background()))
+	require.NoError(t, onePass(t, r))
 
 	meta, err := GetClusterMeta(t.Context(), acctKV, "alpha")
 	require.NoError(t, err)

spinifex/handlers/eks/cluster_reconciler_test.go

@@ -45,6 +45,15 @@ func (s *stubHTTPDoer) setResponse(status int, err error) {
 	s.err = err
 }
 
+// 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 *ClusterReconciler) error {
+	t.Helper()
+	_, err := r.reconcileOnce(t.Context())
+	return err
+}
+
 func newReconcilerHarness(t *testing.T, healthURL string, opts ...ReconcilerOption) (*ClusterReconciler, jetstream.KeyValue, jetstream.KeyValue) {
 	t.Helper()
 	_, nc, _ := testutil.StartTestJetStream(t)
@@ -418,14 +427,14 @@ func TestClusterReconciler_ActiveHealthRecoversClearsIssue(t *testing.T) {
 
 	// First probe fails and records the issue.
 	require.Error(t, r.probeHealthz(context.Background()))
-	require.NoError(t, r.reconcileOnce(context.Background()))
+	require.NoError(t, onePass(t, r))
 	meta, err := GetClusterMeta(t.Context(), acctKV, "alpha")
 	require.NoError(t, err)
 	require.NotEmpty(t, meta.HealthIssue)
 
 	// Probe recovers: the issue must be cleared.
 	stub.setResponse(http.StatusOK, nil)
-	require.NoError(t, r.reconcileOnce(context.Background()))
+	require.NoError(t, onePass(t, r))
 	meta, err = GetClusterMeta(t.Context(), acctKV, "alpha")
 	require.NoError(t, err)
 	assert.Empty(t, meta.HealthIssue, "recovered /healthz must clear the recorded issue")

spinifex/handlers/eks/revisit_test.go

@@ -0,0 +1,183 @@
+package handlers_eks
+
+//test:in-package — asserts on the deadline reconcileOnce returns and on when a
+// state report wakes the loop, both of which are unexported and are the whole
+// subject of these tests.
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// woke reports whether storeReport signalled, draining the channel so a later
+// call in the same test starts from quiet.
+func woke(r *ClusterReconciler) bool {
+	select {
+	case <-r.wake:
+		return true
+	default:
+		return false
+	}
+}
+
+// A CREATING cluster's artifacts are keys in the watched bucket and its first
+// healthy report is a trigger, so the timeout expiring is the only thing left
+// that announces itself to nobody.
+func TestReconcileOnce_ACreatingClusterAsksForWhatIsLeftOfItsCreateTimeout(t *testing.T) {
+	r, _, acctKV := newStateReconcilerHarness(t, WithCreateTimeout(15*time.Minute))
+	freshenClusterCreatedAt(t, acctKV)
+	// No bootstrap state and no report, so the cluster stays CREATING.
+
+	revisit, err := r.reconcileOnce(t.Context())
+	require.NoError(t, err)
+
+	assert.Greater(t, revisit, 14*time.Minute, "a cluster created seconds ago has nearly all of its window left")
+	assert.LessOrEqual(t, revisit, 15*time.Minute)
+}
+
+// A create timeout already past belongs to the pass that fails the cluster.
+// Returning nothing would leave that pass unscheduled.
+func TestReconcileOnce_ACreatingClusterPastItsTimeoutAsksForTheInterval(t *testing.T) {
+	r, _, acctKV := newStateReconcilerHarness(t,
+		WithCreateTimeout(15*time.Minute),
+		WithReconcileInterval(30*time.Second),
+	)
+	meta, err := GetClusterMeta(t.Context(), acctKV, "alpha")
+	require.NoError(t, err)
+	meta.CreatedAt = time.Now().UTC().Add(-time.Hour)
+	require.NoError(t, PutClusterMeta(t.Context(), acctKV, meta))
+
+	// The pass marks it FAILED, which is terminal and reported as such.
+	revisit, err := r.reconcileOnce(t.Context())
+	require.ErrorIs(t, err, ErrReconcilerClusterFailed)
+	assert.Equal(t, 30*time.Second, revisit)
+}
+
+// A healthy cluster is waiting for its control plane to go silent, and silence
+// is published by nobody. The deadline is the instant the newest report expires,
+// which lands the pass on it rather than up to a tick after it.
+func TestReconcileOnce_AHealthyClusterAsksForItsReportsStalenessInstant(t *testing.T) {
+	r, _, acctKV := newStateReconcilerHarness(t, WithStateStaleAfter(90*time.Second))
+	require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusActive))
+	r.latest.Store(freshReport("ok", 3))
+
+	revisit, err := r.reconcileOnce(t.Context())
+	require.NoError(t, err)
+
+	assert.Greater(t, revisit, 60*time.Second, "a report seconds old must not read as nearly stale")
+	assert.LessOrEqual(t, revisit, 90*time.Second)
+}
+
+// A report already past its window is the case the next pass has to notice, so
+// the deadline falls back to the interval rather than to nothing.
+func TestReconcileOnce_AClusterWithAStaleReportAsksForTheInterval(t *testing.T) {
+	r, _, acctKV := newStateReconcilerHarness(t,
+		WithStateStaleAfter(90*time.Second),
+		WithReconcileInterval(30*time.Second),
+	)
+	require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusActive))
+	r.latest.Store(&ServerStateReport{Healthz: "ok", NodeCount: 2, TS: time.Now().Add(-5 * time.Minute).Unix()})
+
+	revisit, err := r.reconcileOnce(t.Context())
+	require.NoError(t, err)
+
+	assert.Equal(t, 30*time.Second, revisit)
+}
+
+// A degraded cluster keeps the old cadence: the recovery ladders run on elapsed
+// time that this deliberately does not restate, and they poll their members'
+// VM state on every pass regardless.
+func TestReconcileOnce_ADegradedClusterAsksForTheInterval(t *testing.T) {
+	r, _, acctKV := newStateReconcilerHarness(t, WithReconcileInterval(30*time.Second))
+	require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusActive))
+	r.latest.Store(freshReport("fail", 0))
+
+	revisit, err := r.reconcileOnce(t.Context())
+	require.NoError(t, err)
+
+	assert.Equal(t, 30*time.Second, revisit)
+}
+
+// Without a state source health comes from an HTTP probe, which answers only
+// when asked, so there is nothing to wait for and the poll has to stay.
+func TestReconcileOnce_AProbedClusterAsksForTheInterval(t *testing.T) {
+	stub := &stubHTTPDoer{status: 200}
+	r, _, acctKV := newReconcilerHarness(t, "https://nlb.example/healthz",
+		WithHTTPClient(stub),
+		WithReconcileInterval(30*time.Second),
+	)
+	require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusActive))
+
+	revisit, err := r.reconcileOnce(t.Context())
+	require.NoError(t, err)
+
+	assert.Equal(t, 30*time.Second, revisit)
+}
+
+// A terminal status ends the loop, so there is no next run to schedule.
+func TestReconcileOnce_ADeletingClusterAsksForNoDeadline(t *testing.T) {
+	r, _, acctKV := newStateReconcilerHarness(t)
+	require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusDeleting))
+
+	revisit, err := r.reconcileOnce(t.Context())
+	require.ErrorIs(t, err, ErrReconcilerClusterDeleting)
+	assert.Zero(t, revisit)
+}
+
+// The first report is always news: there is nothing to compare it against, and
+// it is what flips a CREATING cluster.
+func TestStoreReport_TheFirstReportWakesTheLoop(t *testing.T) {
+	r, _, _ := newStateReconcilerHarness(t)
+
+	r.storeReport(freshReport("ok", 3))
+
+	assert.True(t, woke(r), "the first report must wake the loop")
+}
+
+// The control plane publishes on its own timer whether or not anything changed.
+// Waking on every report would be the old 30s tick under another name.
+func TestStoreReport_ARepeatedReportDoesNotWakeTheLoop(t *testing.T) {
+	r, _, _ := newStateReconcilerHarness(t)
+	r.storeReport(&ServerStateReport{Healthz: "ok", NodeCount: 3, TS: time.Now().Unix()})
+	require.True(t, woke(r))
+
+	r.storeReport(&ServerStateReport{Healthz: "ok", NodeCount: 3, TS: time.Now().Add(30 * time.Second).Unix()})
+
+	assert.False(t, woke(r), "a report saying the same thing later must not wake the loop")
+}
+
+// The newest report still has to be stored even when it does not wake anything,
+// because the staleness deadline is measured from it.
+func TestStoreReport_ARepeatedReportStillRefreshesWhatIsStored(t *testing.T) {
+	r, _, _ := newStateReconcilerHarness(t)
+	r.storeReport(&ServerStateReport{Healthz: "ok", NodeCount: 3, TS: 1000})
+
+	r.storeReport(&ServerStateReport{Healthz: "ok", NodeCount: 3, TS: 2000})
+
+	assert.Equal(t, int64(2000), r.latest.Load().TS)
+}
+
+func TestStoreReport_AChangedReportWakesTheLoop(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		next *ServerStateReport
+	}{
+		{"healthz", &ServerStateReport{Healthz: "fail", NodeCount: 3}},
+		{"reason", &ServerStateReport{Healthz: "ok", NodeCount: 3, Reason: "etcd:unreachable"}},
+		{"node count", &ServerStateReport{Healthz: "ok", NodeCount: 2}},
+		{"nodegroup counts", &ServerStateReport{Healthz: "ok", NodeCount: 3, NodegroupReady: map[string]int{"ng": 2}}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			r, _, _ := newStateReconcilerHarness(t)
+			r.storeReport(&ServerStateReport{Healthz: "ok", NodeCount: 3})
+			require.True(t, woke(r))
+
+			r.storeReport(tc.next)
+
+			assert.True(t, woke(r), "a report that changes health must wake the loop")
+		})
+	}
+}

spinifex/reconciler/reconciler.go

@@ -12,6 +12,10 @@
 // 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.
+//
+// A caller whose inputs are not all KV can supply a Trigger channel as well, for
+// the events that arrive by some other route and would otherwise be seen only
+// when the next pass happened to look.
 package reconciler
 
 import (
@@ -89,6 +93,11 @@ type Config struct {
 	// tombstone carries no value, so a caller whose work key lives in the value
 	// cannot name it, and a whole-set pass needs no attribution.
 	KeyFor func(entry jetstream.KeyValueEntry) (keys []string, ok bool)
+	// Trigger wakes the loop from outside the buckets, for the inputs a KV watch
+	// cannot see because they never reach KV. A receive is treated exactly like a
+	// watch update: debounced with them, then served by a whole-set pass, since an
+	// external signal names no key. Nil means the buckets are the only source.
+	Trigger <-chan struct{}
 	// Resync bounds how stale the world can get when no update arrives, and is
 	// also when Sources are re-enumerated. Zero means DefaultResync.
 	Resync time.Duration
@@ -174,6 +183,10 @@ func Run(ctx context.Context, cfg Config) {
 	// pass that would have seen it ran before the watch existed.
 	w.resync(ctx)
 
+	if cfg.Trigger != nil {
+		go forward(ctx, cfg.Trigger, w)
+	}
+
 	resync := time.NewTicker(cfg.Resync)
 	defer resync.Stop()
 
@@ -213,6 +226,24 @@ func Run(ctx context.Context, cfg Config) {
 	}
 }
 
+// forward turns an external wake into the same signal a watch update produces,
+// so a trigger is debounced and coalesced with them rather than taking a second
+// path through the loop. It queues the whole set because a trigger names no key.
+func forward(ctx context.Context, trigger <-chan struct{}, w *watchSet) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case _, open := <-trigger:
+			if !open {
+				return
+			}
+			w.queue.addWhole()
+			w.signal()
+		}
+	}
+}
+
 // 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.

spinifex/reconciler/trigger_test.go

@@ -0,0 +1,152 @@
+package reconciler_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/reconciler"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// The reason the trigger exists: a loop whose signal arrives by some other route
+// than a KV write would otherwise see it only when the next pass happened to
+// look, which is the poll the watch was supposed to replace.
+func TestRun_ATriggerWakesTheLoopWithNoWatchTraffic(t *testing.T) {
+	store, _ := newBucket(t, "trigger-wakes")
+	p := newPasses()
+	trigger := make(chan struct{}, 1)
+
+	run(t, reconciler.Config{
+		Name:      "trigger",
+		Sources:   []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil },
+		Trigger:   trigger,
+		Resync:    time.Hour,
+	})
+
+	p.waitFor(t, 1, "the startup pass")
+	trigger <- struct{}{}
+
+	p.waitFor(t, 2, "a pass from the trigger, against a bucket nobody is writing to")
+}
+
+// A quiet trigger must not be a source of passes on its own: the loop still
+// falls back to the resync and nothing else.
+func TestRun_AQuietTriggerAddsNoPasses(t *testing.T) {
+	store, _ := newBucket(t, "trigger-quiet")
+	p := newPasses()
+	trigger := make(chan struct{}, 1)
+
+	run(t, reconciler.Config{
+		Name:      "quiet",
+		Sources:   []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil },
+		Trigger:   trigger,
+		Resync:    time.Hour,
+	})
+
+	p.waitFor(t, 1, "the startup pass")
+	p.stillAt(t, 1, "a trigger nobody signalled must not produce passes")
+}
+
+// A burst of triggers is one pass, the same as a burst of watch updates. The
+// caller signalling several times in a row is describing one thing to converge,
+// not several.
+func TestRun_ABurstOfTriggersCoalescesIntoOnePass(t *testing.T) {
+	store, _ := newBucket(t, "trigger-burst")
+	p := newPasses()
+	trigger := make(chan struct{}, 8)
+
+	run(t, reconciler.Config{
+		Name:      "burst",
+		Sources:   []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil },
+		Trigger:   trigger,
+		Resync:    time.Hour,
+		Debounce:  100 * time.Millisecond,
+	})
+
+	p.waitFor(t, 1, "the startup pass")
+	for range 8 {
+		trigger <- struct{}{}
+	}
+
+	p.waitFor(t, 2, "one pass for the burst")
+	p.stillAt(t, 2, "eight triggers inside one debounce window are one pass")
+}
+
+// A trigger names no key, so a per-key loop has to be served by the whole-set
+// pass. Anything else would leave the caller's signal describing nothing.
+func TestRun_ATriggerRunsTheWholeSetOnAPerKeyLoop(t *testing.T) {
+	store, _ := newBucket(t, "trigger-whole-set")
+	whole := newPasses()
+	keyed := newPasses()
+	trigger := make(chan struct{}, 1)
+
+	run(t, reconciler.Config{
+		Name:         "whole-set",
+		Sources:      []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile:    func(context.Context) (time.Duration, error) { whole.record(); return 0, nil },
+		ReconcileKey: func(context.Context, string) (time.Duration, error) { keyed.record(); return 0, nil },
+		Trigger:      trigger,
+		Resync:       time.Hour,
+	})
+
+	whole.waitFor(t, 1, "the startup pass")
+	trigger <- struct{}{}
+
+	whole.waitFor(t, 2, "the trigger served by a whole-set pass")
+	assert.Zero(t, keyed.now(), "a trigger carries no key, so no per-key pass may run")
+}
+
+// The deadline a triggered pass returns has to be armed like any other, or a
+// loop woken by a trigger would silently lose the schedule it just asked for.
+func TestRun_ATriggeredPassStillArmsItsDeadline(t *testing.T) {
+	store, _ := newBucket(t, "trigger-deadline")
+	p := newPasses()
+	trigger := make(chan struct{}, 1)
+
+	run(t, reconciler.Config{
+		Name:      "trigger-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,
+		Trigger:   trigger,
+	})
+
+	p.waitFor(t, 1, "the startup pass")
+	trigger <- struct{}{}
+
+	p.waitFor(t, 5, "deadline passes continuing after the trigger woke the loop")
+}
+
+// Closing the trigger is how a caller says its source is finished. The loop must
+// survive it rather than spin on a closed channel.
+func TestRun_AClosedTriggerDoesNotSpin(t *testing.T) {
+	store, _ := newBucket(t, "trigger-closed")
+	p := newPasses()
+	trigger := make(chan struct{})
+
+	run(t, reconciler.Config{
+		Name:      "closed",
+		Sources:   []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil },
+		Trigger:   trigger,
+		Resync:    time.Hour,
+	})
+
+	p.waitFor(t, 1, "the startup pass")
+	close(trigger)
+
+	// One pass is allowed for the close itself; what must not happen is the
+	// forwarder reading a closed channel forever and passing on every read.
+	time.Sleep(300 * time.Millisecond)
+	require.LessOrEqual(t, p.now(), 2, "a closed trigger must not drive a pass loop")
+
+	// Still alive: a write after the close is still reconciled.
+	before := p.now()
+	require.NoError(t, store.Set(t.Context(), "after-close", &record{Name: "after-close"}))
+	p.waitFor(t, before+1, "a watch update after the trigger closed")
+}