#4 perf(runner): job phase timing, preinstalled Go/Node tool cache and Docker Hub pull-through proxy

M docs/cache-and-artifacts.md +56 -1

@@ -1,6 +1,6 @@
1	1		# Cache and artifacts
2	2
3		-	The runner serves the GitHub Actions cache and artifacts v4 to jobs. Both use one channel per sandbox.
	3	+	The runner serves the GitHub Actions cache, artifacts v4 and a Docker Hub mirror to jobs. All three use one channel per sandbox.
4	4
5	5		## Channel
6	6

@@ -79,3 +79,58 @@ Works with `actions/cache`, `setup-go` and `setup-node` with `cache:`, and anyth
79	79		- Artifacts v3.
80	80		- The `artifact-url` output. It points at the fake host. The run page in ermite lists the artifacts.
81	81		- Downloads from another run (`run-id` with `github-token`).
	82	+
	83	+	## Docker Hub images
	84	+
	85	+	The runner provides an anonymous, pull-only Docker Hub mirror on the same private
	86	+	channel. Each job's dockerd uses `http://127.0.0.1:9089` as a registry mirror.
	87	+	Other registries are not cached. There is no HTTPS interception, and the mirror
	88	+	needs no network policy rule. Guest credentials are never forwarded upstream. Private pulls
	89	+	fall back to Docker Hub directly, where dockerd can use the job's credentials.
	90	+
	91	+	Verified sha256 blobs and manifests live under `<state dir>/registry/`. Tags are
	92	+	revalidated by an upstream HEAD on each request, with the client's Accept headers
	93	+	reduced to the known manifest media types. If Docker Hub is unreachable, times
	94	+	out, returns a server error or rate-limits (429), a previously cached tag can be
	95	+	served stale while its manifest remains stored. Unknown tags return an error so
	96	+	dockerd can fall back.
	97	+
	98	+	A cold blob is downloaded once, whoever asks for it, and is not tied to any one
	99	+	client: every client waits for the digest check and then reads the stored file.
	100	+	A corrupt download returns 502 and leaves no file. A client that disconnects
	101	+	does not stop the download while another client waits for it; once no client
	102	+	has waited for 5 s, the download is cancelled and its temporary file removed.
	103	+	Each job may have started at most 3 cold blob downloads at once (dockerd's
	104	+	default concurrency), and at most 16 run at once across all jobs. A download
	105	+	counts against the job that started it until it ends; a job that asks for a
	106	+	layer another job is already downloading waits for it and uses none of its
	107	+	quota. A request over either limit waits up to 30 s for a slot, its connection
	108	+	kept alive, and then gets 503. A refused layer makes dockerd retry for about
	109	+	50 s before it pulls that layer from Docker Hub directly. Each sandbox may also
	110	+	have at most 8 registry requests in progress; one more gets 503 at once.
	111	+
	112	+	Downloads have no total time limit, but they fail after 30 s without progress.
	113	+	Auth, manifest and HEAD requests are limited to 10 s. Auth and manifest requests
	114	+	never follow redirects; blob requests, HEAD included, follow the CDN redirect
	115	+	without the bearer token. Stored blobs support range requests. Cold blob HEAD
	116	+	requests fetch only upstream metadata, not the layer body.
	117	+
	118	+	`ERMITE_RUNNER_REGISTRY_CACHE_MAX` defaults to `10G`. Blobs, manifests and tag
	119	+	files share one least-recently-used eviction; each is charged at least 4 KiB, so
	120	+	the number of files stays below the cap divided by 4 KiB. An entry that is being served is
	121	+	never evicted. This cap is independent of
	122	+	`ERMITE_RUNNER_CACHE_MAX`. A zero size in any unit (`0`, `0M`, `0G`) disables only
	123	+	the proxy. `ERMITE_RUNNER_CACHE=0` disables the whole channel and therefore also
	124	+	the proxy (logged at startup). Dockerd retains its mirror setting and falls back
	125	+	to Docker Hub if the mirror is unavailable. To clear the cache, stop the runner
	126	+	and remove `<state dir>/registry/`.
	127	+
	128	+	Jobs with neither cache permissions nor artifacts get the loopback channel for
	129	+	dockerd when the proxy is on, but no `ACTIONS_CACHE_URL`, `ACTIONS_RESULTS_URL`, or
	130	+	`ACTIONS_RUNTIME_TOKEN` is exposed to their steps. Registry routes need no token:
	131	+	only public content is served over a channel exclusive to that sandbox.
	132	+
	133	+	The host logs one `registry manifest` line per request with `job_id`, `repo`,
	134	+	`ref`, `hit` and `ms`. Blob misses log `registry blob` at debug level with
	135	+	`job_id`, `digest`, `bytes` and `ms`; `job_id` is the job that started the
	136	+	download.

@@ -1,6 +1,6 @@
1		# Cache and artifacts	1		# Cache and artifacts
2			2
3	-	The runner serves the GitHub Actions cache and artifacts v4 to jobs. Both use one channel per sandbox.	3	+	The runner serves the GitHub Actions cache, artifacts v4 and a Docker Hub mirror to jobs. All three use one channel per sandbox.
4			4
5		## Channel	5		## Channel
6			6

@@ -79,3 +79,58 @@ Works with `actions/cache`, `setup-go` and `setup-node` with `cache:`, and anyth
79		- Artifacts v3.	79		- Artifacts v3.
80		- The `artifact-url` output. It points at the fake host. The run page in ermite lists the artifacts.	80		- The `artifact-url` output. It points at the fake host. The run page in ermite lists the artifacts.
81		- Downloads from another run (`run-id` with `github-token`).	81		- Downloads from another run (`run-id` with `github-token`).
			82	+
			83	+	## Docker Hub images
			84	+
			85	+	The runner provides an anonymous, pull-only Docker Hub mirror on the same private
			86	+	channel. Each job's dockerd uses `http://127.0.0.1:9089` as a registry mirror.
			87	+	Other registries are not cached. There is no HTTPS interception, and the mirror
			88	+	needs no network policy rule. Guest credentials are never forwarded upstream. Private pulls
			89	+	fall back to Docker Hub directly, where dockerd can use the job's credentials.
			90	+
			91	+	Verified sha256 blobs and manifests live under `<state dir>/registry/`. Tags are
			92	+	revalidated by an upstream HEAD on each request, with the client's Accept headers
			93	+	reduced to the known manifest media types. If Docker Hub is unreachable, times
			94	+	out, returns a server error or rate-limits (429), a previously cached tag can be
			95	+	served stale while its manifest remains stored. Unknown tags return an error so
			96	+	dockerd can fall back.
			97	+
			98	+	A cold blob is downloaded once, whoever asks for it, and is not tied to any one
			99	+	client: every client waits for the digest check and then reads the stored file.
			100	+	A corrupt download returns 502 and leaves no file. A client that disconnects
			101	+	does not stop the download while another client waits for it; once no client
			102	+	has waited for 5 s, the download is cancelled and its temporary file removed.
			103	+	Each job may have started at most 3 cold blob downloads at once (dockerd's
			104	+	default concurrency), and at most 16 run at once across all jobs. A download
			105	+	counts against the job that started it until it ends; a job that asks for a
			106	+	layer another job is already downloading waits for it and uses none of its
			107	+	quota. A request over either limit waits up to 30 s for a slot, its connection
			108	+	kept alive, and then gets 503. A refused layer makes dockerd retry for about
			109	+	50 s before it pulls that layer from Docker Hub directly. Each sandbox may also
			110	+	have at most 8 registry requests in progress; one more gets 503 at once.
			111	+
			112	+	Downloads have no total time limit, but they fail after 30 s without progress.
			113	+	Auth, manifest and HEAD requests are limited to 10 s. Auth and manifest requests
			114	+	never follow redirects; blob requests, HEAD included, follow the CDN redirect
			115	+	without the bearer token. Stored blobs support range requests. Cold blob HEAD
			116	+	requests fetch only upstream metadata, not the layer body.
			117	+
			118	+	`ERMITE_RUNNER_REGISTRY_CACHE_MAX` defaults to `10G`. Blobs, manifests and tag
			119	+	files share one least-recently-used eviction; each is charged at least 4 KiB, so
			120	+	the number of files stays below the cap divided by 4 KiB. An entry that is being served is
			121	+	never evicted. This cap is independent of
			122	+	`ERMITE_RUNNER_CACHE_MAX`. A zero size in any unit (`0`, `0M`, `0G`) disables only
			123	+	the proxy. `ERMITE_RUNNER_CACHE=0` disables the whole channel and therefore also
			124	+	the proxy (logged at startup). Dockerd retains its mirror setting and falls back
			125	+	to Docker Hub if the mirror is unavailable. To clear the cache, stop the runner
			126	+	and remove `<state dir>/registry/`.
			127	+
			128	+	Jobs with neither cache permissions nor artifacts get the loopback channel for
			129	+	dockerd when the proxy is on, but no `ACTIONS_CACHE_URL`, `ACTIONS_RESULTS_URL`, or
			130	+	`ACTIONS_RUNTIME_TOKEN` is exposed to their steps. Registry routes need no token:
			131	+	only public content is served over a channel exclusive to that sandbox.
			132	+
			133	+	The host logs one `registry manifest` line per request with `job_id`, `repo`,
			134	+	`ref`, `hit` and `ms`. Blob misses log `registry blob` at debug level with
			135	+	`job_id`, `digest`, `bytes` and `ms`; `job_id` is the job that started the
			136	+	download.

M docs/configuration.md +2 -1


@@ -74,9 +74,10 @@ The token has the scope `runner:register`. It can read the code and the secrets
74	74
75	75		\| Variable \| Default \| Meaning \|
76	76		\| --- \| --- \| --- \|
77		-	\| `ERMITE_RUNNER_CACHE` \| `1` \| `0` turns the Actions cache off. See [cache and artifacts](cache-and-artifacts.md) \|
	77	+	\| `ERMITE_RUNNER_CACHE` \| `1` \| `0` turns the cache channel off, including the Actions cache and Docker Hub proxy. See [cache and artifacts](cache-and-artifacts.md) \|
78	78		\| `ERMITE_RUNNER_CACHE_MAX` \| `10G` \| Size cap of the cache. Least recently used entries are evicted \|
79	79		\| `ERMITE_RUNNER_ACTIONS_CACHE_MAX` \| `2G` \| Size cap of the action archive cache. `0` turns it off \|
	80	+	\| `ERMITE_RUNNER_REGISTRY_CACHE_MAX` \| `10G` \| Docker Hub blob and manifest cache cap, independent of `ERMITE_RUNNER_CACHE_MAX`. A zero size (`0`, `0M`, `0G`) turns only the proxy off; `ERMITE_RUNNER_CACHE=0` also disables it \|
80	81		\| `ERMITE_RUNNER_ARTIFACTS` \| `1` \| `0` turns artifacts v4 off. Artifacts need the cache service \|
81	82		\| `ERMITE_RUNNER_NODE` \| `1` \| `0` turns the Node.js externals off. See [externals](externals.md) \|
82	83		\| `ERMITE_RUNNER_GUEST_FORGE` \| `0` \| `1` lets sandboxes reach the server host and port. Read the security notes first. See [sandboxes](sandboxes.md#guest-forge-access) \|

 | Variable | Default | Meaning |
 | --- | --- | --- |
-| `ERMITE_RUNNER_CACHE` | `1` | `0` turns the Actions cache off. See [cache and artifacts](cache-and-artifacts.md) |
+| `ERMITE_RUNNER_CACHE` | `1` | `0` turns the cache channel off, including the Actions cache and Docker Hub proxy. See [cache and artifacts](cache-and-artifacts.md) |
 | `ERMITE_RUNNER_CACHE_MAX` | `10G` | Size cap of the cache. Least recently used entries are evicted |
 | `ERMITE_RUNNER_ACTIONS_CACHE_MAX` | `2G` | Size cap of the action archive cache. `0` turns it off |
+| `ERMITE_RUNNER_REGISTRY_CACHE_MAX` | `10G` | Docker Hub blob and manifest cache cap, independent of `ERMITE_RUNNER_CACHE_MAX`. A zero size (`0`, `0M`, `0G`) turns only the proxy off; `ERMITE_RUNNER_CACHE=0` also disables it |
 | `ERMITE_RUNNER_ARTIFACTS` | `1` | `0` turns artifacts v4 off. Artifacts need the cache service |
 | `ERMITE_RUNNER_NODE` | `1` | `0` turns the Node.js externals off. See [externals](externals.md) |
 | `ERMITE_RUNNER_GUEST_FORGE` | `0` | `1` lets sandboxes reach the server host and port. Read the security notes first. See [sandboxes](sandboxes.md#guest-forge-access) |

M docs/development.md +26 -0


@@ -23,6 +23,7 @@ They are opt-in. They run the real stub executor. Output and exit codes come bac
23	23		\| `ERMITE_RUNNER_IT_E2E=1` \| End-to-end job tests in `internal/e2e` \|
24	24		\| `ERMITE_RUNNER_IT_NET=1` \| Packet-level firewall test \|
25	25		\| `ERMITE_RUNNER_IT_DOCKER_API=1` \| Docker Engine calls against any daemon, including rootless \|
	26	+	\| `ERMITE_RUNNER_IT_REGISTRY=1` \| Live Docker Hub pulls through the registry proxy with `curl` (`TestDockerHubCurl`). Needs internet access \|
26	27		\| `ERMITE_RUNNER_IT_NODE_DIR` \| Directory with `node-v24.21.0-linux-x64.tar.gz`, for offline hosts \|
27	28		\| `ERMITE_RUNNER_TEST_IMAGE` \| Image used by the container and end-to-end tests \|
28	29

@@ -63,3 +64,28 @@ ERMITE_RUNNER_IT_E2E=1 ERMITE_RUNNER_TEST_IMAGE=ermite-runner-ubuntu-26.04:dev G
63	64		- Docker API. Production `container` mode supports Docker Engine and sysbox. Podman's compatibility API is not a drop-in: it cannot disable DNS on the network that the runner creates.
64	65		- Jailer. The jailer needs `--chroot-base-dir` to exist. The driver creates `<state dir>/jail`. It leaves `<state dir>/jail/<exec name>` and `/sys/fs/cgroup/firecracker` behind.
65	66		- Host firewalls. A firewall that rejects `53` on `ertap` or `erbr` breaks guest DNS. The symptom is `dns silent` in `TestVMNetworkPolicy`.
	67	+
	68	+	## Measuring start-up on a small host
	69	+
	70	+	Build and register the [runner container](container.md) as usual, then constrain
	71	+	its CPU budget to reproduce a small host without changing the boot sequence:
	72	+
	73	+	```sh
	74	+	docker run -d --name runner-timing --cpus=1 \
	75	+	--device /dev/kvm --device /dev/net/tun --cap-add NET_ADMIN \
	76	+	-v runner-timing-state:/var/lib/ermite-runner \
	77	+	--env-file runner.env -e ERMITE_RUNNER_LOG_FORMAT=json ermite-runner:<tag>
	78	+	```
	79	+
	80	+	Use a registered state volume or supply the registration settings in
	81	+	`runner.env`; never put tokens in shared shell history. Podman accepts the same
	82	+	CPU limit (device access may need the CDI setup described in the container
	83	+	guide). Submit representative jobs, then extract completed-job measurements:
	84	+
	85	+	```sh
	86	+	docker logs runner-timing 2>&1 \| jq 'select(.message=="job finished")'
	87	+	```
	88	+
	89	+	Compare `sandbox_ms`, the `sb_` stages and `guest_` boot stages across jobs.
	90	+	`run_ms` includes action downloads and step execution; archive misses have their
	91	+	own counters. See [Job timing](service.md#job-timing) for the full field table.


@@ -23,6 +23,7 @@ They are opt-in. They run the real stub executor. Output and exit codes come bac
23	\| `ERMITE_RUNNER_IT_E2E=1` \| End-to-end job tests in `internal/e2e` \|	23		\| `ERMITE_RUNNER_IT_E2E=1` \| End-to-end job tests in `internal/e2e` \|
24	\| `ERMITE_RUNNER_IT_NET=1` \| Packet-level firewall test \|	24		\| `ERMITE_RUNNER_IT_NET=1` \| Packet-level firewall test \|
25	\| `ERMITE_RUNNER_IT_DOCKER_API=1` \| Docker Engine calls against any daemon, including rootless \|	25		\| `ERMITE_RUNNER_IT_DOCKER_API=1` \| Docker Engine calls against any daemon, including rootless \|
		26	+	\| `ERMITE_RUNNER_IT_REGISTRY=1` \| Live Docker Hub pulls through the registry proxy with `curl` (`TestDockerHubCurl`). Needs internet access \|
26	\| `ERMITE_RUNNER_IT_NODE_DIR` \| Directory with `node-v24.21.0-linux-x64.tar.gz`, for offline hosts \|	27		\| `ERMITE_RUNNER_IT_NODE_DIR` \| Directory with `node-v24.21.0-linux-x64.tar.gz`, for offline hosts \|
27	\| `ERMITE_RUNNER_TEST_IMAGE` \| Image used by the container and end-to-end tests \|	28		\| `ERMITE_RUNNER_TEST_IMAGE` \| Image used by the container and end-to-end tests \|
28		29

@@ -63,3 +64,28 @@ ERMITE_RUNNER_IT_E2E=1 ERMITE_RUNNER_TEST_IMAGE=ermite-runner-ubuntu-26.04:dev G
63	- Docker API. Production `container` mode supports Docker Engine and sysbox. Podman's compatibility API is not a drop-in: it cannot disable DNS on the network that the runner creates.	64		- Docker API. Production `container` mode supports Docker Engine and sysbox. Podman's compatibility API is not a drop-in: it cannot disable DNS on the network that the runner creates.
64	- Jailer. The jailer needs `--chroot-base-dir` to exist. The driver creates `<state dir>/jail`. It leaves `<state dir>/jail/<exec name>` and `/sys/fs/cgroup/firecracker` behind.	65		- Jailer. The jailer needs `--chroot-base-dir` to exist. The driver creates `<state dir>/jail`. It leaves `<state dir>/jail/<exec name>` and `/sys/fs/cgroup/firecracker` behind.
65	- Host firewalls. A firewall that rejects `53` on `ertap` or `erbr` breaks guest DNS. The symptom is `dns silent` in `TestVMNetworkPolicy`.	66		- Host firewalls. A firewall that rejects `53` on `ertap` or `erbr` breaks guest DNS. The symptom is `dns silent` in `TestVMNetworkPolicy`.
		67	+
		68	+	## Measuring start-up on a small host
		69	+
		70	+	Build and register the [runner container](container.md) as usual, then constrain
		71	+	its CPU budget to reproduce a small host without changing the boot sequence:
		72	+
		73	+	```sh
		74	+	docker run -d --name runner-timing --cpus=1 \
		75	+	--device /dev/kvm --device /dev/net/tun --cap-add NET_ADMIN \
		76	+	-v runner-timing-state:/var/lib/ermite-runner \
		77	+	--env-file runner.env -e ERMITE_RUNNER_LOG_FORMAT=json ermite-runner:<tag>
		78	+	```
		79	+
		80	+	Use a registered state volume or supply the registration settings in
		81	+	`runner.env`; never put tokens in shared shell history. Podman accepts the same
		82	+	CPU limit (device access may need the CDI setup described in the container
		83	+	guide). Submit representative jobs, then extract completed-job measurements:
		84	+
		85	+	```sh
		86	+	docker logs runner-timing 2>&1 \| jq 'select(.message=="job finished")'
		87	+	```
		88	+
		89	+	Compare `sandbox_ms`, the `sb_` stages and `guest_` boot stages across jobs.
		90	+	`run_ms` includes action downloads and step execution; archive misses have their
		91	+	own counters. See [Job timing](service.md#job-timing) for the full field table.

M docs/images.md +17 -2


@@ -7,7 +7,7 @@ A job runs in the image named by `ERMITE_RUNNER_IMAGE`.
7	7		\| `container` \| OCI reference \| Default `ermite-runner-ubuntu-26.04:latest` \|
8	8		\| `vm` \| Directory \| `vmlinux` (`Image` on arm64) and `base.ext4`. Default `<state dir>/image` \|
9	9
10		-	Both come from `images/ubuntu-26.04/Dockerfile` (Ubuntu 26.04 LTS). The runner executable and Node.js are attached by the host. They are never part of the image.
	10	+	Both come from `images/ubuntu-26.04/Dockerfile` (Ubuntu 26.04 LTS). The runner executable and action runtime Node.js are attached by the host. The image also includes system Node.js and the cached toolchains listed below.
11	11
12	12		## Build
13	13

@@ -26,9 +26,24 @@ scripts/build-image.sh tmp/image # builds the OCI image and tmp/image/base.ex
26	26		## What the image must contain
27	27
28	28		- User `runner`, uid and gid 1001, home `/home/runner`. Steps run as this user. Without it, steps run as root (test images only).
29		-	- `/home/runner/work` and `/opt/hostedtoolcache`, owned by `runner`.
	29	+	- `/home/runner/work` and `/opt/hostedtoolcache`, owned by `runner`. The tool cache holds the toolchains listed in `images/ubuntu-26.04/toolcache.lock` for the image architecture.
30	30		- `bash` and `sh`. `python3` and a `python` command for `shell: python`.
31	31		- `dockerd` and passwordless `sudo` for `runner`, for jobs that use Docker.
	32	+	- No `registry-mirrors` in `/etc/docker/daemon.json`. The runner starts dockerd with `--registry-mirror` (its Docker Hub mirror), and dockerd refuses to start when an option is set both as a flag and in that file.
32	33		- Common tools: git, curl, jq, build-essential, ca-certificates, tzdata, zip and unzip.
33	34
34	35		`pwsh` is not in the image. A step that asks for it fails with `pwsh: command not found`.
	36	+
	37	+	## Toolchains
	38	+
	39	+	`images/ubuntu-26.04/toolcache.lock` pins Go and Node.js archives by version,
	40	+	action architecture (`x64` or `arm64`), official SHA-256 and download URL.
	41	+	To add a version, add a line for each supported architecture with the checksum
	42	+	from go.dev's download JSON or nodejs.org's `SHASUMS256.txt`, then rebuild with
	43	+	`scripts/build-image.sh tmp/image`. A checksum mismatch fails the build.
	44	+
	45	+	The installer uses the `@actions/tool-cache` layout, including the sibling
	46	+	`<arch>.complete` marker. `actions/setup-go` with an exact version, including
	47	+	one read from `go.mod`, uses the cache only for an exact match. A range such as
	48	+	`1.27` selects the highest matching cached version. Versions not in the cache
	49	+	still download normally (and `check-latest: true` may check upstream).

 | `container` | OCI reference | Default `ermite-runner-ubuntu-26.04:latest` |
 | `vm` | Directory | `vmlinux` (`Image` on arm64) and `base.ext4`. Default `<state dir>/image` |
-Both come from `images/ubuntu-26.04/Dockerfile` (Ubuntu 26.04 LTS). The runner executable and Node.js are attached by the host. They are never part of the image.
+Both come from `images/ubuntu-26.04/Dockerfile` (Ubuntu 26.04 LTS). The runner executable and action runtime Node.js are attached by the host. The image also includes system Node.js and the cached toolchains listed below.
 ## Build
 ## What the image must contain
 - User `runner`, uid and gid 1001, home `/home/runner`. Steps run as this user. Without it, steps run as root (test images only).
-- `/home/runner/work` and `/opt/hostedtoolcache`, owned by `runner`.
+- `/home/runner/work` and `/opt/hostedtoolcache`, owned by `runner`. The tool cache holds the toolchains listed in `images/ubuntu-26.04/toolcache.lock` for the image architecture.
 - `bash` and `sh`. `python3` and a `python` command for `shell: python`.
 - `dockerd` and passwordless `sudo` for `runner`, for jobs that use Docker.
+- No `registry-mirrors` in `/etc/docker/daemon.json`. The runner starts dockerd with `--registry-mirror` (its Docker Hub mirror), and dockerd refuses to start when an option is set both as a flag and in that file.
 - Common tools: git, curl, jq, build-essential, ca-certificates, tzdata, zip and unzip.
 `pwsh` is not in the image. A step that asks for it fails with `pwsh: command not found`.
+## Toolchains
+`images/ubuntu-26.04/toolcache.lock` pins Go and Node.js archives by version,
+action architecture (`x64` or `arm64`), official SHA-256 and download URL.
+To add a version, add a line for each supported architecture with the checksum
+from go.dev's download JSON or nodejs.org's `SHASUMS256.txt`, then rebuild with
+`scripts/build-image.sh tmp/image`. A checksum mismatch fails the build.
+The installer uses the `@actions/tool-cache` layout, including the sibling
+`<arch>.complete` marker. `actions/setup-go` with an exact version, including
+one read from `go.mod`, uses the cache only for an exact match. A range such as
+`1.27` selects the highest matching cached version. Versions not in the cache
+still download normally (and `check-latest: true` may check upstream).

M docs/sandboxes.md +0 -1


@@ -78,7 +78,6 @@ From a sandbox:
78	78
79	79		- Source address and destination checks run before any "established" accept. A guest cannot forge another's flow.
80	80		- The DNS forwarder allows 32 in-flight queries per listener (extra UDP is dropped), 16 TCP connections, and 30 s per TCP connection.
81		-	- Upstream queries use UDP for 1 s, then TCP on the same resolver (also when UDP is truncated). One query takes at most 6 s. A UDP answer larger than the client accepts (512 bytes without EDNS) is replaced by an empty reply with TC=1, so the client retries over TCP.
82	81		- Golden rulesets: `internal/netpolicy/testdata`.
83	82
84	83		Host side effects:


@@ -78,7 +78,6 @@ From a sandbox:
78			78
79		- Source address and destination checks run before any "established" accept. A guest cannot forge another's flow.	79	- Source address and destination checks run before any "established" accept. A guest cannot forge another's flow.
80		- The DNS forwarder allows 32 in-flight queries per listener (extra UDP is dropped), 16 TCP connections, and 30 s per TCP connection.	80	- The DNS forwarder allows 32 in-flight queries per listener (extra UDP is dropped), 16 TCP connections, and 30 s per TCP connection.
81	-	- Upstream queries use UDP for 1 s, then TCP on the same resolver (also when UDP is truncated). One query takes at most 6 s. A UDP answer larger than the client accepts (512 bytes without EDNS) is replaced by an empty reply with TC=1, so the client retries over TCP.
82		- Golden rulesets: `internal/netpolicy/testdata`.	81	- Golden rulesets: `internal/netpolicy/testdata`.
83			82
84		Host side effects:	83	Host side effects:

M docs/service.md +31 -0


@@ -64,3 +64,34 @@ Do not add:
64	64		- `Delegate` and `UMask`: not needed.
65	65
66	66		The security boundary of a job is its microVM or container, not the unit.
	67	+
	68	+	## Job timing
	69	+
	70	+	The `job finished` line carries integer milliseconds (all `*_ms` fields).
	71	+	Phases that did not run are omitted, and the driver (`sb_*`) timings appear only
	72	+	when the sandbox started. `resolve_ms` includes `archive_miss_ms`, and both are
	73	+	part of `run_ms`.
	74	+
	75	+	\| Fields \| Meaning \|
	76	+	\| --- \| --- \|
	77	+	\| `ack_ms` \| Assignment received to successful first heartbeat acknowledgement \|
	78	+	\| `sandbox_ms` \| Host `startSandbox` call \|
	79	+	\| `run_ms` \| Executor session (`drive`) \|
	80	+	\| `stop_ms` \| Sandbox `Stop` call \|
	81	+	\| `complete_ms` \| Completion call, including retries \|
	82	+	\| `total_ms` \| Assignment received to this log line, including log flushing \|
	83	+	\| `resolve_calls`, `resolve_ms` \| Number and summed duration of action resolution requests \|
	84	+	\| `archive_misses`, `archive_miss_ms`, `archive_miss_bytes` \| Archive misses and their summed fetch duration and bytes \|
	85	+	\| `sb_tap_ms`, `sb_net_ms` (VM) \| TAP creation and the rest of network setup \|
	86	+	\| `sb_vmm_ms`, `sb_files_ms`, `sb_scratch_ms` (VM) \| VMM launch/API readiness, shared files excluding scratch, scratch creation \|
	87	+	\| `sb_vsock_ms`, `sb_configure_ms`, `sb_connect_ms` (VM) \| Executor/cache sockets, VM configuration/start, executor connection wait \|
	88	+	\| `sb_socket_ms`, `sb_cache_ms`, `sb_scratch_ms` (container) \| Executor socket, cache socket, scratch setup \|
	89	+	\| `sb_net_ms`, `sb_container_ms`, `sb_connect_ms` (container) \| Network setup, container creation/start, executor connection wait \|
	90	+	\| `guest_kernel_ms`, `guest_setup_ms` (VM only) \| Kernel uptime at init entry, VM filesystem/system setup \|
	91	+	\| `guest_containerd_ms`, `guest_dockerd_ms` \| Containerd start/readiness, dockerd spawn through readiness wait \|
	92	+
	93	+	Guest boot overlaps the driver's executor connection wait: do not sum those
	94	+	fields with `sandbox_ms`. Driver `container running` and `micro-VM running`
	95	+	lines also carry unprefixed phase fields and `total_ms` for their boot sequence.
	96	+	The `action archive cache` line carries `verify_ms` for integrity verification;
	97	+	a miss also carries `fetch_ms` and `bytes`. Archive verification is never skipped.


@@ -64,3 +64,34 @@ Do not add:
64	- `Delegate` and `UMask`: not needed.	64		- `Delegate` and `UMask`: not needed.
65		65
66	The security boundary of a job is its microVM or container, not the unit.	66		The security boundary of a job is its microVM or container, not the unit.
		67	+
		68	+	## Job timing
		69	+
		70	+	The `job finished` line carries integer milliseconds (all `*_ms` fields).
		71	+	Phases that did not run are omitted, and the driver (`sb_*`) timings appear only
		72	+	when the sandbox started. `resolve_ms` includes `archive_miss_ms`, and both are
		73	+	part of `run_ms`.
		74	+
		75	+	\| Fields \| Meaning \|
		76	+	\| --- \| --- \|
		77	+	\| `ack_ms` \| Assignment received to successful first heartbeat acknowledgement \|
		78	+	\| `sandbox_ms` \| Host `startSandbox` call \|
		79	+	\| `run_ms` \| Executor session (`drive`) \|
		80	+	\| `stop_ms` \| Sandbox `Stop` call \|
		81	+	\| `complete_ms` \| Completion call, including retries \|
		82	+	\| `total_ms` \| Assignment received to this log line, including log flushing \|
		83	+	\| `resolve_calls`, `resolve_ms` \| Number and summed duration of action resolution requests \|
		84	+	\| `archive_misses`, `archive_miss_ms`, `archive_miss_bytes` \| Archive misses and their summed fetch duration and bytes \|
		85	+	\| `sb_tap_ms`, `sb_net_ms` (VM) \| TAP creation and the rest of network setup \|
		86	+	\| `sb_vmm_ms`, `sb_files_ms`, `sb_scratch_ms` (VM) \| VMM launch/API readiness, shared files excluding scratch, scratch creation \|
		87	+	\| `sb_vsock_ms`, `sb_configure_ms`, `sb_connect_ms` (VM) \| Executor/cache sockets, VM configuration/start, executor connection wait \|
		88	+	\| `sb_socket_ms`, `sb_cache_ms`, `sb_scratch_ms` (container) \| Executor socket, cache socket, scratch setup \|
		89	+	\| `sb_net_ms`, `sb_container_ms`, `sb_connect_ms` (container) \| Network setup, container creation/start, executor connection wait \|
		90	+	\| `guest_kernel_ms`, `guest_setup_ms` (VM only) \| Kernel uptime at init entry, VM filesystem/system setup \|
		91	+	\| `guest_containerd_ms`, `guest_dockerd_ms` \| Containerd start/readiness, dockerd spawn through readiness wait \|
		92	+
		93	+	Guest boot overlaps the driver's executor connection wait: do not sum those
		94	+	fields with `sandbox_ms`. Driver `container running` and `micro-VM running`
		95	+	lines also carry unprefixed phase fields and `total_ms` for their boot sequence.
		96	+	The `action archive cache` line carries `verify_ms` for integrity verification;
		97	+	a miss also carries `fetch_ms` and `bytes`. Archive verification is never skipped.

M host_cmd.go +42 -6


@@ -8,6 +8,7 @@ import (
8	8		"flag"
9	9		"fmt"
10	10		"net"
	11	+	"net/http"
11	12		"os"
12	13		"path/filepath"
13	14		"strings"

@@ -21,6 +22,7 @@ import (
21	22		"bnema.dev/ermite-runner/internal/cachesvc"
22	23		"bnema.dev/ermite-runner/internal/externals"
23	24		"bnema.dev/ermite-runner/internal/host"
	25	+	"bnema.dev/ermite-runner/internal/registryproxy"
24	26		"bnema.dev/ermite-runner/internal/resultssvc"
25	27		"bnema.dev/ermite-runner/internal/sandbox"
26	28		"bnema.dev/ermite-runner/internal/sandbox/drivers"

@@ -46,20 +48,28 @@ func setupCache(ctx context.Context, scfg sandbox.Config) (cachesvc.Service, *
46	48		switch strings.ToLower(strings.TrimSpace(os.Getenv(envCache))) {
47	49		case "", "1", "true", "yes":
48	50		case "0", "false", "no":
49		-	log.Info().Msg("the cache service is off (" + envCache + "=0)")
	51	+	log.Info().Msg("the cache service and registry proxy are off (" + envCache + "=0)")
50	52		return nil, nil, nil
51	53		default:
52	54		return nil, nil, fmt.Errorf("%s: %q is not 0 or 1", envCache, os.Getenv(envCache))
53	55		}
54		-	max, err := cachesvc.ParseMax(os.Getenv(cachesvc.EnvMax))
	56	+	// The cache API cannot be turned off by its size: 0 is refused.
	57	+	max, err := cachesvc.ParseSize(os.Getenv(cachesvc.EnvMax), cachesvc.DefaultMax)
	58	+	if err == nil && max == 0 {
	59	+	err = errors.New("0 is not a size cap; " + envCache + "=0 turns the cache off")
	60	+	}
55	61		if err != nil {
56		-	return nil, nil, err
	62	+	return nil, nil, fmt.Errorf("%s: %w", cachesvc.EnvMax, err)
57	63		}
58	64		archives, err := setupArchives(ctx, scfg.StateDir)
59	65		if err != nil {
60	66		return nil, nil, err
61	67		}
62		-	svc, err := cachesvc.New(cachesvc.Config{Dir: filepath.Join(scfg.StateDir, "cache"), Max: max, Log: log, Archives: archives})
	68	+	registry, err := setupRegistry(ctx, scfg.StateDir)
	69	+	if err != nil {
	70	+	return nil, nil, err
	71	+	}
	72	+	svc, err := cachesvc.New(cachesvc.Config{Dir: filepath.Join(scfg.StateDir, "cache"), Max: max, Log: log, Archives: archives, Registry: registry})
63	73		if err != nil {
64	74		return nil, nil, err
65	75		}

@@ -73,9 +83,9 @@ func setupCache(ctx context.Context, scfg sandbox.Config) (cachesvc.Service, *
73	83		// service, so it is only used when that is on.
74	84		func setupArchives(ctx context.Context, stateDir string) (*actionscache.Cache, error) {
75	85		log := zerowrap.FromCtx(ctx)
76		-	max, err := actionscache.ParseMax(os.Getenv(actionscache.EnvMax))
	86	+	max, err := cachesvc.ParseSize(os.Getenv(actionscache.EnvMax), actionscache.DefaultMax)
77	87		if err != nil {
78		-	return nil, err
	88	+	return nil, fmt.Errorf("%s: %w", actionscache.EnvMax, err)
79	89		}
80	90		if max == 0 {
81	91		log.Info().Msg("the action archive cache is off (" + actionscache.EnvMax + "=0)")

@@ -89,6 +99,32 @@ func setupArchives(ctx context.Context, stateDir string) (*actionscache.Cache, e
89	99		return c, nil
90	100		}
91	101
	102	+	// envRegistryMax is the size cap of the Docker Hub mirror (default
	103	+	// defaultRegistryMax); 0 turns it off.
	104	+	const envRegistryMax = "ERMITE_RUNNER_REGISTRY_CACHE_MAX"
	105	+
	106	+	const defaultRegistryMax = 10 << 30
	107	+
	108	+	// setupRegistry opens the Docker Hub mirror. The caller only calls this when
	109	+	// the cache channel is enabled; its capacity is independent of the cache API.
	110	+	func setupRegistry(ctx context.Context, stateDir string) (http.Handler, error) {
	111	+	log := zerowrap.FromCtx(ctx)
	112	+	max, err := cachesvc.ParseSize(os.Getenv(envRegistryMax), defaultRegistryMax)
	113	+	if err != nil {
	114	+	return nil, fmt.Errorf("%s: %w", envRegistryMax, err)
	115	+	}
	116	+	if max == 0 {
	117	+	log.Info().Msg("registry proxy off (" + envRegistryMax + "=0)")
	118	+	return nil, nil
	119	+	}
	120	+	p, err := registryproxy.New(registryproxy.Config{Dir: filepath.Join(stateDir, "registry"), Max: max, Log: log})
	121	+	if err != nil {
	122	+	return nil, err
	123	+	}
	124	+	log.Info().Int64("max_bytes", max).Msg("registry proxy on (Docker Hub pull-through)")
	125	+	return p, nil
	126	+	}
	127	+
92	128		// envCache=0 turns the cache service off.
93	129		const envCache = "ERMITE_RUNNER_CACHE"
94	130

 	"flag"
 	"fmt"
 	"net"
+	"net/http"
 	"os"
 	"path/filepath"
 	"strings"
 	"bnema.dev/ermite-runner/internal/cachesvc"
 	"bnema.dev/ermite-runner/internal/externals"
 	"bnema.dev/ermite-runner/internal/host"
+	"bnema.dev/ermite-runner/internal/registryproxy"
 	"bnema.dev/ermite-runner/internal/resultssvc"
 	"bnema.dev/ermite-runner/internal/sandbox"
 	"bnema.dev/ermite-runner/internal/sandbox/drivers"
 	switch strings.ToLower(strings.TrimSpace(os.Getenv(envCache))) {
 	case "", "1", "true", "yes":
 	case "0", "false", "no":
-		log.Info().Msg("the cache service is off (" + envCache + "=0)")
+		log.Info().Msg("the cache service and registry proxy are off (" + envCache + "=0)")
 		return nil, nil, nil
 	default:
 		return nil, nil, fmt.Errorf("%s: %q is not 0 or 1", envCache, os.Getenv(envCache))
 	}
-	max, err := cachesvc.ParseMax(os.Getenv(cachesvc.EnvMax))
+	// The cache API cannot be turned off by its size: 0 is refused.
+	max, err := cachesvc.ParseSize(os.Getenv(cachesvc.EnvMax), cachesvc.DefaultMax)
+	if err == nil && max == 0 {
+		err = errors.New("0 is not a size cap; " + envCache + "=0 turns the cache off")
+	}
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, fmt.Errorf("%s: %w", cachesvc.EnvMax, err)
 	}
 	archives, err := setupArchives(ctx, scfg.StateDir)
 	if err != nil {
 		return nil, nil, err
 	}
-	svc, err := cachesvc.New(cachesvc.Config{Dir: filepath.Join(scfg.StateDir, "cache"), Max: max, Log: log, Archives: archives})
+	registry, err := setupRegistry(ctx, scfg.StateDir)
+	if err != nil {
+		return nil, nil, err
+	}
+	svc, err := cachesvc.New(cachesvc.Config{Dir: filepath.Join(scfg.StateDir, "cache"), Max: max, Log: log, Archives: archives, Registry: registry})
 	if err != nil {
 		return nil, nil, err
 	}
 // service, so it is only used when that is on.
 func setupArchives(ctx context.Context, stateDir string) (*actionscache.Cache, error) {
 	log := zerowrap.FromCtx(ctx)
-	max, err := actionscache.ParseMax(os.Getenv(actionscache.EnvMax))
+	max, err := cachesvc.ParseSize(os.Getenv(actionscache.EnvMax), actionscache.DefaultMax)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("%s: %w", actionscache.EnvMax, err)
 	}
 	if max == 0 {
 		log.Info().Msg("the action archive cache is off (" + actionscache.EnvMax + "=0)")
 	return c, nil
 }
+// envRegistryMax is the size cap of the Docker Hub mirror (default
+// defaultRegistryMax); 0 turns it off.
+const envRegistryMax = "ERMITE_RUNNER_REGISTRY_CACHE_MAX"
+const defaultRegistryMax = 10 << 30
+// setupRegistry opens the Docker Hub mirror. The caller only calls this when
+// the cache channel is enabled; its capacity is independent of the cache API.
+func setupRegistry(ctx context.Context, stateDir string) (http.Handler, error) {
+	log := zerowrap.FromCtx(ctx)
+	max, err := cachesvc.ParseSize(os.Getenv(envRegistryMax), defaultRegistryMax)
+	if err != nil {
+		return nil, fmt.Errorf("%s: %w", envRegistryMax, err)
+	}
+	if max == 0 {
+		log.Info().Msg("registry proxy off (" + envRegistryMax + "=0)")
+		return nil, nil
+	}
+	p, err := registryproxy.New(registryproxy.Config{Dir: filepath.Join(stateDir, "registry"), Max: max, Log: log})
+	if err != nil {
+		return nil, err
+	}
+	log.Info().Int64("max_bytes", max).Msg("registry proxy on (Docker Hub pull-through)")
+	return p, nil
+}
 // envCache=0 turns the cache service off.
 const envCache = "ERMITE_RUNNER_CACHE"

M images/ubuntu-26.04/Dockerfile +4 -0


@@ -78,6 +78,10 @@ RUN useradd --create-home --uid 1001 --shell /bin/bash runner \
78	78		ENV RUNNER_TOOL_CACHE=/opt/hostedtoolcache \
79	79		LANG=C.UTF-8
80	80
	81	+	COPY toolcache.lock install-toolcache.sh /tmp/toolcache/
	82	+	RUN sh /tmp/toolcache/install-toolcache.sh /tmp/toolcache/toolcache.lock \
	83	+	&& rm -rf /tmp/toolcache
	84	+
81	85		# The runner executable is bind-mounted (container) or provided by the init
82	86		# disk (vm) here; its init subcommand starts the job's own Docker daemon.
83	87		ENTRYPOINT ["/opt/ermite/ermite-runner", "init"]


@@ -78,6 +78,10 @@ RUN useradd --create-home --uid 1001 --shell /bin/bash runner \
78	ENV RUNNER_TOOL_CACHE=/opt/hostedtoolcache \	78		ENV RUNNER_TOOL_CACHE=/opt/hostedtoolcache \
79	LANG=C.UTF-8	79		LANG=C.UTF-8
80		80
		81	+	COPY toolcache.lock install-toolcache.sh /tmp/toolcache/
		82	+	RUN sh /tmp/toolcache/install-toolcache.sh /tmp/toolcache/toolcache.lock \
		83	+	&& rm -rf /tmp/toolcache
		84	+
81	# The runner executable is bind-mounted (container) or provided by the init	85		# The runner executable is bind-mounted (container) or provided by the init
82	# disk (vm) here; its init subcommand starts the job's own Docker daemon.	86		# disk (vm) here; its init subcommand starts the job's own Docker daemon.
83	ENTRYPOINT ["/opt/ermite/ermite-runner", "init"]	87		ENTRYPOINT ["/opt/ermite/ermite-runner", "init"]

A images/ubuntu-26.04/install-toolcache.sh +42 -0

@@ -0,0 +1,42 @@
	1	+	#!/bin/sh
	2	+	# SPDX-License-Identifier: MIT
	3	+	# Installs the toolchains of a lock file (tool version arch sha256 url per
	4	+	# line) for the image architecture into the @actions/tool-cache layout.
	5	+	set -eu
	6	+
	7	+	lock=${1:?usage: install-toolcache.sh LOCK_FILE}
	8	+	case "$(dpkg --print-architecture)" in
	9	+	amd64) arch=x64 ;;
	10	+	arm64) arch=arm64 ;;
	11	+	*) echo 'unsupported toolcache architecture' >&2; exit 1 ;;
	12	+	esac
	13	+	cache=${RUNNER_TOOL_CACHE:-/opt/hostedtoolcache}
	14	+	tmp=$(mktemp -d)
	15	+	trap 'rm -rf "$tmp"' EXIT
	16	+	trap 'exit 130' INT TERM
	17	+	installed=0
	18	+	# The last line is read even without a final newline.
	19	+	while read -r tool version entry_arch sha url \|\| [ -n "$tool" ]; do
	20	+	case "$tool" in ''\|'#'*) continue ;; esac
	21	+	case "$tool" in go\|node) ;; *) echo "unsupported tool: $tool" >&2; exit 1 ;; esac
	22	+	case "$entry_arch" in x64\|arm64) ;; *) echo "unsupported arch for $tool $version: $entry_arch" >&2; exit 1 ;; esac
	23	+	[ "$entry_arch" = "$arch" ] \|\| continue
	24	+	case "$url" in
	25	+	*.tar.gz) compression=z ;;
	26	+	*.tar.xz) compression=J ;;
	27	+	*) echo "unsupported archive: $url" >&2; exit 1 ;;
	28	+	esac
	29	+	curl -fsSL --retry 3 "$url" -o "$tmp/archive"
	30	+	printf '%s %s\n' "$sha" "$tmp/archive" \| sha256sum -c -
	31	+	dest="$cache/$tool/$version/$arch"
	32	+	mkdir -p "$dest"
	33	+	tar -x"$compression"f "$tmp/archive" -C "$dest" --strip-components=1 --no-same-owner
	34	+	touch "$dest.complete"
	35	+	rm "$tmp/archive"
	36	+	installed=$((installed + 1))
	37	+	done < "$lock"
	38	+	if [ "$installed" -eq 0 ]; then
	39	+	echo "no toolchain for $arch in $lock" >&2
	40	+	exit 1
	41	+	fi
	42	+	chown -R runner:runner "$cache"

