Repository: dragonflyoss/dragonfly

Issue:
Support preheat for normal files instead of image
### Feature request:
For files to be downloaded from storage like oss, hdfs .et, also need preheat to accelerate download speed.


Discussion:

Solution patch:
diff --git a/client-rs b/client-rs
index 41cd3335..a579e016 160000
--- a/client-rs
+++ b/client-rs
@@ -1 +1 @@
-Subproject commit 41cd33354bf081da8e1d7156141a95debff73a7f
+Subproject commit a579e016c8ed345f6f4e2d435fc65bfa705e328c
diff --git a/go.mod b/go.mod
index d26c70bf..35e1b050 100644
--- a/go.mod
+++ b/go.mod
@@ -4,7 +4,7 @@ go 1.23.8
 
 require (
 	cloud.google.com/go/storage v1.50.0
-	d7y.io/api/v2 v2.1.79
+	d7y.io/api/v2 v2.1.80
 	github.com/MysteriousPotato/go-lockable v1.0.0
 	github.com/Showmax/go-fqdn v1.0.0
 	github.com/VividCortex/mysqlerr v1.0.0
diff --git a/go.sum b/go.sum
index b6b8c9f9..7a6758c2 100644
--- a/go.sum
+++ b/go.sum
@@ -63,8 +63,8 @@ cloud.google.com/go/storage v1.50.0 h1:3TbVkzTooBvnZsk7WaAQfOsNrdoM8QHusXA1cpk6Q
 cloud.google.com/go/storage v1.50.0/go.mod h1:l7XeiD//vx5lfqE3RavfmU9yvk5Pp0Zhcv482poyafY=
 cloud.google.com/go/trace v1.11.6 h1:2O2zjPzqPYAHrn3OKl029qlqG6W8ZdYaOWRyr8NgMT4=
 cloud.google.com/go/trace v1.11.6/go.mod h1:GA855OeDEBiBMzcckLPE2kDunIpC72N+Pq8WFieFjnI=
-d7y.io/api/v2 v2.1.79 h1:LUPpjiuhHZgyezHBH7lG42pkqjcdbiTAHtEJ2Z7tUJg=
-d7y.io/api/v2 v2.1.79/go.mod h1:t6k27g8dFyH6sp3y2J1qHyk2YmoySd88qSbsEaqJ2Sw=
+d7y.io/api/v2 v2.1.80 h1:dH4w0dUXLpZ1sdp6QuWUVXkjQaUSX/pAOFFunkrcsUs=
+d7y.io/api/v2 v2.1.80/go.mod h1:t6k27g8dFyH6sp3y2J1qHyk2YmoySd88qSbsEaqJ2Sw=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
 filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
diff --git a/internal/dflog/logger.go b/internal/dflog/logger.go
index 17b11faa..68b9f9e4 100644
--- a/internal/dflog/logger.go
+++ b/internal/dflog/logger.go
@@ -256,6 +256,24 @@ func WithStatImageAndTaskID(url, taskID string) *SugaredLoggerOnWith {
 	}
 }
 
