Merge pull request #1322 from bnb-chain/develop

 release: prepare release for v1.3.0

.github/workflows/docker-develop.yml

@@ -41,6 +41,13 @@ jobs:
           --label "org.opencontainers.image.revision=$(git rev-parse HEAD)" \
           --label "org.opencontainers.image.licenses=LGPL-3.0,GPL-3.0" \
           -f ./Dockerfile -t "${IMAGE_NAME}"
+          
+          docker build . \
+          --build-arg "GH_TOKEN=${{ secrets.GH_SECRET }}" \
+          --label "org.opencontainers.image.source=${IMAGE_SOURCE}" \
+          --label "org.opencontainers.image.revision=$(git rev-parse HEAD)" \
+          --label "org.opencontainers.image.licenses=LGPL-3.0,GPL-3.0" \
+          -f ./Dockerfile.distroless -t "${IMAGE_NAME}:distroless"
 
       - name: Log into registry
         run: echo "${{ secrets.GH_SECRET }}" | docker login ghcr.io -u ${{ github.actor }} --password-stdin
@@ -49,4 +56,6 @@ jobs:
         run: |
           VERSION=$(echo "${{ github.sha }}" | sed -e 's,.*/\(.*\),\1,')
           docker tag $IMAGE_NAME $IMAGE_NAME:$VERSION
+          docker tag ${IMAGE_NAME}:distroless $IMAGE_NAME:$VERSION-distroless
           docker push $IMAGE_NAME:$VERSION
+          docker push $IMAGE_NAME:$VERSION-distroless

CHANGELOG.md

@@ -1,5 +1,16 @@
 # Changelog
 
