GO-6240 ownership transfer

Makefile

@@ -1,7 +1,7 @@
 .PHONY: build test deps build-dev
 SHELL=/usr/bin/env bash
 export GOPRIVATE=github.com/anyproto
-export PATH:=deps:$(PATH)
+export PATH:=$(CURDIR)/deps:$(PATH)
 export CGO_ENABLED:=1
 BUILD_GOOS:=$(shell go env GOOS)
 BUILD_GOARCH:=$(shell go env GOARCH)
@@ -25,8 +25,27 @@ test:
 
 deps:
 	go mod download
+	go build -o deps google.golang.org/protobuf/cmd/protoc-gen-go
+	go build -o deps github.com/planetscale/vtprotobuf/cmd/protoc-gen-go-vtproto
 	go build -o deps github.com/ahmetb/govvv
-	go build -o deps github.com/gogo/protobuf/protoc-gen-gogofaster
+	go build -o deps go.uber.org/mock/mockgen
+
+mocks:
+	echo 'Generating mocks...'
+	go generate ./...
+
+PROTOC=protoc
+PROTOC_GEN_GO=deps/protoc-gen-go
+PROTOC_GEN_VTPROTO=deps/protoc-gen-go-vtproto
+
+define generate_proto
+	@echo "Generating Protobuf for directory: $(1)"
+	$(PROTOC) \
+    		--go_out=. --plugin protoc-gen-go="$(PROTOC_GEN_GO)" \
+    		--go-vtproto_out=. --plugin protoc-gen-go-vtproto="$(PROTOC_GEN_VTPROTO)" \
+    		--go-vtproto_opt=features=marshal+unmarshal+size \
+    		--proto_path=$(1) $(wildcard $(1)/*.proto)
+endef
 
 proto:
-	protoc --gogofaster_out=:. index/indexproto/protos/*.proto
+	$(call generate_proto,index/indexproto/protos)

deletelog/deletelog.go

@@ -15,6 +15,7 @@ import (
 	"github.com/redis/go-redis/v9"
 	"go.uber.org/zap"
 
+	"github.com/anyproto/any-sync-filenode/filenode"
 	"github.com/anyproto/any-sync-filenode/index"
 	"github.com/anyproto/any-sync-filenode/redisprovider"
 )
@@ -37,6 +38,7 @@ type deleteLog struct {
 	redsync           *redsync.Redsync
 	ticker            periodicsync.PeriodicSync
 	index             index.Index
+	filenode          filenode.Service
 	disableTicker     bool
 }
 
@@ -45,6 +47,7 @@ func (d *deleteLog) Init(a *app.App) (err error) {
 	d.coordinatorClient = a.MustComponent(coordinatorclient.CName).(coordinatorclient.CoordinatorClient)
 	d.redsync = redsync.New(goredis.NewPool(d.redis))
 	d.index = a.MustComponent(index.CName).(index.Index)
+	d.filenode = a.MustComponent(filenode.CName).(filenode.Service)
 	return
 }
 
@@ -79,24 +82,20 @@ func (d *deleteLog) checkLog(ctx context.Context) (err error) {
 		return
 	}
 	var handledCount, deletedCount int
-	var ok bool
 	for _, rec := range recs {
-		if rec.Status == coordinatorproto.DeletionLogRecordStatus_Remove && rec.FileGroup != "" {
-			key := index.Key{
-				GroupId: rec.FileGroup,
-				SpaceId: rec.SpaceId,
-			}
-			ok, err = d.index.SpaceDelete(ctx, key)
-			if err != nil && !errors.Is(err, redis.Nil) && !errors.Is(err, index.ErrSpaceIsDeleted) {
-				return
-			}
-			handledCount++
-			if _, err = d.index.MarkSpaceAsDeleted(ctx, key); err != nil {
-				return
-			}
-			if ok {
-				deletedCount++
-			}
+		var ok bool
+		switch rec.Status {
+		case coordinatorproto.DeletionLogRecordStatus_Remove:
+			ok, err = d.handleDeletion(ctx, rec)
+		case coordinatorproto.DeletionLogRecordStatus_OwnershipChange:
+			err = d.handleOwnershipTransfer(ctx, rec)
+		}
+		if err != nil {
+			return
+		}
+		handledCount++
+		if ok {
+			deletedCount++
 		}
 		if err = d.redis.Set(ctx, lastKey, rec.Id, 0).Err(); err != nil {
 			return
@@ -111,6 +110,28 @@ func (d *deleteLog) checkLog(ctx context.Context) (err error) {
 	return
 }
 
+func (d *deleteLog) handleDeletion(ctx context.Context, rec *coordinatorproto.DeletionLogRecord) (ok bool, err error) {
+	if rec.FileGroup == "" {
+		return
+	}
+	key := index.Key{
+		GroupId: rec.FileGroup,
+		SpaceId: rec.SpaceId,
+	}
+	ok, err = d.index.SpaceDelete(ctx, key)
+	if err != nil && !errors.Is(err, redis.Nil) && !errors.Is(err, index.ErrSpaceIsDeleted) {
+		return
+	}
+	if _, err = d.index.MarkSpaceAsDeleted(ctx, key); err != nil {
+		return
+	}
+	return ok, nil
+}
+
+func (d *deleteLog) handleOwnershipTransfer(ctx context.Context, rec *coordinatorproto.DeletionLogRecord) (err error) {
+	return d.filenode.OwnershipTransfer(ctx, rec.SpaceId, rec.FileGroup, rec.AclRecordId)
+}
+
 func (d *deleteLog) Close(ctx context.Context) (err error) {
 	if d.ticker != nil {
 		d.ticker.Close()

deletelog/deletelog_test.go

@@ -13,6 +13,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.uber.org/mock/gomock"
 
+	"github.com/anyproto/any-sync-filenode/filenode"
+	"github.com/anyproto/any-sync-filenode/filenode/mock_filenode"
 	"github.com/anyproto/any-sync-filenode/index"
 	"github.com/anyproto/any-sync-filenode/index/mock_index"
 	"github.com/anyproto/any-sync-filenode/redisprovider/testredisprovider"
@@ -59,6 +61,26 @@ func TestDeleteLog_checkLog(t *testing.T) {
 		assert.Equal(t, "2", lastId)
 	})
 
+	t.Run("ownership change", func(t *testing.T) {
+		fx := newFixture(t)
+		defer fx.finish(t)
+		now := time.Now().Unix()
+		fx.coord.EXPECT().DeletionLog(ctx, "", recordsLimit).Return([]*coordinatorproto.DeletionLogRecord{
+			{
+				Id:          "1",
+				SpaceId:     "s1",
+				Status:      coordinatorproto.DeletionLogRecordStatus_OwnershipChange,
+				Timestamp:   now,
+				FileGroup:   "f1",
+				AclRecordId: "acl1",
+			},
+		}, nil)
+		fx.filenode.EXPECT().OwnershipTransfer(ctx, "s1", "f1", "acl1").Return(nil)
+		require.NoError(t, fx.checkLog(ctx))
+		lastId, err := fx.redis.Get(ctx, lastKey).Result()
+		require.NoError(t, err)
+		assert.Equal(t, "1", lastId)
+	})
 }
 
 func newFixture(t *testing.T) *fixture {
@@ -68,6 +90,7 @@ func newFixture(t *testing.T) *fixture {
 		a:         new(app.App),
 		coord:     mock_coordinatorclient.NewMockCoordinatorClient(ctrl),
 		index:     mock_index.NewMockIndex(ctrl),
+		filenode:  mock_filenode.NewMockService(ctrl),
 		deleteLog: New().(*deleteLog),
 	}
 	fx.disableTicker = true
@@ -77,18 +100,25 @@ func newFixture(t *testing.T) *fixture {
 	fx.index.EXPECT().Init(gomock.Any()).AnyTimes()
 	fx.index.EXPECT().Run(gomock.Any()).AnyTimes()
 	fx.index.EXPECT().Close(gomock.Any()).AnyTimes()
+	fx.filenode.EXPECT().Name().Return(filenode.CName).AnyTimes()
+	fx.filenode.EXPECT().Init(gomock.Any()).AnyTimes()
 
-	fx.a.Register(testredisprovider.NewTestRedisProviderNum(7)).Register(fx.coord).Register(fx.index).Register(fx.deleteLog)
+	fx.a.Register(testredisprovider.NewTestRedisProviderNum(7)).
+		Register(fx.coord).
+		Register(fx.index).
+		Register(fx.filenode).
+		Register(fx.deleteLog)
 	require.NoError(t, fx.a.Start(ctx))
 
 	return fx
 }
 
 type fixture struct {
-	ctrl  *gomock.Controller
-	a     *app.App
-	coord *mock_coordinatorclient.MockCoordinatorClient
-	index *mock_index.MockIndex
+	ctrl     *gomock.Controller
+	a        *app.App
+	coord    *mock_coordinatorclient.MockCoordinatorClient
+	index    *mock_index.MockIndex
+	filenode *mock_filenode.MockService
 	*deleteLog
 }
 

etc/any-sync-filenode.yml

@@ -69,4 +69,4 @@ network:
 networkStorePath: .
 networkUpdateIntervalSec: 600
 defaultLimit: 1073741824
-persistTtl: 300
+persistTtl: 1800

filenode/filenode.go

@@ -17,6 +17,7 @@ import (
 	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/rpc/server"
 	"github.com/anyproto/any-sync/nodeconf"
+	"github.com/anyproto/any-sync/util/crypto"
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
 	"go.uber.org/zap"
@@ -37,7 +38,9 @@ func New() Service {
 	return new(fileNode)
 }
 
+//go:generate mockgen -destination mock_filenode/mock_filenode.go github.com/anyproto/any-sync-filenode/filenode Service
 type Service interface {
+	OwnershipTransfer(ctx context.Context, spaceId, oldIdentity string, aclRecordId string) (err error)
 	app.Component
 }
 
@@ -180,13 +183,24 @@ func (fn *fileNode) StoreKey(ctx context.Context, spaceId string, checkLimit boo
 		return storageKey, fileprotoerr.ErrForbidden
 	}
 
-	ownerPubKey, err := fn.acl.OwnerPubKey(ctx, spaceId)
-	if err != nil {
-		log.WarnCtx(ctx, "acl ownerPubKey error", zap.Error(err))
-		return storageKey, fileprotoerr.ErrForbidden
-	}
+	var (
+		ownerPubKey      crypto.PubKey
+		ownerRecordIndex int
+		isOneToOne       bool
+	)
 
-	err = fn.acl.ReadState(ctx, spaceId, func(aclState *list.AclState) error {
+	err = fn.acl.ReadList(ctx, spaceId, func(aclList list.AclList) error {
+		aclState := aclList.AclState()
+		var ownerRecordId string
+		if ownerPubKey, ownerRecordId, err = aclState.OwnerPubKeyWithRecordId(); err != nil {
+			log.WarnCtx(ctx, "acl ownerPubKey error", zap.Error(err))
+			return fileprotoerr.ErrForbidden
+		}
+		ownerRecordIndex = aclList.GetRecordIndex(ownerRecordId)
+		if ownerRecordIndex < 0 {
+			log.ErrorCtx(ctx, "acl ownerRecordIndex not found", zap.String("spaceId", spaceId), zap.String("recordId", ownerRecordId), zap.Error(err))
+			return fileprotoerr.ErrUnexpected
+		}
 		if aclState.IsOneToOne() {
 			if aclState.Permissions(identity).NoPermissions() {
 				return fileprotoerr.ErrForbidden
@@ -207,6 +221,7 @@ func (fn *fileNode) StoreKey(ctx context.Context, spaceId string, checkLimit boo
 				GroupId: identity.Account(),
 				SpaceId: newSpaceId,
 			}
+			isOneToOne = true
 		} else {
 			storageKey = index.Key{
 				GroupId: ownerPubKey.Account(),
@@ -231,10 +246,19 @@ func (fn *fileNode) StoreKey(ctx context.Context, spaceId string, checkLimit boo
 		}
 	}
 
-	if e := fn.index.Migrate(ctx, storageKey); e != nil {
-		log.WarnCtx(ctx, "space migrate error", zap.String("spaceId", spaceId), zap.Error(e))
+	if !isOneToOne {
+		if e := fn.index.Migrate(ctx, storageKey); e != nil {
+			log.WarnCtx(ctx, "space migrate error", zap.String("spaceId", spaceId), zap.Error(e))
+		}
+		var oldIdentity string
+		if ownerRecordIndex == 0 {
+			oldIdentity = storageKey.GroupId
+		}
+		if err = fn.index.CheckAndMoveOwnership(ctx, storageKey, oldIdentity, ownerRecordIndex); err != nil {
+			log.ErrorCtx(ctx, "check ownership error", zap.String("spaceId", spaceId), zap.Error(err))
+			return storageKey, fileprotoerr.ErrUnexpected
+		}
 	}
-
 	if checkLimit {
 		if err = fn.index.CheckLimits(ctx, storageKey); err != nil {
 			if errors.Is(err, index.ErrLimitExceed) {
@@ -417,3 +441,35 @@ func (fn *fileNode) FilesGet(ctx context.Context, spaceId string) (fileIds []str
 	}
 	return fn.index.FilesList(ctx, storeKey)
 }
+
+func (fn *fileNode) OwnershipTransfer(ctx context.Context, spaceId, oldIdentity, aclRecordId string) (err error) {
+	var (
+		ownerPubKey      crypto.PubKey
+		ownerRecordIndex int
+	)
+	defer func() {
+		log.InfoCtx(ctx, "ownership transfer", zap.String("spaceId", spaceId), zap.String("recordId", aclRecordId), zap.Error(err))
+	}()
+	err = fn.acl.ReadList(ctx, spaceId, func(aclList list.AclList) error {
+		if !aclList.HasHead(aclRecordId) {
+			log.WarnCtx(ctx, "ownership transfer error: acl record not found", zap.String("spaceId", spaceId), zap.String("recordId", aclRecordId))
+			return fileprotoerr.ErrAclRecordNotFound
+		}
+		aclState := aclList.AclState()
+		var ownerRecordId string
+		if ownerPubKey, ownerRecordId, err = aclState.OwnerPubKeyWithRecordId(); err != nil {
+			log.WarnCtx(ctx, "acl ownerPubKey error", zap.Error(err))
+			return fileprotoerr.ErrForbidden
+		}
+		ownerRecordIndex = aclList.GetRecordIndex(ownerRecordId)
+		if ownerRecordIndex < 0 {
+			log.ErrorCtx(ctx, "acl ownerRecordIndex not found", zap.String("spaceId", spaceId), zap.String("recordId", ownerRecordId), zap.Error(err))
+			return fileprotoerr.ErrUnexpected
+		}
+		return nil
+	})
+	if err != nil {
+		return
+	}
+	return fn.index.CheckAndMoveOwnership(ctx, index.Key{GroupId: ownerPubKey.Account(), SpaceId: spaceId}, oldIdentity, ownerRecordIndex)
+}