From c1becfcae04641f06db5179e5a16a23e71002b3d Mon Sep 17 00:00:00 2001 From: Joshua Scarsbrook Date: Tue, 29 Sep 2026 20:47:29 +1000 Subject: [PATCH] Fix Alpine bootstrap and add agent VM workflows --- AGENTS.star | 203 ++++++++++++++++++++++++++ README.md | 6 +- cmd/cc/daemon_test.go | 88 +++++++++++ cmd/cc/main.go | 18 ++- internal/ccvmd/ccvmd.go | 9 +- internal/ccvmd/kernel_stream_test.go | 100 +++++++++++++ internal/kernel/alpine/kernel.go | 6 +- internal/kernel/alpine/kernel_test.go | 22 ++- 8 files changed, 441 insertions(+), 11 deletions(-) create mode 100644 AGENTS.star create mode 100644 cmd/cc/daemon_test.go create mode 100644 internal/ccvmd/kernel_stream_test.go diff --git a/AGENTS.star b/AGENTS.star new file mode 100644 index 00000000..b02222ab --- /dev/null +++ b/AGENTS.star @@ -0,0 +1,203 @@ +# Repository tooling for tinyrange/cc. Installation requires user approval. +load("//stdlib/git.star", git_tools = "tools") +load("//stdlib/golang.star", go_tools = "tools", go_run = "run", go_test = "test", go_bench = "bench", go_fuzz = "fuzz") +load("//stdlib/github.star", github_tools = "tools", pr_create = "pr_create", pr_comment = "pr_comment", pr_merge = "pr_merge", workflow_run = "workflow_run", run_rerun = "run_rerun", run_cancel = "run_cancel", queue_enqueue = "queue_enqueue") + +workspace = privileged.workspace(".", readonly = False) +git = git_tools +go = go_tools + module("go_execution", run = go_run, test = go_test, bench = go_bench, fuzz = go_fuzz) +github = github_tools + module("github_mutations", pr_create = pr_create, pr_comment = pr_comment, pr_merge = pr_merge, workflow_run = workflow_run, run_rerun = run_rerun, run_cancel = run_cancel, queue_enqueue = queue_enqueue) + +_SUFFIX = ".exe" if platform.startswith("windows/") else "" +_BUILD = "build/staragent" +_CACHE = "local/staragent/cc-cache" + +def _operand(value): + if type(value) != "string" or not value or value.startswith("-"): + fail("expected a nonempty operand not starting with '-'") + return value + +def _argv(values): + if type(values) not in ["list", "tuple"] or not values: + fail("command must be a nonempty list/tuple of strings") + for value in values: + if type(value) != "string": + fail("command arguments must be strings") + return list(values) + +def guestinit(): + """Build embedded guest payloads before compilation or tests.""" + return go.run("./internal/cmd/build-guestinit", cwd = workspace, check = True, timeout_ms = 600000) + +def _mkdirs(path): + # workspace.mkdir creates one directory and rejects existing paths. + parent = "" + for part in path.split("/"): + if part in ["", ".", ".."]: + fail("expected a clean workspace-relative directory path") + child = parent + "/" + part if parent else part + if part not in workspace.list_dir(parent): + workspace.mkdir(child) + # Also reject an existing regular file instead of silently accepting it. + workspace.list_dir(child) + parent = child + +def build(vmsh = False, glass = False): + """Build this checkout, never a downloaded runtime; return build results.""" + _mkdirs(_BUILD) + results = [guestinit()] + for name in ["cc", "ccvm"] + (["glass"] if glass else []): + results.append(go.build_process("./cmd/" + name, output = workspace.path(_BUILD + "/" + name + _SUFFIX), cwd = workspace, check = True, timeout_ms = 600000)) + if vmsh: + results.append(go.build_process("./cmd/vmsh", output = workspace.path(_BUILD + "/vmsh" + _SUFFIX), cwd = workspace.path("frontends/vmsh"), check = True, timeout_ms = 600000)) + return results + +def test(packages = ["./..."], vmsh = False, race = False): + """Build payloads then run selected short tests; opt into vmsh/race.""" + results = [guestinit(), go.test(packages = packages, short = True, race = race, count = 1, timeout = "30m", timeout_ms = 1860000, cwd = workspace, check = True)] + if vmsh: + results.append(go.test("./...", short = True, count = 1, timeout = "3m", timeout_ms = 240000, cwd = workspace.path("frontends/vmsh"), check = True)) + return results + +def cc(args, stdin = None, background = False, timeout_ms = 600000, check = False, env = None): + """Run checkout cc with a dedicated cache. Returns an inspectable process. + + cc auto-starts its daemon. Background handles need explicit wait/kill/close. + A client timeout is not proof the guest command or VM has stopped. + """ + args = _argv(args) + # cc creates the cache recursively and enforces private permissions. + return privileged.run(workspace.path(_BUILD + "/cc" + _SUFFIX), cwd = workspace, stdin = stdin, background = background, timeout_ms = timeout_ms, check = check, env = env, output_limit = 4194304, *(["-ccvm", workspace.path(_BUILD + "/ccvm" + _SUFFIX), "-cache-dir", workspace.path(_CACHE)] + args)) + +def doctor(): + """Download/check the kernel and report virtualization availability.""" + return cc(["doctor"], check = True) + +def images(name = None): + return cc(["images"] + ([_operand(name)] if name != None else []), check = True) + +def pull(name, source, timeout_ms = 1800000): + """Import an OCI reference, local SIMG/SIF, or CVMFS source.""" + return cc(["pull", _operand(name), _operand(source)], timeout_ms = timeout_ms, check = True) + +def start(name, image, network = False, shares = [], memory_mb = None, cpus = None, boot_timeout = "60s", init = None, default_user = None, vnc = False): + """Start a named VM. Shares are explicit WRITABLE HOST:GUEST specs.""" + args = ["vm", "start", "--timeout=" + boot_timeout] + if network: + args.append("--network") + if vnc: + args.append("--vnc") + for flag, value in [("memory-mb", memory_mb), ("cpus", cpus)]: + if value != None: + if type(value) != "int" or value <= 0: + fail(flag + " must be a positive integer") + args.append("--" + flag + "=" + str(value)) + for flag, value in [("init", init), ("default-user", default_user)]: + if value != None: + args.append("--" + flag + "=" + _operand(value)) + if type(shares) not in ["list", "tuple"]: + fail("shares must be a list/tuple of explicit writable HOST:GUEST specs") + for share in shares: + args.append("--share=" + _operand(share)) + return cc(args + ["--", _operand(name), _operand(image)], check = True) + +def vm_list(): + return cc(["vm", "list"], check = True) + +def status(name): + return cc(["vm", "status", _operand(name)], check = True) + +def stop(name): + """Stop only the specified VM; never stop unrelated instances.""" + return cc(["vm", "stop", _operand(name)], check = True) + +def exec_vm(name, argv, stdin = None, background = False, timeout_ms = 600000, check = False): + """Execute argv in a running VM without host-shell interpolation.""" + return cc(["vm", "run", _operand(name), "--"] + _argv(argv), stdin = stdin, background = background, timeout_ms = timeout_ms, check = check) + +def shell(name, script, stdin = None, background = False, timeout_ms = 600000, check = False): + """Explicit guest /bin/sh script execution; requires sh in the image.""" + return exec_vm(name, ["sh", "-lc", script], stdin = stdin, background = background, timeout_ms = timeout_ms, check = check) + +def run_image(image, argv, stdin = None, timeout_ms = 600000, check = False): + """Run via the default-instance CLI flow; prefer named VMs for isolation.""" + return cc(["run", _operand(image), "--"] + _argv(argv), stdin = stdin, timeout_ms = timeout_ms, check = check) + +def forward(name, mapping): + return cc(["vm", "forward", _operand(name), _operand(mapping)], check = True) + +def open_pr(title, body, base, draft = False, remote = "origin"): + """Push an already committed feature branch and open a PR. No auto-commit.""" + branch = git.current_branch(cwd = workspace) + if not branch or branch in [base, "main", "master"]: + fail("switch to a feature branch before opening a PR") + if not git.is_clean(cwd = workspace): + fail("review and commit changes before opening a PR") + git.push(remote = remote, branch = branch, set_upstream = True, cwd = workspace, check = True) + return github.pr_create(title, body, base = base, head = branch, draft = draft, cwd = workspace, check = True) + +def merge_pr(number, expected_head, method = "squash", auto = False): + """Merge an explicitly reviewed SHA; no admin bypass or branch deletion. + + Direct merges require successful reported checks. auto=True requests GitHub + auto-merge subject to repository policy; it does not imply the PR is merged. + """ + if not regexp.compile(r"^([0-9a-fA-F]{40}|[0-9a-fA-F]{64})$").matches(expected_head): + fail("expected_head must be the full reviewed commit SHA") + result = github.pr_view(str(number), json_fields = ["state", "isDraft", "headRefOid"], cwd = workspace, check = True) + if result.stdout_truncated: + fail("PR metadata was truncated") + pr = json.decode(result.stdout) + if pr["state"] != "OPEN" or pr["isDraft"] or pr["headRefOid"] != expected_head.lower(): + fail("PR must be open, ready, and still at the reviewed SHA") + if not auto: + github.pr_checks(str(number), cwd = workspace, check = True) + return github.pr_merge(str(number), method = method, auto = auto, match_head_commit = expected_head.lower(), delete_branch = False, cwd = workspace, check = True) + +vm = module("vm", cc = cc, doctor = doctor, images = images, pull = pull, start = start, list = vm_list, status = status, stop = stop, exec = exec_vm, shell = shell, run_image = run_image, forward = forward) +workflows = module("workflows", guestinit = guestinit, build = build, test = test, open_pr = open_pr, merge_pr = merge_pr) + +def propose_agents_star(content): + """Stage a configuration; only user approval installs it.""" + candidate = privileged.tempdir() + path = candidate.path("candidate.star") + privileged.write_file(path, content) + return privileged.edit_agents_star(path) + +_INSTRUCTIONS = """ +This is tinyrange/cc, an experimental Go microVM runtime. Read local source and +.github/workflows/test.yml for authoritative flags/checks. README can lag source. +Key areas: cmd/cc CLI, cmd/ccvm daemon, client, internal/ccvmd API, +internal/vm orchestration, internal/hv backends, internal/oci image import. +frontends/vmsh is a separate Go module. Match the Go version in go.mod. +Tools: workspace, git, go (including run/test/bench/fuzz), github, vm, workflows. +Use help(tool) for signatures. Processes expose stdout/stderr/success/exit_code; +inspect failures and truncation. Do not print secrets from auth or environment. +Build guest-init before builds/tests (workflows.guestinit or workflows.build). +For local VM work: workflows.build(); vm.doctor(); vm.pull(name, source); +vm.start(unique_name, image); vm.exec(name, ["uname", "-a"]); vm.stop(name). +Use fixtures/alpine.simg on amd64 and fixtures/alpine-arm64.simg on arm64. +vm.shell(name, script) executes only inside the guest. No host shell is exposed. +VM binaries live in build/staragent; cache/daemon state in local/staragent/cc-cache. +Do not rebuild binaries under a running daemon and assume the daemon upgraded. +Named VMs persist: track and stop only instances you created, including on failure. +CLI timeout/cancellation is not guaranteed guest cleanup. Inspect status explicitly. +Never auto-enable network, writable host shares, or change host hypervisor permissions. +Writable shares let guest commands modify host files. Do not mount secrets or the +repository/configuration into untrusted guests. The process tools are not a sandbox. +Run focused checks for the change. CI includes vet, short tests, API contract, +targeted race checks and separate vmsh checks. Expensive KVM tests are opt-in. +Do not install dependencies or invoke release/deploy workflows without authorization. +PR flow: inspect git status/diff, create a feature branch, make focused changes, +run relevant checks, stage explicit paths and commit, then workflows.open_pr with +an explicit base. Review PR diff/checks and head SHA before workflows.merge_pr. +Only push/open/merge when authorized by the user's task. Never bypass protections. +Use github workflow/run inspection to diagnose CI; dispatch/rerun/cancel explicitly. +For merge queues use github.queue_enqueue with the reviewed SHA and verify status. +Auto-merge/queue acceptance is not proof of merge: re-read PR state before reporting. +Never edit AGENTS.star through file/process tools; use propose_agents_star approval. +""" + +environment = {"workspace": workspace, "git": git, "go": go, "github": github, "vm": vm, "workflows": workflows, "propose_agents_star": propose_agents_star} +default_repl = repl(environment) +default = privileged.model("gpt-6-astra").create(default_repl, prompt_addons = [_INSTRUCTIONS]) diff --git a/README.md b/README.md index a7f91a8d..e548dd91 100644 --- a/README.md +++ b/README.md @@ -137,8 +137,10 @@ go build -o "$tmp/glass" ./cmd/glass ## CLI Usage -`cc` starts `ccvm` automatically when needed. If the binaries are not installed -next to each other, pass the daemon path explicitly: +`cc` starts `ccvm` automatically when needed. Auto-started daemon diagnostics +are appended to `ccvm.log` in the selected cache directory (set with +`-cache-dir`), rather than keeping the caller's stderr open after `cc` exits. +If the binaries are not installed next to each other, pass the daemon path explicitly: ```sh cc -ccvm ./ccvm doctor diff --git a/cmd/cc/daemon_test.go b/cmd/cc/daemon_test.go new file mode 100644 index 00000000..c9b22477 --- /dev/null +++ b/cmd/cc/daemon_test.go @@ -0,0 +1,88 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "j5.nz/cc/client" +) + +// Re-execute the test binary as either the short-lived CLI or its long-lived +// daemon so the regression exercises actual inherited OS pipe handles. +func TestMain(m *testing.M) { + if os.Getenv("CC_TEST_DAEMON_HELPER") == "1" && len(os.Args) > 1 && os.Args[1] == "-cache-dir" { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + panic(err) + } + server := &http.Server{ReadHeaderTimeout: time.Second} + mux := http.NewServeMux() + mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { + fmt.Fprintln(os.Stderr, "daemon health check") + fmt.Fprintln(w, `{}`) + }) + mux.HandleFunc("/shutdown", func(w http.ResponseWriter, _ *http.Request) { + fmt.Fprintln(w, `{}`) + go server.Shutdown(context.Background()) + }) + server.Handler = mux + // Bound the helper lifetime even if its parent fails before cleanup. + time.AfterFunc(30*time.Second, func() { server.Close() }) + if err := json.NewEncoder(os.Stdout).Encode(client.ServerHello{Addr: listener.Addr().String(), Scheme: "http"}); err != nil { + panic(err) + } + _ = server.Serve(listener) + os.Exit(0) + } + os.Exit(m.Run()) +} + +func TestConnectBackendDoesNotKeepCallerPipesOpen(t *testing.T) { + if os.Getenv("CC_TEST_DAEMON_HELPER") == "1" { + cache := os.Getenv("CC_TEST_DAEMON_CACHE") + if _, err := connectBackend(os.Args[0], cache, filepath.Join(cache, "ccvm.json")); err != nil { + t.Fatal(err) + } + return + } + cache := t.TempDir() + statePath := filepath.Join(cache, "ccvm.json") + t.Cleanup(func() { + if state, err := readDaemonState(statePath); err == nil { + if err := newClient(state.Addr).Shutdown(); err != nil { + t.Errorf("shutdown helper: %v", err) + } + } + }) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestConnectBackendDoesNotKeepCallerPipesOpen$") + cmd.Env = append(os.Environ(), "CC_TEST_DAEMON_HELPER=1", "CC_TEST_DAEMON_CACHE="+cache) + cmd.WaitDelay = time.Second + if output, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("captured CLI failed: %v\n%s", err, output) + } + state, err := readDaemonState(statePath) + if err != nil { + t.Fatal(err) + } + if err := newClient(state.Addr).HealthCheck(); err != nil { + t.Fatalf("daemon did not survive CLI exit: %v", err) + } + log, err := os.ReadFile(filepath.Join(cache, "ccvm.log")) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(log), "daemon health check") { + t.Fatalf("daemon log = %q", log) + } +} diff --git a/cmd/cc/main.go b/cmd/cc/main.go index 17aeef1d..d698c918 100644 --- a/cmd/cc/main.go +++ b/cmd/cc/main.go @@ -838,20 +838,30 @@ func connectBackend(ccvmPath, cacheDir, statePath string) (*client.Client, error args := []string{"-cache-dir", cacheDir} proc := exec.Command(ccvmPath, args...) - proc.Stderr = os.Stderr + // The daemon outlives cc. Inheriting a captured stderr pipe keeps callers + // waiting for EOF after cc exits (and can SIGPIPE the daemon later). + logPath := filepath.Join(cacheDir, "ccvm.log") + logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600) + if err != nil { + return nil, fmt.Errorf("open ccvm log %s: %w", logPath, err) + } + defer logFile.Close() + proc.Stderr = logFile stdout, err := proc.StdoutPipe() if err != nil { return nil, fmt.Errorf("prepare ccvm stdout pipe for %s: %w", ccvmPath, err) } + defer stdout.Close() if err := proc.Start(); err != nil { - return nil, fmt.Errorf("start ccvm daemon %s with cache %s: %w", ccvmPath, cacheDir, err) + return nil, fmt.Errorf("start ccvm daemon %s with cache %s (log %s): %w", ccvmPath, cacheDir, logPath, err) } var hello client.ServerHello if err := json.NewDecoder(stdout).Decode(&hello); err != nil { + _ = proc.Process.Kill() _ = proc.Wait() - return nil, fmt.Errorf("ccvm daemon did not send a startup banner from %s: %w", ccvmPath, err) + return nil, fmt.Errorf("ccvm daemon did not send a startup banner from %s (log %s): %w", ccvmPath, logPath, err) } if err := validateServerHello(hello, cacheDir); err != nil { _ = proc.Process.Kill() @@ -871,6 +881,8 @@ func connectBackend(ccvmPath, cacheDir, statePath string) (*client.Client, error _ = proc.Wait() return nil, fmt.Errorf("ccvm daemon started at %s but health check failed: %w", hello.Addr, err) } + // Reap the daemon if it exits while this client is still alive. + go func() { _ = proc.Wait() }() return api, nil } diff --git a/internal/ccvmd/ccvmd.go b/internal/ccvmd/ccvmd.go index 3bb42268..8440f048 100644 --- a/internal/ccvmd/ccvmd.go +++ b/internal/ccvmd/ccvmd.go @@ -1591,7 +1591,11 @@ func newMuxWithRoutes(srvState *server, watchdog *watchdogController, shutdown f } if err := srvState.kernel.EnsureWithProgress(r.Context(), report); err != nil { _ = writeProgressEvent(w, client.ProgressEvent{Status: "error", Error: err.Error()}) + return } + // Signal success only after verification, extraction, metadata publication, + // and clearing the manager's downloading state, including cache hits. + _ = writeProgressEvent(w, client.ProgressEvent{Status: "downloaded", Artifact: "kernel"}) return } if err := srvState.kernel.Ensure(r.Context()); err != nil { @@ -2805,9 +2809,8 @@ func bootProgressMessage(prefix string, event client.ProgressEvent) string { func writeProgressEvent(w http.ResponseWriter, event client.ProgressEvent) error { w.Header().Set("Content-Type", "application/x-ndjson") - if _, ok := w.(http.Flusher); ok { - w.WriteHeader(http.StatusOK) - } + // Encode commits status 200 on the first write; subsequent events must not + // call WriteHeader again. if err := json.NewEncoder(w).Encode(event); err != nil { return err } diff --git a/internal/ccvmd/kernel_stream_test.go b/internal/ccvmd/kernel_stream_test.go new file mode 100644 index 00000000..72e7c8df --- /dev/null +++ b/internal/ccvmd/kernel_stream_test.go @@ -0,0 +1,100 @@ +package ccvmd + +import ( + "bytes" + "encoding/json" + "log" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "j5.nz/cc/client" + "j5.nz/cc/internal/kernel/alpine" + "j5.nz/cc/internal/vm" +) + +func TestKernelDownloadStreamCachedAndError(t *testing.T) { + for _, cached := range []bool{true, false} { + name := "error" + if cached { + name = "cached" + } + t.Run(name, func(t *testing.T) { + root := t.TempDir() + if cached { + pkg, symvers := filepath.Join(root, "kernel.tar"), filepath.Join(root, "Module.symvers") + for _, path := range []string{pkg, symvers} { + if err := os.WriteFile(path, nil, 0o600); err != nil { + t.Fatal(err) + } + } + meta, err := json.Marshal(map[string]string{"version": "test", "source": "alpine:test", "package_file": pkg, "module_symvers_file": symvers}) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(root, "kernel.json"), meta, 0o600); err != nil { + t.Fatal(err) + } + } else { + // Fail locally before attempting any network download. + root = filepath.Join(root, "not-a-directory") + if err := os.WriteFile(root, nil, 0o600); err != nil { + t.Fatal(err) + } + } + manager := alpine.NewManager(root) + if cached && manager.Status().Status != "downloaded" { + t.Fatalf("invalid local cache fixture: %+v", manager.Status()) + } + srv := httptest.NewServer(newMux(&server{kernel: manager, vms: vm.NewManagerWithHost(nil)}, nil, func() {}, ServerOptions{})) + defer srv.Close() + api := client.NewClient(srv.URL, nil) + for attempt := 0; attempt < 2; attempt++ { + var events []client.ProgressEvent + err := api.DownloadKernelStreamContext(t.Context(), client.DownloadRequest{}, func(event client.ProgressEvent) error { + events = append(events, event) + return nil + }) + if (err == nil) != cached { + t.Fatalf("cached=%t: error=%v", cached, err) + } + want := "error" + if cached { + want = "downloaded" + } + if len(events) != 1 || events[0].Status != want || events[0].Blob != "" { + t.Fatalf("terminal events = %+v, want one %s", events, want) + } + if cached && manager.Status().Status != "downloaded" { + t.Fatalf("kernel not ready: %+v", manager.Status()) + } + } + }) + } +} + +func TestProgressEventsDoNotRepeatResponseHeaders(t *testing.T) { + var diagnostics bytes.Buffer + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + for _, status := range []string{"downloading", "downloaded"} { + if err := writeProgressEvent(w, client.ProgressEvent{Status: status}); err != nil { + t.Error(err) + } + } + })) + srv.Config.ErrorLog = log.New(&diagnostics, "", 0) + srv.Start() + defer srv.Close() + api := client.NewClient(srv.URL, nil) + if err := api.DownloadKernelStreamContext(t.Context(), client.DownloadRequest{}, nil); err != nil { + t.Fatal(err) + } + // Close waits for the handler before inspecting its log. + srv.Close() + if strings.Contains(diagnostics.String(), "superfluous") { + t.Fatalf("HTTP diagnostics: %s", &diagnostics) + } +} diff --git a/internal/kernel/alpine/kernel.go b/internal/kernel/alpine/kernel.go index 5b0ca6a7..f2fbe4b2 100644 --- a/internal/kernel/alpine/kernel.go +++ b/internal/kernel/alpine/kernel.go @@ -816,8 +816,10 @@ func copyWithProgress(dst io.Writer, src io.Reader, total int64, artifact string eta = float64(total-downloaded) / rate } report(client.ProgressEvent{ - Status: status, - Artifact: artifact, + Status: status, + Artifact: artifact, + // Package transfer completion is not completion of the whole operation. + Blob: artifact, Progress: progress, BytesDownloaded: downloaded, BytesTotal: total, diff --git a/internal/kernel/alpine/kernel_test.go b/internal/kernel/alpine/kernel_test.go index 630347dc..f663816e 100644 --- a/internal/kernel/alpine/kernel_test.go +++ b/internal/kernel/alpine/kernel_test.go @@ -15,6 +15,8 @@ import ( "strings" "sync" "testing" + + "j5.nz/cc/client" ) func TestAPKChecksumUsesControlGzipStream(t *testing.T) { @@ -99,9 +101,27 @@ func TestEnsureDownloadedRecoversFromRepositoryIndexPackageRace(t *testing.T) { manager := NewManager(t.TempDir()) manager.mirror = server.URL manager.httpClient = server.Client() - if err := manager.ensureDownloaded(context.Background(), nil); err != nil { + var events []client.ProgressEvent + if err := manager.EnsureWithProgress(context.Background(), func(event client.ProgressEvent) { + events = append(events, event) + }); err != nil { t.Fatalf("ensure downloaded: %v", err) } + completedPackages := 0 + for _, event := range events { + if event.Status == "downloaded" { + completedPackages++ + if event.Blob == "" { + t.Fatalf("package completion would prematurely terminate client stream: %+v", event) + } + } + } + if completedPackages != 3 { + t.Fatalf("completed package transfers = %d, want 3 (including retry)", completedPackages) + } + if err := manager.Ensure(context.Background()); err != nil { + t.Fatalf("immediate cached ensure: %v", err) + } metadata, err := manager.ReadKernelMetadata() if err != nil { t.Fatalf("read kernel metadata: %v", err)