Merge pull request #1085 from Starnop/retry-report

refactor: handle errors gracefully

dfget/core/core.go

@@ -213,9 +213,9 @@ func doDownload(cfg *config.Config, supernodeAPI api.SupernodeAPI,
 	}
 
 	err := downloader.DoDownloadTimeout(getter, timeout)
+	// report finished task to uploader regardless of the result of downloading from dragonfly
+	reportFinishedTask(cfg, getter)
 	if err == nil {
-		// report finished task to uploader when success to download by dragonfly
-		reportFinishedTask(cfg, getter)
 		return nil
 	}
 

supernode/daemon/mgr/gc/gc_task.go

@@ -116,7 +116,16 @@ func (gcm *Manager) gcCDNByTaskID(ctx context.Context, taskID string, full bool)
 }
 
 func (gcm *Manager) gcTaskByTaskID(ctx context.Context, taskID string) {
-	if err := gcm.progressMgr.DeleteTaskID(ctx, taskID); err != nil {
+	taskInfo, err := gcm.taskMgr.Get(ctx, taskID)
+	if err != nil {
+		logrus.Errorf("gc task: failed to get task info(%s): %v", taskID, err)
+	}
+
+	var pieceTotal int
+	if taskInfo != nil {
+		pieceTotal = int(taskInfo.PieceTotal)
+	}
+	if err := gcm.progressMgr.DeleteTaskID(ctx, taskID, pieceTotal); err != nil {
 		logrus.Errorf("gc task: failed to gc progress info taskID(%s): %v", taskID, err)
 	}
 	if err := gcm.taskMgr.Delete(ctx, taskID); err != nil {

supernode/daemon/mgr/progress/progress_delete.go

@@ -19,13 +19,24 @@ package progress
 import (
 	"context"
 
+	"github.com/dragonflyoss/Dragonfly/pkg/errortypes"
+
 	"github.com/sirupsen/logrus"
 )
 
 // DeleteTaskID deletes the super progress with specified taskID.
-func (pm *Manager) DeleteTaskID(ctx context.Context, taskID string) (err error) {
+func (pm *Manager) DeleteTaskID(ctx context.Context, taskID string, pieceTotal int) (err error) {
 	pm.superLoad.remove(taskID)
-	return pm.superProgress.remove(taskID)
+	pm.superProgress.remove(taskID)
+
+	for i := 0; i < pieceTotal; i++ {
+		key, err := generatePieceProgressKey(taskID, i)
+		if err != nil {
+			return err
+		}
+		pm.pieceProgress.remove(key)
+	}
+	return nil
 }
 
 // DeleteCID deletes the client progress with specified clientID.
@@ -81,9 +92,10 @@ func (pm *Manager) deletePeerIDByPieceNum(ctx context.Context, taskID string, pi
 // the peer no longer provides the service for the pieceNum of taskID.
 func (pm *Manager) deletePeerIDByPieceProgressKey(ctx context.Context, pieceProgressKey string, peerID string) error {
 	ps, err := pm.pieceProgress.getAsPieceState(pieceProgressKey)
-	if err != nil {
+	if err != nil && !errortypes.IsDataNotFound(err) {
 		return err
 	}
 
-	return ps.delete(peerID)
+	ps.delete(peerID)
+	return nil
 }

supernode/daemon/mgr/progress_mgr.go

@@ -88,7 +88,7 @@ type ProgressMgr interface {
 	UpdateSuperLoad(ctx context.Context, taskID string, delta, limit int32) (updated bool, err error)
 
 	// DeleteTaskID deletes the super progress with specified taskID.
-	DeleteTaskID(ctx context.Context, taskID string) (err error)
+	DeleteTaskID(ctx context.Context, taskID string, pieceTotal int) (err error)
 
 	// DeleteCID deletes the super progress with specified clientID.
 	DeleteCID(ctx context.Context, clientID string) (err error)

supernode/daemon/mgr/scheduler/manager.go

@@ -260,7 +260,7 @@ func (sm *Manager) tryGetPID(ctx context.Context, taskID string, pieceNum int, s
 
 func (sm *Manager) deletePeerIDByPieceNum(ctx context.Context, taskID string, pieceNum int, peerID string) {
 	if err := sm.progressMgr.DeletePeerIDByPieceNum(ctx, taskID, pieceNum, peerID); err != nil {
-		logrus.Warnf("scheduler: failed to delete the peerID %s for pieceNum %d of taskID: %s", peerID, pieceNum, taskID)
+		logrus.Warnf("scheduler: failed to delete the peerID %s for pieceNum %d of taskID: %s: %v", peerID, pieceNum, taskID, err)
 	}
 }