fixes to resend payment received messages up to 2 months

woodser · woodser · commit eb733ebf5675 · 2025-07-25T08:37:26.000-04:00
diff --git a/core/src/main/java/haveno/core/offer/OpenOfferManager.java b/core/src/main/java/haveno/core/offer/OpenOfferManager.java
@@ -2206,6 +2206,7 @@ private void startPeriodicRefreshOffersTimer() {
         if (periodicRefreshOffersTimer == null)
             periodicRefreshOffersTimer = UserThread.runPeriodically(() -> {
                         if (!stopped) {
+                            log.info("Refreshing my open offers");
                             synchronized (openOffers.getList()) {
                                 int size = openOffers.size();
                                 //we clone our list as openOffers might change during our delayed call
diff --git a/core/src/main/java/haveno/core/trade/SellerTrade.java b/core/src/main/java/haveno/core/trade/SellerTrade.java
@@ -62,7 +62,7 @@ public boolean confirmPermitted() {
     }
 
     public boolean isFinished() {
-        return super.isFinished() && ((SellerProtocol) getProtocol()).needsToResendPaymentReceivedMessages();
+        return super.isFinished() && !((SellerProtocol) getProtocol()).needsToResendPaymentReceivedMessages();
     }
 }
 
diff --git a/core/src/main/java/haveno/core/trade/Trade.java b/core/src/main/java/haveno/core/trade/Trade.java
@@ -720,7 +720,7 @@ public void initialize(ProcessModelServiceProvider serviceProvider) {
                 if (newValue == Trade.PayoutState.PAYOUT_UNLOCKED) {
                     if (!isInitialized) return;
                     log.info("Payout unlocked for {} {}, deleting multisig wallet", getClass().getSimpleName(), getId());
-                    if (isCompleted()) clearAndShutDown();
+                    if (isInitialized && isFinished()) clearAndShutDown();
                     else deleteWallet();
                 }
             });
@@ -838,7 +838,7 @@ public NodeAddress getArbitratorNodeAddress() {
 
     public void setCompleted(boolean completed) {
         this.isCompleted = completed;
-        if (isPayoutUnlocked()) clearAndShutDown();
+        if (isInitialized && isFinished()) clearAndShutDown();
     }
 
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -1558,6 +1558,11 @@ public boolean mediationResultAppliedPenaltyToSeller() {
     }
 
     public void clearAndShutDown() {
+
+        // unregister p2p message listener immediately
+        getProcessModel().getP2PService().removeDecryptedDirectMessageListener(getProtocol());
+        
+        // clear process data and shut down trade
         ThreadUtils.execute(() -> {
             clearProcessData();
             onShutDownStarted();
@@ -1620,6 +1625,9 @@ public void shutDown() {
         isShutDownStarted = true;
         if (!isPayoutUnlocked()) log.info("Shutting down {} {}", getClass().getSimpleName(), getId());
 
+        // unregister p2p message listener
+        getProcessModel().getP2PService().removeDecryptedDirectMessageListener(getProtocol());
+
         // create task to shut down trade
         Runnable shutDownTask = () -> {
 
diff --git a/core/src/main/java/haveno/core/trade/TradeManager.java b/core/src/main/java/haveno/core/trade/TradeManager.java
@@ -988,29 +988,26 @@ public void onTradeCompleted(Trade trade) {
         if (trade.isCompleted()) throw new RuntimeException("Trade " + trade.getId() + " was already completed");
         closedTradableManager.add(trade);
         trade.setCompleted(true);
-        removeTrade(trade, true);
+        removeTrade(trade);
         xmrWalletService.swapPayoutAddressEntryToAvailable(trade.getId()); // TODO The address entry should have been removed already. Check and if its the case remove that.
         requestPersistence();
     }
 
     public void unregisterTrade(Trade trade) {
         log.warn("Unregistering {} {}", trade.getClass().getSimpleName(), trade.getId());
-        removeTrade(trade, true);
+        removeTrade(trade);
         removeFailedTrade(trade);
         if (!trade.isMaker()) xmrWalletService.swapPayoutAddressEntryToAvailable(trade.getId()); // TODO The address entry should have been removed already. Check and if its the case remove that.
         requestPersistence();
     }
 
-    public void removeTrade(Trade trade, boolean removeDirectMessageListener) {
+    public void removeTrade(Trade trade) {
         log.info("TradeManager.removeTrade() " + trade.getId());
         
         // remove trade
         synchronized (tradableList.getList()) {
             if (!tradableList.remove(trade)) return;
         }
-
-        // unregister message listener and persist
-        if (removeDirectMessageListener) p2PService.removeDecryptedDirectMessageListener(getTradeProtocol(trade));
         requestPersistence();
     }
 
@@ -1077,7 +1074,7 @@ private void updateTradePeriodState() {
     // we move the trade to FailedTradesManager
     public void onMoveInvalidTradeToFailedTrades(Trade trade) {
         failedTradesManager.add(trade);
-        removeTrade(trade, false);
+        removeTrade(trade);
     }
 
     public void onMoveFailedTradeToPendingTrades(Trade trade) {
diff --git a/core/src/main/java/haveno/core/trade/protocol/SellerProtocol.java b/core/src/main/java/haveno/core/trade/protocol/SellerProtocol.java
@@ -34,6 +34,8 @@
 
 package haveno.core.trade.protocol;
 
+import java.util.Date;
+
 import haveno.common.ThreadUtils;
 import haveno.common.handlers.ErrorMessageHandler;
 import haveno.common.handlers.ResultHandler;
@@ -53,6 +55,8 @@
 
 @Slf4j
 public class SellerProtocol extends DisputeProtocol {
+
+    private static final long resendPaymentReceivedMessageDuration = 2L * 30 * 24 * 60 * 60 * 1000; // 2 months
     
     enum SellerEvent implements FluentProtocol.Event {
         STARTUP,
@@ -68,7 +72,7 @@ public SellerProtocol(SellerTrade trade) {
     protected void onInitialized() {
         super.onInitialized();
 
-        // re-send payment received message if payout not published
+        // re-send payment received message if payment received not acked
         ThreadUtils.execute(() -> {
             if (!needsToResendPaymentReceivedMessages()) return;
             synchronized (trade.getLock()) {
@@ -94,13 +98,18 @@ protected void onInitialized() {
     }
 
     public boolean needsToResendPaymentReceivedMessages() {
-        return !trade.isShutDownStarted() && trade.getState().ordinal() >= Trade.State.SELLER_SENT_PAYMENT_RECEIVED_MSG.ordinal() && !trade.getProcessModel().isPaymentReceivedMessagesReceived() && resendPaymentReceivedMessagesEnabled();
+        return !trade.isShutDownStarted() && trade.getState().ordinal() >= Trade.State.SELLER_SENT_PAYMENT_RECEIVED_MSG.ordinal() && !trade.getProcessModel().isPaymentReceivedMessagesReceived() && resendPaymentReceivedMessagesEnabled() && resendPaymentReceivedMessagesWithinDuration();
     }
 
     private boolean resendPaymentReceivedMessagesEnabled() {
         return trade.getOffer().getOfferPayload().getProtocolVersion() >= 2;
     }
 
+    private boolean resendPaymentReceivedMessagesWithinDuration() {
+        Date startDate = trade.getMaxTradePeriodDate(); // TODO: preferably use the date when the payment receipt was confirmed
+        return new Date().getTime() <= (startDate.getTime() + resendPaymentReceivedMessageDuration);
+    }
+
     @Override
     protected void onTradeMessage(TradeMessage message, NodeAddress peer) {
         super.onTradeMessage(message, peer);
diff --git a/core/src/main/java/haveno/core/trade/protocol/TradeProtocol.java b/core/src/main/java/haveno/core/trade/protocol/TradeProtocol.java
@@ -860,6 +860,12 @@ else if (peer == trade.getArbitrator()) {
                 log.warn("Received AckMessage from unexpected peer for {}, sender={}, trade={} {}, messageUid={}, success={}, errorMsg={}", ackMessage.getSourceMsgClassName(), sender, trade.getClass().getSimpleName(), trade.getId(), ackMessage.getSourceUid(), ackMessage.isSuccess(), ackMessage.getErrorMessage());
                 return;
             }
+
+            // clear and shut down trade if completely finished after ack
+            if (trade.isFinished()) {
+                log.info("Trade {} {} is finished after PaymentReceivedMessage ACK, shutting it down", trade.getClass().getSimpleName(), trade.getId());
+                trade.clearAndShutDown();
+            }
         }
 
         // generic handling
diff --git a/core/src/main/java/haveno/core/trade/protocol/tasks/SellerSendPaymentReceivedMessage.java b/core/src/main/java/haveno/core/trade/protocol/tasks/SellerSendPaymentReceivedMessage.java
@@ -48,6 +48,7 @@
 import haveno.core.trade.Trade;
 import haveno.core.trade.messages.PaymentReceivedMessage;
 import haveno.core.trade.messages.TradeMailboxMessage;
+import haveno.core.trade.protocol.SellerProtocol;
 import haveno.core.trade.protocol.TradePeer;
 import haveno.core.util.JsonUtil;
 import haveno.network.p2p.NodeAddress;
@@ -236,6 +237,8 @@ protected boolean isMessageReceived() {
     }
 
     protected boolean stopSending() {
-        return isMessageReceived() || !trade.isPaymentReceived(); // stop if received or trade state reset // TODO: also stop after some number of blocks?
+        if (isMessageReceived()) return true;
+        if (!((SellerProtocol) trade.getProtocol()).needsToResendPaymentReceivedMessages()) return true;
+        return false;
     }
 }
diff --git a/p2p/src/main/java/haveno/network/p2p/P2PService.java b/p2p/src/main/java/haveno/network/p2p/P2PService.java
@@ -249,7 +249,9 @@ public void onTorNodeReady() {
 
         requestDataManager.requestPreliminaryData();
         keepAliveManager.start();
-        p2pServiceListeners.forEach(SetupListener::onTorNodeReady);
+        synchronized (p2pServiceListeners) {
+            p2pServiceListeners.forEach(SetupListener::onTorNodeReady);
+        }
     }
 
     @Override
@@ -258,17 +260,23 @@ public void onHiddenServicePublished() {
 
         hiddenServicePublished.set(true);
 
-        p2pServiceListeners.forEach(SetupListener::onHiddenServicePublished);
+        synchronized (p2pServiceListeners) {
+            p2pServiceListeners.forEach(SetupListener::onHiddenServicePublished);
+        }
     }
 
     @Override
     public void onSetupFailed(Throwable throwable) {
-        p2pServiceListeners.forEach(e -> e.onSetupFailed(throwable));
+        synchronized (p2pServiceListeners) {
+            p2pServiceListeners.forEach(e -> e.onSetupFailed(throwable));
+        }
     }
 
     @Override
     public void onRequestCustomBridges() {
-        p2pServiceListeners.forEach(SetupListener::onRequestCustomBridges);
+        synchronized (p2pServiceListeners) {
+            p2pServiceListeners.forEach(SetupListener::onRequestCustomBridges);
+        }
     }
 
     // Called from networkReadyBinding
@@ -304,7 +312,9 @@ public void onPreliminaryDataReceived() {
 
     @Override
     public void onUpdatedDataReceived() {
-        p2pServiceListeners.forEach(P2PServiceListener::onUpdatedDataReceived);
+        synchronized (p2pServiceListeners) {
+            p2pServiceListeners.forEach(P2PServiceListener::onUpdatedDataReceived);
+        }
     }
 
     @Override
@@ -314,7 +324,9 @@ public void onNoSeedNodeAvailable() {
 
     @Override
     public void onNoPeersAvailable() {
-        p2pServiceListeners.forEach(P2PServiceListener::onNoPeersAvailable);
+        synchronized (p2pServiceListeners) {
+            p2pServiceListeners.forEach(P2PServiceListener::onNoPeersAvailable);
+        }
     }
 
     @Override
@@ -334,7 +346,9 @@ private void applyIsBootstrapped(Consumer<P2PServiceListener> listenerHandler) {
             mailboxMessageService.onBootstrapped();
 
             // Once we have applied the state in the P2P domain we notify our listeners
-            p2pServiceListeners.forEach(listenerHandler);
+            synchronized (p2pServiceListeners) {
+                p2pServiceListeners.forEach(listenerHandler);
+            }
 
             mailboxMessageService.initAfterBootstrapped();
         }
@@ -369,12 +383,14 @@ public void onMessage(NetworkEnvelope networkEnvelope, Connection connection) {
             try {
                 DecryptedMessageWithPubKey decryptedMsg = encryptionService.decryptAndVerify(sealedMsg.getSealedAndSigned());
                 connection.maybeHandleSupportedCapabilitiesMessage(decryptedMsg.getNetworkEnvelope());
-                connection.getPeersNodeAddressOptional().ifPresentOrElse(nodeAddress ->
-                        decryptedDirectMessageListeners.forEach(e -> e.onDirectMessage(decryptedMsg, nodeAddress)),
-                        () -> {
-                            log.error("peersNodeAddress is expected to be available at onMessage for " +
-                                    "processing PrefixedSealedAndSignedMessage.");
-                        });
+                connection.getPeersNodeAddressOptional().ifPresentOrElse(nodeAddress -> {
+                        synchronized (decryptedDirectMessageListeners) {
+                            decryptedDirectMessageListeners.forEach(e -> e.onDirectMessage(decryptedMsg, nodeAddress));
+                        }
+                    }, () -> {
+                        log.error("peersNodeAddress is expected to be available at onMessage for " +
+                                "processing PrefixedSealedAndSignedMessage.");
+                    });
             } catch (CryptoException e) {
                 log.warn("Decryption of a direct message failed. This is not expected as the " +
                         "direct message was sent to our node.");
@@ -503,19 +519,27 @@ public boolean removeData(ProtectedStoragePayload protectedStoragePayload) {
     ///////////////////////////////////////////////////////////////////////////////////////////
 
     public void addDecryptedDirectMessageListener(DecryptedDirectMessageListener listener) {
-        decryptedDirectMessageListeners.add(listener);
+        synchronized (decryptedDirectMessageListeners) {
+            decryptedDirectMessageListeners.add(listener);
+        }
     }
 
     public void removeDecryptedDirectMessageListener(DecryptedDirectMessageListener listener) {
-        decryptedDirectMessageListeners.remove(listener);
+        synchronized (decryptedDirectMessageListeners) {
+            decryptedDirectMessageListeners.remove(listener);
+        }
     }
 
     public void addP2PServiceListener(P2PServiceListener listener) {
-        p2pServiceListeners.add(listener);
+        synchronized (p2pServiceListeners) {
+            p2pServiceListeners.add(listener);
+        }
     }
 
     public void removeP2PServiceListener(P2PServiceListener listener) {
-        p2pServiceListeners.remove(listener);
+        synchronized (p2pServiceListeners) {
+            p2pServiceListeners.remove(listener);
+        }
     }
 
     public void addHashSetChangedListener(HashMapChangedListener hashMapChangedListener) {

Original file line number	Diff line number	Diff line change
`@@ -62,7 +62,7 @@ public boolean confirmPermitted() {`
`62`	`62`	`}`
`63`	`63`
`64`	`64`	`public boolean isFinished() {`
`65`		`- return super.isFinished() && ((SellerProtocol) getProtocol()).needsToResendPaymentReceivedMessages();`
	`65`	`+ return super.isFinished() && !((SellerProtocol) getProtocol()).needsToResendPaymentReceivedMessages();`
`66`	`66`	`}`
`67`	`67`	`}`
`68`	`68`
Original file line number	Diff line number	Diff line change
`@@ -860,6 +860,12 @@ else if (peer == trade.getArbitrator()) {`
`860`	`860`	`log.warn("Received AckMessage from unexpected peer for {}, sender={}, trade={} {}, messageUid={}, success={}, errorMsg={}", ackMessage.getSourceMsgClassName(), sender, trade.getClass().getSimpleName(), trade.getId(), ackMessage.getSourceUid(), ackMessage.isSuccess(), ackMessage.getErrorMessage());`
`861`	`861`	`return;`
`862`	`862`	`}`
	`863`	`+`
	`864`	`+ // clear and shut down trade if completely finished after ack`
	`865`	`+ if (trade.isFinished()) {`
	`866`	`+ log.info("Trade {} {} is finished after PaymentReceivedMessage ACK, shutting it down", trade.getClass().getSimpleName(), trade.getId());`
	`867`	`+ trade.clearAndShutDown();`
	`868`	`+ }`
`863`	`869`	`}`
`864`	`870`
`865`	`871`	`// generic handling`
Original file line number	Diff line number	Diff line change
`@@ -48,6 +48,7 @@`
`48`	`48`	`import haveno.core.trade.Trade;`
`49`	`49`	`import haveno.core.trade.messages.PaymentReceivedMessage;`
`50`	`50`	`import haveno.core.trade.messages.TradeMailboxMessage;`
	`51`	`+import haveno.core.trade.protocol.SellerProtocol;`
`51`	`52`	`import haveno.core.trade.protocol.TradePeer;`
`52`	`53`	`import haveno.core.util.JsonUtil;`
`53`	`54`	`import haveno.network.p2p.NodeAddress;`
`@@ -236,6 +237,8 @@ protected boolean isMessageReceived() {`
`236`	`237`	`}`
`237`	`238`
`238`	`239`	`protected boolean stopSending() {`
`239`		`- return isMessageReceived() \|\| !trade.isPaymentReceived(); // stop if received or trade state reset // TODO: also stop after some number of blocks?`
	`240`	`+ if (isMessageReceived()) return true;`
	`241`	`+ if (!((SellerProtocol) trade.getProtocol()).needsToResendPaymentReceivedMessages()) return true;`
	`242`	`+ return false;`
`240`	`243`	`}`
`241`	`244`	`}`