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) + } +}