Every nixpkgs revision the Nix flake ecosystem pins, joined against every published channel bump

nixpkgs-pins.py

#!/usr/bin/env nix
#! nix shell nixpkgs#python3 --command python3
"""Every nixpkgs revision the flake ecosystem pins, and which channel it was.

Reads `pins.jsonl` from a checkout of https://github.com/fzakaria/omniflake --
one row per indexed flake, carrying the nixpkgs its lock file names -- joins it
against every channel bump the nix release archive ever published, and writes
nixpkgs-pins.csv.

    git clone https://github.com/fzakaria/omniflake
    cd omniflake && ./nixpkgs-pins.py

Companion to https://fzakaria.com/2026/08/31/how-safe-is-follows
"""
import collections
import csv
import datetime
import json
import re
import urllib.parse
import urllib.request

BUCKET = "https://nix-releases.s3.amazonaws.com/"
PREFIX_RE = re.compile(r"<Prefix>([^<]+)</Prefix>")
TRUNC_RE = re.compile(r"<IsTruncated>(true|false)</IsTruncated>")
# A release directory is named for the commit it pointed at. The short rev is
# 7 to 12 characters depending on the era, which is the trap: key on twelve
# and the entire early half of the archive silently stops matching.
REV_RE = re.compile(r"\.([0-9a-f]{7,40})/?$")


def children(prefix):
    """Every common prefix under `prefix`, following pagination."""
    out, marker = [], None
    while True:
        query = {"delimiter": "/", "prefix": prefix}
        if marker:
            query["marker"] = marker
        with urllib.request.urlopen(BUCKET + "?" + urllib.parse.urlencode(query)) as r:
            body = r.read().decode()
        kids = [p for p in PREFIX_RE.findall(body) if p != prefix]
        out.extend(kids)
        truncated = TRUNC_RE.search(body)
        if not (truncated and truncated.group(1) == "true") or not kids:
            return out
        marker = kids[-1]


def channel_map():
    """{short rev -> {channel}} for every bump the archive ever published."""
    out = collections.defaultdict(set)

    # nixos/13.10, nixos/unstable, nixpkgs/25.05-darwin: a directory of
    # releases each.
    for root in ("nixos/", "nixpkgs/"):
        for channel in children(root):
            name = channel.strip("/").split("/", 1)[1]
            for release in children(channel):
                found = REV_RE.search(release.rstrip("/"))
                if found:
                    out[found.group(1)].add(name)

    # nixpkgs-unstable is the exception. Its releases sit flat under nixpkgs/
    # as siblings of the darwin channels rather than in a directory of their
    # own, so the loop above walks straight past the second most pinned
    # channel in the ecosystem and files each release as an empty channel.
    for release in children("nixpkgs/"):
        found = REV_RE.search(release.rstrip("/"))
        if found:
            out[found.group(1)].add("nixpkgs-unstable")

    return out


CHANNELS = channel_map()
LENGTHS = sorted({len(rev) for rev in CHANNELS}, reverse=True)


def channels_for(rev):
    """Probe from the longest prefix down, since short revs vary in length."""
    found = set()
    for n in LENGTHS:
        found |= CHANNELS.get(rev[:n], set())
    return found


def family(names):
    if not names:
        return "never a channel"
    if "unstable" in names:
        return "nixos-unstable"
    if "nixpkgs-unstable" in names:
        return "nixpkgs-unstable"
    if any(n.endswith("-darwin") for n in names):
        return "darwin release"
    if any(n.endswith("-small") for n in names):
        return "small release"
    return "nixos release"


# nim-nix-pkgs is 310 generated repositories, one per Nim package, 305 of them
# behind a single pin -- one decision wearing three hundred hats, and 2.5% of
# the index. Left in, it puts a 314-flake tower on a 2022 master commit that
# nothing else explains. Dropping it moves the age distribution by six days at
# the median and removes the tower entirely.
EXCLUDE = {"nim-nix-pkgs"}

revisions = collections.defaultdict(lambda: {"flakes": 0, "when": 0})
for line in open("pins.jsonl"):
    row = json.loads(line)
    if row["locked"].get("owner") in EXCLUDE:
        continue
    nixpkgs = row.get("lock_nixpkgs") or {}
    if not nixpkgs.get("rev") or not nixpkgs.get("lastModified"):
        continue
    entry = revisions[nixpkgs["rev"]]
    entry["flakes"] += 1
    entry["when"] = nixpkgs["lastModified"]

with open("nixpkgs-pins.csv", "w", newline="") as fh:
    out = csv.writer(fh)
    out.writerow(["rev", "locked", "flakes", "channel"])
    for rev, entry in sorted(revisions.items(), key=lambda kv: kv[1]["when"]):
        out.writerow([
            rev[:12],
            datetime.date.fromtimestamp(entry["when"]).isoformat(),
            entry["flakes"],
            family(channels_for(rev)),
        ])

print(f"{len(revisions)} revisions -> nixpkgs-pins.csv")