Fixed a number of bugs in the handling of persistent ZEO caches:

- Cache records are written in several stems.  If a process exits
    after writing begins and before it is finishes, the cache will be
    corrupt on restart.  The way records are writted was changed to
    make cache record updates atomic.

- There was no lock file to prevent opening a cache multiple times
    at once, which would lead to corruption.  Persistent caches now
    use lock files, in the same way that file storages do.

- A bug in the cache-opening logic led to cache failure in the
    unlikely event that a cache has no free blocks.

NEWS.txt

@@ -3,6 +3,20 @@ Whats new in ZODB 3.8.1
 
 Bugs Fixed:
 
+- Fixed a number of bugs in the handling of persistent ZEO caches:
+
+  - Cache records are written in several stems.  If a process exits
+    after writing begins and before it is finishes, the cache will be
+    corrupt on restart.  The way records are writted was changed to
+    make cache record updates atomic.
+
+  - There was no lock file to prevent opening a cache multiple times
+    at once, which would lead to corruption.  Persistent caches now
+    use lock files, in the same way that file storages do.
+
+  - A bug in the cache-opening logic led to cache failure in the
+    unlikely event that a cache has no free blocks.
+
 - When using ZEO Client Storages, Errors occured when trying to store
   objects too big to fit in the ZEO cache file.
 

src/ZEO/cache.py

@@ -29,6 +29,7 @@ import struct
 import tempfile
 import time
 
+import ZODB.lock_file
 from ZODB.utils import z64, u64
 
 logger = logging.getLogger("ZEO.cache")
@@ -770,6 +771,10 @@ class FileCache(object):
         # (and it sets self.f).
 
         self.fpath = fpath
+
+        if fpath:
+            self._lock_file = ZODB.lock_file.LockFile(fpath + '.lock')
+        
         if fpath and os.path.exists(fpath):
             # Reuse an existing file.  scan() will open & read it.
             self.f = None
@@ -826,8 +831,8 @@ class FileCache(object):
         # file, and tell our parent about it too (via the `install` callback).
         # Remember the location of the largest free block.  That seems a
         # decent place to start currentofs.
-        max_free_size = max_free_offset = 0
-        ofs = ZEC3_HEADER_SIZE
+        max_free_size = 0
+        ofs = max_free_offset = ZEC3_HEADER_SIZE
         while ofs < fsize:
             self.f.seek(ofs)
             ent = None
@@ -895,6 +900,8 @@ class FileCache(object):
     # Close the underlying file.  No methods accessing the cache should be
     # used after this.
     def close(self):
+        if hasattr(self,'_lock_file'):
+            self._lock_file.close()
         if self.f:
             self.sync()
             self.f.close()
@@ -946,11 +953,23 @@ class FileCache(object):
             extra = 'f' + struct.pack(">I", excess)
 
         self.f.seek(self.currentofs)
-        self.f.writelines(('a',
-                           struct.pack(">I8s8s", size,
-                                       obj.key[0], obj.key[1])))
+
+        # Before writing data, we'll write a free block for the space freed.
+        # We'll come back with a last atomic write to rewrite the start of the
+        # allocated-block header.
+        self.f.write('f'+struct.pack(">I", nfreebytes))
+
+        # Now write the rest of the allocation block header and object data.
+        self.f.write(struct.pack(">8s8s", obj.key[0], obj.key[1]))
         obj.serialize(self.f)
         self.f.write(extra)
+
+        # Now, we'll go back and rewrite the beginning of the
+        # allocated block header.
+        self.f.seek(self.currentofs)
+        self.f.write('a'+struct.pack(">I", size))
+        
+        # Update index
         e = Entry(obj.key, self.currentofs)
         self.key2entry[obj.key] = e
         self.filemap[self.currentofs] = size, e

src/ZEO/tests/test_cache.py

@@ -16,6 +16,8 @@
 import os
 import tempfile
 import unittest
+import zope.testing.setupstack
+from zope.testing import doctest
 
 import ZEO.cache
 from ZODB.utils import p64
@@ -181,7 +183,84 @@ class CacheTests(unittest.TestCase):
         # If an object cannot be stored in the cache, it must not be
         # recorded as non-current.
         self.assert_((n2, n3) not in cache.noncurrent[n1])
-  	 
+
+__test__ = dict(
+    kill_does_not_cause_cache_corruption =
+    r"""
+
+    If we kill a process while a cache is being written to, the cache
+    isn't corrupted.  To see this, we'll write a little script that
+    writes records to a cache file repeatedly.
+
+    >>> import os, random, sys, time
+    >>> open('t', 'w').write('''
+    ... import os, random, sys, thread, time
+    ... sys.path = %r
+    ... 
+    ... def suicide():
+    ...     time.sleep(random.random()/10)
+    ...     os._exit(0)
+    ... 
+    ... import ZEO.cache, ZODB.utils
+    ... cache = ZEO.cache.ClientCache('cache')
+    ... oid = 0
+    ... t = 0
+    ... thread.start_new_thread(suicide, ())
+    ... while 1:
+    ...     oid += 1
+    ...     t += 1
+    ...     data = 'X' * random.randint(5000,25000)
+    ...     cache.store(ZODB.utils.p64(oid), '', ZODB.utils.p64(t), None, data)
+    ... 
+    ... ''' % sys.path)
+
+    >>> for i in range(10):
+    ...     _ = os.spawnl(os.P_WAIT, sys.executable, sys.executable, 't')
+    ...     if os.path.exists('cache'):
+    ...         cache = ZEO.cache.ClientCache('cache').open()
+    ...         os.remove('cache')
+    ...         os.remove('cache.lock')
+       
+    
+    """,
+
+    full_cache_is_valid =
+    r"""
+
+    If we fill up the cache without any free space, the cache can
+    still be used.
+
+    >>> import ZEO.cache, ZODB.utils
+    >>> cache = ZEO.cache.ClientCache('cache', 1000)
+    >>> data = 'X' * (1000 - ZEO.cache.ZEC3_HEADER_SIZE
+    ...               - ZEO.cache.OBJECT_HEADER_SIZE
+    ...               - ZEO.cache.Object.TOTAL_FIXED_SIZE)
+    >>> cache.store(ZODB.utils.p64(1), '', ZODB.utils.p64(1), None, data)
+    >>> cache.close()
+    >>> cache = ZEO.cache.ClientCache('cache', 1000)
+    >>> cache.open()
+    >>> cache.store(ZODB.utils.p64(2), '', ZODB.utils.p64(2), None, 'XXX')
+    
+    """,
+
+    cannot_open_same_cache_file_twice =
+    r"""
+    >>> import ZEO.cache
+    >>> cache = ZEO.cache.ClientCache('cache', 1000)
+    >>> cache = ZEO.cache.ClientCache('cache', 1000)
+    Traceback (most recent call last):
+    ...
+    LockError: Couldn't lock 'cache.lock'
+    
+    """,
+    )
+
 
 def test_suite():
-    return unittest.makeSuite(CacheTests)
+    return unittest.TestSuite((
+        unittest.makeSuite(CacheTests),
+        doctest.DocTestSuite(
+            setUp=zope.testing.setupstack.setUpDirectory,
+            tearDown=zope.testing.setupstack.tearDown,
+            ),
+        ))