address review comments

Signed-off-by: everettraven <[email protected]>

Author: everettraven

internal/contentmanager/cache/cache.go

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"slices"
+	"strings"
 	"sync"
 	"time"
 
@@ -39,6 +41,10 @@ type CloserSyncingSource interface {
 }
 
 type sourcerer interface {
+	// Source returns a CloserSyncingSource for the provided
+	// GroupVersionKind. If the CloserSyncingSource encounters an
+	// error after having initially synced, it should requeue the
+	// provided client.Object and call the provided callback function
 	Source(schema.GroupVersionKind, client.Object, func(context.Context)) (CloserSyncingSource, error)
 }
 
@@ -85,6 +91,11 @@ func (c *cache) Close() error {
 			errs = append(errs, err)
 		}
 	}
+
+	slices.SortFunc(errs, func(a, b error) int {
+		return strings.Compare(a.Error(), b.Error())
+	})
+
 	return errors.Join(errs...)
 }
 
@@ -132,11 +143,19 @@ func (c *cache) startNewSources(ctx context.Context, gvks sets.Set[schema.GroupV
 		}
 	}
 
+	slices.SortFunc(sourcesErrors, func(a, b error) int {
+		return strings.Compare(a.Error(), b.Error())
+	})
+
 	return errors.Join(sourcesErrors...)
 }
 
 func (c *cache) startNewSource(ctx context.Context, gvk schema.GroupVersionKind, watcher Watcher) (CloserSyncingSource, error) {
 	s, err := c.sourcerer.Source(gvk, c.owner, func(ctx context.Context) {
+		// this callback function ensures that we remove the source from the
+		// cache if it encounters an error after it initially synced successfully
+		c.mu.Lock()
+		defer c.mu.Unlock()
 		err := c.removeSource(gvk)
 		if err != nil {
 			logr := log.FromContext(ctx)
@@ -180,6 +199,11 @@ func (c *cache) removeStaleSources(gvks sets.Set[schema.GroupVersionKind]) error
 			removeErrs = append(removeErrs, err)
 		}
 	}
+
+	slices.SortFunc(removeErrs, func(a, b error) int {
+		return strings.Compare(a.Error(), b.Error())
+	})
+
 	return errors.Join(removeErrs...)
 }
 

internal/contentmanager/source/dynamicsource.go

@@ -112,7 +112,7 @@ func (dis *dynamicInformerSource) Start(ctx context.Context, q workqueue.TypedRa
 			// We won't be able to update the ClusterExtension status
 			// conditions so instead force a requeue if we
 			// have previously synced and then errored
-			q.Add(reconcile.Request{
+			defer q.Add(reconcile.Request{
 				NamespacedName: types.NamespacedName{
 					Name: dis.cfg.Owner.GetName(),
 				},
@@ -139,7 +139,7 @@ func (dis *dynamicInformerSource) Start(ctx context.Context, q workqueue.TypedRa
 			close(dis.syncedChan)
 		})
 
-		_ = wait.PollUntilContextCancel(dis.informerCtx, time.Second, true, func(ctx context.Context) (bool, error) {
+		_ = wait.PollUntilContextCancel(dis.informerCtx, time.Second, true, func(_ context.Context) (bool, error) {
 			if sharedIndexInf.HasSynced() {
 				syncOnce()
 				return true, nil