refactor: wake the vpcd drift loop on intent changes, not a fixed 5-minute ticker (#959)

* refactor: wake the vpcd drift loop on intent changes, not a fixed 5-minute ticker

DriftLoop was the last ticker-driven reconcile loop in the event-driven
reconciliation work. It runs on reconciler.Run now, watching the eight KV
buckets LoadIntentFromKV reads, with DriftInterval kept as the resync
backstop and the existing backoff ladder returned as the pass deadline.

Repair latency for a dropped vpc.* event drops from a uniform 0-300s to
the debounce plus a 5s floor. The floor is new and deliberate: a pass
loads eight buckets and scans the OVN NB DB, so without it a launch storm
would turn a five-minute scan into a continuous one.

The buckets are enumerated rather than fixed so watching never creates
one: the EC2 handlers own them and set their history depth.

The bootstrap ReconcileApplyOnly pass stays the startup scan. The loop's
own startup pass does no work and returns the seed deadline instead, so a
clean boot still waits a full interval and a failed one still requeues at
5s.

* fix: keep the intent watchers up when a bucket lookup blips

A per-bucket resolve failure was skipped, which drops the bucket out of
the enumerated set and has the reconciler tear its watcher down until the
next resync. Only a genuine ErrBucketNotFound is a skip now; anything else
fails the enumeration, which is what tells the caller to leave the
watchers it already has in place.

* fix: measure the drift pass floor from the end of a pass, not its start

Stamping before the scan means a pass slower than the floor leaves no gap
at all behind it. The floor is meant to be the minimum time between
passes, so it is stamped once the scan returns.

* test: wait for the intent watcher before the write it must catch

spinifex/network/reconcile/drift.go

@@ -3,20 +3,28 @@ package reconcile
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log/slog"
 	"time"
 
+	handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip"
+	handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw"
+	handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw"
+	handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable"
+	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/otelsetup"
+	loop "github.com/mulgadc/spinifex/spinifex/reconciler"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
-// DriftInterval is the gap between drift passes. Var (not const) so
-// integration tests can shrink it.
+// DriftInterval is the resync backstop between drift passes when nothing
+// changes. Var (not const) so integration tests can shrink it.
 var DriftInterval = 5 * time.Minute
 
-// driftBackoffBase is the requeue gap after the first incomplete pass. Var so
-// tests can shrink it.
+// driftBackoffBase is the requeue gap after the first incomplete pass, and the
+// floor on how often a change may drive a pass. Var so tests can shrink it.
 var driftBackoffBase = 5 * time.Second
 
 // driftBackoffFactor grows the requeue gap per consecutive incomplete pass, up
@@ -25,14 +33,57 @@ var driftBackoffBase = 5 * time.Second
 // of a VPC having no external gateway.
 const driftBackoffFactor = 3
 
-// DriftLoop runs Reconcile every DriftInterval, gated on AcquireLeader so
-// only one vpcd scans at a time. An incomplete pass requeues on a short
-// backoff instead. Returns when ctx is cancelled.
+// intentBuckets names every KV bucket LoadIntentFromKV reads. A write to one of
+// them is an intent change and nothing else: none of their writers is periodic,
+// and a pass writes OVN rather than KV, so watching them cannot self-trigger.
+var intentBuckets = []string{
+	handlers_ec2_vpc.KVBucketVPCs,
+	handlers_ec2_vpc.KVBucketSubnets,
+	handlers_ec2_vpc.KVBucketSecurityGroups,
+	handlers_ec2_vpc.KVBucketENIs,
+	handlers_ec2_igw.KVBucketIGW,
+	handlers_ec2_eip.KVBucketEIPs,
+	handlers_ec2_routetable.KVBucketRouteTables,
+	handlers_ec2_natgw.KVBucketNatGateways,
+}
+
+// intentSource watches the intent buckets. Enumerated rather than fixed so a
+// bucket that does not exist yet is left alone: the EC2 handlers own these and
+// set their history depth, and an opening watch would create one at the wrong
+// depth. A bucket that appears later is picked up on the next resync.
+func intentSource(js jetstream.JetStream) loop.Source {
+	return loop.Dynamic(func(ctx context.Context) ([]*kvstore.Bucket, error) {
+		buckets := make([]*kvstore.Bucket, 0, len(intentBuckets))
+		for _, name := range intentBuckets {
+			kv, err := js.KeyValue(ctx, name)
+			switch {
+			case errors.Is(err, jetstream.ErrBucketNotFound):
+				// Nothing of this kind has ever been created here.
+				slog.DebugContext(ctx, "reconcile/drift: intent bucket absent, not watched",
+					"bucket", name)
+				continue
+			case err != nil:
+				// Reported rather than skipped: a blip is not evidence the bucket
+				// went away, and a partial list would have the caller tear down
+				// the watchers it left out.
+				return nil, fmt.Errorf("resolve intent bucket %s: %w", name, err)
+			}
+			// js is passed so a watcher whose stream was lost can reconnect;
+			// RecreateIfMissing stays false so that reconnect never creates.
+			buckets = append(buckets, kvstore.NewOpenBucket(js, kv, kvstore.Config{Name: name}))
+		}
+		return buckets, nil
+	}, ">")
+}
+
+// DriftLoop reconciles on every intent change and every DriftInterval, gated on
+// AcquireLeader so only one vpcd scans at a time. An incomplete pass asks to be
+// revisited on a short backoff instead. Returns when ctx is cancelled.
 //
 // startup is the outcome of the bootstrap ReconcileApplyOnly pass, or nil when
-// this node did not run one. Seeding from it matters because the gateway LRP's
-// first DORA happens in that pass: without the seed a resource that failed
-// before the loop started waits a full DriftInterval for its first retry.
+// this node did not run one. It seeds the first deadline rather than triggering
+// a pass: that bootstrap pass is the startup scan, and running a second one here
+// would duplicate it and pull a scan forward on every clean boot.
 func DriftLoop(ctx context.Context, rec Reconciler, nc *nats.Conn, localAZ, holder string, startup error) {
 	js, err := jetstream.New(nc)
 	if err != nil {
@@ -40,30 +91,53 @@ func DriftLoop(ctx context.Context, rec Reconciler, nc *nats.Conn, localAZ, hold
 		return
 	}
 
-	backoff := nextDriftBackoff(0, startup)
-	timer := time.NewTimer(driftWait(backoff))
-	defer timer.Stop()
+	loop.Run(ctx, loop.Config{
+		Name:      "vpcd/drift",
+		Sources:   []loop.Source{intentSource(js)},
+		Reconcile: driftPass(rec, nc, js, localAZ, holder, startup),
+		Resync:    DriftInterval,
+	})
+}
 
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-timer.C:
-			previous := backoff
-			backoff = nextDriftBackoff(backoff, runDriftCycle(ctx, rec, nc, js, localAZ, holder))
-			wait := driftWait(backoff)
-			if backoff != previous {
-				slog.Info("reconcile/drift: requeue interval changed", "wait_ms", otelsetup.Millis(wait))
-			}
-			timer.Reset(wait)
+// driftPass returns the reconcile pass, closing over the backoff ladder and the
+// last pass time. Run never calls it concurrently with itself, which is why
+// neither needs a lock.
+func driftPass(rec Reconciler, nc *nats.Conn, js jetstream.JetStream, localAZ, holder string, startup error) func(context.Context) (time.Duration, error) {
+	var backoff time.Duration
+	var lastPass time.Time
+	seeded := false
+
+	return func(ctx context.Context) (time.Duration, error) {
+		if !seeded {
+			seeded = true
+			backoff = nextDriftBackoff(0, startup)
+			return backoff, nil
 		}
+		// A change-driven pass is a repair check, not the apply path: the event
+		// subscriber has already applied the change, so coalescing a burst behind
+		// the floor the ladder starts at costs a scan nothing it would have
+		// found. Without it a launch storm scans continuously.
+		if wait := time.Until(lastPass.Add(driftBackoffBase)); wait > 0 {
+			return wait, nil
+		}
+		previous := backoff
+		backoff = nextDriftBackoff(backoff, runDriftCycle(ctx, rec, nc, js, localAZ, holder))
+		// Stamped after the scan, not before: the floor is a gap between passes,
+		// and a pass slower than the floor would otherwise leave none at all.
+		lastPass = time.Now()
+		if backoff != previous {
+			slog.Info("reconcile/drift: requeue interval changed", "backoff_ms", otelsetup.Millis(backoff))
+		}
+		// The outcome is logged by the cycle, and an unconverged pass is a change
+		// of schedule rather than a failure of the loop.
+		return backoff, nil
 	}
 }
 
 // nextDriftBackoff returns the requeue backoff after a pass: zero once a pass
 // converges, otherwise the previous backoff tripled from driftBackoffBase and
-// held at DriftInterval. Tracked separately from the wait it produces, so a
-// backoff sitting at the cap is not mistaken for a converged loop and reset.
+// held at DriftInterval. Zero means no deadline, leaving a converged loop to the
+// resync rather than asking for a revisit it does not need.
 func nextDriftBackoff(current time.Duration, err error) time.Duration {
 	if !errors.Is(err, ErrPassIncomplete) {
 		return 0
@@ -75,16 +149,7 @@ func nextDriftBackoff(current time.Duration, err error) time.Duration {
 	return min(next, DriftInterval)
 }
 
-// driftWait is the gap before the next pass for a backoff; no backoff means a
-// full DriftInterval.
-func driftWait(backoff time.Duration) time.Duration {
-	if backoff <= 0 {
-		return DriftInterval
-	}
-	return backoff
-}
-
-// runDriftCycle is one tick body, split out so tests can drive it directly.
+// runDriftCycle is one pass body, split out so tests can drive it directly.
 // Returns the reconcile outcome so the caller can size the next wait; a cycle
 // this node did not win the election for returns nil.
 func runDriftCycle(ctx context.Context, rec Reconciler, nc *nats.Conn, js jetstream.JetStream, localAZ, holder string) error {

spinifex/network/reconcile/drift_test.go

@@ -1,6 +1,6 @@
 //test:in-package — the backoff ladder is unexported state (driftBackoffBase,
-//nextDriftBackoff, driftWait) and the tests drive runDriftCycle directly to
-//pin what DriftLoop does with its return value.
+//nextDriftBackoff) and the tests drive runDriftCycle directly to pin what
+//DriftLoop does with its return value.
 
 package reconcile
 
@@ -200,34 +200,31 @@ func TestRunDriftCycle_SkipsWhenNotElected(t *testing.T) {
 	}
 }
 
-// The ladder itself, as pure functions: exact gaps are asserted here because the
-// loop tests above run on compressed timings and can only prove ordering.
+// The ladder itself, as a pure function: exact gaps are asserted here because
+// the loop tests above run on compressed timings and can only prove ordering.
+// Zero is the converged value and means no deadline, which leaves the next pass
+// to the resync the loop configures at DriftInterval.
 func TestDrift_IncompletePassBacksOffThenResets(t *testing.T) {
 	incomplete := incompleteErr()
 	outcomes := []error{incomplete, incomplete, incomplete, incomplete, incomplete, incomplete, nil}
 
 	var backoff time.Duration
-	var waits []time.Duration
+	var got []time.Duration
 	for _, outcome := range outcomes {
 		backoff = nextDriftBackoff(backoff, outcome)
-		waits = append(waits, driftWait(backoff))
+		got = append(got, backoff)
 	}
 
 	want := []time.Duration{
 		5 * time.Second, 15 * time.Second, 45 * time.Second,
 		135 * time.Second, DriftInterval, DriftInterval, // capped, then held
-		DriftInterval, // converged: back to the routine interval
+		0, // converged: no deadline, the resync takes over
 	}
 	for i := range want {
-		if waits[i] != want[i] {
-			t.Fatalf("wait after pass %d = %v, want %v (full sequence %v)", i+1, waits[i], want[i], waits)
+		if got[i] != want[i] {
+			t.Fatalf("backoff after pass %d = %v, want %v (full sequence %v)", i+1, got[i], want[i], got)
 		}
 	}
-	// The cap and the reset produce the same wait, so pin the state behind it:
-	// a converged pass must clear the backoff, not sit at the cap.
-	if backoff != 0 {
-		t.Errorf("backoff after a converged pass = %v, want 0", backoff)
-	}
 }
 
 // A pass that converged, or failed for a reason other than non-convergence
@@ -247,9 +244,6 @@ func TestDrift_NonIncompleteOutcomesKeepDriftInterval(t *testing.T) {
 			if got := nextDriftBackoff(45*time.Second, tt.err); got != 0 {
 				t.Errorf("nextDriftBackoff(45s, %v) = %v, want 0", tt.err, got)
 			}
-			if got := driftWait(0); got != DriftInterval {
-				t.Errorf("driftWait(0) = %v, want %v", got, DriftInterval)
-			}
 		})
 	}
 }

spinifex/network/reconcile/drift_watch_test.go

@@ -0,0 +1,234 @@
+//test:in-package — intentSource, intentBuckets and driftBackoffBase are
+//unexported, and the point of these tests is what the loop watches rather than
+//what it does once woken.
+
+package reconcile
+
+import (
+	"os"
+	"regexp"
+	"strconv"
+	"testing"
+	"time"
+
+	handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw"
+	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// createIntentBucket makes one intent bucket the way its owning handler would,
+// so a test can write to it and have the source find it.
+func createIntentBucket(t *testing.T, js jetstream.JetStream, name string) jetstream.KeyValue {
+	t.Helper()
+	kv, err := kvutil.GetOrCreateBucket(t.Context(), js, name, 1)
+	if err != nil {
+		t.Fatalf("create bucket %s: %v", name, err)
+	}
+	return kv
+}
+
+// bucketNames is what intentSource resolved this cycle.
+func bucketNames(t *testing.T, js jetstream.JetStream) []string {
+	t.Helper()
+	buckets, err := intentSource(js).Buckets(t.Context())
+	if err != nil {
+		t.Fatalf("enumerate intent buckets: %v", err)
+	}
+	names := make([]string, 0, len(buckets))
+	for _, b := range buckets {
+		names = append(names, b.Name())
+	}
+	return names
+}
+
+// waitForWatch writes probe keys until one of them wakes a pass, and returns the
+// call count once it has. Watchers attach asynchronously and are UpdatesOnly, so
+// a single write racing startup is lost for good; retrying is what makes a test
+// that needs a live watcher deterministic rather than dependent on how fast the
+// machine got there.
+func waitForWatch(t *testing.T, rec *stubReconciler, kv jetstream.KeyValue, prefix string) int {
+	t.Helper()
+	before := rec.callCount()
+	deadline := time.Now().Add(10 * time.Second)
+	for i := 0; time.Now().Before(deadline); i++ {
+		if _, err := kv.Put(t.Context(), prefix+strconv.Itoa(i), []byte(`{}`)); err != nil {
+			t.Fatalf("probe write %d: %v", i, err)
+		}
+		if got := waitForCalls(t, rec, before+1, 250*time.Millisecond); got > before {
+			return got
+		}
+	}
+	t.Fatal("no intent write woke a pass within 10s, so the loop is still waiting " +
+		"out DriftInterval rather than watching the bucket")
+	return 0
+}
+
+// streamExists reports whether the KV bucket's backing stream is present, which
+// is how a test tells "watched" from "created by being watched".
+func streamExists(t *testing.T, js jetstream.JetStream, name string) bool {
+	t.Helper()
+	_, err := js.Stream(t.Context(), "KV_"+name)
+	return err == nil
+}
+
+// The whole point of the conversion: a write to an intent bucket must reach the
+// loop long before DriftInterval, because that write is the repair signal a
+// dropped vpc.* event would otherwise leave for the next scan.
+func TestDriftLoop_IntentWriteWakesAPass(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+	kv := createIntentBucket(t, js, handlers_ec2_vpc.KVBucketVPCs)
+	// The interval is long so a pass can only have come from the watch, and the
+	// floor is short so the write is not merely deferred behind it.
+	shrinkDriftTiming(t, time.Minute, time.Millisecond)
+
+	rec := &stubReconciler{outcomes: []error{nil}}
+	startDriftLoop(t, rec, nc, nil)
+
+	// After the startup seed the loop is armed at DriftInterval alone, so the
+	// watch has to be what brings the pass forward.
+	time.Sleep(200 * time.Millisecond)
+	if got := rec.callCount(); got != 0 {
+		t.Fatalf("reconcile ran %d times before any write, want 0", got)
+	}
+
+	// Well inside DriftInterval, so the pass can only have come from the write.
+	waitForWatch(t, rec, kv, "vpc-watch-")
+}
+
+// A burst is one pass, not one per key: the debounce coalesces it and the floor
+// keeps a launch storm from turning a five-minute OVN scan into a running one.
+func TestDriftLoop_BurstOfWritesIsOnePass(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+	kv := createIntentBucket(t, js, handlers_ec2_vpc.KVBucketENIs)
+	shrinkDriftTiming(t, time.Minute, 2*time.Second)
+
+	rec := &stubReconciler{outcomes: []error{nil}}
+	startDriftLoop(t, rec, nc, nil)
+
+	// The burst has to land on a watcher that is already attached, or the writes
+	// it is meant to coalesce are simply never seen.
+	base := waitForWatch(t, rec, kv, "eni-probe-")
+
+	for i := range 20 {
+		if _, err := kv.Put(t.Context(), "eni-burst-"+strconv.Itoa(i), []byte(`{}`)); err != nil {
+			t.Fatalf("write %d: %v", i, err)
+		}
+	}
+	if got := waitForCalls(t, rec, base+1, 10*time.Second); got < base+1 {
+		t.Fatal("a burst of intent writes never woke a pass")
+	}
+	// The burst is long spent, so a further pass here could only come from the
+	// loop tracking writes rather than coalescing them.
+	time.Sleep(500 * time.Millisecond)
+	if got := rec.callCount() - base; got != 1 {
+		t.Errorf("reconcile ran %d times for one burst of 20 writes, want 1", got)
+	}
+}
+
+// Watching must not create. These buckets belong to the EC2 handlers, which set
+// their history depth; a bucket vpcd created first would fix the wrong depth in
+// place, and the handler's own get-or-create would then return it rather than
+// correct it.
+func TestIntentSource_DoesNotCreateMissingBuckets(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+
+	if got := bucketNames(t, js); len(got) != 0 {
+		t.Fatalf("intentSource resolved %v against an empty JetStream, want none", got)
+	}
+	for _, name := range intentBuckets {
+		if streamExists(t, js, name) {
+			t.Errorf("bucket %s exists after enumeration, want it left absent", name)
+		}
+	}
+}
+
+// A cluster that has never had an IGW has no IGW bucket, so the source has to
+// pick one up when it appears rather than only at startup. That is why it is a
+// Dynamic source: JetStream has no bucket-created event, so discovery rides the
+// resync.
+func TestIntentSource_PicksUpABucketThatAppearsLater(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+	createIntentBucket(t, js, handlers_ec2_vpc.KVBucketVPCs)
+
+	if got := bucketNames(t, js); len(got) != 1 || got[0] != handlers_ec2_vpc.KVBucketVPCs {
+		t.Fatalf("intentSource resolved %v, want just %s", got, handlers_ec2_vpc.KVBucketVPCs)
+	}
+
+	createIntentBucket(t, js, handlers_ec2_igw.KVBucketIGW)
+	got := bucketNames(t, js)
+	if len(got) != 2 {
+		t.Fatalf("intentSource resolved %v after a second bucket appeared, want both", got)
+	}
+}
+
+// bucketConstants is every handler KV bucket constant referenced by a file.
+func bucketConstants(t *testing.T, path string) map[string]struct{} {
+	t.Helper()
+	src, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read %s: %v", path, err)
+	}
+	out := map[string]struct{}{}
+	for _, m := range regexp.MustCompile(`handlers_ec2_\w+\.KVBucket\w+`).FindAllString(string(src), -1) {
+		out[m] = struct{}{}
+	}
+	return out
+}
+
+// Every bucket LoadIntentFromKV reads must be watched, or a change to it waits
+// out the resync while every other kind of change is repaired in seconds. The
+// two lists are compared at the source rather than by name, so a load added to
+// intent.go and not to intentBuckets fails here instead of going quiet.
+func TestIntentSource_CoversEveryBucketTheIntentLoadReads(t *testing.T) {
+	read := bucketConstants(t, "intent.go")
+	if len(read) == 0 {
+		t.Fatal("no bucket constants found in intent.go — the scan is not matching what it should")
+	}
+	watched := bucketConstants(t, "drift.go")
+	for name := range read {
+		if _, ok := watched[name]; !ok {
+			t.Errorf("%s is read by LoadIntentFromKV but absent from intentBuckets, "+
+				"so a change to it is only repaired on the resync", name)
+		}
+	}
+}
+
+// The floor is what stops a busy cluster scanning continuously, so it is pinned
+// on the pass rather than inferred from the loop's timing.
+func TestDriftPass_FloorsTheChangeDrivenRate(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+	shrinkDriftTiming(t, time.Minute, time.Hour)
+
+	rec := &stubReconciler{outcomes: []error{nil}}
+	pass := driftPass(rec, nc, js, "us-east-1a", "node-1", nil)
+
+	// The seed pass does no work: the bootstrap ReconcileApplyOnly was the
+	// startup scan and this one must not repeat it.
+	if revisit, err := pass(t.Context()); err != nil || revisit != 0 {
+		t.Fatalf("seed pass = (%v, %v), want (0, nil) for a converged startup", revisit, err)
+	}
+	if got := rec.callCount(); got != 0 {
+		t.Fatalf("seed pass ran reconcile %d times, want 0", got)
+	}
+
+	if _, err := pass(t.Context()); err != nil {
+		t.Fatalf("first real pass: %v", err)
+	}
+	if got := rec.callCount(); got != 1 {
+		t.Fatalf("reconcile ran %d times, want 1", got)
+	}
+
+	revisit, err := pass(t.Context())
+	if err != nil {
+		t.Fatalf("second pass: %v", err)
+	}
+	if got := rec.callCount(); got != 1 {
+		t.Errorf("reconcile ran %d times inside the floor, want it deferred to 1", got)
+	}
+	if revisit <= 0 || revisit > driftBackoffBase {
+		t.Errorf("deferred pass asked to be revisited in %v, want a positive wait "+
+			"no longer than the floor (%v)", revisit, driftBackoffBase)
+	}
+}