feat: gc for bucket migration & gc for ZombiePiece & meta

feat: format code

base/gfspapp/app.go

@@ -64,20 +64,22 @@ type GfSpBaseApp struct {
 	replicateSpeed int64
 	receiveSpeed   int64
 
-	sealObjectTimeout int64
-	gcObjectTimeout   int64
-	gcZombieTimeout   int64
-	gcMetaTimeout     int64
-	migrateGVGTimeout int64
-
-	sealObjectRetry     int64
-	replicateRetry      int64
-	receiveConfirmRetry int64
-	gcObjectRetry       int64
-	gcZombieRetry       int64
-	gcMetaRetry         int64
-	recoveryRetry       int64
-	migrateGVGRetry     int64
+	sealObjectTimeout        int64
+	gcObjectTimeout          int64
+	gcZombieTimeout          int64
+	gcMetaTimeout            int64
+	migrateGVGTimeout        int64
+	gcBucketMigrationTimeout int64
+
+	sealObjectRetry        int64
+	replicateRetry         int64
+	receiveConfirmRetry    int64
+	gcObjectRetry          int64
+	gcZombieRetry          int64
+	gcMetaRetry            int64
+	recoveryRetry          int64
+	migrateGVGRetry        int64
+	gcBucketMigrationRetry int64
 }
 
 // AppID returns the GfSpBaseApp ID, the default value is prefix(gfsp) add

base/gfspapp/app_options.go

