From 8b3d8a4fefc31a0d2497bf71d8c5a57b9675ce14 Mon Sep 17 00:00:00 2001 From: mcrakhman Date: Wed, 10 May 2023 23:51:03 +0200 Subject: [PATCH] Update sync protocol to remove transaction add --- .../object/tree/objecttree/objecttree.go | 60 +++++------ .../object/tree/objecttree/objecttree_test.go | 102 +++++++++++++++++- .../object/tree/objecttree/treebuilder.go | 2 +- .../object/tree/synctree/syncprotocol_test.go | 2 +- .../object/tree/treestorage/inmemory.go | 10 +- .../mock_treestorage/mock_treestorage.go | 28 ++--- .../object/tree/treestorage/treestorage.go | 2 +- 7 files changed, 148 insertions(+), 58 deletions(-) diff --git a/commonspace/object/tree/objecttree/objecttree.go b/commonspace/object/tree/objecttree/objecttree.go index d75f76c1..88a24e0d 100644 --- a/commonspace/object/tree/objecttree/objecttree.go +++ b/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 { - if isOldSnapshot(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 + for _, ch := range ot.newChangesBuf { + if snapshotNotInTree(ch) { 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 } } diff --git a/commonspace/object/tree/objecttree/objecttree_test.go b/commonspace/object/tree/objecttree/objecttree_test.go index 9dc5e7be..466a79fd 100644 --- a/commonspace/object/tree/objecttree/objecttree_test.go +++ b/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, diff --git a/commonspace/object/tree/objecttree/treebuilder.go b/commonspace/object/tree/objecttree/treebuilder.go index 43e23042..0abb136b 100644 --- a/commonspace/object/tree/objecttree/treebuilder.go +++ b/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 } diff --git a/commonspace/object/tree/synctree/syncprotocol_test.go b/commonspace/object/tree/synctree/syncprotocol_test.go index 6cac042a..d9e1963a 100644 --- a/commonspace/object/tree/synctree/syncprotocol_test.go +++ b/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, diff --git a/commonspace/object/tree/treestorage/inmemory.go b/commonspace/object/tree/treestorage/inmemory.go index af40561d..4cbb76c7 100644 --- a/commonspace/object/tree/treestorage/inmemory.go +++ b/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 diff --git a/commonspace/object/tree/treestorage/mock_treestorage/mock_treestorage.go b/commonspace/object/tree/treestorage/mock_treestorage/mock_treestorage.go index 2fe6029f..9c99c746 100644 --- a/commonspace/object/tree/treestorage/mock_treestorage/mock_treestorage.go +++ b/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) -} diff --git a/commonspace/object/tree/treestorage/treestorage.go b/commonspace/object/tree/treestorage/treestorage.go index 14fbad78..86a83cc1 100644 --- a/commonspace/object/tree/treestorage/treestorage.go +++ b/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)