diff --git a/core/block/editor/state/change.go b/core/block/editor/state/change.go index e227794dc8..f0e8e456b9 100644 --- a/core/block/editor/state/change.go +++ b/core/block/editor/state/change.go @@ -381,8 +381,54 @@ func (s *State) changeBlockUpdate(update *pb.ChangeBlockUpdate) error { } func (s *State) changeBlockMove(move *pb.ChangeBlockMove) error { + // A concurrent change could have moved the target inside one of the moved blocks. + // Applying such a move would create a cycle detached from the root, and apply() + // would garbage-collect both subtrees. Skip the move instead: the previously + // applied concurrent move wins, deterministically on every device. + if s.isAnyAncestorOf(move.Ids, move.TargetId) { + return fmt.Errorf("move target %s is inside moved blocks", move.TargetId) + } s.UnlinkAll(move.Ids) - return s.InsertTo(move.TargetId, move.Position, move.Ids...) + err := s.InsertTo(move.TargetId, move.Position, move.Ids...) + if err != nil { + // The target could have been removed by a concurrent change. The blocks are + // already unlinked, so without a fallback they would be garbage-collected by + // apply(). Reattach them to the end of the root instead of losing them. + existing := move.Ids[:0:0] + for _, id := range move.Ids { + if s.Exists(id) { + existing = append(existing, id) + } + } + if len(existing) == 0 { + return err + } + if fallbackErr := s.InsertTo("", model.Block_Inner, existing...); fallbackErr != nil { + return fmt.Errorf("reattach moved blocks to root: %w (original move error: %w)", fallbackErr, err) + } + } + return nil +} + +// isAnyAncestorOf reports whether blockId or any of its ancestors is one of ids +func (s *State) isAnyAncestorOf(ids []string, blockId string) bool { + visited := make(map[string]struct{}) + cur := blockId + for cur != "" { + if slice.FindPos(ids, cur) != -1 { + return true + } + if _, ok := visited[cur]; ok { + return false + } + visited[cur] = struct{}{} + parent := s.PickParentOf(cur) + if parent == nil { + return false + } + cur = parent.Model().Id + } + return false } func (s *State) changeStoreKeySet(set *pb.ChangeStoreKeySet) error { diff --git a/core/block/editor/state/change_test.go b/core/block/editor/state/change_test.go index 321bbf8a96..c70548a07c 100644 --- a/core/block/editor/state/change_test.go +++ b/core/block/editor/state/change_test.go @@ -297,6 +297,63 @@ func TestState_ChangesCreate_MoveAdd_Wrap(t *testing.T) { assert.Equal(t, d.(*State).String(), dc.String()) } +func TestState_ApplyChange_MoveConflicts(t *testing.T) { + newCrossMoveDoc := func() Doc { + return NewDoc("root", map[string]simple.Block{ + "root": simple.New(&model.Block{Id: "root", ChildrenIds: []string{"x", "y"}}), + "x": simple.New(&model.Block{Id: "x", ChildrenIds: []string{"x1"}}), + "x1": simple.New(&model.Block{Id: "x1"}), + "y": simple.New(&model.Block{Id: "y", ChildrenIds: []string{"y1"}}), + "y1": simple.New(&model.Block{Id: "y1"}), + }) + } + + reachable := func(s *State) map[string]bool { + found := map[string]bool{} + s.Iterate(func(b simple.Block) bool { + found[b.Model().Id] = true + return true + }) + return found + } + + t.Run("concurrent cross moves must not detach both subtrees", func(t *testing.T) { + // given + d := newCrossMoveDoc() + s := d.NewState() + + // when: replaying concurrent moves from two devices + s.ApplyChangeIgnoreErr(newMoveChange("y", model.Block_Inner, "x")) // device A: x under y + s.ApplyChangeIgnoreErr(newMoveChange("x", model.Block_Inner, "y")) // device B: y under x + _, _, err := ApplyState("", s, false) + + // then: all blocks are still reachable from root + require.NoError(t, err) + found := reachable(d.(*State)) + for _, id := range []string{"x", "x1", "y", "y1"} { + assert.True(t, found[id], "block %s must remain reachable", id) + } + }) + + t.Run("move into concurrently deleted target must not lose moved blocks", func(t *testing.T) { + // given + d := newCrossMoveDoc() + s := d.NewState() + + // when: device A deletes y, device B moves x under y + s.ApplyChangeIgnoreErr(newRemoveChange("y")) + s.ApplyChangeIgnoreErr(newMoveChange("y", model.Block_Inner, "x")) + _, _, err := ApplyState("", s, false) + + // then: x and its subtree are still reachable + require.NoError(t, err) + found := reachable(d.(*State)) + assert.True(t, found["x"], "x must remain reachable") + assert.True(t, found["x1"], "x1 must remain reachable") + assert.False(t, found["y"], "y was removed") + }) +} + func TestState_ChangesCreate_MoveAdd_Side(t *testing.T) { d := NewDoc("root", map[string]simple.Block{ "root": simple.New(&model.Block{Id: "root", ChildrenIds: []string{"a", "b"}}), diff --git a/core/block/editor/state/normalize.go b/core/block/editor/state/normalize.go index 87ecad932b..6915a5bb28 100644 --- a/core/block/editor/state/normalize.go +++ b/core/block/editor/state/normalize.go @@ -5,8 +5,10 @@ import ( "github.com/globalsign/mgo/bson" "github.com/gogo/protobuf/types" + "golang.org/x/exp/slices" "github.com/anyproto/anytype-heart/core/block/simple" + "github.com/anyproto/anytype-heart/core/domain" "github.com/anyproto/anytype-heart/pkg/lib/bundle" "github.com/anyproto/anytype-heart/pkg/lib/pb/model" "github.com/anyproto/anytype-heart/util/pbtypes" @@ -408,10 +410,7 @@ func (s *State) normalizeDetails() { // normalizeRecommendedRelations normalizes recommended relations of Type on state build level, because // these lists mustn't contain similar values, but could be updated by multiple clients independently func (s *State) normalizeRecommendedRelations() { - details := s.details - if details == nil && s.parent != nil { - details = s.parent.details - } + details := s.Details() if details == nil { return } @@ -420,10 +419,16 @@ func (s *State) normalizeRecommendedRelations() { recFeatRelations := details.GetStringList(bundle.RelationKeyRecommendedFeaturedRelations) recHiddenRelations := details.GetStringList(bundle.RelationKeyRecommendedHiddenRelations) - recHiddenRelations = slice.RemoveN(recHiddenRelations, recFeatRelations...) - recHiddenRelations = slice.RemoveN(recHiddenRelations, recRelations...) - recRelations = slice.RemoveN(recRelations, recFeatRelations...) + normalizedHidden := slice.RemoveN(recHiddenRelations, recFeatRelations...) + normalizedHidden = slice.RemoveN(normalizedHidden, recRelations...) + normalizedRec := slice.RemoveN(recRelations, recFeatRelations...) - details.SetStringList(bundle.RelationKeyRecommendedRelations, recRelations) - details.SetStringList(bundle.RelationKeyRecommendedHiddenRelations, recHiddenRelations) + // write through SetDetail to keep copy-on-write semantics: when details are + // inherited, the committed parent state must not be modified in place + if !slices.Equal(normalizedRec, recRelations) { + s.SetDetail(bundle.RelationKeyRecommendedRelations, domain.StringList(normalizedRec)) + } + if !slices.Equal(normalizedHidden, recHiddenRelations) { + s.SetDetail(bundle.RelationKeyRecommendedHiddenRelations, domain.StringList(normalizedHidden)) + } } diff --git a/core/block/editor/state/normalize_test.go b/core/block/editor/state/normalize_test.go index 4fcb9ffaaf..1b305f57c6 100644 --- a/core/block/editor/state/normalize_test.go +++ b/core/block/editor/state/normalize_test.go @@ -571,3 +571,22 @@ func TestNormalizeRecommendedRelations(t *testing.T) { assert.Equal(t, []string{"f1", "f2", "sfh", "fh", "sf"}, child.Details().GetStringList(bundle.RelationKeyRecommendedFeaturedRelations)) assert.Equal(t, []string{"h1", "h2", "h3"}, child.Details().GetStringList(bundle.RelationKeyRecommendedHiddenRelations)) } + +func TestNormalizeRecommendedRelationsDoesNotMutateParent(t *testing.T) { + // given: a committed state of a Type object with a key duplicated across lists + parent := NewDoc("root", nil).(*State) + parent.SetObjectTypeKey(bundle.TypeKeyObjectType) + parent.SetDetails(domain.NewDetailsFromMap(map[domain.RelationKey]domain.Value{ + bundle.RelationKeyRecommendedRelations: domain.StringList([]string{"a", "b"}), + bundle.RelationKeyRecommendedFeaturedRelations: domain.StringList([]string{"b"}), + })) + s := parent.NewState() + + // when: a derived state with no own details is normalized + require.NoError(t, s.Normalize(false)) + + // then: the derived state sees the normalized list + assert.Equal(t, []string{"a"}, s.Details().GetStringList(bundle.RelationKeyRecommendedRelations)) + // and the committed parent state is left untouched + assert.Equal(t, []string{"a", "b"}, parent.Details().GetStringList(bundle.RelationKeyRecommendedRelations)) +} diff --git a/core/block/source/sourceimpl/marshall_change_test.go b/core/block/source/sourceimpl/marshall_change_test.go index fc0680747a..e062e81926 100644 --- a/core/block/source/sourceimpl/marshall_change_test.go +++ b/core/block/source/sourceimpl/marshall_change_test.go @@ -57,6 +57,30 @@ func TestNewUnmarshalTreeChange(t *testing.T) { assert.Nil(t, res2.(*pb.Change).Snapshot) } +func TestNewUnmarshalTreeChange_KeepsSnapshotForSnapshotChanges(t *testing.T) { + // A snapshot change can arrive in an append batch where another change gets + // converted first. The unmarshalled result is cached as Change.Model and the raw + // data is discarded by objecttree, so if the snapshot is stripped here, a later + // in-memory tree reduce that makes this change the root leaves BuildState with + // a nil snapshot, and the object cannot be rebuilt until it is reopened. + + // given + first, firstDataType, err := MarshalChange(changeWithSmallTextUpdate()) + require.NoError(t, err) + snap, snapDataType, err := MarshalChange(changeWithBigSnapshot()) + require.NoError(t, err) + unmarshalF := NewUnmarshalTreeChange() + + // when: a regular change is converted first, then the snapshot change + _, err = unmarshalF(&objecttree.Change{DataType: firstDataType}, first) + require.NoError(t, err) + res, err := unmarshalF(&objecttree.Change{IsSnapshot: true, DataType: snapDataType}, snap) + require.NoError(t, err) + + // then: the snapshot data must survive unmarshalling + assert.NotNil(t, res.(*pb.Change).Snapshot) +} + func TestUnmarshallChange(t *testing.T) { invalidDataType := "invalid" diff --git a/core/block/source/sourceimpl/source.go b/core/block/source/sourceimpl/source.go index f716accd8e..c789b9e4d1 100644 --- a/core/block/source/sourceimpl/source.go +++ b/core/block/source/sourceimpl/source.go @@ -122,11 +122,19 @@ func unmarshalChange(treeChange *objecttree.Change, data []byte, needSnapshot bo } } -// NewUnmarshalTreeChange creates UnmarshalChange func that unmarshalls snapshot only for the first change and ignores it for following. It saves some memory +// NewUnmarshalTreeChange creates UnmarshalChange func that unmarshalls snapshot only for the first change +// and for snapshot changes, ignoring it for the rest. It saves some memory. +// Snapshot changes must keep their snapshot: the unmarshalled result is cached as Change.Model and the raw +// data is discarded by objecttree, so a snapshot change that later becomes the tree root after an in-memory +// reduce would otherwise be left without snapshot data, failing any subsequent state rebuild. func NewUnmarshalTreeChange() objecttree.ChangeConvertFunc { var changeCount atomic.Int32 return func(treeChange *objecttree.Change, data []byte) (result any, err error) { - return unmarshalChange(treeChange, data, changeCount.CompareAndSwap(0, 1)) + needSnapshot := changeCount.CompareAndSwap(0, 1) + if treeChange.IsSnapshot { + needSnapshot = true + } + return unmarshalChange(treeChange, data, needSnapshot) } } @@ -212,15 +220,18 @@ func (s *treeSource) Tree() objecttree.ObjectTree { func (s *treeSource) Update(ot objecttree.ObjectTree) error { // here it should work, because we always have the most common snapshot of the changes in tree s.lastSnapshotId = ot.Root().Id - prevSnapshot := s.lastSnapshotId - // todo: check this one err := s.receiver.StateAppend(func(d state.Doc) (st *state.State, changes []*pb.ChangeContent, err error) { + var ( + sinceSnapshot int + snapshotApplied bool + ) // State will be applied later in smartblock.StateAppend - st, changes, sinceSnapshot, err := BuildState(s.spaceID, d.(*state.State), ot, false) + st, changes, sinceSnapshot, snapshotApplied, err = BuildState(s.spaceID, d.(*state.State), ot, false) if err != nil { return } - if prevSnapshot != s.lastSnapshotId { + if snapshotApplied { + // the batch contained a snapshot change, so the count restarts from it s.changesSinceSnapshot = sinceSnapshot } else { s.changesSinceSnapshot += sinceSnapshot @@ -229,7 +240,8 @@ func (s *treeSource) Update(ot objecttree.ObjectTree) error { }) if err != nil { - log.With(zap.Error(err)).Debug("failed to append the state and send it to receiver") + // the live state is now behind the tree heads and will stay stale until the next update or reopen + log.With("objectID", s.id).With(zap.Error(err)).Error("failed to append the state and send it to receiver") } return nil } @@ -241,13 +253,14 @@ func (s *treeSource) Rebuild(ot objecttree.ObjectTree) error { doc, err := s.buildState() if err != nil { - log.With(zap.Error(err)).Debug("failed to build state") + // the live state is now behind the tree heads and will stay stale until the next update or reopen + log.With("objectID", s.id).With(zap.Error(err)).Error("failed to build state") return nil } st := doc.(*state.State) err = s.receiver.StateRebuild(st) if err != nil { - log.With(zap.Error(err)).Debug("failed to send the state to receiver") + log.With("objectID", s.id).With(zap.Error(err)).Error("failed to send the state to receiver") } return nil } @@ -284,7 +297,7 @@ func (s *treeSource) readDoc(receiver source.ChangeReceiver) (doc state.Doc, err } func (s *treeSource) buildState() (doc state.Doc, err error) { - st, _, changesAppliedSinceSnapshot, err := BuildState(s.spaceID, nil, s.ObjectTree, true) + st, _, changesAppliedSinceSnapshot, _, err := BuildState(s.spaceID, nil, s.ObjectTree, true) if err != nil { return } @@ -602,7 +615,10 @@ func cleanUpChange(objectId string, change *objecttree.Change, model *pb.Change) } } -func BuildState(spaceId string, initState *state.State, ot objecttree.ReadableObjectTree, applyState bool) (st *state.State, appliedContent []*pb.ChangeContent, changesAppliedSinceSnapshot int, err error) { +// BuildState builds or appends the state from the object tree. changesAppliedSinceSnapshot counts the changes +// applied after the last snapshot change; snapshotApplied reports whether a snapshot change was encountered, +// meaning the returned counter restarts from that snapshot instead of continuing the caller's count. +func BuildState(spaceId string, initState *state.State, ot objecttree.ReadableObjectTree, applyState bool) (st *state.State, appliedContent []*pb.ChangeContent, changesAppliedSinceSnapshot int, snapshotApplied bool, err error) { var ( startId string count int @@ -663,6 +679,7 @@ func BuildState(spaceId string, initState *state.State, ot objecttree.ReadableOb if startId == change.Id { if st == nil { changesAppliedSinceSnapshot = 0 + snapshotApplied = true st, iterErr = state.NewDocFromSnapshot(ot.Id(), model.Snapshot, state.WithChangeId(startId), state.WithInternalKey(uniqueKeyInternalKey)) if iterErr != nil { return false @@ -675,6 +692,7 @@ func BuildState(spaceId string, initState *state.State, ot objecttree.ReadableOb } if model.Snapshot != nil { changesAppliedSinceSnapshot = 0 + snapshotApplied = true } else { changesAppliedSinceSnapshot++ } diff --git a/core/block/source/sourceimpl/source_test.go b/core/block/source/sourceimpl/source_test.go index c7ca76b5ec..3e8aa9dc60 100644 --- a/core/block/source/sourceimpl/source_test.go +++ b/core/block/source/sourceimpl/source_test.go @@ -5,12 +5,20 @@ import ( "os" "testing" + "github.com/anyproto/any-sync/commonspace/object/tree/objecttree" + "github.com/anyproto/any-sync/commonspace/object/tree/objecttree/mock_objecttree" + "github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto" + "github.com/anyproto/any-sync/util/crypto" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + "github.com/anyproto/anytype-heart/core/block/editor/state" "github.com/anyproto/anytype-heart/core/block/source" "github.com/anyproto/anytype-heart/pb" "github.com/anyproto/anytype-heart/pkg/lib/bundle" "github.com/anyproto/anytype-heart/pkg/lib/pb/model" + "github.com/anyproto/anytype-heart/space/spacedomain" ) func Test_snapshotChance(t *testing.T) { @@ -50,6 +58,105 @@ func Test_snapshotChance2(t *testing.T) { // https://docs.google.com/spreadsheets/d/1xgH7fUxno5Rm-0VEaSD4LsTHeGeUXQFmHsOm29M6paI } +type stubChangeReceiver struct { + doc state.Doc +} + +func (r *stubChangeReceiver) StateAppend(f func(d state.Doc) (s *state.State, changes []*pb.ChangeContent, err error)) error { + _, _, err := f(r.doc) + return err +} + +func (r *stubChangeReceiver) StateRebuild(d state.Doc) error { + return nil +} + +func TestTreeSource_Update_ChangesSinceSnapshot(t *testing.T) { + newChange := func(id string, prev string, snapshot *pb.ChangeSnapshot) *objecttree.Change { + _, pub, err := crypto.GenerateRandomEd25519KeyPair() + require.NoError(t, err) + return &objecttree.Change{ + Id: id, + PreviousIds: []string{prev}, + Identity: pub, + Timestamp: 1, + Model: &pb.Change{Snapshot: snapshot}, + } + } + + newFixture := func(t *testing.T, changesSinceSnapshot int, batch []*objecttree.Change) *treeSource { + ctrl := gomock.NewController(t) + tree := mock_objecttree.NewMockObjectTree(ctrl) + + payload, err := (&model.ObjectChangePayload{SmartBlockType: model.SmartBlockType_Page}).Marshal() + require.NoError(t, err) + rootChange, err := (&treechangeproto.RootChange{ + ChangeType: spacedomain.ChangeType, + ChangePayload: payload, + }).MarshalVT() + require.NoError(t, err) + rawRoot, err := (&treechangeproto.RawTreeChange{Payload: rootChange}).MarshalVT() + require.NoError(t, err) + + tree.EXPECT().Header().Return(&treechangeproto.RawTreeChangeWithId{RawChange: rawRoot, Id: "treeId"}).AnyTimes() + tree.EXPECT().Id().Return("treeId").AnyTimes() + tree.EXPECT().Root().Return(&objecttree.Change{Id: "snapshotId"}).AnyTimes() + tree.EXPECT().IterateFrom("head0", gomock.Any(), gomock.Any()).DoAndReturn( + func(_ string, _ objecttree.ChangeConvertFunc, f objecttree.ChangeIterateFunc) error { + for _, ch := range batch { + if !f(ch) { + return nil + } + } + return nil + }).AnyTimes() + + doc := state.NewDoc("treeId", nil) + doc.(*state.State).SetChangeId("head0") + + return &treeSource{ + ObjectTree: tree, + id: "treeId", + spaceID: "space1", + receiver: &stubChangeReceiver{doc: doc}, + changesSinceSnapshot: changesSinceSnapshot, + } + } + + t.Run("batch with a snapshot change resets the counter", func(t *testing.T) { + // given + batch := []*objecttree.Change{ + newChange("head0", "treeId", nil), // start change, already applied + newChange("c1", "head0", nil), + newChange("c2", "c1", &pb.ChangeSnapshot{Data: &model.SmartBlockSnapshotBase{}}), + newChange("c3", "c2", nil), + } + src := newFixture(t, 5, batch) + + // when + require.NoError(t, src.Update(src.ObjectTree)) + + // then: only one change was applied after the snapshot + assert.Equal(t, 1, src.changesSinceSnapshot) + }) + + t.Run("batch without snapshot accumulates the counter", func(t *testing.T) { + // given + batch := []*objecttree.Change{ + newChange("head0", "treeId", nil), + newChange("c1", "head0", nil), + newChange("c2", "c1", nil), + } + src := newFixture(t, 5, batch) + + // when + require.NoError(t, src.Update(src.ObjectTree)) + + // then + assert.Equal(t, 7, src.changesSinceSnapshot) + }) +} + func TestSource_CheckChangeSize(t *testing.T) { t.Run("big change", func(t *testing.T) { // given diff --git a/core/debug/exporter/treeimporter.go b/core/debug/exporter/treeimporter.go index 019051e120..5677bc9a98 100644 --- a/core/debug/exporter/treeimporter.go +++ b/core/debug/exporter/treeimporter.go @@ -65,7 +65,7 @@ func (t *treeImporter) State() (*state.State, error) { err error ) - st, _, _, err = sourceimpl.BuildState("", nil, t.objectTree, true) + st, _, _, _, err = sourceimpl.BuildState("", nil, t.objectTree, true) if err != nil { return nil, err } diff --git a/core/history/history.go b/core/history/history.go index 83892c9d90..299b2089c2 100644 --- a/core/history/history.go +++ b/core/history/history.go @@ -573,7 +573,7 @@ func (h *history) buildState(id domain.FullID, versionId string) ( return } - st, _, _, err = sourceimpl.BuildState(id.SpaceID, nil, tree, true) + st, _, _, _, err = sourceimpl.BuildState(id.SpaceID, nil, tree, true) if err != nil { return }