refactor: label kvutil CAS stages and convert routetable (#883)

2026-10-15 17:11:30 +00:00 · 2026-08-26 10:26:31 +10:00
parent ef30b28b49
commit 474dff05d8
2 changed files with 57 additions and 58 deletions
@@ -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)
 }

@@ -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