Validate Checkpoint in ChangePack for PushPull API requests

server/packs/packs.go

@@ -20,6 +20,7 @@ package packs
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 	gotime "time"
@@ -37,6 +38,11 @@ import (
 	"github.com/yorkie-team/yorkie/server/logging"
 )
 
+var (
+	// ErrClientSeqNotSequential is returned when ClientSeq in reqPack.Changes are not sequential
+	ErrClientSeqNotSequential = errors.New("ClientSeq in reqPack.Changes are not sequential")
+)
+
 // PushPullKey creates a new sync.Key of PushPull for the given document.
 func PushPullKey(projectID types.ID, docKey key.Key) sync.Key {
 	return sync.NewKey(fmt.Sprintf("pushpull-%s-%s", projectID, docKey))
@@ -77,6 +83,11 @@ func PushPull(
 
 	// TODO: Changes may be reordered or missing during communication on the network.
 	// We should check the change.pack with checkpoint to make sure the changes are in the correct order.
+	err := validateClientSeqSequential(reqPack.Changes)
+	if err != nil {
+		return nil, err
+	}
+
 	initialServerSeq := docInfo.ServerSeq
 
 	// 01. push changes: filter out the changes that are already saved in the database.
@@ -272,3 +283,24 @@ func BuildDocumentForServerSeq(
 
 	return doc, nil
 }
+
+func validateClientSeqSequential(changes []*change.Change) error {
+	if len(changes) <= 1 {
+		return nil
+	}
+
+	nextClientSeq := changes[0].ClientSeq()
+	for _, cn := range changes[1:] {
+		nextClientSeq++
+
+		if nextClientSeq != cn.ClientSeq() {
+			return fmt.Errorf(
+				"ClientSeq in Changes are not sequential (expected: %d, actual: %d) : %w",
+				nextClientSeq,
+				cn.ClientSeq(),
+				ErrClientSeqNotSequential,
+			)
+		}
+	}
+	return nil
+}

server/packs/packs_test.go

@@ -0,0 +1,261 @@
+package packs_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"golang.org/x/net/context"
+
+	"github.com/yorkie-team/yorkie/api/converter"
+	"github.com/yorkie-team/yorkie/api/types"
+	api "github.com/yorkie-team/yorkie/api/yorkie/v1"
+	"github.com/yorkie-team/yorkie/pkg/document"
+	"github.com/yorkie-team/yorkie/pkg/document/change"
+	"github.com/yorkie-team/yorkie/pkg/document/time"
+	"github.com/yorkie-team/yorkie/server/backend"
+	"github.com/yorkie-team/yorkie/server/backend/database"
+	"github.com/yorkie-team/yorkie/server/clients"
+	"github.com/yorkie-team/yorkie/server/documents"
+	"github.com/yorkie-team/yorkie/server/packs"
+	"github.com/yorkie-team/yorkie/server/profiling/prometheus"
+	"github.com/yorkie-team/yorkie/server/rpc/connecthelper"
+	"github.com/yorkie-team/yorkie/test/helper"
+)
+
+var (
+	clientID = "000000000000000000000001"
+)
+
+func Test(t *testing.T) {
+	t.Run("push/pull sequential ClientSeq test (happy case)", func(t *testing.T) {
+		RunPushPullWithSequentialClientSeqTest(t)
+	})
+
+	t.Run("push/pull not sequential ClientSeq test", func(t *testing.T) {
+		RunPushPullWithNotSequentialClientSeqTest(t)
+	})
+
+	t.Run("push/pull ClientSeq less than ClientInfo's ClientSeq (duplicated request)", func(t *testing.T) {
+		RunPushPullWithClientSeqLessThanClientInfoTest(t)
+	})
+
+	t.Run("push/pull ServerSeq greater than DocInfo's ServerSeq", func(t *testing.T) {
+		RunPushPullWithServerSeqGreaterThanDocInfoTest(t)
+	})
+}
+
+func RunPushPullWithSequentialClientSeqTest(t *testing.T) {
+	ctx := context.Background()
+	be := setUpBackend(t)
+	project, _ := be.DB.FindProjectInfoByID(
+		ctx,
+		database.DefaultProjectID,
+	)
+
+	clientInfo, _ := clients.Activate(ctx, be.DB, project.ToProject(), clientID)
+	actorID, _ := time.ActorIDFromHex(clientID)
+
+	changePackWithSequentialClientSeq, _ :=
+		createChangePackWithSequentialClientSeq(helper.TestDocKey(t).String(), actorID.Bytes())
+
+	docInfo, _ := documents.FindDocInfoByKeyAndOwner(
+		ctx, be, clientInfo, changePackWithSequentialClientSeq.DocumentKey, true)
+	err := clientInfo.AttachDocument(
+		docInfo.ID, changePackWithSequentialClientSeq.IsAttached())
+	if err != nil {
+		assert.Fail(t, "failed to attach document")
+	}
+
+	_, err = packs.PushPull(
+		ctx, be, project.ToProject(), clientInfo, docInfo,
+		changePackWithSequentialClientSeq, packs.PushPullOptions{
+			Mode:   types.SyncModePushPull,
+			Status: document.StatusAttached,
+		})
+	assert.NoError(t, err)
+}
+
+func RunPushPullWithNotSequentialClientSeqTest(t *testing.T) {
+	ctx := context.Background()
+	be := setUpBackend(t)
+	project, _ := be.DB.FindProjectInfoByID(
+		ctx,
+		database.DefaultProjectID,
+	)
+
+	clientInfo, _ := clients.Activate(ctx, be.DB, project.ToProject(), clientID)
+
+	actorID, _ := time.ActorIDFromHex(clientID)
+	changePackWithNotSequentialClientSeq, _ :=
+		createChangePackWithNotSequentialClientSeq(helper.TestDocKey(t).String(), actorID.Bytes())
+
+	docInfo, _ := documents.FindDocInfoByKeyAndOwner(ctx, be, clientInfo,
+		changePackWithNotSequentialClientSeq.DocumentKey, true)
+	err := clientInfo.AttachDocument(
+		docInfo.ID, changePackWithNotSequentialClientSeq.IsAttached())
+	if err != nil {
+		assert.Fail(t, "failed to attach document")
+	}
+
+	_, err = packs.PushPull(
+		ctx, be, project.ToProject(), clientInfo, docInfo,
+		changePackWithNotSequentialClientSeq, packs.PushPullOptions{
+			Mode:   types.SyncModePushPull,
+			Status: document.StatusAttached,
+		})
+	assert.Equal(t, connecthelper.CodeOf(packs.ErrClientSeqNotSequential), connecthelper.CodeOf(err))
+}
+
+func RunPushPullWithClientSeqLessThanClientInfoTest(t *testing.T) {
+	ctx := context.Background()
+	be := setUpBackend(t)
+	project, _ := be.DB.FindProjectInfoByID(
+		ctx,
+		database.DefaultProjectID,
+	)
+
+	clientInfo, _ := clients.Activate(ctx, be.DB, project.ToProject(), clientID)
+
+	actorID, _ := time.ActorIDFromHex(clientID)
+	changePackFixture, _ :=
+		createChangePackFixture(helper.TestDocKey(t).String(), actorID.Bytes())
+
+	docInfo, _ := documents.FindDocInfoByKeyAndOwner(
+		ctx, be, clientInfo, changePackFixture.DocumentKey, true)
+	err := clientInfo.AttachDocument(docInfo.ID, changePackFixture.IsAttached())
+	if err != nil {
+		assert.Fail(t, "failed to attach document")
+	}
+
+	_, err = packs.PushPull(ctx, be, project.ToProject(),
+		clientInfo, docInfo, changePackFixture, packs.PushPullOptions{
+			Mode:   types.SyncModePushPull,
+			Status: document.StatusAttached,
+		})
+	if err != nil {
+		assert.Fail(t, "failed to push pull")
+	}
+
+	changePackWithClientSeqLessThanClientInfo, _ :=
+		createChangePackWithClientSeqLessThanClientInfo(helper.TestDocKey(t).String(), actorID.Bytes())
+	_, err = packs.PushPull(ctx, be, project.ToProject(), clientInfo, docInfo,
+		changePackWithClientSeqLessThanClientInfo, packs.PushPullOptions{
+			Mode:   types.SyncModePushPull,
+			Status: document.StatusAttached,
+		})
+
+	assert.NoError(t, err)
+}
+
+func RunPushPullWithServerSeqGreaterThanDocInfoTest(t *testing.T) {
+	ctx := context.Background()
+	be := setUpBackend(t)
+	project, _ := be.DB.FindProjectInfoByID(
+		ctx,
+		database.DefaultProjectID,
+	)
+
+	clientInfo, _ := clients.Activate(ctx, be.DB, project.ToProject(), clientID)
+
+	actorID, _ := time.ActorIDFromHex(clientID)
+	changePackFixture, _ :=
+		createChangePackFixture(helper.TestDocKey(t).String(), actorID.Bytes())
+
+	docInfo, _ := documents.FindDocInfoByKeyAndOwner(
+		ctx, be, clientInfo, changePackFixture.DocumentKey, true)
+	err := clientInfo.AttachDocument(docInfo.ID, changePackFixture.IsAttached())
+	if err != nil {
+		assert.Fail(t, "failed to attach document")
+	}
+
+	_, _ = packs.PushPull(ctx, be, project.ToProject(), clientInfo, docInfo,
+		changePackFixture, packs.PushPullOptions{
+			Mode:   types.SyncModePushPull,
+			Status: document.StatusAttached,
+		})
+
+	changePackWithServerSeqGreaterThanDocInfo, _ :=
+		createChangePackWithServerSeqGreaterThanDocInfo(helper.TestDocKey(t).String())
+
+	_, err = packs.PushPull(ctx, be, project.ToProject(),
+		clientInfo, docInfo, changePackWithServerSeqGreaterThanDocInfo, packs.PushPullOptions{
+			Mode:   types.SyncModePushPull,
+			Status: document.StatusAttached,
+		})
+
+	assert.Equal(t, connecthelper.CodeOf(packs.ErrInvalidServerSeq), connecthelper.CodeOf(err))
+}
+
+func createChangePackWithSequentialClientSeq(documentKey string, actorID []byte) (*change.Pack, error) {
+	return converter.FromChangePack(&api.ChangePack{
+		DocumentKey: documentKey,
+		Checkpoint:  &api.Checkpoint{ServerSeq: 0, ClientSeq: 2},
+		Changes: []*api.Change{
+			createChange(0, 0, actorID),
+			createChange(1, 1, actorID),
+			createChange(2, 2, actorID),
+		},
+	})
+}
+
+func createChangePackWithNotSequentialClientSeq(documentKey string, actorID []byte) (*change.Pack, error) {
+	return converter.FromChangePack(&api.ChangePack{
+		DocumentKey: documentKey,
+		Checkpoint:  &api.Checkpoint{ServerSeq: 0, ClientSeq: 0},
+		Changes: []*api.Change{
+			createChange(2, 2, actorID),
+			createChange(1, 1, actorID),
+			createChange(0, 0, actorID),
+		},
+	})
+}
+
+func createChangePackWithClientSeqLessThanClientInfo(documentKey string, actorID []byte) (*change.Pack, error) {
+	return converter.FromChangePack(&api.ChangePack{
+		DocumentKey: documentKey,
+		Checkpoint:  &api.Checkpoint{ServerSeq: 2, ClientSeq: 0},
+		Changes: []*api.Change{
+			createChange(0, 0, actorID),
+		},
+	})
+}
+
+func createChangePackFixture(documentKey string, actorID []byte) (*change.Pack, error) {
+	return createChangePackWithSequentialClientSeq(documentKey, actorID)
+}
+
+func createChangePackWithServerSeqGreaterThanDocInfo(documentKey string) (*change.Pack, error) {
+	return converter.FromChangePack(&api.ChangePack{
+		DocumentKey: documentKey,
+		Checkpoint:  &api.Checkpoint{ServerSeq: 1e9, ClientSeq: 2},
+	})
+}
+
+func createChange(clientSeq uint32, lamport int64, actorID []byte) *api.Change {
+	return &api.Change{
+		Id: &api.ChangeID{
+			ClientSeq: clientSeq,
+			Lamport:   lamport,
+			ActorId:   actorID,
+		},
+	}
+}
+
+func setUpBackend(
+	t *testing.T,
+) *backend.Backend {
+	conf := helper.TestConfig()
+
+	metrics, err := prometheus.NewMetrics()
+	assert.NoError(t, err)
+
+	be, err := backend.New(
+		conf.Backend,
+		conf.Mongo,
+		conf.Housekeeping,
+		metrics,
+	)
+	assert.NoError(t, err)
+
+	return be
+}

server/rpc/connecthelper/status.go

@@ -48,6 +48,7 @@ var errorToConnectCode = map[error]connect.Code{
 	clients.ErrInvalidClientKey:     connect.CodeInvalidArgument,
 	key.ErrInvalidKey:               connect.CodeInvalidArgument,
 	types.ErrEmptyProjectFields:     connect.CodeInvalidArgument,
+	packs.ErrClientSeqNotSequential: connect.CodeInvalidArgument,
 
 	// NotFound means the requested resource does not exist.
 	database.ErrProjectNotFound:  connect.CodeNotFound,
@@ -100,6 +101,7 @@ var errorToCode = map[error]string{
 	clients.ErrInvalidClientKey:     "ErrInvalidClientKey",
 	key.ErrInvalidKey:               "ErrInvalidKey",
 	types.ErrEmptyProjectFields:     "ErrEmptyProjectFields",
+	packs.ErrClientSeqNotSequential: "ErrClientSeqNotSequential",
 
 	database.ErrProjectNotFound:  "ErrProjectNotFound",
 	database.ErrClientNotFound:   "ErrClientNotFound",