diff --git a/agent/internal/services/mq/object.go b/agent/internal/services/mq/object.go index 0eec072..06a106f 100644 --- a/agent/internal/services/mq/object.go +++ b/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()) } diff --git a/common/pkgs/db/bucket.go b/common/pkgs/db/bucket.go index cc1d208..5fdda6a 100644 --- a/common/pkgs/db/bucket.go +++ b/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 } diff --git a/common/pkgs/db/object.go b/common/pkgs/db/object.go index 2fa82ca..aae7553 100644 --- a/common/pkgs/db/object.go +++ b/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) diff --git a/common/pkgs/db/object_block.go b/common/pkgs/db/object_block.go index 20d48aa..2ec166c 100644 --- a/common/pkgs/db/object_block.go +++ b/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 } diff --git a/common/pkgs/db/package.go b/common/pkgs/db/package.go index 9ef4baf..1491bec 100644 --- a/common/pkgs/db/package.go +++ b/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) } diff --git a/common/pkgs/db/pinned_object.go b/common/pkgs/db/pinned_object.go index c87da07..2b08e77 100644 --- a/common/pkgs/db/pinned_object.go +++ b/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 diff --git a/common/pkgs/db/user.go b/common/pkgs/db/user.go new file mode 100644 index 0000000..b69290e --- /dev/null +++ b/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 +} diff --git a/common/pkgs/mq/agent/object.go b/common/pkgs/mq/agent/object.go index 7c0994b..a3b56ac 100644 --- a/common/pkgs/mq/agent/object.go +++ b/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 { diff --git a/coordinator/internal/services/bucket.go b/coordinator/internal/services/bucket.go index e755dbf..4ffdab7 100644 --- a/coordinator/internal/services/bucket.go +++ b/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") } diff --git a/coordinator/internal/services/cache.go b/coordinator/internal/services/cache.go index 5c844cc..1cc5a0b 100644 --- a/coordinator/internal/services/cache.go +++ b/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") } diff --git a/coordinator/internal/services/node.go b/coordinator/internal/services/node.go index 3028d69..0614264 100644 --- a/coordinator/internal/services/node.go +++ b/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 { diff --git a/coordinator/internal/services/object.go b/coordinator/internal/services/object.go index 2a73045..43cf601 100644 --- a/coordinator/internal/services/object.go +++ b/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") } diff --git a/coordinator/internal/services/package.go b/coordinator/internal/services/package.go index c41ef9a..30e110c 100644 --- a/coordinator/internal/services/package.go +++ b/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()) + 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) + } - return nil, mq.Failed(errorcode.OperationFailed, "get package failed") - } - - 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). diff --git a/coordinator/internal/services/storage.go b/coordinator/internal/services/storage.go index ee57c39..66b38f0 100644 --- a/coordinator/internal/services/storage.go +++ b/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") } diff --git a/scanner/internal/event/agent_check_storage.go b/scanner/internal/event/agent_check_storage.go index a3d44e2..66e786c 100644 --- a/scanner/internal/event/agent_check_storage.go +++ b/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 diff --git a/scanner/internal/event/check_package_redundancy.go b/scanner/internal/event/check_package_redundancy.go index 72dfbfa..00e331b 100644 --- a/scanner/internal/event/check_package_redundancy.go +++ b/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