Merge pull request #201 from iExecBlockchainComputing/notification-type-for-recovery

Notification type for recovery

Author: james-toussaint

src/main/java/com/iexec/worker/amnesia/AmnesiaRecoveryService.java

@@ -1,27 +1,23 @@
 package com.iexec.worker.amnesia;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Optional;
-
 import com.iexec.common.chain.ContributionAuthorization;
-import com.iexec.common.replicate.AvailableReplicateModel;
-import com.iexec.common.disconnection.InterruptedReplicateModel;
-import com.iexec.common.disconnection.RecoveryAction;
+import com.iexec.common.notification.TaskNotification;
+import com.iexec.common.notification.TaskNotificationType;
+import com.iexec.common.task.TaskDescription;
 import com.iexec.worker.chain.IexecHubService;
-import com.iexec.worker.executor.TaskExecutorService;
 import com.iexec.worker.feign.CustomFeignClient;
 import com.iexec.worker.pubsub.SubscriptionService;
-import com.iexec.worker.replicate.ReplicateService;
 import com.iexec.worker.result.ResultService;
-
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
-import lombok.extern.slf4j.Slf4j;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
 
 
-/* 
+/*
  * This service is used to remind the worker of possible interrupted works
  * after a restart and how to deal with each interruption
  */
@@ -31,140 +27,61 @@ public class AmnesiaRecoveryService {
 
     private CustomFeignClient customFeignClient;
     private SubscriptionService subscriptionService;
-    private ReplicateService replicateService;
     private ResultService resultService;
-    private TaskExecutorService taskExecutorService;
     private IexecHubService iexecHubService;
 
     public AmnesiaRecoveryService(CustomFeignClient customFeignClient,
                                   SubscriptionService subscriptionService,
-                                  ReplicateService replicateService,
                                   ResultService resultService,
-                                  TaskExecutorService taskExecutorService,
                                   IexecHubService iexecHubService) {
         this.customFeignClient = customFeignClient;
         this.subscriptionService = subscriptionService;
-        this.replicateService = replicateService;
         this.resultService = resultService;
-        this.taskExecutorService = taskExecutorService;
         this.iexecHubService = iexecHubService;
     }
 
     public List<String> recoverInterruptedReplicates() {
-        long lasAvailableBlockNumber = iexecHubService.getLatestBlockNumber();
-        List<InterruptedReplicateModel> interruptedReplicates = customFeignClient.getInterruptedReplicates(
-                lasAvailableBlockNumber);
+        long latestAvailableBlockNumber = iexecHubService.getLatestBlockNumber();
+        List<TaskNotification> missedTaskNotifications = customFeignClient.getMissedTaskNotifications(
+                latestAvailableBlockNumber);
         List<String> recoveredChainTaskIds = new ArrayList<>();
 
-        if (interruptedReplicates == null || interruptedReplicates.isEmpty()) {
+        if (missedTaskNotifications == null || missedTaskNotifications.isEmpty()) {
             log.info("No interrupted tasks to recover");
             return Collections.emptyList();
         }
 
-        for (InterruptedReplicateModel interruptedReplicate : interruptedReplicates) {
+        for (TaskNotification missedTaskNotification : missedTaskNotifications) {
+            TaskNotificationType taskNotificationType = missedTaskNotification.getTaskNotificationType();
+            String chainTaskId = missedTaskNotification.getChainTaskId();
+            boolean isResultAvailable = resultService.isResultAvailable(chainTaskId);
 
-            ContributionAuthorization contributionAuth = interruptedReplicate.getContributionAuthorization();
-            RecoveryAction recoveryAction = interruptedReplicate.getRecoveryAction();
-            String chainTaskId = contributionAuth.getChainTaskId();
-            boolean isResultAvailable = isResultAvailable(chainTaskId);
+            log.info("Recovering interrupted task [chainTaskId:{}, taskNotificationType:{}]",
+                    chainTaskId, taskNotificationType);
 
-            log.info("Recovering interrupted task [chainTaskId:{}, recoveryAction:{}]",
-                    chainTaskId, recoveryAction);
-
-            if (!isResultAvailable && recoveryAction != RecoveryAction.CONTRIBUTE) {
-                log.error("Could not recover task, result not found [chainTaskId:{}, RecoveryAction:{}]",
-                        chainTaskId, recoveryAction);
+            if (!isResultAvailable && taskNotificationType != TaskNotificationType.PLEASE_CONTRIBUTE) {
+                log.error("Could not recover task, result not found [chainTaskId:{}, taskNotificationType:{}]",
+                        chainTaskId, taskNotificationType);
                 continue;
             }
 
-            Optional<AvailableReplicateModel> oReplicateModel =
-                    replicateService.retrieveAvailableReplicateModelFromContribAuth(contributionAuth);
+            Optional<TaskDescription> optionalTaskDescription = iexecHubService.getTaskDescriptionFromChain(chainTaskId);
 
-            if (!oReplicateModel.isPresent()) {
-                log.error("Could not recover task, no replicateModel retrieved [chainTaskId:{}, RecoveryAction:{}]",
-                        chainTaskId, recoveryAction);
+            if (!optionalTaskDescription.isPresent()) {
+                log.error("Could not recover task, no TaskDescription retrieved [chainTaskId:{}, taskNotificationType:{}]",
+                        chainTaskId, taskNotificationType);
                 continue;
             }
 
-            AvailableReplicateModel replicateModel = oReplicateModel.get();
-            recoverReplicate(interruptedReplicate, replicateModel);
-            recoveredChainTaskIds.add(chainTaskId);
-        }
+            TaskDescription taskDescription = optionalTaskDescription.get();
 
-        return recoveredChainTaskIds;
-    }
-
-    public void recoverReplicate(InterruptedReplicateModel interruptedReplicate,
-                                 AvailableReplicateModel replicateModel) {
-
-        ContributionAuthorization contributionAuth = interruptedReplicate.getContributionAuthorization();
-        String chainTaskId = contributionAuth.getChainTaskId();
-
-        switch (interruptedReplicate.getRecoveryAction()) {
-            case WAIT:
-                subscriptionService.subscribeToTopic(chainTaskId);
-                resultService.saveResultInfo(chainTaskId, replicateModel);
-                break;
-
-            case CONTRIBUTE:
-                subscriptionService.subscribeToTopic(chainTaskId);
-                recoverReplicateByContributing(contributionAuth, replicateModel);
-                break;
-
-            case ABORT_CONSENSUS_REACHED:
-                taskExecutorService.abortConsensusReached(chainTaskId);
-                break;
-
-            case ABORT_CONTRIBUTION_TIMEOUT:
-                taskExecutorService.abortContributionTimeout(chainTaskId);
-                break;
-
-            case REVEAL:
-                subscriptionService.subscribeToTopic(chainTaskId);
-                resultService.saveResultInfo(chainTaskId, replicateModel);
-                taskExecutorService.reveal(chainTaskId, iexecHubService.getLatestBlockNumber());
-                break;
-
-            case UPLOAD_RESULT:
-                subscriptionService.subscribeToTopic(chainTaskId);
-                resultService.saveResultInfo(chainTaskId, replicateModel);
-                taskExecutorService.uploadResult(chainTaskId);
-                break;
-
-            case COMPLETE:
-                taskExecutorService.completeTask(chainTaskId);
-                break;
-
-            default:
-                break;
-        }
-    }
-
-    private boolean isResultAvailable(String chainTaskId) {
-        boolean isResultZipFound = resultService.isResultZipFound(chainTaskId);
-        boolean isResultFolderFound = resultService.isResultFolderFound(chainTaskId);
-
-        if (!isResultZipFound && !isResultFolderFound) return false;
+            subscriptionService.subscribeToTopic(chainTaskId);
+            resultService.saveResultInfo(chainTaskId, taskDescription);
+            subscriptionService.handleTaskNotification(missedTaskNotification);
 
-        if (!isResultZipFound) resultService.zipResultFolder(chainTaskId);
-
-        return true;
-    }
-
-    public void recoverReplicateByContributing(ContributionAuthorization contributionAuth,
-                                               AvailableReplicateModel replicateModel) {
-
-        String chainTaskId = contributionAuth.getChainTaskId();
-        boolean isResultAvailable = isResultAvailable(chainTaskId);
-
-        if (!isResultAvailable) {
-            log.info("Result not found, re-running computation to recover task " +
-                    "[chainTaskId:{}, recoveryAction:CONTRIBUTE]", chainTaskId);
-            taskExecutorService.addReplicate(replicateModel);
-            return;
+            recoveredChainTaskIds.add(chainTaskId);
         }
 
-        resultService.saveResultInfo(chainTaskId, replicateModel);
-        taskExecutorService.contribute(contributionAuth);
+        return recoveredChainTaskIds;
     }
 }

src/main/java/com/iexec/worker/docker/DockerComputationService.java

@@ -1,6 +1,6 @@
 package com.iexec.worker.docker;
 
-import com.iexec.common.replicate.AvailableReplicateModel;
+import com.iexec.common.task.TaskDescription;
 import com.iexec.worker.config.WorkerConfigurationService;
 import com.spotify.docker.client.messages.ContainerConfig;
 import lombok.extern.slf4j.Slf4j;
@@ -28,9 +28,9 @@ public DockerComputationService(CustomDockerClient dockerClient,
         this.configurationService = configurationService;
     }
 
-    public String dockerRunAndGetLogs(AvailableReplicateModel replicateModel, String datasetFilename) {
-        String chainTaskId = replicateModel.getContributionAuthorization().getChainTaskId();
-        String image = replicateModel.getAppUri();
+    public String dockerRunAndGetLogs(TaskDescription taskDescription, String datasetFilename) {
+        String chainTaskId = taskDescription.getChainTaskId();
+        String image = taskDescription.getAppUri();
         //TODO: check image equals image:tag
         String stdout = "";
 
@@ -41,17 +41,17 @@ public String dockerRunAndGetLogs(AvailableReplicateModel replicateModel, String
         String hostBaseVolume = configurationService.getTaskBaseDir(chainTaskId);
         ContainerConfig containerConfig;
 
-        if (replicateModel.isTrustedExecution()) {
-            containerConfig = getContainerConfig(image, replicateModel.getCmd(), hostBaseVolume,
+        if (taskDescription.isTrustedExecution()) {
+            containerConfig = getContainerConfig(image, taskDescription.getCmd(), hostBaseVolume,
                     TEE_DOCKER_ENV_CHAIN_TASKID + "=" + chainTaskId,
                     TEE_DOCKER_ENV_WORKER_ADDRESS + "=" + configurationService.getWorkerWalletAddress(),
                     DATASET_FILENAME + "=" + datasetFilename);
         } else {
-            containerConfig = getContainerConfig(image, replicateModel.getCmd(), hostBaseVolume,
+            containerConfig = getContainerConfig(image, taskDescription.getCmd(), hostBaseVolume,
                     DATASET_FILENAME + "=" + datasetFilename);
         }
 
-        stdout = startComputationAndGetLogs(chainTaskId, containerConfig, replicateModel.getMaxExecutionTime());
+        stdout = startComputationAndGetLogs(chainTaskId, containerConfig, taskDescription.getMaxExecutionTime());
 
         return stdout;
     }