Only trigger bootstrap on topology change if node receives new shards

src/dbnode/storage/database.go

@@ -398,7 +398,15 @@ func (d *db) AssignShardSet(shardSet sharding.ShardSet) {
 		ns.AssignShardSet(shardSet)
 	}
 
-	d.queueBootstrapWithLock()
+	if receivedNewShards {
+		// Only trigger a bootstrap if the node received new shards otherwise
+		// the nodes will perform lots of small bootstraps (that accomplish nothing)
+		// during topology changes as other nodes mark their shards as available.
+		//
+		// These small bootstraps can significantly delay topology changes as they prevent
+		// the nodes from marking themselves as bootstrapped and durable, for example.
+		d.queueBootstrapWithLock()
+	}
 }
 
 func (d *db) hasReceivedNewShardsWithLock(incoming sharding.ShardSet) bool {

src/dbnode/storage/database_test.go

@@ -404,7 +404,7 @@ func TestDatabaseAssignShardSet(t *testing.T) {
 	wg.Wait()
 }
 
-func TestDatabaseAssignShardSetDoesNotUpdateLastReceivedNewShardsIfNoNewShards(t *testing.T) {
+func TestDatabaseAssignShardSetBehaviorNoNewShards(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
@@ -413,6 +413,11 @@ func TestDatabaseAssignShardSetDoesNotUpdateLastReceivedNewShardsIfNoNewShards(t
 		close(mapCh)
 	}()
 
+	// Set a mock mediator to be certain that bootstrap is not called when
+	// no new shards are assigned.
+	mediator := NewMockdatabaseMediator(ctrl)
+	d.mediator = mediator
+
 	var ns []*MockdatabaseNamespace
 	ns = append(ns, dbAddNewMockNamespace(ctrl, d, "testns1"))
 	ns = append(ns, dbAddNewMockNamespace(ctrl, d, "testns2"))
@@ -427,6 +432,7 @@ func TestDatabaseAssignShardSetDoesNotUpdateLastReceivedNewShardsIfNoNewShards(t
 
 	t1 := d.lastReceivedNewShards
 	d.AssignShardSet(d.shardSet)
+	// Ensure that lastReceivedNewShards is not updated if no new shards are assigned.
 	require.True(t, d.lastReceivedNewShards.Equal(t1))
 
 	wg.Wait()