Merge pull request #4025 from stejbac/make-persistence-thread-safe

Make serialisation in FileManager::saveToFile thread-safe

common/src/main/java/bisq/common/proto/persistable/NavigationPath.java

@@ -21,7 +21,6 @@
 
 import com.google.protobuf.Message;
 
-import java.util.ArrayList;
 import java.util.List;
 
 import lombok.AllArgsConstructor;
@@ -36,7 +35,7 @@
 @Getter
 @Setter
 public class NavigationPath implements PersistableEnvelope {
-    private List<String> path = new ArrayList<>();
+    private List<String> path = List.of();
 
     @Override
     public Message toProtoMessage() {
@@ -45,7 +44,7 @@ public Message toProtoMessage() {
         return protobuf.PersistableEnvelope.newBuilder().setNavigationPath(builder).build();
     }
 
-    public static PersistableEnvelope fromProto(protobuf.NavigationPath proto) {
-        return new NavigationPath(new ArrayList<>(proto.getPathList()));
+    public static NavigationPath fromProto(protobuf.NavigationPath proto) {
+        return new NavigationPath(List.copyOf(proto.getPathList()));
     }
 }

common/src/main/java/bisq/common/proto/persistable/PersistableEnvelope.java

@@ -19,8 +19,14 @@
 
 import bisq.common.Envelope;
 
+import com.google.protobuf.Message;
+
 /**
  * Interface for the outside envelope object persisted to disk.
  */
 public interface PersistableEnvelope extends Envelope {
+
+    default Message toPersistableMessage() {
+        return toProtoMessage();
+    }
 }

common/src/main/java/bisq/common/proto/persistable/ThreadedPersistableEnvelope.java

@@ -0,0 +1,44 @@
+/*
+ * 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.common.proto.persistable;
+
+import com.google.protobuf.Message;
+
+/**
+ * Interface for the outer envelope object persisted to disk, where its serialization
+ * during persistence takes place on a separate thread (for performance).
+ * <p>
+ * To make the serialization thread-safe, all modifications of the object must be
+ * synchronized with it. This may be achieved by wrapping such modifications with the
+ * provided {@link ThreadedPersistableEnvelope#modifySynchronized(Runnable)} method.
+ */
+public interface ThreadedPersistableEnvelope extends PersistableEnvelope {
+
+    @Override
+    default Message toPersistableMessage() {
+        synchronized (this) {
+            return toProtoMessage();
+        }
+    }
+
+    default void modifySynchronized(Runnable modifyTask) {
+        synchronized (this) {
+            modifyTask.run();
+        }
+    }
+}

common/src/main/java/bisq/common/proto/persistable/UserThreadMappedPersistableEnvelope.java

@@ -0,0 +1,45 @@
+/*
+ * 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.common.proto.persistable;
+
+import bisq.common.UserThread;
+
+import com.google.protobuf.Message;
+
+import com.google.common.util.concurrent.Futures;
+
+import java.util.concurrent.FutureTask;
+
+/**
+ * Interface for the outer envelope object persisted to disk, where its serialization
+ * during persistence is forced to take place on the user thread.
+ * <p>
+ * To avoid jitter, this should be only be used for small, safely critical stores. Larger
+ * or frequently written stores should either implement {@link PersistableEnvelope}
+ * directly (where thread-safety isn't needed) or use {@link ThreadedPersistableEnvelope}.
+ */
+public interface UserThreadMappedPersistableEnvelope extends PersistableEnvelope {
+
+    @Override
+    default Message toPersistableMessage() {
+        FutureTask<Message> toProtoOnUserThread = new FutureTask<>(this::toProtoMessage);
+        UserThread.execute(toProtoOnUserThread);
+        //noinspection UnstableApiUsage
+        return Futures.getUnchecked(toProtoOnUserThread);
+    }
+}

common/src/main/java/bisq/common/proto/persistable/UserThreadMappedPersistableList.java

@@ -0,0 +1,31 @@
+/*
+ * 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.common.proto.persistable;
+
+import java.util.List;
+
+public class UserThreadMappedPersistableList<T extends PersistablePayload> extends PersistableList<T>
+        implements UserThreadMappedPersistableEnvelope {
+
+    public UserThreadMappedPersistableList(List<T> list) {
+        super(list);
+    }
+
+    public UserThreadMappedPersistableList() {
+    }
+}

common/src/main/java/bisq/common/storage/FileManager.java

@@ -189,7 +189,7 @@ private synchronized void saveToFile(T persistable, File dir, File storageFile)
             log.debug("Write to disc: {}", storageFile.getName());
             protobuf.PersistableEnvelope protoPersistable;
             try {
-                protoPersistable = (protobuf.PersistableEnvelope) persistable.toProtoMessage();
+                protoPersistable = (protobuf.PersistableEnvelope) persistable.toPersistableMessage();
                 if (protoPersistable.toByteArray().length == 0)
                     log.error("protoPersistable is empty. persistable=" + persistable.getClass().getSimpleName());
             } catch (Throwable e) {

core/src/main/java/bisq/core/account/sign/SignedWitnessStore.java

@@ -21,7 +21,7 @@
 import bisq.network.p2p.storage.P2PDataStorage;
 import bisq.network.p2p.storage.payload.PersistableNetworkPayload;
 
-import bisq.common.proto.persistable.PersistableEnvelope;
+import bisq.common.proto.persistable.ThreadedPersistableEnvelope;
 
 import com.google.protobuf.Message;
 
@@ -40,7 +40,7 @@
  * definition and provide a hashMap for the domain access.
  */
 @Slf4j
-public class SignedWitnessStore implements PersistableEnvelope {
+public class SignedWitnessStore implements ThreadedPersistableEnvelope {
     @Getter
     private Map<P2PDataStorage.ByteArray, PersistableNetworkPayload> map = new ConcurrentHashMap<>();
 
@@ -70,7 +70,7 @@ private protobuf.SignedWitnessStore.Builder getBuilder() {
         return protobuf.SignedWitnessStore.newBuilder().addAllItems(protoList);
     }
 
-    public static PersistableEnvelope fromProto(protobuf.SignedWitnessStore proto) {
+    public static SignedWitnessStore fromProto(protobuf.SignedWitnessStore proto) {
         List<SignedWitness> list = proto.getItemsList().stream()
                 .map(SignedWitness::fromProto).collect(Collectors.toList());
         return new SignedWitnessStore(list);

core/src/main/java/bisq/core/account/witness/AccountAgeWitnessStore.java

@@ -20,7 +20,7 @@
 import bisq.network.p2p.storage.P2PDataStorage;
 import bisq.network.p2p.storage.payload.PersistableNetworkPayload;
 
-import bisq.common.proto.persistable.PersistableEnvelope;
+import bisq.common.proto.persistable.ThreadedPersistableEnvelope;
 
 import com.google.protobuf.Message;
 
@@ -39,7 +39,7 @@
  * definition and provide a hashMap for the domain access.
  */
 @Slf4j
-public class AccountAgeWitnessStore implements PersistableEnvelope {
+public class AccountAgeWitnessStore implements ThreadedPersistableEnvelope {
     @Getter
     private Map<P2PDataStorage.ByteArray, PersistableNetworkPayload> map = new ConcurrentHashMap<>();
 
@@ -69,7 +69,7 @@ private protobuf.AccountAgeWitnessStore.Builder getBuilder() {
         return protobuf.AccountAgeWitnessStore.newBuilder().addAllItems(protoList);
     }
 
-    public static PersistableEnvelope fromProto(protobuf.AccountAgeWitnessStore proto) {
+    public static AccountAgeWitnessStore fromProto(protobuf.AccountAgeWitnessStore proto) {
         List<AccountAgeWitness> list = proto.getItemsList().stream()
                 .map(AccountAgeWitness::fromProto).collect(Collectors.toList());
         return new AccountAgeWitnessStore(list);

core/src/main/java/bisq/core/btc/model/AddressEntryList.java

@@ -17,8 +17,8 @@
 
 package bisq.core.btc.model;
 
-import bisq.common.proto.persistable.PersistableEnvelope;
 import bisq.common.proto.persistable.PersistedDataHost;
+import bisq.common.proto.persistable.UserThreadMappedPersistableEnvelope;
 import bisq.common.storage.Storage;
 
 import com.google.protobuf.Message;
@@ -44,7 +44,7 @@
  */
 @ToString
 @Slf4j
-public final class AddressEntryList implements PersistableEnvelope, PersistedDataHost {
+public final class AddressEntryList implements UserThreadMappedPersistableEnvelope, PersistedDataHost {
     transient private Storage<AddressEntryList> storage;
     transient private Wallet wallet;
     @Getter

core/src/main/java/bisq/core/dao/governance/blindvote/MyBlindVoteList.java

@@ -19,8 +19,7 @@
 
 import bisq.core.dao.governance.ConsensusCritical;
 
-import bisq.common.proto.persistable.PersistableEnvelope;
-import bisq.common.proto.persistable.PersistableList;
+import bisq.common.proto.persistable.UserThreadMappedPersistableList;
 
 import com.google.protobuf.Message;
 
@@ -34,7 +33,7 @@
  * List of my own blind votes. Blind votes received from other voters are stored in the BlindVoteStore.
  */
 @EqualsAndHashCode(callSuper = true)
-public class MyBlindVoteList extends PersistableList<BlindVote> implements ConsensusCritical {
+public class MyBlindVoteList extends UserThreadMappedPersistableList<BlindVote> implements ConsensusCritical {
 
     MyBlindVoteList() {
         super();
@@ -59,7 +58,7 @@ public Message toProtoMessage() {
                 .build();
     }
 
-    public static PersistableEnvelope fromProto(protobuf.MyBlindVoteList proto) {
+    public static MyBlindVoteList fromProto(protobuf.MyBlindVoteList proto) {
         return new MyBlindVoteList(new ArrayList<>(proto.getBlindVoteList().stream()
                 .map(BlindVote::fromProto)
                 .collect(Collectors.toList())));

core/src/main/java/bisq/core/dao/governance/blindvote/storage/BlindVoteStore.java

@@ -20,7 +20,7 @@
 import bisq.network.p2p.storage.P2PDataStorage;
 import bisq.network.p2p.storage.payload.PersistableNetworkPayload;
 
-import bisq.common.proto.persistable.PersistableEnvelope;
+import bisq.common.proto.persistable.ThreadedPersistableEnvelope;
 
 import com.google.protobuf.Message;
 
@@ -39,7 +39,7 @@
  * definition and provide a hashMap for the domain access.
  */
 @Slf4j
-public class BlindVoteStore implements PersistableEnvelope {
+public class BlindVoteStore implements ThreadedPersistableEnvelope {
     @Getter
     private Map<P2PDataStorage.ByteArray, PersistableNetworkPayload> map = new ConcurrentHashMap<>();
 
@@ -69,7 +69,7 @@ private protobuf.BlindVoteStore.Builder getBuilder() {
         return protobuf.BlindVoteStore.newBuilder().addAllItems(protoList);
     }
 
-    public static PersistableEnvelope fromProto(protobuf.BlindVoteStore proto) {
+    public static BlindVoteStore fromProto(protobuf.BlindVoteStore proto) {
         List<BlindVotePayload> list = proto.getItemsList().stream()
                 .map(BlindVotePayload::fromProto).collect(Collectors.toList());
         return new BlindVoteStore(list);

core/src/main/java/bisq/core/dao/governance/bond/reputation/MyReputationList.java

@@ -17,7 +17,7 @@
 
 package bisq.core.dao.governance.bond.reputation;
 
-import bisq.common.proto.persistable.PersistableList;
+import bisq.common.proto.persistable.UserThreadMappedPersistableList;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -29,7 +29,7 @@
  * PersistableEnvelope wrapper for list of MyReputations.
  */
 @EqualsAndHashCode(callSuper = true)
-public class MyReputationList extends PersistableList<MyReputation> {
+public class MyReputationList extends UserThreadMappedPersistableList<MyReputation> {
 
     private MyReputationList(List<MyReputation> list) {
         super(list);
@@ -69,4 +69,3 @@ public String toString() {
                 .collect(Collectors.toList());
     }
 }
-

core/src/main/java/bisq/core/dao/governance/myvote/MyVoteList.java

@@ -17,8 +17,7 @@
 
 package bisq.core.dao.governance.myvote;
 
-import bisq.common.proto.persistable.PersistableEnvelope;
-import bisq.common.proto.persistable.PersistableList;
+import bisq.common.proto.persistable.UserThreadMappedPersistableList;
 
 import com.google.protobuf.Message;
 
@@ -29,9 +28,9 @@
 import lombok.EqualsAndHashCode;
 
 @EqualsAndHashCode(callSuper = true)
-public class MyVoteList extends PersistableList<MyVote> {
+public class MyVoteList extends UserThreadMappedPersistableList<MyVote> {
 
-    public MyVoteList() {
+    MyVoteList() {
         super();
     }
 
@@ -53,7 +52,7 @@ public Message toProtoMessage() {
                 .build();
     }
 
-    public static PersistableEnvelope fromProto(protobuf.MyVoteList proto) {
+    public static MyVoteList fromProto(protobuf.MyVoteList proto) {
         return new MyVoteList(new ArrayList<>(proto.getMyVoteList().stream()
                 .map(MyVote::fromProto)
                 .collect(Collectors.toList())));
@@ -66,4 +65,3 @@ public String toString() {
                 .collect(Collectors.toList());
     }
 }
-

core/src/main/java/bisq/core/dao/governance/proofofburn/MyProofOfBurnList.java

@@ -17,7 +17,7 @@
 
 package bisq.core.dao.governance.proofofburn;
 
-import bisq.common.proto.persistable.PersistableList;
+import bisq.common.proto.persistable.UserThreadMappedPersistableList;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -29,7 +29,7 @@
  * PersistableEnvelope wrapper for list of MyProofOfBurn objects.
  */
 @EqualsAndHashCode(callSuper = true)
-public class MyProofOfBurnList extends PersistableList<MyProofOfBurn> {
+public class MyProofOfBurnList extends UserThreadMappedPersistableList<MyProofOfBurn> {
 
     private MyProofOfBurnList(List<MyProofOfBurn> list) {
         super(list);
@@ -69,4 +69,3 @@ public String toString() {
                 .collect(Collectors.toList());
     }
 }
-