diff --git a/internal/sandbox/sandbox.go b/internal/sandbox/sandbox.go index c2d6cdf60d..99f1aaec0c 100644 --- a/internal/sandbox/sandbox.go +++ b/internal/sandbox/sandbox.go @@ -43,6 +43,13 @@ const ( // concurrent ImportProfile processes deleting and reimporting profiles. providerRetries = 3 providerRetryBackoff = 500 * time.Millisecond + + // updateRetries is the number of times updateProvider retries on + // optimistic concurrency ("provider was modified concurrently") errors. + // Each retry re-invokes openshell which re-reads the current + // resource_version, resolving the conflict. + updateRetries = 3 + updateRetryBaseBackoff = 100 * time.Millisecond ) // RetrySleepFn is the function called between retry attempts in @@ -343,6 +350,18 @@ func isUnsupportedProviderErr(err error) bool { return err != nil && strings.Contains(strings.ToLower(err.Error()), "unsupported provider type or profile") } +// isConcurrentModificationErr reports whether err contains the openshell +// error message indicating a resource_version conflict from concurrent +// provider updates. +// +// NOTE: This matches literal text from the openshell CLI's stderr output. +// If openshell changes its error wording in a future version, this check +// will silently stop matching and retries will no longer trigger. Update +// the substring if the upstream message changes. +func isConcurrentModificationErr(err error) bool { + return err != nil && strings.Contains(strings.ToLower(err.Error()), "provider was modified concurrently") +} + // tryCreateProvider performs a single attempt to create (or update) a // provider via openshell. Extracted from EnsureProvider to support retry. func tryCreateProvider(ctx context.Context, name string, args, extraEnv, secrets []string, credentials, config map[string]string, fromURL bool) error { @@ -368,22 +387,48 @@ func tryCreateProvider(ctx context.Context, name string, args, extraEnv, secrets return nil } -// updateProvider runs openshell provider update for an already-existing provider. +// updateProvider runs openshell provider update for an already-existing +// provider. Concurrent modification errors (resource_version conflicts) +// are retried with jittered exponential backoff. Each retry re-invokes +// openshell, which re-reads the current resource_version. func updateProvider(ctx context.Context, name string, credentials, config map[string]string, extraEnv, secrets []string, fromURL bool) error { args := buildProviderUpdateArgs(name, credentials, config, fromURL) - ctx, cancel := context.WithTimeout(ctx, providerTimeout) - defer cancel() - cmd := exec.CommandContext(ctx, "openshell", args...) - cmd.Env = append(os.Environ(), extraEnv...) - out, err := cmd.CombinedOutput() - if err != nil { + + var lastErr error + for attempt := range updateRetries { + updateCtx, cancel := context.WithTimeout(ctx, providerTimeout) + cmd := exec.CommandContext(updateCtx, "openshell", args...) + cmd.Env = append(os.Environ(), extraEnv...) + out, err := cmd.CombinedOutput() + cancel() + + if err == nil { + return nil + } + outStr := string(out) for _, s := range secrets { outStr = strings.ReplaceAll(outStr, s, "***") } - return fmt.Errorf("provider update %q failed: %w (output: %s)", name, err, outStr) + lastErr = fmt.Errorf("provider update %q failed: %w (output: %s)", name, err, outStr) + + if !isConcurrentModificationErr(lastErr) { + return lastErr + } + + // Retry with jittered exponential backoff: base * 2^attempt ± jitter. + if attempt < updateRetries-1 { + backoff := updateRetryBaseBackoff << attempt // 100ms, 200ms, 400ms + // Add up to 50% jitter to reduce collision probability. + jitter := time.Duration(rand.Int64N(int64(backoff) / 2)) + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoff + jitter): + } + } } - return nil + return fmt.Errorf("retries exhausted after %d attempts: %w", updateRetries, lastErr) } // buildProviderUpdateArgs constructs CLI args for openshell provider update. diff --git a/internal/sandbox/sandbox_test.go b/internal/sandbox/sandbox_test.go index d3a467b8e8..c0e07a20e9 100644 --- a/internal/sandbox/sandbox_test.go +++ b/internal/sandbox/sandbox_test.go @@ -1371,6 +1371,192 @@ fi assert.Contains(t, err.Error(), "***") } +// TestEnsureProvider_RetriesOnConcurrentModification verifies that +// updateProvider retries when openshell returns a concurrent modification +// (resource_version conflict) error during provider update. +func TestEnsureProvider_RetriesOnConcurrentModification(t *testing.T) { + dir := t.TempDir() + markerDir := filepath.Join(dir, "markers") + require.NoError(t, os.MkdirAll(markerDir, 0o755)) + + // Fake openshell: create returns AlreadyExists to trigger update path. + // First 2 update calls fail with concurrent modification error, + // third update call succeeds. + script := fmt.Sprintf(`#!/bin/sh +if [ "$2" = "create" ]; then + echo "code: 'Some entity that we attempted to create already exists', message: \"provider already exists\"" >&2 + exit 1 +elif [ "$2" = "update" ]; then + echo x > "%s/update.$$" + count=0 + for f in "%s"/update.*; do + [ -e "$f" ] && count=$((count + 1)) + done + if [ "$count" -lt 3 ]; then + echo "code: 'The operation was aborted', message: \"provider was modified concurrently (current resource_version: $count)\"" >&2 + exit 1 + fi + exit 0 +fi +exit 0 +`, markerDir, markerDir) + fakePath := filepath.Join(dir, "openshell") + require.NoError(t, os.WriteFile(fakePath, []byte(script), 0o755)) + t.Setenv("PATH", dir) + + err := EnsureProvider(context.Background(), "vertex-ai", "vertex-ai", map[string]string{"TOKEN": "tok"}, nil, false) + assert.NoError(t, err, "should succeed after retrying concurrent modification errors") + + // Verify that 3 update attempts were made. + entries, readErr := os.ReadDir(markerDir) + require.NoError(t, readErr) + assert.Len(t, entries, 3, "should have made 3 update attempts") +} + +// TestEnsureProvider_ConcurrentModification_MaxRetriesExhausted verifies +// that updateProvider propagates the error when all retries are exhausted. +func TestEnsureProvider_ConcurrentModification_MaxRetriesExhausted(t *testing.T) { + dir := t.TempDir() + markerDir := filepath.Join(dir, "markers") + require.NoError(t, os.MkdirAll(markerDir, 0o755)) + + // Fake openshell: create returns AlreadyExists, update always fails + // with concurrent modification. + script := fmt.Sprintf(`#!/bin/sh +if [ "$2" = "create" ]; then + echo "code: 'Some entity that we attempted to create already exists', message: \"provider already exists\"" >&2 + exit 1 +elif [ "$2" = "update" ]; then + echo x > "%s/update.$$" + echo "code: 'The operation was aborted', message: \"provider was modified concurrently (current resource_version: 1)\"" >&2 + exit 1 +fi +exit 0 +`, markerDir) + fakePath := filepath.Join(dir, "openshell") + require.NoError(t, os.WriteFile(fakePath, []byte(script), 0o755)) + t.Setenv("PATH", dir) + + err := EnsureProvider(context.Background(), "vertex-ai", "vertex-ai", map[string]string{"TOKEN": "tok"}, nil, false) + require.Error(t, err) + assert.Contains(t, err.Error(), "retries exhausted after 3 attempts") + assert.Contains(t, err.Error(), "provider was modified concurrently") + + // All 3 retry attempts should have been made. + entries, readErr := os.ReadDir(markerDir) + require.NoError(t, readErr) + assert.Len(t, entries, 3, "should exhaust all retry attempts") +} + +// TestEnsureProvider_ConcurrentModification_NoRetryOnOtherUpdateErrors +// verifies that non-concurrency update errors are not retried. +func TestEnsureProvider_ConcurrentModification_NoRetryOnOtherUpdateErrors(t *testing.T) { + dir := t.TempDir() + markerDir := filepath.Join(dir, "markers") + require.NoError(t, os.MkdirAll(markerDir, 0o755)) + + // Fake openshell: create returns AlreadyExists, update fails with + // a non-concurrency error. + script := fmt.Sprintf(`#!/bin/sh +if [ "$2" = "create" ]; then + echo "code: 'Some entity that we attempted to create already exists', message: \"provider already exists\"" >&2 + exit 1 +elif [ "$2" = "update" ]; then + echo x > "%s/update.$$" + echo "gateway unavailable" >&2 + exit 1 +fi +exit 0 +`, markerDir) + fakePath := filepath.Join(dir, "openshell") + require.NoError(t, os.WriteFile(fakePath, []byte(script), 0o755)) + t.Setenv("PATH", dir) + + err := EnsureProvider(context.Background(), "vertex-ai", "vertex-ai", nil, nil, false) + require.Error(t, err) + assert.Contains(t, err.Error(), "provider update") + assert.NotContains(t, err.Error(), "retries exhausted", + "non-concurrency errors should not be retried") + + // Should have only attempted once. + entries, readErr := os.ReadDir(markerDir) + require.NoError(t, readErr) + assert.Len(t, entries, 1, "should not retry on non-concurrency errors") +} + +// TestEnsureProvider_ConcurrentModification_ContextCancelled verifies that +// context cancellation during the update retry backoff returns the context +// error. +func TestEnsureProvider_ConcurrentModification_ContextCancelled(t *testing.T) { + dir := t.TempDir() + markerDir := filepath.Join(dir, "markers") + require.NoError(t, os.MkdirAll(markerDir, 0o755)) + + // Fake openshell: create returns AlreadyExists, update always fails + // with concurrent modification so retries never succeed. + script := fmt.Sprintf(`#!/bin/sh +if [ "$2" = "create" ]; then + echo "code: 'Some entity that we attempted to create already exists', message: \"provider already exists\"" >&2 + exit 1 +elif [ "$2" = "update" ]; then + echo x > "%s/update.$$" + echo "code: 'The operation was aborted', message: \"provider was modified concurrently (current resource_version: 1)\"" >&2 + exit 1 +fi +exit 0 +`, markerDir) + fakePath := filepath.Join(dir, "openshell") + require.NoError(t, os.WriteFile(fakePath, []byte(script), 0o755)) + t.Setenv("PATH", dir) + + // Cancel the context shortly after the first update attempt so the + // select picks up ctx.Done() during the backoff sleep. The update + // retry base backoff is 100ms, so 80ms is enough for one update + // attempt to run but not enough for the backoff to complete. + ctx, cancel := context.WithTimeout(context.Background(), 80*time.Millisecond) + defer cancel() + + err := EnsureProvider(ctx, "vertex-ai", "vertex-ai", nil, nil, false) + require.Error(t, err) + assert.ErrorIs(t, err, context.DeadlineExceeded, + "should return context error when cancelled during retry sleep") + + // Should have made only 1 update attempt before the context expired + // during the backoff sleep. + entries, readErr := os.ReadDir(markerDir) + require.NoError(t, readErr) + assert.Len(t, entries, 1, "should stop retrying when context is cancelled") +} + +// TestEnsureProvider_ConcurrentModification_SecretRedaction verifies that +// secrets are redacted in error messages from concurrent modification retries. +func TestEnsureProvider_ConcurrentModification_SecretRedaction(t *testing.T) { + dir := t.TempDir() + + // Fake openshell: create returns AlreadyExists, update always fails + // with concurrent modification and includes the secret in output. + script := `#!/bin/sh +if [ "$2" = "create" ]; then + echo "code: 'Some entity that we attempted to create already exists', message: \"provider already exists\"" >&2 + exit 1 +elif [ "$2" = "update" ]; then + echo "provider was modified concurrently supersecret" >&2 + exit 1 +fi +exit 0 +` + fakePath := filepath.Join(dir, "openshell") + require.NoError(t, os.WriteFile(fakePath, []byte(script), 0o755)) + t.Setenv("PATH", dir) + + err := EnsureProvider(context.Background(), "vertex-ai", "vertex-ai", + map[string]string{"TOKEN": "supersecret"}, nil, false) + require.Error(t, err) + assert.NotContains(t, err.Error(), "supersecret", + "secret must be redacted in concurrent modification error") + assert.Contains(t, err.Error(), "***") +} + func TestEnsureProvider_RejectsReservedCredentialKeys(t *testing.T) { tests := []struct { key string