允许重复进行相同的load操作

agent/internal/task/storage_load_package.go

@@ -103,6 +103,7 @@ func (t *StorageLoadPackage) do(task *task.Task[TaskContext], ctx TaskContext) e
 		return fmt.Errorf("loading package to storage: %w", err)
 	}
 
+	// TODO 要防止下载的临时文件被删除
 	return err
 }
 

common/assets/scripts/create_database.sql

@@ -157,8 +157,8 @@ create table StoragePackage (
 );
 
 create table StoragePackageLog (
-  PackageID int not null comment '包ID',
   StorageID int not null comment '存储服务ID',
+  PackageID int not null comment '包ID',
   UserID int not null comment '调度了此文件的用户ID',
   CreateTime timestamp not null comment '加载Package完成的时间'
 );

common/pkgs/db/storage_package.go

@@ -34,8 +34,9 @@ func (*StoragePackageDB) GetAllByStorageID(ctx SQLContext, storageID cdssdk.Stor
 	return ret, err
 }
 
-func (*StoragePackageDB) Create(ctx SQLContext, storageID cdssdk.StorageID, packageID cdssdk.PackageID, userID cdssdk.UserID) error {
-	_, err := ctx.Exec("insert into StoragePackage values(?,?,?,?)", storageID, packageID, userID, model.StoragePackageStateNormal)
+func (*StoragePackageDB) CreateOrUpdate(ctx SQLContext, storageID cdssdk.StorageID, packageID cdssdk.PackageID, userID cdssdk.UserID) error {
+	_, err := ctx.Exec("insert into StoragePackage values(?,?,?,?)"+
+		" on duplicate key update State=values(State)", storageID, packageID, userID, model.StoragePackageStateNormal)
 	return err
 }
 

coordinator/internal/mq/storage.go

@@ -34,7 +34,7 @@ func (svc *Service) StoragePackageLoaded(msg *coormq.StoragePackageLoaded) (*coo
 			return fmt.Errorf("storage is not available to user")
 		}
 
-		err := svc.db.StoragePackage().Create(tx, msg.StorageID, msg.PackageID, msg.UserID)
+		err := svc.db.StoragePackage().CreateOrUpdate(tx, msg.StorageID, msg.PackageID, msg.UserID)
 		if err != nil {
 			return fmt.Errorf("creating storage package: %w", err)
 		}