feat: add AnnouncePeers proto and fix typo (#343)

feat: add AnnouncePeers proto

1. add AnnouncePeers api
2. fix typo of dfdaemon

Signed-off-by: BruceAko <chongzhi@hust.edu.cn>

Cargo.lock

@@ -160,7 +160,7 @@ dependencies = [
 
 [[package]]
 name = "dragonfly-api"
-version = "2.0.132"
+version = "2.0.133"
 dependencies = [
  "prost 0.11.9",
  "prost-types 0.12.6",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dragonfly-api"
-version = "2.0.132"
+version = "2.0.133"
 authors = ["Gaius <gaius.qi@gmail.com>"]
 edition = "2021"
 license = "Apache-2.0"

pkg/apis/common/v2/common.pb.go

@@ -103,8 +103,8 @@ func (SizeScope) EnumDescriptor() ([]byte, []int) {
 type TaskType int32
 
 const (
-	// DFDAEMON is dfdeamon type of task,
+	// DFDAEMON is dfdaemon type of task,
-	// dfdeamon task is a normal p2p task.
+	// dfdaemon task is a normal p2p task.
 	TaskType_DFDAEMON TaskType = 0
 	// DFCACHE is dfcache type of task,
 	// dfcache task is a cache task, and the task url is fake url.
@@ -439,7 +439,7 @@ type CachePeer struct {
 	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
 	// Persistent represents whether the cache peer is persistent.
 	// If the cache peer is persistent, the cache peer will
-	// not be deleted when dfdamon runs garbage collection.
+	// not be deleted when dfdaemon runs garbage collection.
 	Persistent bool `protobuf:"varint,2,opt,name=persistent,proto3" json:"persistent,omitempty"`
 	// Peer downloads costs time.
 	Cost *durationpb.Duration `protobuf:"bytes,3,opt,name=cost,proto3" json:"cost,omitempty"`

pkg/apis/common/v2/common.proto

@@ -44,8 +44,8 @@ enum SizeScope {
 
 // TaskType represents type of task.
 enum TaskType {
-  // DFDAEMON is dfdeamon type of task,
+  // DFDAEMON is dfdaemon type of task,
-  // dfdeamon task is a normal p2p task.
+  // dfdaemon task is a normal p2p task.
   DFDAEMON = 0;
 
   // DFCACHE is dfcache type of task,
@@ -138,7 +138,7 @@ message CachePeer {
   string id = 1 [(validate.rules).string.min_len = 1];
   // Persistent represents whether the cache peer is persistent.
   // If the cache peer is persistent, the cache peer will
-  // not be deleted when dfdamon runs garbage collection.
+  // not be deleted when dfdaemon runs garbage collection.
   bool persistent = 2;
   // Peer downloads costs time.
   google.protobuf.Duration cost = 3 [(validate.rules).duration.required = true];

pkg/apis/dfdaemon/v2/dfdaemon.pb.go

@@ -733,7 +733,7 @@ type DownloadCacheTaskRequest struct {
 	TaskId string `protobuf:"bytes,2,opt,name=task_id,json=taskId,proto3" json:"task_id,omitempty"`
 	// Persistent represents whether the cache task is persistent.
 	// If the cache task is persistent, the cache peer will
-	// not be deleted when dfdamon runs garbage collection.
+	// not be deleted when dfdaemon runs garbage collection.
 	Persistent bool `protobuf:"varint,3,opt,name=persistent,proto3" json:"persistent,omitempty"`
 	// Tag is used to distinguish different cache tasks.
 	Tag *string `protobuf:"bytes,4,opt,name=tag,proto3,oneof" json:"tag,omitempty"`

pkg/apis/dfdaemon/v2/dfdaemon.proto

@@ -133,7 +133,7 @@ message DownloadCacheTaskRequest {
   string task_id = 2 [(validate.rules).string.min_len = 1];
   // Persistent represents whether the cache task is persistent.
   // If the cache task is persistent, the cache peer will
-  // not be deleted when dfdamon runs garbage collection.
+  // not be deleted when dfdaemon runs garbage collection.
   bool persistent = 3;
   // Tag is used to distinguish different cache tasks.
   optional string tag = 4;

pkg/apis/scheduler/v2/mocks/scheduler_mock.go

@@ -104,6 +104,26 @@ func (mr *MockSchedulerClientMockRecorder) AnnouncePeer(ctx any, opts ...any) *g
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AnnouncePeer", reflect.TypeOf((*MockSchedulerClient)(nil).AnnouncePeer), varargs...)
 }
 
+// AnnouncePeers mocks base method.
+func (m *MockSchedulerClient) AnnouncePeers(ctx context.Context, opts ...grpc.CallOption) (scheduler.Scheduler_AnnouncePeersClient, error) {
+	m.ctrl.T.Helper()
+	varargs := []any{ctx}
+	for _, a := range opts {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "AnnouncePeers", varargs...)
+	ret0, _ := ret[0].(scheduler.Scheduler_AnnouncePeersClient)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// AnnouncePeers indicates an expected call of AnnouncePeers.
+func (mr *MockSchedulerClientMockRecorder) AnnouncePeers(ctx any, opts ...any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]any{ctx}, opts...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AnnouncePeers", reflect.TypeOf((*MockSchedulerClient)(nil).AnnouncePeers), varargs...)
+}
+
 // DeleteCachePeer mocks base method.
 func (m *MockSchedulerClient) DeleteCachePeer(ctx context.Context, in *scheduler.DeleteCachePeerRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
 	m.ctrl.T.Helper()
@@ -304,24 +324,64 @@ func (mr *MockSchedulerClientMockRecorder) SyncProbes(ctx any, opts ...any) *gom
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SyncProbes", reflect.TypeOf((*MockSchedulerClient)(nil).SyncProbes), varargs...)
 }
 
-// UploadCacheTask mocks base method.
+// UploadCacheTaskFailed mocks base method.
-func (m *MockSchedulerClient) UploadCacheTask(ctx context.Context, opts ...grpc.CallOption) (scheduler.Scheduler_UploadCacheTaskClient, error) {
+func (m *MockSchedulerClient) UploadCacheTaskFailed(ctx context.Context, in *scheduler.UploadCacheTaskFailedRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
 	m.ctrl.T.Helper()
-	varargs := []any{ctx}
+	varargs := []any{ctx, in}
 	for _, a := range opts {
 		varargs = append(varargs, a)
 	}
-	ret := m.ctrl.Call(m, "UploadCacheTask", varargs...)
+	ret := m.ctrl.Call(m, "UploadCacheTaskFailed", varargs...)
-	ret0, _ := ret[0].(scheduler.Scheduler_UploadCacheTaskClient)
+	ret0, _ := ret[0].(*emptypb.Empty)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
-// UploadCacheTask indicates an expected call of UploadCacheTask.
+// UploadCacheTaskFailed indicates an expected call of UploadCacheTaskFailed.
-func (mr *MockSchedulerClientMockRecorder) UploadCacheTask(ctx any, opts ...any) *gomock.Call {
+func (mr *MockSchedulerClientMockRecorder) UploadCacheTaskFailed(ctx, in any, opts ...any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	varargs := append([]any{ctx}, opts...)
+	varargs := append([]any{ctx, in}, opts...)
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UploadCacheTask", reflect.TypeOf((*MockSchedulerClient)(nil).UploadCacheTask), varargs...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UploadCacheTaskFailed", reflect.TypeOf((*MockSchedulerClient)(nil).UploadCacheTaskFailed), varargs...)
+}
+
+// UploadCacheTaskFinished mocks base method.
+func (m *MockSchedulerClient) UploadCacheTaskFinished(ctx context.Context, in *scheduler.UploadCacheTaskFinishedRequest, opts ...grpc.CallOption) (*common.CacheTask, error) {
+	m.ctrl.T.Helper()
+	varargs := []any{ctx, in}
+	for _, a := range opts {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "UploadCacheTaskFinished", varargs...)
+	ret0, _ := ret[0].(*common.CacheTask)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// UploadCacheTaskFinished indicates an expected call of UploadCacheTaskFinished.
+func (mr *MockSchedulerClientMockRecorder) UploadCacheTaskFinished(ctx, in any, opts ...any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]any{ctx, in}, opts...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UploadCacheTaskFinished", reflect.TypeOf((*MockSchedulerClient)(nil).UploadCacheTaskFinished), varargs...)
+}
+
+// UploadCacheTaskStarted mocks base method.
+func (m *MockSchedulerClient) UploadCacheTaskStarted(ctx context.Context, in *scheduler.UploadCacheTaskStartedRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
+	m.ctrl.T.Helper()
+	varargs := []any{ctx, in}
+	for _, a := range opts {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "UploadCacheTaskStarted", varargs...)
+	ret0, _ := ret[0].(*emptypb.Empty)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// UploadCacheTaskStarted indicates an expected call of UploadCacheTaskStarted.
+func (mr *MockSchedulerClientMockRecorder) UploadCacheTaskStarted(ctx, in any, opts ...any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]any{ctx, in}, opts...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UploadCacheTaskStarted", reflect.TypeOf((*MockSchedulerClient)(nil).UploadCacheTaskStarted), varargs...)
 }
 
 // MockScheduler_AnnouncePeerClient is a mock of Scheduler_AnnouncePeerClient interface.
@@ -461,6 +521,143 @@ func (mr *MockScheduler_AnnouncePeerClientMockRecorder) Trailer() *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Trailer", reflect.TypeOf((*MockScheduler_AnnouncePeerClient)(nil).Trailer))
 }
 
+// MockScheduler_AnnouncePeersClient is a mock of Scheduler_AnnouncePeersClient interface.
+type MockScheduler_AnnouncePeersClient struct {
+	ctrl     *gomock.Controller
+	recorder *MockScheduler_AnnouncePeersClientMockRecorder
+}
+
+// MockScheduler_AnnouncePeersClientMockRecorder is the mock recorder for MockScheduler_AnnouncePeersClient.
+type MockScheduler_AnnouncePeersClientMockRecorder struct {
+	mock *MockScheduler_AnnouncePeersClient
+}
+
+// NewMockScheduler_AnnouncePeersClient creates a new mock instance.
+func NewMockScheduler_AnnouncePeersClient(ctrl *gomock.Controller) *MockScheduler_AnnouncePeersClient {
+	mock := &MockScheduler_AnnouncePeersClient{ctrl: ctrl}
+	mock.recorder = &MockScheduler_AnnouncePeersClientMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockScheduler_AnnouncePeersClient) EXPECT() *MockScheduler_AnnouncePeersClientMockRecorder {
+	return m.recorder
+}
+
+// CloseSend mocks base method.
+func (m *MockScheduler_AnnouncePeersClient) CloseSend() error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CloseSend")
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// CloseSend indicates an expected call of CloseSend.
+func (mr *MockScheduler_AnnouncePeersClientMockRecorder) CloseSend() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CloseSend", reflect.TypeOf((*MockScheduler_AnnouncePeersClient)(nil).CloseSend))
+}
+
+// Context mocks base method.
+func (m *MockScheduler_AnnouncePeersClient) Context() context.Context {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Context")
+	ret0, _ := ret[0].(context.Context)
+	return ret0
+}
+
+// Context indicates an expected call of Context.
+func (mr *MockScheduler_AnnouncePeersClientMockRecorder) Context() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Context", reflect.TypeOf((*MockScheduler_AnnouncePeersClient)(nil).Context))
+}
+
+// Header mocks base method.
+func (m *MockScheduler_AnnouncePeersClient) Header() (metadata.MD, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Header")
+	ret0, _ := ret[0].(metadata.MD)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Header indicates an expected call of Header.
+func (mr *MockScheduler_AnnouncePeersClientMockRecorder) Header() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Header", reflect.TypeOf((*MockScheduler_AnnouncePeersClient)(nil).Header))
+}
+
+// Recv mocks base method.
+func (m *MockScheduler_AnnouncePeersClient) Recv() (*emptypb.Empty, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Recv")
+	ret0, _ := ret[0].(*emptypb.Empty)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Recv indicates an expected call of Recv.
+func (mr *MockScheduler_AnnouncePeersClientMockRecorder) Recv() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Recv", reflect.TypeOf((*MockScheduler_AnnouncePeersClient)(nil).Recv))
+}
+
+// RecvMsg mocks base method.
+func (m_2 *MockScheduler_AnnouncePeersClient) RecvMsg(m any) error {
+	m_2.ctrl.T.Helper()
+	ret := m_2.ctrl.Call(m_2, "RecvMsg", m)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// RecvMsg indicates an expected call of RecvMsg.
+func (mr *MockScheduler_AnnouncePeersClientMockRecorder) RecvMsg(m any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecvMsg", reflect.TypeOf((*MockScheduler_AnnouncePeersClient)(nil).RecvMsg), m)
+}
+
+// Send mocks base method.
+func (m *MockScheduler_AnnouncePeersClient) Send(arg0 *scheduler.AnnouncePeersRequest) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Send", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Send indicates an expected call of Send.
+func (mr *MockScheduler_AnnouncePeersClientMockRecorder) Send(arg0 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Send", reflect.TypeOf((*MockScheduler_AnnouncePeersClient)(nil).Send), arg0)
+}
+
+// SendMsg mocks base method.
+func (m_2 *MockScheduler_AnnouncePeersClient) SendMsg(m any) error {
+	m_2.ctrl.T.Helper()
+	ret := m_2.ctrl.Call(m_2, "SendMsg", m)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SendMsg indicates an expected call of SendMsg.
+func (mr *MockScheduler_AnnouncePeersClientMockRecorder) SendMsg(m any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendMsg", reflect.TypeOf((*MockScheduler_AnnouncePeersClient)(nil).SendMsg), m)
+}
+
+// Trailer mocks base method.
+func (m *MockScheduler_AnnouncePeersClient) Trailer() metadata.MD {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Trailer")
+	ret0, _ := ret[0].(metadata.MD)
+	return ret0
+}
+
+// Trailer indicates an expected call of Trailer.
+func (mr *MockScheduler_AnnouncePeersClientMockRecorder) Trailer() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Trailer", reflect.TypeOf((*MockScheduler_AnnouncePeersClient)(nil).Trailer))
+}
+
 // MockScheduler_SyncProbesClient is a mock of Scheduler_SyncProbesClient interface.
 type MockScheduler_SyncProbesClient struct {
 	ctrl     *gomock.Controller
@@ -735,143 +932,6 @@ func (mr *MockScheduler_AnnounceCachePeerClientMockRecorder) Trailer() *gomock.C
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Trailer", reflect.TypeOf((*MockScheduler_AnnounceCachePeerClient)(nil).Trailer))
 }
 
-// MockScheduler_UploadCacheTaskClient is a mock of Scheduler_UploadCacheTaskClient interface.
-type MockScheduler_UploadCacheTaskClient struct {
-	ctrl     *gomock.Controller
-	recorder *MockScheduler_UploadCacheTaskClientMockRecorder
-}
-
-// MockScheduler_UploadCacheTaskClientMockRecorder is the mock recorder for MockScheduler_UploadCacheTaskClient.
-type MockScheduler_UploadCacheTaskClientMockRecorder struct {
-	mock *MockScheduler_UploadCacheTaskClient
-}
-
-// NewMockScheduler_UploadCacheTaskClient creates a new mock instance.
-func NewMockScheduler_UploadCacheTaskClient(ctrl *gomock.Controller) *MockScheduler_UploadCacheTaskClient {
-	mock := &MockScheduler_UploadCacheTaskClient{ctrl: ctrl}
-	mock.recorder = &MockScheduler_UploadCacheTaskClientMockRecorder{mock}
-	return mock
-}
-
-// EXPECT returns an object that allows the caller to indicate expected use.
-func (m *MockScheduler_UploadCacheTaskClient) EXPECT() *MockScheduler_UploadCacheTaskClientMockRecorder {
-	return m.recorder
-}
-
-// CloseAndRecv mocks base method.
-func (m *MockScheduler_UploadCacheTaskClient) CloseAndRecv() (*common.CacheTask, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "CloseAndRecv")
-	ret0, _ := ret[0].(*common.CacheTask)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// CloseAndRecv indicates an expected call of CloseAndRecv.
-func (mr *MockScheduler_UploadCacheTaskClientMockRecorder) CloseAndRecv() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CloseAndRecv", reflect.TypeOf((*MockScheduler_UploadCacheTaskClient)(nil).CloseAndRecv))
-}
-
-// CloseSend mocks base method.
-func (m *MockScheduler_UploadCacheTaskClient) CloseSend() error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "CloseSend")
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// CloseSend indicates an expected call of CloseSend.
-func (mr *MockScheduler_UploadCacheTaskClientMockRecorder) CloseSend() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CloseSend", reflect.TypeOf((*MockScheduler_UploadCacheTaskClient)(nil).CloseSend))
-}
-
-// Context mocks base method.
-func (m *MockScheduler_UploadCacheTaskClient) Context() context.Context {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Context")
-	ret0, _ := ret[0].(context.Context)
-	return ret0
-}
-
-// Context indicates an expected call of Context.
-func (mr *MockScheduler_UploadCacheTaskClientMockRecorder) Context() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Context", reflect.TypeOf((*MockScheduler_UploadCacheTaskClient)(nil).Context))
-}
-
-// Header mocks base method.
-func (m *MockScheduler_UploadCacheTaskClient) Header() (metadata.MD, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Header")
-	ret0, _ := ret[0].(metadata.MD)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// Header indicates an expected call of Header.
-func (mr *MockScheduler_UploadCacheTaskClientMockRecorder) Header() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Header", reflect.TypeOf((*MockScheduler_UploadCacheTaskClient)(nil).Header))
-}
-
-// RecvMsg mocks base method.
-func (m_2 *MockScheduler_UploadCacheTaskClient) RecvMsg(m any) error {
-	m_2.ctrl.T.Helper()
-	ret := m_2.ctrl.Call(m_2, "RecvMsg", m)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// RecvMsg indicates an expected call of RecvMsg.
-func (mr *MockScheduler_UploadCacheTaskClientMockRecorder) RecvMsg(m any) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecvMsg", reflect.TypeOf((*MockScheduler_UploadCacheTaskClient)(nil).RecvMsg), m)
-}
-
-// Send mocks base method.
-func (m *MockScheduler_UploadCacheTaskClient) Send(arg0 *scheduler.UploadCacheTaskRequest) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Send", arg0)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// Send indicates an expected call of Send.
-func (mr *MockScheduler_UploadCacheTaskClientMockRecorder) Send(arg0 any) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Send", reflect.TypeOf((*MockScheduler_UploadCacheTaskClient)(nil).Send), arg0)
-}
-
-// SendMsg mocks base method.
-func (m_2 *MockScheduler_UploadCacheTaskClient) SendMsg(m any) error {
-	m_2.ctrl.T.Helper()
-	ret := m_2.ctrl.Call(m_2, "SendMsg", m)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// SendMsg indicates an expected call of SendMsg.
-func (mr *MockScheduler_UploadCacheTaskClientMockRecorder) SendMsg(m any) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendMsg", reflect.TypeOf((*MockScheduler_UploadCacheTaskClient)(nil).SendMsg), m)
-}
-
-// Trailer mocks base method.
-func (m *MockScheduler_UploadCacheTaskClient) Trailer() metadata.MD {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Trailer")
-	ret0, _ := ret[0].(metadata.MD)
-	return ret0
-}
-
-// Trailer indicates an expected call of Trailer.
-func (mr *MockScheduler_UploadCacheTaskClientMockRecorder) Trailer() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Trailer", reflect.TypeOf((*MockScheduler_UploadCacheTaskClient)(nil).Trailer))
-}
-
 // MockSchedulerServer is a mock of SchedulerServer interface.
 type MockSchedulerServer struct {
 	ctrl     *gomock.Controller
@@ -938,6 +998,20 @@ func (mr *MockSchedulerServerMockRecorder) AnnouncePeer(arg0 any) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AnnouncePeer", reflect.TypeOf((*MockSchedulerServer)(nil).AnnouncePeer), arg0)
 }
 
+// AnnouncePeers mocks base method.
+func (m *MockSchedulerServer) AnnouncePeers(arg0 scheduler.Scheduler_AnnouncePeersServer) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "AnnouncePeers", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// AnnouncePeers indicates an expected call of AnnouncePeers.
+func (mr *MockSchedulerServerMockRecorder) AnnouncePeers(arg0 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AnnouncePeers", reflect.TypeOf((*MockSchedulerServer)(nil).AnnouncePeers), arg0)
+}
+
 // DeleteCachePeer mocks base method.
 func (m *MockSchedulerServer) DeleteCachePeer(arg0 context.Context, arg1 *scheduler.DeleteCachePeerRequest) (*emptypb.Empty, error) {
 	m.ctrl.T.Helper()
@@ -1087,18 +1161,49 @@ func (mr *MockSchedulerServerMockRecorder) SyncProbes(arg0 any) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SyncProbes", reflect.TypeOf((*MockSchedulerServer)(nil).SyncProbes), arg0)
 }
 
-// UploadCacheTask mocks base method.
+// UploadCacheTaskFailed mocks base method.
-func (m *MockSchedulerServer) UploadCacheTask(arg0 scheduler.Scheduler_UploadCacheTaskServer) error {
+func (m *MockSchedulerServer) UploadCacheTaskFailed(arg0 context.Context, arg1 *scheduler.UploadCacheTaskFailedRequest) (*emptypb.Empty, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "UploadCacheTask", arg0)
+	ret := m.ctrl.Call(m, "UploadCacheTaskFailed", arg0, arg1)
-	ret0, _ := ret[0].(error)
+	ret0, _ := ret[0].(*emptypb.Empty)
-	return ret0
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
 }
 
-// UploadCacheTask indicates an expected call of UploadCacheTask.
+// UploadCacheTaskFailed indicates an expected call of UploadCacheTaskFailed.
-func (mr *MockSchedulerServerMockRecorder) UploadCacheTask(arg0 any) *gomock.Call {
+func (mr *MockSchedulerServerMockRecorder) UploadCacheTaskFailed(arg0, arg1 any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UploadCacheTask", reflect.TypeOf((*MockSchedulerServer)(nil).UploadCacheTask), arg0)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UploadCacheTaskFailed", reflect.TypeOf((*MockSchedulerServer)(nil).UploadCacheTaskFailed), arg0, arg1)
+}
+
+// UploadCacheTaskFinished mocks base method.
+func (m *MockSchedulerServer) UploadCacheTaskFinished(arg0 context.Context, arg1 *scheduler.UploadCacheTaskFinishedRequest) (*common.CacheTask, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "UploadCacheTaskFinished", arg0, arg1)
+	ret0, _ := ret[0].(*common.CacheTask)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// UploadCacheTaskFinished indicates an expected call of UploadCacheTaskFinished.
+func (mr *MockSchedulerServerMockRecorder) UploadCacheTaskFinished(arg0, arg1 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UploadCacheTaskFinished", reflect.TypeOf((*MockSchedulerServer)(nil).UploadCacheTaskFinished), arg0, arg1)
+}
+
+// UploadCacheTaskStarted mocks base method.
+func (m *MockSchedulerServer) UploadCacheTaskStarted(arg0 context.Context, arg1 *scheduler.UploadCacheTaskStartedRequest) (*emptypb.Empty, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "UploadCacheTaskStarted", arg0, arg1)
+	ret0, _ := ret[0].(*emptypb.Empty)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// UploadCacheTaskStarted indicates an expected call of UploadCacheTaskStarted.
+func (mr *MockSchedulerServerMockRecorder) UploadCacheTaskStarted(arg0, arg1 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UploadCacheTaskStarted", reflect.TypeOf((*MockSchedulerServer)(nil).UploadCacheTaskStarted), arg0, arg1)
 }
 
 // MockUnsafeSchedulerServer is a mock of UnsafeSchedulerServer interface.
@@ -1270,6 +1375,140 @@ func (mr *MockScheduler_AnnouncePeerServerMockRecorder) SetTrailer(arg0 any) *go
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetTrailer", reflect.TypeOf((*MockScheduler_AnnouncePeerServer)(nil).SetTrailer), arg0)
 }
 
+// MockScheduler_AnnouncePeersServer is a mock of Scheduler_AnnouncePeersServer interface.
+type MockScheduler_AnnouncePeersServer struct {
+	ctrl     *gomock.Controller
+	recorder *MockScheduler_AnnouncePeersServerMockRecorder
+}
+
+// MockScheduler_AnnouncePeersServerMockRecorder is the mock recorder for MockScheduler_AnnouncePeersServer.
+type MockScheduler_AnnouncePeersServerMockRecorder struct {
+	mock *MockScheduler_AnnouncePeersServer
+}
+
+// NewMockScheduler_AnnouncePeersServer creates a new mock instance.
+func NewMockScheduler_AnnouncePeersServer(ctrl *gomock.Controller) *MockScheduler_AnnouncePeersServer {
+	mock := &MockScheduler_AnnouncePeersServer{ctrl: ctrl}
+	mock.recorder = &MockScheduler_AnnouncePeersServerMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockScheduler_AnnouncePeersServer) EXPECT() *MockScheduler_AnnouncePeersServerMockRecorder {
+	return m.recorder
+}
+
+// Context mocks base method.
+func (m *MockScheduler_AnnouncePeersServer) Context() context.Context {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Context")
+	ret0, _ := ret[0].(context.Context)
+	return ret0
+}
+
+// Context indicates an expected call of Context.
+func (mr *MockScheduler_AnnouncePeersServerMockRecorder) Context() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Context", reflect.TypeOf((*MockScheduler_AnnouncePeersServer)(nil).Context))
+}
+
+// Recv mocks base method.
+func (m *MockScheduler_AnnouncePeersServer) Recv() (*scheduler.AnnouncePeersRequest, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Recv")
+	ret0, _ := ret[0].(*scheduler.AnnouncePeersRequest)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// Recv indicates an expected call of Recv.
+func (mr *MockScheduler_AnnouncePeersServerMockRecorder) Recv() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Recv", reflect.TypeOf((*MockScheduler_AnnouncePeersServer)(nil).Recv))
+}
+
+// RecvMsg mocks base method.
+func (m_2 *MockScheduler_AnnouncePeersServer) RecvMsg(m any) error {
+	m_2.ctrl.T.Helper()
+	ret := m_2.ctrl.Call(m_2, "RecvMsg", m)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// RecvMsg indicates an expected call of RecvMsg.
+func (mr *MockScheduler_AnnouncePeersServerMockRecorder) RecvMsg(m any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecvMsg", reflect.TypeOf((*MockScheduler_AnnouncePeersServer)(nil).RecvMsg), m)
+}
+
+// Send mocks base method.
+func (m *MockScheduler_AnnouncePeersServer) Send(arg0 *emptypb.Empty) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Send", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Send indicates an expected call of Send.
+func (mr *MockScheduler_AnnouncePeersServerMockRecorder) Send(arg0 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Send", reflect.TypeOf((*MockScheduler_AnnouncePeersServer)(nil).Send), arg0)
+}
+
+// SendHeader mocks base method.
+func (m *MockScheduler_AnnouncePeersServer) SendHeader(arg0 metadata.MD) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SendHeader", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SendHeader indicates an expected call of SendHeader.
+func (mr *MockScheduler_AnnouncePeersServerMockRecorder) SendHeader(arg0 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendHeader", reflect.TypeOf((*MockScheduler_AnnouncePeersServer)(nil).SendHeader), arg0)
+}
+
+// SendMsg mocks base method.
+func (m_2 *MockScheduler_AnnouncePeersServer) SendMsg(m any) error {
+	m_2.ctrl.T.Helper()
+	ret := m_2.ctrl.Call(m_2, "SendMsg", m)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SendMsg indicates an expected call of SendMsg.
+func (mr *MockScheduler_AnnouncePeersServerMockRecorder) SendMsg(m any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendMsg", reflect.TypeOf((*MockScheduler_AnnouncePeersServer)(nil).SendMsg), m)
+}
+
+// SetHeader mocks base method.
+func (m *MockScheduler_AnnouncePeersServer) SetHeader(arg0 metadata.MD) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SetHeader", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SetHeader indicates an expected call of SetHeader.
+func (mr *MockScheduler_AnnouncePeersServerMockRecorder) SetHeader(arg0 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetHeader", reflect.TypeOf((*MockScheduler_AnnouncePeersServer)(nil).SetHeader), arg0)
+}
+
+// SetTrailer mocks base method.
+func (m *MockScheduler_AnnouncePeersServer) SetTrailer(arg0 metadata.MD) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "SetTrailer", arg0)
+}
+
+// SetTrailer indicates an expected call of SetTrailer.
+func (mr *MockScheduler_AnnouncePeersServerMockRecorder) SetTrailer(arg0 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetTrailer", reflect.TypeOf((*MockScheduler_AnnouncePeersServer)(nil).SetTrailer), arg0)
+}
+
 // MockScheduler_SyncProbesServer is a mock of Scheduler_SyncProbesServer interface.
 type MockScheduler_SyncProbesServer struct {
 	ctrl     *gomock.Controller
@@ -1537,137 +1776,3 @@ func (mr *MockScheduler_AnnounceCachePeerServerMockRecorder) SetTrailer(arg0 any
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetTrailer", reflect.TypeOf((*MockScheduler_AnnounceCachePeerServer)(nil).SetTrailer), arg0)
 }
-
-// MockScheduler_UploadCacheTaskServer is a mock of Scheduler_UploadCacheTaskServer interface.
-type MockScheduler_UploadCacheTaskServer struct {
-	ctrl     *gomock.Controller
-	recorder *MockScheduler_UploadCacheTaskServerMockRecorder
-}
-
-// MockScheduler_UploadCacheTaskServerMockRecorder is the mock recorder for MockScheduler_UploadCacheTaskServer.
-type MockScheduler_UploadCacheTaskServerMockRecorder struct {
-	mock *MockScheduler_UploadCacheTaskServer
-}
-
-// NewMockScheduler_UploadCacheTaskServer creates a new mock instance.
-func NewMockScheduler_UploadCacheTaskServer(ctrl *gomock.Controller) *MockScheduler_UploadCacheTaskServer {
-	mock := &MockScheduler_UploadCacheTaskServer{ctrl: ctrl}
-	mock.recorder = &MockScheduler_UploadCacheTaskServerMockRecorder{mock}
-	return mock
-}
-
-// EXPECT returns an object that allows the caller to indicate expected use.
-func (m *MockScheduler_UploadCacheTaskServer) EXPECT() *MockScheduler_UploadCacheTaskServerMockRecorder {
-	return m.recorder
-}
-
-// Context mocks base method.
-func (m *MockScheduler_UploadCacheTaskServer) Context() context.Context {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Context")
-	ret0, _ := ret[0].(context.Context)
-	return ret0
-}
-
-// Context indicates an expected call of Context.
-func (mr *MockScheduler_UploadCacheTaskServerMockRecorder) Context() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Context", reflect.TypeOf((*MockScheduler_UploadCacheTaskServer)(nil).Context))
-}
-
-// Recv mocks base method.
-func (m *MockScheduler_UploadCacheTaskServer) Recv() (*scheduler.UploadCacheTaskRequest, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Recv")
-	ret0, _ := ret[0].(*scheduler.UploadCacheTaskRequest)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// Recv indicates an expected call of Recv.
-func (mr *MockScheduler_UploadCacheTaskServerMockRecorder) Recv() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Recv", reflect.TypeOf((*MockScheduler_UploadCacheTaskServer)(nil).Recv))
-}
-
-// RecvMsg mocks base method.
-func (m_2 *MockScheduler_UploadCacheTaskServer) RecvMsg(m any) error {
-	m_2.ctrl.T.Helper()
-	ret := m_2.ctrl.Call(m_2, "RecvMsg", m)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// RecvMsg indicates an expected call of RecvMsg.
-func (mr *MockScheduler_UploadCacheTaskServerMockRecorder) RecvMsg(m any) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecvMsg", reflect.TypeOf((*MockScheduler_UploadCacheTaskServer)(nil).RecvMsg), m)
-}
-
-// SendAndClose mocks base method.
-func (m *MockScheduler_UploadCacheTaskServer) SendAndClose(arg0 *common.CacheTask) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "SendAndClose", arg0)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// SendAndClose indicates an expected call of SendAndClose.
-func (mr *MockScheduler_UploadCacheTaskServerMockRecorder) SendAndClose(arg0 any) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendAndClose", reflect.TypeOf((*MockScheduler_UploadCacheTaskServer)(nil).SendAndClose), arg0)
-}
-
-// SendHeader mocks base method.
-func (m *MockScheduler_UploadCacheTaskServer) SendHeader(arg0 metadata.MD) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "SendHeader", arg0)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// SendHeader indicates an expected call of SendHeader.
-func (mr *MockScheduler_UploadCacheTaskServerMockRecorder) SendHeader(arg0 any) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendHeader", reflect.TypeOf((*MockScheduler_UploadCacheTaskServer)(nil).SendHeader), arg0)
-}
-
-// SendMsg mocks base method.
-func (m_2 *MockScheduler_UploadCacheTaskServer) SendMsg(m any) error {
-	m_2.ctrl.T.Helper()
-	ret := m_2.ctrl.Call(m_2, "SendMsg", m)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// SendMsg indicates an expected call of SendMsg.
-func (mr *MockScheduler_UploadCacheTaskServerMockRecorder) SendMsg(m any) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendMsg", reflect.TypeOf((*MockScheduler_UploadCacheTaskServer)(nil).SendMsg), m)
-}
-
-// SetHeader mocks base method.
-func (m *MockScheduler_UploadCacheTaskServer) SetHeader(arg0 metadata.MD) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "SetHeader", arg0)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// SetHeader indicates an expected call of SetHeader.
-func (mr *MockScheduler_UploadCacheTaskServerMockRecorder) SetHeader(arg0 any) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetHeader", reflect.TypeOf((*MockScheduler_UploadCacheTaskServer)(nil).SetHeader), arg0)
-}
-
-// SetTrailer mocks base method.
-func (m *MockScheduler_UploadCacheTaskServer) SetTrailer(arg0 metadata.MD) {
-	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "SetTrailer", arg0)
-}
-
-// SetTrailer indicates an expected call of SetTrailer.
-func (mr *MockScheduler_UploadCacheTaskServerMockRecorder) SetTrailer(arg0 any) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetTrailer", reflect.TypeOf((*MockScheduler_UploadCacheTaskServer)(nil).SetTrailer), arg0)
-}

