enhance: move sent first timeticksync, make recovery more easier

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

@@ -23,14 +23,14 @@ import (
 var errChannelLifetimeUnrecoverable = errors.New("channel lifetime unrecoverable")
 
 // RecoverWALFlusher recovers the wal flusher.
-func RecoverWALFlusher(param interceptors.InterceptorBuildParam) *WALFlusherImpl {
+func RecoverWALFlusher(param *interceptors.InterceptorBuildParam) *WALFlusherImpl {
 	flusher := &WALFlusherImpl{
 		notifier: syncutil.NewAsyncTaskNotifier[struct{}](),
 		wal:      param.WAL,
 		logger: resource.Resource().Logger().With(
 			log.FieldComponent("flusher"),
-			zap.String("pchannel", param.WALImpls.Channel().Name)),
-		metrics: newFlusherMetrics(param.WALImpls.Channel()),
+			zap.String("pchannel", param.ChannelInfo.Name)),
+		metrics: newFlusherMetrics(param.ChannelInfo),
 	}
 	go flusher.Execute()
 	return flusher

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

@@ -23,7 +23,6 @@ import (
 	internaltypes "github.com/milvus-io/milvus/internal/types"
 	"github.com/milvus-io/milvus/internal/util/streamingutil"
 	"github.com/milvus-io/milvus/pkg/v2/common"
-	"github.com/milvus-io/milvus/pkg/v2/mocks/streaming/mock_walimpls"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/rootcoordpb"
 	"github.com/milvus-io/milvus/pkg/v2/streaming/util/message"
@@ -72,13 +71,9 @@ func TestWALFlusher(t *testing.T) {
 		resource.OptStreamingNodeCatalog(snMeta),
 		resource.OptChunkManager(mock_storage.NewMockChunkManager(t)),
 	)
-	walImpl := mock_walimpls.NewMockWALImpls(t)
-	walImpl.EXPECT().Channel().Return(types.PChannelInfo{Name: "pchannel"})
-
 	l := newMockWAL(t, false)
-	param := interceptors.InterceptorBuildParam{
-		WALImpls: walImpl,
-		WAL:      syncutil.NewFuture[wal.WAL](),
+	param := &interceptors.InterceptorBuildParam{
+		WAL: syncutil.NewFuture[wal.WAL](),
 	}
 	param.WAL.Set(l)
 	flusher := RecoverWALFlusher(param)

internal/streamingnode/server/wal/adaptor/initializing.go

@@ -0,0 +1,110 @@
+package adaptor
+
+import (
+	"context"
+	"time"
+
+	"github.com/cockroachdb/errors"
+	"go.uber.org/zap"
+
+	"github.com/milvus-io/milvus/internal/streamingnode/server/resource"
+	"github.com/milvus-io/milvus/internal/streamingnode/server/wal"
+	"github.com/milvus-io/milvus/internal/streamingnode/server/wal/interceptors"
+	"github.com/milvus-io/milvus/internal/streamingnode/server/wal/interceptors/timetick"
+	"github.com/milvus-io/milvus/internal/streamingnode/server/wal/interceptors/timetick/mvcc"
+	"github.com/milvus-io/milvus/internal/streamingnode/server/wal/interceptors/wab"
+	"github.com/milvus-io/milvus/pkg/v2/streaming/util/message"
+	"github.com/milvus-io/milvus/pkg/v2/streaming/walimpls"
+	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
+	"github.com/milvus-io/milvus/pkg/v2/util/syncutil"
+	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
+)
+
+// buildInterceptorParams builds the interceptor params for the walimpls.
+func buildInterceptorParams(ctx context.Context, underlyingWALImpls walimpls.WALImpls) (*interceptors.InterceptorBuildParam, error) {
+	msg, err := sendFirstTimeTick(ctx, underlyingWALImpls)
+	if err != nil {
+		return nil, err
+	}
+
+	capacity := int(paramtable.Get().StreamingCfg.WALWriteAheadBufferCapacity.GetAsSize())
+	keepalive := paramtable.Get().StreamingCfg.WALWriteAheadBufferKeepalive.GetAsDurationByParse()
+	writeAheadBuffer := wab.NewWirteAheadBuffer(
+		underlyingWALImpls.Channel().Name,
+		resource.Resource().Logger().With(),
+		capacity,
+		keepalive,
+		msg,
+	)
+	mvccManager := mvcc.NewMVCCManager(msg.TimeTick())
+	return &interceptors.InterceptorBuildParam{
+		ChannelInfo:          underlyingWALImpls.Channel(),
+		WAL:                  syncutil.NewFuture[wal.WAL](),
+		InitializedTimeTick:  msg.TimeTick(),
+		InitializedMessageID: msg.MessageID(),
+		WriteAheadBuffer:     writeAheadBuffer,
+		MVCCManager:          mvccManager,
+	}, nil
+}
+
+// sendFirstTimeTick sends the first timetick message to walimpls.
+// It is used to make a fence operation with the underlying walimpls and get the timetick and last message id to recover the wal state.
+func sendFirstTimeTick(ctx context.Context, underlyingWALImpls walimpls.WALImpls) (message.ImmutableMessage, error) {
+	logger := resource.Resource().Logger()
+	logger.Info("start to sync first time tick")
+	defer logger.Info("sync first time tick done")
+
+	backoffTimer := typeutil.NewBackoffTimer(typeutil.BackoffTimerConfig{
+		Default: 5 * time.Second,
+		Backoff: typeutil.BackoffConfig{
+			InitialInterval: 20 * time.Millisecond,
+			Multiplier:      2.0,
+			MaxInterval:     5 * time.Second,
+		},
+	})
+	backoffTimer.EnableBackoff()
+
+	var lastErr error
+	sourceID := paramtable.GetNodeID()
+	// Send first timetick message to wal before interceptor is ready.
+	for count := 0; ; count++ {
+		if count > 0 {
+			nextTimer, nextBalanceInterval := backoffTimer.NextTimer()
+			logger.Warn(
+				"send first time tick failed",
+				zap.Duration("nextBalanceInterval", nextBalanceInterval),
+				zap.Int("retryCount", count),
+				zap.Error(lastErr),
+			)
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			case <-nextTimer:
+			}
+		}
+
+		// Sent first timetick message to wal before ready.
+		// New TT is always greater than all tt on previous streamingnode.
+		// A fencing operation of underlying WAL is needed to make exclusive produce of topic.
+		// Otherwise, the TT principle may be violated.
+		// And sendTsMsg must be done, to help ackManager to get first LastConfirmedMessageID
+		// !!! Send a timetick message into walimpls directly is safe.
+		resource.Resource().TSOAllocator().Sync()
+		ts, err := resource.Resource().TSOAllocator().Allocate(ctx)
+		if err != nil {
+			lastErr = errors.Wrap(err, "allocate timestamp failed")
+			continue
+		}
+		msg, err := timetick.NewTimeTickMsg(ts, nil, sourceID)
+		if err != nil {
+			lastErr = errors.Wrap(err, "at build time tick msg")
+			continue
+		}
+		msgID, err := underlyingWALImpls.Append(ctx, msg)
+		if err != nil {
+			lastErr = errors.Wrap(err, "send first timestamp message failed")
+			continue
+		}
+		return msg.IntoImmutableMessage(msgID), nil
+	}
+}

internal/streamingnode/server/wal/adaptor/opener.go

@@ -56,10 +56,13 @@
 	}
 
 	// wrap the wal into walExtend with cleanup function and interceptors.
-	wal := adaptImplsToWAL(l, o.interceptorBuilders, func() {
+	wal, err := adaptImplsToWAL(ctx, l, o.interceptorBuilders, func() {
 		o.walInstances.Remove(id)
 		logger.Info("wal deleted from opener")
 	})
+	if err != nil {
+		return nil, err
+	}
 
 	o.walInstances.Insert(id, wal)
 	logger.Info("new wal created")

internal/streamingnode/server/wal/adaptor/opener_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 
+	"github.com/milvus-io/milvus/internal/streamingnode/server/resource"
 	"github.com/milvus-io/milvus/internal/streamingnode/server/wal"
 	"github.com/milvus-io/milvus/pkg/v2/mocks/streaming/mock_walimpls"
 	"github.com/milvus-io/milvus/pkg/v2/mocks/streaming/util/mock_message"
@@ -40,6 +41,7 @@ func TestOpenerAdaptorFailure(t *testing.T) {
 }
 
 func TestOpenerAdaptor(t *testing.T) {
+	resource.InitForTest(t)
 	// Build basic opener.
 	basicOpener := mock_walimpls.NewMockOpenerImpls(t)
 	basicOpener.EXPECT().Open(mock.Anything, mock.Anything).RunAndReturn(

internal/streamingnode/server/wal/adaptor/scanner_adaptor.go

@@ -126,15 +126,13 @@ func (s *scannerAdaptorImpl) produceEventLoop(msgChan chan<- message.ImmutableMe
 	var wb wab.ROWriteAheadBuffer
 	var err error
 	if s.Channel().AccessMode == types.AccessModeRW {
-		if wb, err = resource.Resource().TimeTickInspector().MustGetOperator(s.Channel()).WriteAheadBuffer(s.Context()); err != nil {
-			return err
-		}
 		// Trigger a persisted time tick to make sure the timetick is pushed forward.
 		// because the underlying wal may be deleted because of retention policy.
 		// So we cannot get the timetick from the wal.
 		// Trigger the timetick inspector to append a new persisted timetick,
 		// then the catch up scanner can see the latest timetick and make a catchup.
 		resource.Resource().TimeTickInspector().TriggerSync(s.Channel(), true)
+		wb = resource.Resource().TimeTickInspector().MustGetOperator(s.Channel()).WriteAheadBuffer()
 	}
 
 	scanner := newSwithableScanner(s.Name(), s.logger, s.innerWAL, wb, s.readOption.DeliverPolicy, msgChan)