Fix behavior of FileStorage.restore when provided wrong prev_txn

CHANGES.rst

@@ -14,6 +14,9 @@
 - Fix exit code of ``repozo`` script in case of verification error.
   For details see `#396 <https://github.com/zopefoundation/ZODB/pull/396>`_.
 
+- Fix behavior of ``FileStorage.restore`` when provided wrong ``prev_txn``.
+  For details see `#397 <https://github.com/zopefoundation/ZODB/pull/397>`_.
+
 
 5.8.1 (2023-07-18)
 ==================

src/ZODB/FileStorage/FileStorage.py

@@ -672,7 +672,7 @@ def _data_find(self, tpos, oid, data):
         tid, tl, status, ul, dl, el = unpack(TRANS_HDR, h)
         status = as_text(status)
         self._file.read(ul + dl + el)
-        tend = tpos + tl + 8
+        tend = tpos + tl
         pos = self._file.tell()
         while pos < tend:
             h = self._read_data_header(pos)
@@ -1048,6 +1048,8 @@ def undo(self, transaction_id, transaction):
             tid = decodebytes(transaction_id + b'\n')
             assert len(tid) == 8
             tpos = self._txn_find(tid, 1)
+            if not tpos:
+                raise UndoError("Invalid transaction id")
             tindex = self._txn_undo_write(tpos)
             self._tindex.update(tindex)
             return self._tid, tindex.keys()
@@ -1066,7 +1068,7 @@ def _txn_find(self, tid, stop_at_pack):
                 # check the status field of the transaction header
                 if h[16] == b'p':
                     break
-        raise UndoError("Invalid transaction id")
+        return None
 
     def _txn_undo_write(self, tpos):
         # a helper function to write the data records for transactional undo

src/ZODB/tests/testFileStorage.py

@@ -242,6 +242,47 @@ def testRestoreBumpsOid(self):
         # Before ZODB 3.2.6, this failed, with ._oid == z64.
         self.assertEqual(self._storage._oid, giant_oid)
 
+    def testRestorePrevTxnNotExists(self):
+        t = TransactionMetaData()
+        oid = b'\1' * 8
+        self._storage.tpc_begin(t)
+        self._storage.store(oid, b'\0' * 8, b'data1a', b'', t)
+        self._storage.tpc_vote(t)
+        self._storage.tpc_finish(t)
+
+        # restore with a transaction not existing in the storage
+        tid_not_exist = b'\0\0\0\0\0\1\2\3'
+        t = TransactionMetaData()
+        self._storage.tpc_begin(t)
+        self._storage.restore(oid, b'\0' * 8, b'data1b', b'', tid_not_exist, t)
+        self._storage.tpc_vote(t)
+        self._storage.tpc_finish(t)
+        self.assertEqual(self._storage.load(oid)[0], b'data1b')
+
+    def testRestorePrevTxnWithoutOid(self):
+        t = TransactionMetaData()
+        self._storage.tpc_begin(t)
+        self._storage.store(b'\1' * 8, b'\0' * 8, b'data1', b'', t)
+        self._storage.tpc_vote(t)
+        tid1 = self._storage.tpc_finish(t)
+
+        # this transaction is also here to detect problems if restore reads
+        # below the end of tid1
+        t = TransactionMetaData()
+        self._storage.tpc_begin(t)
+        self._storage.store(b'\2' * 8, b'\0' * 8, b'data2a', b'', t)
+        self._storage.tpc_vote(t)
+        self._storage.tpc_finish(t)
+
+        # restore with a transaction not containing the oid
+        t = TransactionMetaData()
+        self._storage.tpc_begin(t)
+        self._storage.restore(b'\2' * 8, b'\0' * 8, b'data2b', b'', tid1, t)
+        self._storage.tpc_vote(t)
+        self._storage.tpc_finish(t)
+        self.assertEqual(self._storage.load(b'\1' * 8)[0], b'data1')
+        self.assertEqual(self._storage.load(b'\2' * 8)[0], b'data2b')
+
     def testCorruptionInPack(self):
         # This sets up a corrupt .fs file, with a redundant transaction
         # length mismatch.  The implementation of pack in many releases of