Merge pull request #392 from iExecBlockchainComputing/feature/final-deadline-handling

Feature/final deadline handling

Author: james-toussaint

src/main/java/com/iexec/core/configuration/CronConfiguration.java

@@ -29,6 +29,9 @@ public class CronConfiguration {
     @Value("${cron.detector.chain.finalize}")
     private int finalize;
 
+    @Value("${cron.detector.chain.final-deadline}")
+    private int finalDeadline;
+
     @Value("${cron.detector.timeout.contribute}")
     private int contributeTimeout;
 

src/main/java/com/iexec/core/detector/task/FinalDeadlineTaskDetector.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2020 IEXEC BLOCKCHAIN TECH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.iexec.core.detector.task;
+
+import com.iexec.core.detector.Detector;
+import com.iexec.core.task.Task;
+import com.iexec.core.task.TaskService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import java.util.Date;
+
+@Slf4j
+@Service
+public class FinalDeadlineTaskDetector implements Detector {
+
+    private final TaskService taskService;
+
+    public FinalDeadlineTaskDetector(TaskService taskService) {
+        this.taskService = taskService;
+    }
+
+    @Scheduled(fixedRateString = "#{@cronConfiguration.getFinalDeadline()}")
+    @Override
+    public void detect() {
+        log.debug("Trying to detect final deadline");
+        for (Task task : taskService.getTasksWhereFinalDeadlineIsPossible()) {
+            Date now = new Date();
+            if (now.after(task.getFinalDeadline())) {
+                log.info("Task after final deadline found [chainTaskId:{}]", task.getChainTaskId());
+                taskService.updateTask(task.getChainTaskId());
+            }
+        }
+    }
+}

src/main/java/com/iexec/core/replicate/ReplicateSupplyService.java

@@ -38,7 +38,6 @@
 import org.springframework.stereotype.Service;
 
 import java.util.*;
-import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 
 import static com.iexec.common.replicate.ReplicateStatus.*;
@@ -359,9 +358,7 @@ private Optional<TaskNotificationType> recoverReplicateInRevealPhase(Task task,
         if (didReplicateStartRevealing && didReplicateRevealOnChain) {
             ReplicateStatusDetails details = new ReplicateStatusDetails(blockNumber);
             replicatesService.updateReplicateStatus(chainTaskId, walletAddress, REVEALED, details);
-
-            CompletableFuture<Void> completableFuture = taskService.updateTask(chainTaskId);
-            completableFuture.join();
+            taskService.updateTask(chainTaskId).join();
         }
 
         // we read the replicate from db to consider the changes added in the previous case

src/main/java/com/iexec/core/task/TaskService.java

@@ -25,50 +25,48 @@
 import com.iexec.core.replicate.Replicate;
 import com.iexec.core.replicate.ReplicatesService;
 import com.iexec.core.task.event.*;
-import com.iexec.core.task.executor.TaskExecutorEngine;
+import com.iexec.core.task.update.TaskUpdateRequestConsumer;
+import com.iexec.core.task.update.TaskUpdateRequestManager;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.tuple.Pair;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
 
-import java.util.Arrays;
-import java.util.Date;
-import java.util.List;
-import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.*;
+import java.util.concurrent.*;
 import java.util.stream.Collectors;
 
 import static com.iexec.core.task.TaskStatus.*;
 
 @Slf4j
 @Service
-public class TaskService {
+public class TaskService implements TaskUpdateRequestConsumer {
 
     private final ConcurrentHashMap<String, Boolean>
             taskAccessForNewReplicateLock = new ConcurrentHashMap<>();
 
-    private TaskRepository taskRepository;
-    private TaskExecutorEngine taskExecutorEngine;
-    private IexecHubService iexecHubService;
-    private ReplicatesService replicatesService;
-    private ApplicationEventPublisher applicationEventPublisher;
-    private Web3jService web3jService;
+    private final TaskRepository taskRepository;
+    private final TaskUpdateRequestManager taskUpdateRequestManager;
+    private final IexecHubService iexecHubService;
+    private final ReplicatesService replicatesService;
+    private final ApplicationEventPublisher applicationEventPublisher;
+    private final Web3jService web3jService;
 
     public TaskService(
         TaskRepository taskRepository,
-        TaskExecutorEngine taskExecutorEngine,
+        TaskUpdateRequestManager taskUpdateRequestManager,
         IexecHubService iexecHubService,
         ReplicatesService replicatesService,
         ApplicationEventPublisher applicationEventPublisher,
         Web3jService web3jService
     ) {
         this.taskRepository = taskRepository;
-        this.taskExecutorEngine = taskExecutorEngine;
+        this.taskUpdateRequestManager = taskUpdateRequestManager;
         this.iexecHubService = iexecHubService;
         this.replicatesService = replicatesService;
         this.applicationEventPublisher = applicationEventPublisher;
         this.web3jService = web3jService;
+        this.taskUpdateRequestManager.setRequestConsumer(this);
     }
 
     /**
@@ -143,6 +141,10 @@ public List<Task> getTasksInNonFinalStatuses() {
         return taskRepository.findByCurrentStatusNotIn(TaskStatus.getFinalStatuses());
     }
 
+    public List<Task> getTasksWhereFinalDeadlineIsPossible() {
+        return taskRepository.findByCurrentStatusNotIn(TaskStatus.getStatusesWhereFinalDeadlineIsImpossible());
+    }
+
     public List<String> getChainTaskIdsOfTasksExpiredBefore(Date expirationDate) {
         return taskRepository.findChainTaskIdsByFinalDeadlineBefore(expirationDate)
                 .stream()
@@ -191,47 +193,22 @@ public Date getTaskFinalDeadline(String chainTaskId) {
     }
 
     /**
-     * Update task asynchronously.
-     * 
+     * Async method for requesting task update
      * @param chainTaskId
      * @return
      */
-    // TODO change this mechanism of update
-    public CompletableFuture<Void> updateTask(String chainTaskId) {
-        Optional<Task> oTask = getTaskByChainTaskId(chainTaskId);
-        if (oTask.isEmpty()) {
-            return CompletableFuture.completedFuture(null);
-        }
-        Task task = oTask.get();
-        Date finalDeadline = task.getFinalDeadline();
-        if (finalDeadline == null) {
-            log.error("Cannot update task without final deadline [chainTaskId:{}]",
-                    chainTaskId);
-            return CompletableFuture.completedFuture(null);
-        }
-        long expiration = finalDeadline.getTime() - new Date().getTime();
-        return taskExecutorEngine.run(
-                chainTaskId,
-                expiration,
-                () -> updateTaskRunnable(chainTaskId)
-        );
+    public CompletableFuture<Boolean> updateTask(String chainTaskId) {
+        return taskUpdateRequestManager.publishRequest(chainTaskId);
     }
 
     /**
-     * Remove task's executor if task is
-     * in final status.
-     * 
-     * @param task
+     * Async called when a task update request is received
+     * @param chainTaskId
      */
-    public void removeTaskExecutor(Task task) {
-        if (!TaskStatus.isFinalStatus(task.getCurrentStatus())) {
-            log.error("Cannot remove executor for unfinished " +
-                    "task [chainTaskId:{}]", task.getChainTaskId());
-            return;
-        }
-        taskExecutorEngine.removeExecutor(task.getChainTaskId());
-        log.info("Removed task executor [chainTaskId:{}]",
-                task.getChainTaskId());
+    @Override
+    public void onTaskUpdateRequest(String chainTaskId) {
+        log.info("Received task update request [chainTaskId:{}]", chainTaskId);
+        this.updateTaskRunnable(chainTaskId);
     }
 
     void updateTaskRunnable(String chainTaskId) {
@@ -240,8 +217,20 @@ void updateTaskRunnable(String chainTaskId) {
             return;
         }
         Task task = optional.get();
+        TaskStatus currentStatus = task.getCurrentStatus();
+
+        boolean isFinalDeadlinePossible =
+                !TaskStatus.getStatusesWhereFinalDeadlineIsImpossible().contains(currentStatus);
+        if (isFinalDeadlinePossible && new Date().after(task.getFinalDeadline())){
+            updateTaskStatusAndSave(task, FINAL_DEADLINE_REACHED);
+            // Eventually should fire a "final deadline reached" notification to worker,
+            // but here let's just trigger an updateTask() leading to a failed status
+            // which will itself fire a generic "abort" notification
+            updateTask(chainTaskId);
+            return;
+        }
 
-        switch (task.getCurrentStatus()) {
+        switch (currentStatus) {
             case RECEIVED:
                 received2Initialized(task);
                 break;
@@ -303,9 +292,11 @@ void updateTaskRunnable(String chainTaskId) {
             case FINALIZE_FAILED:
                 toFailed(task);
                 break;
+            case FINAL_DEADLINE_REACHED:
+                toFailed(task);
+                break;
             case COMPLETED:
             case FAILED:
-                removeTaskExecutor(task);
                 break;
         }
     }
@@ -692,6 +683,7 @@ private void finalizedToCompleted(Task task) {
 
     private void toFailed(Task task) {
         updateTaskStatusAndSave(task, FAILED);
+        applicationEventPublisher.publishEvent(new TaskFailedEvent(task.getChainTaskId()));
     }
     public void initializeTaskAccessForNewReplicateLock(String chainTaskId) {
         taskAccessForNewReplicateLock.putIfAbsent(chainTaskId, false);
@@ -713,4 +705,5 @@ private void setTaskAccessForNewReplicateLock(String chainTaskId, boolean isTask
         taskAccessForNewReplicateLock.replace(chainTaskId, isTaskBeingAccessedForNewReplicate);
     }
 
+
 }

src/main/java/com/iexec/core/task/TaskStatus.java

@@ -16,7 +16,9 @@
 
 package com.iexec.core.task;
 
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 
 public enum TaskStatus {
@@ -39,6 +41,7 @@ public enum TaskStatus {
     FINALIZING,
     FINALIZED,
     FINALIZE_FAILED,
+    FINAL_DEADLINE_REACHED,
     COMPLETED,
     FAILED;
 
@@ -65,6 +68,12 @@ public static List<TaskStatus> getFinalStatuses() {
         return List.of(FAILED, COMPLETED);
     }
 
+    public static List<TaskStatus> getStatusesWhereFinalDeadlineIsImpossible() {
+        List<TaskStatus> excludedStatuses = new ArrayList<>(getFinalStatuses());
+        Collections.addAll(excludedStatuses, FINAL_DEADLINE_REACHED);
+        return excludedStatuses;
+    }
+
     public static boolean isInContributionPhase(TaskStatus status) {
         return getWaitingContributionStatuses().contains(status);
     }

src/main/java/com/iexec/core/task/event/TaskFailedEvent.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2020 IEXEC BLOCKCHAIN TECH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.iexec.core.task.event;
+
+import lombok.*;
+
+@AllArgsConstructor
+@Getter
+public class TaskFailedEvent {
+
+    private String chainTaskId;
+}