Merge pull request #52 from zopefoundation/fs-flush_after_truncate

Fix possible data corruption after FileStorage is truncated to roll back a transaction

CHANGES.rst

@@ -22,6 +22,10 @@
 - Remove useless dependency to `zdaemon` in setup.py. Remove ZEO documentation.
   Both were leftovers from the time where ZEO was part of this repository.
 
+- Fix possible data corruption after FileStorage is truncated to roll back a
+  transaction.
+  https://github.com/zopefoundation/ZODB/pull/52
+
 4.2.0 (2015-06-02)
 ==================
 

src/ZODB/FileStorage/FileStorage.py

@@ -20,6 +20,7 @@
 import errno
 import logging
 import os
+import sys
 import threading
 import time
 from struct import pack
@@ -722,6 +723,7 @@ def tpc_vote(self, transaction):
                 # Hm, an error occurred writing out the data. Maybe the
                 # disk is full. We don't want any turd at the end.
                 self._file.truncate(self._pos)
+                self._files.flush()
                 raise
             self._nextpos = self._pos + (tl + 8)
 
@@ -776,6 +778,7 @@ def _finish_finish(self, tid):
     def _abort(self):
         if self._nextpos:
             self._file.truncate(self._pos)
+            self._files.flush()
             self._nextpos=0
             self._blob_tpc_abort()
 
@@ -2093,6 +2096,22 @@ def empty(self):
         while self._files:
             self._files.pop().close()
 
+
+    def flush(self):
+        """Empty read buffers.
+
+        This is required if they contain data of rolled back transactions.
+        """
+        with self.write_lock():
+            for f in self._files:
+                f.flush()
+
+    # Unfortunately, Python 3.x has no API to flush read buffers.
+    if sys.version_info.major > 2:
+        def flush(self):
+            with self.write_lock():
+                self.empty()
+
     def close(self):
         with self._cond:
             self.closed = True

src/ZODB/tests/testFileStorage.py

@@ -25,6 +25,7 @@
 from ZODB import POSException
 from ZODB import DB
 from ZODB.fsIndex import fsIndex
+from ZODB.utils import U64, p64, z64
 
 from ZODB.tests import StorageTestBase, BasicStorage, TransactionalUndoStorage
 from ZODB.tests import PackableStorage, Synchronization, ConflictResolution
@@ -217,7 +218,6 @@ def checkCorruptionInPack(self):
         # global.
         import time
 
-        from ZODB.utils import U64, p64
         from ZODB.FileStorage.format import CorruptedError
         from ZODB.serialize import referencesf
 
@@ -285,6 +285,27 @@ def check_record_iternext(self):
             else:
                 self.assertNotEqual(next_oid, None)
 
+    def checkFlushAfterTruncate(self, fail=False):
+        r0 = self._dostore(z64)
+        storage = self._storage
+        t = transaction.Transaction()
+        storage.tpc_begin(t)
+        storage.store(z64, r0, b'foo', b'', t)
+        storage.tpc_vote(t)
+        # Read operations are done with separate 'file' objects with their
+        # own buffers: here, the buffer also includes voted data.
+        storage.load(z64)
+        # This must invalidate all read buffers.
+        storage.tpc_abort(t)
+        self._dostore(z64, r0, b'bar', 1)
+        # In the case that read buffers were not invalidated, return value
+        # is based on what was cached during the first load.
+        self.assertEqual(storage.load(z64)[0], b'foo' if fail else b'bar')
+
+    def checkFlushNeededAfterTruncate(self):
+        self._storage._files.flush = lambda: None
+        self.checkFlushAfterTruncate(True)
+
 class FileStorageHexTests(FileStorageTests):
 
     def open(self, **kwargs):