fix: improve ASM error messages (#1879)

- Looking into why we are getting ASM sync errors on staging and it is
hard to tell whether the error is happening in the leader or the
follower. These error changes should help this.
- `secretsCache` should not have any mention of ASM as it is meant to be
generic

common/configuration/asm_follower.go

@@ -27,7 +27,7 @@ var _ asmClient = &asmFollower{}
 func newASMFollower(ctx context.Context, rpcClient ftlv1connect.AdminServiceClient, clock clock.Clock) *asmFollower {
 	f := &asmFollower{
 		client: rpcClient,
-		cache:  newSecretsCache(),
+		cache:  newSecretsCache("asm-follower"),
 	}
 	go f.cache.sync(ctx, asmFollowerSyncInterval, func(ctx context.Context, secrets *xsync.MapOf[Ref, cachedSecret]) error {
 		return f.sync(ctx, secrets)
@@ -43,7 +43,7 @@ func (f *asmFollower) sync(ctx context.Context, secrets *xsync.MapOf[Ref, cached
 		IncludeValues: &includeValues,
 	}))
 	if err != nil {
-		return err
+		return fmt.Errorf("error getting secrets list from leader: %w", err)
 	}
 	visited := map[Ref]bool{}
 	for _, s := range resp.Msg.Secrets {

common/configuration/asm_leader.go

@@ -30,7 +30,7 @@ var _ asmClient = &asmLeader{}
 func newASMLeader(ctx context.Context, client *secretsmanager.Client, clock clock.Clock) *asmLeader {
 	l := &asmLeader{
 		client: client,
-		cache:  newSecretsCache(),
+		cache:  newSecretsCache("asm-leader"),
 	}
 	go l.cache.sync(ctx, asmLeaderSyncInterval, func(ctx context.Context, secrets *xsync.MapOf[Ref, cachedSecret]) error {
 		return l.sync(ctx, secrets)
@@ -56,7 +56,7 @@ func (l *asmLeader) sync(ctx context.Context, secrets *xsync.MapOf[Ref, cachedSe
 			NextToken:  nextToken.Ptr(),
 		})
 		if err != nil {
-			return fmt.Errorf("unable to list secrets: %w", err)
+			return fmt.Errorf("unable to get list of secrets from ASM: %w", err)
 		}
 
 		var activeSecrets = slices.Filter(out.SecretList, func(s types.SecretListEntry) bool {
@@ -65,7 +65,7 @@ func (l *asmLeader) sync(ctx context.Context, secrets *xsync.MapOf[Ref, cachedSe
 		for _, s := range activeSecrets {
 			ref, err := ParseRef(*s.Name)
 			if err != nil {
-				return fmt.Errorf("unable to parse ref: %w", err)
+				return fmt.Errorf("unable to parse ref from ASM secret: %w", err)
 			}
 			seen[ref] = true
 
@@ -103,7 +103,7 @@ func (l *asmLeader) sync(ctx context.Context, secrets *xsync.MapOf[Ref, cachedSe
 			SecretIdList: secretIDs,
 		})
 		if err != nil {
-			return fmt.Errorf("unable to batch get secret values: %w", err)
+			return fmt.Errorf("unable to get batch of secret values from ASM: %w", err)
 		}
 		for _, s := range out.SecretValues {
 			ref, err := ParseRef(*s.Name)
@@ -112,7 +112,7 @@ func (l *asmLeader) sync(ctx context.Context, secrets *xsync.MapOf[Ref, cachedSe
 			}
 			// Expect secrets to be strings, not binary
 			if s.SecretBinary != nil {
-				return fmt.Errorf("secret for %s is not a string", ref)
+				return fmt.Errorf("secret for %s in ASM is not a string", ref)
 			}
 			data := []byte(*s.SecretString)
 			secrets.Store(ref, cachedSecret{
@@ -159,11 +159,11 @@ func (l *asmLeader) store(ctx context.Context, ref Ref, value []byte) (*url.URL,
 			SecretString: aws.String(string(value)),
 		})
 		if err != nil {
-			return nil, fmt.Errorf("unable to update secret: %w", err)
+			return nil, fmt.Errorf("unable to update secret in ASM: %w", err)
 		}
 
 	} else if err != nil {
-		return nil, fmt.Errorf("unable to store secret: %w", err)
+		return nil, fmt.Errorf("unable to store secret in ASM: %w", err)
 	}
 	l.cache.updatedSecret(ref, value)
 	return asmURLForRef(ref), nil
@@ -176,7 +176,7 @@ func (l *asmLeader) delete(ctx context.Context, ref Ref) error {
 		ForceDeleteWithoutRecovery: &t,
 	})
 	if err != nil {
-		return fmt.Errorf("unable to delete secret: %w", err)
+		return fmt.Errorf("unable to delete secret from ASM: %w", err)
 	}
 	l.cache.deletedSecret(ref)
 	return nil

common/configuration/secrets_cache.go

@@ -2,7 +2,6 @@ package configuration
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -36,23 +35,26 @@ type updatedSecretEvent struct {
 }
 
 type secretsCache struct {
-	// closed when secrets have been loaded from ASM for the first time
+	name string
+
+	// closed when secrets have been synced for the first time
 	loaded chan bool
 
-	// secrets is a map of secrets that have been loaded from ASM.
+	// secrets is a map of secrets that have been synced.
 	// optional is nil when not loaded yet
 	// it is updated via:
-	// - polling ASM
-	// - when we write to ASM
+	// - periodic syncs
+	// - when we update or delete a secret
 	secrets *xsync.MapOf[Ref, cachedSecret]
 
 	topic *pubsub.Topic[updatedSecretEvent]
 	// used by tests to wait for events to be processed
 	topicWaitGroup sync.WaitGroup
 }
 
-func newSecretsCache() *secretsCache {
+func newSecretsCache(name string) *secretsCache {
 	return &secretsCache{
+		name:    name,
 		loaded:  make(chan bool),
 		secrets: xsync.NewMapOf[Ref, cachedSecret](),
 		topic:   pubsub.New[updatedSecretEvent](),
@@ -134,7 +136,7 @@ func (c *secretsCache) sync(ctx context.Context, frequency time.Duration, synchr
 				continue
 			}
 			// back off if we fail to sync
-			logger.Warnf("Unable to sync ASM secrets: %v", err)
+			logger.Warnf("Unable to sync %s: %v", c.name, err)
 			nextSync = clock.Now().Add(backOff)
 			if nextSync.After(clock.Now().Add(frequency)) {
 				nextSync = clock.Now().Add(frequency)
@@ -168,7 +170,7 @@ func (c *secretsCache) processEvent(e updatedSecretEvent) {
 func (c *secretsCache) waitForSecrets() error {
 	select {
 	case <-time.After(loadSecretsTimeout):
-		return errors.New("secrets not synced from ASM yet")
+		return fmt.Errorf("secrets not synced for %s yet", c.name)
 	case <-c.loaded:
 		return nil
 	}