fix(vpcd): bounded-retry reconcile-leader instead of fail-open (#202)

spinifex/services/vpcd/reconcile.go

@@ -23,24 +23,52 @@ const (
 	reconcileLeaderTTL    = 60 * time.Second
 )
 
+// Bounded wait for JetStream to come up at vpcd startup. On a cold multi-node
+// start NATS clustering takes a few seconds and js.CreateKeyValue returns
+// "context deadline exceeded" until quorum forms. Retrying makes the election
+// deterministic; without it every vpcd falls through and races the same OVN
+// NB transactions (mulga-js-72). Vars (not consts) so tests can shrink them.
+var (
+	reconcileLeaderRetryFor  = 60 * time.Second
+	reconcileLeaderRetryStep = 1 * time.Second
+)
+
 // AcquireReconcileLeader returns release+true exactly once across all vpcds in
 // a cluster, gating the startup Reconcile/ReconcileFromKV passes. Other vpcds
 // get release=nil, elected=false and skip reconcile (runtime VPC events use the
 // vpcd-workers queue group, so they remain handled cluster-wide).
 //
-// Returns elected=true on KV-bucket failure: first-boot has at most one vpcd
-// up, so falling through is safe and avoids deadlocking the cluster if NATS
-// KV isn't ready.
+// Bounded-retry contract: when JetStream KV is not yet reachable (cold
+// multi-node start, NATS quorum still forming) we retry for
+// reconcileLeaderRetryFor before giving up. On exhaustion we return
+// elected=false rather than running unguarded — concurrent reconciles produce
+// silent OVN NB damage (mulga-js-72). Operators can restart vpcd to retry
+// reconcile once NATS recovers.
 func AcquireReconcileLeader(nc *nats.Conn, holder string) (func(), bool) {
 	js, _ := nc.JetStream()
-	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
-		Bucket:  reconcileLeaderBucket,
-		History: 1,
-		TTL:     reconcileLeaderTTL,
-	})
-	if err != nil {
-		slog.Warn("vpcd reconcile-leader: KV bucket unavailable, running reconcile unguarded", "err", err)
-		return func() {}, true
+
+	var (
+		kv  nats.KeyValue
+		err error
+	)
+	deadline := time.Now().Add(reconcileLeaderRetryFor)
+	for {
+		kv, err = js.CreateKeyValue(&nats.KeyValueConfig{
+			Bucket:  reconcileLeaderBucket,
+			History: 1,
+			TTL:     reconcileLeaderTTL,
+		})
+		if err == nil {
+			break
+		}
+		if time.Now().After(deadline) {
+			slog.Error("vpcd reconcile-leader: JetStream KV unreachable after retry, skipping reconcile",
+				"holder", holder, "waited", reconcileLeaderRetryFor, "err", err)
+			return nil, false
+		}
+		slog.Debug("vpcd reconcile-leader: JetStream KV not ready, retrying",
+			"holder", holder, "err", err)
+		time.Sleep(reconcileLeaderRetryStep)
 	}
 
 	if _, err := kv.Create(reconcileLeaderKey, []byte(holder)); err != nil {

spinifex/services/vpcd/reconcile_test.go

@@ -3,6 +3,8 @@ package vpcd
 import (
 	"context"
 	"encoding/json"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -36,15 +38,89 @@ func TestAcquireReconcileLeader_OnlyOneWinner(t *testing.T) {
 	releaseC()
 }
 
-func TestAcquireReconcileLeader_NoJetStreamFallsThrough(t *testing.T) {
-	// No JetStream on this server — caller must fall through (elected=true)
-	// rather than deadlock the cluster on first boot.
-	_, nc := testutil.StartTestNATS(t)
+// withReconcileLeaderRetry shrinks the bounded-retry window for tests so they
+// run sub-second instead of waiting the production 60s deadline.
+func withReconcileLeaderRetry(t *testing.T, retryFor, retryStep time.Duration) {
+	t.Helper()
+	prevFor, prevStep := reconcileLeaderRetryFor, reconcileLeaderRetryStep
+	reconcileLeaderRetryFor = retryFor
+	reconcileLeaderRetryStep = retryStep
+	t.Cleanup(func() {
+		reconcileLeaderRetryFor = prevFor
+		reconcileLeaderRetryStep = prevStep
+	})
+}
 
+func TestAcquireReconcileLeader_WaitsForJetStream(t *testing.T) {
+	// Cold-start scenario: NATS is up but JetStream is not yet enabled.
+	// AcquireReconcileLeader must retry until JetStream comes online and
+	// then win the election, instead of failing immediately.
+	ns, nc := testutil.StartTestNATS(t)
+	withReconcileLeaderRetry(t, 3*time.Second, 50*time.Millisecond)
+
+	enableAt := time.Now().Add(300 * time.Millisecond)
+	go func() {
+		time.Sleep(time.Until(enableAt))
+		_ = ns.EnableJetStream(nil)
+	}()
+
+	start := time.Now()
 	release, elected := AcquireReconcileLeader(nc, "node-a")
-	assert.True(t, elected, "must fall through when KV unavailable")
-	assert.NotNil(t, release)
-	release() // no-op release on fallthrough path must not panic
+	elapsed := time.Since(start)
+
+	require.True(t, elected, "must win election after JetStream comes up")
+	require.NotNil(t, release)
+	defer release()
+	assert.GreaterOrEqual(t, elapsed, 300*time.Millisecond,
+		"must not return before JetStream is enabled")
+	assert.Less(t, elapsed, 3*time.Second, "must return within retry deadline")
+}
+
+func TestAcquireReconcileLeader_TimeoutReturnsNotElected(t *testing.T) {
+	// JetStream never enabled on this server — bounded retry must exhaust
+	// and return elected=false rather than fall through and run unguarded
+	// (mulga-js-72: concurrent reconciles produce silent OVN NB damage).
+	_, nc := testutil.StartTestNATS(t)
+	withReconcileLeaderRetry(t, 200*time.Millisecond, 20*time.Millisecond)
+
+	start := time.Now()
+	release, elected := AcquireReconcileLeader(nc, "node-a")
+	elapsed := time.Since(start)
+
+	assert.False(t, elected, "must skip reconcile when JetStream KV unreachable")
+	assert.Nil(t, release, "no release closure when not elected")
+	assert.GreaterOrEqual(t, elapsed, 200*time.Millisecond, "must wait full deadline")
+}
+
+func TestAcquireReconcileLeader_ConcurrentExactlyOneElected(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+
+	const goroutines = 5
+	var (
+		wg        sync.WaitGroup
+		electedCt atomic.Int32
+		releases  = make(chan func(), goroutines)
+	)
+	wg.Add(goroutines)
+	for i := range goroutines {
+		go func(id int) {
+			defer wg.Done()
+			release, elected := AcquireReconcileLeader(nc, "node-"+string(rune('a'+id)))
+			if elected {
+				electedCt.Add(1)
+				releases <- release
+			} else {
+				assert.Nil(t, release, "non-leader must return nil release")
+			}
+		}(i)
+	}
+	wg.Wait()
+	close(releases)
+
+	assert.Equal(t, int32(1), electedCt.Load(), "exactly one caller must win")
+	for r := range releases {
+		r()
+	}
 }
 
 func TestReconcile_NoBootstrap(t *testing.T) {