refactor: make compression atomic-or-skip, delete the corrupt-gz reconciliation

_gzip_file now verifies the tmp .gz decompresses BEFORE it os.replaces onto the final path
and removes the plain, so a crash/OOM/failed-verify at any point leaves the plain .log intact
and no .gz - a corrupt/partial .gz is never produced. with that guarantee, the plain-vs-corrupt-gz
reconciliation is dead code: retier's dedupe now drops a plain twin unconditionally (the .gz is
always intact), and both _gz_intact/corrupt-twin guards (the beyond-keep skip added in 4ad066c and
the compression-tier fall-through) are removed. this eliminates the whole 'intact plain vs corrupt
gz twin' class - including the newly-found compression-tier data-loss path - at the source. verified:
killed-mid-gzip leaves the .log intact with no .gz; a successful roll leaves only a decompressible
.gz; no path deletes an intact plain; normal retention still prunes.

Signed-off-by: disqualifier <dev@disqualifier.me>

src/log_setup/rotation.py

@@ -5,12 +5,11 @@ the namer/rotator so rolled files land in `log_dir` and are gzipped when asked,
 the live file at its stable path, and handle the on-start and prune paths the handlers
 don't manage themselves.
 
-FOOTGUN: gzip writes are crash-safe - `_gzip_file` writes to a `.tmp` sibling and
-atomically `os.replace`s it onto the final `.gz` path, so a crash/OOM/power-loss
-mid-write never leaves a truncated `.gz` where retention would trust it. `retier`'s
-plain/gz dedupe additionally verifies a `.gz` decompresses cleanly (`_gz_intact`)
-before removing its plain twin, so a corrupt `.gz` is never preferred over an intact
-plain copy.
+FOOTGUN: compression is atomic - `_gzip_file` writes to a `.tmp` sibling, verifies it
+decompresses, and only then `os.replace`s it onto the final `.gz` and removes the plain.
+a crash/OOM/power-loss or failed verify at any point leaves the plain `.log` intact and
+no `.gz`, so a corrupt/partial `.gz` is never produced and retention never has to choose
+between an intact plain and a corrupt archive.
 """
 
 import gzip
@@ -52,12 +51,13 @@ def _free_dest(dest: str) -> str:
 
 
 def _gzip_file(source: str, dest: str) -> None:
-    """gzip source into dest then remove source (the rolled-file compression idiom)
+    """atomically gzip source into dest, then remove source - or leave source untouched
 
-    FOOTGUN (crash-safe): writes to `dest + ".tmp"` and atomically `os.replace`s it
-    onto `dest` once complete, so a crash/OOM/power-loss mid-write never leaves a
-    truncated `.gz` at `dest` - the partial write stays quarantined in `.tmp` and
-    source is untouched (safe to retry).
+    compression is all-or-nothing: writes to `dest + ".tmp"`, verifies that tmp
+    decompresses cleanly, and only then `os.replace`s it onto `dest` and removes source.
+    a crash/OOM/power-loss or a failed verify at ANY point removes the tmp and raises with
+    source intact and no `.gz` at `dest` - so a corrupt/partial `.gz` is never produced and
+    nothing downstream ever has to choose between an intact plain and a corrupt archive.
 
     the source mtime is carried onto dest so a file keeps its tier position when it
     crosses the plain->gz boundary - a fresh write would otherwise make a
@@ -68,6 +68,8 @@ def _gzip_file(source: str, dest: str) -> None:
     try:
         with open(source, "rb") as src, gzip.open(tmp_dest, "wb") as dst:
             shutil.copyfileobj(src, dst)
+        if not _gz_intact(tmp_dest):
+            raise OSError(f"gzip of {source!r} did not verify")
     except BaseException:
         try:
             os.remove(tmp_dest)
@@ -85,9 +87,8 @@ def _gzip_file(source: str, dest: str) -> None:
 def _gz_intact(path: str) -> bool:
     """return True if the gzip file at path decompresses cleanly end to end
 
-    belt-and-suspenders check before a dedupe site removes a plain twin in favor of its
-    .gz - a corrupt .gz must never be trusted over an intact plain copy. reads the
-    whole stream; any failure means "not intact" so the caller keeps the plain source.
+    used by `_gzip_file` to verify a freshly-written `.gz` before it replaces the plain
+    source; reads the whole stream, any failure means "not intact".
     """
     try:
         with gzip.open(path, "rb") as handle:
@@ -268,34 +269,27 @@ def retier(log_dir: str, stem: str, keep_uncompressed: int, keep_compressed: int
     except OSError:
         return
     entries = [os.path.join(log_dir, name) for name in names]
-    # dedupe plain/gz twins FIRST (a crash between _gzip_file's write and its os.remove
-    # can leave <x>.log beside <x>.log.gz), but ONLY once the .gz is verified intact -
-    # a pre-existing corrupt .gz must never win over an intact plain copy
+    # dedupe plain/gz twins FIRST: a crash between _gzip_file's os.replace and its
+    # os.remove(source) can leave <x>.log beside <x>.log.gz. the .gz is guaranteed intact
+    # (_gzip_file verifies before it replaces, and never produces a partial .gz), so the
+    # plain twin is redundant and dropped unconditionally.
     present = set(entries)
     kept = []
     for p in entries:
         if not p.endswith(".gz") and (p + ".gz") in present:
-            if _gz_intact(p + ".gz"):
             try:
                 os.remove(p)
-                except OSError:
-                    kept.append(p)  # couldn't remove - keep it in the accounting
                 continue
-            # .gz twin is corrupt - keep the intact plain, a later retier retries
+            except OSError:
+                pass  # couldn't remove - keep it in the accounting
         kept.append(p)
     files = [(p, _safe_mtime(p), _roll_counter(p)) for p in kept if os.path.isfile(p)]
     # newest-first: higher mtime first, ties broken by higher (later) roll counter
     files.sort(key=lambda t: (t[1], t[2]), reverse=True)
 
     keep = keep_uncompressed + keep_compressed
-    present = {p for p, _, _ in files}
     for index, (path, _, _) in enumerate(files):
         if index >= keep:
-            # never delete an intact plain whose only surviving twin is a CORRUPT .gz -
-            # that would leave the corrupt archive as the sole copy (data loss). keep the
-            # plain; a later retier retires it once a clean .gz exists.
-            if not path.endswith(".gz") and (path + ".gz") in present and not _gz_intact(path + ".gz"):
-                continue
             try:
                 os.remove(path)
             except OSError:
@@ -303,14 +297,13 @@ def retier(log_dir: str, stem: str, keep_uncompressed: int, keep_compressed: int
         elif index >= keep_uncompressed and not path.endswith(".gz"):
             dest = path + ".gz"
             if os.path.exists(dest):
-                # drop the redundant plain twin, but ONLY once the .gz is intact
-                if _gz_intact(dest):
+                # a verified .gz twin already exists (a prior crashed roll) - drop the
+                # redundant plain rather than re-gzip over it
                 try:
                     os.remove(path)
                 except OSError:
                     pass
                 continue
-                # .gz is corrupt - fall through and re-gzip the plain over the bad dest
             try:
                 _gzip_file(path, dest)
             except OSError: