stored_dict: add support for write batches

Add unit test of atomicity.
Use a write batch for DB upgrades.

Author: ecdsa

electrum/daemon.py

@@ -544,9 +544,9 @@ def _load_wallet(
             force_check_password: bool = False,  # if set, always validate password
     ) -> Optional[Abstract_Wallet]:
         path = standardize_path(path)
-        storage = DictStorage(path, allow_partial_writes=config.WALLET_PARTIAL_WRITES)
-        if not storage.file_exists():
+        if not os.path.exists(path):
             raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), path)
+        storage = DictStorage(path, allow_partial_writes=config.WALLET_PARTIAL_WRITES)
         if storage.is_encrypted():
             if not password:
                 raise InvalidPassword('No password given')

electrum/json_db.py

@@ -26,6 +26,7 @@
 import copy
 import json
 from typing import TYPE_CHECKING, Optional, Sequence, List, Union, Dict, Any
+from contextlib import contextmanager
 
 import jsonpatch
 import jsonpointer
@@ -94,6 +95,7 @@ def __init__(
         self._is_closed = True
         self.lock = threading.RLock()
         self.pending_changes = []  # type: List[str]
+        self._write_batch = False
         self._modified = False
         if self.path:
             self.storage = FileStorage(path, allow_partial_writes=allow_partial_writes)
@@ -309,8 +311,22 @@ def dump(self, *, human_readable: bool = True) -> str:
             sort_keys=bool(human_readable),
         )
 
+    @contextmanager
+    def write_batch(self):
+        assert self._write_batch is False
+        self._write_batch = True
+        try:
+            yield
+        finally:
+            # FIXME: if an exception is raised here, changes will not be
+            # written to disk, but they will be added to the in-memory dict.
+            self._write_batch = False
+        if self.storage:
+            self.write()
+
     @locked
     def write(self):
+        assert self._write_batch is False
         if not self.storage:
             return
         if self.storage.should_do_full_write_next():
@@ -319,7 +335,7 @@ def write(self):
             self._append_pending_changes()
 
     def close(self):
-        # do not call write
+        # do not call write, because we may need to close the DB after an exception was raised during a batch write
         self._is_closed = True
 
     def is_closed(self):

electrum/stored_dict.py

@@ -296,6 +296,9 @@ def __init__(self, db: BaseDB, key: _FLEX_KEY, parent):
     def should_convert(self):
         return self._db._should_convert
 
+    def write_batch(self):
+        return self._db.write_batch()
+
     def dump(self) -> dict:
         data = {}
         for k, v in self.items():

electrum/wallet.py

@@ -4538,8 +4538,6 @@ def restore_wallet_from_text(
         if gap_limit_for_change is not None:
             db.put('gap_limit_for_change', gap_limit_for_change)
         wallet = wallet_factory(db, config=config)
-    if storage:
-        assert not storage.file_exists(), "file was created too soon! plaintext keys might have been written to disk"
     wallet.synchronize()
     msg = ("This wallet was restored offline. It may contain more addresses than displayed. "
            "Start a daemon and use load_wallet to sync its history.")

electrum/wallet_db.py

@@ -1570,14 +1570,15 @@ def upgrade_wallet_db(data: 'StoredDict', do_upgrade: bool) -> Tuple[dict, bool]
         )
 
     data._db._should_convert = False
-    dbu = WalletDBUpgrader(data)
-    if dbu.requires_split():
-        raise WalletRequiresSplit(dbu.get_split_accounts())
-    if dbu.requires_upgrade() and do_upgrade:
-        dbu.upgrade()
-        was_upgraded = True
-    if dbu.requires_upgrade():
-        raise WalletRequiresUpgrade()
+    with data.write_batch():
+        dbu = WalletDBUpgrader(data)
+        if dbu.requires_split():
+            raise WalletRequiresSplit(dbu.get_split_accounts())
+        if dbu.requires_upgrade() and do_upgrade:
+            dbu.upgrade()
+            was_upgraded = True
+        if dbu.requires_upgrade():
+            raise WalletRequiresUpgrade()
     data._db._should_convert = True
     return was_upgraded
 

tests/test_stored_dict.py

@@ -58,6 +58,29 @@ def test_db_iterators(self):
         for i, v in enumerate(sl):
             self.assertEqual(i, v)
 
+    def test_write_batch(self):
+        # test that batches are written atomically
+        sd = DictStorage(self.path)
+        with sd.write_batch():
+            sd['a'] = 0
+        self.assertEqual(len(sd), 1)
+        with sd.write_batch():
+            sd['a'] = 1
+        self.assertEqual(len(sd), 1)
+        try:
+            with sd.write_batch():
+                sd['b'] = 1
+                raise Exception('blah')
+        except Exception as e:
+            pass
+        self.assertEqual(sd._db._write_batch, False)
+        # at this point, the StoredDict length is 2
+        self.assertEqual(len(sd), 2)
+        sd.close()
+        # check that changes have not been written to disk
+        sd = DictStorage(self.path)
+        self.assertEqual(len(sd), 1)
+
     async def test_dangling_dict(self):
         storage = DictStorage(self.path)
         storage['a'] = {'b': {'c': 0}}