From 679087ca63c87a0022d5faeb7afc633e31944f92 Mon Sep 17 00:00:00 2001 From: dawn Date: Fri, 17 Jul 2026 14:10:54 +0300 Subject: [PATCH] spindle/engines/microvm: rename nix cache -> substituters Signed-off-by: dawn --- cmd/spindle-microvm-run/main_linux.go | 10 +-- docs/DOCS.md | 6 +- spindle/engine/manifest.go | 22 +++-- spindle/engines/microvm/README.md | 2 +- spindle/engines/microvm/engine.go | 82 +++++++++---------- spindle/engines/microvm/engine_test.go | 23 ++++++ spindle/engines/microvm/models.go | 12 +-- spindle/engines/microvm/models_test.go | 10 +-- .../{upload_cache_narinfo.go => narinfo.go} | 0 ...os_toplevel_cache.go => nixos_toplevel.go} | 24 +++--- ...est-cache.sh => start-test-substituter.sh} | 0 ...ad_cache_proxy.go => substituter_proxy.go} | 44 +++++----- ...roxy_test.go => substituter_proxy_test.go} | 36 ++++---- ...che_http.go => substituter_upload_http.go} | 10 +-- ...ore.go => substituter_upload_nix_store.go} | 4 +- ...o => substituter_upload_nix_store_test.go} | 12 +-- ...e_proxy.go => substituter_upload_proxy.go} | 22 ++--- ...st.go => substituter_upload_proxy_test.go} | 4 +- .../engines/microvm/test-spindle-microvm.sh | 2 +- spindle/engines/microvm/vm.go | 34 ++++---- 20 files changed, 196 insertions(+), 163 deletions(-) rename spindle/engines/microvm/{upload_cache_narinfo.go => narinfo.go} (100%) rename spindle/engines/microvm/{nixos_toplevel_cache.go => nixos_toplevel.go} (74%) rename spindle/engines/microvm/{start-test-cache.sh => start-test-substituter.sh} (100%) rename spindle/engines/microvm/{read_cache_proxy.go => substituter_proxy.go} (87%) rename spindle/engines/microvm/{read_cache_proxy_test.go => substituter_proxy_test.go} (83%) rename spindle/engines/microvm/{upload_cache_http.go => substituter_upload_http.go} (85%) rename spindle/engines/microvm/{upload_cache_nix_store.go => substituter_upload_nix_store.go} (99%) rename spindle/engines/microvm/{upload_cache_nix_store_test.go => substituter_upload_nix_store_test.go} (97%) rename spindle/engines/microvm/{upload_cache_proxy.go => substituter_upload_proxy.go} (76%) rename spindle/engines/microvm/{upload_cache_proxy_test.go => substituter_upload_proxy_test.go} (96%) diff --git a/cmd/spindle-microvm-run/main_linux.go b/cmd/spindle-microvm-run/main_linux.go index c6988151..3b66fb03 100644 --- a/cmd/spindle-microvm-run/main_linux.go +++ b/cmd/spindle-microvm-run/main_linux.go @@ -190,25 +190,25 @@ func runMicroVMRunDev(ctx context.Context, cmd *cli.Command) error { } defer conn.Close() - upstreams, err := microvm.BuildCacheUpstreams(cmd.StringSlice("cache-read-url"), nil) + upstreams, err := microvm.BuildSubstituterUpstreams(cmd.StringSlice("cache-read-url"), nil) if err != nil { return fmt.Errorf("build cache upstreams: %w", err) } - var readCache *microvm.ReadCacheProxy + var readCache *microvm.SubstituterProxy if len(cmd.StringSlice("cache-read-url")) > 0 { var err error - readCache, err = microvm.StartReadCacheProxy(ctx, vm.CID(), upstreams, logger) + readCache, err = microvm.StartSubstituterProxy(ctx, vm.CID(), upstreams, logger) if err != nil { return fmt.Errorf("start read cache proxy: %w", err) } defer readCache.Close() } - var uploadCache *microvm.UploadCacheProxy + var uploadCache *microvm.SubstituterUploadProxy if cmd.String("cache-upload-url") != "" { var err error - uploadCache, err = microvm.StartUploadCacheProxy(ctx, vm.CID(), cmd.String("cache-upload-url"), upstreams, filepath.Join(vm.WorkDir(), "upload-cache"), logger) + uploadCache, err = microvm.StartSubstituterUploadProxy(ctx, vm.CID(), cmd.String("cache-upload-url"), upstreams, filepath.Join(vm.WorkDir(), "upload-cache"), logger) if err != nil { return fmt.Errorf("start upload cache proxy: %w", err) } diff --git a/docs/DOCS.md b/docs/DOCS.md index abccdbbe..9030c2c9 100644 --- a/docs/DOCS.md +++ b/docs/DOCS.md @@ -1163,15 +1163,15 @@ registry: myflake: github:me/x ``` -#### Caches +#### Substituters -The `caches` field is a map of Nix binary cache URL to its +The `substituters` field is a map of Nix binary cache URL to its trusted public key. These are fed into the spindle's read proxy, so the guest can substitute prebuilt paths from them instead of building everything from scratch. ```yaml -caches: +substituters: https://nix-community.cachix.org: "nix-community.cachix.org-1:mB9FSh9qf2dCimDSUo8Zy7bkq5CX+/rkCWyvRCYg3Fs=" ``` diff --git a/spindle/engine/manifest.go b/spindle/engine/manifest.go index 2cde8694..51d7bca5 100644 --- a/spindle/engine/manifest.go +++ b/spindle/engine/manifest.go @@ -78,7 +78,7 @@ func ignoredKeys(t reflect.Type) map[string]bool { // at the field that's actually mis-shaped. // // returns nil when nothing is structurally wrong. -func DescribeManifestError(raw string, schema any) error { +func DescribeManifestError(raw string, schema any, renamed ...map[string]string) error { var doc yaml.Node if err := yaml.Unmarshal([]byte(raw), &doc); err != nil { return nil @@ -86,7 +86,11 @@ func DescribeManifestError(raw string, schema any) error { if len(doc.Content) == 0 { return nil } - err := checkNode(doc.Content[0], reflect.TypeOf(schema), "", genericWorkflowKeys) + var renames map[string]string + if len(renamed) > 0 { + renames = renamed[0] + } + err := checkNode(doc.Content[0], reflect.TypeOf(schema), "", genericWorkflowKeys, renames) var me *manifestError if !errors.As(err, &me) { return err // nil @@ -100,7 +104,8 @@ func DescribeManifestError(raw string, schema any) error { // checkNode walks a yaml node against the type it's expected to decode into, // recursing through structs, maps and slices. allowExtra names keys that are // valid at this level despite not being in the struct (only the root uses it). -func checkNode(node *yaml.Node, t reflect.Type, path string, allowExtra map[string]bool) error { +// renamed maps removed root-level keys to their new names, for better errors. +func checkNode(node *yaml.Node, t reflect.Type, path string, allowExtra map[string]bool, renamed map[string]string) error { if node.Kind == yaml.AliasNode && node.Alias != nil { node = node.Alias } @@ -141,24 +146,29 @@ func checkNode(node *yaml.Node, t reflect.Type, path string, allowExtra map[stri if allowExtra[key.Value] { continue } + if newName, wasRenamed := renamed[key.Value]; wasRenamed && path == "" { + return &manifestError{line: key.Line, msg: fmt.Sprintf( + "field %s was renamed to %s (line %d)", + describePath(joinKey(path, key.Value)), newName, key.Line)} + } return &manifestError{line: key.Line, msg: fmt.Sprintf( "unknown field %s (line %d)", describePath(joinKey(path, key.Value)), key.Line)} } - if err := checkNode(val, ft, joinKey(path, key.Value), nil); err != nil { + if err := checkNode(val, ft, joinKey(path, key.Value), nil, nil); err != nil { return err } } case reflect.Map: for i := 0; i+1 < len(node.Content); i += 2 { key, val := node.Content[i], node.Content[i+1] - if err := checkNode(val, t.Elem(), joinKey(path, key.Value), nil); err != nil { + if err := checkNode(val, t.Elem(), joinKey(path, key.Value), nil, nil); err != nil { return err } } case reflect.Slice, reflect.Array: for idx, val := range node.Content { - if err := checkNode(val, t.Elem(), fmt.Sprintf("%s[%d]", path, idx), nil); err != nil { + if err := checkNode(val, t.Elem(), fmt.Sprintf("%s[%d]", path, idx), nil, nil); err != nil { return err } } diff --git a/spindle/engines/microvm/README.md b/spindle/engines/microvm/README.md index cea962b3..d10dbbab 100644 --- a/spindle/engines/microvm/README.md +++ b/spindle/engines/microvm/README.md @@ -14,7 +14,7 @@ firecracker) can slot in later. Currently two kinds of images are supported: - NixOS images: these allow configuration such as `dependencies`, `services`, - `virtualisation`, `registry`, `caches` in the workflow file itself. The guest + `virtualisation`, `registry`, `substituters` in the workflow file itself. The guest agent will build (or if it's cached, spindle will send the store path for realization) and activate it before any workflow steps are ran. - Non-NixOS: this is mainly just Alpine for now, but can be anything else. diff --git a/spindle/engines/microvm/engine.go b/spindle/engines/microvm/engine.go index 7bccb4b9..fb953a16 100644 --- a/spindle/engines/microvm/engine.go +++ b/spindle/engines/microvm/engine.go @@ -30,15 +30,15 @@ import ( ) const ( - guestWorkDir = "/workspace/repo" - guestBasePATH = "/run/current-system/sw/bin:/nix/var/nix/profiles/default/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" - guestDevShellEnvPath = "/run/spindle/devshell-env.sh" - activationStepAction = "activate-config" - agentAcceptTimeout = 2 * time.Minute - agentHandshakeTimeout = 30 * time.Second - cacheDrainTimeout = 5 * time.Minute - vmShutdownTimeout = 10 * time.Second - guestTimeoutGrace = 5 * time.Second + guestWorkDir = "/workspace/repo" + guestBasePATH = "/run/current-system/sw/bin:/nix/var/nix/profiles/default/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" + guestDevShellEnvPath = "/run/spindle/devshell-env.sh" + activationStepAction = "activate-config" + agentAcceptTimeout = 2 * time.Minute + agentHandshakeTimeout = 30 * time.Second + substituterDrainTimeout = 5 * time.Minute + vmShutdownTimeout = 10 * time.Second + guestTimeoutGrace = 5 * time.Second ) type cleanupFunc func(context.Context) error @@ -118,7 +118,7 @@ func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipelin swf := &models.Workflow{} var dwf manifestWorkflow - if err := engine.DescribeManifestError(twf.Raw, manifestWorkflow{}); err != nil { + if err := engine.DescribeManifestError(twf.Raw, manifestWorkflow{}, map[string]string{"caches": "substituters"}); err != nil { return nil, err } if err := yaml.Unmarshal([]byte(twf.Raw), &dwf); err != nil { @@ -183,20 +183,20 @@ func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipelin swf.Steps[insertAt] = activationStep } - cacheURLs, cacheKeys, err := workflowCaches(dwf.Caches) + substituterURLs, substituterKeys, err := workflowSubstituters(dwf.Substituters) if err != nil { return nil, err } swf.Data = &workflowState{ - ImageSpec: imageSpec, - ImageSpecPath: imageSpecPath, - Config: config, - ConfigKey: configKey, - Image: imageName, - CacheReadURLs: cacheURLs, - CacheTrustedPublicKeys: cacheKeys, - NixOSToplevelCache: newNixOSToplevelCacheStore(e.db), + ImageSpec: imageSpec, + ImageSpecPath: imageSpecPath, + Config: config, + ConfigKey: configKey, + Image: imageName, + SubstituterReadURLs: substituterURLs, + SubstituterTrustedPublicKeys: substituterKeys, + NixOSToplevels: newNixOSToplevelStore(e.db), } return swf, nil } @@ -257,21 +257,21 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m } }() - upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs) + upstreams, err := BuildSubstituterUpstreams(e.cfg.NixCache.ReadURLs, state.SubstituterReadURLs) if err != nil { return err } - readCache, err := StartReadCacheProxy(ctx, cid, upstreams, l) + substituter, err := StartSubstituterProxy(ctx, cid, upstreams, l) if err != nil { return err } - state.ReadCache = readCache - stagingDir := filepath.Join(workDir, "upload-cache") - uploadCache, err := StartUploadCacheProxy(ctx, cid, e.cfg.NixCache.UploadURL, upstreams, stagingDir, l) + state.Substituter = substituter + stagingDir := filepath.Join(workDir, "substituter-upload") + substituterUpload, err := StartSubstituterUploadProxy(ctx, cid, e.cfg.NixCache.UploadURL, upstreams, stagingDir, l) if err != nil { return err } - state.UploadCache = uploadCache + state.SubstituterUpload = substituterUpload dnsProxy, err := StartDNSProxy(ctx, cid, l) if err != nil { return err @@ -315,9 +315,9 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m defer cancelInit() if err := agentSession.Init(initCtx, &agentv1.Init{ JobId: wid.String(), - CacheTrustedPublicKeys: append(slices.Clone(e.cfg.NixCache.TrustedPublicKeys), state.CacheTrustedPublicKeys...), - CacheReadProxyPort: readCache.Port(), - CacheUploadProxyPort: uploadCache.Port(), + CacheTrustedPublicKeys: append(slices.Clone(e.cfg.NixCache.TrustedPublicKeys), state.SubstituterTrustedPublicKeys...), + CacheReadProxyPort: substituter.Port(), + CacheUploadProxyPort: substituterUpload.Port(), DnsProxyPort: dnsProxy.Port(), }); err != nil { _ = agentSession.Close() @@ -473,15 +473,15 @@ func (e *Engine) activateConfig(ctx context.Context, wid models.WorkflowId, stat var cachedToplevel string if configKey != "" { - if record, ok, err := state.NixOSToplevelCache.Lookup(configKey); err != nil { + if record, ok, err := state.NixOSToplevels.Lookup(configKey); err != nil { return err } else if ok { // todo(dawn): we should probably use gc roots to eliminate TOCTOU // the spindle will have to manage the gc roots, and for remote we have to // ssh in to the host and add / remove gc root. - // we need to have this check anyway since the only check http caches can + // we need to have this check anyway since the only check http substituters can // use is this one, since we cant manage gc roots there... - if e.anyCacheHasPath(ctx, state, record.Toplevel) { + if e.anySubstituterHasPath(ctx, state, record.Toplevel) { cachedToplevel = record.Toplevel fmt.Fprintf(out, "realizing cached NixOS config %s\n", cachedToplevel) } @@ -511,7 +511,7 @@ func (e *Engine) activateConfig(ctx context.Context, wid models.WorkflowId, stat return nil } if e.cfg.NixCache.UploadURL == "" { - e.l.Warn("not committing config cache metadata: no upload URL configured", "workflow", wid, "configKey", configKey, "toplevel", result.Toplevel) + e.l.Warn("not committing toplevel metadata: no upload URL configured", "workflow", wid, "configKey", configKey, "toplevel", result.Toplevel) return nil } @@ -519,20 +519,20 @@ func (e *Engine) activateConfig(ctx context.Context, wid models.WorkflowId, stat // 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) + e.l.Warn("substituter upload drain failed; skipping toplevel metadata commit", "workflow", wid, "configKey", configKey, "toplevel", result.Toplevel, "error", err) return nil } - if err := state.NixOSToplevelCache.Commit(configKey, result.Toplevel); err != nil { + if err := state.NixOSToplevels.Commit(configKey, result.Toplevel); err != nil { return err } - fmt.Fprintf(out, "committed config cache metadata %s -> %s\n", configKey, result.Toplevel) + fmt.Fprintf(out, "committed toplevel metadata %s -> %s\n", configKey, result.Toplevel) return nil } -func (e *Engine) anyCacheHasPath(ctx context.Context, state *workflowState, storePath string) bool { - upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs) +func (e *Engine) anySubstituterHasPath(ctx context.Context, state *workflowState, storePath string) bool { + upstreams, err := BuildSubstituterUpstreams(e.cfg.NixCache.ReadURLs, state.SubstituterReadURLs) if err != nil { - e.l.Warn("config cache check: build upstreams failed; treating as absent", "path", storePath, "error", err) + e.l.Warn("toplevel check: build upstreams failed; treating as absent", "path", storePath, "error", err) return false } if len(upstreams) == 0 { @@ -540,17 +540,17 @@ func (e *Engine) anyCacheHasPath(ctx context.Context, state *workflowState, stor } hash, _, err := parseStorePath(storePath) if err != nil { - e.l.Warn("config cache check: invalid toplevel path; treating as absent", "path", storePath, "error", err) + e.l.Warn("toplevel check: invalid toplevel path; treating as absent", "path", storePath, "error", err) return false } req, err := http.NewRequestWithContext(ctx, http.MethodHead, "http://upstream/"+hash+".narinfo", nil) if err != nil { - e.l.Warn("config cache check: build request failed; treating as absent", "path", storePath, "error", err) + e.l.Warn("toplevel check: build request failed; treating as absent", "path", storePath, "error", err) return false } resp, err := newNarinfoExistenceTransport(upstreams, e.l).RoundTrip(req) if err != nil { - e.l.Warn("config cache check: narinfo probe failed; treating as absent", "path", storePath, "error", err) + e.l.Warn("toplevel check: narinfo probe failed; treating as absent", "path", storePath, "error", err) return false } defer resp.Body.Close() diff --git a/spindle/engines/microvm/engine_test.go b/spindle/engines/microvm/engine_test.go index 13465cd5..18826efb 100644 --- a/spindle/engines/microvm/engine_test.go +++ b/spindle/engines/microvm/engine_test.go @@ -125,3 +125,26 @@ steps: t.Fatalf("first step should be the activation step, got %+v", wf.Steps[0]) } } + +func TestInitWorkflowCachesRenameHint(t *testing.T) { + dir := t.TempDir() + writeTestImageSpec(t, dir, "alpine", validImageSpec()) + + e := testEngine(t, dir) + _, err := e.InitWorkflow(tangled.Pipeline_Workflow{ + Raw: ` +image: alpine +caches: + https://cache.nixos.org: "" +steps: + - name: hello + command: hello +`, + }, tangled.Pipeline{}) + if err == nil { + t.Fatal("expected an error for the old caches key") + } + if !strings.Contains(err.Error(), "renamed to substituters") { + t.Fatalf("error should point at substituters, got: %v", err) + } +} diff --git a/spindle/engines/microvm/models.go b/spindle/engines/microvm/models.go index 02f3c096..6d8ee810 100644 --- a/spindle/engines/microvm/models.go +++ b/spindle/engines/microvm/models.go @@ -12,7 +12,7 @@ type manifestWorkflow struct { Dependencies []string `yaml:"dependencies"` Registry map[string]any `yaml:"registry"` Environment map[string]string `yaml:"environment"` - Caches map[string]string `yaml:"caches"` + Substituters map[string]string `yaml:"substituters"` Steps []struct { Name string `yaml:"name"` Command string `yaml:"command"` @@ -20,16 +20,16 @@ type manifestWorkflow struct { } `yaml:"steps"` } -// flattens the caches map into sorted substituter URLs and trusted public keys -func workflowCaches(caches map[string]string) (urls []string, keys []string, err error) { - for cacheURL, key := range caches { +// sorted so the guest env is deterministic +func workflowSubstituters(substituters map[string]string) (urls []string, keys []string, err error) { + for cacheURL, key := range substituters { urls = append(urls, cacheURL) if key != "" { keys = append(keys, key) } } - if _, err := parseCacheUpstreams(urls); err != nil { - return nil, nil, fmt.Errorf("caches: %w", err) + if _, err := parseSubstituterUpstreams(urls); err != nil { + return nil, nil, fmt.Errorf("substituters: %w", err) } slices.Sort(urls) slices.Sort(keys) diff --git a/spindle/engines/microvm/models_test.go b/spindle/engines/microvm/models_test.go index b103e4f5..b8f71c7c 100644 --- a/spindle/engines/microvm/models_test.go +++ b/spindle/engines/microvm/models_test.go @@ -5,8 +5,8 @@ import ( "testing" ) -func TestWorkflowCaches(t *testing.T) { - urls, keys, err := workflowCaches(map[string]string{ +func TestWorkflowSubstituters(t *testing.T) { + urls, keys, err := workflowSubstituters(map[string]string{ "https://hydra.nixos.org/": "hydra.nixos.org-1:CNHJZBh9K4tP3EKF6FkkgeVYsS3ohTl+oS0Qa8bezVs=", "https://cache.garnix.io/": "cache.garnix.io:CTFPyKSLcx5RMJKfLo5EEPUObbA78b0YQ2DTCJXqr9g=", "https://unsigned.example/": "", @@ -32,10 +32,10 @@ func TestWorkflowCaches(t *testing.T) { } } -func TestWorkflowCachesRejectsBadURLs(t *testing.T) { +func TestWorkflowSubstitutersRejectsBadURLs(t *testing.T) { for _, bad := range []string{"ftp://cache.example/", "not a url"} { - if _, _, err := workflowCaches(map[string]string{bad: ""}); err == nil { - t.Errorf("workflowCaches(%q): expected error, got nil", bad) + if _, _, err := workflowSubstituters(map[string]string{bad: ""}); err == nil { + t.Errorf("workflowSubstituters(%q): expected error, got nil", bad) } } } diff --git a/spindle/engines/microvm/upload_cache_narinfo.go b/spindle/engines/microvm/narinfo.go similarity index 100% rename from spindle/engines/microvm/upload_cache_narinfo.go rename to spindle/engines/microvm/narinfo.go diff --git a/spindle/engines/microvm/nixos_toplevel_cache.go b/spindle/engines/microvm/nixos_toplevel.go similarity index 74% rename from spindle/engines/microvm/nixos_toplevel_cache.go rename to spindle/engines/microvm/nixos_toplevel.go index 8333a65c..f3028bc4 100644 --- a/spindle/engines/microvm/nixos_toplevel_cache.go +++ b/spindle/engines/microvm/nixos_toplevel.go @@ -12,41 +12,41 @@ import ( "tangled.org/core/spindle/db" ) -const nixosToplevelCacheSchemaVersion = 1 +const nixosToplevelSchemaVersion = 1 -type nixosToplevelCacheRecord struct { +type nixosToplevelRecord struct { ConfigKey string `json:"config_key"` Toplevel string `json:"toplevel"` UpdatedAt time.Time `json:"updated_at"` } -type nixosToplevelCacheStore struct { +type nixosToplevelStore struct { db *db.DB } -func newNixOSToplevelCacheStore(d *db.DB) nixosToplevelCacheStore { - return nixosToplevelCacheStore{db: d} +func newNixOSToplevelStore(d *db.DB) nixosToplevelStore { + return nixosToplevelStore{db: d} } -func (s nixosToplevelCacheStore) Lookup(configKey string) (nixosToplevelCacheRecord, bool, error) { +func (s nixosToplevelStore) Lookup(configKey string) (nixosToplevelRecord, bool, error) { if s.db == nil { - return nixosToplevelCacheRecord{}, false, nil + return nixosToplevelRecord{}, false, nil } r, err := s.db.GetNixOSToplevelCacheRecord(configKey) if err != nil { if errors.Is(err, sql.ErrNoRows) { - return nixosToplevelCacheRecord{}, false, nil + return nixosToplevelRecord{}, false, nil } - return nixosToplevelCacheRecord{}, false, err + return nixosToplevelRecord{}, false, err } - return nixosToplevelCacheRecord{ + return nixosToplevelRecord{ ConfigKey: r.ConfigKey, Toplevel: r.Toplevel, UpdatedAt: r.UpdatedAt, }, true, nil } -func (s nixosToplevelCacheStore) Commit(configKey, toplevel string) error { +func (s nixosToplevelStore) Commit(configKey, toplevel string) error { if configKey == "" { return fmt.Errorf("config key is empty") } @@ -82,7 +82,7 @@ func buildConfigKey(imageSpec ImageSpec, cfg manifestConfig) (string, error) { BaseConfig string `json:"base_config"` UserConfig string `json:"user_config"` }{ - Schema: nixosToplevelCacheSchemaVersion, + Schema: nixosToplevelSchemaVersion, BaseConfig: baseHash, UserConfig: userConfigHash(cfg), } diff --git a/spindle/engines/microvm/start-test-cache.sh b/spindle/engines/microvm/start-test-substituter.sh similarity index 100% rename from spindle/engines/microvm/start-test-cache.sh rename to spindle/engines/microvm/start-test-substituter.sh diff --git a/spindle/engines/microvm/read_cache_proxy.go b/spindle/engines/microvm/substituter_proxy.go similarity index 87% rename from spindle/engines/microvm/read_cache_proxy.go rename to spindle/engines/microvm/substituter_proxy.go index 65d838c5..1f61b707 100644 --- a/spindle/engines/microvm/read_cache_proxy.go +++ b/spindle/engines/microvm/substituter_proxy.go @@ -21,18 +21,18 @@ import ( ) const ( - readCacheProxyPortMin = 20000 - readCacheProxyPortMax = 60000 + substituterProxyPortMin = 20000 + substituterProxyPortMax = 60000 ) -type ReadCacheProxy struct { +type SubstituterProxy struct { port uint32 ln *vsock.Listener server *http.Server } -func StartReadCacheProxy(ctx context.Context, cid uint32, upstreams []CacheUpstream, logger *slog.Logger) (*ReadCacheProxy, error) { +func StartSubstituterProxy(ctx context.Context, cid uint32, upstreams []SubstituterUpstream, logger *slog.Logger) (*SubstituterProxy, error) { if logger == nil { logger = slog.Default() } @@ -47,7 +47,7 @@ func StartReadCacheProxy(ctx context.Context, cid uint32, upstreams []CacheUpstr return nil, err } - proxy := &ReadCacheProxy{ + proxy := &SubstituterProxy{ port: port, ln: ln, } @@ -72,14 +72,14 @@ func StartReadCacheProxy(ctx context.Context, cid uint32, upstreams []CacheUpstr return proxy, nil } -func (p *ReadCacheProxy) Port() uint32 { +func (p *SubstituterProxy) Port() uint32 { if p == nil { return 0 } return p.port } -func (p *ReadCacheProxy) Close() error { +func (p *SubstituterProxy) Close() error { if p == nil { return nil } @@ -121,7 +121,7 @@ func (l *cidFilteredVsockListener) Accept() (net.Conn, error) { } } -func parseCacheUpstreams(raw []string) ([]*url.URL, error) { +func parseSubstituterUpstreams(raw []string) ([]*url.URL, error) { upstreams := make([]*url.URL, 0, len(raw)) seen := make(map[string]struct{}, len(raw)) for _, value := range raw { @@ -149,41 +149,41 @@ func parseCacheUpstreams(raw []string) ([]*url.URL, error) { return upstreams, nil } -type CacheUpstream struct { +type SubstituterUpstream struct { url *url.URL // guarded upstreams come from the workflow file // requests to them are refused for special-purpose address ranges guarded bool } -func BuildCacheUpstreams(rawTrusted, rawGuarded []string) ([]CacheUpstream, error) { - trusted, err := parseCacheUpstreams(rawTrusted) +func BuildSubstituterUpstreams(rawTrusted, rawGuarded []string) ([]SubstituterUpstream, error) { + trusted, err := parseSubstituterUpstreams(rawTrusted) if err != nil { return nil, err } - guarded, err := parseCacheUpstreams(rawGuarded) + guarded, err := parseSubstituterUpstreams(rawGuarded) if err != nil { return nil, err } - return mergeCacheUpstreams(trusted, guarded), nil + return mergeSubstituterUpstreams(trusted, guarded), nil } -func mergeCacheUpstreams(trusted, guarded []*url.URL) []CacheUpstream { - merged := make([]CacheUpstream, 0, len(trusted)+len(guarded)) +func mergeSubstituterUpstreams(trusted, guarded []*url.URL) []SubstituterUpstream { + merged := make([]SubstituterUpstream, 0, len(trusted)+len(guarded)) seen := make(map[string]struct{}, len(trusted)+len(guarded)) for _, u := range trusted { if _, ok := seen[u.String()]; ok { continue } seen[u.String()] = struct{}{} - merged = append(merged, CacheUpstream{url: u}) + merged = append(merged, SubstituterUpstream{url: u}) } for _, u := range guarded { if _, ok := seen[u.String()]; ok { continue } seen[u.String()] = struct{}{} - merged = append(merged, CacheUpstream{url: u, guarded: true}) + merged = append(merged, SubstituterUpstream{url: u, guarded: true}) } return merged } @@ -215,8 +215,8 @@ func randomVsockPort() (uint32, error) { if _, err := rand.Read(data[:]); err != nil { return 0, fmt.Errorf("allocate read vsock port: %w", err) } - span := uint32(readCacheProxyPortMax - readCacheProxyPortMin) - return readCacheProxyPortMin + binary.BigEndian.Uint32(data[:])%span, nil + span := uint32(substituterProxyPortMax - substituterProxyPortMin) + return substituterProxyPortMin + binary.BigEndian.Uint32(data[:])%span, nil } var proxyTransport = &http.Transport{ @@ -275,7 +275,7 @@ func refuseSpecialPurposeAddrs(network, address string, _ syscall.RawConn) error // merging) const nixCacheInfo = "StoreDir: /nix/store\nWantMassQuery: 1\nPriority: 40\n" -func cacheProxyHandler(upstreams []CacheUpstream, logger *slog.Logger) http.Handler { +func cacheProxyHandler(upstreams []SubstituterUpstream, logger *slog.Logger) http.Handler { proxy := &httputil.ReverseProxy{ // nothing to do here: the racing transport builds the full URL per // upstream, it just needs the guest's path/query left intact @@ -318,7 +318,7 @@ func mergeQuery(base, extra string) string { } type parallelRacingTransport struct { - upstreams []CacheUpstream + upstreams []SubstituterUpstream underlying http.RoundTripper guardedUnderlying http.RoundTripper logger *slog.Logger @@ -341,7 +341,7 @@ func (t *parallelRacingTransport) RoundTrip(req *http.Request) (*http.Response, ctx, cancel := context.WithCancel(req.Context()) cancels[i] = cancel - go func(idx int, target CacheUpstream, uCtx context.Context) { + go func(idx int, target SubstituterUpstream, uCtx context.Context) { defer wg.Done() raceReq := req.Clone(uCtx) diff --git a/spindle/engines/microvm/read_cache_proxy_test.go b/spindle/engines/microvm/substituter_proxy_test.go similarity index 83% rename from spindle/engines/microvm/read_cache_proxy_test.go rename to spindle/engines/microvm/substituter_proxy_test.go index 9ec84325..5170f479 100644 --- a/spindle/engines/microvm/read_cache_proxy_test.go +++ b/spindle/engines/microvm/substituter_proxy_test.go @@ -21,14 +21,14 @@ func TestCacheProxyFallsBackOnNotFound(t *testing.T) { })) defer second.Close() - upstreams, err := parseCacheUpstreams([]string{first.URL, second.URL}) + upstreams, err := parseSubstituterUpstreams([]string{first.URL, second.URL}) if err != nil { t.Fatal(err) } req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil) rec := httptest.NewRecorder() - cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) + cacheProxyHandler(mergeSubstituterUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) @@ -44,14 +44,14 @@ func TestCacheProxyServesNixCacheInfoItself(t *testing.T) { })) defer upstream.Close() - upstreams, err := parseCacheUpstreams([]string{upstream.URL}) + upstreams, err := parseSubstituterUpstreams([]string{upstream.URL}) if err != nil { t.Fatal(err) } req := httptest.NewRequest(http.MethodGet, "http://guest/nix-cache-info", nil) rec := httptest.NewRecorder() - cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) + cacheProxyHandler(mergeSubstituterUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) @@ -72,14 +72,14 @@ func TestCacheProxyErrorStatusDoesNotWinRace(t *testing.T) { })) defer healthy.Close() - upstreams, err := parseCacheUpstreams([]string{erroring.URL, healthy.URL}) + upstreams, err := parseSubstituterUpstreams([]string{erroring.URL, healthy.URL}) if err != nil { t.Fatal(err) } req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil) rec := httptest.NewRecorder() - cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) + cacheProxyHandler(mergeSubstituterUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) @@ -105,14 +105,14 @@ func TestCacheProxyJoinsSubpathQueryAndAuth(t *testing.T) { defer upstream.Close() upstreamURL := "http://dawn:woof@" + strings.TrimPrefix(upstream.URL, "http://") + "/sub/cache/?token=s3cret" - upstreams, err := parseCacheUpstreams([]string{upstreamURL}) + upstreams, err := parseSubstituterUpstreams([]string{upstreamURL}) if err != nil { t.Fatal(err) } req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil) rec := httptest.NewRecorder() - cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) + cacheProxyHandler(mergeSubstituterUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) @@ -122,12 +122,6 @@ func TestCacheProxyJoinsSubpathQueryAndAuth(t *testing.T) { } } -func TestCacheProxyGuardAllowsPublicIPv4(t *testing.T) { - if err := refuseSpecialPurposeAddrs("tcp", "104.26.13.82:443", nil); err != nil { - t.Fatalf("public IPv4 address was blocked: %v", err) - } -} - func TestCacheProxyGuardedUpstreamCannotReachBlockedRanges(t *testing.T) { // httptest listens on 127.0.0.1, which is in the blocked ranges; reaching // it would mean a workflow-defined cache can hit the host's loopback @@ -136,14 +130,14 @@ func TestCacheProxyGuardedUpstreamCannotReachBlockedRanges(t *testing.T) { })) defer upstream.Close() - upstreams, err := parseCacheUpstreams([]string{upstream.URL}) + upstreams, err := parseSubstituterUpstreams([]string{upstream.URL}) if err != nil { t.Fatal(err) } req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil) rec := httptest.NewRecorder() - cacheProxyHandler(mergeCacheUpstreams(nil, upstreams), slog.Default()).ServeHTTP(rec, req) + cacheProxyHandler(mergeSubstituterUpstreams(nil, upstreams), slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusBadGateway { t.Fatalf("status: got %d, want 502; body=%q", rec.Code, rec.Body.String()) @@ -161,7 +155,7 @@ func TestCacheProxyRewritesHostHeader(t *testing.T) { defer upstream.Close() upstreamHost = strings.TrimPrefix(upstream.URL, "http://") - upstreams, err := parseCacheUpstreams([]string{upstream.URL}) + upstreams, err := parseSubstituterUpstreams([]string{upstream.URL}) if err != nil { t.Fatal(err) } @@ -169,9 +163,15 @@ func TestCacheProxyRewritesHostHeader(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "http://127.0.0.1:10500/abc.narinfo", nil) req.Host = "127.0.0.1:10500" rec := httptest.NewRecorder() - cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) + cacheProxyHandler(mergeSubstituterUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) } } + +func TestCacheProxyGuardAllowsPublicIPv4(t *testing.T) { + if err := refuseSpecialPurposeAddrs("tcp", "104.26.13.82:443", nil); err != nil { + t.Fatalf("public IPv4 address was blocked: %v", err) + } +} diff --git a/spindle/engines/microvm/upload_cache_http.go b/spindle/engines/microvm/substituter_upload_http.go similarity index 85% rename from spindle/engines/microvm/upload_cache_http.go rename to spindle/engines/microvm/substituter_upload_http.go index 0310ef09..5b4fc084 100644 --- a/spindle/engines/microvm/upload_cache_http.go +++ b/spindle/engines/microvm/substituter_upload_http.go @@ -17,7 +17,7 @@ type httpUploadBackend struct { handler http.Handler } -func newHTTPUploadProxyBackend(target *url.URL, readUpstreams []CacheUpstream, logger *slog.Logger) *httpUploadBackend { +func newHTTPUploadProxyBackend(target *url.URL, readUpstreams []SubstituterUpstream, logger *slog.Logger) *httpUploadBackend { return &httpUploadBackend{handler: uploadProxyHandler(target, readUpstreams, logger)} } @@ -27,7 +27,7 @@ func (b *httpUploadBackend) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (b *httpUploadBackend) Close() error { return nil } -func uploadProxyHandler(target *url.URL, readUpstreams []CacheUpstream, logger *slog.Logger) http.Handler { +func uploadProxyHandler(target *url.URL, readUpstreams []SubstituterUpstream, logger *slog.Logger) http.Handler { rp := httputil.NewSingleHostReverseProxy(target) rp.ErrorLog = slog.NewLogLogger(logger.Handler(), slog.LevelError) @@ -46,10 +46,10 @@ func uploadProxyHandler(target *url.URL, readUpstreams []CacheUpstream, logger * // before uploading, nix copy asks the destination whether it already has each // path by GET/HEAD-ing .narinfo and skips the ones it does. we answer - // that check across the upload target *and* the read caches: if any of them + // that check across the upload target *and* the read substituters: if any of them // already serves the path there is no point uploading it (the guest would // just substitute it from there anyway). - narinfoUpstreams := append([]CacheUpstream{{url: target}}, readUpstreams...) + narinfoUpstreams := append([]SubstituterUpstream{{url: target}}, readUpstreams...) exists := newNarinfoExistenceTransport(narinfoUpstreams, logger) return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -61,7 +61,7 @@ func uploadProxyHandler(target *url.URL, readUpstreams []CacheUpstream, logger * }) } -func newNarinfoExistenceTransport(upstreams []CacheUpstream, logger *slog.Logger) http.RoundTripper { +func newNarinfoExistenceTransport(upstreams []SubstituterUpstream, logger *slog.Logger) http.RoundTripper { return ¶llelRacingTransport{ upstreams: upstreams, underlying: proxyTransport, diff --git a/spindle/engines/microvm/upload_cache_nix_store.go b/spindle/engines/microvm/substituter_upload_nix_store.go similarity index 99% rename from spindle/engines/microvm/upload_cache_nix_store.go rename to spindle/engines/microvm/substituter_upload_nix_store.go index 70b9adde..7db6e505 100644 --- a/spindle/engines/microvm/upload_cache_nix_store.go +++ b/spindle/engines/microvm/substituter_upload_nix_store.go @@ -40,13 +40,13 @@ const ( type NixStoreUploadBackend struct { stagingDir string targetStore string - readUpstreams []CacheUpstream + readUpstreams []SubstituterUpstream logger *slog.Logger runner CommandRunner maxNarUploadSize int64 } -func newNixStoreUploadBackend(targetStore, stagingDir string, readUpstreams []CacheUpstream, logger *slog.Logger, runner CommandRunner) (*NixStoreUploadBackend, error) { +func newNixStoreUploadBackend(targetStore, stagingDir string, readUpstreams []SubstituterUpstream, logger *slog.Logger, runner CommandRunner) (*NixStoreUploadBackend, error) { absStaging, err := filepath.Abs(stagingDir) if err != nil { return nil, fmt.Errorf("resolve staging dir %q: %w", stagingDir, err) diff --git a/spindle/engines/microvm/upload_cache_nix_store_test.go b/spindle/engines/microvm/substituter_upload_nix_store_test.go similarity index 97% rename from spindle/engines/microvm/upload_cache_nix_store_test.go rename to spindle/engines/microvm/substituter_upload_nix_store_test.go index 413fb66d..ae0cb980 100644 --- a/spindle/engines/microvm/upload_cache_nix_store_test.go +++ b/spindle/engines/microvm/substituter_upload_nix_store_test.go @@ -23,7 +23,7 @@ const ( testStorePath = "/nix/store/" + testStoreHash + "-abc-output" ) -func TestUploadCacheBackendSchemeDispatch(t *testing.T) { +func TestSubstituterUploadBackendSchemeDispatch(t *testing.T) { staging := t.TempDir() logger := slog.Default() @@ -44,7 +44,7 @@ func TestUploadCacheBackendSchemeDispatch(t *testing.T) { for _, tc := range cases { t.Run(tc.uploadURL, func(t *testing.T) { - backend, err := newUploadCacheBackend(tc.uploadURL, nil, staging, logger) + backend, err := newSubstituterUploadBackend(tc.uploadURL, nil, staging, logger) if tc.wantErr { if err == nil { t.Fatalf("expected error for %q", tc.uploadURL) @@ -62,8 +62,8 @@ func TestUploadCacheBackendSchemeDispatch(t *testing.T) { } } -func TestUploadCacheBackendEmptyURL(t *testing.T) { - backend, err := newUploadCacheBackend("", nil, t.TempDir(), slog.Default()) +func TestSubstituterUploadBackendEmptyURL(t *testing.T) { + backend, err := newSubstituterUploadBackend("", nil, t.TempDir(), slog.Default()) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -245,7 +245,7 @@ func TestNixStoreBackendBackfillsClosureDepNarinfo(t *testing.T) { runner := &fakeRunner{} staging := t.TempDir() - b, err := newNixStoreUploadBackend("ssh-ng://cache-host", staging, []CacheUpstream{{url: upURL}}, slog.Default(), runner) + b, err := newNixStoreUploadBackend("ssh-ng://cache-host", staging, []SubstituterUpstream{{url: upURL}}, slog.Default(), runner) if err != nil { t.Fatalf("newNixStoreUploadBackend: %v", err) } @@ -313,7 +313,7 @@ func TestNixStoreBackendNarinfoReadUpstream(t *testing.T) { } staging := t.TempDir() - b, err := newNixStoreUploadBackend("ssh://cache-host", staging, []CacheUpstream{{url: upURL}}, slog.Default(), nil) + b, err := newNixStoreUploadBackend("ssh://cache-host", staging, []SubstituterUpstream{{url: upURL}}, slog.Default(), nil) if err != nil { t.Fatalf("newNixStoreUploadBackend: %v", err) } diff --git a/spindle/engines/microvm/upload_cache_proxy.go b/spindle/engines/microvm/substituter_upload_proxy.go similarity index 76% rename from spindle/engines/microvm/upload_cache_proxy.go rename to spindle/engines/microvm/substituter_upload_proxy.go index 9b9385d2..df2beab0 100644 --- a/spindle/engines/microvm/upload_cache_proxy.go +++ b/spindle/engines/microvm/substituter_upload_proxy.go @@ -14,20 +14,20 @@ import ( "github.com/mdlayher/vsock" ) -type UploadCacheBackend interface { +type SubstituterUploadBackend interface { http.Handler Close() error } -type UploadCacheProxy struct { +type SubstituterUploadProxy struct { port uint32 ln *vsock.Listener server *http.Server - backend UploadCacheBackend + backend SubstituterUploadBackend } -func StartUploadCacheProxy(ctx context.Context, cid uint32, uploadURL string, readUpstreams []CacheUpstream, stagingDir string, logger *slog.Logger) (*UploadCacheProxy, error) { +func StartSubstituterUploadProxy(ctx context.Context, cid uint32, uploadURL string, readUpstreams []SubstituterUpstream, stagingDir string, logger *slog.Logger) (*SubstituterUploadProxy, error) { if strings.TrimSpace(uploadURL) == "" { return nil, nil } @@ -37,7 +37,7 @@ func StartUploadCacheProxy(ctx context.Context, cid uint32, uploadURL string, re } logger = logger.With("where", "upload_cache_proxy", "cid", cid, "uploadURL", uploadURL) - backend, err := newUploadCacheBackend(uploadURL, readUpstreams, stagingDir, logger) + backend, err := newSubstituterUploadBackend(uploadURL, readUpstreams, stagingDir, logger) if err != nil { return nil, err } @@ -47,7 +47,7 @@ func StartUploadCacheProxy(ctx context.Context, cid uint32, uploadURL string, re return nil, fmt.Errorf("listen for cache upload proxy: %w", err) } - proxy := &UploadCacheProxy{ + proxy := &SubstituterUploadProxy{ port: port, ln: ln, backend: backend, @@ -65,15 +65,15 @@ func StartUploadCacheProxy(ctx context.Context, cid uint32, uploadURL string, re } go func() { if err := proxy.server.Serve(filtered); err != nil && !errors.Is(err, http.ErrServerClosed) && !errors.Is(err, net.ErrClosed) { - logger.Warn("upload cache proxy stopped", "port", port, "error", err) + logger.Warn("substituter upload proxy stopped", "port", port, "error", err) } }() - logger.Info("started upload cache proxy", "port", port, "target", uploadURL, "readUpstreams", len(readUpstreams)) + logger.Info("started substituter upload proxy", "port", port, "target", uploadURL, "readUpstreams", len(readUpstreams)) return proxy, nil } -func newUploadCacheBackend(uploadURL string, readUpstreams []CacheUpstream, stagingDir string, logger *slog.Logger) (UploadCacheBackend, error) { +func newSubstituterUploadBackend(uploadURL string, readUpstreams []SubstituterUpstream, stagingDir string, logger *slog.Logger) (SubstituterUploadBackend, error) { if strings.TrimSpace(uploadURL) == "" { return nil, nil } @@ -106,14 +106,14 @@ func newUploadCacheBackend(uploadURL string, readUpstreams []CacheUpstream, stag } } -func (p *UploadCacheProxy) Port() uint32 { +func (p *SubstituterUploadProxy) Port() uint32 { if p == nil { return 0 } return p.port } -func (p *UploadCacheProxy) Close() error { +func (p *SubstituterUploadProxy) Close() error { if p == nil { return nil } diff --git a/spindle/engines/microvm/upload_cache_proxy_test.go b/spindle/engines/microvm/substituter_upload_proxy_test.go similarity index 96% rename from spindle/engines/microvm/upload_cache_proxy_test.go rename to spindle/engines/microvm/substituter_upload_proxy_test.go index ef4e6405..4f3d29bd 100644 --- a/spindle/engines/microvm/upload_cache_proxy_test.go +++ b/spindle/engines/microvm/substituter_upload_proxy_test.go @@ -69,7 +69,7 @@ func TestUploadProxySkipsNarinfoAvailableUpstream(t *testing.T) { handler := uploadProxyHandler( mustParseURL(t, target.URL), - []CacheUpstream{{url: mustParseURL(t, upstream.URL)}}, + []SubstituterUpstream{{url: mustParseURL(t, upstream.URL)}}, slog.Default(), ) @@ -97,7 +97,7 @@ func TestUploadProxyUploadsNarinfoNobodyHas(t *testing.T) { handler := uploadProxyHandler( mustParseURL(t, target.URL), - []CacheUpstream{{url: mustParseURL(t, upstream.URL)}}, + []SubstituterUpstream{{url: mustParseURL(t, upstream.URL)}}, slog.Default(), ) diff --git a/spindle/engines/microvm/test-spindle-microvm.sh b/spindle/engines/microvm/test-spindle-microvm.sh index 308e5a01..dbcf028c 100755 --- a/spindle/engines/microvm/test-spindle-microvm.sh +++ b/spindle/engines/microvm/test-spindle-microvm.sh @@ -271,7 +271,7 @@ trap 'exit 130' INT trap 'exit 143' TERM CACHE_PORT=$(pick_free_port) -./spindle/engines/microvm/start-test-cache.sh "$TEMP_DIR" "$CACHE_PORT" +./spindle/engines/microvm/start-test-substituter.sh "$TEMP_DIR" "$CACHE_PORT" source "$TEMP_DIR/env.sh" run_vm() { diff --git a/spindle/engines/microvm/vm.go b/spindle/engines/microvm/vm.go index 1fc9e031..3eb26a89 100644 --- a/spindle/engines/microvm/vm.go +++ b/spindle/engines/microvm/vm.go @@ -148,20 +148,20 @@ type VMConfig struct { } type workflowState struct { - ImageSpec ImageSpec - ImageSpecPath string - Config manifestConfig - ConfigKey string - Image string - CacheReadURLs []string - CacheTrustedPublicKeys []string - VM VMHandle - Agent *AgentSession - ReadCache *ReadCacheProxy - UploadCache *UploadCacheProxy - DNSProxy *DNSProxy - WorkDir string - NixOSToplevelCache nixosToplevelCacheStore + ImageSpec ImageSpec + ImageSpecPath string + Config manifestConfig + ConfigKey string + Image string + SubstituterReadURLs []string + SubstituterTrustedPublicKeys []string + VM VMHandle + Agent *AgentSession + Substituter *SubstituterProxy + SubstituterUpload *SubstituterUploadProxy + DNSProxy *DNSProxy + WorkDir string + NixOSToplevels nixosToplevelStore } func (e *Engine) cleanupState(ctx context.Context, wid models.WorkflowId, state *workflowState) error { @@ -178,8 +178,8 @@ func (e *Engine) cleanupState(ctx context.Context, wid models.WorkflowId, state } err = errors.Join(err, e.shutdownVM(ctx, wid, state)) err = errors.Join(err, closeIO(&state.Agent)) - err = errors.Join(err, closeIO(&state.ReadCache)) - err = errors.Join(err, closeIO(&state.UploadCache)) + err = errors.Join(err, closeIO(&state.Substituter)) + err = errors.Join(err, closeIO(&state.SubstituterUpload)) err = errors.Join(err, closeIO(&state.DNSProxy)) err = errors.Join(err, removeWorkDir(state)) return err @@ -190,7 +190,7 @@ func (e *Engine) drainNixCache(ctx context.Context, state *workflowState) error return nil } - drainCtx, cancel := context.WithTimeout(ctx, cacheDrainTimeout) + drainCtx, cancel := context.WithTimeout(ctx, substituterDrainTimeout) defer cancel() if state.Agent != nil { -- 2.51.2