+func WithPreheatFile(url string) *SugaredLoggerOnWith {
+	return &SugaredLoggerOnWith{
+		withArgs: []any{"url", url},
+	}
+}
+
+func WithStatFile(url string) *SugaredLoggerOnWith {
+	return &SugaredLoggerOnWith{
+		withArgs: []any{"url", url},
+	}
+}
+
+func WithStatFileAndTaskID(url, taskID string) *SugaredLoggerOnWith {
+	return &SugaredLoggerOnWith{
+		withArgs: []any{"url", url, "taskID", taskID},
+	}
+}
+
 func (log *SugaredLoggerOnWith) With(args ...any) *SugaredLoggerOnWith {
 	args = append(args, log.withArgs...)
 	return &SugaredLoggerOnWith{
diff --git a/internal/job/types.go b/internal/job/types.go
index 0c91c0ef..ac246977 100644
--- a/internal/job/types.go
+++ b/internal/job/types.go
@@ -18,6 +18,11 @@ package job
 
 import (
 	"time"
+
+	"google.golang.org/protobuf/types/known/durationpb"
+
+	v2 "d7y.io/api/v2/pkg/apis/common/v2"
+	dfdaemonv2 "d7y.io/api/v2/pkg/apis/dfdaemon/v2"
 )
 
 // PreheatRequest defines the request parameters for preheating.
@@ -83,6 +88,24 @@ type GetTaskResponse struct {
 	SchedulerClusterID uint    `json:"scheduler_cluster_id"`
 }
 
+// ListTaskEntriesRequest defines the request parameters for listing task entries.
+type ListTaskEntriesRequest struct {
+	TaskID           string               `json:"task_id" validate:"required"`
+	Url              string               `json:"url" validate:"omitempty"`
+	Timeout          *durationpb.Duration `json:"timeout" validate:"omitempty"`
+	Header           map[string]string    `json:"header" validate:"omitempty"`
+	CertificateChain [][]byte             `json:"certificate_chain" validate:"omitempty"`
+	ObjectStorage    *v2.ObjectStorage    `json:"object_storage" validate:"omitempty"`
+	Hdfs             *v2.HDFS             `json:"hdfs" validate:"omitempty"`
+}
+
+// ListTaskEntriesResponse defines the response parameters for listing task entries.
+type ListTaskEntriesResponse struct {
+	Entries     []*dfdaemonv2.Entry `json:"entries"`
+	Recursive   bool                `json:"recursive"`
+	SchedulerID uint                `json:"scheduler_id"`
+}
+
 // Peer represents the peer information.
 type Peer struct {
 	ID        string    `json:"id"`
diff --git a/pkg/rpc/dfdaemon/client/client_v2.go b/pkg/rpc/dfdaemon/client/client_v2.go
index cac9b8ab..6fc2fa31 100644
--- a/pkg/rpc/dfdaemon/client/client_v2.go
+++ b/pkg/rpc/dfdaemon/client/client_v2.go
@@ -89,6 +89,9 @@ type V2 interface {
 	// DeleteTask deletes task from p2p network.
 	DeleteTask(context.Context, *dfdaemonv2.DeleteTaskRequest, ...grpc.CallOption) error
 
+	// ListTaskEntries lists task entries.
+	ListTaskEntries(context.Context, *dfdaemonv2.ListTaskEntriesRequest, ...grpc.CallOption) (*dfdaemonv2.ListTaskEntriesResponse, error)
+
 	// DownloadPersistentCacheTask downloads persistent cache task from p2p network.
 	DownloadPersistentCacheTask(context.Context, *dfdaemonv2.DownloadPersistentCacheTaskRequest, ...grpc.CallOption) (dfdaemonv2.DfdaemonUpload_DownloadPersistentCacheTaskClient, error)
 
@@ -192,3 +195,11 @@ func (v *v2) DeletePersistentCacheTask(ctx context.Context, req *dfdaemonv2.Dele
 	_, err := v.DfdaemonUploadClient.DeletePersistentCacheTask(ctx, req, opts...)
 	return err
 }
+
+// ListTaskEntries lists task entries from p2p network.
+func (v *v2) ListTaskEntries(ctx context.Context, req *dfdaemonv2.ListTaskEntriesRequest, opts ...grpc.CallOption) (*dfdaemonv2.ListTaskEntriesResponse, error) {
+	ctx, cancel := context.WithTimeout(ctx, contextTimeout)
+	defer cancel()
+
+	return v.DfdaemonUploadClient.ListTaskEntries(ctx, req, opts...)
+}
diff --git a/scheduler/job/job.go b/scheduler/job/job.go
index 9ac412dc..9ac42eb9 100644
--- a/scheduler/job/job.go
+++ b/scheduler/job/job.go
@@ -59,6 +59,9 @@ type Job interface {
 	// GetTask retrieves task information from all hosts in the cluster.
 	GetTask(context.Context, *internaljob.GetTaskRequest, *logger.SugaredLoggerOnWith) (*internaljob.GetTaskResponse, error)
 
+	// ListTaskEntries lists all task entries.
+	ListTaskEntries(context.Context, *internaljob.ListTaskEntriesRequest, *logger.SugaredLoggerOnWith) (*internaljob.ListTaskEntriesResponse, error)
+
 	// PreheatSinglePeer preheats job by single seed peer, scheduler will trigger seed peer to download task.
 	PreheatSingleSeedPeer(context.Context, *internaljob.PreheatRequest, *logger.SugaredLoggerOnWith) (*internaljob.PreheatResponse, error)
 
@@ -1034,3 +1037,47 @@ func (j *job) deleteTask(ctx context.Context, data string) (string, error) {
 		SchedulerClusterID: j.config.Manager.SchedulerClusterID,
 	})
 }
+
+func (j *job) ListTaskEntries(ctx context.Context, req *internaljob.ListTaskEntriesRequest, log *logger.SugaredLoggerOnWith) (*internaljob.ListTaskEntriesResponse, error) {
+	advertiseIP := j.config.Server.AdvertiseIP.String()
+
+	selected, err := j.resource.SeedPeer().Select(ctx, req.TaskID)
+	if err != nil {
+		return nil, err
+	}
+
+	addr := fmt.Sprintf("%s:%d", selected.IP, selected.Port)
+	log.Infof("selected seed peer %s for task %s", addr, req.TaskID)
+
+	dfdaemonClient, err := dfdaemonclient.GetV2ByAddr(ctx, addr, j.dialOptions...)
+	if err != nil {
+		log.Errorf("[list-task-entries] get dfdaemon client failed: %s", err)
+		return nil, err
+	}
+
+	res, err := dfdaemonClient.ListTaskEntries(ctx, &dfdaemonv2.ListTaskEntriesRequest{
+		TaskId:           req.TaskID,
+		Url:              req.Url,
+		RequestHeader:    req.Header,
+		Timeout:          req.Timeout,
+		CertificateChain: req.CertificateChain,
+		ObjectStorage:    req.ObjectStorage,
+		Hdfs:             req.Hdfs,
+		RemoteIp:         &advertiseIP,
+	})
+	if err != nil {
+		log.Errorf("[list-task-entries]list task entries failed: %s", err)
+		return nil, err
+	}
+
+	var recursive bool
+	if len(res.Entries) > 1 {
+		recursive = true
+	}
+
+	return &internaljob.ListTaskEntriesResponse{
+		Recursive:   recursive,
+		Entries:     res.Entries,
+		SchedulerID: j.config.Manager.SchedulerClusterID,
+	}, nil
+}
diff --git a/scheduler/job/mocks/job_mock.go b/scheduler/job/mocks/job_mock.go
index be0bc9fc..abc5458c 100644
--- a/scheduler/job/mocks/job_mock.go
+++ b/scheduler/job/mocks/job_mock.go
@@ -57,6 +57,21 @@ func (mr *MockJobMockRecorder) GetTask(arg0, arg1, arg2 any) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTask", reflect.TypeOf((*MockJob)(nil).GetTask), arg0, arg1, arg2)
 }
 
+// ListTaskEntries mocks base method.
+func (m *MockJob) ListTaskEntries(arg0 context.Context, arg1 *job.ListTaskEntriesRequest, arg2 *logger.SugaredLoggerOnWith) (*job.ListTaskEntriesResponse, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ListTaskEntries", arg0, arg1, arg2)
+	ret0, _ := ret[0].(*job.ListTaskEntriesResponse)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ListTaskEntries indicates an expected call of ListTaskEntries.
+func (mr *MockJobMockRecorder) ListTaskEntries(arg0, arg1, arg2 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListTaskEntries", reflect.TypeOf((*MockJob)(nil).ListTaskEntries), arg0, arg1, arg2)
+}
+
 // PreheatAllPeers mocks base method.
 func (m *MockJob) PreheatAllPeers(arg0 context.Context, arg1 *job.PreheatRequest, arg2 *logger.SugaredLoggerOnWith) (*job.PreheatResponse, error) {
 	m.ctrl.T.Helper()
diff --git a/scheduler/metrics/metrics.go b/scheduler/metrics/metrics.go
index cd1a2e71..dbcb0f6e 100644
--- a/scheduler/metrics/metrics.go
+++ b/scheduler/metrics/metrics.go
@@ -485,6 +485,34 @@ var (
 		Help:      "Counter of the number of failed of the stating image.",
 	})
 
+	PreheatFileCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: types.MetricsNamespace,
+		Subsystem: types.SchedulerMetricsName,
+		Name:      "preheat_file_total",
+		Help:      "Counter of the total preheat file.",
+	})
+
+	PreheatFileFailureCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: types.MetricsNamespace,
+		Subsystem: types.SchedulerMetricsName,
+		Name:      "preheat_file_failure_total",
+		Help:      "Counter of the total preheat file failure.",
+	})
+
+	StatFileCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: types.MetricsNamespace,
+		Subsystem: types.SchedulerMetricsName,
+		Name:      "stat_file_total",
+		Help:      "Counter of the number of the stating file.",
+	})
+
+	StatFileFailureCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: types.MetricsNamespace,
+		Subsystem: types.SchedulerMetricsName,
+		Name:      "stat_file_failure_total",
+		Help:      "Counter of the number of failed of the stating file.",
+	})
+
 	VersionGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
 		Namespace: types.MetricsNamespace,
 		Subsystem: types.SchedulerMetricsName,
