diff --git a/changelog.d/20250818_140720_jb_debug.rst b/changelog.d/20250818_140720_jb_debug.rst
new file mode 100644
index 0000000..906b2c5
--- /dev/null
+++ b/changelog.d/20250818_140720_jb_debug.rst
@@ -0,0 +1,3 @@
+.. A new scriv changelog fragment.
+
+- Add BACKY_DEBUG env (PL-133638)
diff --git a/pyproject.toml b/pyproject.toml
index 9c786d3..8b0c460 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -12,6 +12,7 @@ output_file = "CHANGES.txt"
 version = "literal: pyproject.toml: tool.poetry.version"
 entry_title_template = "{% if version %}{{ version }} {% endif %}({{ date.strftime('%Y-%m-%d') }})"
 categories = ""
+format = "rst"
 
 
 [tool.pytest.ini_options]
diff --git a/src/backy/backends/chunked/chunk.py b/src/backy/backends/chunked/chunk.py
index 78de951..e87d762 100644
--- a/src/backy/backends/chunked/chunk.py
+++ b/src/backy/backends/chunked/chunk.py
@@ -7,6 +7,7 @@
 
 import lzo
 import mmh3
+import structlog
 
 import backy.backends.chunked
 from backy.backends import BackendException
@@ -18,6 +19,9 @@
 }
 
 
+log = structlog.stdlib.get_logger(subsystem="chunk")
+
+
 class InconsistentHash(BackendException):
     def __init__(self, expected, actual):
         self.expected = expected
@@ -140,7 +144,9 @@ def flush(self):
         needs_forced_write = (
             self.store.force_writes and self.hash not in self.store.seen_forced
         )
-        if self.hash not in self.store.known or needs_forced_write:
+        write = self.hash not in self.store.known or needs_forced_write
+        log.trace("flush", id=self.id, write=write, hash=self.hash)
+        if write:
             # Create the tempfile in the right directory to increase locality
             # of our change - avoid renaming between multiple directories to
             # reduce traffic on the directory nodes.
diff --git a/src/backy/backup.py b/src/backy/backup.py
index 5548f87..97bcd7f 100644
--- a/src/backy/backup.py
+++ b/src/backy/backup.py
@@ -6,10 +6,11 @@
 import time
 from typing import IO, Optional, Type
 
+import shortuuid
 import yaml
 from structlog.stdlib import BoundLogger
 
-from backy.utils import min_date
+from backy.utils import DEBUG, min_date
 
 from .backends import BackendException, BackyBackend
 from .backends.chunked import ChunkedFileBackend
@@ -243,6 +244,19 @@ def backup(self, tags, force=False):
                     "verification-failed",
                     revision_uuid=new_revision.uuid,
                 )
