Merge pull request #11498 from filecoin-project/fix/harmony-reclaim

harmony: Fix task reclaim on restart

lib/harmony/harmonytask/harmonytask.go

@@ -176,7 +176,7 @@ func New(
 					continue // not really fatal, but not great
 				}
 			}
-			if !h.considerWork("recovered", []TaskID{TaskID(w.ID)}) {
+			if !h.considerWork(workSourceRecover, []TaskID{TaskID(w.ID)}) {
 				log.Error("Strange: Unable to accept previously owned task: ", w.ID, w.Name)
 			}
 		}
@@ -285,7 +285,7 @@ func (e *TaskEngine) pollerTryAllWork() {
 			continue
 		}
 		if len(unownedTasks) > 0 {
-			accepted := v.considerWork("poller", unownedTasks)
+			accepted := v.considerWork(workSourcePoller, unownedTasks)
 			if accepted {
 				return // accept new work slowly and in priority order
 			}

lib/harmony/harmonytask/task_type_handler.go

@@ -49,6 +49,11 @@ func (h *taskTypeHandler) AddTask(extra func(TaskID, *harmonydb.Tx) (bool, error
 	}
 }
 
+const (
+	workSourcePoller  = "poller"
+	workSourceRecover = "recovered"
+)
+
 // considerWork is called to attempt to start work on a task-id of this task type.
 // It presumes single-threaded calling, so there should not be a multi-threaded re-entry.
 // The only caller should be the one work poller thread. This does spin off other threads,
@@ -87,6 +92,8 @@ top:
 		return false
 	}
 
+	// if recovering we don't need to try to claim anything because those tasks are already claimed by us
+	if from != workSourceRecover {
 		// 4. Can we claim the work for our hostname?
 		ct, err := h.TaskEngine.db.Exec(h.TaskEngine.ctx, "UPDATE harmony_task SET owner_id=$1 WHERE id=$2 AND owner_id IS NULL", h.TaskEngine.ownerID, *tID)
 		if err != nil {
@@ -104,6 +111,7 @@ top:
 			ids = tryAgain
 			goto top
 		}
+	}
 
 	h.Count.Add(1)
 	go func() {