diff --git a/scheduler/resource/standard/task.go b/scheduler/resource/standard/task.go
index d2b9d702..06c91774 100644
--- a/scheduler/resource/standard/task.go
+++ b/scheduler/resource/standard/task.go
@@ -450,7 +450,7 @@ func (t *Task) PeerOutDegree(key string) (int, error) {
 }
 
 // HasAvailablePeer returns whether there is an available peer.
-func (t *Task) HasAvailablePeer(blocklist set.SafeSet[string]) bool {
+func (t *Task) HasAvailablePeer(hostID string, blocklist set.SafeSet[string]) bool {
 	var hasAvailablePeer bool
 	for _, vertex := range t.DAG.GetVertices() {
 		peer := vertex.Value
@@ -458,6 +458,10 @@ func (t *Task) HasAvailablePeer(blocklist set.SafeSet[string]) bool {
 			continue
 		}
 
+		if peer.Host.ID == hostID {
+			continue
+		}
+
 		if blocklist.Contains(peer.ID) {
 			continue
 		}
diff --git a/scheduler/rpcserver/scheduler_server_v2.go b/scheduler/rpcserver/scheduler_server_v2.go
index 7ac92704..f963164e 100644
--- a/scheduler/rpcserver/scheduler_server_v2.go
+++ b/scheduler/rpcserver/scheduler_server_v2.go
@@ -342,12 +342,42 @@ func (s *schedulerServerV2) StatImage(ctx context.Context, req *schedulerv2.Stat
 	return resp, nil
 }
 
-// TODO(EvanCley): Implement the following methods.
+// PreheatFile synchronously triggers an asynchronous preheat task for a file.
+//
+// This is a blocking call. The RPC will not return until the server has completed the
+// initial synchronous work: preparing the file URL.
+//
+// After this call successfully returns, a scheduler on the server begins the actual
+// preheating process, instructing peers to download the file in the background.
+//
+// A successful response (google.protobuf.Empty) confirms that the preparation is complete
+// and the asynchronous download task has been scheduled.
 func (s *schedulerServerV2) PreheatFile(ctx context.Context, req *schedulerv2.PreheatFileRequest) (*emptypb.Empty, error) {
+	// Collect PreheatFileCount metrics.
+	metrics.PreheatFileCount.Inc()
+	if err := s.service.PreheatFile(ctx, req); err != nil {
+		// Collect PreheatFileFailureCount metrics.
+		metrics.PreheatFileFailureCount.Inc()
+		return nil, err
+	}
+
 	return new(emptypb.Empty), nil
 }
 
-// TODO(EvanCley): Implement the following methods.
+// StatFile provides detailed status for files distribution in peers.
+//
+// This is a blocking call that first queries the file/dir entries and then queries
+// all peers to collect the file's download state across the network.
+// The response includes the file status on each peer.
 func (s *schedulerServerV2) StatFile(ctx context.Context, req *schedulerv2.StatFileRequest) (*schedulerv2.StatFileResponse, error) {
-	return nil, nil
+	// Collect StatFileCount metrics.
+	metrics.StatFileCount.Inc()
+	resp, err := s.service.StatFile(ctx, req)
+	if err != nil {
+		// Collect StatFileFailureCount metrics.
+		metrics.StatFileFailureCount.Inc()
+		return nil, err
+	}
+
+	return resp, nil
 }
diff --git a/scheduler/service/service_v1.go b/scheduler/service/service_v1.go
index 2cfbd7f3..6753a0f3 100644
--- a/scheduler/service/service_v1.go
+++ b/scheduler/service/service_v1.go
@@ -431,7 +431,7 @@ func (v *V1) StatTask(ctx context.Context, req *schedulerv1.StatTaskRequest) (*s
 		TotalPieceCount:  task.TotalPieceCount.Load(),
 		State:            task.FSM.Current(),
 		PeerCount:        int32(task.PeerCount()),
-		HasAvailablePeer: task.HasAvailablePeer(set.NewSafeSet[string]()),
+		HasAvailablePeer: task.HasAvailablePeer("", set.NewSafeSet[string]()),
 	}, nil
 }
 
@@ -696,7 +696,7 @@ func (v *V1) triggerTask(ctx context.Context, req *schedulerv1.PeerTaskRequest,
 	blocklist.Add(peer.ID)
 	if (task.FSM.Is(resource.TaskStateRunning) ||
 		task.FSM.Is(resource.TaskStateSucceeded)) &&
-		task.HasAvailablePeer(blocklist) {
+		task.HasAvailablePeer(host.ID, blocklist) {
 		peer.Log.Info("peer does not need to trigger")
 		return nil
 	}
diff --git a/scheduler/service/service_v2.go b/scheduler/service/service_v2.go
index 1d50666a..20940165 100644
--- a/scheduler/service/service_v2.go
+++ b/scheduler/service/service_v2.go
@@ -1123,7 +1123,7 @@ func (v *V2) handleRegisterPeerRequest(ctx context.Context, stream schedulerv2.S
 		task.FSM.Is(standard.TaskStateFailed) ||
 		task.FSM.Is(standard.TaskStateLeave) ||
 		task.FSM.Is(standard.TaskStateSucceeded) &&
-			!task.HasAvailablePeer(blocklist):
+			!task.HasAvailablePeer(hostID, blocklist):
 
 		// If HostType is normal, trigger seed peer download back-to-source.
 		if host.Type == types.HostTypeNormal {
@@ -3159,3 +3159,220 @@ func (v *V2) StatImage(ctx context.Context, req *schedulerv2.StatImageRequest) (
 	log.Infof("stat image finished, total layers: %d, total peers: %d", len(resp.Image.Layers), len(resp.Peers))
 	return resp, nil
 }
+
+// PreheatFile synchronously triggers an asynchronous preheat task for a file.
+//
+// This is a blocking call. The RPC will not return until the server has completed the
+// initial synchronous work: preparing the file URL.
+//
+// After this call successfully returns, a scheduler on the server begins the actual
+// preheating process, instructing peers to download the file in the background.
+//
+// A successful response (google.protobuf.Empty) confirms that the preparation is complete
+// and the asynchronous download task has been scheduled.
+func (v *V2) PreheatFile(ctx context.Context, req *schedulerv2.PreheatFileRequest) error {
+	log := logger.WithPreheatFile(req.Url)
+
+	if req.Scope == "" {
+		req.Scope = managertypes.SingleSeedPeerScope
+	}
+
+	if req.ConcurrentTaskCount == nil {
+		concurrentTaskCount := int64(managertypes.DefaultPreheatConcurrentTaskCount)
+		req.ConcurrentTaskCount = &concurrentTaskCount
+	}
+
+	if req.ConcurrentPeerCount == nil {
+		concurrentPeerCount := int64(managertypes.DefaultPreheatConcurrentPeerCount)
+		req.ConcurrentPeerCount = &concurrentPeerCount
+	}
+
+	if len(req.FilteredQueryParams) == 0 {
+		req.FilteredQueryParams = http.DefaultFilteredQueryParams
+	}
+
+	if req.Timeout == nil {
+		req.Timeout = durationpb.New(managertypes.DefaultJobTimeout)
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, req.GetTimeout().AsDuration())
+	defer cancel()
+
+	// For files preheating, we get entries from client first
+	listResp, err := v.job.ListTaskEntries(ctx, &internaljob.ListTaskEntriesRequest{
+		TaskID:           idgen.TaskIDV2ByURLBased(req.GetUrl(), req.PieceLength, req.GetTag(), req.GetApplication(), req.FilteredQueryParams),
+		Url:              req.GetUrl(),
+		Timeout:          req.GetTimeout(),
+		Header:           req.GetHeader(),
+		CertificateChain: req.GetCertificateChain(),
+		ObjectStorage:    req.GetObjectStorage(),
+		Hdfs:             req.GetHdfs(),
+	}, log)
+	if err != nil {
+		return status.Errorf(codes.InvalidArgument, "failed to list task entries: %s", err)
+	}
+
+	var urls []string
+	for _, entry := range listResp.Entries {
+		urls = append(urls, entry.Url)
+	}
+
+	// Create a preheat request for the job queue.
+	preheatRequest := &internaljob.PreheatRequest{
+		URLs:                urls,
+		Application:         req.GetApplication(),
+		FilteredQueryParams: idgen.FormatFilteredQueryParams(req.GetFilteredQueryParams()),
+		Headers:             req.GetHeader(),
+		PieceLength:         req.PieceLength,
+		Priority:            int32(req.GetPriority()),
+		Scope:               req.GetScope(),
+		IPs:                 req.GetIps(),
+		Count:               req.Count,
+		Percentage:          req.Percentage,
+		ConcurrentTaskCount: req.GetConcurrentTaskCount(),
+		ConcurrentPeerCount: req.GetConcurrentPeerCount(),
+		Timeout:             req.GetTimeout().AsDuration(),
+		CertificateChain:    req.GetCertificateChain(),
+		InsecureSkipVerify:  req.GetInsecureSkipVerify(),
+	}
+
+	switch req.GetScope() {
+	case managertypes.SingleSeedPeerScope:
+		go func() {
+			ctx, cancel := context.WithTimeout(context.Background(), req.GetTimeout().AsDuration())
+			defer cancel()
+
+			log.Info("preheat single seed peer")
+			resp, err := v.job.PreheatSingleSeedPeer(ctx, preheatRequest, log)
+			if err != nil {
+				log.Errorf("preheat single seed peer failed: %s", err.Error())
+				return
+			}
+
+			log.Infof("preheat single seed peer finished, response: %#v", resp)
+		}()
+	case managertypes.AllSeedPeersScope:
+		go func() {
+			ctx, cancel := context.WithTimeout(context.Background(), req.GetTimeout().AsDuration())
+			defer cancel()
+
+			log.Info("preheat all seed peers")
+			resp, err := v.job.PreheatAllSeedPeers(ctx, preheatRequest, log)
+			if err != nil {
+				log.Errorf("preheat all seed peers failed: %s", err.Error())
+				return
+			}
+
+			log.Infof("preheat all seed peers finished, success count: %d, failed count: %d", len(resp.SuccessTasks), len(resp.FailureTasks))
+		}()
+	case managertypes.AllPeersScope:
+		go func() {
+			ctx, cancel := context.WithTimeout(context.Background(), req.GetTimeout().AsDuration())
+			defer cancel()
+
+			log.Info("preheat all peers")
+			resp, err := v.job.PreheatAllPeers(ctx, preheatRequest, log)
+			if err != nil {
+				log.Errorf("preheat all peers failed: %s", err.Error())
+				return
+			}
+
+			log.Infof("preheat all peers finished, success count: %d, failed count: %d", len(resp.SuccessTasks), len(resp.FailureTasks))
+		}()
+	default:
+		return status.Errorf(codes.InvalidArgument, "unsupported preheat scope: %s", req.Scope)
+	}
+
+	return nil
+}
+
+// StatFile provides detailed status for files distribution in peers.
+//
+// This is a blocking call that first queries the file/dir entries and then queries
+// all peers to collect the file's download state across the network.
+// The response includes the file status on each peer.
+func (v *V2) StatFile(ctx context.Context, req *schedulerv2.StatFileRequest) (*schedulerv2.StatFileResponse, error) {
+	log := logger.WithStatFile(req.Url)
+
+	if req.ConcurrentPeerCount == nil {
+		concurrentPeerCount := int64(managertypes.DefaultPreheatConcurrentPeerCount)
+		req.ConcurrentPeerCount = &concurrentPeerCount
+	}
+
+	if len(req.FilteredQueryParams) == 0 {
+		req.FilteredQueryParams = http.DefaultFilteredQueryParams
+	}
+
+	if req.Timeout == nil {
+		req.Timeout = durationpb.New(managertypes.DefaultJobTimeout)
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, req.GetTimeout().AsDuration())
+	defer cancel()
+
+	listResp, err := v.job.ListTaskEntries(ctx, &internaljob.ListTaskEntriesRequest{
+		TaskID:           idgen.TaskIDV2ByURLBased(req.GetUrl(), req.PieceLength, req.GetTag(), req.GetApplication(), req.FilteredQueryParams),
+		Url:              req.GetUrl(),
+		Timeout:          req.GetTimeout(),
+		Header:           req.GetHeader(),
+		CertificateChain: req.GetCertificateChain(),
+		ObjectStorage:    req.GetObjectStorage(),
+		Hdfs:             req.GetHdfs(),
+	}, log)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "failed to list task entries: %s", err)
+	}
+
+	resp := &schedulerv2.StatFileResponse{
+		Peers: make([]*schedulerv2.PeerFile, 0),
+	}
+
+	var mu sync.Mutex
+	peers := map[string]*schedulerv2.PeerFile{}
+	eg, ctx := errgroup.WithContext(ctx)
+	for _, entry := range listResp.Entries {
+		eg.Go(func() error {
+			taskID := idgen.TaskIDV2ByURLBased(entry.Url, req.PieceLength, req.GetTag(), req.GetApplication(), req.FilteredQueryParams)
+			getTaskRequest := &internaljob.GetTaskRequest{
+				TaskID:              taskID,
+				Timeout:             req.GetTimeout().AsDuration(),
+				ConcurrentPeerCount: *req.ConcurrentPeerCount,
+			}
+
+			log := logger.WithStatFileAndTaskID(entry.Url, taskID)
+			log.Infof("get task request: %#v", getTaskRequest)
+			task, err := v.job.GetTask(ctx, getTaskRequest, log)
+			if err != nil {
+				log.Errorf("get task failed: %s", err.Error())
+				return nil
+			}
+			log.Infof("get length of peers: %d", len(task.Peers))
+
+			for _, peer := range task.Peers {
+				hostID := idgen.HostIDV2(peer.IP, peer.Hostname, false)
+				mu.Lock()
+				if _, exists := peers[hostID]; !exists {
+					peers[hostID] = &schedulerv2.PeerFile{
+						Ip:       peer.IP,
+						Hostname: peer.Hostname,
+					}
+				}
+				mu.Unlock()
+			}
+			return nil
+		})
+	}
+
+	// If any of the goroutines return an error, ignore it and continue processing.
+	if err := eg.Wait(); err != nil {
+		logger.Errorf("failed to create get task jobs: %w", err)
+	}
+
+	for _, peer := range peers {
+		resp.Peers = append(resp.Peers, peer)
+		log.Infof("stat file for peer %s", peer.Ip)
+	}
+
+	log.Infof("stat file finished, total files: %d, total peers: %d", len(listResp.Entries), len(resp.Peers))
+	return resp, nil
+}



Test patch:
diff --git a/scheduler/resource/standard/task_test.go b/scheduler/resource/standard/task_test.go
index 470885d3..f04a09bf 100644
--- a/scheduler/resource/standard/task_test.go
+++ b/scheduler/resource/standard/task_test.go
@@ -979,7 +979,18 @@ func TestTask_HasAvailablePeer(t *testing.T) {
 
 				blocklist := set.NewSafeSet[string]()
 				blocklist.Add(mockPeer.ID)
-				assert.Equal(task.HasAvailablePeer(blocklist), false)
+				assert.Equal(task.HasAvailablePeer("", blocklist), false)
+			},
+		},
+		{
+			name: "host id is equal to the peer's host id",
+			expect: func(t *testing.T, task *Task, mockPeer *Peer) {
+				assert := assert.New(t)
+				mockPeer.FSM.SetState(PeerStateSucceeded)
+				task.StorePeer(mockPeer)
+
+				blocklist := set.NewSafeSet[string]()
+				assert.Equal(task.HasAvailablePeer(mockPeer.Host.ID, blocklist), false)
 			},
 		},
 		{
@@ -990,7 +1001,7 @@ func TestTask_HasAvailablePeer(t *testing.T) {
 				mockPeer.ID = idgen.PeerIDV2()
 				mockPeer.FSM.SetState(PeerStateSucceeded)
 				task.StorePeer(mockPeer)
-				assert.Equal(task.HasAvailablePeer(set.NewSafeSet[string]()), true)
+				assert.Equal(task.HasAvailablePeer("", set.NewSafeSet[string]()), true)
 			},
 		},
 		{
@@ -1001,7 +1012,7 @@ func TestTask_HasAvailablePeer(t *testing.T) {
 				mockPeer.ID = idgen.PeerIDV2()
 				mockPeer.FSM.SetState(PeerStateRunning)
 				task.StorePeer(mockPeer)
-				assert.Equal(task.HasAvailablePeer(set.NewSafeSet[string]()), true)
+				assert.Equal(task.HasAvailablePeer("", set.NewSafeSet[string]()), true)
 			},
 		},
 		{
@@ -1012,14 +1023,14 @@ func TestTask_HasAvailablePeer(t *testing.T) {
 				mockPeer.ID = idgen.PeerIDV2()
 				mockPeer.FSM.SetState(PeerStateBackToSource)
 				task.StorePeer(mockPeer)
-				assert.Equal(task.HasAvailablePeer(set.NewSafeSet[string]()), true)
+				assert.Equal(task.HasAvailablePeer("", set.NewSafeSet[string]()), true)
 			},
 		},
 		{
 			name: "peer does not exist",
 			expect: func(t *testing.T, task *Task, mockPeer *Peer) {
 				assert := assert.New(t)
-				assert.Equal(task.HasAvailablePeer(set.NewSafeSet[string]()), false)
+				assert.Equal(task.HasAvailablePeer("", set.NewSafeSet[string]()), false)
 			},
 		},
 	}
diff --git a/scheduler/service/service_v2_test.go b/scheduler/service/service_v2_test.go
index c31cbcb5..f5d3e6cb 100644
--- a/scheduler/service/service_v2_test.go
+++ b/scheduler/service/service_v2_test.go
@@ -3990,3 +3990,345 @@ func TestServiceV2_StatImage(t *testing.T) {
 		})
 	}
 }
