fix: use flusher and recovery checkpoint together to determine the truncate position (#41934)

issue: #41544

- unify the log field of message
- use the minimum one of flusher and recovery storage checkpoint as the
truncate position

Signed-off-by: chyezh <[email protected]>

Author: chyezh

internal/mocks/streamingnode/server/wal/mock_recovery/mock_RecoveryStorage.go

@@ -7,6 +7,7 @@ import (
 	"github.com/samber/lo"
 	"go.uber.org/zap"
 
+	"github.com/milvus-io/milvus-proto/go-api/v2/msgpb"
 	"github.com/milvus-io/milvus/internal/flushcommon/broker"
 	"github.com/milvus-io/milvus/internal/flushcommon/util"
 	"github.com/milvus-io/milvus/internal/streamingnode/server/resource"
@@ -149,7 +150,14 @@ func (impl *WALFlusherImpl) buildFlusherComponents(ctx context.Context, l wal.WA
 	broker := broker.NewCoordBroker(mixc, paramtable.GetNodeID())
 	chunkManager := resource.Resource().ChunkManager()
 
-	cpUpdater := util.NewChannelCheckpointUpdater(broker)
+	cpUpdater := util.NewChannelCheckpointUpdaterWithCallback(broker, func(mp *msgpb.MsgPosition) {
+		messageID := adaptor.MustGetMessageIDFromMQWrapperIDBytes(l.WALName(), mp.MsgID)
+		impl.rs.UpdateFlusherCheckpoint(&recovery.WALCheckpoint{
+			MessageID: messageID,
+			TimeTick:  mp.Timestamp,
+			Magic:     recovery.RecoveryMagicStreamingInitialized,
+		})
+	})
 	go cpUpdater.Start()
 
 	fc := &flusherComponents{

internal/streamingnode/server/flusher/flusherimpl/wal_flusher.go

@@ -67,8 +67,8 @@ func (m *AppendMetrics) StartAppendGuard() *AppendMetricsGuard {
 func (m *AppendMetrics) IntoLogFields() []zap.Field {
 	fields := []zap.Field{
 		log.FieldMessage(m.msg),
-		zap.Duration("append_duration", m.appendDuration),
-		zap.Duration("impl_append_duration", m.implAppendDuration),
+		zap.Duration("appendDuration", m.appendDuration),
+		zap.Duration("implAppendDuration", m.implAppendDuration),
 	}
 	for name, ims := range m.interceptors {
 		for i, im := range ims {
@@ -80,10 +80,10 @@ func (m *AppendMetrics) IntoLogFields() []zap.Field {
 	if m.err != nil {
 		fields = append(fields, zap.Error(m.err))
 	} else {
-		fields = append(fields, zap.String("message_id", m.result.MessageID.String()))
-		fields = append(fields, zap.Uint64("time_tick", m.result.TimeTick))
+		fields = append(fields, zap.String("messageID", m.result.MessageID.String()))
+		fields = append(fields, zap.Uint64("timetick", m.result.TimeTick))
 		if m.result.TxnCtx != nil {
-			fields = append(fields, zap.Int64("txn_id", int64(m.result.TxnCtx.TxnID)))
+			fields = append(fields, zap.Int64("txnID", int64(m.result.TxnCtx.TxnID)))
 		}
 	}
 	return fields

internal/streamingnode/server/wal/metricsutil/append.go

@@ -7,7 +7,7 @@ import (
 )
 
 const (
-	recoveryMagicStreamingInitialized int64 = 1 // the vchannel info is set into the catalog.
+	RecoveryMagicStreamingInitialized int64 = 1 // the vchannel info is set into the catalog.
 	// the checkpoint is set into the catalog.
 )
 

internal/streamingnode/server/wal/recovery/checkpoint.go

@@ -123,10 +123,22 @@ func (rs *recoveryStorageImpl) persistDirtySnapshot(ctx context.Context, snapsho
 
 	// sample the checkpoint for truncator to make wal truncation.
 	rs.metrics.ObServePersistedMetrics(snapshot.Checkpoint.TimeTick)
-	rs.truncator.SampleCheckpoint(snapshot.Checkpoint)
+	rs.sampleTruncateCheckpoint(snapshot.Checkpoint)
 	return
 }
 
+func (rs *recoveryStorageImpl) sampleTruncateCheckpoint(checkpoint *WALCheckpoint) {
+	if rs.flusherCheckpoint == nil {
+		return
+	}
+	// use the smaller one to truncate the wal.
+	if rs.flusherCheckpoint.MessageID.LTE(checkpoint.MessageID) {
+		rs.truncator.SampleCheckpoint(rs.flusherCheckpoint)
+	} else {
+		rs.truncator.SampleCheckpoint(checkpoint)
+	}
+}
+
 // dropAllVirtualChannel drops all virtual channels that are in the dropped state.
 // TODO: DropVirtualChannel will be called twice here,
 // call it in recovery storage is used to promise the drop virtual channel must be called after recovery.

internal/streamingnode/server/wal/recovery/recovery_background_task.go

@@ -119,7 +119,7 @@ func (r *recoveryStorageImpl) initializeRecoverInfo(ctx context.Context, channel
 			Id: untilMessage.LastConfirmedMessageID().Marshal(),
 		},
 		TimeTick:      untilMessage.TimeTick(),
-		RecoveryMagic: recoveryMagicStreamingInitialized,
+		RecoveryMagic: RecoveryMagicStreamingInitialized,
 	}
 	if err := resource.Resource().StreamingNodeCatalog().SaveConsumeCheckpoint(ctx, channelInfo.Name, checkpoint); err != nil {
 		return nil, errors.Wrap(err, "failed to save checkpoint to catalog")

internal/streamingnode/server/wal/recovery/recovery_persisted.go

@@ -46,7 +46,7 @@ func TestInitRecoveryInfoFromMeta(t *testing.T) {
 				Id: rmq.NewRmqID(1).Marshal(),
 			},
 			TimeTick:      1,
-			RecoveryMagic: recoveryMagicStreamingInitialized,
+			RecoveryMagic: RecoveryMagicStreamingInitialized,
 		}, nil)
 	resource.InitForTest(t, resource.OptStreamingNodeCatalog(snCatalog))
 	walName := "rocksmq"
@@ -58,7 +58,7 @@ func TestInitRecoveryInfoFromMeta(t *testing.T) {
 	err := rs.recoverRecoveryInfoFromMeta(context.Background(), walName, channel, lastConfirmed.IntoImmutableMessage(rmq.NewRmqID(1)))
 	assert.NoError(t, err)
 	assert.NotNil(t, rs.checkpoint)
-	assert.Equal(t, recoveryMagicStreamingInitialized, rs.checkpoint.Magic)
+	assert.Equal(t, RecoveryMagicStreamingInitialized, rs.checkpoint.Magic)
 	assert.True(t, rs.checkpoint.MessageID.EQ(rmq.NewRmqID(1)))
 }
 

internal/streamingnode/server/wal/recovery/recovery_persisted_test.go

@@ -61,6 +61,10 @@ type RecoveryStorage interface {
 	// ObserveMessage observes the message from the WAL.
 	ObserveMessage(msg message.ImmutableMessage)
 
+	// UpdateFlusherCheckpoint updates the checkpoint of flusher.
+	// TODO: should be removed in future, after merge the flusher logic into recovery storage.
+	UpdateFlusherCheckpoint(checkpoint *WALCheckpoint)
+
 	// Close closes the recovery storage.
 	Close()
 }

internal/streamingnode/server/wal/recovery/recovery_storage.go

@@ -82,6 +82,7 @@ type recoveryStorageImpl struct {
 	segments               map[int64]*segmentRecoveryInfo
 	vchannels              map[string]*vchannelRecoveryInfo
 	checkpoint             *WALCheckpoint
+	flusherCheckpoint      *WALCheckpoint
 	dirtyCounter           int // records the message count since last persist snapshot.
 	// used to trigger the recovery persist operation.
 	persistNotifier chan struct{}
@@ -90,6 +91,19 @@ type recoveryStorageImpl struct {
 	metrics         *recoveryMetrics
 }
 
+// UpdateFlusherCheckpoint updates the checkpoint of flusher.
+// TODO: should be removed in future, after merge the flusher logic into recovery storage.
+func (r *recoveryStorageImpl) UpdateFlusherCheckpoint(checkpoint *WALCheckpoint) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.flusherCheckpoint == nil || r.flusherCheckpoint.MessageID.LTE(checkpoint.MessageID) {
+		r.flusherCheckpoint = checkpoint
+		r.Logger().Info("update checkpoint of flusher", zap.String("messageID", checkpoint.MessageID.String()))
+		return
+	}
+	r.Logger().Warn("update illegal checkpoint of flusher", zap.String("current", r.flusherCheckpoint.MessageID.String()), zap.String("target", checkpoint.MessageID.String()))
+}
+
 // ObserveMessage is called when a new message is observed.
 func (r *recoveryStorageImpl) ObserveMessage(msg message.ImmutableMessage) {
 	r.mu.Lock()

internal/streamingnode/server/wal/recovery/recovery_storage_impl.go

@@ -23,7 +23,7 @@ func TestTruncator(t *testing.T) {
 	truncator := newSamplingTruncator(&WALCheckpoint{
 		MessageID: rmq.NewRmqID(1),
 		TimeTick:  1,
-		Magic:     recoveryMagicStreamingInitialized,
+		Magic:     RecoveryMagicStreamingInitialized,
 	}, w, newRecoveryStorageMetrics(types.PChannelInfo{Name: "test", Term: 1}))
 
 	for i := 0; i < 20; i++ {
@@ -32,7 +32,7 @@ func TestTruncator(t *testing.T) {
 			truncator.SampleCheckpoint(&WALCheckpoint{
 				MessageID: rmq.NewRmqID(int64(i)),
 				TimeTick:  tsoutil.ComposeTSByTime(time.Now(), 0),
-				Magic:     recoveryMagicStreamingInitialized,
+				Magic:     RecoveryMagicStreamingInitialized,
 			})
 		}
 	}