Implement garbage collection (gc) command

Add garbage collection functionality to Dulwich:

- Add dulwich/gc.py module with core gc implementation including:
  - find_reachable_objects(): Find all objects reachable from refs
  - find_unreachable_objects(): Find objects not reachable from refs
  - prune_unreachable_objects(): Delete unreachable loose objects
  - garbage_collect(): Main gc function with grace period support
  - GCStats dataclass for tracking gc statistics

- Add gc() function to dulwich/porcelain.py that wraps the core gc

- Add cmd_gc class to dulwich/cli.py with command-line argument parsing
  supporting --auto, --aggressive, --prune, --dry-run, and --quiet flags

- Add gc-related methods to ObjectStore:
  - get_object_mtime(): Get modification time of loose objects
  - delete_loose_object(): Delete loose objects (renamed from _remove_loose_object)
  - repack(exclude=...): Modified to support excluding objects during repacking

Fixes #92

NEWS

@@ -1,5 +1,7 @@
 0.22.9	UNRELEASED
 
+ * Add ``gc`` command to ``dulwich.porcelain.`` (Jelmer Vernooij, #92)
+
  * Add ``unpack-objects`` plumbing command to unpack objects from pack files
    into loose objects in the repository. This command extracts all objects
    from a pack file and writes them to the object store as individual files.

dulwich/cli.py

@@ -1044,6 +1044,102 @@ class cmd_merge_tree(Command):
             return 1
 
 
+class cmd_gc(Command):
+    def run(self, args) -> Optional[int]:
+        import datetime
+        import time
+
+        parser = argparse.ArgumentParser()
+        parser.add_argument(
+            "--auto",
+            action="store_true",
+            help="Only run gc if needed",
+        )
+        parser.add_argument(
+            "--aggressive",
+            action="store_true",
+            help="Use more aggressive settings",
+        )
+        parser.add_argument(
+            "--no-prune",
+            action="store_true",
+            help="Do not prune unreachable objects",
+        )
+        parser.add_argument(
+            "--prune",
+            nargs="?",
+            const="now",
+            help="Prune unreachable objects older than date (default: 2 weeks ago)",
+        )
+        parser.add_argument(
+            "--dry-run",
+            "-n",
+            action="store_true",
+            help="Only report what would be done",
+        )
+        parser.add_argument(
+            "--quiet",
+            "-q",
+            action="store_true",
+            help="Only report errors",
+        )
+        args = parser.parse_args(args)
+
+        # Parse prune grace period
+        grace_period = None
+        if args.prune:
+            try:
+                grace_period = parse_relative_time(args.prune)
+            except ValueError:
+                # Try to parse as absolute date
+                try:
+                    date = datetime.datetime.strptime(args.prune, "%Y-%m-%d")
+                    grace_period = int(time.time() - date.timestamp())
+                except ValueError:
+                    print(f"Error: Invalid prune date: {args.prune}")
+                    return 1
+        elif not args.no_prune:
+            # Default to 2 weeks
+            grace_period = 1209600
+
+        # Progress callback
+        def progress(msg):
+            if not args.quiet:
+                print(msg)
+
+        try:
+            stats = porcelain.gc(
+                ".",
+                auto=args.auto,
+                aggressive=args.aggressive,
+                prune=not args.no_prune,
+                grace_period=grace_period,
+                dry_run=args.dry_run,
+                progress=progress if not args.quiet else None,
+            )
+
+            # Report results
+            if not args.quiet:
+                if args.dry_run:
+                    print("\nDry run results:")
+                else:
+                    print("\nGarbage collection complete:")
+
+                if stats.pruned_objects:
+                    print(f"  Pruned {len(stats.pruned_objects)} unreachable objects")
+                    print(f"  Freed {format_bytes(stats.bytes_freed)}")
+
+                if stats.packs_before != stats.packs_after:
+                    print(
+                        f"  Reduced pack files from {stats.packs_before} to {stats.packs_after}"
+                    )
+
+        except porcelain.Error as e:
+            print(f"Error: {e}")
+            return 1
+        return None
+
+
 class cmd_help(Command):
     def run(self, args) -> None:
         parser = argparse.ArgumentParser()
@@ -1090,6 +1186,7 @@ commands = {
     "fetch": cmd_fetch,
     "for-each-ref": cmd_for_each_ref,
     "fsck": cmd_fsck,
+    "gc": cmd_gc,
     "help": cmd_help,
     "init": cmd_init,
     "log": cmd_log,

dulwich/gc.py

@@ -0,0 +1,288 @@
+"""Git garbage collection implementation."""
+
+import collections
+import time
+from dataclasses import dataclass, field
+from typing import Optional, Deque
+
+from dulwich.object_store import BaseObjectStore, PackBasedObjectStore
+from dulwich.objects import Commit, Tag, Tree, ObjectID
+from dulwich.refs import RefsContainer
+
+
+@dataclass
+class GCStats:
+    """Statistics from garbage collection."""
+
+    pruned_objects: set[bytes] = field(default_factory=set)
+    bytes_freed: int = 0
+    packs_before: int = 0
+    packs_after: int = 0
+    loose_objects_before: int = 0
+    loose_objects_after: int = 0
+
+
+def find_reachable_objects(
+    object_store: BaseObjectStore,
+    refs_container: RefsContainer,
+    include_reflogs: bool = True,
+    progress=None,
+) -> set[bytes]:
+    """Find all reachable objects in the repository.
+
+    Args:
+        object_store: Object store to search
+        refs_container: Reference container
+        include_reflogs: Whether to include reflog entries
+        progress: Optional progress callback
+
+    Returns:
+        Set of reachable object SHAs
+    """
+    reachable = set()
+    pending: Deque[ObjectID] = collections.deque()
+
+    # Start with all refs
+    for ref in refs_container.allkeys():
+        try:
+            sha = refs_container[ref]  # This follows symbolic refs
+            if sha and sha not in reachable:
+                pending.append(sha)
+                reachable.add(sha)
+        except KeyError:
+            # Broken ref
+            if progress:
+                progress(f"Warning: Broken ref {ref.decode('utf-8', 'replace')}")
+            continue
+
+    # TODO: Add reflog support when reflog functionality is available
+
+    # Walk all reachable objects
+    while pending:
+        sha = pending.popleft()
+
+        if progress:
+            progress(f"Checking object {sha.decode('ascii', 'replace')}")
+
+        try:
+            obj = object_store[sha]
+        except KeyError:
+            continue
+
+        # Add referenced objects
+        if isinstance(obj, Commit):
+            # Tree
+            if obj.tree not in reachable:
+                pending.append(obj.tree)
+                reachable.add(obj.tree)
+            # Parents
+            for parent in obj.parents:
+                if parent not in reachable:
+                    pending.append(parent)
+                    reachable.add(parent)
+        elif isinstance(obj, Tree):
+            # Tree entries
+            for entry in obj.items():
+                if entry.sha not in reachable:
+                    pending.append(entry.sha)
+                    reachable.add(entry.sha)
+        elif isinstance(obj, Tag):
+            # Tagged object
+            if obj.object[1] not in reachable:
+                pending.append(obj.object[1])
+                reachable.add(obj.object[1])
+
+    return reachable
+
+
+def find_unreachable_objects(
+    object_store: BaseObjectStore,
+    refs_container: RefsContainer,
+    include_reflogs: bool = True,
+    progress=None,
+) -> set[bytes]:
+    """Find all unreachable objects in the repository.
+
+    Args:
+        object_store: Object store to search
+        refs_container: Reference container
+        include_reflogs: Whether to include reflog entries
+        progress: Optional progress callback
+
+    Returns:
+        Set of unreachable object SHAs
+    """
+    reachable = find_reachable_objects(
+        object_store, refs_container, include_reflogs, progress
+    )
+
+    unreachable = set()
+    for sha in object_store:
+        if sha not in reachable:
+            unreachable.add(sha)
+
+    return unreachable
+
+
+def prune_unreachable_objects(
+    object_store: PackBasedObjectStore,
+    refs_container: RefsContainer,
+    grace_period: Optional[int] = None,
+    dry_run: bool = False,
+    progress=None,
+) -> tuple[set[bytes], int]:
+    """Remove unreachable objects from the repository.
+
+    Args:
+        object_store: Object store to prune
+        refs_container: Reference container
+        grace_period: Grace period in seconds (objects newer than this are kept)
+        dry_run: If True, only report what would be deleted
+        progress: Optional progress callback
+
+    Returns:
+        Tuple of (set of pruned object SHAs, total bytes freed)
+    """
+    unreachable = find_unreachable_objects(
+        object_store, refs_container, progress=progress
+    )
+
+    pruned = set()
+    bytes_freed = 0
+
+    for sha in unreachable:
+        try:
+            obj = object_store[sha]
+
+            # Check grace period
+            if grace_period is not None:
+                mtime = object_store.get_object_mtime(sha)
+                if mtime is not None:
+                    age = time.time() - mtime
+                    if age < grace_period:
+                        if progress:
+                            progress(
+                                f"Keeping {sha.decode('ascii', 'replace')} (age: {age:.0f}s < grace period: {grace_period}s)"
+                            )
+                        continue
+
+            if progress:
+                progress(f"Pruning {sha.decode('ascii', 'replace')}")
+
+            # Calculate size before attempting deletion
+            obj_size = len(obj.as_raw_string())
+
+            if not dry_run:
+                object_store.delete_loose_object(sha)
+
+            # Only count as pruned if we get here (deletion succeeded or dry run)
+            pruned.add(sha)
+            bytes_freed += obj_size
+
+        except KeyError:
+            # Object already gone
+            pass
+        except OSError as e:
+            # File system errors during deletion
+            if progress:
+                progress(f"Error pruning {sha.decode('ascii', 'replace')}: {e}")
+    return pruned, bytes_freed
+
+
+def garbage_collect(
+    repo,
+    auto: bool = False,
+    aggressive: bool = False,
+    prune: bool = True,
+    grace_period: Optional[int] = 1209600,  # 2 weeks default
+    dry_run: bool = False,
+    progress=None,
+) -> GCStats:
+    """Run garbage collection on a repository.
+
+    Args:
+        repo: Repository to garbage collect
+        auto: Whether this is an automatic gc
+        aggressive: Whether to use aggressive settings
+        prune: Whether to prune unreachable objects
+        grace_period: Grace period for pruning in seconds
+        dry_run: If True, only report what would be done
+        progress: Optional progress callback
+
+    Returns:
+        GCStats object with garbage collection statistics
+    """
+    stats = GCStats()
+
+    object_store = repo.object_store
+    refs_container = repo.refs
+
+    # Count initial state
+    stats.packs_before = len(list(object_store.packs))
+    # TODO: Count loose objects when we have a method for it
+
+    # Find unreachable objects to exclude from repacking
+    unreachable_to_prune = set()
+    if prune:
+        if progress:
+            progress("Finding unreachable objects")
+        unreachable = find_unreachable_objects(
+            object_store, refs_container, progress=progress
+        )
+
+        # Apply grace period check
+        for sha in unreachable:
+            try:
+                if grace_period is not None:
+                    mtime = object_store.get_object_mtime(sha)
+                    if mtime is not None:
+                        age = time.time() - mtime
+                        if age < grace_period:
+                            if progress:
+                                progress(
+                                    f"Keeping {sha.decode('ascii', 'replace')} (age: {age:.0f}s < grace period: {grace_period}s)"
+                                )
+                            continue
+
+                unreachable_to_prune.add(sha)
+                obj = object_store[sha]
+                stats.bytes_freed += len(obj.as_raw_string())
+            except KeyError:
+                pass
+
+        stats.pruned_objects = unreachable_to_prune
+
+    # Pack refs
+    if progress:
+        progress("Packing references")
+    if not dry_run:
+        from dulwich.porcelain import pack_refs
+
+        pack_refs(repo)
+
+    # Delete loose unreachable objects
+    if prune and not dry_run:
+        for sha in unreachable_to_prune:
+            if object_store.contains_loose(sha):
+                try:
+                    object_store.delete_loose_object(sha)
+                except OSError:
+                    pass
+
+    # Repack everything, excluding unreachable objects
+    # This handles both loose object packing and pack consolidation
+    if progress:
+        progress("Repacking repository")
+    if not dry_run:
+        if prune and unreachable_to_prune:
+            # Repack excluding unreachable objects
+            object_store.repack(exclude=unreachable_to_prune)
+        else:
+            # Normal repack
+            object_store.repack()
+
+    # Count final state
+    stats.packs_after = len(list(object_store.packs))
+    # TODO: Count loose objects when we have a method for it
+
+    return stats

dulwich/object_store.py

@@ -387,6 +387,19 @@ class BaseObjectStore:
         """
         raise NotImplementedError(self.write_commit_graph)
 
+    def get_object_mtime(self, sha):
+        """Get the modification time of an object.
+
+        Args:
+          sha: SHA1 of the object
+
+        Returns:
+          Modification time as seconds since epoch, or None if not available
+        """
+        # Default implementation returns None
+        # Subclasses can override to provide actual mtime
+        return None
+
 
 class PackBasedObjectStore(BaseObjectStore):
     def __init__(self, pack_compression_level=-1) -> None:
@@ -518,8 +531,13 @@ class PackBasedObjectStore(BaseObjectStore):
     def _get_loose_object(self, sha) -> Optional[ShaFile]:
         raise NotImplementedError(self._get_loose_object)
 
-    def _remove_loose_object(self, sha) -> None:
-        raise NotImplementedError(self._remove_loose_object)
+    def delete_loose_object(self, sha) -> None:
+        """Delete a loose object.
+
+        This method only handles loose objects. For packed objects,
+        use repack(exclude=...) to exclude them during repacking.
+        """
+        raise NotImplementedError(self.delete_loose_object)
 
     def _remove_pack(self, name) -> None:
         raise NotImplementedError(self._remove_pack)
@@ -534,32 +552,50 @@ class PackBasedObjectStore(BaseObjectStore):
             objects.add((self._get_loose_object(sha), None))
         self.add_objects(list(objects))
         for obj, path in objects:
-            self._remove_loose_object(obj.id)
+            self.delete_loose_object(obj.id)
         return len(objects)
 
-    def repack(self):
+    def repack(self, exclude=None):
         """Repack the packs in this repository.
 
         Note that this implementation is fairly naive and currently keeps all
         objects in memory while it repacks.
+
+        Args:
+          exclude: Optional set of object SHAs to exclude from repacking
         """
+        if exclude is None:
+            exclude = set()
+
         loose_objects = set()
+        excluded_loose_objects = set()
         for sha in self._iter_loose_objects():
-            loose_objects.add(self._get_loose_object(sha))
+            if sha not in exclude:
+                loose_objects.add(self._get_loose_object(sha))
+            else:
+                excluded_loose_objects.add(sha)
+
         objects = {(obj, None) for obj in loose_objects}
         old_packs = {p.name(): p for p in self.packs}
         for name, pack in old_packs.items():
-            objects.update((obj, None) for obj in pack.iterobjects())
-
-        # The name of the consolidated pack might match the name of a
-        # pre-existing pack. Take care not to remove the newly created
-        # consolidated pack.
+            objects.update(
+                (obj, None) for obj in pack.iterobjects() if obj.id not in exclude
+            )
 
-        consolidated = self.add_objects(objects)
-        old_packs.pop(consolidated.name(), None)
+        # Only create a new pack if there are objects to pack
+        if objects:
+            # The name of the consolidated pack might match the name of a
+            # pre-existing pack. Take care not to remove the newly created
+            # consolidated pack.
+            consolidated = self.add_objects(objects)
+            old_packs.pop(consolidated.name(), None)
 
+        # Delete loose objects that were packed
         for obj in loose_objects:
-            self._remove_loose_object(obj.id)
+            self.delete_loose_object(obj.id)
+        # Delete excluded loose objects
+        for sha in excluded_loose_objects:
+            self.delete_loose_object(sha)
         for name, pack in old_packs.items():
             self._remove_pack(pack)
         self._update_pack_cache()
@@ -893,9 +929,27 @@ class DiskObjectStore(PackBasedObjectStore):
         except FileNotFoundError:
             return None
 
-    def _remove_loose_object(self, sha) -> None:
+    def delete_loose_object(self, sha) -> None:
         os.remove(self._get_shafile_path(sha))
 
+    def get_object_mtime(self, sha):
+        """Get the modification time of a loose object.
+
+        Args:
+          sha: SHA1 of the object
+
+        Returns:
+          Modification time as seconds since epoch, or None if not a loose object
+        """
+        if not self.contains_loose(sha):
+            return None
+
+        path = self._get_shafile_path(sha)
+        try:
+            return os.path.getmtime(path)
+        except (OSError, FileNotFoundError):
+            return None
+
     def _remove_pack(self, pack) -> None:
         try:
             del self._pack_cache[os.path.basename(pack._basename)]
@@ -1781,7 +1835,7 @@ class BucketBasedObjectStore(PackBasedObjectStore):
     def _get_loose_object(self, sha) -> None:
         return None
 
-    def _remove_loose_object(self, sha) -> None:
+    def delete_loose_object(self, sha) -> None:
         # Doesn't exist..
         pass
 

dulwich/porcelain.py

@@ -2893,3 +2893,40 @@ def merge_tree(repo, base_tree, our_tree, their_tree):
         r.object_store.add_object(merged_tree)
 
         return merged_tree.id, conflicts
+
+
+def gc(
+    repo,
+    auto: bool = False,
+    aggressive: bool = False,
+    prune: bool = True,
+    grace_period: Optional[int] = 1209600,  # 2 weeks default
+    dry_run: bool = False,
+    progress=None,
+):
+    """Run garbage collection on a repository.
+
+    Args:
+      repo: Path to the repository or a Repo object
+      auto: If True, only run gc if needed
+      aggressive: If True, use more aggressive settings
+      prune: If True, prune unreachable objects
+      grace_period: Grace period in seconds for pruning (default 2 weeks)
+      dry_run: If True, only report what would be done
+      progress: Optional progress callback
+
+    Returns:
+      GCStats object with garbage collection statistics
+    """
+    from .gc import garbage_collect
+
+    with open_repo_closing(repo) as r:
+        return garbage_collect(
+            r,
+            auto=auto,
+            aggressive=aggressive,
+            prune=prune,
+            grace_period=grace_period,
+            dry_run=dry_run,
+            progress=progress,
+        )

dulwich/tests/test_cli.py

@@ -77,7 +77,9 @@ class ParseRelativeTimeTestCase(TestCase):
         """Test parsing weeks."""
         self.assertEqual(604800, parse_relative_time("1 week ago"))
         self.assertEqual(1209600, parse_relative_time("2 weeks ago"))
-        self.assertEqual(36288000, parse_relative_time("60 weeks ago"))  # 60 * 7 * 24 * 60 * 60
+        self.assertEqual(
+            36288000, parse_relative_time("60 weeks ago")
+        )  # 60 * 7 * 24 * 60 * 60
 
     def test_invalid_format(self):
         """Test invalid time formats."""
@@ -109,8 +111,18 @@ class ParseRelativeTimeTestCase(TestCase):
 
     def test_singular_plural(self):
         """Test that both singular and plural forms work."""
-        self.assertEqual(parse_relative_time("1 second ago"), parse_relative_time("1 seconds ago"))
-        self.assertEqual(parse_relative_time("1 minute ago"), parse_relative_time("1 minutes ago"))
-        self.assertEqual(parse_relative_time("1 hour ago"), parse_relative_time("1 hours ago"))
-        self.assertEqual(parse_relative_time("1 day ago"), parse_relative_time("1 days ago"))
-        self.assertEqual(parse_relative_time("1 week ago"), parse_relative_time("1 weeks ago"))
+        self.assertEqual(
+            parse_relative_time("1 second ago"), parse_relative_time("1 seconds ago")
+        )
+        self.assertEqual(
+            parse_relative_time("1 minute ago"), parse_relative_time("1 minutes ago")
+        )
+        self.assertEqual(
+            parse_relative_time("1 hour ago"), parse_relative_time("1 hours ago")
+        )
+        self.assertEqual(
+            parse_relative_time("1 day ago"), parse_relative_time("1 days ago")
+        )
+        self.assertEqual(
+            parse_relative_time("1 week ago"), parse_relative_time("1 weeks ago")
+        )

dulwich/tests/test_gc.py

@@ -0,0 +1,261 @@
+"""Tests for dulwich.gc."""
+
+import shutil
+import tempfile
+from unittest import TestCase
+
+from dulwich.gc import (
+    GCStats,
+    find_reachable_objects,
+    find_unreachable_objects,
+    garbage_collect,
+    prune_unreachable_objects,
+)
+from dulwich.objects import Blob, Commit, Tree
+from dulwich.repo import Repo
+
+
+class GCTestCase(TestCase):
+    """Tests for garbage collection functionality."""
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+        self.repo = Repo.init(self.tmpdir)
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir)
+
+    def test_find_reachable_objects_empty_repo(self):
+        """Test finding reachable objects in empty repository."""
+        reachable = find_reachable_objects(self.repo.object_store, self.repo.refs)
+        self.assertEqual(set(), reachable)
+
+    def test_find_reachable_objects_with_commit(self):
+        """Test finding reachable objects with a commit."""
+        # Create a blob
+        blob = Blob.from_string(b"test content")
+        self.repo.object_store.add_object(blob)
+
+        # Create a tree
+        tree = Tree()
+        tree.add(b"test.txt", 0o100644, blob.id)
+        self.repo.object_store.add_object(tree)
+
+        # Create a commit
+        commit = Commit()
+        commit.tree = tree.id
+        commit.author = commit.committer = b"Test User <test@example.com>"
+        commit.commit_time = commit.author_time = 1234567890
+        commit.commit_timezone = commit.author_timezone = 0
+        commit.message = b"Test commit"
+        self.repo.object_store.add_object(commit)
+
+        # Set HEAD to the commit
+        self.repo.refs[b"HEAD"] = commit.id
+
+        # Find reachable objects
+        reachable = find_reachable_objects(self.repo.object_store, self.repo.refs)
+
+        # All three objects should be reachable
+        self.assertEqual({blob.id, tree.id, commit.id}, reachable)
+
+    def test_find_unreachable_objects(self):
+        """Test finding unreachable objects."""
+        # Create a reachable blob
+        reachable_blob = Blob.from_string(b"reachable content")
+        self.repo.object_store.add_object(reachable_blob)
+
+        # Create a tree
+        tree = Tree()
+        tree.add(b"reachable.txt", 0o100644, reachable_blob.id)
+        self.repo.object_store.add_object(tree)
+
+        # Create a commit
+        commit = Commit()
+        commit.tree = tree.id
+        commit.author = commit.committer = b"Test User <test@example.com>"
+        commit.commit_time = commit.author_time = 1234567890
+        commit.commit_timezone = commit.author_timezone = 0
+        commit.message = b"Test commit"
+        self.repo.object_store.add_object(commit)
+
+        # Set HEAD to the commit
+        self.repo.refs[b"HEAD"] = commit.id
+
+        # Create an unreachable blob
+        unreachable_blob = Blob.from_string(b"unreachable content")
+        self.repo.object_store.add_object(unreachable_blob)
+
+        # Find unreachable objects
+        unreachable = find_unreachable_objects(self.repo.object_store, self.repo.refs)
+
+        # Only the unreachable blob should be found
+        self.assertEqual({unreachable_blob.id}, unreachable)
+
+    def test_prune_unreachable_objects(self):
+        """Test pruning unreachable objects."""
+        # Create an unreachable blob
+        unreachable_blob = Blob.from_string(b"unreachable content")
+        self.repo.object_store.add_object(unreachable_blob)
+
+        # Verify it exists
+        self.assertIn(unreachable_blob.id, self.repo.object_store)
+
+        # Prune unreachable objects
+        pruned, bytes_freed = prune_unreachable_objects(
+            self.repo.object_store, self.repo.refs, grace_period=0
+        )
+
+        # Verify the blob was pruned
+        self.assertEqual({unreachable_blob.id}, pruned)
+        self.assertGreater(bytes_freed, 0)
+
+        # Note: We can't test that the object is gone because delete()
+        # only supports loose objects and may not be fully implemented
+
+    def test_prune_unreachable_objects_dry_run(self):
+        """Test pruning unreachable objects with dry run."""
+        # Create an unreachable blob
+        unreachable_blob = Blob.from_string(b"unreachable content")
+        self.repo.object_store.add_object(unreachable_blob)
+
+        # Prune with dry run
+        pruned, bytes_freed = prune_unreachable_objects(
+            self.repo.object_store, self.repo.refs, grace_period=0, dry_run=True
+        )
+
+        # Verify the blob would be pruned but still exists
+        self.assertEqual({unreachable_blob.id}, pruned)
+        self.assertGreater(bytes_freed, 0)
+        self.assertIn(unreachable_blob.id, self.repo.object_store)
+
+    def test_garbage_collect(self):
+        """Test full garbage collection."""
+        # Create some reachable objects
+        blob = Blob.from_string(b"test content")
+        self.repo.object_store.add_object(blob)
+
+        tree = Tree()
+        tree.add(b"test.txt", 0o100644, blob.id)
+        self.repo.object_store.add_object(tree)
+
+        commit = Commit()
+        commit.tree = tree.id
+        commit.author = commit.committer = b"Test User <test@example.com>"
+        commit.commit_time = commit.author_time = 1234567890
+        commit.commit_timezone = commit.author_timezone = 0
+        commit.message = b"Test commit"
+        self.repo.object_store.add_object(commit)
+
+        self.repo.refs[b"HEAD"] = commit.id
+
+        # Create an unreachable blob
+        unreachable_blob = Blob.from_string(b"unreachable content")
+        self.repo.object_store.add_object(unreachable_blob)
+
+        # Run garbage collection
+        stats = garbage_collect(self.repo, prune=True, grace_period=0)
+
+        # Check results
+        self.assertIsInstance(stats, GCStats)
+        self.assertEqual({unreachable_blob.id}, stats.pruned_objects)
+        self.assertGreater(stats.bytes_freed, 0)
+
+    def test_garbage_collect_no_prune(self):
+        """Test garbage collection without pruning."""
+        # Create an unreachable blob
+        unreachable_blob = Blob.from_string(b"unreachable content")
+        self.repo.object_store.add_object(unreachable_blob)
+
+        # Run garbage collection without pruning
+        stats = garbage_collect(self.repo, prune=False)
+
+        # Check that nothing was pruned
+        self.assertEqual(set(), stats.pruned_objects)
+        self.assertEqual(0, stats.bytes_freed)
+        self.assertIn(unreachable_blob.id, self.repo.object_store)
+
+    def test_garbage_collect_dry_run(self):
+        """Test garbage collection with dry run."""
+        # Create an unreachable blob
+        unreachable_blob = Blob.from_string(b"unreachable content")
+        self.repo.object_store.add_object(unreachable_blob)
+
+        # Run garbage collection with dry run
+        stats = garbage_collect(self.repo, prune=True, grace_period=0, dry_run=True)
+
+        # Check that object would be pruned but still exists
+        self.assertEqual({unreachable_blob.id}, stats.pruned_objects)
+        self.assertGreater(stats.bytes_freed, 0)
+        self.assertIn(unreachable_blob.id, self.repo.object_store)
+
+    def test_grace_period(self):
+        """Test that grace period prevents pruning recent objects."""
+        # Create an unreachable blob
+        unreachable_blob = Blob.from_string(b"recent unreachable content")
+        self.repo.object_store.add_object(unreachable_blob)
+
+        # Ensure the object is loose
+        self.assertTrue(self.repo.object_store.contains_loose(unreachable_blob.id))
+
+        # Run garbage collection with a 1 hour grace period, but dry run to avoid packing
+        # The object was just created, so it should not be pruned
+        stats = garbage_collect(self.repo, prune=True, grace_period=3600, dry_run=True)
+
+        # Check that the object was NOT pruned
+        self.assertEqual(set(), stats.pruned_objects)
+        self.assertEqual(0, stats.bytes_freed)
+        self.assertIn(unreachable_blob.id, self.repo.object_store)
+
+        # Now test with zero grace period - it should be pruned
+        stats = garbage_collect(self.repo, prune=True, grace_period=0)
+
+        # Check that the object was pruned
+        self.assertEqual({unreachable_blob.id}, stats.pruned_objects)
+        self.assertGreater(stats.bytes_freed, 0)
+
+    def test_grace_period_old_object(self):
+        """Test that old objects are pruned even with grace period."""
+        import os
+        import time
+
+        # Create an unreachable blob
+        old_blob = Blob.from_string(b"old unreachable content")
+        self.repo.object_store.add_object(old_blob)
+
+        # Ensure the object is loose
+        self.assertTrue(self.repo.object_store.contains_loose(old_blob.id))
+
+        # Manually set the mtime to 2 hours ago
+        path = self.repo.object_store._get_shafile_path(old_blob.id)
+        old_time = time.time() - 7200  # 2 hours ago
+        os.utime(path, (old_time, old_time))
+
+        # Run garbage collection with a 1 hour grace period
+        # The object is 2 hours old, so it should be pruned
+        stats = garbage_collect(self.repo, prune=True, grace_period=3600)
+
+        # Check that the object was pruned
+        self.assertEqual({old_blob.id}, stats.pruned_objects)
+        self.assertGreater(stats.bytes_freed, 0)
+
+    def test_packed_objects_pruned(self):
+        """Test that packed objects are pruned via repack with exclusion."""
+        # Create an unreachable blob
+        unreachable_blob = Blob.from_string(b"unreachable packed content")
+        self.repo.object_store.add_object(unreachable_blob)
+
+        # Pack the objects to ensure the blob is in a pack
+        self.repo.object_store.pack_loose_objects()
+
+        # Ensure the object is NOT loose anymore
+        self.assertFalse(self.repo.object_store.contains_loose(unreachable_blob.id))
+        self.assertIn(unreachable_blob.id, self.repo.object_store)
+
+        # Run garbage collection
+        stats = garbage_collect(self.repo, prune=True, grace_period=0)
+
+        # Check that the packed object was pruned
+        self.assertEqual({unreachable_blob.id}, stats.pruned_objects)
+        self.assertGreater(stats.bytes_freed, 0)
+        self.assertNotIn(unreachable_blob.id, self.repo.object_store)

dulwich/tests/test_object_store.py

@@ -55,6 +55,8 @@ class ObjectStoreTests:
     assertNotIn: Callable[[object, object], None]
     assertNotEqual: Callable[[object, object], None]
     assertIn: Callable[[object, object], None]
+    assertTrue: Callable[[bool], None]
+    assertFalse: Callable[[bool], None]
 
     def test_determine_wants_all(self) -> None:
         self.assertEqual(
@@ -353,3 +355,43 @@ class PackBasedObjectStoreTests(ObjectStoreTests):
         self.assertEqual(2, self.store.repack())
         self.assertEqual(1, len(self.store.packs))
         self.assertEqual(0, self.store.pack_loose_objects())
+
+    def test_repack_with_exclude(self) -> None:
+        """Test repacking while excluding specific objects."""
+        b1 = make_object(Blob, data=b"yummy data")
+        self.store.add_object(b1)
+        b2 = make_object(Blob, data=b"more yummy data")
+        self.store.add_object(b2)
+        b3 = make_object(Blob, data=b"even more yummy data")
+        b4 = make_object(Blob, data=b"and more yummy data")
+        self.store.add_objects([(b3, None), (b4, None)])
+
+        self.assertEqual({b1.id, b2.id, b3.id, b4.id}, set(self.store))
+        self.assertEqual(1, len(self.store.packs))
+
+        # Repack, excluding b2 and b3
+        excluded = {b2.id, b3.id}
+        self.assertEqual(2, self.store.repack(exclude=excluded))
+
+        # Should have repacked only b1 and b4
+        self.assertEqual(1, len(self.store.packs))
+        self.assertIn(b1.id, self.store)
+        self.assertNotIn(b2.id, self.store)
+        self.assertNotIn(b3.id, self.store)
+        self.assertIn(b4.id, self.store)
+
+    def test_delete_loose_object(self) -> None:
+        """Test deleting loose objects."""
+        b1 = make_object(Blob, data=b"test data")
+        self.store.add_object(b1)
+
+        # Verify it's loose
+        self.assertTrue(self.store.contains_loose(b1.id))
+        self.assertIn(b1.id, self.store)
+
+        # Delete it
+        self.store.delete_loose_object(b1.id)
+
+        # Verify it's gone
+        self.assertFalse(self.store.contains_loose(b1.id))
+        self.assertNotIn(b1.id, self.store)