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

Add headsync acl logic

Browse Source
This commit is contained in:
mcrakhman 2023-07-03 13:43:54 +02:00
parent 51ac955f1c
commit 145332b0f7
No known key found for this signature in database
GPG Key ID: DED12CFEF5B8396B
4 changed files with 46 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
commonspace/headsync/diffsyncer.go
View File

@ -3,10 +3,13 @@ package headsync
import (
"context"
"fmt"
"time"
"github.com/anyproto/any-sync/app/ldiff"
"github.com/anyproto/any-sync/app/logger"
"github.com/anyproto/any-sync/commonspace/credentialprovider"
"github.com/anyproto/any-sync/commonspace/deletionstate"
"github.com/anyproto/any-sync/commonspace/object/acl/syncacl"
"github.com/anyproto/any-sync/commonspace/object/treemanager"
"github.com/anyproto/any-sync/commonspace/peermanager"
"github.com/anyproto/any-sync/commonspace/spacestorage"
@ -14,8 +17,8 @@ import (
"github.com/anyproto/any-sync/commonspace/syncstatus"
"github.com/anyproto/any-sync/net/peer"
"github.com/anyproto/any-sync/net/rpc/rpcerr"
"github.com/anyproto/any-sync/util/slice"
"go.uber.org/zap"
"time"
)
type DiffSyncer interface {
@ -38,6 +41,7 @@ func newDiffSyncer(hs *headSync) DiffSyncer {
log: log,
syncStatus: hs.syncStatus,
deletionState: hs.deletionState,
syncAcl: hs.syncAcl,
}
}
@ -53,6 +57,7 @@ type diffSyncer struct {
credentialProvider credentialprovider.CredentialProvider
syncStatus syncstatus.StatusUpdater
treeSyncer treemanager.TreeSyncer
syncAcl *syncacl.SyncAcl
}
func (d *diffSyncer) Init() {
@ -116,6 +121,7 @@ func (d *diffSyncer) syncWithPeer(ctx context.Context, p peer.Peer) (err error)
cl = d.clientFactory.Client(conn)
rdiff = NewRemoteDiff(d.spaceId, cl)
stateCounter = d.syncStatus.StateCounter()
syncAclId = d.syncAcl.Id()
)
newIds, changedIds, removedIds, err := d.diff.Diff(ctx, rdiff)
@ -137,6 +143,16 @@ func (d *diffSyncer) syncWithPeer(ctx context.Context, p peer.Peer) (err error)
totalLen := len(newIds) + len(changedIds) + len(removedIds)
// not syncing ids which were removed through settings document
missingIds := d.deletionState.Filter(newIds)
prevLen := len(changedIds)
changedIds = slice.DiscardFromSlice(changedIds, func(s string) bool {
return s == syncAclId
})
// if acl head is different
if len(changedIds) < prevLen {
if syncErr := d.syncAcl.SyncWithPeer(ctx, p.Id()); syncErr != nil {
log.Warn("failed to send acl sync message to peer", zap.String("aclId", syncAclId))
}
}
existingIds := append(d.deletionState.Filter(removedIds), d.deletionState.Filter(changedIds)...)
d.syncStatus.RemoveAllExcept(p.Id(), existingIds, stateCounter)

13
commonspace/headsync/headsync.go
View File

@ -3,12 +3,16 @@ package headsync
import (
"context"
"sync/atomic"
"time"
"github.com/anyproto/any-sync/app"
"github.com/anyproto/any-sync/app/ldiff"
"github.com/anyproto/any-sync/app/logger"
config2 "github.com/anyproto/any-sync/commonspace/config"
"github.com/anyproto/any-sync/commonspace/credentialprovider"
"github.com/anyproto/any-sync/commonspace/deletionstate"
"github.com/anyproto/any-sync/commonspace/object/acl/syncacl"
"github.com/anyproto/any-sync/commonspace/object/treemanager"
"github.com/anyproto/any-sync/commonspace/peermanager"
"github.com/anyproto/any-sync/commonspace/spacestate"
@ -21,8 +25,6 @@ import (
"github.com/anyproto/any-sync/util/slice"
"go.uber.org/zap"
"golang.org/x/exp/slices"
"sync/atomic"
"time"
)
var log = logger.NewNamed(CName)
@ -60,6 +62,7 @@ type headSync struct {
credentialProvider credentialprovider.CredentialProvider
syncStatus syncstatus.StatusService
deletionState deletionstate.ObjectDeletionState
syncAcl *syncacl.SyncAcl
}
func New() HeadSync {
@ -71,6 +74,7 @@ var createDiffSyncer = newDiffSyncer
func (h *headSync) Init(a *app.App) (err error) {
shared := a.MustComponent(spacestate.CName).(*spacestate.SpaceState)
cfg := a.MustComponent("config").(config2.ConfigGetter)
h.syncAcl = a.MustComponent(syncacl.CName).(*syncacl.SyncAcl)
h.spaceId = shared.SpaceId
h.spaceIsDeleted = shared.SpaceIsDeleted
h.syncPeriod = cfg.GetSpace().SyncPeriod
@ -92,6 +96,7 @@ func (h *headSync) Init(a *app.App) (err error) {
return h.syncer.Sync(ctx)
}
h.periodicSync = periodicsync.NewPeriodicSync(h.syncPeriod, time.Minute, sync, h.log)
h.syncAcl.SetHeadUpdater(h)
// TODO: move to run?
h.syncer.Init()
return nil
@ -177,6 +182,10 @@ func (h *headSync) fillDiff(objectIds []string) {
Head: concatStrings(heads),
})
}
els = append(els, ldiff.Element{
Id: h.syncAcl.Id(),
Head: h.syncAcl.Head().Id,
})
h.diff.Set(els...)
if err := h.storage.WriteSpaceHash(h.diff.Hash()); err != nil {
h.log.Error("can't write space hash", zap.Error(err))

4
commonspace/object/acl/list/list.go
View File

@ -57,7 +57,7 @@ type AclList interface {
AddRawRecord(rawRec *consensusproto.RawRecordWithId) (err error)
AddRawRecords(rawRecords []*consensusproto.RawRecordWithId) (err error)
Close() (err error)
Close(ctx context.Context) (err error)
}
type aclList struct {
@ -293,6 +293,6 @@ func (a *aclList) IterateFrom(startId string, iterFunc IterFunc) {
}
}
func (a *aclList) Close() (err error) {
func (a *aclList) Close(ctx context.Context) (err error) {
return nil
}

17
commonspace/object/acl/syncacl/syncacl.go
View File

@ -29,17 +29,30 @@ func New() *SyncAcl {
return &SyncAcl{}
}
type HeadUpdater interface {
UpdateHeads(id string, heads []string)
}
type SyncAcl struct {
list.AclList
syncClient SyncClient
syncHandler synchandler.SyncHandler
headUpdater HeadUpdater
isClosed bool
}
func (s *SyncAcl) Run(ctx context.Context) (err error) {
return
}
func (s *SyncAcl) HandleRequest(ctx context.Context, senderId string, request *spacesyncproto.ObjectSyncMessage) (response *spacesyncproto.ObjectSyncMessage, err error) {
return s.HandleRequest(ctx, senderId, request)
}
func (s *SyncAcl) SetHeadUpdater(updater HeadUpdater) {
s.headUpdater = updater
}
func (s *SyncAcl) HandleMessage(ctx context.Context, senderId string, request *spacesyncproto.ObjectSyncMessage) (err error) {
return s.HandleMessage(ctx, senderId, request)
}
@ -73,6 +86,7 @@ func (s *SyncAcl) AddRawRecord(rawRec *consensusproto.RawRecordWithId) (err erro
return
}
headUpdate := s.syncClient.CreateHeadUpdate(s, []*consensusproto.RawRecordWithId{rawRec})
s.headUpdater.UpdateHeads(s.Id(), []string{rawRec.Id})
s.syncClient.Broadcast(headUpdate)
return
}
@ -86,6 +100,7 @@ func (s *SyncAcl) AddRawRecords(rawRecords []*consensusproto.RawRecordWithId) (e
return
}
headUpdate := s.syncClient.CreateHeadUpdate(s, rawRecords)
s.headUpdater.UpdateHeads(s.Id(), []string{rawRecords[len(rawRecords)-1].Id})
s.syncClient.Broadcast(headUpdate)
return
}
@ -97,7 +112,7 @@ func (s *SyncAcl) SyncWithPeer(ctx context.Context, peerId string) (err error) {
return s.syncClient.SendUpdate(peerId, s.Id(), headUpdate)
}
func (s *SyncAcl) Close() (err error) {
func (s *SyncAcl) Close(ctx context.Context) (err error) {
s.Lock()
defer s.Unlock()
s.isClosed = true