重命名为动态 Runner 并记录协议调度路线

src/gitea_dynamic_runner/__init__.py

@@ -0,0 +1 @@
+"""Dynamic Pod and microVM execution environments for Gitea Actions."""

src/gitea_dynamic_runner/controller.py

@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+"""Gitea workflow_job webhook to NATS JetStream producer."""
+
+import hashlib
+import hmac
+import json
+import os
+import ssl
+from pathlib import Path
+
+import nats
+from aiohttp import web
+from nats.js.api import DiscardPolicy, RetentionPolicy, StorageType, StreamConfig
+from nats.js.errors import NotFoundError
+
+from .models import IdentityBinding, RunnerRequest
+
+
+SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner")
+STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
+NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
+NATS_USER = os.environ.get("NATS_USER", "ci-producer")
+NATS_PASSWORD_FILE = Path(os.environ.get("NATS_PASSWORD_FILE", "/run/secrets/nats/password"))
+NATS_CA_FILE = os.environ.get("NATS_CA_FILE", "/etc/ssl/certs/ca-certificates.crt")
+WEBHOOK_SECRET_FILE = Path(os.environ.get("WEBHOOK_SECRET_FILE", "/run/secrets/gitea/webhook-secret"))
+
+
+def accepts(payload: object) -> tuple[bool, str | None]:
+    """Compatibility helper for callers that only need acceptance and identity."""
+    request = RunnerRequest.from_webhook(payload)
+    return (request is not None, str(request.job_id) if request else None)
+
+
+def valid_signature(body: bytes, signature: str) -> bool:
+    expected = hmac.new(WEBHOOK_SECRET_FILE.read_bytes().strip(), body, hashlib.sha256).hexdigest()
+    return hmac.compare_digest(signature.removeprefix("sha256="), expected)
+
+
+async def ensure_stream(js: object) -> None:
+    config = StreamConfig(
+        name=STREAM,
+        subjects=[f"{SUBJECT_PREFIX}.>"],
+        retention=RetentionPolicy.WORK_QUEUE,
+        storage=StorageType.FILE,
+        discard=DiscardPolicy.OLD,
+        max_age=24 * 60 * 60,
+        max_msgs=10_000,
+        max_bytes=256 * 1024 * 1024,
+        duplicate_window=24 * 60 * 60,
+    )
+    try:
+        await js.stream_info(STREAM)
+    except NotFoundError:
+        await js.add_stream(config=config)
+    else:
+        await js.update_stream(config=config)
+
+
+async def webhook(request: web.Request) -> web.Response:
+    body = await request.read()
+    if not valid_signature(body, request.headers.get("X-Gitea-Signature", "")):
+        raise web.HTTPUnauthorized()
+    try:
+        payload = json.loads(body)
+    except json.JSONDecodeError as error:
+        raise web.HTTPBadRequest(text="invalid JSON\n") from error
+    runner_request = RunnerRequest.from_webhook(payload)
+    if runner_request is not None:
+        await request.app["js"].publish(
+            f"{SUBJECT_PREFIX}.{runner_request.backend}",
+            runner_request.to_json(),
+            headers={"Nats-Msg-Id": f"gitea-workflow-job-{runner_request.job_id}-queued"},
+        )
+        return web.Response(status=202, text="queued\n")
+
+    binding = IdentityBinding.from_webhook(payload)
+    if binding is not None:
+        await request.app["js"].publish(
+            f"{SUBJECT_PREFIX}.{binding.backend}.binding",
+            binding.to_json(),
+            headers={"Nats-Msg-Id": f"gitea-workflow-job-{binding.job_id}-in-progress"},
+        )
+        return web.Response(status=202, text="binding queued\n")
+
+    return web.Response(status=204)
+
+
+async def health(request: web.Request) -> web.Response:
+    connected = request.app["nc"].is_connected
+    return web.Response(text="ok\n" if connected else "disconnected\n", status=200 if connected else 503)
+
+
+async def nats_context(app: web.Application):
+    tls = ssl.create_default_context(cafile=NATS_CA_FILE)
+    nc = await nats.connect(
+        NATS_URL,
+        user=NATS_USER,
+        password=NATS_PASSWORD_FILE.read_text().strip(),
+        tls=tls,
+        name="microvm-runner-controller",
+    )
+    app["nc"] = nc
+    app["js"] = nc.jetstream()
+    await ensure_stream(app["js"])
+    yield
+    await nc.drain()
+
+
+def create_app() -> web.Application:
+    app = web.Application(client_max_size=1024 * 1024)
+    app.cleanup_ctx.append(nats_context)
+    app.router.add_post("/webhook", webhook)
+    app.router.add_get("/healthz", health)
+    return app
+
+
+def main() -> None:
+    web.run_app(create_app(), host=os.environ.get("LISTEN", "0.0.0.0"), port=int(os.environ.get("PORT", "8787")))
+
+
+if __name__ == "__main__":
+    main()

