pack: Clear all non-current entries after pack

Else those non-current entries can be used to serve a loadBefore request
with data, while, after pack that loadBefore request must return "data
deleted" if requested object has current revision >= packtime.

Fixes checkPackVSConnectionGet from ZODB from zopefoundation/ZODB#322
which, without this patch fails as e.g.

    Failure in test checkPackVSConnectionGet (ZEO.tests.testZEO.MappingStorageTests)
    Traceback (most recent call last):
      File "/usr/lib/python2.7/unittest/case.py", line 329, in run
        testMethod()
      File "/home/kirr/src/wendelin/z/ZODB/src/ZODB/tests/PackableStorage.py", line 636, in checkPackVSConnectionGet
        raises(ReadConflictError, conn1.get, oid)
      File "/usr/lib/python2.7/unittest/case.py", line 473, in assertRaises
        callableObj(*args, **kwargs)
      File "/usr/lib/python2.7/unittest/case.py", line 116, in __exit__
        "{0} not raised".format(exc_name))
    AssertionError: ReadConflictError not raised

Author: navytux

src/ZEO/ClientStorage.py

@@ -559,7 +559,16 @@ def pack(self, t=None, referencesf=None, wait=1, days=0):
         if t is None:
             t = time.time()
         t = t - (days * 86400)
-        return self._call('pack', t, wait)
+        ret = self._call('pack', t, wait)
+        # remove all non-current entries from the cache.
+        # This way we make sure that loadBefore with before < packtime, won't
+        # return data from the cache, instead of returning "no data" if requested object
+        # has current revision >= packtime.
+        # By clearing all noncurrent entries we might remove more data from the
+        # cache than is strictly necessary, but since access to noncurrent data
+        # is seldom, that should not cause problems in practice.
+        self._cache.clearAllNonCurrent()
+        return ret
 
     def store(self, oid, serial, data, version, txn):
         """Storage API: store data for an object."""

src/ZEO/cache.py

@@ -246,6 +246,21 @@ def clear(self):
             self.f.truncate()
             self._initfile(ZEC_HEADER_SIZE)
 
+    # clearAllNonCurrent removes all non-current entries from the cache.
+    def clearAllNonCurrent(self):
+        with self._lock:
+            f = self.f
+            for oid, tidofs in self.noncurrent.items():
+                for (tid, ofs) in tidofs.items():
+                    f.seek(ofs)
+                    status = f.read(1)
+                    assert status == b'a', (ofs, f.tell(), oid, tid)
+                    f.seek(ofs)
+                    f.write(b'f')
+                    self._len -= 1
+
+            self.noncurrent.clear()
+
     ##
     # Scan the current contents of the cache file, calling `install`
     # for each object found in the cache.  This method should only

src/ZEO/tests/test_cache.py

@@ -253,6 +253,34 @@ def test_clear_zeo_cache(self):
         self.assertEqual(cache.load(n3), None)
         self.assertEqual(cache.loadBefore(n3, n2), None)
 
+    def testClearAllNonCurrent(self):
+        cache = self.cache
+        cache.store(p64(1), n1, n2,   b'1@1')
+        cache.store(p64(1), n2, n3,   b'1@2')
+        cache.store(p64(1), n3, None, b'1')
+        cache.store(p64(2), n2, n3,   b'2@2')
+        cache.store(p64(2), n3, None, b'2')
+
+        eq = self.assertEqual
+        eq(len(cache), 5)
+        eq(cache.load(p64(1)),              (b'1',   n3))
+        eq(cache.loadBefore(p64(1), n3),    (b'1@2', n2, n3))
+        eq(cache.loadBefore(p64(1), n2),    (b'1@1', n1, n2))
+        eq(cache.loadBefore(p64(1), n1),    None)
+        eq(cache.load(p64(2)),              (b'2',   n3))
+        eq(cache.loadBefore(p64(2), n3),    (b'2@2', n2, n3))
+        eq(cache.loadBefore(p64(2), n2),    None)
+
+        cache.clearAllNonCurrent()
+        eq(len(cache), 2)
+        eq(cache.load(p64(1)),              (b'1',   n3))
+        eq(cache.loadBefore(p64(1), n3),    None)
+        eq(cache.loadBefore(p64(1), n2),    None)
+        eq(cache.loadBefore(p64(1), n1),    None)
+        eq(cache.load(p64(2)),              (b'2',   n3))
+        eq(cache.loadBefore(p64(2), n3),    None)
+        eq(cache.loadBefore(p64(2), n2),    None)
+
     def testChangingCacheSize(self):
         # start with a small cache
         data = b'x'