Update sync protocol to remove transaction add

commonspace/object/tree/objecttree/objecttree.go

@@ -116,13 +116,25 @@ type objectTree struct {
 }
 
 func (ot *objectTree) rebuildFromStorage(theirHeads []string, newChanges []*Change) (err error) {
+	oldTree := ot.tree
 	ot.treeBuilder.Reset()
-
 	ot.tree, err = ot.treeBuilder.Build(theirHeads, newChanges)
 	if err != nil {
 		return
 	}
 
+	// in case there are new heads
+	if theirHeads != nil && oldTree != nil {
+		// checking that old root is still in tree
+		rootCh, rootExists := ot.tree.attached[oldTree.RootId()]
+
+		// checking the case where theirHeads were actually below prevHeads
+		// so if we did load some extra data in the tree, let's reduce it to old root
+		if slice.UnsortedEquals(oldTree.headIds, ot.tree.headIds) && rootExists && ot.tree.RootId() != oldTree.RootId() {
+			ot.tree.makeRootAndRemove(rootCh)
+		}
+	}
+
 	// during building the tree we may have marked some changes as possible roots,
 	// but obviously they are not roots, because of the way how we construct the tree
 	ot.tree.clearPossibleRoots()
@@ -187,7 +199,7 @@ func (ot *objectTree) AddContent(ctx context.Context, content SignableChangeCont
 		panic(err)
 	}
 
-	err = ot.treeStorage.TransactionAdd([]*treechangeproto.RawTreeChangeWithId{rawChange}, []string{objChange.Id})
+	err = ot.treeStorage.AddMany([]*treechangeproto.RawTreeChangeWithId{rawChange}, []string{objChange.Id})
 	if err != nil {
 		return
 	}
@@ -284,7 +296,11 @@ func (ot *objectTree) AddRawChanges(ctx context.Context, changesPayload RawChang
 		addResult.Mode = Rebuild
 	}
 
-	err = ot.treeStorage.TransactionAdd(addResult.Added, addResult.Heads)
+	err = ot.treeStorage.AddMany(addResult.Added, addResult.Heads)
+	if err != nil {
+		// rolling back all changes made to inmemory state
+		ot.rebuildFromStorage(nil, nil)
+	}
 	return
 }
 
@@ -347,7 +363,7 @@ func (ot *objectTree) addRawChanges(ctx context.Context, changesPayload RawChang
 	}
 
 	// checks if we need to go to database
