Repository: dragonflyoss/Dragonfly2
Pull request #1260 (2022-04-18 07:47:36)

fix: client synchronizer report error lock and dial grpc timeout
<!--- Provide a general summary of your changes in the Title above -->

## Description

1. Cancel the peer task in background when encounter report to scheduler err in sync manager
2. Remove superfluous retry to sync and get pieces from other peers to make sync pieces quickly.

## 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: -->

- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] 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/1260?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1260](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1260?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (e04a64e) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/96bdf9f3d0f37c201882f5209d15a9382538f9ff?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (96bdf9f) will **decrease** coverage by `0.16%`.
> The diff coverage is `11.11%`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1260/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/1260?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main    #1260      +/-   ##
==========================================
- Coverage   46.11%   45.95%   -0.17%     
==========================================
  Files         135      135              
  Lines       11245    11248       +3     
==========================================
- Hits         5186     5169      -17     
- Misses       5486     5518      +32     
+ Partials      573      561      -12     
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `45.95% <11.11%> (-0.17%)` | :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/1260?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [client/daemon/peer/peertask\_conductor.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1260/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2NvbmR1Y3Rvci5nbw==) | `57.61% <0.00%> (-0.54%)` | :arrow_down: |
| [...ent/daemon/peer/peertask\_piecetask\_synchronizer.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1260/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19zeW5jaHJvbml6ZXIuZ28=) | `54.44% <14.28%> (+9.81%)` | :arrow_up: |
| [client/daemon/peer/peertask\_piecetask\_poller.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1260/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19wb2xsZXIuZ28=) | `30.37% <0.00%> (-22.97%)` | :arrow_down: |

KubeStacker: LGTM 

Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 0fc067eb..fcd688e5 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -7,6 +7,8 @@ on:
   pull_request:
     branches: [ main, release-* ]
     paths-ignore: [ '**.md', '**.png', '**.jpg', '**.svg', '**/docs/**' ]
+  schedule:
+    - cron: '0 4 * * *'
 
 env:
   GO_VERSION: 1.17
diff --git a/client/config/flags.go b/client/config/flags.go
index d8718fc9..c1c24659 100644
--- a/client/config/flags.go
+++ b/client/config/flags.go
@@ -27,7 +27,7 @@ import (
 	"golang.org/x/time/rate"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 )
 
 // SchedulersValue implements the pflag.Value interface.
diff --git a/client/config/peerhost.go b/client/config/peerhost.go
index 29e79879..b8cc10aa 100644
--- a/client/config/peerhost.go
+++ b/client/config/peerhost.go
@@ -35,7 +35,7 @@ import (
 
 	"d7y.io/dragonfly/v2/client/clientutil"
 	"d7y.io/dragonfly/v2/cmd/dependency/base"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/unit"
 	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
 )
diff --git a/client/config/peerhost_darwin.go b/client/config/peerhost_darwin.go
index 7a295faa..68d05a2a 100644
--- a/client/config/peerhost_darwin.go
+++ b/client/config/peerhost_darwin.go
@@ -26,7 +26,7 @@ import (
 	"golang.org/x/time/rate"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/util/hostutils"
 	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
 )
diff --git a/client/config/peerhost_linux.go b/client/config/peerhost_linux.go
index 5d65e120..da98d305 100644
--- a/client/config/peerhost_linux.go
+++ b/client/config/peerhost_linux.go
@@ -26,7 +26,7 @@ import (
 	"golang.org/x/time/rate"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/util/hostutils"
 	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
 )
diff --git a/client/daemon/daemon.go b/client/daemon/daemon.go
index c0789ed3..fbcaaa6b 100644
--- a/client/daemon/daemon.go
+++ b/client/daemon/daemon.go
@@ -47,7 +47,7 @@ import (
 	"d7y.io/dragonfly/v2/client/daemon/storage"
 	"d7y.io/dragonfly/v2/client/daemon/upload"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/reachable"
diff --git a/client/daemon/peer/peertask_conductor.go b/client/daemon/peer/peertask_conductor.go
index bf057b6c..a4ad5995 100644
--- a/client/daemon/peer/peertask_conductor.go
+++ b/client/daemon/peer/peertask_conductor.go
@@ -107,6 +107,7 @@ type peerTaskConductor struct {
 	// peerPacketStream stands schedulerclient.PeerPacketStream from scheduler
 	peerPacketStream schedulerclient.PeerPacketStream
 	// peerPacket is the latest available peers from peerPacketCh
+	// Deprecated: remove in future release
 	peerPacket      atomic.Value // *scheduler.PeerPacket
 	legacyPeerCount *atomic.Int64
 	// peerPacketReady will receive a ready signal for peerPacket ready
@@ -119,7 +120,6 @@ type peerTaskConductor struct {
 
 	// same actions must be done only once, like close done channel and so on
 	statusOnce sync.Once
-	cancelOnce sync.Once
 	// done channel will be closed when peer task success
 	successCh chan struct{}
 	// fail channel will be closed after peer task fail
@@ -392,20 +392,36 @@ func (pt *peerTaskConductor) Log() *logger.SugaredLoggerOnWith {
 }
 
 func (pt *peerTaskConductor) cancel(code base.Code, reason string) {
-	pt.cancelOnce.Do(func() {
+	pt.statusOnce.Do(func() {
 		pt.failedCode = code
 		pt.failedReason = reason
-		pt.Fail()
+		pt.fail()
 	})
 }
 
+// only use when receive back source code from scheduler
 func (pt *peerTaskConductor) markBackSource() {
 	pt.needBackSource.Store(true)
 	// when close peerPacketReady, pullPiecesFromPeers will invoke backSource
 	close(pt.peerPacketReady)
+	// let legacy mode exit
+	pt.peerPacket.Store(&scheduler.PeerPacket{
+		TaskId:        pt.taskID,
+		SrcPid:        pt.peerID,
+		ParallelCount: 1,
+		MainPeer:      nil,
+		StealPeers: []*scheduler.PeerPacket_DestPeer{
+			{
+				Ip:      pt.host.Ip,
+				RpcPort: pt.host.RpcPort,
+				PeerId:  pt.peerID,
+			},
+		},
+		Code: base.Code_SchedNeedBackSource,
+	})
 }
 
-// only use when schedule timeout
+// only use when legacy get piece from peers schedule timeout
 func (pt *peerTaskConductor) forceBackSource() {
 	pt.needBackSource.Store(true)
 	pt.backSource()
@@ -647,9 +663,9 @@ loop:
 		legacyPeerCount := int64(len(peerPacket.StealPeers))
 		pt.Debugf("connect to %d legacy peers", legacyPeerCount)
 		pt.legacyPeerCount.Store(legacyPeerCount)
-		pt.peerPacket.Store(peerPacket)
 
-		// legacy mode: send peerPacketReady
+		// legacy mode: update peer packet, then send peerPacketReady
+		pt.peerPacket.Store(peerPacket)
 		select {
 		case pt.peerPacketReady <- true:
 		case <-pt.successCh:
@@ -910,8 +926,8 @@ func (pt *peerTaskConductor) waitFirstPeerPacket() (done bool, backSource bool)
 	case _, ok := <-pt.peerPacketReady:
 		if ok {
 			// preparePieceTasksByPeer func already send piece result with error
-			pt.Infof("new peer client ready, scheduler time cost: %dus, main peer: %s",
-				time.Since(pt.startTime).Microseconds(), pt.peerPacket.Load().(*scheduler.PeerPacket).MainPeer)
+			pt.Infof("new peer client ready, scheduler time cost: %dus, peer count: %d",
+				time.Since(pt.startTime).Microseconds(), len(pt.peerPacket.Load().(*scheduler.PeerPacket).StealPeers))
 			return true, false
 		}
 		// when scheduler says base.Code_SchedNeedBackSource, receivePeerPacket will close pt.peerPacketReady
@@ -946,7 +962,7 @@ func (pt *peerTaskConductor) waitAvailablePeerPacket() (int32, bool) {
 	case _, ok := <-pt.peerPacketReady:
 		if ok {
 			// preparePieceTasksByPeer func already send piece result with error
-			pt.Infof("new peer client ready, main peer: %s", pt.peerPacket.Load().(*scheduler.PeerPacket).MainPeer)
+			pt.Infof("new peer client ready, peer count: %d", len(pt.peerPacket.Load().(*scheduler.PeerPacket).StealPeers))
 			// research from piece 0
 			return 0, true
 		}
diff --git a/client/daemon/peer/peertask_dummy.go b/client/daemon/peer/peertask_dummy.go
index 14df5bff..c684cbdc 100644
--- a/client/daemon/peer/peertask_dummy.go
+++ b/client/daemon/peer/peertask_dummy.go
@@ -22,7 +22,7 @@ import (
 	"google.golang.org/grpc"
 
 	"d7y.io/dragonfly/v2/internal/dferrors"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	schedulerclient "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
diff --git a/client/daemon/peer/peertask_piecetask_poller.go b/client/daemon/peer/peertask_piecetask_poller.go
index 8d1cf877..77f6261c 100644
--- a/client/daemon/peer/peertask_piecetask_poller.go
+++ b/client/daemon/peer/peertask_piecetask_poller.go
@@ -88,6 +88,7 @@ func (poller *pieceTaskPoller) preparePieceTasksByPeer(
 	span.SetAttributes(config.AttributeGetPieceLimit.Int(int(request.Limit)))
 	defer span.End()
 
+	var maxRetries = 60
 	// when cdn returns base.Code_CDNTaskNotFound, report it to scheduler and wait cdn download it.
 retry:
 	ptc.Debugf("try get piece task from peer %s, piece num: %d, limit: %d\"", peer.PeerId, request.StartNum, request.Limit)
@@ -140,8 +141,10 @@ retry:
 	}
 
 	// currently, before cdn gc tasks, it did not notify scheduler, when cdn complains Code_CDNTaskNotFound, retry
-	if code == base.Code_CDNTaskNotFound && curPeerPacket == ptc.peerPacket.Load().(*scheduler.PeerPacket) {
+	if maxRetries > 0 && code == base.Code_CDNTaskNotFound && curPeerPacket == ptc.peerPacket.Load().(*scheduler.PeerPacket) {
 		span.AddEvent("retry for CdnTaskNotFound")
+		time.Sleep(time.Second)
+		maxRetries--
 		goto retry
 	}
 	return nil, err
diff --git a/client/daemon/peer/peertask_piecetask_synchronizer.go b/client/daemon/peer/peertask_piecetask_synchronizer.go
index e1d6f06d..3515da51 100644
--- a/client/daemon/peer/peertask_piecetask_synchronizer.go
+++ b/client/daemon/peer/peertask_piecetask_synchronizer.go
@@ -21,6 +21,7 @@ import (
 	"io"
 	"sync"
 
+	"github.com/pkg/errors"
 	"go.uber.org/atomic"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -138,6 +139,11 @@ func (s *pieceTaskSyncManager) newPieceTaskSynchronizer(
 
 	request.DstPid = dstPeer.PeerId
 	client, err := dfclient.SyncPieceTasks(ctx, dstPeer, request)
+	// Refer: https://github.com/grpc/grpc-go/blob/v1.44.0/stream.go#L104
+	// When receive io.EOF, the real error should be discovered using RecvMsg, here is client.Recv() here
+	if err == io.EOF && client != nil {
+		_, err = client.Recv()
+	}
 	if err != nil {
 		s.peerTaskConductor.Errorf("call SyncPieceTasks error: %s, dest peer: %s", err, dstPeer.PeerId)
 		return err
@@ -183,13 +189,23 @@ func (s *pieceTaskSyncManager) newMultiPieceTaskSynchronizer(
 			s.peerTaskConductor.Infof("connected to peer: %s", peer.PeerId)
 			continue
 		}
-		legacyPeers = append(legacyPeers, peer)
 		// when err is codes.Unimplemented, fallback to legacy get piece grpc
 		stat, ok := status.FromError(err)
 		if ok && stat.Code() == codes.Unimplemented {
+			// for legacy peers, when get pieces error, will report the error
 			s.peerTaskConductor.Warnf("connect peer %s error: %s, fallback to legacy get piece grpc", peer.PeerId, err)
+			legacyPeers = append(legacyPeers, peer)
+			continue
+		}
+
+		// other errors, report to scheduler
+		if errors.Is(err, context.DeadlineExceeded) {
+			// connect timeout error, report to scheduler to get more available peers
+			s.reportInvalidPeer(peer, base.Code_ClientConnectionError)
+			s.peerTaskConductor.Infof("connect to peer %s with error: %s, peer is invalid, skip legacy grpc", peer.PeerId, err)
 		} else {
-			s.reportError(peer)
+			// other errors, report to scheduler to get more available peers
+			s.reportInvalidPeer(peer, base.Code_ClientPieceRequestFail)
 			s.peerTaskConductor.Errorf("connect peer %s error: %s, not codes.Unimplemented", peer.PeerId, err)
 		}
 	}
@@ -197,24 +213,26 @@ func (s *pieceTaskSyncManager) newMultiPieceTaskSynchronizer(
 	return legacyPeers
 }
 
-func compositePieceResult(peerTaskConductor *peerTaskConductor, destPeer *scheduler.PeerPacket_DestPeer) *scheduler.PieceResult {
+func compositePieceResult(peerTaskConductor *peerTaskConductor, destPeer *scheduler.PeerPacket_DestPeer, code base.Code) *scheduler.PieceResult {
 	return &scheduler.PieceResult{
 		TaskId:        peerTaskConductor.taskID,
 		SrcPid:        peerTaskConductor.peerID,
 		DstPid:        destPeer.PeerId,
 		PieceInfo:     &base.PieceInfo{},
 		Success:       false,
-		Code:          base.Code_ClientPieceRequestFail,
+		Code:          code,
 		HostLoad:      nil,
 		FinishedCount: peerTaskConductor.readyPieces.Settled(),
 	}
 }
 
-func (s *pieceTaskSyncManager) reportError(destPeer *scheduler.PeerPacket_DestPeer) {
-	sendError := s.peerTaskConductor.sendPieceResult(compositePieceResult(s.peerTaskConductor, destPeer))
+func (s *pieceTaskSyncManager) reportInvalidPeer(destPeer *scheduler.PeerPacket_DestPeer, code base.Code) {
+	sendError := s.peerTaskConductor.sendPieceResult(compositePieceResult(s.peerTaskConductor, destPeer, code))
 	if sendError != nil {
-		s.peerTaskConductor.cancel(base.Code_SchedError, sendError.Error())
 		s.peerTaskConductor.Errorf("connect peer %s failed and send piece result with error: %s", destPeer.PeerId, sendError)
+		go s.peerTaskConductor.cancel(base.Code_SchedError, sendError.Error())
+	} else {
+		s.peerTaskConductor.Debugf("report invalid peer %s/%d to scheduler", destPeer.PeerId, code)
 	}
 }
 
@@ -302,12 +320,12 @@ func (s *pieceTaskSynchronizer) receive(piecePacket *base.PiecePacket) {
 	if err == io.EOF {
 		s.Debugf("synchronizer receives io.EOF")
 	} else if s.canceled(err) {
-		s.error.Store(&pieceTaskSynchronizerError{err})
 		s.Debugf("synchronizer receives canceled")
+		s.error.Store(&pieceTaskSynchronizerError{err})
 	} else {
+		s.Errorf("synchronizer receives with error: %s", err)
 		s.error.Store(&pieceTaskSynchronizerError{err})
 		s.reportError()
-		s.Errorf("synchronizer receives with error: %s", err)
 	}
 }
 
@@ -331,10 +349,12 @@ func (s *pieceTaskSynchronizer) acquire(request *base.PieceTaskRequest) error {
 }
 
 func (s *pieceTaskSynchronizer) reportError() {
-	sendError := s.peerTaskConductor.sendPieceResult(compositePieceResult(s.peerTaskConductor, s.dstPeer))
+	sendError := s.peerTaskConductor.sendPieceResult(compositePieceResult(s.peerTaskConductor, s.dstPeer, base.Code_ClientPieceRequestFail))
 	if sendError != nil {
-		s.peerTaskConductor.cancel(base.Code_SchedError, sendError.Error())
 		s.Errorf("sync piece info failed and send piece result with error: %s", sendError)
+		go s.peerTaskConductor.cancel(base.Code_SchedError, sendError.Error())
+	} else {
+		s.Debugf("report sync piece error to scheduler")
 	}
 }
 
diff --git a/cmd/dependency/dependency.go b/cmd/dependency/dependency.go
index ac872f57..526d6b03 100644
--- a/cmd/dependency/dependency.go
+++ b/cmd/dependency/dependency.go
@@ -46,7 +46,7 @@ import (
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/cmd/dependency/base"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/unit"
 	"d7y.io/dragonfly/v2/pkg/util/hostutils"
diff --git a/cmd/dfcache/cmd/root.go b/cmd/dfcache/cmd/root.go
index 6ea13ac6..92a374c3 100644
--- a/cmd/dfcache/cmd/root.go
+++ b/cmd/dfcache/cmd/root.go
@@ -29,7 +29,7 @@ import (
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/cmd/dependency"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/client"
 	"d7y.io/dragonfly/v2/version"
diff --git a/cmd/dfget/cmd/daemon.go b/cmd/dfget/cmd/daemon.go
index c577c9bd..746553cd 100644
--- a/cmd/dfget/cmd/daemon.go
+++ b/cmd/dfget/cmd/daemon.go
@@ -33,7 +33,7 @@ import (
 	server "d7y.io/dragonfly/v2/client/daemon"
 	"d7y.io/dragonfly/v2/cmd/dependency"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/client"
 	"d7y.io/dragonfly/v2/version"
diff --git a/cmd/dfget/cmd/root.go b/cmd/dfget/cmd/root.go
index 202ed94c..eccdd61e 100644
--- a/cmd/dfget/cmd/root.go
+++ b/cmd/dfget/cmd/root.go
@@ -36,8 +36,8 @@ import (
 	"d7y.io/dragonfly/v2/cmd/dependency"
 	"d7y.io/dragonfly/v2/internal/constants"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
 	"d7y.io/dragonfly/v2/pkg/basic"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/client"
 	"d7y.io/dragonfly/v2/pkg/source"
diff --git a/internal/dfnet/dfnet.go b/pkg/dfnet/dfnet.go
similarity index 100%
rename from internal/dfnet/dfnet.go
rename to pkg/dfnet/dfnet.go
diff --git a/pkg/rpc/cdnsystem/client/client.go b/pkg/rpc/cdnsystem/client/client.go
index 1fc3236a..4ce71ff1 100644
--- a/pkg/rpc/cdnsystem/client/client.go
+++ b/pkg/rpc/cdnsystem/client/client.go
@@ -25,7 +25,7 @@ import (
 	"google.golang.org/grpc"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
@@ -103,32 +103,22 @@ func (cc *cdnClient) ObtainSeeds(ctx context.Context, sr *cdnsystem.SeedRequest,
 }
 
 func (cc *cdnClient) GetPieceTasks(ctx context.Context, addr dfnet.NetAddr, req *base.PieceTaskRequest, opts ...grpc.CallOption) (*base.PiecePacket, error) {
-	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, err := cc.getSeederClientWithTarget(addr.GetEndpoint())
-		if err != nil {
-			return nil, err
-		}
-		return client.GetPieceTasks(ctx, req, opts...)
-	}, 0.2, 2.0, 3, nil)
+	client, err := cc.getSeederClientWithTarget(addr.GetEndpoint())
 	if err != nil {
-		logger.WithTaskID(req.TaskId).Infof("GetPieceTasks: invoke cdn node %s GetPieceTasks failed: %v", addr.GetEndpoint(), err)
 		return nil, err
 	}
-	return res.(*base.PiecePacket), nil
+	return client.GetPieceTasks(ctx, req, opts...)
 }
 
 func (cc *cdnClient) SyncPieceTasks(ctx context.Context, addr dfnet.NetAddr, req *base.PieceTaskRequest, opts ...grpc.CallOption) (cdnsystem.Seeder_SyncPieceTasksClient, error) {
-	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, err := cc.getSeederClientWithTarget(addr.GetEndpoint())
-		if err != nil {
-			return nil, err
-		}
-		return client.SyncPieceTasks(ctx, opts...)
-	}, 0.2, 2.0, 3, nil)
+	client, err := cc.getSeederClientWithTarget(addr.GetEndpoint())
+	if err != nil {
+		return nil, err
+	}
+	syncClient, err := client.SyncPieceTasks(ctx, opts...)
 	if err != nil {
 		logger.WithTaskID(req.TaskId).Infof("SyncPieceTasks: invoke cdn node %s SyncPieceTasks failed: %v", addr.GetEndpoint(), err)
 		return nil, err
 	}
-	client := res.(cdnsystem.Seeder_SyncPieceTasksClient)
-	return client, client.Send(req)
+	return syncClient, syncClient.Send(req)
 }
diff --git a/pkg/rpc/client.go b/pkg/rpc/client.go
index dff07941..cea09795 100644
--- a/pkg/rpc/client.go
+++ b/pkg/rpc/client.go
@@ -33,7 +33,7 @@ import (
 
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 )
 
@@ -269,7 +269,6 @@ type candidateClient struct {
 }
 
 func (conn *Connection) createClient(target string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
-	// should not retry
 	ctx, cancel := context.WithTimeout(context.Background(), conn.dialTimeout)
 	defer cancel()
 	return grpc.DialContext(ctx, target, opts...)
diff --git a/pkg/rpc/dfdaemon/cdnsystem/mocks/client_mock.go b/pkg/rpc/dfdaemon/cdnsystem/mocks/client_mock.go
index 654f6f35..22794b47 100644
--- a/pkg/rpc/dfdaemon/cdnsystem/mocks/client_mock.go
+++ b/pkg/rpc/dfdaemon/cdnsystem/mocks/client_mock.go
@@ -8,7 +8,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	dfnet "d7y.io/dragonfly/v2/internal/dfnet"
+	dfnet "d7y.io/dragonfly/v2/pkg/dfnet"
 	base "d7y.io/dragonfly/v2/pkg/rpc/base"
 	cdnsystem "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
 	client "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem/client"
diff --git a/pkg/rpc/dfdaemon/client/client.go b/pkg/rpc/dfdaemon/client/client.go
index 4131cc35..560f7d35 100644
--- a/pkg/rpc/dfdaemon/client/client.go
+++ b/pkg/rpc/dfdaemon/client/client.go
@@ -28,7 +28,7 @@ import (
 	"google.golang.org/protobuf/types/known/emptypb"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
@@ -119,35 +119,25 @@ func (dc *daemonClient) Download(ctx context.Context, req *dfdaemon.DownRequest,
 
 func (dc *daemonClient) GetPieceTasks(ctx context.Context, target dfnet.NetAddr, ptr *base.PieceTaskRequest, opts ...grpc.CallOption) (*base.PiecePacket,
 	error) {
-	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, err := dc.getDaemonClientWithTarget(target.GetEndpoint())
-		if err != nil {
-			return nil, err
-		}
-		return client.GetPieceTasks(ctx, ptr, opts...)
-	}, 0.2, 2.0, 3, nil)
+	client, err := dc.getDaemonClientWithTarget(target.GetEndpoint())
 	if err != nil {
-		logger.WithTaskID(ptr.TaskId).Infof("GetPieceTasks: invoke daemon node %s GetPieceTasks failed: %v", target, err)
 		return nil, err
 	}
-	return res.(*base.PiecePacket), nil
+	return client.GetPieceTasks(ctx, ptr, opts...)
 }
 
 func (dc *daemonClient) SyncPieceTasks(ctx context.Context, target dfnet.NetAddr, ptr *base.PieceTaskRequest, opts ...grpc.CallOption) (dfdaemon.Daemon_SyncPieceTasksClient, error) {
-	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, err := dc.getDaemonClientWithTarget(target.GetEndpoint())
-		if err != nil {
-			return nil, err
-		}
-		return client.SyncPieceTasks(ctx, opts...)
-	}, 0.2, 2.0, 3, nil)
+	client, err := dc.getDaemonClientWithTarget(target.GetEndpoint())
+	if err != nil {
+		return nil, err
+	}
+	syncClient, err := client.SyncPieceTasks(ctx, opts...)
 	if err != nil {
 		logger.WithTaskID(ptr.TaskId).Infof("SyncPieceTasks: invoke daemon node %s SyncPieceTasks failed: %v", target, err)
 		return nil, err
 	}
 
-	client := res.(dfdaemon.Daemon_SyncPieceTasksClient)
-	return client, client.Send(ptr)
+	return syncClient, syncClient.Send(ptr)
 }
 
 func (dc *daemonClient) CheckHealth(ctx context.Context, target dfnet.NetAddr, opts ...grpc.CallOption) (err error) {
diff --git a/pkg/rpc/dfdaemon/client/mocks/client_mock.go b/pkg/rpc/dfdaemon/client/mocks/client_mock.go
index c100e9c0..14b76f9e 100644
--- a/pkg/rpc/dfdaemon/client/mocks/client_mock.go
+++ b/pkg/rpc/dfdaemon/client/mocks/client_mock.go
@@ -8,7 +8,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	dfnet "d7y.io/dragonfly/v2/internal/dfnet"
+	dfnet "d7y.io/dragonfly/v2/pkg/dfnet"
 	base "d7y.io/dragonfly/v2/pkg/rpc/base"
 	dfdaemon "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
 	client "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/client"
diff --git a/pkg/rpc/dfdaemon/client/peer.go b/pkg/rpc/dfdaemon/client/peer.go
index c2b3fb13..77463e76 100644
--- a/pkg/rpc/dfdaemon/client/peer.go
+++ b/pkg/rpc/dfdaemon/client/peer.go
@@ -23,7 +23,7 @@ import (
 
 	"google.golang.org/grpc"
 
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/base/common"
diff --git a/pkg/rpc/manager/client/client.go b/pkg/rpc/manager/client/client.go
index e2e04167..4daf9a85 100644
--- a/pkg/rpc/manager/client/client.go
+++ b/pkg/rpc/manager/client/client.go
@@ -29,7 +29,7 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/reachable"
 	"d7y.io/dragonfly/v2/pkg/rpc/manager"
 )
@@ -102,10 +102,10 @@ func NewWithAddrs(netAddrs []dfnet.NetAddr) (Client, error) {
 			logger.Infof("use %s address for manager grpc client", netAddr.Addr)
 			return New(netAddr.Addr)
 		}
-		logger.Warnf("%s address can not reachable", netAddr.Addr)
+		logger.Warnf("%s manager address can not reachable", netAddr.Addr)
 	}
 
-	return nil, errors.New("can not find available addresses")
+	return nil, errors.New("can not find available manager addresses")
 }
 
 func (c *client) GetScheduler(req *manager.GetSchedulerRequest) (*manager.Scheduler, error) {
diff --git a/pkg/rpc/scheduler/client/client.go b/pkg/rpc/scheduler/client/client.go
index 4ce5b3f9..7968d14e 100644
--- a/pkg/rpc/scheduler/client/client.go
+++ b/pkg/rpc/scheduler/client/client.go
@@ -27,7 +27,7 @@ import (
 
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
diff --git a/pkg/rpc/scheduler/client/mocks/client_mock.go b/pkg/rpc/scheduler/client/mocks/client_mock.go
index 7807c4f6..529f55d6 100644
--- a/pkg/rpc/scheduler/client/mocks/client_mock.go
+++ b/pkg/rpc/scheduler/client/mocks/client_mock.go
@@ -8,7 +8,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	dfnet "d7y.io/dragonfly/v2/internal/dfnet"
+	dfnet "d7y.io/dragonfly/v2/pkg/dfnet"
 	scheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	client "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
 	gomock "github.com/golang/mock/gomock"
diff --git a/pkg/rpc/server_listen.go b/pkg/rpc/server_listen.go
index 98d4152b..0cce05b2 100644
--- a/pkg/rpc/server_listen.go
+++ b/pkg/rpc/server_listen.go
@@ -23,7 +23,7 @@ import (
 	"syscall"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 )
 
 // Listen wraps net.Listen with dfnet.NetAddr
diff --git a/scheduler/resource/cdn.go b/scheduler/resource/cdn.go
index 7a169055..2cd9f178 100644
--- a/scheduler/resource/cdn.go
+++ b/scheduler/resource/cdn.go
@@ -27,7 +27,7 @@ import (
 	"google.golang.org/grpc"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base/common"
 	"d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
diff --git a/scheduler/resource/cdn_mock.go b/scheduler/resource/cdn_mock.go
index 67a461af..17138375 100644
--- a/scheduler/resource/cdn_mock.go
+++ b/scheduler/resource/cdn_mock.go
@@ -8,7 +8,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	dfnet "d7y.io/dragonfly/v2/internal/dfnet"
+	dfnet "d7y.io/dragonfly/v2/pkg/dfnet"
 	base "d7y.io/dragonfly/v2/pkg/rpc/base"
 	cdnsystem "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
 	client "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem/client"
diff --git a/scheduler/resource/peer.go b/scheduler/resource/peer.go
index 4648b029..6cc5696e 100644
--- a/scheduler/resource/peer.go
+++ b/scheduler/resource/peer.go
@@ -369,7 +369,7 @@ func (p *Peer) Depth() int {
 
 		// Prevent traversal tree from infinite loop
 		if p.ID == parent.ID {
-			p.Log.Info("tree structure produces an infinite loop")
+			p.Log.Error("tree structure produces an infinite loop")
 			break
 		}
 
@@ -381,16 +381,22 @@ func (p *Peer) Depth() int {
 
 // IsDescendant determines whether it is ancestor of peer
 func (p *Peer) IsDescendant(ancestor *Peer) bool {
-	return isDescendant(ancestor, p)
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+
+	return p.isDescendant(ancestor, p)
 }
 
 // IsAncestor determines whether it is descendant of peer
 func (p *Peer) IsAncestor(descendant *Peer) bool {
-	return isDescendant(p, descendant)
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+
+	return p.isDescendant(p, descendant)
 }
 
 // isDescendant determines whether it is ancestor of peer
-func isDescendant(ancestor, descendant *Peer) bool {
+func (p *Peer) isDescendant(ancestor, descendant *Peer) bool {
 	node := descendant
 	for node != nil {
 		parent, ok := node.LoadParent()
@@ -402,6 +408,12 @@ func isDescendant(ancestor, descendant *Peer) bool {
 			return true
 		}
 
+		// Prevent traversal tree from infinite loop
+		if parent.ID == descendant.ID {
+			p.Log.Error("tree structure produces an infinite loop")
+			return true
+		}
+
 		node = parent
 	}
 


Test patch:
diff --git a/.github/workflows/compatibility-e2e.yml b/.github/workflows/compatibility-e2e.yml
index cec1839e..3bb8e09b 100644
--- a/.github/workflows/compatibility-e2e.yml
+++ b/.github/workflows/compatibility-e2e.yml
@@ -7,6 +7,8 @@ on:
   pull_request:
     branches: [main, release-*]
     paths-ignore: ["**.md", "**.png", "**.jpg", "**.svg", "**/docs/**"]
+  schedule:
+    - cron: '0 4 * * *'
 
 env:
   GO_VERSION: 1.17
diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml
index 1f9cbcb4..4a44f34b 100644
--- a/.github/workflows/e2e.yml
+++ b/.github/workflows/e2e.yml
@@ -7,6 +7,8 @@ on:
   pull_request:
     branches: [main, release-*]
     paths-ignore: ["**.md", "**.png", "**.jpg", "**.svg", "**/docs/**"]
+  schedule:
+    - cron: '0 4 * * *'
 
 env:
   GO_VERSION: 1.17
diff --git a/client/config/peerhost_test.go b/client/config/peerhost_test.go
index 7a9c2840..6bf2f334 100644
--- a/client/config/peerhost_test.go
+++ b/client/config/peerhost_test.go
@@ -26,7 +26,7 @@ import (
 	"gopkg.in/yaml.v3"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/unit"
 )
 
diff --git a/client/daemon/daemon_test.go b/client/daemon/daemon_test.go
index f2ed6847..96a40294 100644
--- a/client/daemon/daemon_test.go
+++ b/client/daemon/daemon_test.go
@@ -22,7 +22,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc/manager"
 )
 
diff --git a/client/daemon/peer/peertask_manager_test.go b/client/daemon/peer/peertask_manager_test.go
index 38d2fa65..a8b86c99 100644
--- a/client/daemon/peer/peertask_manager_test.go
+++ b/client/daemon/peer/peertask_manager_test.go
@@ -49,7 +49,7 @@ import (
 	mock_daemon "d7y.io/dragonfly/v2/client/daemon/test/mock/daemon"
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
diff --git a/client/daemon/peer/peertask_stream_backsource_partial_test.go b/client/daemon/peer/peertask_stream_backsource_partial_test.go
index 607e328c..7242dd49 100644
--- a/client/daemon/peer/peertask_stream_backsource_partial_test.go
+++ b/client/daemon/peer/peertask_stream_backsource_partial_test.go
@@ -44,7 +44,7 @@ import (
 	"d7y.io/dragonfly/v2/client/daemon/test"
 	mock_daemon "d7y.io/dragonfly/v2/client/daemon/test/mock/daemon"
 	"d7y.io/dragonfly/v2/internal/dferrors"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
diff --git a/client/daemon/rpcserver/rpcserver_test.go b/client/daemon/rpcserver/rpcserver_test.go
index 8b8e6d82..942c35a2 100644
--- a/client/daemon/rpcserver/rpcserver_test.go
+++ b/client/daemon/rpcserver/rpcserver_test.go
@@ -34,7 +34,7 @@ import (
 	"d7y.io/dragonfly/v2/client/daemon/peer"
 	mock_peer "d7y.io/dragonfly/v2/client/daemon/test/mock/peer"
 	mock_storage "d7y.io/dragonfly/v2/client/daemon/test/mock/storage"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	dfdaemongrpc "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
diff --git a/client/daemon/test/mock/scheduler/scheduler_client.go b/client/daemon/test/mock/scheduler/scheduler_client.go
index 2821c1c9..4aa17da3 100644
--- a/client/daemon/test/mock/scheduler/scheduler_client.go
+++ b/client/daemon/test/mock/scheduler/scheduler_client.go
@@ -8,7 +8,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	dfnet "d7y.io/dragonfly/v2/internal/dfnet"
+	dfnet "d7y.io/dragonfly/v2/pkg/dfnet"
 	scheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	client "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
 	gomock "github.com/golang/mock/gomock"
diff --git a/scheduler/resource/cdn_test.go b/scheduler/resource/cdn_test.go
index 2fe1c30d..879f1a96 100644
--- a/scheduler/resource/cdn_test.go
+++ b/scheduler/resource/cdn_test.go
@@ -27,8 +27,8 @@ import (
 	gomock "github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 
-	"d7y.io/dragonfly/v2/internal/dfnet"
 	"d7y.io/dragonfly/v2/manager/types"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	rpcscheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	"d7y.io/dragonfly/v2/scheduler/config"
 	configmocks "d7y.io/dragonfly/v2/scheduler/config/mocks"
diff --git a/test/e2e/dfget_test.go b/test/e2e/dfget_test.go
index 8b389428..54b016b4 100644
--- a/test/e2e/dfget_test.go
+++ b/test/e2e/dfget_test.go
@@ -99,7 +99,8 @@ func singleDfgetTest(name, ns, label, podNamePrefix, container string, fileDetai
 		pod := e2eutil.NewPodExec(ns, podName, container)
 
 		// install curl
-		_, err = pod.Command("apk", "add", "-U", "curl").CombinedOutput()
+		out, err = pod.Command("apk", "add", "-U", "curl").CombinedOutput()
+		fmt.Println("apk output: " + string(out))
 		Expect(err).NotTo(HaveOccurred())
 
 		for path, size := range fileDetails {