src/gitea_dynamic_runner/jwt_broker.py

@@ -0,0 +1,90 @@
+#!/usr/bin/env python3
+"""Fixed-audience JWT-SVID broker for nested Gitea job containers."""
+
+import asyncio
+import json
+import os
+
+from aiohttp import web
+
+SPIRE_AGENT = os.environ.get("SPIRE_AGENT", "/opt/spire/bin/spire-agent")
+SPIRE_SOCKET = os.environ.get(
+    "SPIRE_SOCKET", "/run/spire/agent-sockets/spire-agent.sock"
+)
+AUDIENCE = os.environ.get("JWT_AUDIENCE", "zot")
+MAX_CONCURRENCY = int(os.environ.get("MAX_CONCURRENCY", "4"))
+semaphore = asyncio.Semaphore(MAX_CONCURRENCY)
+
+
+def extract_svid(document: object) -> str:
+    if not isinstance(document, list):
+        raise ValueError("unexpected SPIRE response")
+    for item in document:
+        if not isinstance(item, dict):
+            continue
+        svids = item.get("svids")
+        if not isinstance(svids, list):
+            continue
+        for svid in svids:
+            if isinstance(svid, dict) and isinstance(svid.get("svid"), str):
+                return svid["svid"]
+    raise ValueError("SPIRE response contains no JWT-SVID")
+
+
+async def fetch_svid() -> str:
+    async with semaphore:
+        process = await asyncio.create_subprocess_exec(
+            SPIRE_AGENT,
+            "api",
+            "fetch",
+            "jwt",
+            "-output",
+            "json",
+            "-audience",
+            AUDIENCE,
+            "-socketPath",
+            SPIRE_SOCKET,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=10)
+    if process.returncode != 0:
+        message = stderr.decode(errors="replace").strip()
+        raise RuntimeError(f"SPIRE agent failed: {message}")
+    return extract_svid(json.loads(stdout))
+
+
+async def token(_: web.Request) -> web.Response:
+    try:
+        value = await fetch_svid()
+    except (RuntimeError, ValueError, json.JSONDecodeError, asyncio.TimeoutError):
+        raise web.HTTPServiceUnavailable(text="identity unavailable\n")
+    return web.Response(
+        text=f"{value}\n",
+        content_type="text/plain",
+        headers={"Cache-Control": "no-store"},
+    )
+
+
+async def health(_: web.Request) -> web.Response:
+    return web.Response(text="ok\n")
+
+
+def create_app() -> web.Application:
+    app = web.Application(client_max_size=1024)
+    app.router.add_post("/token", token)
+    app.router.add_get("/healthz", health)
+    return app
+
+
+def main() -> None:
+    web.run_app(
+        create_app(),
+        host=os.environ.get("LISTEN", "0.0.0.0"),
+        port=int(os.environ.get("PORT", "8788")),
+        access_log=None,
+    )
+
+
+if __name__ == "__main__":
+    main()

src/gitea_dynamic_runner/kubernetes.py

