diff --git a/spinifex/handlers/ec2/routetable/service_impl.go b/spinifex/handlers/ec2/routetable/service_impl.go index e3ffa81e1..8a9a4590b 100644 --- a/spinifex/handlers/ec2/routetable/service_impl.go +++ b/spinifex/handlers/ec2/routetable/service_impl.go @@ -123,56 +123,44 @@ func (s *RouteTableServiceImpl) putRouteTable(ctx context.Context, accountID str // parallel AssociateRouteTable calls Terraform fans out per route table. const rtbCASMaxRetries = 16 +// errRTBAbsent and errRTBContended mark the two kvutil outcomes that map to +// specific AWS errors rather than a generic internal failure. Neither +// escapes this file. +var ( + errRTBAbsent = errors.New("routetable: record absent") + errRTBContended = errors.New("routetable: CAS retries exhausted") +) + // mutateRouteTableCAS applies mutate to a route table under optimistic -// concurrency: read with revision, mutate, then Update guarded by that -// revision, retrying when a concurrent writer wins. A blind read-modify-Put -// loses updates when callers associate several subnets with one route table at -// once. mutate reports whether it changed the record; a false return commits -// nothing. +// concurrency. A blind read-modify-Put loses updates when callers associate +// several subnets with one route table at once. mutate reports whether +// it changed the record; a false return commits nothing. func (s *RouteTableServiceImpl) mutateRouteTableCAS(ctx context.Context, accountID, rtbID string, mutate func(*RouteTableRecord) (bool, error)) error { - key := utils.AccountKey(accountID, rtbID) - for range rtbCASMaxRetries { - entry, err := s.rtbKV.Get(ctx, key) - if err != nil { - if errors.Is(err, jetstream.ErrKeyNotFound) { - return errors.New(awserrors.ErrorInvalidRouteTableIDNotFound) - } - slog.ErrorContext(ctx, "Failed to read route table from KV", "routeTableId", rtbID, "err", err) - return errors.New(awserrors.ErrorServerInternal) - } - - var record RouteTableRecord - if err := json.Unmarshal(entry.Value(), &record); err != nil { - slog.ErrorContext(ctx, "Corrupt route table record in KV", "routeTableId", rtbID, "err", err) - return errors.New(awserrors.ErrorServerInternal) - } - - changed, err := mutate(&record) - if err != nil { - return err - } - if !changed { - return nil - } - - data, err := json.Marshal(&record) - if err != nil { - slog.ErrorContext(ctx, "Failed to marshal route table record", "routeTableId", rtbID, "err", err) - return errors.New(awserrors.ErrorServerInternal) - } - if _, err := s.rtbKV.Update(ctx, key, data, entry.Revision()); err != nil { - if errors.Is(err, jetstream.ErrKeyRevisionMismatch) { - continue // CAS conflict — another writer won, re-read and retry. - } - slog.ErrorContext(ctx, "Failed to write route table to KV", "routeTableId", rtbID, "err", err) - return errors.New(awserrors.ErrorServerInternal) - } + _, err := kvutil.Update(ctx, s.rtbKV, utils.AccountKey(accountID, rtbID), kvutil.CASConfig{ + Attempts: rtbCASMaxRetries, + NotFound: errRTBAbsent, + Exhausted: func(string, int) error { return errRTBContended }, + }, mutate) + switch { + case err == nil: return nil + case errors.Is(err, errRTBAbsent): + return errors.New(awserrors.ErrorInvalidRouteTableIDNotFound) + case errors.Is(err, errRTBContended): + // Contended rather than broken. + slog.ErrorContext(ctx, "Route table CAS retries exhausted under contention", + "routeTableId", rtbID, "attempts", rtbCASMaxRetries) + case errors.Is(err, kvutil.ErrRead): + slog.ErrorContext(ctx, "Failed to read route table from KV", "routeTableId", rtbID, "err", err) + case errors.Is(err, kvutil.ErrDecode): + slog.ErrorContext(ctx, "Corrupt route table record in KV", "routeTableId", rtbID, "err", err) + case errors.Is(err, kvutil.ErrEncode): + slog.ErrorContext(ctx, "Failed to marshal route table record", "routeTableId", rtbID, "err", err) + case errors.Is(err, kvutil.ErrWrite): + slog.ErrorContext(ctx, "Failed to write route table to KV", "routeTableId", rtbID, "err", err) + default: + return err } - // Only a CAS conflict reaches here, so the record is contended rather than - // broken; say so, or the caller's InternalError has no cause anywhere. - slog.ErrorContext(ctx, "Route table CAS retries exhausted under contention", - "routeTableId", rtbID, "attempts", rtbCASMaxRetries) return errors.New(awserrors.ErrorServerInternal) } diff --git a/spinifex/kvutil/cas.go b/spinifex/kvutil/cas.go index 3b94c0371..a2cd7ac92 100644 --- a/spinifex/kvutil/cas.go +++ b/spinifex/kvutil/cas.go @@ -17,12 +17,20 @@ const casAttempts = 5 // casBackoffBase scales the pause between attempts. const casBackoffBase = 2 * time.Millisecond -// CASConfig tunes an optimistic-concurrency loop. The zero value retreis casAttempts times and fails when the key is absent. +// Stage labels for a failed CAS cycle, so a caller can log which step broke. +var ( + ErrRead = errors.New("kvutil: read") + ErrDecode = errors.New("kvutil: decode") + ErrEncode = errors.New("kvutil: encode") + ErrWrite = errors.New("kvutil: write") +) + +// CASConfig tunes an optimistic-concurrency loop. The zero value retries casAttempts times and fails when the key is absent. type CASConfig struct { Attempts int // 0 selects casAttempts CreateIfAbsent bool // start from a zero value when the key is missing - // NotFounc replaces the raw jetstream error when the key is absent and + // NotFound replaces the raw jetstream error when the key is absent and // CreateIfAbsent is unset, so a caller can map it to its own API error. NotFound error @@ -37,7 +45,7 @@ func isConflict(err error) bool { return errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) } -// retryCAS runs attempt until is succeeds, returns a non-conflict error, or spends its buudget. It pauses between attempts, +// retryCAS runs attempt until it succeeds, returns a non-conflict error, or spends its budget. It pauses between attempts, // jittered so contending writers do not re-collide on the same revision. func retryCAS(ctx context.Context, cfg CASConfig, op, key string, attempt func() error) error { attempts := cfg.Attempts @@ -80,7 +88,7 @@ func Update[T any](ctx context.Context, kv jetstream.KeyValue, key string, cfg C switch { case err == nil: if uerr := json.Unmarshal(entry.Value(), &value); uerr != nil { - return uerr + return fmt.Errorf("%w: %w", ErrDecode, uerr) } revision = entry.Revision() case errors.Is(err, jetstream.ErrKeyNotFound) && cfg.CreateIfAbsent: @@ -88,7 +96,7 @@ func Update[T any](ctx context.Context, kv jetstream.KeyValue, key string, cfg C case errors.Is(err, jetstream.ErrKeyNotFound) && cfg.NotFound != nil: return cfg.NotFound default: - return err + return fmt.Errorf("%w: %w", ErrRead, err) } changed, merr := mutate(&value) @@ -102,7 +110,7 @@ func Update[T any](ctx context.Context, kv jetstream.KeyValue, key string, cfg C data, merr := json.Marshal(&value) if merr != nil { - return merr + return fmt.Errorf("%w: %w", ErrEncode, merr) } if revision == 0 { _, err = kv.Create(ctx, key, data) @@ -110,7 +118,7 @@ func Update[T any](ctx context.Context, kv jetstream.KeyValue, key string, cfg C _, err = kv.Update(ctx, key, data, revision) } if err != nil { - return err + return fmt.Errorf("%w: %w", ErrWrite, err) } result = &value return nil @@ -134,14 +142,17 @@ func Put(ctx context.Context, kv jetstream.KeyValue, key string, cfg CASConfig, case errors.Is(err, jetstream.ErrKeyNotFound) && cfg.NotFound != nil: return cfg.NotFound default: - return err + return fmt.Errorf("%w: %w", ErrRead, err) } if revision == 0 { _, err = kv.Create(ctx, key, data) } else { _, err = kv.Update(ctx, key, data, revision) } - return err + if err != nil { + return fmt.Errorf("%w: %w", ErrWrite, err) + } + return nil }) } @@ -157,14 +168,14 @@ func Claim[T any](ctx context.Context, kv jetstream.KeyValue, key string, cfg CA absent = true return nil } - return gerr + return fmt.Errorf("%w: %w", ErrRead, gerr) } var v T if uerr := json.Unmarshal(entry.Value(), &v); uerr != nil { - return uerr + return fmt.Errorf("%w: %w", ErrDecode, uerr) } if derr := kv.Delete(ctx, key, jetstream.LastRevision(entry.Revision())); derr != nil { - return derr + return fmt.Errorf("%w: %w", ErrWrite, derr) } result = &v return nil