refactor: simplify fetching channels

docs/data_ingestion_and_matching.md

@@ -14,19 +14,19 @@ This document shows how to fetch channels, run evaluations, ingest CVEs and prod
 manage fetch_all_channels
 ```
 
-When running this command for first time, this would take a couple of minutes.
-In the meantime you can watch some cat videos.
 The output will look like this:
 
 ```console
-{'channel_branch': 'nixos-25.05',
- 'head_sha1_commit': 'ac62194c3917d5f474c1a844b6fd6da2db95077d',
- 'release_branch': 'release-25.05',
- 'state': NixChannel.ChannelState.END_OF_LIFE}
-{'channel_branch': 'nixos-25.05-small',
- 'head_sha1_commit': 'ac62194c3917d5f474c1a844b6fd6da2db95077d',
- 'release_branch': 'release-25.05',
- 'state': NixChannel.ChannelState.END_OF_LIFE}
+{'channel': 'nixos-25.11-small',
+ 'release_branch': 'release-25.11',
+ 'revision': 'c9bfd86ed684d27e63b0ff9ebb18699f84f27a3b',
+ 'status': NixChannel.ChannelState.END_OF_LIFE,
+ 'variant': NixChannel.Variant.SMALL}
+{'channel': 'nixpkgs-25.11-darwin',
+ 'release_branch': 'release-25.11',
+ 'revision': '0921fdb3e13e40fe25fbc52b89661a9d6d32ac68',
+ 'status': NixChannel.ChannelState.END_OF_LIFE,
+ 'variant': NixChannel.Variant.DARWIN}
 ```
 
 2. Select a `head_sha1_commit` from the output of `fetch_all_channels` command and run evaluation on that:

src/shared/management/commands/fetch_all_channels.py

@@ -1,40 +1,54 @@
-import asyncio
-import sys
-from collections.abc import Coroutine
-from dataclasses import dataclass
+import re
+from io import StringIO
 from pprint import pprint
-from typing import Any
+from typing import Annotated, Any
 
 import requests
 from django.conf import settings
 from django.core.management.base import BaseCommand
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
-from shared.git import GitRepo
-from shared.models.nix_evaluation import NixChannel, release_branch
+from shared.models.nix_evaluation import NixChannel
 
 
-@dataclass
-class MonitoredChannel:
-    name: str
-    revision: str
-    status: str
-    variant: str | None
+class MonitoredChannel(BaseModel):
+    model_config = ConfigDict(extra="ignore")
 
+    channel: str
+    release_branch: str
+    revision: Annotated[str, Field(pattern="[0-9a-f]{40}")]
+    status: NixChannel.ChannelState
+    variant: NixChannel.Variant | None = None
 
-def aggregate_by_channels(data: list[dict[str, Any]]) -> dict[str, MonitoredChannel]:
-    channels = {}
-    for metric in data:
-        m = metric["metric"]
-        channels[m["channel"]] = MonitoredChannel(
-            name=m["channel"],
-            revision=m["revision"],
-            status=m["status"],
-            variant=m.get("variant"),
+    @model_validator(mode="before")
+    @classmethod
+    def parse_channel_name(cls, data: Any) -> Any:
+        """
+        This is a heuristic for deriving the release branch from a channel's name.
+        The Technically Correct way to obtain that value would be to
+        1. Take the underlying `channels.nix` from `github:NixOS/infra` for jobsets and channel states
+        2. Resolve the associated branch names and commits through Hydra
+        because those are the authoritative sources.
+
+        But that would require hitting Hydra with 3 roundtrips per channel and parsing all the responses.
+        Therefore we'll only implement that if the flexibility is actually required.
+        """
+        name = data.get("channel", "")
+        match = re.match(
+            r"^(?P<jobset>nixos|nixpkgs)-(?P<release>\d\d\.\d\d|unstable)(?:-(?P<variant>primary|small|darwin))?$",
+            name,
         )
-    return channels
+        if match is None:
+            raise ValueError(f"unexpected channel name: {name!r}")
+        release = match.group("release")
+        release_branch = "master" if release == "unstable" else f"release-{release}"
+        return {
+            **data,
+            "release_branch": release_branch,
+        }
 
 
-def fetch_from_monitoring() -> dict[str, MonitoredChannel]:
+def fetch_from_monitoring() -> list[MonitoredChannel]:
     resp = requests.get(
         # XXX(@fricklerhandwerk): The sources for this are declared in the `NixOS/infra` repo. [tag:channel-structure]
         # exporter logic:
@@ -46,41 +60,28 @@ def fetch_from_monitoring() -> dict[str, MonitoredChannel]:
         settings.CHANNEL_MONITORING_URL
     )
     resp.raise_for_status()
-    return aggregate_by_channels(resp.json()["data"]["result"])
-
-
-async def wait_for_parallel_fetches(
-    parallel_fetches: list[Coroutine[Any, Any, bool]],
-) -> list[Any]:
-    return await asyncio.gather(*parallel_fetches, return_exceptions=True)
+    channels = []
+    for metric in resp.json()["data"]["result"]:
+        channels.append(MonitoredChannel.model_validate(metric["metric"]))
+    return channels
 
 
 class Command(BaseCommand):
-    help = "Register Nix channels"
+    help = "Fetch current channel tips"
 
     def handle(self, *args: Any, **kwargs: Any) -> str | None:
-        fresh_channels = fetch_from_monitoring()
-        for channel in fresh_channels.values():
-            channel_branch = channel.name
-            branch_info = {
-                "release_branch": release_branch(channel.name),
-                "state": NixChannel.ChannelState(channel.status),
-                "head_sha1_commit": channel.revision,
-                "variant": channel.variant,
-            }
-            pprint(branch_info | {"channel_branch": channel.name})
-            NixChannel.objects.update_or_create(
-                branch_info, channel_branch=channel_branch
+        for monitored in fetch_from_monitoring():
+            channel, _ = NixChannel.objects.update_or_create(
+                channel_branch=monitored.channel,
+                defaults=dict(
+                    release_branch=monitored.release_branch,
+                    head_sha1_commit=monitored.revision,
+                    state=monitored.status,
+                    variant=monitored.variant,
+                ),
             )
 
-        repo = GitRepo(
-            settings.LOCAL_NIXPKGS_CHECKOUT,
-            stderr=sys.stderr.fileno(),
-        )
-        parallel_fetches = []
-        for channel in NixChannel.objects.iterator():
-            parallel_fetches.append(repo.update_from_ref(channel.head_sha1_commit))
-
-        results = asyncio.run(wait_for_parallel_fetches(parallel_fetches))
-        # FIXME(@fricklerhandwerk): Fold that into `branch_info`, so there's only one output.
-        print("Parallel fetches results", results)
+            # Can't `pprint()` to `self.stdout` directly...
+            stream = StringIO()
+            pprint(monitored.__dict__, stream=stream)
+            self.stdout.write(stream.getvalue())

src/shared/models/nix_evaluation.py

@@ -1,5 +1,3 @@
-import re
-
 from django.conf import settings
 from django.contrib.postgres import fields
 from django.contrib.postgres.indexes import BTreeIndex, GinIndex
@@ -330,18 +328,3 @@ def get_major_channel(branch_name: str) -> str | None:
         if mc in branch_name:
             return f"nixos-{mc}"
     return None
-
-
-def get_release(channel_branch: str) -> str:
-    match = re.match(
-        r"^(?P<jobset>[a-z]+)-(?P<release>\d\d\.\d\d|unstable)(?:-(?P<variant>[a-z]+))?$",
-        channel_branch,
-    )
-    if match is None:
-        raise ValueError(f"unexpected channel branch name: {channel_branch!r}")
-    return match.group("release")
-
-
-def release_branch(channel_branch: str) -> str:
-    release = get_release(channel_branch)
-    return "master" if release == "unstable" else f"release-{release}"

src/shared/tests/conftest.py

@@ -37,7 +37,6 @@
     NixDerivationMeta,
     NixEvaluation,
     NixMaintainer,
-    release_branch,
 )
 from shared.models.package import Package, PackageAttrpath
 from shared.notify_users import create_package_subscription_notifications
@@ -135,20 +134,19 @@ def cve(make_container: Callable[..., Container]) -> Container:
 
 @pytest.fixture
 def make_channel(db: None) -> Callable[..., NixChannel]:
-    # FIXME(@fricklerhandwerk): This will fall apart when we obtain the channel structure dynamically [ref:channel-structure]
     def wrapped(
         channel_branch: str = settings.TRACKING_BRANCH,
+        release_branch: str = "master",
         state: NixChannel.ChannelState = NixChannel.ChannelState.UNSTABLE,
         variant: NixChannel.Variant | None = None,
     ) -> NixChannel:
         channel, _ = NixChannel.objects.get_or_create(
             channel_branch=channel_branch,
             defaults=dict(
-                head_sha1_commit=secrets.token_hex(16),
+                head_sha1_commit=secrets.token_hex(20),
                 state=state,
                 variant=variant,
-                release_branch=release_branch(channel_branch),
-                repository="https://github.com/NixOS/nixpkgs",
+                release_branch=release_branch,
             ),
         )
         return channel
@@ -177,7 +175,7 @@ def wrapped(
             channel=channel,
             commit_sha1=commit_sha1
             if commit_sha1 is not None
-            else secrets.token_hex(16),
+            else secrets.token_hex(20),
             state=state,
         )
 

src/shared/tests/test_fetch_all_channels.py

@@ -0,0 +1,75 @@
+from unittest.mock import Mock, patch
+
+import pytest
+
+from shared.management.commands.fetch_all_channels import fetch_from_monitoring
+
+
+def monitoring_response(name: str, revision: str, status: str) -> Mock:
+    resp = Mock()
+    resp.json.return_value = {
+        "data": {
+            "result": [
+                {
+                    "metric": {
+                        "channel": name,
+                        "revision": revision,
+                        "status": status,
+                    }
+                }
+            ]
+        }
+    }
+    return resp
+
+
+@pytest.mark.parametrize(
+    "name,name_valid",
+    [
+        ("nixos-25.05", True),
+        ("nixos-unstable", True),
+        ("nixos", False),
+        ("NixOS-25.05", False),
+        ("--evil", False),
+        ("", False),
+    ],
+)
+@pytest.mark.parametrize(
+    "revision,revision_valid",
+    [
+        ("a" * 40, True),
+        ("abc123", False),
+        ("z" * 40, False),
+        ("--upload-pack=evil" + "a" * 21, False),
+        ("", False),
+    ],
+)
+@pytest.mark.parametrize(
+    "status,status_valid",
+    [
+        ("stable", True),
+        ("rolling", True),
+        ("STABLE", False),
+        ("1", False),
+        ("", False),
+    ],
+)
+def test_input_validation(
+    name: str,
+    name_valid: bool,
+    revision: str,
+    revision_valid: bool,
+    status: str,
+    status_valid: bool,
+) -> None:
+    with patch(
+        "requests.get", return_value=monitoring_response(name, revision, status)
+    ):
+        if name_valid and revision_valid and status_valid:
+            [channel] = fetch_from_monitoring()
+            assert channel.channel == name
+            assert channel.revision == revision
+            assert channel.status == status
+        else:
+            with pytest.raises(Exception):
+                fetch_from_monitoring()

src/shared/tests/test_garbage_collect.py

@@ -237,8 +237,8 @@ def test_deletes_empty_old_evaluations(
     """
     channel = make_channel(
         state=channel_state,
-        # Make a unique channel for each state; the concrete name doesn't matter here.
-        channel_branch=f"{channel_state.value}-unstable",
+        # FIXME(@fricklerhandwerk): When leaving this at the default value, the channel is created via `make_drv()` at parse time...
+        channel_branch="nixos-unstable",
     )
     evaluation = make_evaluation(
         channel=channel,