[dbnode] Skip bootstrapping shards from aggregation

src/dbnode/storage/namespace.go

@@ -1766,6 +1766,7 @@ func (n *dbNamespace) aggregateTiles(
 		processedTileCount int64
 		aggregationSuccess bool
 	)
+
 	defer func() {
 		if aggregationSuccess {
 			return
@@ -1776,7 +1777,15 @@ func (n *dbNamespace) aggregateTiles(
 				zap.Stringer("sourceNs", sourceNs.ID()), zap.Error(err))
 		}
 	}()
+
 	for _, targetShard := range n.OwnedShards() {
+		if !targetShard.IsBootstrapped() {
+			n.log.
+				With(zap.Uint32("shard", targetShard.ID())).
+				Debug("skipping aggregateTiles due to shard not bootstrapped")
+			continue
+		}
+
 		shardProcessedTileCount, err := targetShard.AggregateTiles(
 			ctx, sourceNs, n, targetShard.ID(), onColdFlushNs, opts)
 

src/dbnode/storage/namespace_test.go

@@ -1629,6 +1629,9 @@ func TestNamespaceAggregateTiles(t *testing.T) {
 	targetNs.shards[0] = targetShard0
 	targetNs.shards[1] = targetShard1
 
+	targetShard0.EXPECT().IsBootstrapped().Return(true)
+	targetShard1.EXPECT().IsBootstrapped().Return(true)
+
 	targetShard0.EXPECT().ID().Return(shard0ID)
 	targetShard1.EXPECT().ID().Return(shard1ID)
 
@@ -1646,6 +1649,56 @@ func TestNamespaceAggregateTiles(t *testing.T) {
 	assert.Equal(t, int64(3+2), processedTileCount)
 }
 
+func TestNamespaceAggregateTilesShipBootstrappingShards(t *testing.T) {
+	ctrl := xtest.NewController(t)
+	defer ctrl.Finish()
+
+	ctx := context.NewBackground()
+	defer ctx.Close()
+
+	var (
+		sourceNsID      = ident.StringID("source")
+		targetNsID      = ident.StringID("target")
+		sourceBlockSize = time.Hour
+		targetBlockSize = 2 * time.Hour
+		start           = time.Now().Truncate(targetBlockSize)
+		shard0ID        = uint32(10)
+		shard1ID        = uint32(20)
+		insOpts         = instrument.NewOptions()
+	)
+
+	opts, err := NewAggregateTilesOptions(start, start.Add(targetBlockSize), time.Second, targetNsID, insOpts)
+	require.NoError(t, err)
+
+	sourceNs, sourceCloser := newTestNamespaceWithIDOpts(t, sourceNsID, namespace.NewOptions())
+	defer sourceCloser()
+	sourceNs.bootstrapState = Bootstrapped
+	sourceRetentionOpts := sourceNs.nopts.RetentionOptions().SetBlockSize(sourceBlockSize)
+	sourceNs.nopts = sourceNs.nopts.SetRetentionOptions(sourceRetentionOpts)
+
+	targetNs, targetCloser := newTestNamespaceWithIDOpts(t, targetNsID, namespace.NewOptions())
+	defer targetCloser()
+	targetNs.bootstrapState = Bootstrapped
+	targetRetentionOpts := targetNs.nopts.RetentionOptions().SetBlockSize(targetBlockSize)
+	targetNs.nopts = targetNs.nopts.SetColdWritesEnabled(true).SetRetentionOptions(targetRetentionOpts)
+
+	targetShard0 := NewMockdatabaseShard(ctrl)
+	targetShard1 := NewMockdatabaseShard(ctrl)
+	targetNs.shards[0] = targetShard0
+	targetNs.shards[1] = targetShard1
+
+	targetShard0.EXPECT().IsBootstrapped().Return(false)
+	targetShard1.EXPECT().IsBootstrapped().Return(false)
+
+	targetShard0.EXPECT().ID().Return(shard0ID)
+	targetShard1.EXPECT().ID().Return(shard1ID)
+
+	processedTileCount, err := targetNs.AggregateTiles(ctx, sourceNs, opts)
+
+	require.NoError(t, err)
+	assert.Zero(t, processedTileCount)
+}
+
 func waitForStats(
 	reporter xmetrics.TestStatsReporter,
 	check func(xmetrics.TestStatsReporter) bool,