diff --git a/spinifex/handlers/ecs/store.go b/spinifex/handlers/ecs/store.go index c6b43f026..10978ea57 100644 --- a/spinifex/handlers/ecs/store.go +++ b/spinifex/handlers/ecs/store.go @@ -221,19 +221,21 @@ func GetOrCreateAccountBucket(ctx context.Context, js jetstream.JetStream, accou // used for per-cluster scheduler leader-lease CAS locks. The bucket is configured // with History=1 and a 60s TTL so stale leases expire on their own when a leader // dies mid-cycle. kvutil.GetOrCreateBucket doesn't expose a TTL knob, so this -// takes the direct js.CreateKeyValue path and falls back to js.KeyValue on -// already-exists. +// attaches directly and creates only when absent. func InitLeaderBucket(ctx context.Context, js jetstream.JetStream) (jetstream.KeyValue, error) { - kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{ - Bucket: KVBucketECSLeader, - History: 1, - TTL: KVBucketECSLeaderTTL, - }) + // Attach before create. This runs on every scheduler tick, and CreateKeyValue + // against a bucket that exists with any other config is a STREAM.CREATE the + // meta leader answers with an error, so creating first bills one per tick. + kv, err := js.KeyValue(ctx, KVBucketECSLeader) + if errors.Is(err, jetstream.ErrBucketNotFound) { + kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{ + Bucket: KVBucketECSLeader, + History: 1, + TTL: KVBucketECSLeaderTTL, + }) + } if err != nil { - kv, err = js.KeyValue(ctx, KVBucketECSLeader) - if err != nil { - return nil, fmt.Errorf("failed to create or open ECS leader bucket %s: %w", KVBucketECSLeader, err) - } + return nil, fmt.Errorf("failed to create or open ECS leader bucket %s: %w", KVBucketECSLeader, err) } if err := migrate.DefaultRegistry.RunKV(ctx, KVBucketECSLeader, kv, KVBucketECSLeaderVersion); err != nil { return nil, fmt.Errorf("migrate %s: %w", KVBucketECSLeader, err) diff --git a/spinifex/handlers/ecs/store_test.go b/spinifex/handlers/ecs/store_test.go index de121cfcb..1caceff64 100644 --- a/spinifex/handlers/ecs/store_test.go +++ b/spinifex/handlers/ecs/store_test.go @@ -1,9 +1,13 @@ package handlers_ecs import ( + "encoding/json" + "strings" "testing" + "time" "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -57,6 +61,48 @@ func TestInitLeaderBucket_Idempotent(t *testing.T) { assert.Equal(t, KVBucketECSLeader, kv2.Bucket()) } +// The scheduler calls InitLeaderBucket on every tick. Creating first and falling +// back on already-exists reaches the same handle, so idempotency alone cannot +// tell the two orders apart — only the API traffic can. +// +// On a multi-node cluster the stored bucket is replicated while this call asks +// for the default, and JetStream answers the mismatch with an error rather than +// a no-op: 10058, once per tick, forever. That is invisible from the caller and +// showed up only as a 16% JetStream API error rate on the meta leader. +func TestInitLeaderBucket_AttachesWithoutCreating(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + + _, err := InitLeaderBucket(t.Context(), js) + require.NoError(t, err) + + // Subscribe only after the bucket exists, so the first, legitimate create is + // not counted. The server drops advisories when nothing is listening, so the + // subscription has to be flushed before the call under test. + creates := make(chan string, 8) + sub, err := nc.Subscribe("$JS.EVENT.ADVISORY.API", func(m *nats.Msg) { + var adv struct { + Subject string `json:"subject"` + } + if json.Unmarshal(m.Data, &adv) == nil && + strings.HasPrefix(adv.Subject, "$JS.API.STREAM.CREATE.") { + creates <- adv.Subject + } + }) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + require.NoError(t, nc.Flush()) + + _, err = InitLeaderBucket(t.Context(), js) + require.NoError(t, err) + require.NoError(t, nc.Flush()) + + select { + case subject := <-creates: + t.Fatalf("InitLeaderBucket issued %s on an existing bucket; attach before create", subject) + case <-time.After(250 * time.Millisecond): + } +} + // Key-path helpers must produce the ecs-v1.md Q2 layout exactly: prefixes are // what the List* enumerations watch, so a drift here silently breaks listing. func TestKeyPaths(t *testing.T) { diff --git a/spinifex/handlers/eks/store.go b/spinifex/handlers/eks/store.go index aadce170c..fb108d780 100644 --- a/spinifex/handlers/eks/store.go +++ b/spinifex/handlers/eks/store.go @@ -229,20 +229,23 @@ func GetOrCreateAccountBucket(ctx context.Context, js jetstream.JetStream, accou // replica count (clamped to a minimum of 1). The bucket is configured with // History=1 and a 60s TTL so stale leases expire on their own when a leader // dies mid-cycle. kvutil.GetOrCreateBucketWithReplicas doesn't expose a TTL -// knob, so this function sets Replicas directly on its own js.CreateKeyValue -// call and falls back to js.KeyValue on already-exists. +// knob, so this function sets Replicas on its own create call, which runs only +// when the bucket is absent. func InitLeaderBucket(ctx context.Context, js jetstream.JetStream, replicas int) (jetstream.KeyValue, error) { - kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{ - Bucket: KVBucketEKSLeader, - History: 1, - TTL: KVBucketEKSLeaderTTL, - Replicas: max(replicas, 1), - }) + // Attach before create, so replicas applies to a genuine first creation only. + // CreateKeyValue against a bucket that exists with any other config is a + // STREAM.CREATE the meta leader answers with an error rather than a no-op. + kv, err := js.KeyValue(ctx, KVBucketEKSLeader) + if errors.Is(err, jetstream.ErrBucketNotFound) { + kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{ + Bucket: KVBucketEKSLeader, + History: 1, + TTL: KVBucketEKSLeaderTTL, + Replicas: max(replicas, 1), + }) + } if err != nil { - kv, err = js.KeyValue(ctx, KVBucketEKSLeader) - if err != nil { - return nil, fmt.Errorf("failed to create or open EKS leader bucket %s: %w", KVBucketEKSLeader, err) - } + return nil, fmt.Errorf("failed to create or open EKS leader bucket %s: %w", KVBucketEKSLeader, err) } if err := migrate.DefaultRegistry.RunKV(ctx, KVBucketEKSLeader, kv, KVBucketEKSLeaderVersion); err != nil { return nil, fmt.Errorf("migrate %s: %w", KVBucketEKSLeader, err) diff --git a/spinifex/handlers/rds/store.go b/spinifex/handlers/rds/store.go index 1e1944698..abe214e7b 100644 --- a/spinifex/handlers/rds/store.go +++ b/spinifex/handlers/rds/store.go @@ -215,19 +215,22 @@ func GetOrCreateSystemBucket(ctx context.Context, js jetstream.JetStream) (jetst return kv, nil } -// kvutil.GetOrCreateBucket exposes no TTL knob, so the lease bucket takes the -// direct CreateKeyValue path and falls back to an open on already-exists. +// kvutil.GetOrCreateBucket exposes no TTL knob, so the lease bucket attaches +// directly and creates only when absent. func InitLeaderBucket(ctx context.Context, js jetstream.JetStream) (jetstream.KeyValue, error) { - kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{ - Bucket: KVBucketRDSLeader, - History: 1, - TTL: KVBucketRDSLeaderTTL, - }) + // Attach before create. This runs on every reconcile tick, and CreateKeyValue + // against a bucket that exists with any other config is a STREAM.CREATE the + // meta leader answers with an error, so creating first bills one per tick. + kv, err := js.KeyValue(ctx, KVBucketRDSLeader) + if errors.Is(err, jetstream.ErrBucketNotFound) { + kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{ + Bucket: KVBucketRDSLeader, + History: 1, + TTL: KVBucketRDSLeaderTTL, + }) + } if err != nil { - kv, err = js.KeyValue(ctx, KVBucketRDSLeader) - if err != nil { - return nil, fmt.Errorf("failed to create or open RDS leader bucket %s: %w", KVBucketRDSLeader, err) - } + return nil, fmt.Errorf("failed to create or open RDS leader bucket %s: %w", KVBucketRDSLeader, err) } if err := migrate.DefaultRegistry.RunKV(ctx, KVBucketRDSLeader, kv, KVBucketRDSLeaderVersion); err != nil { return nil, fmt.Errorf("migrate %s: %w", KVBucketRDSLeader, err)