fix: allocate mgmt IPs cluster-wide via NATS KV CAS (#576)

MgmtIPAllocator allocated from a process-local map, but br-mgmt is
trunked onto a shared physical NIC making the /24 one flat L2 segment
across the whole cluster. Every node independently allocated from
.10, causing collisions and ARP-race losers being blackholed.

Rewrite the allocator on the spinifex-cluster-state NATS KV bucket
using compare-and-swap, keyed by mgmt-ipam.<subnet>, modeled on IPAM.
Bind the KV handle in startCluster (not startLocal) to respect the
DDIL boundary. Serve already-allocated instance IDs from the local
cache when KV is unhealthy, but refuse new allocations until it
recovers. Rebuild now reconciles this node's running VMs into the
cluster-wide record instead of just refreshing the local cache.

spinifex/daemon/daemon.go

@@ -1615,8 +1615,13 @@ func (d *Daemon) startCluster() error {
 		return fmt.Errorf("restore instances: %w", err)
 	}
 
-	// Rebuild mgmt IP allocator so already-allocated IPs aren't reused.
-	if d.mgmtIPAllocator != nil {
+	// Bind the mgmt IP allocator to cluster KV now that JetStream is up (DDIL
+	// §1e-audit: startLocal built it KV-less). Rebuild then reconciles this
+	// node's already-running VMs into the cluster-wide record instead of
+	// just refreshing the local cache, so already-allocated IPs aren't
+	// reused by another node either.
+	if d.mgmtIPAllocator != nil && d.jsManager != nil {
+		d.mgmtIPAllocator.BindKV(d.jsManager, d.node)
 		d.mgmtIPAllocator.Rebuild(d.vmMgr.SnapshotMap())
 		slog.Info("Rebuilt mgmt IP allocator from restored instances", "allocated", d.mgmtIPAllocator.AllocatedCount())
 	}

spinifex/daemon/daemon_mgmt_ip.go

@@ -1,27 +1,107 @@
 package daemon
 
 import (
+	"errors"
 	"fmt"
 	"log/slog"
 	"net"
 	"sync"
 
 	"github.com/mulgadc/spinifex/spinifex/vm"
+	"github.com/nats-io/nats.go"
 )
 
-// MgmtIPAllocator manages static IP allocation for management NICs on system instances.
-// IPs are derived from the host's management bridge subnet (.10–.249 range).
-// Thread-safe via its own mutex — LaunchSystemInstance does not hold a global lock.
+// mgmtIPAMKeyPrefix namespaces mgmt-IP allocation records within the shared
+// spinifex-cluster-state KV bucket (see JetStreamManager.UpdateMgmtIPAM). One
+// record per management subnet, so clusters running distinct br-mgmt CIDRs
+// never contend on the same key.
+const mgmtIPAMKeyPrefix = "mgmt-ipam."
+
+// mgmtCASOuterRetries bounds the outer retry loop Allocate/Release/Rebuild
+// wrap around UpdateMgmtIPAM. casUpdate already retries a revision conflict
+// internally (maxCASRetries), but that window only absorbs one overlapping
+// writer per call; a burst of many nodes or VMs allocating on the same
+// subnet at once (e.g. several system instances launched together at
+// cluster bootstrap) can exhaust it. Retrying the whole read-mutate-write
+// cycle from a fresh Get here converges on success under that contention
+// instead of surfacing a transient conflict as a hard allocation failure.
+const mgmtCASOuterRetries = 20
+
+// updateMgmtIPAMWithRetry wraps jsManager.UpdateMgmtIPAM with the outer
+// retry described above. Only a revision-conflict exhaustion is retried —
+// any other error (KV not initialized, context errors, etc.) is returned
+// immediately.
+func updateMgmtIPAMWithRetry(jsManager *JetStreamManager, subnet string, mutate func(*MgmtIPRecord), createIfAbsent bool) (*MgmtIPRecord, error) {
+	var lastErr error
+	for range mgmtCASOuterRetries {
+		record, err := jsManager.UpdateMgmtIPAM(subnet, mutate, createIfAbsent)
+		if err == nil {
+			return record, nil
+		}
+		if !errors.Is(err, nats.ErrKeyExists) {
+			return nil, err
+		}
+		lastErr = err
+	}
+	return nil, fmt.Errorf("mgmt IPAM update for subnet %s exhausted %d outer retries under contention: %w",
+		subnet, mgmtCASOuterRetries, lastErr)
+}
+
+// MgmtIPEntry records one management IP allocation: the address, the
+// instance it belongs to, and the node that made the allocation. Node is
+// carried for audit and for a future reclaim sweep against dead nodes'
+// heartbeats — it plays no part in allocation itself, which must consider
+// every entry regardless of which node owns it.
+type MgmtIPEntry struct {
+	IP         string `json:"ip"`
+	InstanceID string `json:"instance_id"`
+	Node       string `json:"node"`
+}
+
+// MgmtIPRecord is the KV-CAS record for one management subnet. br-mgmt is
+// trunked onto a shared physical NIC on the tofu/ansible/baremetal cluster
+// paths, so the /24 it lives on is one flat L2 segment across every host —
+// every node allocating on that subnet reads and writes this same record.
+type MgmtIPRecord struct {
+	Subnet    string        `json:"subnet"` // e.g. "10.28.8.0/24"
+	Allocated []MgmtIPEntry `json:"allocated"`
+}
+
+// MgmtIPAllocator manages static IP allocation for management NICs on
+// system instances. IPs are derived from the host's management bridge
+// subnet (.10-.249 range).
+//
+// Allocation is backed by a cluster-wide NATS KV compare-and-swap record
+// once BindKV has been called, so nodes sharing the same br-mgmt L2 segment
+// never hand out the same address (mulga-f3j2x): every node independently
+// scanning from .10 was the bug, since br-mgmt is one flat subnet across the
+// whole cluster, not a per-host range.
+//
+// allocated is a write-through local cache mirroring this node's view of the
+// KV record, so AllocatedCount and lookups for an already-known instance ID
+// keep working even when KV is unreachable. Allocate refuses to mint a NEW
+// address in that case — this node's view of the cluster-wide record may be
+// stale — but a cached instance ID and Release both still work locally.
 type MgmtIPAllocator struct {
 	mu        sync.Mutex
-	allocated map[string]string // instanceID → IP
+	allocated map[string]string // instanceID -> IP (write-through cache)
 	baseIP    net.IP            // network base (e.g. 10.15.8.0)
+	subnet    string            // "10.15.8.0/24", the CAS record's key suffix
 	rangeMin  byte              // first host byte (10)
 	rangeMax  byte              // last host byte (249)
+
+	// jsManager / node are bound by BindKV once the daemon reaches
+	// startCluster. Both stay nil/empty through startLocal — the DDIL
+	// boundary there (daemon.go's assertNoClusterServicesInitialised) means
+	// startLocal must never touch JetStream KV, so the allocator constructed
+	// there is local-cache-only until BindKV is called.
+	jsManager *JetStreamManager
+	node      string
 }
 
 // NewMgmtIPAllocator creates an allocator for the /24 subnet of bridgeIP.
-// Allocates from .10 to .249 (240 addresses).
+// Allocates from .10 to .249 (240 addresses). The allocator is KV-less until
+// BindKV is called; see the DDIL boundary note above.
 func NewMgmtIPAllocator(bridgeIP string) (*MgmtIPAllocator, error) {
 	ip := net.ParseIP(bridgeIP)
 	if ip == nil {
@@ -40,70 +120,205 @@ func NewMgmtIPAllocator(bridgeIP string) (*MgmtIPAllocator, error) {
 	return &MgmtIPAllocator{
 		allocated: make(map[string]string),
 		baseIP:    base,
+		subnet:    fmt.Sprintf("%d.%d.%d.0/24", base[0], base[1], base[2]),
 		rangeMin:  10,
 		rangeMax:  249,
 	}, nil
 }
 
-// Allocate assigns a management IP to the given instance.
-// Returns the allocated IP string (e.g. "10.15.8.10").
-func (a *MgmtIPAllocator) Allocate(instanceID string) (string, error) {
+// BindKV attaches the cluster-wide KV backing store. Called once from
+// daemon.startCluster after JetStream is up — never from startLocal (DDIL
+// §1e-audit). node identifies this node's entries in the shared record for
+// audit; it plays no part in allocation, which considers every entry in the
+// record regardless of which node owns it.
+func (a *MgmtIPAllocator) BindKV(jsManager *JetStreamManager, node string) {
 	a.mu.Lock()
 	defer a.mu.Unlock()
+	a.jsManager = jsManager
+	a.node = node
+}
 
-	// Check if already allocated
+// Allocate assigns a management IP to the given instance.
+// Returns the allocated IP string (e.g. "10.15.8.10").
+//
+// A known instance ID is served from the local cache without touching KV, so
+// a node that already owns an address can always answer for it. A brand new
+// allocation requires a healthy cluster KV — this node's cache cannot prove
+// an address is free cluster-wide on its own — and is refused otherwise.
+func (a *MgmtIPAllocator) Allocate(instanceID string) (string, error) {
+	a.mu.Lock()
 	if ip, ok := a.allocated[instanceID]; ok {
+		a.mu.Unlock()
 		return ip, nil
 	}
+	jsManager, node, subnet := a.jsManager, a.node, a.subnet
+	a.mu.Unlock()
 
-	// Build set of used IPs for fast lookup
-	used := make(map[string]struct{}, len(a.allocated))
-	for _, ip := range a.allocated {
-		used[ip] = struct{}{}
+	if jsManager == nil || !jsManager.KVHealthy() {
+		return "", fmt.Errorf("mgmt IP allocation unavailable: cluster state KV unreachable for subnet %s", subnet)
 	}
 
-	// Find first free IP in range
-	for i := a.rangeMin; i <= a.rangeMax; i++ {
-		candidate := fmt.Sprintf("%d.%d.%d.%d", a.baseIP[0], a.baseIP[1], a.baseIP[2], i)
+	// mutate has no error return (casUpdate's contract), so exhaustion and
+	// idempotent hits are threaded back out through these closed-over
+	// locals. On either path mutate leaves the record unchanged, so the
+	// CAS write that follows is a harmless no-op against the same revision.
+	var allocatedIP string
+	var mutateErr error
+	_, err := updateMgmtIPAMWithRetry(jsManager, subnet, func(r *MgmtIPRecord) {
+		if r.Subnet == "" {
+			r.Subnet = subnet
+		}
+
+		// Idempotent: an instance that already holds an address (allocated
+		// by this node earlier, or by another node before a restart lost
+		// the local cache) keeps it rather than taking a second one.
+		for _, e := range r.Allocated {
+			if e.InstanceID == instanceID {
+				allocatedIP = e.IP
+				return
+			}
+		}
+
+		ip, findErr := nextFreeIP(a.baseIP, a.rangeMin, a.rangeMax, r.Allocated)
+		if findErr != nil {
+			mutateErr = findErr
+			return
+		}
+		r.Allocated = append(r.Allocated, MgmtIPEntry{IP: ip, InstanceID: instanceID, Node: node})
+		allocatedIP = ip
+	}, true)
+	if err != nil {
+		return "", fmt.Errorf("mgmt IP CAS update for subnet %s: %w", subnet, err)
+	}
+	if mutateErr != nil {
+		return "", mutateErr
+	}
+
+	a.mu.Lock()
+	a.allocated[instanceID] = allocatedIP
+	a.mu.Unlock()
+
+	slog.Debug("Management IP allocated", "instance", instanceID, "ip", allocatedIP, "node", node)
+	return allocatedIP, nil
+}
+
+// nextFreeIP scans rangeMin-rangeMax on the /24 rooted at baseIP, skipping
+// every IP already present in allocated regardless of which entry's Node it
+// belongs to — that is the whole fix for mulga-f3j2x: the range is shared
+// across the cluster, not partitioned per host.
+func nextFreeIP(baseIP net.IP, rangeMin, rangeMax byte, allocated []MgmtIPEntry) (string, error) {
+	used := make(map[string]struct{}, len(allocated))
+	for _, e := range allocated {
+		used[e.IP] = struct{}{}
+	}
+
+	for i := rangeMin; i <= rangeMax; i++ {
+		candidate := fmt.Sprintf("%d.%d.%d.%d", baseIP[0], baseIP[1], baseIP[2], i)
 		if _, taken := used[candidate]; !taken {
-			a.allocated[instanceID] = candidate
-			slog.Debug("Management IP allocated", "instance", instanceID, "ip", candidate)
 			return candidate, nil
 		}
 	}
 
-	return "", fmt.Errorf("management IP range exhausted (%d.%d.%d.%d–%d.%d.%d.%d)",
-		a.baseIP[0], a.baseIP[1], a.baseIP[2], a.rangeMin,
-		a.baseIP[0], a.baseIP[1], a.baseIP[2], a.rangeMax)
+	return "", fmt.Errorf("management IP range exhausted across the cluster (%d.%d.%d.%d-%d.%d.%d.%d, %d addresses)",
+		baseIP[0], baseIP[1], baseIP[2], rangeMin,
+		baseIP[0], baseIP[1], baseIP[2], rangeMax,
+		int(rangeMax-rangeMin)+1)
 }
 
-// Release frees the management IP for the given instance.
+// Release frees the management IP for the given instance, cluster-wide.
+// Best-effort against KV: the local cache is always updated so this node
+// never reoffers the address, but a KV write failure is logged rather than
+// returned — callers (LaunchSystemInstance rollback, CleanupMgmtNetwork)
+// cannot act on a Release error, matching the pre-existing signature.
 func (a *MgmtIPAllocator) Release(instanceID string) {
 	a.mu.Lock()
-	defer a.mu.Unlock()
+	ip, hadLocal := a.allocated[instanceID]
+	delete(a.allocated, instanceID)
+	jsManager, subnet := a.jsManager, a.subnet
+	a.mu.Unlock()
 
-	if ip, ok := a.allocated[instanceID]; ok {
-		delete(a.allocated, instanceID)
-		slog.Debug("Management IP released", "instance", instanceID, "ip", ip)
+	if hadLocal {
+		slog.Debug("Management IP released locally", "instance", instanceID, "ip", ip)
+	}
+
+	if jsManager == nil {
+		return
+	}
+
+	// createIfAbsent=false: releasing an instance this node never allocated
+	// (or that no node has a record for) must not conjure an empty record
+	// into existence — nats.ErrKeyNotFound just means there was nothing to
+	// release.
+	_, err := updateMgmtIPAMWithRetry(jsManager, subnet, func(r *MgmtIPRecord) {
+		for i, e := range r.Allocated {
+			if e.InstanceID == instanceID {
+				r.Allocated = append(r.Allocated[:i], r.Allocated[i+1:]...)
+				return
+			}
+		}
+	}, false)
+	if err != nil && !errors.Is(err, nats.ErrKeyNotFound) {
+		slog.Warn("Failed to release mgmt IP in cluster KV; address remains reserved until reconciled",
+			"instance", instanceID, "err", err)
 	}
 }
 
-// Rebuild reconstructs the allocated set from existing VMs.
-// Called on daemon startup after loading VMs from KV store.
+// Rebuild reconciles the allocator against vms (this node's own VM
+// snapshot). It always refreshes the local write-through cache first, then
+// — once KV is bound (see BindKV) and healthy — CAS-inserts an entry for
+// every local VM carrying a MgmtIP, idempotent by instance ID. This never
+// removes another node's entries: reclaiming addresses of a dead node is a
+// separate concern (tracked apart from mulga-f3j2x).
+//
+// Called on daemon startup (startLocal, cache-only since KV isn't bound yet)
+// and again after cluster bootstrap (startCluster, once BindKV has run) so
+// already-allocated IPs are never handed out a second time.
 func (a *MgmtIPAllocator) Rebuild(vms map[string]*vm.VM) {
 	a.mu.Lock()
-	defer a.mu.Unlock()
-
 	for id, v := range vms {
 		if v.MgmtIP != "" {
 			a.allocated[id] = v.MgmtIP
 		}
 	}
+	count := len(a.allocated)
+	jsManager, node, subnet := a.jsManager, a.node, a.subnet
+	a.mu.Unlock()
 
-	slog.Info("Management IP allocator rebuilt", "count", len(a.allocated))
+	slog.Info("Management IP allocator rebuilt", "count", count)
+
+	if jsManager == nil {
+		// startLocal: no cluster KV yet, cache-only rebuild.
+		return
+	}
+	if !jsManager.KVHealthy() {
+		slog.Warn("Skipping mgmt IP KV reconcile: cluster state KV unreachable", "subnet", subnet)
+		return
+	}
+
+	for id, v := range vms {
+		if v.MgmtIP == "" {
+			continue
+		}
+		mgmtIP := v.MgmtIP
+		_, err := updateMgmtIPAMWithRetry(jsManager, subnet, func(r *MgmtIPRecord) {
+			if r.Subnet == "" {
+				r.Subnet = subnet
+			}
+			for _, e := range r.Allocated {
+				if e.InstanceID == id {
+					return // already reconciled
+				}
+			}
+			r.Allocated = append(r.Allocated, MgmtIPEntry{IP: mgmtIP, InstanceID: id, Node: node})
+		}, true)
+		if err != nil {
+			slog.Warn("Failed to reconcile mgmt IP into cluster KV", "instance", id, "ip", mgmtIP, "err", err)
+		}
+	}
 }
 
-// AllocatedCount returns the number of currently allocated IPs.
+// AllocatedCount returns the number of currently allocated IPs known to this
+// node's local cache.
 func (a *MgmtIPAllocator) AllocatedCount() int {
 	a.mu.Lock()
 	defer a.mu.Unlock()

spinifex/daemon/daemon_mgmt_ip_test.go

@@ -2,12 +2,46 @@ package daemon
 
 import (
 	"fmt"
+	"strings"
 	"sync"
 	"testing"
 
 	"github.com/mulgadc/spinifex/spinifex/vm"
+	"github.com/nats-io/nats.go"
 )
 
+// newTestMgmtJSM returns a JetStreamManager backed by the shared per-package
+// JetStream test server, with the cluster-state bucket initialised — the
+// same bucket UpdateMgmtIPAM writes into (namespaced under "mgmt-ipam.").
+func newTestMgmtJSM(t *testing.T) *JetStreamManager {
+	t.Helper()
+	nc, err := nats.Connect(sharedJSNATSURL)
+	if err != nil {
+		t.Fatalf("connect NATS: %v", err)
+	}
+	t.Cleanup(nc.Close)
+
+	jsm, err := NewJetStreamManager(nc, 1)
+	if err != nil {
+		t.Fatalf("new JetStreamManager: %v", err)
+	}
+	if err := jsm.InitClusterStateBucket(); err != nil {
+		t.Fatalf("init cluster state bucket: %v", err)
+	}
+	return jsm
+}
+
+// cleanupMgmtIPAM deletes the mgmt-ipam record for a's subnet at test end.
+// The cluster-state bucket is shared by the whole package's test binary, so
+// without this a record left behind by one test (or one -count repetition)
+// would leak into the next test/run that reuses the same bridge subnet.
+func cleanupMgmtIPAM(t *testing.T, jsm *JetStreamManager, a *MgmtIPAllocator) {
+	t.Helper()
+	t.Cleanup(func() {
+		_ = jsm.clusterKV.Delete(mgmtIPAMKeyPrefix + a.subnet)
+	})
+}
+
 func TestNewMgmtIPAllocator(t *testing.T) {
 	tests := []struct {
 		name      string
@@ -40,6 +74,10 @@ func TestNewMgmtIPAllocator(t *testing.T) {
 	}
 }
 
+// TestMgmtIPAllocator_Allocate exercises the same scan-from-.10 contract as
+// before the KV rewrite, but now through a bound cluster KV — Allocate
+// refuses new addresses without one (see fail-closed tests below), so every
+// allocator under test here binds one first.
 func TestMgmtIPAllocator_Allocate(t *testing.T) {
 	tests := []struct {
 		name     string
@@ -52,10 +90,13 @@ func TestMgmtIPAllocator_Allocate(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			jsm := newTestMgmtJSM(t)
 			a, err := NewMgmtIPAllocator(tt.bridgeIP)
 			if err != nil {
 				t.Fatal(err)
 			}
+			a.BindKV(jsm, "node-a")
+			cleanupMgmtIPAM(t, jsm, a)
 
 			ip, err := a.Allocate("i-first")
 			if err != nil {
@@ -90,10 +131,13 @@ func TestMgmtIPAllocator_Allocate(t *testing.T) {
 }
 
 func TestMgmtIPAllocator_Release(t *testing.T) {
-	a, err := NewMgmtIPAllocator("10.15.8.1")
+	jsm := newTestMgmtJSM(t)
+	a, err := NewMgmtIPAllocator("10.16.8.1")
 	if err != nil {
 		t.Fatal(err)
 	}
+	a.BindKV(jsm, "node-a")
+	cleanupMgmtIPAM(t, jsm, a)
 
 	a.Allocate("i-one")
 	a.Allocate("i-two")
@@ -108,11 +152,14 @@ func TestMgmtIPAllocator_Release(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if ip != "10.15.8.10" {
-		t.Errorf("reused IP = %q, want 10.15.8.10", ip)
+	if ip != "10.16.8.10" {
+		t.Errorf("reused IP = %q, want 10.16.8.10", ip)
 	}
 }
 
+// TestMgmtIPAllocator_ReleaseNonexistent asserts Release never panics or
+// blocks on an instance ID with no allocation — including with no KV bound
+// at all, the shape startLocal constructs.
 func TestMgmtIPAllocator_ReleaseNonexistent(t *testing.T) {
 	a, err := NewMgmtIPAllocator("10.15.8.1")
 	if err != nil {
@@ -123,12 +170,15 @@ func TestMgmtIPAllocator_ReleaseNonexistent(t *testing.T) {
 }
 
 func TestMgmtIPAllocator_Exhaustion(t *testing.T) {
-	a, err := NewMgmtIPAllocator("10.15.8.1")
+	jsm := newTestMgmtJSM(t)
+	a, err := NewMgmtIPAllocator("10.17.8.1")
 	if err != nil {
 		t.Fatal(err)
 	}
+	a.BindKV(jsm, "node-a")
+	cleanupMgmtIPAM(t, jsm, a)
 
-	// Fill all 240 slots (.10–.249)
+	// Fill all 240 slots (.10-.249)
 	for i := range 240 {
 		_, err := a.Allocate(fmt.Sprintf("i-%d", i))
 		if err != nil {
@@ -136,11 +186,14 @@ func TestMgmtIPAllocator_Exhaustion(t *testing.T) {
 		}
 	}
 
-	// Next should fail
+	// Next should fail, naming the cluster rather than a host-local limit.
 	_, err = a.Allocate("i-overflow")
 	if err == nil {
 		t.Fatal("expected exhaustion error")
 	}
+	if !strings.Contains(err.Error(), "exhausted across the cluster") {
+		t.Errorf("exhaustion error = %q, want it to name the cluster", err.Error())
+	}
 
 	// Release one, then it should work
 	a.Release("i-0")
@@ -148,22 +201,28 @@ func TestMgmtIPAllocator_Exhaustion(t *testing.T) {
 	if err != nil {
 		t.Fatalf("after release: %v", err)
 	}
-	if ip != "10.15.8.10" {
-		t.Errorf("reused IP = %q, want 10.15.8.10", ip)
+	if ip != "10.17.8.10" {
+		t.Errorf("reused IP = %q, want 10.17.8.10", ip)
 	}
 }
 
+// TestMgmtIPAllocator_Rebuild binds KV before Rebuild so it exercises the
+// reconcile path (CAS-inserting local VMs into the shared record), not just
+// the local-cache refresh that runs unbound in startLocal.
 func TestMgmtIPAllocator_Rebuild(t *testing.T) {
-	a, err := NewMgmtIPAllocator("10.15.8.1")
+	jsm := newTestMgmtJSM(t)
+	a, err := NewMgmtIPAllocator("10.18.8.1")
 	if err != nil {
 		t.Fatal(err)
 	}
+	a.BindKV(jsm, "node-a")
+	cleanupMgmtIPAM(t, jsm, a)
 
 	vms := map[string]*vm.VM{
-		"i-a": {MgmtIP: "10.15.8.10"},
-		"i-b": {MgmtIP: "10.15.8.15"},
+		"i-a": {MgmtIP: "10.18.8.10"},
+		"i-b": {MgmtIP: "10.18.8.15"},
 		"i-c": {MgmtIP: ""}, // no mgmt NIC
-		"i-d": {MgmtIP: "10.15.8.20"},
+		"i-d": {MgmtIP: "10.18.8.20"},
 	}
 
 	a.Rebuild(vms)
@@ -177,16 +236,46 @@ func TestMgmtIPAllocator_Rebuild(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if ip != "10.15.8.11" {
-		t.Errorf("next IP after rebuild = %q, want 10.15.8.11", ip)
+	if ip != "10.18.8.11" {
+		t.Errorf("next IP after rebuild = %q, want 10.18.8.11", ip)
+	}
+}
+
+// TestMgmtIPAllocator_Rebuild_UnboundIsCacheOnly is the startLocal shape:
+// Rebuild with no KV bound must only refresh the local cache and must never
+// panic or block waiting on a cluster that doesn't exist yet.
+func TestMgmtIPAllocator_Rebuild_UnboundIsCacheOnly(t *testing.T) {
+	a, err := NewMgmtIPAllocator("10.19.8.1")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	vms := map[string]*vm.VM{
+		"i-a": {MgmtIP: "10.19.8.10"},
+	}
+	a.Rebuild(vms)
+
+	if a.AllocatedCount() != 1 {
+		t.Errorf("count after unbound rebuild = %d, want 1", a.AllocatedCount())
+	}
+	// A cached ID still resolves without KV.
+	ip, err := a.Allocate("i-a")
+	if err != nil {
+		t.Fatalf("cached instance should resolve without KV: %v", err)
+	}
+	if ip != "10.19.8.10" {
+		t.Errorf("cached IP = %q, want 10.19.8.10", ip)
 	}
 }
 
 func TestMgmtIPAllocator_Concurrent(t *testing.T) {
-	a, err := NewMgmtIPAllocator("10.15.8.1")
+	jsm := newTestMgmtJSM(t)
+	a, err := NewMgmtIPAllocator("10.20.8.1")
 	if err != nil {
 		t.Fatal(err)
 	}
+	a.BindKV(jsm, "node-a")
+	cleanupMgmtIPAM(t, jsm, a)
 
 	var wg sync.WaitGroup
 	results := make(map[string]string)
@@ -226,3 +315,304 @@ func TestMgmtIPAllocator_Concurrent(t *testing.T) {
 		seen[ip] = true
 	}
 }
+
+// --- Cluster-safety coverage (mulga-f3j2x) ---
+
+// TestMgmtIPAllocator_SharedKV_NoDuplicates is the core regression test:
+// two allocators standing in for two nodes, sharing one KV, must never both
+// hand out the same address even though each scans from .10 independently.
+func TestMgmtIPAllocator_SharedKV_NoDuplicates(t *testing.T) {
+	jsm := newTestMgmtJSM(t)
+
+	node1, err := NewMgmtIPAllocator("10.21.8.1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	node1.BindKV(jsm, "node-1")
+	cleanupMgmtIPAM(t, jsm, node1)
+
+	node2, err := NewMgmtIPAllocator("10.21.8.2")
+	if err != nil {
+		t.Fatal(err)
+	}
+	node2.BindKV(jsm, "node-2")
+
+	seen := make(map[string]string) // ip -> owning instance
+	allocate := func(a *MgmtIPAllocator, id string) {
+		ip, err := a.Allocate(id)
+		if err != nil {
+			t.Fatalf("allocate %s: %v", id, err)
+		}
+		if owner, dup := seen[ip]; dup {
+			t.Fatalf("duplicate address %s: held by %s and %s", ip, owner, id)
+		}
+		seen[ip] = id
+	}
+
+	// Alternate allocation across the two "nodes" — each locally believes it
+	// is scanning a fresh range from .10, but the shared KV record must
+	// still serialise them onto distinct addresses.
+	for i := range 10 {
+		allocate(node1, fmt.Sprintf("node1-i-%d", i))
+		allocate(node2, fmt.Sprintf("node2-i-%d", i))
+	}
+
+	if len(seen) != 20 {
+		t.Errorf("got %d unique addresses, want 20", len(seen))
+	}
+}
+
+// TestMgmtIPAllocator_ConcurrentAcrossNodes fires concurrent allocations
+// from three simulated nodes against the same KV record, forcing genuine
+// CAS conflicts (not just the happy path) as their Get/mutate/Update
+// windows overlap. Every address handed out across all three must be
+// unique — this is the exact shape of CI run 29912697556, where independent
+// per-node allocation raced onto the same br-mgmt address.
+func TestMgmtIPAllocator_ConcurrentAcrossNodes(t *testing.T) {
+	jsm := newTestMgmtJSM(t)
+
+	const nodeCount = 3
+	const perNode = 20
+	nodes := make([]*MgmtIPAllocator, nodeCount)
+	for n := range nodeCount {
+		a, err := NewMgmtIPAllocator("10.22.8.1")
+		if err != nil {
+			t.Fatal(err)
+		}
+		a.BindKV(jsm, fmt.Sprintf("node-%d", n))
+		if n == 0 {
+			cleanupMgmtIPAM(t, jsm, a)
+		}
+		nodes[n] = a
+	}
+
+	type result struct {
+		id  string
+		ip  string
+		err error
+	}
+	results := make(chan result, nodeCount*perNode)
+	var wg sync.WaitGroup
+	for n := range nodeCount {
+		for i := range perNode {
+			wg.Add(1)
+			go func(n, i int) {
+				defer wg.Done()
+				id := fmt.Sprintf("node%d-i-%d", n, i)
+				ip, err := nodes[n].Allocate(id)
+				results <- result{id: id, ip: ip, err: err}
+			}(n, i)
+		}
+	}
+	wg.Wait()
+	close(results)
+
+	seen := make(map[string]string)
+	for r := range results {
+		if r.err != nil {
+			t.Fatalf("allocate %s: %v", r.id, r.err)
+		}
+		if owner, dup := seen[r.ip]; dup {
+			t.Fatalf("duplicate address %s: held by %s and %s", r.ip, owner, r.id)
+		}
+		seen[r.ip] = r.id
+	}
+	if len(seen) != nodeCount*perNode {
+		t.Errorf("got %d unique addresses, want %d", len(seen), nodeCount*perNode)
+	}
+}
+
+// TestMgmtIPAllocator_Allocate_IdempotentAcrossNodes proves idempotency is a
+// property of the shared KV record, not of one allocator's local cache: a
+// second allocator (standing in for the instance's owning node restarting
+// inside a partition and losing its cache) re-requesting a known instance ID
+// gets back the same address rather than a new one.
+func TestMgmtIPAllocator_Allocate_IdempotentAcrossNodes(t *testing.T) {
+	jsm := newTestMgmtJSM(t)
+
+	node1, err := NewMgmtIPAllocator("10.23.8.1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	node1.BindKV(jsm, "node-1")
+	cleanupMgmtIPAM(t, jsm, node1)
+
+	ip, err := node1.Allocate("i-reattach")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// A second allocator, simulating a fresh process on the same node after
+	// a restart with an empty local cache.
+	node1Restarted, err := NewMgmtIPAllocator("10.23.8.1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	node1Restarted.BindKV(jsm, "node-1")
+
+	ip2, err := node1Restarted.Allocate("i-reattach")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if ip2 != ip {
+		t.Errorf("re-allocate from fresh allocator = %q, want %q (idempotent by instance ID)", ip2, ip)
+	}
+}
+
+// TestMgmtIPAllocator_Release_FreesAddressClusterWide proves Release's
+// effect is visible cluster-wide: a second allocator, not the one that
+// released the address, can immediately take it.
+func TestMgmtIPAllocator_Release_FreesAddressClusterWide(t *testing.T) {
+	jsm := newTestMgmtJSM(t)
+
+	node1, err := NewMgmtIPAllocator("10.24.8.1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	node1.BindKV(jsm, "node-1")
+	cleanupMgmtIPAM(t, jsm, node1)
+
+	node2, err := NewMgmtIPAllocator("10.24.8.2")
+	if err != nil {
+		t.Fatal(err)
+	}
+	node2.BindKV(jsm, "node-2")
+
+	ip, err := node1.Allocate("i-node1-only")
+	if err != nil {
+		t.Fatal(err)
+	}
+	node1.Release("i-node1-only")
+
+	// node2 must now be able to take the address node1 just released.
+	ip2, err := node2.Allocate("i-node2-new")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if ip2 != ip {
+		t.Errorf("node2 got %q, want the released address %q", ip2, ip)
+	}
+}
+
+// TestMgmtIPAllocator_Exhaustion_ClusterWide proves the exhaustion check is
+// against the cluster-wide record, not this allocator's own local cache:
+// two allocators together hold all 240 addresses (120 each, so neither
+// locally believes the range is full) and a third allocation from either
+// must still fail.
+func TestMgmtIPAllocator_Exhaustion_ClusterWide(t *testing.T) {
+	jsm := newTestMgmtJSM(t)
+
+	node1, err := NewMgmtIPAllocator("10.25.8.1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	node1.BindKV(jsm, "node-1")
+	cleanupMgmtIPAM(t, jsm, node1)
+
+	node2, err := NewMgmtIPAllocator("10.25.8.2")
+	if err != nil {
+		t.Fatal(err)
+	}
+	node2.BindKV(jsm, "node-2")
+
+	for i := range 120 {
+		if _, err := node1.Allocate(fmt.Sprintf("node1-i-%d", i)); err != nil {
+			t.Fatalf("node1 allocation %d failed: %v", i, err)
+		}
+	}
+	for i := range 120 {
+		if _, err := node2.Allocate(fmt.Sprintf("node2-i-%d", i)); err != nil {
+			t.Fatalf("node2 allocation %d failed: %v", i, err)
+		}
+	}
+
+	// Each allocator's local cache only knows about its own 120 — but the
+	// cluster-wide record holds all 240, so the next allocation from either
+	// must fail.
+	if node1.AllocatedCount() != 120 {
+		t.Fatalf("node1 local cache = %d, want 120 (proves the check isn't local)", node1.AllocatedCount())
+	}
+	if _, err := node1.Allocate("i-overflow-from-node1"); err == nil {
+		t.Fatal("expected exhaustion error from node1")
+	}
+	if _, err := node2.Allocate("i-overflow-from-node2"); err == nil {
+		t.Fatal("expected exhaustion error from node2")
+	}
+}
+
+// TestMgmtIPAllocator_Allocate_RefusedWhenKVUnhealthy is the fail-closed
+// contract: once KV is unreachable, a brand new allocation is refused, but
+// an instance ID already resolved (and cached) before the outage keeps
+// resolving from the local cache without touching KV at all.
+func TestMgmtIPAllocator_Allocate_RefusedWhenKVUnhealthy(t *testing.T) {
+	nc, err := nats.Connect(sharedJSNATSURL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	jsm, err := NewJetStreamManager(nc, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := jsm.InitClusterStateBucket(); err != nil {
+		t.Fatal(err)
+	}
+
+	a, err := NewMgmtIPAllocator("10.26.8.1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	a.BindKV(jsm, "node-1")
+
+	ip, err := a.Allocate("i-known")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Sever the connection backing this JetStreamManager: KVHealthy's
+	// AccountInfo round-trip will now fail, simulating a partition.
+	nc.Close()
+
+	if jsm.KVHealthy() {
+		t.Fatal("expected KVHealthy to report false after closing the connection")
+	}
+
+	// A known instance ID still resolves — served entirely from the local
+	// cache, no KV round-trip required.
+	cachedIP, err := a.Allocate("i-known")
+	if err != nil {
+		t.Fatalf("cached instance should resolve without KV: %v", err)
+	}
+	if cachedIP != ip {
+		t.Errorf("cached IP = %q, want %q", cachedIP, ip)
+	}
+
+	// A brand new allocation must be refused rather than risk a duplicate
+	// against a cluster-wide record this node can no longer trust.
+	if _, err := a.Allocate("i-new-during-partition"); err == nil {
+		t.Fatal("expected allocation to be refused while KV is unhealthy")
+	}
+}
+
+// TestMgmtIPAllocator_Allocate_RefusedWhenKVNeverBound is the startLocal
+// shape: an allocator with no KV bound at all (as constructed before
+// startCluster runs) must refuse new allocations exactly like an unhealthy
+// bound KV, while still serving whatever Rebuild already cached.
+func TestMgmtIPAllocator_Allocate_RefusedWhenKVNeverBound(t *testing.T) {
+	a, err := NewMgmtIPAllocator("10.27.8.1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	a.Rebuild(map[string]*vm.VM{"i-known": {MgmtIP: "10.27.8.10"}})
+
+	ip, err := a.Allocate("i-known")
+	if err != nil {
+		t.Fatalf("cached instance should resolve without KV: %v", err)
+	}
+	if ip != "10.27.8.10" {
+		t.Errorf("cached IP = %q, want 10.27.8.10", ip)
+	}
+
+	if _, err := a.Allocate("i-brand-new"); err == nil {
+		t.Fatal("expected allocation to be refused with no KV bound")
+	}
+}

spinifex/daemon/jetstream.go

@@ -478,6 +478,22 @@ func (m *JetStreamManager) UpdateClusterShutdown(mutate func(*ClusterShutdownSta
 	return casUpdate(m.clusterKV, "cluster.shutdown", mutate, false)
 }
 
+// UpdateMgmtIPAM atomically applies mutate to the mgmt-ipam record for
+// subnet and writes it back with optimistic concurrency, retrying on a
+// concurrent writer's revision conflict. createIfAbsent lets the first
+// allocation for a subnet create its record; releases pass false so a
+// release of an instance nobody holds an address for does not conjure an
+// empty record into existence. Every node sharing the management bridge's
+// L2 segment reads and writes the same record, keyed by subnet, which is
+// what makes mgmt IP allocation safe across the whole cluster rather than
+// just this process (mulga-f3j2x).
+func (m *JetStreamManager) UpdateMgmtIPAM(subnet string, mutate func(*MgmtIPRecord), createIfAbsent bool) (*MgmtIPRecord, error) {
+	if m.clusterKV == nil {
+		return nil, errors.New("cluster state KV not initialized")
+	}
+	return casUpdate(m.clusterKV, mgmtIPAMKeyPrefix+subnet, mutate, createIfAbsent)
+}
+
 // ReadClusterShutdown reads the cluster shutdown state from KV.
 func (m *JetStreamManager) ReadClusterShutdown() (*ClusterShutdownState, error) {
 	if m.clusterKV == nil {

spinifex/daemon/jetstream_test.go

@@ -1634,3 +1634,90 @@ func TestJetStreamManager_BestEffort_NilObserver_NoPanic(t *testing.T) {
 	// No observer set — must not panic on success or failure paths.
 	jsm.WriteStateBytesBestEffort("obs-nil", []byte(`{"vms":{}}`), 5*time.Second)
 }
+
+// --- UpdateMgmtIPAM CAS tests ---
+
+// TestJetStreamManager_UpdateMgmtIPAM_ConflictRetry locks the CAS contract
+// for mgmt-ipam records: a concurrent writer landing between Get and Update
+// must be detected via a revision conflict and retried, and both the
+// injected concurrent write and the retried write must land — mirroring
+// UpdateStoppedInstance/UpdateTerminatedInstance's conflict-retry tests,
+// applied to the allocator's backing store (mulga-f3j2x).
+func TestJetStreamManager_UpdateMgmtIPAM_ConflictRetry(t *testing.T) {
+	nc, err := nats.Connect(sharedJSNATSURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	jsm, err := NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, jsm.InitClusterStateBucket())
+
+	const subnet = "10.99.8.0/24"
+	defer func() { _ = jsm.clusterKV.Delete(mgmtIPAMKeyPrefix + subnet) }()
+
+	_, err = jsm.UpdateMgmtIPAM(subnet, func(r *MgmtIPRecord) {
+		r.Subnet = subnet
+		r.Allocated = append(r.Allocated, MgmtIPEntry{IP: "10.99.8.10", InstanceID: "i-seed", Node: "node-seed"})
+	}, true)
+	require.NoError(t, err)
+
+	var attempts int
+	var injectOnce sync.Once
+	updated, err := jsm.UpdateMgmtIPAM(subnet, func(r *MgmtIPRecord) {
+		attempts++
+		r.Allocated = append(r.Allocated, MgmtIPEntry{IP: "10.99.8.11", InstanceID: "i-retried", Node: "node-a"})
+
+		// Simulate a concurrent writer landing between our Get and Update on
+		// the first attempt only, forcing exactly one revision conflict.
+		injectOnce.Do(func() {
+			_, cerr := jsm.UpdateMgmtIPAM(subnet, func(cr *MgmtIPRecord) {
+				cr.Allocated = append(cr.Allocated, MgmtIPEntry{IP: "10.99.8.12", InstanceID: "i-concurrent", Node: "node-b"})
+			}, true)
+			require.NoError(t, cerr)
+		})
+	}, true)
+	require.NoError(t, err, "the retried Update must eventually succeed against the fresh revision")
+	require.NotNil(t, updated)
+	assert.Equal(t, 2, attempts, "the injected concurrent write must force exactly one retry")
+
+	ids := make(map[string]bool, len(updated.Allocated))
+	for _, e := range updated.Allocated {
+		ids[e.InstanceID] = true
+	}
+	assert.True(t, ids["i-seed"], "the original entry must survive")
+	assert.True(t, ids["i-concurrent"], "the concurrent writer's entry must survive, not be clobbered")
+	assert.True(t, ids["i-retried"], "the retried update must also land")
+}
+
+// TestJetStreamManager_UpdateMgmtIPAM_NotFound verifies createIfAbsent=false
+// surfaces nats.ErrKeyNotFound instead of silently creating a record —
+// MgmtIPAllocator.Release relies on this to no-op cleanly when there is
+// nothing to release.
+func TestJetStreamManager_UpdateMgmtIPAM_NotFound(t *testing.T) {
+	nc, err := nats.Connect(sharedJSNATSURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	jsm, err := NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, jsm.InitClusterStateBucket())
+
+	_, err = jsm.UpdateMgmtIPAM("10.98.8.0/24-nonexistent", func(r *MgmtIPRecord) {}, false)
+	assert.ErrorIs(t, err, nats.ErrKeyNotFound)
+}
+
+// TestJetStreamManager_UpdateMgmtIPAM_ClusterKVNotInitialized tests error
+// handling when the cluster-state KV bucket has not been initialized.
+func TestJetStreamManager_UpdateMgmtIPAM_ClusterKVNotInitialized(t *testing.T) {
+	nc, err := nats.Connect(sharedJSNATSURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	jsm, err := NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	// Don't call InitClusterStateBucket
+
+	_, err = jsm.UpdateMgmtIPAM("10.97.8.0/24", func(r *MgmtIPRecord) {}, true)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "cluster state KV not initialized")
+}