Merge pull request #80 from authzed/feat/enhanced-fgam-retry-logic

feat: enhance FGAM retry logic with exponential backoff for concurrent operations

Author: Verolop

docs/index.md

@@ -27,6 +27,9 @@ provider "authzed" {
   token       = var.authzed_api_token
   # Uncomment to specify a different API version
   # api_version = "25r1"
+  
+  # Uncomment to enable serialization for high-concurrency scenarios
+  # fgam_serialization = true
 }
 ```
 
@@ -46,6 +49,7 @@ To obtain a token, contact your AuthZed account team. They will provide you with
 * `endpoint` - (Required) The host address of the AuthZed Cloud API. Default is `https://api.admin.stage.aws.authzed.net`.
 * `token` - (Required) The bearer token for authentication with AuthZed.
 * `api_version` - (Optional) The version of the API to use. Default is "25r1".
+* `fgam_serialization` - (Optional) Enable serialization of operations to prevent conflicts. When enabled, resources within the same permission system will be created/updated sequentially instead of in parallel. Default is `false`.
 
 ## Important Notes
 

internal/client/cloud_client.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -109,87 +110,82 @@ func (c *CloudClient) Do(req *http.Request) (*ResponseWithETag, error) {
 }
 
 // UpdateResource updates any resource that implements the Resource interface
