anyproto/any-sync
1
0
Fork 0
mirror of https://github.com/anyproto/any-sync.git synced 2025-06-08 05:57:03 +09:00
Code Activity

Add last write

Browse source
This commit is contained in:
mcrakhman 2025-02-19 20:29:49 +01:00
parent 48a67d9e71
commit 649e5a13a8
No known key found for this signature in database
GPG key ID: DED12CFEF5B8396B
2 changed files with 8 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

8
commonspace/headsync/statestorage/statestorage.go
View file

@ -60,6 +60,9 @@ func (s *stateStorage) SetObserver(observer Observer) {
func (s *stateStorage) SetHash(ctx context.Context, hash string) (err error) { func (s *stateStorage) SetHash(ctx context.Context, hash string) (err error) {
defer func() { defer func() {
if last, ok := s.store.(lastWriteDb); ok {
last.SetLastWrite()
}
if s.observer != nil && err == nil { if s.observer != nil && err == nil {
s.observer.OnHashChange(hash) s.observer.OnHashChange(hash)
} }
@ -80,6 +83,11 @@ func (s *stateStorage) SetHash(ctx context.Context, hash string) (err error) {
return tx.Commit() return tx.Commit()
} }
type lastWriteDb interface {
SetLastWrite()
anystore.DB
}
func New(ctx context.Context, spaceId string, store anystore.DB) (StateStorage, error) { func New(ctx context.Context, spaceId string, store anystore.DB) (StateStorage, error) {
stateCollection, err := store.Collection(ctx, stateCollectionKey) stateCollection, err := store.Collection(ctx, stateCollectionKey)
if err != nil { if err != nil {

20
commonspace/object/tree/objecttree/storage.go
View file

@ -4,13 +4,11 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"sync"
"time" "time"
anystore "github.com/anyproto/any-store" anystore "github.com/anyproto/any-store"
"github.com/anyproto/any-store/anyenc" "github.com/anyproto/any-store/anyenc"
"github.com/anyproto/any-store/query" "github.com/anyproto/any-store/query"
"go.uber.org/atomic"
"github.com/anyproto/any-sync/commonspace/headsync/headstorage" "github.com/anyproto/any-sync/commonspace/headsync/headstorage"
"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto" "github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
@ -208,11 +206,6 @@ func (s *storage) GetAfterOrder(ctx context.Context, orderId string, storageIter
return nil return nil
} }
var (
lastCheckpoint = atomic.Time{}
lock = sync.Mutex{}
)
func (s *storage) AddAll(ctx context.Context, changes []StorageChange, heads []string, commonSnapshot string) error { func (s *storage) AddAll(ctx context.Context, changes []StorageChange, heads []string, commonSnapshot string) error {
arena := s.arena arena := s.arena
defer arena.Reset() defer arena.Reset()
@ -229,19 +222,6 @@ func (s *storage) AddAll(ctx context.Context, changes []StorageChange, heads []s
tx.Rollback() tx.Rollback()
return err return err
} }
now := time.Now()
checkpoint := lastCheckpoint.Load()
if now.Sub(checkpoint) > time.Second {
lock.Lock()
checkpoint := lastCheckpoint.Load()
now = time.Now()
if now.Sub(checkpoint) > time.Second {
s.store.Checkpoint(context.Background(), false)
lastCheckpoint.Store(time.Now())
}
lock.Unlock()
}
} }
update := headstorage.HeadsUpdate{ update := headstorage.HeadsUpdate{
Id: s.id, Id: s.id,