From 0b79697c15383a1d4d44ef024972440ae5a42683 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Wed, 10 Jun 2026 15:43:37 +1000 Subject: [PATCH] feat: atomic create idempotency across EKS, ELBv2 and EC2 ClientToken (267.3) (#340) * feat: atomic create idempotency across EKS, ELBv2 and EC2 ClientToken Every create handler took a non-atomic Get-check then a blind Put, so a gateway/NATS retry that spawned a second concurrent handler could pass the check in both and double-launch + leak VMs (267 failure-mode C). Make each create claim its identity key atomically before any launching work: - EKS CreateCluster: claim the cluster meta via kv.Create; a prior FAILED attempt is reclaimed via a CREATING CAS so exactly one concurrent retry wins. purgeClusterInfra gains a deleteMeta flag so reclaim purges infra while keeping its claim key. - EKS CreateNodegroup: claim the nodegroup record before SG provisioning or worker launch; duplicates return ResourceInUse without provisioning. - ELBv2 CreateLoadBalancer: account-scoped name claim (lbname..) taken before ENI/VM work, released on every failure path and in DeleteLoadBalancer; a name claim whose owner lbID has no live record is a crashed-create orphan and is reclaimed. - EC2 RunInstances: ClientToken dedup over a dedicated TTL bucket with a two-phase claim (in-flight -> done) and reservation replay; a token reused with different params returns IdempotentParameterMismatch. Race tests under -race assert exactly one owner and a single launch for concurrent same-name creates on all paths. * test: cover ClientToken idempotency; extract orchestration for testability Extract the claim->replay->launch->finalize/abort flow out of RunInstances into runInstancesWithClientToken(store, ..., launch func) so the idempotency path is unit-testable without a live gateway/NATS. RunInstances now just resolves the store, hashes params, and passes a launch closure. Make the in-flight poll timeout/step package vars so a duplicate-caller wait-timeout is deterministically testable. Adds tests: replay-skips-launch, owner-launches-once-then-replay, launch-failure-aborts, param-mismatch maps to AWS error, getClientTokenStore binds once, and in-flight wait-timeout. --- spinifex/gateway/ec2/instance/RunInstances.go | 31 ++- spinifex/gateway/ec2/instance/clienttoken.go | 242 +++++++++++++++++ .../gateway/ec2/instance/clienttoken_test.go | 251 ++++++++++++++++++ spinifex/handlers/eks/claim.go | 34 +++ spinifex/handlers/eks/claim_race_test.go | 113 ++++++++ spinifex/handlers/eks/nodegroup.go | 39 ++- spinifex/handlers/eks/service_impl.go | 118 ++++++-- spinifex/handlers/elbv2/name_claim_test.go | 94 +++++++ spinifex/handlers/elbv2/service_impl.go | 33 +++ spinifex/handlers/elbv2/store.go | 67 +++++ 10 files changed, 988 insertions(+), 34 deletions(-) create mode 100644 spinifex/gateway/ec2/instance/clienttoken.go create mode 100644 spinifex/gateway/ec2/instance/clienttoken_test.go create mode 100644 spinifex/handlers/eks/claim.go create mode 100644 spinifex/handlers/eks/claim_race_test.go create mode 100644 spinifex/handlers/elbv2/name_claim_test.go diff --git a/spinifex/gateway/ec2/instance/RunInstances.go b/spinifex/gateway/ec2/instance/RunInstances.go index 42fa6260c..d40ff528e 100644 --- a/spinifex/gateway/ec2/instance/RunInstances.go +++ b/spinifex/gateway/ec2/instance/RunInstances.go @@ -2,6 +2,7 @@ package gateway_ec2_instance import ( "errors" + "log/slog" "strings" "time" @@ -83,12 +84,36 @@ func ValidateRunInstancesInput(input *ec2.RunInstancesInput) (err error) { // unit tests that don't exercise the policy path). func RunInstances(input *ec2.RunInstancesInput, natsConn *nats.Conn, iamSvc handlers_iam.IAMService, accountID string, passRoleCheck PassRoleChecker) (reservation ec2.Reservation, err error) { // Validate input - err = ValidateRunInstancesInput(input) - - if err != nil { + if err = ValidateRunInstancesInput(input); err != nil { return reservation, err } + // No ClientToken ⇒ no idempotency contract; launch directly. + token := aws.StringValue(input.ClientToken) + if token == "" { + return runInstancesInner(input, natsConn, iamSvc, accountID, passRoleCheck) + } + + // ClientToken set: dedup concurrent/retried launches. The caller asked for + // idempotency, so a store failure fails the call rather than risking a + // double launch on their retry. + store, serr := getClientTokenStore(natsConn) + if serr != nil { + slog.Error("RunInstances: client-token store unavailable", "err", serr) + return reservation, errors.New(awserrors.ErrorServerInternal) + } + // Hash the request BEFORE any mutation (resolveAndAuthorizeInstanceProfile + // rewrites IamInstanceProfile) so the same token+params always matches. + paramHash := clientTokenParamHash(input) + return runInstancesWithClientToken(store, accountID, token, paramHash, func() (ec2.Reservation, error) { + return runInstancesInner(input, natsConn, iamSvc, accountID, passRoleCheck) + }) +} + +// runInstancesInner performs the actual launch (profile resolution, placement +// routing, capacity-aware distribution). It is wrapped by RunInstances, which +// layers ClientToken idempotency on top. +func runInstancesInner(input *ec2.RunInstancesInput, natsConn *nats.Conn, iamSvc handlers_iam.IAMService, accountID string, passRoleCheck PassRoleChecker) (reservation ec2.Reservation, err error) { resolvedProfile, err := resolveAndAuthorizeInstanceProfile(input, iamSvc, accountID, passRoleCheck) if err != nil { return reservation, err diff --git a/spinifex/gateway/ec2/instance/clienttoken.go b/spinifex/gateway/ec2/instance/clienttoken.go new file mode 100644 index 000000000..04067b4f2 --- /dev/null +++ b/spinifex/gateway/ec2/instance/clienttoken.go @@ -0,0 +1,242 @@ +package gateway_ec2_instance + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "log/slog" + "sync" + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/nats-io/nats.go" +) + +const ( + // kvBucketClientTokens holds RunInstances ClientToken idempotency records. + kvBucketClientTokens = "spinifex-ec2-clienttokens" //nolint:gosec // G101 false positive: KV bucket name, not a credential + + // clientTokenTTL bounds how long a token replays. It must outlast normal + // SDK retry windows yet be short enough that a crashed in-flight record + // (winner died mid-launch) ages out and frees the token for a fresh launch. + clientTokenTTL = 15 * time.Minute + + tokenStatusInFlight = "in-flight" + tokenStatusDone = "done" +) + +// clientTokenWaitTimeout caps how long a duplicate caller polls an in-flight +// winner before giving up (the SDK retries the whole call); clientTokenPollStep +// is the inter-poll sleep. Vars (not consts) so tests can shrink them. +var ( + clientTokenWaitTimeout = 30 * time.Second + clientTokenPollStep = 250 * time.Millisecond +) + +// errIdempotentParamMismatch signals the same token was reused with different +// request parameters (AWS IdempotentParameterMismatch). +var errIdempotentParamMismatch = errors.New("clienttoken: idempotent parameter mismatch") + +// errClientTokenWaitTimeout signals a duplicate caller polled an in-flight +// winner past clientTokenWaitTimeout without the winner finishing. +var errClientTokenWaitTimeout = errors.New("clienttoken: timed out waiting for in-flight launch") + +// tokenRecord is the per-token idempotency record. ParamHash binds the token to +// the original request so a reuse with different params is rejected; Reservation +// is populated only once Status is done so a duplicate caller can replay it. +type tokenRecord struct { + Status string `json:"status"` + ParamHash string `json:"paramHash"` + StartedAt time.Time `json:"startedAt"` + Reservation *ec2.Reservation `json:"reservation,omitempty"` +} + +// ClientTokenStore implements RunInstances ClientToken idempotency over a +// dedicated TTL KV bucket. The first caller to Create a token record owns the +// launch; duplicates either replay the stored reservation (done) or poll until +// the owner finishes (in-flight). +type ClientTokenStore struct { + kv nats.KeyValue +} + +var ( + ctStore *ClientTokenStore + ctOnce sync.Once + errCTStoreInit error +) + +// getClientTokenStore lazily opens (or creates) the process-wide client-token +// store. The gateway holds a single long-lived NATS connection, so a sync.Once +// bind is sufficient; the bucket is get-or-created with a TTL. +func getClientTokenStore(nc *nats.Conn) (*ClientTokenStore, error) { + ctOnce.Do(func() { + js, err := nc.JetStream() + if err != nil { + errCTStoreInit = fmt.Errorf("clienttoken jetstream: %w", err) + return + } + ctStore, errCTStoreInit = newClientTokenStore(js) + }) + return ctStore, errCTStoreInit +} + +func newClientTokenStore(js nats.JetStreamContext) (*ClientTokenStore, error) { + kv, err := js.KeyValue(kvBucketClientTokens) + if errors.Is(err, nats.ErrBucketNotFound) { + kv, err = js.CreateKeyValue(&nats.KeyValueConfig{ + Bucket: kvBucketClientTokens, + History: 1, + TTL: clientTokenTTL, + }) + } + if err != nil { + return nil, fmt.Errorf("open client-token bucket: %w", err) + } + return &ClientTokenStore{kv: kv}, nil +} + +func clientTokenKey(accountID, token string) string { + return accountID + "." + token +} + +// clientTokenParamHash hashes the caller's request, ignoring ClientToken, so the +// same token with identical params matches and with different params mismatches. +// It MUST run before any mutation of input (e.g. instance-profile ARN rewrite). +func clientTokenParamHash(input *ec2.RunInstancesInput) string { + clone := *input + clone.ClientToken = nil + b, err := json.Marshal(&clone) + if err != nil { + // Marshal of an SDK struct does not fail in practice; fall back to a + // constant so a token still claims (idempotency degrades to name-only). + b = []byte(clientTokenKey("", "")) + } + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]) +} + +// Claim runs the two-phase token claim. It returns: +// - (nil, true, nil) ⇒ caller owns the launch; must Finalize or Abort. +// - (reservation, false, nil) ⇒ replay a completed launch; return it directly. +// - (nil, false, err) ⇒ errIdempotentParamMismatch (token reused with +// different params) or errClientTokenWaitTimeout (in-flight winner too slow). +func (c *ClientTokenStore) Claim(accountID, token, paramHash string) (*ec2.Reservation, bool, error) { + key := clientTokenKey(accountID, token) + inflight, err := json.Marshal(tokenRecord{ + Status: tokenStatusInFlight, + ParamHash: paramHash, + StartedAt: time.Now().UTC(), + }) + if err != nil { + return nil, false, fmt.Errorf("clienttoken marshal: %w", err) + } + if _, cerr := c.kv.Create(key, inflight); cerr == nil { + return nil, true, nil + } else if !errors.Is(cerr, nats.ErrKeyExists) { + return nil, false, fmt.Errorf("clienttoken create %s: %w", key, cerr) + } + + deadline := time.Now().Add(clientTokenWaitTimeout) + for { + entry, gerr := c.kv.Get(key) + if gerr != nil { + if errors.Is(gerr, nats.ErrKeyNotFound) { + // Owner aborted (transient launch failure) or the record aged + // out: race for ownership again. + if _, rcerr := c.kv.Create(key, inflight); rcerr == nil { + return nil, true, nil + } else if !errors.Is(rcerr, nats.ErrKeyExists) { + return nil, false, fmt.Errorf("clienttoken recreate %s: %w", key, rcerr) + } + continue + } + return nil, false, fmt.Errorf("clienttoken get %s: %w", key, gerr) + } + var rec tokenRecord + if uerr := json.Unmarshal(entry.Value(), &rec); uerr != nil { + return nil, false, fmt.Errorf("clienttoken unmarshal %s: %w", key, uerr) + } + if rec.ParamHash != paramHash { + return nil, false, errIdempotentParamMismatch + } + if rec.Status == tokenStatusDone { + return rec.Reservation, false, nil + } + if time.Now().After(deadline) { + return nil, false, errClientTokenWaitTimeout + } + time.Sleep(clientTokenPollStep) + } +} + +// Finalize stamps the token record done with the launched reservation so +// duplicate callers replay it. The owner is the sole writer, so a plain Put is +// safe and refreshes the TTL. +func (c *ClientTokenStore) Finalize(accountID, token, paramHash string, res *ec2.Reservation) error { + key := clientTokenKey(accountID, token) + data, err := json.Marshal(tokenRecord{ + Status: tokenStatusDone, + ParamHash: paramHash, + StartedAt: time.Now().UTC(), + Reservation: res, + }) + if err != nil { + return fmt.Errorf("clienttoken finalize marshal: %w", err) + } + if _, err := c.kv.Put(key, data); err != nil { + return fmt.Errorf("clienttoken finalize put %s: %w", key, err) + } + return nil +} + +// runInstancesWithClientToken wraps a launch in ClientToken idempotency: it +// claims the token, replays a completed reservation, or (as the owner) runs +// launch and finalizes the result — aborting the token on a launch failure so a +// retry can re-launch. AWS error codes are mapped here; the launch closure +// returns the raw reservation/error. Extracted from RunInstances so the +// idempotency flow is unit-testable without a live gateway. +func runInstancesWithClientToken( + store *ClientTokenStore, + accountID, token, paramHash string, + launch func() (ec2.Reservation, error), +) (ec2.Reservation, error) { + var zero ec2.Reservation + replay, owned, cerr := store.Claim(accountID, token, paramHash) + if cerr != nil { + if errors.Is(cerr, errIdempotentParamMismatch) { + return zero, errors.New(awserrors.ErrorIdempotentParameterMismatch) + } + slog.Error("RunInstances: client-token claim failed", "token", token, "err", cerr) + return zero, errors.New(awserrors.ErrorServerInternal) + } + if replay != nil { + return *replay, nil + } + if !owned { + return zero, errors.New(awserrors.ErrorServerInternal) + } + + res, rerr := launch() + if rerr != nil { + store.Abort(accountID, token) + return zero, rerr + } + if ferr := store.Finalize(accountID, token, paramHash, &res); ferr != nil { + // Launch succeeded; failing to persist the replay record only weakens a + // future retry's dedup, so do not fail the response. + slog.Warn("RunInstances: failed to finalize client-token record", "token", token, "err", ferr) + } + return res, nil +} + +// Abort drops an in-flight token after a launch failure so a retry with the +// same token can re-launch instead of replaying a non-existent reservation. +func (c *ClientTokenStore) Abort(accountID, token string) { + key := clientTokenKey(accountID, token) + if err := c.kv.Delete(key); err != nil && !errors.Is(err, nats.ErrKeyNotFound) { + slog.Warn("clienttoken: failed to abort token record", "key", key, "err", err) + } +} diff --git a/spinifex/gateway/ec2/instance/clienttoken_test.go b/spinifex/gateway/ec2/instance/clienttoken_test.go new file mode 100644 index 000000000..c28ac4acb --- /dev/null +++ b/spinifex/gateway/ec2/instance/clienttoken_test.go @@ -0,0 +1,251 @@ +package gateway_ec2_instance + +import ( + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ctTestAccount = "111122223333" + +func newTestClientTokenStore(t *testing.T) *ClientTokenStore { + t.Helper() + _, _, js := testutil.StartTestJetStream(t) + store, err := newClientTokenStore(js) + require.NoError(t, err) + return store +} + +// A completed token replays the stored reservation for a duplicate caller with +// matching params. +func TestClientToken_ReplaysCompletedReservation(t *testing.T) { + store := newTestClientTokenStore(t) + const tok, hash = "tok-1", "hash-a" + + _, owned, err := store.Claim(ctTestAccount, tok, hash) + require.NoError(t, err) + require.True(t, owned, "first caller owns the launch") + + res := &ec2.Reservation{ReservationId: aws.String("r-123")} + require.NoError(t, store.Finalize(ctTestAccount, tok, hash, res)) + + replay, owned2, err := store.Claim(ctTestAccount, tok, hash) + require.NoError(t, err) + assert.False(t, owned2, "duplicate caller does not own the launch") + require.NotNil(t, replay) + assert.Equal(t, "r-123", aws.StringValue(replay.ReservationId)) +} + +// Reusing a token with different params is IdempotentParameterMismatch. +func TestClientToken_ParamMismatchRejected(t *testing.T) { + store := newTestClientTokenStore(t) + const tok = "tok-2" + + _, owned, err := store.Claim(ctTestAccount, tok, "hash-a") + require.NoError(t, err) + require.True(t, owned) + require.NoError(t, store.Finalize(ctTestAccount, tok, "hash-a", &ec2.Reservation{ReservationId: aws.String("r-1")})) + + _, _, err = store.Claim(ctTestAccount, tok, "hash-DIFFERENT") + assert.ErrorIs(t, err, errIdempotentParamMismatch) +} + +// A failed launch aborts the token so a later retry with the same token +// re-launches instead of replaying a non-existent reservation. +func TestClientToken_AbortAllowsRelaunch(t *testing.T) { + store := newTestClientTokenStore(t) + const tok, hash = "tok-3", "hash-a" + + _, owned, err := store.Claim(ctTestAccount, tok, hash) + require.NoError(t, err) + require.True(t, owned) + + store.Abort(ctTestAccount, tok) + + _, owned2, err := store.Claim(ctTestAccount, tok, hash) + require.NoError(t, err) + assert.True(t, owned2, "after abort the token is free to re-own") +} + +// Concurrent callers with the same token must yield exactly one owner; the +// others replay the owner's reservation. Proves single-launch under -race. +func TestClientToken_ConcurrentSingleOwner(t *testing.T) { + store := newTestClientTokenStore(t) + const tok, hash = "tok-4", "hash-a" + res := &ec2.Reservation{ReservationId: aws.String("r-only")} + + const n = 4 + var owners int32 + var wg sync.WaitGroup + errs := make([]error, n) + replays := make([]*ec2.Reservation, n) + for i := range n { + wg.Add(1) + go func(i int) { + defer wg.Done() + replay, owned, err := store.Claim(ctTestAccount, tok, hash) + if err != nil { + errs[i] = err + return + } + if owned { + atomic.AddInt32(&owners, 1) + // Simulate the launch then publish the reservation. + errs[i] = store.Finalize(ctTestAccount, tok, hash, res) + replays[i] = res + return + } + replays[i] = replay + }(i) + } + wg.Wait() + + for _, e := range errs { + require.NoError(t, e) + } + assert.Equal(t, int32(1), owners, "exactly one caller launches") + for _, r := range replays { + require.NotNil(t, r, "every caller ends with the single reservation") + assert.Equal(t, "r-only", aws.StringValue(r.ReservationId)) + } +} + +// clientTokenParamHash ignores ClientToken (same params, different token → +// same hash) but reflects a real parameter change. +func TestClientTokenParamHash_IgnoresTokenReflectsParams(t *testing.T) { + base := &ec2.RunInstancesInput{ + ImageId: aws.String("ami-123"), + InstanceType: aws.String("t3.micro"), + MinCount: aws.Int64(1), + MaxCount: aws.Int64(1), + ClientToken: aws.String("tok-A"), + } + sameParamsDiffToken := *base + sameParamsDiffToken.ClientToken = aws.String("tok-B") + diffParams := *base + diffParams.InstanceType = aws.String("t3.large") + + assert.Equal(t, clientTokenParamHash(base), clientTokenParamHash(&sameParamsDiffToken), + "token must not affect the param hash") + assert.NotEqual(t, clientTokenParamHash(base), clientTokenParamHash(&diffParams), + "a real param change must change the hash") +} + +// --- runInstancesWithClientToken (extracted orchestration) --- + +// A completed token replays its reservation and must NOT invoke the launcher. +func TestRunInstancesWithClientToken_ReplaySkipsLaunch(t *testing.T) { + store := newTestClientTokenStore(t) + const tok, hash = "rt-1", "h" + _, owned, err := store.Claim(ctTestAccount, tok, hash) + require.NoError(t, err) + require.True(t, owned) + require.NoError(t, store.Finalize(ctTestAccount, tok, hash, &ec2.Reservation{ReservationId: aws.String("r-x")})) + + launched := false + res, err := runInstancesWithClientToken(store, ctTestAccount, tok, hash, func() (ec2.Reservation, error) { + launched = true + return ec2.Reservation{}, nil + }) + require.NoError(t, err) + assert.False(t, launched, "replay must not launch") + assert.Equal(t, "r-x", aws.StringValue(res.ReservationId)) +} + +// The owner launches once and finalizes; a duplicate replays without launching. +func TestRunInstancesWithClientToken_OwnerLaunchesOnceThenReplay(t *testing.T) { + store := newTestClientTokenStore(t) + const tok, hash = "rt-2", "h" + launches := 0 + launch := func() (ec2.Reservation, error) { + launches++ + return ec2.Reservation{ReservationId: aws.String("r-own")}, nil + } + + res, err := runInstancesWithClientToken(store, ctTestAccount, tok, hash, launch) + require.NoError(t, err) + assert.Equal(t, "r-own", aws.StringValue(res.ReservationId)) + + res2, err := runInstancesWithClientToken(store, ctTestAccount, tok, hash, launch) + require.NoError(t, err) + assert.Equal(t, "r-own", aws.StringValue(res2.ReservationId)) + assert.Equal(t, 1, launches, "duplicate must replay, not relaunch") +} + +// A launch failure aborts the token so a retry re-launches. +func TestRunInstancesWithClientToken_LaunchFailureAborts(t *testing.T) { + store := newTestClientTokenStore(t) + const tok, hash = "rt-3", "h" + + _, err := runInstancesWithClientToken(store, ctTestAccount, tok, hash, func() (ec2.Reservation, error) { + return ec2.Reservation{}, errors.New("no capacity") + }) + require.Error(t, err) + + relaunched := false + res, err := runInstancesWithClientToken(store, ctTestAccount, tok, hash, func() (ec2.Reservation, error) { + relaunched = true + return ec2.Reservation{ReservationId: aws.String("r-retry")}, nil + }) + require.NoError(t, err) + assert.True(t, relaunched, "after abort the retry must launch") + assert.Equal(t, "r-retry", aws.StringValue(res.ReservationId)) +} + +// Token reuse with different params maps to the AWS IdempotentParameterMismatch +// error code and never launches. +func TestRunInstancesWithClientToken_ParamMismatchMapsAWSError(t *testing.T) { + store := newTestClientTokenStore(t) + const tok = "rt-4" + _, owned, err := store.Claim(ctTestAccount, tok, "hA") + require.NoError(t, err) + require.True(t, owned) + require.NoError(t, store.Finalize(ctTestAccount, tok, "hA", &ec2.Reservation{ReservationId: aws.String("r")})) + + launched := false + _, err = runInstancesWithClientToken(store, ctTestAccount, tok, "hB", func() (ec2.Reservation, error) { + launched = true + return ec2.Reservation{}, nil + }) + require.EqualError(t, err, awserrors.ErrorIdempotentParameterMismatch) + assert.False(t, launched) +} + +// getClientTokenStore binds the process-wide store once and returns the same +// instance on subsequent calls. +func TestGetClientTokenStore_BindsOnce(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + s1, err := getClientTokenStore(nc) + require.NoError(t, err) + require.NotNil(t, s1) + s2, err := getClientTokenStore(nc) + require.NoError(t, err) + assert.Same(t, s1, s2, "store binds once") +} + +// A duplicate caller polling an in-flight winner that never finishes bails out +// with the wait-timeout sentinel (exercises the Claim poll loop + deadline). +func TestClientToken_InFlightWaitTimesOut(t *testing.T) { + store := newTestClientTokenStore(t) + const tok, hash = "wt-1", "h" + + _, owned, err := store.Claim(ctTestAccount, tok, hash) + require.NoError(t, err) + require.True(t, owned, "owner holds the in-flight record and never finalizes") + + origTimeout, origStep := clientTokenWaitTimeout, clientTokenPollStep + clientTokenWaitTimeout, clientTokenPollStep = 30*time.Millisecond, 10*time.Millisecond + defer func() { clientTokenWaitTimeout, clientTokenPollStep = origTimeout, origStep }() + + _, _, err = store.Claim(ctTestAccount, tok, hash) + assert.ErrorIs(t, err, errClientTokenWaitTimeout) +} diff --git a/spinifex/handlers/eks/claim.go b/spinifex/handlers/eks/claim.go new file mode 100644 index 000000000..51ecb8a22 --- /dev/null +++ b/spinifex/handlers/eks/claim.go @@ -0,0 +1,34 @@ +package handlers_eks + +import ( + "errors" + "fmt" + + "github.com/nats-io/nats.go" +) + +// claimKey atomically creates key with payload. owned is true when this caller +// created it (won the claim). On an existing key it returns owned=false with the +// current value + revision so the caller can decide reclaim-vs-conflict; a key +// that vanished between the Create and the Get returns owned=false, existing=nil +// so the caller can retry the claim. +// +// This is the shared idempotency primitive for create handlers: the first +// caller to Create a resource's identity key wins and does the launching work; +// a duplicate (SDK retry, gateway re-publish that spawned a second handler) +// loses and never launches, so concurrent creates cannot double-launch + leak. +func claimKey(kv nats.KeyValue, key string, payload []byte) (owned bool, existing []byte, rev uint64, err error) { + if _, cerr := kv.Create(key, payload); cerr == nil { + return true, nil, 0, nil + } else if !errors.Is(cerr, nats.ErrKeyExists) { + return false, nil, 0, fmt.Errorf("kv create %s: %w", key, cerr) + } + entry, gerr := kv.Get(key) + if gerr != nil { + if errors.Is(gerr, nats.ErrKeyNotFound) { + return false, nil, 0, nil + } + return false, nil, 0, fmt.Errorf("kv get %s: %w", key, gerr) + } + return false, entry.Value(), entry.Revision(), nil +} diff --git a/spinifex/handlers/eks/claim_race_test.go b/spinifex/handlers/eks/claim_race_test.go new file mode 100644 index 000000000..6cdf0ef64 --- /dev/null +++ b/spinifex/handlers/eks/claim_race_test.go @@ -0,0 +1,113 @@ +package handlers_eks + +import ( + "sync" + "testing" + + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// classifyCreateErrs splits a slice of create results into success vs +// ResourceInUse counts, failing the test on any other error. +func classifyCreateErrs(t *testing.T, errs []error) (ok, inUse int) { + t.Helper() + for _, e := range errs { + switch { + case e == nil: + ok++ + case e.Error() == awserrors.ErrorEKSResourceInUse: + inUse++ + default: + t.Fatalf("unexpected create error: %v", e) + } + } + return ok, inUse +} + +// Two concurrent CreateCluster calls for the same name must resolve to exactly +// one owner (CREATING) and one ResourceInUse reject. The atomic name claim is +// the first mutating step, so the loser returns before any VM/NLB launch — +// asserted structurally here and enforced under -race (a double-claim would +// race the launcher fakes the winner mutates). +func TestCreateCluster_ConcurrentSameNameSingleOwner(t *testing.T) { + f := newEKSServiceFixture(t) + + const n = 2 + errs := make([]error, n) + var wg sync.WaitGroup + for i := range n { + wg.Add(1) + go func(i int) { + defer wg.Done() + _, errs[i] = f.svc.CreateCluster(createInput("race"), testAccountID, "") + }(i) + } + wg.Wait() + + ok, inUse := classifyCreateErrs(t, errs) + assert.Equal(t, 1, ok, "exactly one create owns the name") + assert.Equal(t, 1, inUse, "the duplicate is rejected ResourceInUse") + + meta, err := GetClusterMeta(f.kv, "race") + require.NoError(t, err) + assert.Equal(t, ClusterStatusCreating, meta.Status) +} + +// A FAILED cluster plus two concurrent retries must reclaim exactly once: one +// caller CAS-flips FAILED→CREATING and proceeds, the other loses the CAS and is +// rejected ResourceInUse. No double purge/relaunch. +func TestCreateCluster_ConcurrentReclaimSingleOwner(t *testing.T) { + f := newEKSServiceFixture(t) + + meta := sampleClusterMeta("race") + meta.Status = ClusterStatusFailed + require.NoError(t, PutClusterMeta(f.kv, meta)) + + const n = 2 + errs := make([]error, n) + var wg sync.WaitGroup + for i := range n { + wg.Add(1) + go func(i int) { + defer wg.Done() + _, errs[i] = f.svc.CreateCluster(createInput("race"), testAccountID, "") + }(i) + } + wg.Wait() + + ok, inUse := classifyCreateErrs(t, errs) + assert.Equal(t, 1, ok, "exactly one retry reclaims the FAILED cluster") + assert.Equal(t, 1, inUse, "the concurrent retry is rejected") + + got, err := GetClusterMeta(f.kv, "race") + require.NoError(t, err) + assert.Equal(t, ClusterStatusCreating, got.Status) +} + +// Two concurrent CreateNodegroup calls for the same name must resolve to one +// owner and one ResourceInUse reject, and the worker launcher must record +// exactly one launch (the loser returns at the record claim before +// launchWorkers). +func TestCreateNodegroup_ConcurrentSameNameSingleOwner(t *testing.T) { + f := newEKSServiceFixture(t) + seedActiveClusterWithToken(t, f, "c1") + + const n = 2 + errs := make([]error, n) + var wg sync.WaitGroup + for i := range n { + wg.Add(1) + go func(i int) { + defer wg.Done() + _, errs[i] = f.svc.CreateNodegroup(createNGInput("c1", "ng1", 1), testAccountID) + }(i) + } + wg.Wait() + + ok, inUse := classifyCreateErrs(t, errs) + assert.Equal(t, 1, ok, "exactly one create owns the nodegroup") + assert.Equal(t, 1, inUse, "the duplicate is rejected ResourceInUse") + assert.Len(t, f.worker.runCalls, 1, "only the owner launches a worker; no double-launch") +} diff --git a/spinifex/handlers/eks/nodegroup.go b/spinifex/handlers/eks/nodegroup.go index 38e4cf081..46cfc3286 100644 --- a/spinifex/handlers/eks/nodegroup.go +++ b/spinifex/handlers/eks/nodegroup.go @@ -54,6 +54,25 @@ func PutNodegroupRecord(kv nats.KeyValue, rec *NodegroupRecord) error { return nil } +// ClaimNodegroupRecord atomically creates the nodegroup record, returning +// owned=false when a record already exists. This is the idempotency barrier for +// CreateNodegroup — a duplicate request loses the claim and never launches +// workers. Owner updates after the claim use PutNodegroupRecord. +func ClaimNodegroupRecord(kv nats.KeyValue, rec *NodegroupRecord) (bool, error) { + if rec == nil { + return false, errors.New("eks: ClaimNodegroupRecord nil record") + } + if rec.ClusterName == "" || rec.Name == "" { + return false, errors.New("eks: ClaimNodegroupRecord missing cluster or name") + } + data, err := json.Marshal(rec) + if err != nil { + return false, fmt.Errorf("marshal nodegroup %s: %w", rec.Name, err) + } + owned, _, _, err := claimKey(kv, NodegroupKey(rec.ClusterName, rec.Name), data) + return owned, err +} + // GetNodegroupRecord reads one record. Returns ErrNodegroupNotFound if absent. func GetNodegroupRecord(kv nats.KeyValue, cluster, ng string) (*NodegroupRecord, error) { if cluster == "" || ng == "" { @@ -207,18 +226,30 @@ func (s *EKSServiceImpl) createNodegroup(acctKV nats.KeyValue, input *eks.Create ModifiedAt: now, } + // Atomically claim the nodegroup record before any SG provisioning or worker + // launch. This is the idempotency barrier: a duplicate CreateNodegroup (SDK + // retry / gateway re-publish that spawned a second handler) loses the claim + // here and returns without doing any provisioning work, so concurrent creates + // cannot double-launch workers. The L169 read above stays only as a cheap + // fast-path reject. + owned, err := ClaimNodegroupRecord(acctKV, rec) + if err != nil { + return nil, err + } + if !owned { + return nil, errors.New(awserrors.ErrorEKSResourceInUse) + } + cpSGID, ngSGID, err := EnsureClusterSGs(s.deps.VPCSG, accountID, cluster, meta.ResourcesVpcConfig.VpcId) if err != nil { + s.markNodegroupFailed(acctKV, cluster, ng, "ensure cluster SGs: "+err.Error()) return nil, fmt.Errorf("ensure cluster SGs: %w", err) } if err := EnsureNodegroupSGRules(s.deps.VPCSG, accountID, cluster, cpSGID, ngSGID); err != nil { + s.markNodegroupFailed(acctKV, cluster, ng, "ensure nodegroup SG rules: "+err.Error()) return nil, fmt.Errorf("ensure nodegroup SG rules: %w", err) } - if err := PutNodegroupRecord(acctKV, rec); err != nil { - return nil, err - } - token, err := s.decryptNodeToken(acctKV, cluster) if err != nil { s.markNodegroupFailed(acctKV, cluster, ng, "decrypt node token: "+err.Error()) diff --git a/spinifex/handlers/eks/service_impl.go b/spinifex/handlers/eks/service_impl.go index ec4e74a81..70176e834 100644 --- a/spinifex/handlers/eks/service_impl.go +++ b/spinifex/handlers/eks/service_impl.go @@ -2,6 +2,7 @@ package handlers_eks import ( "context" + "encoding/json" "errors" "fmt" "log/slog" @@ -326,27 +327,6 @@ func (s *EKSServiceImpl) CreateCluster(input *eks.CreateClusterInput, accountID, return nil, logCreateErr(name, accountID, "get account bucket", err) } - if existing, err := GetClusterMeta(acctKV, name); err == nil { - if existing.Status != ClusterStatusFailed { - slog.Info("CreateCluster: cluster already exists", - "name", name, "accountID", accountID, "status", existing.Status) - return nil, errors.New(awserrors.ErrorEKSResourceInUse) - } - // A prior attempt failed. Reclaim whatever it left behind and clear its - // state so this create starts clean — makes a retry (including the - // SDK's own auto-retry) idempotent instead of masking the original - // failure with a spurious "already exists" error. If teardown of the - // failed attempt's resources fails, surface that rather than recreating - // on top of orphans. - slog.Info("CreateCluster: reclaiming FAILED cluster before recreate", - "name", name, "accountID", accountID) - if perr := s.purgeClusterInfra(accountID, name, existing, acctKV); perr != nil { - return nil, logCreateErr(name, accountID, "reclaim failed cluster", perr) - } - } else if !errors.Is(err, ErrClusterNotFound) { - return nil, logCreateErr(name, accountID, "preflight get meta", err) - } - vpcID, err := s.deps.VPCSubnet.GetSubnetVPC(accountID, subnetIDs[0]) if err != nil { // Subnet is a caller-supplied parameter — a resolve failure is a client @@ -379,8 +359,13 @@ func (s *EKSServiceImpl) CreateCluster(input *eks.CreateClusterInput, accountID, BuiltinIngress: builtinIngressEnabled(input.Tags), CreatedAt: time.Now().UTC(), } - if err := PutClusterMeta(acctKV, meta); err != nil { - return nil, logCreateErr(name, accountID, "persist initial meta", err) + // Atomically claim the cluster name before any launching/fallible work. A + // duplicate CreateCluster (the SDK's own retry, or a gateway re-publish that + // spawns a second concurrent handler) loses the claim and is rejected — or, + // for a prior FAILED attempt, reclaims it — so exactly one handler ever + // launches control-plane VMs for a given name. + if err := s.claimClusterName(accountID, acctKV, meta); err != nil { + return nil, err } cpSG, ngSG, err := EnsureClusterSGs(s.deps.VPCSG, accountID, name, vpcID) @@ -708,7 +693,7 @@ func (s *EKSServiceImpl) DeleteCluster(input *eks.DeleteClusterInput, accountID } meta.Status = ClusterStatusDeleting - if err := s.purgeClusterInfra(accountID, name, meta, acctKV); err != nil { + if err := s.purgeClusterInfra(accountID, name, meta, acctKV, true); err != nil { slog.Error("DeleteCluster: teardown incomplete; leaving cluster DELETING for retry", "cluster", name, "err", err) return nil, fmt.Errorf("eks: DeleteCluster %s: %w", name, err) @@ -717,6 +702,80 @@ func (s *EKSServiceImpl) DeleteCluster(input *eks.DeleteClusterInput, accountID return &eks.DeleteClusterOutput{Cluster: clusterMetaToAWS(meta)}, nil } +// claimClusterName atomically takes ownership of the cluster's meta key before +// CreateCluster does any launching work, closing the duplicate-handler race (an +// SDK retry or a gateway re-publish spawning a second concurrent CreateCluster). +// The fresh CREATING meta is laid down with kv.Create: the winner owns it. A +// loser finds the key already present and either rejects (a live cluster) or, +// for a prior FAILED attempt, reclaims it — flipping FAILED→CREATING via CAS so +// exactly one concurrent reclaimer wins, then purging the old attempt's infra +// (keeping the meta key as its lock) before laying down the fresh record. The +// old resource refs stay in the record until teardown confirms the infra is +// gone, so a crash mid-purge leaves them recoverable rather than orphaned. +// Returns an AWS-shaped error ready to hand back to the caller. +func (s *EKSServiceImpl) claimClusterName(accountID string, acctKV nats.KeyValue, meta *ClusterMeta) error { + name := meta.Name + data, err := json.Marshal(meta) + if err != nil { + return logCreateErr(name, accountID, "marshal initial meta", err) + } + + if _, cerr := acctKV.Create(ClusterMetaKey(name), data); cerr == nil { + return nil // won a fresh claim + } else if !errors.Is(cerr, nats.ErrKeyExists) { + return logCreateErr(name, accountID, "claim cluster name", cerr) + } + + // Key already present. Reclaim only a prior FAILED attempt; flip its status + // to CREATING via CAS so exactly one concurrent reclaimer wins. The mutate + // changes only Status, so the old resource refs survive in the record for + // teardown below (and for a retry if we crash before clearing them). + var oldRefs *ClusterMeta + reclaimed := false + if err := casUpdateMeta(acctKV, name, func(m *ClusterMeta) bool { + // casUpdateMeta re-runs this closure on every CAS-conflict retry, so reset + // the outputs each pass: a retry that now observes CREATING (a concurrent + // reclaimer won) must leave reclaimed=false, or we double-own the cluster. + reclaimed = false + oldRefs = nil + if m.Status != ClusterStatusFailed { + return false + } + snapshot := *m + oldRefs = &snapshot + m.Status = ClusterStatusCreating + reclaimed = true + return true + }); err != nil { + if errors.Is(err, ErrClusterNotFound) { + // The FAILED record was swept underneath us (a concurrent reclaim or + // delete). Treat as in-use; the caller may retry. + return errors.New(awserrors.ErrorEKSResourceInUse) + } + return logCreateErr(name, accountID, "reclaim failed cluster", err) + } + if !reclaimed { + slog.Info("CreateCluster: cluster already exists", + "name", name, "accountID", accountID) + return errors.New(awserrors.ErrorEKSResourceInUse) + } + + slog.Info("CreateCluster: reclaiming FAILED cluster before recreate", + "name", name, "accountID", accountID) + // Purge the failed attempt's infra but KEEP the meta claim (deleteMeta=false) + // — the CREATING record is our lock for the rest of this create. + if perr := s.purgeClusterInfra(accountID, name, oldRefs, acctKV, false); perr != nil { + s.markFailed(acctKV, name) + return logCreateErr(name, accountID, "reclaim failed cluster", perr) + } + // Infra gone: overwrite with the fresh CREATING meta (clears the old refs). + if err := PutClusterMeta(acctKV, meta); err != nil { + s.markFailed(acctKV, name) + return logCreateErr(name, accountID, "persist reclaimed meta", err) + } + return nil +} + // purgeClusterInfra tears down a cluster's live AWS resources and erases its // per-cluster KV state. Zeroize (a security guarantee — no recoverable key // material) plus NLB and VM teardown are blocking: their failures are joined @@ -724,7 +783,7 @@ func (s *EKSServiceImpl) DeleteCluster(input *eks.DeleteClusterInput, accountID // without the meta record that owns its ARNs/IDs. SG cleanup is best-effort // (a leaked SG strands no owning record). Shared by DeleteCluster and the // FAILED-cluster reclaim path in CreateCluster. -func (s *EKSServiceImpl) purgeClusterInfra(accountID, name string, meta *ClusterMeta, acctKV nats.KeyValue) error { +func (s *EKSServiceImpl) purgeClusterInfra(accountID, name string, meta *ClusterMeta, acctKV nats.KeyValue, deleteMeta bool) error { s.registry.Stop(accountID, name) var teardownErrs []error @@ -806,8 +865,13 @@ func (s *EKSServiceImpl) purgeClusterInfra(accountID, name string, meta *Cluster s.teardownSpreadGroup(meta) // Only now, with NLB + VM confirmed gone, erase the meta + per-cluster KV. - if err := DeleteClusterPrefix(acctKV, name); err != nil { - return fmt.Errorf("delete cluster prefix: %w", err) + // The reclaim path passes deleteMeta=false: it has already CAS-claimed the + // meta key (status CREATING) as its lock for the recreate, so the key must + // survive — only the failed attempt's infra is purged here. + if deleteMeta { + if err := DeleteClusterPrefix(acctKV, name); err != nil { + return fmt.Errorf("delete cluster prefix: %w", err) + } } return nil } diff --git a/spinifex/handlers/elbv2/name_claim_test.go b/spinifex/handlers/elbv2/name_claim_test.go new file mode 100644 index 000000000..139f0d4cf --- /dev/null +++ b/spinifex/handlers/elbv2/name_claim_test.go @@ -0,0 +1,94 @@ +package handlers_elbv2 + +import ( + "sync" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/elbv2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Two concurrent CreateLoadBalancer calls for the same name resolve to exactly +// one success and one DuplicateLoadBalancerName. The atomic name claim is the +// barrier; under -race a double-claim would surface as a launcher/store race. +func TestCreateLoadBalancer_ConcurrentSameNameSingleOwner(t *testing.T) { + svc := setupTestService(t) + + const n = 2 + errs := make([]error, n) + var wg sync.WaitGroup + for i := range n { + wg.Add(1) + go func(i int) { + defer wg.Done() + _, errs[i] = svc.CreateLoadBalancer(&elbv2.CreateLoadBalancerInput{ + Name: aws.String("race-lb"), + }, testAccountID) + }(i) + } + wg.Wait() + + var ok, dup int + for _, e := range errs { + switch { + case e == nil: + ok++ + case e.Error() == "DuplicateLoadBalancerName": + dup++ + default: + t.Fatalf("unexpected create error: %v", e) + } + } + assert.Equal(t, 1, ok, "exactly one create owns the name") + assert.Equal(t, 1, dup, "the duplicate is rejected DuplicateLoadBalancerName") + + rec, err := svc.store.GetLoadBalancerByName("race-lb", testAccountID) + require.NoError(t, err) + require.NotNil(t, rec) +} + +// A name claim whose owner lbID resolves to no live record is a crashed prior +// create. A fresh CreateLoadBalancer with that name must reclaim the orphan and +// succeed, not wedge on DuplicateLoadBalancerName forever. +func TestCreateLoadBalancer_ReclaimsCrashOrphanNameClaim(t *testing.T) { + svc := setupTestService(t) + + // Seed a name claim pointing at a non-existent LB (crashed create left no + // record). ClaimLBName writes the key with this bogus owner. + ok, dup, err := svc.store.ClaimLBName("orphan-lb", testAccountID, "lb-doesnotexist") + require.NoError(t, err) + require.True(t, ok) + require.False(t, dup) + + // A fresh create for the same name reclaims the orphan and succeeds. + out, err := svc.CreateLoadBalancer(&elbv2.CreateLoadBalancerInput{ + Name: aws.String("orphan-lb"), + }, testAccountID) + require.NoError(t, err) + require.Len(t, out.LoadBalancers, 1) +} + +// Deleting a load balancer releases its name claim so the name is immediately +// reusable. +func TestDeleteLoadBalancer_ReleasesNameForReuse(t *testing.T) { + svc := setupTestService(t) + + out, err := svc.CreateLoadBalancer(&elbv2.CreateLoadBalancerInput{ + Name: aws.String("reuse-lb"), + }, testAccountID) + require.NoError(t, err) + arn := out.LoadBalancers[0].LoadBalancerArn + + _, err = svc.DeleteLoadBalancer(&elbv2.DeleteLoadBalancerInput{ + LoadBalancerArn: arn, + }, testAccountID) + require.NoError(t, err) + + // The name is free: a second create with the same name succeeds. + _, err = svc.CreateLoadBalancer(&elbv2.CreateLoadBalancerInput{ + Name: aws.String("reuse-lb"), + }, testAccountID) + require.NoError(t, err) +} diff --git a/spinifex/handlers/elbv2/service_impl.go b/spinifex/handlers/elbv2/service_impl.go index ab838ffe8..4348ae43f 100644 --- a/spinifex/handlers/elbv2/service_impl.go +++ b/spinifex/handlers/elbv2/service_impl.go @@ -1142,6 +1142,23 @@ func (s *ELBv2ServiceImpl) CreateLoadBalancer(input *elbv2.CreateLoadBalancerInp } dnsName := fmt.Sprintf("%s%s-%s.%s.elb.spinifex.local", dnsPrefix, name, lbID, s.region) + // Atomically claim the LB name before any ENI/VM work. A duplicate + // CreateLoadBalancer (SDK retry / gateway re-publish that spawned a second + // handler) loses the claim and never launches an LB VM; a claim left by a + // crashed prior create (no live record) is reclaimed. Every failure path + // below releases the claim so a failed create does not block the name. + claimOK, claimDup, claimErr := s.store.ClaimLBName(name, accountID, lbID) + if claimErr != nil { + slog.Error("CreateLoadBalancer: name claim failed", "name", name, "err", claimErr) + return nil, errors.New(awserrors.ErrorServerInternal) + } + if claimDup { + return nil, errors.New(awserrors.ErrorELBv2DuplicateLoadBalancer) + } + if !claimOK { + return nil, errors.New(awserrors.ErrorServerInternal) + } + var subnets []string for _, sn := range input.Subnets { if sn != nil { @@ -1173,6 +1190,7 @@ func (s *ELBv2ServiceImpl) CreateLoadBalancer(input *elbv2.CreateLoadBalancerInp sgID, sgErr := s.createNLBManagedSG(lbID, lbArn, subnets[0], accountID) if sgErr != nil { slog.Error("CreateLoadBalancer: failed to create managed NLB SG", "lbId", lbID, "err", sgErr) + s.releaseLBNameClaim(name, accountID) return nil, errors.New(awserrors.ErrorServerInternal) } nlbManagedSGID = sgID @@ -1211,6 +1229,7 @@ func (s *ELBv2ServiceImpl) CreateLoadBalancer(input *elbv2.CreateLoadBalancerInp } } s.deleteNLBManagedSG(nlbManagedSGID, accountID) + s.releaseLBNameClaim(name, accountID) slog.Error("CreateLoadBalancer: failed to create ENI", "subnet", subnetID, "err", eniErr) return nil, errors.New(awserrors.ErrorELBv2SubnetNotFound) } @@ -1298,6 +1317,7 @@ func (s *ELBv2ServiceImpl) CreateLoadBalancer(input *elbv2.CreateLoadBalancerInp if err := s.store.PutLoadBalancer(record); err != nil { slog.Error("CreateLoadBalancer: failed to persist record", "lbId", lbID, "err", err) + s.releaseLBNameClaim(name, accountID) return nil, errors.New(awserrors.ErrorServerInternal) } @@ -1384,11 +1404,24 @@ func (s *ELBv2ServiceImpl) DeleteLoadBalancer(input *elbv2.DeleteLoadBalancerInp return nil, errors.New(awserrors.ErrorServerInternal) } + // Release the name claim so the name is reusable. Idempotent on a missing + // key, so a delete that races the record removal still converges. + s.releaseLBNameClaim(lb.Name, accountID) + slog.Info("DeleteLoadBalancer completed", "lbArn", *input.LoadBalancerArn, "enis", len(lb.ENIs), "accountID", accountID) return &elbv2.DeleteLoadBalancerOutput{}, nil } +// releaseLBNameClaim drops the per-account LB name claim. Failure is logged but +// not fatal: a leaked claim is reclaimed by the crash-orphan path on the next +// same-name create (its owner lbID resolves to no live record). +func (s *ELBv2ServiceImpl) releaseLBNameClaim(name, accountID string) { + if err := s.store.ReleaseLBName(name, accountID); err != nil { + slog.Warn("failed to release LB name claim", "name", name, "accountID", accountID, "err", err) + } +} + // reapFloatingIPNAT removes the OVN dnat_and_snat rule mapping an // internet-facing LB's floating (public) IP to its VM's VPC IP. DeleteEIP // matches on external_ip+logical_ip and is idempotent, so a redundant publish diff --git a/spinifex/handlers/elbv2/store.go b/spinifex/handlers/elbv2/store.go index e2c04c012..0b76867e2 100644 --- a/spinifex/handlers/elbv2/store.go +++ b/spinifex/handlers/elbv2/store.go @@ -22,6 +22,15 @@ const ( KeyPrefixTG = "tg." KeyPrefixListener = "listener." KeyPrefixRule = "rule." + // KeyPrefixLBName is the per-account LB-name claim used for + // CreateLoadBalancer idempotency: it makes the LB name (not the generated + // lbID) an atomically claimable identity so concurrent same-name creates + // cannot both launch an LB VM. + KeyPrefixLBName = "lbname." + + // maxLBNameClaimRetries bounds the crash-orphan CAS-reclaim loop in + // ClaimLBName (Create raced a delete, or a CAS-reclaim lost the revision). + maxLBNameClaimRetries = 5 ) // Store provides CRUD operations for ELBv2 resources backed by JetStream KV. @@ -51,6 +60,64 @@ func NewStore(nc *nats.Conn) (*Store, error) { // --- Load Balancer CRUD --- +// LBNameKey returns the per-account name-claim key for an LB name. +func LBNameKey(name, accountID string) string { + return KeyPrefixLBName + accountID + "." + name +} + +// ClaimLBName atomically claims the LB name for accountID with lbID as owner. +// ok=true ⇒ this caller owns the name and may proceed. dup=true ⇒ a live LB +// already holds the name (a genuine duplicate). A name claim whose owner lbID +// resolves to no live record is a crashed prior create — it is reclaimed for +// this caller (ok=true). This is the idempotency barrier for CreateLoadBalancer: +// a duplicate concurrent create loses the claim and never launches a VM. +func (s *Store) ClaimLBName(name, accountID, lbID string) (ok bool, dup bool, err error) { + key := LBNameKey(name, accountID) + for range maxLBNameClaimRetries { + if _, cerr := s.kv.Create(key, []byte(lbID)); cerr == nil { + return true, false, nil + } else if !errors.Is(cerr, nats.ErrKeyExists) { + return false, false, fmt.Errorf("kv create %s: %w", key, cerr) + } + entry, gerr := s.kv.Get(key) + if gerr != nil { + if errors.Is(gerr, nats.ErrKeyNotFound) { + continue // raced vanish; retry the create + } + return false, false, fmt.Errorf("kv get %s: %w", key, gerr) + } + ownerID := string(entry.Value()) + if ownerID != "" && ownerID != lbID { + rec, rerr := s.GetLoadBalancer(ownerID) + if rerr != nil { + return false, false, fmt.Errorf("resolve LB name owner %s: %w", ownerID, rerr) + } + if rec != nil { + return false, true, nil // live LB holds the name + } + } + // Orphaned (crashed prior create) or already ours: CAS-take the claim. + if _, uerr := s.kv.Update(key, []byte(lbID), entry.Revision()); uerr != nil { + if errors.Is(uerr, nats.ErrKeyExists) { + continue // lost the CAS race; re-read + } + return false, false, fmt.Errorf("kv update %s: %w", key, uerr) + } + return true, false, nil + } + return false, false, fmt.Errorf("elbv2: claim LB name %s exhausted retries", name) +} + +// ReleaseLBName deletes the name claim. A missing key is success (idempotent), +// so create-rollback paths and DeleteLoadBalancer can call it unconditionally. +func (s *Store) ReleaseLBName(name, accountID string) error { + key := LBNameKey(name, accountID) + if err := s.kv.Delete(key); err != nil && !errors.Is(err, nats.ErrKeyNotFound) { + return fmt.Errorf("kv delete %s: %w", key, err) + } + return nil +} + // PutLoadBalancer stores a load balancer record. func (s *Store) PutLoadBalancer(lb *LoadBalancerRecord) error { data, err := json.Marshal(lb)