mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-19 10:52:04 +00:00
fix(nats): create lazy KV buckets at cluster-size replicas, not R1 (#529)
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.
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user