1
0
Fork 0
mirror of https://github.com/anyproto/any-sync.git synced 2025-06-08 14:07:02 +09:00

Change deletion logic in headsync and objectsync

This commit is contained in:
mcrakhman 2023-02-23 23:58:09 +01:00 committed by Mikhail Iudin
parent 73b27c7dac
commit a0aa02bb8a
No known key found for this signature in database
GPG key ID: FAAAA8BAABDFF1C0
4 changed files with 28 additions and 13 deletions

View file

@ -11,6 +11,7 @@ import (
"github.com/anytypeio/any-sync/commonspace/spacestorage" "github.com/anytypeio/any-sync/commonspace/spacestorage"
"github.com/anytypeio/any-sync/commonspace/spacesyncproto" "github.com/anytypeio/any-sync/commonspace/spacesyncproto"
"github.com/anytypeio/any-sync/commonspace/syncstatus" "github.com/anytypeio/any-sync/commonspace/syncstatus"
"github.com/anytypeio/any-sync/nodeconf"
"github.com/anytypeio/any-sync/util/periodicsync" "github.com/anytypeio/any-sync/util/periodicsync"
"go.uber.org/zap" "go.uber.org/zap"
"strings" "strings"
@ -51,6 +52,7 @@ func NewHeadSync(
spaceId string, spaceId string,
spaceIsDeleted *atomic.Bool, spaceIsDeleted *atomic.Bool,
syncPeriod int, syncPeriod int,
configuration nodeconf.Configuration,
storage spacestorage.SpaceStorage, storage spacestorage.SpaceStorage,
peerManager peermanager.PeerManager, peerManager peermanager.PeerManager,
cache treegetter.TreeGetter, cache treegetter.TreeGetter,
@ -62,7 +64,7 @@ func NewHeadSync(
factory := spacesyncproto.ClientFactoryFunc(spacesyncproto.NewDRPCSpaceSyncClient) factory := spacesyncproto.ClientFactoryFunc(spacesyncproto.NewDRPCSpaceSyncClient)
syncer := newDiffSyncer(spaceId, diff, peerManager, cache, storage, factory, syncStatus, l) syncer := newDiffSyncer(spaceId, diff, peerManager, cache, storage, factory, syncStatus, l)
sync := func(ctx context.Context) (err error) { sync := func(ctx context.Context) (err error) {
if spaceIsDeleted.Load() { if spaceIsDeleted.Load() && !configuration.IsResponsible(spaceId) {
return spacesyncproto.ErrSpaceIsDeleted return spacesyncproto.ErrSpaceIsDeleted
} }
return syncer.Sync(ctx) return syncer.Sync(ctx)
@ -88,10 +90,6 @@ func (d *headSync) Init(objectIds []string, deletionState settingsstate.ObjectDe
} }
func (d *headSync) HandleRangeRequest(ctx context.Context, req *spacesyncproto.HeadSyncRequest) (resp *spacesyncproto.HeadSyncResponse, err error) { func (d *headSync) HandleRangeRequest(ctx context.Context, req *spacesyncproto.HeadSyncRequest) (resp *spacesyncproto.HeadSyncResponse, err error) {
if d.spaceIsDeleted.Load() {
err = spacesyncproto.ErrSpaceIsDeleted
return
}
return HandleRangeRequest(ctx, d.diff, req) return HandleRangeRequest(ctx, d.diff, req)
} }

View file

@ -8,7 +8,9 @@ import (
"github.com/anytypeio/any-sync/commonspace/objectsync/synchandler" "github.com/anytypeio/any-sync/commonspace/objectsync/synchandler"
"github.com/anytypeio/any-sync/commonspace/peermanager" "github.com/anytypeio/any-sync/commonspace/peermanager"
"github.com/anytypeio/any-sync/commonspace/spacesyncproto" "github.com/anytypeio/any-sync/commonspace/spacesyncproto"
"github.com/anytypeio/any-sync/nodeconf"
"go.uber.org/zap" "go.uber.org/zap"
"golang.org/x/exp/slices"
"sync/atomic" "sync/atomic"
"time" "time"
) )
@ -27,8 +29,9 @@ type ObjectSync interface {
type objectSync struct { type objectSync struct {
spaceId string spaceId string
messagePool MessagePool messagePool MessagePool
objectGetter syncobjectgetter.SyncObjectGetter objectGetter syncobjectgetter.SyncObjectGetter
configuration nodeconf.Configuration
syncCtx context.Context syncCtx context.Context
cancelSync context.CancelFunc cancelSync context.CancelFunc
@ -38,12 +41,14 @@ type objectSync struct {
func NewObjectSync( func NewObjectSync(
spaceId string, spaceId string,
spaceIsDeleted *atomic.Bool, spaceIsDeleted *atomic.Bool,
configuration nodeconf.Configuration,
peerManager peermanager.PeerManager, peerManager peermanager.PeerManager,
objectGetter syncobjectgetter.SyncObjectGetter) ObjectSync { objectGetter syncobjectgetter.SyncObjectGetter) ObjectSync {
syncCtx, cancel := context.WithCancel(context.Background()) syncCtx, cancel := context.WithCancel(context.Background())
os := newObjectSync( os := newObjectSync(
spaceId, spaceId,
spaceIsDeleted, spaceIsDeleted,
configuration,
objectGetter, objectGetter,
syncCtx, syncCtx,
cancel) cancel)
@ -55,6 +60,7 @@ func NewObjectSync(
func newObjectSync( func newObjectSync(
spaceId string, spaceId string,
spaceIsDeleted *atomic.Bool, spaceIsDeleted *atomic.Bool,
configuration nodeconf.Configuration,
objectGetter syncobjectgetter.SyncObjectGetter, objectGetter syncobjectgetter.SyncObjectGetter,
syncCtx context.Context, syncCtx context.Context,
cancel context.CancelFunc, cancel context.CancelFunc,
@ -65,6 +71,7 @@ func newObjectSync(
syncCtx: syncCtx, syncCtx: syncCtx,
cancelSync: cancel, cancelSync: cancel,
spaceIsDeleted: spaceIsDeleted, spaceIsDeleted: spaceIsDeleted,
configuration: configuration,
} }
} }
@ -85,8 +92,13 @@ func (s *objectSync) HandleMessage(ctx context.Context, senderId string, message
} }
func (s *objectSync) handleMessage(ctx context.Context, senderId string, msg *spacesyncproto.ObjectSyncMessage) (err error) { func (s *objectSync) handleMessage(ctx context.Context, senderId string, msg *spacesyncproto.ObjectSyncMessage) (err error) {
log := log.With(zap.String("objectId", msg.ObjectId), zap.String("replyId", msg.ReplyId))
if s.spaceIsDeleted.Load() { if s.spaceIsDeleted.Load() {
return spacesyncproto.ErrSpaceIsDeleted log = log.With(zap.Bool("isDeleted", true))
// preventing sync with other clients
if !slices.Contains(s.configuration.NodeIds(s.spaceId), senderId) {
return spacesyncproto.ErrSpaceIsDeleted
}
} }
log.With(zap.String("objectId", msg.ObjectId), zap.String("replyId", msg.ReplyId)).DebugCtx(ctx, "handling message") log.With(zap.String("objectId", msg.ObjectId), zap.String("replyId", msg.ReplyId)).DebugCtx(ctx, "handling message")
obj, err := s.objectGetter.GetObject(ctx, msg.ObjectId) obj, err := s.objectGetter.GetObject(ctx, msg.ObjectId)

View file

@ -9,6 +9,7 @@ import (
type SpaceIdsProvider interface { type SpaceIdsProvider interface {
AllIds() []string AllIds() []string
RemoveObjects(ids []string)
} }
type SpaceDeleter interface { type SpaceDeleter interface {
@ -60,13 +61,17 @@ func (d *deletionManager) UpdateState(state *settingsstate.State) (err error) {
if ok { if ok {
spaceDeleter.DeleteSpace(d.spaceId) spaceDeleter.DeleteSpace(d.spaceId)
} }
d.onSpaceDelete()
if d.isResponsible { if d.isResponsible {
allIds := slice.DiscardFromSlice(d.provider.AllIds(), func(s string) bool { allIds := slice.DiscardFromSlice(d.provider.AllIds(), func(id string) bool {
return s == d.settingsId return id == d.settingsId
}) })
err = d.deletionState.Add(allIds) err = d.deletionState.Add(allIds)
if err != nil {
return
}
d.provider.RemoveObjects(allIds)
} }
d.onSpaceDelete()
} }
return return
} }

View file

@ -139,8 +139,8 @@ func (s *spaceService) NewSpace(ctx context.Context, id string) (Space, error) {
return nil, err return nil, err
} }
headSync := headsync.NewHeadSync(id, spaceIsDeleted, s.config.SyncPeriod, st, peerManager, getter, syncStatus, log) headSync := headsync.NewHeadSync(id, spaceIsDeleted, s.config.SyncPeriod, lastConfiguration, st, peerManager, getter, syncStatus, log)
objectSync := objectsync.NewObjectSync(id, spaceIsDeleted, peerManager, getter) objectSync := objectsync.NewObjectSync(id, spaceIsDeleted, lastConfiguration, peerManager, getter)
sp := &space{ sp := &space{
id: id, id: id,
objectSync: objectSync, objectSync: objectSync,