From ab7fcd18b821499d4627423007fd5e9b2c17d37f Mon Sep 17 00:00:00 2001 From: Rohithmatham12 Date: Thu, 6 Aug 2026 20:54:24 -0700 Subject: [PATCH 1/2] Fix blocksToDelete shipper initialization race Signed-off-by: Rohithmatham12 --- pkg/ingester/ingester.go | 36 ++++++++++++++++++++---------------- 1 file changed, 20 insertions(+), 16 deletions(-) diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go index 3a6c03449c6..6813cea9e71 100644 --- a/pkg/ingester/ingester.go +++ b/pkg/ingester/ingester.go @@ -3113,26 +3113,14 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) { return nil, errors.Wrapf(err, "failed to compact TSDB: %s", udir) } - userDB.db = db - // We set the limiter here because we don't want to limit - // series during WAL replay. - userDB.limiter = i.limiter - - if db.Head().NumSeries() > 0 { - // If there are series in the head, use max time from head. If this time is too old, - // TSDB will be eligible for flushing and closing sooner, unless more data is pushed to it quickly. - userDB.setLastUpdate(util.TimeFromMillis(db.Head().MaxTime())) - } else { - // If head is empty (eg. new TSDB), don't close it right after. - userDB.setLastUpdate(time.Now()) - } - // Thanos shipper requires at least 1 external label to be set. For this reason, // we set the tenant ID as external label and we'll filter it out when reading // the series from the storage. l := labels.FromStrings(cortex_tsdb.TenantIDExternalLabel, userID, cortex_tsdb.IngesterIDExternalLabel, i.TSDBState.shipperIngesterID) - // Create a new shipper for this database + // Create a new shipper for this database before making the TSDB visible to + // callbacks. TSDB's reload goroutine can call blocksToDelete concurrently, + // and blocksToDelete reads userDB.shipper once userDB.db is set. if i.cfg.BlocksStorageConfig.TSDB.IsBlocksShippingEnabled() { udirRoot, err := os.OpenRoot(udir) if err != nil { @@ -3151,8 +3139,24 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) { shipper.WithAllowOutOfOrderUploads(true), // Allow out of order uploads. It's fine in Cortex's context. shipper.WithSkipCorruptedBlocks(true), // We allow out of order uploads. This is the same behavior. We should track error with metrics ) - userDB.shipperMetadataFilePath = filepath.Join(userDB.db.Dir(), filepath.Clean(shipper.DefaultMetaFilename)) + userDB.shipperMetadataFilePath = filepath.Join(db.Dir(), filepath.Clean(shipper.DefaultMetaFilename)) + } + + userDB.db = db + // We set the limiter here because we don't want to limit + // series during WAL replay. + userDB.limiter = i.limiter + + if db.Head().NumSeries() > 0 { + // If there are series in the head, use max time from head. If this time is too old, + // TSDB will be eligible for flushing and closing sooner, unless more data is pushed to it quickly. + userDB.setLastUpdate(util.TimeFromMillis(db.Head().MaxTime())) + } else { + // If head is empty (eg. new TSDB), don't close it right after. + userDB.setLastUpdate(time.Now()) + } + if userDB.shipper != nil { // Initialise the shipper blocks cache. if err := userDB.updateCachedShippedBlocks(); err != nil { level.Error(userLogger).Log("msg", "failed to update cached shipped blocks after shipper initialisation", "err", err) From 4294d28fbfc307a08e0ca04a29becfae77fc6779 Mon Sep 17 00:00:00 2001 From: Rohithmatham12 Date: Sun, 1 Nov 2026 21:17:56 -0800 Subject: [PATCH 2/2] Synchronize blocksToDelete publication --- pkg/ingester/ingester.go | 4 +++- pkg/ingester/ingester_test.go | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go index 6813cea9e71..2fc2ea4a03e 100644 --- a/pkg/ingester/ingester.go +++ b/pkg/ingester/ingester.go @@ -377,6 +377,7 @@ func (r tsdbCloseCheckResult) shouldClose() bool { type userTSDB struct { db *tsdb.DB + blocksToDeleteReady atomic.Bool userID string logger log.Logger activeSeries *ActiveSeries @@ -600,7 +601,7 @@ func (u *userTSDB) PostDeletion(metrics map[chunks.HeadSeriesRef]labels.Labels) // blocksToDelete filters the input blocks and returns the blocks which are safe to be deleted from the ingester. func (u *userTSDB) blocksToDelete(blocks []*tsdb.Block) map[ulid.ULID]struct{} { - if u.db == nil { + if !u.blocksToDeleteReady.Load() { return nil } deletable := tsdb.DefaultBlocksToDelete(u.db)(blocks) @@ -3143,6 +3144,7 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) { } userDB.db = db + userDB.blocksToDeleteReady.Store(true) // We set the limiter here because we don't want to limit // series during WAL replay. userDB.limiter = i.limiter diff --git a/pkg/ingester/ingester_test.go b/pkg/ingester/ingester_test.go index c56e79d25b8..dcd6d64a670 100644 --- a/pkg/ingester/ingester_test.go +++ b/pkg/ingester/ingester_test.go @@ -7525,6 +7525,7 @@ func TestIngester_UserTSDB_BlocksToDelete(t *testing.T) { shippedBlocks: shippedBlocks, blockRetentionPeriod: 2 * time.Hour.Milliseconds(), } + userDB.blocksToDeleteReady.Store(true) blocksToDelete := userDB.blocksToDelete(blocks) require.Equal(t, 4, len(blocksToDelete)) @@ -7557,6 +7558,7 @@ func TestIngester_UserTSDB_BlocksToDelete(t *testing.T) { shippedBlocks: shippedBlocks, blockRetentionPeriod: 2 * time.Hour.Milliseconds(), } + userDB.blocksToDeleteReady.Store(true) blocksToDelete := userDB.blocksToDelete(blocks) require.Equal(t, 3, len(blocksToDelete)) @@ -7585,6 +7587,7 @@ func TestIngester_UserTSDB_BlocksToDelete(t *testing.T) { shippedBlocks: shippedBlocks, blockRetentionPeriod: 2 * time.Hour.Milliseconds(), } + userDB.blocksToDeleteReady.Store(true) blocksToDelete := userDB.blocksToDelete(blocks) require.Equal(t, 0, len(blocksToDelete)) @@ -7617,6 +7620,7 @@ func TestIngester_UserTSDB_BlocksToDelete(t *testing.T) { shippedBlocks: shippedBlocks, blockRetentionPeriod: 2 * time.Hour.Milliseconds(), } + userDB.blocksToDeleteReady.Store(true) blocksToDelete := userDB.blocksToDelete(blocks) require.Equal(t, 3, len(blocksToDelete)) @@ -7650,6 +7654,7 @@ func TestIngester_UserTSDB_BlocksToDelete(t *testing.T) { shippedBlocks: shippedBlocks, blockRetentionPeriod: 2 * time.Hour.Milliseconds(), } + userDB.blocksToDeleteReady.Store(true) blocksToDelete := userDB.blocksToDelete(blocks) require.Equal(t, 3, len(blocksToDelete)) @@ -7674,6 +7679,7 @@ func TestIngester_UserTSDB_BlocksToDelete(t *testing.T) { db: db, blockRetentionPeriod: 2 * time.Hour.Milliseconds(), } + userDB.blocksToDeleteReady.Store(true) blocksToDelete := userDB.blocksToDelete(blocks) require.Equal(t, 4, len(blocksToDelete)) @@ -7684,6 +7690,35 @@ func TestIngester_UserTSDB_BlocksToDelete(t *testing.T) { }) } +func TestIngester_UserTSDB_BlocksToDeletePublication(t *testing.T) { + db, err := tsdb.Open(t.TempDir(), promslog.NewNopLogger(), prometheus.NewPedanticRegistry(), &tsdb.Options{}, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, db.Close()) }) + + userDB := &userTSDB{} + stop := make(chan struct{}) + done := make(chan struct{}) + + go func() { + defer close(done) + for { + select { + case <-stop: + return + default: + userDB.blocksToDelete(nil) + } + } + }() + + userDB.db = db + userDB.shipper = &shipperMock{} + userDB.blocksToDeleteReady.Store(true) + + close(stop) + <-done +} + func TestIngester_UpdateLabelSetMetrics(t *testing.T) { cfg := defaultIngesterTestConfig(t) cfg.BlocksStorageConfig.TSDB.BlockRanges = []time.Duration{2 * time.Hour}