From be5c5e4fd17adcabb4c0664b8d7f0638b1820ed2 Mon Sep 17 00:00:00 2001 From: dawn Date: Sat, 20 Jun 2026 12:41:47 +0000 Subject: [PATCH] spindle/microvm: fix "path is not valid" on nix cache pushes, make cache drain not fatal Signed-off-by: dawn --- spindle/engines/microvm/engine.go | 6 +++++- spindle/engines/microvm/upload_cache_narinfo.go | 4 ++++ spindle/engines/microvm/upload_cache_nix_store.go | 111 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++---- spindle/engines/microvm/upload_cache_nix_store_test.go | 56 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ spindle/engines/microvm/vm.go | 5 ++++- 5 file(s) changed, 176 insertion(s)(+), 6 deletion(s)(-) diff --git a/spindle/engines/microvm/engine.go b/spindle/engines/microvm/engine.go --- a/spindle/engines/microvm/engine.go +++ b/spindle/engines/microvm/engine.go @@ -454,7 +454,11 @@ return nil } if err := e.drainNixCache(ctx, state); err != nil { - return fmt.Errorf("drain config cache uploads before metadata commit: %w", err) + // a partial upload would leave the cache unable to realize this toplevel, + // so skip the metadata commit rather than poison it with an un-realizable + // key. the config still activated fine, so don't fail the workflow. + e.l.Warn("cache drain failed; skipping config cache metadata commit", "workflow", wid, "configKey", configKey, "toplevel", result.Toplevel, "error", err) + return nil } if err := state.NixOSToplevelCache.Commit(configKey, result.Toplevel); err != nil { return err diff --git a/spindle/engines/microvm/upload_cache_narinfo.go b/spindle/engines/microvm/upload_cache_narinfo.go --- a/spindle/engines/microvm/upload_cache_narinfo.go +++ b/spindle/engines/microvm/upload_cache_narinfo.go @@ -15,6 +15,8 @@ StorePath string URL string NarHash string NarSize int64 + // paths this path depends on + References []string } const ( @@ -62,6 +64,8 @@ if err != nil { return nil, fmt.Errorf("invalid NarSize %q: %w", value, err) } info.NarSize = n + case "References": + info.References = strings.Fields(value) } } if err := scanner.Err(); err != nil { diff --git a/spindle/engines/microvm/upload_cache_nix_store.go b/spindle/engines/microvm/upload_cache_nix_store.go --- a/spindle/engines/microvm/upload_cache_nix_store.go +++ b/spindle/engines/microvm/upload_cache_nix_store.go @@ -278,10 +278,7 @@ http.Error(w, "invalid path", http.StatusBadRequest) return } - if _, err := writeFileAtomic(dst, ".tmp-narinfo", func(f *os.File) (int64, error) { - n, err := f.Write(body) - return int64(n), err - }); err != nil { + if _, err := writeNarinfoFile(dst, body); err != nil { b.logger.Warn("stage narinfo upload failed", "path", relPath, "error", err) http.Error(w, "internal error", http.StatusInternalServerError) return @@ -346,9 +343,108 @@ return filepath.Join(b.stagingDir, local), nil } +// makes the full reference graph of rootStorePath resolvable in the staging +// cache. `nix copy` computes the closure from the --from store, so every +// referenced narinfo must be present there or the walk fails with "path ... is +// not valid". newly-built deps are already staged by the guest, but deps that +// live only in a read cache were skipped during upload, so we backfill their +// narinfos here. only the narinfos (the reference graph) are needed: the +// destination supplies the NAR data via --substitute-on-destination. +func (b *NixStoreUploadBackend) ensureClosureStaged(ctx context.Context, rootStorePath string) error { + visited := map[string]bool{} + queue := []string{rootStorePath} + for len(queue) > 0 { + storePath := queue[0] + queue = queue[1:] + if visited[storePath] { + continue + } + visited[storePath] = true + + info, err := b.resolveStagedNarinfo(ctx, storePath) + if err != nil { + // the root must resolve (the guest just staged it); a dep we can't + // find anywhere is left for `nix copy` to surface with its own error. + if storePath == rootStorePath { + return fmt.Errorf("resolve narinfo for %s: %w", storePath, err) + } + b.logger.Warn("closure dep narinfo unresolved; leaving to nix copy", "storePath", storePath, "error", err) + continue + } + + for _, ref := range info.References { + refPath := storePrefix + ref + if refPath == storePath { + continue // self-reference + } + if !visited[refPath] { + queue = append(queue, refPath) + } + } + } + return nil +} + +// returns parsed narinfo for store path, backfilling from readUpstreams if not found +func (b *NixStoreUploadBackend) resolveStagedNarinfo(ctx context.Context, storePath string) (*narinfo, error) { + hash, _, err := parseStorePath(storePath) + if err != nil { + return nil, err + } + localPath := filepath.Join(b.stagingDir, hash+".narinfo") + info, err := readNarinfoFile(localPath) + if err == nil { + return info, nil + } + if !errors.Is(err, os.ErrNotExist) { + return nil, err + } + + // dep missing from staging because it was skipped during upload + // (lives on a read cache) so we backfill it from readUpstreams. + body, err := b.fetchUpstreamNarinfo(ctx, hash) + if err != nil { + return nil, err + } + written, err := writeNarinfoFile(localPath, body) + if err != nil { + return nil, err + } + b.logger.Debug("backfilled narinfo", "hash", hash, "bytes", written) + + return parseNarinfo(bytes.NewReader(body)) +} + +// fetches narinfo from readUpstreams +func (b *NixStoreUploadBackend) fetchUpstreamNarinfo(ctx context.Context, hash string) ([]byte, error) { + if len(b.readUpstreams) == 0 { + return nil, os.ErrNotExist + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://upstream/"+hash+".narinfo", nil) + if err != nil { + return nil, err + } + resp, err := newNarinfoExistenceTransport(b.readUpstreams, b.logger).RoundTrip(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode == http.StatusNotFound { + return nil, os.ErrNotExist + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("upstream narinfo %s: status %d", hash, resp.StatusCode) + } + return io.ReadAll(io.LimitReader(resp.Body, maxNarinfoSize+1)) +} + // todo(dawn): ideally we don't use `nix copy` here but instead have our own // `nix copy` impl so we don't need nix on host. but that's a far stretch goal :p func (b *NixStoreUploadBackend) importStorePath(ctx context.Context, storePath string) error { + if err := b.ensureClosureStaged(ctx, storePath); err != nil { + return fmt.Errorf("stage closure for %s: %w", storePath, err) + } + fromURL := url.URL{Scheme: "file", Path: b.stagingDir} args := []string{ "copy", @@ -375,6 +471,13 @@ return nil, err } defer f.Close() return parseNarinfo(f) +} + +func writeNarinfoFile(path string, body []byte) (int64, error) { + return writeFileAtomic(path, ".tmp-narinfo", func(f *os.File) (int64, error) { + n, err := f.Write(body) + return int64(n), err + }) } func writeFileAtomic(dst, tempPrefix string, write func(*os.File) (int64, error)) (written int64, err error) { diff --git a/spindle/engines/microvm/upload_cache_nix_store_test.go b/spindle/engines/microvm/upload_cache_nix_store_test.go --- a/spindle/engines/microvm/upload_cache_nix_store_test.go +++ b/spindle/engines/microvm/upload_cache_nix_store_test.go @@ -223,6 +223,62 @@ t.Fatalf("staged narinfo contents: got %q, want %q", string(data), narinfo) } } +func TestNixStoreBackendBackfillsClosureDepNarinfo(t *testing.T) { + const depHash = "abcdfghijklmnpqrsvwxyz0123456789" + depNarinfo := "StorePath: /nix/store/" + depHash + "-dep\nURL: nar/dep.nar.zst\nNarHash: sha256:dep\nNarSize: 1\n" + + var depRequests int + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/"+depHash+".narinfo" { + depRequests++ + _, _ = io.WriteString(w, depNarinfo) + return + } + w.WriteHeader(http.StatusNotFound) + })) + defer upstream.Close() + + upURL, err := url.Parse(upstream.URL) + if err != nil { + t.Fatal(err) + } + + runner := &fakeRunner{} + staging := t.TempDir() + b, err := newNixStoreUploadBackend("ssh-ng://cache-host", staging, []CacheUpstream{{url: upURL}}, slog.Default(), runner) + if err != nil { + t.Fatalf("newNixStoreUploadBackend: %v", err) + } + + mustUploadNar(t, b, "foo.nar.zst", "nar-body") + + // the top path references the dep, which is absent from staging. + narinfo := "StorePath: " + testStorePath + "\nURL: nar/foo.nar.zst\nNarHash: sha256:abc\nNarSize: 123\nReferences: " + depHash + "-dep\n" + rec := httptest.NewRecorder() + b.ServeHTTP(rec, httptest.NewRequest(http.MethodPut, "/"+testStoreHash+".narinfo", strings.NewReader(narinfo))) + if rec.Code != http.StatusOK { + t.Fatalf("PUT narinfo status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) + } + + // the dep narinfo must have been fetched from the upstream and staged... + if depRequests == 0 { + t.Fatalf("expected the dep narinfo to be fetched from the upstream") + } + staged, err := os.ReadFile(filepath.Join(staging, depHash+".narinfo")) + if err != nil { + t.Fatalf("dep narinfo not backfilled into staging: %v", err) + } + if string(staged) != depNarinfo { + t.Fatalf("backfilled dep narinfo contents: got %q, want %q", string(staged), depNarinfo) + } + + // ...and the import still copies just the requested top path. + calls := runner.Calls() + if len(calls) != 1 || calls[0][len(calls[0])-1] != testStorePath { + t.Fatalf("expected a single nix copy for %s, got %v", testStorePath, calls) + } +} + func TestNixStoreBackendRemovesNarinfoOnImportFailure(t *testing.T) { runner := &fakeRunner{nextErr: errors.New("nix copy failed")} b, staging := newTestNixStoreBackend(t, "ssh://cache-host", runner) diff --git a/spindle/engines/microvm/vm.go b/spindle/engines/microvm/vm.go --- a/spindle/engines/microvm/vm.go +++ b/spindle/engines/microvm/vm.go @@ -171,7 +171,10 @@ ctx = context.WithoutCancel(ctx) var err error - err = errors.Join(err, e.drainNixCache(ctx, state)) + // todo(dawn): expose this error to the user as a warning + if drainErr := e.drainNixCache(ctx, state); drainErr != nil { + e.l.Warn("cache drain failed during cleanup; continuing", "workflow", wid, "error", drainErr) + } err = errors.Join(err, e.shutdownVM(ctx, wid, state)) err = errors.Join(err, closeIO(&state.Agent)) err = errors.Join(err, closeIO(&state.ReadCache)) -- tangled.sh