broker: ActiveStore.Store may be nil if it didn't initialize

Account for this in broker/fragment/stores.go

Also, don't embed ActiveStore.Store to make it more obvious to the
reader that it could be nil.

Author: jgraettinger

broker/fragment/stores.go

@@ -47,12 +47,12 @@ func Persist(ctx context.Context, spool Spool, spec *pb.JournalSpec, isExiting b
 			spool.PathPostfix = postfix
 		}
 
-		var store = stores.Get(spool.Fragment.BackingStore)
-		store.Mark.Store(true)
+		var active = stores.Get(spool.Fragment.BackingStore)
+		active.Mark.Store(true)
 
-		exists, err := store.Exists(ctx, spool.ContentPath())
+		exists, err := active.Exists(ctx, spool.ContentPath())
 		if err != nil {
-			if store.IsAuthError(err) && isExiting && len(fs) != 0 {
+			if isExiting && len(fs) != 0 && active.Store != nil && active.Store.IsAuthError(err) {
 				continue // Fall back to the next store.
 			}
 			return err
@@ -91,9 +91,9 @@ func Persist(ctx context.Context, spool Spool, spec *pb.JournalSpec, isExiting b
 			contentLength = spool.ContentLength()
 		}
 
-		err = store.Put(timeoutCtx, spool.ContentPath(), reader, contentLength, contentEncoding)
+		err = active.Put(timeoutCtx, spool.ContentPath(), reader, contentLength, contentEncoding)
 		if err != nil {
-			if store.IsAuthError(err) && isExiting && len(fs) != 0 {
+			if isExiting && len(fs) != 0 && active.Store != nil && active.Store.IsAuthError(err) {
 				continue // During shutdown with auth errors, try next store.
 			}
 			return err

broker/stores/active_store.go

@@ -13,9 +13,9 @@ import (
 
 // ActiveStore wraps a Store implementation with mark-and-sweep GC support and instrumentation.
 type ActiveStore struct {
-	Key pb.FragmentStore // FragmentStore from which this ActiveStore was built
-	Store
-	Mark atomic.Bool // Mark bit for GC
+	Key   pb.FragmentStore // FragmentStore from which this ActiveStore was built
+	Store Store
+	Mark  atomic.Bool // Mark bit for GC
 
 	initErr error // Initialization error (if any) - checked by all methods
 

broker/stores/health_check.go

@@ -58,11 +58,11 @@ func checkLoop(stores map[pb.FragmentStore]*ActiveStore, fs pb.FragmentStore, at
 		}
 
 		log.WithFields(log.Fields{
-			"store":    fs,
-			"error":    err,
-			"attempt":  attempt,
-			"interval": checkInterval,
-			"authError": active.IsAuthError(err),
+			"store":     fs,
+			"error":     err,
+			"attempt":   attempt,
+			"interval":  checkInterval,
+			"authError": active.Store.IsAuthError(err),
 		}).Warn("store health check failed")
 		attempt += 1
 	}