refactor(logstream): use beginLLSN in Replicate method

This commit replaces llsnList with beginLLSN in the Replicate method. Since
LLSNs in llsnList are strictly sequential, beginLLSN and the length of dataList
are sufficient. The corresponding tests have been updated to reflect this
change. The llsn field in ReplicateRequest has been deprecated and will be
removed soon.

internal/storagenode/logstream/executor.go

@@ -160,7 +160,7 @@ func NewExecutor(opts ...ExecutorOption) (lse *Executor, err error) {
 	return lse, err
 }
 
-func (lse *Executor) Replicate(ctx context.Context, llsnList []types.LLSN, dataList [][]byte) error {
+func (lse *Executor) Replicate(ctx context.Context, beginLLSN types.LLSN, dataList [][]byte) error {
 	lse.inflight.Add(1)
 	defer lse.inflight.Add(-1)
 
@@ -178,7 +178,7 @@ func (lse *Executor) Replicate(ctx context.Context, llsnList []types.LLSN, dataL
 	var preparationDuration time.Duration
 	startTime := time.Now()
 	dataBytes := int64(0)
-	batchSize := len(llsnList)
+	batchSize := len(dataList)
 	defer func() {
 		if lse.lsm == nil {
 			return
@@ -190,11 +190,11 @@ func (lse *Executor) Replicate(ctx context.Context, llsnList []types.LLSN, dataL
 		lse.lsm.ReplicatePreparationMicro.Add(preparationDuration.Microseconds())
 	}()
 
-	oldLLSN, newLLSN := llsnList[0], llsnList[batchSize-1]+1
+	oldLLSN, newLLSN := beginLLSN, beginLLSN+types.LLSN(batchSize)
 	wb := lse.stg.NewWriteBatch()
 	cwts := newListQueue()
-	for i := 0; i < len(llsnList); i++ {
-		_ = wb.Set(llsnList[i], dataList[i])
+	for i := 0; i < batchSize; i++ {
+		_ = wb.Set(beginLLSN+types.LLSN(i), dataList[i])
 		dataBytes += int64(len(dataList[i]))
 		cwts.PushFront(newCommitWaitTask(nil))
 	}
@@ -329,7 +329,7 @@ func (lse *Executor) Unseal(_ context.Context, replicas []varlogpb.LogStreamRepl
 				replica:       replicas[i],
 				rpcConn:       rpcConn,
 				queueCapacity: lse.replicateClientQueueCapacity,
-				//grpcDialOptions: lse.replicateClientGRPCOptions,
+				// grpcDialOptions: lse.replicateClientGRPCOptions,
 				lse:    lse,
 				logger: lse.logger.Named("replicate client"),
 			})

internal/storagenode/logstream/executor_test.go

@@ -84,7 +84,7 @@ func TestExecutor_Closed(t *testing.T) {
 	_, err := lse.Append(context.Background(), TestNewBatchData(t, 1, 0))
 	assert.ErrorIs(t, err, verrors.ErrClosed)
 
-	err = lse.Replicate(context.Background(), []types.LLSN{1}, TestNewBatchData(t, 1, 0))
+	err = lse.Replicate(context.Background(), types.LLSN(1), TestNewBatchData(t, 1, 0))
 	assert.ErrorIs(t, err, verrors.ErrClosed)
 
 	_, _, err = lse.Seal(context.Background(), types.MinGLSN)
@@ -182,7 +182,7 @@ func TestExecutor_Sealing(t *testing.T) {
 				assert.Equal(t, varlogpb.LogStreamStatusSealing, st)
 				assert.Equal(t, executorStateSealing, lse.esm.load())
 
-				err = lse.Replicate(context.Background(), []types.LLSN{1}, TestNewBatchData(t, 1, 0))
+				err = lse.Replicate(context.Background(), types.LLSN(1), TestNewBatchData(t, 1, 0))
 				assert.ErrorIs(t, err, verrors.ErrSealed)
 			},
 		},
@@ -286,7 +286,7 @@ func TestExecutor_Sealed(t *testing.T) {
 	_, err = lse.Append(context.Background(), TestNewBatchData(t, 1, 0))
 	assert.ErrorIs(t, err, verrors.ErrSealed)
 
-	err = lse.Replicate(context.Background(), []types.LLSN{1}, TestNewBatchData(t, 1, 0))
+	err = lse.Replicate(context.Background(), types.LLSN(1), TestNewBatchData(t, 1, 0))
 	assert.ErrorIs(t, err, verrors.ErrSealed)
 }
 
@@ -616,22 +616,18 @@ func TestExecutor_Replicate(t *testing.T) {
 
 			// primary
 			if tc.isErr {
-				err := lse.Replicate(context.Background(), []types.LLSN{1}, [][]byte{nil})
+				err := lse.Replicate(context.Background(), types.LLSN(1), [][]byte{nil})
 				assert.Error(t, err)
 				return
 			}
 
 			// backup
-			var llsn types.LLSN
+			llsn := types.MinLLSN
 			for _, batchLen := range batchlet.LengthClasses {
 				dataList := TestNewBatchData(t, batchLen, 0)
-				llsnList := make([]types.LLSN, batchLen)
-				for i := 0; i < batchLen; i++ {
-					llsn++
-					llsnList[i] = llsn
-				}
-				err := lse.Replicate(context.Background(), llsnList, dataList)
+				err := lse.Replicate(context.Background(), llsn, dataList)
 				assert.NoError(t, err)
+				llsn += types.LLSN(len(dataList))
 			}
 
 			// Commit
@@ -875,7 +871,7 @@ func TestExecutor_ReplicateSeal(t *testing.T) {
 	go func() {
 		defer wg.Done()
 		for llsn := lastLLSN + 1; llsn < types.MaxLLSN; llsn++ {
-			err := lse.Replicate(context.Background(), []types.LLSN{llsn}, [][]byte{nil})
+			err := lse.Replicate(context.Background(), llsn, [][]byte{nil})
 			if err != nil {
 				break
 			}
@@ -935,7 +931,7 @@ func TestExecutor_ReplicateSeal(t *testing.T) {
 	go func() {
 		defer wg.Done()
 		for llsn := lastLLSN + 1; llsn < types.MaxLLSN; llsn++ {
-			err := lse.Replicate(context.Background(), []types.LLSN{llsn}, [][]byte{nil})
+			err := lse.Replicate(context.Background(), llsn, [][]byte{nil})
 			if err != nil {
 				break
 			}
@@ -3091,7 +3087,6 @@ func TestExecutorSyncInit(t *testing.T) {
 						rpt.UncommittedLLSNLength == 0 &&
 						rpt.HighWatermark == types.GLSN(dstLastLSN) &&
 						rpt.Version == types.Version(1)
-
 				}, time.Second, 10*time.Millisecond)
 				wg.Wait()
 
@@ -3500,7 +3495,6 @@ func TestExecutorSyncReplicate(t *testing.T) {
 					rpt.UncommittedLLSNLength == 0 &&
 					rpt.HighWatermark == types.GLSN(numLogs) &&
 					rpt.Version == types.Version(1)
-
 			}, time.Second, 10*time.Millisecond)
 			wg.Wait()
 

internal/storagenode/logstream/replicate_client.go

@@ -26,7 +26,7 @@ type replicateClient struct {
 
 	rpcClient    snpb.ReplicatorClient
 	streamClient snpb.Replicator_ReplicateClient
-	//req          *snpb.ReplicateRequest
+	// req          *snpb.ReplicateRequest
 }
 
 // newReplicateClient creates a new client to replicate logs to backup replica.
@@ -52,7 +52,7 @@ func newReplicateClient(ctx context.Context, cfg replicateClientConfig) (*replic
 		replicateClientConfig: cfg,
 		queue:                 make(chan *replicateTask, cfg.queueCapacity),
 		runner:                runner.New("replicate client", cfg.logger),
-		//rpcConn:               rpcConn,
+		// rpcConn:               rpcConn,
 		rpcClient:    rpcClient,
 		streamClient: streamClient,
 		// NOTE: To reuse the request struct, we need to initialize the field LLSN.
@@ -135,11 +135,15 @@ func (rc *replicateClient) sendLoop(ctx context.Context) {
 func (rc *replicateClient) sendLoopInternal(_ context.Context, rt *replicateTask, req *snpb.ReplicateRequest) error {
 	// Remove maxAppendSubBatchSize, since rt already has batched data.
 	startTime := time.Now()
+	// TODO(jun): Since (snpb.ReplicateRequest).LLSN is deprecated, it will disappear soon.
 	// NOTE: We need to copy the LLSN array, since the array is reused.
 	req.LLSN = req.LLSN[0:len(rt.llsnList)]
 	copy(req.LLSN, rt.llsnList)
-	//req.LLSN = rt.llsnList
+	// req.LLSN = rt.llsnList
 	req.Data = rt.dataList
+	if len(rt.llsnList) > 0 {
+		req.BeginLLSN = rt.llsnList[0]
+	}
 	rt.release()
 	err := rc.streamClient.Send(req)
 	inflight := rc.inflight.Add(-1)

internal/storagenode/replication_server.go

@@ -173,7 +173,7 @@ func (rs *replicationServer) replicate(ctx context.Context, requestC <-chan *rep
 
 			lse.Metrics().ReplicateServerOperations.Add(1)
 
-			err = lse.Replicate(ctx, rst.req.LLSN, rst.req.Data)
+			err = lse.Replicate(ctx, rst.req.BeginLLSN, rst.req.Data)
 			if err != nil {
 				rst.release()
 				return