Only cleanup partial transactions if an existing transaction is found

attic/remote.py

@@ -132,6 +132,8 @@ def fetch_from_cache(args):
                             raise Repository.DoesNotExist(self.location.orig)
                         elif error == b'AlreadyExists':
                             raise Repository.AlreadyExists(self.location.orig)
+                        elif error == b'CheckNeeded':
+                            raise Repository.CheckNeeded(self.location.orig)
                         raise self.RPCError(error)
                     else:
                         yield res

attic/repository.py

@@ -42,6 +42,9 @@ class AlreadyExists(Error):
     class InvalidRepository(Error):
         """{} is not a valid repository"""
 
+    class CheckNeeded(Error):
+        '''Inconsistency detected. Please "run attic check {}"'''
+
 
     def __init__(self, path, create=False):
         self.path = path
@@ -90,7 +93,9 @@ def open(self, path):
 
     def close(self):
         if self.lock:
-            self.rollback()
+            if self.io:
+                self.io.close()
+            self.io = None
             self.lock.release()
             self.lock = None
 
@@ -235,7 +240,7 @@ def report_progress(msg, error=False):
                 elif tag == TAG_COMMIT:
                     continue
                 else:
-                    raise self.RepositoryCheckFailed(self.path, 'Unexpected tag {} in segment {}'.format(tag, segment))
+                    report_progress('Unexpected tag {} in segment {}'.format(tag, segment), error=True)
         if len(self.index) != len(seen):
             report_progress('Index object count mismatch. {} != {}'.format(len(self.index), len(seen)), error=True)
         if not error_found:
@@ -350,15 +355,23 @@ def cleanup(self):
         """
         self.head = None
         self.segment = 0
-        # FIXME: Only delete segments if we're sure there's at least
-        # one complete segment somewhere
+        to_delete = []
         for segment, filename in self._segment_names(reverse=True):
             if self.is_complete_segment(filename):
                 self.head = segment
                 self.segment = self.head + 1
-                return
+                for filename in to_delete:
+                    os.unlink(filename)
+                break
             else:
-                os.unlink(filename)
+                to_delete.append(filename)
+        else:
+            # Abort if no transaction is found, otherwise all segments
+            # would be deleted
+            if to_delete:
+                raise Repository.CheckNeeded(self.path)
+
+
 
     def is_complete_segment(self, filename):
         with open(filename, 'rb') as fd:

attic/testsuite/repository.py

@@ -107,6 +107,11 @@ class RepositoryCheckTestCase(AtticTestCase):
     def open(self, create=False):
         return Repository(os.path.join(self.tmppath, 'repository'), create=create)
 
+    def reopen(self):
+        if self.repository:
+            self.repository.close()
+        self.repository = self.open()
+
     def setUp(self):
         self.tmppath = tempfile.mkdtemp()
         self.repository = self.open(create=True)
@@ -120,9 +125,11 @@ def add_objects(self, ids):
             self.repository.put(('%032d' % id_).encode('ascii'), b'data')
         self.repository.commit()
 
+    def get_head(self):
+        return sorted(int(n) for n in os.listdir(os.path.join(self.tmppath, 'repository', 'data', '0')))[-1]
+
     def open_index(self):
-        head = sorted(int(n[6:]) for n in os.listdir(os.path.join(self.tmppath, 'repository')) if n.startswith('index') and n[6:].isdigit())[0]
-        return NSIndex(os.path.join(self.tmppath, 'repository', 'index.{}'.format(head)))
+        return NSIndex(os.path.join(self.tmppath, 'repository', 'index.{}'.format(self.get_head())))
 
     def corrupt_object(self, id_):
         idx = self.open_index()
@@ -140,9 +147,16 @@ def test_check(self):
         self.assert_equal(set([1, 2, 3, 4, 5, 6]), self.list_objects())
         self.assert_equal(True, self.repository.check())
         self.corrupt_object(5)
+        self.reopen()
         self.assert_equal(False, self.repository.check())
         self.assert_equal(set([1, 2, 3, 4, 5, 6]), self.list_objects())
 
+    def test_check_missing_or_corrupt_commit_tag(self):
+        self.add_objects([1, 2, 3])
+        self.assert_equal(set([1, 2, 3]), self.list_objects())
+        with open(os.path.join(self.tmppath, 'repository', 'data', '0', str(self.get_head())), 'ab') as fd:
+            fd.write(b'X')
+        self.assert_raises(Repository.CheckNeeded, self.reopen)
 
 class RemoteRepositoryTestCase(RepositoryTestCase):