-func (c *CloudClient) UpdateResource(resource Resource, endpoint string, body any) (Resource, error) {
+func (c *CloudClient) UpdateResource(ctx context.Context, resource Resource, endpoint string, body any) (Resource, error) {
 	// Define a function to get the latest ETag
 	getLatestETag := func() (string, error) {
-		getReq, err := c.NewRequest(http.MethodGet, endpoint, nil)
+		req, err := c.NewRequest(http.MethodGet, endpoint, nil)
 		if err != nil {
 			return "", fmt.Errorf("failed to create GET request: %w", err)
 		}
 
-		getResp, err := c.Do(getReq)
+		resp, err := c.Do(req)
 		if err != nil {
 			return "", fmt.Errorf("failed to send GET request: %w", err)
 		}
 		defer func() {
-			if getResp.Response.Body != nil {
-				_ = getResp.Response.Body.Close()
+			if resp.Response.Body != nil {
+				_ = resp.Response.Body.Close()
 			}
 		}()
 
-		if getResp.Response.StatusCode != http.StatusOK {
-			return "", fmt.Errorf("failed to get latest ETag, status: %d", getResp.Response.StatusCode)
+		if resp.Response.StatusCode != http.StatusOK {
+			return "", NewAPIError(resp)
 		}
 
-		// Extract ETag from response header
-		return getResp.ETag, nil
+		return resp.ETag, nil
 	}
 
-	// Try update with current ETag
+	// Define a function to update with a specific ETag
 	updateWithETag := func(currentETag string) (*ResponseWithETag, error) {
-		req, err := c.NewRequest(http.MethodPut, endpoint, body, WithETag(currentETag))
+		req, err := c.NewRequest(http.MethodPut, endpoint, body)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to create request: %w", err)
 		}
 
-		return c.Do(req)
-	}
-
-	respWithETag, err := updateWithETag(resource.GetETag())
-	if err != nil {
-		return nil, err
-	}
+		// Set the If-Match header for optimistic concurrency control
+		req.Header.Set("If-Match", currentETag)
 
-	// Handle retryable errors (412 Precondition Failed, 409 Conflict) by refreshing ETag and retrying
-	retryWithFreshETag := func(errorContext string) error {
-		// Close the body of the first response
-		if respWithETag.Response.Body != nil {
-			_ = respWithETag.Response.Body.Close()
+		resp, err := c.Do(req)
+		if err != nil {
+			return nil, fmt.Errorf("failed to send request: %w", err)
 		}
 
-		// Get the latest ETag
-		latestETag, err := getLatestETag()
-		if err != nil {
-			return fmt.Errorf("failed to get latest ETag for retry (%s): %w", errorContext, err)
+		if resp.Response.StatusCode != http.StatusOK {
+			defer func() {
+				if resp.Response.Body != nil {
+					_ = resp.Response.Body.Close()
+				}
+			}()
+			return nil, NewAPIError(resp)
 		}
 
-		// Retry the update with the latest ETag
-		respWithETag, err = updateWithETag(latestETag)
-		return err
+		return resp, nil
 	}
 
-	switch respWithETag.Response.StatusCode {
-	case http.StatusPreconditionFailed:
-		if err := retryWithFreshETag("precondition failed"); err != nil {
-			return nil, err
-		}
-	case http.StatusConflict:
-		if err := retryWithFreshETag("FGAM configuration change"); err != nil {
-			return nil, err
-		}
+	// Use retry logic with exponential backoff
+	retryConfig := DefaultRetryConfig()
+	respWithETag, err := retryConfig.RetryWithExponentialBackoffLegacy(
+		ctx,
+		func() (*ResponseWithETag, error) {
+			return updateWithETag(resource.GetETag())
+		},
+		getLatestETag,
+		updateWithETag,
+		"resource update",
+	)
+	if err != nil {
+		return nil, err
 	}
 
 	defer func() {
-		// ignore the error
-		_ = respWithETag.Response.Body.Close()
+		if respWithETag.Response.Body != nil {
+			_ = respWithETag.Response.Body.Close()
+		}
 	}()
 
-	if respWithETag.Response.StatusCode != http.StatusOK {
-		return nil, NewAPIError(respWithETag)
-	}
-
 	// Update the resource's ETag
 	resource.SetETag(respWithETag.ETag)
+
 	return resource, nil
 }
 
@@ -354,16 +350,47 @@ func (c *CloudClient) GetResourceWithFactory(endpoint string, dest any, factory
 }
 
 // CreateResourceWithFactory combines CreateResource with a factory to create a proper Resource
-func (c *CloudClient) CreateResourceWithFactory(endpoint string, body any, dest any, factory ResourceFactory) (Resource, error) {
-	req, err := c.NewRequest(http.MethodPost, endpoint, body)
-	if err != nil {
-		return nil, err
+func (c *CloudClient) CreateResourceWithFactory(ctx context.Context, endpoint string, body any, dest any, factory ResourceFactory) (Resource, error) {
+	// Use retry logic with exponential backoff
+	retryConfig := DefaultRetryConfig()
+
+	// Define the create operation
+	createOperation := func() (*ResponseWithETag, error) {
+		req, err := c.NewRequest(http.MethodPost, endpoint, body)
+		if err != nil {
+			return nil, err
+		}
+
+		respWithETag, err := c.Do(req)
+		if err != nil {
+			return nil, err
+		}
+
+		return respWithETag, nil
 	}
 
-	respWithETag, err := c.Do(req)
+	// For CREATE operations, we don't need to get fresh ETags since there's no existing resource
+	// We just retry the same operation
+	getLatestETag := func() (string, error) {
+		return "", nil // Not used for CREATE operations
+	}
+
+	createWithETag := func(etag string) (*ResponseWithETag, error) {
+		return createOperation() // Retry the same CREATE operation
+	}
+
+	// Execute with retry logic
+	respWithETag, err := retryConfig.RetryWithExponentialBackoffLegacy(
+		ctx,
+		createOperation,
+		getLatestETag,
+		createWithETag,
+		"resource create",
+	)
 	if err != nil {
 		return nil, err
 	}
+
 	defer func() {
 		// ignore the error
 		_ = respWithETag.Response.Body.Close()

internal/client/constants.go

@@ -11,4 +11,11 @@ const (
 	DefaultDeleteTimeout = 5 * time.Minute
 	// DefaultDeletePollInterval is the default interval between polling attempts during delete operations
 	DefaultDeletePollInterval = 2 * time.Second
+
+	// Retry configuration for handling concurrent operations
+	// DefaultMaxRetries is the default number of retry attempts
+	DefaultMaxRetries     = 8
+	DefaultBaseRetryDelay = 100 * time.Millisecond
+	DefaultMaxRetryDelay  = 5 * time.Second
+	DefaultMaxJitter      = 500 * time.Millisecond
 )

internal/client/errors.go

@@ -37,7 +37,7 @@ type APIError struct {
 
 func (e *APIError) Error() string {
 	if e.Message != "" {
-		// Check if this is a FGAM configuration conflict and provide helpful context
+		// Check if this is a configuration conflict and provide helpful context
 		if e.StatusCode == 409 && containsFGAMConfigConflict(e.Message) {
 			return fmt.Sprintf("API error (status %d): %s\n\nThis error occurs when the Fine-Grained Access Management (FGAM) configuration for the permission system has been modified by another process. The Terraform provider will automatically retry this operation.", e.StatusCode, e.Message)
 		}
@@ -46,7 +46,7 @@ func (e *APIError) Error() string {
 	return fmt.Sprintf("API error (status %d)", e.StatusCode)
 }
 
-// containsFGAMConfigConflict checks if the error message indicates a FGAM configuration conflict
+// containsFGAMConfigConflict checks if the error message indicates a configuration conflict
 func containsFGAMConfigConflict(message string) bool {
 	lowerMessage := strings.ToLower(message)
 

internal/client/policy.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -81,11 +82,11 @@ func (c *CloudClient) GetPolicy(permissionsSystemID, policyID string) (*PolicyWi
 }
 
 // CreatePolicy creates a new policy
-func (c *CloudClient) CreatePolicy(policy *models.Policy) (*PolicyWithETag, error) {
+func (c *CloudClient) CreatePolicy(ctx context.Context, policy *models.Policy) (*PolicyWithETag, error) {
 	path := fmt.Sprintf("/ps/%s/access/policies", policy.PermissionsSystemID)
 
 	var createdPolicy models.Policy
-	resource, err := c.CreateResourceWithFactory(path, policy, &createdPolicy, NewPolicyResource)
+	resource, err := c.CreateResourceWithFactory(ctx, path, policy, &createdPolicy, NewPolicyResource)
 	if err != nil {
 		return nil, err
 	}
@@ -94,7 +95,7 @@ func (c *CloudClient) CreatePolicy(policy *models.Policy) (*PolicyWithETag, erro
 }
 
 // UpdatePolicy updates an existing policy using the PUT method
-func (c *CloudClient) UpdatePolicy(policy *models.Policy, etag string) (*PolicyWithETag, error) {
+func (c *CloudClient) UpdatePolicy(ctx context.Context, policy *models.Policy, etag string) (*PolicyWithETag, error) {
 	path := fmt.Sprintf("/ps/%s/access/policies/%s", policy.PermissionsSystemID, policy.ID)
 
 	// Define a function to get the latest ETag
@@ -143,41 +144,21 @@ func (c *CloudClient) UpdatePolicy(policy *models.Policy, etag string) (*PolicyW
 		return respWithETag, nil
 	}
 
-	// First attempt with the provided ETag
-	respWithETag, err := updateWithETag(etag)
+	// Use retry logic with exponential backoff
+	retryConfig := DefaultRetryConfig()
+	respWithETag, err := retryConfig.RetryWithExponentialBackoffLegacy(
+		ctx,
+		func() (*ResponseWithETag, error) {
+			return updateWithETag(etag)
+		},
+		getLatestETag,
+		updateWithETag,
+		"policy update",
+	)
 	if err != nil {
 		return nil, err
 	}
 
-	// Handle retryable errors (412 Precondition Failed, 409 Conflict) by refreshing ETag and retrying
-	retryWithFreshETag := func(errorContext string) error {
-		// Close the body of the first response
-		if respWithETag.Response.Body != nil {
-			_ = respWithETag.Response.Body.Close()
-		}
-
-		// Get the latest ETag
-		latestETag, err := getLatestETag()
-		if err != nil {
-			return fmt.Errorf("failed to get latest ETag for retry (%s): %w", errorContext, err)
-		}
-
-		// Retry the update with the latest ETag
-		respWithETag, err = updateWithETag(latestETag)
-		return err
-	}
-
-	switch respWithETag.Response.StatusCode {
-	case http.StatusPreconditionFailed:
-		if err := retryWithFreshETag("precondition failed"); err != nil {
-			return nil, err
-		}
-	case http.StatusConflict:
-		if err := retryWithFreshETag("FGAM configuration change"); err != nil {
-			return nil, err
-		}
-	}
-
 	// Keep the response body for potential error reporting
 	var respBody []byte
 	if respWithETag.Response.Body != nil {