Merge pull request kakao#199 from ijsong/change-sync

feat(storagenode): change the synchronization method to accept only the last commit context

internal/storage/testing.go

@@ -8,6 +8,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/kakao/varlog/pkg/types"
+	"github.com/kakao/varlog/proto/varlogpb"
 )
 
 func TestNewStorage(tb testing.TB, opts ...Option) *Storage {
@@ -44,3 +45,27 @@ func TestSetCommitContext(tb testing.TB, stg *Storage, cc CommitContext) {
 	require.NoError(tb, batch.Apply())
 	require.NoError(tb, batch.Close())
 }
+
+func TestDeleteCommitContext(tb testing.TB, stg *Storage) {
+	err := stg.db.Delete(commitContextKey, pebble.Sync)
+	require.NoError(tb, err)
+}
+
+func TestDeleteLogEntry(tb testing.TB, stg *Storage, lsn varlogpb.LogSequenceNumber) {
+	batch := stg.db.NewBatch()
+	defer func() {
+		err := batch.Close()
+		require.NoError(tb, err)
+	}()
+
+	dk := make([]byte, dataKeyLength)
+	err := batch.Delete(encodeDataKeyInternal(lsn.LLSN, dk), nil)
+	require.NoError(tb, err)
+
+	ck := make([]byte, commitKeyLength)
+	err = batch.Delete(encodeCommitKeyInternal(lsn.GLSN, ck), nil)
+	require.NoError(tb, err)
+
+	err = batch.Commit(pebble.Sync)
+	require.NoError(tb, err)
+}

internal/storagenode/logstream/config.go

@@ -16,7 +16,7 @@ const (
 	DefaultWriteQueueCapacity           = 1024
 	DefaultCommitQueueCapacity          = 1024
 	DefaultReplicateClientQueueCapacity = 1024
-	DefaultSyncInitTimeout              = 10 * time.Second
+	DefaultSyncTimeout                  = 10 * time.Second
 )
 
 type executorConfig struct {
@@ -33,7 +33,7 @@ type executorConfig struct {
 	replicateClientGRPCOptions   []grpc.DialOption
 	logger                       *zap.Logger
 	lsm                          *telemetry.LogStreamMetrics
-	syncInitTimeout              time.Duration
+	syncTimeout                  time.Duration
 }
 
 func newExecutorConfig(opts []ExecutorOption) (executorConfig, error) {
@@ -43,7 +43,7 @@ func newExecutorConfig(opts []ExecutorOption) (executorConfig, error) {
 		commitQueueCapacity:          DefaultCommitQueueCapacity,
 		replicateClientQueueCapacity: DefaultReplicateClientQueueCapacity,
 		logger:                       zap.NewNop(),
-		syncInitTimeout:              DefaultSyncInitTimeout,
+		syncTimeout:                  DefaultSyncTimeout,
 	}
 	for _, opt := range opts {
 		opt.applyExecutor(&cfg)
@@ -174,8 +174,11 @@ func WithLogStreamMetrics(lsm *telemetry.LogStreamMetrics) ExecutorOption {
 	})
 }
 
-func WithSyncInitTimeout(syncInitTimeout time.Duration) ExecutorOption {
+// WithSyncTimeout sets timeout for synchronization in the destination replica.
+// If the destination replica doesn't receive the SyncReplicate RPC within
+// syncTimeout, other SyncInit RPC can cancel the synchronization.
+func WithSyncTimeout(syncTimeout time.Duration) ExecutorOption {
 	return newFuncExecutorOption(func(cfg *executorConfig) {
-		cfg.syncInitTimeout = syncInitTimeout
+		cfg.syncTimeout = syncTimeout
 	})
 }

internal/storagenode/logstream/executor.go

@@ -53,8 +53,14 @@ type Executor struct {
 	// primaryBackups is a slice of replicas of a log stream.
 	// It is updated by Unseal and is read by many codes.
 	primaryBackups []varlogpb.LogStreamReplica
-	// sync replicate buffer
-	srb        *syncReplicateBuffer
+	dstSyncInfo    struct {
+		// lastSyncTime is when the replica handles SyncInit or SyncReplicate
+		// successfully. If the elapsed time since the last RPC is greater than
+		// configured syncDurationTimeout, the synchronization process can be
+		// canceled by another SyncInit.
+		lastSyncTime time.Time
+		srcReplica   types.StorageNodeID
+	}
 	sts        map[types.StorageNodeID]*syncTracker
 	syncRunner *runner.Runner
 
@@ -241,6 +247,10 @@ func (lse *Executor) Seal(_ context.Context, lastCommittedGLSN types.GLSN) (stat
 		// log stream.
 		lse.logger.Panic("log stream: seal: unexpected last committed GLSN")
 	}
+	if lse.esm.load() == executorStateLearning {
+		return varlogpb.LogStreamStatusSealing, localHWM, nil
+	}
+
 	if localHighWatermark.GLSN < lastCommittedGLSN || invalid {
 		status = varlogpb.LogStreamStatusSealing
 		return status, localHWM, nil