fix: only trigger task status update after successful replicate status when relevant (#760)

Author: jbern0rd

gradle.properties

@@ -1,8 +1,8 @@
 # x-release-please-start-version
 version=9.0.0
 # x-release-please-end
-iexecCommonsPocoVersion=5.0.0
-iexecCommonVersion=9.0.0
+iexecCommonsPocoVersion=5.1.0
+iexecCommonVersion=9.1.0
 iexecBlockchainAdapterVersion=9.0.0
 iexecResultVersion=9.0.0
 iexecSmsVersion=9.0.0

src/main/java/com/iexec/core/replicate/listener/ReplicateListeners.java

@@ -28,7 +28,9 @@
 import org.springframework.context.event.EventListener;
 import org.springframework.stereotype.Component;
 
-import static com.iexec.common.replicate.ReplicateStatus.FAILED;
+import java.util.List;
+
+import static com.iexec.common.replicate.ReplicateStatus.*;
 import static com.iexec.common.replicate.ReplicateStatusCause.TASK_NOT_ACTIVE;
 
 @Slf4j
@@ -51,23 +53,31 @@ public ReplicateListeners(final WorkerService workerService,
     }
 
     @EventListener
-    public void onReplicateUpdatedEvent(ReplicateUpdatedEvent event) {
-        log.debug("Received ReplicateUpdatedEvent [chainTaskId:{}] ", event.getChainTaskId());
+    public void onReplicateUpdatedEvent(final ReplicateUpdatedEvent event) {
+        log.debug("Received ReplicateUpdatedEvent [chainTaskId:{}, workerAddress:{}, status:{}]",
+                event.getChainTaskId(), event.getWalletAddress(), event.getReplicateStatusUpdate().getStatus());
         final ReplicateStatusUpdate statusUpdate = event.getReplicateStatusUpdate();
         final ReplicateStatus newStatus = statusUpdate.getStatus();
         final ReplicateStatusCause cause = statusUpdate.getDetails() != null ? statusUpdate.getDetails().getCause() : null;
 
-        taskUpdateRequestManager.publishRequest(event.getChainTaskId());
+        // Those are the only transitions justifying to update a task status
+        // When one worker updates its replicate status to STARTED, its task status can be updated to RUNNING
+        // When one worker updates its replicate status to RESULT_UPLOADED, its task status can be updated to RESULT_UPLOADED
+        // The other statuses denote major PoCo events allowing to see an update of the associated task status
+        // The check against isFailedBeforeComputed allows to trigger the running2runningFailed check in TaskUpdateManager
+        // This protection allows to avoid unnecessary loops in update manager as well as unnecessary on-chain calls
+        if (List.of(STARTED, CONTRIBUTE_AND_FINALIZE_DONE, CONTRIBUTED, REVEALED, RESULT_UPLOADED).contains(newStatus)
+                || newStatus.isFailedBeforeComputed()) {
+            taskUpdateRequestManager.publishRequest(event.getChainTaskId());
+        }
 
         /*
          * Should release 1 CPU of given worker for this replicate if status is
          * "COMPUTED" or "*_FAILED" before COMPUTED
          */
-        if (newStatus == ReplicateStatus.START_FAILED
-                || newStatus == ReplicateStatus.APP_DOWNLOAD_FAILED
-                || newStatus == ReplicateStatus.DATA_DOWNLOAD_FAILED
-                || newStatus == ReplicateStatus.COMPUTED
-                || newStatus == ReplicateStatus.COMPUTE_FAILED) {
+        if (newStatus.isFailedBeforeComputed() || newStatus == ReplicateStatus.COMPUTED) {
+            log.info("End of replicate computation detected [chainTaskId:{}, workerAddress:{}]",
+                    event.getChainTaskId(), event.getWalletAddress());
             workerService.removeComputedChainTaskIdFromWorker(event.getChainTaskId(), event.getWalletAddress());
         }
 

src/main/java/com/iexec/core/task/update/TaskUpdateManager.java

@@ -45,7 +45,6 @@
 import java.util.Date;
 import java.util.List;
 import java.util.Optional;
-import java.util.function.Predicate;
 
 import static com.iexec.core.task.TaskStatus.*;
 
@@ -496,20 +495,19 @@ private void running2RunningFailed(final Task task, final ReplicatesList replica
         }
 
         // If not all alive workers have failed while running the task, that's not a running failure.
-        boolean notAllReplicatesFailed = replicatesOfAliveWorkers
+        // FIXME use lambda when deprecated code has been removed in iexec-common
+        final boolean allReplicatesFailed = replicatesOfAliveWorkers
                 .stream()
                 .map(Replicate::getLastRelevantStatus)
-                .anyMatch(Predicate.not(ReplicateStatus::isFailedBeforeComputed));
-
-        if (notAllReplicatesFailed) {
-            return;
-        }
+                .allMatch(replicateStatus -> replicateStatus.isFailedBeforeComputed());
 
         // If all alive workers have failed on this task, its computation should be stopped.
         // It could denote that the task is wrong
         // - e.g. failing script, dataset can't be retrieved, app can't be downloaded, ...
-        updateTaskStatusesAndSave(task, RUNNING_FAILED, FAILED);
-        applicationEventPublisher.publishEvent(new TaskRunningFailedEvent(task.getChainTaskId()));
+        if (allReplicatesFailed) {
+            updateTaskStatusesAndSave(task, RUNNING_FAILED, FAILED);
+            applicationEventPublisher.publishEvent(new TaskRunningFailedEvent(task.getChainTaskId()));
+        }
     }
 
     // endregion

src/test/java/com/iexec/core/replicate/listener/ReplicateListenersTests.java

@@ -58,19 +58,33 @@ class ReplicateListenersTests {
     private ReplicatesService replicatesService;
     @Mock
     private TaskUpdateRequestManager taskUpdateRequestManager;
-
     @InjectMocks
     private ReplicateListeners replicateListeners;
 
+    private final List<ReplicateStatus> statusesTriggeringTaskUpdate = List.of(
+            STARTED, CONTRIBUTE_AND_FINALIZE_DONE, CONTRIBUTED, REVEALED, RESULT_UPLOADED,
+            START_FAILED, APP_DOWNLOAD_FAILED, DATA_DOWNLOAD_FAILED, COMPUTE_FAILED);
+
     @Test
     void shouldUpdateTaskOnReplicateUpdate() {
-        final List<ReplicateStatus> someStatuses = ReplicateStatus.getSuccessStatuses(); //not exhaustive
+        statusesTriggeringTaskUpdate.stream()
+                .map(this::getMockReplicate)
+                .forEach(replicateListeners::onReplicateUpdatedEvent);
+
+        verify(taskUpdateRequestManager, times(statusesTriggeringTaskUpdate.size())).publishRequest(any());
+    }
+
+    @Test
+    void shouldNotUpdateTaskOnReplicateUpdate() {
+        final List<ReplicateStatus> nonTriggeringStatuses = Arrays.stream(ReplicateStatus.values())
+                .filter(status -> !statusesTriggeringTaskUpdate.contains(status))
+                .toList();
 
-        someStatuses.stream()
+        nonTriggeringStatuses.stream()
                 .map(this::getMockReplicate)
                 .forEach(replicateListeners::onReplicateUpdatedEvent);
 
-        verify(taskUpdateRequestManager, times(someStatuses.size())).publishRequest(any());
+        verifyNoInteractions(taskUpdateRequestManager);
     }
 
     @Test