Repository: dragonflyoss/Dragonfly2
Pull request #1443 (2022-07-08 01:46:46)

feat: reload proxy option
<!--- Provide a general summary of your changes in the Title above -->

## Description

Support reload proxy rules when config file changed.

## Related Issue

<!--- This project only accepts pull requests related to open issues -->
<!--- If suggesting a new feature or change, please discuss it in an issue first -->
<!--- If fixing a bug, there should be an issue describing it with steps to reproduce -->
<!--- Please link to the issue here: -->

## Motivation and Context

<!--- Why is this change required? What problem does it solve? -->

## Screenshots (if appropriate)

## Types of changes

<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)
- [ ] Documentation Update (if none of the other choices apply)

## Checklist

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [ ] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1443](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (c9c928c) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/902d4a18c22a72ef7e20b74c8ee91a031fd137ae?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (902d4a1) will **decrease** coverage by `0.43%`.
> The diff coverage is `1.32%`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443/graphs/tree.svg?width=650&height=150&src=pr&token=PUICJYSAT7&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main    #1443      +/-   ##
==========================================
- Coverage   44.30%   43.86%   -0.44%     
==========================================
  Files         114      115       +1     
  Lines       10464    10619     +155     
==========================================
+ Hits         4636     4658      +22     
- Misses       5359     5497     +138     
+ Partials      469      464       -5     
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| unittests | `43.86% <1.32%> (-0.44%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [client/config/peerhost\_linux.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2NvbmZpZy9wZWVyaG9zdF9saW51eC5nbw==) | `0.00% <0.00%> (ø)` | |
| [client/daemon/daemon.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9kYWVtb24uZ28=) | `5.11% <0.00%> (-0.18%)` | :arrow_down: |
| [client/daemon/proxy/proxy\_manager.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wcm94eS9wcm94eV9tYW5hZ2VyLmdv) | `0.00% <0.00%> (ø)` | |
| [client/config/peerhost.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2NvbmZpZy9wZWVyaG9zdC5nbw==) | `32.62% <25.00%> (-0.24%)` | :arrow_down: |
| [client/daemon/proxy/proxy.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wcm94eS9wcm94eS5nbw==) | `13.35% <100.00%> (+0.08%)` | :arrow_up: |
| [client/daemon/peer/peertask\_conductor.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2NvbmR1Y3Rvci5nbw==) | `57.76% <0.00%> (+1.19%)` | :arrow_up: |
| [...ent/daemon/peer/peertask\_piecetask\_synchronizer.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19zeW5jaHJvbml6ZXIuZ28=) | `47.61% <0.00%> (+2.16%)` | :arrow_up: |
| [client/daemon/peer/peertask\_piecetask\_poller.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1443/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19wb2xsZXIuZ28=) | `50.72% <0.00%> (+3.62%)` | :arrow_up: |


Patch:
diff --git a/client/config/peerhost.go b/client/config/peerhost.go
index 1f3b3cd3..32f58d81 100644
--- a/client/config/peerhost.go
+++ b/client/config/peerhost.go
@@ -66,12 +66,11 @@ type DaemonOption struct {
 	ObjectStorage ObjectStorageOption `mapstructure:"objectStorage" yaml:"objectStorage"`
 	Storage       StorageOption       `mapstructure:"storage" yaml:"storage"`
 	Health        *HealthOption       `mapstructure:"health" yaml:"health"`
-	// TODO WIP, did not use
-	Reload ReloadOption `mapstructure:"reloadOption" yaml:"reloadOption"`
+	Reload        ReloadOption        `mapstructure:"reload" yaml:"reload"`
 }
 
 func NewDaemonConfig() *DaemonOption {
-	return &peerHostConfig
+	return peerHostConfig()
 }
 
 func (p *DaemonOption) Load(path string) error {
@@ -145,6 +144,10 @@ func (p *DaemonOption) Validate() error {
 		}
 	}
 
+	if p.Reload.Interval.Duration > 0 && p.Reload.Interval.Duration < time.Second {
+		return errors.New("reload interval too short, must great than 1 second")
+	}
+
 	switch p.Download.DefaultPattern {
 	case PatternP2P, PatternSeedPeer, PatternSource:
 	default:
@@ -269,7 +272,7 @@ type ProxyOption struct {
 	MaxConcurrency  int64           `mapstructure:"maxConcurrency" yaml:"maxConcurrency"`
 	RegistryMirror  *RegistryMirror `mapstructure:"registryMirror" yaml:"registryMirror"`
 	WhiteList       []*WhiteList    `mapstructure:"whiteList" yaml:"whiteList"`
-	Proxies         []*ProxyRule    `mapstructure:"proxies" yaml:"proxies"`
+	ProxyRules      []*ProxyRule    `mapstructure:"proxies" yaml:"proxies"`
 	HijackHTTPS     *HijackConfig   `mapstructure:"hijackHTTPS" yaml:"hijackHTTPS"`
 	DumpHTTPContent bool            `mapstructure:"dumpHTTPContent" yaml:"dumpHTTPContent"`
 	// ExtraRegistryMirrors add more mirror for different ports
@@ -368,7 +371,7 @@ func (p *ProxyOption) unmarshal(unmarshal func(in []byte, out any) (err error),
 	p.ExtraRegistryMirrors = pt.ExtraRegistryMirrors
 	p.ListenOption = pt.ListenOption
 	p.RegistryMirror = pt.RegistryMirror
-	p.Proxies = pt.Proxies
+	p.ProxyRules = pt.Proxies
 	p.HijackHTTPS = pt.HijackHTTPS
 	p.WhiteList = pt.WhiteList
 	p.MaxConcurrency = pt.MaxConcurrency
diff --git a/client/config/peerhost_darwin.go b/client/config/peerhost_darwin.go
index 06663a36..b881240e 100644
--- a/client/config/peerhost_darwin.go
+++ b/client/config/peerhost_darwin.go
@@ -32,133 +32,135 @@ import (
 	"d7y.io/dragonfly/v2/pkg/net/ip"
 )
 
-var peerHostConfig = DaemonOption{
-	AliveTime:   util.Duration{Duration: DefaultDaemonAliveTime},
-	GCInterval:  util.Duration{Duration: DefaultGCInterval},
-	KeepStorage: false,
-	Scheduler: SchedulerOption{
-		Manager: ManagerOption{
-			Enable:          false,
-			RefreshInterval: 5 * time.Minute,
-			SeedPeer: SeedPeerOption{
-				Enable:    false,
-				Type:      model.SeedPeerTypeSuperSeed,
-				ClusterID: 1,
-				KeepAlive: KeepAliveOption{
-					Interval: 5 * time.Second,
+var peerHostConfig = func() *DaemonOption {
+	return &DaemonOption{
+		AliveTime:   util.Duration{Duration: DefaultDaemonAliveTime},
+		GCInterval:  util.Duration{Duration: DefaultGCInterval},
+		KeepStorage: false,
+		Scheduler: SchedulerOption{
+			Manager: ManagerOption{
+				Enable:          false,
+				RefreshInterval: 5 * time.Minute,
+				SeedPeer: SeedPeerOption{
+					Enable:    false,
+					Type:      model.SeedPeerTypeSuperSeed,
+					ClusterID: 1,
+					KeepAlive: KeepAliveOption{
+						Interval: 5 * time.Second,
+					},
 				},
 			},
-		},
-		NetAddrs: []dfnet.NetAddr{
-			{
-				Type: dfnet.TCP,
-				Addr: "127.0.0.1:8002",
+			NetAddrs: []dfnet.NetAddr{
+				{
+					Type: dfnet.TCP,
+					Addr: "127.0.0.1:8002",
+				},
 			},
+			ScheduleTimeout: util.Duration{Duration: DefaultScheduleTimeout},
 		},
-		ScheduleTimeout: util.Duration{Duration: DefaultScheduleTimeout},
-	},
-	Host: HostOption{
-		Hostname:       fqdn.FQDNHostname,
-		ListenIP:       net.IPv4zero.String(),
-		AdvertiseIP:    ip.IPv4,
-		SecurityDomain: "",
-		Location:       "",
-		IDC:            "",
-		NetTopology:    "",
-	},
-	Download: DownloadOption{
-		DefaultPattern:       PatternP2P,
-		CalculateDigest:      true,
-		PieceDownloadTimeout: 30 * time.Second,
-		GetPiecesMaxRetry:    100,
-		TotalRateLimit: util.RateLimit{
-			Limit: rate.Limit(DefaultTotalDownloadLimit),
+		Host: HostOption{
+			Hostname:       fqdn.FQDNHostname,
+			ListenIP:       net.IPv4zero.String(),
+			AdvertiseIP:    ip.IPv4,
+			SecurityDomain: "",
+			Location:       "",
+			IDC:            "",
+			NetTopology:    "",
 		},
-		PerPeerRateLimit: util.RateLimit{
-			Limit: rate.Limit(DefaultPerPeerDownloadLimit),
-		},
-		DownloadGRPC: ListenOption{
-			Security: SecurityOption{
-				Insecure:  true,
-				TLSVerify: true,
+		Download: DownloadOption{
+			DefaultPattern:       PatternP2P,
+			CalculateDigest:      true,
+			PieceDownloadTimeout: 30 * time.Second,
+			GetPiecesMaxRetry:    100,
+			TotalRateLimit: util.RateLimit{
+				Limit: rate.Limit(DefaultTotalDownloadLimit),
 			},
-			UnixListen: &UnixListenOption{
-				Socket: "/tmp/dfdaemon.sock",
+			PerPeerRateLimit: util.RateLimit{
+				Limit: rate.Limit(DefaultPerPeerDownloadLimit),
 			},
-		},
-		PeerGRPC: ListenOption{
-			Security: SecurityOption{
-				Insecure:  true,
-				TLSVerify: true,
+			DownloadGRPC: ListenOption{
+				Security: SecurityOption{
+					Insecure:  true,
+					TLSVerify: true,
+				},
+				UnixListen: &UnixListenOption{
+					Socket: "/tmp/dfdaemon.sock",
+				},
 			},
-			TCPListen: &TCPListenOption{
-				Listen: net.IPv4zero.String(),
-				PortRange: TCPListenPortRange{
-					Start: DefaultPeerStartPort,
-					End:   DefaultEndPort,
+			PeerGRPC: ListenOption{
+				Security: SecurityOption{
+					Insecure:  true,
+					TLSVerify: true,
+				},
+				TCPListen: &TCPListenOption{
+					Listen: net.IPv4zero.String(),
+					PortRange: TCPListenPortRange{
+						Start: DefaultPeerStartPort,
+						End:   DefaultEndPort,
+					},
 				},
 			},
 		},
-	},
-	Upload: UploadOption{
-		RateLimit: util.RateLimit{
-			Limit: rate.Limit(DefaultUploadLimit),
-		},
-		ListenOption: ListenOption{
-			Security: SecurityOption{
-				Insecure:  true,
-				TLSVerify: false,
+		Upload: UploadOption{
+			RateLimit: util.RateLimit{
+				Limit: rate.Limit(DefaultUploadLimit),
 			},
-			TCPListen: &TCPListenOption{
-				Listen: net.IPv4zero.String(),
-				PortRange: TCPListenPortRange{
-					Start: DefaultUploadStartPort,
-					End:   DefaultEndPort,
+			ListenOption: ListenOption{
+				Security: SecurityOption{
+					Insecure:  true,
+					TLSVerify: false,
+				},
+				TCPListen: &TCPListenOption{
+					Listen: net.IPv4zero.String(),
+					PortRange: TCPListenPortRange{
+						Start: DefaultUploadStartPort,
+						End:   DefaultEndPort,
+					},
 				},
 			},
 		},
-	},
-	ObjectStorage: ObjectStorageOption{
-		Enable:      false,
-		Filter:      "Expires&Signature&ns",
-		MaxReplicas: DefaultObjectMaxReplicas,
-		ListenOption: ListenOption{
-			Security: SecurityOption{
-				Insecure:  true,
-				TLSVerify: true,
-			},
-			TCPListen: &TCPListenOption{
-				Listen: net.IPv4zero.String(),
-				PortRange: TCPListenPortRange{
-					Start: DefaultObjectStorageStartPort,
-					End:   DefaultEndPort,
+		ObjectStorage: ObjectStorageOption{
+			Enable:      false,
+			Filter:      "Expires&Signature&ns",
+			MaxReplicas: DefaultObjectMaxReplicas,
+			ListenOption: ListenOption{
+				Security: SecurityOption{
+					Insecure:  true,
+					TLSVerify: true,
+				},
+				TCPListen: &TCPListenOption{
+					Listen: net.IPv4zero.String(),
+					PortRange: TCPListenPortRange{
+						Start: DefaultObjectStorageStartPort,
+						End:   DefaultEndPort,
+					},
 				},
 			},
 		},
-	},
-	Proxy: &ProxyOption{
-		ListenOption: ListenOption{
-			Security: SecurityOption{
-				Insecure:  true,
-				TLSVerify: false,
-			},
-			TCPListen: &TCPListenOption{
-				Listen:    net.IPv4zero.String(),
-				PortRange: TCPListenPortRange{},
+		Proxy: &ProxyOption{
+			ListenOption: ListenOption{
+				Security: SecurityOption{
+					Insecure:  true,
+					TLSVerify: false,
+				},
+				TCPListen: &TCPListenOption{
+					Listen:    net.IPv4zero.String(),
+					PortRange: TCPListenPortRange{},
+				},
 			},
 		},
-	},
-	Storage: StorageOption{
-		TaskExpireTime: util.Duration{
-			Duration: DefaultTaskExpireTime,
+		Storage: StorageOption{
+			TaskExpireTime: util.Duration{
+				Duration: DefaultTaskExpireTime,
+			},
+			StoreStrategy:          AdvanceLocalTaskStoreStrategy,
+			Multiplex:              false,
+			DiskGCThresholdPercent: 95,
 		},
-		StoreStrategy:          AdvanceLocalTaskStoreStrategy,
-		Multiplex:              false,
-		DiskGCThresholdPercent: 95,
-	},
-	Reload: ReloadOption{
-		Interval: util.Duration{
-			Duration: time.Minute,
+		Reload: ReloadOption{
+			Interval: util.Duration{
+				Duration: time.Minute,
+			},
 		},
-	},
+	}
 }
diff --git a/client/config/peerhost_linux.go b/client/config/peerhost_linux.go
index de4d3769..22713e92 100644
--- a/client/config/peerhost_linux.go
+++ b/client/config/peerhost_linux.go
@@ -32,132 +32,134 @@ import (
 	"d7y.io/dragonfly/v2/pkg/net/ip"
 )
 
-var peerHostConfig = DaemonOption{
-	AliveTime:   util.Duration{Duration: DefaultDaemonAliveTime},
-	GCInterval:  util.Duration{Duration: DefaultGCInterval},
-	KeepStorage: false,
-	Scheduler: SchedulerOption{
-		Manager: ManagerOption{
-			Enable:          false,
-			RefreshInterval: 5 * time.Minute,
-			SeedPeer: SeedPeerOption{
-				Enable:    false,
-				Type:      model.SeedPeerTypeSuperSeed,
-				ClusterID: 1,
-				KeepAlive: KeepAliveOption{
-					Interval: 5 * time.Second,
+var peerHostConfig = func() *DaemonOption {
+	return &DaemonOption{
+		AliveTime:   util.Duration{Duration: DefaultDaemonAliveTime},
+		GCInterval:  util.Duration{Duration: DefaultGCInterval},
+		KeepStorage: false,
+		Scheduler: SchedulerOption{
+			Manager: ManagerOption{
+				Enable:          false,
+				RefreshInterval: 5 * time.Minute,
+				SeedPeer: SeedPeerOption{
+					Enable:    false,
+					Type:      model.SeedPeerTypeSuperSeed,
+					ClusterID: 1,
+					KeepAlive: KeepAliveOption{
+						Interval: 5 * time.Second,
+					},
 				},
 			},
-		},
-		NetAddrs: []dfnet.NetAddr{
-			{
-				Type: dfnet.TCP,
-				Addr: "127.0.0.1:8002",
+			NetAddrs: []dfnet.NetAddr{
+				{
+					Type: dfnet.TCP,
+					Addr: "127.0.0.1:8002",
+				},
 			},
+			ScheduleTimeout: util.Duration{Duration: DefaultScheduleTimeout},
 		},
-		ScheduleTimeout: util.Duration{Duration: DefaultScheduleTimeout},
-	},
-	Host: HostOption{
-		Hostname:       fqdn.FQDNHostname,
-		ListenIP:       "0.0.0.0",
-		AdvertiseIP:    ip.IPv4,
-		SecurityDomain: "",
-		Location:       "",
-		IDC:            "",
-		NetTopology:    "",
-	},
-	Download: DownloadOption{
-		DefaultPattern:       PatternP2P,
-		CalculateDigest:      true,
-		PieceDownloadTimeout: 30 * time.Second,
-		GetPiecesMaxRetry:    100,
-		TotalRateLimit: util.RateLimit{
-			Limit: rate.Limit(DefaultTotalDownloadLimit),
+		Host: HostOption{
+			Hostname:       fqdn.FQDNHostname,
+			ListenIP:       "0.0.0.0",
+			AdvertiseIP:    ip.IPv4,
+			SecurityDomain: "",
+			Location:       "",
+			IDC:            "",
+			NetTopology:    "",
 		},
-		PerPeerRateLimit: util.RateLimit{
-			Limit: rate.Limit(DefaultPerPeerDownloadLimit),
-		},
-		DownloadGRPC: ListenOption{
-			Security: SecurityOption{
-				Insecure:  true,
-				TLSVerify: true,
+		Download: DownloadOption{
+			DefaultPattern:       PatternP2P,
+			CalculateDigest:      true,
+			PieceDownloadTimeout: 30 * time.Second,
+			GetPiecesMaxRetry:    100,
+			TotalRateLimit: util.RateLimit{
+				Limit: rate.Limit(DefaultTotalDownloadLimit),
 			},
-			UnixListen: &UnixListenOption{
-				Socket: "/var/run/dfdaemon.sock",
+			PerPeerRateLimit: util.RateLimit{
+				Limit: rate.Limit(DefaultPerPeerDownloadLimit),
 			},
-		},
-		PeerGRPC: ListenOption{
-			Security: SecurityOption{
-				Insecure:  true,
-				TLSVerify: true,
+			DownloadGRPC: ListenOption{
+				Security: SecurityOption{
+					Insecure:  true,
+					TLSVerify: true,
+				},
+				UnixListen: &UnixListenOption{
+					Socket: "/var/run/dfdaemon.sock",
+				},
 			},
-			TCPListen: &TCPListenOption{
-				PortRange: TCPListenPortRange{
-					Start: DefaultPeerStartPort,
-					End:   DefaultEndPort,
+			PeerGRPC: ListenOption{
+				Security: SecurityOption{
+					Insecure:  true,
+					TLSVerify: true,
+				},
+				TCPListen: &TCPListenOption{
+					PortRange: TCPListenPortRange{
+						Start: DefaultPeerStartPort,
+						End:   DefaultEndPort,
+					},
 				},
 			},
 		},
-	},
-	Upload: UploadOption{
-		RateLimit: util.RateLimit{
-			Limit: rate.Limit(DefaultUploadLimit),
-		},
-		ListenOption: ListenOption{
-			Security: SecurityOption{
-				Insecure:  true,
-				TLSVerify: false,
+		Upload: UploadOption{
+			RateLimit: util.RateLimit{
+				Limit: rate.Limit(DefaultUploadLimit),
 			},
-			TCPListen: &TCPListenOption{
-				Listen: net.IPv4zero.String(),
-				PortRange: TCPListenPortRange{
-					Start: DefaultUploadStartPort,
-					End:   DefaultEndPort,
+			ListenOption: ListenOption{
+				Security: SecurityOption{
+					Insecure:  true,
+					TLSVerify: false,
+				},
+				TCPListen: &TCPListenOption{
+					Listen: net.IPv4zero.String(),
+					PortRange: TCPListenPortRange{
+						Start: DefaultUploadStartPort,
+						End:   DefaultEndPort,
+					},
 				},
 			},
 		},
-	},
-	ObjectStorage: ObjectStorageOption{
-		Enable:      false,
-		Filter:      "Expires&Signature&ns",
-		MaxReplicas: DefaultObjectMaxReplicas,
-		ListenOption: ListenOption{
-			Security: SecurityOption{
-				Insecure:  true,
-				TLSVerify: true,
-			},
-			TCPListen: &TCPListenOption{
-				Listen: net.IPv4zero.String(),
-				PortRange: TCPListenPortRange{
-					Start: DefaultObjectStorageStartPort,
-					End:   DefaultEndPort,
+		ObjectStorage: ObjectStorageOption{
+			Enable:      false,
+			Filter:      "Expires&Signature&ns",
+			MaxReplicas: DefaultObjectMaxReplicas,
+			ListenOption: ListenOption{
+				Security: SecurityOption{
+					Insecure:  true,
+					TLSVerify: true,
+				},
+				TCPListen: &TCPListenOption{
+					Listen: net.IPv4zero.String(),
+					PortRange: TCPListenPortRange{
+						Start: DefaultObjectStorageStartPort,
+						End:   DefaultEndPort,
+					},
 				},
 			},
 		},
-	},
-	Proxy: &ProxyOption{
-		ListenOption: ListenOption{
-			Security: SecurityOption{
-				Insecure:  true,
-				TLSVerify: false,
-			},
-			TCPListen: &TCPListenOption{
-				Listen:    net.IPv4zero.String(),
-				PortRange: TCPListenPortRange{},
+		Proxy: &ProxyOption{
+			ListenOption: ListenOption{
+				Security: SecurityOption{
+					Insecure:  true,
+					TLSVerify: false,
+				},
+				TCPListen: &TCPListenOption{
+					Listen:    net.IPv4zero.String(),
+					PortRange: TCPListenPortRange{},
+				},
 			},
 		},
-	},
-	Storage: StorageOption{
-		TaskExpireTime: util.Duration{
-			Duration: DefaultTaskExpireTime,
+		Storage: StorageOption{
+			TaskExpireTime: util.Duration{
+				Duration: DefaultTaskExpireTime,
+			},
+			StoreStrategy:          AdvanceLocalTaskStoreStrategy,
+			Multiplex:              false,
+			DiskGCThresholdPercent: 95,
 		},
-		StoreStrategy:          AdvanceLocalTaskStoreStrategy,
-		Multiplex:              false,
-		DiskGCThresholdPercent: 95,
-	},
-	Reload: ReloadOption{
-		Interval: util.Duration{
-			Duration: time.Minute,
+		Reload: ReloadOption{
+			Interval: util.Duration{
+				Duration: time.Minute,
+			},
 		},
-	},
+	}
 }
diff --git a/client/daemon/daemon.go b/client/daemon/daemon.go
index 36b4449a..f63a8102 100644
--- a/client/daemon/daemon.go
+++ b/client/daemon/daemon.go
@@ -47,6 +47,7 @@ import (
 	"d7y.io/dragonfly/v2/client/daemon/storage"
 	"d7y.io/dragonfly/v2/client/daemon/upload"
 	"d7y.io/dragonfly/v2/client/util"
+	"d7y.io/dragonfly/v2/cmd/dependency"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/dfpath"
@@ -358,6 +359,10 @@ func (*clientDaemon) prepareTCPListener(opt config.ListenOption, withTLS bool) (
 }
 
 func (cd *clientDaemon) Serve() error {
+	var (
+		watchers []func(daemon *config.DaemonOption)
+		interval = cd.Option.Reload.Interval.Duration
+	)
 	cd.GCManager.Start()
 	// prepare download service listen
 	if cd.Option.Download.DownloadGRPC.UnixListen == nil {
@@ -478,6 +483,9 @@ func (cd *clientDaemon) Serve() error {
 				})
 			}
 		}
+		watchers = append(watchers, func(daemon *config.DaemonOption) {
+			cd.ProxyManager.Watch(daemon.Proxy)
+		})
 	}
 
 	// serve upload service
@@ -610,6 +618,19 @@ func (cd *clientDaemon) Serve() error {
 		}()
 	}
 
+	if len(watchers) > 0 && interval > 0 {
+		go func() {
+			dependency.WatchConfig(interval, func() any {
+				return config.NewDaemonConfig()
+			}, func(cfg any) {
+				daemonConfig := cfg.(*config.DaemonOption)
+				for _, w := range watchers {
+					w(daemonConfig)
+				}
+			})
+		}()
+	}
+
 	werr := g.Wait()
 	cd.Stop()
 	return werr
diff --git a/client/daemon/peer/peertask_conductor.go b/client/daemon/peer/peertask_conductor.go
index 4ce65848..c5dfbe8d 100644
--- a/client/daemon/peer/peertask_conductor.go
+++ b/client/daemon/peer/peertask_conductor.go
@@ -30,6 +30,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/atomic"
 	"golang.org/x/time/rate"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
 	"d7y.io/dragonfly/v2/client/config"
@@ -805,6 +806,16 @@ func (pt *peerTaskConductor) isExitPeerPacketCode(pp *scheduler.PeerPacket) bool
 		pt.failedCode = pp.Code
 		pt.failedReason = fmt.Sprintf("receive exit peer packet with code %d", pp.Code)
 		return true
+	case base.Code_BackToSourceAborted:
+		st := status.Newf(codes.Aborted, "response is not valid")
+		st, err := st.WithDetails(pp.GetSourceError())
+		if err != nil {
+			pt.Errorf("convert source error details error: %s", err.Error())
+			return false
+		}
+
+		pt.sourceErrorStatus = st
+		return true
 	}
 	return false
 }
diff --git a/client/daemon/proxy/proxy.go b/client/daemon/proxy/proxy.go
index 1dead26c..3ee88cad 100644
--- a/client/daemon/proxy/proxy.go
+++ b/client/daemon/proxy/proxy.go
@@ -36,6 +36,7 @@ import (
 	"go.opentelemetry.io/otel/propagation"
 	semconv "go.opentelemetry.io/otel/semconv/v1.7.0"
 	"go.opentelemetry.io/otel/trace"
+	"go.uber.org/atomic"
 	"golang.org/x/sync/semaphore"
 
 	"d7y.io/dragonfly/v2/client/config"
@@ -66,7 +67,7 @@ type Proxy struct {
 	registry *config.RegistryMirror
 
 	// proxy rules
-	rules []*config.ProxyRule
+	rules atomic.Value
 
 	// httpsHosts is the list of hosts whose https requests will be hijacked
 	httpsHosts []*config.HijackHost
@@ -185,7 +186,7 @@ func WithDirectHandler(h *http.ServeMux) Option {
 // WithRules sets the proxy rules
 func WithRules(rules []*config.ProxyRule) Option {
 	return func(p *Proxy) *Proxy {
-		p.rules = rules
+		p.rules.Store(rules)
 		return p
 	}
 }
@@ -540,12 +541,6 @@ func (proxy *Proxy) remoteConfig(host string) *tls.Config {
 	return nil
 }
 
-// setRules changes the rule lists of the proxy to the given rules.
-func (proxy *Proxy) setRules(rules []*config.ProxyRule) error {
-	proxy.rules = rules
-	return nil
-}
-
 // checkWhiteList check proxy white list.
 func (proxy *Proxy) checkWhiteList(r *http.Request) bool {
 	whiteList := proxy.whiteList
@@ -577,14 +572,14 @@ func (proxy *Proxy) checkWhiteList(r *http.Request) bool {
 }
 
 // shouldUseDragonfly returns whether we should use dragonfly to proxy a request. It
-// also change the scheme of the given request if the matched rule has
+// also changes the scheme of the given request if the matched rule has
 // UseHTTPS = true
 func (proxy *Proxy) shouldUseDragonfly(req *http.Request) bool {
 	if req.Method != http.MethodGet {
 		return false
 	}
 
-	for _, rule := range proxy.rules {
+	for _, rule := range proxy.rules.Load().([]*config.ProxyRule) {
 		if rule.Match(req.URL.String()) {
 			if rule.UseHTTPS {
 				req.URL.Scheme = schemaHTTPS
diff --git a/client/daemon/proxy/proxy_manager.go b/client/daemon/proxy/proxy_manager.go
index 2dec985b..c21235bc 100644
--- a/client/daemon/proxy/proxy_manager.go
+++ b/client/daemon/proxy/proxy_manager.go
@@ -30,6 +30,7 @@ import (
 	"reflect"
 
 	"github.com/spf13/viper"
+	"gopkg.in/yaml.v3"
 
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/client/daemon/peer"
@@ -39,12 +40,17 @@ import (
 )
 
 type Manager interface {
+	ConfigWatcher
 	Serve(net.Listener) error
 	ServeSNI(net.Listener) error
 	Stop() error
 	IsEnabled() bool
 }
 
+type ConfigWatcher interface {
+	Watch(*config.ProxyOption)
+}
+
 type proxyManager struct {
 	*http.Server
 	*Proxy
@@ -53,28 +59,28 @@ type proxyManager struct {
 
 var _ Manager = (*proxyManager)(nil)
 
-func NewProxyManager(peerHost *scheduler.PeerHost, peerTaskManager peer.TaskManager, defaultPattern base.Pattern, opts *config.ProxyOption) (Manager, error) {
+func NewProxyManager(peerHost *scheduler.PeerHost, peerTaskManager peer.TaskManager, defaultPattern base.Pattern, proxyOption *config.ProxyOption) (Manager, error) {
 	// proxy is option, when nil, just disable it
-	if opts == nil {
+	if proxyOption == nil {
 		logger.Infof("proxy config is empty, disabled")
 		return &proxyManager{}, nil
 	}
-	registry := opts.RegistryMirror
-	proxies := opts.Proxies
-	hijackHTTPS := opts.HijackHTTPS
-	whiteList := opts.WhiteList
+	registry := proxyOption.RegistryMirror
+	proxyRules := proxyOption.ProxyRules
+	hijackHTTPS := proxyOption.HijackHTTPS
+	whiteList := proxyOption.WhiteList
 
 	options := []Option{
 		WithPeerHost(peerHost),
 		WithPeerIDGenerator(peer.NewPeerIDGenerator(peerHost.Ip)),
 		WithPeerTaskManager(peerTaskManager),
-		WithRules(proxies),
+		WithRules(proxyRules),
 		WithWhiteList(whiteList),
-		WithMaxConcurrency(opts.MaxConcurrency),
-		WithDefaultFilter(opts.DefaultFilter),
+		WithMaxConcurrency(proxyOption.MaxConcurrency),
+		WithDefaultFilter(proxyOption.DefaultFilter),
 		WithDefaultPattern(defaultPattern),
-		WithBasicAuth(opts.BasicAuth),
-		WithDumpHTTPContent(opts.DumpHTTPContent),
+		WithBasicAuth(proxyOption.BasicAuth),
+		WithDumpHTTPContent(proxyOption.DumpHTTPContent),
 	}
 
 	if registry != nil {
@@ -82,9 +88,9 @@ func NewProxyManager(peerHost *scheduler.PeerHost, peerTaskManager peer.TaskMana
 		options = append(options, WithRegistryMirror(registry))
 	}
 
-	if len(proxies) > 0 {
-		logger.Infof("load %d proxy rules", len(proxies))
-		for i, r := range proxies {
+	if len(proxyRules) > 0 {
+		logger.Infof("load %d proxy rules", len(proxyRules))
+		for i, r := range proxyRules {
 			method := "with dragonfly"
 			if r.Direct {
 				method = "directly"
@@ -119,7 +125,7 @@ func NewProxyManager(peerHost *scheduler.PeerHost, peerTaskManager peer.TaskMana
 	return &proxyManager{
 		Server:       &http.Server{},
 		Proxy:        p,
-		ListenOption: opts.ListenOption,
+		ListenOption: proxyOption.ListenOption,
 	}, nil
 }
 
@@ -141,6 +147,26 @@ func (pm *proxyManager) IsEnabled() bool {
 	return pm.ListenOption.TCPListen != nil && pm.ListenOption.TCPListen.PortRange.Start != 0
 }
 
+func (pm *proxyManager) Watch(opt *config.ProxyOption) {
+	old, err := yaml.Marshal(pm.Proxy.rules.Load().([]*config.ProxyRule))
+	if err != nil {
+		logger.Errorf("yaml marshal proxy rules error: %s", err.Error())
+		return
+	}
+
+	fresh, err := yaml.Marshal(opt.ProxyRules)
+	if err != nil {
+		logger.Errorf("yaml marshal proxy rules error: %s", err.Error())
+		return
+	}
+	logger.Infof("previous rules: %s", string(old))
+	logger.Infof("current rules: %s", string(fresh))
+	if string(old) != string(fresh) {
+		logger.Infof("update proxy rules")
+		pm.Proxy.rules.Store(opt.ProxyRules)
+	}
+}
+
 func newDirectHandler() *http.ServeMux {
 	s := http.DefaultServeMux
 	s.HandleFunc("/args", getArgs)
@@ -195,7 +221,6 @@ func getArgs(w http.ResponseWriter, r *http.Request) {
 }
 
 func certFromFile(certFile string, keyFile string) (*tls.Certificate, error) {
-
 	// cert.Certificate is a chain of one or more certificates, leaf first.
 	cert, err := tls.LoadX509KeyPair(certFile, keyFile)
 	if err != nil {
diff --git a/client/daemon/storage/local_storage.go b/client/daemon/storage/local_storage.go
index b2ee529e..ffbe9ace 100644
--- a/client/daemon/storage/local_storage.go
+++ b/client/daemon/storage/local_storage.go
@@ -216,7 +216,7 @@ func (t *localTaskStore) UpdateTask(ctx context.Context, req *UpdateTaskRequest)
 		t.PieceMd5Sign = req.PieceMd5Sign
 		t.Debugf("update piece md5 sign: %s", t.PieceMd5Sign)
 	}
-	if t.Header == nil && req.Header != nil {
+	if t.Header == nil && req.Header != nil && len(*req.Header) > 0 {
 		t.Header = req.Header
 		t.Debugf("update header: %#v", t.Header)
 	}
diff --git a/cmd/dependency/dependency.go b/cmd/dependency/dependency.go
index 0f058533..326633e3 100644
--- a/cmd/dependency/dependency.go
+++ b/cmd/dependency/dependency.go
@@ -19,6 +19,7 @@ package dependency
 import (
 	"context"
 	"fmt"
+	"io/ioutil"
 	"os"
 	"os/signal"
 	"path/filepath"
@@ -53,9 +54,9 @@ import (
 	"d7y.io/dragonfly/v2/version"
 )
 
-// InitCobra initializes flags binding and common sub cmds.
+// InitCommandAndConfig initializes flags binding and common sub cmds.
 // config is a pointer to configuration struct.
-func InitCobra(cmd *cobra.Command, useConfigFile bool, config any) {
+func InitCommandAndConfig(cmd *cobra.Command, useConfigFile bool, config any) {
 	rootName := cmd.Root().Name()
 	cobra.OnInitialize(func() { initConfig(useConfigFile, rootName, config) })
 
@@ -186,6 +187,48 @@ func initConfig(useConfigFile bool, name string, config any) {
 	}
 }
 
+func LoadConfig(config any) error {
+	if err := viper.ReadInConfig(); err != nil {
+		return err
+	}
+	return viper.Unmarshal(config, initDecoderConfig)
+}
+
+func WatchConfig(interval time.Duration, newConfig func() (cfg any), watcher func(cfg any)) {
+	var oldData string
+	file := viper.ConfigFileUsed()
+
+	data, err := ioutil.ReadFile(file)
+	if err != nil {
+		logger.Errorf("read file %s error: %v", file, err)
+	}
+	oldData = string(data)
+loop:
+	for {
+		select {
+		case <-time.After(interval):
+			// for k8s configmap case, the config file is symbol link
+			// reload file instead use fsnotify
+			data, err = ioutil.ReadFile(file)
+			if err != nil {
+				logger.Errorf("read file %s error: %v", file, err)
+				continue loop
+			}
+			if oldData != string(data) {
+				cfg := newConfig()
+				err = LoadConfig(cfg)
+				if err != nil {
+					logger.Errorf("load config file %s error: %v", file, err)
+					continue loop
+				}
+				logger.Infof("config file %s changed", file)
+				watcher(cfg)
+				oldData = string(data)
+			}
+		}
+	}
+}
+
 func initDecoderConfig(dc *mapstructure.DecoderConfig) {
 	dc.DecodeHook = mapstructure.ComposeDecodeHookFunc(func(from, to reflect.Type, v any) (any, error) {
 		switch to {
diff --git a/cmd/dfcache/cmd/root.go b/cmd/dfcache/cmd/root.go
index b6bd7c07..57c3036c 100644
--- a/cmd/dfcache/cmd/root.go
+++ b/cmd/dfcache/cmd/root.go
@@ -73,8 +73,8 @@ func Execute() {
 func init() {
 	// Initialize default dfcache config
 	dfcacheConfig = config.NewDfcacheConfig()
-	// Initialize cobra
-	dependency.InitCobra(rootCmd, false, dfcacheConfig)
+	// Initialize command and config
+	dependency.InitCommandAndConfig(rootCmd, false, dfcacheConfig)
 
 	// Bind more cache specific persistent flags
 	flags := rootCmd.PersistentFlags()
diff --git a/cmd/dfget/cmd/daemon.go b/cmd/dfget/cmd/daemon.go
index d7adab58..836bfe91 100644
--- a/cmd/dfget/cmd/daemon.go
+++ b/cmd/dfget/cmd/daemon.go
@@ -89,7 +89,7 @@ func init() {
 		// Initialize default daemon config
 		cfg = config.NewDaemonConfig()
 		// Initialize cobra
-		dependency.InitCobra(daemonCmd, true, cfg)
+		dependency.InitCommandAndConfig(daemonCmd, true, cfg)
 
 		flags := daemonCmd.Flags()
 		flags.Int("launcher", -1, "pid of process launching daemon, a negative number implies that the daemon is started directly by the user")
@@ -104,17 +104,21 @@ func initDaemonDfpath(cfg *config.DaemonOption) (dfpath.Dfpath, error) {
 		options = append(options, dfpath.WithWorkHome(cfg.WorkHome))
 	}
 
-	if cfg.CacheDir != "" {
-		options = append(options, dfpath.WithCacheDir(cfg.CacheDir))
-	}
-
 	if cfg.LogDir != "" {
 		options = append(options, dfpath.WithLogDir(cfg.LogDir))
 	}
 
+	cacheDir := dfpath.DefaultCacheDir
+	if cfg.CacheDir != "" {
+		cacheDir = cfg.CacheDir
+	}
+	options = append(options, dfpath.WithCacheDir(cacheDir))
+
+	dataDir := dfpath.DefaultDataDir
 	if cfg.DataDir != "" {
-		options = append(options, dfpath.WithDataDir(cfg.DataDir))
+		dataDir = cfg.DataDir
 	}
+	options = append(options, dfpath.WithDataDir(dataDir))
 
 	return dfpath.New(options...)
 }
diff --git a/cmd/dfget/cmd/root.go b/cmd/dfget/cmd/root.go
index 3313e2fc..e062910d 100644
--- a/cmd/dfget/cmd/root.go
+++ b/cmd/dfget/cmd/root.go
@@ -126,8 +126,8 @@ func Execute() {
 func init() {
 	// Initialize default dfget config
 	dfgetConfig = config.NewDfgetConfig()
-	// Initialize cobra
-	dependency.InitCobra(rootCmd, false, dfgetConfig)
+	// Initialize command and config
+	dependency.InitCommandAndConfig(rootCmd, false, dfgetConfig)
 
 	// Add flags
 	flagSet := rootCmd.Flags()
diff --git a/cmd/manager/cmd/root.go b/cmd/manager/cmd/root.go
index 9a964f4d..5aced092 100644
--- a/cmd/manager/cmd/root.go
+++ b/cmd/manager/cmd/root.go
@@ -80,8 +80,8 @@ func init() {
 	// Initialize default manager config
 	cfg = config.New()
 
-	// Initialize cobra
-	dependency.InitCobra(rootCmd, true, cfg)
+	// Initialize command and config
+	dependency.InitCommandAndConfig(rootCmd, true, cfg)
 }
 
 func initDfpath(cfg *config.ServerConfig) (dfpath.Dfpath, error) {
diff --git a/cmd/scheduler/cmd/root.go b/cmd/scheduler/cmd/root.go
index b6d90b82..452cd22b 100644
--- a/cmd/scheduler/cmd/root.go
+++ b/cmd/scheduler/cmd/root.go
@@ -83,8 +83,8 @@ func Execute() {
 func init() {
 	// Initialize default scheduler config
 	cfg = config.New()
-	// Initialize cobra
-	dependency.InitCobra(rootCmd, true, cfg)
+	// Initialize command and config
+	dependency.InitCommandAndConfig(rootCmd, true, cfg)
 }
 
 func initDfpath(cfg *config.ServerConfig) (dfpath.Dfpath, error) {
@@ -93,9 +93,11 @@ func initDfpath(cfg *config.ServerConfig) (dfpath.Dfpath, error) {
 		options = append(options, dfpath.WithLogDir(cfg.LogDir))
 	}
 
+	cacheDir := dfpath.DefaultCacheDir
 	if cfg.CacheDir != "" {
-		options = append(options, dfpath.WithCacheDir(cfg.CacheDir))
+		cacheDir = cfg.CacheDir
 	}
+	options = append(options, dfpath.WithCacheDir(cacheDir))
 
 	return dfpath.New(options...)
 }
diff --git a/go.mod b/go.mod
index 6e4c3de3..87e998c6 100644
--- a/go.mod
+++ b/go.mod
@@ -35,6 +35,7 @@ require (
 	github.com/google/uuid v1.3.0
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
+	github.com/hashicorp/go-multierror v1.1.1
 	github.com/jarcoal/httpmock v1.0.8
 	github.com/looplab/fsm v0.3.0
 	github.com/mcuadros/go-gin-prometheus v0.1.0
@@ -122,7 +123,6 @@ require (
 	github.com/googleapis/enterprise-certificate-proxy v0.1.0 // indirect
 	github.com/googleapis/gax-go/v2 v2.4.0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
-	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/go-uuid v1.0.3 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
diff --git a/manager/searcher/searcher.go b/manager/searcher/searcher.go
index d6deec19..828dff94 100644
--- a/manager/searcher/searcher.go
+++ b/manager/searcher/searcher.go
@@ -236,7 +236,7 @@ func calculateMultiElementAffinityScore(dst, src string) float64 {
 	var score, elementLen int
 	dstElements := strings.Split(dst, "|")
 	srcElements := strings.Split(src, "|")
-	elementLen = math.MaxInt(len(dstElements), len(srcElements))
+	elementLen = math.Min(len(dstElements), len(srcElements))
 
 	// Maximum element length is 5
 	if elementLen > maxElementLen {
diff --git a/pkg/dfpath/dfpath.go b/pkg/dfpath/dfpath.go
index a5548529..f3052ea1 100644
--- a/pkg/dfpath/dfpath.go
+++ b/pkg/dfpath/dfpath.go
@@ -19,14 +19,15 @@
 package dfpath
 
 import (
-	"fmt"
 	"io/fs"
 	"os"
 	"path/filepath"
 	"sync"
+
+	"github.com/hashicorp/go-multierror"
 )
 
-// Dfpath is the interface used for init project path
+// Dfpath is the interface used for init project path.
 type Dfpath interface {
 	WorkHome() string
 	CacheDir() string
@@ -38,7 +39,7 @@ type Dfpath interface {
 	DfgetLockPath() string
 }
 
-// Dfpath provides init project path function
+// Dfpath provides init project path function.
 type dfpath struct {
 	workHome       string
 	cacheDir       string
@@ -50,76 +51,103 @@ type dfpath struct {
 	dfgetLockPath  string
 }
 
-// Cache of the dfpath
+// Cache of the dfpath.
 var cache struct {
 	sync.Once
-	d    *dfpath
-	errs []error
+	d   *dfpath
+	err *multierror.Error
 }
 
-// Option is a functional option for configuring the dfpath
+// Option is a functional option for configuring the dfpath.
 type Option func(d *dfpath)
 
-// WithWorkHome set the workhome directory
+// WithWorkHome set the workhome directory.
 func WithWorkHome(dir string) Option {
 	return func(d *dfpath) {
 		d.workHome = dir
 	}
 }
 
-// WithCacheDir set the cache directory
+// WithCacheDir set the cache directory.
 func WithCacheDir(dir string) Option {
 	return func(d *dfpath) {
 		d.cacheDir = dir
 	}
 }
 
-// WithLogDir set the log directory
+// WithLogDir set the log directory.
 func WithLogDir(dir string) Option {
 	return func(d *dfpath) {
 		d.logDir = dir
 	}
 }
 
-// WithDataDir set download data directory
+// WithDataDir set download data directory.
 func WithDataDir(dir string) Option {
 	return func(d *dfpath) {
 		d.dataDir = dir
 	}
 }
 
-// New returns a new dfpath interface
+// WithPluginDir set plugin directory.
+func WithPluginDir(dir string) Option {
+	return func(d *dfpath) {
+		d.pluginDir = dir
+	}
+}
+
+// New returns a new dfpath interface.
 func New(options ...Option) (Dfpath, error) {
 	cache.Do(func() {
 		d := &dfpath{
-			workHome: DefaultWorkHome,
-			cacheDir: DefaultCacheDir,
-			logDir:   DefaultLogDir,
-			dataDir:  DefaultDataDir,
+			workHome:  DefaultWorkHome,
+			logDir:    DefaultLogDir,
+			pluginDir: DefaultPluginDir,
 		}
 
 		for _, opt := range options {
 			opt(d)
 		}
 
-		d.pluginDir = filepath.Join(d.workHome, "plugins")
+		// Initialize dfdaemon path.
 		d.daemonSockPath = filepath.Join(d.workHome, "daemon.sock")
 		d.daemonLockPath = filepath.Join(d.workHome, "daemon.lock")
 		d.dfgetLockPath = filepath.Join(d.workHome, "dfget.lock")
 
-		// Create directories
-		for name, dir := range map[string]string{"workHome": d.workHome, "cacheDir": d.cacheDir, "logDir": d.logDir, "dataDir": d.dataDir,
-			"pluginDir": d.pluginDir} {
-			if err := os.MkdirAll(dir, fs.FileMode(0755)); err != nil {
-				cache.errs = append(cache.errs, fmt.Errorf("create %s dir %s failed: %v", name, dir, err))
+		// Create workhome directory.
+		if err := os.MkdirAll(d.workHome, fs.FileMode(0755)); err != nil {
+			cache.err = multierror.Append(cache.err, err)
+		}
+
+		// Create log directory.
+		if err := os.MkdirAll(d.logDir, fs.FileMode(0755)); err != nil {
+			cache.err = multierror.Append(cache.err, err)
+		}
+
+		// Create plugin directory.
+		if err := os.MkdirAll(d.pluginDir, fs.FileMode(0755)); err != nil {
+			cache.err = multierror.Append(cache.err, err)
+		}
+
+		// Create cache directory.
+		if d.cacheDir != "" {
+			if err := os.MkdirAll(d.cacheDir, fs.FileMode(0755)); err != nil {
+				cache.err = multierror.Append(cache.err, err)
+			}
+		}
+
+		// Create data directory.
+		if d.dataDir != "" {
+			if err := os.MkdirAll(d.dataDir, fs.FileMode(0755)); err != nil {
+				cache.err = multierror.Append(cache.err, err)
 			}
 		}
 
 		cache.d = d
 	})
 
-	if len(cache.errs) > 0 {
-		return nil, fmt.Errorf("create dfpath failed: %s", cache.errs)
+	if cache.err.ErrorOrNil() != nil {
+		return nil, cache.err
 	}
 
 	d := *cache.d
diff --git a/pkg/dfpath/dfpath_darwin.go b/pkg/dfpath/dfpath_darwin.go
index 41590a72..1e52f44f 100644
--- a/pkg/dfpath/dfpath_darwin.go
+++ b/pkg/dfpath/dfpath_darwin.go
@@ -30,3 +30,4 @@ var DefaultCacheDir = filepath.Join(DefaultWorkHome, "cache")
 var DefaultConfigDir = filepath.Join(DefaultWorkHome, "config")
 var DefaultLogDir = filepath.Join(DefaultWorkHome, "logs")
 var DefaultDataDir = filepath.Join(DefaultWorkHome, "data")
+var DefaultPluginDir = filepath.Join(DefaultWorkHome, "plugins")
diff --git a/pkg/dfpath/dfpath_linux.go b/pkg/dfpath/dfpath_linux.go
index 589f72fc..62350489 100644
--- a/pkg/dfpath/dfpath_linux.go
+++ b/pkg/dfpath/dfpath_linux.go
@@ -24,3 +24,4 @@ var DefaultCacheDir = "/var/cache/dragonfly"
 var DefaultConfigDir = "/etc/dragonfly"
 var DefaultLogDir = "/var/log/dragonfly"
 var DefaultDataDir = "/var/lib/dragonfly"
+var DefaultPluginDir = "/usr/local/dragonfly/plugins"
diff --git a/pkg/math/math.go b/pkg/math/math.go
index c3c211a5..f78543e8 100644
--- a/pkg/math/math.go
+++ b/pkg/math/math.go
@@ -16,50 +16,28 @@
 
 package math
 
-func MaxInt(a, b int) int {
-	if a < b {
-		return a
+import "golang.org/x/exp/constraints"
+
+// Max returns the maximum of values.
+func Max[T constraints.Ordered](values ...T) T {
+	max := values[0]
+	for _, value := range values {
+		if value > max {
+			max = value
+		}
 	}
 
-	return b
+	return max
 }
 
-func MaxInt32(a, b int32) int32 {
-	if a > b {
-		return a
+// Min returns the minimum of values.
+func Min[T constraints.Ordered](values ...T) T {
+	min := values[0]
+	for _, value := range values {
+		if value < min {
+			min = value
+		}
 	}
 
-	return b
-}
-
-func MaxInt64(a, b int64) int64 {
-	if a > b {
-		return a
-	}
-
-	return b
-}
-
-func MinInt(a, b int) int {
-	if a < b {
-		return a
-	}
-
-	return b
-}
-
-func MinInt32(a, b int32) int32 {
-	if a < b {
-		return a
-	}
-
-	return b
-}
-
-func MinInt64(a, b int64) int64 {
-	if a < b {
-		return a
-	}
-
-	return b
+	return min
 }
diff --git a/pkg/rpc/base/base.pb.go b/pkg/rpc/base/base.pb.go
index 23900ffa..b8d5937a 100644
--- a/pkg/rpc/base/base.pb.go
+++ b/pkg/rpc/base/base.pb.go
@@ -46,11 +46,12 @@ const (
 	Code_ServerUnavailable Code = 500
 	// common response error 1000-1999
 	// client can be migrated to another scheduler/CDN
-	Code_ResourceLacked   Code = 1000
-	Code_BadRequest       Code = 1400
-	Code_PeerTaskNotFound Code = 1404
-	Code_UnknownError     Code = 1500
-	Code_RequestTimeOut   Code = 1504
+	Code_ResourceLacked      Code = 1000
+	Code_BackToSourceAborted Code = 1001
+	Code_BadRequest          Code = 1400
+	Code_PeerTaskNotFound    Code = 1404
+	Code_UnknownError        Code = 1500
+	Code_RequestTimeOut      Code = 1504
 	// client response error 4000-4999
 	Code_ClientError             Code = 4000
 	Code_ClientPieceRequestFail  Code = 4001 // get piece task from other peer error
@@ -83,6 +84,7 @@ var (
 		200:  "Success",
 		500:  "ServerUnavailable",
 		1000: "ResourceLacked",
+		1001: "BackToSourceAborted",
 		1400: "BadRequest",
 		1404: "PeerTaskNotFound",
 		1500: "UnknownError",
@@ -112,6 +114,7 @@ var (
 		"Success":                        200,
 		"ServerUnavailable":              500,
 		"ResourceLacked":                 1000,
+		"BackToSourceAborted":            1001,
 		"BadRequest":                     1400,
 		"PeerTaskNotFound":               1404,
 		"UnknownError":                   1500,
@@ -1036,61 +1039,63 @@ var file_pkg_rpc_base_base_proto_rawDesc = []byte{
 	0x75, 0x74, 0x65, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x62, 0x61, 0x73, 0x65,
 	0x2e, 0x45, 0x78, 0x74, 0x65, 0x6e, 0x64, 0x41, 0x74, 0x74, 0x72, 0x69, 0x62, 0x75, 0x74, 0x65,
 	0x52, 0x0f, 0x65, 0x78, 0x74, 0x65, 0x6e, 0x64, 0x41, 0x74, 0x74, 0x72, 0x69, 0x62, 0x75, 0x74,
-	0x65, 0x2a, 0x94, 0x05, 0x0a, 0x04, 0x43, 0x6f, 0x64, 0x65, 0x12, 0x11, 0x0a, 0x0d, 0x58, 0x5f,
+	0x65, 0x2a, 0xae, 0x05, 0x0a, 0x04, 0x43, 0x6f, 0x64, 0x65, 0x12, 0x11, 0x0a, 0x0d, 0x58, 0x5f,
 	0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x0c, 0x0a,
 	0x07, 0x53, 0x75, 0x63, 0x63, 0x65, 0x73, 0x73, 0x10, 0xc8, 0x01, 0x12, 0x16, 0x0a, 0x11, 0x53,
 	0x65, 0x72, 0x76, 0x65, 0x72, 0x55, 0x6e, 0x61, 0x76, 0x61, 0x69, 0x6c, 0x61, 0x62, 0x6c, 0x65,
 	0x10, 0xf4, 0x03, 0x12, 0x13, 0x0a, 0x0e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x4c,
-	0x61, 0x63, 0x6b, 0x65, 0x64, 0x10, 0xe8, 0x07, 0x12, 0x0f, 0x0a, 0x0a, 0x42, 0x61, 0x64, 0x52,
-	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x10, 0xf8, 0x0a, 0x12, 0x15, 0x0a, 0x10, 0x50, 0x65, 0x65,
-	0x72, 0x54, 0x61, 0x73, 0x6b, 0x4e, 0x6f, 0x74, 0x46, 0x6f, 0x75, 0x6e, 0x64, 0x10, 0xfc, 0x0a,
-	0x12, 0x11, 0x0a, 0x0c, 0x55, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x45, 0x72, 0x72, 0x6f, 0x72,
-	0x10, 0xdc, 0x0b, 0x12, 0x13, 0x0a, 0x0e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x69,
-	0x6d, 0x65, 0x4f, 0x75, 0x74, 0x10, 0xe0, 0x0b, 0x12, 0x10, 0x0a, 0x0b, 0x43, 0x6c, 0x69, 0x65,
-	0x6e, 0x74, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0xa0, 0x1f, 0x12, 0x1b, 0x0a, 0x16, 0x43, 0x6c,
-	0x69, 0x65, 0x6e, 0x74, 0x50, 0x69, 0x65, 0x63, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
-	0x46, 0x61, 0x69, 0x6c, 0x10, 0xa1, 0x1f, 0x12, 0x1a, 0x0a, 0x15, 0x43, 0x6c, 0x69, 0x65, 0x6e,
-	0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74,
-	0x10, 0xa2, 0x1f, 0x12, 0x1a, 0x0a, 0x15, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x43, 0x6f, 0x6e,
-	0x74, 0x65, 0x78, 0x74, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x65, 0x64, 0x10, 0xa3, 0x1f, 0x12,
-	0x19, 0x0a, 0x14, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x57, 0x61, 0x69, 0x74, 0x50, 0x69, 0x65,
-	0x63, 0x65, 0x52, 0x65, 0x61, 0x64, 0x79, 0x10, 0xa4, 0x1f, 0x12, 0x1c, 0x0a, 0x17, 0x43, 0x6c,
-	0x69, 0x65, 0x6e, 0x74, 0x50, 0x69, 0x65, 0x63, 0x65, 0x44, 0x6f, 0x77, 0x6e, 0x6c, 0x6f, 0x61,
-	0x64, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xa5, 0x1f, 0x12, 0x1b, 0x0a, 0x16, 0x43, 0x6c, 0x69, 0x65,
-	0x6e, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x46, 0x61,
-	0x69, 0x6c, 0x10, 0xa6, 0x1f, 0x12, 0x1a, 0x0a, 0x15, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x43,
-	0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0xa7,
-	0x1f, 0x12, 0x1a, 0x0a, 0x15, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x42, 0x61, 0x63, 0x6b, 0x53,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0xa8, 0x1f, 0x12, 0x18, 0x0a,
-	0x13, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x50, 0x69, 0x65, 0x63, 0x65, 0x4e, 0x6f, 0x74, 0x46,
-	0x6f, 0x75, 0x6e, 0x64, 0x10, 0xb4, 0x22, 0x12, 0x0f, 0x0a, 0x0a, 0x53, 0x63, 0x68, 0x65, 0x64,
-	0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0x88, 0x27, 0x12, 0x18, 0x0a, 0x13, 0x53, 0x63, 0x68, 0x65,
-	0x64, 0x4e, 0x65, 0x65, 0x64, 0x42, 0x61, 0x63, 0x6b, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x10,
-	0x89, 0x27, 0x12, 0x12, 0x0a, 0x0d, 0x53, 0x63, 0x68, 0x65, 0x64, 0x50, 0x65, 0x65, 0x72, 0x47,
-	0x6f, 0x6e, 0x65, 0x10, 0x8a, 0x27, 0x12, 0x16, 0x0a, 0x11, 0x53, 0x63, 0x68, 0x65, 0x64, 0x50,
-	0x65, 0x65, 0x72, 0x4e, 0x6f, 0x74, 0x46, 0x6f, 0x75, 0x6e, 0x64, 0x10, 0x8c, 0x27, 0x12, 0x23,
-	0x0a, 0x1e, 0x53, 0x63, 0x68, 0x65, 0x64, 0x50, 0x65, 0x65, 0x72, 0x50, 0x69, 0x65, 0x63, 0x65,
-	0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x52, 0x65, 0x70, 0x6f, 0x72, 0x74, 0x46, 0x61, 0x69, 0x6c,
-	0x10, 0x8d, 0x27, 0x12, 0x19, 0x0a, 0x14, 0x53, 0x63, 0x68, 0x65, 0x64, 0x54, 0x61, 0x73, 0x6b,
-	0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0x8e, 0x27, 0x12, 0x18,
-	0x0a, 0x13, 0x43, 0x44, 0x4e, 0x54, 0x61, 0x73, 0x6b, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x72,
-	0x79, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xf1, 0x2e, 0x12, 0x14, 0x0a, 0x0f, 0x43, 0x44, 0x4e, 0x54,
-	0x61, 0x73, 0x6b, 0x4e, 0x6f, 0x74, 0x46, 0x6f, 0x75, 0x6e, 0x64, 0x10, 0x84, 0x32, 0x12, 0x18,
-	0x0a, 0x13, 0x49, 0x6e, 0x76, 0x61, 0x6c, 0x69, 0x64, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
-	0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xd9, 0x36, 0x2a, 0x17, 0x0a, 0x0a, 0x50, 0x69, 0x65, 0x63,
-	0x65, 0x53, 0x74, 0x79, 0x6c, 0x65, 0x12, 0x09, 0x0a, 0x05, 0x50, 0x4c, 0x41, 0x49, 0x4e, 0x10,
-	0x00, 0x2a, 0x2c, 0x0a, 0x09, 0x53, 0x69, 0x7a, 0x65, 0x53, 0x63, 0x6f, 0x70, 0x65, 0x12, 0x0a,
-	0x0a, 0x06, 0x4e, 0x4f, 0x52, 0x4d, 0x41, 0x4c, 0x10, 0x00, 0x12, 0x09, 0x0a, 0x05, 0x53, 0x4d,
-	0x41, 0x4c, 0x4c, 0x10, 0x01, 0x12, 0x08, 0x0a, 0x04, 0x54, 0x49, 0x4e, 0x59, 0x10, 0x02, 0x2a,
-	0x2d, 0x0a, 0x07, 0x50, 0x61, 0x74, 0x74, 0x65, 0x72, 0x6e, 0x12, 0x07, 0x0a, 0x03, 0x50, 0x32,
-	0x50, 0x10, 0x00, 0x12, 0x0d, 0x0a, 0x09, 0x53, 0x45, 0x45, 0x44, 0x5f, 0x50, 0x45, 0x45, 0x52,
-	0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06, 0x53, 0x4f, 0x55, 0x52, 0x43, 0x45, 0x10, 0x02, 0x2a, 0x30,
-	0x0a, 0x08, 0x54, 0x61, 0x73, 0x6b, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x4e, 0x6f,
-	0x72, 0x6d, 0x61, 0x6c, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x44, 0x66, 0x43, 0x61, 0x63, 0x68,
-	0x65, 0x10, 0x01, 0x12, 0x0b, 0x0a, 0x07, 0x44, 0x66, 0x53, 0x74, 0x6f, 0x72, 0x65, 0x10, 0x02,
-	0x42, 0x22, 0x5a, 0x20, 0x64, 0x37, 0x79, 0x2e, 0x69, 0x6f, 0x2f, 0x64, 0x72, 0x61, 0x67, 0x6f,
-	0x6e, 0x66, 0x6c, 0x79, 0x2f, 0x76, 0x32, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x72, 0x70, 0x63, 0x2f,
-	0x62, 0x61, 0x73, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x61, 0x63, 0x6b, 0x65, 0x64, 0x10, 0xe8, 0x07, 0x12, 0x18, 0x0a, 0x13, 0x42, 0x61, 0x63, 0x6b,
+	0x54, 0x6f, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x41, 0x62, 0x6f, 0x72, 0x74, 0x65, 0x64, 0x10,
+	0xe9, 0x07, 0x12, 0x0f, 0x0a, 0x0a, 0x42, 0x61, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x10, 0xf8, 0x0a, 0x12, 0x15, 0x0a, 0x10, 0x50, 0x65, 0x65, 0x72, 0x54, 0x61, 0x73, 0x6b, 0x4e,
+	0x6f, 0x74, 0x46, 0x6f, 0x75, 0x6e, 0x64, 0x10, 0xfc, 0x0a, 0x12, 0x11, 0x0a, 0x0c, 0x55, 0x6e,
+	0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0xdc, 0x0b, 0x12, 0x13, 0x0a,
+	0x0e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x4f, 0x75, 0x74, 0x10,
+	0xe0, 0x0b, 0x12, 0x10, 0x0a, 0x0b, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x45, 0x72, 0x72, 0x6f,
+	0x72, 0x10, 0xa0, 0x1f, 0x12, 0x1b, 0x0a, 0x16, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x50, 0x69,
+	0x65, 0x63, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xa1,
+	0x1f, 0x12, 0x1a, 0x0a, 0x15, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64,
+	0x75, 0x6c, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x10, 0xa2, 0x1f, 0x12, 0x1a, 0x0a,
+	0x15, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x43, 0x61,
+	0x6e, 0x63, 0x65, 0x6c, 0x65, 0x64, 0x10, 0xa3, 0x1f, 0x12, 0x19, 0x0a, 0x14, 0x43, 0x6c, 0x69,
+	0x65, 0x6e, 0x74, 0x57, 0x61, 0x69, 0x74, 0x50, 0x69, 0x65, 0x63, 0x65, 0x52, 0x65, 0x61, 0x64,
+	0x79, 0x10, 0xa4, 0x1f, 0x12, 0x1c, 0x0a, 0x17, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x50, 0x69,
+	0x65, 0x63, 0x65, 0x44, 0x6f, 0x77, 0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x46, 0x61, 0x69, 0x6c, 0x10,
+	0xa5, 0x1f, 0x12, 0x1b, 0x0a, 0x16, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x71, 0x75,
+	0x65, 0x73, 0x74, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xa6, 0x1f, 0x12,
+	0x1a, 0x0a, 0x15, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
+	0x69, 0x6f, 0x6e, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0xa7, 0x1f, 0x12, 0x1a, 0x0a, 0x15, 0x43,
+	0x6c, 0x69, 0x65, 0x6e, 0x74, 0x42, 0x61, 0x63, 0x6b, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x45,
+	0x72, 0x72, 0x6f, 0x72, 0x10, 0xa8, 0x1f, 0x12, 0x18, 0x0a, 0x13, 0x43, 0x6c, 0x69, 0x65, 0x6e,
+	0x74, 0x50, 0x69, 0x65, 0x63, 0x65, 0x4e, 0x6f, 0x74, 0x46, 0x6f, 0x75, 0x6e, 0x64, 0x10, 0xb4,
+	0x22, 0x12, 0x0f, 0x0a, 0x0a, 0x53, 0x63, 0x68, 0x65, 0x64, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x10,
+	0x88, 0x27, 0x12, 0x18, 0x0a, 0x13, 0x53, 0x63, 0x68, 0x65, 0x64, 0x4e, 0x65, 0x65, 0x64, 0x42,
+	0x61, 0x63, 0x6b, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x10, 0x89, 0x27, 0x12, 0x12, 0x0a, 0x0d,
+	0x53, 0x63, 0x68, 0x65, 0x64, 0x50, 0x65, 0x65, 0x72, 0x47, 0x6f, 0x6e, 0x65, 0x10, 0x8a, 0x27,
+	0x12, 0x16, 0x0a, 0x11, 0x53, 0x63, 0x68, 0x65, 0x64, 0x50, 0x65, 0x65, 0x72, 0x4e, 0x6f, 0x74,
+	0x46, 0x6f, 0x75, 0x6e, 0x64, 0x10, 0x8c, 0x27, 0x12, 0x23, 0x0a, 0x1e, 0x53, 0x63, 0x68, 0x65,
+	0x64, 0x50, 0x65, 0x65, 0x72, 0x50, 0x69, 0x65, 0x63, 0x65, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74,
+	0x52, 0x65, 0x70, 0x6f, 0x72, 0x74, 0x46, 0x61, 0x69, 0x6c, 0x10, 0x8d, 0x27, 0x12, 0x19, 0x0a,
+	0x14, 0x53, 0x63, 0x68, 0x65, 0x64, 0x54, 0x61, 0x73, 0x6b, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73,
+	0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0x8e, 0x27, 0x12, 0x18, 0x0a, 0x13, 0x43, 0x44, 0x4e, 0x54,
+	0x61, 0x73, 0x6b, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x72, 0x79, 0x46, 0x61, 0x69, 0x6c, 0x10,
+	0xf1, 0x2e, 0x12, 0x14, 0x0a, 0x0f, 0x43, 0x44, 0x4e, 0x54, 0x61, 0x73, 0x6b, 0x4e, 0x6f, 0x74,
+	0x46, 0x6f, 0x75, 0x6e, 0x64, 0x10, 0x84, 0x32, 0x12, 0x18, 0x0a, 0x13, 0x49, 0x6e, 0x76, 0x61,
+	0x6c, 0x69, 0x64, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10,
+	0xd9, 0x36, 0x2a, 0x17, 0x0a, 0x0a, 0x50, 0x69, 0x65, 0x63, 0x65, 0x53, 0x74, 0x79, 0x6c, 0x65,
+	0x12, 0x09, 0x0a, 0x05, 0x50, 0x4c, 0x41, 0x49, 0x4e, 0x10, 0x00, 0x2a, 0x2c, 0x0a, 0x09, 0x53,
+	0x69, 0x7a, 0x65, 0x53, 0x63, 0x6f, 0x70, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x4e, 0x4f, 0x52, 0x4d,
+	0x41, 0x4c, 0x10, 0x00, 0x12, 0x09, 0x0a, 0x05, 0x53, 0x4d, 0x41, 0x4c, 0x4c, 0x10, 0x01, 0x12,
+	0x08, 0x0a, 0x04, 0x54, 0x49, 0x4e, 0x59, 0x10, 0x02, 0x2a, 0x2d, 0x0a, 0x07, 0x50, 0x61, 0x74,
+	0x74, 0x65, 0x72, 0x6e, 0x12, 0x07, 0x0a, 0x03, 0x50, 0x32, 0x50, 0x10, 0x00, 0x12, 0x0d, 0x0a,
+	0x09, 0x53, 0x45, 0x45, 0x44, 0x5f, 0x50, 0x45, 0x45, 0x52, 0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06,
+	0x53, 0x4f, 0x55, 0x52, 0x43, 0x45, 0x10, 0x02, 0x2a, 0x30, 0x0a, 0x08, 0x54, 0x61, 0x73, 0x6b,
+	0x54, 0x79, 0x70, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x4e, 0x6f, 0x72, 0x6d, 0x61, 0x6c, 0x10, 0x00,
+	0x12, 0x0b, 0x0a, 0x07, 0x44, 0x66, 0x43, 0x61, 0x63, 0x68, 0x65, 0x10, 0x01, 0x12, 0x0b, 0x0a,
+	0x07, 0x44, 0x66, 0x53, 0x74, 0x6f, 0x72, 0x65, 0x10, 0x02, 0x42, 0x22, 0x5a, 0x20, 0x64, 0x37,
+	0x79, 0x2e, 0x69, 0x6f, 0x2f, 0x64, 0x72, 0x61, 0x67, 0x6f, 0x6e, 0x66, 0x6c, 0x79, 0x2f, 0x76,
+	0x32, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x72, 0x70, 0x63, 0x2f, 0x62, 0x61, 0x73, 0x65, 0x62, 0x06,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
diff --git a/pkg/rpc/base/base.proto b/pkg/rpc/base/base.proto
index b9a78bda..e0f39f8f 100644
--- a/pkg/rpc/base/base.proto
+++ b/pkg/rpc/base/base.proto
@@ -32,6 +32,7 @@ enum Code{
   // common response error 1000-1999
   // client can be migrated to another scheduler/CDN
   ResourceLacked = 1000;
+  BackToSourceAborted = 1001;
   BadRequest = 1400;
   PeerTaskNotFound = 1404;
   UnknownError = 1500;
diff --git a/pkg/rpc/scheduler/scheduler.pb.go b/pkg/rpc/scheduler/scheduler.pb.go
index 95e6efbb..1b8131ab 100644
--- a/pkg/rpc/scheduler/scheduler.pb.go
+++ b/pkg/rpc/scheduler/scheduler.pb.go
@@ -623,6 +623,11 @@ type PeerPacket struct {
 	StealPeers []*PeerPacket_DestPeer `protobuf:"bytes,6,rep,name=steal_peers,json=stealPeers,proto3" json:"steal_peers,omitempty"`
 	// Result code.
 	Code base.Code `protobuf:"varint,7,opt,name=code,proto3,enum=base.Code" json:"code,omitempty"`
+	// Error detail.
+	//
+	// Types that are assignable to ErrorDetail:
+	//	*PeerPacket_SourceError
+	ErrorDetail isPeerPacket_ErrorDetail `protobuf_oneof:"error_detail"`
 }
 
 func (x *PeerPacket) Reset() {
@@ -699,6 +704,31 @@ func (x *PeerPacket) GetCode() base.Code {
 	return base.Code(0)
 }
 
+func (m *PeerPacket) GetErrorDetail() isPeerPacket_ErrorDetail {
+	if m != nil {
+		return m.ErrorDetail
+	}
+	return nil
+}
+
+func (x *PeerPacket) GetSourceError() *errordetails.SourceError {
+	if x, ok := x.GetErrorDetail().(*PeerPacket_SourceError); ok {
+		return x.SourceError
+	}
+	return nil
+}
+
+type isPeerPacket_ErrorDetail interface {
+	isPeerPacket_ErrorDetail()
+}
+
+type PeerPacket_SourceError struct {
+	// Source error.
+	SourceError *errordetails.SourceError `protobuf:"bytes,8,opt,name=source_error,json=sourceError,proto3,oneof"`
+}
+
+func (*PeerPacket_SourceError) isPeerPacket_ErrorDetail() {}
+
 // PeerResult represents response of ReportPeerResult.
 type PeerResult struct {
 	state         protoimpl.MessageState
@@ -729,6 +759,8 @@ type PeerResult struct {
 	Code base.Code `protobuf:"varint,11,opt,name=code,proto3,enum=base.Code" json:"code,omitempty"`
 	// Task total piece count.
 	TotalPieceCount int32 `protobuf:"varint,12,opt,name=total_piece_count,json=totalPieceCount,proto3" json:"total_piece_count,omitempty"`
+	// Error detail.
+	//
 	// Types that are assignable to ErrorDetail:
 	//	*PeerResult_SourceError
 	ErrorDetail isPeerResult_ErrorDetail `protobuf_oneof:"error_detail"`
@@ -1354,7 +1386,7 @@ var file_pkg_rpc_scheduler_scheduler_proto_rawDesc = []byte{
 	0x74, 0x65, 0x6e, 0x64, 0x5f, 0x61, 0x74, 0x74, 0x72, 0x69, 0x62, 0x75, 0x74, 0x65, 0x18, 0x0b,
 	0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x62, 0x61, 0x73, 0x65, 0x2e, 0x45, 0x78, 0x74, 0x65,
 	0x6e, 0x64, 0x41, 0x74, 0x74, 0x72, 0x69, 0x62, 0x75, 0x74, 0x65, 0x52, 0x0f, 0x65, 0x78, 0x74,
-	0x65, 0x6e, 0x64, 0x41, 0x74, 0x74, 0x72, 0x69, 0x62, 0x75, 0x74, 0x65, 0x22, 0x8e, 0x03, 0x0a,
+	0x65, 0x6e, 0x64, 0x41, 0x74, 0x74, 0x72, 0x69, 0x62, 0x75, 0x74, 0x65, 0x22, 0xde, 0x03, 0x0a,
 	0x0a, 0x50, 0x65, 0x65, 0x72, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x12, 0x20, 0x0a, 0x07, 0x74,
 	0x61, 0x73, 0x6b, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42,
 	0x04, 0x72, 0x02, 0x10, 0x01, 0x52, 0x06, 0x74, 0x61, 0x73, 0x6b, 0x49, 0x64, 0x12, 0x20, 0x0a,
@@ -1372,14 +1404,19 @@ var file_pkg_rpc_scheduler_scheduler_proto_rawDesc = []byte{
 	0x65, 0x72, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x2e, 0x44, 0x65, 0x73, 0x74, 0x50, 0x65, 0x65,
 	0x72, 0x52, 0x0a, 0x73, 0x74, 0x65, 0x61, 0x6c, 0x50, 0x65, 0x65, 0x72, 0x73, 0x12, 0x1e, 0x0a,
 	0x04, 0x63, 0x6f, 0x64, 0x65, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0a, 0x2e, 0x62, 0x61,
-	0x73, 0x65, 0x2e, 0x43, 0x6f, 0x64, 0x65, 0x52, 0x04, 0x63, 0x6f, 0x64, 0x65, 0x1a, 0x6e, 0x0a,
+	0x73, 0x65, 0x2e, 0x43, 0x6f, 0x64, 0x65, 0x52, 0x04, 0x63, 0x6f, 0x64, 0x65, 0x12, 0x3e, 0x0a,
+	0x0c, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x08, 0x20,
+	0x01, 0x28, 0x0b, 0x32, 0x19, 0x2e, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x64, 0x65, 0x74, 0x61, 0x69,
+	0x6c, 0x73, 0x2e, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x48, 0x00,
+	0x52, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x1a, 0x6e, 0x0a,
 	0x08, 0x44, 0x65, 0x73, 0x74, 0x50, 0x65, 0x65, 0x72, 0x12, 0x17, 0x0a, 0x02, 0x69, 0x70, 0x18,
 	0x01, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x70, 0x01, 0x52, 0x02,
 	0x69, 0x70, 0x12, 0x27, 0x0a, 0x08, 0x72, 0x70, 0x63, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x02,
 	0x20, 0x01, 0x28, 0x05, 0x42, 0x0c, 0xfa, 0x42, 0x09, 0x1a, 0x07, 0x10, 0xff, 0xff, 0x03, 0x28,
 	0x80, 0x08, 0x52, 0x07, 0x72, 0x70, 0x63, 0x50, 0x6f, 0x72, 0x74, 0x12, 0x20, 0x0a, 0x07, 0x70,
 	0x65, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42,
-	0x04, 0x72, 0x02, 0x10, 0x01, 0x52, 0x06, 0x70, 0x65, 0x65, 0x72, 0x49, 0x64, 0x22, 0xf6, 0x03,
+	0x04, 0x72, 0x02, 0x10, 0x01, 0x52, 0x06, 0x70, 0x65, 0x65, 0x72, 0x49, 0x64, 0x42, 0x0e, 0x0a,
+	0x0c, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x5f, 0x64, 0x65, 0x74, 0x61, 0x69, 0x6c, 0x22, 0xf6, 0x03,
 	0x0a, 0x0a, 0x50, 0x65, 0x65, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x20, 0x0a, 0x07,
 	0x74, 0x61, 0x73, 0x6b, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa,
 	0x42, 0x04, 0x72, 0x02, 0x10, 0x01, 0x52, 0x06, 0x74, 0x61, 0x73, 0x6b, 0x49, 0x64, 0x12, 0x20,
@@ -1543,30 +1580,31 @@ var file_pkg_rpc_scheduler_scheduler_proto_depIdxs = []int32{
 	11, // 13: scheduler.PeerPacket.main_peer:type_name -> scheduler.PeerPacket.DestPeer
 	11, // 14: scheduler.PeerPacket.steal_peers:type_name -> scheduler.PeerPacket.DestPeer
 	19, // 15: scheduler.PeerPacket.code:type_name -> base.Code
-	19, // 16: scheduler.PeerResult.code:type_name -> base.Code
-	20, // 17: scheduler.PeerResult.source_error:type_name -> errordetails.SourceError
-	15, // 18: scheduler.Task.type:type_name -> base.TaskType
-	12, // 19: scheduler.AnnounceTaskRequest.url_meta:type_name -> base.UrlMeta
-	3,  // 20: scheduler.AnnounceTaskRequest.peer_host:type_name -> scheduler.PeerHost
-	21, // 21: scheduler.AnnounceTaskRequest.piece_packet:type_name -> base.PiecePacket
-	15, // 22: scheduler.AnnounceTaskRequest.task_type:type_name -> base.TaskType
-	0,  // 23: scheduler.Scheduler.RegisterPeerTask:input_type -> scheduler.PeerTaskRequest
-	4,  // 24: scheduler.Scheduler.ReportPieceResult:input_type -> scheduler.PieceResult
-	6,  // 25: scheduler.Scheduler.ReportPeerResult:input_type -> scheduler.PeerResult
-	7,  // 26: scheduler.Scheduler.LeaveTask:input_type -> scheduler.PeerTarget
-	8,  // 27: scheduler.Scheduler.StatTask:input_type -> scheduler.StatTaskRequest
-	10, // 28: scheduler.Scheduler.AnnounceTask:input_type -> scheduler.AnnounceTaskRequest
-	1,  // 29: scheduler.Scheduler.RegisterPeerTask:output_type -> scheduler.RegisterResult
-	5,  // 30: scheduler.Scheduler.ReportPieceResult:output_type -> scheduler.PeerPacket
-	22, // 31: scheduler.Scheduler.ReportPeerResult:output_type -> google.protobuf.Empty
-	22, // 32: scheduler.Scheduler.LeaveTask:output_type -> google.protobuf.Empty
-	9,  // 33: scheduler.Scheduler.StatTask:output_type -> scheduler.Task
-	22, // 34: scheduler.Scheduler.AnnounceTask:output_type -> google.protobuf.Empty
-	29, // [29:35] is the sub-list for method output_type
-	23, // [23:29] is the sub-list for method input_type
-	23, // [23:23] is the sub-list for extension type_name
-	23, // [23:23] is the sub-list for extension extendee
-	0,  // [0:23] is the sub-list for field type_name
+	20, // 16: scheduler.PeerPacket.source_error:type_name -> errordetails.SourceError
+	19, // 17: scheduler.PeerResult.code:type_name -> base.Code
+	20, // 18: scheduler.PeerResult.source_error:type_name -> errordetails.SourceError
+	15, // 19: scheduler.Task.type:type_name -> base.TaskType
+	12, // 20: scheduler.AnnounceTaskRequest.url_meta:type_name -> base.UrlMeta
+	3,  // 21: scheduler.AnnounceTaskRequest.peer_host:type_name -> scheduler.PeerHost
+	21, // 22: scheduler.AnnounceTaskRequest.piece_packet:type_name -> base.PiecePacket
+	15, // 23: scheduler.AnnounceTaskRequest.task_type:type_name -> base.TaskType
+	0,  // 24: scheduler.Scheduler.RegisterPeerTask:input_type -> scheduler.PeerTaskRequest
+	4,  // 25: scheduler.Scheduler.ReportPieceResult:input_type -> scheduler.PieceResult
+	6,  // 26: scheduler.Scheduler.ReportPeerResult:input_type -> scheduler.PeerResult
+	7,  // 27: scheduler.Scheduler.LeaveTask:input_type -> scheduler.PeerTarget
+	8,  // 28: scheduler.Scheduler.StatTask:input_type -> scheduler.StatTaskRequest
+	10, // 29: scheduler.Scheduler.AnnounceTask:input_type -> scheduler.AnnounceTaskRequest
+	1,  // 30: scheduler.Scheduler.RegisterPeerTask:output_type -> scheduler.RegisterResult
+	5,  // 31: scheduler.Scheduler.ReportPieceResult:output_type -> scheduler.PeerPacket
+	22, // 32: scheduler.Scheduler.ReportPeerResult:output_type -> google.protobuf.Empty
+	22, // 33: scheduler.Scheduler.LeaveTask:output_type -> google.protobuf.Empty
+	9,  // 34: scheduler.Scheduler.StatTask:output_type -> scheduler.Task
+	22, // 35: scheduler.Scheduler.AnnounceTask:output_type -> google.protobuf.Empty
+	30, // [30:36] is the sub-list for method output_type
+	24, // [24:30] is the sub-list for method input_type
+	24, // [24:24] is the sub-list for extension type_name
+	24, // [24:24] is the sub-list for extension extendee
+	0,  // [0:24] is the sub-list for field type_name
 }
 
 func init() { file_pkg_rpc_scheduler_scheduler_proto_init() }
@@ -1724,6 +1762,9 @@ func file_pkg_rpc_scheduler_scheduler_proto_init() {
 		(*RegisterResult_SinglePiece)(nil),
 		(*RegisterResult_PieceContent)(nil),
 	}
+	file_pkg_rpc_scheduler_scheduler_proto_msgTypes[5].OneofWrappers = []interface{}{
+		(*PeerPacket_SourceError)(nil),
+	}
 	file_pkg_rpc_scheduler_scheduler_proto_msgTypes[6].OneofWrappers = []interface{}{
 		(*PeerResult_SourceError)(nil),
 	}
diff --git a/pkg/rpc/scheduler/scheduler.pb.validate.go b/pkg/rpc/scheduler/scheduler.pb.validate.go
index b13ffd38..8987d1bb 100644
--- a/pkg/rpc/scheduler/scheduler.pb.validate.go
+++ b/pkg/rpc/scheduler/scheduler.pb.validate.go
@@ -697,6 +697,22 @@ func (m *PeerPacket) Validate() error {
 
 	// no validation rules for Code
 
+	switch m.ErrorDetail.(type) {
+
+	case *PeerPacket_SourceError:
+
+		if v, ok := interface{}(m.GetSourceError()).(interface{ Validate() error }); ok {
+			if err := v.Validate(); err != nil {
+				return PeerPacketValidationError{
+					field:  "SourceError",
+					reason: "embedded message failed validation",
+					cause:  err,
+				}
+			}
+		}
+
+	}
+
 	return nil
 }
 
diff --git a/pkg/rpc/scheduler/scheduler.proto b/pkg/rpc/scheduler/scheduler.proto
index 47631172..e7d064bc 100644
--- a/pkg/rpc/scheduler/scheduler.proto
+++ b/pkg/rpc/scheduler/scheduler.proto
@@ -147,6 +147,11 @@ message PeerPacket{
   repeated DestPeer steal_peers = 6;
   // Result code.
   base.Code code = 7;
+  // Error detail.
+  oneof error_detail{
+    // Source error.
+    errordetails.SourceError source_error = 8;
+  }
 }
 
 // PeerResult represents response of ReportPeerResult.
@@ -175,6 +180,7 @@ message PeerResult{
   base.Code code = 11;
   // Task total piece count.
   int32 total_piece_count = 12 [(validate.rules).int32.gte = -1];
+  // Error detail.
   oneof error_detail{
     // Source error.
     errordetails.SourceError source_error = 13;
diff --git a/scheduler/resource/task.go b/scheduler/resource/task.go
index 0511dab9..81c50a0d 100644
--- a/scheduler/resource/task.go
+++ b/scheduler/resource/task.go
@@ -325,7 +325,7 @@ func (t *Task) CanBackToSource() bool {
 }
 
 // NotifyPeers notify all peers in the task with the state code.
-func (t *Task) NotifyPeers(code base.Code, event string) {
+func (t *Task) NotifyPeers(peerPacket *rpcscheduler.PeerPacket, event string) {
 	t.Peers.Range(func(_, value any) bool {
 		peer := value.(*Peer)
 		if peer.FSM.Is(PeerStateRunning) {
@@ -334,11 +334,11 @@ func (t *Task) NotifyPeers(code base.Code, event string) {
 				return true
 			}
 
-			if err := stream.Send(&rpcscheduler.PeerPacket{Code: code}); err != nil {
+			if err := stream.Send(peerPacket); err != nil {
 				t.Log.Errorf("send packet to peer %s failed: %s", peer.ID, err.Error())
 				return true
 			}
-			t.Log.Infof("task notify peer %s code %s", peer.ID, code)
+			t.Log.Infof("task notify peer %s code %s", peer.ID, peerPacket.Code)
 
 			if err := peer.FSM.Event(event); err != nil {
 				peer.Log.Errorf("peer fsm event failed: %s", err.Error())
diff --git a/scheduler/scheduler/evaluator/evaluator_base.go b/scheduler/scheduler/evaluator/evaluator_base.go
index 61327b05..ed08364e 100644
--- a/scheduler/scheduler/evaluator/evaluator_base.go
+++ b/scheduler/scheduler/evaluator/evaluator_base.go
@@ -159,7 +159,7 @@ func calculateMultiElementAffinityScore(dst, src string) float64 {
 	var score, elementLen int
 	dstElements := strings.Split(dst, "|")
 	srcElements := strings.Split(src, "|")
-	elementLen = math.MaxInt(len(dstElements), len(srcElements))
+	elementLen = math.Min(len(dstElements), len(srcElements))
 
 	// Maximum element length is 5.
 	if elementLen > maxElementLen {
diff --git a/scheduler/service/service.go b/scheduler/service/service.go
index 49c21333..6c039f5c 100644
--- a/scheduler/service/service.go
+++ b/scheduler/service/service.go
@@ -23,12 +23,14 @@ import (
 	"time"
 
 	"go.opentelemetry.io/otel/trace"
+	"google.golang.org/grpc/status"
 
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/container/set"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/base/common"
+	"d7y.io/dragonfly/v2/pkg/rpc/errordetails"
 	rpcscheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	pkgtime "d7y.io/dragonfly/v2/pkg/time"
 	"d7y.io/dragonfly/v2/scheduler/config"
@@ -348,7 +350,7 @@ func (s *Service) ReportPeerResult(ctx context.Context, req *rpcscheduler.PeerRe
 			s.createRecord(peer, storage.PeerStateBackToSourceFailed, req)
 			metrics.DownloadFailureCount.WithLabelValues(peer.BizTag, metrics.DownloadFailureBackToSourceType).Inc()
 
-			s.handleTaskFail(ctx, peer.Task)
+			s.handleTaskFail(ctx, peer.Task, req.GetSourceError(), nil)
 			s.handlePeerFail(ctx, peer)
 			return nil
 		}
@@ -570,7 +572,7 @@ func (s *Service) triggerSeedPeerTask(ctx context.Context, task *resource.Task)
 		trace.ContextWithSpanContext(context.Background(), trace.SpanContextFromContext(ctx)), task)
 	if err != nil {
 		task.Log.Errorf("trigger seed peer download task failed: %s", err.Error())
-		s.handleTaskFail(ctx, task)
+		s.handleTaskFail(ctx, task, nil, err)
 		return
 	}
 
@@ -786,12 +788,47 @@ func (s *Service) handleTaskSuccess(ctx context.Context, task *resource.Task, re
 // Conditions for the task to switch to the TaskStateSucceeded are:
 // 1. Seed peer downloads the resource falied.
 // 2. Dfdaemon back-to-source to download failed.
-func (s *Service) handleTaskFail(ctx context.Context, task *resource.Task) {
-	// If the number of failed peers in the task is greater than FailedPeerCountLimit,
-	// then scheduler notifies running peers of failure.
-	if task.PeerFailedCount.Load() > resource.FailedPeerCountLimit {
+func (s *Service) handleTaskFail(ctx context.Context, task *resource.Task, backToSourceErr *errordetails.SourceError, seedPeerErr error) {
+	// If peer back-to-source fails due to an unrecoverable error,
+	// notify other peers of the failure,
+	// and return the source metadata to peer.
+	if backToSourceErr != nil {
+		if !backToSourceErr.Temporary {
+			task.NotifyPeers(&rpcscheduler.PeerPacket{
+				Code: base.Code_BackToSourceAborted,
+				ErrorDetail: &rpcscheduler.PeerPacket_SourceError{
+					SourceError: backToSourceErr,
+				},
+			}, resource.PeerEventDownloadFailed)
+			task.PeerFailedCount.Store(0)
+		}
+	} else if seedPeerErr != nil {
+		// If seed peer back-to-source fails due to an unrecoverable error,
+		// notify other peers of the failure,
+		// and return the source metadata to peer.
+		if st, ok := status.FromError(seedPeerErr); ok {
+			for _, detail := range st.Details() {
+				switch d := detail.(type) {
+				case *errordetails.SourceError:
+					if !d.Temporary {
+						task.NotifyPeers(&rpcscheduler.PeerPacket{
+							Code: base.Code_BackToSourceAborted,
+							ErrorDetail: &rpcscheduler.PeerPacket_SourceError{
+								SourceError: d,
+							},
+						}, resource.PeerEventDownloadFailed)
+						task.PeerFailedCount.Store(0)
+					}
+				}
+			}
+		}
+	} else if task.PeerFailedCount.Load() > resource.FailedPeerCountLimit {
+		// If the number of failed peers in the task is greater than FailedPeerCountLimit,
+		// then scheduler notifies running peers of failure.
+		task.NotifyPeers(&rpcscheduler.PeerPacket{
+			Code: base.Code_SchedTaskStatusError,
+		}, resource.PeerEventDownloadFailed)
 		task.PeerFailedCount.Store(0)
-		task.NotifyPeers(base.Code_SchedTaskStatusError, resource.PeerEventDownloadFailed)
 	}
 
 	if task.FSM.Is(resource.TaskStateFailed) {


Test patch:
diff --git a/client/config/peerhost_test.go b/client/config/peerhost_test.go
index 09c6b47a..d6a0b2f5 100644
--- a/client/config/peerhost_test.go
+++ b/client/config/peerhost_test.go
@@ -443,7 +443,7 @@ func TestPeerHostOption_Load(t *testing.T) {
 					},
 				},
 			},
-			Proxies: []*ProxyRule{
+			ProxyRules: []*ProxyRule{
 				{
 					Regx:     proxyExp,
 					UseHTTPS: false,
diff --git a/client/config/testdata/config/daemon.yaml b/client/config/testdata/config/daemon.yaml
index 92c534c7..00dc7f12 100644
--- a/client/config/testdata/config/daemon.yaml
+++ b/client/config/testdata/config/daemon.yaml
@@ -165,5 +165,5 @@ proxy:
         - "1000"
         - "2000"
   dumpHTTPContent: true
-reloadOption:
+reload:
   interval: 3m0s
diff --git a/pkg/dfpath/dfpath_test.go b/pkg/dfpath/dfpath_test.go
index 16975028..c5ac694f 100644
--- a/pkg/dfpath/dfpath_test.go
+++ b/pkg/dfpath/dfpath_test.go
@@ -20,6 +20,7 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/hashicorp/go-multierror"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -43,13 +44,14 @@ func TestNew(t *testing.T) {
 			expect: func(t *testing.T, options []Option) {
 				assert := assert.New(t)
 				cache.Once = sync.Once{}
-				cache.errs = []error{}
+				cache.err = &multierror.Error{}
 				d, err := New(options...)
 				assert.NoError(err)
 				assert.Equal(d.WorkHome(), DefaultWorkHome)
-				assert.Equal(d.CacheDir(), DefaultCacheDir)
+				assert.Equal(d.CacheDir(), "")
 				assert.Equal(d.LogDir(), DefaultLogDir)
-				assert.Equal(d.DataDir(), DefaultDataDir)
+				assert.Equal(d.DataDir(), "")
+				assert.Equal(d.PluginDir(), DefaultPluginDir)
 			},
 		},
 		{
@@ -58,13 +60,14 @@ func TestNew(t *testing.T) {
 			expect: func(t *testing.T, options []Option) {
 				assert := assert.New(t)
 				cache.Once = sync.Once{}
-				cache.errs = []error{}
+				cache.err = &multierror.Error{}
 				d, err := New(options...)
 				assert.NoError(err)
 				assert.Equal(d.WorkHome(), "foo")
-				assert.Equal(d.CacheDir(), DefaultCacheDir)
+				assert.Equal(d.CacheDir(), "")
 				assert.Equal(d.LogDir(), DefaultLogDir)
-				assert.Equal(d.DataDir(), DefaultDataDir)
+				assert.Equal(d.DataDir(), "")
+				assert.Equal(d.PluginDir(), DefaultPluginDir)
 			},
 		},
 		{
@@ -73,13 +76,14 @@ func TestNew(t *testing.T) {
 			expect: func(t *testing.T, options []Option) {
 				assert := assert.New(t)
 				cache.Once = sync.Once{}
-				cache.errs = []error{}
+				cache.err = &multierror.Error{}
 				d, err := New(options...)
 				assert.NoError(err)
 				assert.Equal(d.WorkHome(), DefaultWorkHome)
 				assert.Equal(d.CacheDir(), "foo")
 				assert.Equal(d.LogDir(), DefaultLogDir)
-				assert.Equal(d.DataDir(), DefaultDataDir)
+				assert.Equal(d.DataDir(), "")
+				assert.Equal(d.PluginDir(), DefaultPluginDir)
 			},
 		},
 		{
@@ -88,13 +92,30 @@ func TestNew(t *testing.T) {
 			expect: func(t *testing.T, options []Option) {
 				assert := assert.New(t)
 				cache.Once = sync.Once{}
-				cache.errs = []error{}
+				cache.err = &multierror.Error{}
 				d, err := New(options...)
 				assert.NoError(err)
 				assert.Equal(d.WorkHome(), DefaultWorkHome)
-				assert.Equal(d.CacheDir(), DefaultCacheDir)
+				assert.Equal(d.CacheDir(), "")
 				assert.Equal(d.LogDir(), "foo")
-				assert.Equal(d.DataDir(), DefaultDataDir)
+				assert.Equal(d.DataDir(), "")
+				assert.Equal(d.PluginDir(), DefaultPluginDir)
+			},
+		},
+		{
+			name:    "new dfpath by pluginDir",
+			options: []Option{WithPluginDir("foo")},
+			expect: func(t *testing.T, options []Option) {
+				assert := assert.New(t)
+				cache.Once = sync.Once{}
+				cache.err = &multierror.Error{}
+				d, err := New(options...)
+				assert.NoError(err)
+				assert.Equal(d.WorkHome(), DefaultWorkHome)
+				assert.Equal(d.CacheDir(), "")
+				assert.Equal(d.LogDir(), DefaultLogDir)
+				assert.Equal(d.DataDir(), "")
+				assert.Equal(d.PluginDir(), "foo")
 			},
 		},
 	}
diff --git a/pkg/math/math_test.go b/pkg/math/math_test.go
new file mode 100644
index 00000000..ad5f6741
--- /dev/null
+++ b/pkg/math/math_test.go
@@ -0,0 +1,59 @@
+/*
+ *     Copyright 2022 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package math
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMax(t *testing.T) {
+	assert := assert.New(t)
+	assert.Equal(1, Max(1))
+	assert.Equal(3, Max(1, 2, 3))
+	assert.Equal(int8(3), Max(int8(1), int8(2), int8(3)))
+	assert.Equal(int16(3), Max(int16(1), int16(2), int16(3)))
+	assert.Equal(int32(3), Max(int32(1), int32(2), int32(3)))
+	assert.Equal(int64(3), Max(int64(1), int64(2), int64(3)))
+	assert.Equal(uint(3), Max(uint(1), uint(2), uint(3)))
+	assert.Equal(uint8(3), Max(uint8(1), uint8(2), uint8(3)))
+	assert.Equal(uint16(3), Max(uint16(1), uint16(2), uint16(3)))
+	assert.Equal(uint32(3), Max(uint32(1), uint32(2), uint32(3)))
+	assert.Equal(uint64(3), Max(uint64(1), uint64(2), uint64(3)))
+	assert.Equal(float32(1.3), Max(float32(1.1), float32(1.2), float32(1.3)))
+	assert.Equal(float64(1.3), Max(1.1, 1.2, 1.3))
+	assert.Equal("c", Max("a", "b", "c"))
+}
+
+func TestMin(t *testing.T) {
+	assert := assert.New(t)
+	assert.Equal(1, Min(1))
+	assert.Equal(1, Min(1, 2, 3))
+	assert.Equal(int8(1), Min(int8(1), int8(2), int8(3)))
+	assert.Equal(int16(1), Min(int16(1), int16(2), int16(3)))
+	assert.Equal(int32(1), Min(int32(1), int32(2), int32(3)))
+	assert.Equal(int64(1), Min(int64(1), int64(2), int64(3)))
+	assert.Equal(uint(1), Min(uint(1), uint(2), uint(3)))
+	assert.Equal(uint8(1), Min(uint8(1), uint8(2), uint8(3)))
+	assert.Equal(uint16(1), Min(uint16(1), uint16(2), uint16(3)))
+	assert.Equal(uint32(1), Min(uint32(1), uint32(2), uint32(3)))
+	assert.Equal(uint64(1), Min(uint64(1), uint64(2), uint64(3)))
+	assert.Equal(float32(1.1), Min(float32(1.1), float32(1.2), float32(1.3)))
+	assert.Equal(float64(1.1), Min(1.1, 1.2, 1.3))
+	assert.Equal("a", Min("a", "b", "c"))
+}
diff --git a/scheduler/resource/task_test.go b/scheduler/resource/task_test.go
index 02a5a214..c7fa5f61 100644
--- a/scheduler/resource/task_test.go
+++ b/scheduler/resource/task_test.go
@@ -885,7 +885,7 @@ func TestTask_NotifyPeers(t *testing.T) {
 			name: "peer state is PeerStatePending",
 			run: func(t *testing.T, task *Task, mockPeer *Peer, stream rpcscheduler.Scheduler_ReportPieceResultServer, ms *rpcschedulermocks.MockScheduler_ReportPieceResultServerMockRecorder) {
 				mockPeer.FSM.SetState(PeerStatePending)
-				task.NotifyPeers(base.Code_SchedTaskStatusError, PeerEventDownloadFailed)
+				task.NotifyPeers(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError}, PeerEventDownloadFailed)
 
 				assert := assert.New(t)
 				assert.True(mockPeer.FSM.Is(PeerStatePending))
@@ -895,7 +895,7 @@ func TestTask_NotifyPeers(t *testing.T) {
 			name: "peer state is PeerStateRunning and stream is empty",
 			run: func(t *testing.T, task *Task, mockPeer *Peer, stream rpcscheduler.Scheduler_ReportPieceResultServer, ms *rpcschedulermocks.MockScheduler_ReportPieceResultServerMockRecorder) {
 				mockPeer.FSM.SetState(PeerStateRunning)
-				task.NotifyPeers(base.Code_SchedTaskStatusError, PeerEventDownloadFailed)
+				task.NotifyPeers(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError}, PeerEventDownloadFailed)
 
 				assert := assert.New(t)
 				assert.True(mockPeer.FSM.Is(PeerStateRunning))
@@ -908,7 +908,7 @@ func TestTask_NotifyPeers(t *testing.T) {
 				mockPeer.StoreStream(stream)
 				ms.Send(gomock.Eq(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError})).Return(errors.New("foo")).Times(1)
 
-				task.NotifyPeers(base.Code_SchedTaskStatusError, PeerEventDownloadFailed)
+				task.NotifyPeers(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError}, PeerEventDownloadFailed)
 
 				assert := assert.New(t)
 				assert.True(mockPeer.FSM.Is(PeerStateRunning))
@@ -921,7 +921,7 @@ func TestTask_NotifyPeers(t *testing.T) {
 				mockPeer.StoreStream(stream)
 				ms.Send(gomock.Eq(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError})).Return(errors.New("foo")).Times(1)
 
-				task.NotifyPeers(base.Code_SchedTaskStatusError, PeerEventRegisterNormal)
+				task.NotifyPeers(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError}, PeerEventDownloadFailed)
 
 				assert := assert.New(t)
 				assert.True(mockPeer.FSM.Is(PeerStateRunning))
@@ -934,7 +934,7 @@ func TestTask_NotifyPeers(t *testing.T) {
 				mockPeer.StoreStream(stream)
 				ms.Send(gomock.Eq(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError})).Return(nil).Times(1)
 
-				task.NotifyPeers(base.Code_SchedTaskStatusError, PeerEventDownloadFailed)
+				task.NotifyPeers(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError}, PeerEventDownloadFailed)
 
 				assert := assert.New(t)
 				assert.True(mockPeer.FSM.Is(PeerStateFailed))
diff --git a/scheduler/service/service_test.go b/scheduler/service/service_test.go
index a798ed0e..425bf736 100644
--- a/scheduler/service/service_test.go
+++ b/scheduler/service/service_test.go
@@ -32,6 +32,8 @@ import (
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	"d7y.io/dragonfly/v2/manager/types"
@@ -39,6 +41,7 @@ import (
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/base/common"
+	"d7y.io/dragonfly/v2/pkg/rpc/errordetails"
 	rpcscheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	rpcschedulermocks "d7y.io/dragonfly/v2/pkg/rpc/scheduler/mocks"
 	"d7y.io/dragonfly/v2/scheduler/config"
@@ -2893,10 +2896,24 @@ func TestService_handleTaskSuccess(t *testing.T) {
 }
 
 func TestService_handleTaskFail(t *testing.T) {
+	rst := status.Newf(codes.Aborted, "response is not valid")
+	st, err := rst.WithDetails(&errordetails.SourceError{Temporary: false})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rtst := status.Newf(codes.Aborted, "response is not valid")
+	tst, err := rtst.WithDetails(&errordetails.SourceError{Temporary: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	tests := []struct {
-		name   string
-		mock   func(task *resource.Task)
-		expect func(t *testing.T, task *resource.Task)
+		name            string
+		backToSourceErr *errordetails.SourceError
+		seedPeerErr     error
+		mock            func(task *resource.Task)
+		expect          func(t *testing.T, task *resource.Task)
 	}{
 		{
 			name: "task state is TaskStatePending",
@@ -2938,10 +2955,57 @@ func TestService_handleTaskFail(t *testing.T) {
 				assert.True(task.FSM.Is(resource.TaskStateFailed))
 			},
 		},
+		{
+			name:            "peer back-to-source fails due to an unrecoverable error",
+			backToSourceErr: &errordetails.SourceError{Temporary: false},
+			mock: func(task *resource.Task) {
+				task.FSM.SetState(resource.TaskStateRunning)
+			},
+			expect: func(t *testing.T, task *resource.Task) {
+				assert := assert.New(t)
+				assert.True(task.FSM.Is(resource.TaskStateFailed))
+				assert.Equal(task.PeerFailedCount.Load(), int32(0))
+			},
+		},
+		{
+			name:            "peer back-to-source fails due to an temporary error",
+			backToSourceErr: &errordetails.SourceError{Temporary: true},
+			mock: func(task *resource.Task) {
+				task.FSM.SetState(resource.TaskStateRunning)
+			},
+			expect: func(t *testing.T, task *resource.Task) {
+				assert := assert.New(t)
+				assert.True(task.FSM.Is(resource.TaskStateFailed))
+			},
+		},
+		{
+			name:        "seed peer back-to-source fails due to an unrecoverable error",
+			seedPeerErr: st.Err(),
+			mock: func(task *resource.Task) {
+				task.FSM.SetState(resource.TaskStateRunning)
+			},
+			expect: func(t *testing.T, task *resource.Task) {
+				assert := assert.New(t)
+				assert.True(task.FSM.Is(resource.TaskStateFailed))
+				assert.Equal(task.PeerFailedCount.Load(), int32(0))
+			},
+		},
+		{
+			name:        "seed peer back-to-source fails due to an temporary error",
+			seedPeerErr: tst.Err(),
+			mock: func(task *resource.Task) {
+				task.FSM.SetState(resource.TaskStateRunning)
+			},
+			expect: func(t *testing.T, task *resource.Task) {
+				assert := assert.New(t)
+				assert.True(task.FSM.Is(resource.TaskStateFailed))
+				assert.Equal(task.PeerFailedCount.Load(), int32(0))
+			},
+		},
 		{
 			name: "number of failed peers in the task is greater than FailedPeerCountLimit",
 			mock: func(task *resource.Task) {
-				task.FSM.SetState(resource.TaskStateFailed)
+				task.FSM.SetState(resource.TaskStateRunning)
 				task.PeerFailedCount.Store(201)
 			},
 			expect: func(t *testing.T, task *resource.Task) {
@@ -2964,7 +3028,7 @@ func TestService_handleTaskFail(t *testing.T) {
 			task := resource.NewTask(mockTaskID, mockTaskURL, base.TaskType_Normal, mockTaskURLMeta, resource.WithBackToSourceLimit(mockTaskBackToSourceLimit))
 
 			tc.mock(task)
-			svc.handleTaskFail(context.Background(), task)
+			svc.handleTaskFail(context.Background(), task, tc.backToSourceErr, tc.seedPeerErr)
 			tc.expect(t, task)
 		})
 	}