@@ -0,0 +1,89 @@
+"""Small in-cluster Kubernetes API client used by the Pod backend."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from urllib.parse import quote
+
+from aiohttp import ClientResponseError, ClientSession, TCPConnector
+import ssl
+
+
+class KubernetesClient:
+    def __init__(
+        self,
+        *,
+        api_url: str,
+        token_file: Path,
+        ca_file: Path,
+        namespace: str,
+    ) -> None:
+        self.api_url = api_url.rstrip("/")
+        self.token_file = token_file
+        self.ca_file = ca_file
+        self.namespace = namespace
+        self.session: ClientSession | None = None
+
+    async def __aenter__(self) -> KubernetesClient:
+        context = ssl.create_default_context(cafile=self.ca_file)
+        self.session = ClientSession(
+            connector=TCPConnector(ssl=context),
+            headers={
+                "Authorization": f"Bearer {self.token_file.read_text().strip()}",
+            },
+            raise_for_status=True,
+        )
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        if self.session is not None:
+            await self.session.close()
+
+    async def create_pod(self, manifest: dict[str, object]) -> dict[str, object]:
+        response = await self._request("POST", self._pods_path(), json=manifest)
+        return await response.json()
+
+    async def get_pod(self, name: str) -> dict[str, object] | None:
+        try:
+            response = await self._request("GET", f"{self._pods_path()}/{quote(name)}")
+        except ClientResponseError as error:
+            if error.status == 404:
+                return None
+            raise
+        return await response.json()
+
+    async def bind_identity(self, name: str, identity_path: str) -> None:
+        patch = {
+            "metadata": {
+                "labels": {"ci.ddupan.top/identity-bound": "true"},
+                "annotations": {"ci.ddupan.top/spiffe-path": identity_path},
+            }
+        }
+        response = await self._request(
+            "PATCH",
+            f"{self._pods_path()}/{quote(name)}",
+            data=json.dumps(patch),
+            headers={"Content-Type": "application/merge-patch+json"},
+        )
+        response.release()
+
+    async def delete_pod(self, name: str) -> None:
+        try:
+            response = await self._request(
+                "DELETE",
+                f"{self._pods_path()}/{quote(name)}",
+                json={"gracePeriodSeconds": 30, "propagationPolicy": "Background"},
+            )
+            response.release()
+        except ClientResponseError as error:
+            if error.status != 404:
+                raise
+
+    async def _request(self, method: str, path: str, **kwargs: object):
+        if self.session is None:
+            raise RuntimeError("KubernetesClient is not open")
+        return await self.session.request(method, f"{self.api_url}{path}", **kwargs)
+
+    def _pods_path(self) -> str:
+        return f"/api/v1/namespaces/{quote(self.namespace)}/pods"

src/gitea_dynamic_runner/models.py

@@ -0,0 +1,176 @@
+"""Validated messages shared by the controller and runner backends."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+import json
+
+
+BACKENDS = frozenset({"pod", "vm"})
+REQUIRED_LABEL = "self-hosted"
+
+
+@dataclass(frozen=True, slots=True)
+class RunnerRequest:
+    """A queued Gitea job that needs one disposable runner environment."""
+
+    job_id: int
+    run_id: int
+    backend: str
+    repository: str
+    job_name: str
+    labels: tuple[str, ...]
+    schema_version: int = 1
+
+    @classmethod
+    def from_webhook(cls, payload: object) -> RunnerRequest | None:
+        if not isinstance(payload, dict) or payload.get("action") != "queued":
+            return None
+
+        job = payload.get("workflow_job")
+        repository = payload.get("repository")
+        if not isinstance(job, dict) or not isinstance(repository, dict):
+            return None
+
+        labels_value = job.get("labels")
+        if not isinstance(labels_value, list) or not all(
+            isinstance(label, str) for label in labels_value
+        ):
+            return None
+        labels = tuple(dict.fromkeys(labels_value))
+        selected = BACKENDS.intersection(labels)
+        if REQUIRED_LABEL not in labels or len(selected) != 1:
+            return None
+
+        job_id = job.get("id")
+        run_id = job.get("run_id")
+        job_name = job.get("name")
+        full_name = repository.get("full_name")
+        if not _positive_int(job_id) or not _positive_int(run_id):
+            return None
+        if not all(_nonempty(value) for value in (job_name, full_name)):
+            return None
+
+        return cls(
+            job_id=job_id,
+            run_id=run_id,
+            backend=next(iter(selected)),
+            repository=full_name.strip(),
+            job_name=job_name.strip(),
+            labels=labels,
+        )
+
+    def to_json(self) -> bytes:
+        document = asdict(self)
+        document["labels"] = list(self.labels)
+        return json.dumps(
+            document,
+            ensure_ascii=False,
+            separators=(",", ":"),
+            sort_keys=True,
+        ).encode()
+
+    @classmethod
+    def from_json(cls, body: bytes) -> RunnerRequest:
+        document = json.loads(body)
+        if not isinstance(document, dict) or document.get("schema_version") != 1:
+            raise ValueError("unsupported runner request")
+        labels = document.get("labels")
+        if not isinstance(labels, list) or not all(
+            isinstance(label, str) for label in labels
+        ):
+            raise ValueError("invalid runner labels")
+        try:
+            request = cls(
+                job_id=document["job_id"],
+                run_id=document["run_id"],
+                backend=document["backend"],
+                repository=document["repository"],
+                job_name=document["job_name"],
+                labels=tuple(labels),
+            )
+        except KeyError as error:
+            raise ValueError(f"missing runner request field: {error.args[0]}") from error
+        if (
+            not _positive_int(request.job_id)
+            or not _positive_int(request.run_id)
+            or request.backend not in BACKENDS
+            or not _nonempty(request.repository)
+            or not _nonempty(request.job_name)
+            or REQUIRED_LABEL not in request.labels
+            or request.backend not in request.labels
+        ):
+            raise ValueError("invalid runner request")
+        return request
+
+
+@dataclass(frozen=True, slots=True)
+class IdentityBinding:
+    """The actual task claimed by an ephemeral runner."""
+
+    job_id: int
+    run_id: int
+    backend: str
+    runner_name: str
+    repository: str
+    job_name: str
+    schema_version: int = 1
+
+    @classmethod
+    def from_webhook(cls, payload: object) -> IdentityBinding | None:
+        if not isinstance(payload, dict) or payload.get("action") != "in_progress":
+            return None
+        job = payload.get("workflow_job")
+        repository = payload.get("repository")
+        if not isinstance(job, dict) or not isinstance(repository, dict):
+            return None
+
+        labels = job.get("labels")
+        if not isinstance(labels, list) or not all(
+            isinstance(label, str) for label in labels
+        ):
+            return None
+        selected = BACKENDS.intersection(labels)
+        if REQUIRED_LABEL not in labels or len(selected) != 1:
+            return None
+        backend = next(iter(selected))
+
+        job_id = job.get("id")
+        run_id = job.get("run_id")
+        runner_name = job.get("runner_name")
+        job_name = job.get("name")
+        full_name = repository.get("full_name")
+        if not _positive_int(job_id) or not _positive_int(run_id):
+            return None
+        if not all(
+            _nonempty(value)
+            for value in (runner_name, job_name, full_name)
+        ):
+            return None
+        if not runner_name.startswith(f"gitea-{backend}-"):
+            return None
+
+        return cls(
+            job_id=job_id,
+            run_id=run_id,
+            backend=backend,
+            runner_name=runner_name.strip(),
+            repository=full_name.strip(),
+            job_name=job_name.strip(),
+        )
+
+    def to_json(self) -> bytes:
+        return json.dumps(
+            asdict(self),
+            ensure_ascii=False,
+            separators=(",", ":"),
+            sort_keys=True,
+        ).encode()
+
+
+def _positive_int(value: object) -> bool:
+    return isinstance(value, int) and not isinstance(value, bool) and value > 0
+
+
+def _nonempty(value: object) -> bool:
+    return isinstance(value, str) and bool(value.strip())

