Delete loose files per pack creation incrementally (#193)

This allows to delete loose objects immediately after each pack is
closed and committed to the database, instead of deleting all in a
manual cleanup step later.

---------

Co-authored-by: Giovanni Pizzi <gio.piz@gmail.com>

Author: GeigerJ2

disk_objectstore/container.py

@@ -1261,6 +1261,7 @@ def pack_all_loose(  # pylint: disable=too-many-locals,too-many-branches,too-man
         validate_objects: bool = True,
         do_fsync: bool = True,
         callback: None | (Callable[[str, Any], None]) = None,
+        clean_loose_per_pack: bool = False,
     ) -> None:
         """Pack all loose objects.
 
@@ -1276,10 +1277,11 @@ def pack_all_loose(  # pylint: disable=too-many-locals,too-many-branches,too-man
            Note that since loose objects are always uncompressed, `CompressMode.KEEP` is equivalent to
            `CompressMode.NO` in this function.
         :param validate_objects: if True, recompute the hash while packing, and raises if there is a problem.
-        :param do_fsync: if True, calls a flush to disk of the pack files before closing it.
+        :param do_fsync: if True, calls a flush to disk of each pack file just before closing it.
             Needed to guarantee that data will be there even in the case of a power loss.
-            Set this to False if you don't need such a guarantee (major risk of data loss if power
-            supply stops during packing operation-we don't recommend this).
+            Set to False if you don't need such a guarantee (major risk of data loss if this is set to False, and power
+            supply stops during packing operation). If you set `do_fsync=False` at least use
+            `clean_loose_per_pack=False`, and do a manual `fsync` before cleaning the loose objects via `clean_storage`.
         :param callback: a callback function that can be used to report progress.
             The callback function should accept two arguments: a string with the action being performed
             and the value of the action. The action can be "init" (initialization),
@@ -1289,6 +1291,8 @@ def pack_all_loose(  # pylint: disable=too-many-locals,too-many-branches,too-man
             In case of "update", the value is amount of the operation that has been completed.
             In case of "close", the value is None.
             return value of the callback function is ignored.
+        :param clean_loose_per_pack: if True, the loose files that went into a `pack` are deleted immediately after this
+            `pack` is created.
         """
         hash_type = self.hash_type if validate_objects else None
 
@@ -1336,9 +1340,6 @@ def pack_all_loose(  # pylint: disable=too-many-locals,too-many-branches,too-man
         # Here, I could do some clean up of loose objects that are already in the packs,
         # by removing all loose objects with hash key `existing_packed_hashkeys`, that are
         # already packed.
-        # HOWEVER, while this would work fine on Linux, there are concurrency issues both
-        # on Mac and on Windows (see issues #37 and #43). Therefore, I do NOT delete them,
-        # and deletion is deferred to a manual clean-up operation.
 
         if callback:
             callback(
@@ -1357,6 +1358,9 @@ def pack_all_loose(  # pylint: disable=too-many-locals,too-many-branches,too-man
         while loose_objects:
             # Store the last pack integer ID, needed to know later if I need to open a new pack
             last_pack_int_id = pack_int_id
+            # Track which objects were added to this pack for cleanup
+            packed_in_current_pack: list[str] = []
+
             # Avoid concurrent writes on the pack file
             with self.lock_pack(str(pack_int_id)) as pack_handle:
                 # Inner loop: continue until when there is a file, or
@@ -1431,6 +1435,9 @@ def pack_all_loose(  # pylint: disable=too-many-locals,too-many-branches,too-man
                     # Appending for later bulk commit - see comments in add_streamed_objects_to_pack
                     obj_dicts.append(obj_dict)
 
+                    # Track this object for potential cleanup
+                    packed_in_current_pack.append(loose_hashkey)
+
                     if callback:
                         callback(
                             'update',
@@ -1459,6 +1466,8 @@ def pack_all_loose(  # pylint: disable=too-many-locals,too-many-branches,too-man
                     )
 
             # OK, if we are here, file was flushed, synced to disk and closed.
+            # Note that this is only the case if do_fsync is True. This is the default,
+            # and we warn users that they should not set it to False.
             # Let's commit then the information to the DB, so it's officially a
             # packed object. Note: committing as soon as we are done with one pack,
             # so if there's a problem with one pack we don't start operating on the next one
@@ -1469,12 +1478,31 @@ def pack_all_loose(  # pylint: disable=too-many-locals,too-many-branches,too-man
             # Then, it would be safe to already do some clean up of loose objects that are now packed,
             # and by doing it here we would do it after each pack.
             # This would mean keeping track of the loose objects added to packs, and removing them.
-            # HOWEVER, while this would work fine on Linux, there are concurrency issues both
-            # on Mac and on Windows (see issues #37 and #43). Therefore, I do NOT delete them,
-            # and deletion is deferred to a manual clean-up operation.
+
+            # Clean up loose objects for this pack if requested
+            if clean_loose_per_pack and packed_in_current_pack:
+                self._clean_loose_objects(packed_in_current_pack)
         if callback:
             callback('close', None)
 
+    def _clean_loose_objects(self, hashkeys: list[str]) -> None:
+        """Clean up specific loose objects that have been successfully packed.
+
+        :param hashkeys: List of hashkeys to clean up from loose objects.
+            These should be objects that were just packed.
+        """
+
+        # Simply remove the loose files for the given hashkeys
+        # We trust that the caller has already ensured these are packed
+        for obj_hashkey in hashkeys:
+            try:
+                os.remove(self._get_loose_path_from_hashkey(obj_hashkey))
+            except (FileNotFoundError, PermissionError):
+                # FileNotFoundError: file might already be removed by another process
+                # PermissionError: file might be locked (especially on Windows)
+                # In both cases, we just continue - the file will be cleaned up later
+                pass
+
     def add_streamed_object_to_pack(  # pylint: disable=too-many-arguments
         self,
         stream: StreamSeekBytesType,

disk_objectstore/utils.py

@@ -1302,7 +1302,7 @@ def safe_flush_to_disk(
     real_path: Path,
     use_fullsync: bool = False,
 ) -> None:
-    """Tries to to its best to safely commit to disk.
+    """Tries to do its best to safely commit to disk.
 
     Note that calling this is needed to reduce the risk of data loss due to, e.g., a power failure.
     However, this call is typically expensive so should be called only if the guarantees are really needed, and

tests/concurrent_tests/periodic_packer.py

@@ -26,8 +26,13 @@ def timestamp():
 )
 @click.option('-r', '--repetitions', default=10, help='Number of repetitions before stopping.')
 @click.option('-w', '--wait-time', default=0.83, help='Time to wait between iterations.')
+@click.option(
+    '--clean-loose-per-pack/--no-clean-loose-per-pack',
+    default=False,
+    help='If set, clean loose objects after every pack operation.',
+)
 @click.help_option('-h', '--help')
-def main(path, repetitions, wait_time):
+def main(path, repetitions, wait_time, clean_loose_per_pack):
     """Periodically pack container."""
     container = Container(path)
     if not container.is_initialised:
@@ -39,7 +44,10 @@ def main(path, repetitions, wait_time):
 
         compress_packs = iteration % 2 == 0
         start_counts = container.count_objects()
-        container.pack_all_loose(compress=compress_packs)
+        container.pack_all_loose(
+            compress=compress_packs,
+            clean_loose_per_pack=clean_loose_per_pack,
+        )
         print(f'[PACKER {timestamp()}] Done packing!')
         end_counts = container.count_objects()
 

tests/test_basic.py

@@ -355,6 +355,88 @@ def test_exclusive_mode_windows(temp_dir, lock_file_on_windows):
         os.close(fd)
 
 
+def test_clean_loose_objects_empty(temp_container):
+    """Test early return."""
+    result = temp_container._clean_loose_objects([])
+    assert result is None
+
+
+def test_clean_loose_objects_file_not_found(temp_container):
+    """Test _clean_loose_objects when loose file doesn't exist."""
+    # Create a fake hashkey that won't have a corresponding loose file
+    fake_hashkey = 'a' * 64  # Valid hex string of typical hash length
+
+    # This should hit the FileNotFoundError exception, which, however, is being captured, so no raise
+    temp_container._clean_loose_objects([fake_hashkey])
+
+
+def test_clean_loose_objects_successful_cleanup(temp_container):
+    """Test _clean_loose_objects successfully removes loose files."""
+    contents = [b'content1', b'content2', b'content3']
+    hashkeys = []
+    loose_paths = []
+
+    for content in contents:
+        hashkey = temp_container.add_object(content)
+        hashkeys.append(hashkey)
+        loose_paths.append(temp_container._get_loose_path_from_hashkey(hashkey))
+
+    # Verify all loose files exist
+    assert all(path.exists() for path in loose_paths)
+
+    temp_container._clean_loose_objects(hashkeys)
+
+    # Verify all loose files are removed
+    assert all(not path.exists() for path in loose_paths)
+
+
+def test_clean_loose_objects_mixed_scenarios(temp_container):
+    """Test _clean_loose_objects with mix of existing and non-existing files."""
+    # Create one real loose object
+    content = b'real content'
+    real_hashkey = temp_container.add_object(content)
+    real_path = temp_container._get_loose_path_from_hashkey(real_hashkey)
+    assert real_path.exists()
+
+    # Create fake hashkeys
+    fake_hashkey1 = 'b' * 64
+    fake_hashkey2 = 'c' * 64
+
+    # Mix real and fake hashkeys
+    mixed_hashkeys = [fake_hashkey1, real_hashkey, fake_hashkey2]
+
+    # This should handle both successful deletion and FileNotFoundError
+    temp_container._clean_loose_objects(mixed_hashkeys)
+
+    # Real file should be deleted, fake ones just ignored
+    assert not real_path.exists()
+
+
+@pytest.mark.skipif(os.name != 'nt', reason='PermissionError test only relevant on Windows')
+def test_clean_loose_objects_permission_error_windows(temp_container, lock_file_on_windows):
+    """Test _clean_loose_objects when file is locked (Windows only)."""
+    # Create a loose object
+    content = b'test content'
+    hashkey = temp_container.add_object(content)
+
+    # Verify the loose file exists
+    loose_path = temp_container._get_loose_path_from_hashkey(hashkey)
+    assert loose_path.exists()
+
+    # Lock the file on Windows
+    fd = os.open(loose_path, os.O_RDONLY)
+    try:
+        lock_file_on_windows(fd)
+
+        # This should hit the PermissionError exception path
+        temp_container._clean_loose_objects([hashkey])
+
+        # File should still exist since it was locked
+        assert loose_path.exists()
+    finally:
+        os.close(fd)
+
+
 def test_get_pack_id_to_write_to_with_known_sizes(temp_container):
     """Unit test: Verify _get_pack_id_to_write_to correctly uses known_sizes parameter.