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


@@ -121,11 +121,11 @@ func TestStartCacheRelay(t *testing.T) {
121	121		}
122	122		}
123	123
124		-	// No cache service, no relay and no variables; a port that is taken leaves the
125		-	// job without a cache instead of failing it.
	124	+	// No port, no relay and no variables; a port that is taken leaves the job
	125	+	// without a cache instead of failing it.
126	126		func TestStartCacheRelayWithoutServiceOrPort(t *testing.T) {
127	127		target := Target{Network: "unix", Addr: filepath.Join(t.TempDir(), "exec.sock")}
128		-	for _, spec := range []*execpb.JobSpec{{}, {CachePort: 9089}, {CacheToken: "tok"}} {
	128	+	for _, spec := range []*execpb.JobSpec{{}, {CacheToken: "tok"}, {CacheToken: "tok", CacheMode: "write"}} {
129	129		if got := startCacheRelay(context.Background(), target, spec, noBridge); got != (cacheRelay{}) {
130	130		t.Errorf("%+v: relay %+v", spec, got)
131	131		}


@@ -121,11 +121,11 @@ func TestStartCacheRelay(t *testing.T) {
121		}	121		}
122		}	122		}
123			123
124	-	// No cache service, no relay and no variables; a port that is taken leaves the	124	+	// No port, no relay and no variables; a port that is taken leaves the job
125	-	// job without a cache instead of failing it.	125	+	// without a cache instead of failing it.
126		func TestStartCacheRelayWithoutServiceOrPort(t *testing.T) {	126		func TestStartCacheRelayWithoutServiceOrPort(t *testing.T) {
127		target := Target{Network: "unix", Addr: filepath.Join(t.TempDir(), "exec.sock")}	127		target := Target{Network: "unix", Addr: filepath.Join(t.TempDir(), "exec.sock")}
128	-	for _, spec := range []*execpb.JobSpec{{}, {CachePort: 9089}, {CacheToken: "tok"}} {	128	+	for _, spec := range []*execpb.JobSpec{{}, {CacheToken: "tok"}, {CacheToken: "tok", CacheMode: "write"}} {
129		if got := startCacheRelay(context.Background(), target, spec, noBridge); got != (cacheRelay{}) {	129		if got := startCacheRelay(context.Background(), target, spec, noBridge); got != (cacheRelay{}) {
130		t.Errorf("%+v: relay %+v", spec, got)	130		t.Errorf("%+v: relay %+v", spec, got)
131		}	131		}

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

@@ -0,0 +1,15 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package guest
	4	+
	5	+	import (
	6	+	"fmt"
	7	+
	8	+	"bnema.dev/ermite-runner/internal/sandbox"
	9	+	)
	10	+
	11	+	func defaultDockerdArgs() []string {
	12	+	return []string{"/usr/bin/dockerd", "--data-root", sandbox.GuestScratch + "/docker",
	13	+	"--storage-driver", "overlay2", "--host", "unix:///var/run/docker.sock",
	14	+	"--registry-mirror", fmt.Sprintf("http://127.0.0.1:%d", sandbox.CacheVsockPort)}
	15	+	}

@@ -0,0 +1,15 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package guest
	4	+
	5	+	import (
	6	+	"fmt"
	7	+
	8	+	"bnema.dev/ermite-runner/internal/sandbox"
	9	+	)
	10	+
	11	+	func defaultDockerdArgs() []string {
	12	+	return []string{"/usr/bin/dockerd", "--data-root", sandbox.GuestScratch + "/docker",
	13	+	"--storage-driver", "overlay2", "--host", "unix:///var/run/docker.sock",
	14	+	"--registry-mirror", fmt.Sprintf("http://127.0.0.1:%d", sandbox.CacheVsockPort)}
	15	+	}

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


@@ -33,6 +33,7 @@ const (
33	33
34	34		// ExecConfig configures the executor.
35	35		type ExecConfig struct {
	36	+	BootMs map[string]int64
36	37		Target Target
37	38		Sandbox string // "vm" or "container", reported in Hello
38	39		Version string

@@ -189,7 +190,7 @@ func serve(ctx context.Context, rw io.ReadWriteCloser, cfg ExecConfig) (int, err
189	190		conn := execstream.NewConn(rw)
190	191		host, _ := os.Hostname()
191	192		if err := conn.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{
192		-	Version: cfg.Version, Sandbox: cfg.Sandbox, Hostname: host,
	193	+	Version: cfg.Version, Sandbox: cfg.Sandbox, Hostname: host, BootMs: cfg.BootMs,
193	194		}}}); err != nil {
194	195		return 1, fmt.Errorf("send hello: %w", err)
195	196		}

 // ExecConfig configures the executor.
 type ExecConfig struct {
+	BootMs  map[string]int64
 	Target  Target
 	Sandbox string // "vm" or "container", reported in Hello
 	Version string
 	conn := execstream.NewConn(rw)
 	host, _ := os.Hostname()
 	if err := conn.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{
-		Version: cfg.Version, Sandbox: cfg.Sandbox, Hostname: host,
+		Version: cfg.Version, Sandbox: cfg.Sandbox, Hostname: host, BootMs: cfg.BootMs,
 	}}}); err != nil {
 		return 1, fmt.Errorf("send hello: %w", err)
 	}

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


@@ -11,6 +11,7 @@ import (
11	11		"net/http"
12	12		"os"
13	13		"os/signal"
	14	+	"strconv"
14	15		"strings"
15	16		"syscall"
16	17		"time"

@@ -62,6 +63,10 @@ type InitConfig struct {
62	63		// (`ermite-runner exec`) as a child, reap orphans, stop dockerd, and finish by
63	64		// powering the VM off, or exiting with the executor's code in a container.
64	65		func Init(ctx context.Context, cfg InitConfig) int {
	66	+	entry := time.Now()
	67	+	// bootMs is the boot phase timing handed to the executor (ERMITE_BOOT_MS),
	68	+	// which reports it in its Hello.
	69	+	bootMs := make(map[string]int64)
65	70		log := zerowrap.FromCtx(ctx)
66	71		// The kernel starts a VM's init with umask 0, which dockerd, the executor
67	72		// and every step would inherit: files and t.TempDir() directories would be

@@ -93,13 +98,22 @@ func Init(ctx context.Context, cfg InitConfig) int {
93	98		mode := string(sandbox.Container)
94	99		if inVM {
95	100		mode = string(sandbox.VM)
	101	+	if b, err := os.ReadFile("/proc/uptime"); err == nil {
	102	+	if fields := strings.Fields(string(b)); len(fields) > 0 {
	103	+	if seconds, err := strconv.ParseFloat(fields[0], 64); err == nil && seconds >= 0 {
	104	+	bootMs["kernel"] = max(0, int64(seconds*1000)-time.Since(entry).Milliseconds())
	105	+	}
	106	+	}
	107	+	}
	108	+	start := time.Now()
96	109		if err := setupVM(log, args); err != nil {
97	110		log.Error().Err(err).Msg("init: vm setup failed")
98	111		powerOff()
99	112		return 1
100	113		}
	114	+	bootMs["setup"] = time.Since(start).Milliseconds()
101	115		}
102		-	code := runPayload(ctx, cfg, mode, args)
	116	+	code := runPayload(ctx, cfg, mode, args, bootMs)
103	117		if inVM {
104	118		powerOff()
105	119		}

@@ -114,7 +128,9 @@ const StepUmask = 0o022
114	128		// with an empty environment.
115	129		const defaultPath = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
116	130
117		-	func runPayload(ctx context.Context, cfg InitConfig, mode string, args map[string]string) int {
	131	+	// runPayload starts dockerd and the executor and waits for them. It adds its
	132	+	// boot phases to bootMs, which the executor gets.
	133	+	func runPayload(ctx context.Context, cfg InitConfig, mode string, args map[string]string, bootMs map[string]int64) int {
118	134		log := zerowrap.FromCtx(ctx)
119	135		if os.Getenv("PATH") == "" {
120	136		_ = os.Setenv("PATH", defaultPath)

@@ -136,23 +152,31 @@ func runPayload(ctx context.Context, cfg InitConfig, mode string, args map[strin
136	152		return 1
137	153		}
138	154
	155	+	var dockerStart time.Time
139	156		var dockerPid, containerdPid int
140	157		_, statErr := os.Stat("/usr/bin/dockerd")
141	158		if !cfg.SkipDocker && (statErr == nil \|\| cfg.DockerdArgs != nil) {
142	159		_ = os.MkdirAll(sandbox.GuestScratch+"/docker", 0o711)
143	160		argv := cfg.DockerdArgs
144	161		if argv == nil {
145		-	argv = []string{"/usr/bin/dockerd", "--data-root", sandbox.GuestScratch + "/docker",
146		-	"--storage-driver", "overlay2", "--host", "unix:///var/run/docker.sock"}
	162	+	argv = defaultDockerdArgs()
147	163		// dockerd sets its umask to 0, and a containerd it starts itself
148	164		// inherits it, then every `docker exec` (the steps of a job
149	165		// container): their files would be world-writable. Start containerd
150	166		// here with our umask instead, as systemd does on GitHub's runners.
151		-	if pid, ok := startContainerd(ctx, log); ok {
	167	+	start := time.Now()
	168	+	pid, ok := startContainerd(ctx, log)
	169	+	// A containerd that never came up still cost its wait: keep it in the
	170	+	// breakdown. A missing binary returns at once (0 ms) and is omitted.
	171	+	if ms := time.Since(start).Milliseconds(); ok \|\| ms > 0 {
	172	+	bootMs["containerd"] = ms
	173	+	}
	174	+	if ok {
152	175		containerdPid = pid
153	176		argv = append(argv, "--containerd", containerdSocket)
154	177		}
155	178		}
	179	+	dockerStart = time.Now()
156	180		pid, err := spawn(argv, nil)
157	181		if err != nil {
158	182		log.Error().Err(err).Msg("init: start dockerd")

@@ -186,6 +210,7 @@ func runPayload(ctx context.Context, cfg InitConfig, mode string, args map[strin
186	210		timeout = dockerReadyTimeout
187	211		}
188	212		err := waitDockerReady(ctx, log, sock, dockerPid, timeout, sigs)
	213	+	bootMs["dockerd"] = time.Since(dockerStart).Milliseconds()
189	214		var exited *dockerdExitError
190	215		var abort *abortBootError
191	216		switch {

@@ -198,7 +223,7 @@ func runPayload(ctx context.Context, cfg InitConfig, mode string, args map[strin
198	223		}
199	224		}
200	225
201		-	execEnv := append(os.Environ(), "ERMITE_CONNECT="+target, "ERMITE_SANDBOX="+mode, "ERMITE_VERSION="+cfg.Version)
	226	+	execEnv := append(os.Environ(), "ERMITE_CONNECT="+target, "ERMITE_SANDBOX="+mode, "ERMITE_VERSION="+cfg.Version, "ERMITE_BOOT_MS="+formatBootMs(bootMs))
202	227		execPid, err := spawn(cfg.ExecArgs, execEnv)
203	228		if err != nil {
204	229		log.Error().Err(err).Msg("init: start executor")

 	"net/http"
 	"os"
 	"os/signal"
+	"strconv"
 	"strings"
 	"syscall"
 	"time"
 // (`ermite-runner exec`) as a child, reap orphans, stop dockerd, and finish by
 // powering the VM off, or exiting with the executor's code in a container.
 func Init(ctx context.Context, cfg InitConfig) int {
+	entry := time.Now()
+	// bootMs is the boot phase timing handed to the executor (ERMITE_BOOT_MS),
+	// which reports it in its Hello.
+	bootMs := make(map[string]int64)
 	log := zerowrap.FromCtx(ctx)
 	// The kernel starts a VM's init with umask 0, which dockerd, the executor
 	// and every step would inherit: files and t.TempDir() directories would be
 	mode := string(sandbox.Container)
 	if inVM {
 		mode = string(sandbox.VM)
+		if b, err := os.ReadFile("/proc/uptime"); err == nil {
+			if fields := strings.Fields(string(b)); len(fields) > 0 {
+				if seconds, err := strconv.ParseFloat(fields[0], 64); err == nil && seconds >= 0 {
+					bootMs["kernel"] = max(0, int64(seconds*1000)-time.Since(entry).Milliseconds())
+				}
+			}
+		}
+		start := time.Now()
 		if err := setupVM(log, args); err != nil {
 			log.Error().Err(err).Msg("init: vm setup failed")
 			powerOff()
 			return 1
 		}
+		bootMs["setup"] = time.Since(start).Milliseconds()
 	}
-	code := runPayload(ctx, cfg, mode, args)
+	code := runPayload(ctx, cfg, mode, args, bootMs)
 	if inVM {
 		powerOff()
 	}
 // with an empty environment.
 const defaultPath = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
-func runPayload(ctx context.Context, cfg InitConfig, mode string, args map[string]string) int {
+// runPayload starts dockerd and the executor and waits for them. It adds its
+// boot phases to bootMs, which the executor gets.
+func runPayload(ctx context.Context, cfg InitConfig, mode string, args map[string]string, bootMs map[string]int64) int {
 	log := zerowrap.FromCtx(ctx)
 	if os.Getenv("PATH") == "" {
 		_ = os.Setenv("PATH", defaultPath)
 		return 1
 	}
+	var dockerStart time.Time
 	var dockerPid, containerdPid int
 	_, statErr := os.Stat("/usr/bin/dockerd")
 	if !cfg.SkipDocker && (statErr == nil || cfg.DockerdArgs != nil) {
 		_ = os.MkdirAll(sandbox.GuestScratch+"/docker", 0o711)
 		argv := cfg.DockerdArgs
 		if argv == nil {
-			argv = []string{"/usr/bin/dockerd", "--data-root", sandbox.GuestScratch + "/docker",
+			argv = defaultDockerdArgs()
-				"--storage-driver", "overlay2", "--host", "unix:///var/run/docker.sock"}
 			// dockerd sets its umask to 0, and a containerd it starts itself
 			// inherits it, then every `docker exec` (the steps of a job
 			// container): their files would be world-writable. Start containerd
 			// here with our umask instead, as systemd does on GitHub's runners.
-			if pid, ok := startContainerd(ctx, log); ok {
+			start := time.Now()
+			pid, ok := startContainerd(ctx, log)
+			// A containerd that never came up still cost its wait: keep it in the
+			// breakdown. A missing binary returns at once (0 ms) and is omitted.
+			if ms := time.Since(start).Milliseconds(); ok || ms > 0 {
+				bootMs["containerd"] = ms
+			}
+			if ok {
 				containerdPid = pid
 				argv = append(argv, "--containerd", containerdSocket)
 			}
 		}
+		dockerStart = time.Now()
 		pid, err := spawn(argv, nil)
 		if err != nil {
 			log.Error().Err(err).Msg("init: start dockerd")
 			timeout = dockerReadyTimeout
 		}
 		err := waitDockerReady(ctx, log, sock, dockerPid, timeout, sigs)
+		bootMs["dockerd"] = time.Since(dockerStart).Milliseconds()
 		var exited *dockerdExitError
 		var abort *abortBootError
 		switch {
 		}
 	}
-	execEnv := append(os.Environ(), "ERMITE_CONNECT="+target, "ERMITE_SANDBOX="+mode, "ERMITE_VERSION="+cfg.Version)
+	execEnv := append(os.Environ(), "ERMITE_CONNECT="+target, "ERMITE_SANDBOX="+mode, "ERMITE_VERSION="+cfg.Version, "ERMITE_BOOT_MS="+formatBootMs(bootMs))
 	execPid, err := spawn(cfg.ExecArgs, execEnv)
 	if err != nil {
 		log.Error().Err(err).Msg("init: start executor")

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


@@ -353,7 +353,7 @@ func (c cacheRelay) close() {
353	353		func startCacheRelay(ctx context.Context, target Target, spec *execpb.JobSpec, bridge bridgeLookup) cacheRelay {
354	354		port := spec.GetCachePort()
355	355		url := cacheURL(port)
356		-	if url == "" \|\| spec.GetCacheToken() == "" {
	356	+	if url == "" {
357	357		return cacheRelay{}
358	358		}
359	359		log := zerowrap.FromCtx(ctx)

@@ -365,6 +365,12 @@ func startCacheRelay(ctx context.Context, target Target, spec *execpb.JobSpec, b
365	365		return cacheRelay{}
366	366		}
367	367		results := spec.GetResults()
	368	+	// A registry-only job (no cache scope, no results service: the host sends
	369	+	// no token) needs dockerd's loopback relay, but no action runtime URLs or
	370	+	// token: neither node nor docker actions get ACTIONS_* variables.
	371	+	if spec.GetCacheToken() == "" \|\| (!results && spec.GetCacheMode() == "") {
	372	+	return cacheRelay{relay: r}
	373	+	}
368	374		c := cacheRelay{relay: r, url: url, token: spec.GetCacheToken(), mode: spec.GetCacheMode(),
369	375		results: results, cacheOff: results && spec.GetCacheMode() == ""}
370	376		if bridge == nil {

 func startCacheRelay(ctx context.Context, target Target, spec *execpb.JobSpec, bridge bridgeLookup) cacheRelay {
 	port := spec.GetCachePort()
 	url := cacheURL(port)
-	if url == "" || spec.GetCacheToken() == "" {
+	if url == "" {
 		return cacheRelay{}
 	}
 	log := zerowrap.FromCtx(ctx)
 		return cacheRelay{}
 	}
 	results := spec.GetResults()
+	// A registry-only job (no cache scope, no results service: the host sends
+	// no token) needs dockerd's loopback relay, but no action runtime URLs or
+	// token: neither node nor docker actions get ACTIONS_* variables.
+	if spec.GetCacheToken() == "" || (!results && spec.GetCacheMode() == "") {
+		return cacheRelay{relay: r}
+	}
 	c := cacheRelay{relay: r, url: url, token: spec.GetCacheToken(), mode: spec.GetCacheMode(),
 		results: results, cacheOff: results && spec.GetCacheMode() == ""}
 	if bridge == nil {

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

@@ -0,0 +1,35 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package guest
	4	+
	5	+	import (
	6	+	"context"
	7	+	"strings"
	8	+	"testing"
	9	+
	10	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	11	+	)
	12	+
	13	+	func TestDefaultDockerdMirror(t *testing.T) {
	14	+	args := strings.Join(defaultDockerdArgs(), " ")
	15	+	if !strings.Contains(args, "--registry-mirror http://127.0.0.1:9089") {
	16	+	t.Fatal(args)
	17	+	}
	18	+	}
	19	+
	20	+	// A registry-only job (a port, no token, no mode, no results) gets the
	21	+	// loopback relay for dockerd and no runtime environment for its steps.
	22	+	func TestRegistryOnlyRelayHasNoRuntimeEnvironment(t *testing.T) {
	23	+	target := serveCache(t, t.TempDir())
	24	+	got := startCacheRelay(context.Background(), target, &execpb.JobSpec{CachePort: freePort(t)}, noBridge)
	25	+	defer got.close()
	26	+	if got.relay == nil {
	27	+	t.Fatal("missing loopback relay")
	28	+	}
	29	+	if got.token != "" {
	30	+	t.Fatalf("token %q for a registry-only job", got.token)
	31	+	}
	32	+	if got.url != "" \|\| got.containerURL != "" \|\| got.mode != "" \|\| got.results \|\| got.bridge != nil {
	33	+	t.Fatalf("runtime environment leaked: %+v", got)
	34	+	}
	35	+	}

@@ -0,0 +1,35 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package guest
	4	+
	5	+	import (
	6	+	"context"
	7	+	"strings"
	8	+	"testing"
	9	+
	10	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	11	+	)
	12	+
	13	+	func TestDefaultDockerdMirror(t *testing.T) {
	14	+	args := strings.Join(defaultDockerdArgs(), " ")
	15	+	if !strings.Contains(args, "--registry-mirror http://127.0.0.1:9089") {
	16	+	t.Fatal(args)
	17	+	}
	18	+	}
	19	+
	20	+	// A registry-only job (a port, no token, no mode, no results) gets the
	21	+	// loopback relay for dockerd and no runtime environment for its steps.
	22	+	func TestRegistryOnlyRelayHasNoRuntimeEnvironment(t *testing.T) {
	23	+	target := serveCache(t, t.TempDir())
	24	+	got := startCacheRelay(context.Background(), target, &execpb.JobSpec{CachePort: freePort(t)}, noBridge)
	25	+	defer got.close()
	26	+	if got.relay == nil {
	27	+	t.Fatal("missing loopback relay")
	28	+	}
	29	+	if got.token != "" {
	30	+	t.Fatalf("token %q for a registry-only job", got.token)
	31	+	}
	32	+	if got.url != "" \|\| got.containerURL != "" \|\| got.mode != "" \|\| got.results \|\| got.bridge != nil {
	33	+	t.Fatalf("runtime environment leaked: %+v", got)
	34	+	}
	35	+	}

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

@@ -0,0 +1,38 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package guest
	4	+
	5	+	import (
	6	+	"sort"
	7	+	"strconv"
	8	+	"strings"
	9	+	)
	10	+
	11	+	func formatBootMs(ms map[string]int64) string {
	12	+	keys := make([]string, 0, len(ms))
	13	+	for k := range ms {
	14	+	keys = append(keys, k)
	15	+	}
	16	+	sort.Strings(keys)
	17	+	pairs := make([]string, 0, len(keys))
	18	+	for _, k := range keys {
	19	+	pairs = append(pairs, k+"="+strconv.FormatInt(ms[k], 10))
	20	+	}
	21	+	return strings.Join(pairs, ",")
	22	+	}
	23	+
	24	+	// ParseBootMs reads init's boot timing environment, ignoring malformed pairs.
	25	+	func ParseBootMs(value string) map[string]int64 {
	26	+	ms := make(map[string]int64)
	27	+	for _, pair := range strings.Split(value, ",") {
	28	+	key, value, ok := strings.Cut(pair, "=")
	29	+	if !ok \|\| key == "" {
	30	+	continue
	31	+	}
	32	+	n, err := strconv.ParseInt(value, 10, 64)
	33	+	if err == nil && n >= 0 {
	34	+	ms[key] = n
	35	+	}
	36	+	}
	37	+	return ms
	38	+	}

@@ -0,0 +1,38 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package guest
	4	+
	5	+	import (
	6	+	"sort"
	7	+	"strconv"
	8	+	"strings"
	9	+	)
	10	+
	11	+	func formatBootMs(ms map[string]int64) string {
	12	+	keys := make([]string, 0, len(ms))
	13	+	for k := range ms {
	14	+	keys = append(keys, k)
	15	+	}
	16	+	sort.Strings(keys)
	17	+	pairs := make([]string, 0, len(keys))
	18	+	for _, k := range keys {
	19	+	pairs = append(pairs, k+"="+strconv.FormatInt(ms[k], 10))
	20	+	}
	21	+	return strings.Join(pairs, ",")
	22	+	}
	23	+
	24	+	// ParseBootMs reads init's boot timing environment, ignoring malformed pairs.
	25	+	func ParseBootMs(value string) map[string]int64 {
	26	+	ms := make(map[string]int64)
	27	+	for _, pair := range strings.Split(value, ",") {
	28	+	key, value, ok := strings.Cut(pair, "=")
	29	+	if !ok \|\| key == "" {
	30	+	continue
	31	+	}
	32	+	n, err := strconv.ParseInt(value, 10, 64)
	33	+	if err == nil && n >= 0 {
	34	+	ms[key] = n
	35	+	}
	36	+	}
	37	+	return ms
	38	+	}

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

@@ -0,0 +1,33 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package guest
	4	+
	5	+	import (
	6	+	"reflect"
	7	+	"testing"
	8	+	)
	9	+
	10	+	func TestBootMs(t *testing.T) {
	11	+	want := map[string]int64{"setup": 40, "dockerd": 2300, "kernel": 812}
	12	+	encoded := formatBootMs(want)
	13	+	if encoded != "dockerd=2300,kernel=812,setup=40" {
	14	+	t.Fatal(encoded)
	15	+	}
	16	+	if got := ParseBootMs(encoded); !reflect.DeepEqual(got, want) {
	17	+	t.Fatal(got)
	18	+	}
	19	+	if got := ParseBootMs("a=x,b,=3,c=-1,d=5"); !reflect.DeepEqual(got, map[string]int64{"d": 5}) {
	20	+	t.Fatal(got)
	21	+	}
	22	+	}
	23	+
	24	+	func TestHelloBootMs(t *testing.T) {
	25	+	want := map[string]int64{"dockerd": 23}
	26	+	_, _, _, hello, _, err := host(t, "true", 0, func(cfg *ExecConfig) { cfg.BootMs = want })
	27	+	if err != nil {
	28	+	t.Fatal(err)
	29	+	}
	30	+	if !reflect.DeepEqual(hello.BootMs, want) {
	31	+	t.Fatalf("Hello boot_ms = %v", hello.BootMs)
	32	+	}
	33	+	}

@@ -0,0 +1,33 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package guest
	4	+
	5	+	import (
	6	+	"reflect"
	7	+	"testing"
	8	+	)
	9	+
	10	+	func TestBootMs(t *testing.T) {
	11	+	want := map[string]int64{"setup": 40, "dockerd": 2300, "kernel": 812}
	12	+	encoded := formatBootMs(want)
	13	+	if encoded != "dockerd=2300,kernel=812,setup=40" {
	14	+	t.Fatal(encoded)
	15	+	}
	16	+	if got := ParseBootMs(encoded); !reflect.DeepEqual(got, want) {
	17	+	t.Fatal(got)
	18	+	}
	19	+	if got := ParseBootMs("a=x,b,=3,c=-1,d=5"); !reflect.DeepEqual(got, map[string]int64{"d": 5}) {
	20	+	t.Fatal(got)
	21	+	}
	22	+	}
	23	+
	24	+	func TestHelloBootMs(t *testing.T) {
	25	+	want := map[string]int64{"dockerd": 23}
	26	+	_, _, _, hello, _, err := host(t, "true", 0, func(cfg *ExecConfig) { cfg.BootMs = want })
	27	+	if err != nil {
	28	+	t.Fatal(err)
	29	+	}
	30	+	if !reflect.DeepEqual(hello.BootMs, want) {
	31	+	t.Fatalf("Hello boot_ms = %v", hello.BootMs)
	32	+	}
	33	+	}

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


@@ -30,7 +30,7 @@ const (
30	30		// busy runner does not trip the forge's rate limit (codeload 429).
31	31		//
32	32		// It only ever improves a resolution: an action that cannot be cached (no cache,
33		-	// no channel for this job, a ref that did not resolve to a sha, a download that
	33	+	// no cache scope for this job, a ref that did not resolve to a sha, a download that
34	34		// failed, a full disk) keeps its original URL and token, and the sandbox
35	35		// downloads it as before, with its retries and its messages. The sandbox prints
36	36		// "Download action repository" in either case, so a hit reads like a miss in the

@@ -41,7 +41,7 @@ func (j jobRun) cacheArchives(ctx context.Context, orig []ermite.ResolvedActio
41	41		return
42	42		}
43	43		if _, ok, err := cacheScope(j.job, j.contexts); err != nil \|\| !ok {
44		-	return // no channel for this job: the guest has no relay either
	44	+	return // action archives are served only to jobs with a cache scope
45	45		}
46	46		ttl := time.Duration(j.timeoutMinutes())*time.Minute + cacheTokenMargin
47	47		for i, a := range orig {

@@ -54,9 +54,27 @@ func (j jobRun) cacheArchives(ctx context.Context, orig []ermite.ResolvedActio
54	54		if !ok {
55	55		continue
56	56		}
57		-	hit, err := rt.Archives.Ensure(ctx, key, func(ctx context.Context, w io.Writer) error {
58		-	return j.fetchArchive(ctx, a, origin, w)
	57	+	// fetched: this call ran the download itself. An invalid key, a wait
	58	+	// that was cancelled or a joined download that failed are no misses.
	59	+	var fetched bool
	60	+	var fetchTime time.Duration
	61	+	var bytes int64
	62	+	hit, verify, err := rt.Archives.Ensure(ctx, key, func(ctx context.Context, w io.Writer) error {
	63	+	fetched = true
	64	+	start := time.Now()
	65	+	cw := &archiveCountWriter{Writer: w}
	66	+	err := j.fetchArchive(ctx, a, origin, cw)
	67	+	fetchTime += time.Since(start)
	68	+	bytes += cw.n
	69	+	return err
59	70		})
	71	+	if fetched {
	72	+	j.mu.Lock()
	73	+	j.timeline.addMs("archive_miss_ms", fetchTime)
	74	+	j.timeline.add("archive_misses", 1)
	75	+	j.timeline.add("archive_miss_bytes", bytes)
	76	+	j.mu.Unlock()
	77	+	}
60	78		log := j.log.With().Str("action", a.NameWithOwner).Str("sha", a.ResolvedSha).Logger()
61	79		if err != nil {
62	80		log.Warn().Err(err).Msg("action archive not cached; the sandbox downloads it")

@@ -66,7 +84,11 @@ func (j jobRun) cacheArchives(ctx context.Context, orig []ermite.ResolvedActio
66	84		if u == "" {
67	85		continue
68	86		}
69		-	log.Info().Bool("hit", hit).Msg("action archive cache")
	87	+	event := log.Info().Bool("hit", hit).Int64("verify_ms", verify.Milliseconds())
	88	+	if fetched {
	89	+	event = event.Int64("fetch_ms", fetchTime.Milliseconds()).Int64("bytes", bytes)
	90	+	}
	91	+	event.Msg("action archive cache")
70	92		c.TarballUrl, c.Token, c.AuthOrigin = u, "", ""
71	93		}
72	94		}

@@ -102,3 +124,15 @@ func (j jobRun) fetchArchive(ctx context.Context, a ermite.ResolvedAction, ori
102	124		}
103	125		return err
104	126		}
	127	+
	128	+	// archiveCountWriter counts bytes accepted by the cache.
	129	+	type archiveCountWriter struct {
	130	+	io.Writer
	131	+	n int64
	132	+	}
	133	+
	134	+	func (w *archiveCountWriter) Write(p []byte) (int, error) {
	135	+	n, err := w.Writer.Write(p)
	136	+	w.n += int64(n)
	137	+	return n, err
	138	+	}


@@ -30,7 +30,7 @@ const (
30		// busy runner does not trip the forge's rate limit (codeload 429).	30		// busy runner does not trip the forge's rate limit (codeload 429).
31		//	31		//
32		// It only ever improves a resolution: an action that cannot be cached (no cache,	32		// It only ever improves a resolution: an action that cannot be cached (no cache,
33	-	// no channel for this job, a ref that did not resolve to a sha, a download that	33	+	// no cache scope for this job, a ref that did not resolve to a sha, a download that
34		// failed, a full disk) keeps its original URL and token, and the sandbox	34		// failed, a full disk) keeps its original URL and token, and the sandbox
35		// downloads it as before, with its retries and its messages. The sandbox prints	35		// downloads it as before, with its retries and its messages. The sandbox prints
36		// "Download action repository" in either case, so a hit reads like a miss in the	36		// "Download action repository" in either case, so a hit reads like a miss in the

@@ -41,7 +41,7 @@ func (j jobRun) cacheArchives(ctx context.Context, orig []ermite.ResolvedActio
41		return	41		return
42		}	42		}
43		if _, ok, err := cacheScope(j.job, j.contexts); err != nil \|\| !ok {	43		if _, ok, err := cacheScope(j.job, j.contexts); err != nil \|\| !ok {
44	-	return // no channel for this job: the guest has no relay either	44	+	return // action archives are served only to jobs with a cache scope
45		}	45		}
46		ttl := time.Duration(j.timeoutMinutes())*time.Minute + cacheTokenMargin	46		ttl := time.Duration(j.timeoutMinutes())*time.Minute + cacheTokenMargin
47		for i, a := range orig {	47		for i, a := range orig {

@@ -54,9 +54,27 @@ func (j jobRun) cacheArchives(ctx context.Context, orig []ermite.ResolvedActio
54		if !ok {	54		if !ok {
55		continue	55		continue
56		}	56		}
57	-	hit, err := rt.Archives.Ensure(ctx, key, func(ctx context.Context, w io.Writer) error {	57	+	// fetched: this call ran the download itself. An invalid key, a wait
58	-	return j.fetchArchive(ctx, a, origin, w)	58	+	// that was cancelled or a joined download that failed are no misses.
			59	+	var fetched bool
			60	+	var fetchTime time.Duration
			61	+	var bytes int64
			62	+	hit, verify, err := rt.Archives.Ensure(ctx, key, func(ctx context.Context, w io.Writer) error {
			63	+	fetched = true
			64	+	start := time.Now()
			65	+	cw := &archiveCountWriter{Writer: w}
			66	+	err := j.fetchArchive(ctx, a, origin, cw)
			67	+	fetchTime += time.Since(start)
			68	+	bytes += cw.n
			69	+	return err
59		})	70		})
			71	+	if fetched {
			72	+	j.mu.Lock()
			73	+	j.timeline.addMs("archive_miss_ms", fetchTime)
			74	+	j.timeline.add("archive_misses", 1)
			75	+	j.timeline.add("archive_miss_bytes", bytes)
			76	+	j.mu.Unlock()
			77	+	}
60		log := j.log.With().Str("action", a.NameWithOwner).Str("sha", a.ResolvedSha).Logger()	78		log := j.log.With().Str("action", a.NameWithOwner).Str("sha", a.ResolvedSha).Logger()
61		if err != nil {	79		if err != nil {
62		log.Warn().Err(err).Msg("action archive not cached; the sandbox downloads it")	80		log.Warn().Err(err).Msg("action archive not cached; the sandbox downloads it")

@@ -66,7 +84,11 @@ func (j jobRun) cacheArchives(ctx context.Context, orig []ermite.ResolvedActio
66		if u == "" {	84		if u == "" {
67		continue	85		continue
68		}	86		}
69	-	log.Info().Bool("hit", hit).Msg("action archive cache")	87	+	event := log.Info().Bool("hit", hit).Int64("verify_ms", verify.Milliseconds())
			88	+	if fetched {
			89	+	event = event.Int64("fetch_ms", fetchTime.Milliseconds()).Int64("bytes", bytes)
			90	+	}
			91	+	event.Msg("action archive cache")
70		c.TarballUrl, c.Token, c.AuthOrigin = u, "", ""	92		c.TarballUrl, c.Token, c.AuthOrigin = u, "", ""
71		}	93		}
72		}	94		}

@@ -102,3 +124,15 @@ func (j jobRun) fetchArchive(ctx context.Context, a ermite.ResolvedAction, ori
102		}	124		}
103		return err	125		return err
104		}	126		}
			127	+
			128	+	// archiveCountWriter counts bytes accepted by the cache.
			129	+	type archiveCountWriter struct {
			130	+	io.Writer
			131	+	n int64
			132	+	}
			133	+
			134	+	func (w *archiveCountWriter) Write(p []byte) (int, error) {
			135	+	n, err := w.Writer.Write(p)
			136	+	w.n += int64(n)
			137	+	return n, err
			138	+	}

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


@@ -3,7 +3,10 @@
3	3		package host
4	4
5	5		import (
	6	+	"bytes"
6	7		"context"
	8	+	"encoding/json"
	9	+	"io"
7	10		"net/http"
8	11		"net/http/httptest"
9	12		"path/filepath"

@@ -58,7 +61,7 @@ func newArchiveRig(t testing.T) archiveRig {
58	61		func (a archiveRig) job(id string) jobRun {
59	62		j := &jobRun{r: a.r, log: zerowrap.Logger{}, ctx: context.Background(),
60	63		job: &ermite.Job{JobId: id, Ref: "refs/heads/main"},
61		-	contexts: `{"github":{"repository":"o/r"}}`}
	64	+	contexts: `{"github":{"repository":"o/r"}}`, timeline: timeline{}}
62	65		return j
63	66		}
64	67

@@ -128,7 +131,7 @@ func TestCacheArchivesFallsBack(t *testing.T) {
128	131		}
129	132		}
130	133
131		-	// No cache channel for the job (no repository): the guest has no relay.
	134	+	// No cache scope for the job (no repository): archives are not served.
132	135		o, c = a.action(strings.Repeat("d", 40))
133	136		j := a.job("j")
134	137		j.contexts = `{}`

@@ -145,3 +148,81 @@ func TestCacheArchivesFallsBack(t *testing.T) {
145	148		t.Errorf("cache off: %+v", c)
146	149		}
147	150		}
	151	+
	152	+	func TestArchiveTimingLogs(t *testing.T) {
	153	+	a := newArchiveRig(t)
	154	+	var logs bytes.Buffer
	155	+	j := a.job("timing")
	156	+	j.log = zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &logs})
	157	+	for i := 0; i < 2; i++ {
	158	+	logs.Reset()
	159	+	o, c := a.action(strings.Repeat("a", 40))
	160	+	j.cacheArchives(context.Background(), []ermite.ResolvedAction{o}, []execpb.ResolvedAction{c})
	161	+	var fields map[string]any
	162	+	if err := json.Unmarshal(logs.Bytes(), &fields); err != nil {
	163	+	t.Fatal(err, logs.String())
	164	+	}
	165	+	if value, ok := fields["verify_ms"].(float64); !ok \|\| value < 0 {
	166	+	t.Fatal(fields)
	167	+	}
	168	+	if fields["hit"] != (i == 1) {
	169	+	t.Fatal(fields)
	170	+	}
	171	+	if i == 0 {
	172	+	if value, ok := fields["fetch_ms"].(float64); !ok \|\| value < 0 {
	173	+	t.Fatal(fields)
	174	+	}
	175	+	if value, ok := fields["bytes"].(float64); !ok \|\| value <= 0 {
	176	+	t.Fatal(fields)
	177	+	}
	178	+	} else if _, ok := fields["fetch_ms"]; ok {
	179	+	t.Fatal("hit has fetch_ms", fields)
	180	+	}
	181	+	}
	182	+	if j.timeline["archive_misses"] != 1 \|\| j.timeline["archive_miss_bytes"] <= 0 {
	183	+	t.Fatal(j.timeline)
	184	+	}
	185	+	}
	186	+
	187	+	// Only a download the job ran itself is a miss: a failed one counts, a job
	188	+	// whose wait for another job's download is cancelled does not.
	189	+	func TestArchiveMissesCountFetches(t *testing.T) {
	190	+	a := newArchiveRig(t)
	191	+	a.status.Store(http.StatusTooManyRequests)
	192	+	failed := a.job("failed")
	193	+	o, c := a.action(strings.Repeat("a", 40))
	194	+	failed.cacheArchives(context.Background(), []ermite.ResolvedAction{o}, []execpb.ResolvedAction{c})
	195	+	if failed.timeline["archive_misses"] != 1 {
	196	+	t.Fatalf("failed fetch: %v", failed.timeline)
	197	+	}
	198	+
	199	+	a.status.Store(http.StatusOK)
	200	+	// A joiner whose wait is cancelled while another job downloads.
	201	+	release := make(chan struct{})
	202	+	started, done := make(chan struct{}), make(chan struct{})
	203	+	sha := strings.Repeat("b", 40)
	204	+	key, ok := actionscache.Key(originOf(a.srv.URL), "x/y", sha)
	205	+	if !ok {
	206	+	t.Fatal("key")
	207	+	}
	208	+	go func() {
	209	+	defer close(done)
	210	+	_, _, _ = a.r.Archives.Ensure(context.Background(), key, func(_ context.Context, w io.Writer) error {
	211	+	close(started)
	212	+	<-release
	213	+	_, err := io.WriteString(w, "tarball")
	214	+	return err
	215	+	})
	216	+	}()
	217	+	<-started
	218	+	ctx, cancel := context.WithCancel(context.Background())
	219	+	cancel()
	220	+	joiner := a.job("joiner")
	221	+	o, c = a.action(sha)
	222	+	joiner.cacheArchives(ctx, []ermite.ResolvedAction{o}, []execpb.ResolvedAction{c})
	223	+	close(release)
	224	+	<-done // the download writes under t.TempDir: let it finish before cleanup
	225	+	if _, ok := joiner.timeline["archive_misses"]; ok {
	226	+	t.Fatalf("cancelled joiner: %v", joiner.timeline)
	227	+	}
	228	+	}


@@ -3,7 +3,10 @@
3		package host	3		package host
4			4
5		import (	5		import (
			6	+	"bytes"
6		"context"	7		"context"
			8	+	"encoding/json"
			9	+	"io"
7		"net/http"	10		"net/http"
8		"net/http/httptest"	11		"net/http/httptest"
9		"path/filepath"	12		"path/filepath"

@@ -58,7 +61,7 @@ func newArchiveRig(t testing.T) archiveRig {
58		func (a archiveRig) job(id string) jobRun {	61		func (a archiveRig) job(id string) jobRun {
59		j := &jobRun{r: a.r, log: zerowrap.Logger{}, ctx: context.Background(),	62		j := &jobRun{r: a.r, log: zerowrap.Logger{}, ctx: context.Background(),
60		job: &ermite.Job{JobId: id, Ref: "refs/heads/main"},	63		job: &ermite.Job{JobId: id, Ref: "refs/heads/main"},
61	-	contexts: `{"github":{"repository":"o/r"}}`}	64	+	contexts: `{"github":{"repository":"o/r"}}`, timeline: timeline{}}
62		return j	65		return j
63		}	66		}
64			67

@@ -128,7 +131,7 @@ func TestCacheArchivesFallsBack(t *testing.T) {
128		}	131		}
129		}	132		}
130			133
131	-	// No cache channel for the job (no repository): the guest has no relay.	134	+	// No cache scope for the job (no repository): archives are not served.
132		o, c = a.action(strings.Repeat("d", 40))	135		o, c = a.action(strings.Repeat("d", 40))
133		j := a.job("j")	136		j := a.job("j")
134		j.contexts = `{}`	137		j.contexts = `{}`

@@ -145,3 +148,81 @@ func TestCacheArchivesFallsBack(t *testing.T) {
145		t.Errorf("cache off: %+v", c)	148		t.Errorf("cache off: %+v", c)
146		}	149		}
147		}	150		}
			151	+
			152	+	func TestArchiveTimingLogs(t *testing.T) {
			153	+	a := newArchiveRig(t)
			154	+	var logs bytes.Buffer
			155	+	j := a.job("timing")
			156	+	j.log = zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &logs})
			157	+	for i := 0; i < 2; i++ {
			158	+	logs.Reset()
			159	+	o, c := a.action(strings.Repeat("a", 40))
			160	+	j.cacheArchives(context.Background(), []ermite.ResolvedAction{o}, []execpb.ResolvedAction{c})
			161	+	var fields map[string]any
			162	+	if err := json.Unmarshal(logs.Bytes(), &fields); err != nil {
			163	+	t.Fatal(err, logs.String())
			164	+	}
			165	+	if value, ok := fields["verify_ms"].(float64); !ok \|\| value < 0 {
			166	+	t.Fatal(fields)
			167	+	}
			168	+	if fields["hit"] != (i == 1) {
			169	+	t.Fatal(fields)
			170	+	}
			171	+	if i == 0 {
			172	+	if value, ok := fields["fetch_ms"].(float64); !ok \|\| value < 0 {
			173	+	t.Fatal(fields)
			174	+	}
			175	+	if value, ok := fields["bytes"].(float64); !ok \|\| value <= 0 {
			176	+	t.Fatal(fields)
			177	+	}
			178	+	} else if _, ok := fields["fetch_ms"]; ok {
			179	+	t.Fatal("hit has fetch_ms", fields)
			180	+	}
			181	+	}
			182	+	if j.timeline["archive_misses"] != 1 \|\| j.timeline["archive_miss_bytes"] <= 0 {
			183	+	t.Fatal(j.timeline)
			184	+	}
			185	+	}
			186	+
			187	+	// Only a download the job ran itself is a miss: a failed one counts, a job
			188	+	// whose wait for another job's download is cancelled does not.
			189	+	func TestArchiveMissesCountFetches(t *testing.T) {
			190	+	a := newArchiveRig(t)
			191	+	a.status.Store(http.StatusTooManyRequests)
			192	+	failed := a.job("failed")
			193	+	o, c := a.action(strings.Repeat("a", 40))
			194	+	failed.cacheArchives(context.Background(), []ermite.ResolvedAction{o}, []execpb.ResolvedAction{c})
			195	+	if failed.timeline["archive_misses"] != 1 {
			196	+	t.Fatalf("failed fetch: %v", failed.timeline)
			197	+	}
			198	+
			199	+	a.status.Store(http.StatusOK)
			200	+	// A joiner whose wait is cancelled while another job downloads.
			201	+	release := make(chan struct{})
			202	+	started, done := make(chan struct{}), make(chan struct{})
			203	+	sha := strings.Repeat("b", 40)
			204	+	key, ok := actionscache.Key(originOf(a.srv.URL), "x/y", sha)
			205	+	if !ok {
			206	+	t.Fatal("key")
			207	+	}
			208	+	go func() {
			209	+	defer close(done)
			210	+	_, _, _ = a.r.Archives.Ensure(context.Background(), key, func(_ context.Context, w io.Writer) error {
			211	+	close(started)
			212	+	<-release
			213	+	_, err := io.WriteString(w, "tarball")
			214	+	return err
			215	+	})
			216	+	}()
			217	+	<-started
			218	+	ctx, cancel := context.WithCancel(context.Background())
			219	+	cancel()
			220	+	joiner := a.job("joiner")
			221	+	o, c = a.action(sha)
			222	+	joiner.cacheArchives(ctx, []ermite.ResolvedAction{o}, []execpb.ResolvedAction{c})
			223	+	close(release)
			224	+	<-done // the download writes under t.TempDir: let it finish before cleanup
			225	+	if _, ok := joiner.timeline["archive_misses"]; ok {
			226	+	t.Fatalf("cancelled joiner: %v", joiner.timeline)
			227	+	}
			228	+	}

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


@@ -111,10 +111,11 @@ func githubRepository(contexts string) string {
111	111		// runtimeFor returns what the executor needs of the job's runtime channel: the
112	112		// port it has in the guest, the job's ACTIONS_RUNTIME_TOKEN, the cache mode and
113	113		// whether the channel serves the results service. The channel exists when the
114		-	// runner serves a cache and the job has a cache scope or the results service;
115		-	// without it, 0, "", "" and false. A job with the results service and no cache
116		-	// scope gets an empty mode (the executor then sets no ACTIONS_CACHE_URL) and a
117		-	// token that the cache refuses.
	114	+	// runner serves a cache and the job has a cache scope, the results service,
	115	+	// or the cache serves a registry proxy. Without it, 0, "", "" and false.
	116	+	// A job with the results service and no cache scope gets an empty mode and a
	117	+	// token that the cache refuses. A registry-only job (neither) gets the port
	118	+	// alone: no token, and so no runtime environment variables for its steps.
118	119		func (j jobRun) runtimeFor(rt Runner) (port uint32, token, mode string, results bool) {
119	120		if rt.Cache == nil {
120	121		return 0, "", "", false

@@ -135,6 +136,9 @@ func (j jobRun) runtimeFor(rt Runner) (port uint32, token, mode string, result
135	136		}
136	137		}
137	138		if !ok && !j.results {
	139	+	if rt.Cache.RegistryEnabled() {
	140	+	return cachesvc.DefaultPort, "", "", false
	141	+	}
138	142		return 0, "", "", false
139	143		}
140	144		var scopes []string

 // runtimeFor returns what the executor needs of the job's runtime channel: the
 // port it has in the guest, the job's ACTIONS_RUNTIME_TOKEN, the cache mode and
 // whether the channel serves the results service. The channel exists when the
-// runner serves a cache and the job has a cache scope or the results service;
+// runner serves a cache and the job has a cache scope, the results service,
-// without it, 0, "", "" and false. A job with the results service and no cache
+// or the cache serves a registry proxy. Without it, 0, "", "" and false.
-// scope gets an empty mode (the executor then sets no ACTIONS_CACHE_URL) and a
+// A job with the results service and no cache scope gets an empty mode and a
-// token that the cache refuses.
+// token that the cache refuses. A registry-only job (neither) gets the port
+// alone: no token, and so no runtime environment variables for its steps.
 func (j *jobRun) runtimeFor(rt *Runner) (port uint32, token, mode string, results bool) {
 	if rt.Cache == nil {
 		return 0, "", "", false
 		}
 	}
 	if !ok && !j.results {
+		if rt.Cache.RegistryEnabled() {
+			return cachesvc.DefaultPort, "", "", false
+		}
 		return 0, "", "", false
 	}
 	var scopes []string

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


@@ -351,6 +351,7 @@ type fakeDriver struct {
351	351		blockStart chan struct{} // when set, Start waits on it
352	352		specs []sandbox.JobSpec // what each Start was asked for
353	353		iso sandbox.Isolation
	354	+	bootMs map[string]int64 // the guest's Hello.boot_ms; nil: dockerd=2
354	355		}
355	356
356	357		func (d *fakeDriver) Mode() sandbox.Mode { return sandbox.Container }

@@ -380,7 +381,8 @@ type fakeSandbox struct {
380	381		stoped bool
381	382		}
382	383
383		-	func (s *fakeSandbox) ID() string { return "er00000000" }
	384	+	func (s *fakeSandbox) ID() string { return "er00000000" }
	385	+	func (s *fakeSandbox) StartTimings() map[string]int64 { return map[string]int64{"socket_ms": 1} }
384	386		func (s *fakeSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
385	387		d := s.d
386	388		d.mu.Lock()

@@ -401,6 +403,10 @@ func (s *fakeSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.
401	403		}
402	404		d.started++
403	405		sc := d.script
	406	+	bootMs := d.bootMs
	407	+	if bootMs == nil {
	408	+	bootMs = map[string]int64{"dockerd": 2}
	409	+	}
404	410		d.mu.Unlock()
405	411		host, guest := net.Pipe()
406	412		s.mu.Lock()

@@ -409,7 +415,7 @@ func (s *fakeSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.
409	415		go func() {
410	416		defer func() { _ = guest.Close() }()
411	417		c := execstream.NewConn(guest)
412		-	_ = c.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Version: "test", Sandbox: "container"}}})
	418	+	_ = c.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Version: "test", Sandbox: "container", BootMs: bootMs}}})
413	419		// NamedValue frames (secrets, variables) come before Run.
414	420		vals := &seenValues{secrets: map[string]string{}, vars: map[string]string{}}
415	421		var f *execpb.Frame

 	blockStart chan struct{}     // when set, Start waits on it
 	specs      []sandbox.JobSpec // what each Start was asked for
 	iso        sandbox.Isolation
+	bootMs     map[string]int64 // the guest's Hello.boot_ms; nil: dockerd=2
 }
 func (d *fakeDriver) Mode() sandbox.Mode           { return sandbox.Container }
 	stoped bool
 }
-func (s *fakeSandbox) ID() string { return "er00000000" }
+func (s *fakeSandbox) ID() string                     { return "er00000000" }
+func (s *fakeSandbox) StartTimings() map[string]int64 { return map[string]int64{"socket_ms": 1} }
 func (s *fakeSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
 	d := s.d
 	d.mu.Lock()
 	}
 	d.started++
 	sc := d.script
+	bootMs := d.bootMs
+	if bootMs == nil {
+		bootMs = map[string]int64{"dockerd": 2}
+	}
 	d.mu.Unlock()
 	host, guest := net.Pipe()
 	s.mu.Lock()
 	go func() {
 		defer func() { _ = guest.Close() }()
 		c := execstream.NewConn(guest)
-		_ = c.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Version: "test", Sandbox: "container"}}})
+		_ = c.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Version: "test", Sandbox: "container", BootMs: bootMs}}})
 		// NamedValue frames (secrets, variables) come before Run.
 		vals := &seenValues{secrets: map[string]string{}, vars: map[string]string{}}
 		var f *execpb.Frame

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


@@ -62,6 +62,8 @@ type jobRun struct {
62	62
63	63		mu sync.Mutex
64	64		steps map[int32]*ermite.StepUpdate
	65	+	// timeline is the job's phase timing, guarded by mu.
	66	+	timeline timeline
65	67		// What the guest has told us, for the validation of the next frame: the
66	68		// guest is untrusted code, its step numbers and lines are checked before
67	69		// they reach the server (see violate).

@@ -494,6 +496,7 @@ func (j *jobRun) acknowledge(ctx context.Context, shutdown <-chan struct{}) bool
494	496		}
495	497		}()
496	498		if j.ack(ackCtx) {
	499	+	j.timeSet("ack_ms", time.Since(j.receivedAt))
497	500		return true
498	501		}
499	502		select {

@@ -627,7 +630,16 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
627	630		j.features = j.loadFeatures(ctx, rt)
628	631		j.results = j.resultsOn(rt)
629	632
	633	+	start := time.Now()
630	634		conn, err := j.startSandbox(ctx, rt, sb, stopHard)
	635	+	j.timeSet("sandbox_ms", time.Since(start))
	636	+	if timed, ok := sb.(sandbox.Timed); ok && err == nil {
	637	+	j.mu.Lock()
	638	+	for k, v := range timed.StartTimings() {
	639	+	j.timeline["sb_"+k] = max(0, v)
	640	+	}
	641	+	j.mu.Unlock()
	642	+	}
631	643		if err != nil {
632	644		if ctx.Err() != nil \|\| j.cancelRequested() {
633	645		return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, shutdown: j.isStopping()}

@@ -651,7 +663,9 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
651	663		}
652	664		}()
653	665
	666	+	start = time.Now()
654	667		exit, err := j.drive(conn, rt)
	668	+	j.timeSet("run_ms", time.Since(start))
655	669		return j.outcomeOf(ctx, exit, err)
656	670		}
657	671

@@ -660,6 +674,8 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
660	674		func (j jobRun) stopSandbox(ctx context.Context, rt Runner, sb sandbox.Sandbox) {
661	675		sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.StopDeadline)
662	676		defer cancel()
	677	+	start := time.Now()
	678	+	defer func() { j.timeSet("stop_ms", time.Since(start)) }()
663	679		if err := sb.Stop(sctx); err != nil {
664	680		j.log.Error().Err(err).Msg("stopping the sandbox")
665	681		if errors.Is(err, sandbox.ErrLeaked) {

@@ -706,6 +722,11 @@ func (j jobRun) drive(conn sandbox.Conn, rt Runner) (*execpb.Exit, error) {
706	722		}
707	723		run.Job.CachePort, run.Job.CacheToken, run.Job.CacheMode, run.Job.Results = j.runtimeFor(rt)
708	724		_, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{
	725	+	Hello: func(h *execpb.Hello) {
	726	+	j.mu.Lock()
	727	+	defer j.mu.Unlock()
	728	+	j.timeline.addGuestBoot(h.GetBootMs())
	729	+	},
709	730		StepStarted: j.stepStarted, StepCompleted: j.stepCompleted,
710	731		Log: j.logLine,
711	732		JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },

@@ -818,6 +839,8 @@ func (j *jobRun) sandboxError(msg string) {
818	839		// complete reports the end of the job. CompleteJob is idempotent, so a lost
819	840		// response is retried until it is answered (applied or not).
820	841		func (j *jobRun) complete(ctx context.Context, conclusion ermite.CiConclusion, outputs map[string]string, shutdown bool) {
	842	+	start := time.Now()
	843	+	defer func() { j.timeSet("complete_ms", time.Since(start)) }()
821	844		req := &ermite.CompleteJobRequest{
822	845		JobId: j.job.JobId, Conclusion: conclusion, Outputs: outputs,
823	846		RunnerShutdown: shutdown && conclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED,


@@ -62,6 +62,8 @@ type jobRun struct {
62		62
63	mu sync.Mutex	63		mu sync.Mutex
64	steps map[int32]*ermite.StepUpdate	64		steps map[int32]*ermite.StepUpdate
		65	+	// timeline is the job's phase timing, guarded by mu.
		66	+	timeline timeline
65	// What the guest has told us, for the validation of the next frame: the	67		// What the guest has told us, for the validation of the next frame: the
66	// guest is untrusted code, its step numbers and lines are checked before	68		// guest is untrusted code, its step numbers and lines are checked before
67	// they reach the server (see violate).	69		// they reach the server (see violate).

@@ -494,6 +496,7 @@ func (j *jobRun) acknowledge(ctx context.Context, shutdown <-chan struct{}) bool
494	}	496		}
495	}()	497		}()
496	if j.ack(ackCtx) {	498		if j.ack(ackCtx) {
		499	+	j.timeSet("ack_ms", time.Since(j.receivedAt))
497	return true	500		return true
498	}	501		}
499	select {	502		select {

@@ -627,7 +630,16 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
627	j.features = j.loadFeatures(ctx, rt)	630		j.features = j.loadFeatures(ctx, rt)
628	j.results = j.resultsOn(rt)	631		j.results = j.resultsOn(rt)
629		632
		633	+	start := time.Now()
630	conn, err := j.startSandbox(ctx, rt, sb, stopHard)	634		conn, err := j.startSandbox(ctx, rt, sb, stopHard)
		635	+	j.timeSet("sandbox_ms", time.Since(start))
		636	+	if timed, ok := sb.(sandbox.Timed); ok && err == nil {
		637	+	j.mu.Lock()
		638	+	for k, v := range timed.StartTimings() {
		639	+	j.timeline["sb_"+k] = max(0, v)
		640	+	}
		641	+	j.mu.Unlock()
		642	+	}
631	if err != nil {	643		if err != nil {
632	if ctx.Err() != nil \|\| j.cancelRequested() {	644		if ctx.Err() != nil \|\| j.cancelRequested() {
633	return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, shutdown: j.isStopping()}	645		return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, shutdown: j.isStopping()}

@@ -651,7 +663,9 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
651	}	663		}
652	}()	664		}()
653		665
		666	+	start = time.Now()
654	exit, err := j.drive(conn, rt)	667		exit, err := j.drive(conn, rt)
		668	+	j.timeSet("run_ms", time.Since(start))
655	return j.outcomeOf(ctx, exit, err)	669		return j.outcomeOf(ctx, exit, err)
656	}	670		}
657		671

@@ -660,6 +674,8 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
660	func (j jobRun) stopSandbox(ctx context.Context, rt Runner, sb sandbox.Sandbox) {	674		func (j jobRun) stopSandbox(ctx context.Context, rt Runner, sb sandbox.Sandbox) {
661	sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.StopDeadline)	675		sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.StopDeadline)
662	defer cancel()	676		defer cancel()
		677	+	start := time.Now()
		678	+	defer func() { j.timeSet("stop_ms", time.Since(start)) }()
663	if err := sb.Stop(sctx); err != nil {	679		if err := sb.Stop(sctx); err != nil {
664	j.log.Error().Err(err).Msg("stopping the sandbox")	680		j.log.Error().Err(err).Msg("stopping the sandbox")
665	if errors.Is(err, sandbox.ErrLeaked) {	681		if errors.Is(err, sandbox.ErrLeaked) {

@@ -706,6 +722,11 @@ func (j jobRun) drive(conn sandbox.Conn, rt Runner) (*execpb.Exit, error) {
706	}	722		}
707	run.Job.CachePort, run.Job.CacheToken, run.Job.CacheMode, run.Job.Results = j.runtimeFor(rt)	723		run.Job.CachePort, run.Job.CacheToken, run.Job.CacheMode, run.Job.Results = j.runtimeFor(rt)
708	_, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{	724		_, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{
		725	+	Hello: func(h *execpb.Hello) {
		726	+	j.mu.Lock()
		727	+	defer j.mu.Unlock()
		728	+	j.timeline.addGuestBoot(h.GetBootMs())
		729	+	},
709	StepStarted: j.stepStarted, StepCompleted: j.stepCompleted,	730		StepStarted: j.stepStarted, StepCompleted: j.stepCompleted,
710	Log: j.logLine,	731		Log: j.logLine,
711	JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },	732		JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },

@@ -818,6 +839,8 @@ func (j *jobRun) sandboxError(msg string) {
818	// complete reports the end of the job. CompleteJob is idempotent, so a lost	839		// complete reports the end of the job. CompleteJob is idempotent, so a lost
819	// response is retried until it is answered (applied or not).	840		// response is retried until it is answered (applied or not).
820	func (j *jobRun) complete(ctx context.Context, conclusion ermite.CiConclusion, outputs map[string]string, shutdown bool) {	841		func (j *jobRun) complete(ctx context.Context, conclusion ermite.CiConclusion, outputs map[string]string, shutdown bool) {
		842	+	start := time.Now()
		843	+	defer func() { j.timeSet("complete_ms", time.Since(start)) }()
821	req := &ermite.CompleteJobRequest{	844		req := &ermite.CompleteJobRequest{
822	JobId: j.job.JobId, Conclusion: conclusion, Outputs: outputs,	845		JobId: j.job.JobId, Conclusion: conclusion, Outputs: outputs,
823	RunnerShutdown: shutdown && conclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED,	846		RunnerShutdown: shutdown && conclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED,

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

@@ -0,0 +1,49 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import (
	6	+	"testing"
	7	+
	8	+	"bnema.dev/ermite-api/gen/ermite"
	9	+
	10	+	"bnema.dev/ermite-runner/internal/cachesvc"
	11	+	"bnema.dev/ermite-runner/internal/registryproxy"
	12	+	)
	13	+
	14	+	// A job with neither a cache scope nor the results service gets the channel
	15	+	// for the registry proxy alone: its port and no token. Without the proxy, or
	16	+	// without the cache, it gets no channel.
	17	+	func TestRuntimeForRegistryOnly(t *testing.T) {
	18	+	proxy, err := registryproxy.New(registryproxy.Config{Dir: t.TempDir(), Max: 1 << 20})
	19	+	if err != nil {
	20	+	t.Fatal(err)
	21	+	}
	22	+	withRegistry, err := cachesvc.New(cachesvc.Config{Dir: t.TempDir(), Registry: proxy})
	23	+	if err != nil {
	24	+	t.Fatal(err)
	25	+	}
	26	+	defer func() { _ = withRegistry.CloseAll() }()
	27	+	without, err := cachesvc.New(cachesvc.Config{Dir: t.TempDir()})
	28	+	if err != nil {
	29	+	t.Fatal(err)
	30	+	}
	31	+	defer func() { _ = without.CloseAll() }()
	32	+	// No repository in the contexts: the job has no cache scope.
	33	+	j := &jobRun{job: &ermite.Job{JobId: "j1", TimeoutMinutes: 5}}
	34	+	port, tok, mode, results := j.runtimeFor(&Runner{Cache: withRegistry})
	35	+	if port != cachesvc.DefaultPort \|\| tok != "" \|\| mode != "" \|\| results {
	36	+	t.Fatalf("registry only: %d %q %q %v", port, tok, mode, results)
	37	+	}
	38	+	if p, _, _, _ := j.runtimeFor(&Runner{Cache: without}); p != 0 {
	39	+	t.Fatal("proxy off still supplies a channel")
	40	+	}
	41	+	if p, _, _, _ := j.runtimeFor(&Runner{}); p != 0 {
	42	+	t.Fatal("cache off still supplies a channel")
	43	+	}
	44	+	// A job with a cache scope still gets its token, proxy or not.
	45	+	scoped := &jobRun{job: &ermite.Job{JobId: "j1", Ref: "refs/heads/main", TimeoutMinutes: 5}, contexts: `{"github":{"repository":"o/r"}}`}
	46	+	if _, tok, mode, _ := scoped.runtimeFor(&Runner{Cache: withRegistry}); tok == "" \|\| mode != "write" {
	47	+	t.Fatalf("scoped job: %q %q", tok, mode)
	48	+	}
	49	+	}

@@ -0,0 +1,49 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import (
	6	+	"testing"
	7	+
	8	+	"bnema.dev/ermite-api/gen/ermite"
	9	+
	10	+	"bnema.dev/ermite-runner/internal/cachesvc"
	11	+	"bnema.dev/ermite-runner/internal/registryproxy"
	12	+	)
	13	+
	14	+	// A job with neither a cache scope nor the results service gets the channel
	15	+	// for the registry proxy alone: its port and no token. Without the proxy, or
	16	+	// without the cache, it gets no channel.
	17	+	func TestRuntimeForRegistryOnly(t *testing.T) {
	18	+	proxy, err := registryproxy.New(registryproxy.Config{Dir: t.TempDir(), Max: 1 << 20})
	19	+	if err != nil {
	20	+	t.Fatal(err)
	21	+	}
	22	+	withRegistry, err := cachesvc.New(cachesvc.Config{Dir: t.TempDir(), Registry: proxy})
	23	+	if err != nil {
	24	+	t.Fatal(err)
	25	+	}
	26	+	defer func() { _ = withRegistry.CloseAll() }()
	27	+	without, err := cachesvc.New(cachesvc.Config{Dir: t.TempDir()})
	28	+	if err != nil {
	29	+	t.Fatal(err)
	30	+	}
	31	+	defer func() { _ = without.CloseAll() }()
	32	+	// No repository in the contexts: the job has no cache scope.
	33	+	j := &jobRun{job: &ermite.Job{JobId: "j1", TimeoutMinutes: 5}}
	34	+	port, tok, mode, results := j.runtimeFor(&Runner{Cache: withRegistry})
	35	+	if port != cachesvc.DefaultPort \|\| tok != "" \|\| mode != "" \|\| results {
	36	+	t.Fatalf("registry only: %d %q %q %v", port, tok, mode, results)
	37	+	}
	38	+	if p, _, _, _ := j.runtimeFor(&Runner{Cache: without}); p != 0 {
	39	+	t.Fatal("proxy off still supplies a channel")
	40	+	}
	41	+	if p, _, _, _ := j.runtimeFor(&Runner{}); p != 0 {
	42	+	t.Fatal("cache off still supplies a channel")
	43	+	}
	44	+	// A job with a cache scope still gets its token, proxy or not.
	45	+	scoped := &jobRun{job: &ermite.Job{JobId: "j1", Ref: "refs/heads/main", TimeoutMinutes: 5}, contexts: `{"github":{"repository":"o/r"}}`}
	46	+	if _, tok, mode, _ := scoped.runtimeFor(&Runner{Cache: withRegistry}); tok == "" \|\| mode != "write" {
	47	+	t.Fatalf("scoped job: %q %q", tok, mode)
	48	+	}
	49	+	}

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


@@ -8,6 +8,7 @@ import (
8	8		"errors"
9	9		"fmt"
10	10		"strings"
	11	+	"time"
11	12		"unicode/utf8"
12	13
13	14		"connectrpc.com/connect"

@@ -111,6 +112,13 @@ func badRefRune(r rune) bool {
111	112		// (ActionManager: FailedToResolveActionDownloadInfoException). ctx ends with the
112	113		// session.
113	114		func (j jobRun) resolveActions(ctx context.Context, req execpb.ResolveRequest) *execpb.ResolveResponse {
	115	+	start := time.Now()
	116	+	defer func() {
	117	+	j.mu.Lock()
	118	+	defer j.mu.Unlock()
	119	+	j.timeline.addMs("resolve_ms", time.Since(start))
	120	+	j.timeline.add("resolve_calls", 1)
	121	+	}()
114	122		fail := func(format string, a ...any) *execpb.ResolveResponse {
115	123		return &execpb.ResolveResponse{Error: fmt.Sprintf(format, a...)}
116	124		}


@@ -8,6 +8,7 @@ import (
8	"errors"	8		"errors"
9	"fmt"	9		"fmt"
10	"strings"	10		"strings"
		11	+	"time"
11	"unicode/utf8"	12		"unicode/utf8"
12		13
13	"connectrpc.com/connect"	14		"connectrpc.com/connect"

@@ -111,6 +112,13 @@ func badRefRune(r rune) bool {
111	// (ActionManager: FailedToResolveActionDownloadInfoException). ctx ends with the	112		// (ActionManager: FailedToResolveActionDownloadInfoException). ctx ends with the
112	// session.	113		// session.
113	func (j jobRun) resolveActions(ctx context.Context, req execpb.ResolveRequest) *execpb.ResolveResponse {	114		func (j jobRun) resolveActions(ctx context.Context, req execpb.ResolveRequest) *execpb.ResolveResponse {
		115	+	start := time.Now()
		116	+	defer func() {
		117	+	j.mu.Lock()
		118	+	defer j.mu.Unlock()
		119	+	j.timeline.addMs("resolve_ms", time.Since(start))
		120	+	j.timeline.add("resolve_calls", 1)
		121	+	}()
114	fail := func(format string, a ...any) *execpb.ResolveResponse {	122		fail := func(format string, a ...any) *execpb.ResolveResponse {
115	return &execpb.ResolveResponse{Error: fmt.Sprintf(format, a...)}	123		return &execpb.ResolveResponse{Error: fmt.Sprintf(format, a...)}
116	}	124		}

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


@@ -48,7 +48,7 @@ func (c resolveStub) ResolveActions(_ context.Context, r connect.Request[ermit
48	48		}
49	49
50	50		func resolveJob(c jobClient) *jobRun {
51		-	j := &jobRun{client: c, ctx: context.Background(), job: &ermite.Job{JobId: "j1"}, log: zerowrap.Logger{},
	51	+	j := &jobRun{client: c, ctx: context.Background(), job: &ermite.Job{JobId: "j1"}, log: zerowrap.Logger{}, timeline: timeline{},
52	52		t: Timing{CallTimeout: time.Second, BackoffMin: time.Millisecond, BackoffMax: time.Millisecond}}
53	53		j.setOrigins(`{"github":{"server_url":"https://ermite.example"}}`, "https://ermite.example", "")
54	54		return j


@@ -48,7 +48,7 @@ func (c resolveStub) ResolveActions(_ context.Context, r connect.Request[ermit
48		}	48		}
49			49
50		func resolveJob(c jobClient) *jobRun {	50		func resolveJob(c jobClient) *jobRun {
51	-	j := &jobRun{client: c, ctx: context.Background(), job: &ermite.Job{JobId: "j1"}, log: zerowrap.Logger{},	51	+	j := &jobRun{client: c, ctx: context.Background(), job: &ermite.Job{JobId: "j1"}, log: zerowrap.Logger{}, timeline: timeline{},
52		t: Timing{CallTimeout: time.Second, BackoffMin: time.Millisecond, BackoffMax: time.Millisecond}}	52		t: Timing{CallTimeout: time.Second, BackoffMin: time.Millisecond, BackoffMax: time.Millisecond}}
53		j.setOrigins(`{"github":{"server_url":"https://ermite.example"}}`, "https://ermite.example", "")	53		j.setOrigins(`{"github":{"server_url":"https://ermite.example"}}`, "https://ermite.example", "")
54		return j	54		return j

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


@@ -485,7 +485,7 @@ func TestAnnotationsWaitingForTheServerAreBounded(t *testing.T) {
485	485		tm := fastTiming()
486	486		ctx, cancel := context.WithCancel(context.Background())
487	487		defer cancel()
488		-	j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{},
	488	+	j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{}, timeline: timeline{},
489	489		annSeen: map[int32]bool{}, sumSeen: map[int32]bool{}, features: jobFeatures{annotations: true},
490	490		cancelCh: make(chan *execpb.Cancel, 1), log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: logs})}
491	491		j.outbox = newOutbox(cl, "j1", tm, func(error) {}, j.log)


@@ -485,7 +485,7 @@ func TestAnnotationsWaitingForTheServerAreBounded(t *testing.T) {
485		tm := fastTiming()	485		tm := fastTiming()
486		ctx, cancel := context.WithCancel(context.Background())	486		ctx, cancel := context.WithCancel(context.Background())
487		defer cancel()	487		defer cancel()
488	-	j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{},	488	+	j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{}, timeline: timeline{},
489		annSeen: map[int32]bool{}, sumSeen: map[int32]bool{}, features: jobFeatures{annotations: true},	489		annSeen: map[int32]bool{}, sumSeen: map[int32]bool{}, features: jobFeatures{annotations: true},
490		cancelCh: make(chan *execpb.Cancel, 1), log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: logs})}	490		cancelCh: make(chan *execpb.Cancel, 1), log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: logs})}
491		j.outbox = newOutbox(cl, "j1", tm, func(error) {}, j.log)	491		j.outbox = newOutbox(cl, "j1", tm, func(error) {}, j.log)

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


@@ -211,7 +211,7 @@ func (r Runner) startJob(ctx context.Context, log zerowrap.Logger, job ermite.
211	211		r: r, job: job, client: r.Job(job.JobToken), t: t, ctx: jctx, cancel: cancel,
212	212		log: zerowrap.Logger{Logger: log.With().Str("job_id", job.JobId).Int64("run_id", job.RunId).Int32("attempt", job.Attempt).Logger()},
213	213		receivedAt: r.clock(), cancelCh: make(chan execpb.Cancel, 1), steps: map[int32]ermite.StepUpdate{},
214		-	annSeen: map[int32]bool{}, sumSeen: map[int32]bool{},
	214	+	annSeen: map[int32]bool{}, sumSeen: map[int32]bool{}, timeline: timeline{},
215	215		}
216	216		jr.log.Info().Str("job_key", job.JobKey).Msg("job assigned")
217	217		r.active.Add(1)

@@ -219,7 +219,7 @@ func (r Runner) startJob(ctx context.Context, log zerowrap.Logger, job ermite.
219	219		defer r.active.Done()
220	220		defer func() { <-slots }()
221	221		jr.run(r, shutdown)
222		-	jr.log.Info().Msg("job finished")
	222	+	jr.log.Info().Fields(jr.timingFields()).Msg("job finished")
223	223		}()
224	224		}
225	225


@@ -211,7 +211,7 @@ func (r Runner) startJob(ctx context.Context, log zerowrap.Logger, job ermite.
211		r: r, job: job, client: r.Job(job.JobToken), t: t, ctx: jctx, cancel: cancel,	211		r: r, job: job, client: r.Job(job.JobToken), t: t, ctx: jctx, cancel: cancel,
212		log: zerowrap.Logger{Logger: log.With().Str("job_id", job.JobId).Int64("run_id", job.RunId).Int32("attempt", job.Attempt).Logger()},	212		log: zerowrap.Logger{Logger: log.With().Str("job_id", job.JobId).Int64("run_id", job.RunId).Int32("attempt", job.Attempt).Logger()},
213		receivedAt: r.clock(), cancelCh: make(chan execpb.Cancel, 1), steps: map[int32]ermite.StepUpdate{},	213		receivedAt: r.clock(), cancelCh: make(chan execpb.Cancel, 1), steps: map[int32]ermite.StepUpdate{},
214	-	annSeen: map[int32]bool{}, sumSeen: map[int32]bool{},	214	+	annSeen: map[int32]bool{}, sumSeen: map[int32]bool{}, timeline: timeline{},
215		}	215		}
216		jr.log.Info().Str("job_key", job.JobKey).Msg("job assigned")	216		jr.log.Info().Str("job_key", job.JobKey).Msg("job assigned")
217		r.active.Add(1)	217		r.active.Add(1)

@@ -219,7 +219,7 @@ func (r Runner) startJob(ctx context.Context, log zerowrap.Logger, job ermite.
219		defer r.active.Done()	219		defer r.active.Done()
220		defer func() { <-slots }()	220		defer func() { <-slots }()
221		jr.run(r, shutdown)	221		jr.run(r, shutdown)
222	-	jr.log.Info().Msg("job finished")	222	+	jr.log.Info().Fields(jr.timingFields()).Msg("job finished")
223		}()	223		}()
224		}	224		}
225			225

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


@@ -39,6 +39,18 @@ func TestJobRunsAndReports(t *testing.T) {
39	39		if err := stop(); err != nil {
40	40		t.Fatal(err)
41	41		}
	42	+	fields := jobFinished(t, h.logs.String())
	43	+	ms := map[string]float64{}
	44	+	for _, key := range []string{"ack_ms", "sandbox_ms", "run_ms", "stop_ms", "complete_ms", "total_ms", "sb_socket_ms", "guest_dockerd_ms"} {
	45	+	value, ok := fields[key].(float64)
	46	+	if !ok \|\| value < 0 {
	47	+	t.Errorf("%s = %v", key, fields[key])
	48	+	}
	49	+	ms[key] = value
	50	+	}
	51	+	if ms["total_ms"] < ms["sandbox_ms"]+ms["run_ms"] {
	52	+	t.Error("total shorter than sandbox + run")
	53	+	}
42	54		s := h.srv
43	55		s.mu.Lock()
44	56		defer s.mu.Unlock()


@@ -39,6 +39,18 @@ func TestJobRunsAndReports(t *testing.T) {
39	if err := stop(); err != nil {	39		if err := stop(); err != nil {
40	t.Fatal(err)	40		t.Fatal(err)
41	}	41		}
		42	+	fields := jobFinished(t, h.logs.String())
		43	+	ms := map[string]float64{}
		44	+	for _, key := range []string{"ack_ms", "sandbox_ms", "run_ms", "stop_ms", "complete_ms", "total_ms", "sb_socket_ms", "guest_dockerd_ms"} {
		45	+	value, ok := fields[key].(float64)
		46	+	if !ok \|\| value < 0 {
		47	+	t.Errorf("%s = %v", key, fields[key])
		48	+	}
		49	+	ms[key] = value
		50	+	}
		51	+	if ms["total_ms"] < ms["sandbox_ms"]+ms["run_ms"] {
		52	+	t.Error("total shorter than sandbox + run")
		53	+	}
42	s := h.srv	54		s := h.srv
43	s.mu.Lock()	55		s.mu.Lock()
44	defer s.mu.Unlock()	56		defer s.mu.Unlock()

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


@@ -81,7 +81,7 @@ func TestReportBudgetFollowsTheShutdownChannel(t *testing.T) {
81	81		cl := &stubClient{}
82	82		ctx, cancel := context.WithCancel(context.Background())
83	83		defer cancel()
84		-	j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{},
	84	+	j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{}, timeline: timeline{},
85	85		cancelCh: make(chan *execpb.Cancel, 1), log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logBuf{}})}
86	86		j.outbox = newOutbox(cl, "j1", tm, func(error) {}, j.log)
87	87		j.outbox.start(ctx)


@@ -81,7 +81,7 @@ func TestReportBudgetFollowsTheShutdownChannel(t *testing.T) {
81		cl := &stubClient{}	81		cl := &stubClient{}
82		ctx, cancel := context.WithCancel(context.Background())	82		ctx, cancel := context.WithCancel(context.Background())
83		defer cancel()	83		defer cancel()
84	-	j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{},	84	+	j := &jobRun{client: cl, t: tm, ctx: ctx, job: &ermite.Job{JobId: "j1"}, steps: map[int32]*ermite.StepUpdate{}, timeline: timeline{},
85		cancelCh: make(chan *execpb.Cancel, 1), log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logBuf{}})}	85		cancelCh: make(chan *execpb.Cancel, 1), log: zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logBuf{}})}
86		j.outbox = newOutbox(cl, "j1", tm, func(error) {}, j.log)	86		j.outbox = newOutbox(cl, "j1", tm, func(error) {}, j.log)
87		j.outbox.start(ctx)	87		j.outbox.start(ctx)

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

@@ -0,0 +1,52 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import "time"
	6	+
	7	+	// timeline is the per-phase timing of a job, logged on its "job finished"
	8	+	// line: durations in milliseconds (the *_ms keys) and counters. A jobRun
	9	+	// creates it once, in startJob; it is guarded by jobRun.mu.
	10	+	type timeline map[string]int64
	11	+
	12	+	// setMs records d as the duration of a phase, in whole milliseconds.
	13	+	func (t timeline) setMs(key string, d time.Duration) { t[key] = max(0, d.Milliseconds()) }
	14	+
	15	+	// addMs adds d to the summed duration of a phase that may run several times.
	16	+	func (t timeline) addMs(key string, d time.Duration) { t[key] += max(0, d.Milliseconds()) }
	17	+
	18	+	// add adds n to a counter.
	19	+	func (t timeline) add(key string, n int64) { t[key] += n }
	20	+
	21	+	// guestBootKeys are the boot phases the guest may report in its Hello: the
	22	+	// host logs these as guest_<key>_ms and ignores any other key, as it does not
	23	+	// trust the guest.
	24	+	var guestBootKeys = []string{"kernel", "setup", "containerd", "dockerd"}
	25	+
	26	+	// addGuestBoot records the guest's boot phases: known keys with a value of
	27	+	// at least 0.
	28	+	func (t timeline) addGuestBoot(ms map[string]int64) {
	29	+	for _, k := range guestBootKeys {
	30	+	if v, ok := ms[k]; ok && v >= 0 {
	31	+	t["guest_"+k+"_ms"] = v
	32	+	}
	33	+	}
	34	+	}
	35	+
	36	+	func (j *jobRun) timeSet(key string, d time.Duration) {
	37	+	j.mu.Lock()
	38	+	defer j.mu.Unlock()
	39	+	j.timeline.setMs(key, d)
	40	+	}
	41	+
	42	+	// timingFields sets total_ms and returns a copy of the timeline for the log.
	43	+	func (j *jobRun) timingFields() map[string]any {
	44	+	j.mu.Lock()
	45	+	defer j.mu.Unlock()
	46	+	j.timeline.setMs("total_ms", time.Since(j.receivedAt))
	47	+	fields := make(map[string]any, len(j.timeline))
	48	+	for k, v := range j.timeline {
	49	+	fields[k] = v
	50	+	}
	51	+	return fields
	52	+	}

@@ -0,0 +1,52 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import "time"
	6	+
	7	+	// timeline is the per-phase timing of a job, logged on its "job finished"
	8	+	// line: durations in milliseconds (the *_ms keys) and counters. A jobRun
	9	+	// creates it once, in startJob; it is guarded by jobRun.mu.
	10	+	type timeline map[string]int64
	11	+
	12	+	// setMs records d as the duration of a phase, in whole milliseconds.
	13	+	func (t timeline) setMs(key string, d time.Duration) { t[key] = max(0, d.Milliseconds()) }
	14	+
	15	+	// addMs adds d to the summed duration of a phase that may run several times.
	16	+	func (t timeline) addMs(key string, d time.Duration) { t[key] += max(0, d.Milliseconds()) }
	17	+
	18	+	// add adds n to a counter.
	19	+	func (t timeline) add(key string, n int64) { t[key] += n }
	20	+
	21	+	// guestBootKeys are the boot phases the guest may report in its Hello: the
	22	+	// host logs these as guest_<key>_ms and ignores any other key, as it does not
	23	+	// trust the guest.
	24	+	var guestBootKeys = []string{"kernel", "setup", "containerd", "dockerd"}
	25	+
	26	+	// addGuestBoot records the guest's boot phases: known keys with a value of
	27	+	// at least 0.
	28	+	func (t timeline) addGuestBoot(ms map[string]int64) {
	29	+	for _, k := range guestBootKeys {
	30	+	if v, ok := ms[k]; ok && v >= 0 {
	31	+	t["guest_"+k+"_ms"] = v
	32	+	}
	33	+	}
	34	+	}
	35	+
	36	+	func (j *jobRun) timeSet(key string, d time.Duration) {
	37	+	j.mu.Lock()
	38	+	defer j.mu.Unlock()
	39	+	j.timeline.setMs(key, d)
	40	+	}
	41	+
	42	+	// timingFields sets total_ms and returns a copy of the timeline for the log.
	43	+	func (j *jobRun) timingFields() map[string]any {
	44	+	j.mu.Lock()
	45	+	defer j.mu.Unlock()
	46	+	j.timeline.setMs("total_ms", time.Since(j.receivedAt))
	47	+	fields := make(map[string]any, len(j.timeline))
	48	+	for k, v := range j.timeline {
	49	+	fields[k] = v
	50	+	}
	51	+	return fields
	52	+	}

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

@@ -0,0 +1,74 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import (
	6	+	"encoding/json"
	7	+	"fmt"
	8	+	"strings"
	9	+	"testing"
	10	+
	11	+	ermite "bnema.dev/ermite-api/gen/ermite"
	12	+	)
	13	+
	14	+	// jobFinished returns the fields of the one "job finished" line of logs.
	15	+	func jobFinished(t *testing.T, logs string) map[string]any {
	16	+	t.Helper()
	17	+	var found map[string]any
	18	+	for _, line := range strings.Split(logs, "\n") {
	19	+	var fields map[string]any
	20	+	if json.Unmarshal([]byte(line), &fields) != nil \|\| fields["message"] != "job finished" {
	21	+	continue
	22	+	}
	23	+	if found != nil {
	24	+	t.Fatal("two job finished lines")
	25	+	}
	26	+	found = fields
	27	+	}
	28	+	if found == nil {
	29	+	t.Fatalf("no job finished line: %s", logs)
	30	+	}
	31	+	return found
	32	+	}
	33	+
	34	+	// The guest is untrusted: of its Hello.boot_ms, the job finished line keeps
	35	+	// only the known phases with a value of at least 0, whatever else it sends.
	36	+	func TestHostileGuestBootTimings(t *testing.T) {
	37	+	h := newHarness(t, simpleScript("hello"))
	38	+	hostile := map[string]int64{
	39	+	"kernel": 5,
	40	+	"dockerd": -7, // negative: dropped
	41	+	"setup": 0,
	42	+	"total": 1, // would overwrite a host field as guest_total_ms
	43	+	"": 3,
	44	+	"x\"y": 4,
	45	+	"../../etc": 5,
	46	+	"containerd": 9,
	47	+	}
	48	+	for i := range 500 {
	49	+	hostile[fmt.Sprint("junk", i)] = int64(i)
	50	+	}
	51	+	h.driver.bootMs = hostile
	52	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	53	+	stop := h.run(t)
	54	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
	55	+	if err := stop(); err != nil {
	56	+	t.Fatal(err)
	57	+	}
	58	+	fields := jobFinished(t, h.logs.String())
	59	+	var guest []string
	60	+	for k := range fields {
	61	+	if strings.HasPrefix(k, "guest_") {
	62	+	guest = append(guest, k)
	63	+	}
	64	+	}
	65	+	want := map[string]float64{"guest_kernel_ms": 5, "guest_setup_ms": 0, "guest_containerd_ms": 9}
	66	+	if len(guest) != len(want) {
	67	+	t.Fatalf("guest fields %v, want %v", guest, want)
	68	+	}
	69	+	for k, v := range want {
	70	+	if got, ok := fields[k].(float64); !ok \|\| got != v {
	71	+	t.Errorf("%s = %v, want %v", k, fields[k], v)
	72	+	}
	73	+	}
	74	+	}

@@ -0,0 +1,74 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import (
	6	+	"encoding/json"
	7	+	"fmt"
	8	+	"strings"
	9	+	"testing"
	10	+
	11	+	ermite "bnema.dev/ermite-api/gen/ermite"
	12	+	)
	13	+
	14	+	// jobFinished returns the fields of the one "job finished" line of logs.
	15	+	func jobFinished(t *testing.T, logs string) map[string]any {
	16	+	t.Helper()
	17	+	var found map[string]any
	18	+	for _, line := range strings.Split(logs, "\n") {
	19	+	var fields map[string]any
	20	+	if json.Unmarshal([]byte(line), &fields) != nil \|\| fields["message"] != "job finished" {
	21	+	continue
	22	+	}
	23	+	if found != nil {
	24	+	t.Fatal("two job finished lines")
	25	+	}
	26	+	found = fields
	27	+	}
	28	+	if found == nil {
	29	+	t.Fatalf("no job finished line: %s", logs)
	30	+	}
	31	+	return found
	32	+	}
	33	+
	34	+	// The guest is untrusted: of its Hello.boot_ms, the job finished line keeps
	35	+	// only the known phases with a value of at least 0, whatever else it sends.
	36	+	func TestHostileGuestBootTimings(t *testing.T) {
	37	+	h := newHarness(t, simpleScript("hello"))
	38	+	hostile := map[string]int64{
	39	+	"kernel": 5,
	40	+	"dockerd": -7, // negative: dropped
	41	+	"setup": 0,
	42	+	"total": 1, // would overwrite a host field as guest_total_ms
	43	+	"": 3,
	44	+	"x\"y": 4,
	45	+	"../../etc": 5,
	46	+	"containerd": 9,
	47	+	}
	48	+	for i := range 500 {
	49	+	hostile[fmt.Sprint("junk", i)] = int64(i)
	50	+	}
	51	+	h.driver.bootMs = hostile
	52	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	53	+	stop := h.run(t)
	54	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
	55	+	if err := stop(); err != nil {
	56	+	t.Fatal(err)
	57	+	}
	58	+	fields := jobFinished(t, h.logs.String())
	59	+	var guest []string
	60	+	for k := range fields {
	61	+	if strings.HasPrefix(k, "guest_") {
	62	+	guest = append(guest, k)
	63	+	}
	64	+	}
	65	+	want := map[string]float64{"guest_kernel_ms": 5, "guest_setup_ms": 0, "guest_containerd_ms": 9}
	66	+	if len(guest) != len(want) {
	67	+	t.Fatalf("guest fields %v, want %v", guest, want)
	68	+	}
	69	+	for k, v := range want {
	70	+	if got, ok := fields[k].(float64); !ok \|\| got != v {
	71	+	t.Errorf("%s = %v, want %v", k, fields[k], v)
	72	+	}
	73	+	}
	74	+	}

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

@@ -0,0 +1,41 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package registryproxy
	4	+
	5	+	import (
	6	+	"sort"
	7	+	"strings"
	8	+	)
	9	+
	10	+	var manifestTypes = []string{
	11	+	"application/vnd.docker.distribution.manifest.v2+json",
	12	+	"application/vnd.docker.distribution.manifest.list.v2+json",
	13	+	"application/vnd.oci.image.manifest.v1+json",
	14	+	"application/vnd.oci.image.index.v1+json",
	15	+	}
	16	+
	17	+	// normalizeAccept bounds both negotiation and metadata cardinality to the
	18	+	// fifteen nonempty subsets of the supported manifest types.
	19	+	func normalizeAccept(raw string) string {
	20	+	seen := map[string]bool{}
	21	+	for _, part := range strings.Split(raw, ",") {
	22	+	typ, _, _ := strings.Cut(part, ";")
	23	+	typ = strings.ToLower(strings.TrimSpace(typ))
	24	+	for _, known := range manifestTypes {
	25	+	if typ == known {
	26	+	seen[typ] = true
	27	+	}
	28	+	}
	29	+	}
	30	+	if len(seen) == 0 {
	31	+	for _, typ := range manifestTypes {
	32	+	seen[typ] = true
	33	+	}
	34	+	}
	35	+	types := make([]string, 0, len(seen))
	36	+	for typ := range seen {
	37	+	types = append(types, typ)
	38	+	}
	39	+	sort.Strings(types)
	40	+	return strings.Join(types, ",")
	41	+	}

@@ -0,0 +1,41 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package registryproxy
	4	+
	5	+	import (
	6	+	"sort"
	7	+	"strings"
	8	+	)
	9	+
	10	+	var manifestTypes = []string{
	11	+	"application/vnd.docker.distribution.manifest.v2+json",
	12	+	"application/vnd.docker.distribution.manifest.list.v2+json",
	13	+	"application/vnd.oci.image.manifest.v1+json",
	14	+	"application/vnd.oci.image.index.v1+json",
	15	+	}
	16	+
	17	+	// normalizeAccept bounds both negotiation and metadata cardinality to the
	18	+	// fifteen nonempty subsets of the supported manifest types.
	19	+	func normalizeAccept(raw string) string {
	20	+	seen := map[string]bool{}
	21	+	for _, part := range strings.Split(raw, ",") {
	22	+	typ, _, _ := strings.Cut(part, ";")
	23	+	typ = strings.ToLower(strings.TrimSpace(typ))
	24	+	for _, known := range manifestTypes {
	25	+	if typ == known {
	26	+	seen[typ] = true
	27	+	}
	28	+	}
	29	+	}
	30	+	if len(seen) == 0 {
	31	+	for _, typ := range manifestTypes {
	32	+	seen[typ] = true
	33	+	}
	34	+	}
	35	+	types := make([]string, 0, len(seen))
	36	+	for typ := range seen {
	37	+	types = append(types, typ)
	38	+	}
	39	+	sort.Strings(types)
	40	+	return strings.Join(types, ",")
	41	+	}

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

@@ -0,0 +1,112 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package registryproxy
	4	+
	5	+	import (
	6	+	"fmt"
	7	+	"io"
	8	+	"net/http"
	9	+	"net/http/httptest"
	10	+	"strings"
	11	+	"sync"
	12	+	"testing"
	13	+	)
	14	+
	15	+	// Accept is reduced to the known manifest types, so its variants map to
	16	+	// a bounded set of keys.
	17	+	func TestNormalizeAccept(t *testing.T) {
	18	+	const index, v2 = "application/vnd.oci.image.index.v1+json", "application/vnd.docker.distribution.manifest.v2+json"
	19	+	all := normalizeAccept("")
	20	+	if strings.Count(all, ",") != len(manifestTypes)-1 {
	21	+	t.Fatalf("default key %q", all)
	22	+	}
	23	+	for _, tc := range []struct{ in, want string }{
	24	+	{index + ";q=0.9, " + strings.ToUpper(v2) + " , " + index, v2 + "," + index},
	25	+	{" " + index + " ;q=0.1", index},
	26	+	{"text/html, /", all},
	27	+	{"", all},
	28	+	} {
	29	+	if got := normalizeAccept(tc.in); got != tc.want {
	30	+	t.Errorf("normalizeAccept(%q) = %q, want %q", tc.in, got, tc.want)
	31	+	}
	32	+	}
	33	+	}
	34	+
	35	+	// The Accept variants a job sends for one tag reach upstream with every known
	36	+	// type they name, and make at most one tag file per subset of the known types.
	37	+	// dockerd's legacy client sends one Accept line per type; the 8 KiB bound
	38	+	// applies to the joined lines.
	39	+	func TestAcceptTagFiles(t *testing.T) {
	40	+	types := []string{
	41	+	"application/vnd.oci.image.index.v1+json",
	42	+	"application/vnd.docker.distribution.manifest.list.v2+json",
	43	+	"application/vnd.docker.distribution.manifest.v2+json",
	44	+	}
	45	+	for _, tc := range []struct {
	46	+	name string
	47	+	accepts func(i int) []string // the Accept lines of request i
	48	+	requests int
	49	+	maxTags int
	50	+	code int
	51	+	}{
	52	+	{"multi-line", func(int) []string { return types }, 2, 1, http.StatusOK},
	53	+	{"variants", func(i int) []string {
	54	+	switch i % 3 {
	55	+	case 1:
	56	+	return []string{fmt.Sprintf("x/only-junk-%d", i)}
	57	+	case 2:
	58	+	return []string{fmt.Sprintf("%s;q=0.%d, x/junk-%d, %s", types[0], i%10, i, types[1])}
	59	+	}
	60	+	return []string{fmt.Sprintf("%s;q=0.%d, x/junk-%d", types[0], i%10, i)}
	61	+	}, 300, 3, http.StatusOK},
	62	+	{"oversized", func(int) []string {
	63	+	lines := make([]string, 200)
	64	+	for i := range lines {
	65	+	lines[i] = strings.Repeat("x", 50)
	66	+	}
	67	+	return lines
	68	+	}, 1, 0, http.StatusNotFound},
	69	+	} {
	70	+	t.Run(tc.name, func(t *testing.T) {
	71	+	var mu sync.Mutex
	72	+	var seen []string
	73	+	srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
	74	+	mu.Lock()
	75	+	seen = append(seen, strings.Join(r.Header.Values("Accept"), ","))
	76	+	mu.Unlock()
	77	+	w.Header().Set("Content-Type", types[0])
	78	+	w.Header().Set("Docker-Content-Digest", sum(manifestBody))
	79	+	if r.Method == http.MethodGet {
	80	+	_, _ = io.WriteString(w, manifestBody)
	81	+	}
	82	+	})
	83	+	p := proxyWith(t, srv, Config{})
	84	+	for i := range tc.requests {
	85	+	r := httptest.NewRequest(http.MethodGet, "/v2/library/alpine/manifests/3", nil)
	86	+	lines := tc.accepts(i)
	87	+	for _, line := range lines {
	88	+	r.Header.Add("Accept", line)
	89	+	}
	90	+	w := httptest.NewRecorder()
	91	+	p.ServeHTTP(w, r)
	92	+	if w.Code != tc.code {
	93	+	t.Fatalf("request %d: %d", i, w.Code)
	94	+	}
	95	+	if tc.code != http.StatusOK {
	96	+	continue
	97	+	}
	98	+	mu.Lock()
	99	+	got := seen[len(seen)-1]
	100	+	mu.Unlock()
	101	+	for _, typ := range types {
	102	+	if strings.Contains(strings.Join(lines, ","), typ) && !strings.Contains(got, typ) {
	103	+	t.Errorf("upstream Accept %q lacks %s", got, typ)
	104	+	}
	105	+	}
	106	+	}
	107	+	if n := accounted(t, p.s); n > tc.maxTags \|\| (tc.maxTags > 0 && n == 0) {
	108	+	t.Fatalf("%d tag files for %d requests", n, tc.requests)
	109	+	}
	110	+	})
	111	+	}
	112	+	}

@@ -0,0 +1,112 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package registryproxy
	4	+
	5	+	import (
	6	+	"fmt"
	7	+	"io"
	8	+	"net/http"
	9	+	"net/http/httptest"
	10	+	"strings"
	11	+	"sync"
	12	+	"testing"
	13	+	)
	14	+
	15	+	// Accept is reduced to the known manifest types, so its variants map to
	16	+	// a bounded set of keys.
	17	+	func TestNormalizeAccept(t *testing.T) {
	18	+	const index, v2 = "application/vnd.oci.image.index.v1+json", "application/vnd.docker.distribution.manifest.v2+json"
	19	+	all := normalizeAccept("")
	20	+	if strings.Count(all, ",") != len(manifestTypes)-1 {
	21	+	t.Fatalf("default key %q", all)
	22	+	}
	23	+	for _, tc := range []struct{ in, want string }{
	24	+	{index + ";q=0.9, " + strings.ToUpper(v2) + " , " + index, v2 + "," + index},
	25	+	{" " + index + " ;q=0.1", index},
	26	+	{"text/html, /", all},
	27	+	{"", all},
	28	+	} {
	29	+	if got := normalizeAccept(tc.in); got != tc.want {
	30	+	t.Errorf("normalizeAccept(%q) = %q, want %q", tc.in, got, tc.want)
	31	+	}
	32	+	}
	33	+	}
	34	+
	35	+	// The Accept variants a job sends for one tag reach upstream with every known
	36	+	// type they name, and make at most one tag file per subset of the known types.
	37	+	// dockerd's legacy client sends one Accept line per type; the 8 KiB bound
	38	+	// applies to the joined lines.
	39	+	func TestAcceptTagFiles(t *testing.T) {
	40	+	types := []string{
	41	+	"application/vnd.oci.image.index.v1+json",
	42	+	"application/vnd.docker.distribution.manifest.list.v2+json",
	43	+	"application/vnd.docker.distribution.manifest.v2+json",
	44	+	}
	45	+	for _, tc := range []struct {
	46	+	name string
	47	+	accepts func(i int) []string // the Accept lines of request i
	48	+	requests int
	49	+	maxTags int
	50	+	code int
	51	+	}{
	52	+	{"multi-line", func(int) []string { return types }, 2, 1, http.StatusOK},
	53	+	{"variants", func(i int) []string {
	54	+	switch i % 3 {
	55	+	case 1:
	56	+	return []string{fmt.Sprintf("x/only-junk-%d", i)}
	57	+	case 2:
	58	+	return []string{fmt.Sprintf("%s;q=0.%d, x/junk-%d, %s", types[0], i%10, i, types[1])}
	59	+	}
	60	+	return []string{fmt.Sprintf("%s;q=0.%d, x/junk-%d", types[0], i%10, i)}
	61	+	}, 300, 3, http.StatusOK},
	62	+	{"oversized", func(int) []string {
	63	+	lines := make([]string, 200)
	64	+	for i := range lines {
	65	+	lines[i] = strings.Repeat("x", 50)
	66	+	}
	67	+	return lines
	68	+	}, 1, 0, http.StatusNotFound},
	69	+	} {
	70	+	t.Run(tc.name, func(t *testing.T) {
	71	+	var mu sync.Mutex
	72	+	var seen []string
	73	+	srv := upstreamWith(t, func(w http.ResponseWriter, r *http.Request) {
	74	+	mu.Lock()
	75	+	seen = append(seen, strings.Join(r.Header.Values("Accept"), ","))
	76	+	mu.Unlock()
	77	+	w.Header().Set("Content-Type", types[0])
	78	+	w.Header().Set("Docker-Content-Digest", sum(manifestBody))
	79	+	if r.Method == http.MethodGet {
	80	+	_, _ = io.WriteString(w, manifestBody)
	81	+	}
	82	+	})
	83	+	p := proxyWith(t, srv, Config{})
	84	+	for i := range tc.requests {
	85	+	r := httptest.NewRequest(http.MethodGet, "/v2/library/alpine/manifests/3", nil)
	86	+	lines := tc.accepts(i)
	87	+	for _, line := range lines {
	88	+	r.Header.Add("Accept", line)
	89	+	}
	90	+	w := httptest.NewRecorder()
	91	+	p.ServeHTTP(w, r)
	92	+	if w.Code != tc.code {
	93	+	t.Fatalf("request %d: %d", i, w.Code)
	94	+	}
	95	+	if tc.code != http.StatusOK {
	96	+	continue
	97	+	}
	98	+	mu.Lock()
	99	+	got := seen[len(seen)-1]
	100	+	mu.Unlock()
	101	+	for _, typ := range types {
	102	+	if strings.Contains(strings.Join(lines, ","), typ) && !strings.Contains(got, typ) {
	103	+	t.Errorf("upstream Accept %q lacks %s", got, typ)
	104	+	}
	105	+	}
	106	+	}
	107	+	if n := accounted(t, p.s); n > tc.maxTags \|\| (tc.maxTags > 0 && n == 0) {
	108	+	t.Fatalf("%d tag files for %d requests", n, tc.requests)
	109	+	}
	110	+	})
	111	+	}
	112	+	}

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	23h 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	33h ago	GPG Verified
2b98cb5	feat(timing): log the duration of each job phase on job finished	brice	33h ago	GPG Verified
be87239	feat(images): preinstall Go and Node.js in the tool cache of the job image	brice	33h ago	GPG Verified