Check entry empty state to ensure GC eligible

src/dbnode/integration/index_active_block_rotate_test.go

@@ -51,7 +51,7 @@ func TestIndexActiveBlockRotate(t *testing.T) {
 		numWrites       = 50
 		numTags         = 10
 		blockSize       = 2 * time.Hour
-		indexBlockSize  = blockSize
+		indexBlockSize  = blockSize * 2
 		retentionPeriod = 12 * blockSize
 		bufferPast      = 10 * time.Minute
 		rOpts           = retention.NewOptions().

src/dbnode/storage/flush.go

@@ -54,6 +54,20 @@ const (
 	flushManagerIndexFlushInProgress
 )
 
+type namespaceFlushes map[string]namespaceFlush
+
+type shardFlushes map[shardFlush]bool
+
+type namespaceFlush struct {
+	namespace    databaseNamespace
+	shardFlushes map[shardFlush]bool
+}
+
+type shardFlush struct {
+	time  xtime.UnixNano
+	shard databaseShard
+}
+
 type flushManagerMetrics struct {
 	isFlushing      tally.Gauge
 	isSnapshotting  tally.Gauge
@@ -144,7 +158,8 @@ func (m *flushManager) Flush(startTime xtime.UnixNano) error {
 	// will attempt to snapshot blocks w/ unflushed data which would be wasteful if
 	// the block is already flushable.
 	multiErr := xerrors.NewMultiError()
-	if err = m.dataWarmFlush(namespaces, startTime); err != nil {
+	dataFlushes, err := m.dataWarmFlush(namespaces, startTime)
+	if err != nil {
 		multiErr = multiErr.Add(err)
 	}
 
@@ -159,26 +174,69 @@ func (m *flushManager) Flush(startTime xtime.UnixNano) error {
 		multiErr = multiErr.Add(fmt.Errorf("error rotating commitlog in mediator tick: %v", err))
 	}
 
-	if err = m.indexFlush(namespaces); err != nil {
+	indexFlushes, err := m.indexFlush(namespaces)
+	if err != nil {
 		multiErr = multiErr.Add(err)
 	}
 
-	return multiErr.FinalError()
+	err = multiErr.FinalError()
+
+	// Mark all flushed shards as such.
+	// If index is not enabled, then a shard+blockStart is "flushed" if the data has been flushed.
+	// If index is enabled, then a shard+blockStart is "flushed" if the data AND index has been flushed.
+	for _, n := range namespaces {
+		var (
+			indexEnabled  = n.Options().IndexOptions().Enabled()
+			flushedShards map[shardFlush]bool
+		)
+		if indexEnabled {
+			flushesForNs, ok := indexFlushes[n.ID().String()]
+			if !ok {
+				continue
+			}
+			flushedShards = flushesForNs.shardFlushes
+		} else {
+			flushesForNs, ok := dataFlushes[n.ID().String()]
+			if !ok {
+				continue
+			}
+			flushedShards = flushesForNs.shardFlushes
+		}
+
+		for s := range flushedShards {
+			// Block sizes for data and index can differ and so if we are driving the flushing by
+			// the index blockStarts, we must expand them to mark all containing data blockStarts.
+			// E.g. if blockSize == 2h and indexBlockSize == 4h and the flushed index time is 6:00pm,
+			// we should mark as flushed [6:00pm, 8:00pm].
+			if indexEnabled {
+				blockSize := n.Options().RetentionOptions().BlockSize()
+				indexBlockSize := n.Options().IndexOptions().BlockSize()
+				for start := s.time; start < s.time.Add(indexBlockSize); start = start.Add(blockSize) {
+					s.shard.MarkWarmFlushStateSuccessOrError(start, err)
+				}
+			} else {
+				s.shard.MarkWarmFlushStateSuccessOrError(s.time, err)
+			}
+		}
+	}
+
+	return err
 }
 
 func (m *flushManager) dataWarmFlush(
 	namespaces []databaseNamespace,
 	startTime xtime.UnixNano,
-) error {
+) (namespaceFlushes, error) {
 	flushPersist, err := m.pm.StartFlushPersist()
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	m.setState(flushManagerFlushInProgress)
 	var (
-		start    = m.nowFn()
-		multiErr = xerrors.NewMultiError()
+		start      = m.nowFn()
+		multiErr   = xerrors.NewMultiError()
+		allFlushes = make(map[string]namespaceFlush)
 	)
 	for _, ns := range namespaces {
 		// Flush first because we will only snapshot if there are no outstanding flushes.
@@ -187,10 +245,11 @@ func (m *flushManager) dataWarmFlush(
 			multiErr = multiErr.Add(err)
 			continue
 		}
-		err = m.flushNamespaceWithTimes(ns, flushTimes, flushPersist)
+		flush, err := m.flushNamespaceWithTimes(ns, flushTimes, flushPersist)
 		if err != nil {
 			multiErr = multiErr.Add(err)
 		}
+		allFlushes[ns.ID().String()] = flush
 	}
 
 	err = flushPersist.DoneFlush()
@@ -199,7 +258,7 @@ func (m *flushManager) dataWarmFlush(
 	}
 
 	m.metrics.dataWarmFlushDuration.Record(m.nowFn().Sub(start))
-	return multiErr.FinalError()
+	return allFlushes, multiErr.FinalError()
 }
 
 func (m *flushManager) dataSnapshot(
@@ -253,16 +312,17 @@ func (m *flushManager) dataSnapshot(
 
 func (m *flushManager) indexFlush(
 	namespaces []databaseNamespace,
-) error {
+) (namespaceFlushes, error) {
 	indexFlush, err := m.pm.StartIndexPersist()
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	m.setState(flushManagerIndexFlushInProgress)
 	var (
-		start    = m.nowFn()
-		multiErr = xerrors.NewMultiError()
+		start            = m.nowFn()
+		multiErr         = xerrors.NewMultiError()
+		namespaceFlushes = make(map[string]namespaceFlush)
 	)
 	for _, ns := range namespaces {
 		var (
@@ -272,12 +332,21 @@ func (m *flushManager) indexFlush(
 		if !indexEnabled {
 			continue
 		}
-		multiErr = multiErr.Add(ns.FlushIndex(indexFlush))
+
+		flushes, err := ns.FlushIndex(indexFlush)
+		if err != nil {
+			multiErr = multiErr.Add(err)
+		} else {
+			namespaceFlushes[ns.ID().String()] = namespaceFlush{
+				namespace:    ns,
+				shardFlushes: flushes,
+			}
+		}
 	}
 	multiErr = multiErr.Add(indexFlush.DoneIndex())
 
 	m.metrics.indexFlushDuration.Record(m.nowFn().Sub(start))
-	return multiErr.FinalError()
+	return namespaceFlushes, multiErr.FinalError()
 }
 
 func (m *flushManager) Report() {
@@ -361,18 +430,30 @@ func (m *flushManager) flushNamespaceWithTimes(
 	ns databaseNamespace,
 	times []xtime.UnixNano,
 	flushPreparer persist.FlushPreparer,
-) error {
+) (namespaceFlush, error) {
+	flushes := make(map[shardFlush]bool, 0)
 	multiErr := xerrors.NewMultiError()
 	for _, t := range times {
 		// NB(xichen): we still want to proceed if a namespace fails to flush its data.
 		// Probably want to emit a counter here, but for now just log it.
-		if err := ns.WarmFlush(t, flushPreparer); err != nil {
+		shards, err := ns.WarmFlush(t, flushPreparer)
+		if err != nil {
 			detailedErr := fmt.Errorf("namespace %s failed to flush data: %v",
 				ns.ID().String(), err)
 			multiErr = multiErr.Add(detailedErr)
+		} else {
+			for _, s := range shards {
+				flushes[shardFlush{
+					shard: s,
+					time:  t,
+				}] = true
+			}
 		}
 	}
-	return multiErr.FinalError()
+	return namespaceFlush{
+		namespace:    ns,
+		shardFlushes: flushes,
+	}, multiErr.FinalError()
 }
 
 func (m *flushManager) LastSuccessfulSnapshotStartTime() (xtime.UnixNano, bool) {

src/dbnode/storage/flush_test.go

@@ -315,12 +315,16 @@ func TestFlushManagerSkipNamespaceIndexingDisabled(t *testing.T) {
 	defer ctrl.Finish()
 
 	nsOpts := defaultTestNs1Opts.SetIndexOptions(namespace.NewIndexOptions().SetEnabled(false))
+	s1 := NewMockdatabaseShard(ctrl)
+	s2 := NewMockdatabaseShard(ctrl)
 	ns := NewMockdatabaseNamespace(ctrl)
 	ns.EXPECT().Options().Return(nsOpts).AnyTimes()
 	ns.EXPECT().ID().Return(defaultTestNs1ID).AnyTimes()
 	ns.EXPECT().NeedsFlush(gomock.Any(), gomock.Any()).Return(true, nil).AnyTimes()
-	ns.EXPECT().WarmFlush(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	ns.EXPECT().WarmFlush(gomock.Any(), gomock.Any()).Return([]databaseShard{s1, s2}, nil).AnyTimes()
 	ns.EXPECT().Snapshot(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	s1.EXPECT().MarkWarmFlushStateSuccessOrError(gomock.Any(), nil).AnyTimes()
+	s2.EXPECT().MarkWarmFlushStateSuccessOrError(gomock.Any(), nil).AnyTimes()
 
 	var (
 		mockFlushPersist    = persist.NewMockFlushPreparer(ctrl)
@@ -357,14 +361,35 @@ func TestFlushManagerNamespaceIndexingEnabled(t *testing.T) {
 	ctrl := xtest.NewController(t)
 	defer ctrl.Finish()
 
+	blocks := 24
 	nsOpts := defaultTestNs1Opts.SetIndexOptions(namespace.NewIndexOptions().SetEnabled(true))
+	s1 := NewMockdatabaseShard(ctrl)
+	s2 := NewMockdatabaseShard(ctrl)
 	ns := NewMockdatabaseNamespace(ctrl)
 	ns.EXPECT().Options().Return(nsOpts).AnyTimes()
 	ns.EXPECT().ID().Return(defaultTestNs1ID).AnyTimes()
 	ns.EXPECT().NeedsFlush(gomock.Any(), gomock.Any()).Return(true, nil).AnyTimes()
-	ns.EXPECT().WarmFlush(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
-	ns.EXPECT().Snapshot(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
-	ns.EXPECT().FlushIndex(gomock.Any()).Return(nil)
+
+	// Validate that the flush state is marked as successful only AFTER all prequisite steps have been run.
+	// Order is important to avoid any edge case where data is GCed from memory without all flushing operations
+	// being completed.
+	mockFlushedShards := map[shardFlush]bool{
+		shardFlush{shard: s1, time: xtime.Now().Add(time.Minute * 1)}: true,
+		shardFlush{shard: s1, time: xtime.Now().Add(time.Minute * 2)}: true,
+		shardFlush{shard: s2, time: xtime.Now().Add(time.Minute * 1)}: true,
+		shardFlush{shard: s2, time: xtime.Now().Add(time.Minute * 2)}: true,
+	}
+	steps := make([]*gomock.Call, 0)
+	steps = append(steps,
+		ns.EXPECT().WarmFlush(gomock.Any(), gomock.Any()).Return([]databaseShard{s1, s2}, nil).Times(blocks),
+		ns.EXPECT().Snapshot(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes(),
+		ns.EXPECT().FlushIndex(gomock.Any()).Return(mockFlushedShards, nil),
+		s1.EXPECT().MarkWarmFlushStateSuccessOrError(gomock.Any(), nil),
+		s1.EXPECT().MarkWarmFlushStateSuccessOrError(gomock.Any(), nil),
+		s2.EXPECT().MarkWarmFlushStateSuccessOrError(gomock.Any(), nil),
+		s2.EXPECT().MarkWarmFlushStateSuccessOrError(gomock.Any(), nil),
+	)
+	gomock.InOrder(steps...)
 
 	var (
 		mockFlushPersist    = persist.NewMockFlushPreparer(ctrl)