Add test coverage for shard.Bootstrap/Tick concurrency (#1616)

src/dbnode/storage/shard_race_prop_test.go

@@ -30,6 +30,7 @@ import (
 
 	"github.com/m3db/m3/src/dbnode/runtime"
 	"github.com/m3db/m3/src/dbnode/storage/block"
+	"github.com/m3db/m3/src/dbnode/storage/bootstrap/result"
 	"github.com/m3db/m3/src/dbnode/storage/series"
 	"github.com/m3db/m3/src/x/context"
 	"github.com/m3db/m3/src/x/ident"
@@ -132,7 +133,7 @@ func TestShardTickWriteRace(t *testing.T) {
 	}()
 
 	ids := []ident.ID{}
-	for i := 0; i < 1; i++ {
+	for i := 0; i < 10; i++ {
 		ids = append(ids, ident.StringID(fmt.Sprintf("foo.%d", i)))
 	}
 
@@ -174,7 +175,102 @@ func TestShardTickWriteRace(t *testing.T) {
 	go func() {
 		defer doneFn()
 		<-barrier
-		shard.Tick(context.NewNoOpCanncellable(), time.Now())
+		_, err := shard.Tick(context.NewNoOpCanncellable(), time.Now())
+		assert.NoError(t, err)
+	}()
+
+	for i := 0; i < numRoutines; i++ {
+		barrier <- struct{}{}
+	}
+
+	wg.Wait()
+}
+
+func TestShardTickBootstrapWriteRace(t *testing.T) {
+	shard, opts := propTestDatabaseShard(t, 10)
+	defer func() {
+		if r := recover(); r != nil {
+			assert.Fail(t, "unexpected panic: %v", r)
+		}
+		shard.Close()
+		opts.RuntimeOptionsManager().Close()
+	}()
+
+	// distribute ids into 3 categories
+	// (1) existing in the shard prior to bootstrap (for w/e reason)
+	// (2) actively being written to by Write()
+	// (3) inserted via Bootstrap()
+	// further, we ensure there's pairwise overlaps between each pair of categories.
+
+	// total ids = 30, splitting id space into following
+	// (1) - existingIDs - [0, 20)
+	// (2) - writeIDs - [10, 30)
+	// (3) - bootstrapIDs - [0, 10) U [] [20, 30)
+
+	var writeIDs []ident.ID
+	bootstrapResult := result.NewMap(result.MapOptions{})
+
+	for i := 0; i < 30; i++ {
+		id := ident.StringID(fmt.Sprintf("foo.%d", i))
+		// existing ids
+		if i < 20 {
+			addTestSeriesWithCountAndBootstrap(shard, id, 0, false)
+		}
+		// write ids
+		if i >= 10 {
+			writeIDs = append(writeIDs, id)
+		}
+		// botstrap ids
+		if i < 10 || i >= 20 {
+			bootstrapResult.Set(id, result.DatabaseSeriesBlocks{
+				ID:     id,
+				Tags:   ident.NewTags(),
+				Blocks: block.NewDatabaseSeriesBlocks(3),
+			})
+		}
+	}
+
+	var (
+		numRoutines = 1 + /* Bootstrap */ +1 /* Tick */ + len(writeIDs) /* Write(s) */
+		barrier     = make(chan struct{}, numRoutines)
+		wg          sync.WaitGroup
+	)
+
+	wg.Add(numRoutines)
+
+	doneFn := func() {
+		if r := recover(); r != nil {
+			assert.Fail(t, "unexpected panic: %v", r)
+		}
+		wg.Done()
+	}
+
+	for _, id := range writeIDs {
+		id := id
+		go func() {
+			defer doneFn()
+			<-barrier
+			ctx := context.NewContext()
+			now := time.Now()
+			_, wasWritten, err := shard.Write(ctx, id, now, 1.0, xtime.Second, nil, series.WriteOptions{})
+			assert.NoError(t, err)
+			assert.True(t, wasWritten)
+			ctx.BlockingClose()
+		}()
+	}
+
+	go func() {
+		defer doneFn()
+		<-barrier
+		err := shard.Bootstrap(bootstrapResult)
+		assert.NoError(t, err)
+	}()
+
+	go func() {
+		defer doneFn()
+		<-barrier
+		_, err := shard.Tick(context.NewNoOpCanncellable(), time.Now())
+		assert.NoError(t, err)
 	}()
 
 	for i := 0; i < numRoutines; i++ {

src/dbnode/storage/shard_test.go

@@ -404,8 +404,14 @@ func addTestSeries(shard *dbShard, id ident.ID) series.DatabaseSeries {
 }
 
 func addTestSeriesWithCount(shard *dbShard, id ident.ID, count int32) series.DatabaseSeries {
+	return addTestSeriesWithCountAndBootstrap(shard, id, count, true)
+}
+
+func addTestSeriesWithCountAndBootstrap(shard *dbShard, id ident.ID, count int32, bootstrap bool) series.DatabaseSeries {
 	series := series.NewDatabaseSeries(id, ident.Tags{}, shard.seriesOpts)
-	series.Bootstrap(nil)
+	if bootstrap {
+		series.Bootstrap(nil)
+	}
 	shard.Lock()
 	entry := lookup.NewEntry(series, 0)
 	for i := int32(0); i < count; i++ {