From 072a3c2e177631ab4bb351ba00d4528f60f3d055 Mon Sep 17 00:00:00 2001 From: Nate Broyles Date: Fri, 5 Nov 2021 15:11:23 -0400 Subject: [PATCH] [dbnode] Gracefully handle reads including documents with stale index state In db nodes, state is kept for cheap lookups to determine if a series is indexed for a specific block start. This state exists on the document in the index and in the shard object. On concurrent writes to the same time series, it's possible for these two sources to get out of sync. Basically, the state associated with the document on the index is a different instance than the one that gets placed in the shard's lookup map. This can result in issues where we don't accurately track that a series in indexed for a specific block start as only the entry in the lookup map will get updated in the future. To resolve this, we introduce reconciliation methods when querying these objects. If it's noticed on read that we're querying index state that never got inserted into the shard lookup map, then we query the shard for the appropriate entry and use the results on that object. --- .../index_block_orphaned_entry_test.go | 207 ++++++++++++++++++ src/dbnode/integration/index_helpers.go | 23 +- src/dbnode/storage/entry.go | 40 +++- src/dbnode/storage/entry_blackbox_test.go | 16 +- src/dbnode/storage/index/block.go | 4 +- src/dbnode/storage/index/block_bench_test.go | 6 + src/dbnode/storage/index/block_test.go | 26 +-- .../storage/index_query_concurrent_test.go | 4 +- .../storage/index_queue_forward_write_test.go | 8 +- src/dbnode/storage/index_queue_test.go | 4 +- src/m3ninx/doc/doc_mock.go | 44 ++-- src/m3ninx/doc/types.go | 11 +- 12 files changed, 343 insertions(+), 50 deletions(-) create mode 100644 src/dbnode/integration/index_block_orphaned_entry_test.go diff --git a/src/dbnode/integration/index_block_orphaned_entry_test.go b/src/dbnode/integration/index_block_orphaned_entry_test.go new file mode 100644 index 0000000000..f19e93d9c2 --- /dev/null +++ b/src/dbnode/integration/index_block_orphaned_entry_test.go @@ -0,0 +1,207 @@ +// +build integration +// +// Copyright (c) 2021 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package integration + +import ( + "fmt" + "math/rand" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/m3db/m3/src/dbnode/client" + "github.com/m3db/m3/src/dbnode/namespace" + "github.com/m3db/m3/src/dbnode/persist/fs" + "github.com/m3db/m3/src/dbnode/storage" + "github.com/m3db/m3/src/dbnode/storage/index/compaction" + xclock "github.com/m3db/m3/src/x/clock" + "github.com/m3db/m3/src/x/ident" + xsync "github.com/m3db/m3/src/x/sync" + xtime "github.com/m3db/m3/src/x/time" +) + +const ( + numTestSeries = 5 + concurrentWorkers = 25 + writesPerWorker = 5 + blockSize = 2 * time.Hour +) + +func TestIndexBlockOrphanedEntry(t *testing.T) { + setup := generateTestSetup(t) + defer setup.Close() + + // Start the server + log := setup.StorageOpts().InstrumentOptions().Logger() + require.NoError(t, setup.StartServer()) + + // Stop the server + defer func() { + assert.NoError(t, setup.StopServer()) + log.Debug("server is now down") + }() + + client := setup.M3DBClient() + session, err := client.DefaultSession() + require.NoError(t, err) + + // Write concurrent metrics to generate multiple entries for the same series + ids := make([]ident.ID, 0, numTestSeries) + for i := 0; i < numTestSeries; i++ { + fooID := ident.StringID(fmt.Sprintf("foo.%v", i)) + ids = append(ids, fooID) + + writeConcurrentMetrics(t, setup, session, fooID) + } + + // Write metrics for a different series to push current foreground segment + // to the background. After this, all documents for foo.X exist in background segments + barID := ident.StringID("bar") + writeConcurrentMetrics(t, setup, session, barID) + + // Fast-forward to a block rotation + newBlock := xtime.Now().Truncate(blockSize).Add(blockSize) + newCurrentTime := newBlock.Add(30 * time.Minute) // Add extra to account for buffer past + setup.SetNowFn(newCurrentTime) + + // Wait for flush + log.Info("waiting for block rotation to complete") + nsID := setup.Namespaces()[0].ID() + found := xclock.WaitUntil(func() bool { + filesets, err := fs.IndexFileSetsAt(setup.FilePathPrefix(), nsID, newBlock.Add(-blockSize)) + require.NoError(t, err) + return len(filesets) == 1 + }, 30*time.Second) + require.True(t, found) + + // Do post-block rotation writes + for _, id := range ids { + writeMetric(t, session, nsID, id, newCurrentTime, 999.0) + } + writeMetric(t, session, nsID, barID, newCurrentTime, 999.0) + + // Foreground segments should be in the background again which means updated index entry + // is now behind the orphaned entry so index reads should fail. + log.Info("waiting for metrics to be indexed") + var ( + missing string + ok bool + ) + found = xclock.WaitUntil(func() bool { + for _, id := range ids { + ok, err = isIndexedCheckedWithTime( + t, session, nsID, id, genTags(id), newCurrentTime, + ) + if !ok { + missing = id.String() + return false + } + } + return true + }, 30*time.Second) + assert.True(t, found, fmt.Sprintf("series %s never indexed\n", missing)) + assert.NoError(t, err) +} + +func writeConcurrentMetrics( + t *testing.T, + setup TestSetup, + session client.Session, + seriesID ident.ID, +) { + var wg sync.WaitGroup + nowFn := setup.DB().Options().ClockOptions().NowFn() + + workerPool := xsync.NewWorkerPool(concurrentWorkers) + workerPool.Init() + + mdID := setup.Namespaces()[0].ID() + for i := 0; i < concurrentWorkers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + + for j := 0; j < writesPerWorker; j++ { + j := j + wg.Add(1) + workerPool.Go(func() { + defer wg.Done() + writeMetric(t, session, mdID, seriesID, xtime.ToUnixNano(nowFn()), float64(j)) + }) + } + }() + } + + wg.Wait() +} + +func genTags(seriesID ident.ID) ident.TagsIterator { + return ident.NewTagsIterator(ident.NewTags(ident.StringTag("tagName", seriesID.String()))) +} + +func writeMetric( + t *testing.T, + session client.Session, + nsID ident.ID, + seriesID ident.ID, + timestamp xtime.UnixNano, + value float64, +) { + err := session.WriteTagged(nsID, seriesID, genTags(seriesID), + timestamp, value, xtime.Second, nil) + require.NoError(t, err) +} + +func generateTestSetup(t *testing.T) TestSetup { + md, err := namespace.NewMetadata(testNamespaces[0], + namespace.NewOptions(). + SetRetentionOptions(DefaultIntegrationTestRetentionOpts). + SetIndexOptions(namespace.NewIndexOptions().SetEnabled(true))) + require.NoError(t, err) + + testOpts := NewTestOptions(t). + SetNamespaces([]namespace.Metadata{md}). + SetWriteNewSeriesAsync(true) + testSetup, err := NewTestSetup(t, testOpts, nil, + func(s storage.Options) storage.Options { + s = s.SetCoreFn(func() int { + return rand.Intn(4) //nolint:gosec + }) + compactionOpts := s.IndexOptions().ForegroundCompactionPlannerOptions() + compactionOpts.Levels = []compaction.Level{ + { + MinSizeInclusive: 0, + MaxSizeExclusive: 1, + }, + } + return s.SetIndexOptions( + s.IndexOptions(). + SetForegroundCompactionPlannerOptions(compactionOpts)) + }) + require.NoError(t, err) + + return testSetup +} diff --git a/src/dbnode/integration/index_helpers.go b/src/dbnode/integration/index_helpers.go index e3debcf00e..d3c8796a85 100644 --- a/src/dbnode/integration/index_helpers.go +++ b/src/dbnode/integration/index_helpers.go @@ -291,13 +291,30 @@ func isIndexed(t *testing.T, s client.Session, ns ident.ID, id ident.ID, tags id return result } -func isIndexedChecked(t *testing.T, s client.Session, ns ident.ID, id ident.ID, tags ident.TagIterator) (bool, error) { +func isIndexedChecked( + t *testing.T, + s client.Session, + ns ident.ID, + id ident.ID, + tags ident.TagIterator, +) (bool, error) { + return isIndexedCheckedWithTime(t, s, ns, id, tags, xtime.Now()) +} + +func isIndexedCheckedWithTime( + t *testing.T, + s client.Session, + ns ident.ID, + id ident.ID, + tags ident.TagIterator, + queryTime xtime.UnixNano, +) (bool, error) { q := newQuery(t, tags) iter, _, err := s.FetchTaggedIDs(ContextWithDefaultTimeout(), ns, index.Query{Query: q}, index.QueryOptions{ - StartInclusive: xtime.Now(), - EndExclusive: xtime.Now(), + StartInclusive: queryTime, + EndExclusive: queryTime.Add(time.Nanosecond), SeriesLimit: 10, }) if err != nil { diff --git a/src/dbnode/storage/entry.go b/src/dbnode/storage/entry.go index 78bae50c56..a399f5922a 100644 --- a/src/dbnode/storage/entry.go +++ b/src/dbnode/storage/entry.go @@ -141,13 +141,43 @@ func (entry *Entry) IndexedForBlockStart(indexBlockStart xtime.UnixNano) bool { return isIndexed } -// IndexedRange returns minimum and maximum blockStart values covered by index entry. +// ReconciledIndexedForBlockStart returns true if the blockStart has been indexed. +// Attempts to retrieve the most recent index entry from the shard if this entry +// was never inserted there. If there is an error, returns the value for this entry. +func (entry *Entry) ReconciledIndexedForBlockStart(indexBlockStart xtime.UnixNano) bool { + if entry.insertTime.Load() > 0 { + return entry.IndexedForBlockStart(indexBlockStart) + } + e, _, err := entry.Shard.TryRetrieveSeriesAndIncrementReaderWriterCount(entry.ID) + if err != nil || e == nil { + return entry.IndexedForBlockStart(indexBlockStart) + } + defer e.DecrementReaderWriterCount() + + return e.IndexedForBlockStart(indexBlockStart) +} + +// ReconciledIndexedRange returns minimum and maximum blockStart values covered by index entry. // The range is inclusive. Note that there may be uncovered gaps within the range. +// Attempts to retrieve the most recent index entry from the shard if this entry +// was never inserted there. If there is an error, returns the range on this entry. // Returns (0, 0) for an empty range. -func (entry *Entry) IndexedRange() (xtime.UnixNano, xtime.UnixNano) { - entry.reverseIndex.RLock() - min, max := entry.reverseIndex.indexedRangeWithRLock() - entry.reverseIndex.RUnlock() +func (entry *Entry) ReconciledIndexedRange() (xtime.UnixNano, xtime.UnixNano) { + var min, max xtime.UnixNano + if entry.insertTime.Load() > 0 { + min, max = entry.reverseIndex.indexedRangeWithRLock() + } else { + e, _, err := entry.Shard.TryRetrieveSeriesAndIncrementReaderWriterCount(entry.ID) + if err != nil || e == nil { + return entry.reverseIndex.indexedRangeWithRLock() + } + defer e.DecrementReaderWriterCount() + + e.reverseIndex.RLock() + min, max = e.reverseIndex.indexedRangeWithRLock() + e.reverseIndex.RUnlock() + } + return min, max } diff --git a/src/dbnode/storage/entry_blackbox_test.go b/src/dbnode/storage/entry_blackbox_test.go index f48b9a2401..20f32a3535 100644 --- a/src/dbnode/storage/entry_blackbox_test.go +++ b/src/dbnode/storage/entry_blackbox_test.go @@ -171,10 +171,22 @@ func TestEntryIndexedRange(t *testing.T) { ctrl := xtest.NewController(t) defer ctrl.Finish() - entry := NewEntry(NewEntryOptions{Series: newMockSeries(ctrl)}) + opts := DefaultTestOptions() + ctx := opts.ContextPool().Get() + defer ctx.Close() + + shard := testDatabaseShard(t, opts) + defer func() { + require.NoError(t, shard.Close()) + }() + + entry := NewEntry(NewEntryOptions{ + Shard: shard, + Series: newMockSeries(ctrl), + }) assertRange := func(expectedMin, expectedMax xtime.UnixNano) { - min, max := entry.IndexedRange() + min, max := entry.ReconciledIndexedRange() assert.Equal(t, expectedMin, min) assert.Equal(t, expectedMax, max) } diff --git a/src/dbnode/storage/index/block.go b/src/dbnode/storage/index/block.go index 07fa7ac963..1b5664faa9 100644 --- a/src/dbnode/storage/index/block.go +++ b/src/dbnode/storage/index/block.go @@ -547,7 +547,7 @@ func (b *block) queryWithSpan( inBlock bool currentBlock = opts.StartInclusive.Truncate(b.blockSize) endExclusive = opts.EndExclusive - minIndexed, maxIndexed = md.OnIndexSeries.IndexedRange() + minIndexed, maxIndexed = md.OnIndexSeries.ReconciledIndexedRange() ) if maxIndexed == 0 { @@ -566,7 +566,7 @@ func (b *block) queryWithSpan( } for !inBlock && currentBlock.Before(endExclusive) { - inBlock = md.OnIndexSeries.IndexedForBlockStart(currentBlock) + inBlock = md.OnIndexSeries.ReconciledIndexedForBlockStart(currentBlock) currentBlock = currentBlock.Add(b.blockSize) } diff --git a/src/dbnode/storage/index/block_bench_test.go b/src/dbnode/storage/index/block_bench_test.go index 5fd6790ca9..3a6bcbb2ca 100644 --- a/src/dbnode/storage/index/block_bench_test.go +++ b/src/dbnode/storage/index/block_bench_test.go @@ -134,3 +134,9 @@ func (m mockOnIndexSeries) NeedsIndexGarbageCollected() bool { return func (m mockOnIndexSeries) IndexedRange() (xtime.UnixNano, xtime.UnixNano) { return 0, 0 } + +func (m mockOnIndexSeries) ReconciledIndexedRange() (xtime.UnixNano, xtime.UnixNano) { + return 0, 0 +} + +func (m mockOnIndexSeries) ReconciledIndexedForBlockStart(_ xtime.UnixNano) bool { return false } diff --git a/src/dbnode/storage/index/block_test.go b/src/dbnode/storage/index/block_test.go index 81d762b96f..e87a0596bd 100644 --- a/src/dbnode/storage/index/block_test.go +++ b/src/dbnode/storage/index/block_test.go @@ -1500,8 +1500,8 @@ func TestBlockE2EInsertQueryLimit(t *testing.T) { h1 := doc.NewMockOnIndexSeries(ctrl) h1.EXPECT().OnIndexFinalize(blockStart) h1.EXPECT().OnIndexSuccess(blockStart) - h1.EXPECT().IndexedRange().Return(blockStart, blockStart) - h1.EXPECT().IndexedForBlockStart(blockStart).Return(true) + h1.EXPECT().ReconciledIndexedRange().Return(blockStart, blockStart) + h1.EXPECT().ReconciledIndexedForBlockStart(blockStart).Return(true) h2 := doc.NewMockOnIndexSeries(ctrl) h2.EXPECT().OnIndexFinalize(blockStart) @@ -1588,14 +1588,14 @@ func TestBlockE2EInsertAddResultsQuery(t *testing.T) { h1 := doc.NewMockOnIndexSeries(ctrl) h1.EXPECT().OnIndexFinalize(blockStart) h1.EXPECT().OnIndexSuccess(blockStart) - h1.EXPECT().IndexedRange().Return(blockStart, blockStart) - h1.EXPECT().IndexedForBlockStart(blockStart).Return(true) + h1.EXPECT().ReconciledIndexedRange().Return(blockStart, blockStart) + h1.EXPECT().ReconciledIndexedForBlockStart(blockStart).Return(true) h2 := doc.NewMockOnIndexSeries(ctrl) h2.EXPECT().OnIndexFinalize(blockStart) h2.EXPECT().OnIndexSuccess(blockStart) - h2.EXPECT().IndexedRange().Return(blockStart, blockStart) - h2.EXPECT().IndexedForBlockStart(blockStart).Return(true) + h2.EXPECT().ReconciledIndexedRange().Return(blockStart, blockStart) + h2.EXPECT().ReconciledIndexedForBlockStart(blockStart).Return(true) batch := NewWriteBatch(WriteBatchOptions{ IndexBlockSize: blockSize, @@ -1688,8 +1688,8 @@ func TestBlockE2EInsertAddResultsMergeQuery(t *testing.T) { h1 := doc.NewMockOnIndexSeries(ctrl) h1.EXPECT().OnIndexFinalize(blockStart) h1.EXPECT().OnIndexSuccess(blockStart) - h1.EXPECT().IndexedRange().Return(blockStart, blockStart) - h1.EXPECT().IndexedForBlockStart(blockStart).Return(true) + h1.EXPECT().ReconciledIndexedRange().Return(blockStart, blockStart) + h1.EXPECT().ReconciledIndexedForBlockStart(blockStart).Return(true) batch := NewWriteBatch(WriteBatchOptions{ IndexBlockSize: blockSize, @@ -1778,15 +1778,15 @@ func TestBlockE2EInsertAddResultsQueryNarrowingBlockRange(t *testing.T) { h1 := doc.NewMockOnIndexSeries(ctrl) h1.EXPECT().OnIndexFinalize(blockStart) h1.EXPECT().OnIndexSuccess(blockStart) - h1.EXPECT().IndexedRange().Return(blockStart, blockStart.Add(2*blockSize)) - h1.EXPECT().IndexedForBlockStart(blockStart).Return(false) - h1.EXPECT().IndexedForBlockStart(blockStart.Add(1 * blockSize)).Return(false) - h1.EXPECT().IndexedForBlockStart(blockStart.Add(2 * blockSize)).Return(true) + h1.EXPECT().ReconciledIndexedRange().Return(blockStart, blockStart.Add(2*blockSize)) + h1.EXPECT().ReconciledIndexedForBlockStart(blockStart).Return(false) + h1.EXPECT().ReconciledIndexedForBlockStart(blockStart.Add(1 * blockSize)).Return(false) + h1.EXPECT().ReconciledIndexedForBlockStart(blockStart.Add(2 * blockSize)).Return(true) h2 := doc.NewMockOnIndexSeries(ctrl) h2.EXPECT().OnIndexFinalize(blockStart) h2.EXPECT().OnIndexSuccess(blockStart) - h2.EXPECT().IndexedRange().Return(xtime.UnixNano(0), xtime.UnixNano(0)) + h2.EXPECT().ReconciledIndexedRange().Return(xtime.UnixNano(0), xtime.UnixNano(0)) batch := NewWriteBatch(WriteBatchOptions{ IndexBlockSize: blockSize, diff --git a/src/dbnode/storage/index_query_concurrent_test.go b/src/dbnode/storage/index_query_concurrent_test.go index 3729d7d9af..3f60dd823a 100644 --- a/src/dbnode/storage/index_query_concurrent_test.go +++ b/src/dbnode/storage/index_query_concurrent_test.go @@ -176,11 +176,11 @@ func testNamespaceIndexHighConcurrentQueries( IfAlreadyIndexedMarkIndexSuccessAndFinalize(gomock.Any()). Times(idsPerBlock) onIndexSeries.EXPECT(). - IndexedRange(). + ReconciledIndexedRange(). Return(min, max). AnyTimes() onIndexSeries.EXPECT(). - IndexedForBlockStart(gomock.Any()). + ReconciledIndexedForBlockStart(gomock.Any()). DoAndReturn(func(ts xtime.UnixNano) bool { return ts.Equal(st) }). diff --git a/src/dbnode/storage/index_queue_forward_write_test.go b/src/dbnode/storage/index_queue_forward_write_test.go index 82e15a510b..773f50dbb7 100644 --- a/src/dbnode/storage/index_queue_forward_write_test.go +++ b/src/dbnode/storage/index_queue_forward_write_test.go @@ -126,11 +126,11 @@ func setupForwardIndex( ) if !expectAggregateQuery { - lifecycle.EXPECT().IndexedRange().Return(ts, ts) - lifecycle.EXPECT().IndexedForBlockStart(ts).Return(true) + lifecycle.EXPECT().ReconciledIndexedRange().Return(ts, ts) + lifecycle.EXPECT().ReconciledIndexedForBlockStart(ts).Return(true) - lifecycle.EXPECT().IndexedRange().Return(next, next) - lifecycle.EXPECT().IndexedForBlockStart(next).Return(true) + lifecycle.EXPECT().ReconciledIndexedRange().Return(next, next) + lifecycle.EXPECT().ReconciledIndexedForBlockStart(next).Return(true) } entry, doc := testWriteBatchEntry(id, tags, now, lifecycle) diff --git a/src/dbnode/storage/index_queue_test.go b/src/dbnode/storage/index_queue_test.go index 28b67eb712..36ffe67df4 100644 --- a/src/dbnode/storage/index_queue_test.go +++ b/src/dbnode/storage/index_queue_test.go @@ -353,8 +353,8 @@ func setupIndex(t *testing.T, lifecycleFns.EXPECT().IfAlreadyIndexedMarkIndexSuccessAndFinalize(gomock.Any()).Return(false) if !expectAggregateQuery { - lifecycleFns.EXPECT().IndexedRange().Return(ts, ts) - lifecycleFns.EXPECT().IndexedForBlockStart(ts).Return(true) + lifecycleFns.EXPECT().ReconciledIndexedRange().Return(ts, ts) + lifecycleFns.EXPECT().ReconciledIndexedForBlockStart(ts).Return(true) } entry, doc := testWriteBatchEntry(id, tags, now, lifecycleFns) diff --git a/src/m3ninx/doc/doc_mock.go b/src/m3ninx/doc/doc_mock.go index 75cffd5945..fdb8f7ad3d 100644 --- a/src/m3ninx/doc/doc_mock.go +++ b/src/m3ninx/doc/doc_mock.go @@ -334,21 +334,6 @@ func (mr *MockOnIndexSeriesMockRecorder) IndexedForBlockStart(blockStart interfa return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IndexedForBlockStart", reflect.TypeOf((*MockOnIndexSeries)(nil).IndexedForBlockStart), blockStart) } -// IndexedRange mocks base method. -func (m *MockOnIndexSeries) IndexedRange() (time.UnixNano, time.UnixNano) { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "IndexedRange") - ret0, _ := ret[0].(time.UnixNano) - ret1, _ := ret[1].(time.UnixNano) - return ret0, ret1 -} - -// IndexedRange indicates an expected call of IndexedRange. -func (mr *MockOnIndexSeriesMockRecorder) IndexedRange() *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IndexedRange", reflect.TypeOf((*MockOnIndexSeries)(nil).IndexedRange)) -} - // NeedsIndexGarbageCollected mocks base method. func (m *MockOnIndexSeries) NeedsIndexGarbageCollected() bool { m.ctrl.T.Helper() @@ -413,6 +398,35 @@ func (mr *MockOnIndexSeriesMockRecorder) OnIndexSuccess(blockStart interface{}) return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnIndexSuccess", reflect.TypeOf((*MockOnIndexSeries)(nil).OnIndexSuccess), blockStart) } +// ReconciledIndexedForBlockStart mocks base method. +func (m *MockOnIndexSeries) ReconciledIndexedForBlockStart(blockStart time.UnixNano) bool { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ReconciledIndexedForBlockStart", blockStart) + ret0, _ := ret[0].(bool) + return ret0 +} + +// ReconciledIndexedForBlockStart indicates an expected call of ReconciledIndexedForBlockStart. +func (mr *MockOnIndexSeriesMockRecorder) ReconciledIndexedForBlockStart(blockStart interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReconciledIndexedForBlockStart", reflect.TypeOf((*MockOnIndexSeries)(nil).ReconciledIndexedForBlockStart), blockStart) +} + +// ReconciledIndexedRange mocks base method. +func (m *MockOnIndexSeries) ReconciledIndexedRange() (time.UnixNano, time.UnixNano) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ReconciledIndexedRange") + ret0, _ := ret[0].(time.UnixNano) + ret1, _ := ret[1].(time.UnixNano) + return ret0, ret1 +} + +// ReconciledIndexedRange indicates an expected call of ReconciledIndexedRange. +func (mr *MockOnIndexSeriesMockRecorder) ReconciledIndexedRange() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReconciledIndexedRange", reflect.TypeOf((*MockOnIndexSeries)(nil).ReconciledIndexedRange)) +} + // TryMarkIndexGarbageCollected mocks base method. func (m *MockOnIndexSeries) TryMarkIndexGarbageCollected() bool { m.ctrl.T.Helper() diff --git a/src/m3ninx/doc/types.go b/src/m3ninx/doc/types.go index 7782b21917..6e38449575 100644 --- a/src/m3ninx/doc/types.go +++ b/src/m3ninx/doc/types.go @@ -125,8 +125,15 @@ type OnIndexSeries interface { // IndexedForBlockStart returns true if the blockStart has been indexed. IndexedForBlockStart(blockStart xtime.UnixNano) bool - // IndexedRange returns minimum and maximum blockStart values covered by index entry. + // ReconciledIndexedForBlockStart returns true if the blockStart has been indexed. + // Attempts to retrieve the most recent index entry from the shard if this entry + // was never inserted there. If there is an error, returns the value for this entry. + ReconciledIndexedForBlockStart(blockStart xtime.UnixNano) bool + + // ReconciledIndexedRange returns minimum and maximum blockStart values covered by index entry. // The range is inclusive. Note that there may be uncovered gaps within the range. + // Attempts to retrieve the most recent index entry from the shard if this entry + // was never inserted there. If there is an error, returns the range on this entry. // Returns (0, 0) for an empty range. - IndexedRange() (xtime.UnixNano, xtime.UnixNano) + ReconciledIndexedRange() (xtime.UnixNano, xtime.UnixNano) }