Include responders missing hashes from our historical data store

Author: alvasw

p2p/src/main/java/bisq/network/p2p/peers/getdata/RequestDataHandler.java

@@ -43,6 +43,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -73,7 +74,7 @@ class RequestDataHandler implements MessageListener {
     ///////////////////////////////////////////////////////////////////////////////////////////
 
     public interface Listener {
-        void onComplete(boolean wasTruncated);
+        void onComplete(boolean wasTruncated, String responderVersion);
 
         @SuppressWarnings("UnusedParameters")
         void onFault(String errorMessage, @SuppressWarnings("SameParameterValue") @Nullable Connection connection);
@@ -87,6 +88,7 @@ public interface Listener {
     private final NetworkNode networkNode;
     private final P2PDataStorage dataStorage;
     private final PeerManager peerManager;
+    private final Optional<String> responderVersion;
     private final Listener listener;
     private Timer timeoutTimer;
     private final int nonce = new Random().nextInt();
@@ -100,10 +102,12 @@ public interface Listener {
     RequestDataHandler(NetworkNode networkNode,
                        P2PDataStorage dataStorage,
                        PeerManager peerManager,
+                       Optional<String> responderVersion,
                        Listener listener) {
         this.networkNode = networkNode;
         this.dataStorage = dataStorage;
         this.peerManager = peerManager;
+        this.responderVersion = responderVersion;
         this.listener = listener;
     }
 
@@ -124,7 +128,7 @@ void requestData(NodeAddress nodeAddress, boolean isPreliminaryDataRequest) {
             if (isPreliminaryDataRequest)
                 getDataRequest = dataStorage.buildPreliminaryGetDataRequest(nonce);
             else
-                getDataRequest = dataStorage.buildGetUpdatedDataRequest(networkNode.getNodeAddress(), nonce);
+                getDataRequest = dataStorage.buildGetUpdatedDataRequest(networkNode.getNodeAddress(), nonce, responderVersion);
 
             if (timeoutTimer == null) {
                 timeoutTimer = UserThread.runAfter(() -> {  // setup before sending to avoid race conditions
@@ -201,7 +205,7 @@ public void onMessage(NetworkEnvelope networkEnvelope, Connection connection) {
                                 connection.getPeersNodeAddressOptional().get());
 
                         cleanup();
-                        listener.onComplete(getDataResponse.isWasTruncated());
+                        listener.onComplete(getDataResponse.isWasTruncated(), getDataResponse.getVersion());
                     } else {
                         log.warn("Nonce not matching. That can happen rarely if we get a response after a canceled " +
                                         "handshake (timeout causes connection close but peer might have sent a msg before " +

p2p/src/main/java/bisq/network/p2p/peers/getdata/RequestDataManager.java

@@ -177,7 +177,7 @@ public void requestPreliminaryData() {
                 nodeAddresses.remove(nodeAddress);
                 // We clone list to avoid mutable change during iterations
                 List<NodeAddress> remainingNodeAddresses = new ArrayList<>(nodeAddresses);
-                UserThread.runAfter(() -> requestData(nodeAddress, remainingNodeAddresses), (i * 200 + 1), TimeUnit.MILLISECONDS);
+                UserThread.runAfter(() -> requestData(nodeAddress, remainingNodeAddresses, Optional.empty()), (i * 200 + 1), TimeUnit.MILLISECONDS);
             }
 
             isPreliminaryDataRequest = true;
@@ -195,7 +195,7 @@ public void requestUpdateData() {
             // We use the node we have already connected to to request again
             nodeAddressOfPreliminaryDataRequest.ifPresent(candidate -> {
                 nodeAddresses.remove(candidate);
-                requestData(candidate, nodeAddresses);
+                requestData(candidate, nodeAddresses, Optional.empty());
 
                 ArrayList<NodeAddress> finalNodeAddresses = new ArrayList<>(nodeAddresses);
                 int numRequests = 0;
@@ -205,7 +205,7 @@ public void requestUpdateData() {
 
                     // It might be that we have a prelim. request open for the same seed, if so we skip to the next.
                     if (!handlerMap.containsKey(nodeAddress)) {
-                        UserThread.runAfter(() -> requestData(nodeAddress, nodeAddresses), (i * 200 + 1), TimeUnit.MILLISECONDS);
+                        UserThread.runAfter(() -> requestData(nodeAddress, nodeAddresses, Optional.empty()), (i * 200 + 1), TimeUnit.MILLISECONDS);
                         numRequests++;
                     }
                 }
@@ -333,13 +333,15 @@ public void onFault(String errorMessage, @Nullable Connection connection) {
     // RequestData
     ///////////////////////////////////////////////////////////////////////////////////////////
 
-    private void requestData(NodeAddress nodeAddress, List<NodeAddress> remainingNodeAddresses) {
+    private void requestData(NodeAddress nodeAddress, List<NodeAddress> remainingNodeAddresses,
+                             Optional<String> responderVersion) {
         if (!stopped) {
             if (!handlerMap.containsKey(nodeAddress)) {
                 RequestDataHandler requestDataHandler = new RequestDataHandler(networkNode, dataStorage, peerManager,
+                        responderVersion,
                         new RequestDataHandler.Listener() {
                             @Override
-                            public void onComplete(boolean wasTruncated) {
+                            public void onComplete(boolean wasTruncated, String responderVersion) {
                                 log.trace("RequestDataHandshake of outbound connection complete. nodeAddress={}",
                                         nodeAddress);
                                 stopRetryTimer();
@@ -363,12 +365,20 @@ public void onComplete(boolean wasTruncated) {
                                 }
 
                                 if (wasTruncated) {
-                                    if (numRepeatedRequests < MAX_REPEATED_REQUESTS) {
+                                    boolean isResponderOlderVersion = Version.isNewVersion(Version.VERSION, responderVersion);
+                                    if (isResponderOlderVersion) {
+                                        log.info("The responder's version is older than hours. " +
+                                                "We need to include the hashes from our historical store (since {}) " +
+                                                "to the next request.", responderVersion);
+                                        UserThread.runAfter(() -> requestData(nodeAddress, remainingNodeAddresses, Optional.of(responderVersion)), 2);
+                                    }
+
+                                    else if (numRepeatedRequests < MAX_REPEATED_REQUESTS) {
                                         // If we had allDataReceived already set to true but get a response with truncated flag,
                                         // we still repeat the request to that node for higher redundancy. Otherwise, one seed node
                                         // providing incomplete data would stop others to fill the gaps.
                                         log.info("DataResponse did not contain all data, so we repeat request until we got all data");
-                                        UserThread.runAfter(() -> requestData(nodeAddress, remainingNodeAddresses), 2);
+                                        UserThread.runAfter(() -> requestData(nodeAddress, remainingNodeAddresses, Optional.of(responderVersion)), 2);
                                     } else if (!allDataReceived) {
                                         allDataReceived = true;
                                         log.warn("\n#################################################################\n" +
@@ -398,7 +408,7 @@ public void onFault(String errorMessage, @Nullable Connection connection) {
                                             "We will try requestDataFromPeers again.");
                                     NodeAddress nextCandidate = remainingNodeAddresses.get(0);
                                     remainingNodeAddresses.remove(nextCandidate);
-                                    requestData(nextCandidate, remainingNodeAddresses);
+                                    requestData(nextCandidate, remainingNodeAddresses, Optional.empty());
                                 } else if (handlerMap.isEmpty()) {
                                     // If not other connection attempts are in the handlerMap we assume that no seed
                                     // nodes are available.
@@ -455,7 +465,7 @@ private void requestFromNonSeedNodePeers() {
         if (!list.isEmpty()) {
             NodeAddress nextCandidate = list.get(0);
             list.remove(nextCandidate);
-            requestData(nextCandidate, list);
+            requestData(nextCandidate, list, Optional.empty());
         }
     }
 
@@ -482,7 +492,7 @@ private void restart() {
                         if (!list.isEmpty()) {
                             NodeAddress nextCandidate = list.get(0);
                             list.remove(nextCandidate);
-                            requestData(nextCandidate, list);
+                            requestData(nextCandidate, list, Optional.empty());
                         }
                     },
                     RETRY_DELAY_SEC);

p2p/src/main/java/bisq/network/p2p/storage/P2PDataStorage.java

@@ -104,6 +104,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArraySet;
@@ -271,25 +272,26 @@ public void addProtectedMailboxStorageEntryToMap(ProtectedStorageEntry protected
      * Returns a PreliminaryGetDataRequest that can be sent to a peer node to request missing Payload data.
      */
     public PreliminaryGetDataRequest buildPreliminaryGetDataRequest(int nonce) {
-        return new PreliminaryGetDataRequest(nonce, getKnownPayloadHashes());
+        return new PreliminaryGetDataRequest(nonce, getKnownPayloadHashes(Optional.empty()));
     }
 
     /**
      * Returns a GetUpdatedDataRequest that can be sent to a peer node to request missing Payload data.
      */
-    public GetUpdatedDataRequest buildGetUpdatedDataRequest(NodeAddress senderNodeAddress, int nonce) {
-        return new GetUpdatedDataRequest(senderNodeAddress, nonce, getKnownPayloadHashes());
+    public GetUpdatedDataRequest buildGetUpdatedDataRequest(NodeAddress senderNodeAddress, int nonce,
+                                                            Optional<String> responderVersion) {
+        return new GetUpdatedDataRequest(senderNodeAddress, nonce, getKnownPayloadHashes(responderVersion));
     }
 
     /**
      * Returns the set of known payload hashes. This is used in the GetData path to request missing data from peer nodes
      */
-    private Set<byte[]> getKnownPayloadHashes() {
+    private Set<byte[]> getKnownPayloadHashes(Optional<String> responderVersion) {
         // We collect the keys of the PersistableNetworkPayload items so we exclude them in our request.
         // PersistedStoragePayload items don't get removed, so we don't have an issue with the case that
         // an object gets removed in between PreliminaryGetDataRequest and the GetUpdatedDataRequest and we would
         // miss that event if we do not load the full set or use some delta handling.
-        Map<ByteArray, PersistableNetworkPayload> mapForDataRequest = getMapForDataRequest();
+        Map<ByteArray, PersistableNetworkPayload> mapForDataRequest = getMapForDataRequest(responderVersion);
         Set<byte[]> excludedKeys = getKeysAsByteSet(mapForDataRequest);
         Set<byte[]> excludedKeysFromProtectedStorageEntryMap = getKeysAsByteSet(map);
         excludedKeys.addAll(excludedKeysFromProtectedStorageEntryMap);
@@ -374,7 +376,7 @@ public GetDataResponse buildGetDataResponse(
     // Utils for collecting the exclude hashes
     ///////////////////////////////////////////////////////////////////////////////////////////
 
-    private Map<ByteArray, PersistableNetworkPayload> getMapForDataRequest() {
+    private Map<ByteArray, PersistableNetworkPayload> getMapForDataRequest(Optional<String> responderVersion) {
         Map<ByteArray, PersistableNetworkPayload> map = new HashMap<>();
         appendOnlyDataStoreService.getServices()
                 .forEach(service -> {
@@ -383,7 +385,11 @@ private Map<ByteArray, PersistableNetworkPayload> getMapForDataRequest() {
                         var historicalDataStoreService = (HistoricalDataStoreService<? extends PersistableNetworkPayloadStore>) service;
                         // As we add the version to our request we only use the live data. Eventually missing data will be
                         // derived from the version.
-                        serviceMap = historicalDataStoreService.getMapOfLiveData();
+                        if (responderVersion.isEmpty()) {
+                            serviceMap = historicalDataStoreService.getMapOfLiveData();
+                        } else {
+                            serviceMap = historicalDataStoreService.getMapSinceVersion(responderVersion.get(), Collections.emptySet(), new AtomicBoolean());
+                        }
                     } else {
                         serviceMap = service.getMap();
                     }
@@ -547,7 +553,7 @@ static private <T extends NetworkPayload> Set<T> filterKnownHashes(
     }
 
     public Collection<PersistableNetworkPayload> getPersistableNetworkPayloadCollection() {
-        return getMapForDataRequest().values();
+        return getMapForDataRequest(Optional.empty()).values();
     }
 
     private Set<byte[]> getKeysAsByteSet(Map<ByteArray, ? extends PersistablePayload> map) {

p2p/src/test/java/bisq/network/p2p/storage/P2PDataStorageRequestDataTest.java

@@ -33,6 +33,7 @@
 import java.security.KeyPair;
 import java.security.NoSuchAlgorithmException;
 
+import java.util.Optional;
 import java.util.Set;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -105,7 +106,7 @@ public void buildPreliminaryGetDataRequest_EmptyP2PDataStore() {
     @Test
     public void buildGetUpdatedDataRequest_EmptyP2PDataStore() {
         GetUpdatedDataRequest getDataRequest =
-                this.testState.mockedStorage.buildGetUpdatedDataRequest(this.localNodeAddress, 1);
+                this.testState.mockedStorage.buildGetUpdatedDataRequest(this.localNodeAddress, 1, Optional.empty());
 
         assertEquals(getDataRequest.getNonce(), 1);
         assertEquals(getDataRequest.getSenderNodeAddress(), this.localNodeAddress);
@@ -156,7 +157,7 @@ public void requestData_FilledP2PDataStore_GetUpdatedDataRequest() throws NoSuch
         this.testState.mockedStorage.addProtectedStorageEntry(toAdd4, this.localNodeAddress, null);
 
         GetUpdatedDataRequest getDataRequest =
-                this.testState.mockedStorage.buildGetUpdatedDataRequest(this.localNodeAddress, 1);
+                this.testState.mockedStorage.buildGetUpdatedDataRequest(this.localNodeAddress, 1, Optional.empty());
 
         assertEquals(getDataRequest.getNonce(), 1);
         assertEquals(getDataRequest.getSenderNodeAddress(), this.localNodeAddress);