src/gitea_dynamic_runner/pod_worker.py

@@ -0,0 +1,307 @@
+#!/usr/bin/env python3
+"""JetStream worker that creates one disposable Kubernetes Pod per job."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+from pathlib import Path
+import ssl
+from urllib.parse import quote
+import uuid
+
+import nats
+from aiohttp import ClientResponseError
+from nats.errors import TimeoutError
+from nats.js.api import AckPolicy, ConsumerConfig
+
+from .kubernetes import KubernetesClient
+from .models import IdentityBinding, RunnerRequest
+
+
+LOG = logging.getLogger(__name__)
+STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
+REQUEST_SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.pod")
+BINDING_SUBJECT = os.environ.get("NATS_BINDING_SUBJECT", "ci.runner.pod.binding")
+NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
+NATS_USER = os.environ.get("NATS_USER", "ci-worker")
+NATS_PASSWORD_FILE = Path(os.environ.get("NATS_PASSWORD_FILE", "/run/secrets/nats/password"))
+NATS_CA_FILE = os.environ.get("NATS_CA_FILE", "/etc/ssl/certs/ca-certificates.crt")
+KUBERNETES_API = os.environ.get("KUBERNETES_API", "https://kubernetes.default.svc")
+KUBERNETES_TOKEN_FILE = Path(
+    os.environ.get(
+        "KUBERNETES_TOKEN_FILE",
+        "/var/run/secrets/kubernetes.io/serviceaccount/token",
+    )
+)
+KUBERNETES_CA_FILE = Path(
+    os.environ.get(
+        "KUBERNETES_CA_FILE",
+        "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
+    )
+)
+NAMESPACE = os.environ.get("RUNNER_NAMESPACE", "gitea-actions")
+RUNNER_IMAGE = os.environ.get("RUNNER_IMAGE", "docker.io/gitea/runner:2")
+RUNNER_SERVICE_ACCOUNT = os.environ.get("RUNNER_SERVICE_ACCOUNT", "gitea-dynamic-runner")
+RUNNER_TOKEN_SECRET = os.environ.get("RUNNER_TOKEN_SECRET", "gitea-dynamic-runner")
+GITEA_INSTANCE = os.environ.get("GITEA_INSTANCE", "https://git.ddupan.top")
+CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "4"))
+POD_TIMEOUT = int(os.environ.get("RUNNER_POD_TIMEOUT", str(4 * 60 * 60)))
+
+
+def identity_path(repository: str, job_name: str) -> str:
+    parts = repository.split("/")
+    if len(parts) != 2 or not all(parts):
+        raise ValueError("repository must be owner/name")
+    encoded = [quote(part, safe="-._~") for part in (*parts, job_name)]
+    if not job_name.strip() or any(not part for part in encoded):
+        raise ValueError("identity components must not be empty")
+    return "/".join(encoded)
+
+
+def pod_manifest(request: RunnerRequest, pod_name: str) -> dict[str, object]:
+    if request.backend != "pod":
+        raise ValueError("Pod backend only accepts pod requests")
+    return {
+        "apiVersion": "v1",
+        "kind": "Pod",
+        "metadata": {
+            "name": pod_name,
+            "namespace": NAMESPACE,
+            "labels": {
+                "app.kubernetes.io/name": "gitea-dynamic-runner",
+                "app.kubernetes.io/component": "runner",
+                "ci.ddupan.top/backend": "pod",
+            },
+            "annotations": {
+                "ci.ddupan.top/queued-job-id": str(request.job_id),
+                "ci.ddupan.top/queued-run-id": str(request.run_id),
+            },
+        },
+        "spec": {
+            "serviceAccountName": RUNNER_SERVICE_ACCOUNT,
+            "restartPolicy": "Never",
+            "terminationGracePeriodSeconds": 30,
+            "containers": [
+                {
+                    "name": "runner",
+                    "image": RUNNER_IMAGE,
+                    "imagePullPolicy": "IfNotPresent",
+                    "securityContext": {"privileged": True},
+                    "env": [
+                        {"name": "GITEA_INSTANCE_URL", "value": GITEA_INSTANCE},
+                        {
+                            "name": "GITEA_RUNNER_NAME",
+                            "valueFrom": {"fieldRef": {"fieldPath": "metadata.name"}},
+                        },
+                        {
+                            "name": "GITEA_RUNNER_REGISTRATION_TOKEN_FILE",
+                            "value": "/run/secrets/gitea/token",
+                        },
+                        {"name": "GITEA_RUNNER_LABELS", "value": "self-hosted:host,pod:host"},
+                        {"name": "GITEA_RUNNER_EPHEMERAL", "value": "1"},
+                        {"name": "GITEA_RUNNER_ONCE", "value": "1"},
+                        {"name": "CONFIG_FILE", "value": "/etc/gitea-runner/config.yaml"},
+                    ],
+                    "volumeMounts": [
+                        {
+                            "name": "registration-token",
+                            "mountPath": "/run/secrets/gitea",
+                            "readOnly": True,
+                        },
+                        {
+                            "name": "spire-agent-socket",
+                            "mountPath": "/run/spire/agent-sockets",
+                            "readOnly": True,
+                        },
+                    ],
+                }
+            ],
+            "volumes": [
+                {
+                    "name": "registration-token",
+                    "secret": {
+                        "secretName": RUNNER_TOKEN_SECRET,
+                        "items": [{"key": "token", "path": "token"}],
+                    },
+                },
+                {
+                    "name": "spire-agent-socket",
+                    "csi": {"driver": "csi.spiffe.io", "readOnly": True},
+                },
+            ],
+        },
+    }
+
+
+async def heartbeat(message: object, stop: asyncio.Event) -> None:
+    while True:
+        try:
+            await asyncio.wait_for(stop.wait(), timeout=60)
+            return
+        except asyncio.TimeoutError:
+            await message.in_progress()
+
+
+async def wait_for_pod(client: KubernetesClient, name: str) -> bool:
+    deadline = asyncio.get_running_loop().time() + POD_TIMEOUT
+    while asyncio.get_running_loop().time() < deadline:
+        pod = await client.get_pod(name)
+        if pod is None:
+            raise RuntimeError(f"runner Pod {name} disappeared")
+        status = pod.get("status")
+        phase = status.get("phase") if isinstance(status, dict) else None
+        if phase == "Succeeded":
+            return True
+        if phase == "Failed":
+            return False
+        await asyncio.sleep(2)
+    raise asyncio.TimeoutError(f"runner Pod {name} timed out")
+
+
+async def run_request(message: object, client: KubernetesClient) -> None:
+    try:
+        request = RunnerRequest.from_json(message.data)
+    except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as error:
+        LOG.error("discarding invalid Pod request: %s", error)
+        await message.ack()
+        return
+    if request.backend != "pod":
+        LOG.error("discarding %s request received by Pod worker", request.backend)
+        await message.ack()
+        return
+
+    pod_name = f"gitea-pod-{uuid.uuid4().hex[:12]}"
+    stop = asyncio.Event()
+    pulse = asyncio.create_task(heartbeat(message, stop))
+    try:
+        await client.create_pod(pod_manifest(request, pod_name))
+        if await wait_for_pod(client, pod_name):
+            await message.ack()
+        else:
+            LOG.error("runner Pod %s failed", pod_name)
+            await message.nak(delay=30)
+    except Exception:
+        await message.nak(delay=30)
+        raise
+    finally:
+        stop.set()
+        await pulse
+        await client.delete_pod(pod_name)
+
+
+async def bind_request(message: object, client: KubernetesClient) -> None:
+    try:
+        document = json.loads(message.data)
+        binding = IdentityBinding(**document)
+        if binding.backend != "pod" or not binding.runner_name.startswith("gitea-pod-"):
+            raise ValueError("invalid Pod identity binding")
+        path = identity_path(binding.repository, binding.job_name)
+        await client.bind_identity(binding.runner_name, path)
+    except ClientResponseError as error:
+        if error.status == 404:
+            await message.nak(delay=2)
+            return
+        await message.nak(delay=30)
+        raise
+    except (json.JSONDecodeError, TypeError, ValueError) as error:
+        LOG.error("discarding invalid identity binding: %s", error)
+        await message.ack()
+        return
+    await message.ack()
+
+
+async def consume_requests(subscription: object, client: KubernetesClient) -> None:
+    active: set[asyncio.Task[None]] = set()
+    while True:
+        active = {task for task in active if not task.done()}
+        free = CAPACITY - len(active)
+        if free < 1:
+            await asyncio.wait(active, return_when=asyncio.FIRST_COMPLETED)
+            continue
+        try:
+            messages = await subscription.fetch(batch=free, timeout=5)
+        except TimeoutError:
+            continue
+        for message in messages:
+            task = asyncio.create_task(run_request(message, client))
+            task.add_done_callback(_report_task)
+            active.add(task)
+
+
+async def consume_bindings(subscription: object, client: KubernetesClient) -> None:
+    while True:
+        try:
+            messages = await subscription.fetch(batch=16, timeout=5)
+        except TimeoutError:
+            continue
+        await asyncio.gather(*(bind_request(message, client) for message in messages))
+
+
+def _report_task(task: asyncio.Task[None]) -> None:
+    if not task.cancelled() and (error := task.exception()) is not None:
+        LOG.error("Pod runner task failed", exc_info=(type(error), error, error.__traceback__))
+
+
+async def main() -> None:
+    if CAPACITY < 1:
+        raise ValueError("RUNNER_CAPACITY must be at least 1")
+    context = ssl.create_default_context(cafile=NATS_CA_FILE)
+    nc = await nats.connect(
+        NATS_URL,
+        user=NATS_USER,
+        password=NATS_PASSWORD_FILE.read_text().strip(),
+        tls=context,
+        name="pod-runner-worker",
+    )
+    js = nc.jetstream()
+    requests = await js.pull_subscribe(
+        REQUEST_SUBJECT,
+        durable="pod",
+        stream=STREAM,
+        config=ConsumerConfig(
+            durable_name="pod",
+            filter_subject=REQUEST_SUBJECT,
+            ack_policy=AckPolicy.EXPLICIT,
+            ack_wait=5 * 60,
+            max_ack_pending=max(CAPACITY, 1),
+            max_deliver=5,
+        ),
+    )
+    bindings = await js.pull_subscribe(
+        BINDING_SUBJECT,
+        durable="pod-binding",
+        stream=STREAM,
+        config=ConsumerConfig(
+            durable_name="pod-binding",
+            filter_subject=BINDING_SUBJECT,
+            ack_policy=AckPolicy.EXPLICIT,
+            ack_wait=30,
+            max_ack_pending=64,
+            max_deliver=10,
+        ),
+    )
+    async with KubernetesClient(
+        api_url=KUBERNETES_API,
+        token_file=KUBERNETES_TOKEN_FILE,
+        ca_file=KUBERNETES_CA_FILE,
+        namespace=NAMESPACE,
+    ) as client:
+        try:
+            await asyncio.gather(
+                consume_requests(requests, client),
+                consume_bindings(bindings, client),
+            )
+        finally:
+            await nc.drain()
+
+
+def cli() -> None:
+    logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
+    asyncio.run(main())
+
+
+if __name__ == "__main__":
+    cli()

src/gitea_dynamic_runner/worker.py

@@ -0,0 +1,180 @@
+#!/usr/bin/env python3
+"""Capacity-bounded JetStream consumer that launches one ephemeral VM per job."""
+
+import asyncio
+import json
+import logging
+import os
+import secrets
+import ssl
+import uuid
+from pathlib import Path
+
+import nats
+from aiohttp import web
+from nats.errors import TimeoutError
+from nats.js.api import AckPolicy, ConsumerConfig
+
+from .models import RunnerRequest
+
+LOG = logging.getLogger(__name__)
+CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "1"))
+SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.vm")
+STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
+DURABLE = os.environ.get("NATS_DURABLE", "vm")
+MAX_INFLIGHT = int(os.environ.get("RUNNER_MAX_INFLIGHT", "64"))
+NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
+NATS_USER = os.environ.get("NATS_USER", "ci-worker")
+NATS_PASSWORD_FILE = Path(os.environ.get("NATS_PASSWORD_FILE", "/etc/microvm-runner/nats-password"))
+NATS_CA_FILE = os.environ.get("NATS_CA_FILE", "/etc/ssl/certs/ca-certificates.crt")
+REGISTRATION_TOKEN_FILE = Path(os.environ.get("REGISTRATION_TOKEN_FILE", "/etc/microvm-runner/registration-token"))
+LAUNCHER = os.environ.get("LAUNCHER", "/usr/local/libexec/microvm-runner-launch")
+TOKEN_LISTEN = os.environ.get("TOKEN_LISTEN", "172.30.0.1")
+TOKEN_PORT = int(os.environ.get("TOKEN_PORT", "8787"))
+
+tokens: dict[str, bytes] = {}
+token_lock = asyncio.Lock()
+
+
+async def token(request: web.Request) -> web.Response:
+    nonce = request.match_info["nonce"]
+    async with token_lock:
+        value = tokens.pop(nonce, None)
+    if value is None:
+        raise web.HTTPNotFound()
+    return web.Response(body=value, headers={"Cache-Control": "no-store"})
+
+
+async def heartbeat(message: object, stop: asyncio.Event) -> None:
+    while True:
+        try:
+            await asyncio.wait_for(stop.wait(), timeout=60)
+            return
+        except asyncio.TimeoutError:
+            await message.in_progress()
+
+
+async def run_one(message: object) -> None:
+    try:
+        request = RunnerRequest.from_json(message.data)
+    except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as error:
+        LOG.error("discarding invalid runner request: %s", error)
+        await message.ack()
+        return
+    if request.backend != "vm":
+        LOG.error("discarding %s request received by VM worker", request.backend)
+        await message.ack()
+        return
+
+    instance_id = str(uuid.uuid4())
+    nonce = secrets.token_urlsafe(32)
+    async with token_lock:
+        tokens[nonce] = REGISTRATION_TOKEN_FILE.read_bytes().strip()
+    stop = asyncio.Event()
+    pulse = asyncio.create_task(heartbeat(message, stop))
+    try:
+        process = await asyncio.create_subprocess_exec(
+            LAUNCHER,
+            instance_id,
+            nonce,
+            env={
+                **os.environ,
+                "RUNNER_JOB_ID": str(request.job_id),
+                "RUNNER_RUN_ID": str(request.run_id),
+                "RUNNER_REPOSITORY": request.repository,
+                "RUNNER_JOB_NAME": request.job_name,
+            },
+        )
+        return_code = await process.wait()
+        if return_code == 0:
+            await message.ack()
+        else:
+            LOG.error("launcher for %s exited with %d", instance_id, return_code)
+            await message.nak(delay=30)
+    except Exception:
+        await message.nak(delay=30)
+        raise
+    finally:
+        stop.set()
+        await pulse
+        async with token_lock:
+            tokens.pop(nonce, None)
+
+
+def _report_task(task: asyncio.Task[None]) -> None:
+    if task.cancelled():
+        return
+    error = task.exception()
+    if error is not None:
+        LOG.error(
+            "runner task failed",
+            exc_info=(type(error), error, error.__traceback__),
+        )
+
+
+async def consume() -> None:
+    if CAPACITY < 1:
+        raise ValueError("RUNNER_CAPACITY must be at least 1")
+    tls = ssl.create_default_context(cafile=NATS_CA_FILE)
+    nc = await nats.connect(
+        NATS_URL,
+        user=NATS_USER,
+        password=NATS_PASSWORD_FILE.read_text().strip(),
+        tls=tls,
+        name=DURABLE,
+    )
+    js = nc.jetstream()
+    subscription = await js.pull_subscribe(
+        SUBJECT,
+        durable=DURABLE,
+        stream=STREAM,
+        config=ConsumerConfig(
+            durable_name=DURABLE,
+            filter_subject=SUBJECT,
+            ack_policy=AckPolicy.EXPLICIT,
+            ack_wait=5 * 60,
+            max_ack_pending=MAX_INFLIGHT,
+            max_deliver=5,
+        ),
+    )
+    active: set[asyncio.Task[None]] = set()
+    try:
+        while True:
+            active = {task for task in active if not task.done()}
+            free = CAPACITY - len(active)
+            if free == 0:
+                await asyncio.wait(active, return_when=asyncio.FIRST_COMPLETED)
+                continue
+            try:
+                messages = await subscription.fetch(batch=free, timeout=5)
+            except TimeoutError:
+                continue
+            for message in messages:
+                task = asyncio.create_task(run_one(message))
+                task.add_done_callback(_report_task)
+                active.add(task)
+    finally:
+        if active:
+            await asyncio.gather(*active, return_exceptions=True)
+        await nc.drain()
+
+
+async def main() -> None:
+    app = web.Application()
+    app.router.add_get("/token/{nonce}", token)
+    runner = web.AppRunner(app)
+    await runner.setup()
+    await web.TCPSite(runner, TOKEN_LISTEN, TOKEN_PORT).start()
+    try:
+        await consume()
+    finally:
+        await runner.cleanup()
+
+
+def cli() -> None:
+    logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
+    asyncio.run(main())
+
+
+if __name__ == "__main__":
+    cli()