fix: replace retry mechanism for requesting mnlists

dashpay · May 6, 2024 · bd7390d · bd7390d
1 parent c1e3fad
commit bd7390d
Show file tree

Hide file tree

Showing 3 changed files with 35 additions and 23 deletions.
diff --git a/core/src/main/java/org/bitcoinj/evolution/AbstractQuorumState.java b/core/src/main/java/org/bitcoinj/evolution/AbstractQuorumState.java
@@ -17,7 +17,6 @@
 package org.bitcoinj.evolution;
 
 import com.google.common.base.Preconditions;
-import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import org.bitcoinj.core.AbstractBlockChain;
 import org.bitcoinj.core.BlockQueue;
@@ -47,6 +46,7 @@
 import org.bitcoinj.quorums.SigningManager;
 import org.bitcoinj.quorums.SimplifiedQuorumList;
 import org.bitcoinj.store.BlockStoreException;
+import org.bitcoinj.utils.ContextPropagatingThreadFactory;
 import org.bitcoinj.utils.Threading;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -63,6 +63,9 @@
 import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.locks.ReentrantLock;
@@ -701,22 +704,28 @@ public void onFirstSaveComplete() {
         }
     }
 
+    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(
+            1,
+            new ContextPropagatingThreadFactory("quorum-state-" + getClass().getSimpleName())
+    );
+    ScheduledFuture<?> retryFuture = null;
+
     protected void sendRequestWithRetry(Peer peer) {
-        ListenableFuture sendMessageFuture = peer.sendMessage(lastRequest.getRequestMessage());
-        sendMessageFuture.addListener(new Runnable() {
-            @Override
-            public void run() {
-                try {
-                    // throws an exception if there was a problem sending
-                    sendMessageFuture.get(10, TimeUnit.SECONDS);
-                } catch (ExecutionException | TimeoutException e) {
-                    // send the message again
-                    retryLastRequest(peer, e);
-                } catch (InterruptedException e) {
-                    log.info("sendMessageFuture interrupted", e);
-                }
+        peer.sendMessage(lastRequest.getRequestMessage());
+
+        if (retryFuture != null) {
+            log.info("sendMessageFuture cancel: {}", lastRequest.request.getClass().getSimpleName());
+            retryFuture.cancel(true);
+            retryFuture = null;
+        }
+        retryFuture = scheduledExecutorService.schedule(() -> {
+            if (!lastRequest.getReceived()) {
+                log.info("sendMessageFuture check: last request not received {}", lastRequest.request.getClass().getSimpleName());
+                retryLastRequest(peer, new TimeoutException("last request not received"));
+            } else {
+                log.info("sendMessageFuture check: last request received {}", lastRequest.request.getClass().getSimpleName());
             }
-        }, Threading.THREAD_POOL);
+        }, 10, TimeUnit.SECONDS);
     }
 
     private void retryLastRequest(Peer peer, Exception e) {
@@ -865,5 +874,10 @@ Sha256Hash getHashModifier(LLMQParameters llmqParams, StoredBlock quorumBaseBloc
     public void close() {
         // reset the state of any sync operation
         waitingForMNListDiff = false;
+        if (retryFuture != null) {
+            log.info("cancel: {}", lastRequest.request.getClass().getSimpleName());
+            retryFuture.cancel(true);
+            retryFuture = null;
+        }
     }
 }
diff --git a/core/src/main/java/org/bitcoinj/evolution/QuorumRotationState.java b/core/src/main/java/org/bitcoinj/evolution/QuorumRotationState.java
@@ -1246,6 +1246,7 @@ public void processDiff(@Nullable Peer peer, QuorumRotationInfo quorumRotationIn
         boolean isSyncingHeadersFirst = syncStage == PeerGroup.SyncStage.MNLIST;
 
         quorumRotationInfo.dump(getMnListTip().getHeight(), newHeight);
+        lastRequest.setReceived();
 
         lock.lock();
         try {

diff --git a/core/src/main/java/org/bitcoinj/evolution/QuorumState.java b/core/src/main/java/org/bitcoinj/evolution/QuorumState.java
@@ -114,11 +114,6 @@ public void requestReset(Peer peer, StoredBlock nextBlock) {
     public void requestUpdate(Peer peer, StoredBlock nextBlock) {
         lastRequest = new QuorumUpdateRequest<>(getMasternodeListDiffRequest(nextBlock), peer.getAddress());
         sendRequestWithRetry(peer);
-//        try {
-//            Thread.sleep(100000);
-//        } catch (InterruptedException x) {
-//            // none
-//        }
     }
 
     public void applyDiff(Peer peer, DualBlockChain blockChain,
@@ -271,9 +266,10 @@ public void processDiff(@Nullable Peer peer, SimplifiedMasternodeListDiff mnlist
 
         mnlistdiff.dump(mnList.getHeight(), newHeight);
         lastRequest.setReceived();
-        if (lock.isLocked() && !initChainTipSyncComplete()) {
-            Threading.dump();
-        }
+        // TODO: remove this
+//        if (lock.isLocked() && !initChainTipSyncComplete()) {
+//            Threading.dump();
+//        }
         lock.lock();
         try {
             log.info("lock acquired when processing mnlistdiff");
@@ -326,6 +322,7 @@ public void processDiff(@Nullable Peer peer, SimplifiedMasternodeListDiff mnlist
                     resetMNList(true);
                 }
             }
+            lastRequest.setFulfilled();
             finishDiff(isLoadingBootStrap);
         } catch(VerificationException x) {
             //request this block again and close this peer