Persist and republish mailbox messages

common/src/main/java/bisq/common/config/Config.java

@@ -120,6 +120,7 @@ public class Config {
     public static final String API_PASSWORD = "apiPassword";
     public static final String API_PORT = "apiPort";
     public static final String PREVENT_PERIODIC_SHUTDOWN_AT_SEED_NODE = "preventPeriodicShutdownAtSeedNode";
+    public static final String REPUBLISH_MAILBOX_ENTRIES = "republishMailboxEntries";
 
     // Default values for certain options
     public static final int UNSPECIFIED_PORT = -1;
@@ -207,6 +208,7 @@ public class Config {
     public final String apiPassword;
     public final int apiPort;
     public final boolean preventPeriodicShutdownAtSeedNode;
+    public final boolean republishMailboxEntries;
 
     // Properties derived from options but not exposed as options themselves
     public final File torDir;
@@ -648,6 +650,13 @@ public Config(String defaultAppName, File defaultUserDataDir, String... args) {
                         .ofType(boolean.class)
                         .defaultsTo(false);
 
+        ArgumentAcceptingOptionSpec<Boolean> republishMailboxEntriesOpt =
+                parser.accepts(REPUBLISH_MAILBOX_ENTRIES,
+                        "Republish mailbox messages at startup")
+                        .withRequiredArg()
+                        .ofType(boolean.class)
+                        .defaultsTo(false);
+
         try {
             CompositeOptionSet options = new CompositeOptionSet();
 
@@ -764,6 +773,7 @@ public Config(String defaultAppName, File defaultUserDataDir, String... args) {
             this.apiPassword = options.valueOf(apiPasswordOpt);
             this.apiPort = options.valueOf(apiPortOpt);
             this.preventPeriodicShutdownAtSeedNode = options.valueOf(preventPeriodicShutdownAtSeedNodeOpt);
+            this.republishMailboxEntries = options.valueOf(republishMailboxEntriesOpt);
         } catch (OptionException ex) {
             throw new ConfigException("problem parsing option '%s': %s",
                     ex.options().get(0),

core/src/main/java/bisq/core/alert/Alert.java

@@ -47,6 +47,8 @@
 @ToString
 @Slf4j
 public final class Alert implements ProtectedStoragePayload, ExpirablePayload {
+    public static final long TTL = TimeUnit.DAYS.toMillis(90);
+
     private final String message;
     private final boolean isUpdateInfo;
     private final String version;
@@ -131,7 +133,7 @@ public static Alert fromProto(protobuf.Alert proto) {
 
     @Override
     public long getTTL() {
-        return TimeUnit.DAYS.toMillis(90);
+        return TTL;
     }
 
     public void setSigAndPubKey(String signatureAsBase64, PublicKey ownerPubKey) {

core/src/main/java/bisq/core/alert/PrivateNotificationManager.java

@@ -21,6 +21,7 @@
 import bisq.network.p2p.NodeAddress;
 import bisq.network.p2p.P2PService;
 import bisq.network.p2p.SendMailboxMessageListener;
+import bisq.network.p2p.mailbox.MailboxMessageService;
 
 import bisq.common.app.DevEnv;
 import bisq.common.config.Config;
@@ -49,20 +50,24 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import static org.bitcoinj.core.Utils.HEX;
 
 public class PrivateNotificationManager {
     private static final Logger log = LoggerFactory.getLogger(PrivateNotificationManager.class);
 
     private final P2PService p2PService;
+    private final MailboxMessageService mailboxMessageService;
     private final KeyRing keyRing;
     private final ObjectProperty<PrivateNotificationPayload> privateNotificationMessageProperty = new SimpleObjectProperty<>();
 
     // Pub key for developer global privateNotification message
     private final String pubKeyAsHex;
 
     private ECKey privateNotificationSigningKey;
-    private DecryptedMessageWithPubKey decryptedMessageWithPubKey;
+    @Nullable
+    private PrivateNotificationMessage privateNotificationMessage;
 
 
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -71,26 +76,27 @@ public class PrivateNotificationManager {
 
     @Inject
     public PrivateNotificationManager(P2PService p2PService,
+                                      MailboxMessageService mailboxMessageService,
                                       KeyRing keyRing,
                                       @Named(Config.IGNORE_DEV_MSG) boolean ignoreDevMsg,
                                       @Named(Config.USE_DEV_PRIVILEGE_KEYS) boolean useDevPrivilegeKeys) {
         this.p2PService = p2PService;
+        this.mailboxMessageService = mailboxMessageService;
         this.keyRing = keyRing;
 
         if (!ignoreDevMsg) {
             this.p2PService.addDecryptedDirectMessageListener(this::handleMessage);
-            this.p2PService.addDecryptedMailboxListener(this::handleMessage);
+            this.mailboxMessageService.addDecryptedMailboxListener(this::handleMessage);
         }
         pubKeyAsHex = useDevPrivilegeKeys ?
                 DevEnv.DEV_PRIVILEGE_PUB_KEY :
                 "02ba7c5de295adfe57b60029f3637a2c6b1d0e969a8aaefb9e0ddc3a7963f26925";
     }
 
     private void handleMessage(DecryptedMessageWithPubKey decryptedMessageWithPubKey, NodeAddress senderNodeAddress) {
-        this.decryptedMessageWithPubKey = decryptedMessageWithPubKey;
         NetworkEnvelope networkEnvelope = decryptedMessageWithPubKey.getNetworkEnvelope();
         if (networkEnvelope instanceof PrivateNotificationMessage) {
-            PrivateNotificationMessage privateNotificationMessage = (PrivateNotificationMessage) networkEnvelope;
+            privateNotificationMessage = (PrivateNotificationMessage) networkEnvelope;
             log.info("Received PrivateNotificationMessage from {} with uid={}",
                     senderNodeAddress, privateNotificationMessage.getUid());
             if (privateNotificationMessage.getSenderNodeAddress().equals(senderNodeAddress)) {
@@ -112,8 +118,11 @@ public ReadOnlyObjectProperty<PrivateNotificationPayload> privateNotificationPro
         return privateNotificationMessageProperty;
     }
 
-    public boolean sendPrivateNotificationMessageIfKeyIsValid(PrivateNotificationPayload privateNotification, PubKeyRing pubKeyRing, NodeAddress peersNodeAddress,
-                                                              String privKeyString, SendMailboxMessageListener sendMailboxMessageListener) {
+    public boolean sendPrivateNotificationMessageIfKeyIsValid(PrivateNotificationPayload privateNotification,
+                                                              PubKeyRing pubKeyRing,
+                                                              NodeAddress peersNodeAddress,
+                                                              String privKeyString,
+                                                              SendMailboxMessageListener sendMailboxMessageListener) {
         boolean isKeyValid = isKeyValid(privKeyString);
         if (isKeyValid) {
             signAndAddSignatureToPrivateNotificationMessage(privateNotification);
@@ -123,7 +132,7 @@ public boolean sendPrivateNotificationMessageIfKeyIsValid(PrivateNotificationPay
                     UUID.randomUUID().toString());
             log.info("Send {} to peer {}. uid={}",
                     message.getClass().getSimpleName(), peersNodeAddress, message.getUid());
-            p2PService.sendEncryptedMailboxMessage(peersNodeAddress,
+            mailboxMessageService.sendEncryptedMailboxMessage(peersNodeAddress,
                     pubKeyRing,
                     message,
                     sendMailboxMessageListener);
@@ -133,7 +142,9 @@ public boolean sendPrivateNotificationMessageIfKeyIsValid(PrivateNotificationPay
     }
 
     public void removePrivateNotification() {
-        p2PService.removeMailboxMsg(decryptedMessageWithPubKey);
+        if (privateNotificationMessage != null) {
+            mailboxMessageService.removeMailboxMsg(privateNotificationMessage);
+        }
     }
 
     private boolean isKeyValid(String privKeyString) {

core/src/main/java/bisq/core/alert/PrivateNotificationMessage.java

@@ -17,8 +17,8 @@
 
 package bisq.core.alert;
 
-import bisq.network.p2p.MailboxMessage;
 import bisq.network.p2p.NodeAddress;
+import bisq.network.p2p.mailbox.MailboxMessage;
 
 import bisq.common.app.Version;
 import bisq.common.proto.network.NetworkEnvelope;

core/src/main/java/bisq/core/dao/governance/proposal/storage/temp/TempProposalPayload.java

@@ -56,6 +56,9 @@
 @FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE)
 public class TempProposalPayload implements ProcessOncePersistableNetworkPayload, ProtectedStoragePayload,
         ExpirablePayload, PersistablePayload {
+    // We keep data 2 months to be safe if we increase durations of cycle. Also give a bit more resilience in case
+    // of any issues with the append-only data store
+    public static final long TTL = TimeUnit.DAYS.toMillis(60);
 
     protected final Proposal proposal;
     protected final byte[] ownerPubKeyEncoded;
@@ -124,8 +127,6 @@ public PublicKey getOwnerPubKey() {
 
     @Override
     public long getTTL() {
-        // We keep data 2 months to be safe if we increase durations of cycle. Also give a bit more resilience in case
-        // of any issues with the append-only data store
-        return TimeUnit.DAYS.toMillis(60);
+        return TTL;
     }
 }

core/src/main/java/bisq/core/dao/node/full/RpcService.java

@@ -155,8 +155,8 @@ void setup(ResultHandler resultHandler, Consumer<Throwable> errorHandler) {
                 nodeConfig.setProperty("node.bitcoind.rpc.port", Integer.toString(rpcPort));
                 nodeConfig.setProperty("node.bitcoind.notification.block.port", Integer.toString(rpcBlockPort));
                 nodeConfig.setProperty("node.bitcoind.notification.block.host", rpcBlockHost);
-                nodeConfig.setProperty("node.bitcoind.notification.alert.port", Integer.toString(bisq.network.p2p.Utils.findFreeSystemPort()));
-                nodeConfig.setProperty("node.bitcoind.notification.wallet.port", Integer.toString(bisq.network.p2p.Utils.findFreeSystemPort()));
+                nodeConfig.setProperty("node.bitcoind.notification.alert.port", Integer.toString(bisq.network.utils.Utils.findFreeSystemPort()));
+                nodeConfig.setProperty("node.bitcoind.notification.wallet.port", Integer.toString(bisq.network.utils.Utils.findFreeSystemPort()));
 
                 nodeConfig.setProperty("node.bitcoind.http.auth_scheme", "Basic");
                 BtcdClientImpl client = new BtcdClientImpl(httpProvider, nodeConfig);

core/src/main/java/bisq/core/filter/Filter.java

@@ -47,6 +47,8 @@
 @Slf4j
 @Value
 public final class Filter implements ProtectedStoragePayload, ExpirablePayload {
+    public static final long TTL = TimeUnit.DAYS.toMillis(180);
+
     private final List<String> bannedOfferIds;
     private final List<String> nodeAddressesBannedFromTrading;
     private final List<String> bannedAutoConfExplorers;
@@ -361,7 +363,7 @@ public static Filter fromProto(protobuf.Filter proto) {
 
     @Override
     public long getTTL() {
-        return TimeUnit.DAYS.toMillis(180);
+        return TTL;
     }
 
     @Override

core/src/main/java/bisq/core/network/p2p/inventory/GetInventoryRequester.java

@@ -41,7 +41,7 @@
 
 @Slf4j
 public class GetInventoryRequester implements MessageListener, ConnectionListener {
-    private final static int TIMEOUT_SEC = 90;
+    private final static int TIMEOUT_SEC = 180;
 
     private final NetworkNode networkNode;
     private final NodeAddress nodeAddress;

core/src/main/java/bisq/core/offer/OfferPayload.java

@@ -55,6 +55,7 @@
 @Getter
 @Slf4j
 public final class OfferPayload implements ProtectedStoragePayload, ExpirablePayload, RequiresOwnerIsOnlinePayload {
+    public static final long TTL = TimeUnit.MINUTES.toMillis(9);
 
     ///////////////////////////////////////////////////////////////////////////////////////////
     // Enum
@@ -373,7 +374,7 @@ public static OfferPayload fromProto(protobuf.OfferPayload proto) {
 
     @Override
     public long getTTL() {
-        return TimeUnit.MINUTES.toMillis(9);
+        return TTL;
     }
 
     @Override

core/src/main/java/bisq/core/proto/persistable/CorePersistenceProtoResolver.java

@@ -43,9 +43,10 @@
 import bisq.core.user.PreferencesPayload;
 import bisq.core.user.UserPayload;
 
-import bisq.network.p2p.MailboxMessageList;
 import bisq.network.p2p.mailbox.IgnoredMailboxMap;
+import bisq.network.p2p.mailbox.MailboxMessageList;
 import bisq.network.p2p.peers.peerexchange.PeerList;
+import bisq.network.p2p.storage.persistence.RemovedPayloadsMap;
 import bisq.network.p2p.storage.persistence.SequenceNumberMap;
 
 import bisq.common.proto.ProtobufferRuntimeException;
@@ -135,6 +136,8 @@ public PersistableEnvelope fromProto(protobuf.PersistableEnvelope proto) {
                     return MailboxMessageList.fromProto(proto.getMailboxMessageList(), networkProtoResolver);
                 case IGNORED_MAILBOX_MAP:
                     return IgnoredMailboxMap.fromProto(proto.getIgnoredMailboxMap());
+                case REMOVED_PAYLOADS_MAP:
+                    return RemovedPayloadsMap.fromProto(proto.getRemovedPayloadsMap());
                 default:
                     throw new ProtobufferRuntimeException("Unknown proto message case(PB.PersistableEnvelope). " +
                             "messageCase=" + proto.getMessageCase() + "; proto raw data=" + proto.toString());

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

@@ -35,10 +35,11 @@
 import bisq.core.user.Preferences;
 import bisq.core.user.User;
 
-import bisq.network.p2p.P2PService;
 import bisq.network.p2p.mailbox.IgnoredMailboxService;
+import bisq.network.p2p.mailbox.MailboxMessageService;
 import bisq.network.p2p.peers.PeerManager;
 import bisq.network.p2p.storage.P2PDataStorage;
+import bisq.network.p2p.storage.persistence.RemovedPayloadsService;
 
 import bisq.common.config.Config;
 import bisq.common.proto.persistable.PersistedDataHost;
@@ -68,8 +69,9 @@ public static List<PersistedDataHost> getPersistedDataHosts(Injector injector) {
         persistedDataHosts.add(injector.getInstance(RefundDisputeListService.class));
         persistedDataHosts.add(injector.getInstance(P2PDataStorage.class));
         persistedDataHosts.add(injector.getInstance(PeerManager.class));
-        persistedDataHosts.add(injector.getInstance(P2PService.class));
+        persistedDataHosts.add(injector.getInstance(MailboxMessageService.class));
         persistedDataHosts.add(injector.getInstance(IgnoredMailboxService.class));
+        persistedDataHosts.add(injector.getInstance(RemovedPayloadsService.class));
 
         if (injector.getInstance(Config.class).daoActivated) {
             persistedDataHosts.add(injector.getInstance(BallotListService.class));

core/src/main/java/bisq/core/support/SupportManager.java

@@ -27,6 +27,7 @@
 import bisq.network.p2p.NodeAddress;
 import bisq.network.p2p.P2PService;
 import bisq.network.p2p.SendMailboxMessageListener;
+import bisq.network.p2p.mailbox.MailboxMessageService;
 
 import bisq.common.Timer;
 import bisq.common.UserThread;
@@ -49,6 +50,7 @@ public abstract class SupportManager {
     protected final Map<String, Timer> delayMsgMap = new HashMap<>();
     private final CopyOnWriteArraySet<DecryptedMessageWithPubKey> decryptedMailboxMessageWithPubKeys = new CopyOnWriteArraySet<>();
     private final CopyOnWriteArraySet<DecryptedMessageWithPubKey> decryptedDirectMessageWithPubKeys = new CopyOnWriteArraySet<>();
+    protected final MailboxMessageService mailboxMessageService;
     private boolean allServicesInitialized;
 
 
@@ -58,6 +60,8 @@ public abstract class SupportManager {
 
     public SupportManager(P2PService p2PService, WalletsSetup walletsSetup) {
         this.p2PService = p2PService;
+        mailboxMessageService = p2PService.getMailboxMessageService();
+
         this.walletsSetup = walletsSetup;
 
         // We get first the message handler called then the onBootstrapped
@@ -67,7 +71,7 @@ public SupportManager(P2PService p2PService, WalletsSetup walletsSetup) {
             decryptedDirectMessageWithPubKeys.add(decryptedMessageWithPubKey);
             tryApplyMessages();
         });
-        p2PService.addDecryptedMailboxListener((decryptedMessageWithPubKey, senderAddress) -> {
+        mailboxMessageService.addDecryptedMailboxListener((decryptedMessageWithPubKey, senderAddress) -> {
             // As decryptedMailboxMessageWithPubKeys is a CopyOnWriteArraySet we do not need to check if it was
             // already stored
             decryptedMailboxMessageWithPubKeys.add(decryptedMessageWithPubKey);
@@ -80,7 +84,7 @@ public SupportManager(P2PService p2PService, WalletsSetup walletsSetup) {
     // Abstract methods
     ///////////////////////////////////////////////////////////////////////////////////////////
 
-    protected abstract void dispatchMessage(SupportMessage networkEnvelope);
+    protected abstract void onSupportMessage(SupportMessage networkEnvelope);
 
     public abstract NodeAddress getPeerNodeAddress(ChatMessage message);
 
@@ -155,8 +159,7 @@ protected void onChatMessage(ChatMessage chatMessage) {
             sendAckMessage(chatMessage, receiverPubKeyRing, true, null);
     }
 
-    private void onAckMessage(AckMessage ackMessage,
-                              @Nullable DecryptedMessageWithPubKey decryptedMessageWithPubKey) {
+    private void onAckMessage(AckMessage ackMessage) {
         if (ackMessage.getSourceType() == getAckMessageSourceType()) {
             if (ackMessage.isSuccess()) {
                 log.info("Received AckMessage for {} with tradeId {} and uid {}",
@@ -175,9 +178,6 @@ private void onAckMessage(AckMessage ackMessage,
                             msg.setAckError(ackMessage.getErrorMessage());
                     });
             requestPersistence();
-
-            if (decryptedMessageWithPubKey != null)
-                p2PService.removeMailboxMsg(decryptedMessageWithPubKey);
         }
     }
 
@@ -195,7 +195,7 @@ public ChatMessage sendChatMessage(ChatMessage message) {
             log.info("Send {} to peer {}. tradeId={}, uid={}",
                     message.getClass().getSimpleName(), peersNodeAddress, message.getTradeId(), message.getUid());
 
-            p2PService.sendEncryptedMailboxMessage(peersNodeAddress,
+            mailboxMessageService.sendEncryptedMailboxMessage(peersNodeAddress,
                     receiverPubKeyRing,
                     message,
                     new SendMailboxMessageListener() {
@@ -243,7 +243,7 @@ protected void sendAckMessage(SupportMessage supportMessage, PubKeyRing peersPub
         final NodeAddress peersNodeAddress = supportMessage.getSenderNodeAddress();
         log.info("Send AckMessage for {} to peer {}. tradeId={}, uid={}",
                 ackMessage.getSourceMsgClassName(), peersNodeAddress, tradeId, uid);
-        p2PService.sendEncryptedMailboxMessage(
+        mailboxMessageService.sendEncryptedMailboxMessage(
                 peersNodeAddress,
                 peersPubKeyRing,
                 ackMessage,
@@ -302,9 +302,9 @@ private void applyMessages() {
         decryptedDirectMessageWithPubKeys.forEach(decryptedMessageWithPubKey -> {
             NetworkEnvelope networkEnvelope = decryptedMessageWithPubKey.getNetworkEnvelope();
             if (networkEnvelope instanceof SupportMessage) {
-                dispatchMessage((SupportMessage) networkEnvelope);
+                onSupportMessage((SupportMessage) networkEnvelope);
             } else if (networkEnvelope instanceof AckMessage) {
-                onAckMessage((AckMessage) networkEnvelope, null);
+                onAckMessage((AckMessage) networkEnvelope);
             }
         });
         decryptedDirectMessageWithPubKeys.clear();
@@ -313,10 +313,13 @@ private void applyMessages() {
             NetworkEnvelope networkEnvelope = decryptedMessageWithPubKey.getNetworkEnvelope();
             log.debug("decryptedMessageWithPubKey.message " + networkEnvelope);
             if (networkEnvelope instanceof SupportMessage) {
-                dispatchMessage((SupportMessage) networkEnvelope);
-                p2PService.removeMailboxMsg(decryptedMessageWithPubKey);
+                SupportMessage supportMessage = (SupportMessage) networkEnvelope;
+                onSupportMessage(supportMessage);
+                mailboxMessageService.removeMailboxMsg(supportMessage);
             } else if (networkEnvelope instanceof AckMessage) {
-                onAckMessage((AckMessage) networkEnvelope, decryptedMessageWithPubKey);
+                AckMessage ackMessage = (AckMessage) networkEnvelope;
+                onAckMessage(ackMessage);
+                mailboxMessageService.removeMailboxMsg(ackMessage);
             }
         });
         decryptedMailboxMessageWithPubKeys.clear();