From 6253c3d0ae4e2f7d9d3364ff191529ebdc8c4ccf Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Tue, 14 Jul 2026 00:32:38 +1000 Subject: [PATCH] fix(nats): create lazy KV buckets at cluster-size replicas, not R1 (#529) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit utils.GetOrCreateKVBucket hardcoded Replicas=1, so the ~38 handlers that create their KV buckets lazily (VPC/subnet/ENI/EIP/ELBv2/ECS/ECR/ACM/ natgw/snapshot/placementgroup/IPAM/account-settings/dhcp/static-pool …) created them at R1 on a multi-node cluster. upgradeJetStreamReplicas runs once at startup, so any bucket created lazily after boot stayed R1 permanently — and an R1 stream's single Raft replica is transiently leaderless during formation/join/rebalance, surfacing as ErrNoResponders/ErrTimeout on KV ops. Add a package-level default replica count that the daemon sets once at boot (startCluster, after initJetStream) to len(clusterConfig.Nodes), before any handler creates a bucket. GetOrCreateKVBucket now creates at that default, so lazily-created buckets are born quorate. The one-shot startup UpdateReplicas remains as a safety net for pre-existing R1 buckets from older deploys. Zero call-site churn: the default flows through the existing GetOrCreateKVBucketWithReplicas path already used by eks/ecs/iam. --- spinifex/daemon/daemon.go | 6 ++++ spinifex/utils/account.go | 25 ++++++++++++-- spinifex/utils/account_replicas_test.go | 43 +++++++++++++++++++++++++ 3 files changed, 72 insertions(+), 2 deletions(-) create mode 100644 spinifex/utils/account_replicas_test.go diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index eecde6573..59fcd22e9 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1282,6 +1282,12 @@ func (d *Daemon) startCluster() error { return fmt.Errorf("initialize JetStream: %w", err) } + // Set the default KV replica count before any handler creates a bucket, so + // lazily-created buckets are born at cluster-size replication instead of R1. + if d.clusterConfig != nil { + utils.SetDefaultKVReplicas(len(d.clusterConfig.Nodes)) + } + // Remove the obsolete spinifex-dhcp-leases bucket (idempotent). if js, jsErr := d.natsConn.JetStream(); jsErr == nil { if err := utils.DeleteKVBucketIfExists(js, "spinifex-dhcp-leases"); err != nil { diff --git a/spinifex/utils/account.go b/spinifex/utils/account.go index b04e9ae18..85a4cf32b 100644 --- a/spinifex/utils/account.go +++ b/spinifex/utils/account.go @@ -4,10 +4,29 @@ import ( "errors" "fmt" "strconv" + "sync/atomic" "github.com/nats-io/nats.go" ) +// defaultKVReplicas is the replica count applied to buckets created via +// GetOrCreateKVBucket. The daemon sets it once at boot to the cluster size so +// lazily-created buckets are born quorate on multi-node; 0 means unset and is +// treated as 1 (single-node / tests). +var defaultKVReplicas atomic.Int64 + +// SetDefaultKVReplicas sets the replica count for buckets subsequently created +// via GetOrCreateKVBucket. Clamped to a minimum of 1. Called once at daemon +// boot with the cluster size, before any handler creates a bucket. +func SetDefaultKVReplicas(n int) { + defaultKVReplicas.Store(int64(max(n, 1))) +} + +// DefaultKVReplicas returns the current default replica count (minimum 1). +func DefaultKVReplicas() int { + return max(int(defaultKVReplicas.Load()), 1) +} + // VersionKey is the well-known KV key used to store a bucket's schema version. const VersionKey = "_version" @@ -69,9 +88,11 @@ func IsAccountID(s string) bool { return true } -// GetOrCreateKVBucket creates or retrieves a NATS KV bucket at Replicas=1. +// GetOrCreateKVBucket creates or retrieves a NATS KV bucket at the cluster's +// default replica count (see SetDefaultKVReplicas), so buckets created lazily +// after boot are quorate on multi-node rather than stuck at R1. func GetOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) { - return GetOrCreateKVBucketWithReplicas(js, bucketName, history, 1) + return GetOrCreateKVBucketWithReplicas(js, bucketName, history, DefaultKVReplicas()) } // GetOrCreateKVBucketWithReplicas creates or retrieves a NATS KV bucket at the diff --git a/spinifex/utils/account_replicas_test.go b/spinifex/utils/account_replicas_test.go new file mode 100644 index 000000000..ef6d04c02 --- /dev/null +++ b/spinifex/utils/account_replicas_test.go @@ -0,0 +1,43 @@ +package utils + +import ( + "testing" + + "github.com/mulgadc/spinifex/spinifex/testutil" +) + +func TestDefaultKVReplicas(t *testing.T) { + t.Cleanup(func() { defaultKVReplicas.Store(0) }) + + if got := DefaultKVReplicas(); got != 1 { + t.Fatalf("unset default = %d, want 1", got) + } + SetDefaultKVReplicas(0) + if got := DefaultKVReplicas(); got != 1 { + t.Fatalf("SetDefaultKVReplicas(0) = %d, want 1 (clamped)", got) + } + SetDefaultKVReplicas(3) + if got := DefaultKVReplicas(); got != 3 { + t.Fatalf("SetDefaultKVReplicas(3) = %d, want 3", got) + } +} + +func TestGetOrCreateKVBucketUsesDefaultReplicas(t *testing.T) { + t.Cleanup(func() { defaultKVReplicas.Store(0) }) + _, _, js := testutil.StartTestJetStream(t) + + // A single-node test server can only host R1, so assert the default is + // threaded into the created stream config rather than the hardcoded 1. + // The R>1 case is exercised on a live multi-node cluster. + SetDefaultKVReplicas(1) + if _, err := GetOrCreateKVBucket(js, "test-default-replicas", 1); err != nil { + t.Fatalf("GetOrCreateKVBucket: %v", err) + } + info, err := js.StreamInfo("KV_test-default-replicas") + if err != nil { + t.Fatalf("StreamInfo: %v", err) + } + if info.Config.Replicas != 1 { + t.Fatalf("bucket replicas = %d, want 1", info.Config.Replicas) + } +}