benjamyn/any-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update sync protocol to remove transaction add

Browse Source
This commit is contained in:
mcrakhman 2023-05-10 23:51:03 +02:00
parent 182b047dce
commit 8b3d8a4fef
No known key found for this signature in database
GPG Key ID: DED12CFEF5B8396B
7 changed files with 148 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
commonspace/object/tree/objecttree/objecttree.go
View File

@ -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
}
}

102
commonspace/object/tree/objecttree/objecttree_test.go
View File

@ -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,

2
commonspace/object/tree/objecttree/treebuilder.go
View File

@ -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
}

2
commonspace/object/tree/synctree/syncprotocol_test.go
View File

@ -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,

10
commonspace/object/tree/treestorage/inmemory.go
View File

@ -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

28
commonspace/object/tree/treestorage/mock_treestorage/mock_treestorage.go
View File

@ -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)
}

2
commonspace/object/tree/treestorage/treestorage.go
View File

@ -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)