@@ -0,0 +1,42 @@
	1	+	#!/bin/sh
	2	+	# SPDX-License-Identifier: MIT
	3	+	# Installs the toolchains of a lock file (tool version arch sha256 url per
	4	+	# line) for the image architecture into the @actions/tool-cache layout.
	5	+	set -eu
	6	+
	7	+	lock=${1:?usage: install-toolcache.sh LOCK_FILE}
	8	+	case "$(dpkg --print-architecture)" in
	9	+	amd64) arch=x64 ;;
	10	+	arm64) arch=arm64 ;;
	11	+	*) echo 'unsupported toolcache architecture' >&2; exit 1 ;;
	12	+	esac
	13	+	cache=${RUNNER_TOOL_CACHE:-/opt/hostedtoolcache}
	14	+	tmp=$(mktemp -d)
	15	+	trap 'rm -rf "$tmp"' EXIT
	16	+	trap 'exit 130' INT TERM
	17	+	installed=0
	18	+	# The last line is read even without a final newline.
	19	+	while read -r tool version entry_arch sha url \|\| [ -n "$tool" ]; do
	20	+	case "$tool" in ''\|'#'*) continue ;; esac
	21	+	case "$tool" in go\|node) ;; *) echo "unsupported tool: $tool" >&2; exit 1 ;; esac
	22	+	case "$entry_arch" in x64\|arm64) ;; *) echo "unsupported arch for $tool $version: $entry_arch" >&2; exit 1 ;; esac
	23	+	[ "$entry_arch" = "$arch" ] \|\| continue
	24	+	case "$url" in
	25	+	*.tar.gz) compression=z ;;
	26	+	*.tar.xz) compression=J ;;
	27	+	*) echo "unsupported archive: $url" >&2; exit 1 ;;
	28	+	esac
	29	+	curl -fsSL --retry 3 "$url" -o "$tmp/archive"
	30	+	printf '%s %s\n' "$sha" "$tmp/archive" \| sha256sum -c -
	31	+	dest="$cache/$tool/$version/$arch"
	32	+	mkdir -p "$dest"
	33	+	tar -x"$compression"f "$tmp/archive" -C "$dest" --strip-components=1 --no-same-owner
	34	+	touch "$dest.complete"
	35	+	rm "$tmp/archive"
	36	+	installed=$((installed + 1))
	37	+	done < "$lock"
	38	+	if [ "$installed" -eq 0 ]; then
	39	+	echo "no toolchain for $arch in $lock" >&2
	40	+	exit 1
	41	+	fi
	42	+	chown -R runner:runner "$cache"

A images/ubuntu-26.04/toolcache.lock +11 -0

@@ -0,0 +1,11 @@
	1	+	# tool version arch sha256 url
	2	+	go 1.27.3 x64 65efbc80fdc1fa2cd2631cad5b481777a68e8725f34e89f50fc6ac714e379b40 https://go.dev/dl/go1.27.3.linux-amd64.tar.gz
	3	+	go 1.27.3 arm64 a06e99309dbcbdca1a8ccd2282c3ee146e55930886e26b4f1193269587165f49 https://go.dev/dl/go1.27.3.linux-arm64.tar.gz
	4	+	go 1.27.1 x64 63d339f0da5ab53635a56f2490a7984dfe12dfcff22ad749f63edaf590168445 https://go.dev/dl/go1.27.1.linux-amd64.tar.gz
	5	+	go 1.27.1 arm64 3450b45a3f9ee8568792736a5c5e70a1f2e9b36c35a8f74958c03e51d7d92bec https://go.dev/dl/go1.27.1.linux-arm64.tar.gz
	6	+	go 1.26.10 x64 ccfe7ae822d2c2f41f4fe7c956199967a4dac22d0ee0d2b2bfdb440aab3b20c5 https://go.dev/dl/go1.26.10.linux-amd64.tar.gz
	7	+	go 1.26.10 arm64 3f4227ce7b206a513e794f1393adb473a07b46e94d18bd15d1c684bbb7319797 https://go.dev/dl/go1.26.10.linux-arm64.tar.gz
	8	+	node 24.21.0 x64 fd8e59d5a511510f6a298afb548f18c7d2b1be404d8b4a27d94fbe49f56cb2d6 https://nodejs.org/dist/v24.21.0/node-v24.21.0-linux-x64.tar.xz
	9	+	node 24.21.0 arm64 6ad1325edbdb5649c379b75a237147a666c95d4f9ae8d340fef2d1575d289ad2 https://nodejs.org/dist/v24.21.0/node-v24.21.0-linux-arm64.tar.xz
	10	+	node 22.23.3 x64 df450af89261115ef9f9e3830c3eeb2cc9213b63c720b1af623cb5dcbe2e02de https://nodejs.org/dist/v22.23.3/node-v22.23.3-linux-x64.tar.xz
	11	+	node 22.23.3 arm64 a44aeb94849a299b22df10b9e622ec2f605c2183501bc40590705131de7c740f https://nodejs.org/dist/v22.23.3/node-v22.23.3-linux-arm64.tar.xz

@@ -0,0 +1,11 @@
	1	+	# tool version arch sha256 url
	2	+	go 1.27.3 x64 65efbc80fdc1fa2cd2631cad5b481777a68e8725f34e89f50fc6ac714e379b40 https://go.dev/dl/go1.27.3.linux-amd64.tar.gz
	3	+	go 1.27.3 arm64 a06e99309dbcbdca1a8ccd2282c3ee146e55930886e26b4f1193269587165f49 https://go.dev/dl/go1.27.3.linux-arm64.tar.gz
	4	+	go 1.27.1 x64 63d339f0da5ab53635a56f2490a7984dfe12dfcff22ad749f63edaf590168445 https://go.dev/dl/go1.27.1.linux-amd64.tar.gz
	5	+	go 1.27.1 arm64 3450b45a3f9ee8568792736a5c5e70a1f2e9b36c35a8f74958c03e51d7d92bec https://go.dev/dl/go1.27.1.linux-arm64.tar.gz
	6	+	go 1.26.10 x64 ccfe7ae822d2c2f41f4fe7c956199967a4dac22d0ee0d2b2bfdb440aab3b20c5 https://go.dev/dl/go1.26.10.linux-amd64.tar.gz
	7	+	go 1.26.10 arm64 3f4227ce7b206a513e794f1393adb473a07b46e94d18bd15d1c684bbb7319797 https://go.dev/dl/go1.26.10.linux-arm64.tar.gz
	8	+	node 24.21.0 x64 fd8e59d5a511510f6a298afb548f18c7d2b1be404d8b4a27d94fbe49f56cb2d6 https://nodejs.org/dist/v24.21.0/node-v24.21.0-linux-x64.tar.xz
	9	+	node 24.21.0 arm64 6ad1325edbdb5649c379b75a237147a666c95d4f9ae8d340fef2d1575d289ad2 https://nodejs.org/dist/v24.21.0/node-v24.21.0-linux-arm64.tar.xz
	10	+	node 22.23.3 x64 df450af89261115ef9f9e3830c3eeb2cc9213b63c720b1af623cb5dcbe2e02de https://nodejs.org/dist/v22.23.3/node-v22.23.3-linux-x64.tar.xz
	11	+	node 22.23.3 arm64 a44aeb94849a299b22df10b9e622ec2f605c2183501bc40590705131de7c740f https://nodejs.org/dist/v22.23.3/node-v22.23.3-linux-arm64.tar.xz

M internal/actionscache/cache.go +17 -43