@@ -286,6 +286,7 @@ func defaultGfSpDB(cfg *config.SQLDBConfig) {
 	if cfg.MaxOpenConns == 0 {
 		cfg.MaxOpenConns = sqldb.DefaultMaxOpenConns
 	}
+	cfg.EnableTracePutEvent = sqldb.DefaultEnableTracePutEvent
 	if cfg.User == "" {
 		cfg.User = "root"
 	}

base/gfspapp/manage_server.go

@@ -168,6 +168,10 @@ func (g *GfSpBaseApp) GfSpAskTask(ctx context.Context, req *gfspserver.GfSpAskTa
 		resp.Response = &gfspserver.GfSpAskTaskResponse_MigrateGvgTask{
 			MigrateGvgTask: t,
 		}
+	case *gfsptask.GfSpGCBucketMigrationTask:
+		resp.Response = &gfspserver.GfSpAskTaskResponse_GcBucketMigrationTask{
+			GcBucketMigrationTask: t,
+		}
 	default:
 		log.CtxErrorw(ctx, "[BUG] Unsupported task type to dispatch")
 		return &gfspserver.GfSpAskTaskResponse{Err: ErrUnsupportedTaskType}, nil
@@ -300,6 +304,12 @@ func (g *GfSpBaseApp) GfSpReportTask(ctx context.Context, req *gfspserver.GfSpRe
 		task.SetAddress(util.GetRPCRemoteAddress(ctx))
 		log.CtxInfow(ctx, "begin to handle reported migrate gvg task", "task_info", task.Info())
 		err = g.manager.HandleMigrateGVGTask(ctx, t.MigrateGvgTask)
+	case *gfspserver.GfSpReportTaskRequest_GcBucketMigrationTask:
+		task := t.GcBucketMigrationTask
+		ctx = log.WithValue(ctx, log.CtxKeyTask, task.Key().String())
+		task.SetAddress(util.GetRPCRemoteAddress(ctx))
+		log.CtxInfow(ctx, "begin to handle reported gc bucket migration task", "task_info", task.Info())
+		err = g.manager.HandleGCBucketMigrationTask(ctx, t.GcBucketMigrationTask)
 	default:
 		log.CtxError(ctx, "receive unsupported task type")
 		return &gfspserver.GfSpReportTaskResponse{Err: ErrUnsupportedTaskType}, nil
@@ -358,9 +368,10 @@ func (g *GfSpBaseApp) GfSpNotifyPreMigrate(ctx context.Context, req *gfspserver.
 
 	return &gfspserver.GfSpNotifyPreMigrateBucketResponse{}, nil
 }
+
 func (g *GfSpBaseApp) GfSpNotifyPostMigrate(ctx context.Context, req *gfspserver.GfSpNotifyPostMigrateBucketRequest) (
 	*gfspserver.GfSpNotifyPostMigrateBucketResponse, error) {
-	if err := g.manager.NotifyPostMigrateBucket(ctx, req.GetBucketId()); err != nil {
+	if err := g.manager.NotifyPostMigrateBucket(ctx, req.GetBucketMigrationInfo()); err != nil {
 		return nil, err
 	}
 

base/gfspapp/task_options.go

@@ -51,6 +51,10 @@ const (
 	MinMigrateGVGTime int64 = 1800 // 0.5 hour
 	// MaxMigrateGVGTime defines the max timeout to migrate gvg.
 	MaxMigrateGVGTime int64 = 3600 // 1 hour
+	// MinGCBucketMigrationTime defines the min timeout to gc bucket migration.
+	MinGCBucketMigrationTime int64 = 300 // 0.5 hour
+	// MaxGCBucketMigrationTime defines the max timeout to gc bucket migration.
+	MaxGCBucketMigrationTime int64 = 600 // 1 hour
 
 	// NotUseRetry defines the default task max retry.
 	NotUseRetry int64 = 0
@@ -78,6 +82,10 @@ const (
 	MinMigrateGVGRetry = 2
 	// MaxMigrateGVGRetry  defines the max retry number to migrate gvg.
 	MaxMigrateGVGRetry = 3
+	// MinGCBucketMigrationRetry defines the min retry number to gc bucket migration.
+	MinGCBucketMigrationRetry = 3
+	// MaxBucketMigrationRetry  defines the max retry number to gc bucket migration.
+	MaxBucketMigrationRetry = 5
 )
 
 // TaskTimeout returns the task timeout by task type and some task need payload size
@@ -184,6 +192,14 @@ func (g *GfSpBaseApp) TaskTimeout(task coretask.Task, size uint64) int64 {
 			return MaxMigrateGVGTime
 		}
 		return g.migrateGVGTimeout
+	case coretask.TypeTaskGCBucketMigration:
+		if g.gcBucketMigrationTimeout < MinGCBucketMigrationTime {
+			return MinGCBucketMigrationTime
+		}
+		if g.gcBucketMigrationTimeout > MaxGCBucketMigrationTime {
+			return MaxGCBucketMigrationTime
+		}
+		return g.gcBucketMigrationTimeout
 	}
 	return NotUseTimeout
 }
@@ -267,6 +283,14 @@ func (g *GfSpBaseApp) TaskMaxRetry(task coretask.Task) int64 {
 			return MaxMigrateGVGRetry
 		}
 		return g.migrateGVGRetry
+	case coretask.TypeTaskGCBucketMigration:
+		if g.gcBucketMigrationRetry < MinGCBucketMigrationRetry {
+			return MinMigrateGVGRetry
+		}
+		if g.gcBucketMigrationRetry > MaxBucketMigrationRetry {
+			return MaxMigrateGVGRetry
+		}
+		return g.gcBucketMigrationRetry
 	default:
 		return NotUseRetry
 	}
@@ -299,13 +323,15 @@ func (g *GfSpBaseApp) TaskPriority(task coretask.Task) coretask.TPriority {
 	case coretask.TypeTaskGCObject:
 		return coretask.UnSchedulingPriority
 	case coretask.TypeTaskGCZombiePiece:
-		return coretask.UnSchedulingPriority
+		return coretask.DefaultSmallerPriority / 4
 	case coretask.TypeTaskGCMeta:
-		return coretask.UnSchedulingPriority
+		return coretask.DefaultSmallerPriority / 4
 	case coretask.TypeTaskRecoverPiece:
 		return coretask.DefaultSmallerPriority / 4
 	case coretask.TypeTaskMigrateGVG:
 		return coretask.DefaultSmallerPriority
+	case coretask.TypeTaskGCBucketMigration:
+		return coretask.DefaultSmallerPriority
 	default:
 		return coretask.UnKnownTaskPriority
 	}

base/gfspapp/task_options_test.go

@@ -801,12 +801,12 @@ func TestGfSpBaseApp_TaskPriority(t *testing.T) {
 		{
 			name:         "gc zombie piece task",
 			task:         &gfsptask.GfSpGCZombiePieceTask{},
-			wantedResult: coretask.UnSchedulingPriority,
+			wantedResult: coretask.DefaultSmallerPriority / 4,
 		},
 		{
 			name:         "gc meta task",
 			task:         &gfsptask.GfSpGCMetaTask{},
-			wantedResult: coretask.UnSchedulingPriority,
+			wantedResult: coretask.DefaultSmallerPriority / 4,
 		},
 		{
 			name:         "recover piece task",

base/gfspclient/interface.go

@@ -89,13 +89,14 @@ type ManagerAPI interface {
 	NotifyMigrateSwapOut(ctx context.Context, swapOut *virtualgrouptypes.MsgSwapOut) error
 	GetTasksStats(ctx context.Context) (*gfspserver.TasksStats, error)
 	NotifyPreMigrateBucket(ctx context.Context, bucketID uint64) error
-	NotifyPostMigrateBucket(ctx context.Context, bucketID uint64) error
+	NotifyPostMigrateBucket(ctx context.Context, bmInfo *gfsptask.GfSpBucketMigrationInfo) error
 }
 
 // MetadataAPI for mock sue
 type MetadataAPI interface {
 	GetUserBucketsCount(ctx context.Context, account string, includeRemoved bool, opts ...grpc.DialOption) (int64, error)
 	ListDeletedObjectsByBlockNumberRange(ctx context.Context, spOperatorAddress string, startBlockNumber uint64, endBlockNumber uint64, includePrivate bool, opts ...grpc.DialOption) ([]*types.Object, uint64, error)
+	ListObjectsByBlockNumberRange(ctx context.Context, spOperatorAddress string, startBlockNumber uint64, endBlockNumber uint64, includePrivate bool, opts ...grpc.DialOption) ([]*types.Object, uint64, error)
 	GetUserBuckets(ctx context.Context, account string, includeRemoved bool, opts ...grpc.DialOption) ([]*types.VGFInfoBucket, error)
 	ListObjectsByBucketName(ctx context.Context, bucketName string, accountID string, maxKeys uint64, startAfter string, continuationToken string, delimiter string, prefix string, includeRemoved bool,
 		opts ...grpc.DialOption) (objects []*types.Object, keyCount, maxKeysRe uint64, isTruncated bool, nextContinuationToken, name, prefixRe, delimiterRe string, commonPrefixes []string, continuationTokenRe string, err error)

base/gfspclient/manager.go

@@ -90,6 +90,8 @@ func (s *GfSpClient) AskTask(ctx context.Context, limit corercmgr.Limit) (coreta
 		return t.RecoverPieceTask, nil
 	case *gfspserver.GfSpAskTaskResponse_MigrateGvgTask:
 		return t.MigrateGvgTask, nil
+	case *gfspserver.GfSpAskTaskResponse_GcBucketMigrationTask:
+		return t.GcBucketMigrationTask, nil
 	default:
 		return nil, ErrTypeMismatch
 	}
@@ -127,6 +129,8 @@ func (s *GfSpClient) ReportTask(ctx context.Context, report coretask.Task) error
 		req.Request = &gfspserver.GfSpReportTaskRequest_RecoverPieceTask{RecoverPieceTask: t}
 	case *gfsptask.GfSpMigrateGVGTask:
 		req.Request = &gfspserver.GfSpReportTaskRequest_MigrateGvgTask{MigrateGvgTask: t}
+	case *gfsptask.GfSpGCBucketMigrationTask:
+		req.Request = &gfspserver.GfSpReportTaskRequest_GcBucketMigrationTask{GcBucketMigrationTask: t}
 	default:
 		log.CtxErrorw(ctx, "unsupported task type to report")
 		return ErrTypeMismatch
@@ -218,14 +222,14 @@ func (s *GfSpClient) NotifyPreMigrateBucket(ctx context.Context, bucketID uint64
 	return nil
 }
 
-func (s *GfSpClient) NotifyPostMigrateBucket(ctx context.Context, bucketID uint64) error {
+func (s *GfSpClient) NotifyPostMigrateBucket(ctx context.Context, bmInfo *gfsptask.GfSpBucketMigrationInfo) error {
 	conn, connErr := s.ManagerConn(ctx)
 	if connErr != nil {
 		log.CtxErrorw(ctx, "client failed to connect manager", "error", connErr)
 		return ErrRPCUnknownWithDetail("client failed to connect manager, error: " + connErr.Error())
 	}
 	req := &gfspserver.GfSpNotifyPostMigrateBucketRequest{
-		BucketId: bucketID,
+		BucketMigrationInfo: bmInfo,
 	}
 	resp, err := gfspserver.NewGfSpManageServiceClient(conn).GfSpNotifyPostMigrate(ctx, req)
 	if err != nil {

base/gfspclient/metadata.go

@@ -54,6 +54,26 @@ func (s *GfSpClient) ListDeletedObjectsByBlockNumberRange(ctx context.Context, s
 	return resp.GetObjects(), uint64(resp.GetEndBlockNumber()), nil
 }
 
+func (s *GfSpClient) ListObjectsByBlockNumberRange(ctx context.Context, spOperatorAddress string, startBlockNumber uint64,
+	endBlockNumber uint64, includePrivate bool, opts ...grpc.DialOption) ([]*types.Object, uint64, error) {
+	conn, err := s.Connection(ctx, s.metadataEndpoint, opts...)
+	if err != nil {
+		return nil, uint64(0), ErrRPCUnknownWithDetail("client failed to connect metadata, error: " + err.Error())
+	}
+	defer conn.Close()
+	req := &types.GfSpListDeletedObjectsByBlockNumberRangeRequest{
+		StartBlockNumber: int64(startBlockNumber),
+		EndBlockNumber:   int64(endBlockNumber),
+		IncludePrivate:   includePrivate,
+	}
+	resp, err := types.NewGfSpMetadataServiceClient(conn).GfSpListDeletedObjectsByBlockNumberRange(ctx, req)
+	if err != nil {
+		log.CtxErrorw(ctx, "failed to list deleted objects by block number range", "error", err)
+		return nil, uint64(0), ErrRPCUnknownWithDetail("failed to list deleted objects by block number range, error: " + err.Error())
+	}
+	return resp.GetObjects(), uint64(resp.GetEndBlockNumber()), nil
+}
+
 func (s *GfSpClient) GetUserBuckets(ctx context.Context, account string, includeRemoved bool, opts ...grpc.DialOption) ([]*types.VGFInfoBucket, error) {
 	conn, err := s.Connection(ctx, s.metadataEndpoint, opts...)
 	if err != nil {
@@ -568,15 +588,15 @@ func (s *GfSpClient) ListObjectsInGVG(ctx context.Context, gvgID uint32, startAf
 	return resp.Objects, nil
 }
 
-func (s *GfSpClient) ListObjectsByGVGAndBucketForGC(ctx context.Context, gvgID uint32, bucketID uint64, startAfter uint64, limit uint32, opts ...grpc.DialOption) ([]*types.ObjectDetails, error) {
+func (s *GfSpClient) ListObjectsByGVGAndBucketForGC(ctx context.Context, dstGvgID uint32, bucketID uint64, startAfter uint64, limit uint32, opts ...grpc.DialOption) ([]*types.ObjectDetails, error) {
 	conn, connErr := s.Connection(ctx, s.metadataEndpoint, opts...)
 	if connErr != nil {
 		log.CtxErrorw(ctx, "client failed to connect metadata", "error", connErr)
 		return nil, ErrRPCUnknownWithDetail("client failed to connect metadata, error: " + connErr.Error())
 	}
 	defer conn.Close()
 	req := &types.GfSpListObjectsByGVGAndBucketForGCRequest{
-		GvgId:      gvgID,
+		DstGvgId:   dstGvgID,
 		BucketId:   bucketID,
 		StartAfter: startAfter,
 		Limit:      limit,