Merge pull request #442 from iExecBlockchainComputing/fix/optimistic-locking-failure-exception

mcornaton · web-flow · commit d32a577f3d1f · 2021-10-12T09:52:03.000+02:00
Fix optimistic locking failure exception
diff --git a/gradle.properties b/gradle.properties
@@ -1,4 +1,4 @@
-version=6.1.3
+version=6.1.4
 iexecCommonVersion=5.5.1
 nexusUser=fake
 nexusPassword=fake
diff --git a/src/main/java/com/iexec/core/detector/replicate/ContributionUnnotifiedDetector.java b/src/main/java/com/iexec/core/detector/replicate/ContributionUnnotifiedDetector.java
@@ -34,13 +34,15 @@
 @Service
 public class ContributionUnnotifiedDetector extends UnnotifiedAbstractDetector {
 
-    private static final int DETECTOR_MULTIPLIER = 10;
+    private static final int LESS_OFTEN_DETECTOR_FREQUENCY = 10;
     private final List<TaskStatus> dectectWhenOffchainTaskStatuses;
     private final ReplicateStatus offchainCompleting;
     private final ReplicateStatus offchainCompleted;
     private final ChainContributionStatus onchainCompleted;
     private final int detectorRate;
 
+    private int detectorOccurrence;
+
     public ContributionUnnotifiedDetector(TaskService taskService,
                                           ReplicatesService replicatesService,
                                           IexecHubService iexecHubService,
@@ -54,12 +56,26 @@ public ContributionUnnotifiedDetector(TaskService taskService,
         this.detectorRate = cronConfiguration.getContribute();
     }
 
+    /**
+     * Detects onchain CONTRIBUTED only if replicates are offchain CONTRIBUTING and
+     * onchain CONTRIBUTED if replicates are not CONTRIBUTED.
+     * The second detection is not always ran, depending on the detector run occurrences.
+     */
+    @Scheduled(fixedRateString = "#{@cronConfiguration.getContribute()}")
+    public void detectOnChainChanges() {
+        detectOnchainContributedWhenOffchainContributing();
+
+        detectorOccurrence++;
+        if (detectorOccurrence % LESS_OFTEN_DETECTOR_FREQUENCY == 0) {
+            detectOnchainContributed();
+        }
+    }
+
     /*
      * Detecting onchain CONTRIBUTED only if replicates are offchain CONTRIBUTING
      * (worker didn't notify last offchain CONTRIBUTED)
      * We want to detect them very often since it's highly probable
      */
-    @Scheduled(fixedRateString = "#{@cronConfiguration.getContribute()}")
     public void detectOnchainContributedWhenOffchainContributing() {
         log.debug("Detect onchain Contributed (when offchain Contributing) [retryIn:{}]",
                 this.detectorRate);
@@ -78,9 +94,8 @@ public void detectOnchainContributedWhenOffchainContributing() {
      * - Frequently but no so often since it's eth node resource consuming and less probable
      * - When we receive a CANT_CONTRIBUTE_SINCE_TASK_NOT_ACTIVE
      */
-    @Scheduled(fixedRateString = "#{@cronConfiguration.getContribute() * " + DETECTOR_MULTIPLIER + "}")
     public void detectOnchainContributed() {
-        log.debug("Detect onchain Contributed [retryIn:{}]", this.detectorRate * DETECTOR_MULTIPLIER);
+        log.debug("Detect onchain Contributed [retryIn:{}]", this.detectorRate * LESS_OFTEN_DETECTOR_FREQUENCY);
         dectectOnchainCompleted(
                 dectectWhenOffchainTaskStatuses,
                 offchainCompleting,
diff --git a/src/main/java/com/iexec/core/detector/replicate/RevealUnnotifiedDetector.java b/src/main/java/com/iexec/core/detector/replicate/RevealUnnotifiedDetector.java
@@ -34,13 +34,15 @@
 @Service
 public class RevealUnnotifiedDetector extends UnnotifiedAbstractDetector {
 
-    private static final int DETECTOR_MULTIPLIER = 10;
+    private static final int LESS_OFTEN_DETECTOR_FREQUENCY = 10;
     private final List<TaskStatus> dectectWhenTaskStatuses;
     private final ReplicateStatus offchainCompleting;
     private final ReplicateStatus offchainCompleted;
     private final ChainContributionStatus onchainCompleted;
     private final CronConfiguration cronConfiguration;
 
+    private int detectorOccurrence;
+
     public RevealUnnotifiedDetector(TaskService taskService,
                                     ReplicatesService replicatesService,
                                     IexecHubService iexecHubService,
@@ -54,12 +56,21 @@ public RevealUnnotifiedDetector(TaskService taskService,
         onchainCompleted = ChainContributionStatus.REVEALED;
     }
 
+    @Scheduled(fixedRateString = "#{@cronConfiguration.getReveal()}")
+    public void detectOnChainChanges() {
+        detectOnchainRevealedWhenOffchainRevealed();
+
+        detectorOccurrence++;
+        if (detectorOccurrence % LESS_OFTEN_DETECTOR_FREQUENCY == 0) {
+            detectOnchainRevealed();
+        }
+    }
+
     /*
      * Detecting onchain REVEALED only if replicates are offchain REVEALING
      * (worker didn't notify last offchain REVEALED)
      * We want to detect them very often since it's highly probable
      */
-    @Scheduled(fixedRateString = "#{@cronConfiguration.getReveal()}")
     public void detectOnchainRevealedWhenOffchainRevealed() {
         log.debug("Detect onchain Revealed (when offchain Revealing) [retryIn:{}]",
                 cronConfiguration.getReveal());
@@ -78,10 +89,9 @@ public void detectOnchainRevealedWhenOffchainRevealed() {
      * - Frequently but no so often since it's eth node resource consuming and less probable
      * - When we receive a CANT_REVEAL
      */
-    @Scheduled(fixedRateString = "#{@cronConfiguration.getReveal() * " + DETECTOR_MULTIPLIER + "}")
     public void detectOnchainRevealed() {
         log.debug("Detect onchain Revealed [retryIn:{}]",
-                cronConfiguration.getReveal() * DETECTOR_MULTIPLIER);
+                cronConfiguration.getReveal() * LESS_OFTEN_DETECTOR_FREQUENCY);
         dectectOnchainCompleted(
                 dectectWhenTaskStatuses,
                 offchainCompleting,
diff --git a/src/main/java/com/iexec/core/replicate/ReplicatesService.java b/src/main/java/com/iexec/core/replicate/ReplicatesService.java
@@ -226,17 +226,31 @@ private boolean isStatusBeforeWorkerLostEqualsTo(Replicate replicate, ReplicateS
 
     /*
      * This implicitly sets the modifier to POOL_MANAGER
+     *
+     * @Retryable is needed as it isn't triggered by a call from within the class itself.
      */
+    @Retryable(value = {OptimisticLockingFailureException.class}, maxAttempts = 100)
     public void updateReplicateStatus(String chainTaskId,
                                       String walletAddress,
                                       ReplicateStatus newStatus) {
         ReplicateStatusUpdate statusUpdate = ReplicateStatusUpdate.poolManagerRequest(newStatus);
         updateReplicateStatus(chainTaskId, walletAddress, statusUpdate);
     }
 
+    @Recover
+    public void updateReplicateStatus(OptimisticLockingFailureException exception,
+                                      String chainTaskId,
+                                      String walletAddress,
+                                      ReplicateStatus newStatus) {
+        logUpdateReplicateStatusRecover(exception);
+    }
+
     /*
      * This implicitly sets the modifier to POOL_MANAGER
+     *
+     * @Retryable is needed as it isn't triggered by a call from within the class itself.
      */
+    @Retryable(value = {OptimisticLockingFailureException.class}, maxAttempts = 100)
     public void updateReplicateStatus(String chainTaskId,
                                       String walletAddress,
                                       ReplicateStatus newStatus,
@@ -245,6 +259,15 @@ public void updateReplicateStatus(String chainTaskId,
         updateReplicateStatus(chainTaskId, walletAddress, statusUpdate);
     }
 
+    @Recover
+    public void updateReplicateStatus(OptimisticLockingFailureException exception,
+                                      String chainTaskId,
+                                      String walletAddress,
+                                      ReplicateStatus newStatus,
+                                      ReplicateStatusDetails details) {
+        logUpdateReplicateStatusRecover(exception);
+    }
+
     /*
      * We retry up to 100 times in case the task has been modified between
      * reading and writing it.
@@ -333,6 +356,10 @@ public void updateReplicateStatus(OptimisticLockingFailureException exception,
                                       String chainTaskId,
                                       String walletAddress,
                                       ReplicateStatusUpdate statusUpdate) {
+        logUpdateReplicateStatusRecover(exception);
+    }
+
+    private void logUpdateReplicateStatusRecover(OptimisticLockingFailureException exception) {
         log.error("Could not update replicate status, maximum number of retries reached");
         exception.printStackTrace();
     }
diff --git a/src/test/java/com/iexec/core/detector/replicate/ContributionUnnotifiedDetectorTests.java b/src/test/java/com/iexec/core/detector/replicate/ContributionUnnotifiedDetectorTests.java
@@ -35,6 +35,7 @@
 import java.util.Collections;
 
 import static com.iexec.common.replicate.ReplicateStatus.*;
+import static com.iexec.common.replicate.ReplicateStatusModifier.WORKER;
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.when;
 
@@ -68,6 +69,32 @@ public void init() {
         MockitoAnnotations.initMocks(this);
     }
 
+    // Detector aggregator
+    @Test
+    public void shouldDetectBothChangesOnChain() {
+        Task task = Task.builder().chainTaskId(CHAIN_TASK_ID).build();
+        when(taskService.findByCurrentStatus(TaskStatus.getWaitingContributionStatuses())).thenReturn(Collections.singletonList(task));
+
+        Replicate replicate = new Replicate(WALLET_ADDRESS, CHAIN_TASK_ID);
+        ReplicateStatusUpdate statusUpdate = ReplicateStatusUpdate.builder().status(CONTRIBUTING).modifier(WORKER).build();
+        replicate.setStatusUpdateList(Collections.singletonList(statusUpdate));
+
+        when(replicatesService.getReplicates(any())).thenReturn(Collections.singletonList(replicate));
+        when(iexecHubService.isStatusTrueOnChain(any(), any(), any())).thenReturn(true);
+        when(web3jService.getLatestBlockNumber()).thenReturn(11L);
+        when(iexecHubService.getContributionBlock(anyString(), anyString(), anyLong())).thenReturn(ChainReceipt.builder()
+                .blockNumber(10L)
+                .txHash("0xabcef")
+                .build());
+
+        for (int i = 0; i < 10; i++) {
+            contributionDetector.detectOnChainChanges();
+        }
+
+        Mockito.verify(replicatesService, Mockito.times(11))    // 10 detectors #1 & 1 detector #2
+                .updateReplicateStatus(any(), any(), any(), any(ReplicateStatusDetails.class));
+    }
+
 
     //Detector#1 after contributing
 
diff --git a/src/test/java/com/iexec/core/detector/replicate/RevealUnnotifiedDetectorTests.java b/src/test/java/com/iexec/core/detector/replicate/RevealUnnotifiedDetectorTests.java
@@ -68,6 +68,32 @@ public void init() {
         MockitoAnnotations.initMocks(this);
     }
 
+    // Detector aggregator
+    @Test
+    public void shouldDetectBothChangesOnChain() {
+        Task task = Task.builder().chainTaskId(CHAIN_TASK_ID).build();
+        when(taskService.findByCurrentStatus(TaskStatus.getWaitingRevealStatuses())).thenReturn(Collections.singletonList(task));
+
+        Replicate replicate = new Replicate(WALLET_ADDRESS, CHAIN_TASK_ID);
+        ReplicateStatusUpdate statusUpdate = ReplicateStatusUpdate.builder().status(REVEALING).modifier(WORKER).build();
+        replicate.setStatusUpdateList(Collections.singletonList(statusUpdate));
+
+        when(replicatesService.getReplicates(any())).thenReturn(Collections.singletonList(replicate));
+        when(iexecHubService.isStatusTrueOnChain(any(), any(), any())).thenReturn(true);
+        when(web3jService.getLatestBlockNumber()).thenReturn(11L);
+        when(iexecHubService.getRevealBlock(anyString(), anyString(), anyLong())).thenReturn(ChainReceipt.builder()
+                .blockNumber(10L)
+                .txHash("0xabcef")
+                .build());
+
+        for (int i = 0; i < 10; i++) {
+            revealDetector.detectOnChainChanges();
+        }
+
+        Mockito.verify(replicatesService, Mockito.times(11))    // 10 detectors #1 & 1 detector #2
+                .updateReplicateStatus(any(), any(), any(), any(ReplicateStatusDetails.class));
+    }
+
 
     //Detector#1 after contributing
 