+
+func TestServiceV2_PreheatFile(t *testing.T) {
+	tests := []struct {
+		name string
+		req  *schedulerv2.PreheatFileRequest
+		run  func(t *testing.T, svc *V2, req *schedulerv2.PreheatFileRequest, mj *jobmocks.MockJobMockRecorder)
+	}{
+		{
+			name: "unsupported preheat scope",
+			req: &schedulerv2.PreheatFileRequest{
+				Url:   "https://example.com/file.txt",
+				Scope: "invalid-scope",
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.PreheatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{
+					Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/file.txt"}},
+				}, nil).Times(1)
+
+				assert := assert.New(t)
+				assert.ErrorIs(svc.PreheatFile(context.Background(), req), status.Errorf(codes.InvalidArgument, "unsupported preheat scope: invalid-scope"))
+			},
+		},
+		{
+			name: "preheat scope is empty",
+			req: &schedulerv2.PreheatFileRequest{
+				Url: "https://example.com/file.txt",
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.PreheatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(1)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/file.txt"}}}, nil).Times(1),
+					mj.PreheatSingleSeedPeer(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.PreheatRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(nil, nil).Times(1),
+				)
+
+				assert := assert.New(t)
+				assert.NoError(svc.PreheatFile(context.Background(), req))
+			},
+		},
+		{
+			name: "preheat single_seed_peer",
+			req: &schedulerv2.PreheatFileRequest{
+				Url:   "https://example.com/file.txt",
+				Scope: managertypes.SingleSeedPeerScope,
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.PreheatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(1)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/file.txt"}}}, nil).Times(1),
+					mj.PreheatSingleSeedPeer(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.PreheatRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(nil, nil).Times(1),
+				)
+				assert := assert.New(t)
+				assert.NoError(svc.PreheatFile(context.Background(), req))
+			},
+		},
+		{
+			name: "preheat single_seed_peer failed",
+			req: &schedulerv2.PreheatFileRequest{
+				Url:   "https://example.com/file.txt",
+				Scope: managertypes.SingleSeedPeerScope,
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.PreheatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(1)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/file.txt"}}}, nil).Times(1),
+					mj.PreheatSingleSeedPeer(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.PreheatRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(nil, nil).Times(1),
+				)
+				assert := assert.New(t)
+				assert.NoError(svc.PreheatFile(context.Background(), req))
+			},
+		},
+		{
+			name: "preheat all_seed_peers",
+			req: &schedulerv2.PreheatFileRequest{
+				Url:   "https://example.com/file.txt",
+				Scope: managertypes.AllSeedPeersScope,
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.PreheatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(1)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/file.txt"}}}, nil).Times(1),
+					mj.PreheatAllSeedPeers(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.PreheatRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(&internaljob.PreheatResponse{SuccessTasks: make([]*internaljob.PreheatSuccessTask, 0), FailureTasks: make([]*internaljob.PreheatFailureTask, 0)}, nil).Times(1),
+				)
+				assert := assert.New(t)
+				assert.NoError(svc.PreheatFile(context.Background(), req))
+			},
+		},
+		{
+			name: "preheat all_seed_peers failed",
+			req: &schedulerv2.PreheatFileRequest{
+				Url:   "https://example.com/file.txt",
+				Scope: managertypes.AllSeedPeersScope,
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.PreheatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(1)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/file.txt"}}}, nil).Times(1),
+					mj.PreheatAllSeedPeers(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.PreheatRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(&internaljob.PreheatResponse{SuccessTasks: make([]*internaljob.PreheatSuccessTask, 0), FailureTasks: make([]*internaljob.PreheatFailureTask, 0)}, nil).Times(1),
+				)
+				assert := assert.New(t)
+				assert.NoError(svc.PreheatFile(context.Background(), req))
+			},
+		},
+		{
+			name: "preheat all_peers",
+			req: &schedulerv2.PreheatFileRequest{
+				Url:   "https://example.com/file.txt",
+				Scope: managertypes.AllPeersScope,
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.PreheatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(1)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/file.txt"}}}, nil).Times(1),
+					mj.PreheatAllPeers(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.PreheatRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(&internaljob.PreheatResponse{SuccessTasks: make([]*internaljob.PreheatSuccessTask, 0), FailureTasks: make([]*internaljob.PreheatFailureTask, 0)}, nil).Times(1),
+				)
+				assert := assert.New(t)
+				assert.NoError(svc.PreheatFile(context.Background(), req))
+			},
+		},
+		{
+			name: "preheat all_peers failed",
+			req: &schedulerv2.PreheatFileRequest{
+				Url:   "https://example.com/file.txt",
+				Scope: managertypes.AllPeersScope,
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.PreheatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(1)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/file.txt"}}}, nil).Times(1),
+					mj.PreheatAllPeers(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.PreheatRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(&internaljob.PreheatResponse{SuccessTasks: make([]*internaljob.PreheatSuccessTask, 0), FailureTasks: make([]*internaljob.PreheatFailureTask, 0)}, nil).Times(1),
+				)
+
+				assert := assert.New(t)
+				assert.NoError(svc.PreheatFile(context.Background(), req))
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctl := gomock.NewController(t)
+			defer ctl.Finish()
+			scheduling := schedulingmocks.NewMockScheduling(ctl)
+			resource := standard.NewMockResource(ctl)
+			persistentCacheResource := persistentcache.NewMockResource(ctl)
+			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
+			job := jobmocks.NewMockJob(ctl)
+
+			svc := NewV2(&config.Config{Scheduler: mockSchedulerConfig, Metrics: config.MetricsConfig{EnableHost: true}}, resource, persistentCacheResource, scheduling, job, nil, dynconfig)
+
+			tc.run(t, svc, tc.req, job.EXPECT())
+		})
+	}
+}
+
+func TestServiceV2_StatFile(t *testing.T) {
+	tests := []struct {
+		name string
+		req  *schedulerv2.StatFileRequest
+		run  func(t *testing.T, svc *V2, req *schedulerv2.StatFileRequest, mj *jobmocks.MockJobMockRecorder)
+	}{
+		{
+			name: "list task entries failed",
+			req: &schedulerv2.StatFileRequest{
+				Url: "https://example.com/file.txt",
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.StatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, errors.New("list task entries failed")).Times(1)
+
+				resp, err := svc.StatFile(context.Background(), req)
+				assert := assert.New(t)
+				assert.Nil(resp)
+				assert.ErrorIs(err, status.Errorf(codes.InvalidArgument, "failed to list task entries: list task entries failed"))
+			},
+		},
+		{
+			name: "get task failed",
+			req: &schedulerv2.StatFileRequest{
+				Url: "https://example.com/file.txt",
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.StatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(1)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/file.txt"}}}, nil).Times(1),
+					mj.GetTask(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.GetTaskRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(nil, errors.New("get task failed")).Times(1),
+				)
+
+				resp, err := svc.StatFile(context.Background(), req)
+				assert := assert.New(t)
+				assert.NoError(err)
+				assert.Equal(0, len(resp.Peers))
+			},
+		},
+		{
+			name: "stat file by peer",
+			req: &schedulerv2.StatFileRequest{
+				Url: "https://example.com/file.txt",
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.StatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(1)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/file.txt"}}}, nil).Times(1),
+					mj.GetTask(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.GetTaskRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(&internaljob.GetTaskResponse{Peers: []*internaljob.Peer{{IP: "127.0.0.1", Hostname: "peer-1"}}}, nil).Times(1),
+				)
+
+				resp, err := svc.StatFile(context.Background(), req)
+				assert := assert.New(t)
+				assert.NoError(err)
+				assert.Equal(1, len(resp.Peers))
+			},
+		},
+		{
+			name: "stat multi files by different peer",
+			req: &schedulerv2.StatFileRequest{
+				Url: "https://example.com/dir/",
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.StatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(2)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/dir/file1.txt"}, {Url: "https://example.com/dir/file2.txt"}}}, nil).Times(1),
+					mj.GetTask(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.GetTaskRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(&internaljob.GetTaskResponse{Peers: []*internaljob.Peer{{IP: "127.0.0.1", Hostname: "peer-1"}}}, nil).Times(1),
+					mj.GetTask(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.GetTaskRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(&internaljob.GetTaskResponse{Peers: []*internaljob.Peer{{IP: "127.0.0.1", Hostname: "peer-2"}}}, nil).Times(1),
+				)
+
+				resp, err := svc.StatFile(context.Background(), req)
+				assert := assert.New(t)
+				assert.NoError(err)
+				assert.Equal(2, len(resp.Peers))
+			},
+		},
+		{
+			name: "stat multi files by peers",
+			req: &schedulerv2.StatFileRequest{
+				Url: "https://example.com/dir/",
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.StatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(2)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/dir/file1.txt"}, {Url: "https://example.com/dir/file2.txt"}}}, nil).Times(1),
+					mj.GetTask(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.GetTaskRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(&internaljob.GetTaskResponse{Peers: []*internaljob.Peer{{IP: "127.0.0.1", Hostname: "peer-1"}, {IP: "127.0.0.1", Hostname: "peer-2"}}}, nil).Times(1),
+					mj.GetTask(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.GetTaskRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(&internaljob.GetTaskResponse{Peers: []*internaljob.Peer{{IP: "127.0.0.1", Hostname: "peer-2"}, {IP: "127.0.0.1", Hostname: "peer-1"}}}, nil).Times(1),
+				)
+
+				resp, err := svc.StatFile(context.Background(), req)
+				assert := assert.New(t)
+				assert.NoError(err)
+				assert.Equal(2, len(resp.Peers))
+			},
+		},
+		{
+			name: "stat multi files by peers, but one of the get task failed",
+			req: &schedulerv2.StatFileRequest{
+				Url: "https://example.com/dir/",
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.StatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(2)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/dir/file1.txt"}, {Url: "https://example.com/dir/file2.txt"}}}, nil).Times(1),
+					mj.GetTask(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.GetTaskRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(&internaljob.GetTaskResponse{Peers: []*internaljob.Peer{{IP: "127.0.0.1", Hostname: "peer-1"}, {IP: "127.0.0.1", Hostname: "peer-2"}}}, nil).Times(1),
+					mj.GetTask(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.GetTaskRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(nil, errors.New("get task failed")).Times(1),
+				)
+
+				resp, err := svc.StatFile(context.Background(), req)
+				assert := assert.New(t)
+				assert.NoError(err)
+				assert.Equal(2, len(resp.Peers))
+			},
+		},
+		{
+			name: "stat multi files by peers, but the get task failed",
+			req: &schedulerv2.StatFileRequest{
+				Url: "https://example.com/dir/",
+			},
+			run: func(t *testing.T, svc *V2, req *schedulerv2.StatFileRequest, mj *jobmocks.MockJobMockRecorder) {
+				var wg sync.WaitGroup
+				wg.Add(1)
+				defer wg.Wait()
+
+				gomock.InOrder(
+					mj.ListTaskEntries(gomock.Any(), gomock.Any(), gomock.Any()).Return(&internaljob.ListTaskEntriesResponse{Entries: []*dfdaemonv2.Entry{{Url: "https://example.com/file.txt"}}}, nil).Times(1),
+					mj.GetTask(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(context.Context, *internaljob.GetTaskRequest, *logger.SugaredLoggerOnWith) { wg.Done() }).Return(&internaljob.GetTaskResponse{Peers: []*internaljob.Peer{{IP: "127.0.0.1", Hostname: "peer-1"}}}, nil).Times(1),
+				)
+
+				resp, err := svc.StatFile(context.Background(), req)
+				assert := assert.New(t)
+				assert.NoError(err)
+				assert.Equal(1, len(resp.Peers))
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctl := gomock.NewController(t)
+			defer ctl.Finish()
+			scheduling := schedulingmocks.NewMockScheduling(ctl)
+			resource := standard.NewMockResource(ctl)
+			persistentCacheResource := persistentcache.NewMockResource(ctl)
+			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
+			job := jobmocks.NewMockJob(ctl)
+
+			svc := NewV2(&config.Config{Scheduler: mockSchedulerConfig, Metrics: config.MetricsConfig{EnableHost: true}}, resource, persistentCacheResource, scheduling, job, nil, dynconfig)
+
+			tc.run(t, svc, tc.req, job.EXPECT())
+		})
+	}
+}