Add synchronized keyword on abort method to avoid concurrency issues on file system (#643)

Author: jbern0rd

CHANGELOG.md

@@ -10,13 +10,18 @@ All notable changes to this project will be documented in this file.
 - Validate authorization proof for pre/post-compute requests. (#635)
 - Add `WebSocketBlockchainListener` to fetch latest block without polling the blockchain network. (#639)
 
+### Bug Fixes
+
+- Add synchronized keyword on abort method to avoid concurrency issues on file system. (#643)
+
 ### Quality
 
 - Refactor `RestTemplateConfig` to use `HttpClient 5` and improve proxy handling. (#626)
 - Replace deprecated `connect` with `connectAsync` in `StompClientService`. (#627)
 - Remove redundant blockchain calls to diminish pressure on Ethereum JSON-RPC API. (#632)
 - Stop using `TestUtils` in `ContributionServiceTests`. (#640)
 - Fix several issues raised by SonarQube Cloud. (#642)
+- Improve JavaDoc comments in `ComputeManagerService` and `DockerService`. (#643)
 
 ### Breaking API changes
 

src/main/java/com/iexec/worker/compute/ComputeManagerService.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2024 IEXEC BLOCKCHAIN TECH
+ * Copyright 2020-2025 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.
@@ -41,7 +41,6 @@
 import java.time.temporal.ChronoUnit;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.function.Predicate;
 
 @Slf4j
 @Service
@@ -77,6 +76,15 @@ public ComputeManagerService(
         this.resultService = resultService;
     }
 
+    /**
+     * Download OCI image of the application to execute.
+     * <p>
+     * The download fails for a bad task description or if a timeout is reached.
+     * The timeout is computed by calling {@link #computeImagePullTimeout(TaskDescription)}.
+     *
+     * @param taskDescription Task description containing application type and download URI
+     * @return true if download succeeded, false otherwise
+     */
     public boolean downloadApp(TaskDescription taskDescription) {
         if (taskDescription == null || taskDescription.getAppType() == null) {
             return false;
@@ -88,8 +96,9 @@ public boolean downloadApp(TaskDescription taskDescription) {
         }
 
         final long pullTimeout = computeImagePullTimeout(taskDescription);
-        return dockerService.getClient(taskDescription.getAppUri())
+        dockerService.getClient(taskDescription.getAppUri())
                 .pullImage(taskDescription.getAppUri(), Duration.of(pullTimeout, ChronoUnit.MINUTES));
+        return dockerService.getClient(taskDescription.getAppUri()).isImagePresent(taskDescription.getAppUri());
     }
 
     /**
@@ -139,35 +148,45 @@ public boolean isAppDownloaded(String imageUri) {
     }
 
     /**
-     * Standard tasks: download secrets && decrypt dataset (TODO: rewritte or remove)
-     * <p>
+     * Execute pre-compute stage for standard and TEE tasks.
+     * <ul>
+     * <li>Standard tasks: Nothing is executed, an empty result is returned
+     * <li>TEE tasks: Call {@link PreComputeService#runTeePreCompute(TaskDescription, WorkerpoolAuthorization)}
+     * </ul>
      * TEE tasks: download pre-compute and post-compute images,
      * create SCONE secure session, and run pre-compute container.
      *
-     * @param taskDescription
-     * @param workerpoolAuth
-     * @return
+     * @param taskDescription Description of the task
+     * @param workerpoolAuth  Authorization to contribute delivered by the scheduler for the given task
+     * @return {@code PreComputeResponse} instance
+     * @see PreComputeService#runTeePreCompute(TaskDescription, WorkerpoolAuthorization)
      */
-    public PreComputeResponse runPreCompute(TaskDescription taskDescription,
-                                            WorkerpoolAuthorization workerpoolAuth) {
+    public PreComputeResponse runPreCompute(final TaskDescription taskDescription,
+                                            final WorkerpoolAuthorization workerpoolAuth) {
         log.info("Running pre-compute [chainTaskId:{}, isTee:{}]",
-                taskDescription.getChainTaskId(),
-                taskDescription.isTeeTask());
+                taskDescription.getChainTaskId(), taskDescription.isTeeTask());
 
         if (taskDescription.isTeeTask()) {
-            return preComputeService.runTeePreCompute(taskDescription,
-                            workerpoolAuth);
+            return preComputeService.runTeePreCompute(taskDescription, workerpoolAuth);
         }
         return PreComputeResponse.builder().build();
     }
 
-    public AppComputeResponse runCompute(TaskDescription taskDescription,
-                                         TeeSessionGenerationResponse secureSession) {
-        String chainTaskId = taskDescription.getChainTaskId();
-        log.info("Running compute [chainTaskId:{}, isTee:{}]", chainTaskId,
-                taskDescription.isTeeTask());
+    /**
+     * Execute application stage for standard and TEE tasks.
+     *
+     * @param taskDescription Description of the task
+     * @param secureSession   Session ID and session storage URL for TEE tasks
+     * @return {@code AppComputeResponse} instance
+     * @see AppComputeService#runCompute(TaskDescription, TeeSessionGenerationResponse)
+     */
+    public AppComputeResponse runCompute(final TaskDescription taskDescription,
+                                         final TeeSessionGenerationResponse secureSession) {
+        final String chainTaskId = taskDescription.getChainTaskId();
+        log.info("Running compute [chainTaskId:{}, isTee:{}]",
+                chainTaskId, taskDescription.isTeeTask());
 
-        AppComputeResponse appComputeResponse =
+        final AppComputeResponse appComputeResponse =
                 appComputeService.runCompute(taskDescription, secureSession);
 
         if (appComputeResponse.isSuccessful()) {
@@ -179,55 +198,58 @@ public AppComputeResponse runCompute(TaskDescription taskDescription,
 
     private void writeLogs(String chainTaskId, String filename, String logs) {
         if (!logs.isEmpty()) {
-            String filePath = workerConfigService.getTaskIexecOutDir(chainTaskId) + File.separator + filename;
-            File file = FileHelper.createFileWithContent(filePath, logs);
-            log.info("Saved logs file [path:{}]",
-                    file.getAbsolutePath());
+            final String filePath = workerConfigService.getTaskIexecOutDir(chainTaskId) + File.separator + filename;
+            final File file = FileHelper.createFileWithContent(filePath, logs);
+            log.info("Saved logs file [path:{}]", file.getAbsolutePath());
             //TODO Make sure file is properly written
         }
     }
 
-    /*
-     * - Copy computed.json file produced by the compute stage to /output
-     * - Zip iexec_out folder
-     * For TEE tasks, worker-tee-post-compute will do those two steps since
-     * all files in are protected.
+    /**
+     * Execute post-compute stage for standard and TEE tasks.
+     * <p>
+     * This method calls methods from {@code PostComputeService} depending on the Task type.
      *
-     * - Save stdout file
+     * @param taskDescription Description of the task
+     * @param secureSession   Session ID and session storage URL for TEE tasks
+     * @return {@code PostComputeResponse} instance
+     * @see PostComputeService#runStandardPostCompute(TaskDescription)
+     * @see PostComputeService#runTeePostCompute(TaskDescription, TeeSessionGenerationResponse)
      */
-    public PostComputeResponse runPostCompute(TaskDescription taskDescription,
-                                              TeeSessionGenerationResponse secureSession) {
-        String chainTaskId = taskDescription.getChainTaskId();
+    public PostComputeResponse runPostCompute(final TaskDescription taskDescription,
+                                              final TeeSessionGenerationResponse secureSession) {
+        final String chainTaskId = taskDescription.getChainTaskId();
         log.info("Running post-compute [chainTaskId:{}, isTee:{}]",
                 chainTaskId, taskDescription.isTeeTask());
-        PostComputeResponse postComputeResponse = PostComputeResponse.builder()
-                .exitCause(ReplicateStatusCause.POST_COMPUTE_FAILED_UNKNOWN_ISSUE)
-                .build();
 
+        final PostComputeResponse postComputeResponse;
         if (!taskDescription.isTeeTask()) {
             postComputeResponse = postComputeService.runStandardPostCompute(taskDescription);
         } else if (secureSession != null) {
             postComputeResponse = postComputeService.runTeePostCompute(taskDescription, secureSession);
+        } else {
+            postComputeResponse = PostComputeResponse.builder()
+                    .exitCause(ReplicateStatusCause.POST_COMPUTE_FAILED_UNKNOWN_ISSUE)
+                    .build();
         }
         if (!postComputeResponse.isSuccessful()) {
             return postComputeResponse;
         }
-        ComputedFile computedFile = resultService.readComputedFile(chainTaskId);
+        final ComputedFile computedFile = resultService.readComputedFile(chainTaskId);
         if (computedFile == null) {
             postComputeResponse.setExitCause(ReplicateStatusCause.POST_COMPUTE_COMPUTED_FILE_NOT_FOUND);
             return postComputeResponse;
         }
-        String resultDigest = resultService.computeResultDigest(computedFile);
+        final String resultDigest = resultService.computeResultDigest(computedFile);
         if (resultDigest.isEmpty()) {
             postComputeResponse.setExitCause(ReplicateStatusCause.POST_COMPUTE_RESULT_DIGEST_COMPUTATION_FAILED);
         }
-        resultService.saveResultInfo(chainTaskId, taskDescription, computedFile);
+        resultService.saveResultInfo(taskDescription, computedFile);
         return postComputeResponse;
     }
 
-    public boolean abort(String chainTaskId) {
-        Predicate<String> containsChainTaskId = name -> name.contains(chainTaskId);
-        long remaining = dockerService.stopRunningContainersWithNamePredicate(containsChainTaskId);
+    public boolean abort(final String chainTaskId) {
+        final long remaining = dockerService.stopRunningContainersWithNameContaining(chainTaskId);
         log.info("Stopped task containers [chainTaskId:{}, remaining:{}]", chainTaskId, remaining);
         return remaining == 0L;
     }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2024 IEXEC BLOCKCHAIN TECH
+ * Copyright 2020-2025 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.
@@ -202,18 +202,18 @@ public void stopAllRunningContainers() {
     }
 
     /**
-     * Stop running containers with names that match the provided predicate and
+     * Stop running containers with names that contain the provided pattern and
      * remove them from the running containers record. This is typically used when
      * the worker aborts a task and needs to stop its pre-compute, compute, or
      * post-compute containers. The container itself is not removed here as it is
      * removed by its watcher thread.
      *
-     * @param containerNamePredicate predicate that contains a condition on the
-     *                               container name.
+     * @param pattern containers whose name contains this pattern will be removed.
      * @return The remaining count of containers matching the provided predicate.
      */
-    public long stopRunningContainersWithNamePredicate(Predicate<String> containerNamePredicate) {
-        log.info("Stopping containers with names matching the provided predicate");
+    public long stopRunningContainersWithNameContaining(final String pattern) {
+        log.info("Stopping containers with names containing the following pattern [pattern:{}]", pattern);
+        final Predicate<String> containerNamePredicate = name -> name.contains(pattern);
         List.copyOf(runningContainersRecord).stream()
                 .filter(containerNamePredicate)
                 .forEach(this::stopRunningContainer);

src/main/java/com/iexec/worker/replicate/ReplicateRecoveryService.java

@@ -96,7 +96,7 @@ boolean canReplicateBeRecovered(TaskNotification missedTaskNotification) {
         }
 
         final ComputedFile computedFile = resultService.getComputedFile(chainTaskId);
-        resultService.saveResultInfo(chainTaskId, taskDescription, computedFile);
+        resultService.saveResultInfo(taskDescription, computedFile);
         subscriptionService.subscribeToTopic(chainTaskId);
         applicationEventPublisher.publishEvent(missedTaskNotification);
 

src/main/java/com/iexec/worker/result/ResultService.java

@@ -116,7 +116,7 @@ public boolean writeErrorToIexecOut(final String chainTaskId, final ReplicateSta
                 + IexecFileHelper.COMPUTED_JSON, computedFileJsonAsString.getBytes());
     }
 
-    public void saveResultInfo(final String chainTaskId, final TaskDescription taskDescription,
+    public void saveResultInfo(final TaskDescription taskDescription,
                                final ComputedFile computedFile) {
         final ResultInfo resultInfo = ResultInfo.builder()
                 .image(taskDescription.getAppUri())
@@ -125,7 +125,7 @@ public void saveResultInfo(final String chainTaskId, final TaskDescription taskD
                 .datasetUri(taskDescription.getDatasetUri())
                 .build();
 
-        resultInfoMap.put(chainTaskId, resultInfo);
+        resultInfoMap.put(taskDescription.getChainTaskId(), resultInfo);
     }
 
     public ResultModel getResultModelWithZip(final String chainTaskId) {
@@ -409,12 +409,11 @@ public boolean purgeTask(final String chainTaskId) {
         final boolean deletedInMap = !resultInfoMap.containsKey(chainTaskId);
         final boolean deletedTaskFolder = !new File(taskBaseDir).exists();
 
-        boolean deleted = deletedInMap && deletedTaskFolder;
-        if (deletedTaskFolder) {
+        final boolean deleted = deletedInMap && deletedTaskFolder;
+        if (deleted) {
             log.info("The result of the chainTaskId has been deleted [chainTaskId:{}]", chainTaskId);
         } else {
-            log.warn("The result of the chainTaskId couldn't be deleted [chainTaskId:{}, deletedInMap:{}, " +
-                            "deletedTaskFolder:{}]",
+            log.warn("The result of the chainTaskId couldn't be deleted [chainTaskId:{}, deletedInMap:{}, deletedTaskFolder:{}]",
                     chainTaskId, deletedInMap, deletedTaskFolder);
         }
 

src/main/java/com/iexec/worker/task/TaskManagerService.java

@@ -452,10 +452,10 @@ ReplicateActionResponse complete(String chainTaskId) {
      * @param chainTaskId Task ID
      * @return {@literal true} if all cleanup operations went well, {@literal false} otherwise
      */
-    boolean abort(String chainTaskId) {
+    synchronized boolean abort(final String chainTaskId) {
         log.info("Aborting task [chainTaskId:{}]", chainTaskId);
-        boolean allContainersStopped = computeManagerService.abort(chainTaskId);
-        boolean allServicesPurged = purgeService.purgeAllServices(chainTaskId);
+        final boolean allContainersStopped = computeManagerService.abort(chainTaskId);
+        final boolean allServicesPurged = purgeService.purgeAllServices(chainTaskId);
         final boolean isSuccess = allContainersStopped && allServicesPurged;
         if (!isSuccess) {
             log.error("Failed to abort task [chainTaskId:{}, containers:{}, services:{}]",

src/test/java/com/iexec/worker/compute/ComputeManagerServiceTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2024 IEXEC BLOCKCHAIN TECH
+ * Copyright 2020-2025 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.
@@ -54,7 +54,6 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.*;
 
 @ExtendWith(MockitoExtension.class)
@@ -119,6 +118,7 @@ void shouldDownloadApp() {
         when(dockerRegistryConfiguration.getMaxPullTimeout()).thenReturn(Duration.of(30, ChronoUnit.MINUTES));
         when(dockerService.getClient(taskDescription.getAppUri())).thenReturn(dockerClient);
         when(dockerClient.pullImage(taskDescription.getAppUri(), Duration.of(7, ChronoUnit.MINUTES))).thenReturn(true);
+        when(dockerClient.isImagePresent(taskDescription.getAppUri())).thenReturn(true);
         assertThat(computeManagerService.downloadApp(taskDescription)).isTrue();
     }
 
@@ -127,6 +127,7 @@ void shouldNotDownloadAppSincePullImageFailed() {
         final TaskDescription taskDescription = createTaskDescriptionBuilder(true).build();
         when(dockerService.getClient(taskDescription.getAppUri())).thenReturn(dockerClient);
         when(dockerClient.pullImage(taskDescription.getAppUri(), Duration.ofMinutes(0))).thenReturn(false);
+        when(dockerClient.isImagePresent(taskDescription.getAppUri())).thenReturn(false);
         assertThat(computeManagerService.downloadApp(taskDescription)).isFalse();
     }
 
@@ -349,7 +350,7 @@ void shouldRunStandardPostCompute() {
         verify(postComputeService).runStandardPostCompute(taskDescription);
         verify(resultService).readComputedFile(CHAIN_TASK_ID);
         verify(resultService).computeResultDigest(computedFile);
-        verify(resultService).saveResultInfo(anyString(), any(), any());
+        verify(resultService).saveResultInfo(any(), any());
     }
 
     @ParameterizedTest
@@ -390,7 +391,7 @@ void shouldRunTeePostCompute() {
         verify(postComputeService).runTeePostCompute(taskDescription, SECURE_SESSION);
         verify(resultService).readComputedFile(CHAIN_TASK_ID);
         verify(resultService).computeResultDigest(computedFile);
-        verify(resultService).saveResultInfo(anyString(), any(), any());
+        verify(resultService).saveResultInfo(any(), any());
     }
 
     @Test
@@ -458,13 +459,13 @@ void computeImagePullTimeout(long maxExecutionTime,
     // region abort
     @Test
     void shouldNotAbortWhenContainersAreStillRunning() {
-        when(dockerService.stopRunningContainersWithNamePredicate(any())).thenReturn(1L);
+        when(dockerService.stopRunningContainersWithNameContaining(any())).thenReturn(1L);
         assertThat(computeManagerService.abort(CHAIN_TASK_ID)).isFalse();
     }
 
     @Test
     void shouldAbortTask() {
-        when(dockerService.stopRunningContainersWithNamePredicate(any())).thenReturn(0L);
+        when(dockerService.stopRunningContainersWithNameContaining(any())).thenReturn(0L);
         assertThat(computeManagerService.abort(CHAIN_TASK_ID)).isTrue();
     }
     // endregion

src/test/java/com/iexec/worker/docker/DockerServiceTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2024 IEXEC BLOCKCHAIN TECH
+ * Copyright 2020-2025 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.
@@ -39,7 +39,6 @@
 
 import java.io.File;
 import java.util.Optional;
-import java.util.function.Predicate;
 import java.util.stream.Stream;
 
 import static com.iexec.commons.containers.client.DockerClientInstance.DEFAULT_DOCKER_REGISTRY;
@@ -422,8 +421,7 @@ void shouldStopRunningContainersWithNamePattern() {
         dockerService.addToRunningContainersRecord("containerName1");
         dockerService.addToRunningContainersRecord("containerName2");
 
-        Predicate<String> containsChainTaskId = name -> name.contains(CHAIN_TASK_ID);
-        dockerService.stopRunningContainersWithNamePredicate(containsChainTaskId);
+        dockerService.stopRunningContainersWithNameContaining(CHAIN_TASK_ID);
         // Verify we removed all containers matching the predicate
         verify(dockerService, times(3)).stopRunningContainer(anyString());
         // Verify we removed only containers matching the predicate