fix: attach to leader KV buckets before creating them (#837)

The RDS reconciler and the ECS scheduler call InitLeaderBucket on every
tick, and it created before attaching. CreateKeyValue against a bucket that
already exists is only a no-op when the stored config matches the requested
one; otherwise JetStream answers 10058 and the caller silently falls back to
an attach.

On a multi-node cluster it never matches. The stored buckets are replicated
while the RDS and ECS calls ask for the default replica count, so every tick
billed an error. That is the whole of the 14-18% JetStream API error rate
seen on env1 and env19, and it landed entirely on the meta leader because
that is the node that serves STREAM.CREATE. Single-node clusters matched and
so sat at 0%, which is why this never showed up in CI.

Attaching first removes the per-tick create outright, so it no longer matters
whether the configs agree. EKS already passed a replica count and so did not
error, but it made the same wasted call every tick and is changed with them;
its replica count now applies to a genuine first creation, which is the only
time it was ever read.

The regression test asserts on the API traffic rather than the return value.
Both orders reach the same handle, so idempotency cannot tell them apart, and
on a single-node test server the old order raised no error at all - it just
issued the create. Watching for the STREAM.CREATE advisory catches it there.

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)

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) {

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)

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)