Release v1.15.1

aws-source/adapters/adapterhelpers_always_get_source.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sync"
+	"sync/atomic"
 	"time"
 
 	"buf.build/go/protovalidate"
@@ -80,7 +80,7 @@ type AlwaysGetAdapter[ListInput InputType, ListOutput OutputType, GetInput Input
 	ListFuncOutputMapper func(output ListOutput, input ListInput) ([]GetInput, error)
 
 	CacheDuration time.Duration  // How long to cache items for
-	cache         sdpcache.Cache // The cache for this adapter (set during creation, can be nil for tests)
+	cache         sdpcache.Cache // This is mandatory
 }
 
 func (s *AlwaysGetAdapter[ListInput, ListOutput, GetInput, GetOutput, ClientStruct, Options]) cacheDuration() time.Duration {
@@ -91,21 +91,6 @@ func (s *AlwaysGetAdapter[ListInput, ListOutput, GetInput, GetOutput, ClientStru
 	return s.CacheDuration
 }
 
-var (
-	noOpCacheAlwaysGetOnce sync.Once
-	noOpCacheAlwaysGet     sdpcache.Cache
-)
-
-func (s *AlwaysGetAdapter[ListInput, ListOutput, GetInput, GetOutput, ClientStruct, Options]) Cache() sdpcache.Cache {
-	if s.cache == nil {
-		noOpCacheAlwaysGetOnce.Do(func() {
-			noOpCacheAlwaysGet = sdpcache.NewNoOpCache()
-		})
-		return noOpCacheAlwaysGet
-	}
-	return s.cache
-}
-
 // Validate Checks that the adapter has been set up correctly
 func (s *AlwaysGetAdapter[ListInput, ListOutput, GetInput, GetOutput, ClientStruct, Options]) Validate() error {
 	if !s.DisableList {
@@ -168,7 +153,7 @@ func (s *AlwaysGetAdapter[ListInput, ListOutput, GetInput, GetOutput, ClientStru
 		return nil, WrapAWSError(err)
 	}
 
-	cacheHit, ck, cachedItems, qErr, done := s.Cache().Lookup(ctx, s.Name(), sdp.QueryMethod_GET, scope, s.ItemType, query, ignoreCache)
+	cacheHit, ck, cachedItems, qErr, done := s.cache.Lookup(ctx, s.Name(), sdp.QueryMethod_GET, scope, s.ItemType, query, ignoreCache)
 	defer done()
 	if qErr != nil {
 		return nil, qErr
@@ -187,12 +172,12 @@ func (s *AlwaysGetAdapter[ListInput, ListOutput, GetInput, GetOutput, ClientStru
 	if err != nil {
 		err := WrapAWSError(err)
 		if !CanRetry(err) {
-			s.Cache().StoreError(ctx, err, s.cacheDuration(), ck)
+			s.cache.StoreError(ctx, err, s.cacheDuration(), ck)
 		}
 		return nil, err
 	}
 
-	s.Cache().StoreItem(ctx, item, s.cacheDuration(), ck)
+	s.cache.StoreItem(ctx, item, s.cacheDuration(), ck)
 	return item, nil
 }
 
@@ -218,9 +203,13 @@ func (s *AlwaysGetAdapter[ListInput, ListOutput, GetInput, GetOutput, ClientStru
 		return
 	}
 
-	cacheHit, ck, cachedItems, qErr, done := s.Cache().Lookup(ctx, s.Name(), sdp.QueryMethod_LIST, scope, s.ItemType, "", ignoreCache)
+	cacheHit, ck, cachedItems, qErr, done := s.cache.Lookup(ctx, s.Name(), sdp.QueryMethod_LIST, scope, s.ItemType, "", ignoreCache)
 	defer done()
 	if qErr != nil {
+		// For better semantics, convert cached NOTFOUND into empty result
+		if qErr.GetErrorType() == sdp.QueryError_NOTFOUND {
+			return
+		}
 		stream.SendError(qErr)
 		return
 	}
@@ -238,30 +227,53 @@ func (s *AlwaysGetAdapter[ListInput, ListOutput, GetInput, GetOutput, ClientStru
 	paginator := s.ListFuncPaginatorBuilder(s.Client, input)
 	var newGetInputs []GetInput
 	p := pool.New().WithContext(ctx).WithMaxGoroutines(s.MaxParallel.Value())
+
+	// Track whether any items were found and if we had an error
+	var itemsSent atomic.Int64
+	var hadError atomic.Bool
+
 	defer func() {
 		// Always wait for everything to be completed before returning
 		err := p.Wait()
 		if err != nil {
 			sentry.CaptureException(err)
 		}
+
+		// Only cache not-found when no items were found AND no error occurred
+		// If we had an error, that error is already cached, don't overwrite it
+		shouldCacheNotFound := itemsSent.Load() == 0 && !hadError.Load()
+
+		if shouldCacheNotFound {
+			notFoundErr := &sdp.QueryError{
+				ErrorType:     sdp.QueryError_NOTFOUND,
+				ErrorString:   fmt.Sprintf("no %s found in scope %s", s.ItemType, scope),
+				Scope:         scope,
+				SourceName:    s.Name(),
+				ItemType:      s.ItemType,
+				ResponderName: s.Name(),
+			}
+			s.cache.StoreError(ctx, notFoundErr, s.cacheDuration(), ck)
+		}
 	}()
 
 	for paginator.HasMorePages() {
 		output, err := paginator.NextPage(ctx)
 		if err != nil {
+			hadError.Store(true)
 			err := WrapAWSError(err)
 			if !CanRetry(err) {
-				s.Cache().StoreError(ctx, err, s.cacheDuration(), ck)
+				s.cache.StoreError(ctx, err, s.cacheDuration(), ck)
 			}
 			stream.SendError(err)
 			return
 		}
 
 		newGetInputs, err = s.ListFuncOutputMapper(output, input)
 		if err != nil {
+			hadError.Store(true)
 			err := WrapAWSError(err)
 			if !CanRetry(err) {
-				s.Cache().StoreError(ctx, err, s.cacheDuration(), ck)
+				s.cache.StoreError(ctx, err, s.cacheDuration(), ck)
 			}
 			stream.SendError(err)
 			return
@@ -276,10 +288,13 @@ func (s *AlwaysGetAdapter[ListInput, ListOutput, GetInput, GetOutput, ClientStru
 				if err != nil {
 					// Don't cache individual errors as they are cheap to re-run
 					stream.SendError(WrapAWSError(err))
+					// Mark that we had an error so we don't cache NOTFOUND
+					hadError.Store(true)
 				}
 				if item != nil {
-					s.Cache().StoreItem(ctx, item, s.cacheDuration(), ck)
+					s.cache.StoreItem(ctx, item, s.cacheDuration(), ck)
 					stream.SendItem(item)
+					itemsSent.Add(1)
 				}
 
 				return nil
@@ -322,9 +337,13 @@ func (s *AlwaysGetAdapter[ListInput, ListOutput, GetInput, GetOutput, ClientStru
 // SearchCustom Searches using custom mapping logic. The SearchInputMapper is
 // used to create an input for ListFunc, at which point the usual logic is used
 func (s *AlwaysGetAdapter[ListInput, ListOutput, GetInput, GetOutput, ClientStruct, Options]) SearchCustom(ctx context.Context, scope string, query string, ignoreCache bool, stream discovery.QueryResultStream) {
-	cacheHit, ck, cachedItems, qErr, done := s.Cache().Lookup(ctx, s.Name(), sdp.QueryMethod_SEARCH, scope, s.ItemType, query, ignoreCache)
+	cacheHit, ck, cachedItems, qErr, done := s.cache.Lookup(ctx, s.Name(), sdp.QueryMethod_SEARCH, scope, s.ItemType, query, ignoreCache)
 	defer done()
 	if qErr != nil {
+		// For better semantics, convert cached NOTFOUND into empty result
+		if qErr.GetErrorType() == sdp.QueryError_NOTFOUND {
+			return
+		}
 		stream.SendError(qErr)
 		return
 	}
@@ -356,15 +375,26 @@ func (s *AlwaysGetAdapter[ListInput, ListOutput, GetInput, GetOutput, ClientStru
 		if err != nil {
 			err := WrapAWSError(err)
 			if !CanRetry(err) {
-				s.Cache().StoreError(ctx, err, s.cacheDuration(), ck)
+				s.cache.StoreError(ctx, err, s.cacheDuration(), ck)
 			}
 			stream.SendError(err)
 			return
 		}
 
 		if item != nil {
-			s.Cache().StoreItem(ctx, item, s.cacheDuration(), ck)
+			s.cache.StoreItem(ctx, item, s.cacheDuration(), ck)
 			stream.SendItem(item)
+		} else {
+			// Cache not-found when item is nil
+			notFoundErr := &sdp.QueryError{
+				ErrorType:     sdp.QueryError_NOTFOUND,
+				ErrorString:   fmt.Sprintf("%s not found for search query '%s'", s.ItemType, query),
+				Scope:         scope,
+				SourceName:    s.Name(),
+				ItemType:      s.ItemType,
+				ResponderName: s.Name(),
+			}
+			s.cache.StoreError(ctx, notFoundErr, s.cacheDuration(), ck)
 		}
 	} else {
 		stream.SendError(errors.New("SearchCustom called without SearchInputMapper or SearchGetInputMapper"))

aws-source/adapters/adapterhelpers_always_get_source_test.go

@@ -76,6 +76,7 @@ func TestAlwaysGetSourceGet(t *testing.T) {
 			GetInputMapper: func(scope, query string) string {
 				return ""
 			},
+			cache: sdpcache.NewNoOpCache(),
 		}
 
 		_, err := lgs.Get(context.Background(), "foo.bar", "", false)
@@ -108,6 +109,7 @@ func TestAlwaysGetSourceGet(t *testing.T) {
 			GetInputMapper: func(scope, query string) string {
 				return ""
 			},
+			cache: sdpcache.NewNoOpCache(),
 		}
 
 		_, err := lgs.Get(context.Background(), "foo.bar", "", false)
@@ -144,6 +146,7 @@ func TestAlwaysGetSourceList(t *testing.T) {
 			GetInputMapper: func(scope, query string) string {
 				return ""
 			},
+			cache: sdpcache.NewNoOpCache(),
 		}
 
 		stream := discovery.NewRecordingQueryResultStream()
@@ -183,6 +186,7 @@ func TestAlwaysGetSourceList(t *testing.T) {
 			GetInputMapper: func(scope, query string) string {
 				return ""
 			},
+			cache: sdpcache.NewNoOpCache(),
 		}
 
 		stream := discovery.NewRecordingQueryResultStream()
@@ -228,6 +232,7 @@ func TestAlwaysGetSourceList(t *testing.T) {
 			GetInputMapper: func(scope, query string) string {
 				return ""
 			},
+			cache: sdpcache.NewNoOpCache(),
 		}
 
 		stream := discovery.NewRecordingQueryResultStream()
@@ -275,6 +280,7 @@ func TestAlwaysGetSourceSearch(t *testing.T) {
 			GetInputMapper: func(scope, query string) string {
 				return scope + "." + query
 			},
+			cache: sdpcache.NewNoOpCache(),
 		}
 
 		t.Run("bad ARN", func(t *testing.T) {
@@ -338,6 +344,7 @@ func TestAlwaysGetSourceSearch(t *testing.T) {
 			GetInputMapper: func(scope, query string) string {
 				return scope + "." + query
 			},
+			cache: sdpcache.NewNoOpCache(),
 		}
 
 		t.Run("ARN", func(t *testing.T) {
@@ -400,6 +407,7 @@ func TestAlwaysGetSourceSearch(t *testing.T) {
 			GetInputMapper: func(scope, query string) string {
 				return ""
 			},
+			cache: sdpcache.NewNoOpCache(),
 		}
 
 		stream := discovery.NewRecordingQueryResultStream()
@@ -448,6 +456,7 @@ func TestAlwaysGetSourceSearch(t *testing.T) {
 			GetInputMapper: func(scope, query string) string {
 				return scope + "." + query
 			},
+			cache: sdpcache.NewNoOpCache(),
 		}
 
 		stream := discovery.NewRecordingQueryResultStream()
@@ -475,7 +484,7 @@ func TestAlwaysGetSourceCaching(t *testing.T) {
 		Region:          "eu-west-2",
 		Client:          struct{}{},
 		ListInput:       "",
-		cache:        sdpcache.NewCache(ctx),
+		cache:           sdpcache.NewMemoryCache(),
 		ListFuncPaginatorBuilder: func(client struct{}, input string) Paginator[string, struct{}] {
 			return &TestPaginator{
 				DataFunc: func() string {