Fix small p2p network issues

common/src/main/java/bisq/common/app/Capabilities.java

@@ -39,6 +39,10 @@ public class Capabilities {
      */
     public static final Capabilities app = new Capabilities();
 
+    // Defines which most recent capability any node need to support.
+    // This helps to clean network from very old inactive but still running nodes.
+    private static final Capability mandatoryCapability = Capability.DAO_STATE;
+
     protected final Set<Capability> capabilities = new HashSet<>();
 
     public Capabilities(Capability... capabilities) {
@@ -71,7 +75,7 @@ public void addAll(Capability... capabilities) {
     }
 
     public void addAll(Capabilities capabilities) {
-        if(capabilities != null)
+        if (capabilities != null)
             this.capabilities.addAll(capabilities.capabilities);
     }
 
@@ -111,6 +115,10 @@ public static Capabilities fromIntList(List<Integer> capabilities) {
                 .collect(Collectors.toSet()));
     }
 
+    public static boolean hasMandatoryCapability(Capabilities capabilities) {
+        return capabilities.capabilities.stream().anyMatch(c -> c == mandatoryCapability);
+    }
+
     @Override
     public String toString() {
         return Arrays.toString(Capabilities.toIntList(this).toArray());

core/src/main/java/bisq/core/setup/CoreNetworkCapabilities.java

@@ -28,11 +28,20 @@
 @Slf4j
 public class CoreNetworkCapabilities {
     public static void setSupportedCapabilities(BisqEnvironment bisqEnvironment) {
-        Capabilities.app.addAll(Capability.TRADE_STATISTICS, Capability.TRADE_STATISTICS_2, Capability.ACCOUNT_AGE_WITNESS, Capability.ACK_MSG);
-        Capabilities.app.addAll(Capability.BUNDLE_OF_ENVELOPES, Capability.SIGNED_ACCOUNT_AGE_WITNESS);
+        Capabilities.app.addAll(
+                Capability.TRADE_STATISTICS,
+                Capability.TRADE_STATISTICS_2,
+                Capability.ACCOUNT_AGE_WITNESS,
+                Capability.ACK_MSG,
+                Capability.BUNDLE_OF_ENVELOPES
+        );
 
         if (BisqEnvironment.isDaoActivated(bisqEnvironment)) {
-            Capabilities.app.addAll(Capability.PROPOSAL, Capability.BLIND_VOTE, Capability.DAO_STATE);
+            Capabilities.app.addAll(
+                    Capability.PROPOSAL,
+                    Capability.BLIND_VOTE,
+                    Capability.DAO_STATE
+            );
 
             maybeApplyDaoFullMode(bisqEnvironment);
         }

p2p/src/main/java/bisq/network/p2p/network/Connection.java

@@ -121,8 +121,8 @@ public enum PeerType {
     private static ConnectionConfig connectionConfig;
 
     // Leaving some constants package-private for tests to know limits.
-    static final int PERMITTED_MESSAGE_SIZE = 200 * 1024;                       // 200 kb
-    static final int MAX_PERMITTED_MESSAGE_SIZE = 10 * 1024 * 1024;         // 10 MB (425 offers resulted in about 660 kb, mailbox msg will add more to it) offer has usually 2 kb, mailbox 3kb.
+    private static final int PERMITTED_MESSAGE_SIZE = 200 * 1024;                       // 200 kb
+    private static final int MAX_PERMITTED_MESSAGE_SIZE = 10 * 1024 * 1024;             // 10 MB (425 offers resulted in about 660 kb, mailbox msg will add more to it) offer has usually 2 kb, mailbox 3kb.
     //TODO decrease limits again after testing
     private static final int SOCKET_TIMEOUT = (int) TimeUnit.SECONDS.toMillis(120);
 
@@ -172,7 +172,7 @@ public static int getPermittedMessageSize() {
     private RuleViolation ruleViolation;
     private final ConcurrentHashMap<RuleViolation, Integer> ruleViolations = new ConcurrentHashMap<>();
 
-    private Capabilities capabilities = new Capabilities();
+    private final Capabilities capabilities = new Capabilities();
 
 
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -233,9 +233,9 @@ public Capabilities getCapabilities() {
         return capabilities;
     }
 
-    Object lock = new Object();
-    Queue<BundleOfEnvelopes> queueOfBundles = new ConcurrentLinkedQueue<>();
-    ScheduledExecutorService bundleSender = Executors.newSingleThreadScheduledExecutor();
+    private final Object lock = new Object();
+    private final Queue<BundleOfEnvelopes> queueOfBundles = new ConcurrentLinkedQueue<>();
+    private final ScheduledExecutorService bundleSender = Executors.newSingleThreadScheduledExecutor();
 
     // Called from various threads
     public void sendMessage(NetworkEnvelope networkEnvelope) {
@@ -250,7 +250,7 @@ public void sendMessage(NetworkEnvelope networkEnvelope) {
                     log.debug("Sending message: {}", Utilities.toTruncatedString(proto.toString(), 10000));
 
                     if (networkEnvelope instanceof Ping | networkEnvelope instanceof RefreshOfferMessage) {
-                        // pings and offer refresh msg we dont want to log in production
+                        // pings and offer refresh msg we don't want to log in production
                         log.trace("\n\n>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>\n" +
                                         "Sending direct message to peer" +
                                         "Write object to outputStream to peer: {} (uid={})\ntruncated message={} / size={}" +
@@ -298,10 +298,13 @@ public void sendMessage(NetworkEnvelope networkEnvelope) {
                                         if (!stopped) {
                                             synchronized (lock) {
                                                 BundleOfEnvelopes current = queueOfBundles.poll();
-                                                if(current.getEnvelopes().size() == 1)
-                                                    protoOutputStream.writeEnvelope(current.getEnvelopes().get(0));
-                                                else
-                                                    protoOutputStream.writeEnvelope(current);
+                                                if (current != null) {
+                                                    if (current.getEnvelopes().size() == 1) {
+                                                        protoOutputStream.writeEnvelope(current.getEnvelopes().get(0));
+                                                    } else {
+                                                        protoOutputStream.writeEnvelope(current);
+                                                    }
+                                                }
                                             }
                                         }
                                     }, lastSendTimeStamp - now, TimeUnit.MILLISECONDS);
@@ -386,10 +389,10 @@ private boolean violatesThrottleLimit() {
         messageTimeStamps.add(now);
 
         // clean list
-        while(messageTimeStamps.size() > msgThrottlePer10Sec)
+        while (messageTimeStamps.size() > msgThrottlePer10Sec)
             messageTimeStamps.remove(0);
 
-        return violatesThrottleLimit(now,1, msgThrottlePerSec) || violatesThrottleLimit(now,10, msgThrottlePer10Sec);
+        return violatesThrottleLimit(now, 1, msgThrottlePerSec) || violatesThrottleLimit(now, 10, msgThrottlePer10Sec);
     }
 
     private boolean violatesThrottleLimit(long now, int seconds, int messageCountLimit) {
@@ -399,7 +402,7 @@ private boolean violatesThrottleLimit(long now, int seconds, int messageCountLim
             long compareValue = messageTimeStamps.get(messageTimeStamps.size() - messageCountLimit);
 
             // if duration < seconds sec we received too much network_messages
-            if(now - compareValue < TimeUnit.SECONDS.toMillis(seconds)) {
+            if (now - compareValue < TimeUnit.SECONDS.toMillis(seconds)) {
                 log.error("violatesThrottleLimit {}/{} second(s)", messageCountLimit, seconds);
 
                 return true;
@@ -436,7 +439,7 @@ public void setPeerType(PeerType peerType) {
         this.peerType = peerType;
     }
 
-    public void setPeersNodeAddress(NodeAddress peerNodeAddress) {
+    private void setPeersNodeAddress(NodeAddress peerNodeAddress) {
         checkNotNull(peerNodeAddress, "peerAddress must not be null");
         peersNodeAddressOptional = Optional.of(peerNodeAddress);
 
@@ -494,6 +497,7 @@ public void shutDown(CloseConnectionReason closeConnectionReason, @Nullable Runn
 
                         stopped = true;
 
+                        //noinspection UnstableApiUsage
                         Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
                     } catch (Throwable t) {
                         log.error(t.getMessage());
@@ -534,6 +538,7 @@ private void doShutDown(CloseConnectionReason closeConnectionReason, @Nullable R
                 e.printStackTrace();
             }
 
+            //noinspection UnstableApiUsage
             MoreExecutors.shutdownAndAwaitTermination(singleThreadExecutor, 500, TimeUnit.MILLISECONDS);
 
             log.debug("Connection shutdown complete " + this.toString());
@@ -705,7 +710,7 @@ public void run() {
                         Thread.sleep(20);
                     }
 
-                    // Reading the protobuffer message from the inputstream
+                    // Reading the protobuffer message from the inputStream
                     protobuf.NetworkEnvelope proto = protobuf.NetworkEnvelope.parseDelimitedFrom(protoInputStream);
 
                     if (proto == null) {
@@ -793,7 +798,20 @@ && reportInvalidRequest(RuleViolation.WRONG_NETWORK_ID)) {
                     if (networkEnvelope instanceof SupportedCapabilitiesMessage) {
                         Capabilities supportedCapabilities = ((SupportedCapabilitiesMessage) networkEnvelope).getSupportedCapabilities();
                         if (supportedCapabilities != null) {
-                            capabilities.set(supportedCapabilities);
+                            if (!capabilities.equals(supportedCapabilities)) {
+                                if (!Capabilities.hasMandatoryCapability(capabilities)) {
+                                    shutDown(CloseConnectionReason.RULE_VIOLATION);
+                                    return;
+                                }
+
+                                capabilities.set(supportedCapabilities);
+                                capabilitiesListeners.forEach(weakListener -> {
+                                    SupportedCapabilitiesListener supportedCapabilitiesListener = weakListener.get();
+                                    if (supportedCapabilitiesListener != null) {
+                                        UserThread.execute(() -> supportedCapabilitiesListener.onChanged(supportedCapabilities));
+                                    }
+                                });
+                            }
                         }
                     }
 

p2p/src/main/java/bisq/network/p2p/network/NetworkNode.java

@@ -96,7 +96,8 @@ public abstract class NetworkNode implements MessageListener {
     // when the events happen.
     abstract public void start(@Nullable SetupListener setupListener);
 
-    public SettableFuture<Connection> sendMessage(@NotNull NodeAddress peersNodeAddress, NetworkEnvelope networkEnvelope) {
+    public SettableFuture<Connection> sendMessage(@NotNull NodeAddress peersNodeAddress,
+                                                  NetworkEnvelope networkEnvelope) {
         log.debug("sendMessage: peersNodeAddress=" + peersNodeAddress + "\n\tmessage=" + Utilities.toTruncatedString(networkEnvelope));
         checkNotNull(peersNodeAddress, "peerAddress must not be null");
 
@@ -112,9 +113,9 @@ public SettableFuture<Connection> sendMessage(@NotNull NodeAddress peersNodeAddr
 
             final SettableFuture<Connection> resultFuture = SettableFuture.create();
             ListenableFuture<Connection> future = executorService.submit(() -> {
-                Thread.currentThread().setName("NetworkNode:SendMessage-to-" + peersNodeAddress);
+                Thread.currentThread().setName("NetworkNode:SendMessage-to-" + peersNodeAddress.getFullAddress());
 
-                if(peersNodeAddress.equals(getNodeAddress())){
+                if (peersNodeAddress.equals(getNodeAddress())) {
                     throw new ConnectException("We do not send a message to ourselves");
                 }
 
@@ -162,7 +163,8 @@ public void onConnection(Connection connection) {
                             }
 
                             @Override
-                            public void onDisconnect(CloseConnectionReason closeConnectionReason, Connection connection) {
+                            public void onDisconnect(CloseConnectionReason closeConnectionReason,
+                                                     Connection connection) {
                                 log.trace("onDisconnect connectionListener\n\tconnection={}" + connection);
                                 //noinspection SuspiciousMethodCalls
                                 outBoundConnections.remove(connection);
@@ -264,7 +266,8 @@ public Socks5Proxy getSocksProxy() {
     public SettableFuture<Connection> sendMessage(Connection connection, NetworkEnvelope networkEnvelope) {
         // connection.sendMessage might take a bit (compression, write to stream), so we use a thread to not block
         ListenableFuture<Connection> future = executorService.submit(() -> {
-            Thread.currentThread().setName("NetworkNode:SendMessage-to-" + connection.getUid());
+            String id = connection.getPeersNodeAddressOptional().isPresent() ? connection.getPeersNodeAddressOptional().get().getFullAddress() : connection.getUid();
+            Thread.currentThread().setName("NetworkNode:SendMessage-to-" + id);
             connection.sendMessage(networkEnvelope);
             return connection;
         });

p2p/src/main/java/bisq/network/p2p/peers/peerexchange/Peer.java

@@ -97,8 +97,9 @@ public Date getDate() {
 
     @Override
     public void onChanged(Capabilities supportedCapabilities) {
-        if (!supportedCapabilities.isEmpty())
+        if (!supportedCapabilities.isEmpty()) {
             capabilities.set(supportedCapabilities);
+        }
     }