+                if DEBUG:
+                    new_revision.stats["duration"] = time.time() - start
+                    new_revision.write_info()
+
+                    archive_id = shortuuid.uuid()
+                    self.log.info("archive", id=archive_id)
+                    archive = os.path.join(self.path, "archives", archive_id)
+                    os.makedirs(archive)
+                    self.scan()
+                    for r in self.history:
+                        r.archive(archive)
+                    source.archive()
+
                 new_revision.remove()
             else:
                 self.log.info(
diff --git a/src/backy/logging.py b/src/backy/logging.py
index 4c37a5a..d13a452 100644
--- a/src/backy/logging.py
+++ b/src/backy/logging.py
@@ -3,6 +3,7 @@
 # repository for complete details.
 
 import io
+import logging
 import os
 import string
 import sys
@@ -330,7 +331,12 @@ def init_logging(
     verbose: bool,
     logfile: Optional[Path] = None,
     default_job_name: str = "",
+    trace=False,
 ):
+    TRACE = 3
+
+    structlog.stdlib.LEVEL_TO_NAME[TRACE] = "trace"
+    structlog.stdlib.NAME_TO_LEVEL["trace"] = TRACE
 
     console_file_renderer = ConsoleFileRenderer(
         min_level="trace" if verbose else "info",
@@ -356,7 +362,9 @@ def init_logging(
 
     structlog.configure(
         processors=processors,
-        wrapper_class=structlog.stdlib.BoundLogger,
+        wrapper_class=structlog._native._make_filtering_bound_logger(
+            TRACE if trace else logging.DEBUG
+        ),
         logger_factory=MultiOptimisticLoggerFactory(**loggers),
         cache_logger_on_first_use=False,
     )
diff --git a/src/backy/main.py b/src/backy/main.py
index 6eb9d66..39e2291 100644
--- a/src/backy/main.py
+++ b/src/backy/main.py
@@ -15,7 +15,7 @@
 
 import backy.backup
 import backy.daemon
-from backy.utils import format_datetime_local
+from backy.utils import DEBUG, format_datetime_local
 
 from . import logging
 
@@ -334,9 +334,11 @@ def main():
         args.verbose,
         args.logfile or default_logfile,
         default_job_name="-" if is_daemon else "",
+        trace=DEBUG,
     )
     log = structlog.stdlib.get_logger(subsystem="command")
     log.debug("invoked", args=" ".join(sys.argv))
+    log.trace("tracing-enabled")
 
     command = Command(args.backupdir, log)
     func = getattr(command, args.func)
diff --git a/src/backy/revision.py b/src/backy/revision.py
index 97ed1a9..471d5bc 100644
--- a/src/backy/revision.py
+++ b/src/backy/revision.py
@@ -1,7 +1,7 @@
 import datetime
 import glob
 import os
-import os.path as p
+import shutil
 from enum import Enum
 from typing import Optional
 
@@ -130,6 +130,12 @@ def remove(self):
         if self in self.backup.history:
             self.backup.history.remove(self)
 
+    def archive(self, path):
+        self.log.info("archive")
+        for filename in glob.glob(self.filename + "*"):
+            if os.path.exists(filename):
+                shutil.copy(filename, path)
+
     def writable(self):
         if os.path.exists(self.filename):
             os.chmod(self.filename, 0o640)
diff --git a/src/backy/sources/__init__.py b/src/backy/sources/__init__.py
index faa9948..8bef735 100644
--- a/src/backy/sources/__init__.py
+++ b/src/backy/sources/__init__.py
@@ -14,6 +14,9 @@ def backup(self, target: "backy.backends.BackyBackend") -> None: ...
     @abstractmethod
     def verify(self, target: "backy.backends.BackyBackend") -> bool: ...
 
+    def archive(self) -> None:
+        pass
+
 
 class BackySourceContext(ABC):
     @abstractmethod
diff --git a/src/backy/sources/ceph/diff.py b/src/backy/sources/ceph/diff.py
index 453b904..b6cc94b 100644
--- a/src/backy/sources/ceph/diff.py
+++ b/src/backy/sources/ceph/diff.py
@@ -2,6 +2,8 @@
 from collections import namedtuple
 from typing import IO, Optional
 
+import structlog
+
 from backy.fallocate import punch_hole
 
 
@@ -17,6 +19,8 @@ def unpack_from(fmt, f):
 FromSnap = namedtuple("FromSnap", ["snapshot"])
 ToSnap = namedtuple("ToSnap", ["snapshot"])
 
+log = structlog.stdlib.get_logger(subsystem="rbddiff")
+
 
 class RBDDiffV1(object):
     f: IO
@@ -137,19 +141,28 @@ def integrate(self, target, snapshot_from, snapshot_to, clean=True):
 
         for record in self.read_metadata():
             if isinstance(record, SnapSize):
+                log.trace("new-size", size=record.size)
                 target.truncate(record.size)
             elif isinstance(record, FromSnap):
+                log.trace("new-from", snapshot=record.snapshot)
                 assert record.snapshot == snapshot_from
             elif isinstance(record, ToSnap):
+                log.trace("new-to", snapshot=record.snapshot)
                 assert record.snapshot == snapshot_to
+            else:
+                assert False
 
         for record in self.read_data():
             target.seek(record.start)
             if isinstance(record, Zero):
+                log.trace("new-zero", start=record.start, length=record.length)
                 punch_hole(target, target.tell(), record.length)
             elif isinstance(record, Data):
+                log.trace("new-data", start=record.start, length=record.length)
                 for chunk in record.stream():
                     target.write(chunk)
+            else:
+                assert False
             bytes += record.length
 
         self.f.close()
diff --git a/src/backy/sources/ceph/rbd.py b/src/backy/sources/ceph/rbd.py
index 7bd74a0..8bfa9bf 100644
--- a/src/backy/sources/ceph/rbd.py
+++ b/src/backy/sources/ceph/rbd.py
@@ -135,6 +135,9 @@ def snap_ls(self, image):
     def snap_rm(self, image):
         return self._rbd(["snap", "rm", image])
 
+    def snap_protect(self, image):
+        return self._rbd(["snap", "protect", image])
+
     @contextlib.contextmanager
     def export_diff(self, new: str, old: str) -> Iterator[RBDDiffV1]:
         if self._supports_whole_object:
diff --git a/src/backy/sources/ceph/source.py b/src/backy/sources/ceph/source.py
index cd4c3d1..0acde4a 100644
--- a/src/backy/sources/ceph/source.py
+++ b/src/backy/sources/ceph/source.py
@@ -65,6 +65,21 @@ def _image_name(self):
     def __exit__(self, exc_type=None, exc_val=None, exc_tb=None):
         self._delete_old_snapshots()
 
+    def archive(self):
+        # protect all snapshots
+        for snapshot in self.rbd.snap_ls(self._image_name):
+            if not snapshot["name"].startswith("backy-"):
+                # Do not touch non-backy snapshots
+                continue
+            self.log.info("snapshot-protect", snapshot_name=snapshot["name"])
+            try:
+                self.rbd.snap_protect(self._image_name + "@" + snapshot["name"])
+            except Exception:
+                self.log.exception(
+                    "snapshot-protect-failed",
+                    snapshot_name=snapshot["name"],
+                )
+
     def backup(self, target):
         if self.always_full:
             self.log.info("backup-always-full")
@@ -143,7 +158,12 @@ def verify(self, target) -> bool:
 
         with s as source, t as target:
             self.log.info("verify")
-            return backy.utils.files_are_roughly_equal(
+            if backy.utils.DEBUG:
+                f = backy.utils.files_are_equal
+            else:
+                f = backy.utils.files_are_roughly_equal  # type: ignore
+
+            return f(
                 source,
                 target,
                 report=lambda s, t, o: self.revision.backup.quarantine.add_report(
@@ -167,6 +187,12 @@ def _delete_old_snapshots(self):
             if not snapshot["name"].startswith("backy-"):
                 # Do not touch non-backy snapshots
                 continue
+            if snapshot["protected"] == "true":
+                self.log.info(
+                    "delete-old-snapshot-protected",
+                    snapshot_name=snapshot["name"],
+                )
+                continue
             uuid = snapshot["name"].replace("backy-", "")
             if uuid != keep_snapshot_revision:
                 time.sleep(3)  # avoid race condition while unmapping
diff --git a/src/backy/utils.py b/src/backy/utils.py
index 6ff1b03..6ac4a77 100644
--- a/src/backy/utils.py
+++ b/src/backy/utils.py
@@ -36,6 +36,8 @@
 
 END = object()
 
+DEBUG = os.environ.get("BACKY_DEBUG") == "1"
+
 
 def report_status(f):
     def wrapped(*args, **kw):