archive (#19): fill the snapshot gaps the first CI-artifact ingest found

docs/archive.md

@@ -67,19 +67,26 @@ Only files that proven routes and boxes reference. Never a mirror of a whole dis
    index entry names it (`"source": {"url", "digest"}`):
    - **Debian, Ubuntu:** the `.dsc` and every file it lists, from the `<comp>/source/Sources` index the pinned
      `Release` lists, so the chain runs from the run's pin as the binaries' does. The binary's `Source:` field names
-     the source and its version.
+     the source and its version. A suite whose run pins only its package index (wheezy) has its `Release` read live,
+     and used only when it lists that pinned index at the pinned digest; it is recorded `"live": true`. A source
+     `Sources` names only by md5 (lenny and squeeze still list pre-2008 uploads that way) is stored by its md5: good
+     enough for a licence copy nobody installs, and accepted nowhere else.
    - **EL:** the `.src.rpm` its `primary.xml` names (`<rpm:sourcerpm>`), or for a package a run pins directly (old
      EPEL 5 builds no repo serves) the one its own RPM header names, from the repo's source repo, checked against
      that repo's own metadata. Source repos aren't pinned by any run, so their `repomd.xml` is taken live and recorded
      in the index, marked `"live": true`; the index signature (step 8) is what vouches for it. Where each repo's sources live is a table in
      `tools/archive.py` (`SOURCE_REPOS`): `vault.centos.org/<v>/<repo>/Source/` (`SRPMS/` for CentOS 5), AlmaLinux's
      current sources under `vault.almalinux.org/<major>/<repo>/Source/`, EPEL's `source/tree/`.
 
-   When upstream no longer has a source (a rotted vault, an EPEL build since replaced, etch's md5-only `Sources`,
-   wheezy whose `Release` is gone), the index records `"source": {"missing": why}` rather than inventing one, and
+   When upstream no longer has a source (a rotted vault, an EPEL build since replaced), the index records `"source": {"missing": why}` rather than inventing one, and
    `ingest` lists the gaps. They don't fail an ingest. They are what has to be filled (from snapshot.debian.org,
    Koji's source RPMs, or a minor-release vault) before that release goes public.
 
+   One gap is known and stays: runs recorded before the capture shim (#105) pin 28 AlmaLinux 8-10 and EPEL 8-10
+   `repomd.xml` digests (under 9 URLs; `ingest` lists gaps by URL) that upstream replaced before anything kept them,
+   and no CI artifact or copy on remi holds them. Their packages are stored, but `wend walk --run` can't replay those
+   runs from the archive. The next recording of each EL route is captured whole and closes it for that route.
+
 "Referenced" means: by a run file on main whose `result` is `pass`, or by a box file on main. The archive never
 evicts a blob that such a file references. No eviction at all in the first version: storage is cheap and the set is
 small. A blob a later run no longer references can be swept once there are numbers saying it's worth the code.

tools/archive.py

@@ -77,6 +77,11 @@ def digest(alg, hexdigest):
     return None
 
 
+def md5(hexdigest):
+    """("md5", hex) for a source file a Sources index names only by md5: see sources_entries."""
+    return ("md5", hexdigest.lower()) if re.fullmatch("[0-9a-fA-F]{32}", hexdigest) else None
+
+
 def plain_index(url):
     """An index pinned uncompressed (the client's `plain` rule): upstream may serve only its .gz."""
     return "/dists/" in url and not url.endswith(("Release", ".gz", ".xz", ".bz2", ".zst") + UNREADABLE)
@@ -246,8 +251,9 @@ def source_names(url, data):
 
 
 def sources_entries(text):
-    """A Sources index: {(package, version): [(directory/name, digest)]}, by Checksums-Sha256, else Checksums-Sha1.
-    A stanza with only md5 sums (etch's) vouches for nothing."""
+    """A Sources index: {(package, version): [(directory/name, digest)]}, by Checksums-Sha256, else Checksums-Sha1,
+    else Files' md5 (lenny and squeeze still list sources uploaded before dpkg wrote sha sums). md5 is good enough
+    here and nowhere else: these are the distro's own 2000s uploads, kept for their licences and never installed."""
     out = {}
     for stanza in text.split("\n\n"):
         f, cur = {}, None
@@ -261,9 +267,10 @@ def sources_entries(text):
                 cur = k
                 f[k] = v.strip() if v.strip() else []
         files = []
-        for field, alg in (("Checksums-Sha256", "sha256"), ("Checksums-Sha1", "sha1")):
+        for field, alg in (("Checksums-Sha256", "sha256"), ("Checksums-Sha1", "sha1"), ("Files", "md5")):
             if isinstance(f.get(field), list):
-                files = [(f"{f.get('Directory', '')}/{x[2]}", digest(alg, x[0])) for x in f[field] if len(x) == 3]
+                files = [(f"{f.get('Directory', '')}/{x[2]}", md5(x[0]) if alg == "md5" else digest(alg, x[0]))
+                         for x in f[field] if len(x) == 3]
                 break
         if files and all(d for _, d in files) and isinstance(f.get("Package"), str):
             out[(f["Package"], f.get("Version"))] = files
@@ -274,9 +281,13 @@ def stem(u):
     return re.sub(r"\.(gz|xz|bz2|lzma|zst)$", "", u)
 
 
-def bounded_gunzip(f):
-    """A .gz's contents, refusing past MAX_PLAIN: these bytes aren't checked until they're whole."""
-    data = gzip.GzipFile(fileobj=f).read(MAX_PLAIN + 1)
+PLAIN_FORMS = (".gz", ".bz2", ".xz")  # jessie..buster's Translation-en: snapshot serves only the .bz2
+
+
+def bounded_unpack(f, name):
+    """A compressed index's contents, refusing past MAX_PLAIN: these bytes aren't checked until they're whole."""
+    opener = {".gz": gzip.open, ".bz2": bz2.open, ".xz": lzma.open}[name[name.rindex("."):]]
+    data = opener(f).read(MAX_PLAIN + 1)
     if len(data) > MAX_PLAIN:
         raise ValueError("decompresses past the limit")
     return data
@@ -312,6 +323,8 @@ def __init__(self, path, fetched=(), blobs=(), trees=(), offline=False):
         self.built_from = {}  # (package url, digest) -> source_names() value, for every package stored
         self.deb_sources = {}  # "<.../dists/X>/<comp>" -> [(Sources url, digest, the Release that lists it)]
         self.no_source = {}  # (package url, digest) -> why no source was stored
+        self.deb_indexes = {}  # "<.../dists/X>" -> {(url, digest)} of the package indexes pinned under it
+        self.live_releases = {}  # "<.../dists/X>" -> [(url, "sha256:<hex>")]: a live Release read for its Sources
 
     def blob(self, hexdigest):
         return f"{self.path}/blob/sha256/{hexdigest[:2]}/{hexdigest}"
@@ -335,15 +348,15 @@ def candidates(self, url, digest):
                 for d in self.blob_dirs:
                     yield f"{d}/{h}", None
         host_path = re.sub(r"^[a-z]+://", "", url)
-        plain = plain_index(url)  # wheezy: the pin is the plain index, the archive serves only its .gz
+        forms = PLAIN_FORMS if plain_index(url) else ()  # the pin is the plain index, upstream serves it compressed
         for t in self.trees:
             yield f"{t}/{host_path}", None
-            if plain:
-                yield f"{t}/{host_path}.gz", None
+            for ext in forms:
+                yield f"{t}/{host_path}{ext}", None
         if not self.offline:
             yield None, url
-            if plain:
-                yield None, url + ".gz"
+            for ext in forms:
+                yield None, url + ext
 
     def get(self, url, digest):
         """(the verified bytes of url at digest, or None; why the origin didn't have them)."""
@@ -355,10 +368,10 @@ def get(self, url, digest):
                     if not os.path.isfile(path):
                         continue
                     with open(path, "rb") as f:
-                        data = bounded_gunzip(f) if plain_index(url) and path.endswith(".gz") else f.read()
+                        data = bounded_unpack(f, path) if plain_index(url) and path.endswith(PLAIN_FORMS) else f.read()
                 else:
                     with urllib.request.urlopen(src, timeout=300) as r:
-                        data = r.read() if src == url else bounded_gunzip(r)
+                        data = r.read() if src == url else bounded_unpack(r, src)
             except Exception as e:  # urllib's HTTPError, a timeout, a bad .gz: this source hasn't got it
                 if src:
                     why = f"{src}: {e}"
@@ -428,6 +441,8 @@ def ingest(self, pins):
                 names = source_names(url, data) if PACKAGE_INDEX.search(url) else {}
             except Exception:  # its packages still go in; their sources show up as gaps
                 names = {}
+            if "/dists/" in url and "/binary-" in url and PACKAGE_INDEX.search(url):
+                self.deb_indexes.setdefault(url[:url.index("/", url.index("/dists/") + 7)], set()).add((url, d))
             if url.endswith(("/Release", "/InRelease")):
                 for child, cd in listed:
                     m = re.match(r"(.*/dists/[^/]+)/(.+)/source/Sources(\.[a-z0-9]+)?$", child)
@@ -496,7 +511,7 @@ def sources(self):
                     groups.setdefault(base, {})[key] = what
                 else:
                     self.no_source[key] = f"no source repo known for {what[2]}"
-            elif any(c.rsplit("/", 1)[0] == what[4].rsplit("/", 1)[0] for c in self.deb_sources):
+            elif self.has_sources(what[4].rsplit("/", 1)[0]):
                 groups.setdefault(("deb", what[3], what[4].rsplit("/", 1)[0]), {})[key] = what
             else:
                 self.no_source[key] = "its suite's Release lists no Sources"
@@ -508,6 +523,7 @@ def sources(self):
                 lookup = lambda what: found.get(what[1])
             else:
                 lookup, why = self.deb_lookup(group[1], group[2], packages.values(), meta), ""
+                meta += self.live_releases.get(group[2], [])
             runs_of_group = set()
             for key, what in packages.items():
                 files = lookup(what)
@@ -538,6 +554,28 @@ def sources(self):
         for key, why in self.no_source.items():
             self.index["files"][key[0]][self.kept[key]].setdefault("source", {"missing": why})
 
+    def has_sources(self, suite):
+        """Whether a Release lists Sources for suite. A suite whose run pinned only a package index (wheezy) gets its
+        live Release, used only when it lists that pinned index at the pinned digest: then it is the Release the
+        run's index came from, and its Sources entries vouch as a pinned Release's would."""
+        if not any(c.rsplit("/", 1)[0] == suite for c in self.deb_sources) and suite not in self.live_releases:
+            self.live_releases[suite] = []
+            url = f"{suite}/Release"
+            data = self.live(url)
+            try:
+                listed = children(url, data) if data is not None else []
+            except ValueError:
+                listed = []
+            if self.deb_indexes.get(suite, set()) & set(listed):
+                h = self.keep(url, data)
+                self.index["files"][url][f"sha256:{h}"]["live"] = True
+                self.live_releases[suite].append((url, f"sha256:{h}"))
+                for child, cd in listed:
+                    m = re.match(r"(.*/dists/[^/]+)/(.+)/source/Sources(\.[a-z0-9]+)?$", child)
+                    if m:
+                        self.deb_sources.setdefault(f"{m[1]}/{m[2]}", []).append((child, cd, (url, ("sha256", h))))
+        return any(c.rsplit("/", 1)[0] == suite for c in self.deb_sources)
+
     def rpm_sources(self, base, meta):
         """({"<name>.src.rpm": [(url, digest)]}, why it's empty) from a source repo's live metadata. Its repomd.xml
         is pinned by no run: its index entry says "live", and the archive's signature is what vouches for it."""

tools/test_archive.py

@@ -1,6 +1,7 @@
 """The archive's store by behaviour (docs/archive.md): offline, on a made-up repo. Only bytes that hash right
 against a passing run's pins go in, only what a download list names comes along, and verify catches a bad blob.
 python3 tools/test_archive.py"""
+import bz2
 import gzip
 import hashlib
 import json
@@ -126,6 +127,9 @@ def put(rel, data):
 plain = b"Package: x\nFilename: pool/x.deb\nSHA256: " + sha(b"x").encode() + b"\n"
 wheezy = "archive.example.org/debian/dists/wheezy/main/binary-amd64/Packages"
 put(f"{wheezy}.gz", gzip.compress(plain))
+tr = b"Package: x\nDescription-md5: 0\nDescription-en: x\n"
+buster = "archive.example.org/debian/dists/buster/main/i18n/Translation-en"
+put(f"{buster}.bz2", bz2.compress(tr))
 zck = b"(zchunk bytes)"
 repomd3 = (f'<repomd><data type="primary"><checksum type="sha256">{sha(zck)}</checksum>'
            f'<location href="repodata/primary.xml.zck"/></data></repomd>').encode()
@@ -134,11 +138,14 @@ def put(rel, data):
 os.makedirs(f"{w}/runs2")
 json.dump({"result": "pass", "box": "x", "started": "4", "hops": [{"sources": {
     f"http://{wheezy}": f"sha256:{sha(plain)}",
+    f"http://{buster}": f"sha256:{sha(tr)}",
     "https://new.example.org/10/repodata/repomd.xml": f"sha256:{sha(repomd3)}"}}]}, open(f"{w}/runs2/r.json", "w"))
 p = subprocess.run([sys.executable, "-B", f"{HERE}/archive.py", "ingest", f"{w}/store2", "--offline", "--tree",
                     f"{w}/tree", "--runs", f"{w}/runs2"], capture_output=True, text=True)
 check(open(f"{w}/store2/blob/sha256/{sha(plain)[:2]}/{sha(plain)}", "rb").read() == plain,
       "a plain index pinned uncompressed is found through the .gz a tree holds")
+check(os.path.isfile(f"{w}/store2/blob/sha256/{sha(tr)[:2]}/{sha(tr)}"),
+      "a plain index served only as .bz2 (buster's Translation-en) is found through it")
 check(p.returncode == 1 and "unreadable https://new.example.org/10/repodata/primary.xml.zck" in p.stdout,
       "an index in a compression nothing reads is reported, not silently skipped")
 # sources: an EL repo's .src.rpm from its source repo (SOURCE_REPOS), a Debian package's .dsc and files from the
@@ -213,6 +220,41 @@ def put(rel, data):
 check(f"no source https://{deb}/pool/main/x/xsrc/x_1.0-1+b1_amd64.deb" in p.stdout and
       not os.path.exists(f"{w}/store4/blob/sha256/{sha(b'evil')[:2]}/{sha(b'evil')}"),
       "a source file that doesn't hash to what Sources lists is refused")
+# wheezy: a run pins only a package index, no Release. The suite's live Release is read for its Sources only when it
+# lists that pinned index at the pinned digest; and a lenny-era source listed only by md5 is stored by it
+md5 = lambda b: hashlib.md5(b).hexdigest()
+obin, odsc, ebin = b"!<arch> o", b"Format: 1.0\n", b"!<arch> e"
+opk = f"Package: o\nVersion: 1\nFilename: pool/main/o/o/o_1_amd64.deb\nSHA256: {sha(obin)}\n".encode()
+epk = f"Package: e\nVersion: 1\nFilename: pool/main/e/e/e_1_amd64.deb\nSHA256: {sha(ebin)}\n".encode()
+osrc = gzip.compress(f"Package: o\nVersion: 1\nDirectory: pool/main/o/o\nFiles:\n {md5(odsc)} {len(odsc)} o_1.dsc\n".encode())
+put(f"{deb}/dists/old/main/binary-amd64/Packages", opk)
+put(f"{deb}/dists/old/main/source/Sources.gz", osrc)
+put(f"{deb}/dists/old/Release", (f"SHA256:\n {sha(opk)} {len(opk)} main/binary-amd64/Packages\n"
+                                 f" {sha(osrc)} {len(osrc)} main/source/Sources.gz\n").encode())
+put(f"{deb}/pool/main/o/o/o_1_amd64.deb", obin)
+put(f"{deb}/pool/main/o/o/o_1.dsc", odsc)
+put(f"{deb}/dists/else/main/binary-amd64/Packages", epk)
+put(f"{deb}/dists/else/main/source/Sources.gz", osrc)
+put(f"{deb}/dists/else/Release", (f"SHA256:\n {sha(b'newer')} 5 main/binary-amd64/Packages\n"
+                                  f" {sha(osrc)} {len(osrc)} main/source/Sources.gz\n").encode())
+put(f"{deb}/pool/main/e/e/e_1_amd64.deb", ebin)
+os.makedirs(f"{w}/runsw")
+os.makedirs(f"{w}/fetchedw/r.fetched")
+open(f"{w}/fetchedw/r.fetched/fetched.txt", "w").write(f"https://{deb}/pool/main/o/o/o_1_amd64.deb sha256:{sha(obin)} 1\n"
+                                                       f"https://{deb}/pool/main/e/e/e_1_amd64.deb sha256:{sha(ebin)} 1\n")
+json.dump({"result": "pass", "box": "w", "started": "6", "hops": [{"sources": {
+    f"https://{deb}/dists/old/main/binary-amd64/Packages": f"sha256:{sha(opk)}",
+    f"https://{deb}/dists/else/main/binary-amd64/Packages": f"sha256:{sha(epk)}"}}]}, open(f"{w}/runsw/r.json", "w"))
+p = subprocess.run([sys.executable, "-B", f"{HERE}/archive.py", "ingest", f"{w}/storew", "--offline", "--tree",
+                    f"{w}/tree", "--runs", f"{w}/runsw", "--fetched", f"{w}/fetchedw"], capture_output=True, text=True)
+idx = json.load(open(f"{w}/storew/index/latest.json"))["files"]
+check(idx[f"https://{deb}/pool/main/o/o/o_1_amd64.deb"][f"sha256:{sha(obin)}"]["source"] ==
+      {"url": f"https://{deb}/pool/main/o/o/o_1.dsc", "digest": f"sha256:{sha(odsc)}"} and
+      [(e["live"], e["runs"]) for e in idx[f"https://{deb}/dists/old/Release"].values()] == [(True, ["w@6"])],
+      "a suite with only its package index pinned gets its sources through the live Release that lists that index")
+check(f"no source https://{deb}/pool/main/e/e/e_1_amd64.deb (its suite's Release lists no Sources)" in p.stdout and
+      f"https://{deb}/dists/else/Release" not in idx,
+      "a live Release that doesn't list the pinned index at its digest is not used, nor stored")
 # a point release: two runs pin the same Release url at two digests, each listing its own Packages and Sources; a
 # universe binary built from a main source; a package a run pins directly
 def put5(rel, data):