pkg/apis/scheduler/v2/scheduler.pb.validate.go

@@ -3134,6 +3134,153 @@ var _ interface {
 	ErrorName() string
 } = DeletePeerRequestValidationError{}
 
+// Validate checks the field values on AnnouncePeersRequest with the rules
+// defined in the proto definition for this message. If any rules are
+// violated, the first error encountered is returned, or nil if there are no violations.
+func (m *AnnouncePeersRequest) Validate() error {
+	return m.validate(false)
+}
+
+// ValidateAll checks the field values on AnnouncePeersRequest with the rules
+// defined in the proto definition for this message. If any rules are
+// violated, the result is a list of violation errors wrapped in
+// AnnouncePeersRequestMultiError, or nil if none found.
+func (m *AnnouncePeersRequest) ValidateAll() error {
+	return m.validate(true)
+}
+
+func (m *AnnouncePeersRequest) validate(all bool) error {
+	if m == nil {
+		return nil
+	}
+
+	var errors []error
+
+	if len(m.GetPeers()) < 1 {
+		err := AnnouncePeersRequestValidationError{
+			field:  "Peers",
+			reason: "value must contain at least 1 item(s)",
+		}
+		if !all {
+			return err
+		}
+		errors = append(errors, err)
+	}
+
+	for idx, item := range m.GetPeers() {
+		_, _ = idx, item
+
+		if all {
+			switch v := interface{}(item).(type) {
+			case interface{ ValidateAll() error }:
+				if err := v.ValidateAll(); err != nil {
+					errors = append(errors, AnnouncePeersRequestValidationError{
+						field:  fmt.Sprintf("Peers[%v]", idx),
+						reason: "embedded message failed validation",
+						cause:  err,
+					})
+				}
+			case interface{ Validate() error }:
+				if err := v.Validate(); err != nil {
+					errors = append(errors, AnnouncePeersRequestValidationError{
+						field:  fmt.Sprintf("Peers[%v]", idx),
+						reason: "embedded message failed validation",
+						cause:  err,
+					})
+				}
+			}
+		} else if v, ok := interface{}(item).(interface{ Validate() error }); ok {
+			if err := v.Validate(); err != nil {
+				return AnnouncePeersRequestValidationError{
+					field:  fmt.Sprintf("Peers[%v]", idx),
+					reason: "embedded message failed validation",
+					cause:  err,
+				}
+			}
+		}
+
+	}
+
+	if len(errors) > 0 {
+		return AnnouncePeersRequestMultiError(errors)
+	}
+
+	return nil
+}
+
+// AnnouncePeersRequestMultiError is an error wrapping multiple validation
+// errors returned by AnnouncePeersRequest.ValidateAll() if the designated
+// constraints aren't met.
+type AnnouncePeersRequestMultiError []error
+
+// Error returns a concatenation of all the error messages it wraps.
+func (m AnnouncePeersRequestMultiError) Error() string {
+	var msgs []string
+	for _, err := range m {
+		msgs = append(msgs, err.Error())
+	}
+	return strings.Join(msgs, "; ")
+}
+
+// AllErrors returns a list of validation violation errors.
+func (m AnnouncePeersRequestMultiError) AllErrors() []error { return m }
+
+// AnnouncePeersRequestValidationError is the validation error returned by
+// AnnouncePeersRequest.Validate if the designated constraints aren't met.
+type AnnouncePeersRequestValidationError struct {
+	field  string
+	reason string
+	cause  error
+	key    bool
+}
+
+// Field function returns field value.
+func (e AnnouncePeersRequestValidationError) Field() string { return e.field }
+
+// Reason function returns reason value.
+func (e AnnouncePeersRequestValidationError) Reason() string { return e.reason }
+
+// Cause function returns cause value.
+func (e AnnouncePeersRequestValidationError) Cause() error { return e.cause }
+
+// Key function returns key value.
+func (e AnnouncePeersRequestValidationError) Key() bool { return e.key }
+
+// ErrorName returns error name.
+func (e AnnouncePeersRequestValidationError) ErrorName() string {
+	return "AnnouncePeersRequestValidationError"
+}
+
+// Error satisfies the builtin error interface
+func (e AnnouncePeersRequestValidationError) Error() string {
+	cause := ""
+	if e.cause != nil {
+		cause = fmt.Sprintf(" | caused by: %v", e.cause)
+	}
+
+	key := ""
+	if e.key {
+		key = "key for "
+	}
+
+	return fmt.Sprintf(
+		"invalid %sAnnouncePeersRequest.%s: %s%s",
+		key,
+		e.field,
+		e.reason,
+		cause)
+}
+
+var _ error = AnnouncePeersRequestValidationError{}
+
+var _ interface {
+	Field() string
+	Reason() string
+	Key() bool
+	Cause() error
+	ErrorName() string
+} = AnnouncePeersRequestValidationError{}
+
 // Validate checks the field values on StatTaskRequest with the rules defined
 // in the proto definition for this message. If any rules are violated, the
 // first error encountered is returned, or nil if there are no violations.

pkg/apis/scheduler/v2/scheduler.proto

@@ -188,6 +188,12 @@ message DeletePeerRequest {
   string peer_id = 3 [(validate.rules).string.min_len = 1];
 }
 
+// AnnouncePeersRequest represents request of AnnouncePeers.
+message AnnouncePeersRequest{
+  // Peers information.
+  repeated common.v2.Peer peers = 1 [(validate.rules).repeated.min_items = 1];
+}
+
 // StatTaskRequest represents request of StatTask.
 message StatTaskRequest {
   // Host id.
@@ -448,6 +454,9 @@ service Scheduler {
   // DeletePeer releases peer in scheduler.
   rpc DeletePeer(DeletePeerRequest)returns(google.protobuf.Empty);
 
+  // A host announces that it has the announced peers to scheduler at startup.
+  rpc AnnouncePeers(stream AnnouncePeersRequest) returns(stream google.protobuf.Empty);
+
   // Checks information of task.
   rpc StatTask(StatTaskRequest)returns(common.v2.Task);
 

pkg/apis/scheduler/v2/scheduler_grpc.pb.go

@@ -30,6 +30,8 @@ type SchedulerClient interface {
 	StatPeer(ctx context.Context, in *StatPeerRequest, opts ...grpc.CallOption) (*v2.Peer, error)
 	// DeletePeer releases peer in scheduler.
 	DeletePeer(ctx context.Context, in *DeletePeerRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
+	// A host announces that it has the announced peers to scheduler at startup.
+	AnnouncePeers(ctx context.Context, opts ...grpc.CallOption) (Scheduler_AnnouncePeersClient, error)
 	// Checks information of task.
 	StatTask(ctx context.Context, in *StatTaskRequest, opts ...grpc.CallOption) (*v2.Task, error)
 	// DeleteTask releases task in scheduler.
@@ -115,6 +117,37 @@ func (c *schedulerClient) DeletePeer(ctx context.Context, in *DeletePeerRequest,
 	return out, nil
 }
 
+func (c *schedulerClient) AnnouncePeers(ctx context.Context, opts ...grpc.CallOption) (Scheduler_AnnouncePeersClient, error) {
+	stream, err := c.cc.NewStream(ctx, &Scheduler_ServiceDesc.Streams[1], "/scheduler.v2.Scheduler/AnnouncePeers", opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &schedulerAnnouncePeersClient{stream}
+	return x, nil
+}
+
+type Scheduler_AnnouncePeersClient interface {
+	Send(*AnnouncePeersRequest) error
+	Recv() (*emptypb.Empty, error)
+	grpc.ClientStream
+}
+
+type schedulerAnnouncePeersClient struct {
+	grpc.ClientStream
+}
+
+func (x *schedulerAnnouncePeersClient) Send(m *AnnouncePeersRequest) error {
+	return x.ClientStream.SendMsg(m)
+}
+
+func (x *schedulerAnnouncePeersClient) Recv() (*emptypb.Empty, error) {
+	m := new(emptypb.Empty)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 func (c *schedulerClient) StatTask(ctx context.Context, in *StatTaskRequest, opts ...grpc.CallOption) (*v2.Task, error) {
 	out := new(v2.Task)
 	err := c.cc.Invoke(ctx, "/scheduler.v2.Scheduler/StatTask", in, out, opts...)
@@ -152,7 +185,7 @@ func (c *schedulerClient) DeleteHost(ctx context.Context, in *DeleteHostRequest,
 }
 
 func (c *schedulerClient) SyncProbes(ctx context.Context, opts ...grpc.CallOption) (Scheduler_SyncProbesClient, error) {
-	stream, err := c.cc.NewStream(ctx, &Scheduler_ServiceDesc.Streams[1], "/scheduler.v2.Scheduler/SyncProbes", opts...)
+	stream, err := c.cc.NewStream(ctx, &Scheduler_ServiceDesc.Streams[2], "/scheduler.v2.Scheduler/SyncProbes", opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -183,7 +216,7 @@ func (x *schedulerSyncProbesClient) Recv() (*SyncProbesResponse, error) {
 }
 
 func (c *schedulerClient) AnnounceCachePeer(ctx context.Context, opts ...grpc.CallOption) (Scheduler_AnnounceCachePeerClient, error) {
-	stream, err := c.cc.NewStream(ctx, &Scheduler_ServiceDesc.Streams[2], "/scheduler.v2.Scheduler/AnnounceCachePeer", opts...)
+	stream, err := c.cc.NewStream(ctx, &Scheduler_ServiceDesc.Streams[3], "/scheduler.v2.Scheduler/AnnounceCachePeer", opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -286,6 +319,8 @@ type SchedulerServer interface {
 	StatPeer(context.Context, *StatPeerRequest) (*v2.Peer, error)
 	// DeletePeer releases peer in scheduler.
 	DeletePeer(context.Context, *DeletePeerRequest) (*emptypb.Empty, error)
+	// A host announces that it has the announced peers to scheduler at startup.
+	AnnouncePeers(Scheduler_AnnouncePeersServer) error
 	// Checks information of task.
 	StatTask(context.Context, *StatTaskRequest) (*v2.Task, error)
 	// DeleteTask releases task in scheduler.
@@ -327,6 +362,9 @@ func (UnimplementedSchedulerServer) StatPeer(context.Context, *StatPeerRequest)
 func (UnimplementedSchedulerServer) DeletePeer(context.Context, *DeletePeerRequest) (*emptypb.Empty, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method DeletePeer not implemented")
 }
+func (UnimplementedSchedulerServer) AnnouncePeers(Scheduler_AnnouncePeersServer) error {
+	return status.Errorf(codes.Unimplemented, "method AnnouncePeers not implemented")
+}
 func (UnimplementedSchedulerServer) StatTask(context.Context, *StatTaskRequest) (*v2.Task, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method StatTask not implemented")
 }
@@ -440,6 +478,32 @@ func _Scheduler_DeletePeer_Handler(srv interface{}, ctx context.Context, dec fun
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Scheduler_AnnouncePeers_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(SchedulerServer).AnnouncePeers(&schedulerAnnouncePeersServer{stream})
+}
+
+type Scheduler_AnnouncePeersServer interface {
+	Send(*emptypb.Empty) error
+	Recv() (*AnnouncePeersRequest, error)
+	grpc.ServerStream
+}
+
+type schedulerAnnouncePeersServer struct {
+	grpc.ServerStream
+}
+
+func (x *schedulerAnnouncePeersServer) Send(m *emptypb.Empty) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func (x *schedulerAnnouncePeersServer) Recv() (*AnnouncePeersRequest, error) {
+	m := new(AnnouncePeersRequest)
+	if err := x.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 func _Scheduler_StatTask_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(StatTaskRequest)
 	if err := dec(in); err != nil {
@@ -757,6 +821,12 @@ var Scheduler_ServiceDesc = grpc.ServiceDesc{
 			ServerStreams: true,
 			ClientStreams: true,
 		},
+		{
+			StreamName:    "AnnouncePeers",
+			Handler:       _Scheduler_AnnouncePeers_Handler,
+			ServerStreams: true,
+			ClientStreams: true,
+		},
 		{
 			StreamName:    "SyncProbes",
 			Handler:       _Scheduler_SyncProbes_Handler,

proto/common.proto

@@ -38,8 +38,8 @@ enum SizeScope {
 
 // TaskType represents type of task.
 enum TaskType {
-  // DFDAEMON is dfdeamon type of task,
+  // DFDAEMON is dfdaemon type of task,
-  // dfdeamon task is a normal p2p task.
+  // dfdaemon task is a normal p2p task.
   DFDAEMON = 0;
 
   // DFCACHE is dfcache type of task,
@@ -132,7 +132,7 @@ message CachePeer {
   string id = 1;
   // Persistent represents whether the cache peer is persistent.
   // If the cache peer is persistent, the cache peer will
-  // not be deleted when dfdamon runs garbage collection.
+  // not be deleted when dfdaemon runs garbage collection.
   bool persistent = 2;
   // Peer downloads costs time.
   google.protobuf.Duration cost = 3;

proto/dfdaemon.proto

@@ -122,7 +122,7 @@ message DownloadCacheTaskRequest {
   string task_id = 2;
   // Persistent represents whether the cache task is persistent.
   // If the cache task is persistent, the cache peer will
-  // not be deleted when dfdamon runs garbage collection.
+  // not be deleted when dfdaemon runs garbage collection.
   bool persistent = 3;
   // Tag is used to distinguish different cache tasks.
   optional string tag = 4;

proto/scheduler.proto

@@ -178,6 +178,12 @@ message DeletePeerRequest {
   string peer_id = 3;
 }
 
+// AnnouncePeersRequest represents request of AnnouncePeers.
+message AnnouncePeersRequest{
+  // Peers information.
+  repeated common.v2.Peer peers = 1;
+}
+
 // StatTaskRequest represents request of StatTask.
 message StatTaskRequest {
   // Host id.
@@ -432,6 +438,9 @@ service Scheduler{
   // DeletePeer releases peer in scheduler.
   rpc DeletePeer(DeletePeerRequest)returns(google.protobuf.Empty);
 
+  // A host announces that it has the announced peers to scheduler at startup.
+  rpc AnnouncePeers(stream AnnouncePeersRequest) returns(stream google.protobuf.Empty);
+
   // Checks information of task.
   rpc StatTask(StatTaskRequest)returns(common.v2.Task);
 

src/common.v2.rs

@@ -47,7 +47,7 @@ pub struct CachePeer {
     pub id: ::prost::alloc::string::String,
     /// Persistent represents whether the cache peer is persistent.
     /// If the cache peer is persistent, the cache peer will
-    /// not be deleted when dfdamon runs garbage collection.
+    /// not be deleted when dfdaemon runs garbage collection.
     #[prost(bool, tag = "2")]
     pub persistent: bool,
     /// Peer downloads costs time.
@@ -558,8 +558,8 @@ impl SizeScope {
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
 #[repr(i32)]
 pub enum TaskType {
-    /// DFDAEMON is dfdeamon type of task,
+    /// DFDAEMON is dfdaemon type of task,
-    /// dfdeamon task is a normal p2p task.
+    /// dfdaemon task is a normal p2p task.
     Dfdaemon = 0,
     /// DFCACHE is dfcache type of task,
     /// dfcache task is a cache task, and the task url is fake url.

src/dfdaemon.v2.rs

@@ -153,7 +153,7 @@ pub struct DownloadCacheTaskRequest {
     pub task_id: ::prost::alloc::string::String,
     /// Persistent represents whether the cache task is persistent.
     /// If the cache task is persistent, the cache peer will
-    /// not be deleted when dfdamon runs garbage collection.
+    /// not be deleted when dfdaemon runs garbage collection.
     #[prost(bool, tag = "3")]
     pub persistent: bool,
     /// Tag is used to distinguish different cache tasks.

src/scheduler.v2.rs

@@ -272,6 +272,15 @@ pub struct DeletePeerRequest {
     #[prost(string, tag = "3")]
     pub peer_id: ::prost::alloc::string::String,
 }
+/// AnnouncePeersRequest represents request of AnnouncePeers.
+#[derive(serde::Serialize, serde::Deserialize)]
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct AnnouncePeersRequest {
+    /// Peers information.
+    #[prost(message, repeated, tag = "1")]
+    pub peers: ::prost::alloc::vec::Vec<super::super::common::v2::Peer>,
+}
 /// StatTaskRequest represents request of StatTask.
 #[derive(serde::Serialize, serde::Deserialize)]
 #[allow(clippy::derive_partial_eq_without_eq)]
@@ -824,6 +833,34 @@ pub mod scheduler_client {
                 .insert(GrpcMethod::new("scheduler.v2.Scheduler", "DeletePeer"));
             self.inner.unary(req, path, codec).await
         }
+        /// A host announces that it has the announced peers to scheduler at startup.
+        pub async fn announce_peers(
+            &mut self,
+            request: impl tonic::IntoStreamingRequest<
+                Message = super::AnnouncePeersRequest,
+            >,
+        ) -> std::result::Result<
+            tonic::Response<tonic::codec::Streaming<()>>,
+            tonic::Status,
+        > {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| {
+                    tonic::Status::new(
+                        tonic::Code::Unknown,
+                        format!("Service was not ready: {}", e.into()),
+                    )
+                })?;
+            let codec = tonic::codec::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static(
+                "/scheduler.v2.Scheduler/AnnouncePeers",
+            );
+            let mut req = request.into_streaming_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("scheduler.v2.Scheduler", "AnnouncePeers"));
+            self.inner.streaming(req, path, codec).await
+        }
         /// Checks information of task.
         pub async fn stat_task(
             &mut self,
@@ -1185,6 +1222,20 @@ pub mod scheduler_server {
             &self,
             request: tonic::Request<super::DeletePeerRequest>,
         ) -> std::result::Result<tonic::Response<()>, tonic::Status>;
+        /// Server streaming response type for the AnnouncePeers method.
+        type AnnouncePeersStream: futures_core::Stream<
+                Item = std::result::Result<(), tonic::Status>,
+            >
+            + Send
+            + 'static;
+        /// A host announces that it has the announced peers to scheduler at startup.
+        async fn announce_peers(
+            &self,
+            request: tonic::Request<tonic::Streaming<super::AnnouncePeersRequest>>,
+        ) -> std::result::Result<
+            tonic::Response<Self::AnnouncePeersStream>,
+            tonic::Status,
+        >;
         /// Checks information of task.
         async fn stat_task(
             &self,
@@ -1498,6 +1549,55 @@ pub mod scheduler_server {
                     };
                     Box::pin(fut)
                 }
+                "/scheduler.v2.Scheduler/AnnouncePeers" => {
+                    #[allow(non_camel_case_types)]
+                    struct AnnouncePeersSvc<T: Scheduler>(pub Arc<T>);
+                    impl<
+                        T: Scheduler,
+                    > tonic::server::StreamingService<super::AnnouncePeersRequest>
+                    for AnnouncePeersSvc<T> {
+                        type Response = ();
+                        type ResponseStream = T::AnnouncePeersStream;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::ResponseStream>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<
+                                tonic::Streaming<super::AnnouncePeersRequest>,
+                            >,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                (*inner).announce_peers(request).await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let inner = inner.0;
+                        let method = AnnouncePeersSvc(inner);
+                        let codec = tonic::codec::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.streaming(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
                 "/scheduler.v2.Scheduler/StatTask" => {
                     #[allow(non_camel_case_types)]
                     struct StatTaskSvc<T: Scheduler>(pub Arc<T>);