数据库操作使用事务

agent/internal/services/mq/object.go

@@ -19,7 +19,7 @@ func (svc *Service) PinObject(msg *agtmq.PinObject) (*agtmq.PinObjectResp, *mq.C
 		return nil, mq.Failed(errorcode.OperationFailed, "pin object failed")
 	}
 
-	if msg.Async {
+	if msg.IsBackground {
 		return mq.ReplyOK(agtmq.RespPinObject())
 	}
 

common/pkgs/db/bucket.go

@@ -111,11 +111,13 @@ func (db *BucketDB) Delete(ctx SQLContext, bucketID cdssdk.BucketID) error {
 	}
 
 	for _, pkgID := range pkgIDs {
-		// TODO 不一定所有的错误都要中断后续过程
 		err = db.Package().SoftDelete(ctx, pkgID)
 		if err != nil {
 			return fmt.Errorf("set package seleted failed, err: %w", err)
 		}
+
+		// 失败也没关系，会有定时任务再次尝试
+		db.Package().DeleteUnused(ctx, pkgID)
 	}
 	return nil
 }

common/pkgs/db/object.go

@@ -107,9 +107,14 @@ func (db *ObjectDB) BatchAdd(ctx SQLContext, packageID cdssdk.PackageID, objs []
 
 		if !isCreate {
 			// 删除原本所有的编码块记录，重新添加
-			if err = db.ObjectBlock().DeleteObjectAll(ctx, objID); err != nil {
+			if err = db.ObjectBlock().DeleteByObjectID(ctx, objID); err != nil {
 				return nil, fmt.Errorf("deleting all object block: %w", err)
 			}
+
+			// 删除原本Pin住的Object。暂不考虑FileHash没有变化的情况
+			if err = db.PinnedObject().DeleteByObjectID(ctx, objID); err != nil {
+				return nil, fmt.Errorf("deleting all pinned object: %w", err)
+			}
 		}
 
 		// 首次上传默认使用不分块的none模式
@@ -136,10 +141,15 @@ func (db *ObjectDB) BatchUpdateRedundancy(ctx SQLContext, objs []coormq.ChangeOb
 		}
 
 		// 删除原本所有的编码块记录，重新添加
-		if err = db.ObjectBlock().DeleteObjectAll(ctx, obj.ObjectID); err != nil {
+		if err = db.ObjectBlock().DeleteByObjectID(ctx, obj.ObjectID); err != nil {
 			return fmt.Errorf("deleting all object block: %w", err)
 		}
 
+		// 删除原本Pin住的Object。暂不考虑FileHash没有变化的情况
+		if err = db.PinnedObject().DeleteByObjectID(ctx, obj.ObjectID); err != nil {
+			return fmt.Errorf("deleting all pinned object: %w", err)
+		}
+
 		for _, block := range obj.Blocks {
 			// 首次上传默认使用不分块的rep模式
 			err = db.ObjectBlock().Create(ctx, obj.ObjectID, block.Index, block.NodeID, block.FileHash)

common/pkgs/db/object_block.go

@@ -31,7 +31,7 @@ func (db *ObjectBlockDB) Create(ctx SQLContext, objectID cdssdk.ObjectID, index
 	return err
 }
 
-func (db *ObjectBlockDB) DeleteObjectAll(ctx SQLContext, objectID cdssdk.ObjectID) error {
+func (db *ObjectBlockDB) DeleteByObjectID(ctx SQLContext, objectID cdssdk.ObjectID) error {
 	_, err := ctx.Exec("delete from ObjectBlock where ObjectID = ?", objectID)
 	return err
 }

common/pkgs/db/package.go

@@ -127,6 +127,10 @@ func (db *PackageDB) SoftDelete(ctx SQLContext, packageID cdssdk.PackageID) erro
 		return fmt.Errorf("delete from object rep failed, err: %w", err)
 	}
 
+	if err := db.PinnedObject().DeleteInPackage(ctx, packageID); err != nil {
+		return fmt.Errorf("deleting pinned objects in package: %w", err)
+	}
+
 	if err := db.Object().DeleteInPackage(ctx, packageID); err != nil {
 		return fmt.Errorf("deleting objects in package: %w", err)
 	}

common/pkgs/db/pinned_object.go

@@ -47,6 +47,16 @@ func (*PinnedObjectDB) Delete(ctx SQLContext, nodeID cdssdk.NodeID, objectID cds
 	return err
 }
 
+func (*PinnedObjectDB) DeleteByObjectID(ctx SQLContext, objectID cdssdk.ObjectID) error {
+	_, err := ctx.Exec("delete from PinnedObject where and ObjectID = ?")
+	return err
+}
+
+func (*PinnedObjectDB) DeleteInPackage(ctx SQLContext, packageID cdssdk.PackageID) error {
+	_, err := ctx.Exec("delete PinnedObject from PinnedObject inner join Object on PinnedObject.ObjectID = Object.ObjectID where PackageID = ?", packageID)
+	return err
+}
+
 func (*PinnedObjectDB) NodeBatchDelete(ctx SQLContext, nodeID cdssdk.NodeID, objectIDs []cdssdk.ObjectID) error {
 	_, err := ctx.Exec("delete from PinnedObject where NodeID = ? and ObjectID in (?)", objectIDs)
 	return err

common/pkgs/db/user.go

@@ -0,0 +1,21 @@
+package db
+
+import (
+	"github.com/jmoiron/sqlx"
+	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/db/model"
+)
+
+type UserDB struct {
+	*DB
+}
+
+func (db *DB) User() *UserDB {
+	return &UserDB{DB: db}
+}
+
+func (db *UserDB) GetByID(ctx SQLContext, userID cdssdk.UserID) (model.User, error) {
+	var ret model.User
+	err := sqlx.Get(ctx, &ret, "select * from User where UserID = ?", userID)
+	return ret, err
+}

common/pkgs/mq/agent/object.go

@@ -11,17 +11,17 @@ var _ = Register(Service.PinObject)
 
 type PinObject struct {
 	mq.MessageBodyBase
-	FileHash string `json:"fileHash"`
-	Async    bool   `json:"async"`
+	FileHash     string `json:"fileHash"`
+	IsBackground bool   `json:"isBackground"`
 }
 type PinObjectResp struct {
 	mq.MessageBodyBase
 }
 
-func ReqPinObject(fileHash string, async bool) *PinObject {
+func ReqPinObject(fileHash string, isBackground bool) *PinObject {
 	return &PinObject{
-		FileHash: fileHash,
-		Async:    async,
+		FileHash:     fileHash,
+		IsBackground: isBackground,
 	}
 }
 func RespPinObject() *PinObjectResp {

coordinator/internal/services/bucket.go

@@ -2,6 +2,7 @@ package services
 
 import (
 	"database/sql"
+	"fmt"
 
 	"github.com/jmoiron/sqlx"
 	"gitlink.org.cn/cloudream/common/consts/errorcode"
@@ -44,16 +45,23 @@ func (svc *Service) GetBucketPackages(msg *coormq.GetBucketPackages) (*coormq.Ge
 
 func (svc *Service) CreateBucket(msg *coormq.CreateBucket) (*coormq.CreateBucketResp, *mq.CodeMessage) {
 	var bucketID cdssdk.BucketID
-	var err error
-	svc.db.DoTx(sql.LevelDefault, func(tx *sqlx.Tx) error {
-		// 这里用的是外部的err
+	err := svc.db.DoTx(sql.LevelLinearizable, func(tx *sqlx.Tx) error {
+		_, err := svc.db.User().GetByID(tx, msg.UserID)
+		if err != nil {
+			return fmt.Errorf("getting user by id: %w", err)
+		}
+
 		bucketID, err = svc.db.Bucket().Create(tx, msg.UserID, msg.BucketName)
-		return err
+		if err != nil {
+			return fmt.Errorf("creating bucket: %w", err)
+		}
+
+		return nil
 	})
 	if err != nil {
 		logger.WithField("UserID", msg.UserID).
 			WithField("BucketName", msg.BucketName).
-			Warnf("create bucket failed, err: %s", err.Error())
+			Warn(err.Error())
 		return nil, mq.Failed(errorcode.OperationFailed, "create bucket failed")
 	}
 
@@ -61,13 +69,23 @@ func (svc *Service) CreateBucket(msg *coormq.CreateBucket) (*coormq.CreateBucket
 }
 
 func (svc *Service) DeleteBucket(msg *coormq.DeleteBucket) (*coormq.DeleteBucketResp, *mq.CodeMessage) {
-	err := svc.db.DoTx(sql.LevelDefault, func(tx *sqlx.Tx) error {
-		return svc.db.Bucket().Delete(tx, msg.BucketID)
+	err := svc.db.DoTx(sql.LevelLinearizable, func(tx *sqlx.Tx) error {
+		isAvai, _ := svc.db.Bucket().IsAvailable(tx, msg.BucketID, msg.UserID)
+		if !isAvai {
+			return fmt.Errorf("bucket is not avaiable to the user")
+		}
+
+		err := svc.db.Bucket().Delete(tx, msg.BucketID)
+		if err != nil {
+			return fmt.Errorf("deleting bucket: %w", err)
+		}
+
+		return nil
 	})
 	if err != nil {
 		logger.WithField("UserID", msg.UserID).
 			WithField("BucketID", msg.BucketID).
-			Warnf("delete bucket failed, err: %s", err.Error())
+			Warn(err.Error())
 		return nil, mq.Failed(errorcode.OperationFailed, "delete bucket failed")
 	}
 

coordinator/internal/services/cache.go

@@ -1,6 +1,10 @@
 package services
 
 import (
+	"database/sql"
+	"fmt"
+
+	"github.com/jmoiron/sqlx"
 	"gitlink.org.cn/cloudream/common/consts/errorcode"
 	"gitlink.org.cn/cloudream/common/pkgs/logger"
 	"gitlink.org.cn/cloudream/common/pkgs/mq"
@@ -8,8 +12,26 @@ import (
 )
 
 func (svc *Service) CachePackageMoved(msg *coormq.CachePackageMoved) (*coormq.CachePackageMovedResp, *mq.CodeMessage) {
-	if err := svc.db.PinnedObject().CreateFromPackage(svc.db.SQLCtx(), msg.PackageID, msg.NodeID); err != nil {
-		logger.Warnf("create package pinned objects: %s", err.Error())
+	err := svc.db.DoTx(sql.LevelLinearizable, func(tx *sqlx.Tx) error {
+		_, err := svc.db.Package().GetByID(tx, msg.PackageID)
+		if err != nil {
+			return fmt.Errorf("getting package by id: %w", err)
+		}
+
+		_, err = svc.db.Node().GetByID(tx, msg.NodeID)
+		if err != nil {
+			return fmt.Errorf("getting node by id: %w", err)
+		}
+
+		err = svc.db.PinnedObject().CreateFromPackage(tx, msg.PackageID, msg.NodeID)
+		if err != nil {
+			return fmt.Errorf("creating pinned objects from package: %w", err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		logger.WithField("PackageID", msg.PackageID).WithField("NodeID", msg.NodeID).Warn(err.Error())
 		return nil, mq.Failed(errorcode.OperationFailed, "create package pinned objects failed")
 	}
 

coordinator/internal/services/node.go

@@ -31,6 +31,7 @@ func (svc *Service) GetNodes(msg *coormq.GetNodes) (*coormq.GetNodesResp, *mq.Co
 		}
 
 	} else {
+		// 可以不用事务
 		for _, id := range msg.NodeIDs {
 			node, err := svc.db.Node().GetByID(svc.db.SQLCtx(), id)
 			if err != nil {

coordinator/internal/services/object.go

@@ -1,9 +1,14 @@
 package services
 
 import (
+	"database/sql"
+	"fmt"
+
+	"github.com/jmoiron/sqlx"
 	"gitlink.org.cn/cloudream/common/consts/errorcode"
 	"gitlink.org.cn/cloudream/common/pkgs/logger"
 	"gitlink.org.cn/cloudream/common/pkgs/mq"
+	stgmod "gitlink.org.cn/cloudream/storage/common/models"
 	coormq "gitlink.org.cn/cloudream/storage/common/pkgs/mq/coordinator"
 )
 
@@ -21,22 +26,37 @@ func (svc *Service) GetPackageObjects(msg *coormq.GetPackageObjects) (*coormq.Ge
 }
 
 func (svc *Service) GetPackageObjectDetails(msg *coormq.GetPackageObjectDetails) (*coormq.GetPackageObjectDetailsResp, *mq.CodeMessage) {
-	data, err := svc.db.ObjectBlock().GetPackageBlockDetails(svc.db.SQLCtx(), msg.PackageID)
-	if err != nil {
-		logger.WithField("PackageID", msg.PackageID).
-			Warnf("getting package block details: %s", err.Error())
+	var details []stgmod.ObjectDetail
+	// 必须放在事务里进行，因为GetPackageBlockDetails是由多次数据库操作组成，必须保证数据的一致性
+	err := svc.db.DoTx(sql.LevelLinearizable, func(tx *sqlx.Tx) error {
+		var err error
+		_, err = svc.db.Package().GetByID(tx, msg.PackageID)
+		if err != nil {
+			return fmt.Errorf("getting package by id: %w", err)
+		}
+
+		details, err = svc.db.ObjectBlock().GetPackageBlockDetails(tx, msg.PackageID)
+		if err != nil {
+			return fmt.Errorf("getting package block details: %w", err)
+		}
 
+		return nil
+	})
+
+	if err != nil {
+		logger.WithField("PackageID", msg.PackageID).Warn(err.Error())
 		return nil, mq.Failed(errorcode.OperationFailed, "get package object block details failed")
 	}
 
-	return mq.ReplyOK(coormq.NewGetPackageObjectDetailsResp(data))
+	return mq.ReplyOK(coormq.NewGetPackageObjectDetailsResp(details))
 }
 
 func (svc *Service) ChangeObjectRedundancy(msg *coormq.ChangeObjectRedundancy) (*coormq.ChangeObjectRedundancyResp, *mq.CodeMessage) {
-	err := svc.db.Object().BatchUpdateRedundancy(svc.db.SQLCtx(), msg.Entries)
+	err := svc.db.DoTx(sql.LevelLinearizable, func(tx *sqlx.Tx) error {
+		return svc.db.Object().BatchUpdateRedundancy(tx, msg.Entries)
+	})
 	if err != nil {
 		logger.Warnf("batch updating redundancy: %s", err.Error())
-
 		return nil, mq.Failed(errorcode.OperationFailed, "batch update redundancy failed")
 	}
 

coordinator/internal/services/package.go

@@ -28,16 +28,25 @@ func (svc *Service) GetPackage(msg *coormq.GetPackage) (*coormq.GetPackageResp,
 
 func (svc *Service) CreatePackage(msg *coormq.CreatePackage) (*coormq.CreatePackageResp, *mq.CodeMessage) {
 	var pkgID cdssdk.PackageID
-	err := svc.db.DoTx(sql.LevelDefault, func(tx *sqlx.Tx) error {
+	err := svc.db.DoTx(sql.LevelLinearizable, func(tx *sqlx.Tx) error {
 		var err error
-		pkgID, err = svc.db.Package().Create(svc.db.SQLCtx(), msg.BucketID, msg.Name)
-		return err
+
+		isAvai, _ := svc.db.Bucket().IsAvailable(tx, msg.BucketID, msg.UserID)
+		if !isAvai {
+			return fmt.Errorf("bucket is not avaiable to the user")
+		}
+
+		pkgID, err = svc.db.Package().Create(tx, msg.BucketID, msg.Name)
+		if err != nil {
+			return fmt.Errorf("creating package: %w", err)
+		}
+
+		return nil
 	})
 	if err != nil {
 		logger.WithField("BucketID", msg.BucketID).
 			WithField("Name", msg.Name).
-			Warnf("creating package: %s", err.Error())
-
+			Warn(err.Error())
 		return nil, mq.Failed(errorcode.OperationFailed, "creating package failed")
 	}
 
@@ -45,15 +54,12 @@ func (svc *Service) CreatePackage(msg *coormq.CreatePackage) (*coormq.CreatePack
 }
 
 func (svc *Service) UpdatePackage(msg *coormq.UpdatePackage) (*coormq.UpdatePackageResp, *mq.CodeMessage) {
-	_, err := svc.db.Package().GetByID(svc.db.SQLCtx(), msg.PackageID)
-	if err != nil {
-		logger.WithField("PackageID", msg.PackageID).
-			Warnf("get package: %s", err.Error())
-
-		return nil, mq.Failed(errorcode.OperationFailed, "get package failed")
-	}
+	err := svc.db.DoTx(sql.LevelLinearizable, func(tx *sqlx.Tx) error {
+		_, err := svc.db.Package().GetByID(tx, msg.PackageID)
+		if err != nil {
+			return fmt.Errorf("getting package by id: %w", err)
+		}
 
-	err = svc.db.DoTx(sql.LevelDefault, func(tx *sqlx.Tx) error {
 		// 先执行删除操作
 		if len(msg.Deletes) > 0 {
 			if err := svc.db.Object().BatchDelete(tx, msg.Deletes); err != nil {
@@ -71,7 +77,7 @@ func (svc *Service) UpdatePackage(msg *coormq.UpdatePackage) (*coormq.UpdatePack
 		return nil
 	})
 	if err != nil {
-		logger.Warn(err.Error())
+		logger.WithField("PackageID", msg.PackageID).Warn(err.Error())
 		return nil, mq.Failed(errorcode.OperationFailed, "update package failed")
 	}
 
@@ -79,21 +85,12 @@ func (svc *Service) UpdatePackage(msg *coormq.UpdatePackage) (*coormq.UpdatePack
 }
 
 func (svc *Service) DeletePackage(msg *coormq.DeletePackage) (*coormq.DeletePackageResp, *mq.CodeMessage) {
-	isAva, err := svc.db.Package().IsAvailable(svc.db.SQLCtx(), msg.UserID, msg.PackageID)
-	if err != nil {
-		logger.WithField("UserID", msg.UserID).
-			WithField("PackageID", msg.PackageID).
-			Warnf("check package available failed, err: %s", err.Error())
-		return nil, mq.Failed(errorcode.OperationFailed, "check package available failed")
-	}
-	if !isAva {
-		logger.WithField("UserID", msg.UserID).
-			WithField("PackageID", msg.PackageID).
-			Warnf("package is not available to the user")
-		return nil, mq.Failed(errorcode.OperationFailed, "package is not available to the user")
-	}
+	err := svc.db.DoTx(sql.LevelLinearizable, func(tx *sqlx.Tx) error {
+		isAvai, _ := svc.db.Package().IsAvailable(tx, msg.UserID, msg.PackageID)
+		if !isAvai {
+			return fmt.Errorf("package is not available to the user")
+		}
 
-	err = svc.db.DoTx(sql.LevelDefault, func(tx *sqlx.Tx) error {
 		err := svc.db.Package().SoftDelete(tx, msg.PackageID)
 		if err != nil {
 			return fmt.Errorf("soft delete package: %w", err)
@@ -111,8 +108,8 @@ func (svc *Service) DeletePackage(msg *coormq.DeletePackage) (*coormq.DeletePack
 	if err != nil {
 		logger.WithField("UserID", msg.UserID).
 			WithField("PackageID", msg.PackageID).
-			Warnf("set package deleted failed, err: %s", err.Error())
-		return nil, mq.Failed(errorcode.OperationFailed, "set package deleted failed")
+			Warnf(err.Error())
+		return nil, mq.Failed(errorcode.OperationFailed, "delete package failed")
 	}
 
 	return mq.ReplyOK(coormq.NewDeletePackageResp())
@@ -133,6 +130,7 @@ func (svc *Service) GetPackageCachedNodes(msg *coormq.GetPackageCachedNodes) (*c
 		return nil, mq.Failed(errorcode.OperationFailed, "package is not available to the user")
 	}
 
+	// 这个函数只是统计哪些节点缓存了Package中的数据，不需要多么精确，所以可以不用事务
 	objDetails, err := svc.db.ObjectBlock().GetPackageBlockDetails(svc.db.SQLCtx(), msg.PackageID)
 	if err != nil {
 		logger.WithField("PackageID", msg.PackageID).

coordinator/internal/services/storage.go

@@ -24,8 +24,7 @@ func (svc *Service) GetStorageInfo(msg *coormq.GetStorageInfo) (*coormq.GetStora
 }
 
 func (svc *Service) StoragePackageLoaded(msg *coormq.StoragePackageLoaded) (*coormq.StoragePackageLoadedResp, *mq.CodeMessage) {
-	// TODO: 对于的storage中已经存在的文件，直接覆盖已有文件
-	err := svc.db.DoTx(sql.LevelDefault, func(tx *sqlx.Tx) error {
+	err := svc.db.DoTx(sql.LevelLinearizable, func(tx *sqlx.Tx) error {
 		err := svc.db.StoragePackage().Create(tx, msg.StorageID, msg.PackageID, msg.UserID)
 		if err != nil {
 			return fmt.Errorf("creating storage package: %w", err)
@@ -42,7 +41,7 @@ func (svc *Service) StoragePackageLoaded(msg *coormq.StoragePackageLoaded) (*coo
 		logger.WithField("UserID", msg.UserID).
 			WithField("StorageID", msg.StorageID).
 			WithField("PackageID", msg.PackageID).
-			Warnf("user load package to storage failed, err: %s", err.Error())
+			Warn(err.Error())
 		return nil, mq.Failed(errorcode.OperationFailed, "user load package to storage failed")
 	}
 

scanner/internal/event/agent_check_storage.go

@@ -89,7 +89,7 @@ func (t *AgentCheckStorage) Execute(execCtx ExecuteContext) {
 	}
 
 	execCtx.Args.DB.DoTx(sql.LevelLinearizable, func(tx *sqlx.Tx) error {
-		packages, err := execCtx.Args.DB.StoragePackage().GetAllByStorageID(execCtx.Args.DB.SQLCtx(), t.StorageID)
+		packages, err := execCtx.Args.DB.StoragePackage().GetAllByStorageID(tx, t.StorageID)
 		if err != nil {
 			log.Warnf("getting storage package: %s", err.Error())
 			return nil

scanner/internal/event/check_package_redundancy.go

@@ -11,6 +11,7 @@ import (
 	stgglb "gitlink.org.cn/cloudream/storage/common/globals"
 	stgmod "gitlink.org.cn/cloudream/storage/common/models"
 	"gitlink.org.cn/cloudream/storage/common/pkgs/db/model"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/distlock/reqbuilder"
 	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch/plans"
 	agtmq "gitlink.org.cn/cloudream/storage/common/pkgs/mq/agent"
 	coormq "gitlink.org.cn/cloudream/storage/common/pkgs/mq/coordinator"
@@ -112,6 +113,21 @@ func (t *CheckPackageRedundancy) Execute(execCtx ExecuteContext) {
 	newRepNodes := t.chooseNewNodesForRep(&defRep, allNodes)
 	newECNodes := t.chooseNewNodesForEC(&defEC, allNodes)
 
+	// 加锁
+	builder := reqbuilder.NewBuilder()
+	for _, node := range newRepNodes {
+		builder.IPFS().Buzy(node.Node.NodeID)
+	}
+	for _, node := range newECNodes {
+		builder.IPFS().Buzy(node.Node.NodeID)
+	}
+	mutex, err := builder.MutexLock(execCtx.Args.DistLock)
+	if err != nil {
+		log.Warnf("acquiring dist lock: %s", err.Error())
+		return
+	}
+	defer mutex.Unlock()
+
 	for _, obj := range getObjs.Objects {
 		var entry *coormq.ChangeObjectRedundancyEntry
 		var err error