export-tar/import-tar: zstd (de)compression in-process, see #10067

With --tar-filter=auto, .tar.zst/.tar.zstd/.tzst tarballs (.tzst newly
recognized) are now (de)compressed in-process using the zstd module
(Python >= 3.14 stdlib / backports.zstd) instead of piping through an
external zstd process. borg requires the zstd module anyway, libzstd
multithreading (BORG_ZSTD_MT_WORKERS) runs outside the GIL, and no
external tool needs to be installed.

Compression uses level 3 (the zstd tool's default). An explicit
--tar-filter program is still run as an external filter, also for zstd.

Co-Authored-By: Claude Fable 5 <[email protected]>

docs/usage/export-tar.rst.inc

@@ -95,9 +95,13 @@ before writing it to FILE:
 - .tar.gz or .tgz: gzip
 - .tar.bz2 or .tbz: bzip2
 - .tar.xz or .txz: xz
-- .tar.zstd or .tar.zst: zstd
+- .tar.zstd, .tar.zst or .tzst: zstd (in-process, level 3)
 - .tar.lz4: lz4
 
+For zstd, Borg compresses in-process using libzstd instead of piping through an
+external program (set BORG_ZSTD_MT_WORKERS to use multiple compression threads).
+The other formats are piped through the respective external filter program.
+
 Alternatively, a ``--tar-filter`` program may be explicitly specified. It should
 read the uncompressed tar stream from stdin and write a compressed/filtered
 tar stream to stdout.

docs/usage/import-tar.rst.inc

@@ -96,9 +96,13 @@ based on its file extension and pipe the file through an appropriate filter:
 - .tar.gz or .tgz: gzip -d
 - .tar.bz2 or .tbz: bzip2 -d
 - .tar.xz or .txz: xz -d
-- .tar.zstd or .tar.zst: zstd -d
+- .tar.zstd, .tar.zst or .tzst: zstd (in-process)
 - .tar.lz4: lz4 -d
 
+For zstd, Borg decompresses in-process using libzstd instead of piping through
+an external program. The other formats are piped through the respective
+external filter program.
+
 Alternatively, a --tar-filter program may be explicitly specified. It should
 read compressed data from stdin and output an uncompressed tar stream on
 stdout.

src/borg/archiver/tar_cmds.py

@@ -1,11 +1,20 @@
 import base64
+import contextlib
 import logging
 import os
 import stat
+import sys
 import tarfile
 
+if sys.version_info >= (3, 14):
+    from compression import zstd
+else:
+    from backports import zstd
+
 from ..archive import Archive, TarfileObjectProcessors, ChunksProcessor
+from ..compress import get_zstd_mt_workers
 from ..constants import *  # NOQA
+from ..helpers import Error
 from ..helpers import HardLinkManager, IncludePatternNeverMatchedWarning
 from ..helpers import ProgressIndicatorPercent
 from ..helpers import dash_open
@@ -146,6 +155,14 @@ def item_to_paxheaders(format, item):
     return ph
 
 
+# Sentinel returned by get_tar_filter for zstd suffixes: (de)compress in-process
+# via the zstd module instead of piping through an external filter program.
+IN_PROCESS_ZSTD = "zstd (in-process)"
+
+# Default zstd compression level, same as the zstd command line tool's default.
+ZSTD_TAR_LEVEL = 3
+
+
 def get_tar_filter(fname, decompress):
     # Note that filter is None if fname is '-'.
     if fname.endswith((".tar.gz", ".tgz")):
@@ -156,14 +173,39 @@ def get_tar_filter(fname, decompress):
         filter = "xz -d" if decompress else "xz"
     elif fname.endswith((".tar.lz4",)):
         filter = "lz4 -d" if decompress else "lz4"
-    elif fname.endswith((".tar.zstd", ".tar.zst")):
-        filter = "zstd -d" if decompress else "zstd"
+    elif fname.endswith((".tar.zstd", ".tar.zst", ".tzst")):
+        filter = IN_PROCESS_ZSTD
     else:
         filter = None
     logger.debug("Automatically determined tar filter: %s", filter)
     return filter
 
 
+@contextlib.contextmanager
+def create_zstd_filter(stream, stream_close, decompress):
+    """In-process zstd (de)compression wrapper around *stream*, same contract as create_filter_process."""
+    if decompress:
+        zstream = zstd.ZstdFile(stream, "rb")
+    else:
+        workers = get_zstd_mt_workers()
+        if workers > 1:
+            params = zstd.CompressionParameter
+            options = {params.compression_level: ZSTD_TAR_LEVEL, params.nb_workers: workers}
+            zstream = zstd.ZstdFile(stream, "wb", options=options)
+        else:
+            zstream = zstd.ZstdFile(stream, "wb", level=ZSTD_TAR_LEVEL)
+    try:
+        yield zstream
+    except zstd.ZstdError as e:
+        raise Error(f"in-process zstd filter failed: {e}") from None
+    finally:
+        # for compression, closing also finishes the zstd frame.
+        # ZstdFile does not close a fileobj it was given, so close stream separately.
+        zstream.close()
+        if stream_close:
+            stream.close()
+
+
 class TarMixIn:
     @with_repository(compatibility=(Manifest.Operation.READ,))
     @with_archive
@@ -188,13 +230,25 @@ class TarMixIn:
         # none of these drawbacks. The only issue of using an external filter is
         # that it has to be installed -- hardly a problem, considering that
         # the decompressor must be installed as well to make use of the exported tarball!
+        #
+        # zstd is the exception: borg requires the zstd module anyway (Python >= 3.14
+        # stdlib, backports.zstd otherwise), libzstd multithreading (see the
+        # BORG_ZSTD_MT_WORKERS env var) runs outside the GIL, and there is no
+        # significantly more optimized external tool - so zstd tarballs are
+        # (de)compressed in-process and need no external program.
 
         filter = get_tar_filter(args.tarfile, decompress=False) if args.tar_filter == "auto" else args.tar_filter
 
         tarstream = dash_open(args.tarfile, "wb")
         tarstream_close = args.tarfile != "-"
 
-        with create_filter_process(filter, stream=tarstream, stream_close=tarstream_close, inbound=False) as _stream:
+        if filter is IN_PROCESS_ZSTD:
+            filter_context = create_zstd_filter(tarstream, stream_close=tarstream_close, decompress=False)
+        else:
+            filter_context = create_filter_process(
+                filter, stream=tarstream, stream_close=tarstream_close, inbound=False
+            )
+        with filter_context as _stream:
             self._export_tar(args, archive, _stream)
 
     def _export_tar(self, args, archive, tarstream):
@@ -267,7 +321,11 @@ class TarMixIn:
         tarstream = dash_open(args.tarfile, "rb")
         tarstream_close = args.tarfile != "-"
 
-        with create_filter_process(filter, stream=tarstream, stream_close=tarstream_close, inbound=True) as _stream:
+        if filter is IN_PROCESS_ZSTD:
+            filter_context = create_zstd_filter(tarstream, stream_close=tarstream_close, decompress=True)
+        else:
+            filter_context = create_filter_process(filter, stream=tarstream, stream_close=tarstream_close, inbound=True)
+        with filter_context as _stream:
             self._import_tar(args, repository, manifest, manifest.key, cache, _stream)
 
     def _import_tar(self, args, repository, manifest, key, cache, tarstream):
@@ -350,9 +408,13 @@ class TarMixIn:
         - .tar.gz or .tgz: gzip
         - .tar.bz2 or .tbz: bzip2
         - .tar.xz or .txz: xz
-        - .tar.zstd or .tar.zst: zstd
+        - .tar.zstd, .tar.zst or .tzst: zstd (in-process, level 3)
         - .tar.lz4: lz4
 
+        For zstd, Borg compresses in-process using libzstd instead of piping through an
+        external program (set BORG_ZSTD_MT_WORKERS to use multiple compression threads).
+        The other formats are piped through the respective external filter program.
+
         Alternatively, a ``--tar-filter`` program may be explicitly specified. It should
         read the uncompressed tar stream from stdin and write a compressed/filtered
         tar stream to stdout.
@@ -425,9 +487,13 @@ class TarMixIn:
         - .tar.gz or .tgz: gzip -d
         - .tar.bz2 or .tbz: bzip2 -d
         - .tar.xz or .txz: xz -d
-        - .tar.zstd or .tar.zst: zstd -d
+        - .tar.zstd, .tar.zst or .tzst: zstd (in-process)
         - .tar.lz4: lz4 -d
 
+        For zstd, Borg decompresses in-process using libzstd instead of piping through
+        an external program. The other formats are piped through the respective
+        external filter program.
+
         Alternatively, a --tar-filter program may be explicitly specified. It should
         read compressed data from stdin and output an uncompressed tar stream on
         stdout.

src/borg/testsuite/archiver/tar_cmds_test.py

@@ -26,6 +26,9 @@ def have_gnutar():
 
 requires_gnutar = pytest.mark.skipif(not have_gnutar(), reason="GNU tar must be installed for this test.")
 requires_gzip = pytest.mark.skipif(not shutil.which("gzip"), reason="gzip must be installed for this test.")
+requires_zstd = pytest.mark.skipif(not shutil.which("zstd"), reason="zstd must be installed for this test.")
+
+ZSTD_MAGIC = b"\x28\xb5\x2f\xfd"
 
 
 @requires_gnutar
@@ -159,6 +162,72 @@ def test_import_tar_gz(archivers, request, tar_format="GNU"):
     assert_dirs_equal("input", "output/input", ignore_ns=True, ignore_xattrs=True)
 
 
+@pytest.mark.parametrize("suffix", ["tar.zst", "tar.zstd", "tzst"])
+def test_export_import_tar_zst(archivers, request, suffix):
+    # zstd tarballs are (de)compressed in-process, no external zstd binary is needed.
+    archiver = request.getfixturevalue(archivers)
+    create_test_files(archiver.input_path, create_hardlinks=False)  # hard links become separate files
+    os.unlink("input/flagfile")
+    cmd(archiver, "repo-create", "--encryption=none")
+    cmd(archiver, "create", "src", "input")
+    cmd(archiver, "export-tar", "src", f"simple.{suffix}")
+    with open(f"simple.{suffix}", "rb") as fd:
+        assert fd.read(4) == ZSTD_MAGIC
+    cmd(archiver, "import-tar", "dst", f"simple.{suffix}")
+    with changedir(archiver.output_path):
+        cmd(archiver, "extract", "dst")
+    assert_dirs_equal("input", "output/input", ignore_ns=True, ignore_xattrs=True)
+
+
+def test_export_import_tar_zst_mt(archivers, request, monkeypatch):
+    # exercise the multi-threaded compression path (nb_workers) of the in-process zstd filter.
+    archiver = request.getfixturevalue(archivers)
+    monkeypatch.setenv("BORG_ZSTD_MT_WORKERS", "2")
+    monkeypatch.setattr("borg.compress._zstd_mt_workers", None)  # drop the cache
+    create_test_files(archiver.input_path, create_hardlinks=False)  # hard links become separate files
+    os.unlink("input/flagfile")
+    cmd(archiver, "repo-create", "--encryption=none")
+    cmd(archiver, "create", "src", "input")
+    cmd(archiver, "export-tar", "src", "simple.tar.zst")
+    cmd(archiver, "import-tar", "dst", "simple.tar.zst")
+    with changedir(archiver.output_path):
+        cmd(archiver, "extract", "dst")
+    assert_dirs_equal("input", "output/input", ignore_ns=True, ignore_xattrs=True)
+
+
+@requires_gnutar
+@requires_zstd
+def test_export_tar_zst_interop(archivers, request):
+    # in-process zstd output must be readable by the standard zstd tool.
+    archiver = request.getfixturevalue(archivers)
+    create_test_files(archiver.input_path)
+    os.unlink("input/flagfile")
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    cmd(archiver, "create", "test", "input")
+    cmd(archiver, "export-tar", "test", "simple.tar.zst", "--tar-format=GNU")
+    subprocess.check_call(["zstd", "-d", "simple.tar.zst", "-o", "simple.tar"])
+    with changedir("output"):
+        subprocess.check_call(["tar", "xpf", "../simple.tar", "--warning=no-timestamp"])
+    assert_dirs_equal("input", "output/input", ignore_flags=True, ignore_xattrs=True, ignore_ns=True)
+
+
+@requires_zstd
+def test_tar_filter_zstd_external(archivers, request):
+    # an explicit --tar-filter always runs the external filter program, also for zstd.
+    archiver = request.getfixturevalue(archivers)
+    create_test_files(archiver.input_path, create_hardlinks=False)  # hard links become separate files
+    os.unlink("input/flagfile")
+    cmd(archiver, "repo-create", "--encryption=none")
+    cmd(archiver, "create", "src", "input")
+    cmd(archiver, "export-tar", "src", "simple.tar.zst", "--tar-filter=zstd")
+    with open("simple.tar.zst", "rb") as fd:
+        assert fd.read(4) == ZSTD_MAGIC
+    cmd(archiver, "import-tar", "dst", "simple.tar.zst", "--tar-filter=zstd -d")
+    with changedir(archiver.output_path):
+        cmd(archiver, "extract", "dst")
+    assert_dirs_equal("input", "output/input", ignore_ns=True, ignore_xattrs=True)
+
+
 @requires_gnutar
 def test_import_concatenated_tar_with_ignore_zeros(archivers, request):
     archiver = request.getfixturevalue(archivers)