fix: chunk store_load over RPC to avoid msgpack BufferFull

Large repositories can have a chunks index exceeding the msgpack
unpacker buffer size, causing BufferFull errors over SSH/RPC.

Instead of loading the entire value in one RPC call, add two new
server-side methods (store_get_size, store_load_chunk) and override
store_load in RemoteRepository to fetch data in MAX_DATA_SIZE pieces,
reassembling on the client side.

Fixes #8440

Author: hiijoshi

src/borg/archiver/create_cmd.py

@@ -576,8 +576,7 @@ def build_parser_create(self, subparsers, common_parser, mid_common_parser):
         from ._common import process_epilog
         from ._common import define_exclusion_group
 
-        create_epilog = process_epilog(
-            """
+        create_epilog = process_epilog("""
         This command creates a backup archive containing all files found while recursively
         traversing all specified paths. Paths are added to the archive as they are given,
         which means that if relative paths are desired, the command must be run from the correct
@@ -783,8 +782,7 @@ def build_parser_create(self, subparsers, common_parser, mid_common_parser):
 
         Borg supports paths with the slashdot hack to strip path prefixes here also.
         So, be careful not to unintentionally trigger that.
-        """
-        )
+        """)
 
         subparser = ArgumentParser(parents=[common_parser], description=self.do_create.__doc__, epilog=create_epilog)
         subparsers.add_subcommand("create", subparser, help="create a backup")

src/borg/helpers/parseformat.py

@@ -300,11 +300,24 @@ def ChunkerParams(s):
 def FilesCacheMode(s):
     ENTRIES_MAP = dict(ctime="c", mtime="m", size="s", inode="i", rechunk="r", disabled="d")
     VALID_MODES = ("cis", "ims", "cs", "ms", "cr", "mr", "d", "s")  # letters in alpha order
+    WIN32_INVALID_MODES = ("cis", "cs", "cr")  # modes containing ctime, invalid on Windows
     if s in VALID_MODES:
+        if is_win32 and s in WIN32_INVALID_MODES:
+            raise ArgumentTypeError(
+                "ctime is not supported in files-cache mode on Windows "
+                "(ctime means file creation time on Windows, not inode change time). "
+                "Use an mtime-based mode instead."
+            )
         return s
     entries = set(s.strip().split(","))
     if not entries <= set(ENTRIES_MAP):
         raise ArgumentTypeError("cache mode must be a comma-separated list of: %s" % ",".join(sorted(ENTRIES_MAP)))
+    if is_win32 and "ctime" in entries:
+        raise ArgumentTypeError(
+            "ctime is not supported in files-cache mode on Windows "
+            "(ctime means file creation time on Windows, not inode change time). "
+            "Use an mtime-based mode instead."
+        )
     short_entries = {ENTRIES_MAP[entry] for entry in entries}
     mode = "".join(sorted(short_entries))
     if mode not in VALID_MODES:

src/borg/remote.py

@@ -161,6 +161,8 @@ class RepositoryServer:  # pragma: no cover
         "put_manifest",
         "store_list",
         "store_load",
+        "store_load_chunk",
+        "store_get_size",
         "store_store",
         "store_delete",
         "store_move",
@@ -1049,9 +1051,16 @@ def put_manifest(self, data):
     def store_list(self, name, *, deleted=False):
         """actual remoting is done via self.call in the @api decorator"""
 
-    @api(since=parse_version("2.0.0b8"))
     def store_load(self, name):
-        """actual remoting is done via self.call in the @api decorator"""
+        # chunked fetch to avoid msgpack BufferFull on large repositories
+        total_size = self.call("store_get_size", {"name": name})
+        data = bytearray()
+        offset = 0
+        while offset < total_size:
+            chunk = self.call("store_load_chunk", {"name": name, "offset": offset, "size": MAX_DATA_SIZE})
+            data += chunk
+            offset += len(chunk)
+        return bytes(data)
 
     @api(since=parse_version("2.0.0b8"))
     def store_store(self, name, value):

src/borg/repository.py

@@ -560,6 +560,16 @@ def store_store(self, name, value):
         self._lock_refresh()
         return self.store.store(name, value)
 
+    def store_get_size(self, name):
+        self._lock_refresh()
+        data = self.store.load(name)
+        return len(data)
+
+    def store_load_chunk(self, name, offset, size):
+        self._lock_refresh()
+        data = self.store.load(name)
+        return data[offset : offset + size]
+
     def store_delete(self, name, *, deleted=False):
         self._lock_refresh()
         return self.store.delete(name, deleted=deleted)