Add PeerGroupStore

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

@@ -133,7 +133,9 @@ public ServiceNode(Config config,
             }
 
             if (services.contains(Service.MONITOR)) {
-                monitorService = Optional.of(new MonitorService(defaultNode, peerGroupService));
+                monitorService = Optional.of(new MonitorService(defaultNode,
+                        peerGroupService.getPeerGroup(), 
+                        peerGroupService.getPeerGroupStore()));
             }
         }
 

network/src/main/java/bisq/network/p2p/services/monitor/MonitorService.java

@@ -22,7 +22,7 @@
 import bisq.network.p2p.node.Node;
 import bisq.network.p2p.services.peergroup.Peer;
 import bisq.network.p2p.services.peergroup.PeerGroup;
-import bisq.network.p2p.services.peergroup.PeerGroupService;
+import bisq.network.p2p.services.peergroup.PeerGroupStore;
 
 import java.text.SimpleDateFormat;
 import java.util.Comparator;
@@ -32,10 +32,12 @@
 public class MonitorService {
     private final Node node;
     private final PeerGroup peerGroup;
+    private final PeerGroupStore peerGroupStore;
 
-    public MonitorService(Node node, PeerGroupService peerGroupService) {
+    public MonitorService(Node node, PeerGroup peerGroup, PeerGroupStore peerGroupStore) {
         this.node = node;
-        this.peerGroup = peerGroupService.getPeerGroup();
+        this.peerGroup = peerGroup;
+        this.peerGroupStore = peerGroupStore;
     }
 
     public CompletableFuture<Void> shutdown() {
@@ -62,7 +64,7 @@ public String getPeerGroupInfo() {
                 .forEach(connection -> appendConnectionInfo(sb, connection, false));
         sb.append("\n").append("Reported peers (").append(peerGroup.getReportedPeers().size()).append("): ").append(peerGroup.getReportedPeers().stream()
                 .map(Peer::getAddress).sorted(Comparator.comparing(Address::getPort)).collect(Collectors.toList()));
-        sb.append("\n").append("Persisted peers: ").append(peerGroup.getPersistedPeers().stream()
+        sb.append("\n").append("Persisted peers: ").append(peerGroupStore.getPersistedPeers().stream()
                 .map(Peer::getAddress).sorted(Comparator.comparing(Address::getPort)).collect(Collectors.toList()));
         return sb.append("\n").toString();
     }

network/src/main/java/bisq/network/p2p/services/peergroup/PeerGroup.java

@@ -66,17 +66,16 @@ public static Config from(com.typesafe.config.Config typesafeConfig) {
     @Getter
     private final List<Address> seedNodeAddresses;
     private final BanList banList;
+    private final PeerGroupStore peerGroupStore;
     @Getter
     private final Set<Peer> reportedPeers = new CopyOnWriteArraySet<>();
-    //todo persist
-    @Getter
-    private final Set<Peer> persistedPeers = new CopyOnWriteArraySet<>();
 
-    public PeerGroup(Node node, Config config, List<Address> seedNodeAddresses, BanList banList) {
+    public PeerGroup(Node node, Config config, List<Address> seedNodeAddresses, BanList banList, PeerGroupStore peerGroupStore) {
         this.node = node;
         this.config = config;
         this.seedNodeAddresses = seedNodeAddresses;
         this.banList = banList;
+        this.peerGroupStore = peerGroupStore;
     }
 
     ///////////////////////////////////////////////////////////////////////////////////////////////////
@@ -91,10 +90,6 @@ public void removeReportedPeers(Collection<Peer> peers) {
         reportedPeers.removeAll(peers);
     }
 
-    public void removePersistedPeers(Collection<Peer> peers) {
-        persistedPeers.removeAll(peers);
-    }
-
 
     ///////////////////////////////////////////////////////////////////////////////////////////////////
     // Connections

network/src/main/java/bisq/network/p2p/services/peergroup/PeerGroupService.java

@@ -27,6 +27,8 @@
 import bisq.network.p2p.services.peergroup.exchange.PeerExchangeStrategy;
 import bisq.network.p2p.services.peergroup.keepalive.KeepAliveService;
 import bisq.network.p2p.services.peergroup.validateaddress.AddressValidationService;
+import bisq.persistence.Persistence;
+import bisq.persistence.PersistenceClient;
 import bisq.persistence.PersistenceService;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -42,7 +44,7 @@
 import static java.util.concurrent.TimeUnit.*;
 
 @Slf4j
-public class PeerGroupService {
+public class PeerGroupService implements PersistenceClient<PeerGroupStore>, PersistedPeersHandler {
     public enum State {
         NEW,
         STARTING,
@@ -64,9 +66,14 @@ public interface Listener {
     private final KeepAliveService keepAliveService;
     private final AddressValidationService addressValidationService;
     private Optional<Scheduler> scheduler = Optional.empty();
+
+    @Getter
+    private final PeerGroupStore persistableStore = new PeerGroupStore();
     @Getter
     public AtomicReference<PeerGroupService.State> state = new AtomicReference<>(PeerGroupService.State.NEW);
     private final Set<Listener> listeners = new CopyOnWriteArraySet<>();
+    @Getter
+    private final Persistence<PeerGroupStore> persistence;
 
     public static record Config(PeerGroup.Config peerGroupConfig,
                                 PeerExchangeStrategy.Config peerExchangeConfig,
@@ -101,10 +108,14 @@ public PeerGroupService(PersistenceService persistenceService, Node node, BanLis
         this.node = node;
         this.banList = banList;
         this.config = config;
-        peerGroup = new PeerGroup(node, config.peerGroupConfig, seedNodeAddresses, banList);
-        peerExchangeService = new PeerExchangeService(node, new PeerExchangeStrategy(peerGroup, config.peerExchangeConfig()));
+        peerGroup = new PeerGroup(node, config.peerGroupConfig, seedNodeAddresses, banList, persistableStore);
+        PeerExchangeStrategy peerExchangeStrategy = new PeerExchangeStrategy(peerGroup,
+                config.peerExchangeConfig(),
+                persistableStore);
+        peerExchangeService = new PeerExchangeService(node, peerExchangeStrategy, this);
         keepAliveService = new KeepAliveService(node, peerGroup, config.keepAliveServiceConfig());
         addressValidationService = new AddressValidationService(node, banList);
+        persistence = persistenceService.getOrCreatePersistence(this, persistableStore);
     }
 
     public void start() {
@@ -120,6 +131,15 @@ public void start() {
         setState(State.RUNNING);
     }
 
+    ///////////////////////////////////////////////////////////////////////////////////////////////////
+    // PersistedPeersHandler
+    ///////////////////////////////////////////////////////////////////////////////////////////////////
+
+    public void addPersistedPeers(Set<Peer> peers) {
+        persistableStore.getPersistedPeers().addAll(peers);
+        persist();
+    }
+
     private void runBlockingTasks() {
         log.debug("Node {} called runBlockingTasks", node);
         try {
@@ -245,7 +265,6 @@ private void maybeCloseExceedingInboundConnections() {
 
     }
 
-
     private void maybeCloseExceedingConnections() {
         log.debug("Node {} called maybeCloseExceedingConnections", node);
         Comparator<Connection> comparator = peerGroup.getConnectionAgeComparator().reversed();
@@ -311,13 +330,15 @@ private void maybeRemoveReportedPeers() {
     }
 
     private void maybeRemovePersistedPeers() {
-        List<Peer> persistedPeers = new ArrayList<>(peerGroup.getPersistedPeers());
+        List<Peer> persistedPeers = new ArrayList<>(persistableStore.getPersistedPeers());
         int exceeding = persistedPeers.size() - config.maxPersisted();
         if (exceeding > 0) {
             persistedPeers.sort(Comparator.comparing(Peer::getDate));
             List<Peer> candidates = persistedPeers.subList(0, Math.min(exceeding, persistedPeers.size()));
             log.info("Remove {} persisted peers: {}", candidates.size(), candidates);
-            peerGroup.removePersistedPeers(candidates);
+
+            persistableStore.getPersistedPeers().removeAll(candidates);
+            persist();
         }
     }
 
@@ -337,6 +358,11 @@ private void setState(PeerGroupService.State newState) {
         runAsync(() -> listeners.forEach(e -> e.onStateChanged(newState)), NetworkService.DISPATCHER);
     }
 
+
+    public PeerGroupStore getPeerGroupStore() {
+        return persistableStore;
+    }
+
     ///////////////////////////////////////////////////////////////////////////////////////////////////
     // Utils
     ///////////////////////////////////////////////////////////////////////////////////////////////////
@@ -354,4 +380,5 @@ private boolean notBootstrapping(Connection connection) {
     private int getMissingOutboundConnections() {
         return peerGroup.getMinOutboundConnections() - (int) peerGroup.getOutboundConnections().count();
     }
+
 }

network/src/main/java/bisq/network/p2p/services/peergroup/PeerGroupStore.java

@@ -0,0 +1,76 @@
+/*
+ * This file is part of Bisq.
+ *
+ * Bisq is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or (at
+ * your option) any later version.
+ *
+ * Bisq is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public
+ * License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with Bisq. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package bisq.network.p2p.services.peergroup;
+import bisq.common.proto.ProtoResolver;
+import bisq.common.proto.UnresolvableProtobufMessageException;
+import bisq.persistence.PersistableStore;
+import com.google.protobuf.InvalidProtocolBufferException;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class PeerGroupStore implements PersistableStore<PeerGroupStore> {
+    @Getter
+    private final Set<Peer> persistedPeers = new CopyOnWriteArraySet<>();
+
+    public PeerGroupStore() {
+    }
+
+    private PeerGroupStore(Set<Peer> persistedPeers) {
+        this.persistedPeers.addAll(persistedPeers);
+    }
+
+    @Override
+    public bisq.network.protobuf.PeerGroupStore toProto() {
+        return bisq.network.protobuf.PeerGroupStore.newBuilder().addAllPersistedPeers(persistedPeers.stream()
+                        .map(Peer::toProto)
+                        .collect(Collectors.toSet()))
+                .build();
+    }
+
+    public static PeerGroupStore fromProto(bisq.network.protobuf.PeerGroupStore proto) {
+        return new PeerGroupStore(proto.getPersistedPeersList().stream()
+                .map(Peer::fromProto).collect(Collectors.toSet()));
+    }
+
+    @Override
+    public ProtoResolver<PersistableStore<?>> getResolver() {
+        return any -> {
+            try {
+                return fromProto(any.unpack(bisq.network.protobuf.PeerGroupStore.class));
+            } catch (InvalidProtocolBufferException e) {
+                throw new UnresolvableProtobufMessageException(e);
+            }
+        };
+    }
+
+    @Override
+    public PeerGroupStore getClone() {
+        return new PeerGroupStore(persistedPeers);
+    }
+
+    @Override
+    public void applyPersisted(PeerGroupStore persisted) {
+        persistedPeers.clear();
+        persistedPeers.addAll(persisted.getPersistedPeers());
+    }
+}

network/src/main/java/bisq/network/p2p/services/peergroup/PersistedPeersHandler.java

@@ -0,0 +1,24 @@
+/*
+ * This file is part of Bisq.
+ *
+ * Bisq is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or (at
+ * your option) any later version.
+ *
+ * Bisq is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public
+ * License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with Bisq. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package bisq.network.p2p.services.peergroup;
+
+import java.util.Set;
+
+public interface PersistedPeersHandler {
+    void addPersistedPeers(Set<Peer> peers);
+}

network/src/main/java/bisq/network/p2p/services/peergroup/exchange/PeerExchangeService.java

@@ -26,6 +26,7 @@
 import bisq.network.p2p.node.Connection;
 import bisq.network.p2p.node.Node;
 import bisq.network.p2p.services.peergroup.Peer;
+import bisq.network.p2p.services.peergroup.PersistedPeersHandler;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
@@ -52,14 +53,20 @@ public class PeerExchangeService implements Node.Listener {
 
     private final Node node;
     private final PeerExchangeStrategy peerExchangeStrategy;
+
+    // todo if persisted peer needs to be written from that class we can use the addPersistedPeerHandler to delegate it 
+    // to the PeerGroupService. We do not want a dependency from PeerExchangeService to PeerGroupService as 
+    // PeerExchangeService got created by PeerGroupService
+    private final PersistedPeersHandler persistedPeersHandler;
     private final Map<String, PeerExchangeRequestHandler> requestHandlerMap = new ConcurrentHashMap<>();
     private int doInitialPeerExchangeDelaySec = 1; //todo move to config
     private volatile boolean isStopped;
     private Optional<Scheduler> scheduler = Optional.empty();
 
-    public PeerExchangeService(Node node, PeerExchangeStrategy peerExchangeStrategy) {
+    public PeerExchangeService(Node node, PeerExchangeStrategy peerExchangeStrategy, PersistedPeersHandler persistedPeersHandler) {
         this.node = node;
         this.peerExchangeStrategy = peerExchangeStrategy;
+        this.persistedPeersHandler = persistedPeersHandler;
         this.node.addListener(this);
     }
 

network/src/main/java/bisq/network/p2p/services/peergroup/exchange/PeerExchangeStrategy.java

@@ -20,6 +20,7 @@
 import bisq.network.p2p.node.Address;
 import bisq.network.p2p.services.peergroup.Peer;
 import bisq.network.p2p.services.peergroup.PeerGroup;
+import bisq.network.p2p.services.peergroup.PeerGroupStore;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
@@ -62,11 +63,13 @@ public static Config from(com.typesafe.config.Config typesafeConfig) {
 
     private final PeerGroup peerGroup;
     private final Config config;
+    private final PeerGroupStore peerGroupStore;
     private final Set<Address> usedAddresses = new CopyOnWriteArraySet<>();
 
-    public PeerExchangeStrategy(PeerGroup peerGroup, Config config) {
+    public PeerExchangeStrategy(PeerGroup peerGroup, Config config, PeerGroupStore peerGroupStore) {
         this.peerGroup = peerGroup;
         this.config = config;
+        this.peerGroupStore = peerGroupStore;
     }
 
     List<Address> getAddressesForInitialPeerExchange() {
@@ -203,7 +206,7 @@ private Set<Address> getReported() {
     }
 
     private Set<Address> getPersisted() {
-        return peerGroup.getPersistedPeers().stream()
+        return peerGroupStore.getPersistedPeers().stream()
                 .filter(peerGroup::isNotInQuarantine)
                 .sorted(Comparator.comparing(Peer::getDate))
                 .map(Peer::getAddress)

network/src/main/java/bisq/network/p2p/services/peergroup/keepalive/KeepAliveService.java

@@ -115,5 +115,4 @@ public void onDisconnect(Connection connection, CloseReason closeReason) {
     private boolean isRequired(Connection connection) {
         return System.currentTimeMillis() - connection.getMetrics().getLastUpdate().get() > config.maxIdleTime();
     }
-
 }

network/src/main/proto/network.proto

@@ -263,11 +263,16 @@ message NetworkEnvelope {
 message DataStore {
   message MapEntry {
     common.ByteArray key = 1;
-    DataRequest value = 2; 
+    DataRequest value = 2;
   }
   repeated MapEntry mapEntries = 1;
 }
 
 message NetworkIdStore {
   map<string, NetworkId> networkIdByNodeId = 1;
-}
+}
+
+message PeerGroupStore {
+  repeated Peer persistedPeers = 1;
+}
+