do not resend payment sent message to arbitrator after ack

core/src/main/java/haveno/core/trade/protocol/ProcessModel.java

@@ -144,6 +144,8 @@ public class ProcessModel implements Model, PersistablePayload {
     // To enable that even after restart we persist the state.
     @Setter
     private ObjectProperty<MessageState> paymentSentMessageStateProperty = new SimpleObjectProperty<>(MessageState.UNDEFINED);
+    @Setter
+    private ObjectProperty<MessageState> paymentSentMessageStatePropertyArbitrator = new SimpleObjectProperty<>(MessageState.UNDEFINED);
 
     public ProcessModel(String offerId, String accountId, PubKeyRing pubKeyRing) {
         this(offerId, accountId, pubKeyRing, new TradePeer(), new TradePeer(), new TradePeer());
@@ -181,6 +183,7 @@ public class ProcessModel implements Model, PersistablePayload {
                 .setUseSavingsWallet(useSavingsWallet)
                 .setFundsNeededForTrade(fundsNeededForTrade)
                 .setPaymentSentMessageState(paymentSentMessageStateProperty.get().name())
+                .setPaymentSentMessageStateArbitrator(paymentSentMessageStatePropertyArbitrator.get().name())
                 .setBuyerPayoutAmountFromMediation(buyerPayoutAmountFromMediation)
                 .setSellerPayoutAmountFromMediation(sellerPayoutAmountFromMediation)
                 .setDeleteBackupsHeight(deleteBackupsHeight);
@@ -218,6 +221,10 @@ public class ProcessModel implements Model, PersistablePayload {
         MessageState paymentSentMessageState = ProtoUtil.enumFromProto(MessageState.class, paymentSentMessageStateString);
         processModel.setPaymentSentMessageState(paymentSentMessageState);
 
+        String paymentSentMessageStateArbitratorString = ProtoUtil.stringOrNullFromProto(proto.getPaymentSentMessageStateArbitrator());
+        MessageState paymentSentMessageStateArbitrator = ProtoUtil.enumFromProto(MessageState.class, paymentSentMessageStateArbitratorString);
+        processModel.setPaymentSentMessageStateArbitrator(paymentSentMessageStateArbitrator);
+
         return processModel;
     }
 
@@ -251,6 +258,13 @@ public class ProcessModel implements Model, PersistablePayload {
         setPaymentSentMessageState(messageState);
     }
 
+    void setPaymentSentAckMessageArbitrator(AckMessage ackMessage) {
+        MessageState messageState = ackMessage.isSuccess() ?
+                MessageState.ACKNOWLEDGED :
+                MessageState.FAILED;
+        setPaymentSentMessageStateArbitrator(messageState);
+    }
+
     public void setPaymentSentMessageState(MessageState paymentSentMessageStateProperty) {
         this.paymentSentMessageStateProperty.set(paymentSentMessageStateProperty);
         if (tradeManager != null) {
@@ -258,6 +272,13 @@ public class ProcessModel implements Model, PersistablePayload {
         }
     }
 
+    public void setPaymentSentMessageStateArbitrator(MessageState paymentSentMessageStateProperty) {
+        this.paymentSentMessageStatePropertyArbitrator.set(paymentSentMessageStateProperty);
+        if (tradeManager != null) {
+            tradeManager.requestPersistence();
+        }
+    }
+
     void setDepositTxSentAckMessage(AckMessage ackMessage) {
         MessageState messageState = ackMessage.isSuccess() ?
                 MessageState.ACKNOWLEDGED :

core/src/main/java/haveno/core/trade/protocol/TradeProtocol.java

@@ -663,6 +663,8 @@ public abstract class TradeProtocol implements DecryptedDirectMessageListener, D
         if (ackMessage.getSourceMsgClassName().equals(PaymentSentMessage.class.getSimpleName())) {
             if (trade.getTradePeer(sender) == trade.getSeller()) {
                 processModel.setPaymentSentAckMessage(ackMessage);
+            } else if (trade.getTradePeer(sender) == trade.getArbitrator()) {
+                processModel.setPaymentSentAckMessageArbitrator(ackMessage);
             } else if (!ackMessage.isSuccess()) {
                 String err = "Received AckMessage with error state for " + ackMessage.getSourceMsgClassName() + " from "+ sender + " with tradeId " + trade.getId() + " and errorMessage=" + ackMessage.getErrorMessage();
                 log.warn(err);

core/src/main/java/haveno/core/trade/protocol/tasks/BuyerSendPaymentSentMessage.java

@@ -73,6 +73,13 @@ public abstract class BuyerSendPaymentSentMessage extends SendMailboxMessageTask
     protected void run() {
         try {
             runInterceptHook();
+
+            // skip if already acked by receiver
+            if (isAckedByReceiver()) {
+                if (!isCompleted()) complete();
+                return;
+            }
+
             super.run();
         } catch (Throwable t) {
             failed(t);
@@ -153,7 +160,7 @@ public abstract class BuyerSendPaymentSentMessage extends SendMailboxMessageTask
     private void tryToSendAgainLater() {
 
         // skip if already acked
-        if (trade.getState().ordinal() >= Trade.State.SELLER_RECEIVED_PAYMENT_SENT_MSG.ordinal()) return;
+        if (isAckedByReceiver()) return;
 
         if (resendCounter >= MAX_RESEND_ATTEMPTS) {
             cleanup();
@@ -185,4 +192,6 @@ public abstract class BuyerSendPaymentSentMessage extends SendMailboxMessageTask
             cleanup();
         }
     }
+
+    protected abstract boolean isAckedByReceiver();
 }

core/src/main/java/haveno/core/trade/protocol/tasks/BuyerSendPaymentSentMessageToArbitrator.java

@@ -18,6 +18,7 @@
 package haveno.core.trade.protocol.tasks;
 
 import haveno.common.taskrunner.TaskRunner;
+import haveno.core.network.MessageState;
 import haveno.core.trade.Trade;
 import haveno.core.trade.protocol.TradePeer;
 import lombok.EqualsAndHashCode;
@@ -55,4 +56,9 @@ public class BuyerSendPaymentSentMessageToArbitrator extends BuyerSendPaymentSen
     protected void setStateArrived() {
         // state only updated on seller message
     }
+
+    @Override
+    protected boolean isAckedByReceiver() {
+        return trade.getProcessModel().getPaymentSentMessageStatePropertyArbitrator().get() == MessageState.ACKNOWLEDGED;
+    }
 }

core/src/main/java/haveno/core/trade/protocol/tasks/BuyerSendPaymentSentMessageToSeller.java

@@ -44,4 +44,9 @@ public class BuyerSendPaymentSentMessageToSeller extends BuyerSendPaymentSentMes
         appendToErrorMessage("Sending message failed: message=" + message + "\nerrorMessage=" + errorMessage);
         complete();
     }
+
+    @Override
+    protected boolean isAckedByReceiver() {
+        return trade.getState().ordinal() >= Trade.State.SELLER_RECEIVED_PAYMENT_SENT_MSG.ordinal();
+    }
 }

core/src/main/java/haveno/core/trade/protocol/tasks/SendDepositsConfirmedMessage.java

@@ -53,7 +53,7 @@ public abstract class SendDepositsConfirmedMessage extends SendMailboxMessageTas
             runInterceptHook();
 
             // skip if already acked by receiver
-            if (ackedByReceiver()) {
+            if (isAckedByReceiver()) {
                 if (!isCompleted()) complete();
                 return;
             }
@@ -126,7 +126,7 @@ public abstract class SendDepositsConfirmedMessage extends SendMailboxMessageTas
     private void tryToSendAgainLater() {
 
         // skip if already acked or payout published
-        if (ackedByReceiver() || trade.isPayoutPublished()) return;
+        if (isAckedByReceiver() || trade.isPayoutPublished()) return;
 
         if (resendCounter >= MAX_RESEND_ATTEMPTS) {
             cleanup();
@@ -151,7 +151,7 @@ public abstract class SendDepositsConfirmedMessage extends SendMailboxMessageTas
         resendCounter++;
     }
 
-    private boolean ackedByReceiver() {
+    private boolean isAckedByReceiver() {
         TradePeer peer = trade.getTradePeer(getReceiverNodeAddress());
         return peer.isDepositsConfirmedMessageAcked();
     }

proto/src/main/proto/pb.proto

@@ -1538,16 +1538,17 @@ message ProcessModel {
     bool use_savings_wallet = 5;
     int64 funds_needed_for_trade = 6;
     string payment_sent_message_state = 7;
-    bytes maker_signature = 8;
+    string payment_sent_message_state_arbitrator = 8;
-    TradePeer maker = 9;
+    bytes maker_signature = 9;
-    TradePeer taker = 10;
+    TradePeer maker = 10;
-    TradePeer arbitrator = 11;
+    TradePeer taker = 11;
-    NodeAddress temp_trade_peer_node_address = 12;
+    TradePeer arbitrator = 12;
-    string multisig_address = 13;
+    NodeAddress temp_trade_peer_node_address = 13;
-    bytes mediated_payout_tx_signature = 17; // placeholder if mediation used in future
+    string multisig_address = 14;
-    int64 buyer_payout_amount_from_mediation = 18;
+    bytes mediated_payout_tx_signature = 15; // placeholder if mediation used in future
-    int64 seller_payout_amount_from_mediation = 19;
+    int64 buyer_payout_amount_from_mediation = 16;
-    int64 delete_backups_height = 31;
+    int64 seller_payout_amount_from_mediation = 17;
+    int64 delete_backups_height = 18;
 }
 
 message TradePeer {