diff --git a/contracts/admission/v1/README.md b/contracts/admission/v1/README.md new file mode 100644 index 000000000..770d3494a --- /dev/null +++ b/contracts/admission/v1/README.md @@ -0,0 +1,42 @@ +# Admission contract, version 1 + +This package owns the cross-process contract for the admission domain's +quota override. It is not an in-process service interface or the owner of +quota state. + +## Seal account quota override + +`SealAccountQuotaOverrideSubject` (`admission.v1.SealAccountQuotaOverride`) is +a request/reply route on which a gateway, in the `awsgw-quota-seal` queue +group, irreversibly seals one account's quota override record. +`domains/admission/quota` owns it: `awsgw` serves it. +`accountteardown` is the consumer: it calls this subject to fence every +override writer at the quota owner before the account record is removed. + +The request names the account in its body (`account_id`) and may carry an +opaque `correlation_ref` that is echoed for logs. +`awsgw` ignores `X-Account-ID` for this subject: the account sealed is the one +named in the body. +A seal is refused with `IncorrectState` unless IAM already holds the account +as `TERMINATING`; a precondition read failure, or no termination checker +wired, answers `InternalError`. +Both refusals write nothing and never open the override bucket. + +A reply is sent only on success: `disposition` is always `sealed` and +`revision` is the stream leader's accepted write for that seal. +A failure is the standard awserrors error payload, not this reply shape. +Sealing is idempotent: a seal on an already-sealed record re-writes the same +value and returns a fresh, higher revision. +Any current gateway may answer; the leader-accepted revision is the evidence +whichever one does, and the override bucket is opened at seal time even on a +gateway whose quotas were disabled at startup. + +### Compatibility boundary + +The subject name carries the contract major; `account_quota_seal_test.go` +preserves the version-1 JSON shape. +Producer and consumer are in the coordinated Spinifex release, so the window +is N and N-1. +A gateway that predates the subject has no subscription, so the request gets +no responder. +An incompatible change to the subject or fields needs a new major. diff --git a/contracts/admission/v1/account_quota_seal.go b/contracts/admission/v1/account_quota_seal.go new file mode 100644 index 000000000..c23dded3c --- /dev/null +++ b/contracts/admission/v1/account_quota_seal.go @@ -0,0 +1,31 @@ +// Package admissionv1 holds the admission domain's cross-process contract: +// the subjects, request/reply shapes and wire vocabulary the quota owner and +// its consumers share. +package admissionv1 + +// SealAccountQuotaOverrideSubject is the request/reply subject on which a +// gateway seals one account's quota override record. Sealing is irreversible +// and refused unless IAM already holds the account as TERMINATING. +const SealAccountQuotaOverrideSubject = "admission.v1.SealAccountQuotaOverride" + +// AccountQuotaOverrideDispositionSealed is the only disposition a successful +// reply carries: a reply is sent only on success, so there is no other value. +const AccountQuotaOverrideDispositionSealed = "sealed" + +// SealAccountQuotaOverrideRequest asks for one account's quota override to be +// sealed. CorrelationRef is an opaque caller reference echoed for logs; it +// grants nothing and is never a fence or authority to write. +type SealAccountQuotaOverrideRequest struct { + AccountID string `json:"account_id"` + CorrelationRef string `json:"correlation_ref,omitempty"` +} + +// SealAccountQuotaOverrideReply confirms accountID's quota override record +// was sealed at Revision, the stream leader's accepted write for that seal. +// A failure is the standard awserrors error payload, not this type. +type SealAccountQuotaOverrideReply struct { + AccountID string `json:"account_id"` + CorrelationRef string `json:"correlation_ref,omitempty"` + Disposition string `json:"disposition"` + Revision uint64 `json:"revision"` +} diff --git a/contracts/admission/v1/account_quota_seal_test.go b/contracts/admission/v1/account_quota_seal_test.go new file mode 100644 index 000000000..a2292d7bd --- /dev/null +++ b/contracts/admission/v1/account_quota_seal_test.go @@ -0,0 +1,43 @@ +package admissionv1 + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestSealAccountQuotaOverrideSubject(t *testing.T) { + require.Equal(t, "admission.v1.SealAccountQuotaOverride", SealAccountQuotaOverrideSubject) +} + +func TestSealAccountQuotaOverrideRequestJSONContract(t *testing.T) { + data, err := json.Marshal(SealAccountQuotaOverrideRequest{AccountID: "111122223333", CorrelationRef: "req-1"}) + require.NoError(t, err) + require.JSONEq(t, `{"account_id":"111122223333","correlation_ref":"req-1"}`, string(data)) + + data, err = json.Marshal(SealAccountQuotaOverrideRequest{AccountID: "111122223333"}) + require.NoError(t, err) + require.JSONEq(t, `{"account_id":"111122223333"}`, string(data)) +} + +func TestSealAccountQuotaOverrideReplyJSONContract(t *testing.T) { + reply := SealAccountQuotaOverrideReply{ + AccountID: "111122223333", + CorrelationRef: "req-1", + Disposition: AccountQuotaOverrideDispositionSealed, + Revision: 42, + } + data, err := json.Marshal(reply) + require.NoError(t, err) + require.JSONEq(t, `{ + "account_id": "111122223333", + "correlation_ref": "req-1", + "disposition": "sealed", + "revision": 42 + }`, string(data)) + + data, err = json.Marshal(SealAccountQuotaOverrideReply{AccountID: "111122223333", Disposition: AccountQuotaOverrideDispositionSealed}) + require.NoError(t, err) + require.JSONEq(t, `{"account_id":"111122223333","disposition":"sealed","revision":0}`, string(data)) +} diff --git a/spinifex/domains/admission/quota/cleanup.go b/spinifex/domains/admission/quota/cleanup.go deleted file mode 100644 index 494411753..000000000 --- a/spinifex/domains/admission/quota/cleanup.go +++ /dev/null @@ -1,93 +0,0 @@ -package quota - -import ( - "context" - "errors" - "fmt" - - awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" - awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" - "github.com/nats-io/nats.go/jetstream" -) - -// OverrideCleanupDisposition is CleanupAccount's outcome for an account's -// quota override key. -type OverrideCleanupDisposition string - -const ( - OverrideCleanupDeleted OverrideCleanupDisposition = "deleted" - OverrideCleanupAbsent OverrideCleanupDisposition = "absent" - OverrideCleanupFailed OverrideCleanupDisposition = "failed" -) - -// evidenceRevisionGuardedDelete and evidenceObservedAbsent are the fixed -// evidence bases CleanupAccount reports; see kvstore/bucket.go:112-120 for why -// a plain read cannot support a stronger absence claim. -const ( - evidenceRevisionGuardedDelete = "revision-guarded-delete" - evidenceObservedAbsent = "observed-absent-at-scan" -) - -// cleanupMaxAttempts bounds retries against a revision that keeps moving -// under concurrent writes, so cleanup fails closed instead of spinning. -const cleanupMaxAttempts = 5 - -// OverrideCleanup is CleanupAccount's result: what happened to an account's -// quota override key, and the fixed evidence basis for that claim. -type OverrideCleanup struct { - Disposition OverrideCleanupDisposition - Evidence string - Revision uint64 -} - -// CleanupAccount removes accountID's quota override key so a reused account ID -// does not inherit a deleted account's raised or lowered limits. It is -// unwired: nothing calls it yet. -func (s *Service) CleanupAccount(ctx context.Context, accountID string) (*OverrideCleanup, error) { - if accountID == "" { - return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "accountID is required for quota override cleanup") - } - if !awsidentifiers.IsAccountID(accountID) { - return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "accountID %q is not a valid AWS account ID", accountID) - } - if accountID == awsidentifiers.GlobalAccountID { - return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "quota override cleanup must not target the global account") - } - if s == nil || s.overrides == nil { - return &OverrideCleanup{Disposition: OverrideCleanupFailed}, errors.New("quota: override bucket not configured") - } - - // accountID is never resolved through any fallback here: cleanup must act - // on exactly accountID and never fall back to the global key. - for range cleanupMaxAttempts { - entry, err := s.overrides.Get(ctx, accountID) - if err != nil { - if errors.Is(err, jetstream.ErrKeyNotFound) { - return &OverrideCleanup{Disposition: OverrideCleanupAbsent, Evidence: evidenceObservedAbsent}, nil - } - return &OverrideCleanup{Disposition: OverrideCleanupFailed}, fmt.Errorf("get quota override for cleanup: %w", err) - } - rev := entry.Revision() - - err = s.overrides.Delete(ctx, accountID, jetstream.LastRevision(rev)) - switch { - case err == nil: - return &OverrideCleanup{ - Disposition: OverrideCleanupDeleted, - Evidence: evidenceRevisionGuardedDelete, - Revision: rev, - }, nil - case errors.Is(err, jetstream.ErrKeyNotFound): - return &OverrideCleanup{Disposition: OverrideCleanupAbsent, Evidence: evidenceObservedAbsent}, nil - case errors.Is(err, jetstream.ErrKeyRevisionMismatch): - // A concurrent write landed between the read and the guarded - // delete; retry against whatever revision now holds. - continue - default: - return &OverrideCleanup{Disposition: OverrideCleanupFailed}, fmt.Errorf("delete quota override for cleanup: %w", err) - } - } - - return &OverrideCleanup{Disposition: OverrideCleanupFailed}, - fmt.Errorf("quota override cleanup for account: exhausted %d attempts against concurrent writes", cleanupMaxAttempts) -} diff --git a/spinifex/domains/admission/quota/cleanup_test.go b/spinifex/domains/admission/quota/cleanup_test.go deleted file mode 100644 index 42c36252d..000000000 --- a/spinifex/domains/admission/quota/cleanup_test.go +++ /dev/null @@ -1,184 +0,0 @@ -//test:in-package — CleanupAccount deletes through the unexported overrides field, and the retry/race behaviour is only observable from inside the package. -package quota - -import ( - "context" - "errors" - "testing" - - awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" - "github.com/nats-io/nats.go/jetstream" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// overrideCleanupFakeKV wraps a real KeyValue so a test can inject a race -// between CleanupAccount's read and its guarded delete, or a store failure, -// deterministically rather than through goroutine timing. -type overrideCleanupFakeKV struct { - jetstream.KeyValue - - deleteCalls int - beforeDelete func(attempt int) error - getErr error -} - -func (f *overrideCleanupFakeKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) { - if f.getErr != nil { - return nil, f.getErr - } - return f.KeyValue.Get(ctx, key) -} - -func (f *overrideCleanupFakeKV) Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error { - f.deleteCalls++ - if f.beforeDelete != nil { - if err := f.beforeDelete(f.deleteCalls); err != nil { - return err - } - } - return f.KeyValue.Delete(ctx, key, opts...) -} - -func TestCleanupAccount_DeletesTheOverrideKey(t *testing.T) { - svc, _ := newOverrideService(t) - ctx := t.Context() - require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator")) - - result, err := svc.CleanupAccount(ctx, testAccount) - require.NoError(t, err) - assert.Equal(t, OverrideCleanupDeleted, result.Disposition) - assert.Equal(t, evidenceRevisionGuardedDelete, result.Evidence) - assert.NotZero(t, result.Revision) - - _, err = svc.overrides.Get(ctx, testAccount) - assert.ErrorIs(t, err, jetstream.ErrKeyNotFound) -} - -func TestCleanupAccount_RepeatCallIsIdempotent(t *testing.T) { - svc, _ := newOverrideService(t) - ctx := t.Context() - require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator")) - - first, err := svc.CleanupAccount(ctx, testAccount) - require.NoError(t, err) - assert.Equal(t, OverrideCleanupDeleted, first.Disposition) - - second, err := svc.CleanupAccount(ctx, testAccount) - require.NoError(t, err) - assert.Equal(t, OverrideCleanupAbsent, second.Disposition) - assert.Equal(t, evidenceObservedAbsent, second.Evidence) -} - -func TestCleanupAccount_AbsentKeyIsIdempotent(t *testing.T) { - svc, _ := newOverrideService(t) - result, err := svc.CleanupAccount(t.Context(), "333333333333") - require.NoError(t, err) - assert.Equal(t, OverrideCleanupAbsent, result.Disposition) - assert.Equal(t, evidenceObservedAbsent, result.Evidence) -} - -func TestCleanupAccount_OtherAccountsUntouched(t *testing.T) { - svc, _ := newOverrideService(t) - ctx := t.Context() - accountA, accountB := testAccount, "222222222222" - require.NoError(t, svc.PutAccountQuota(ctx, accountA, Overrides{VCPUs: ptr(32)}, "operator")) - require.NoError(t, svc.PutAccountQuota(ctx, accountB, Overrides{VCPUs: ptr(8)}, "operator")) - - result, err := svc.CleanupAccount(ctx, accountA) - require.NoError(t, err) - assert.Equal(t, OverrideCleanupDeleted, result.Disposition) - - over, _, err := svc.GetAccountQuota(ctx, accountB) - require.NoError(t, err) - require.NotNil(t, over.VCPUs) - assert.Equal(t, 8, *over.VCPUs) -} - -func TestCleanupAccount_RefusesGlobalAccountID(t *testing.T) { - svc, _ := newOverrideService(t) - result, err := svc.CleanupAccount(t.Context(), awsidentifiers.GlobalAccountID) - assert.Error(t, err) - assert.Nil(t, result) -} - -func TestCleanupAccount_RefusesEmptyAccountID(t *testing.T) { - svc, _ := newOverrideService(t) - result, err := svc.CleanupAccount(t.Context(), "") - assert.Error(t, err) - assert.Nil(t, result) -} - -func TestCleanupAccount_RefusesInvalidAccountID(t *testing.T) { - svc, _ := newOverrideService(t) - result, err := svc.CleanupAccount(t.Context(), "not-an-account") - assert.Error(t, err) - assert.Nil(t, result) -} - -func TestCleanupAccount_WithoutBucketFails(t *testing.T) { - svc := New(baseLimits(), nil) - result, err := svc.CleanupAccount(t.Context(), testAccount) - assert.Error(t, err) - require.NotNil(t, result) - assert.Equal(t, OverrideCleanupFailed, result.Disposition) -} - -// A concurrent Put landing between CleanupAccount's read and its -// revision-guarded delete must not be lost silently: the delete is rejected -// and retried against the revision the write left behind. -func TestCleanupAccount_ConcurrentPutDuringCleanupRetriesAndDeletes(t *testing.T) { - svc, _ := newOverrideService(t) - ctx := t.Context() - require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator")) - - fake := &overrideCleanupFakeKV{KeyValue: svc.overrides} - fake.beforeDelete = func(attempt int) error { - if attempt != 1 { - return nil - } - // Simulate a racing override write landing after CleanupAccount's - // read but before its guarded delete, through the real KV so the - // delete's expected revision is now stale. - return svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator") - } - svc.overrides = fake - - result, err := svc.CleanupAccount(ctx, testAccount) - require.NoError(t, err) - assert.Equal(t, OverrideCleanupDeleted, result.Disposition) - assert.Equal(t, 2, fake.deleteCalls, "the first delete is rejected by the race and retried once") -} - -// A revision that keeps moving under sustained concurrent writes must not -// spin forever: cleanup gives up and reports failed, bounded by attempts. -func TestCleanupAccount_PersistentConcurrentWritesFailsBounded(t *testing.T) { - svc, _ := newOverrideService(t) - ctx := t.Context() - require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator")) - - fake := &overrideCleanupFakeKV{KeyValue: svc.overrides} - fake.beforeDelete = func(int) error { - return svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator") - } - svc.overrides = fake - - result, err := svc.CleanupAccount(ctx, testAccount) - assert.Error(t, err) - require.NotNil(t, result) - assert.Equal(t, OverrideCleanupFailed, result.Disposition) - assert.Equal(t, cleanupMaxAttempts, fake.deleteCalls) -} - -// A store error must report failed, never absent: absence is a claim about -// what the store holds, which an error gives no basis to make. -func TestCleanupAccount_StoreErrorIsFailedNotAbsent(t *testing.T) { - svc, _ := newOverrideService(t) - fake := &overrideCleanupFakeKV{KeyValue: svc.overrides, getErr: errors.New("kv unavailable")} - svc.overrides = fake - - result, err := svc.CleanupAccount(t.Context(), testAccount) - assert.Error(t, err) - require.NotNil(t, result) - assert.Equal(t, OverrideCleanupFailed, result.Disposition) -} diff --git a/spinifex/domains/admission/quota/live.go b/spinifex/domains/admission/quota/live.go index f4ab03277..b0cf39d96 100644 --- a/spinifex/domains/admission/quota/live.go +++ b/spinifex/domains/admission/quota/live.go @@ -39,7 +39,11 @@ func (s *Service) EnforceVPCs(ctx context.Context, natsConn *nats.Conn, accountI if err != nil { return err } - return exceeds(len(out.Vpcs), want, s.limitsFor(ctx, accountID).VPCs) + limits, err := s.limitsFor(ctx, accountID) + if err != nil { + return err + } + return exceeds(len(out.Vpcs), want, limits.VPCs) } // EnforceSubnets gates CreateSubnet on the account's live DescribeSubnets count. @@ -52,7 +56,11 @@ func (s *Service) EnforceSubnets(ctx context.Context, natsConn *nats.Conn, accou if err != nil { return err } - return exceeds(len(out.Subnets), want, s.limitsFor(ctx, accountID).Subnets) + limits, err := s.limitsFor(ctx, accountID) + if err != nil { + return err + } + return exceeds(len(out.Subnets), want, limits.Subnets) } // EnforceEIPs gates AllocateAddress on the account's live DescribeAddresses count. @@ -64,7 +72,11 @@ func (s *Service) EnforceEIPs(ctx context.Context, natsConn *nats.Conn, accountI if err != nil { return err } - return exceeds(len(out.Addresses), want, s.limitsFor(ctx, accountID).EIPs) + limits, err := s.limitsFor(ctx, accountID) + if err != nil { + return err + } + return exceeds(len(out.Addresses), want, limits.EIPs) } // EnforceLoadBalancers gates CreateLoadBalancer on the account's live count. @@ -77,7 +89,11 @@ func (s *Service) EnforceLoadBalancers(ctx context.Context, natsConn *nats.Conn, if err != nil { return err } - return exceeds(len(out.LoadBalancers), want, s.limitsFor(ctx, accountID).LoadBalancers) + limits, err := s.limitsFor(ctx, accountID) + if err != nil { + return err + } + return exceeds(len(out.LoadBalancers), want, limits.LoadBalancers) } // EnforceRDSInstances gates CreateDBInstance on the account's live count. A DB @@ -91,5 +107,9 @@ func (s *Service) EnforceRDSInstances(ctx context.Context, natsConn *nats.Conn, if err != nil { return err } - return exceeds(len(out.DBInstances), want, s.limitsFor(ctx, accountID).RDSInstances) + limits, err := s.limitsFor(ctx, accountID) + if err != nil { + return err + } + return exceeds(len(out.DBInstances), want, limits.RDSInstances) } diff --git a/spinifex/domains/admission/quota/override.go b/spinifex/domains/admission/quota/override.go index a127728b4..99f23dd86 100644 --- a/spinifex/domains/admission/quota/override.go +++ b/spinifex/domains/admission/quota/override.go @@ -8,12 +8,16 @@ import ( "log/slog" "time" + awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" + "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" "github.com/nats-io/nats.go/jetstream" ) // Overrides raises or lowers individual dimensions for one account. Every field // is a pointer so an explicit 0 (a limit that denies everything) is distinct -// from an absent field, which inherits the configured value. +// from an absent field, which inherits the configured value. This is also the +// admin wire type: an admin request body can carry only these fields, so it can +// never set or clear a stored record's state. type Overrides struct { VCPUs *int `json:"vcpus,omitempty"` VPCs *int `json:"vpcs,omitempty"` @@ -29,11 +33,62 @@ type Overrides struct { UpdatedBy string `json:"updated_by,omitempty"` } -// Empty reports whether no dimension is overridden, in which case the record -// carries nothing but provenance and the account resolves to the config. -func (o Overrides) Empty() bool { - return o.VCPUs == nil && o.VPCs == nil && o.Subnets == nil && o.EIPs == nil && - o.VolumesGiB == nil && o.Volumes == nil && o.RDSInstances == nil && o.LoadBalancers == nil +// OverrideState is a stored override record's lifecycle state. "" is active; +// a record written before the state field existed has none and stays active. +type OverrideState string + +// OverrideStateSealed marks a record no writer may change or recreate. +const OverrideStateSealed OverrideState = "sealed" + +// overrideRecord is the stored form: the admin-settable fields inline, plus a +// state no admin request can read or write. +type overrideRecord struct { + Overrides + + State OverrideState `json:"state,omitempty"` +} + +// errOverridesSealed marks a read or write that found a sealed record, so a +// caller maps it to Blocked without logging it as a fault. +var errOverridesSealed = errors.New(awserrors.ErrorBlocked) + +// errOverridesContended marks a write whose CAS retries were exhausted under +// concurrent writers, bounding the retry loop instead of spinning. +var errOverridesContended = errors.New("quota override: CAS retries exhausted") + +// overrideMaxAttempts bounds a write's retries against a revision that keeps +// moving under concurrent writers, so it fails closed instead of spinning. +const overrideMaxAttempts = 5 + +// checkActive rejects a sealed or unknown-state record; "" is active and +// passes through. A caller must never read or mutate a record that failed +// this as if it were active. +func checkActive(rec overrideRecord) error { + switch rec.State { + case "": + return nil + case OverrideStateSealed: + return errOverridesSealed + default: + return fmt.Errorf("quota override: unknown state %q", rec.State) + } +} + +// mapOverrideError turns a recordFor, PutAccountQuota or clear failure into +// the AWS error a caller returns: sealed is an expected, unlogged Blocked; +// everything else is InternalError, logged with the account ID. +func mapOverrideError(ctx context.Context, accountID string, err error) error { + switch { + case err == nil: + return nil + case errors.Is(err, errOverridesSealed): + return errors.New(awserrors.ErrorBlocked) + case errors.Is(err, errOverridesContended): + slog.ErrorContext(ctx, "Quota override CAS retries exhausted under contention", "accountID", accountID) + default: + slog.ErrorContext(ctx, "Quota override operation failed", "accountID", accountID, "err", err) + } + return errors.New(awserrors.ErrorInternalError) } // apply overlays the set fields onto base, leaving Enabled and the Bedrock @@ -60,72 +115,128 @@ func (o Overrides) apply(base Limits) Limits { return out } -// limitsFor resolves the effective limits for accountID: its overrides where -// set, the configured limits everywhere else. A read failure falls back to the -// configured limits rather than to no limit, so a KV outage cannot lift a cap. -func (s *Service) limitsFor(ctx context.Context, accountID string) Limits { - over, err := s.overridesFor(ctx, accountID) - if err != nil { - slog.WarnContext(ctx, "quota: override read failed, using configured limits", - "account", accountID, "err", err) - return s.limits - } - return over.apply(s.limits) +// Empty reports whether no dimension is overridden, in which case the record +// carries nothing but provenance and the account resolves to the config. +func (o Overrides) Empty() bool { + return o.VCPUs == nil && o.VPCs == nil && o.Subnets == nil && o.EIPs == nil && + o.VolumesGiB == nil && o.Volumes == nil && o.RDSInstances == nil && o.LoadBalancers == nil } -// overridesFor reads accountID's override record. An absent key is the normal -// case for nearly every account and yields the zero value, not an error; only a -// genuine read or decode failure is returned. -func (s *Service) overridesFor(ctx context.Context, accountID string) (Overrides, error) { +// recordFor reads and decodes accountID's stored override record. An absent +// key is the active zero value, not an error; only a genuine read or decode +// failure is returned. +func (s *Service) recordFor(ctx context.Context, accountID string) (overrideRecord, error) { if s == nil || s.overrides == nil { - return Overrides{}, nil + return overrideRecord{}, nil } entry, err := s.overrides.Get(ctx, accountID) if errors.Is(err, jetstream.ErrKeyNotFound) { - return Overrides{}, nil + return overrideRecord{}, nil } if err != nil { - return Overrides{}, fmt.Errorf("read quota override for %s: %w", accountID, err) + return overrideRecord{}, fmt.Errorf("read quota override for %s: %w", accountID, err) } - var over Overrides - if err := json.Unmarshal(entry.Value(), &over); err != nil { - return Overrides{}, fmt.Errorf("decode quota override for %s: %w", accountID, err) + var rec overrideRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + return overrideRecord{}, fmt.Errorf("decode quota override for %s: %w", accountID, err) } - return over, nil + return rec, nil } -// GetAccountQuota returns accountID's stored overrides and the limits they -// resolve to. A read failure is surfaced here rather than masked, because an -// operator inspecting a quota must not be shown the fallback as if it were real. +// limitsFor resolves accountID's effective limits: overrides where set, the +// config elsewhere. A read or decode failure falls back to the config so an +// outage cannot lift a cap; a sealed or unknown-state record refuses instead. +func (s *Service) limitsFor(ctx context.Context, accountID string) (Limits, error) { + rec, err := s.recordFor(ctx, accountID) + if err != nil { + slog.WarnContext(ctx, "quota: override read failed, using configured limits", + "account", accountID, "err", err) + return s.limits, nil + } + if err := checkActive(rec); err != nil { + return Limits{}, mapOverrideError(ctx, accountID, err) + } + return rec.apply(s.limits), nil +} + +// GetAccountQuota returns accountID's stored overrides and resolved limits. +// Read failures surface rather than show the fallback as real; a sealed or +// unknown-state record returns Blocked or InternalError. func (s *Service) GetAccountQuota(ctx context.Context, accountID string) (Overrides, Limits, error) { - over, err := s.overridesFor(ctx, accountID) + rec, err := s.recordFor(ctx, accountID) if err != nil { return Overrides{}, Limits{}, err } - return over, over.apply(s.limits), nil + if err := checkActive(rec); err != nil { + return Overrides{}, Limits{}, mapOverrideError(ctx, accountID, err) + } + return rec.Overrides, rec.apply(s.limits), nil } -// PutAccountQuota stores accountID's overrides, stamping provenance. An empty -// set deletes the record so the account resolves to the config, which keeps -// "has this account been touched?" a single existence check. +// PutAccountQuota stores or clears accountID's overrides, stamping provenance +// on a store. Both branches are revision-guarded and refuse a sealed record, +// so a seal landing between a read and a write fails the CAS. func (s *Service) PutAccountQuota(ctx context.Context, accountID string, over Overrides, updatedBy string) error { if s == nil || s.overrides == nil { return errors.New("quota: override bucket not configured") } if over.Empty() { - if err := s.overrides.Delete(ctx, accountID); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { - return fmt.Errorf("clear quota override for %s: %w", accountID, err) - } - return nil + return s.clearAccountQuota(ctx, accountID) } + over.UpdatedAt = time.Now().UTC().Format(time.RFC3339) over.UpdatedBy = updatedBy - data, err := json.Marshal(over) - if err != nil { - return fmt.Errorf("encode quota override for %s: %w", accountID, err) - } - if _, err := s.overrides.Put(ctx, accountID, data); err != nil { - return fmt.Errorf("store quota override for %s: %w", accountID, err) - } - return nil + + _, err := kvutil.Update(ctx, s.overrides, accountID, kvutil.CASConfig{ + CreateIfAbsent: true, + Attempts: overrideMaxAttempts, + Exhausted: func(string, int) error { return errOverridesContended }, + }, func(rec *overrideRecord) (bool, error) { + if err := checkActive(*rec); err != nil { + return false, err + } + rec.Overrides = over + // Every success is a leader-accepted write over an active value: a + // seal landing between the read and the write fails this CAS. + return true, nil + }) + return mapOverrideError(ctx, accountID, err) +} + +// clearAccountQuota deletes an active override at the revision it read. An +// observed-absent key writes nothing, so a stale read cannot unseal; a sealed +// record refuses rather than being erased and later recreated active. +func (s *Service) clearAccountQuota(ctx context.Context, accountID string) error { + for range overrideMaxAttempts { + entry, err := s.overrides.Get(ctx, accountID) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil + } + if err != nil { + return mapOverrideError(ctx, accountID, fmt.Errorf("read quota override for clear: %w", err)) + } + + var rec overrideRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + return mapOverrideError(ctx, accountID, fmt.Errorf("decode quota override for clear: %w", err)) + } + if err := checkActive(rec); err != nil { + return mapOverrideError(ctx, accountID, err) + } + + err = s.overrides.Delete(ctx, accountID, jetstream.LastRevision(entry.Revision())) + switch { + case err == nil: + return nil + case errors.Is(err, jetstream.ErrKeyNotFound): + return nil + case isRevisionConflict(err): + // A concurrent write landed between the read and the guarded + // delete; retry against whatever revision now holds. + continue + default: + return mapOverrideError(ctx, accountID, fmt.Errorf("clear quota override: %w", err)) + } + } + return mapOverrideError(ctx, accountID, errOverridesContended) } diff --git a/spinifex/domains/admission/quota/override_kv_test.go b/spinifex/domains/admission/quota/override_kv_test.go index 5c0c1eebd..3ee302a0c 100644 --- a/spinifex/domains/admission/quota/override_kv_test.go +++ b/spinifex/domains/admission/quota/override_kv_test.go @@ -31,7 +31,8 @@ func newOverrideService(t *testing.T) (*Service, jetstream.KeyValue) { func TestLimitsForAbsentKeyInherits(t *testing.T) { svc, _ := newOverrideService(t) - got := svc.limitsFor(t.Context(), "123456789012") + got, err := svc.limitsFor(t.Context(), "123456789012") + require.NoError(t, err) require.Equal(t, baseLimits(), got) } @@ -44,10 +45,14 @@ func TestLimitsForIsPerAccount(t *testing.T) { require.NoError(t, svc.PutAccountQuota(t.Context(), upgraded, Overrides{VCPUs: ptr(32), RDSInstances: ptr(8)}, "operator")) - require.Equal(t, 32, svc.limitsFor(t.Context(), upgraded).VCPUs) - require.Equal(t, 8, svc.limitsFor(t.Context(), upgraded).RDSInstances) - require.Equal(t, baseLimits().VCPUs, svc.limitsFor(t.Context(), sandbox).VCPUs) - require.Equal(t, baseLimits().RDSInstances, svc.limitsFor(t.Context(), sandbox).RDSInstances) + upgradedLimits, err := svc.limitsFor(t.Context(), upgraded) + require.NoError(t, err) + sandboxLimits, err := svc.limitsFor(t.Context(), sandbox) + require.NoError(t, err) + require.Equal(t, 32, upgradedLimits.VCPUs) + require.Equal(t, 8, upgradedLimits.RDSInstances) + require.Equal(t, baseLimits().VCPUs, sandboxLimits.VCPUs) + require.Equal(t, baseLimits().RDSInstances, sandboxLimits.RDSInstances) } // A stored override must round-trip through KV with its provenance, and an @@ -79,7 +84,9 @@ func TestPutAccountQuotaEmptyClears(t *testing.T) { _, err := kv.Get(t.Context(), account) require.ErrorIs(t, err, jetstream.ErrKeyNotFound) - require.Equal(t, baseLimits(), svc.limitsFor(t.Context(), account)) + limits, err := svc.limitsFor(t.Context(), account) + require.NoError(t, err) + require.Equal(t, baseLimits(), limits) } // Clearing an account that never had a record must succeed rather than fail on @@ -97,8 +104,10 @@ func TestPutAccountQuotaExplicitZero(t *testing.T) { require.NoError(t, svc.PutAccountQuota(t.Context(), account, Overrides{VPCs: ptr(0)}, "operator")) - require.Equal(t, 0, svc.limitsFor(t.Context(), account).VPCs) - require.Error(t, exceeds(0, 1, svc.limitsFor(t.Context(), account).VPCs)) + limits, err := svc.limitsFor(t.Context(), account) + require.NoError(t, err) + require.Equal(t, 0, limits.VPCs) + require.Error(t, exceeds(0, 1, limits.VPCs)) } // A corrupt record is a read failure, not an absent one: it must fall back to @@ -110,7 +119,9 @@ func TestLimitsForCorruptRecordFailsClosed(t *testing.T) { _, err := kv.Put(t.Context(), account, []byte("{not json")) require.NoError(t, err) - require.Equal(t, baseLimits(), svc.limitsFor(t.Context(), account), + limits, err := svc.limitsFor(t.Context(), account) + require.NoError(t, err, "a decode failure must fall back, not error, from limitsFor") + require.Equal(t, baseLimits(), limits, "a corrupt override must fall back to the configured limits") _, _, err = svc.GetAccountQuota(t.Context(), account) diff --git a/spinifex/domains/admission/quota/override_test.go b/spinifex/domains/admission/quota/override_test.go index fa23fe933..2a3c2f6f4 100644 --- a/spinifex/domains/admission/quota/override_test.go +++ b/spinifex/domains/admission/quota/override_test.go @@ -126,7 +126,11 @@ func TestOverridesEmpty(t *testing.T) { // fall through to the configured limits rather than panicking. func TestLimitsForWithoutBucket(t *testing.T) { svc := New(baseLimits(), nil) - if got := svc.limitsFor(t.Context(), "123456789012"); got != baseLimits() { + got, err := svc.limitsFor(t.Context(), "123456789012") + if err != nil { + t.Fatalf("limitsFor error = %v, want nil", err) + } + if got != baseLimits() { t.Fatalf("limitsFor = %+v, want the configured limits", got) } } diff --git a/spinifex/domains/admission/quota/override_writes_guarded_test.go b/spinifex/domains/admission/quota/override_writes_guarded_test.go new file mode 100644 index 000000000..8e33c8228 --- /dev/null +++ b/spinifex/domains/admission/quota/override_writes_guarded_test.go @@ -0,0 +1,75 @@ +package quota + +//test:in-package — pins what AccountQuotaOverrideSealCapabilityVersion 1 will +// assert: nothing in this package writes the override bucket except through +// PutAccountQuota's CAS loop, clearAccountQuota's guarded delete or +// SealAccountOverrides' CAS loop. + +import ( + "go/ast" + "go/parser" + "go/token" + "path/filepath" + "runtime" + "testing" +) + +// overrideWriteFiles are the only files in this package that touch the +// override bucket. The package also holds the vCPU usage and Bedrock usage +// buckets, which write through their own stores and are out of scope here; +// scanning the whole package directory, as the EC2 settings equivalent does, +// would false-flag those unrelated writes. +var overrideWriteFiles = []string{"override.go", "seal.go"} + +func quotaPackageDir(t *testing.T) string { + t.Helper() + _, file, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("cannot locate test source") + } + return filepath.Dir(file) +} + +// No file owning the override bucket may call Put, PutString, or Delete +// without a revision option: every write must go through PutAccountQuota's +// CAS loop, clearAccountQuota's guarded delete, or SealAccountOverrides' CAS +// loop. +func TestOverrideWritesAreRevisionGuarded(t *testing.T) { + root := quotaPackageDir(t) + fset := token.NewFileSet() + scanned := 0 + + for _, name := range overrideWriteFiles { + path := filepath.Join(root, name) + f, err := parser.ParseFile(fset, path, nil, 0) + if err != nil { + t.Fatalf("parse %s: %v", path, err) + } + scanned++ + + ast.Inspect(f, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + sel, ok := call.Fun.(*ast.SelectorExpr) + if !ok { + return true + } + switch sel.Sel.Name { + case "Put", "PutString": + t.Errorf("%s: unconditional %s call bypasses the revision-guarded override write; use PutAccountQuota or SealAccountOverrides instead", + fset.Position(call.Pos()), sel.Sel.Name) + case "Delete": + if len(call.Args) < 3 { + t.Errorf("%s: Delete call with no revision option bypasses the guarded clear; pin it to the read revision", + fset.Position(call.Pos())) + } + } + return true + }) + } + if scanned == 0 { + t.Fatalf("scanned no override-write files under %s; the invariant would pass vacuously", root) + } +} diff --git a/spinifex/domains/admission/quota/quota.go b/spinifex/domains/admission/quota/quota.go index e6e1b0c6f..8eafc3198 100644 --- a/spinifex/domains/admission/quota/quota.go +++ b/spinifex/domains/admission/quota/quota.go @@ -90,6 +90,14 @@ type Service struct { // overrides holds the per-account limit overrides. Nil leaves every account // on the configured limits, which is also what an absent key means. overrides jetstream.KeyValue + + // js opens the override bucket for SealAccountOverrides when quotas were + // disabled at startup; nil makes such a seal refuse rather than create it. + js jetstream.KeyValueManager + + // terminationChecker is the seal precondition; SealAccountOverrides + // refuses until it is wired. + terminationChecker AccountTerminationChecker } // New constructs a quota Service from the configured limits and the gateway-owned @@ -125,6 +133,16 @@ func (s *Service) SetOverrides(kv jetstream.KeyValue) { s.overrides = kv } +// SetJetStream wires the client SealAccountOverrides uses to get-or-create +// the override bucket when this gateway disabled quotas. Called once at +// wiring time, regardless of whether quotas are enabled here. +func (s *Service) SetJetStream(js jetstream.KeyValueManager) { + if s == nil { + return + } + s.js = js +} + // Exempt returns true for the global/system account and whenever quotas are // disabled, so those callers bypass every quota check. A nil Service (quotas // never built) is treated as disabled so callers need no nil guard. diff --git a/spinifex/domains/admission/quota/seal.go b/spinifex/domains/admission/quota/seal.go new file mode 100644 index 000000000..be2e8db56 --- /dev/null +++ b/spinifex/domains/admission/quota/seal.go @@ -0,0 +1,163 @@ +package quota + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + + awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" + awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" + "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" + "github.com/nats-io/nats.go/jetstream" +) + +// sealMaxAttempts bounds SealAccountOverrides' retry loop against a revision +// that keeps moving under concurrent writes, so it fails closed instead of +// spinning. +const sealMaxAttempts = 5 + +// sealedRecord is the fixed value a seal writes. No dimension or provenance +// survives, so no tenant commercial setting outlives the seal. +var sealedRecord = &overrideRecord{State: OverrideStateSealed} + +// OverrideSeal is SealAccountOverrides' result: the revision at which the +// stream leader accepted the sealed value. +type OverrideSeal struct { + Revision uint64 +} + +// AccountTerminationChecker reports whether IAM holds accountID as +// TERMINATING. A missing record is (false, nil); only a failed read is an +// error. +type AccountTerminationChecker interface { + AccountTerminating(ctx context.Context, accountID string) (bool, error) +} + +// SetAccountTerminationChecker wires the seal precondition; until it is set, +// SealAccountOverrides refuses. +func (s *Service) SetAccountTerminationChecker(c AccountTerminationChecker) { + if s == nil { + return + } + s.terminationChecker = c +} + +// checkTerminating runs before any override-KV call, so a refusal never +// writes or opens the bucket. Not TERMINATING (incl. missing) is +// IncorrectState; a read failure or unwired checker is InternalError. +func (s *Service) checkTerminating(ctx context.Context, accountID string) error { + if s.terminationChecker == nil { + slog.ErrorContext(ctx, "Quota override seal has no termination checker wired", "accountID", accountID) + return errors.New(awserrors.ErrorInternalError) + } + terminating, err := s.terminationChecker.AccountTerminating(ctx, accountID) + if err != nil { + slog.ErrorContext(ctx, "Failed to read account status for seal precondition", "accountID", accountID, "err", err) + return errors.New(awserrors.ErrorInternalError) + } + if !terminating { + return awserrors.Errorf(awserrors.ErrorIncorrectState, "account %s is not TERMINATING", accountID) + } + return nil +} + +// isRevisionConflict reports a lost CAS race to retry, including the bare +// wrong-last-sequence APIError from Create's retry against a delete marker. +// A private copy of domains/ec2/account's helper, which this package may not import. +func isRevisionConflict(err error) bool { + if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + return true + } + if apiErr, ok := errors.AsType[*jetstream.APIError](err); ok { + return apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence || + apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequenceConstant + } + return false +} + +// sealBucket returns this gateway's override bucket, or get-or-creates it +// when quotas are disabled here. Called only after checkTerminating accepts, +// so a refused seal never creates the bucket. +func (s *Service) sealBucket(ctx context.Context) (jetstream.KeyValue, error) { + if s.overrides != nil { + return s.overrides, nil + } + if s.js == nil { + slog.ErrorContext(ctx, "Quota override seal has no JetStream client wired to open the override bucket") + return nil, errors.New(awserrors.ErrorInternalError) + } + kv, err := kvutil.GetOrCreateBucket(ctx, s.js, KVBucketAccountQuota, 1) + if err != nil { + slog.ErrorContext(ctx, "Failed to open quota override bucket for seal", "err", err) + return nil, errors.New(awserrors.ErrorInternalError) + } + return kv, nil +} + +// SealAccountOverrides seals accountID's override record so no later writer +// can recreate it. Success is only a leader-accepted write, evidenced by the +// revision; a missing key is created sealed, never reported from absence. +func (s *Service) SealAccountOverrides(ctx context.Context, accountID string) (*OverrideSeal, error) { + if accountID == "" { + return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "accountID is required to seal quota overrides") + } + if !awsidentifiers.IsAccountID(accountID) { + return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "accountID %q is not a valid AWS account ID", accountID) + } + if accountID == awsidentifiers.GlobalAccountID { + return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "quota override seal must not target the global account") + } + + if err := s.checkTerminating(ctx, accountID); err != nil { + return nil, err + } + + overrides, err := s.sealBucket(ctx) + if err != nil { + return nil, err + } + + data, err := json.Marshal(sealedRecord) + if err != nil { + slog.ErrorContext(ctx, "Failed to encode sealed quota override record", "accountID", accountID, "err", err) + return nil, errors.New(awserrors.ErrorInternalError) + } + + // accountID is never resolved through any fallback here: a seal must act + // on exactly accountID and never fall back to the global key. + for range sealMaxAttempts { + var rev uint64 + entry, getErr := overrides.Get(ctx, accountID) + switch { + case getErr == nil: + rev = entry.Revision() + case errors.Is(getErr, jetstream.ErrKeyNotFound): + rev = 0 + default: + slog.ErrorContext(ctx, "Failed to read quota override for seal", "accountID", accountID, "err", getErr) + return nil, errors.New(awserrors.ErrorInternalError) + } + + var newRev uint64 + var writeErr error + if rev == 0 { + newRev, writeErr = overrides.Create(ctx, accountID, data) + } else { + newRev, writeErr = overrides.Update(ctx, accountID, data, rev) + } + + switch { + case writeErr == nil: + return &OverrideSeal{Revision: newRev}, nil + case isRevisionConflict(writeErr): + continue + default: + slog.ErrorContext(ctx, "Failed to write sealed quota override record", "accountID", accountID, "err", writeErr) + return nil, errors.New(awserrors.ErrorInternalError) + } + } + + slog.ErrorContext(ctx, "Quota override seal exhausted attempts against concurrent writes", "accountID", accountID, "attempts", sealMaxAttempts) + return nil, errors.New(awserrors.ErrorInternalError) +} diff --git a/spinifex/domains/admission/quota/seal_test.go b/spinifex/domains/admission/quota/seal_test.go new file mode 100644 index 000000000..dfc9701c6 --- /dev/null +++ b/spinifex/domains/admission/quota/seal_test.go @@ -0,0 +1,806 @@ +package quota + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "testing" + + "github.com/mulgadc/spinifex/internal/testkit" + awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" + awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" + "github.com/nats-io/nats.go/jetstream" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// stubTerminationChecker answers AccountTerminating from a fixed set, for +// tests that exercise the KV paths rather than IAM itself. +type stubTerminationChecker struct { + terminating map[string]bool + err error +} + +func (c *stubTerminationChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) { + if c.err != nil { + return false, c.err + } + return c.terminating[accountID], nil +} + +// terminatingChecker answers true for exactly accountID. +func terminatingChecker(accountID string) *stubTerminationChecker { + return &stubTerminationChecker{terminating: map[string]bool{accountID: true}} +} + +// newSealService builds a Service over a real JetStream KV bucket with the +// override bucket already attached, mirroring a quota-enabled gateway. +func newSealService(t *testing.T) (*Service, jetstream.JetStream) { + t.Helper() + _, _, js := testutil.StartTestJetStream(t) + kv, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{ + Bucket: KVBucketAccountQuota, + History: 1, + }) + require.NoError(t, err) + usage, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{ + Bucket: KVBucketAccountUsage, + History: 1, + }) + require.NoError(t, err) + + svc := New(baseLimits(), usage) + svc.SetOverrides(kv) + svc.SetJetStream(js) + return svc, js +} + +// newDisabledSealService builds a Service with only the JetStream client +// wired, mirroring a quota-disabled gateway: the override bucket does not +// exist until a seal creates it. +func newDisabledSealService(t *testing.T) (*Service, jetstream.JetStream) { + t.Helper() + _, _, js := testutil.StartTestJetStream(t) + svc := New(baseLimits(), nil) + svc.SetJetStream(js) + return svc, js +} + +// lastOverrideSeq reports the override stream's last sequence, so a test can +// assert a failed attempt wrote nothing. +func lastOverrideSeq(t *testing.T, js jetstream.JetStream) uint64 { + t.Helper() + stream, err := js.Stream(context.Background(), "KV_"+KVBucketAccountQuota) + require.NoError(t, err) + info, err := stream.Info(context.Background()) + require.NoError(t, err) + return info.State.LastSeq +} + +// mutationCase is one of PutAccountQuota's two mutations, called uniformly so +// a test can table over them. +type mutationCase struct { + name string + call func(svc *Service, ctx context.Context) error +} + +func mutationTable() []mutationCase { + return []mutationCase{ + {"Set", func(svc *Service, ctx context.Context) error { + return svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator") + }}, + {"Clear", func(svc *Service, ctx context.Context) error { + return svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator") + }}, + } +} + +// seedOverrideState prepares testAccount's record for a race test: present +// (an active record), absent (never written) or tombstone (written then +// deleted, leaving a delete marker Get hides). +func seedOverrideState(t *testing.T, svc *Service, state string) { + t.Helper() + ctx := context.Background() + switch state { + case "present": + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")) + case "tombstone": + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")) + require.NoError(t, svc.overrides.Delete(ctx, testAccount)) + case "absent": + default: + t.Fatalf("unknown seed state %q", state) + } +} + +// sealRaceKV wraps a real KeyValue so a test can inject a deterministic race +// between a caller's read and its guarded write (Create, Update or Delete), +// instead of relying on goroutine timing. +type sealRaceKV struct { + jetstream.KeyValue + + writeCalls int + beforeWrite func(attempt int) error +} + +func (f *sealRaceKV) Create(ctx context.Context, key string, value []byte, opts ...jetstream.KVCreateOpt) (uint64, error) { + f.writeCalls++ + if f.beforeWrite != nil { + if err := f.beforeWrite(f.writeCalls); err != nil { + return 0, err + } + } + return f.KeyValue.Create(ctx, key, value, opts...) +} + +func (f *sealRaceKV) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) { + f.writeCalls++ + if f.beforeWrite != nil { + if err := f.beforeWrite(f.writeCalls); err != nil { + return 0, err + } + } + return f.KeyValue.Update(ctx, key, value, revision) +} + +func (f *sealRaceKV) Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error { + f.writeCalls++ + if f.beforeWrite != nil { + if err := f.beforeWrite(f.writeCalls); err != nil { + return err + } + } + return f.KeyValue.Delete(ctx, key, opts...) +} + +// staleGetKV always answers Get with a fixed snapshot taken at construction, +// so a caller's CAS keeps retrying against a revision that can never win. +type staleGetKV struct { + jetstream.KeyValue + + entry jetstream.KeyValueEntry +} + +func (f *staleGetKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) { + return f.entry, nil +} + +// staleNotFoundKV always answers Get as absent, regardless of what the real +// bucket holds. +type staleNotFoundKV struct { + jetstream.KeyValue +} + +func (f *staleNotFoundKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) { + return nil, jetstream.ErrKeyNotFound +} + +// errGetKV always fails Get with a fixed store error. +type errGetKV struct { + jetstream.KeyValue + + err error +} + +func (f *errGetKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) { + return nil, f.err +} + +func decodeOverride(t *testing.T, entry jetstream.KeyValueEntry) overrideRecord { + t.Helper() + var rec overrideRecord + require.NoError(t, json.Unmarshal(entry.Value(), &rec)) + return rec +} + +// A seal landing between a mutation's read and its guarded write is not +// outlived: the mutation's retry sees the seal and is rejected, and nothing +// it wrote survives. Clear on an absent or tombstoned key never reaches a +// write at all (see TestClearOnMissingKeyWritesNothing), so only the four +// cells with a real write are exercised here. +func TestSealRaceMutation_SealLandsBetweenReadAndWrite(t *testing.T) { + cases := []struct { + name string + state string + call func(svc *Service, ctx context.Context) error + }{ + {"Set_present", "present", mutationTable()[0].call}, + {"Set_absent", "absent", mutationTable()[0].call}, + {"Set_tombstone", "tombstone", mutationTable()[0].call}, + {"Clear_present", "present", mutationTable()[1].call}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + seedOverrideState(t, svc, tc.state) + + realKV := svc.overrides + sealer := New(baseLimits(), nil) + sealer.SetOverrides(realKV) + sealer.SetAccountTerminationChecker(terminatingChecker(testAccount)) + + var sealed *OverrideSeal + race := &sealRaceKV{KeyValue: realKV} + race.beforeWrite = func(attempt int) error { + if attempt != 1 { + return nil + } + var sealErr error + sealed, sealErr = sealer.SealAccountOverrides(ctx, testAccount) + return sealErr + } + svc.overrides = race + + err := tc.call(svc, ctx) + require.Error(t, err) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "got %v", err) + require.NotNil(t, sealed) + + entry, err := realKV.Get(ctx, testAccount) + require.NoError(t, err) + assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State) + assert.Equal(t, sealed.Revision, entry.Revision(), "no write landed after the seal") + }) + } +} + +// A mutation landing inside a seal's read-write window does not survive it: +// the seal retries against the mutation's write and wins, leaving the sealed +// value with no overridden dimension. +func TestSealRaceMutation_MutationLandsBetweenSealReadAndWrite(t *testing.T) { + for _, m := range mutationTable() { + for _, state := range []string{"present", "absent", "tombstone"} { + t.Run(m.name+"_"+state, func(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + seedOverrideState(t, svc, state) + + realKV := svc.overrides + race := &sealRaceKV{KeyValue: realKV} + race.beforeWrite = func(attempt int) error { + if attempt != 1 { + return nil + } + _ = m.call(svc, ctx) + return nil + } + + sealer := New(baseLimits(), nil) + sealer.SetOverrides(race) + sealer.SetAccountTerminationChecker(terminatingChecker(testAccount)) + + seal, err := sealer.SealAccountOverrides(ctx, testAccount) + require.NoError(t, err) + + entry, err := realKV.Get(ctx, testAccount) + require.NoError(t, err) + rec := decodeOverride(t, entry) + assert.Equal(t, OverrideStateSealed, rec.State) + assert.Nil(t, rec.VCPUs) + assert.Equal(t, seal.Revision, entry.Revision()) + }) + } + } +} + +// Neither mutation can act on an already-sealed record. +func TestMutationOnSealedRecordIsRejected(t *testing.T) { + for _, m := range mutationTable() { + t.Run(m.name, func(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + _, err := svc.SealAccountOverrides(ctx, testAccount) + require.NoError(t, err) + + before, err := svc.overrides.Get(ctx, testAccount) + require.NoError(t, err) + + err = m.call(svc, ctx) + require.Error(t, err) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "got %v", err) + + after, err := svc.overrides.Get(ctx, testAccount) + require.NoError(t, err) + assert.Equal(t, before.Revision(), after.Revision()) + }) + } +} + +// Every override mutation races one seal on a real bucket; afterwards the +// record is sealed and stays sealed, so no success among them could have +// landed after the seal. +func TestSealConcurrentMutations(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")) + + mutations := mutationTable() + errs := make([]error, len(mutations)) + var wg sync.WaitGroup + for i, m := range mutations { + wg.Add(1) + go func(i int, m mutationCase) { + defer wg.Done() + errs[i] = m.call(svc, ctx) + }(i, m) + } + + var sealResult *OverrideSeal + var sealErr error + wg.Go(func() { + sealResult, sealErr = svc.SealAccountOverrides(ctx, testAccount) + }) + wg.Wait() + + require.NoError(t, sealErr) + require.NotNil(t, sealResult) + for i, err := range errs { + if err != nil { + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "mutation %s: got %v", mutations[i].name, err) + } + } + + entry, err := svc.overrides.Get(ctx, testAccount) + require.NoError(t, err) + assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State) + assert.Equal(t, sealResult.Revision, entry.Revision()) + + _, _, err = svc.GetAccountQuota(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked)) +} + +// A Get pinned to a revision the record has already moved past exhausts the +// mutation's retries without ever writing. +func TestMutationStaleReadExhaustsWithoutWriting(t *testing.T) { + svc, js := newSealService(t) + ctx := context.Background() + + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")) + stale, err := svc.overrides.Get(ctx, testAccount) + require.NoError(t, err) + + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(16)}, "operator")) + + before := lastOverrideSeq(t, js) + svc.overrides = &staleGetKV{KeyValue: svc.overrides, entry: stale} + + err = svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(32)}, "operator") + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Equal(t, before, lastOverrideSeq(t, js)) +} + +// A Get that always reports absent, while a sealed record actually exists, +// never gets to recreate it: every Create attempt conflicts. +func TestMutationStaleAbsentDoesNotRecreate(t *testing.T) { + svc, js := newSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + _, err := svc.SealAccountOverrides(ctx, testAccount) + require.NoError(t, err) + + before := lastOverrideSeq(t, js) + svc.overrides = &staleNotFoundKV{KeyValue: svc.overrides} + + err = svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator") + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Equal(t, before, lastOverrideSeq(t, js)) +} + +// A revision that keeps moving under sustained concurrent writes must not let +// the seal spin forever: it gives up, bounded by its attempt budget, and its +// own write never lands. +func TestSealSustainedConflictFailsWithoutWriting(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")) + + realKV := svc.overrides + race := &sealRaceKV{KeyValue: realKV} + race.beforeWrite = func(int) error { + data, merr := json.Marshal(&overrideRecord{Overrides: Overrides{VCPUs: ptr(16)}}) + require.NoError(t, merr) + _, perr := realKV.Put(ctx, testAccount, data) + return perr + } + svc.overrides = race + + _, err := svc.SealAccountOverrides(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Equal(t, sealMaxAttempts, race.writeCalls) + + entry, err := realKV.Get(ctx, testAccount) + require.NoError(t, err) + assert.NotEqual(t, OverrideStateSealed, decodeOverride(t, entry).State, "the seal's own write never landed") +} + +// A store error on the seal's read reports failed, never written. +func TestSealStoreErrorWritesNothing(t *testing.T) { + svc, js := newSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + + before := lastOverrideSeq(t, js) + svc.overrides = &errGetKV{KeyValue: svc.overrides, err: errors.New("kv unavailable")} + + _, err := svc.SealAccountOverrides(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Equal(t, before, lastOverrideSeq(t, js)) +} + +func TestIsRevisionConflictRecognisesTombstoneRetry(t *testing.T) { + assert.True(t, isRevisionConflict(jetstream.ErrKeyExists)) + assert.True(t, isRevisionConflict(jetstream.ErrKeyRevisionMismatch)) + assert.True(t, isRevisionConflict(&jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequence})) + assert.True(t, isRevisionConflict(&jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequenceConstant})) + assert.True(t, isRevisionConflict(fmt.Errorf("wrapped: %w", &jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequence}))) + assert.False(t, isRevisionConflict(errors.New("unrelated failure"))) + assert.False(t, isRevisionConflict(nil)) +} + +// Clearing never writes when there is nothing active to clear: an absent key, +// a tombstoned key, and a sealed key hidden behind a stale not-found Get all +// leave the stream exactly where they found it. +func TestClearOnMissingKeyWritesNothing(t *testing.T) { + t.Run("absent", func(t *testing.T) { + svc, js := newSealService(t) + ctx := context.Background() + before := lastOverrideSeq(t, js) + + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator")) + assert.Equal(t, before, lastOverrideSeq(t, js)) + }) + + t.Run("tombstone", func(t *testing.T) { + svc, js := newSealService(t) + ctx := context.Background() + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")) + require.NoError(t, svc.overrides.Delete(ctx, testAccount)) + before := lastOverrideSeq(t, js) + + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator")) + assert.Equal(t, before, lastOverrideSeq(t, js)) + }) + + t.Run("sealed_hidden_by_stale_not_found", func(t *testing.T) { + svc, js := newSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + sealed, err := svc.SealAccountOverrides(ctx, testAccount) + require.NoError(t, err) + before := lastOverrideSeq(t, js) + + realKV := svc.overrides + svc.overrides = &staleNotFoundKV{KeyValue: realKV} + + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator")) + assert.Equal(t, before, lastOverrideSeq(t, js)) + + entry, err := realKV.Get(ctx, testAccount) + require.NoError(t, err) + rec := decodeOverride(t, entry) + assert.Equal(t, OverrideStateSealed, rec.State) + assert.Equal(t, sealed.Revision, entry.Revision()) + }) +} + +// A seal against an absent or a tombstoned key creates the sealed record. +func TestSealOnMissingKeyCreatesSealedRecord(t *testing.T) { + for _, state := range []string{"absent", "tombstone"} { + t.Run(state, func(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + if state == "tombstone" { + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")) + require.NoError(t, svc.overrides.Delete(ctx, testAccount)) + } + _, err := svc.overrides.Get(ctx, testAccount) + require.ErrorIs(t, err, jetstream.ErrKeyNotFound) + + seal, err := svc.SealAccountOverrides(ctx, testAccount) + require.NoError(t, err) + + entry, err := svc.overrides.Get(ctx, testAccount) + require.NoError(t, err) + assert.Equal(t, seal.Revision, entry.Revision()) + assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State) + }) + } +} + +// alwaysConflictThenErrorKV answers Get as absent and every Create attempt +// with a conflict, then a store error on the final attempt: a recording +// wrapper proving the seal tried a real write on every attempt before giving +// up, rather than ever reporting success from an observed absence. +type alwaysConflictThenErrorKV struct { + jetstream.KeyValue + + calls int +} + +func (f *alwaysConflictThenErrorKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) { + return nil, jetstream.ErrKeyNotFound +} + +func (f *alwaysConflictThenErrorKV) Create(context.Context, string, []byte, ...jetstream.KVCreateOpt) (uint64, error) { + f.calls++ + if f.calls < sealMaxAttempts { + return 0, jetstream.ErrKeyExists + } + return 0, errors.New("store unavailable") +} + +func TestSealNeverSucceedsWithoutAWrite(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + fake := &alwaysConflictThenErrorKV{} + svc.overrides = fake + + seal, err := svc.SealAccountOverrides(ctx, testAccount) + assert.Error(t, err) + assert.Nil(t, seal) + assert.Equal(t, sealMaxAttempts, fake.calls, "every attempt must have tried a real write before giving up") +} + +// An account IAM does not hold as TERMINATING refuses the seal before any +// override-KV write, whether the key is present or absent. +func TestSealRefusedForActiveAccount(t *testing.T) { + for _, state := range []string{"present", "absent"} { + t.Run(state, func(t *testing.T) { + svc, js := newSealService(t) + ctx := context.Background() + if state == "present" { + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator")) + } + svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}}) + + before := lastOverrideSeq(t, js) + _, err := svc.SealAccountOverrides(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err) + assert.Equal(t, before, lastOverrideSeq(t, js)) + }) + } +} + +func TestSealRefusedForMissingIAMRecord(t *testing.T) { + svc, js := newSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}}) + + before := lastOverrideSeq(t, js) + _, err := svc.SealAccountOverrides(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err) + assert.Equal(t, before, lastOverrideSeq(t, js)) +} + +func TestSealRefusedWhenIAMReadFails(t *testing.T) { + svc, js := newSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(&stubTerminationChecker{err: errors.New("iam unavailable")}) + + before := lastOverrideSeq(t, js) + _, err := svc.SealAccountOverrides(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Equal(t, before, lastOverrideSeq(t, js)) +} + +func TestSealRefusedWithNoCheckerWired(t *testing.T) { + svc, js := newSealService(t) + ctx := context.Background() + + before := lastOverrideSeq(t, js) + _, err := svc.SealAccountOverrides(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Equal(t, before, lastOverrideSeq(t, js)) +} + +func TestSealAcceptedForTerminatingAccount(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + + seal, err := svc.SealAccountOverrides(ctx, testAccount) + require.NoError(t, err) + assert.NotZero(t, seal.Revision) + + entry, err := svc.overrides.Get(ctx, testAccount) + require.NoError(t, err) + assert.Equal(t, seal.Revision, entry.Revision()) +} + +// A refused seal, on a gateway that disabled quotas and so holds no override +// bucket, never creates one. +func TestSealRefusalDoesNotCreateBucket(t *testing.T) { + t.Run("not_terminating", func(t *testing.T) { + svc, js := newDisabledSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}}) + + _, err := svc.SealAccountOverrides(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err) + _, err = js.KeyValue(context.Background(), KVBucketAccountQuota) + assert.ErrorIs(t, err, jetstream.ErrBucketNotFound) + }) + + t.Run("iam_read_fails", func(t *testing.T) { + svc, js := newDisabledSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(&stubTerminationChecker{err: errors.New("iam unavailable")}) + + _, err := svc.SealAccountOverrides(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + _, err = js.KeyValue(context.Background(), KVBucketAccountQuota) + assert.ErrorIs(t, err, jetstream.ErrBucketNotFound) + }) +} + +// A seal accepted on a gateway whose quotas were disabled at startup creates +// the override bucket at seal time, never at startup. +func TestSealOnQuotaDisabledGatewayCreatesBucket(t *testing.T) { + svc, js := newDisabledSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + + seal, err := svc.SealAccountOverrides(ctx, testAccount) + require.NoError(t, err) + + kv, err := js.KeyValue(context.Background(), KVBucketAccountQuota) + require.NoError(t, err) + entry, err := kv.Get(context.Background(), testAccount) + require.NoError(t, err) + assert.Equal(t, seal.Revision, entry.Revision()) + assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State) +} + +// A sealed record stays sealed after the service is rebuilt over the same +// JetStream, every read and write refuses it, and a re-seal gets a higher +// revision. +func TestSealSurvivesServiceReconstruction(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + ctx := context.Background() + + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: KVBucketAccountQuota, History: 1}) + require.NoError(t, err) + + svc := New(baseLimits(), nil) + svc.SetOverrides(kv) + svc.SetJetStream(js) + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + first, err := svc.SealAccountOverrides(ctx, testAccount) + require.NoError(t, err) + + rebuiltKV, err := js.KeyValue(ctx, KVBucketAccountQuota) + require.NoError(t, err) + rebuilt := New(baseLimits(), nil) + rebuilt.SetOverrides(rebuiltKV) + rebuilt.SetJetStream(js) + + entry, err := rebuilt.overrides.Get(ctx, testAccount) + require.NoError(t, err) + assert.Equal(t, OverrideStateSealed, decodeOverride(t, entry).State) + + _, _, err = rebuilt.GetAccountQuota(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked)) + assert.True(t, awserrors.IsErrorCode(rebuilt.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"), awserrors.ErrorBlocked)) + assert.True(t, awserrors.IsErrorCode(rebuilt.PutAccountQuota(ctx, testAccount, Overrides{}, "operator"), awserrors.ErrorBlocked)) + + limits, err := rebuilt.limitsFor(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked)) + assert.Zero(t, limits) + + rebuilt.SetAccountTerminationChecker(terminatingChecker(testAccount)) + second, err := rebuilt.SealAccountOverrides(ctx, testAccount) + require.NoError(t, err) + assert.Greater(t, second.Revision, first.Revision) +} + +func TestSealRefusesEmptyInvalidAndGlobalAccount(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + + _, err := svc.SealAccountOverrides(ctx, "") + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorMissingParameter), "got %v", err) + + _, err = svc.SealAccountOverrides(ctx, "not-an-account") + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), "got %v", err) + + _, err = svc.SealAccountOverrides(ctx, awsidentifiers.GlobalAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), "got %v", err) +} + +func TestSealLeavesOtherAccountsUntouched(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + accountB := "222222222222" + require.NoError(t, svc.PutAccountQuota(ctx, accountB, Overrides{VCPUs: ptr(8)}, "operator")) + + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + _, err := svc.SealAccountOverrides(ctx, testAccount) + require.NoError(t, err) + + over, _, err := svc.GetAccountQuota(ctx, accountB) + require.NoError(t, err) + require.NotNil(t, over.VCPUs) + assert.Equal(t, 8, *over.VCPUs) +} + +// A record written before the state field existed has none and is treated as +// active by both reads and writes; a write on it preserves the old shape. +func TestRecordWithoutStateIsActive(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + + data, err := json.Marshal(Overrides{VCPUs: ptr(8)}) + require.NoError(t, err) + _, err = svc.overrides.Put(ctx, testAccount, data) + require.NoError(t, err) + + over, limits, err := svc.GetAccountQuota(ctx, testAccount) + require.NoError(t, err) + require.NotNil(t, over.VCPUs) + assert.Equal(t, 8, limits.VCPUs) + + require.NoError(t, svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(16)}, "operator")) + entry, err := svc.overrides.Get(ctx, testAccount) + require.NoError(t, err) + assert.NotContains(t, string(entry.Value()), `"state"`) +} + +// A record whose state is neither active nor sealed fails every read and +// write closed, rather than being treated as active. +func TestUnknownStateFailsClosed(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + + _, err := svc.overrides.Put(ctx, testAccount, []byte(`{"state":"frozen"}`)) + require.NoError(t, err) + + _, _, err = svc.GetAccountQuota(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + + limits, err := svc.limitsFor(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Zero(t, limits) + + assert.True(t, awserrors.IsErrorCode(svc.PutAccountQuota(ctx, testAccount, Overrides{VCPUs: ptr(8)}, "operator"), awserrors.ErrorInternalError)) + assert.True(t, awserrors.IsErrorCode(svc.PutAccountQuota(ctx, testAccount, Overrides{}, "operator"), awserrors.ErrorInternalError)) +} + +// A sealed record is Blocked for admission's own resolver, which every +// caller (admin read, ELBv2 DescribeAccountLimits, every Enforce* method) +// shares. +func TestAdmissionOnSealedRecordIsBlocked(t *testing.T) { + svc, _ := newSealService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccount)) + _, err := svc.SealAccountOverrides(ctx, testAccount) + require.NoError(t, err) + + limits, err := svc.limitsFor(ctx, testAccount) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "got %v", err) + assert.Zero(t, limits) + + assert.Error(t, svc.CheckVCPU(ctx, testAccount, 1)) + assert.True(t, awserrors.IsErrorCode(svc.CheckVCPU(ctx, testAccount, 1), awserrors.ErrorBlocked)) +} + +// The existing read-failure fallback in limitsFor (a decode or store error +// falls back to the configured limits) is unchanged by the seal work: it is +// distinct from a successfully read sealed record, which refuses instead. +func TestAdmissionReadFailureFallbackUnchanged(t *testing.T) { + svc, kv := newOverrideService(t) + + _, err := kv.Put(t.Context(), testAccount, []byte("{not json")) + require.NoError(t, err) + + limits, err := svc.limitsFor(t.Context(), testAccount) + require.NoError(t, err, "a decode failure must still fall back, not error, from limitsFor") + require.Equal(t, baseLimits(), limits) +} diff --git a/spinifex/domains/admission/quota/storage.go b/spinifex/domains/admission/quota/storage.go index 3b79bc979..ac990b507 100644 --- a/spinifex/domains/admission/quota/storage.go +++ b/spinifex/domains/admission/quota/storage.go @@ -23,7 +23,10 @@ func (s *Service) EnforceVolumeCreate(ctx context.Context, natsConn *nats.Conn, if err != nil { return err } - limits := s.limitsFor(ctx, accountID) + limits, err := s.limitsFor(ctx, accountID) + if err != nil { + return err + } if err := exceeds(len(out.Volumes), 1, limits.Volumes); err != nil { return err } @@ -43,7 +46,11 @@ func (s *Service) EnforceVolumeModify(ctx context.Context, natsConn *nats.Conn, if err != nil { return err } - return exceeds(total-oldGiB, newGiB, s.limitsFor(ctx, accountID).VolumesGiB) + limits, err := s.limitsFor(ctx, accountID) + if err != nil { + return err + } + return exceeds(total-oldGiB, newGiB, limits.VolumesGiB) } // volumeUsage sums the account's volume sizes in GiB via a single DescribeVolumes diff --git a/spinifex/domains/admission/quota/vcpu.go b/spinifex/domains/admission/quota/vcpu.go index 502c5b1e7..aec0d1ec9 100644 --- a/spinifex/domains/admission/quota/vcpu.go +++ b/spinifex/domains/admission/quota/vcpu.go @@ -37,7 +37,11 @@ func (s *Service) CheckVCPU(ctx context.Context, accountID string, want int) err if err != nil { return err } - return exceeds(current, want, s.limitsFor(ctx, accountID).VCPUs) + limits, err := s.limitsFor(ctx, accountID) + if err != nil { + return err + } + return exceeds(current, want, limits.VCPUs) } // casVCPU applies next to accountID's counter under bounded JetStream CAS so diff --git a/spinifex/gateway/admin_accountquota.go b/spinifex/gateway/admin_accountquota.go index 5a2b3c8e0..e9e3c29e6 100644 --- a/spinifex/gateway/admin_accountquota.go +++ b/spinifex/gateway/admin_accountquota.go @@ -43,6 +43,9 @@ func (gw *GatewayConfig) adminGetAccountQuota(ctx context.Context, body []byte) over, limits, err := gw.Quota.GetAccountQuota(ctx, accountID) if err != nil { + if awserrors.IsErrorCode(err, awserrors.ErrorBlocked) { + return nil, err + } slog.ErrorContext(ctx, "GetAccountQuota: failed to read override", "accountID", accountID, "err", err) return nil, errors.New(awserrors.ErrorInternalError) } @@ -61,6 +64,9 @@ func (gw *GatewayConfig) adminPutAccountQuota(ctx context.Context, body []byte) // The caller's own identity, so a quota change names who made it. updatedBy, _ := ctx.Value(ctxIdentity).(string) if err := gw.Quota.PutAccountQuota(ctx, accountID, over, updatedBy); err != nil { + if awserrors.IsErrorCode(err, awserrors.ErrorBlocked) { + return nil, err + } slog.ErrorContext(ctx, "PutAccountQuota: failed to store override", "accountID", accountID, "err", err) return nil, errors.New(awserrors.ErrorInternalError) } @@ -69,6 +75,9 @@ func (gw *GatewayConfig) adminPutAccountQuota(ctx context.Context, body []byte) stored, limits, err := gw.Quota.GetAccountQuota(ctx, accountID) if err != nil { + if awserrors.IsErrorCode(err, awserrors.ErrorBlocked) { + return nil, err + } slog.ErrorContext(ctx, "PutAccountQuota: failed to read back override", "accountID", accountID, "err", err) return nil, errors.New(awserrors.ErrorInternalError) } diff --git a/spinifex/gateway/admin_accountquota_test.go b/spinifex/gateway/admin_accountquota_test.go index 79b355b7b..e7edeee6b 100644 --- a/spinifex/gateway/admin_accountquota_test.go +++ b/spinifex/gateway/admin_accountquota_test.go @@ -205,3 +205,56 @@ func TestAccountQuotaMethodsAreRegistered(t *testing.T) { require.Contains(t, AdminMethodNames(), "GetAccountQuota") require.Contains(t, AdminMethodNames(), "PutAccountQuota") } + +// quotaStubChecker answers AccountTerminating from a fixed set, so these +// tests can seal a record without a real IAM bucket. +type quotaStubChecker struct { + terminating map[string]bool +} + +func (c *quotaStubChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) { + return c.terminating[accountID], nil +} + +// A request body cannot carry or alter the stored state: an attempted +// "state":"sealed" in the overrides object is simply unknown to the wire +// type, the stored record stays active, and the response never shows state. +func TestAdminRequestCannotCarryState(t *testing.T) { + gw, account := quotaTestGateway(t) + ctx := context.WithValue(t.Context(), ctxIdentity, "operator") + + body := []byte(`{"accountId":"` + account + `","overrides":{"vcpus":32,"state":"sealed"}}`) + output, err := gw.adminPutAccountQuota(ctx, body) + require.NoError(t, err) + + resp, ok := output.(*AccountQuotaResponse) + require.True(t, ok) + require.Equal(t, 32, resp.Limits["vcpus"]) + + data, err := json.Marshal(resp.Overrides) + require.NoError(t, err) + require.NotContains(t, string(data), "state") + + _, _, err = gw.Quota.GetAccountQuota(ctx, account) + require.NoError(t, err, "the record must still be active, not sealed") +} + +// Admin get and the put read-back on a sealed record must return Blocked, +// not the generic InternalError every other read failure maps to. +func TestAdminQuotaReadsMapBlocked(t *testing.T) { + gw, account := quotaTestGateway(t) + gw.Quota.SetAccountTerminationChecker("aStubChecker{terminating: map[string]bool{account: true}}) + _, err := gw.Quota.SealAccountOverrides(t.Context(), account) + require.NoError(t, err) + + _, err = gw.adminGetAccountQuota(t.Context(), quotaRequestBody(t, AccountQuotaRequest{AccountID: account})) + require.Error(t, err) + require.Equal(t, awserrors.ErrorBlocked, err.Error()) + + ctx := context.WithValue(t.Context(), ctxIdentity, "operator") + _, err = gw.adminPutAccountQuota(ctx, quotaRequestBody(t, AccountQuotaRequest{ + AccountID: account, Overrides: admissionquota.Overrides{VCPUs: quotaPtr(8)}, + })) + require.Error(t, err) + require.Equal(t, awserrors.ErrorBlocked, err.Error()) +} diff --git a/spinifex/gateway/elbv2.go b/spinifex/gateway/elbv2.go index d17782804..cbc4cb907 100644 --- a/spinifex/gateway/elbv2.go +++ b/spinifex/gateway/elbv2.go @@ -187,6 +187,9 @@ func accountLoadBalancerLimit(ctx context.Context, gw *GatewayConfig, accountID } _, limits, err := gw.Quota.GetAccountQuota(ctx, accountID) if err != nil { + if awserrors.IsErrorCode(err, awserrors.ErrorBlocked) { + return 0, err + } slog.ErrorContext(ctx, "DescribeAccountLimits: failed to read account quota", "accountID", accountID, "err", err) return 0, errors.New(awserrors.ErrorInternalError) } diff --git a/spinifex/gateway/elbv2_accountlimits_test.go b/spinifex/gateway/elbv2_accountlimits_test.go index 48929d840..102994e11 100644 --- a/spinifex/gateway/elbv2_accountlimits_test.go +++ b/spinifex/gateway/elbv2_accountlimits_test.go @@ -9,6 +9,7 @@ import ( "github.com/aws/aws-sdk-go/service/elbv2" "github.com/mulgadc/spinifex/spinifex/domains/admission/quota" + "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" gateway_elbv2 "github.com/mulgadc/spinifex/spinifex/gateway/elbv2" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -121,6 +122,20 @@ func TestELBv2AccountLoadBalancerLimitDefaultsForExemptAccount(t *testing.T) { assert.Equal(t, gateway_elbv2.DefaultLoadBalancerLimit, limit) } +// A sealed override must refuse the load balancer cap with Blocked rather +// than the generic InternalError every other read failure maps to. +func TestELBv2AccountLoadBalancerLimitBlockedOnSealedRecord(t *testing.T) { + gw, account := quotaTestGateway(t) + gw.Quota.SetAccountTerminationChecker("aStubChecker{terminating: map[string]bool{account: true}}) + _, err := gw.Quota.SealAccountOverrides(t.Context(), account) + require.NoError(t, err) + + limit, err := accountLoadBalancerLimit(t.Context(), gw, account) + require.Error(t, err) + require.Equal(t, awserrors.ErrorBlocked, err.Error()) + assert.Zero(t, limit) +} + // The dispatch entry must accept the paging parameters the model defines and // still answer with the whole list. func TestELBv2DescribeAccountLimitsAcceptsPaging(t *testing.T) { diff --git a/spinifex/handlers/iam/account_status_reader.go b/spinifex/handlers/iam/account_status_reader.go index 96e5d25b0..fcd1e0443 100644 --- a/spinifex/handlers/iam/account_status_reader.go +++ b/spinifex/handlers/iam/account_status_reader.go @@ -10,8 +10,8 @@ import ( ) // AccountStatusReader reads account status from spinifex-accounts and never -// writes. It is the read-only capability EC2's settings seal uses to check -// its TERMINATING precondition. +// writes. It is the read-only capability EC2's settings seal and the quota +// override seal both use to check their TERMINATING precondition. type AccountStatusReader struct { accounts *kvstore.Bucket // AttachOnly: never creates the bucket } diff --git a/spinifex/runtime/roles/awsgw/awsgw.go b/spinifex/runtime/roles/awsgw/awsgw.go index af18c4733..c92ecaa6e 100644 --- a/spinifex/runtime/roles/awsgw/awsgw.go +++ b/spinifex/runtime/roles/awsgw/awsgw.go @@ -60,6 +60,11 @@ var serviceName = "awsgw" // handlers/eks, because handlers/eks must not import handlers/sts. var _ handlers_sts.ClusterJWKSResolver = (*handlers_eks.OIDCJWKSReader)(nil) +// AccountStatusReader also serves as the quota override seal's TERMINATING +// precondition, wired here because handlers/iam must not import +// domains/admission/quota. +var _ admissionquota.AccountTerminationChecker = (*handlers_iam.AccountStatusReader)(nil) + // Version and Commit are set by the cmd package before Start() to pass // build-time ldflags to the gateway without creating an import cycle. var ( @@ -561,6 +566,25 @@ func launchService(config *config.ClusterConfig) error { gw.Quota = admissionquota.New(quotaCfg, usageBucket) gw.Quota.SetOverrides(quotaOverrides) + // Seal precondition: the quota owner's only fact from IAM, read directly + // rather than through the gateway's cached IAMService, so a seal sees a + // TERMINATING flip the moment IAM commits it. + gw.Quota.SetAccountTerminationChecker(handlers_iam.NewAccountStatusReader(js)) + // SealAccountOverrides opens the override bucket itself when this + // gateway's quotas are disabled, so the seal subject works regardless of + // quotaCfg.Enabled and never touches the bucket at startup. + gw.Quota.SetJetStream(js) + + quotaSealSub, err := registerQuotaOverrideSeal(natsConn, gw.Quota) + if err != nil { + return fmt.Errorf("subscribe quota override seal: %w", err) + } + defer func() { + if err := quotaSealSub.Unsubscribe(); err != nil { + slog.Warn("Quota override seal: unsubscribe failed", "err", err) + } + }() + // Bedrock token quota reads the stream-fed usage counters built above, // independent of whether the standing-infra dimensions are enabled. gw.Quota.SetBedrockUsage(bedrockUsage) diff --git a/spinifex/runtime/roles/awsgw/quota_override_seal.go b/spinifex/runtime/roles/awsgw/quota_override_seal.go new file mode 100644 index 000000000..1db7473ab --- /dev/null +++ b/spinifex/runtime/roles/awsgw/quota_override_seal.go @@ -0,0 +1,39 @@ +package awsgw + +import ( + "context" + + admissionv1 "github.com/mulgadc/spinifex/contracts/admission/v1" + admissionquota "github.com/mulgadc/spinifex/spinifex/domains/admission/quota" + natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats" + "github.com/nats-io/nats.go" +) + +// quotaOverrideSealQueue is the NATS queue group for the quota override seal +// responders so the request is load-balanced across awsgw nodes. +const quotaOverrideSealQueue = "awsgw-quota-seal" + +// registerQuotaOverrideSeal serves SealAccountQuotaOverrideSubject on this +// gateway. Any current gateway may answer; the leader-accepted revision is +// the evidence. +func registerQuotaOverrideSeal(nc *nats.Conn, quota *admissionquota.Service) (*nats.Subscription, error) { + return nc.QueueSubscribe(admissionv1.SealAccountQuotaOverrideSubject, quotaOverrideSealQueue, + func(msg *nats.Msg) { + natsmsg.ServeNATSRequestCtx(msg, handleSealAccountQuotaOverride(quota)) + }) +} + +func handleSealAccountQuotaOverride(quota *admissionquota.Service) func(context.Context, *admissionv1.SealAccountQuotaOverrideRequest) (*admissionv1.SealAccountQuotaOverrideReply, error) { + return func(ctx context.Context, req *admissionv1.SealAccountQuotaOverrideRequest) (*admissionv1.SealAccountQuotaOverrideReply, error) { + seal, err := quota.SealAccountOverrides(ctx, req.AccountID) + if err != nil { + return nil, err + } + return &admissionv1.SealAccountQuotaOverrideReply{ + AccountID: req.AccountID, + CorrelationRef: req.CorrelationRef, + Disposition: admissionv1.AccountQuotaOverrideDispositionSealed, + Revision: seal.Revision, + }, nil + } +} diff --git a/spinifex/runtime/roles/awsgw/quota_override_seal_test.go b/spinifex/runtime/roles/awsgw/quota_override_seal_test.go new file mode 100644 index 000000000..2aa41236b --- /dev/null +++ b/spinifex/runtime/roles/awsgw/quota_override_seal_test.go @@ -0,0 +1,84 @@ +package awsgw + +import ( + "context" + "testing" + "time" + + admissionv1 "github.com/mulgadc/spinifex/contracts/admission/v1" + "github.com/mulgadc/spinifex/internal/testkit" + admissionquota "github.com/mulgadc/spinifex/spinifex/domains/admission/quota" + natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats" + "github.com/nats-io/nats.go/jetstream" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// stubChecker answers AccountTerminating from a fixed set, standing in for +// handlers_iam.AccountStatusReader so this test does not need a real IAM +// bucket. +type stubChecker struct { + terminating map[string]bool +} + +func (c *stubChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) { + return c.terminating[accountID], nil +} + +func newQuotaSealService(t *testing.T) (*admissionquota.Service, jetstream.JetStream) { + t.Helper() + _, _, js := testutil.StartTestJetStream(t) + quota := admissionquota.New(admissionquota.Limits{}, nil) + quota.SetJetStream(js) + return quota, js +} + +func TestQuotaOverrideSeal_SealsTerminatingAccount(t *testing.T) { + quota, js := newQuotaSealService(t) + _, nc := testutil.StartTestNATS(t) + quota.SetAccountTerminationChecker(&stubChecker{terminating: map[string]bool{"111122223333": true}}) + + sub, err := registerQuotaOverrideSeal(nc, quota) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + + req := admissionv1.SealAccountQuotaOverrideRequest{AccountID: "111122223333", CorrelationRef: "req-1"} + resp, err := natsmsg.NATSRequest[admissionv1.SealAccountQuotaOverrideReply](context.Background(), + nc, admissionv1.SealAccountQuotaOverrideSubject, req, 2*time.Second, "") + require.NoError(t, err) + + assert.Equal(t, "111122223333", resp.AccountID) + assert.Equal(t, "req-1", resp.CorrelationRef) + assert.Equal(t, admissionv1.AccountQuotaOverrideDispositionSealed, resp.Disposition) + assert.NotZero(t, resp.Revision) + + kv, err := js.KeyValue(context.Background(), admissionquota.KVBucketAccountQuota) + require.NoError(t, err) + entry, err := kv.Get(context.Background(), "111122223333") + require.NoError(t, err) + assert.Equal(t, resp.Revision, entry.Revision()) +} + +func TestQuotaOverrideSeal_RefusesActiveAccount(t *testing.T) { + quota, _ := newQuotaSealService(t) + _, nc := testutil.StartTestNATS(t) + quota.SetAccountTerminationChecker(&stubChecker{terminating: map[string]bool{}}) + + sub, err := registerQuotaOverrideSeal(nc, quota) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + + req := admissionv1.SealAccountQuotaOverrideRequest{AccountID: "111122223333"} + _, err = natsmsg.NATSRequest[admissionv1.SealAccountQuotaOverrideReply](context.Background(), + nc, admissionv1.SealAccountQuotaOverrideSubject, req, 2*time.Second, "") + require.Error(t, err, "a non-TERMINATING account must be refused") +} + +func TestQuotaOverrideSeal_NoResponderWithoutGateway(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + + req := admissionv1.SealAccountQuotaOverrideRequest{AccountID: "111122223333"} + _, err := natsmsg.NATSRequest[admissionv1.SealAccountQuotaOverrideReply](context.Background(), + nc, admissionv1.SealAccountQuotaOverrideSubject, req, 500*time.Millisecond, "") + require.Error(t, err, "no subscriber must surface as a no-responders error") +}