fix: Ensure successive WAL replays don't overwrite each other (#14848)

Author: benclive

pkg/ingester/recovery.go

@@ -196,9 +196,6 @@ func (r *ingesterRecoverer) Close() {
 			s.chunkMtx.Lock()
 			defer s.chunkMtx.Unlock()
 
-			// reset all the incrementing stream counters after a successful WAL replay.
-			s.resetCounter()
-
 			if len(s.chunks) == 0 {
 				inst.removeStream(s)
 				return nil

pkg/ingester/recovery_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/tsdb/chunks"
 	"github.com/prometheus/prometheus/tsdb/record"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/grafana/loki/v3/pkg/distributor/writefailures"
@@ -302,3 +303,42 @@ func TestSeriesRecoveryNoDuplicates(t *testing.T) {
 	}
 	require.Equal(t, expected, result.resps[0].Streams)
 }
+
+func TestRecoveryWritesContinuesEntryCountAfterWALReplay(t *testing.T) {
+	ingesterConfig := defaultIngesterTestConfig(t)
+	limits, err := validation.NewOverrides(defaultLimitsTestConfig(), nil)
+	require.NoError(t, err)
+
+	store := &mockStore{
+		chunks: map[string][]chunk.Chunk{},
+	}
+
+	readRingMock := mockReadRingWithOneActiveIngester()
+
+	i, err := New(ingesterConfig, client.Config{}, store, limits, loki_runtime.DefaultTenantConfigs(), nil, writefailures.Cfg{}, constants.Loki, log.NewNopLogger(), nil, readRingMock, nil)
+	require.NoError(t, err)
+
+	var (
+		users            = 10
+		streamsCt        = 1000
+		entriesPerStream = 50
+	)
+
+	reader, _ := buildMemoryReader(users, streamsCt, entriesPerStream, true)
+
+	recoverer := newIngesterRecoverer(i)
+
+	err = RecoverWAL(context.Background(), reader, recoverer)
+	require.NoError(t, err)
+
+	recoverer.Close()
+
+	// Check that the entry count continues counting from the last WAL entry to avoid overwriting existing entries in the WAL on future replays.
+	for _, inst := range i.getInstances() {
+		err := inst.forAllStreams(context.Background(), func(s *stream) error {
+			assert.Equal(t, int64(entriesPerStream), s.entryCt)
+			return nil
+		})
+		require.NoError(t, err)
+	}
+}

pkg/ingester/stream.go

@@ -649,10 +649,6 @@ func (s *stream) addTailer(t *tailer) {
 	s.tailers[t.getID()] = t
 }
 
-func (s *stream) resetCounter() {
-	s.entryCt = 0
-}
-
 func headBlockType(chunkfmt byte, unorderedWrites bool) chunkenc.HeadBlockFmt {
 	if unorderedWrites {
 		if chunkfmt >= chunkenc.ChunkFormatV3 {

pkg/ingester/wal.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/go-kit/log/level"
+	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/prometheus/tsdb/wlog"
 
@@ -108,7 +109,7 @@ func (w *walWrapper) Log(record *wal.Record) error {
 	}
 	select {
 	case <-w.quit:
-		return nil
+		return errors.New("wal is stopped")
 	default:
 		buf := recordPool.GetBytes()
 		defer func() {