refactor: split branches from channels in data model

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:

nix/tests/default.nix

@@ -44,6 +44,8 @@ pkgs.testers.runNixOSTest {
             git init --initial-branch=master
             git add -A
             git -c user.name=test -c user.email=test@test commit -m "test"
+            git branch release-25.11
+            git branch release-25.05
             git rev-parse HEAD > REVISION
           '';
     in
@@ -138,8 +140,10 @@ pkgs.testers.runNixOSTest {
       with subtest("Check that channels are fetched and only small ones get enqueued for evaluation"):
         server.succeed("wst-manage fetch_all_channels")
         ${in-shell "succeed" ''
-          from shared.models import NixChannel
-          assert NixChannel.objects.count() == 6
+          from shared.models import NixChannel, NixpkgsBranch
+
+          assert NixpkgsBranch.objects.count() == 3, f"expected 3 branches, got {NixpkgsBranch.objects.count()}"
+          assert NixChannel.objects.count() == 6, f"expected 6 channels, got {NixChannel.objects.count()}"
         ''}
         ${
           # Give it some time to queue up the evaluations...

src/project/settings.py

@@ -108,6 +108,12 @@ class DjangoSettings(BaseModel):
                 "https://monitoring.nixos.org/prometheus/api/v1/query?query=channel_revision"
             ),
         )
+        NETWORK_REQUEST_TIMEOUT: int = Field(
+            description="""
+            Timeout in seconds for outbound network requests.
+            """,
+            default=60,
+        )
         SYNC_GITHUB_STATE_AT_STARTUP: bool = Field(
             description="""
             Connect to GitHub when the service is started and update

src/shared/git.py

@@ -4,17 +4,33 @@
 import os.path
 import pathlib
 import random
+import subprocess
 import time
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from typing import IO, Any
 
 from django.conf import settings
+from pydantic import AnyUrl
 
 logger = logging.getLogger(__name__)
 
 
+def get_head_sha1(url: AnyUrl, branch: str) -> str:
+    result = subprocess.run(
+        ["git", "ls-remote", str(url), f"refs/heads/{branch}"],
+        capture_output=True,
+        text=True,
+        check=True,
+        timeout=settings.NETWORK_REQUEST_TIMEOUT,
+    )
+    line = result.stdout.strip()
+    if not line:
+        raise ValueError(f"branch {branch!r} not found at {url!r}")
+    return line.split()[0]
+
+
 @dataclass
 class Worktree:
     path: pathlib.Path

src/shared/management/commands/fetch_all_channels.py

@@ -1,40 +1,56 @@
-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 django.db import transaction
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
-from shared.git import GitRepo
-from shared.models.nix_evaluation import NixChannel, release_branch
+from shared.git import get_head_sha1
+from shared.models.nix_evaluation import NixChannel, NixpkgsBranch
 
 
-@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 +62,44 @@ 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
-            )
-
-        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))
+        channels = fetch_from_monitoring()
+
+        branch_tips: dict[str, str] = {
+            release_branch: get_head_sha1(settings.GIT_CLONE_URL, release_branch)
+            for release_branch in {c.release_branch for c in channels}
+        }
+
+        with transaction.atomic():
+            branches: dict[str, NixpkgsBranch] = {}
+            for name, head in branch_tips.items():
+                branch, _ = NixpkgsBranch.objects.update_or_create(
+                    name=name,
+                    defaults={"head_sha1_commit": head},
+                )
+                branches[name] = branch
+
+            for monitored in channels:
+                NixChannel.objects.update_or_create(
+                    channel_branch=monitored.channel,
+                    defaults=dict(
+                        release_branch=branches[monitored.release_branch],
+                        head_sha1_commit=monitored.revision,
+                        state=monitored.status,
+                        variant=monitored.variant,
+                    ),
+                )
 
-        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/migrations/0098_nixpkgsbranch_nixchannel_release_branch.py

@@ -0,0 +1,82 @@
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+def populate_release_branch(apps, schema_editor):
+    NixChannel = apps.get_model("shared", "NixChannel")
+    NixpkgsBranch = apps.get_model("shared", "NixpkgsBranch")
+    for channel in NixChannel.objects.all():
+        branch, _ = NixpkgsBranch.objects.get_or_create(
+            name=channel.release_branch_name,
+            defaults={"head_sha1_commit": channel.head_sha1_commit},
+        )
+        channel.release_branch = branch
+        channel.save(update_fields=["release_branch"])
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("shared", "0097_nixchannel_variant"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="NixpkgsBranch",
+            fields=[
+                (
+                    "name",
+                    models.CharField(max_length=126, primary_key=True, serialize=False),
+                ),
+                ("head_sha1_commit", models.CharField(max_length=40)),
+            ],
+            options={
+                "constraints": [
+                    models.CheckConstraint(
+                        check=models.Q(head_sha1_commit__regex="^[0-9a-f]{40}$"),
+                        name="nixpkgsbranch_head_sha1_commit_valid",
+                    )
+                ]
+            },
+        ),
+        migrations.RenameField(
+            model_name="nixchannel",
+            old_name="release_branch",
+            new_name="release_branch_name",
+        ),
+        migrations.AddField(
+            model_name="nixchannel",
+            name="release_branch",
+            field=models.ForeignKey(
+                null=True,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="channels",
+                to="shared.nixpkgsbranch",
+            ),
+        ),
+        # The table has triggers registered.
+        # Altering it within the migration will defer them to after the transaction, thus failing the whole thing.
+        migrations.RunSQL("ALTER TABLE shared_nixchannel DISABLE TRIGGER USER"),
+        migrations.RunPython(
+            code=populate_release_branch,
+            reverse_code=migrations.RunPython.noop,
+        ),
+        migrations.AlterField(
+            model_name="nixchannel",
+            name="release_branch",
+            field=models.ForeignKey(
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="channels",
+                to="shared.nixpkgsbranch",
+            ),
+        ),
+        migrations.RemoveField(
+            model_name="nixchannel",
+            name="release_branch_name",
+        ),
+        migrations.RemoveField(
+            model_name="nixchannel",
+            name="repository",
+        ),
+        migrations.RunSQL("ALTER TABLE shared_nixchannel ENABLE TRIGGER USER"),
+    ]