Only detect worker lost when possible

Author: Jérémy James Toussaint

src/main/java/com/iexec/core/detector/WorkerLostDetector.java

@@ -16,6 +16,8 @@
 
 package com.iexec.core.detector;
 
+import com.google.common.collect.ImmutableSet;
+import com.iexec.common.replicate.ReplicateStatus;
 import com.iexec.core.replicate.ReplicatesService;
 import com.iexec.core.task.TaskService;
 import com.iexec.core.worker.Worker;
@@ -30,14 +32,20 @@
 @Slf4j
 public class WorkerLostDetector implements Detector {
 
+    public static final ImmutableSet<ReplicateStatus> SHOULD_NOT_UPDATE_STATUSES =
+            new ImmutableSet.Builder<ReplicateStatus>()
+                    .add(WORKER_LOST)
+                    .addAll(ReplicateStatus.getFinalStatuses())
+                    .build();
+
     private final ReplicatesService replicatesService;
     private final WorkerService workerService;
     private final TaskService taskService;
 
     public WorkerLostDetector(
-        ReplicatesService replicatesService,
-        WorkerService workerService,
-        TaskService taskService
+            ReplicatesService replicatesService,
+            WorkerService workerService,
+            TaskService taskService
     ) {
         this.replicatesService = replicatesService;
         this.workerService = workerService;
@@ -57,13 +65,13 @@ public void detect() {
                 replicatesService
                         .getReplicate(chainTaskId, workerWallet)
                         .ifPresent(replicate -> {
-                                if (!replicate.getCurrentStatus().equals(WORKER_LOST)) {
-                                    replicatesService.updateReplicateStatus(
+                            if (!SHOULD_NOT_UPDATE_STATUSES.contains(replicate.getCurrentStatus())) {
+                                replicatesService.updateReplicateStatus(
                                         chainTaskId,
                                         workerWallet,
                                         WORKER_LOST
-                                    );
-                                }
+                                );
+                            }
                         });
             }
         }

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

@@ -34,11 +34,10 @@
 @Component
 public class ReplicateListeners {
 
-    private TaskService taskService;
-    private WorkerService workerService;
-    private ContributionUnnotifiedDetector contributionUnnotifiedDetector;
-    private ReplicatesService replicatesService;
-
+    private final TaskService taskService;
+    private final WorkerService workerService;
+    private final ContributionUnnotifiedDetector contributionUnnotifiedDetector;
+    private final ReplicatesService replicatesService;
 
     public ReplicateListeners(WorkerService workerService,
                               TaskService taskService,
@@ -55,10 +54,13 @@ public void onReplicateUpdatedEvent(ReplicateUpdatedEvent event) {
         log.debug("Received ReplicateUpdatedEvent [chainTaskId:{}] ", event.getChainTaskId());
         ReplicateStatusUpdate statusUpdate = event.getReplicateStatusUpdate();
         ReplicateStatus newStatus = statusUpdate.getStatus();
-        ReplicateStatusCause cause = statusUpdate.getDetails() != null ? statusUpdate.getDetails().getCause(): null;
+        ReplicateStatusCause cause = statusUpdate.getDetails() != null ? statusUpdate.getDetails().getCause() : null;
 
         taskService.updateTask(event.getChainTaskId());
 
+        /*
+         * Should release 1 CPU of given worker for this replicate if status is COMPUTED
+         * */
         if (newStatus.equals(ReplicateStatus.COMPUTED)) {
             workerService.removeComputedChainTaskIdFromWorker(event.getChainTaskId(), event.getWalletAddress());
         }
@@ -83,9 +85,9 @@ public void onReplicateUpdatedEvent(ReplicateUpdatedEvent event) {
         }
 
         /*
-         * Should release one CPU for this replicate if status is FAILED
+         * Should release given worker for this replicate if status is COMPLETED or FAILED
          * */
-        if (newStatus.equals(ReplicateStatus.FAILED)) {
+        if (ReplicateStatus.getFinalStatuses().contains(newStatus)) {
             workerService.removeChainTaskIdFromWorker(event.getChainTaskId(), event.getWalletAddress());
         }
     }

src/test/java/com/iexec/core/detector/WorkerLostDetectorTests.java

@@ -37,8 +37,7 @@
 
 import static com.iexec.common.utils.DateTimeUtils.addMinutesToDate;
 import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
 
 public class WorkerLostDetectorTests {
 
@@ -97,12 +96,9 @@ public void shouldUpdateOneReplicateToWorkerLost(){
 
     // similar test with previous except that the Replicate is already is WORKER_LOST status.
     @Test
-    public void shouldNotUpdateOneReplicateToWorkerLostSinceAlreadyUpdated(){
-        Date twoMinutesAgo = addMinutesToDate(new Date(), -2);
-
+    public void shouldNotUpdateToWorkerLostSinceAlreadyUpdated(){
         Worker worker = Worker.builder()
                 .walletAddress(WALLET_WORKER)
-                .lastAliveDate(twoMinutesAgo)
                 .participatingChainTaskIds(Collections.singletonList(CHAIN_TASK_ID))
                 .build();
 
@@ -115,13 +111,48 @@ public void shouldNotUpdateOneReplicateToWorkerLostSinceAlreadyUpdated(){
         when(taskService.isExpired(anyString())).thenReturn(false);
 
         workerLostDetector.detect();
-        // verify that the call on the update is correct
-        Mockito.verify(replicatesService, Mockito.times(0))
+        Mockito.verify(replicatesService, never())
                 .updateReplicateStatus(CHAIN_TASK_ID, WALLET_WORKER, ReplicateStatus.WORKER_LOST);
+    }
 
-        // verify that the worker should remove the taskId from its current tasks
-        Mockito.verify(workerService, Mockito.times(0))
-                .removeChainTaskIdFromWorker(CHAIN_TASK_ID, WALLET_WORKER);
+    @Test
+    public void shouldNotUpdateToWorkerLostSinceFailed(){
+        Worker worker = Worker.builder()
+                .walletAddress(WALLET_WORKER)
+                .participatingChainTaskIds(Collections.singletonList(CHAIN_TASK_ID))
+                .build();
+
+        Replicate replicate = new Replicate(WALLET_WORKER, CHAIN_TASK_ID);
+        replicate.updateStatus(ReplicateStatus.STARTING, ReplicateStatusModifier.WORKER);
+        replicate.updateStatus(ReplicateStatus.FAILED, ReplicateStatusModifier.POOL_MANAGER);
+
+        when(workerService.getLostWorkers()).thenReturn(Collections.singletonList(worker));
+        when(replicatesService.getReplicate(CHAIN_TASK_ID, WALLET_WORKER)).thenReturn(Optional.of(replicate));
+        when(taskService.isExpired(anyString())).thenReturn(false);
+
+        workerLostDetector.detect();
+        Mockito.verify(replicatesService, never())
+                .updateReplicateStatus(CHAIN_TASK_ID, WALLET_WORKER, ReplicateStatus.WORKER_LOST);
+    }
+
+    @Test
+    public void shouldNotUpdateToWorkerLostSinceCompleted(){
+        Worker worker = Worker.builder()
+                .walletAddress(WALLET_WORKER)
+                .participatingChainTaskIds(Collections.singletonList(CHAIN_TASK_ID))
+                .build();
+
+        Replicate replicate = new Replicate(WALLET_WORKER, CHAIN_TASK_ID);
+        replicate.updateStatus(ReplicateStatus.STARTING, ReplicateStatusModifier.WORKER);
+        replicate.updateStatus(ReplicateStatus.COMPLETED, ReplicateStatusModifier.POOL_MANAGER);
+
+        when(workerService.getLostWorkers()).thenReturn(Collections.singletonList(worker));
+        when(replicatesService.getReplicate(CHAIN_TASK_ID, WALLET_WORKER)).thenReturn(Optional.of(replicate));
+        when(taskService.isExpired(anyString())).thenReturn(false);
+
+        workerLostDetector.detect();
+        Mockito.verify(replicatesService, never())
+                .updateReplicateStatus(CHAIN_TASK_ID, WALLET_WORKER, ReplicateStatus.WORKER_LOST);
     }
 
     @Test

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

@@ -145,7 +145,21 @@ public void shouldNotAddFailedStatusSinceCompletableReplicateStatus() {
     }
 
     @Test
-    public void shouldRemoveComputedChainTaskIdFromWorkerSinceFailed() {
+    public void shouldRemoveChainTaskIdFromWorkerSinceCompleted() {
+        ReplicateUpdatedEvent replicateUpdatedEvent = ReplicateUpdatedEvent.builder()
+                .chainTaskId(CHAIN_TASK_ID)
+                .walletAddress(WORKER_WALLET)
+                .replicateStatusUpdate(new ReplicateStatusUpdate(COMPLETED))
+                .build();
+
+        replicateListeners.onReplicateUpdatedEvent(replicateUpdatedEvent);
+
+        Mockito.verify(workerService, Mockito.times(1))
+                .removeChainTaskIdFromWorker(CHAIN_TASK_ID, WORKER_WALLET);
+    }
+
+    @Test
+    public void shouldRemoveChainTaskIdFromWorkerSinceFailed() {
         ReplicateUpdatedEvent replicateUpdatedEvent = ReplicateUpdatedEvent.builder()
                 .chainTaskId(CHAIN_TASK_ID)
                 .walletAddress(WORKER_WALLET)
@@ -159,8 +173,9 @@ public void shouldRemoveComputedChainTaskIdFromWorkerSinceFailed() {
     }
 
     @Test
-    public void shouldNotRemoveComputedChainTaskIdFromWorker() {
+    public void shouldNotRemoveChainTaskIdFromWorker() {
         List<ReplicateStatus> someStatuses = ReplicateStatus.getSuccessStatuses(); //not exhaustive
+        someStatuses.remove(COMPLETED);
         someStatuses.remove(FAILED);
 
         for (ReplicateStatus randomStatus: someStatuses){