-	isOldSnapshot := func(ch *Change) bool {
+	snapshotNotInTree := func(ch *Change) bool {
 		if ch.SnapshotId == ot.tree.RootId() {
 			return false
 		}
@@ -362,26 +378,12 @@ func (ot *objectTree) addRawChanges(ctx context.Context, changesPayload RawChang
 
 	shouldRebuildFromStorage := false
 	// checking if we have some changes with different snapshot and then rebuilding
-	for idx, ch := range ot.newChangesBuf {
+	for _, ch := range ot.newChangesBuf {
-		if isOldSnapshot(ch) {
+		if snapshotNotInTree(ch) {
-			var exists bool
-			// checking if it exists in the storage, if yes, then at some point it was added to the tree
-			// thus we don't need to look at this change
-			exists, err = ot.treeStorage.HasChange(ctx, ch.Id)
-			if err != nil {
-				return
-			}
-			if exists {
-				// marking as nil to delete after
-				ot.newChangesBuf[idx] = nil
-				continue
-			}
-			// we haven't seen the change, and it refers to old snapshot, so we should rebuild
 			shouldRebuildFromStorage = true
+			break
 		}
 	}
-	// discarding all previously seen changes
-	ot.newChangesBuf = slice.DiscardFromSlice(ot.newChangesBuf, func(ch *Change) bool { return ch == nil })
 
 	if shouldRebuildFromStorage {
 		err = ot.rebuildFromStorage(changesPayload.NewHeads, ot.newChangesBuf)
@@ -549,22 +551,8 @@ func (ot *objectTree) IterateFrom(id string, convert ChangeConvertFunc, iterate
 }
 
 func (ot *objectTree) HasChanges(chs ...string) bool {
-	hasChange := func(s string) bool {
-		_, attachedExists := ot.tree.attached[s]
-		if attachedExists {
-			return attachedExists
-		}
-
-		has, err := ot.treeStorage.HasChange(context.Background(), s)
-		if err != nil {
-			return false
-		}
-
-		return has
-	}
-
 	for _, ch := range chs {
-		if !hasChange(ch) {
+		if _, attachedExists := ot.tree.attached[ch]; !attachedExists {
 			return false
 		}
 	}

commonspace/object/tree/objecttree/objecttree_test.go

@@ -2,6 +2,7 @@ package objecttree
 
 import (
 	"context"
+	"fmt"
 	"github.com/anytypeio/any-sync/commonspace/object/accountdata"
 	"github.com/anytypeio/any-sync/commonspace/object/acl/list"
 	"github.com/anytypeio/any-sync/commonspace/object/tree/treechangeproto"
@@ -64,7 +65,7 @@ func prepareContext(
 	treeStorage := changeCreator.CreateNewTreeStorage("0", aclList.Head().Id)
 	if additionalChanges != nil {
 		payload := additionalChanges(changeCreator)
-		err := treeStorage.TransactionAdd(payload.RawChanges, payload.NewHeads)
+		err := treeStorage.AddMany(payload.RawChanges, payload.NewHeads)
 		require.NoError(t, err)
 	}
 	objTree, err := objTreeBuilder(treeStorage, aclList)
@@ -421,6 +422,99 @@ func TestObjectTree(t *testing.T) {
 		})
 	})
 
+	t.Run("rollback when add to storage returns error", func(t *testing.T) {
+		ctx := prepareTreeContext(t, aclList)
+		changeCreator := ctx.changeCreator
+		objTree := ctx.objTree
+		store := ctx.treeStorage.(*treestorage.InMemoryTreeStorage)
+		addErr := fmt.Errorf("error saving")
+		store.SetReturnErrorOnAdd(addErr)
+
+		rawChanges := []*treechangeproto.RawTreeChangeWithId{
+			changeCreator.CreateRaw("1", aclList.Head().Id, "0", true, "0"),
+		}
+		payload := RawChangesPayload{
+			NewHeads:   []string{"1"},
+			RawChanges: rawChanges,
+		}
+		_, err := objTree.AddRawChanges(context.Background(), payload)
+		require.Error(t, err, addErr)
+		require.Equal(t, "0", objTree.Root().Id)
+	})
+
+	t.Run("their heads before common snapshot", func(t *testing.T) {
+		// checking that adding old changes did not affect the tree
+		ctx := prepareTreeContext(t, aclList)
+		changeCreator := ctx.changeCreator
+		objTree := ctx.objTree
+
+		rawChanges := []*treechangeproto.RawTreeChangeWithId{
+			changeCreator.CreateRaw("1", aclList.Head().Id, "0", true, "0"),
+			changeCreator.CreateRaw("2", aclList.Head().Id, "1", false, "1"),
+			changeCreator.CreateRaw("3", aclList.Head().Id, "1", true, "2"),
+			changeCreator.CreateRaw("4", aclList.Head().Id, "1", false, "2"),
+			changeCreator.CreateRaw("5", aclList.Head().Id, "1", false, "1"),
+			changeCreator.CreateRaw("6", aclList.Head().Id, "1", true, "3", "4", "5"),
+		}
+		payload := RawChangesPayload{
+			NewHeads:   []string{rawChanges[len(rawChanges)-1].Id},
+			RawChanges: rawChanges,
+		}
+		_, err := objTree.AddRawChanges(context.Background(), payload)
+		require.NoError(t, err, "adding changes should be without error")
+		require.Equal(t, "6", objTree.Root().Id)
+
+		rawChangesPrevious := []*treechangeproto.RawTreeChangeWithId{
+			changeCreator.CreateRaw("1", aclList.Head().Id, "0", true, "0"),
+		}
+		payload = RawChangesPayload{
+			NewHeads:   []string{"1"},
+			RawChanges: rawChangesPrevious,
+		}
+		_, err = objTree.AddRawChanges(context.Background(), payload)
+		require.NoError(t, err, "adding changes should be without error")
+		require.Equal(t, "6", objTree.Root().Id)
+	})
+
+	t.Run("stored changes will not break the pipeline if heads were not updated", func(t *testing.T) {
+		ctx := prepareTreeContext(t, aclList)
+		changeCreator := ctx.changeCreator
+		objTree := ctx.objTree
+		store := ctx.treeStorage.(*treestorage.InMemoryTreeStorage)
+
+		rawChanges := []*treechangeproto.RawTreeChangeWithId{
+			changeCreator.CreateRaw("1", aclList.Head().Id, "0", true, "0"),
+		}
+		payload := RawChangesPayload{
+			NewHeads:   []string{rawChanges[len(rawChanges)-1].Id},
+			RawChanges: rawChanges,
+		}
+		_, err := objTree.AddRawChanges(context.Background(), payload)
+		require.NoError(t, err, "adding changes should be without error")
+		require.Equal(t, "1", objTree.Root().Id)
+
+		// creating changes to save in the storage
+		// to imitate the condition where all changes are in the storage
+		// but the head was not updated
+		storageChanges := []*treechangeproto.RawTreeChangeWithId{
+			changeCreator.CreateRaw("2", aclList.Head().Id, "1", false, "1"),
+			changeCreator.CreateRaw("3", aclList.Head().Id, "1", true, "2"),
+			changeCreator.CreateRaw("4", aclList.Head().Id, "1", false, "2"),
+			changeCreator.CreateRaw("5", aclList.Head().Id, "1", false, "1"),
+			changeCreator.CreateRaw("6", aclList.Head().Id, "1", true, "3", "4", "5"),
+		}
+		store.AddMany(storageChanges, []string{"1"})
+
+		// updating with subset of those changes to see that everything will still work
+		payload = RawChangesPayload{
+			NewHeads:   []string{"6"},
+			RawChanges: storageChanges,
+		}
+		_, err = objTree.AddRawChanges(context.Background(), payload)
+		require.NoError(t, err, "adding changes should be without error")
+		require.Equal(t, "6", objTree.Root().Id)
+	})
+
 	t.Run("changes from tree after common snapshot complex", func(t *testing.T) {
 		ctx := prepareTreeContext(t, aclList)
 		changeCreator := ctx.changeCreator
@@ -654,7 +748,7 @@ func TestObjectTree(t *testing.T) {
 			changeCreator.CreateRaw("5", aclList.Head().Id, "0", false, "1"),
 			changeCreator.CreateRaw("6", aclList.Head().Id, "0", false, "3", "4", "5"),
 		}
-		deps.treeStorage.TransactionAdd(rawChanges, []string{"6"})
+		deps.treeStorage.AddMany(rawChanges, []string{"6"})
 		hTree, err := buildHistoryTree(deps, HistoryTreeParams{
 			BeforeId:        "6",
 			IncludeBeforeId: false,
@@ -686,7 +780,7 @@ func TestObjectTree(t *testing.T) {
 			changeCreator.CreateRaw("5", aclList.Head().Id, "1", true, "3", "4"),
 			changeCreator.CreateRaw("6", aclList.Head().Id, "5", false, "5"),
 		}
-		deps.treeStorage.TransactionAdd(rawChanges, []string{"6"})
+		deps.treeStorage.AddMany(rawChanges, []string{"6"})
 		hTree, err := buildHistoryTree(deps, HistoryTreeParams{
 			BuildFullTree: true,
 		})
@@ -716,7 +810,7 @@ func TestObjectTree(t *testing.T) {
 			changeCreator.CreateRaw("5", aclList.Head().Id, "0", false, "1"),
 			changeCreator.CreateRaw("6", aclList.Head().Id, "0", false, "3", "4", "5"),
 		}
-		deps.treeStorage.TransactionAdd(rawChanges, []string{"6"})
+		deps.treeStorage.AddMany(rawChanges, []string{"6"})
 		hTree, err := buildHistoryTree(deps, HistoryTreeParams{
 			BeforeId:        "6",
 			IncludeBeforeId: true,

commonspace/object/tree/objecttree/treebuilder.go

@@ -121,8 +121,8 @@ func (tb *treeBuilder) buildTree(heads []string, breakpoint string) (err error)
 	if err != nil {
 		return
 	}
-	tb.tree.AddFast(ch)
 	changes := tb.dfs(heads, breakpoint)
+	tb.tree.AddFast(ch)
 	tb.tree.AddFast(changes...)
 	return
 }

commonspace/object/tree/synctree/syncprotocol_test.go

@@ -112,7 +112,7 @@ func testTreeMerge(t *testing.T, levels, perLevel int, hasData bool, isSnapshot
 	}
 	// generating initial tree
 	initialRes := genChanges(changeCreator, params)
-	err = storage.TransactionAdd(initialRes.changes, initialRes.heads)
+	err = storage.AddMany(initialRes.changes, initialRes.heads)
 	require.NoError(t, err)
 	deps := fixtureDeps{
 		aclList:     aclList,

commonspace/object/tree/treestorage/inmemory.go

@@ -13,13 +13,21 @@ type InMemoryTreeStorage struct {
 	root    *treechangeproto.RawTreeChangeWithId
 	heads   []string
 	Changes map[string]*treechangeproto.RawTreeChangeWithId
+	addErr  error
 
 	sync.RWMutex
 }
 
-func (t *InMemoryTreeStorage) TransactionAdd(changes []*treechangeproto.RawTreeChangeWithId, heads []string) error {
+func (t *InMemoryTreeStorage) SetReturnErrorOnAdd(err error) {
+	t.addErr = err
+}
+
+func (t *InMemoryTreeStorage) AddMany(changes []*treechangeproto.RawTreeChangeWithId, heads []string) error {
 	t.RLock()
 	defer t.RUnlock()
+	if t.addErr != nil {
+		return t.addErr
+	}
 
 	for _, ch := range changes {
 		t.Changes[ch.Id] = ch

commonspace/object/tree/treestorage/mock_treestorage/mock_treestorage.go

@@ -35,6 +35,20 @@ func (m *MockTreeStorage) EXPECT() *MockTreeStorageMockRecorder {
 	return m.recorder
 }
 
+// AddMany mocks base method.
+func (m *MockTreeStorage) AddMany(arg0 []*treechangeproto.RawTreeChangeWithId, arg1 []string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "AddMany", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// AddMany indicates an expected call of AddMany.
+func (mr *MockTreeStorageMockRecorder) AddMany(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddMany", reflect.TypeOf((*MockTreeStorage)(nil).AddMany), arg0, arg1)
+}
+
 // AddRawChange mocks base method.
 func (m *MockTreeStorage) AddRawChange(arg0 *treechangeproto.RawTreeChangeWithId) error {
 	m.ctrl.T.Helper()
@@ -150,17 +164,3 @@ func (mr *MockTreeStorageMockRecorder) SetHeads(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetHeads", reflect.TypeOf((*MockTreeStorage)(nil).SetHeads), arg0)
 }
-
-// TransactionAdd mocks base method.
-func (m *MockTreeStorage) TransactionAdd(arg0 []*treechangeproto.RawTreeChangeWithId, arg1 []string) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "TransactionAdd", arg0, arg1)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// TransactionAdd indicates an expected call of TransactionAdd.
-func (mr *MockTreeStorageMockRecorder) TransactionAdd(arg0, arg1 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TransactionAdd", reflect.TypeOf((*MockTreeStorage)(nil).TransactionAdd), arg0, arg1)
-}

commonspace/object/tree/treestorage/treestorage.go

@@ -31,7 +31,7 @@ type TreeStorage interface {
 	Heads() ([]string, error)
 	SetHeads(heads []string) error
 	AddRawChange(change *treechangeproto.RawTreeChangeWithId) error
-	TransactionAdd(changes []*treechangeproto.RawTreeChangeWithId, heads []string) error
+	AddMany(changes []*treechangeproto.RawTreeChangeWithId, heads []string) error
 
 	GetRawChange(ctx context.Context, id string) (*treechangeproto.RawTreeChangeWithId, error)
 	HasChange(ctx context.Context, id string) (bool, error)