diff --git a/spinifex/services/vpcd/reconcile.go b/spinifex/services/vpcd/reconcile.go index b1e603432..cbe65bca2 100644 --- a/spinifex/services/vpcd/reconcile.go +++ b/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 { diff --git a/spinifex/services/vpcd/reconcile_test.go b/spinifex/services/vpcd/reconcile_test.go index ca602b7a9..cbbe619e2 100644 --- a/spinifex/services/vpcd/reconcile_test.go +++ b/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) {