@@ -184,14 +184,16 @@ type Fetch func(ctx context.Context, w io.Writer) error
184	184		// download it when it does not (a miss: hit is false). Concurrent calls for the
185	185		// same key share one download; the others report a hit once it is in place. The
186	186		// entry is pinned for PinFor.
187		-	func (c *Cache) Ensure(ctx context.Context, key string, fetch Fetch) (hit bool, err error) {
	187	+	func (c *Cache) Ensure(ctx context.Context, key string, fetch Fetch) (hit bool, verify time.Duration, err error) {
188	188		if !validKey(key) {
189		-	return false, ErrNotFound
	189	+	return false, verify, ErrNotFound
190	190		}
191	191		for {
192		-	if c.verify(key) == nil {
	192	+	duration, err := c.verify(key)
	193	+	verify += duration
	194	+	if err == nil {
193	195		c.touch(key, true)
194		-	return true, nil
	196	+	return true, verify, nil
195	197		}
196	198		c.mu.Lock()
197	199		if f, ok := c.flights[key]; ok {

@@ -199,10 +201,10 @@ func (c *Cache) Ensure(ctx context.Context, key string, fetch Fetch) (hit bool,
199	201		select {
200	202		case <-f.done:
201	203		case <-ctx.Done():
202		-	return false, ctx.Err()
	204	+	return false, verify, ctx.Err()
203	205		}
204	206		if f.err != nil {
205		-	return false, f.err
	207	+	return false, verify, f.err
206	208		}
207	209		continue
208	210		}

@@ -215,10 +217,10 @@ func (c *Cache) Ensure(ctx context.Context, key string, fetch Fetch) (hit bool,
215	217		c.mu.Unlock()
216	218		close(f.done)
217	219		if f.err != nil {
218		-	return false, f.err
	220	+	return false, verify, f.err
219	221		}
220	222		c.touch(key, true)
221		-	return false, nil
	223	+	return false, verify, nil
222	224		}
223	225		}
224	226

@@ -314,11 +316,12 @@ func (w *countWriter) Write(p []byte) (int, error) {
314	316
315	317		// verify re-hashes the archive of key against its sum. Any mismatch (or a
316	318		// missing file) drops the entry and returns ErrNotFound.
317		-	func (c *Cache) verify(key string) error {
	319	+	func (c *Cache) verify(key string) (time.Duration, error) {
	320	+	start := time.Now()
318	321		sb, err := os.ReadFile(c.sumPath(key))
319	322		if err != nil {
320	323		c.drop(key)
321		-	return ErrNotFound
	324	+	return time.Since(start), ErrNotFound
322	325		}
323	326		want, sizeS, _ := strings.Cut(strings.TrimSpace(string(sb)), " ")
324	327		size, perr := strconv.ParseInt(sizeS, 10, 64)

@@ -328,7 +331,7 @@ func (c *Cache) verify(key string) error {
328	331		_ = f.Close()
329	332		}
330	333		c.drop(key)
331		-	return ErrNotFound
	334	+	return time.Since(start), ErrNotFound
332	335		}
333	336		defer func() { _ = f.Close() }()
334	337		h := sha256.New()

@@ -336,9 +339,9 @@ func (c *Cache) verify(key string) error {
336	339		if err != nil \|\| n != size \|\| hex.EncodeToString(h.Sum(nil)) != want {
337	340		c.log.Warn().Str("key", key).Msg("action archive failed its integrity check; dropped")
338	341		c.drop(key)
339		-	return ErrNotFound
	342	+	return time.Since(start), ErrNotFound
340	343		}
341		-	return nil
	344	+	return time.Since(start), nil
342	345		}
343	346
344	347		func (c *Cache) drop(key string) {

@@ -396,7 +399,7 @@ func (c Cache) Open(key string) (os.File, error) {
396	399		if !validKey(key) {
397	400		return nil, ErrNotFound
398	401		}
399		-	if err := c.verify(key); err != nil {
	402	+	if _, err := c.verify(key); err != nil {
400	403		return nil, err
401	404		}
402	405		f, err := os.Open(c.archivePath(key))

@@ -415,34 +418,5 @@ func (c *Cache) Size() int64 {
415	418		return c.total
416	419		}
417	420
418		-	// ParseMax reads ERMITE_RUNNER_ACTIONS_CACHE_MAX: empty is DefaultMax, 0 is off
419		-	// (returned as 0), otherwise a number of bytes or a size with a K, M, G or T
420		-	// suffix (optionally followed by iB or B).
421		-	func ParseMax(v string) (int64, error) {
422		-	v = strings.TrimSpace(v)
423		-	if v == "" {
424		-	return DefaultMax, nil
425		-	}
426		-	s := strings.TrimRight(strings.ToUpper(v), "IB")
427		-	mult := int64(1)
428		-	if n := len(s); n > 0 {
429		-	switch s[n-1] {
430		-	case 'K':
431		-	mult, s = 1<<10, s[:n-1]
432		-	case 'M':
433		-	mult, s = 1<<20, s[:n-1]
434		-	case 'G':
435		-	mult, s = 1<<30, s[:n-1]
436		-	case 'T':
437		-	mult, s = 1<<40, s[:n-1]
438		-	}
439		-	}
440		-	n, err := strconv.ParseInt(s, 10, 64)
441		-	if err != nil \|\| n < 0 \|\| n > (1<<62)/mult {
442		-	return 0, fmt.Errorf("%s: %q is not a size (examples: 2G, 500M, 0)", EnvMax, v)
443		-	}
444		-	return n * mult, nil
445		-	}
446		-
447	421		// Dir is the directory of the cache (tests).
448	422		func (c *Cache) Dir() string { return c.dir }

 // download it when it does not (a miss: hit is false). Concurrent calls for the
 // same key share one download; the others report a hit once it is in place. The
 // entry is pinned for PinFor.
-func (c *Cache) Ensure(ctx context.Context, key string, fetch Fetch) (hit bool, err error) {
+func (c *Cache) Ensure(ctx context.Context, key string, fetch Fetch) (hit bool, verify time.Duration, err error) {
 	if !validKey(key) {
-		return false, ErrNotFound
+		return false, verify, ErrNotFound
 	}
 	for {
-		if c.verify(key) == nil {
+		duration, err := c.verify(key)
+		verify += duration
+		if err == nil {
 			c.touch(key, true)
-			return true, nil
+			return true, verify, nil
 		}
 		c.mu.Lock()
 		if f, ok := c.flights[key]; ok {
 			select {
 			case <-f.done:
 			case <-ctx.Done():
-				return false, ctx.Err()
+				return false, verify, ctx.Err()
 			}
 			if f.err != nil {
-				return false, f.err
+				return false, verify, f.err
 			}
 			continue
 		}
 		c.mu.Unlock()
 		close(f.done)
 		if f.err != nil {
-			return false, f.err
+			return false, verify, f.err
 		}
 		c.touch(key, true)
-		return false, nil
+		return false, verify, nil
 	}
 }
 // verify re-hashes the archive of key against its sum. Any mismatch (or a
 // missing file) drops the entry and returns ErrNotFound.
-func (c *Cache) verify(key string) error {
+func (c *Cache) verify(key string) (time.Duration, error) {
+	start := time.Now()
 	sb, err := os.ReadFile(c.sumPath(key))
 	if err != nil {
 		c.drop(key)
-		return ErrNotFound
+		return time.Since(start), ErrNotFound
 	}
 	want, sizeS, _ := strings.Cut(strings.TrimSpace(string(sb)), " ")
 	size, perr := strconv.ParseInt(sizeS, 10, 64)
 			_ = f.Close()
 		}
 		c.drop(key)
-		return ErrNotFound
+		return time.Since(start), ErrNotFound
 	}
 	defer func() { _ = f.Close() }()
 	h := sha256.New()
 	if err != nil || n != size || hex.EncodeToString(h.Sum(nil)) != want {
 		c.log.Warn().Str("key", key).Msg("action archive failed its integrity check; dropped")
 		c.drop(key)
-		return ErrNotFound
+		return time.Since(start), ErrNotFound
 	}
-	return nil
+	return time.Since(start), nil
 }
 func (c *Cache) drop(key string) {
 	if !validKey(key) {
 		return nil, ErrNotFound
 	}
-	if err := c.verify(key); err != nil {
+	if _, err := c.verify(key); err != nil {
 		return nil, err
 	}
 	f, err := os.Open(c.archivePath(key))
 	return c.total
 }
-// ParseMax reads ERMITE_RUNNER_ACTIONS_CACHE_MAX: empty is DefaultMax, 0 is off
-// (returned as 0), otherwise a number of bytes or a size with a K, M, G or T
-// suffix (optionally followed by iB or B).
-func ParseMax(v string) (int64, error) {
-	v = strings.TrimSpace(v)
-	if v == "" {
-		return DefaultMax, nil
-	}
-	s := strings.TrimRight(strings.ToUpper(v), "IB")
-	mult := int64(1)
-	if n := len(s); n > 0 {
-		switch s[n-1] {
-		case 'K':
-			mult, s = 1<<10, s[:n-1]
-		case 'M':
-			mult, s = 1<<20, s[:n-1]
-		case 'G':
-			mult, s = 1<<30, s[:n-1]
-		case 'T':
-			mult, s = 1<<40, s[:n-1]
-		}
-	}
-	n, err := strconv.ParseInt(s, 10, 64)
-	if err != nil || n < 0 || n > (1<<62)/mult {
-		return 0, fmt.Errorf("%s: %q is not a size (examples: 2G, 500M, 0)", EnvMax, v)
-	}
-	return n * mult, nil
-}
 // Dir is the directory of the cache (tests).
 func (c *Cache) Dir() string { return c.dir }

M internal/actionscache/cache_test.go +17 -30


@@ -55,10 +55,10 @@ func TestHitMiss(t *testing.T) {
55	55		k := keyOf(t, 0)
56	56		var fetches int
57	57		f := func(ctx context.Context, w io.Writer) error { fetches++; return body("archive")(ctx, w) }
58		-	if hit, err := c.Ensure(context.Background(), k, f); err != nil \|\| hit {
	58	+	if hit, _, err := c.Ensure(context.Background(), k, f); err != nil \|\| hit {
59	59		t.Fatalf("first: hit=%v err=%v", hit, err)
60	60		}
61		-	if hit, err := c.Ensure(context.Background(), k, f); err != nil \|\| !hit {
	61	+	if hit, verify, err := c.Ensure(context.Background(), k, f); err != nil \|\| !hit \|\| verify <= 0 {
62	62		t.Fatalf("second: hit=%v err=%v", hit, err)
63	63		}
64	64		if fetches != 1 \|\| read(t, c, k) != "archive" {

@@ -69,7 +69,7 @@ func TestHitMiss(t *testing.T) {
69	69		if err != nil {
70	70		t.Fatal(err)
71	71		}
72		-	if hit, _ := c2.Ensure(context.Background(), k, f); !hit \|\| fetches != 1 {
	72	+	if hit, _, _ := c2.Ensure(context.Background(), k, f); !hit \|\| fetches != 1 {
73	73		t.Fatalf("restart: hit=%v fetches=%d", hit, fetches)
74	74		}
75	75		}

@@ -78,7 +78,7 @@ func TestFailedFetchLeavesNothing(t *testing.T) {
78	78		c, _ := newCache(t, 1<<20)
79	79		k := keyOf(t, 0)
80	80		boom := errors.New("429")
81		-	_, err := c.Ensure(context.Background(), k, func(_ context.Context, w io.Writer) error { _, _ = io.WriteString(w, "part"); return boom })
	81	+	_, _, err := c.Ensure(context.Background(), k, func(_ context.Context, w io.Writer) error { _, _ = io.WriteString(w, "part"); return boom })
82	82		if !errors.Is(err, boom) {
83	83		t.Fatal(err)
84	84		}

@@ -107,7 +107,7 @@ func TestConcurrentSingleFlight(t *testing.T) {
107	107		wg.Add(1)
108	108		go func() {
109	109		defer wg.Done()
110		-	hit, err := c.Ensure(context.Background(), k, f)
	110	+	hit, _, err := c.Ensure(context.Background(), k, f)
111	111		if err != nil {
112	112		t.Error(err)
113	113		}

@@ -130,7 +130,7 @@ func TestConcurrentSingleFlight(t *testing.T) {
130	130		func TestCorruptEntryIsDroppedAndRefetched(t *testing.T) {
131	131		c, _ := newCache(t, 1<<20)
132	132		k := keyOf(t, 0)
133		-	if _, err := c.Ensure(context.Background(), k, body("good bytes")); err != nil {
	133	+	if _, _, err := c.Ensure(context.Background(), k, body("good bytes")); err != nil {
134	134		t.Fatal(err)
135	135		}
136	136		// Same size, other content: only the hash can tell.

@@ -143,7 +143,7 @@ func TestCorruptEntryIsDroppedAndRefetched(t *testing.T) {
143	143		if c.Size() != 0 {
144	144		t.Errorf("size %d after the drop", c.Size())
145	145		}
146		-	hit, err := c.Ensure(context.Background(), k, body("good bytes"))
	146	+	hit, _, err := c.Ensure(context.Background(), k, body("good bytes"))
147	147		if err != nil \|\| hit {
148	148		t.Fatalf("hit=%v err=%v", hit, err)
149	149		}

@@ -165,17 +165,17 @@ func TestLRUEviction(t *testing.T) {
165	165		step := func() { *now = now.Add(time.Hour + PinFor) } // past every pin
166	166		a, b, d := keyOf(t, 0), keyOf(t, 1), keyOf(t, 2)
167	167		for _, k := range []string{a, b} {
168		-	if _, err := c.Ensure(ctx, k, body(strings.Repeat("x", 10))); err != nil {
	168	+	if _, _, err := c.Ensure(ctx, k, body(strings.Repeat("x", 10))); err != nil {
169	169		t.Fatal(err)
170	170		}
171	171		step()
172	172		}
173	173		// Use a again: b is now the least recently used.
174		-	if hit, _ := c.Ensure(ctx, a, body("unused")); !hit {
	174	+	if hit, _, _ := c.Ensure(ctx, a, body("unused")); !hit {
175	175		t.Fatal("a should hit")
176	176		}
177	177		step()
178		-	if _, err := c.Ensure(ctx, d, body(strings.Repeat("y", 10))); err != nil {
	178	+	if _, _, err := c.Ensure(ctx, d, body(strings.Repeat("y", 10))); err != nil {
179	179		t.Fatal(err)
180	180		}
181	181		if _, err := c.Open(b); !errors.Is(err, ErrNotFound) {

@@ -197,10 +197,10 @@ func TestPinnedEntryIsNotEvicted(t *testing.T) {
197	197		c, _ := newCache(t, 15)
198	198		ctx := context.Background()
199	199		a, b := keyOf(t, 0), keyOf(t, 1)
200		-	if _, err := c.Ensure(ctx, a, body(strings.Repeat("x", 10))); err != nil {
	200	+	if _, _, err := c.Ensure(ctx, a, body(strings.Repeat("x", 10))); err != nil {
201	201		t.Fatal(err)
202	202		}
203		-	if _, err := c.Ensure(ctx, b, body(strings.Repeat("y", 10))); !errors.Is(err, ErrTooLarge) {
	203	+	if _, _, err := c.Ensure(ctx, b, body(strings.Repeat("y", 10))); !errors.Is(err, ErrTooLarge) {
204	204		t.Fatalf("got %v", err)
205	205		}
206	206		if _, err := c.Open(a); err != nil {

@@ -257,7 +257,7 @@ func TestConcurrentDownloadsStayUnderTheCap(t *testing.T) {
257	257		go func() {
258	258		defer wg.Done()
259	259		k := keyOf(t, i)
260		-	_, _ = c.Ensure(context.Background(), k, func(_ context.Context, w io.Writer) error {
	260	+	_, _, _ = c.Ensure(context.Background(), k, func(_ context.Context, w io.Writer) error {
261	261		for range 6 { // 6 x 10 bytes, in pieces
262	262		if _, err := w.Write([]byte(strings.Repeat("z", 10))); err != nil {
263	263		return err

@@ -307,7 +307,7 @@ func TestConcurrentDownloadsStayUnderTheCap(t *testing.T) {
307	307		func TestTemporaryFilesAreCountedAndCleaned(t *testing.T) {
308	308		c, _ := newCache(t, 50)
309	309		seen := int64(-1)
310		-	_, err := c.Ensure(context.Background(), keyOf(t, 0), func(_ context.Context, w io.Writer) error {
	310	+	_, _, err := c.Ensure(context.Background(), keyOf(t, 0), func(_ context.Context, w io.Writer) error {
311	311		_, _ = io.WriteString(w, strings.Repeat("a", 30))
312	312		c.mu.Lock()
313	313		seen = c.pending

@@ -318,7 +318,7 @@ func TestTemporaryFilesAreCountedAndCleaned(t *testing.T) {
318	318		t.Fatalf("err=%v pending during=%d", err, seen)
319	319		}
320	320		// Refused midway for size: the part already written is released too.
321		-	_, err = c.Ensure(context.Background(), keyOf(t, 1), body(strings.Repeat("b", 60)))
	321	+	_, _, err = c.Ensure(context.Background(), keyOf(t, 1), body(strings.Repeat("b", 60)))
322	322		if !errors.Is(err, ErrTooLarge) {
323	323		t.Fatalf("got %v", err)
324	324		}

@@ -326,7 +326,7 @@ func TestTemporaryFilesAreCountedAndCleaned(t *testing.T) {
326	326		t.Fatalf("pending=%d size=%d disk=%d", c.pending, c.Size(), diskUse(t, c))
327	327		}
328	328		// The size is what was written, whatever the server announced.
329		-	if _, err := c.Ensure(context.Background(), keyOf(t, 2), body(strings.Repeat("c", 50))); err != nil {
	329	+	if _, _, err := c.Ensure(context.Background(), keyOf(t, 2), body(strings.Repeat("c", 50))); err != nil {
330	330		t.Fatal(err)
331	331		}
332	332		if c.Size() != 50 {

@@ -336,7 +336,7 @@ func TestTemporaryFilesAreCountedAndCleaned(t *testing.T) {
336	336
337	337		func TestTooLarge(t *testing.T) {
338	338		c, _ := newCache(t, 10)
339		-	_, err := c.Ensure(context.Background(), keyOf(t, 0), body(strings.Repeat("x", 11)))
	339	+	_, _, err := c.Ensure(context.Background(), keyOf(t, 0), body(strings.Repeat("x", 11)))
340	340		if !errors.Is(err, ErrTooLarge) {
341	341		t.Fatalf("got %v", err)
342	342		}

@@ -383,16 +383,3 @@ func TestKeyNeverUsesARef(t *testing.T) {
383	383		t.Error("shape")
384	384		}
385	385		}
386		-
387		-	func TestParseMax(t *testing.T) {
388		-	for in, want := range map[string]int64{"": DefaultMax, "0": 0, "500M": 500 << 20, "2G": 2 << 30, "2GiB": 2 << 30, "1024": 1024} {
389		-	if got, err := ParseMax(in); err != nil \|\| got != want {
390		-	t.Errorf("%q: %d %v", in, got, err)
391		-	}
392		-	}
393		-	for _, in := range []string{"-1", "x", "G"} {
394		-	if _, err := ParseMax(in); err == nil {
395		-	t.Errorf("%q accepted", in)
396		-	}
397		-	}
398		-	}

 	k := keyOf(t, 0)
 	var fetches int
 	f := func(ctx context.Context, w io.Writer) error { fetches++; return body("archive")(ctx, w) }
-	if hit, err := c.Ensure(context.Background(), k, f); err != nil || hit {
+	if hit, _, err := c.Ensure(context.Background(), k, f); err != nil || hit {
 		t.Fatalf("first: hit=%v err=%v", hit, err)
 	}
-	if hit, err := c.Ensure(context.Background(), k, f); err != nil || !hit {
+	if hit, verify, err := c.Ensure(context.Background(), k, f); err != nil || !hit || verify <= 0 {
 		t.Fatalf("second: hit=%v err=%v", hit, err)
 	}
 	if fetches != 1 || read(t, c, k) != "archive" {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if hit, _ := c2.Ensure(context.Background(), k, f); !hit || fetches != 1 {
+	if hit, _, _ := c2.Ensure(context.Background(), k, f); !hit || fetches != 1 {
 		t.Fatalf("restart: hit=%v fetches=%d", hit, fetches)
 	}
 }
 	c, _ := newCache(t, 1<<20)
 	k := keyOf(t, 0)
 	boom := errors.New("429")
-	_, err := c.Ensure(context.Background(), k, func(_ context.Context, w io.Writer) error { _, _ = io.WriteString(w, "part"); return boom })
+	_, _, err := c.Ensure(context.Background(), k, func(_ context.Context, w io.Writer) error { _, _ = io.WriteString(w, "part"); return boom })
 	if !errors.Is(err, boom) {
 		t.Fatal(err)
 	}
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			hit, err := c.Ensure(context.Background(), k, f)
+			hit, _, err := c.Ensure(context.Background(), k, f)
 			if err != nil {
 				t.Error(err)
 			}
 func TestCorruptEntryIsDroppedAndRefetched(t *testing.T) {
 	c, _ := newCache(t, 1<<20)
 	k := keyOf(t, 0)
-	if _, err := c.Ensure(context.Background(), k, body("good bytes")); err != nil {
+	if _, _, err := c.Ensure(context.Background(), k, body("good bytes")); err != nil {
 		t.Fatal(err)
 	}
 	// Same size, other content: only the hash can tell.
 	if c.Size() != 0 {
 		t.Errorf("size %d after the drop", c.Size())
 	}
-	hit, err := c.Ensure(context.Background(), k, body("good bytes"))
+	hit, _, err := c.Ensure(context.Background(), k, body("good bytes"))
 	if err != nil || hit {
 		t.Fatalf("hit=%v err=%v", hit, err)
 	}
 	step := func() { *now = now.Add(time.Hour + PinFor) } // past every pin
 	a, b, d := keyOf(t, 0), keyOf(t, 1), keyOf(t, 2)
 	for _, k := range []string{a, b} {
-		if _, err := c.Ensure(ctx, k, body(strings.Repeat("x", 10))); err != nil {
+		if _, _, err := c.Ensure(ctx, k, body(strings.Repeat("x", 10))); err != nil {
 			t.Fatal(err)
 		}
 		step()
 	}
 	// Use a again: b is now the least recently used.
-	if hit, _ := c.Ensure(ctx, a, body("unused")); !hit {
+	if hit, _, _ := c.Ensure(ctx, a, body("unused")); !hit {
 		t.Fatal("a should hit")
 	}
 	step()
-	if _, err := c.Ensure(ctx, d, body(strings.Repeat("y", 10))); err != nil {
+	if _, _, err := c.Ensure(ctx, d, body(strings.Repeat("y", 10))); err != nil {
 		t.Fatal(err)
 	}
 	if _, err := c.Open(b); !errors.Is(err, ErrNotFound) {
 	c, _ := newCache(t, 15)
 	ctx := context.Background()
 	a, b := keyOf(t, 0), keyOf(t, 1)
-	if _, err := c.Ensure(ctx, a, body(strings.Repeat("x", 10))); err != nil {
+	if _, _, err := c.Ensure(ctx, a, body(strings.Repeat("x", 10))); err != nil {
 		t.Fatal(err)
 	}
-	if _, err := c.Ensure(ctx, b, body(strings.Repeat("y", 10))); !errors.Is(err, ErrTooLarge) {
+	if _, _, err := c.Ensure(ctx, b, body(strings.Repeat("y", 10))); !errors.Is(err, ErrTooLarge) {
 		t.Fatalf("got %v", err)
 	}
 	if _, err := c.Open(a); err != nil {
 		go func() {
 			defer wg.Done()
 			k := keyOf(t, i)
-			_, _ = c.Ensure(context.Background(), k, func(_ context.Context, w io.Writer) error {
+			_, _, _ = c.Ensure(context.Background(), k, func(_ context.Context, w io.Writer) error {
 				for range 6 { // 6 x 10 bytes, in pieces
 					if _, err := w.Write([]byte(strings.Repeat("z", 10))); err != nil {
 						return err
 func TestTemporaryFilesAreCountedAndCleaned(t *testing.T) {
 	c, _ := newCache(t, 50)
 	seen := int64(-1)
-	_, err := c.Ensure(context.Background(), keyOf(t, 0), func(_ context.Context, w io.Writer) error {
+	_, _, err := c.Ensure(context.Background(), keyOf(t, 0), func(_ context.Context, w io.Writer) error {
 		_, _ = io.WriteString(w, strings.Repeat("a", 30))
 		c.mu.Lock()
 		seen = c.pending
 		t.Fatalf("err=%v pending during=%d", err, seen)
 	}
 	// Refused midway for size: the part already written is released too.
-	_, err = c.Ensure(context.Background(), keyOf(t, 1), body(strings.Repeat("b", 60)))
+	_, _, err = c.Ensure(context.Background(), keyOf(t, 1), body(strings.Repeat("b", 60)))
 	if !errors.Is(err, ErrTooLarge) {
 		t.Fatalf("got %v", err)
 	}
 		t.Fatalf("pending=%d size=%d disk=%d", c.pending, c.Size(), diskUse(t, c))
 	}
 	// The size is what was written, whatever the server announced.
-	if _, err := c.Ensure(context.Background(), keyOf(t, 2), body(strings.Repeat("c", 50))); err != nil {
+	if _, _, err := c.Ensure(context.Background(), keyOf(t, 2), body(strings.Repeat("c", 50))); err != nil {
 		t.Fatal(err)
 	}
 	if c.Size() != 50 {
 func TestTooLarge(t *testing.T) {
 	c, _ := newCache(t, 10)
-	_, err := c.Ensure(context.Background(), keyOf(t, 0), body(strings.Repeat("x", 11)))
+	_, _, err := c.Ensure(context.Background(), keyOf(t, 0), body(strings.Repeat("x", 11)))
 	if !errors.Is(err, ErrTooLarge) {
 		t.Fatalf("got %v", err)
 	}
 		t.Error("shape")
 	}
 }
-func TestParseMax(t *testing.T) {
-	for in, want := range map[string]int64{"": DefaultMax, "0": 0, "500M": 500 << 20, "2G": 2 << 30, "2GiB": 2 << 30, "1024": 1024} {
-		if got, err := ParseMax(in); err != nil || got != want {
-			t.Errorf("%q: %d %v", in, got, err)
-		}
-	}
-	for _, in := range []string{"-1", "x", "G"} {
-		if _, err := ParseMax(in); err == nil {
-			t.Errorf("%q accepted", in)
-		}
-	}
-}

M internal/cachesvc/archives_test.go +1 -1


@@ -24,7 +24,7 @@ func archiveRig(t testing.T) (rig, *actionscache.Cache, string) {
24	24		}
25	25		r := newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, Archives: ac})
26	26		key, _ := actionscache.Key("https://github.com", "o/r", strings.Repeat("a", 40))
27		-	if _, err := ac.Ensure(context.Background(), key, func(_ context.Context, w io.Writer) error {
	27	+	if _, _, err := ac.Ensure(context.Background(), key, func(_ context.Context, w io.Writer) error {
28	28		_, err := io.WriteString(w, "tarball")
29	29		return err
30	30		}); err != nil {


@@ -24,7 +24,7 @@ func archiveRig(t testing.T) (rig, *actionscache.Cache, string) {
24		}	24		}
25		r := newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, Archives: ac})	25		r := newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, Archives: ac})
26		key, _ := actionscache.Key("https://github.com", "o/r", strings.Repeat("a", 40))	26		key, _ := actionscache.Key("https://github.com", "o/r", strings.Repeat("a", 40))
27	-	if _, err := ac.Ensure(context.Background(), key, func(_ context.Context, w io.Writer) error {	27	+	if _, _, err := ac.Ensure(context.Background(), key, func(_ context.Context, w io.Writer) error {
28		_, err := io.WriteString(w, "tarball")	28		_, err := io.WriteString(w, "tarball")
29		return err	29		return err
30		}); err != nil {	30		}); err != nil {

M internal/cachesvc/handler.go +51 -3


@@ -13,6 +13,8 @@ import (
13	13		"strconv"
14	14		"strings"
15	15		"time"
	16	+
	17	+	"bnema.dev/ermite-runner/internal/registryproxy"
16	18		)
17	19
18	20		// APIPath is the prefix of the cache API: the toolkit appends "_apis/

@@ -45,6 +47,11 @@ func (s *Service) Handler() http.Handler { return s.handler("") }
45	47		// any other sandbox's channel.
46	48		func (s *Service) handler(job string) http.Handler {
47	49		mux := http.NewServeMux()
	50	+	if s.registry != nil {
	51	+	registry := s.serveRegistry(job, make(chan struct{}, s.registryPerChannel), registryproxy.NewQuota(RegistryDownloadsPerChannel))
	52	+	mux.HandleFunc("GET "+RegistryPath, registry)
	53	+	mux.HandleFunc("HEAD "+RegistryPath, registry)
	54	+	}
48	55		// The routes are those of the toolkit's v1 client (cacheHttpClient.ts):
49	56		// GET cache?keys=k1,k2&version=v :89 lookup
50	57		// GET caches?key=k :132 diagnostics (core.isDebug())

@@ -73,8 +80,8 @@ func (s *Service) handler(job string) http.Handler {
73	80		})
74	81		}
75	82
76		-	// channelHandler is the handler of a sandbox's channel: the cache API and the
77		-	// action archives, and when results is not nil everything else goes to it (the results service of the
	83	+	// channelHandler is the handler of a sandbox's channel: the cache API, the
	84	+	// action archives and the registry, and when results is not nil everything else goes to it (the results service of the
78	85		// job: Twirp and blob paths). The results service gets the same progress
79	86		// deadlines as the cache: its bodies and replies must keep moving.
80	87		func (s *Service) channelHandler(job string, results http.Handler) http.Handler {

@@ -83,7 +90,7 @@ func (s *Service) channelHandler(job string, results http.Handler) http.Handler
83	90		return cache
84	91		}
85	92		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
86		-	if strings.HasPrefix(r.URL.Path, APIPath) \|\| strings.HasPrefix(r.URL.Path, ArchivePath) {
	93	+	if strings.HasPrefix(r.URL.Path, APIPath) \|\| strings.HasPrefix(r.URL.Path, ArchivePath) \|\| strings.HasPrefix(r.URL.Path, RegistryPath) {
87	94		cache.ServeHTTP(w, r)
88	95		return
89	96		}

@@ -388,6 +395,47 @@ func (p *progressWriter) Write(b []byte) (int, error) {
388	395		// Unwrap lets http.ResponseController reach the real writer.
389	396		func (p *progressWriter) Unwrap() http.ResponseWriter { return p.ResponseWriter }
390	397
	398	+	var _ registryproxy.KeepAliver = (*progressWriter)(nil)
	399	+
	400	+	// KeepAlive moves the deadlines without writing. The registry proxy calls it
	401	+	// while the client waits for a shared upstream download that is progressing,
	402	+	// before the first byte of the reply. The read deadline moves too: once it
	403	+	// passes, net/http cancels the request's context and the wait would end.
	404	+	func (p *progressWriter) KeepAlive() {
	405	+	d := deadline(p.idle, p.remaining)
	406	+	_ = p.rc.SetReadDeadline(d)
	407	+	_ = p.rc.SetWriteDeadline(d)
	408	+	}
	409	+
	410	+	// serveRegistry serves the Docker Hub mirror with the same sliding progress
	411	+	// deadline as the other downloads: a layer of GiB keeps going as long as it
	412	+	// moves, and a client that stops reading is cut after one progress period.
	413	+	// slots caps the requests of this channel in progress at once; one over the
	414	+	// cap gets 503 at once, which dockerd retries or answers by pulling from
	415	+	// Docker Hub, so the channel's connections stay free for the cache. quota is
	416	+	// the channel's share of the proxy's cold downloads. job, when not empty, is
	417	+	// the channel's job, which the registry logs.
	418	+	func (s Service) serveRegistry(job string, slots chan struct{}, quota registryproxy.Quota) http.HandlerFunc {
	419	+	return func(w http.ResponseWriter, r *http.Request) {
	420	+	select {
	421	+	case slots <- struct{}{}:
	422	+	defer func() { <-slots }()
	423	+	default:
	424	+	w.Header().Set("Retry-After", "1")
	425	+	reply(w, http.StatusServiceUnavailable, "registry busy")
	426	+	return
	427	+	}
	428	+	rc := http.NewResponseController(w)
	429	+	defer func() { _ = rc.SetWriteDeadline(deadline(s.progress, nil)) }()
	430	+	ctx := registryproxy.WithQuota(r.Context(), quota)
	431	+	if job != "" {
	432	+	ctx = registryproxy.WithJob(ctx, job)
	433	+	}
	434	+	r = r.WithContext(ctx)
	435	+	s.registry.ServeHTTP(&progressWriter{ResponseWriter: w, rc: rc, idle: s.progress}, r)
	436	+	}
	437	+	}
	438	+
391	439		func (s Service) commit(w http.ResponseWriter, r http.Request, c Claims) {
392	440		id, ok := parseID(r)
393	441		var req struct {


@@ -13,6 +13,8 @@ import (
13		"strconv"	13		"strconv"
14		"strings"	14		"strings"
15		"time"	15		"time"
			16	+
			17	+	"bnema.dev/ermite-runner/internal/registryproxy"
16		)	18		)
17			19
18		// APIPath is the prefix of the cache API: the toolkit appends "_apis/	20		// APIPath is the prefix of the cache API: the toolkit appends "_apis/

@@ -45,6 +47,11 @@ func (s *Service) Handler() http.Handler { return s.handler("") }
45		// any other sandbox's channel.	47		// any other sandbox's channel.
46		func (s *Service) handler(job string) http.Handler {	48		func (s *Service) handler(job string) http.Handler {
47		mux := http.NewServeMux()	49		mux := http.NewServeMux()
			50	+	if s.registry != nil {
			51	+	registry := s.serveRegistry(job, make(chan struct{}, s.registryPerChannel), registryproxy.NewQuota(RegistryDownloadsPerChannel))
			52	+	mux.HandleFunc("GET "+RegistryPath, registry)
			53	+	mux.HandleFunc("HEAD "+RegistryPath, registry)
			54	+	}
48		// The routes are those of the toolkit's v1 client (cacheHttpClient.ts):	55		// The routes are those of the toolkit's v1 client (cacheHttpClient.ts):
49		// GET cache?keys=k1,k2&version=v :89 lookup	56		// GET cache?keys=k1,k2&version=v :89 lookup
50		// GET caches?key=k :132 diagnostics (core.isDebug())	57		// GET caches?key=k :132 diagnostics (core.isDebug())

@@ -73,8 +80,8 @@ func (s *Service) handler(job string) http.Handler {
73		})	80		})
74		}	81		}
75			82
76	-	// channelHandler is the handler of a sandbox's channel: the cache API and the	83	+	// channelHandler is the handler of a sandbox's channel: the cache API, the
77	-	// action archives, and when results is not nil everything else goes to it (the results service of the	84	+	// action archives and the registry, and when results is not nil everything else goes to it (the results service of the
78		// job: Twirp and blob paths). The results service gets the same progress	85		// job: Twirp and blob paths). The results service gets the same progress
79		// deadlines as the cache: its bodies and replies must keep moving.	86		// deadlines as the cache: its bodies and replies must keep moving.
80		func (s *Service) channelHandler(job string, results http.Handler) http.Handler {	87		func (s *Service) channelHandler(job string, results http.Handler) http.Handler {

@@ -83,7 +90,7 @@ func (s *Service) channelHandler(job string, results http.Handler) http.Handler
83		return cache	90		return cache
84		}	91		}
85		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {	92		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
86	-	if strings.HasPrefix(r.URL.Path, APIPath) \|\| strings.HasPrefix(r.URL.Path, ArchivePath) {	93	+	if strings.HasPrefix(r.URL.Path, APIPath) \|\| strings.HasPrefix(r.URL.Path, ArchivePath) \|\| strings.HasPrefix(r.URL.Path, RegistryPath) {
87		cache.ServeHTTP(w, r)	94		cache.ServeHTTP(w, r)
88		return	95		return
89		}	96		}

@@ -388,6 +395,47 @@ func (p *progressWriter) Write(b []byte) (int, error) {
388		// Unwrap lets http.ResponseController reach the real writer.	395		// Unwrap lets http.ResponseController reach the real writer.
389		func (p *progressWriter) Unwrap() http.ResponseWriter { return p.ResponseWriter }	396		func (p *progressWriter) Unwrap() http.ResponseWriter { return p.ResponseWriter }
390			397
			398	+	var _ registryproxy.KeepAliver = (*progressWriter)(nil)
			399	+
			400	+	// KeepAlive moves the deadlines without writing. The registry proxy calls it
			401	+	// while the client waits for a shared upstream download that is progressing,
			402	+	// before the first byte of the reply. The read deadline moves too: once it
			403	+	// passes, net/http cancels the request's context and the wait would end.
			404	+	func (p *progressWriter) KeepAlive() {
			405	+	d := deadline(p.idle, p.remaining)
			406	+	_ = p.rc.SetReadDeadline(d)
			407	+	_ = p.rc.SetWriteDeadline(d)
			408	+	}
			409	+
			410	+	// serveRegistry serves the Docker Hub mirror with the same sliding progress
			411	+	// deadline as the other downloads: a layer of GiB keeps going as long as it
			412	+	// moves, and a client that stops reading is cut after one progress period.
			413	+	// slots caps the requests of this channel in progress at once; one over the
			414	+	// cap gets 503 at once, which dockerd retries or answers by pulling from
			415	+	// Docker Hub, so the channel's connections stay free for the cache. quota is
			416	+	// the channel's share of the proxy's cold downloads. job, when not empty, is
			417	+	// the channel's job, which the registry logs.
			418	+	func (s Service) serveRegistry(job string, slots chan struct{}, quota registryproxy.Quota) http.HandlerFunc {
			419	+	return func(w http.ResponseWriter, r *http.Request) {
			420	+	select {
			421	+	case slots <- struct{}{}:
			422	+	defer func() { <-slots }()
			423	+	default:
			424	+	w.Header().Set("Retry-After", "1")
			425	+	reply(w, http.StatusServiceUnavailable, "registry busy")
			426	+	return
			427	+	}
			428	+	rc := http.NewResponseController(w)
			429	+	defer func() { _ = rc.SetWriteDeadline(deadline(s.progress, nil)) }()
			430	+	ctx := registryproxy.WithQuota(r.Context(), quota)
			431	+	if job != "" {
			432	+	ctx = registryproxy.WithJob(ctx, job)
			433	+	}
			434	+	r = r.WithContext(ctx)
			435	+	s.registry.ServeHTTP(&progressWriter{ResponseWriter: w, rc: rc, idle: s.progress}, r)
			436	+	}
			437	+	}
			438	+
391		func (s Service) commit(w http.ResponseWriter, r http.Request, c Claims) {	439		func (s Service) commit(w http.ResponseWriter, r http.Request, c Claims) {
392		id, ok := parseID(r)	440		id, ok := parseID(r)
393		var req struct {	441		var req struct {

A internal/cachesvc/registry_deadline_test.go +127 -0

@@ -0,0 +1,127 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package cachesvc
	4	+
	5	+	import (
	6	+	"bufio"
	7	+	"fmt"
	8	+	"io"
	9	+	"net"
	10	+	"net/http"
	11	+	"strings"
	12	+	"testing"
	13	+	"time"
	14	+
	15	+	"bnema.dev/ermite-runner/internal/registryproxy"
	16	+	)
	17	+
	18	+	const registryProgress = 300 * time.Millisecond
	19	+
	20	+	// registryRig serves a fake registry behind the service's real HTTP server, so
	21	+	// that the connection deadlines apply.
	22	+	func registryRig(t testing.T, h http.HandlerFunc) rig {
	23	+	t.Helper()
	24	+	return newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, ProgressTimeout: registryProgress, Registry: h})
	25	+	}
	26	+
	27	+	// A registry download that keeps moving may last longer than the progress
	28	+	// period set when the request started.
	29	+	func TestRegistryProgressingTransferSucceeds(t *testing.T) {
	30	+	const chunks = 14
	31	+	r := registryRig(t, func(w http.ResponseWriter, _ *http.Request) {
	32	+	w.Header().Set("Content-Length", fmt.Sprint(chunks*1024))
	33	+	for range chunks {
	34	+	if _, err := w.Write([]byte(strings.Repeat("x", 1024))); err != nil {
	35	+	return
	36	+	}
	37	+	_ = http.NewResponseController(w).Flush()
	38	+	time.Sleep(registryProgress / 6)
	39	+	}
	40	+	})
	41	+	start := time.Now()
	42	+	resp, err := http.Get(r.srv.URL + RegistryPath + "library/alpine/blobs/x")
	43	+	if err != nil {
	44	+	t.Fatal(err)
	45	+	}
	46	+	defer func() { _ = resp.Body.Close() }()
	47	+	b, err := io.ReadAll(resp.Body)
	48	+	if err != nil \|\| len(b) != chunks*1024 {
	49	+	t.Fatalf("read %d bytes: %v", len(b), err)
	50	+	}
	51	+	if d := time.Since(start); d < 2*registryProgress {
	52	+	t.Fatalf("transfer took %v, not longer than the progress period", d)
	53	+	}
	54	+	}
	55	+
	56	+	// The proxy keeps a client alive while it waits for a shared upstream
	57	+	// download, before the first byte; the read deadline moves too, so the request
	58	+	// context stays alive.
	59	+	func TestRegistryKeepAliveWhileWaiting(t *testing.T) {
	60	+	r := registryRig(t, func(w http.ResponseWriter, req *http.Request) {
	61	+	k, ok := w.(registryproxy.KeepAliver)
	62	+	if !ok {
	63	+	t.Error("the registry writer is no KeepAliver")
	64	+	return
	65	+	}
	66	+	for range 12 { // two progress periods
	67	+	time.Sleep(registryProgress / 6)
	68	+	k.KeepAlive()
	69	+	if req.Context().Err() != nil {
	70	+	t.Error("request context ended while kept alive")
	71	+	return
	72	+	}
	73	+	}
	74	+	_, _ = io.WriteString(w, "verified")
	75	+	})
	76	+	resp, err := http.Get(r.srv.URL + RegistryPath + "library/alpine/blobs/x")
	77	+	if err != nil {
	78	+	t.Fatal(err)
	79	+	}
	80	+	defer func() { _ = resp.Body.Close() }()
	81	+	if b, err := io.ReadAll(resp.Body); err != nil \|\| string(b) != "verified" {
	82	+	t.Fatalf("%q %v", b, err)
	83	+	}
	84	+	}
	85	+
	86	+	// A client that stops reading is cut after one progress period.
	87	+	func TestRegistryStalledReaderIsCut(t *testing.T) {
	88	+	result := make(chan error, 1)
	89	+	r := registryRig(t, func(w http.ResponseWriter, _ *http.Request) {
	90	+	chunk := []byte(strings.Repeat("x", 64<<10))
	91	+	deadline := time.Now().Add(5 * time.Second)
	92	+	for time.Now().Before(deadline) {
	93	+	if _, err := w.Write(chunk); err != nil {
	94	+	result <- err
	95	+	return
	96	+	}
	97	+	}
	98	+	result <- nil
	99	+	})
	100	+	conn, err := net.Dial("tcp", strings.TrimPrefix(r.srv.URL, "http://"))
	101	+	if err != nil {
	102	+	t.Fatal(err)
	103	+	}
	104	+	defer func() { _ = conn.Close() }()
	105	+	if tc, ok := conn.(*net.TCPConn); ok {
	106	+	_ = tc.SetReadBuffer(4096)
	107	+	}
	108	+	if _, err := fmt.Fprint(conn, "GET "+RegistryPath+"library/alpine/blobs/x HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil {
	109	+	t.Fatal(err)
	110	+	}
	111	+	resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
	112	+	if err != nil {
	113	+	t.Fatal(err)
	114	+	}
	115	+	if resp.StatusCode != http.StatusOK {
	116	+	t.Fatal(resp.StatusCode)
	117	+	}
	118	+	// Read nothing more.
	119	+	select {
	120	+	case err := <-result:
	121	+	if err == nil {
	122	+	t.Fatal("a stalled reader was not cut")
	123	+	}
	124	+	case <-time.After(10 * time.Second):
	125	+	t.Fatal("handler still blocked")
	126	+	}
	127	+	}

@@ -0,0 +1,127 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package cachesvc
	4	+
	5	+	import (
	6	+	"bufio"
	7	+	"fmt"
	8	+	"io"
	9	+	"net"
	10	+	"net/http"
	11	+	"strings"
	12	+	"testing"
	13	+	"time"
	14	+
	15	+	"bnema.dev/ermite-runner/internal/registryproxy"
	16	+	)
	17	+
	18	+	const registryProgress = 300 * time.Millisecond
	19	+
	20	+	// registryRig serves a fake registry behind the service's real HTTP server, so
	21	+	// that the connection deadlines apply.
	22	+	func registryRig(t testing.T, h http.HandlerFunc) rig {
	23	+	t.Helper()
	24	+	return newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, ProgressTimeout: registryProgress, Registry: h})
	25	+	}
	26	+
	27	+	// A registry download that keeps moving may last longer than the progress
	28	+	// period set when the request started.
	29	+	func TestRegistryProgressingTransferSucceeds(t *testing.T) {
	30	+	const chunks = 14
	31	+	r := registryRig(t, func(w http.ResponseWriter, _ *http.Request) {
	32	+	w.Header().Set("Content-Length", fmt.Sprint(chunks*1024))
	33	+	for range chunks {
	34	+	if _, err := w.Write([]byte(strings.Repeat("x", 1024))); err != nil {
	35	+	return
	36	+	}
	37	+	_ = http.NewResponseController(w).Flush()
	38	+	time.Sleep(registryProgress / 6)
	39	+	}
	40	+	})
	41	+	start := time.Now()
	42	+	resp, err := http.Get(r.srv.URL + RegistryPath + "library/alpine/blobs/x")
	43	+	if err != nil {
	44	+	t.Fatal(err)
	45	+	}
	46	+	defer func() { _ = resp.Body.Close() }()
	47	+	b, err := io.ReadAll(resp.Body)
	48	+	if err != nil \|\| len(b) != chunks*1024 {
	49	+	t.Fatalf("read %d bytes: %v", len(b), err)
	50	+	}
	51	+	if d := time.Since(start); d < 2*registryProgress {
	52	+	t.Fatalf("transfer took %v, not longer than the progress period", d)
	53	+	}
	54	+	}
	55	+
	56	+	// The proxy keeps a client alive while it waits for a shared upstream
	57	+	// download, before the first byte; the read deadline moves too, so the request
	58	+	// context stays alive.
	59	+	func TestRegistryKeepAliveWhileWaiting(t *testing.T) {
	60	+	r := registryRig(t, func(w http.ResponseWriter, req *http.Request) {
	61	+	k, ok := w.(registryproxy.KeepAliver)
	62	+	if !ok {
	63	+	t.Error("the registry writer is no KeepAliver")
	64	+	return
	65	+	}
	66	+	for range 12 { // two progress periods
	67	+	time.Sleep(registryProgress / 6)
	68	+	k.KeepAlive()
	69	+	if req.Context().Err() != nil {
	70	+	t.Error("request context ended while kept alive")
	71	+	return
	72	+	}
	73	+	}
	74	+	_, _ = io.WriteString(w, "verified")
	75	+	})
	76	+	resp, err := http.Get(r.srv.URL + RegistryPath + "library/alpine/blobs/x")
	77	+	if err != nil {
	78	+	t.Fatal(err)
	79	+	}
	80	+	defer func() { _ = resp.Body.Close() }()
	81	+	if b, err := io.ReadAll(resp.Body); err != nil \|\| string(b) != "verified" {
	82	+	t.Fatalf("%q %v", b, err)
	83	+	}
	84	+	}
	85	+
	86	+	// A client that stops reading is cut after one progress period.
	87	+	func TestRegistryStalledReaderIsCut(t *testing.T) {
	88	+	result := make(chan error, 1)
	89	+	r := registryRig(t, func(w http.ResponseWriter, _ *http.Request) {
	90	+	chunk := []byte(strings.Repeat("x", 64<<10))
	91	+	deadline := time.Now().Add(5 * time.Second)
	92	+	for time.Now().Before(deadline) {
	93	+	if _, err := w.Write(chunk); err != nil {
	94	+	result <- err
	95	+	return
	96	+	}
	97	+	}
	98	+	result <- nil
	99	+	})
	100	+	conn, err := net.Dial("tcp", strings.TrimPrefix(r.srv.URL, "http://"))
	101	+	if err != nil {
	102	+	t.Fatal(err)
	103	+	}
	104	+	defer func() { _ = conn.Close() }()
	105	+	if tc, ok := conn.(*net.TCPConn); ok {
	106	+	_ = tc.SetReadBuffer(4096)
	107	+	}
	108	+	if _, err := fmt.Fprint(conn, "GET "+RegistryPath+"library/alpine/blobs/x HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil {
	109	+	t.Fatal(err)
	110	+	}
	111	+	resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
	112	+	if err != nil {
	113	+	t.Fatal(err)
	114	+	}
	115	+	if resp.StatusCode != http.StatusOK {
	116	+	t.Fatal(resp.StatusCode)
	117	+	}
	118	+	// Read nothing more.
	119	+	select {
	120	+	case err := <-result:
	121	+	if err == nil {
	122	+	t.Fatal("a stalled reader was not cut")
	123	+	}
	124	+	case <-time.After(10 * time.Second):
	125	+	t.Fatal("handler still blocked")
	126	+	}
	127	+	}

A internal/cachesvc/registry_limits_test.go +195 -0

@@ -0,0 +1,195 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package cachesvc
	4	+
	5	+	import (
	6	+	"bytes"
	7	+	"context"
	8	+	"crypto/sha256"
	9	+	"fmt"
	10	+	"io"
	11	+	"net/http"
	12	+	"net/http/httptest"
	13	+	"strings"
	14	+	"sync"
	15	+	"sync/atomic"
	16	+	"testing"
	17	+	"time"
	18	+
	19	+	"github.com/bnema/zerowrap"
	20	+
	21	+	"bnema.dev/ermite-runner/internal/registryproxy"
	22	+	)
	23	+
	24	+	// The registry requests of one channel in progress at once are capped: one
	25	+	// over the cap gets 503 at once, the cache API of that channel still answers,
	26	+	// and another channel has its own slots.
	27	+	func TestRegistryPerChannelCap(t *testing.T) {
	28	+	const limit = 2
	29	+	var inside atomic.Int32
	30	+	release := make(chan struct{})
	31	+	svc, err := New(Config{Dir: t.TempDir(), RegistryPerChannel: limit, Registry: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
	32	+	inside.Add(1)
	33	+	<-release
	34	+	_, _ = io.WriteString(w, "ok")
	35	+	})})
	36	+	if err != nil {
	37	+	t.Fatal(err)
	38	+	}
	39	+	t.Cleanup(func() { _ = svc.CloseAll() })
	40	+	var once sync.Once
	41	+	unblock := func() { once.Do(func() { close(release) }) }
	42	+	t.Cleanup(unblock)
	43	+	a, b := channel(t, svc, "sb-a", "job-a"), channel(t, svc, "sb-b", "job-b")
	44	+	get := func(c *http.Client) int {
	45	+	resp, err := c.Get("http://cache" + RegistryPath + "library/alpine/blobs/x")
	46	+	if err != nil {
	47	+	t.Error(err)
	48	+	return 0
	49	+	}
	50	+	_, _ = io.Copy(io.Discard, resp.Body)
	51	+	_ = resp.Body.Close()
	52	+	return resp.StatusCode
	53	+	}
	54	+	var wg sync.WaitGroup
	55	+	codes := make(chan int, limit)
	56	+	for range limit {
	57	+	wg.Go(func() { codes <- get(a) })
	58	+	}
	59	+	waitUntil(t, func() bool { return inside.Load() == limit })
	60	+	if code := get(a); code != http.StatusServiceUnavailable {
	61	+	t.Fatalf("over the cap: %d", code)
	62	+	}
	63	+	if code := lookupStatus(t, a, ""); code != http.StatusForbidden {
	64	+	t.Fatalf("cache API behind the registry cap: %d", code)
	65	+	}
	66	+	other := make(chan int, 1)
	67	+	go func() { other <- get(b) }()
	68	+	waitUntil(t, func() bool { return inside.Load() == limit+1 })
	69	+	unblock()
	70	+	wg.Wait()
	71	+	close(codes)
	72	+	for code := range codes {
	73	+	if code != http.StatusOK {
	74	+	t.Fatalf("within the cap: %d", code)
	75	+	}
	76	+	}
	77	+	if code := <-other; code != http.StatusOK {
	78	+	t.Fatalf("other channel: %d", code)
	79	+	}
	80	+	if code := get(a); code != http.StatusOK {
	81	+	t.Fatalf("slots not released: %d", code)
	82	+	}
	83	+	}
	84	+
	85	+	// stallingHub fakes Docker Hub for a real registryproxy: a manifest, and blobs
	86	+	// whose download stalls until the test ends. It counts the blob downloads
	87	+	// started.
	88	+	func stallingHub(t testing.T) (httptest.Server, *atomic.Int32, string) {
	89	+	t.Helper()
	90	+	const manifest = `{"schemaVersion":2}`
	91	+	digest := fmt.Sprintf("sha256:%x", sha256.Sum256([]byte(manifest)))
	92	+	var started atomic.Int32
	93	+	done := make(chan struct{})
	94	+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	95	+	switch {
	96	+	case r.URL.Path == "/token":
	97	+	_, _ = io.WriteString(w, `{"token":"public"}`)
	98	+	case strings.Contains(r.URL.Path, "/manifests/"):
	99	+	w.Header().Set("Content-Type", "application/vnd.oci.image.index.v1+json")
	100	+	w.Header().Set("Docker-Content-Digest", digest)
	101	+	_, _ = io.WriteString(w, manifest)
	102	+	default:
	103	+	started.Add(1)
	104	+	w.Header().Set("Content-Length", "1000000")
	105	+	_, _ = io.WriteString(w, "partial")
	106	+	w.(http.Flusher).Flush()
	107	+	select {
	108	+	case <-r.Context().Done():
	109	+	case <-done:
	110	+	}
	111	+	}
	112	+	}))
	113	+	t.Cleanup(srv.Close)
	114	+	t.Cleanup(func() { close(done) }) // before srv.Close, which waits for the handlers
	115	+	return srv, &started, digest
	116	+	}
	117	+
	118	+	// Each channel has its own quota of cold downloads: a channel that has
	119	+	// started RegistryDownloadsPerChannel of them waits for its own to end, while
	120	+	// another channel starts its download at once.
	121	+	func TestRegistryQuotaPerChannel(t *testing.T) {
	122	+	hub, started, _ := stallingHub(t)
	123	+	proxy, err := registryproxy.New(registryproxy.Config{Dir: t.TempDir(), Max: 1 << 30, Upstream: hub.URL, AuthURL: hub.URL + "/token",
	124	+	SlotWait: time.Hour, AbandonGrace: 10 * time.Millisecond})
	125	+	if err != nil {
	126	+	t.Fatal(err)
	127	+	}
	128	+	svc, err := New(Config{Dir: t.TempDir(), Registry: proxy})
	129	+	if err != nil {
	130	+	t.Fatal(err)
	131	+	}
	132	+	t.Cleanup(func() { _ = svc.CloseAll() })
	133	+	ctx, cancel := context.WithCancel(context.Background())
	134	+	defer cancel()
	135	+	var wg sync.WaitGroup
	136	+	defer wg.Wait()
	137	+	defer cancel()
	138	+	pull := func(c *http.Client, i int) {
	139	+	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://cache%slibrary/alpine/blobs/sha256:%x", RegistryPath, sha256.Sum256(fmt.Appendf(nil, "blob-%d", i))), nil)
	140	+	wg.Go(func() {
	141	+	if resp, err := c.Do(req); err == nil {
	142	+	_ = resp.Body.Close()
	143	+	}
	144	+	})
	145	+	}
	146	+	a, b := channel(t, svc, "sb-a", "job-a"), channel(t, svc, "sb-b", "job-b")
	147	+	for i := range RegistryDownloadsPerChannel + 1 {
	148	+	pull(a, i)
	149	+	}
	150	+	waitUntil(t, func() bool { return started.Load() == RegistryDownloadsPerChannel })
	151	+	time.Sleep(50 * time.Millisecond) // the one over A's quota would have started by now
	152	+	if n := started.Load(); n != RegistryDownloadsPerChannel {
	153	+	t.Fatalf("channel A started %d downloads, its quota is %d", n, RegistryDownloadsPerChannel)
	154	+	}
	155	+	pull(b, 100)
	156	+	waitUntil(t, func() bool { return started.Load() == RegistryDownloadsPerChannel+1 })
	157	+	}
	158	+
	159	+	// The registry lines of a channel carry its job.
	160	+	func TestRegistryLogsJob(t *testing.T) {
	161	+	hub, _, digest := stallingHub(t)
	162	+	var logs bytes.Buffer
	163	+	var mu sync.Mutex
	164	+	proxy, err := registryproxy.New(registryproxy.Config{Dir: t.TempDir(), Max: 1 << 20, Upstream: hub.URL, AuthURL: hub.URL + "/token",
	165	+	Log: zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: writerFunc(func(p []byte) (int, error) {
	166	+	mu.Lock()
	167	+	defer mu.Unlock()
	168	+	return logs.Write(p)
	169	+	})})})
	170	+	if err != nil {
	171	+	t.Fatal(err)
	172	+	}
	173	+	svc, err := New(Config{Dir: t.TempDir(), Registry: proxy})
	174	+	if err != nil {
	175	+	t.Fatal(err)
	176	+	}
	177	+	t.Cleanup(func() { _ = svc.CloseAll() })
	178	+	resp, err := channel(t, svc, "sb-a", "job-a").Get("http://cache" + RegistryPath + "library/alpine/manifests/" + digest)
	179	+	if err != nil {
	180	+	t.Fatal(err)
	181	+	}
	182	+	_ = resp.Body.Close()
	183	+	if resp.StatusCode != http.StatusOK {
	184	+	t.Fatal(resp.StatusCode)
	185	+	}
	186	+	mu.Lock()
	187	+	defer mu.Unlock()
	188	+	if !strings.Contains(logs.String(), `"job_id":"job-a"`) \|\| !strings.Contains(logs.String(), `"message":"registry manifest"`) {
	189	+	t.Fatalf("logs: %s", logs.String())
	190	+	}
	191	+	}
	192	+
	193	+	type writerFunc func(p []byte) (int, error)
	194	+
	195	+	func (f writerFunc) Write(p []byte) (int, error) { return f(p) }

@@ -0,0 +1,195 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package cachesvc
	4	+
	5	+	import (
	6	+	"bytes"
	7	+	"context"
	8	+	"crypto/sha256"
	9	+	"fmt"
	10	+	"io"
	11	+	"net/http"
	12	+	"net/http/httptest"
	13	+	"strings"
	14	+	"sync"
	15	+	"sync/atomic"
	16	+	"testing"
	17	+	"time"
	18	+
	19	+	"github.com/bnema/zerowrap"
	20	+
	21	+	"bnema.dev/ermite-runner/internal/registryproxy"
	22	+	)
	23	+
	24	+	// The registry requests of one channel in progress at once are capped: one
	25	+	// over the cap gets 503 at once, the cache API of that channel still answers,
	26	+	// and another channel has its own slots.
	27	+	func TestRegistryPerChannelCap(t *testing.T) {
	28	+	const limit = 2
	29	+	var inside atomic.Int32
	30	+	release := make(chan struct{})
	31	+	svc, err := New(Config{Dir: t.TempDir(), RegistryPerChannel: limit, Registry: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
	32	+	inside.Add(1)
	33	+	<-release
	34	+	_, _ = io.WriteString(w, "ok")
	35	+	})})
	36	+	if err != nil {
	37	+	t.Fatal(err)
	38	+	}
	39	+	t.Cleanup(func() { _ = svc.CloseAll() })
	40	+	var once sync.Once
	41	+	unblock := func() { once.Do(func() { close(release) }) }
	42	+	t.Cleanup(unblock)
	43	+	a, b := channel(t, svc, "sb-a", "job-a"), channel(t, svc, "sb-b", "job-b")
	44	+	get := func(c *http.Client) int {
	45	+	resp, err := c.Get("http://cache" + RegistryPath + "library/alpine/blobs/x")
	46	+	if err != nil {
	47	+	t.Error(err)
	48	+	return 0
	49	+	}
	50	+	_, _ = io.Copy(io.Discard, resp.Body)
	51	+	_ = resp.Body.Close()
	52	+	return resp.StatusCode
	53	+	}
	54	+	var wg sync.WaitGroup
	55	+	codes := make(chan int, limit)
	56	+	for range limit {
	57	+	wg.Go(func() { codes <- get(a) })
	58	+	}
	59	+	waitUntil(t, func() bool { return inside.Load() == limit })
	60	+	if code := get(a); code != http.StatusServiceUnavailable {
	61	+	t.Fatalf("over the cap: %d", code)
	62	+	}
	63	+	if code := lookupStatus(t, a, ""); code != http.StatusForbidden {
	64	+	t.Fatalf("cache API behind the registry cap: %d", code)
	65	+	}
	66	+	other := make(chan int, 1)
	67	+	go func() { other <- get(b) }()
	68	+	waitUntil(t, func() bool { return inside.Load() == limit+1 })
	69	+	unblock()
	70	+	wg.Wait()
	71	+	close(codes)
	72	+	for code := range codes {
	73	+	if code != http.StatusOK {
	74	+	t.Fatalf("within the cap: %d", code)
	75	+	}
	76	+	}
	77	+	if code := <-other; code != http.StatusOK {
	78	+	t.Fatalf("other channel: %d", code)
	79	+	}
	80	+	if code := get(a); code != http.StatusOK {
	81	+	t.Fatalf("slots not released: %d", code)
	82	+	}
	83	+	}
	84	+
	85	+	// stallingHub fakes Docker Hub for a real registryproxy: a manifest, and blobs
	86	+	// whose download stalls until the test ends. It counts the blob downloads
	87	+	// started.
	88	+	func stallingHub(t testing.T) (httptest.Server, *atomic.Int32, string) {
	89	+	t.Helper()
	90	+	const manifest = `{"schemaVersion":2}`
	91	+	digest := fmt.Sprintf("sha256:%x", sha256.Sum256([]byte(manifest)))
	92	+	var started atomic.Int32
	93	+	done := make(chan struct{})
	94	+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	95	+	switch {
	96	+	case r.URL.Path == "/token":
	97	+	_, _ = io.WriteString(w, `{"token":"public"}`)
	98	+	case strings.Contains(r.URL.Path, "/manifests/"):
	99	+	w.Header().Set("Content-Type", "application/vnd.oci.image.index.v1+json")
	100	+	w.Header().Set("Docker-Content-Digest", digest)
	101	+	_, _ = io.WriteString(w, manifest)
	102	+	default:
	103	+	started.Add(1)
	104	+	w.Header().Set("Content-Length", "1000000")
	105	+	_, _ = io.WriteString(w, "partial")
	106	+	w.(http.Flusher).Flush()
	107	+	select {
	108	+	case <-r.Context().Done():
	109	+	case <-done:
	110	+	}
	111	+	}
	112	+	}))
	113	+	t.Cleanup(srv.Close)
	114	+	t.Cleanup(func() { close(done) }) // before srv.Close, which waits for the handlers
	115	+	return srv, &started, digest
	116	+	}
	117	+
	118	+	// Each channel has its own quota of cold downloads: a channel that has
	119	+	// started RegistryDownloadsPerChannel of them waits for its own to end, while
	120	+	// another channel starts its download at once.
	121	+	func TestRegistryQuotaPerChannel(t *testing.T) {
	122	+	hub, started, _ := stallingHub(t)
	123	+	proxy, err := registryproxy.New(registryproxy.Config{Dir: t.TempDir(), Max: 1 << 30, Upstream: hub.URL, AuthURL: hub.URL + "/token",
	124	+	SlotWait: time.Hour, AbandonGrace: 10 * time.Millisecond})
	125	+	if err != nil {
	126	+	t.Fatal(err)
	127	+	}
	128	+	svc, err := New(Config{Dir: t.TempDir(), Registry: proxy})
	129	+	if err != nil {
	130	+	t.Fatal(err)
	131	+	}
	132	+	t.Cleanup(func() { _ = svc.CloseAll() })
	133	+	ctx, cancel := context.WithCancel(context.Background())
	134	+	defer cancel()
	135	+	var wg sync.WaitGroup
	136	+	defer wg.Wait()
	137	+	defer cancel()
	138	+	pull := func(c *http.Client, i int) {
	139	+	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://cache%slibrary/alpine/blobs/sha256:%x", RegistryPath, sha256.Sum256(fmt.Appendf(nil, "blob-%d", i))), nil)
	140	+	wg.Go(func() {
	141	+	if resp, err := c.Do(req); err == nil {
	142	+	_ = resp.Body.Close()
	143	+	}
	144	+	})
	145	+	}
	146	+	a, b := channel(t, svc, "sb-a", "job-a"), channel(t, svc, "sb-b", "job-b")
	147	+	for i := range RegistryDownloadsPerChannel + 1 {
	148	+	pull(a, i)
	149	+	}
	150	+	waitUntil(t, func() bool { return started.Load() == RegistryDownloadsPerChannel })
	151	+	time.Sleep(50 * time.Millisecond) // the one over A's quota would have started by now
	152	+	if n := started.Load(); n != RegistryDownloadsPerChannel {
	153	+	t.Fatalf("channel A started %d downloads, its quota is %d", n, RegistryDownloadsPerChannel)
	154	+	}
	155	+	pull(b, 100)
	156	+	waitUntil(t, func() bool { return started.Load() == RegistryDownloadsPerChannel+1 })
	157	+	}
	158	+
	159	+	// The registry lines of a channel carry its job.
	160	+	func TestRegistryLogsJob(t *testing.T) {
	161	+	hub, _, digest := stallingHub(t)
	162	+	var logs bytes.Buffer
	163	+	var mu sync.Mutex
	164	+	proxy, err := registryproxy.New(registryproxy.Config{Dir: t.TempDir(), Max: 1 << 20, Upstream: hub.URL, AuthURL: hub.URL + "/token",
	165	+	Log: zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: writerFunc(func(p []byte) (int, error) {
	166	+	mu.Lock()
	167	+	defer mu.Unlock()
	168	+	return logs.Write(p)
	169	+	})})})
	170	+	if err != nil {
	171	+	t.Fatal(err)
	172	+	}
	173	+	svc, err := New(Config{Dir: t.TempDir(), Registry: proxy})
	174	+	if err != nil {
	175	+	t.Fatal(err)
	176	+	}
	177	+	t.Cleanup(func() { _ = svc.CloseAll() })
	178	+	resp, err := channel(t, svc, "sb-a", "job-a").Get("http://cache" + RegistryPath + "library/alpine/manifests/" + digest)
	179	+	if err != nil {
	180	+	t.Fatal(err)
	181	+	}
	182	+	_ = resp.Body.Close()
	183	+	if resp.StatusCode != http.StatusOK {
	184	+	t.Fatal(resp.StatusCode)
	185	+	}
	186	+	mu.Lock()
	187	+	defer mu.Unlock()
	188	+	if !strings.Contains(logs.String(), `"job_id":"job-a"`) \|\| !strings.Contains(logs.String(), `"message":"registry manifest"`) {
	189	+	t.Fatalf("logs: %s", logs.String())
	190	+	}
	191	+	}
	192	+
	193	+	type writerFunc func(p []byte) (int, error)
	194	+
	195	+	func (f writerFunc) Write(p []byte) (int, error) { return f(p) }

A internal/cachesvc/registry_test.go +37 -0

@@ -0,0 +1,37 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package cachesvc
	4	+
	5	+	import (
	6	+	"io"
	7	+	"net/http"
	8	+	"net/http/httptest"
	9	+	"testing"
	10	+	)
	11	+
	12	+	// The registry is served under RegistryPath on the cache handler and the
	13	+	// channel, never by the results service; without one, those paths are 404.
	14	+	func TestRegistryRoute(t *testing.T) {
	15	+	for _, on := range []bool{false, true} {
	16	+	cfg := Config{}
	17	+	if on {
	18	+	cfg.Registry = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = io.WriteString(w, "registry") })
	19	+	}
	20	+	r := newRigWith(t, cfg)
	21	+	results := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	22	+	t.Error("registry fell through to results")
	23	+	w.WriteHeader(http.StatusInternalServerError)
	24	+	})
	25	+	for _, h := range []http.Handler{r.svc.Handler(), r.svc.channelHandler("job", results)} {
	26	+	w := httptest.NewRecorder()
	27	+	h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, RegistryPath, nil))
	28	+	if on {
	29	+	if w.Code != http.StatusOK \|\| w.Body.String() != "registry" {
	30	+	t.Fatalf("%d %s", w.Code, w.Body)
	31	+	}
	32	+	} else if w.Code != http.StatusNotFound {
	33	+	t.Fatal(w.Code)
	34	+	}
	35	+	}
	36	+	}
	37	+	}

@@ -0,0 +1,37 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package cachesvc
	4	+
	5	+	import (
	6	+	"io"
	7	+	"net/http"
	8	+	"net/http/httptest"
	9	+	"testing"
	10	+	)
	11	+
	12	+	// The registry is served under RegistryPath on the cache handler and the
	13	+	// channel, never by the results service; without one, those paths are 404.
	14	+	func TestRegistryRoute(t *testing.T) {
	15	+	for _, on := range []bool{false, true} {
	16	+	cfg := Config{}
	17	+	if on {
	18	+	cfg.Registry = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = io.WriteString(w, "registry") })
	19	+	}
	20	+	r := newRigWith(t, cfg)
	21	+	results := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	22	+	t.Error("registry fell through to results")
	23	+	w.WriteHeader(http.StatusInternalServerError)
	24	+	})
	25	+	for _, h := range []http.Handler{r.svc.Handler(), r.svc.channelHandler("job", results)} {
	26	+	w := httptest.NewRecorder()
	27	+	h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, RegistryPath, nil))
	28	+	if on {
	29	+	if w.Code != http.StatusOK \|\| w.Body.String() != "registry" {
	30	+	t.Fatalf("%d %s", w.Code, w.Body)
	31	+	}
	32	+	} else if w.Code != http.StatusNotFound {
	33	+	t.Fatal(w.Code)
	34	+	}
	35	+	}
	36	+	}
	37	+	}

M internal/cachesvc/service.go +39 -13


@@ -58,6 +58,17 @@ const (
58	58		// blocks at once by default, besides downloads and the cache.
59	59		DefaultConnsPerListener = 32
60	60		DefaultConns = 128
	61	+	// DefaultRegistryPerChannel bounds the registry requests of one channel in
	62	+	// progress at once, so that one sandbox's pulls leave room for its cache and
	63	+	// artifact traffic. dockerd pulls 3 layers at once by default.
	64	+	DefaultRegistryPerChannel = 8
	65	+	// RegistryDownloadsPerChannel bounds the cold Docker Hub downloads one
	66	+	// channel may have started at once (dockerd's default
	67	+	// max-concurrent-downloads), so that one sandbox cannot take every slot of
	68	+	// the registry proxy.
	69	+	RegistryDownloadsPerChannel = 3
	70	+	// RegistryPath is the prefix of the Docker registry API on the channel.
	71	+	RegistryPath = "/v2/"
61	72		)
62	73
63	74		// Config configures a Service.

@@ -78,10 +89,13 @@ type Config struct {
78	89		// Archives, when not nil, is the host's cache of action archives, served on
79	90		// the channels (see ArchiveURL).
80	91		Archives *actionscache.Cache
	92	+	// Registry is an anonymous pull-only registry on the private channel.
	93	+	Registry http.Handler
81	94		// The limits below take the Default… values (or those of the store) when
82	95		// zero; tests set them.
83	96		ProgressTimeout time.Duration
84	97		ConnsPerListener, Conns int
	98	+	RegistryPerChannel int
85	99		MinCharge int64
86	100		MaxPending, MaxPendingPerJob int
87	101		// hookChunk, when set, is called by an upload once its range is admitted and

@@ -102,11 +116,14 @@ type Service struct {
102	116		wg sync.WaitGroup
103	117		closed bool
104	118
105		-	archives *actionscache.Cache
106		-	hookChunk func(id int64)
107		-	progress time.Duration
108		-	perListener int
109		-	connsGlobal chan struct{}
	119	+	archives *actionscache.Cache
	120	+	registry http.Handler
	121	+	// registryPerChannel caps the registry requests in progress per handler.
	122	+	registryPerChannel int
	123	+	hookChunk func(id int64)
	124	+	progress time.Duration
	125	+	perListener int
	126	+	connsGlobal chan struct{}
110	127		}
111	128
112	129		type listener struct {

@@ -144,8 +161,11 @@ func New(cfg Config) (*Service, error) {
144	161		if cfg.Conns <= 0 {
145	162		cfg.Conns = DefaultConns
146	163		}
	164	+	if cfg.RegistryPerChannel <= 0 {
	165	+	cfg.RegistryPerChannel = DefaultRegistryPerChannel
	166	+	}
147	167		s := &Service{store: st, sign: sg, now: cfg.Now, log: cfg.Log, channels: map[string]*listener{}, stop: make(chan struct{}),
148		-	archives: cfg.Archives, hookChunk: cfg.hookChunk, progress: cfg.ProgressTimeout, perListener: cfg.ConnsPerListener, connsGlobal: make(chan struct{}, cfg.Conns)}
	168	+	archives: cfg.Archives, registry: cfg.Registry, hookChunk: cfg.hookChunk, progress: cfg.ProgressTimeout, perListener: cfg.ConnsPerListener, connsGlobal: make(chan struct{}, cfg.Conns), registryPerChannel: cfg.RegistryPerChannel}
149	169		s.wg.Add(1)
150	170		go s.housekeeping()
151	171		return s, nil

@@ -169,6 +189,9 @@ func (s *Service) housekeeping() {
169	189		}
170	190		}
171	191
	192	+	// RegistryEnabled reports whether the channel serves an anonymous registry.
	193	+	func (s *Service) RegistryEnabled() bool { return s.registry != nil }
	194	+
172	195		// Token returns the ACTIONS_RUNTIME_TOKEN of a job. It is valid for ttl, which
173	196		// should be the job timeout plus a margin. sc is the job's cache scope (the
174	197		// zero Scope for a job without a cache: the cache then refuses the token);

@@ -292,14 +315,17 @@ func (s *Service) CloseAll() error {
292	315		return s.store.flush()
293	316		}
294	317
295		-	// ParseMax reads ERMITE_RUNNER_CACHE_MAX: a number of bytes or a size with a K,
296		-	// M, G or T suffix (optionally followed by iB or B). Empty means DefaultMax.
297		-	func ParseMax(v string) (int64, error) {
	318	+	// ParseSize reads a size: a number of bytes or a number with a K, M, G or T
	319	+	// suffix (powers of 1024), optionally followed by iB or B, such as 10G, 500M
	320	+	// or 2GiB. Surrounding spaces are ignored; empty means def. Zero in any unit
	321	+	// (0, 0G, 0GiB) is 0. The error names no variable: the caller prefixes it.
	322	+	func ParseSize(v string, def int64) (int64, error) {
	323	+	v = strings.TrimSpace(v)
298	324		if v == "" {
299		-	return DefaultMax, nil
	325	+	return def, nil
300	326		}
301	327		s := []byte(v)
302		-	for len(s) > 0 && (s[len(s)-1] == 'B' \|\| s[len(s)-1] == 'b' \|\| s[len(s)-1] == 'i') {
	328	+	for len(s) > 0 && strings.IndexByte("bBiI", s[len(s)-1]) >= 0 {
303	329		s = s[:len(s)-1]
304	330		}
305	331		mult := int64(1)

@@ -316,8 +342,8 @@ func ParseMax(v string) (int64, error) {
316	342		}
317	343		}
318	344		n, err := strconv.ParseInt(string(s), 10, 64)
319		-	if err != nil \|\| n <= 0 \|\| n > (1<<62)/mult {
320		-	return 0, fmt.Errorf("%s: %q is not a size (examples: 10G, 500M)", EnvMax, v)
	345	+	if err != nil \|\| n < 0 \|\| n > (1<<62)/mult {
	346	+	return 0, fmt.Errorf("%q is not a size (examples: 10G, 500M)", v)
321	347		}
322	348		return n * mult, nil
323	349		}

 	// blocks at once by default, besides downloads and the cache.
 	DefaultConnsPerListener = 32
 	DefaultConns            = 128
+	// DefaultRegistryPerChannel bounds the registry requests of one channel in
+	// progress at once, so that one sandbox's pulls leave room for its cache and
+	// artifact traffic. dockerd pulls 3 layers at once by default.
+	DefaultRegistryPerChannel = 8
+	// RegistryDownloadsPerChannel bounds the cold Docker Hub downloads one
+	// channel may have started at once (dockerd's default
+	// max-concurrent-downloads), so that one sandbox cannot take every slot of
+	// the registry proxy.
+	RegistryDownloadsPerChannel = 3
+	// RegistryPath is the prefix of the Docker registry API on the channel.
+	RegistryPath = "/v2/"
 )
 // Config configures a Service.
 	// Archives, when not nil, is the host's cache of action archives, served on
 	// the channels (see ArchiveURL).
 	Archives *actionscache.Cache
+	// Registry is an anonymous pull-only registry on the private channel.
+	Registry http.Handler
 	// The limits below take the Default… values (or those of the store) when
 	// zero; tests set them.
 	ProgressTimeout              time.Duration
 	ConnsPerListener, Conns      int
+	RegistryPerChannel           int
 	MinCharge                    int64
 	MaxPending, MaxPendingPerJob int
 	// hookChunk, when set, is called by an upload once its range is admitted and
 	wg       sync.WaitGroup
 	closed   bool
-	archives    *actionscache.Cache
+	archives *actionscache.Cache
-	hookChunk   func(id int64)
+	registry http.Handler
-	progress    time.Duration
+	// registryPerChannel caps the registry requests in progress per handler.
-	perListener int
+	registryPerChannel int
-	connsGlobal chan struct{}
+	hookChunk          func(id int64)
+	progress           time.Duration
+	perListener        int
+	connsGlobal        chan struct{}
 }
 type listener struct {
 	if cfg.Conns <= 0 {
 		cfg.Conns = DefaultConns
 	}
+	if cfg.RegistryPerChannel <= 0 {
+		cfg.RegistryPerChannel = DefaultRegistryPerChannel
+	}
 	s := &Service{store: st, sign: sg, now: cfg.Now, log: cfg.Log, channels: map[string]*listener{}, stop: make(chan struct{}),
-		archives: cfg.Archives, hookChunk: cfg.hookChunk, progress: cfg.ProgressTimeout, perListener: cfg.ConnsPerListener, connsGlobal: make(chan struct{}, cfg.Conns)}
+		archives: cfg.Archives, registry: cfg.Registry, hookChunk: cfg.hookChunk, progress: cfg.ProgressTimeout, perListener: cfg.ConnsPerListener, connsGlobal: make(chan struct{}, cfg.Conns), registryPerChannel: cfg.RegistryPerChannel}
 	s.wg.Add(1)
 	go s.housekeeping()
 	return s, nil
 	}
 }
+// RegistryEnabled reports whether the channel serves an anonymous registry.
+func (s *Service) RegistryEnabled() bool { return s.registry != nil }
 // Token returns the ACTIONS_RUNTIME_TOKEN of a job. It is valid for ttl, which
 // should be the job timeout plus a margin. sc is the job's cache scope (the
 // zero Scope for a job without a cache: the cache then refuses the token);
 	return s.store.flush()
 }
-// ParseMax reads ERMITE_RUNNER_CACHE_MAX: a number of bytes or a size with a K,
+// ParseSize reads a size: a number of bytes or a number with a K, M, G or T
-// M, G or T suffix (optionally followed by iB or B). Empty means DefaultMax.
+// suffix (powers of 1024), optionally followed by iB or B, such as 10G, 500M
-func ParseMax(v string) (int64, error) {
+// or 2GiB. Surrounding spaces are ignored; empty means def. Zero in any unit
+// (0, 0G, 0GiB) is 0. The error names no variable: the caller prefixes it.
+func ParseSize(v string, def int64) (int64, error) {
+	v = strings.TrimSpace(v)
 	if v == "" {
-		return DefaultMax, nil
+		return def, nil
 	}
 	s := []byte(v)
-	for len(s) > 0 && (s[len(s)-1] == 'B' || s[len(s)-1] == 'b' || s[len(s)-1] == 'i') {
+	for len(s) > 0 && strings.IndexByte("bBiI", s[len(s)-1]) >= 0 {
 		s = s[:len(s)-1]
 	}
 	mult := int64(1)
 		}
 	}
 	n, err := strconv.ParseInt(string(s), 10, 64)
-	if err != nil || n <= 0 || n > (1<<62)/mult {
+	if err != nil || n < 0 || n > (1<<62)/mult {
-		return 0, fmt.Errorf("%s: %q is not a size (examples: 10G, 500M)", EnvMax, v)
+		return 0, fmt.Errorf("%q is not a size (examples: 10G, 500M)", v)
 	}
 	return n * mult, nil
 }

M internal/cachesvc/service_test.go +10 -5


@@ -600,15 +600,20 @@ func TestListDiagnostics(t *testing.T) {
600	600		}
601	601		}
602	602
603		-	func TestParseMax(t *testing.T) {
604		-	for in, want := range map[string]int64{"": DefaultMax, "10G": 10 << 30, "500m": 500 << 20, "1GiB": 1 << 30, "2048": 2048, "3TB": 3 << 40} {
605		-	if got, err := ParseMax(in); err != nil \|\| got != want {
	603	+	func TestParseSize(t *testing.T) {
	604	+	const def = 7
	605	+	for in, want := range map[string]int64{"": def, " ": def, "10G": 10 << 30, "500m": 500 << 20, "1GiB": 1 << 30, "2048": 2048, "3TB": 3 << 40, " 2G ": 2 << 30,
	606	+	"0": 0, "00": 0, "0G": 0, "0GiB": 0, "2GIB": 2 << 30} {
	607	+	if got, err := ParseSize(in, def); err != nil \|\| got != want {
606	608		t.Errorf("%q: %d %v", in, got, err)
607	609		}
608	610		}
609		-	for _, in := range []string{"0", "-1", "G", "ten", "1.5G", "99999999999T"} {
610		-	if _, err := ParseMax(in); err == nil {
	611	+	for _, in := range []string{"-1", "G", "ten", "1.5G", "99999999999T"} {
	612	+	_, err := ParseSize(in, def)
	613	+	if err == nil {
611	614		t.Errorf("%q accepted", in)
	615	+	} else if strings.Contains(err.Error(), "ERMITE_") {
	616	+	t.Errorf("%q: the error names a variable: %v", in, err)
612	617		}
613	618		}
614	619		}

 	}
 }
-func TestParseMax(t *testing.T) {
+func TestParseSize(t *testing.T) {
-	for in, want := range map[string]int64{"": DefaultMax, "10G": 10 << 30, "500m": 500 << 20, "1GiB": 1 << 30, "2048": 2048, "3TB": 3 << 40} {
+	const def = 7
-		if got, err := ParseMax(in); err != nil || got != want {
+	for in, want := range map[string]int64{"": def, "  ": def, "10G": 10 << 30, "500m": 500 << 20, "1GiB": 1 << 30, "2048": 2048, "3TB": 3 << 40, " 2G ": 2 << 30,
+		"0": 0, "00": 0, "0G": 0, "0GiB": 0, "2GIB": 2 << 30} {
+		if got, err := ParseSize(in, def); err != nil || got != want {
 			t.Errorf("%q: %d %v", in, got, err)
 		}
 	}
-	for _, in := range []string{"0", "-1", "G", "ten", "1.5G", "99999999999T"} {
+	for _, in := range []string{"-1", "G", "ten", "1.5G", "99999999999T"} {
-		if _, err := ParseMax(in); err == nil {
+		_, err := ParseSize(in, def)
+		if err == nil {
 			t.Errorf("%q accepted", in)
+		} else if strings.Contains(err.Error(), "ERMITE_") {
+			t.Errorf("%q: the error names a variable: %v", in, err)
 		}
 	}
 }

M internal/dnsfwd/dnsfwd.go +20 -140


@@ -39,15 +39,8 @@ import (
39	39		const (
40	40		// TCPLifetime is the absolute lifetime of a TCP connection, set at accept.
41	41		TCPLifetime = 30 * time.Second
42		-	// UpstreamTimeout is the deadline of one upstream TCP attempt.
	42	+	// UpstreamTimeout is the deadline of one upstream attempt.
43	43		UpstreamTimeout = 3 * time.Second
44		-	// UDPAttemptTimeout is the deadline of one upstream UDP attempt. It is short
45		-	// on purpose: glibc guests use "timeout:2 attempts:2", so the reply (or its
46		-	// TC=1 / TCP fallback) must reach them well inside the first 2 s window.
47		-	UDPAttemptTimeout = time.Second
48		-	// MaxExchange caps one whole exchange over all upstreams and transports, so
49		-	// a long upstream list cannot hold an in-flight slot for long.
50		-	MaxExchange = 6 * time.Second
51	44		// TCPIdleTimeout closes TCP client connections that stay silent.
52	45		TCPIdleTimeout = 10 * time.Second
53	46

@@ -60,9 +53,6 @@ const (
60	53
61	54		maxUDP = 4096 // largest datagram relayed (EDNS0 sizes up to 4096)
62	55		hdrLen = 12
63		-
64		-	// plainUDPSize is the largest UDP reply for a client without EDNS (RFC 1035).
65		-	plainUDPSize = 512
66	56		)
67	57
68	58		// Forwarder relays DNS queries. The zero value is not usable; use New.

@@ -71,11 +61,6 @@ type Forwarder struct {
71	61		port int
72	62		// Lifetime overrides TCPLifetime; set before Listen (tests).
73	63		Lifetime time.Duration
74		-	// Upstream deadlines, set to the constants above by New; tests shorten
75		-	// them before Listen.
76		-	udpTimeout, tcpTimeout, maxExchange time.Duration
77		-	// dial opens upstream connections; tests replace it with in-memory fakes.
78		-	dial func(ctx context.Context, network, addr string) (net.Conn, error)
79	64		// FreeBind binds with IP_FREEBIND, so that the gateway address may be
80	65		// configured after the listener. Off by default: with Docker the address
81	66		// exists when the network is created and a missing one is an error to

@@ -112,12 +97,7 @@ func New(upstreams func() []string, port int) *Forwarder {
112	97		upstreams: upstreams,
113	98		port: port,
114	99		Lifetime: TCPLifetime,
115		-
116		-	udpTimeout: UDPAttemptTimeout,
117		-	tcpTimeout: UpstreamTimeout,
118		-	maxExchange: MaxExchange,
119		-	dial: (&net.Dialer{}).DialContext,
120		-	listeners: map[netip.Addr]*listener{},
	100	+	listeners: map[netip.Addr]*listener{},
121	101		}
122	102		}
123	103

@@ -252,11 +232,6 @@ func (f Forwarder) serveUDP(ctx context.Context, l listener, log zerowrap.Logg
252	232		go func() {
253	233		defer func() { <-l.inflight }()
254	234		resp := f.exchange(ctx, l, q, false)
255		-	// A reply larger than the client accepts would be dropped or cut
256		-	// by its stack: send TC=1 so that it retries over TCP at once.
257		-	if len(resp) > clientUDPSize(q) {
258		-	resp = truncated(resp)
259		-	}
260	235		if len(resp) > 0 {
261	236		_, _ = pc.WriteTo(resp, from)
262	237		}

@@ -336,130 +311,34 @@ func (f Forwarder) serveConn(ctx context.Context, l listener, c net.Conn, dead
336	311		}
337	312		}
338	313
339		-	// exchange relays q to the upstreams in order and returns the first answer,
340		-	// SERVFAIL when none answers, or nil when ctx is cancelled. A TCP client is
341		-	// served over TCP. A UDP client is served over UDP for at most udpTimeout; if
342		-	// that fails, times out or returns TC=1, the same upstream is retried over TCP
343		-	// (at most tcpTimeout) before the next upstream is tried, because some
344		-	// resolvers or network paths lose large plain-UDP answers. Each upstream gets
345		-	// its own budget, and the whole exchange is capped at maxExchange. The
	314	+	// exchange relays q to the upstreams in order and returns the first answer, or
	315	+	// a SERVFAIL when none answers. The transport towards the upstream matches the
	316	+	// client's: a client that fell back to TCP gets a complete answer. The
346	317		// overridden name, if any, is answered without asking the upstreams.
347	318		func (f Forwarder) exchange(ctx context.Context, l listener, q []byte, overTCP bool) []byte {
348	319		if resp := f.overrideAnswer(l.addr, q); resp != nil {
349	320		return resp
350	321		}
351		-	cctx, cancel := context.WithTimeout(ctx, f.maxExchange)
352		-	defer cancel()
353		-	valid := func(resp []byte, err error) bool {
354		-	return err == nil && len(resp) >= hdrLen && resp[0] == q[0] && resp[1] == q[1]
355		-	}
356	322		for _, up := range f.upstreams() {
357		-	if cctx.Err() != nil {
358		-	break
359		-	}
	323	+	var resp []byte
	324	+	var err error
360	325		if overTCP {
361		-	if resp, err := f.exchangeTCP(cctx, up, q, f.tcpTimeout); valid(resp, err) {
362		-	return resp
363		-	}
364		-	continue
365		-	}
366		-	resp, err := f.exchangeUDP(cctx, up, q, f.udpTimeout)
367		-	var tc []byte
368		-	if valid(resp, err) {
369		-	if resp[2]&0x02 == 0 {
370		-	return resp
371		-	}
372		-	tc = resp
373		-	}
374		-	if full, err := f.exchangeTCP(cctx, up, q, f.tcpTimeout); valid(full, err) {
375		-	return full
	326	+	resp, err = exchangeTCP(ctx, up, q)
	327	+	} else {
	328	+	resp, err = exchangeUDP(ctx, up, q)
376	329		}
377		-	if tc != nil {
378		-	return tc // the client retries over TCP itself
	330	+	if err == nil && len(resp) >= hdrLen && resp[0] == q[0] && resp[1] == q[1] {
	331	+	return resp
379	332		}
380	333		}
381		-	if ctx.Err() != nil {
382		-	return nil
383		-	}
384	334		return servfail(q)
385	335		}
386	336
387		-	// clientUDPSize returns the largest UDP reply the client of q accepts: 512
388		-	// without EDNS, else its advertised OPT size (at least 512, at most maxUDP).
389		-	func clientUDPSize(q []byte) int {
390		-	size := plainUDPSize
391		-	if len(q) < hdrLen {
392		-	return size
393		-	}
394		-	off := hdrLen
395		-	for i := 0; i < int(binary.BigEndian.Uint16(q[4:])); i++ { // questions
396		-	if off = skipName(q, off); off < 0 \|\| off+4 > len(q) {
397		-	return size
398		-	}
399		-	off += 4
400		-	}
401		-	rrs := int(binary.BigEndian.Uint16(q[6:])) + int(binary.BigEndian.Uint16(q[8:])) + int(binary.BigEndian.Uint16(q[10:]))
402		-	for i := 0; i < rrs; i++ {
403		-	if off = skipName(q, off); off < 0 \|\| off+10 > len(q) {
404		-	return size
405		-	}
406		-	if binary.BigEndian.Uint16(q[off:]) == 41 { // OPT: class is the UDP size
407		-	size = max(size, int(binary.BigEndian.Uint16(q[off+2:])))
408		-	break
409		-	}
410		-	off += 10 + int(binary.BigEndian.Uint16(q[off+8:]))
411		-	}
412		-	return min(size, maxUDP)
413		-	}
414		-
415		-	// skipName returns the offset after the name at off, or -1 if malformed.
416		-	func skipName(m []byte, off int) int {
417		-	for off < len(m) {
418		-	switch c := int(m[off]); {
419		-	case c == 0:
420		-	return off + 1
421		-	case c&0xc0 == 0xc0:
422		-	if off+2 > len(m) {
423		-	return -1
424		-	}
425		-	return off + 2
426		-	case c&0xc0 != 0:
427		-	return -1
428		-	default:
429		-	off += 1 + c
430		-	}
431		-	}
432		-	return -1
433		-	}
434		-
435		-	// truncated returns the header and question of resp with TC=1 and no records.
436		-	func truncated(resp []byte) []byte {
437		-	if len(resp) < hdrLen {
438		-	return nil
439		-	}
440		-	end := hdrLen
441		-	for i := 0; i < int(binary.BigEndian.Uint16(resp[4:])); i++ {
442		-	if end = skipName(resp, end); end < 0 \|\| end+4 > len(resp) {
443		-	end = hdrLen
444		-	break
445		-	}
446		-	end += 4
447		-	}
448		-	r := append([]byte(nil), resp[:end]...)
449		-	r[2] \|= 0x02 // TC
450		-	if end == hdrLen {
451		-	r[4], r[5] = 0, 0
452		-	}
453		-	for i := 6; i < hdrLen; i++ { // ANCOUNT, NSCOUNT, ARCOUNT
454		-	r[i] = 0
455		-	}
456		-	return r
457		-	}
458		-
459		-	func (f *Forwarder) exchangeUDP(ctx context.Context, up string, q []byte, timeout time.Duration) ([]byte, error) {
460		-	ctx, cancel := context.WithTimeout(ctx, timeout)
	337	+	func exchangeUDP(ctx context.Context, up string, q []byte) ([]byte, error) {
	338	+	ctx, cancel := context.WithTimeout(ctx, UpstreamTimeout)
461	339		defer cancel()
462		-	c, err := f.dial(ctx, "udp", up)
	340	+	var d net.Dialer
	341	+	c, err := d.DialContext(ctx, "udp", up)
463	342		if err != nil {
464	343		return nil, err
465	344		}

@@ -482,10 +361,11 @@ func (f *Forwarder) exchangeUDP(ctx context.Context, up string, q []byte, timeou
482	361		}
483	362		}
484	363
485		-	func (f *Forwarder) exchangeTCP(ctx context.Context, up string, q []byte, timeout time.Duration) ([]byte, error) {
486		-	ctx, cancel := context.WithTimeout(ctx, timeout)
	364	+	func exchangeTCP(ctx context.Context, up string, q []byte) ([]byte, error) {
	365	+	ctx, cancel := context.WithTimeout(ctx, UpstreamTimeout)
487	366		defer cancel()
488		-	c, err := f.dial(ctx, "tcp", up)
	367	+	var d net.Dialer
	368	+	c, err := d.DialContext(ctx, "tcp", up)
489	369		if err != nil {
490	370		return nil, err
491	371		}


@@ -39,15 +39,8 @@ import (
39		const (	39		const (
40		// TCPLifetime is the absolute lifetime of a TCP connection, set at accept.	40		// TCPLifetime is the absolute lifetime of a TCP connection, set at accept.
41		TCPLifetime = 30 * time.Second	41		TCPLifetime = 30 * time.Second
42	-	// UpstreamTimeout is the deadline of one upstream TCP attempt.	42	+	// UpstreamTimeout is the deadline of one upstream attempt.
43		UpstreamTimeout = 3 * time.Second	43		UpstreamTimeout = 3 * time.Second
44	-	// UDPAttemptTimeout is the deadline of one upstream UDP attempt. It is short
45	-	// on purpose: glibc guests use "timeout:2 attempts:2", so the reply (or its
46	-	// TC=1 / TCP fallback) must reach them well inside the first 2 s window.
47	-	UDPAttemptTimeout = time.Second
48	-	// MaxExchange caps one whole exchange over all upstreams and transports, so
49	-	// a long upstream list cannot hold an in-flight slot for long.
50	-	MaxExchange = 6 * time.Second
51		// TCPIdleTimeout closes TCP client connections that stay silent.	44		// TCPIdleTimeout closes TCP client connections that stay silent.
52		TCPIdleTimeout = 10 * time.Second	45		TCPIdleTimeout = 10 * time.Second
53			46

@@ -60,9 +53,6 @@ const (
60			53
61		maxUDP = 4096 // largest datagram relayed (EDNS0 sizes up to 4096)	54		maxUDP = 4096 // largest datagram relayed (EDNS0 sizes up to 4096)
62		hdrLen = 12	55		hdrLen = 12
63	-
64	-	// plainUDPSize is the largest UDP reply for a client without EDNS (RFC 1035).
65	-	plainUDPSize = 512
66		)	56		)
67			57
68		// Forwarder relays DNS queries. The zero value is not usable; use New.	58		// Forwarder relays DNS queries. The zero value is not usable; use New.

@@ -71,11 +61,6 @@ type Forwarder struct {
71		port int	61		port int
72		// Lifetime overrides TCPLifetime; set before Listen (tests).	62		// Lifetime overrides TCPLifetime; set before Listen (tests).
73		Lifetime time.Duration	63		Lifetime time.Duration
74	-	// Upstream deadlines, set to the constants above by New; tests shorten
75	-	// them before Listen.
76	-	udpTimeout, tcpTimeout, maxExchange time.Duration
77	-	// dial opens upstream connections; tests replace it with in-memory fakes.
78	-	dial func(ctx context.Context, network, addr string) (net.Conn, error)
79		// FreeBind binds with IP_FREEBIND, so that the gateway address may be	64		// FreeBind binds with IP_FREEBIND, so that the gateway address may be
80		// configured after the listener. Off by default: with Docker the address	65		// configured after the listener. Off by default: with Docker the address
81		// exists when the network is created and a missing one is an error to	66		// exists when the network is created and a missing one is an error to

@@ -112,12 +97,7 @@ func New(upstreams func() []string, port int) *Forwarder {
112		upstreams: upstreams,	97		upstreams: upstreams,
113		port: port,	98		port: port,
114		Lifetime: TCPLifetime,	99		Lifetime: TCPLifetime,
115	-		100	+	listeners: map[netip.Addr]*listener{},
116	-	udpTimeout: UDPAttemptTimeout,
117	-	tcpTimeout: UpstreamTimeout,
118	-	maxExchange: MaxExchange,
119	-	dial: (&net.Dialer{}).DialContext,
120	-	listeners: map[netip.Addr]*listener{},
121		}	101		}
122		}	102		}
123			103

@@ -252,11 +232,6 @@ func (f Forwarder) serveUDP(ctx context.Context, l listener, log zerowrap.Logg
252		go func() {	232		go func() {
253		defer func() { <-l.inflight }()	233		defer func() { <-l.inflight }()
254		resp := f.exchange(ctx, l, q, false)	234		resp := f.exchange(ctx, l, q, false)
255	-	// A reply larger than the client accepts would be dropped or cut
256	-	// by its stack: send TC=1 so that it retries over TCP at once.
257	-	if len(resp) > clientUDPSize(q) {
258	-	resp = truncated(resp)
259	-	}
260		if len(resp) > 0 {	235		if len(resp) > 0 {
261		_, _ = pc.WriteTo(resp, from)	236		_, _ = pc.WriteTo(resp, from)
262		}	237		}

@@ -336,130 +311,34 @@ func (f Forwarder) serveConn(ctx context.Context, l listener, c net.Conn, dead
336		}	311		}
337		}	312		}
338			313
339	-	// exchange relays q to the upstreams in order and returns the first answer,	314	+	// exchange relays q to the upstreams in order and returns the first answer, or
340	-	// SERVFAIL when none answers, or nil when ctx is cancelled. A TCP client is	315	+	// a SERVFAIL when none answers. The transport towards the upstream matches the
341	-	// served over TCP. A UDP client is served over UDP for at most udpTimeout; if	316	+	// client's: a client that fell back to TCP gets a complete answer. The
342	-	// that fails, times out or returns TC=1, the same upstream is retried over TCP
343	-	// (at most tcpTimeout) before the next upstream is tried, because some
344	-	// resolvers or network paths lose large plain-UDP answers. Each upstream gets
345	-	// its own budget, and the whole exchange is capped at maxExchange. The
346		// overridden name, if any, is answered without asking the upstreams.	317		// overridden name, if any, is answered without asking the upstreams.
347		func (f Forwarder) exchange(ctx context.Context, l listener, q []byte, overTCP bool) []byte {	318		func (f Forwarder) exchange(ctx context.Context, l listener, q []byte, overTCP bool) []byte {
348		if resp := f.overrideAnswer(l.addr, q); resp != nil {	319		if resp := f.overrideAnswer(l.addr, q); resp != nil {
349		return resp	320		return resp
350		}	321		}
351	-	cctx, cancel := context.WithTimeout(ctx, f.maxExchange)
352	-	defer cancel()
353	-	valid := func(resp []byte, err error) bool {
354	-	return err == nil && len(resp) >= hdrLen && resp[0] == q[0] && resp[1] == q[1]
355	-	}
356		for _, up := range f.upstreams() {	322		for _, up := range f.upstreams() {
357	-	if cctx.Err() != nil {	323	+	var resp []byte
358	-	break	324	+	var err error
359	-	}
360		if overTCP {	325		if overTCP {
361	-	if resp, err := f.exchangeTCP(cctx, up, q, f.tcpTimeout); valid(resp, err) {	326	+	resp, err = exchangeTCP(ctx, up, q)
362	-	return resp	327	+	} else {
363	-	}	328	+	resp, err = exchangeUDP(ctx, up, q)
364	-	continue
365	-	}
366	-	resp, err := f.exchangeUDP(cctx, up, q, f.udpTimeout)
367	-	var tc []byte
368	-	if valid(resp, err) {
369	-	if resp[2]&0x02 == 0 {
370	-	return resp
371	-	}
372	-	tc = resp
373	-	}
374	-	if full, err := f.exchangeTCP(cctx, up, q, f.tcpTimeout); valid(full, err) {
375	-	return full
376		}	329		}
377	-	if tc != nil {	330	+	if err == nil && len(resp) >= hdrLen && resp[0] == q[0] && resp[1] == q[1] {
378	-	return tc // the client retries over TCP itself	331	+	return resp
379		}	332		}
380		}	333		}
381	-	if ctx.Err() != nil {
382	-	return nil
383	-	}
384		return servfail(q)	334		return servfail(q)
385		}	335		}
386			336
387	-	// clientUDPSize returns the largest UDP reply the client of q accepts: 512	337	+	func exchangeUDP(ctx context.Context, up string, q []byte) ([]byte, error) {
388	-	// without EDNS, else its advertised OPT size (at least 512, at most maxUDP).	338	+	ctx, cancel := context.WithTimeout(ctx, UpstreamTimeout)
389	-	func clientUDPSize(q []byte) int {
390	-	size := plainUDPSize
391	-	if len(q) < hdrLen {
392	-	return size
393	-	}
394	-	off := hdrLen
395	-	for i := 0; i < int(binary.BigEndian.Uint16(q[4:])); i++ { // questions
396	-	if off = skipName(q, off); off < 0 \|\| off+4 > len(q) {
397	-	return size
398	-	}
399	-	off += 4
400	-	}
401	-	rrs := int(binary.BigEndian.Uint16(q[6:])) + int(binary.BigEndian.Uint16(q[8:])) + int(binary.BigEndian.Uint16(q[10:]))
402	-	for i := 0; i < rrs; i++ {
403	-	if off = skipName(q, off); off < 0 \|\| off+10 > len(q) {
404	-	return size
405	-	}
406	-	if binary.BigEndian.Uint16(q[off:]) == 41 { // OPT: class is the UDP size
407	-	size = max(size, int(binary.BigEndian.Uint16(q[off+2:])))
408	-	break
409	-	}
410	-	off += 10 + int(binary.BigEndian.Uint16(q[off+8:]))
411	-	}
412	-	return min(size, maxUDP)
413	-	}
414	-
415	-	// skipName returns the offset after the name at off, or -1 if malformed.
416	-	func skipName(m []byte, off int) int {
417	-	for off < len(m) {
418	-	switch c := int(m[off]); {
419	-	case c == 0:
420	-	return off + 1
421	-	case c&0xc0 == 0xc0:
422	-	if off+2 > len(m) {
423	-	return -1
424	-	}
425	-	return off + 2
426	-	case c&0xc0 != 0:
427	-	return -1
428	-	default:
429	-	off += 1 + c
430	-	}
431	-	}
432	-	return -1
433	-	}
434	-
435	-	// truncated returns the header and question of resp with TC=1 and no records.
436	-	func truncated(resp []byte) []byte {
437	-	if len(resp) < hdrLen {
438	-	return nil
439	-	}
440	-	end := hdrLen
441	-	for i := 0; i < int(binary.BigEndian.Uint16(resp[4:])); i++ {
442	-	if end = skipName(resp, end); end < 0 \|\| end+4 > len(resp) {
443	-	end = hdrLen
444	-	break
445	-	}
446	-	end += 4
447	-	}
448	-	r := append([]byte(nil), resp[:end]...)
449	-	r[2] \|= 0x02 // TC
450	-	if end == hdrLen {
451	-	r[4], r[5] = 0, 0
452	-	}
453	-	for i := 6; i < hdrLen; i++ { // ANCOUNT, NSCOUNT, ARCOUNT
454	-	r[i] = 0
455	-	}
456	-	return r
457	-	}
458	-
459	-	func (f *Forwarder) exchangeUDP(ctx context.Context, up string, q []byte, timeout time.Duration) ([]byte, error) {
460	-	ctx, cancel := context.WithTimeout(ctx, timeout)
461		defer cancel()	339		defer cancel()
462	-	c, err := f.dial(ctx, "udp", up)	340	+	var d net.Dialer
			341	+	c, err := d.DialContext(ctx, "udp", up)
463		if err != nil {	342		if err != nil {
464		return nil, err	343		return nil, err
465		}	344		}

@@ -482,10 +361,11 @@ func (f *Forwarder) exchangeUDP(ctx context.Context, up string, q []byte, timeou
482		}	361		}
483		}	362		}
484			363
485	-	func (f *Forwarder) exchangeTCP(ctx context.Context, up string, q []byte, timeout time.Duration) ([]byte, error) {	364	+	func exchangeTCP(ctx context.Context, up string, q []byte) ([]byte, error) {
486	-	ctx, cancel := context.WithTimeout(ctx, timeout)	365	+	ctx, cancel := context.WithTimeout(ctx, UpstreamTimeout)
487		defer cancel()	366		defer cancel()
488	-	c, err := f.dial(ctx, "tcp", up)	367	+	var d net.Dialer
			368	+	c, err := d.DialContext(ctx, "tcp", up)
489		if err != nil {	369		if err != nil {
490		return nil, err	370		return nil, err
491		}	371		}

M internal/dnsfwd/dnsfwd_test.go +31 -374


@@ -16,7 +16,6 @@ import (
16	16		"sync/atomic"
17	17		"syscall"
18	18		"testing"
19		-	"testing/synctest"
20	19		"time"
21	20		)
22	21

@@ -40,32 +39,18 @@ func answer(q []byte) []byte {
40	39		return append(r, "ANSWER"...)
41	40		}
42	41
43		-	// listenUDPAndTCP binds UDP on a free loopback port and TCP on the same port.
44		-	// The TCP port may already be taken (outbound connections use the same
45		-	// ephemeral range), so it picks again a few times, like Listen does.
46		-	func listenUDPAndTCP(t *testing.T) (net.PacketConn, net.Listener) {
47		-	t.Helper()
48		-	for attempt := 0; ; attempt++ {
49		-	pc, err := net.ListenPacket("udp", "127.0.0.1:0")
50		-	if err != nil {
51		-	t.Skip("no loopback:", err)
52		-	}
53		-	tl, err := net.Listen("tcp", pc.LocalAddr().String())
54		-	if err == nil {
55		-	return pc, tl
56		-	}
57		-	_ = pc.Close()
58		-	if attempt == 49 \|\| !errors.Is(err, syscall.EADDRINUSE) {
59		-	t.Fatal(err)
60		-	}
61		-	}
62		-	}
63		-
64	42		// fakeUpstream serves UDP and TCP on one port of 127.0.0.1 and returns it.
65	43		func fakeUpstream(t *testing.T, handle func(q []byte) []byte) string {
66	44		t.Helper()
67		-	pc, tl := listenUDPAndTCP(t)
	45	+	pc, err := net.ListenPacket("udp", "127.0.0.1:0")
	46	+	if err != nil {
	47	+	t.Skip("no loopback:", err)
	48	+	}
68	49		addr := pc.LocalAddr().String()
	50	+	tl, err := net.Listen("tcp", addr)
	51	+	if err != nil {
	52	+	t.Fatal(err)
	53	+	}
69	54		t.Cleanup(func() { _ = pc.Close(); _ = tl.Close() })
70	55		go func() {
71	56		buf := make([]byte, 4096)

@@ -107,17 +92,8 @@ func fakeUpstream(t *testing.T, handle func(q []byte) []byte) string {
107	92		}
108	93
109	94		func start(t testing.T, ups ...string) (Forwarder, string) {
110		-	t.Helper()
111		-	return startWith(t, nil, ups...)
112		-	}
113		-
114		-	// startWith is start with a hook to adjust the Forwarder before Listen.
115		-	func startWith(t testing.T, cfg func(Forwarder), ups ...string) (*Forwarder, string) {
116	95		t.Helper()
117	96		f := New(func() []string { return ups }, 0)
118		-	if cfg != nil {
119		-	cfg(f)
120		-	}
121	97		ip := netip.MustParseAddr("127.0.0.1")
122	98		if err := f.Listen(context.Background(), ip); err != nil {
123	99		t.Fatal(err)

@@ -204,6 +180,29 @@ func TestServfailWhenNoUpstreamAnswers(t *testing.T) {
204	180		}
205	181		}
206	182
	183	+	func TestMismatchedUpstreamIDIgnored(t *testing.T) {
	184	+	// Upstream answers with a wrong ID: treated as no answer, client gets SERVFAIL.
	185	+	up := fakeUpstream(t, func(q []byte) []byte {
	186	+	r := answer(q)
	187	+	r[0] ^= 0xff
	188	+	return r
	189	+	})
	190	+	f, addr := start(t, up)
	191	+	_ = f
	192	+	c, _ := net.Dial("udp", addr)
	193	+	defer func() { _ = c.Close() }()
	194	+	_ = c.SetDeadline(time.Now().Add(2UpstreamTimeout + 2time.Second))
	195	+	_, _ = c.Write(query(5, "a.test"))
	196	+	buf := make([]byte, 512)
	197	+	n, err := c.Read(buf)
	198	+	if err != nil {
	199	+	t.Fatal(err)
	200	+	}
	201	+	if buf[3]&0x0f != 2 {
	202	+	t.Fatalf("want SERVFAIL, got %x", buf[:n])
	203	+	}
	204	+	}
	205	+
207	206		func TestShortPacketsIgnored(t *testing.T) {
208	207		up := fakeUpstream(t, answer)
209	208		_, addr := start(t, up)

@@ -613,345 +612,3 @@ func TestOverrideIgnoresMalformedAndNonQueries(t *testing.T) {
613	612		}
614	613		}
615	614		}
616		-
617		-	// bigAnswer returns an answer to q padded to size bytes, with TC set if tc.
618		-	func bigAnswer(q []byte, size int, tc bool) []byte {
619		-	r := answer(q)
620		-	if tc {
621		-	r[2] \|= 0x02
622		-	}
623		-	for len(r) < size {
624		-	r = append(r, 'x')
625		-	}
626		-	return r
627		-	}
628		-
629		-	func TestLargeAnswerNonEDNSClientGetsTC(t *testing.T) {
630		-	up := fakeUpstream(t, func(q []byte) []byte { return bigAnswer(q, 700, false) })
631		-	_, addr := start(t, up)
632		-	q := queryType(0x3333, "storage.googleapis.com", 1, false)
633		-	resp := udpExchange(t, addr, q)
634		-	if resp[2]&0x02 == 0 {
635		-	t.Fatalf("TC not set: %x", resp)
636		-	}
637		-	if len(resp) != len(q) \|\| binary.BigEndian.Uint16(resp[4:]) != 1 \|\| binary.BigEndian.Uint16(resp[6:]) != 0 {
638		-	t.Fatalf("want header+question only, got %x", resp)
639		-	}
640		-	if resp[0] != q[0] \|\| resp[1] != q[1] {
641		-	t.Fatal("ID changed")
642		-	}
643		-	}
644		-
645		-	func TestLargeAnswerEDNSClientGetsFullAnswer(t *testing.T) {
646		-	up := fakeUpstream(t, func(q []byte) []byte { return bigAnswer(q, 700, false) })
647		-	_, addr := start(t, up)
648		-	// queryType advertises 0x1000 (4096) bytes.
649		-	resp := udpExchange(t, addr, queryType(0x4444, "storage.googleapis.com", 1, true))
650		-	if len(resp) != 700 \|\| resp[2]&0x02 != 0 {
651		-	t.Fatalf("want the full 700-byte answer, got len %d tc=%d", len(resp), resp[2]&0x02)
652		-	}
653		-	}
654		-
655		-	// withOPT appends an OPT record (root name, type 41) with the given class and
656		-	// RDLEN to a query and bumps ARCOUNT.
657		-	func withOPT(q []byte, size, rdlen uint16, body int) []byte {
658		-	q = append([]byte(nil), q...)
659		-	q[11] = 1
660		-	q = append(q, 0, 0, 41)
661		-	q = binary.BigEndian.AppendUint16(q, size)
662		-	q = append(q, 0, 0, 0, 0)
663		-	q = binary.BigEndian.AppendUint16(q, rdlen)
664		-	return append(q, make([]byte, body)...)
665		-	}
666		-
667		-	func TestClientUDPSize(t *testing.T) {
668		-	plain := query(1, "a.b")
669		-	bigQD := append([]byte(nil), plain[:hdrLen]...)
670		-	bigQD[4], bigQD[5] = 0xff, 0xff
671		-	bigQD = append(bigQD, 1, 'a', 0, 0, 1, 0, 1)
672		-	ptrQ := append([]byte(nil), plain[:hdrLen]...)
673		-	ptrQ = append(ptrQ, 0xc0, hdrLen, 0, 1, 0, 1) // compression pointer in question
674		-	tests := []struct {
675		-	name string
676		-	q []byte
677		-	want int
678		-	}{
679		-	{"no edns", plain, 512},
680		-	{"edns 1232", withOPT(plain, 1232, 0, 0), 1232},
681		-	{"edns below 512", withOPT(plain, 100, 0, 0), 512},
682		-	{"edns above cap", withOPT(plain, 65535, 0, 0), maxUDP},
683		-	{"edns 4096", withOPT(plain, 4096, 0, 0), 4096},
684		-	{"qdcount 65535 short body", bigQD, 512},
685		-	{"compression pointer in question", ptrQ, 512},
686		-	{"OPT rdlen overruns buffer", withOPT(plain, 4096, 500, 3), 4096},
687		-	{"truncated header", plain[:15], 512},
688		-	{"short", plain[:5], 512},
689		-	}
690		-	for _, tt := range tests {
691		-	t.Run(tt.name, func(t *testing.T) {
692		-	if got := clientUDPSize(tt.q); got != tt.want {
693		-	t.Fatalf("got %d, want %d", got, tt.want)
694		-	}
695		-	})
696		-	}
697		-	}
698		-
699		-	func TestTruncated(t *testing.T) {
700		-	q := queryType(7, "a.b", 1, false)
701		-	full := bigAnswer(q, 700, false)
702		-	full[7] = 5 // ANCOUNT
703		-	got := truncated(full)
704		-	if len(got) != len(q) \|\| got[2]&0x02 == 0 \|\| got[6] != 0 \|\| got[7] != 0 \|\| got[4] != 0 \|\| got[5] != 1 {
705		-	t.Fatalf("bad truncated reply %x", got)
706		-	}
707		-	bad := append([]byte(nil), full...)
708		-	bad[4], bad[5] = 0xff, 0xff // QDCOUNT lies, question parse fails
709		-	got = truncated(bad)
710		-	if len(got) != hdrLen \|\| got[2]&0x02 == 0 \|\| got[4] != 0 \|\| got[5] != 0 {
711		-	t.Fatalf("bad-question case: %x", got)
712		-	}
713		-	if truncated(full[:5]) != nil {
714		-	t.Fatal("short input must give nil")
715		-	}
716		-	}
717		-
718		-	func TestServfailAndOverrideRepliesAreNotTruncated(t *testing.T) {
719		-	dead := "127.0.0.1:1" // nothing listens: UDP and TCP both fail at once
720		-	_, addr := startOverride(t, "forge.test", dead)
721		-	resp := udpExchange(t, addr, queryType(9, "x.test", 1, false))
722		-	if len(resp) != hdrLen \|\| resp[3]&0x0f != 2 \|\| resp[2]&0x02 != 0 {
723		-	t.Fatalf("want a plain 12-byte SERVFAIL, got %x", resp)
724		-	}
725		-	resp = udpExchange(t, addr, queryType(10, "forge.test", 1, false))
726		-	if resp[2]&0x02 != 0 \|\| resp[3]&0x0f != 0 \|\| binary.BigEndian.Uint16(resp[6:]) != 1 {
727		-	t.Fatalf("want the override A answer, got %x", resp)
728		-	}
729		-	}
730		-
731		-	// fakeNet is an in-memory upstream network for Forwarder.dial. One Write on a
732		-	// UDP conn is one datagram (net.Pipe delivers a Write to a single Read when the
733		-	// reader's buffer is large enough). A handler returning nil never answers.
734		-	type fakeNet struct {
735		-	udp, tcp func(addr string, q []byte) []byte
736		-	udpCalls, tcpCalls atomic.Int64
737		-	}
738		-
739		-	func (n *fakeNet) dial(ctx context.Context, network, addr string) (net.Conn, error) {
740		-	if err := ctx.Err(); err != nil { // like net.Dialer
741		-	return nil, err
742		-	}
743		-	c, s := net.Pipe()
744		-	switch network {
745		-	case "udp":
746		-	go func() {
747		-	defer func() { _ = s.Close() }()
748		-	buf := make([]byte, 4096)
749		-	for {
750		-	k, err := s.Read(buf)
751		-	if err != nil {
752		-	return
753		-	}
754		-	n.udpCalls.Add(1)
755		-	if r := n.udp(addr, append([]byte(nil), buf[:k]...)); r != nil {
756		-	if _, err := s.Write(r); err != nil {
757		-	return
758		-	}
759		-	}
760		-	}
761		-	}()
762		-	case "tcp":
763		-	go func() {
764		-	defer func() { _ = s.Close() }()
765		-	var l [2]byte
766		-	if _, err := io.ReadFull(s, l[:]); err != nil {
767		-	return
768		-	}
769		-	q := make([]byte, binary.BigEndian.Uint16(l[:]))
770		-	if _, err := io.ReadFull(s, q); err != nil {
771		-	return
772		-	}
773		-	n.tcpCalls.Add(1)
774		-	if r := n.tcp(addr, q); r != nil {
775		-	_, _ = s.Write(append(binary.BigEndian.AppendUint16(nil, uint16(len(r))), r...))
776		-	}
777		-	_, _ = io.Copy(io.Discard, s) // wait for the client to hang up
778		-	}()
779		-	default:
780		-	return nil, errors.New("unexpected network " + network)
781		-	}
782		-	return c, nil
783		-	}
784		-
785		-	// silent is a handler that never answers.
786		-	func silent(string, []byte) []byte { return nil }
787		-
788		-	// fakeForwarder returns a Forwarder with production timeouts whose upstreams
789		-	// are served by n, and a listener for exchange.
790		-	func fakeForwarder(n fakeNet, ups ...string) (Forwarder, *listener) {
791		-	f := New(func() []string { return ups }, 0)
792		-	f.dial = n.dial
793		-	return f, &listener{addr: netip.MustParseAddr("127.0.0.1")}
794		-	}
795		-
796		-	func isServfail(r []byte, id uint16) bool {
797		-	return len(r) == hdrLen && binary.BigEndian.Uint16(r) == id && r[3]&0x0f == 2
798		-	}
799		-
800		-	// The tests below run in a synctest bubble: the fake clock makes the real
801		-	// production timeouts (1 s UDP, 3 s TCP, 6 s cap) instant and exact.
802		-
803		-	func TestExchangeSilentUDPFallsBackToTCPAfterUDPTimeout(t *testing.T) {
804		-	synctest.Test(t, func(t *testing.T) {
805		-	n := &fakeNet{udp: silent, tcp: func(_ string, q []byte) []byte { return answer(q) }}
806		-	f, l := fakeForwarder(n, "up:53")
807		-	start := time.Now()
808		-	resp := f.exchange(t.Context(), l, query(1, "storage.googleapis.com"), false)
809		-	if !strings.HasSuffix(string(resp), "ANSWER") {
810		-	t.Fatalf("want the TCP answer, got %x", resp)
811		-	}
812		-	if got := time.Since(start); got != UDPAttemptTimeout {
813		-	t.Fatalf("answered after %v, want %v", got, UDPAttemptTimeout)
814		-	}
815		-	if n.udpCalls.Load() != 1 \|\| n.tcpCalls.Load() != 1 {
816		-	t.Fatalf("udp=%d tcp=%d, want 1 and 1", n.udpCalls.Load(), n.tcpCalls.Load())
817		-	}
818		-	})
819		-	}
820		-
821		-	func TestExchangeTruncatedUDPRetriedOverTCP(t *testing.T) {
822		-	synctest.Test(t, func(t *testing.T) {
823		-	n := &fakeNet{
824		-	udp: func(_ string, q []byte) []byte { return bigAnswer(q, 0, true) },
825		-	tcp: func(_ string, q []byte) []byte { return bigAnswer(q, 700, false) },
826		-	}
827		-	f, l := fakeForwarder(n, "up:53")
828		-	start := time.Now()
829		-	resp := f.exchange(t.Context(), l, query(2, "example.com"), false)
830		-	if len(resp) != 700 \|\| resp[2]&0x02 != 0 {
831		-	t.Fatalf("want the full TCP answer, got len %d tc=%d", len(resp), resp[2]&0x02)
832		-	}
833		-	if time.Since(start) != 0 {
834		-	t.Fatalf("took %v, want no waiting", time.Since(start))
835		-	}
836		-	})
837		-	}
838		-
839		-	func TestExchangeKeepsTruncatedReplyWhenTCPFails(t *testing.T) {
840		-	synctest.Test(t, func(t *testing.T) {
841		-	n := &fakeNet{udp: func(_ string, q []byte) []byte { return bigAnswer(q, 0, true) }, tcp: silent}
842		-	f, l := fakeForwarder(n, "up:53")
843		-	resp := f.exchange(t.Context(), l, query(3, "example.com"), false)
844		-	if len(resp) < hdrLen \|\| resp[2]&0x02 == 0 {
845		-	t.Fatalf("want the TC=1 reply, got %x", resp)
846		-	}
847		-	})
848		-	}
849		-
850		-	func TestExchangeSecondUpstreamNotStarved(t *testing.T) {
851		-	synctest.Test(t, func(t *testing.T) {
852		-	n := &fakeNet{
853		-	udp: func(a string, q []byte) []byte {
854		-	if a == "dead:53" {
855		-	return nil
856		-	}
857		-	return answer(q)
858		-	},
859		-	tcp: func(a string, q []byte) []byte {
860		-	if a == "dead:53" {
861		-	return nil
862		-	}
863		-	return answer(q)
864		-	},
865		-	}
866		-	f, l := fakeForwarder(n, "dead:53", "live:53")
867		-	start := time.Now()
868		-	resp := f.exchange(t.Context(), l, query(4, "a.test"), false)
869		-	if !strings.HasSuffix(string(resp), "ANSWER") {
870		-	t.Fatalf("want the second upstream's answer, got %x", resp)
871		-	}
872		-	if got, want := time.Since(start), UDPAttemptTimeout+UpstreamTimeout; got != want {
873		-	t.Fatalf("took %v, want %v", got, want)
874		-	}
875		-	})
876		-	}
877		-
878		-	func TestExchangeMismatchedUpstreamIDIgnored(t *testing.T) {
879		-	synctest.Test(t, func(t *testing.T) {
880		-	wrong := func(_ string, q []byte) []byte {
881		-	r := answer(q)
882		-	r[0] ^= 0xff
883		-	return r
884		-	}
885		-	f, l := fakeForwarder(&fakeNet{udp: wrong, tcp: wrong}, "up:53")
886		-	start := time.Now()
887		-	resp := f.exchange(t.Context(), l, query(5, "a.test"), false)
888		-	if !isServfail(resp, 5) {
889		-	t.Fatalf("want SERVFAIL, got %x", resp)
890		-	}
891		-	// UDP keeps listening for the right ID until its timeout; the TCP
892		-	// reply is read once and rejected.
893		-	if got := time.Since(start); got != UDPAttemptTimeout {
894		-	t.Fatalf("took %v, want %v", got, UDPAttemptTimeout)
895		-	}
896		-	})
897		-	}
898		-
899		-	func TestExchangeCappedAtMaxExchange(t *testing.T) {
900		-	synctest.Test(t, func(t *testing.T) {
901		-	n := &fakeNet{udp: silent, tcp: silent}
902		-	f, l := fakeForwarder(n, "a:53", "b:53", "c:53")
903		-	start := time.Now()
904		-	resp := f.exchange(t.Context(), l, query(6, "a.test"), false)
905		-	if !isServfail(resp, 6) {
906		-	t.Fatalf("want SERVFAIL, got %x", resp)
907		-	}
908		-	if got := time.Since(start); got != MaxExchange {
909		-	t.Fatalf("took %v, want %v", got, MaxExchange)
910		-	}
911		-	// a: 1 s UDP + 3 s TCP; b: 1 s UDP + 1 s TCP (cut by the cap); c: skipped.
912		-	if n.udpCalls.Load() != 2 \|\| n.tcpCalls.Load() != 2 {
913		-	t.Fatalf("udp=%d tcp=%d, want 2 and 2", n.udpCalls.Load(), n.tcpCalls.Load())
914		-	}
915		-	})
916		-	}
917		-
918		-	func TestExchangeTCPClientUsesTCPOnly(t *testing.T) {
919		-	synctest.Test(t, func(t *testing.T) {
920		-	n := &fakeNet{udp: silent, tcp: func(_ string, q []byte) []byte { return answer(q) }}
921		-	f, l := fakeForwarder(n, "up:53")
922		-	resp := f.exchange(t.Context(), l, query(7, "a.test"), true)
923		-	if !strings.HasSuffix(string(resp), "ANSWER") \|\| n.udpCalls.Load() != 0 {
924		-	t.Fatalf("resp=%x udp=%d", resp, n.udpCalls.Load())
925		-	}
926		-	})
927		-	}
928		-
929		-	func TestExchangeCancelledReturnsNil(t *testing.T) {
930		-	synctest.Test(t, func(t *testing.T) {
931		-	f, l := fakeForwarder(&fakeNet{udp: silent, tcp: silent}, "up:53")
932		-	ctx, cancel := context.WithCancel(t.Context())
933		-	time.AfterFunc(500*time.Millisecond, cancel)
934		-	start := time.Now()
935		-	if resp := f.exchange(ctx, l, query(8, "a.test"), false); resp != nil {
936		-	t.Fatalf("want nil on cancellation, got %x", resp)
937		-	}
938		-	// Cancellation is noticed at the next attempt boundary: the blocked UDP
939		-	// read runs to its own deadline, then no TCP attempt is started.
940		-	if got := time.Since(start); got != UDPAttemptTimeout {
941		-	t.Fatalf("returned after %v, want %v", got, UDPAttemptTimeout)
942		-	}
943		-	})
944		-	}
945		-
946		-	// End to end through real client-facing sockets; the upstream is in memory.
947		-	func TestUDPClientGetsTCPAnswerWhenUpstreamUDPSilent(t *testing.T) {
948		-	n := &fakeNet{udp: silent, tcp: func(_ string, q []byte) []byte { return answer(q) }}
949		-	_, addr := startWith(t, func(f *Forwarder) {
950		-	f.dial = n.dial
951		-	f.udpTimeout = 50 * time.Millisecond
952		-	}, "up:53")
953		-	resp := udpExchange(t, addr, query(0x1111, "storage.googleapis.com"))
954		-	if !strings.HasSuffix(string(resp), "ANSWER") \|\| resp[3]&0x0f != 0 {
955		-	t.Fatalf("want the TCP answer, got %x", resp)
956		-	}
957		-	}


@@ -16,7 +16,6 @@ import (
16		"sync/atomic"	16		"sync/atomic"
17		"syscall"	17		"syscall"
18		"testing"	18		"testing"
19	-	"testing/synctest"
20		"time"	19		"time"
21		)	20		)
22			21

@@ -40,32 +39,18 @@ func answer(q []byte) []byte {
40		return append(r, "ANSWER"...)	39		return append(r, "ANSWER"...)
41		}	40		}
42			41
43	-	// listenUDPAndTCP binds UDP on a free loopback port and TCP on the same port.
44	-	// The TCP port may already be taken (outbound connections use the same
45	-	// ephemeral range), so it picks again a few times, like Listen does.
46	-	func listenUDPAndTCP(t *testing.T) (net.PacketConn, net.Listener) {
47	-	t.Helper()
48	-	for attempt := 0; ; attempt++ {
49	-	pc, err := net.ListenPacket("udp", "127.0.0.1:0")
50	-	if err != nil {
51	-	t.Skip("no loopback:", err)
52	-	}
53	-	tl, err := net.Listen("tcp", pc.LocalAddr().String())
54	-	if err == nil {
55	-	return pc, tl
56	-	}
57	-	_ = pc.Close()
58	-	if attempt == 49 \|\| !errors.Is(err, syscall.EADDRINUSE) {
59	-	t.Fatal(err)
60	-	}
61	-	}
62	-	}
63	-
64		// fakeUpstream serves UDP and TCP on one port of 127.0.0.1 and returns it.	42		// fakeUpstream serves UDP and TCP on one port of 127.0.0.1 and returns it.
65		func fakeUpstream(t *testing.T, handle func(q []byte) []byte) string {	43		func fakeUpstream(t *testing.T, handle func(q []byte) []byte) string {
66		t.Helper()	44		t.Helper()
67	-	pc, tl := listenUDPAndTCP(t)	45	+	pc, err := net.ListenPacket("udp", "127.0.0.1:0")
			46	+	if err != nil {
			47	+	t.Skip("no loopback:", err)
			48	+	}
68		addr := pc.LocalAddr().String()	49		addr := pc.LocalAddr().String()
			50	+	tl, err := net.Listen("tcp", addr)
			51	+	if err != nil {
			52	+	t.Fatal(err)
			53	+	}
69		t.Cleanup(func() { _ = pc.Close(); _ = tl.Close() })	54		t.Cleanup(func() { _ = pc.Close(); _ = tl.Close() })
70		go func() {	55		go func() {
71		buf := make([]byte, 4096)	56		buf := make([]byte, 4096)

@@ -107,17 +92,8 @@ func fakeUpstream(t *testing.T, handle func(q []byte) []byte) string {
107		}	92		}
108			93
109		func start(t testing.T, ups ...string) (Forwarder, string) {	94		func start(t testing.T, ups ...string) (Forwarder, string) {
110	-	t.Helper()
111	-	return startWith(t, nil, ups...)
112	-	}
113	-
114	-	// startWith is start with a hook to adjust the Forwarder before Listen.
115	-	func startWith(t testing.T, cfg func(Forwarder), ups ...string) (*Forwarder, string) {
116		t.Helper()	95		t.Helper()
117		f := New(func() []string { return ups }, 0)	96		f := New(func() []string { return ups }, 0)
118	-	if cfg != nil {
119	-	cfg(f)
120	-	}
121		ip := netip.MustParseAddr("127.0.0.1")	97		ip := netip.MustParseAddr("127.0.0.1")
122		if err := f.Listen(context.Background(), ip); err != nil {	98		if err := f.Listen(context.Background(), ip); err != nil {
123		t.Fatal(err)	99		t.Fatal(err)

@@ -204,6 +180,29 @@ func TestServfailWhenNoUpstreamAnswers(t *testing.T) {
204		}	180		}
205		}	181		}
206			182
			183	+	func TestMismatchedUpstreamIDIgnored(t *testing.T) {
			184	+	// Upstream answers with a wrong ID: treated as no answer, client gets SERVFAIL.
			185	+	up := fakeUpstream(t, func(q []byte) []byte {
			186	+	r := answer(q)
			187	+	r[0] ^= 0xff
			188	+	return r
			189	+	})
			190	+	f, addr := start(t, up)
			191	+	_ = f
			192	+	c, _ := net.Dial("udp", addr)
			193	+	defer func() { _ = c.Close() }()
			194	+	_ = c.SetDeadline(time.Now().Add(2UpstreamTimeout + 2time.Second))
			195	+	_, _ = c.Write(query(5, "a.test"))
			196	+	buf := make([]byte, 512)
			197	+	n, err := c.Read(buf)
			198	+	if err != nil {
			199	+	t.Fatal(err)
			200	+	}
			201	+	if buf[3]&0x0f != 2 {
			202	+	t.Fatalf("want SERVFAIL, got %x", buf[:n])
			203	+	}
			204	+	}
			205	+
207		func TestShortPacketsIgnored(t *testing.T) {	206		func TestShortPacketsIgnored(t *testing.T) {
208		up := fakeUpstream(t, answer)	207		up := fakeUpstream(t, answer)
209		_, addr := start(t, up)	208		_, addr := start(t, up)

@@ -613,345 +612,3 @@ func TestOverrideIgnoresMalformedAndNonQueries(t *testing.T) {
613		}	612		}
614		}	613		}
615		}	614		}
616	-
617	-	// bigAnswer returns an answer to q padded to size bytes, with TC set if tc.
618	-	func bigAnswer(q []byte, size int, tc bool) []byte {
619	-	r := answer(q)
620	-	if tc {
621	-	r[2] \|= 0x02
622	-	}
623	-	for len(r) < size {
624	-	r = append(r, 'x')
625	-	}
626	-	return r
627	-	}
628	-
629	-	func TestLargeAnswerNonEDNSClientGetsTC(t *testing.T) {
630	-	up := fakeUpstream(t, func(q []byte) []byte { return bigAnswer(q, 700, false) })
631	-	_, addr := start(t, up)
632	-	q := queryType(0x3333, "storage.googleapis.com", 1, false)
633	-	resp := udpExchange(t, addr, q)
634	-	if resp[2]&0x02 == 0 {
635	-	t.Fatalf("TC not set: %x", resp)
636	-	}
637	-	if len(resp) != len(q) \|\| binary.BigEndian.Uint16(resp[4:]) != 1 \|\| binary.BigEndian.Uint16(resp[6:]) != 0 {
638	-	t.Fatalf("want header+question only, got %x", resp)
639	-	}
640	-	if resp[0] != q[0] \|\| resp[1] != q[1] {
641	-	t.Fatal("ID changed")
642	-	}
643	-	}
644	-
645	-	func TestLargeAnswerEDNSClientGetsFullAnswer(t *testing.T) {
646	-	up := fakeUpstream(t, func(q []byte) []byte { return bigAnswer(q, 700, false) })
647	-	_, addr := start(t, up)
648	-	// queryType advertises 0x1000 (4096) bytes.
649	-	resp := udpExchange(t, addr, queryType(0x4444, "storage.googleapis.com", 1, true))
650	-	if len(resp) != 700 \|\| resp[2]&0x02 != 0 {
651	-	t.Fatalf("want the full 700-byte answer, got len %d tc=%d", len(resp), resp[2]&0x02)
652	-	}
653	-	}
654	-
655	-	// withOPT appends an OPT record (root name, type 41) with the given class and
656	-	// RDLEN to a query and bumps ARCOUNT.
657	-	func withOPT(q []byte, size, rdlen uint16, body int) []byte {
658	-	q = append([]byte(nil), q...)
659	-	q[11] = 1
660	-	q = append(q, 0, 0, 41)
661	-	q = binary.BigEndian.AppendUint16(q, size)
662	-	q = append(q, 0, 0, 0, 0)
663	-	q = binary.BigEndian.AppendUint16(q, rdlen)
664	-	return append(q, make([]byte, body)...)
665	-	}
666	-
667	-	func TestClientUDPSize(t *testing.T) {
668	-	plain := query(1, "a.b")
669	-	bigQD := append([]byte(nil), plain[:hdrLen]...)
670	-	bigQD[4], bigQD[5] = 0xff, 0xff
671	-	bigQD = append(bigQD, 1, 'a', 0, 0, 1, 0, 1)
672	-	ptrQ := append([]byte(nil), plain[:hdrLen]...)
673	-	ptrQ = append(ptrQ, 0xc0, hdrLen, 0, 1, 0, 1) // compression pointer in question
674	-	tests := []struct {
675	-	name string
676	-	q []byte
677	-	want int
678	-	}{
679	-	{"no edns", plain, 512},
680	-	{"edns 1232", withOPT(plain, 1232, 0, 0), 1232},
681	-	{"edns below 512", withOPT(plain, 100, 0, 0), 512},
682	-	{"edns above cap", withOPT(plain, 65535, 0, 0), maxUDP},
683	-	{"edns 4096", withOPT(plain, 4096, 0, 0), 4096},
684	-	{"qdcount 65535 short body", bigQD, 512},
685	-	{"compression pointer in question", ptrQ, 512},
686	-	{"OPT rdlen overruns buffer", withOPT(plain, 4096, 500, 3), 4096},
687	-	{"truncated header", plain[:15], 512},
688	-	{"short", plain[:5], 512},
689	-	}
690	-	for _, tt := range tests {
691	-	t.Run(tt.name, func(t *testing.T) {
692	-	if got := clientUDPSize(tt.q); got != tt.want {
693	-	t.Fatalf("got %d, want %d", got, tt.want)
694	-	}
695	-	})
696	-	}
697	-	}
698	-
699	-	func TestTruncated(t *testing.T) {
700	-	q := queryType(7, "a.b", 1, false)
701	-	full := bigAnswer(q, 700, false)
702	-	full[7] = 5 // ANCOUNT
703	-	got := truncated(full)
704	-	if len(got) != len(q) \|\| got[2]&0x02 == 0 \|\| got[6] != 0 \|\| got[7] != 0 \|\| got[4] != 0 \|\| got[5] != 1 {
705	-	t.Fatalf("bad truncated reply %x", got)
706	-	}
707	-	bad := append([]byte(nil), full...)
708	-	bad[4], bad[5] = 0xff, 0xff // QDCOUNT lies, question parse fails
709	-	got = truncated(bad)
710	-	if len(got) != hdrLen \|\| got[2]&0x02 == 0 \|\| got[4] != 0 \|\| got[5] != 0 {
711	-	t.Fatalf("bad-question case: %x", got)
712	-	}
713	-	if truncated(full[:5]) != nil {
714	-	t.Fatal("short input must give nil")
715	-	}
716	-	}
717	-
718	-	func TestServfailAndOverrideRepliesAreNotTruncated(t *testing.T) {
719	-	dead := "127.0.0.1:1" // nothing listens: UDP and TCP both fail at once
720	-	_, addr := startOverride(t, "forge.test", dead)
721	-	resp := udpExchange(t, addr, queryType(9, "x.test", 1, false))
722	-	if len(resp) != hdrLen \|\| resp[3]&0x0f != 2 \|\| resp[2]&0x02 != 0 {
723	-	t.Fatalf("want a plain 12-byte SERVFAIL, got %x", resp)
724	-	}
725	-	resp = udpExchange(t, addr, queryType(10, "forge.test", 1, false))
726	-	if resp[2]&0x02 != 0 \|\| resp[3]&0x0f != 0 \|\| binary.BigEndian.Uint16(resp[6:]) != 1 {
727	-	t.Fatalf("want the override A answer, got %x", resp)
728	-	}
729	-	}
730	-
731	-	// fakeNet is an in-memory upstream network for Forwarder.dial. One Write on a
732	-	// UDP conn is one datagram (net.Pipe delivers a Write to a single Read when the
733	-	// reader's buffer is large enough). A handler returning nil never answers.
734	-	type fakeNet struct {
735	-	udp, tcp func(addr string, q []byte) []byte
736	-	udpCalls, tcpCalls atomic.Int64
737	-	}
738	-
739	-	func (n *fakeNet) dial(ctx context.Context, network, addr string) (net.Conn, error) {
740	-	if err := ctx.Err(); err != nil { // like net.Dialer
741	-	return nil, err
742	-	}
743	-	c, s := net.Pipe()
744	-	switch network {
745	-	case "udp":
746	-	go func() {
747	-	defer func() { _ = s.Close() }()
748	-	buf := make([]byte, 4096)
749	-	for {
750	-	k, err := s.Read(buf)
751	-	if err != nil {
752	-	return
753	-	}
754	-	n.udpCalls.Add(1)
755	-	if r := n.udp(addr, append([]byte(nil), buf[:k]...)); r != nil {
756	-	if _, err := s.Write(r); err != nil {
757	-	return
758	-	}
759	-	}
760	-	}
761	-	}()
762	-	case "tcp":
763	-	go func() {
764	-	defer func() { _ = s.Close() }()
765	-	var l [2]byte
766	-	if _, err := io.ReadFull(s, l[:]); err != nil {
767	-	return
768	-	}
769	-	q := make([]byte, binary.BigEndian.Uint16(l[:]))
770	-	if _, err := io.ReadFull(s, q); err != nil {
771	-	return
772	-	}
773	-	n.tcpCalls.Add(1)
774	-	if r := n.tcp(addr, q); r != nil {
775	-	_, _ = s.Write(append(binary.BigEndian.AppendUint16(nil, uint16(len(r))), r...))
776	-	}
777	-	_, _ = io.Copy(io.Discard, s) // wait for the client to hang up
778	-	}()
779	-	default:
780	-	return nil, errors.New("unexpected network " + network)
781	-	}
782	-	return c, nil
783	-	}
784	-
785	-	// silent is a handler that never answers.
786	-	func silent(string, []byte) []byte { return nil }
787	-
788	-	// fakeForwarder returns a Forwarder with production timeouts whose upstreams
789	-	// are served by n, and a listener for exchange.
790	-	func fakeForwarder(n fakeNet, ups ...string) (Forwarder, *listener) {
791	-	f := New(func() []string { return ups }, 0)
792	-	f.dial = n.dial
793	-	return f, &listener{addr: netip.MustParseAddr("127.0.0.1")}
794	-	}
795	-
796	-	func isServfail(r []byte, id uint16) bool {
797	-	return len(r) == hdrLen && binary.BigEndian.Uint16(r) == id && r[3]&0x0f == 2
798	-	}
799	-
800	-	// The tests below run in a synctest bubble: the fake clock makes the real
801	-	// production timeouts (1 s UDP, 3 s TCP, 6 s cap) instant and exact.
802	-
803	-	func TestExchangeSilentUDPFallsBackToTCPAfterUDPTimeout(t *testing.T) {
804	-	synctest.Test(t, func(t *testing.T) {
805	-	n := &fakeNet{udp: silent, tcp: func(_ string, q []byte) []byte { return answer(q) }}
806	-	f, l := fakeForwarder(n, "up:53")
807	-	start := time.Now()
808	-	resp := f.exchange(t.Context(), l, query(1, "storage.googleapis.com"), false)
809	-	if !strings.HasSuffix(string(resp), "ANSWER") {
810	-	t.Fatalf("want the TCP answer, got %x", resp)
811	-	}
812	-	if got := time.Since(start); got != UDPAttemptTimeout {
813	-	t.Fatalf("answered after %v, want %v", got, UDPAttemptTimeout)
814	-	}
815	-	if n.udpCalls.Load() != 1 \|\| n.tcpCalls.Load() != 1 {
816	-	t.Fatalf("udp=%d tcp=%d, want 1 and 1", n.udpCalls.Load(), n.tcpCalls.Load())
817	-	}
818	-	})
819	-	}
820	-
821	-	func TestExchangeTruncatedUDPRetriedOverTCP(t *testing.T) {
822	-	synctest.Test(t, func(t *testing.T) {
823	-	n := &fakeNet{
824	-	udp: func(_ string, q []byte) []byte { return bigAnswer(q, 0, true) },
825	-	tcp: func(_ string, q []byte) []byte { return bigAnswer(q, 700, false) },
826	-	}
827	-	f, l := fakeForwarder(n, "up:53")
828	-	start := time.Now()
829	-	resp := f.exchange(t.Context(), l, query(2, "example.com"), false)
830	-	if len(resp) != 700 \|\| resp[2]&0x02 != 0 {
831	-	t.Fatalf("want the full TCP answer, got len %d tc=%d", len(resp), resp[2]&0x02)
832	-	}
833	-	if time.Since(start) != 0 {
834	-	t.Fatalf("took %v, want no waiting", time.Since(start))
835	-	}
836	-	})
837	-	}
838	-
839	-	func TestExchangeKeepsTruncatedReplyWhenTCPFails(t *testing.T) {
840	-	synctest.Test(t, func(t *testing.T) {
841	-	n := &fakeNet{udp: func(_ string, q []byte) []byte { return bigAnswer(q, 0, true) }, tcp: silent}
842	-	f, l := fakeForwarder(n, "up:53")
843	-	resp := f.exchange(t.Context(), l, query(3, "example.com"), false)
844	-	if len(resp) < hdrLen \|\| resp[2]&0x02 == 0 {
845	-	t.Fatalf("want the TC=1 reply, got %x", resp)
846	-	}
847	-	})
848	-	}
849	-
850	-	func TestExchangeSecondUpstreamNotStarved(t *testing.T) {
851	-	synctest.Test(t, func(t *testing.T) {
852	-	n := &fakeNet{
853	-	udp: func(a string, q []byte) []byte {
854	-	if a == "dead:53" {
855	-	return nil
856	-	}
857	-	return answer(q)
858	-	},
859	-	tcp: func(a string, q []byte) []byte {
860	-	if a == "dead:53" {
861	-	return nil
862	-	}
863	-	return answer(q)
864	-	},
865	-	}
866	-	f, l := fakeForwarder(n, "dead:53", "live:53")
867	-	start := time.Now()
868	-	resp := f.exchange(t.Context(), l, query(4, "a.test"), false)
869	-	if !strings.HasSuffix(string(resp), "ANSWER") {
870	-	t.Fatalf("want the second upstream's answer, got %x", resp)
871	-	}
872	-	if got, want := time.Since(start), UDPAttemptTimeout+UpstreamTimeout; got != want {
873	-	t.Fatalf("took %v, want %v", got, want)
874	-	}
875	-	})
876	-	}
877	-
878	-	func TestExchangeMismatchedUpstreamIDIgnored(t *testing.T) {
879	-	synctest.Test(t, func(t *testing.T) {
880	-	wrong := func(_ string, q []byte) []byte {
881	-	r := answer(q)
882	-	r[0] ^= 0xff
883	-	return r
884	-	}
885	-	f, l := fakeForwarder(&fakeNet{udp: wrong, tcp: wrong}, "up:53")
886	-	start := time.Now()
887	-	resp := f.exchange(t.Context(), l, query(5, "a.test"), false)
888	-	if !isServfail(resp, 5) {
889	-	t.Fatalf("want SERVFAIL, got %x", resp)
890	-	}
891	-	// UDP keeps listening for the right ID until its timeout; the TCP
892	-	// reply is read once and rejected.
893	-	if got := time.Since(start); got != UDPAttemptTimeout {
894	-	t.Fatalf("took %v, want %v", got, UDPAttemptTimeout)
895	-	}
896	-	})
897	-	}
898	-
899	-	func TestExchangeCappedAtMaxExchange(t *testing.T) {
900	-	synctest.Test(t, func(t *testing.T) {
901	-	n := &fakeNet{udp: silent, tcp: silent}
902	-	f, l := fakeForwarder(n, "a:53", "b:53", "c:53")
903	-	start := time.Now()
904	-	resp := f.exchange(t.Context(), l, query(6, "a.test"), false)
905	-	if !isServfail(resp, 6) {
906	-	t.Fatalf("want SERVFAIL, got %x", resp)
907	-	}
908	-	if got := time.Since(start); got != MaxExchange {
909	-	t.Fatalf("took %v, want %v", got, MaxExchange)
910	-	}
911	-	// a: 1 s UDP + 3 s TCP; b: 1 s UDP + 1 s TCP (cut by the cap); c: skipped.
912	-	if n.udpCalls.Load() != 2 \|\| n.tcpCalls.Load() != 2 {
913	-	t.Fatalf("udp=%d tcp=%d, want 2 and 2", n.udpCalls.Load(), n.tcpCalls.Load())
914	-	}
915	-	})
916	-	}
917	-
918	-	func TestExchangeTCPClientUsesTCPOnly(t *testing.T) {
919	-	synctest.Test(t, func(t *testing.T) {
920	-	n := &fakeNet{udp: silent, tcp: func(_ string, q []byte) []byte { return answer(q) }}
921	-	f, l := fakeForwarder(n, "up:53")
922	-	resp := f.exchange(t.Context(), l, query(7, "a.test"), true)
923	-	if !strings.HasSuffix(string(resp), "ANSWER") \|\| n.udpCalls.Load() != 0 {
924	-	t.Fatalf("resp=%x udp=%d", resp, n.udpCalls.Load())
925	-	}
926	-	})
927	-	}
928	-
929	-	func TestExchangeCancelledReturnsNil(t *testing.T) {
930	-	synctest.Test(t, func(t *testing.T) {
931	-	f, l := fakeForwarder(&fakeNet{udp: silent, tcp: silent}, "up:53")
932	-	ctx, cancel := context.WithCancel(t.Context())
933	-	time.AfterFunc(500*time.Millisecond, cancel)
934	-	start := time.Now()
935	-	if resp := f.exchange(ctx, l, query(8, "a.test"), false); resp != nil {
936	-	t.Fatalf("want nil on cancellation, got %x", resp)
937	-	}
938	-	// Cancellation is noticed at the next attempt boundary: the blocked UDP
939	-	// read runs to its own deadline, then no TCP attempt is started.
940	-	if got := time.Since(start); got != UDPAttemptTimeout {
941	-	t.Fatalf("returned after %v, want %v", got, UDPAttemptTimeout)
942	-	}
943	-	})
944	-	}
945	-
946	-	// End to end through real client-facing sockets; the upstream is in memory.
947	-	func TestUDPClientGetsTCPAnswerWhenUpstreamUDPSilent(t *testing.T) {
948	-	n := &fakeNet{udp: silent, tcp: func(_ string, q []byte) []byte { return answer(q) }}
949	-	_, addr := startWith(t, func(f *Forwarder) {
950	-	f.dial = n.dial
951	-	f.udpTimeout = 50 * time.Millisecond
952	-	}, "up:53")
953	-	resp := udpExchange(t, addr, query(0x1111, "storage.googleapis.com"))
954	-	if !strings.HasSuffix(string(resp), "ANSWER") \|\| resp[3]&0x0f != 0 {
955	-	t.Fatalf("want the TCP answer, got %x", resp)
956	-	}
957	-	}

A internal/e2e/registry_test.go +114 -0

@@ -0,0 +1,114 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package e2e
	4	+
	5	+	import (
	6	+	"bytes"
	7	+	"encoding/json"
	8	+	"net/http"
	9	+	"strings"
	10	+	"sync"
	11	+	"testing"
	12	+	"time"
	13	+
	14	+	"github.com/bnema/zerowrap"
	15	+
	16	+	"bnema.dev/ermite-runner/internal/cachesvc"
	17	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	18	+	"bnema.dev/ermite-runner/internal/registryproxy"
	19	+	)
	20	+
	21	+	type registryLogs struct {
	22	+	mu sync.Mutex
	23	+	b bytes.Buffer
	24	+	}
	25	+
	26	+	func (b *registryLogs) Write(p []byte) (int, error) {
	27	+	b.mu.Lock()
	28	+	defer b.mu.Unlock()
	29	+	return b.b.Write(p)
	30	+	}
	31	+	func (b *registryLogs) text() string { b.mu.Lock(); defer b.mu.Unlock(); return b.b.String() }
	32	+	func (b *registryLogs) reset() { b.mu.Lock(); defer b.mu.Unlock(); b.b.Reset() }
	33	+
	34	+	func TestDockerHubRegistryAcrossJobs(t *testing.T) {
	35	+	image := needE2E(t)
	36	+	cl := &http.Client{Timeout: 10 * time.Second}
	37	+	resp, err := cl.Get("https://registry-1.docker.io/v2/")
	38	+	if err != nil {
	39	+	t.Skipf("Docker Hub unreachable: %v", err)
	40	+	}
	41	+	_ = resp.Body.Close()
	42	+	if resp.StatusCode != http.StatusUnauthorized && resp.StatusCode != http.StatusOK {
	43	+	t.Skipf("Docker Hub unavailable: HTTP %d", resp.StatusCode)
	44	+	}
	45	+	var logs registryLogs
	46	+	p, err := registryproxy.New(registryproxy.Config{Dir: t.TempDir(), Max: 256 << 20, Log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logs})})
	47	+	if err != nil {
	48	+	t.Fatal(err)
	49	+	}
	50	+	svc, err := cachesvc.New(cachesvc.Config{Dir: t.TempDir(), Registry: p})
	51	+	if err != nil {
	52	+	t.Fatal(err)
	53	+	}
	54	+	defer func() { _ = svc.CloseAll() }()
	55	+	d := newCacheDriver(t, image, svc)
	56	+	probe := runJob(t, d, jobSpec(`on: push
	57	+	jobs:
	58	+	j:
	59	+	runs-on: x
	60	+	steps:
	61	+	- run: if docker info >/dev/null 2>&1; then echo nested-dockerd-ready; else echo nested-dockerd-unavailable; fi
	62	+	`), nil)
	63	+	if !strings.Contains(strings.Join(probe.Logs, "\n"), "nested-dockerd-ready") {
	64	+	t.Skip("nested dockerd unavailable in sandbox; rootless Docker harness cannot run job containers")
	65	+	}
	66	+	for i := 0; i < 2; i++ {
	67	+	logs.reset()
	68	+	spec := jobSpec(`on: push
	69	+	jobs:
	70	+	j:
	71	+	runs-on: x
	72	+	container: alpine:3
	73	+	steps:
	74	+	- run: echo alpine-job
	75	+	`)
	76	+	// A registry-only job: the host sends the port and no token.
	77	+	spec.CachePort = cachesvc.DefaultPort
	78	+	s := runJob(t, d, spec, nil)
	79	+	if s.Result == nil \|\| s.Result.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS {
	80	+	t.Fatalf("pull job %d: %v\n%s", i, s.Result, strings.Join(s.Logs, "\n"))
	81	+	}
	82	+	all := logs.text()
	83	+	manifestHit := false
	84	+	for _, line := range strings.Split(all, "\n") {
	85	+	var event struct {
	86	+	Message string `json:"message"`
	87	+	Hit bool `json:"hit"`
	88	+	}
	89	+	if json.Unmarshal([]byte(line), &event) == nil && event.Message == "registry manifest" && event.Hit {
	90	+	manifestHit = true
	91	+	}
	92	+	}
	93	+	if i == 1 && !manifestHit {
	94	+	t.Fatalf("second job missing manifest hit:\n%s", all)
	95	+	}
	96	+	t.Logf("pull job %d proxy logs:\n%s", i+1, all)
	97	+	if i == 1 && strings.Contains(all, `"message":"registry blob"`) {
	98	+	t.Fatalf("second job fetched upstream blobs:\n%s", all)
	99	+	}
	100	+	}
	101	+	// The mirror argv is unconditional: a job without the runtime channel
	102	+	// must still pull directly from Docker Hub.
	103	+	off := runJob(t, d, jobSpec(`on: push
	104	+	jobs:
	105	+	j:
	106	+	runs-on: x
	107	+	container: alpine:3
	108	+	steps:
	109	+	- run: echo direct-pull
	110	+	`), nil)
	111	+	if off.Result == nil \|\| off.Result.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS {
	112	+	t.Fatalf("proxy-off pull: %v\n%s", off.Result, strings.Join(off.Logs, "\n"))
	113	+	}
	114	+	}

@@ -0,0 +1,114 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package e2e
	4	+
	5	+	import (
	6	+	"bytes"
	7	+	"encoding/json"
	8	+	"net/http"
	9	+	"strings"
	10	+	"sync"
	11	+	"testing"
	12	+	"time"
	13	+
	14	+	"github.com/bnema/zerowrap"
	15	+
	16	+	"bnema.dev/ermite-runner/internal/cachesvc"
	17	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	18	+	"bnema.dev/ermite-runner/internal/registryproxy"
	19	+	)
	20	+
	21	+	type registryLogs struct {
	22	+	mu sync.Mutex
	23	+	b bytes.Buffer
	24	+	}
	25	+
	26	+	func (b *registryLogs) Write(p []byte) (int, error) {
	27	+	b.mu.Lock()
	28	+	defer b.mu.Unlock()
	29	+	return b.b.Write(p)
	30	+	}
	31	+	func (b *registryLogs) text() string { b.mu.Lock(); defer b.mu.Unlock(); return b.b.String() }
	32	+	func (b *registryLogs) reset() { b.mu.Lock(); defer b.mu.Unlock(); b.b.Reset() }
	33	+
	34	+	func TestDockerHubRegistryAcrossJobs(t *testing.T) {
	35	+	image := needE2E(t)
	36	+	cl := &http.Client{Timeout: 10 * time.Second}
	37	+	resp, err := cl.Get("https://registry-1.docker.io/v2/")
	38	+	if err != nil {
	39	+	t.Skipf("Docker Hub unreachable: %v", err)
	40	+	}
	41	+	_ = resp.Body.Close()
	42	+	if resp.StatusCode != http.StatusUnauthorized && resp.StatusCode != http.StatusOK {
	43	+	t.Skipf("Docker Hub unavailable: HTTP %d", resp.StatusCode)
	44	+	}
	45	+	var logs registryLogs
	46	+	p, err := registryproxy.New(registryproxy.Config{Dir: t.TempDir(), Max: 256 << 20, Log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logs})})
	47	+	if err != nil {
	48	+	t.Fatal(err)
	49	+	}
	50	+	svc, err := cachesvc.New(cachesvc.Config{Dir: t.TempDir(), Registry: p})
	51	+	if err != nil {
	52	+	t.Fatal(err)
	53	+	}
	54	+	defer func() { _ = svc.CloseAll() }()
	55	+	d := newCacheDriver(t, image, svc)
	56	+	probe := runJob(t, d, jobSpec(`on: push
	57	+	jobs:
	58	+	j:
	59	+	runs-on: x
	60	+	steps:
	61	+	- run: if docker info >/dev/null 2>&1; then echo nested-dockerd-ready; else echo nested-dockerd-unavailable; fi
	62	+	`), nil)
	63	+	if !strings.Contains(strings.Join(probe.Logs, "\n"), "nested-dockerd-ready") {
	64	+	t.Skip("nested dockerd unavailable in sandbox; rootless Docker harness cannot run job containers")
	65	+	}
	66	+	for i := 0; i < 2; i++ {
	67	+	logs.reset()
	68	+	spec := jobSpec(`on: push
	69	+	jobs:
	70	+	j:
	71	+	runs-on: x
	72	+	container: alpine:3
	73	+	steps:
	74	+	- run: echo alpine-job
	75	+	`)
	76	+	// A registry-only job: the host sends the port and no token.
	77	+	spec.CachePort = cachesvc.DefaultPort
	78	+	s := runJob(t, d, spec, nil)
	79	+	if s.Result == nil \|\| s.Result.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS {
	80	+	t.Fatalf("pull job %d: %v\n%s", i, s.Result, strings.Join(s.Logs, "\n"))
	81	+	}
	82	+	all := logs.text()
	83	+	manifestHit := false
	84	+	for _, line := range strings.Split(all, "\n") {
	85	+	var event struct {
	86	+	Message string `json:"message"`
	87	+	Hit bool `json:"hit"`
	88	+	}
	89	+	if json.Unmarshal([]byte(line), &event) == nil && event.Message == "registry manifest" && event.Hit {
	90	+	manifestHit = true
	91	+	}
	92	+	}
	93	+	if i == 1 && !manifestHit {
	94	+	t.Fatalf("second job missing manifest hit:\n%s", all)
	95	+	}
	96	+	t.Logf("pull job %d proxy logs:\n%s", i+1, all)
	97	+	if i == 1 && strings.Contains(all, `"message":"registry blob"`) {
	98	+	t.Fatalf("second job fetched upstream blobs:\n%s", all)
	99	+	}
	100	+	}
	101	+	// The mirror argv is unconditional: a job without the runtime channel
	102	+	// must still pull directly from Docker Hub.
	103	+	off := runJob(t, d, jobSpec(`on: push
	104	+	jobs:
	105	+	j:
	106	+	runs-on: x
	107	+	container: alpine:3
	108	+	steps:
	109	+	- run: echo direct-pull
	110	+	`), nil)
	111	+	if off.Result == nil \|\| off.Result.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS {
	112	+	t.Fatalf("proxy-off pull: %v\n%s", off.Result, strings.Join(off.Logs, "\n"))
	113	+	}
	114	+	}

A internal/e2e/timing_test.go +83 -0

@@ -0,0 +1,83 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package e2e
	4	+
	5	+	import (
	6	+	"context"
	7	+	"encoding/json"
	8	+	"net/http"
	9	+	"sync"
	10	+	"testing"
	11	+	"time"
	12	+
	13	+	"github.com/bnema/zerowrap"
	14	+
	15	+	ermite "bnema.dev/ermite-api/gen/ermite"
	16	+	"bnema.dev/ermite-api/gen/ermite/ermiteconnect"
	17	+
	18	+	"bnema.dev/ermite-runner/internal/host"
	19	+	"bnema.dev/ermite-runner/internal/host/hosttest"
	20	+	)
	21	+
	22	+	type timingWriter struct {
	23	+	mu sync.Mutex
	24	+	lines [][]byte
	25	+	}
	26	+
	27	+	func (w *timingWriter) Write(p []byte) (int, error) {
	28	+	w.mu.Lock()
	29	+	defer w.mu.Unlock()
	30	+	w.lines = append(w.lines, append([]byte(nil), p...))
	31	+	return len(p), nil
	32	+	}
	33	+
	34	+	func TestJobTiming(t *testing.T) {
	35	+	image := needE2E(t)
	36	+	drv := newDriver(t, image)
	37	+	srv := hosttest.New()
	38	+	defer srv.Close()
	39	+	srv.Queue(&ermite.Job{JobId: "timing", RunId: 1, Attempt: 1, JobToken: "ermitejob_timing", WorkflowPath: ".github/workflows/ci.yml", WorkflowYaml: "name: timing\non: push\njobs:\n main:\n runs-on: ubuntu-latest\n steps:\n - run: echo timing\n", JobKey: "main", Matrix: "{}", Contexts: chainContexts("main"), Repo: "hello", TimeoutMinutes: 2})
	40	+	r := &host.Runner{
	41	+	Cfg: host.Config{URL: srv.URL, Name: "timing", Capacity: 1, Version: "test"}, Driver: drv,
	42	+	Poll: host.NewClient(srv.URL, hosttest.RunnerToken, http.DefaultTransport),
	43	+	Job: func(tok string) ermiteconnect.RunnerServiceClient {
	44	+	return host.NewClient(srv.URL, tok, http.DefaultTransport)
	45	+	},
	46	+	Timing: host.Timing{LogFlush: 20 * time.Millisecond, NoJobPause: 100 * time.Millisecond},
	47	+	}
	48	+	var logs timingWriter
	49	+	ctx, cancel := context.WithCancel(zerowrap.WithCtx(context.Background(), zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &logs})))
	50	+	done := make(chan error, 1)
	51	+	go func() { done <- r.Run(ctx) }()
	52	+	select {
	53	+	case <-srv.Done():
	54	+	case <-time.After(3 * time.Minute):
	55	+	cancel()
	56	+	t.Fatal("job did not complete")
	57	+	}
	58	+	cancel()
	59	+	select {
	60	+	case err := <-done:
	61	+	if err != nil {
	62	+	t.Fatal(err)
	63	+	}
	64	+	case <-time.After(time.Minute):
	65	+	t.Fatal("runner did not stop")
	66	+	}
	67	+	logs.mu.Lock()
	68	+	defer logs.mu.Unlock()
	69	+	for _, line := range logs.lines {
	70	+	var fields map[string]any
	71	+	if json.Unmarshal(line, &fields) != nil \|\| fields["message"] != "job finished" {
	72	+	continue
	73	+	}
	74	+	for _, key := range []string{"sandbox_ms", "guest_dockerd_ms"} {
	75	+	value, ok := fields[key].(float64)
	76	+	if !ok \|\| value < 0 {
	77	+	t.Errorf("%s missing or negative: %v", key, fields)
	78	+	}
	79	+	}
	80	+	return
	81	+	}
	82	+	t.Fatal("no job finished log")
	83	+	}

@@ -0,0 +1,83 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package e2e
	4	+
	5	+	import (
	6	+	"context"
	7	+	"encoding/json"
	8	+	"net/http"
	9	+	"sync"
	10	+	"testing"
	11	+	"time"
	12	+
	13	+	"github.com/bnema/zerowrap"
	14	+
	15	+	ermite "bnema.dev/ermite-api/gen/ermite"
	16	+	"bnema.dev/ermite-api/gen/ermite/ermiteconnect"
	17	+
	18	+	"bnema.dev/ermite-runner/internal/host"
	19	+	"bnema.dev/ermite-runner/internal/host/hosttest"
	20	+	)
	21	+
	22	+	type timingWriter struct {
	23	+	mu sync.Mutex
	24	+	lines [][]byte
	25	+	}
	26	+
	27	+	func (w *timingWriter) Write(p []byte) (int, error) {
	28	+	w.mu.Lock()
	29	+	defer w.mu.Unlock()
	30	+	w.lines = append(w.lines, append([]byte(nil), p...))
	31	+	return len(p), nil
	32	+	}
	33	+
	34	+	func TestJobTiming(t *testing.T) {
	35	+	image := needE2E(t)
	36	+	drv := newDriver(t, image)
	37	+	srv := hosttest.New()
	38	+	defer srv.Close()
	39	+	srv.Queue(&ermite.Job{JobId: "timing", RunId: 1, Attempt: 1, JobToken: "ermitejob_timing", WorkflowPath: ".github/workflows/ci.yml", WorkflowYaml: "name: timing\non: push\njobs:\n main:\n runs-on: ubuntu-latest\n steps:\n - run: echo timing\n", JobKey: "main", Matrix: "{}", Contexts: chainContexts("main"), Repo: "hello", TimeoutMinutes: 2})
	40	+	r := &host.Runner{
	41	+	Cfg: host.Config{URL: srv.URL, Name: "timing", Capacity: 1, Version: "test"}, Driver: drv,
	42	+	Poll: host.NewClient(srv.URL, hosttest.RunnerToken, http.DefaultTransport),
	43	+	Job: func(tok string) ermiteconnect.RunnerServiceClient {
	44	+	return host.NewClient(srv.URL, tok, http.DefaultTransport)
	45	+	},
	46	+	Timing: host.Timing{LogFlush: 20 * time.Millisecond, NoJobPause: 100 * time.Millisecond},
	47	+	}
	48	+	var logs timingWriter
	49	+	ctx, cancel := context.WithCancel(zerowrap.WithCtx(context.Background(), zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &logs})))
	50	+	done := make(chan error, 1)
	51	+	go func() { done <- r.Run(ctx) }()
	52	+	select {
	53	+	case <-srv.Done():
	54	+	case <-time.After(3 * time.Minute):
	55	+	cancel()
	56	+	t.Fatal("job did not complete")
	57	+	}
	58	+	cancel()
	59	+	select {
	60	+	case err := <-done:
	61	+	if err != nil {
	62	+	t.Fatal(err)
	63	+	}
	64	+	case <-time.After(time.Minute):
	65	+	t.Fatal("runner did not stop")
	66	+	}
	67	+	logs.mu.Lock()
	68	+	defer logs.mu.Unlock()
	69	+	for _, line := range logs.lines {
	70	+	var fields map[string]any
	71	+	if json.Unmarshal(line, &fields) != nil \|\| fields["message"] != "job finished" {
	72	+	continue
	73	+	}
	74	+	for _, key := range []string{"sandbox_ms", "guest_dockerd_ms"} {
	75	+	value, ok := fields[key].(float64)
	76	+	if !ok \|\| value < 0 {
	77	+	t.Errorf("%s missing or negative: %v", key, fields)
	78	+	}
	79	+	}
	80	+	return
	81	+	}
	82	+	t.Fatal("no job finished log")
	83	+	}

A internal/e2e/toolcache_test.go +82 -0

@@ -0,0 +1,82 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package e2e
	4	+
	5	+	import (
	6	+	"context"
	7	+	"os/exec"
	8	+	"strings"
	9	+	"testing"
	10	+	"time"
	11	+
	12	+	"bnema.dev/ermite-runner/internal/execstream"
	13	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	14	+	"bnema.dev/ermite-runner/internal/sandbox"
	15	+	)
	16	+
	17	+	func TestContainerSetupGoFindsToolcache(t *testing.T) {
	18	+	image := needE2E(t)
	19	+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	20	+	defer cancel()
	21	+	// Distinguish a missing cache from an unavailable image or Docker daemon.
	22	+	out, err := exec.CommandContext(ctx, "docker", "run", "--rm", "--entrypoint", "sh", image, "-c", "if test -d /opt/hostedtoolcache/go/1.27.1; then echo present; else echo absent; fi").CombinedOutput()
	23	+	if err != nil {
	24	+	t.Fatalf("probe toolcache: %v\n%s", err, out)
	25	+	}
	26	+	if strings.TrimSpace(string(out)) == "absent" {
	27	+	t.Skip("test image has no Go 1.27.1 toolcache")
	28	+	}
	29	+	d := newDriver(t, image)
	30	+	sb, err := d.New()
	31	+	if err != nil {
	32	+	t.Fatal(err)
	33	+	}
	34	+	defer func() {
	35	+	if err := sb.Stop(context.Background()); err != nil {
	36	+	t.Error(err)
	37	+	}
	38	+	}()
	39	+	conn, err := sb.Start(ctx, sandbox.JobSpec{JobID: "e2e-toolcache"})
	40	+	if err != nil {
	41	+	t.Fatal(err)
	42	+	}
	43	+	spec := jobSpec(`on: push
	44	+	jobs:
	45	+	j:
	46	+	runs-on: x
	47	+	steps:
	48	+	- uses: actions/setup-go@v7
	49	+	with:
	50	+	go-version: '1.27.1'
	51	+	cache: false
	52	+	- run: go version
	53	+	`)
	54	+	var logs []string
	55	+	var result *execpb.JobResult
	56	+	_, exit, err := execstream.DriveJob(conn, &execpb.Run{Job: spec}, nil, execstream.JobHandler{
	57	+	Resolve: func(_ context.Context, r execpb.ResolveRequest) execpb.ResolveResponse {
	58	+	resp := &execpb.ResolveResponse{}
	59	+	for _, ref := range r.Uses {
	60	+	if ref != "actions/setup-go@v7" {
	61	+	return &execpb.ResolveResponse{Error: "unexpected action: " + ref}
	62	+	}
	63	+	resp.Actions = append(resp.Actions, &execpb.ResolvedAction{Ref: ref, NameWithOwner: "actions/setup-go", ResolvedSha: "v7", TarballUrl: "https://codeload.github.com/actions/setup-go/tar.gz/refs/tags/v7"})
	64	+	}
	65	+	return resp
	66	+	},
	67	+	Log: func(l *execpb.Log) { logs = append(logs, l.Text) },
	68	+	JobResult: func(r *execpb.JobResult) { result = r },
	69	+	})
	70	+	all := strings.Join(logs, "\n")
	71	+	if err != nil {
	72	+	t.Fatalf("drive: %v\n%s", err, all)
	73	+	}
	74	+	if result == nil \|\| result.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS \|\| exit == nil \|\| exit.Code != 0 {
	75	+	t.Fatalf("result %v exit %v\n%s", result, exit, all)
	76	+	}
	77	+	for _, want := range []string{"Found in cache", "go version go1.27.1"} {
	78	+	if !strings.Contains(all, want) {
	79	+	t.Errorf("log lacks %q\n%s", want, all)
	80	+	}
	81	+	}
	82	+	}

@@ -0,0 +1,82 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package e2e
	4	+
	5	+	import (
	6	+	"context"
	7	+	"os/exec"
	8	+	"strings"
	9	+	"testing"
	10	+	"time"
	11	+
	12	+	"bnema.dev/ermite-runner/internal/execstream"
	13	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	14	+	"bnema.dev/ermite-runner/internal/sandbox"
	15	+	)
	16	+
	17	+	func TestContainerSetupGoFindsToolcache(t *testing.T) {
	18	+	image := needE2E(t)
	19	+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	20	+	defer cancel()
	21	+	// Distinguish a missing cache from an unavailable image or Docker daemon.
	22	+	out, err := exec.CommandContext(ctx, "docker", "run", "--rm", "--entrypoint", "sh", image, "-c", "if test -d /opt/hostedtoolcache/go/1.27.1; then echo present; else echo absent; fi").CombinedOutput()
	23	+	if err != nil {
	24	+	t.Fatalf("probe toolcache: %v\n%s", err, out)
	25	+	}
	26	+	if strings.TrimSpace(string(out)) == "absent" {
	27	+	t.Skip("test image has no Go 1.27.1 toolcache")
	28	+	}
	29	+	d := newDriver(t, image)
	30	+	sb, err := d.New()
	31	+	if err != nil {
	32	+	t.Fatal(err)
	33	+	}
	34	+	defer func() {
	35	+	if err := sb.Stop(context.Background()); err != nil {
	36	+	t.Error(err)
	37	+	}
	38	+	}()
	39	+	conn, err := sb.Start(ctx, sandbox.JobSpec{JobID: "e2e-toolcache"})
	40	+	if err != nil {
	41	+	t.Fatal(err)
	42	+	}
	43	+	spec := jobSpec(`on: push
	44	+	jobs:
	45	+	j:
	46	+	runs-on: x
	47	+	steps:
	48	+	- uses: actions/setup-go@v7
	49	+	with:
	50	+	go-version: '1.27.1'
	51	+	cache: false
	52	+	- run: go version
	53	+	`)
	54	+	var logs []string
	55	+	var result *execpb.JobResult
	56	+	_, exit, err := execstream.DriveJob(conn, &execpb.Run{Job: spec}, nil, execstream.JobHandler{
	57	+	Resolve: func(_ context.Context, r execpb.ResolveRequest) execpb.ResolveResponse {
	58	+	resp := &execpb.ResolveResponse{}
	59	+	for _, ref := range r.Uses {
	60	+	if ref != "actions/setup-go@v7" {
	61	+	return &execpb.ResolveResponse{Error: "unexpected action: " + ref}
	62	+	}
	63	+	resp.Actions = append(resp.Actions, &execpb.ResolvedAction{Ref: ref, NameWithOwner: "actions/setup-go", ResolvedSha: "v7", TarballUrl: "https://codeload.github.com/actions/setup-go/tar.gz/refs/tags/v7"})
	64	+	}
	65	+	return resp
	66	+	},
	67	+	Log: func(l *execpb.Log) { logs = append(logs, l.Text) },
	68	+	JobResult: func(r *execpb.JobResult) { result = r },
	69	+	})
	70	+	all := strings.Join(logs, "\n")
	71	+	if err != nil {
	72	+	t.Fatalf("drive: %v\n%s", err, all)
	73	+	}
	74	+	if result == nil \|\| result.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS \|\| exit == nil \|\| exit.Code != 0 {
	75	+	t.Fatalf("result %v exit %v\n%s", result, exit, all)
	76	+	}
	77	+	for _, want := range []string{"Found in cache", "go version go1.27.1"} {
	78	+	if !strings.Contains(all, want) {
	79	+	t.Errorf("log lacks %q\n%s", want, all)
	80	+	}
	81	+	}
	82	+	}

M internal/execstream/exec.proto +1 -0


@@ -48,6 +48,7 @@ message Hello {
48	48		string version = 1; // runner version of the executor
49	49		string sandbox = 2; // "vm" or "container"
50	50		string hostname = 3;
	51	+	map<string, int64> boot_ms = 4;
51	52		}
52	53
53	54		// Run asks the executor to run a script or a job. Without job it is the stub


@@ -48,6 +48,7 @@ message Hello {
48	string version = 1; // runner version of the executor	48		string version = 1; // runner version of the executor
49	string sandbox = 2; // "vm" or "container"	49		string sandbox = 2; // "vm" or "container"
50	string hostname = 3;	50		string hostname = 3;
		51	+	map<string, int64> boot_ms = 4;
51	}	52		}
52		53
53	// Run asks the executor to run a script or a job. Without job it is the stub	54		// Run asks the executor to run a script or a job. Without job it is the stub

M internal/execstream/execpb/exec.pb.go +39 -25

show this file

M internal/guest/cacheurl_test.go +3 -3

show this file

A internal/guest/dockerd.go +15 -0

show this file

M internal/guest/exec.go +2 -1

show this file

M internal/guest/init.go +31 -6

show this file

M internal/guest/jobexec.go +7 -1

show this file

A internal/guest/registry_test.go +35 -0

show this file

A internal/guest/timing.go +38 -0

show this file

A internal/guest/timing_test.go +33 -0

show this file

M internal/host/archives.go +39 -5

show this file

M internal/host/archives_test.go +83 -2

show this file

M internal/host/cache.go +8 -4

show this file

M internal/host/fake_test.go +8 -2

show this file

M internal/host/job.go +23 -0

show this file

A internal/host/registry_test.go +49 -0

show this file

M internal/host/resolve.go +8 -0

show this file

M internal/host/resolve_test.go +1 -1

show this file

M internal/host/results_test.go +1 -1

show this file

M internal/host/runner.go +2 -2

show this file

M internal/host/runner_test.go +12 -0

show this file

M internal/host/shutdown_test.go +1 -1

show this file

A internal/host/timeline.go +52 -0

show this file

A internal/host/timeline_test.go +74 -0

show this file

A internal/registryproxy/accept.go +41 -0

show this file

A internal/registryproxy/accept_test.go +112 -0

show this file

A internal/registryproxy/clients.go +105 -0

show this file

A internal/registryproxy/flights_test.go +248 -0

show this file

A internal/registryproxy/helpers_test.go +375 -0

show this file

A internal/registryproxy/hub_test.go +80 -0

show this file

A internal/registryproxy/proxy.go +338 -0

show this file

A internal/registryproxy/proxy_test.go +117 -0

show this file

A internal/registryproxy/quota.go +61 -0

show this file

A internal/registryproxy/slots_test.go +215 -0

show this file

A internal/registryproxy/store.go +784 -0

show this file

A internal/registryproxy/store_test.go +327 -0

show this file

A internal/registryproxy/upstream.go +169 -0

show this file

A internal/registryproxy/upstream_test.go +172 -0

show this file

M internal/sandbox/container/container_test.go +11 -0

show this file

M internal/sandbox/container/driver.go +34 -6

show this file

M internal/sandbox/sandbox.go +7 -0

show this file

M internal/sandbox/vm/driver.go +42 -7

show this file

M internal/sandbox/vm/vm_test.go +10 -0

show this file

M main.go +1 -0

show this file

A registry_setup_test.go +75 -0

show this file

Merge settings

Comments1 open / 1

Commit	Subject	Author	Date	Signature
2b9f5c9	refactor(registryproxy): simplify the proxy store, guest timing contract and size parsing	brice	22h ago	GPG Verified
57617df	feat(registryproxy): cache Docker Hub pulls of the sandboxes on the runner host	brice	29h ago	GPG Verified
09df1ab	Merge branch 'perf/p1-timing' into perf/job-timing-and-caches	brice	32h ago	GPG Verified
2b98cb5	feat(timing): log the duration of each job phase on job finished	brice	32h ago	GPG Verified
be87239	feat(images): preinstall Go and Node.js in the tool cache of the job image	brice	32h ago	GPG Verified