+## v1.3.0
+This release contains 2 bugfix and 2 feature.
+
+BUGFIXES
+* [#1304](https://github.com/bnb-chain/greenfield-storage-provider/pull/1304) fix: fix sp unhealthy bug
+* [#1311](https://github.com/bnb-chain/greenfield-storage-provider/pull/1311) fix: fix health check
+
+FEATURES
+* [#1279](https://github.com/bnb-chain/greenfield-storage-provider/pull/1279) feat: re-implementation of sp exit
+* [#1312](https://github.com/bnb-chain/greenfield-storage-provider/pull/1312) perf: blocksyncer sql commit
+
 ## v1.2.5
 This release contains 1 bugfixes.
 

base/gfspapp/app_options.go

@@ -153,6 +153,12 @@ const (
 	SignerFailureGfSpBucketMigrateInfo      = "signer_gfsp_bucket_migrate_info_failure"
 	SignerSuccessRejectMigrateBucket        = "signer_reject_migrate_bucket_success"
 	SignerFailureRejectMigrateBucket        = "signer_reject_migrate_bucket_failure"
+	SignerSuccessSwapIn                     = "signer_swap_in_success"
+	SignerFailureSwapIn                     = "signer_swap_in_failure"
+	SignerSuccessCompleteSwapIn             = "signer_complete_swap_in_success"
+	SignerFailureCompleteSwapIn             = "signer_complete_swap_in_failure"
+	SignerSuccessCancelSwapIn               = "signer_cancel_swap_in_success"
+	SignerFailureCancelSwapIn               = "signer_cancel_swap_in_failure"
 
 	SignerSuccessDeposit = "signer_deposit_success"
 	SignerFailureDeposit = "signer_deposit_failure"

base/gfspapp/manage_server.go

@@ -360,22 +360,48 @@ func (g *GfSpBaseApp) GfSpQueryTasksStats(ctx context.Context, _ *gfspserver.GfS
 	}, nil
 }
 
-func (g *GfSpBaseApp) GfSpNotifyPreMigrate(ctx context.Context, req *gfspserver.GfSpNotifyPreMigrateBucketRequest) (
+func (g *GfSpBaseApp) GfSpQueryBucketMigrationProgress(ctx context.Context, req *gfspserver.GfSpQueryBucketMigrationProgressRequest) (
+	*gfspserver.GfSpQueryBucketMigrationProgressResponse, error) {
+	var (
+		progress *gfspserver.MigrateBucketProgressMeta
+		err      error
+	)
+	if progress, err = g.manager.QueryBucketMigrationProgress(ctx, req.GetBucketId()); err != nil {
+		log.CtxErrorw(ctx, "failed to query bucket migration progress", "bucket_id", req.GetBucketId(), "error", err)
+		return nil, err
+	}
+
+	return &gfspserver.GfSpQueryBucketMigrationProgressResponse{
+		Progress: progress,
+	}, nil
+}
+
+func (g *GfSpBaseApp) GfSpNotifyPreMigrateBucketAndDeductQuota(ctx context.Context, req *gfspserver.GfSpNotifyPreMigrateBucketRequest) (
 	*gfspserver.GfSpNotifyPreMigrateBucketResponse, error) {
-	if err := g.manager.NotifyPreMigrateBucket(ctx, req.GetBucketId()); err != nil {
+	var (
+		quota *gfsptask.GfSpBucketQuotaInfo
+		err   error
+	)
+	if quota, err = g.manager.NotifyPreMigrateBucketAndDeductQuota(ctx, req.GetBucketId()); err != nil {
+		log.CtxErrorw(ctx, "failed to notify pre migrate bucket and deduct quota", "bucket_id", req.GetBucketId(), "error", err)
 		return nil, err
 	}
 
-	return &gfspserver.GfSpNotifyPreMigrateBucketResponse{}, nil
+	return &gfspserver.GfSpNotifyPreMigrateBucketResponse{Quota: quota}, nil
 }
 
-func (g *GfSpBaseApp) GfSpNotifyPostMigrate(ctx context.Context, req *gfspserver.GfSpNotifyPostMigrateBucketRequest) (
+func (g *GfSpBaseApp) GfSpNotifyPostMigrateAndRecoupQuota(ctx context.Context, req *gfspserver.GfSpNotifyPostMigrateBucketRequest) (
 	*gfspserver.GfSpNotifyPostMigrateBucketResponse, error) {
-	if err := g.manager.NotifyPostMigrateBucket(ctx, req.GetBucketMigrationInfo()); err != nil {
+	var (
+		quota *gfsptask.GfSpBucketQuotaInfo
+		err   error
+	)
+	if quota, err = g.manager.NotifyPostMigrateBucketAndRecoupQuota(ctx, req.GetBucketMigrationInfo()); err != nil {
+		log.CtxErrorw(ctx, "failed to notify post migrate bucket and recoup quota", "bucket_migration_info", req.GetBucketMigrationInfo(), "error", err)
 		return nil, err
 	}
 
-	return &gfspserver.GfSpNotifyPostMigrateBucketResponse{}, nil
+	return &gfspserver.GfSpNotifyPostMigrateBucketResponse{Quota: quota}, nil
 }
 
 func (g *GfSpBaseApp) GfSpResetRecoveryFailedList(ctx context.Context, _ *gfspserver.GfSpResetRecoveryFailedListRequest) (
@@ -385,3 +411,23 @@ func (g *GfSpBaseApp) GfSpResetRecoveryFailedList(ctx context.Context, _ *gfspse
 		RecoveryFailedList: recoveryFailedList,
 	}, nil
 }
+
+func (g *GfSpBaseApp) GfSpTriggerRecoverForSuccessorSP(ctx context.Context, req *gfspserver.GfSpTriggerRecoverForSuccessorSPRequest) (
+	*gfspserver.GfSpTriggerRecoverForSuccessorSPResponse, error) {
+	err := g.manager.TriggerRecoverForSuccessorSP(ctx, req.GetVgfId(), req.GetGvgId(), req.ReplicateIndex)
+	if err != nil {
+		return nil, err
+	}
+	return &gfspserver.GfSpTriggerRecoverForSuccessorSPResponse{}, nil
+}
+
+func (g *GfSpBaseApp) GfSpQueryRecoverProcess(ctx context.Context, req *gfspserver.GfSpQueryRecoverProcessRequest) (*gfspserver.GfSpQueryRecoverProcessResponse, error) {
+	gvgStats, flag, err := g.manager.QueryRecoverProcess(ctx, req.GetVgfId(), req.GetGvgId())
+	if err != nil {
+		return nil, err
+	}
+	return &gfspserver.GfSpQueryRecoverProcessResponse{
+		RecoverProcesses: gvgStats,
+		Executing:        flag,
+	}, nil
+}

base/gfspapp/manage_server_test.go

@@ -632,3 +632,53 @@ func TestGfSpBaseApp_GfSpNotifyMigrateSwapOutFailure(t *testing.T) {
 	assert.Equal(t, mockErr, err)
 	assert.Nil(t, result)
 }
+
+func TestGfSpBaseApp_GfSpNotifyPreMigrateBucketAndDeductQuotaSuccess(t *testing.T) {
+	g := setup(t)
+	ctrl := gomock.NewController(t)
+	m := module.NewMockManager(ctrl)
+	g.manager = m
+	quota := &gfsptask.GfSpBucketQuotaInfo{BucketId: 2}
+	m.EXPECT().NotifyPreMigrateBucketAndDeductQuota(gomock.Any(), gomock.Any()).Return(quota, nil).Times(1)
+	req := &gfspserver.GfSpNotifyPreMigrateBucketRequest{BucketId: 2}
+	result, err := g.GfSpNotifyPreMigrateBucketAndDeductQuota(context.TODO(), req)
+	assert.Nil(t, err)
+	assert.Equal(t, &gfspserver.GfSpNotifyPreMigrateBucketResponse{Err: nil, Quota: quota}, result)
+}
+
+func TestGfSpBaseApp_GfSpNotifyPreMigrateBucketAndDeductQuotaFailure(t *testing.T) {
+	g := setup(t)
+	ctrl := gomock.NewController(t)
+	m := module.NewMockManager(ctrl)
+	g.manager = m
+	m.EXPECT().NotifyPreMigrateBucketAndDeductQuota(gomock.Any(), gomock.Any()).Return(nil, mockErr).Times(1)
+	req := &gfspserver.GfSpNotifyPreMigrateBucketRequest{BucketId: 2}
+	result, err := g.GfSpNotifyPreMigrateBucketAndDeductQuota(context.TODO(), req)
+	assert.Equal(t, mockErr, err)
+	assert.Nil(t, result)
+}
+
+func TestGfSpBaseApp_GfSpNotifyPostMigrateBucketAndRecoupQuotaSuccess(t *testing.T) {
+	g := setup(t)
+	ctrl := gomock.NewController(t)
+	m := module.NewMockManager(ctrl)
+	g.manager = m
+	quota := &gfsptask.GfSpBucketQuotaInfo{BucketId: 2}
+	m.EXPECT().NotifyPostMigrateBucketAndRecoupQuota(gomock.Any(), gomock.Any()).Return(quota, nil).Times(1)
+	req := &gfspserver.GfSpNotifyPostMigrateBucketRequest{BucketId: 2}
+	result, err := g.GfSpNotifyPostMigrateAndRecoupQuota(context.TODO(), req)
+	assert.Nil(t, err)
+	assert.Equal(t, &gfspserver.GfSpNotifyPostMigrateBucketResponse{Quota: quota}, result)
+}
+
+func TestGfSpBaseApp_GfSpPostMigrateBucketAndRecoupQuotaFailure(t *testing.T) {
+	g := setup(t)
+	ctrl := gomock.NewController(t)
+	m := module.NewMockManager(ctrl)
+	g.manager = m
+	m.EXPECT().NotifyPostMigrateBucketAndRecoupQuota(gomock.Any(), gomock.Any()).Return(nil, mockErr).Times(1)
+	req := &gfspserver.GfSpNotifyPostMigrateBucketRequest{BucketId: 2}
+	result, err := g.GfSpNotifyPostMigrateAndRecoupQuota(context.TODO(), req)
+	assert.Equal(t, mockErr, err)
+	assert.Nil(t, result)
+}

base/gfspapp/sign_server.go

@@ -291,6 +291,36 @@ func (g *GfSpBaseApp) GfSpSign(ctx context.Context, req *gfspserver.GfSpSignRequ
 			metrics.ReqCounter.WithLabelValues(SignerSuccessRejectMigrateBucket).Inc()
 			metrics.ReqTime.WithLabelValues(SignerSuccessRejectMigrateBucket).Observe(time.Since(startTime).Seconds())
 		}
+	case *gfspserver.GfSpSignRequest_ReserveSwapIn:
+		txHash, err = g.signer.ReserveSwapIn(ctx, t.ReserveSwapIn)
+		if err != nil {
+			log.CtxErrorw(ctx, "failed to reserve swap in", "error", err)
+			metrics.ReqCounter.WithLabelValues(SignerFailureSwapIn).Inc()
+			metrics.ReqTime.WithLabelValues(SignerFailureSwapIn).Observe(time.Since(startTime).Seconds())
+		} else {
+			metrics.ReqCounter.WithLabelValues(SignerSuccessSwapIn).Inc()
+			metrics.ReqTime.WithLabelValues(SignerSuccessSwapIn).Observe(time.Since(startTime).Seconds())
+		}
+	case *gfspserver.GfSpSignRequest_CompleteSwapIn:
+		txHash, err = g.signer.CompleteSwapIn(ctx, t.CompleteSwapIn)
+		if err != nil {
+			log.CtxErrorw(ctx, "failed to complete swap in", "error", err)
+			metrics.ReqCounter.WithLabelValues(SignerFailureCompleteSwapIn).Inc()
+			metrics.ReqTime.WithLabelValues(SignerFailureCompleteSwapIn).Observe(time.Since(startTime).Seconds())
+		} else {
+			metrics.ReqCounter.WithLabelValues(SignerSuccessCompleteSwapIn).Inc()
+			metrics.ReqTime.WithLabelValues(SignerSuccessCompleteSwapIn).Observe(time.Since(startTime).Seconds())
+		}
+	case *gfspserver.GfSpSignRequest_CancelSwapIn:
+		txHash, err = g.signer.CancelSwapIn(ctx, t.CancelSwapIn)
+		if err != nil {
+			log.CtxErrorw(ctx, "failed to cancel swap in", "error", err)
+			metrics.ReqCounter.WithLabelValues(SignerFailureCancelSwapIn).Inc()
+			metrics.ReqTime.WithLabelValues(SignerFailureCancelSwapIn).Observe(time.Since(startTime).Seconds())
+		} else {
+			metrics.ReqCounter.WithLabelValues(SignerSuccessCancelSwapIn).Inc()
+			metrics.ReqTime.WithLabelValues(SignerSuccessCancelSwapIn).Observe(time.Since(startTime).Seconds())
+		}
 	case *gfspserver.GfSpSignRequest_Deposit:
 		txHash, err = g.signer.Deposit(ctx, t.Deposit)
 		if err != nil {

base/gfspclient/gater.go

@@ -67,6 +67,8 @@ const (
 	GnfdUnsignedApprovalMsgHeader = "X-Gnfd-Unsigned-Msg"
 	// GnfdSignedApprovalMsgHeader defines signed msg, which is used by get-approval
 	GnfdSignedApprovalMsgHeader = "X-Gnfd-Signed-Msg"
+	// GnfdQuotaInfoHeader defines quota info, which is used by sp
+	GnfdQuotaInfoHeader = "X-Gnfd-Quota-Info"
 )
 
 func (s *GfSpClient) ReplicatePieceToSecondary(ctx context.Context, endpoint string, receive coretask.ReceivePieceTask,
@@ -243,13 +245,13 @@ func (s *GfSpClient) QueryLatestBucketQuota(ctx context.Context, endpoint string
 		return gfsptask.GfSpBucketQuotaInfo{}, fmt.Errorf("failed to query latest bucket quota, bucket(%s), status_code(%d), endpoint(%s)", queryMsg, resp.StatusCode, endpoint)
 	}
 
-	signedMsg, err := hex.DecodeString(resp.Header.Get(GnfdSignedApprovalMsgHeader))
+	quotaInfoMsg, err := hex.DecodeString(resp.Header.Get(GnfdQuotaInfoHeader))
 	if err != nil {
 		return gfsptask.GfSpBucketQuotaInfo{}, err
 	}
 
 	quotaResult := gfsptask.GfSpBucketQuotaInfo{}
-	if err = proto.Unmarshal(signedMsg, &quotaResult); err != nil {
+	if err = proto.Unmarshal(quotaInfoMsg, &quotaResult); err != nil {
 		return gfsptask.GfSpBucketQuotaInfo{}, err
 	}
 
@@ -280,13 +282,13 @@ func (s *GfSpClient) PreMigrateBucket(ctx context.Context, srcSPEndpoint string,
 		return gfsptask.GfSpBucketQuotaInfo{}, fmt.Errorf("failed to pre migrate bucket, bucket_migration_info(%s), status_code(%d), endpoint(%s)", preMsg, resp.StatusCode, srcSPEndpoint)
 	}
 
-	signedMsg, err := hex.DecodeString(resp.Header.Get(GnfdSignedApprovalMsgHeader))
+	quotaInfoMsg, err := hex.DecodeString(resp.Header.Get(GnfdQuotaInfoHeader))
 	if err != nil {
 		return gfsptask.GfSpBucketQuotaInfo{}, err
 	}
 
 	quotaResult := gfsptask.GfSpBucketQuotaInfo{}
-	if err = proto.Unmarshal(signedMsg, &quotaResult); err != nil {
+	if err = proto.Unmarshal(quotaInfoMsg, &quotaResult); err != nil {
 		return gfsptask.GfSpBucketQuotaInfo{}, err
 	}
 
@@ -319,13 +321,13 @@ func (s *GfSpClient) PostMigrateBucket(ctx context.Context, srcSPEndpoint string
 		return gfsptask.GfSpBucketQuotaInfo{}, fmt.Errorf("failed to post migrate bucket to src sp, bucket(%d), status_code(%d), endpoint(%s)", bucketID, resp.StatusCode, srcSPEndpoint)
 	}
 
-	signedMsg, err := hex.DecodeString(resp.Header.Get(GnfdSignedApprovalMsgHeader))
+	quotaInfoMsg, err := hex.DecodeString(resp.Header.Get(GnfdQuotaInfoHeader))
 	if err != nil {
 		return gfsptask.GfSpBucketQuotaInfo{}, err
 	}
 
 	quotaResult := gfsptask.GfSpBucketQuotaInfo{}
-	if err = proto.Unmarshal(signedMsg, &quotaResult); err != nil {
+	if err = proto.Unmarshal(quotaInfoMsg, &quotaResult); err != nil {
 		return gfsptask.GfSpBucketQuotaInfo{}, err
 	}
 
@@ -358,13 +360,13 @@ func (s *GfSpClient) QuerySPHasEnoughQuotaForMigrateBucket(ctx context.Context,
 			bucketID, resp.StatusCode, srcSPEndpoint)
 	}
 
-	signedMsg, err := hex.DecodeString(resp.Header.Get(GnfdSignedApprovalMsgHeader))
+	quotaInfoMsg, err := hex.DecodeString(resp.Header.Get(GnfdQuotaInfoHeader))
 	if err != nil {
 		return err
 	}
 
 	quotaResult := gfsptask.GfSpBucketQuotaInfo{}
-	if err = proto.Unmarshal(signedMsg, &quotaResult); err != nil {
+	if err = proto.Unmarshal(quotaInfoMsg, &quotaResult); err != nil {
 		return err
 	}
 

base/gfspclient/helper_test.go

@@ -236,6 +236,14 @@ func (mockDownloaderServer) GfSpDeductQuotaForBucketMigrate(ctx context.Context,
 
 type mockManagerServer struct{}
 
+func (s mockManagerServer) GfSpTriggerRecoverForSuccessorSP(ctx context.Context, request *gfspserver.GfSpTriggerRecoverForSuccessorSPRequest) (*gfspserver.GfSpTriggerRecoverForSuccessorSPResponse, error) {
+	return &gfspserver.GfSpTriggerRecoverForSuccessorSPResponse{Err: ErrExceptionsStream}, nil
+}
+
+func (s mockManagerServer) GfSpQueryRecoverProcess(ctx context.Context, request *gfspserver.GfSpQueryRecoverProcessRequest) (*gfspserver.GfSpQueryRecoverProcessResponse, error) {
+	return &gfspserver.GfSpQueryRecoverProcessResponse{Err: ErrExceptionsStream}, nil
+}
+
 func (mockManagerServer) GfSpBeginTask(ctx context.Context, req *gfspserver.GfSpBeginTaskRequest) (
 	*gfspserver.GfSpBeginTaskResponse, error) {
 	switch req.Request.(type) {
@@ -348,14 +356,23 @@ func (s mockManagerServer) GfSpQueryTasksStats(ctx context.Context, req *gfspser
 	}, nil
 }
 
-func (s mockManagerServer) GfSpNotifyPreMigrate(ctx context.Context, req *gfspserver.GfSpNotifyPreMigrateBucketRequest) (*gfspserver.GfSpNotifyPreMigrateBucketResponse, error) {
+func (s mockManagerServer) GfSpQueryBucketMigrationProgress(ctx context.Context, req *gfspserver.GfSpQueryBucketMigrationProgressRequest) (*gfspserver.GfSpQueryBucketMigrationProgressResponse, error) {
+	if req == nil {
+		return nil, mockRPCErr
+	}
+	return &gfspserver.GfSpQueryBucketMigrationProgressResponse{
+		Progress: &gfspserver.MigrateBucketProgressMeta{},
+	}, nil
+}
+
+func (s mockManagerServer) GfSpNotifyPreMigrateBucketAndDeductQuota(ctx context.Context, req *gfspserver.GfSpNotifyPreMigrateBucketRequest) (*gfspserver.GfSpNotifyPreMigrateBucketResponse, error) {
 	if req == nil {
 		return nil, mockRPCErr
 	}
 	return &gfspserver.GfSpNotifyPreMigrateBucketResponse{}, nil
 }
 
-func (s mockManagerServer) GfSpNotifyPostMigrate(ctx context.Context, req *gfspserver.GfSpNotifyPostMigrateBucketRequest) (*gfspserver.GfSpNotifyPostMigrateBucketResponse, error) {
+func (s mockManagerServer) GfSpNotifyPostMigrateAndRecoupQuota(ctx context.Context, req *gfspserver.GfSpNotifyPostMigrateBucketRequest) (*gfspserver.GfSpNotifyPostMigrateBucketResponse, error) {
 	if req == nil {
 		return nil, mockRPCErr
 	}

base/gfspclient/interface.go

@@ -88,8 +88,9 @@ type ManagerAPI interface {
 	PickVirtualGroupFamilyID(ctx context.Context, task coretask.ApprovalCreateBucketTask) (uint32, error)
 	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, bmInfo *gfsptask.GfSpBucketMigrationInfo) error
+	GetMigrateBucketProgress(ctx context.Context, bucketID uint64) (*gfspserver.MigrateBucketProgressMeta, error)
+	NotifyPostMigrateBucketAndRecoupQuota(ctx context.Context, bmInfo *gfsptask.GfSpBucketMigrationInfo) (*gfsptask.GfSpBucketQuotaInfo, error)
+	NotifyPreMigrateBucketAndDeductQuota(ctx context.Context, bucketID uint64) (*gfsptask.GfSpBucketQuotaInfo, error)
 }
 
 // MetadataAPI for mock sue
@@ -110,7 +111,7 @@ type MetadataAPI interface {
 	GetBucketMeta(ctx context.Context, bucketName string, includePrivate bool, opts ...grpc.DialOption) (*types.VGFInfoBucket, *payment_types.StreamRecord, error)
 	GetEndpointBySpID(ctx context.Context, spID uint32, opts ...grpc.DialOption) (string, error)
 	GetBucketReadQuota(ctx context.Context, bucket *storagetypes.BucketInfo, yearMonth string, opts ...grpc.DialOption) (uint64, uint64, uint64, uint64, error)
-	GetLatestBucketReadQuota(ctx context.Context, bucketID uint64, opts ...grpc.DialOption) (gfsptask.GfSpBucketQuotaInfo, error)
+	GetLatestBucketReadQuota(ctx context.Context, bucketID uint64, opts ...grpc.DialOption) (*gfsptask.GfSpBucketQuotaInfo, error)
 	ListBucketReadRecord(ctx context.Context, bucket *storagetypes.BucketInfo, startTimestampUs, endTimestampUs, maxRecordNum int64, opts ...grpc.DialOption) ([]*types.ReadRecord, int64, error)
 	GetUploadObjectState(ctx context.Context, objectID uint64, opts ...grpc.DialOption) (int32, string, error)
 	GetUploadObjectSegment(ctx context.Context, objectID uint64, opts ...grpc.DialOption) (uint32, error)
@@ -127,6 +128,7 @@ type MetadataAPI interface {
 	ListGlobalVirtualGroupsByBucket(ctx context.Context, bucketID uint64, opts ...grpc.DialOption) ([]*virtualgrouptypes.GlobalVirtualGroup, error)
 	ListGlobalVirtualGroupsBySecondarySP(ctx context.Context, spID uint32, opts ...grpc.DialOption) ([]*virtualgrouptypes.GlobalVirtualGroup, error)
 	ListMigrateBucketEvents(ctx context.Context, blockID uint64, spID uint32, opts ...grpc.DialOption) ([]*types.ListMigrateBucketEvents, error)
+	ListCompleteMigrationBucketEvents(ctx context.Context, blockID uint64, srcSpID uint32, opts ...grpc.DialOption) ([]*storagetypes.EventCompleteMigrationBucket, error)
 	ListSwapOutEvents(ctx context.Context, blockID uint64, spID uint32, opts ...grpc.DialOption) ([]*types.ListSwapOutEvents, error)
 	ListSpExitEvents(ctx context.Context, blockID uint64, spID uint32, opts ...grpc.DialOption) (*types.ListSpExitEvents, error)
 	GetObjectByID(ctx context.Context, objectID uint64, opts ...grpc.DialOption) (*storagetypes.ObjectInfo, error)
@@ -192,6 +194,8 @@ type SignerAPI interface {
 	SignMigrateGVG(ctx context.Context, task *gfsptask.GfSpMigrateGVGTask) ([]byte, error)
 	SignBucketMigrationInfo(ctx context.Context, task *gfsptask.GfSpBucketMigrationInfo) ([]byte, error)
 	RejectMigrateBucket(ctx context.Context, rejectMigrateBucket *storagetypes.MsgRejectMigrateBucket) (string, error)
+	ReserveSwapIn(ctx context.Context, reserveSwapIn *virtualgrouptypes.MsgReserveSwapIn) (string, error)
+	CompleteSwapIn(ctx context.Context, completeSwpIn *virtualgrouptypes.MsgCompleteSwapIn) (string, error)
 	Deposit(ctx context.Context, deposit *virtualgrouptypes.MsgDeposit) (string, error)
 	DeleteGlobalVirtualGroup(ctx context.Context, deleteGVG *virtualgrouptypes.MsgDeleteGlobalVirtualGroup) (string, error)
 }