alice.pds.demo.boltless.dev
This repository has no description
spindle/engines/microvm: rename nix cache -> substituters
Signed-off-by: dawn <dawn@tangled.org>
author
dawn
date
(Jul 25, 2026, 1:27 PM +0300)
commit
679087ca
679087ca63c87a0022d5faeb7afc633e31944f92
parent
3341c2e9
3341c2e992248e68511a3e499e42d679a4a0d4b2
change-id
rxlsswpv
rxlsswpvlqmzplvtunwymootvqvnorpl
+196
-163
+5
-5
cmd/spindle-microvm-run/main_linux.go
+5
-5
cmd/spindle-microvm-run/main_linux.go
···
190
190
}
191
191
defer conn.Close()
192
192
193
-
upstreams, err := microvm.BuildCacheUpstreams(cmd.StringSlice("cache-read-url"), nil)
193
+
upstreams, err := microvm.BuildSubstituterUpstreams(cmd.StringSlice("cache-read-url"), nil)
194
194
if err != nil {
195
195
return fmt.Errorf("build cache upstreams: %w", err)
196
196
}
197
197
198
-
var readCache *microvm.ReadCacheProxy
198
+
var readCache *microvm.SubstituterProxy
199
199
if len(cmd.StringSlice("cache-read-url")) > 0 {
200
200
var err error
201
-
readCache, err = microvm.StartReadCacheProxy(ctx, vm.CID(), upstreams, logger)
201
+
readCache, err = microvm.StartSubstituterProxy(ctx, vm.CID(), upstreams, logger)
202
202
if err != nil {
203
203
return fmt.Errorf("start read cache proxy: %w", err)
204
204
}
205
205
defer readCache.Close()
206
206
}
207
207
208
-
var uploadCache *microvm.UploadCacheProxy
208
+
var uploadCache *microvm.SubstituterUploadProxy
209
209
if cmd.String("cache-upload-url") != "" {
210
210
var err error
211
-
uploadCache, err = microvm.StartUploadCacheProxy(ctx, vm.CID(), cmd.String("cache-upload-url"), upstreams, filepath.Join(vm.WorkDir(), "upload-cache"), logger)
211
+
uploadCache, err = microvm.StartSubstituterUploadProxy(ctx, vm.CID(), cmd.String("cache-upload-url"), upstreams, filepath.Join(vm.WorkDir(), "upload-cache"), logger)
212
212
if err != nil {
213
213
return fmt.Errorf("start upload cache proxy: %w", err)
214
214
}
+3
-3
docs/DOCS.md
+3
-3
docs/DOCS.md
···
1163
1163
myflake: github:me/x
1164
1164
```
1165
1165
1166
-
#### Caches
1166
+
#### Substituters
1167
1167
1168
-
The `caches` field is a map of Nix binary cache URL to its
1168
+
The `substituters` field is a map of Nix binary cache URL to its
1169
1169
trusted public key. These are fed into the spindle's read
1170
1170
proxy, so the guest can substitute prebuilt paths from them
1171
1171
instead of building everything from scratch.
1172
1172
1173
1173
```yaml
1174
-
caches:
1174
+
substituters:
1175
1175
https://nix-community.cachix.org: "nix-community.cachix.org-1:mB9FSh9qf2dCimDSUo8Zy7bkq5CX+/rkCWyvRCYg3Fs="
1176
1176
```
1177
1177
+16
-6
spindle/engine/manifest.go
+16
-6
spindle/engine/manifest.go
···
78
78
// at the field that's actually mis-shaped.
79
79
//
80
80
// returns nil when nothing is structurally wrong.
81
-
func DescribeManifestError(raw string, schema any) error {
81
+
func DescribeManifestError(raw string, schema any, renamed ...map[string]string) error {
82
82
var doc yaml.Node
83
83
if err := yaml.Unmarshal([]byte(raw), &doc); err != nil {
84
84
return nil
···
86
86
if len(doc.Content) == 0 {
87
87
return nil
88
88
}
89
-
err := checkNode(doc.Content[0], reflect.TypeOf(schema), "", genericWorkflowKeys)
89
+
var renames map[string]string
90
+
if len(renamed) > 0 {
91
+
renames = renamed[0]
92
+
}
93
+
err := checkNode(doc.Content[0], reflect.TypeOf(schema), "", genericWorkflowKeys, renames)
90
94
var me *manifestError
91
95
if !errors.As(err, &me) {
92
96
return err // nil
···
100
104
// checkNode walks a yaml node against the type it's expected to decode into,
101
105
// recursing through structs, maps and slices. allowExtra names keys that are
102
106
// valid at this level despite not being in the struct (only the root uses it).
103
-
func checkNode(node *yaml.Node, t reflect.Type, path string, allowExtra map[string]bool) error {
107
+
// renamed maps removed root-level keys to their new names, for better errors.
108
+
func checkNode(node *yaml.Node, t reflect.Type, path string, allowExtra map[string]bool, renamed map[string]string) error {
104
109
if node.Kind == yaml.AliasNode && node.Alias != nil {
105
110
node = node.Alias
106
111
}
···
141
146
if allowExtra[key.Value] {
142
147
continue
143
148
}
149
+
if newName, wasRenamed := renamed[key.Value]; wasRenamed && path == "" {
150
+
return &manifestError{line: key.Line, msg: fmt.Sprintf(
151
+
"field %s was renamed to %s (line %d)",
152
+
describePath(joinKey(path, key.Value)), newName, key.Line)}
153
+
}
144
154
return &manifestError{line: key.Line, msg: fmt.Sprintf(
145
155
"unknown field %s (line %d)",
146
156
describePath(joinKey(path, key.Value)), key.Line)}
147
157
}
148
-
if err := checkNode(val, ft, joinKey(path, key.Value), nil); err != nil {
158
+
if err := checkNode(val, ft, joinKey(path, key.Value), nil, nil); err != nil {
149
159
return err
150
160
}
151
161
}
152
162
case reflect.Map:
153
163
for i := 0; i+1 < len(node.Content); i += 2 {
154
164
key, val := node.Content[i], node.Content[i+1]
155
-
if err := checkNode(val, t.Elem(), joinKey(path, key.Value), nil); err != nil {
165
+
if err := checkNode(val, t.Elem(), joinKey(path, key.Value), nil, nil); err != nil {
156
166
return err
157
167
}
158
168
}
159
169
case reflect.Slice, reflect.Array:
160
170
for idx, val := range node.Content {
161
-
if err := checkNode(val, t.Elem(), fmt.Sprintf("%s[%d]", path, idx), nil); err != nil {
171
+
if err := checkNode(val, t.Elem(), fmt.Sprintf("%s[%d]", path, idx), nil, nil); err != nil {
162
172
return err
163
173
}
164
174
}
+1
-1
spindle/engines/microvm/README.md
+1
-1
spindle/engines/microvm/README.md
···
14
14
Currently two kinds of images are supported:
15
15
16
16
- NixOS images: these allow configuration such as `dependencies`, `services`,
17
-
`virtualisation`, `registry`, `caches` in the workflow file itself. The guest
17
+
`virtualisation`, `registry`, `substituters` in the workflow file itself. The guest
18
18
agent will build (or if it's cached, spindle will send the store path for
19
19
realization) and activate it before any workflow steps are ran.
20
20
- Non-NixOS: this is mainly just Alpine for now, but can be anything else.
+41
-41
spindle/engines/microvm/engine.go
+41
-41
spindle/engines/microvm/engine.go
···
30
30
)
31
31
32
32
const (
33
-
guestWorkDir = "/workspace/repo"
34
-
guestBasePATH = "/run/current-system/sw/bin:/nix/var/nix/profiles/default/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
35
-
guestDevShellEnvPath = "/run/spindle/devshell-env.sh"
36
-
activationStepAction = "activate-config"
37
-
agentAcceptTimeout = 2 * time.Minute
38
-
agentHandshakeTimeout = 30 * time.Second
39
-
cacheDrainTimeout = 5 * time.Minute
40
-
vmShutdownTimeout = 10 * time.Second
41
-
guestTimeoutGrace = 5 * time.Second
33
+
guestWorkDir = "/workspace/repo"
34
+
guestBasePATH = "/run/current-system/sw/bin:/nix/var/nix/profiles/default/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
35
+
guestDevShellEnvPath = "/run/spindle/devshell-env.sh"
36
+
activationStepAction = "activate-config"
37
+
agentAcceptTimeout = 2 * time.Minute
38
+
agentHandshakeTimeout = 30 * time.Second
39
+
substituterDrainTimeout = 5 * time.Minute
40
+
vmShutdownTimeout = 10 * time.Second
41
+
guestTimeoutGrace = 5 * time.Second
42
42
)
43
43
44
44
type cleanupFunc func(context.Context) error
···
118
118
swf := &models.Workflow{}
119
119
var dwf manifestWorkflow
120
120
121
-
if err := engine.DescribeManifestError(twf.Raw, manifestWorkflow{}); err != nil {
121
+
if err := engine.DescribeManifestError(twf.Raw, manifestWorkflow{}, map[string]string{"caches": "substituters"}); err != nil {
122
122
return nil, err
123
123
}
124
124
if err := yaml.Unmarshal([]byte(twf.Raw), &dwf); err != nil {
···
183
183
swf.Steps[insertAt] = activationStep
184
184
}
185
185
186
-
cacheURLs, cacheKeys, err := workflowCaches(dwf.Caches)
186
+
substituterURLs, substituterKeys, err := workflowSubstituters(dwf.Substituters)
187
187
if err != nil {
188
188
return nil, err
189
189
}
190
190
191
191
swf.Data = &workflowState{
192
-
ImageSpec: imageSpec,
193
-
ImageSpecPath: imageSpecPath,
194
-
Config: config,
195
-
ConfigKey: configKey,
196
-
Image: imageName,
197
-
CacheReadURLs: cacheURLs,
198
-
CacheTrustedPublicKeys: cacheKeys,
199
-
NixOSToplevelCache: newNixOSToplevelCacheStore(e.db),
192
+
ImageSpec: imageSpec,
193
+
ImageSpecPath: imageSpecPath,
194
+
Config: config,
195
+
ConfigKey: configKey,
196
+
Image: imageName,
197
+
SubstituterReadURLs: substituterURLs,
198
+
SubstituterTrustedPublicKeys: substituterKeys,
199
+
NixOSToplevels: newNixOSToplevelStore(e.db),
200
200
}
201
201
return swf, nil
202
202
}
···
257
257
}
258
258
}()
259
259
260
-
upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs)
260
+
upstreams, err := BuildSubstituterUpstreams(e.cfg.NixCache.ReadURLs, state.SubstituterReadURLs)
261
261
if err != nil {
262
262
return err
263
263
}
264
-
readCache, err := StartReadCacheProxy(ctx, cid, upstreams, l)
264
+
substituter, err := StartSubstituterProxy(ctx, cid, upstreams, l)
265
265
if err != nil {
266
266
return err
267
267
}
268
-
state.ReadCache = readCache
269
-
stagingDir := filepath.Join(workDir, "upload-cache")
270
-
uploadCache, err := StartUploadCacheProxy(ctx, cid, e.cfg.NixCache.UploadURL, upstreams, stagingDir, l)
268
+
state.Substituter = substituter
269
+
stagingDir := filepath.Join(workDir, "substituter-upload")
270
+
substituterUpload, err := StartSubstituterUploadProxy(ctx, cid, e.cfg.NixCache.UploadURL, upstreams, stagingDir, l)
271
271
if err != nil {
272
272
return err
273
273
}
274
-
state.UploadCache = uploadCache
274
+
state.SubstituterUpload = substituterUpload
275
275
dnsProxy, err := StartDNSProxy(ctx, cid, l)
276
276
if err != nil {
277
277
return err
···
315
315
defer cancelInit()
316
316
if err := agentSession.Init(initCtx, &agentv1.Init{
317
317
JobId: wid.String(),
318
-
CacheTrustedPublicKeys: append(slices.Clone(e.cfg.NixCache.TrustedPublicKeys), state.CacheTrustedPublicKeys...),
319
-
CacheReadProxyPort: readCache.Port(),
320
-
CacheUploadProxyPort: uploadCache.Port(),
318
+
CacheTrustedPublicKeys: append(slices.Clone(e.cfg.NixCache.TrustedPublicKeys), state.SubstituterTrustedPublicKeys...),
319
+
CacheReadProxyPort: substituter.Port(),
320
+
CacheUploadProxyPort: substituterUpload.Port(),
321
321
DnsProxyPort: dnsProxy.Port(),
322
322
}); err != nil {
323
323
_ = agentSession.Close()
···
473
473
474
474
var cachedToplevel string
475
475
if configKey != "" {
476
-
if record, ok, err := state.NixOSToplevelCache.Lookup(configKey); err != nil {
476
+
if record, ok, err := state.NixOSToplevels.Lookup(configKey); err != nil {
477
477
return err
478
478
} else if ok {
479
479
// todo(dawn): we should probably use gc roots to eliminate TOCTOU
480
480
// the spindle will have to manage the gc roots, and for remote we have to
481
481
// ssh in to the host and add / remove gc root.
482
-
// we need to have this check anyway since the only check http caches can
482
+
// we need to have this check anyway since the only check http substituters can
483
483
// use is this one, since we cant manage gc roots there...
484
-
if e.anyCacheHasPath(ctx, state, record.Toplevel) {
484
+
if e.anySubstituterHasPath(ctx, state, record.Toplevel) {
485
485
cachedToplevel = record.Toplevel
486
486
fmt.Fprintf(out, "realizing cached NixOS config %s\n", cachedToplevel)
487
487
}
···
511
511
return nil
512
512
}
513
513
if e.cfg.NixCache.UploadURL == "" {
514
-
e.l.Warn("not committing config cache metadata: no upload URL configured", "workflow", wid, "configKey", configKey, "toplevel", result.Toplevel)
514
+
e.l.Warn("not committing toplevel metadata: no upload URL configured", "workflow", wid, "configKey", configKey, "toplevel", result.Toplevel)
515
515
return nil
516
516
}
517
517
···
519
519
// a partial upload would leave the cache unable to realize this toplevel,
520
520
// so skip the metadata commit rather than poison it with an un-realizable
521
521
// key. the config still activated fine, so don't fail the workflow.
522
-
e.l.Warn("cache drain failed; skipping config cache metadata commit", "workflow", wid, "configKey", configKey, "toplevel", result.Toplevel, "error", err)
522
+
e.l.Warn("substituter upload drain failed; skipping toplevel metadata commit", "workflow", wid, "configKey", configKey, "toplevel", result.Toplevel, "error", err)
523
523
return nil
524
524
}
525
-
if err := state.NixOSToplevelCache.Commit(configKey, result.Toplevel); err != nil {
525
+
if err := state.NixOSToplevels.Commit(configKey, result.Toplevel); err != nil {
526
526
return err
527
527
}
528
-
fmt.Fprintf(out, "committed config cache metadata %s -> %s\n", configKey, result.Toplevel)
528
+
fmt.Fprintf(out, "committed toplevel metadata %s -> %s\n", configKey, result.Toplevel)
529
529
return nil
530
530
}
531
531
532
-
func (e *Engine) anyCacheHasPath(ctx context.Context, state *workflowState, storePath string) bool {
533
-
upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs)
532
+
func (e *Engine) anySubstituterHasPath(ctx context.Context, state *workflowState, storePath string) bool {
533
+
upstreams, err := BuildSubstituterUpstreams(e.cfg.NixCache.ReadURLs, state.SubstituterReadURLs)
534
534
if err != nil {
535
-
e.l.Warn("config cache check: build upstreams failed; treating as absent", "path", storePath, "error", err)
535
+
e.l.Warn("toplevel check: build upstreams failed; treating as absent", "path", storePath, "error", err)
536
536
return false
537
537
}
538
538
if len(upstreams) == 0 {
···
540
540
}
541
541
hash, _, err := parseStorePath(storePath)
542
542
if err != nil {
543
-
e.l.Warn("config cache check: invalid toplevel path; treating as absent", "path", storePath, "error", err)
543
+
e.l.Warn("toplevel check: invalid toplevel path; treating as absent", "path", storePath, "error", err)
544
544
return false
545
545
}
546
546
req, err := http.NewRequestWithContext(ctx, http.MethodHead, "http://upstream/"+hash+".narinfo", nil)
547
547
if err != nil {
548
-
e.l.Warn("config cache check: build request failed; treating as absent", "path", storePath, "error", err)
548
+
e.l.Warn("toplevel check: build request failed; treating as absent", "path", storePath, "error", err)
549
549
return false
550
550
}
551
551
resp, err := newNarinfoExistenceTransport(upstreams, e.l).RoundTrip(req)
552
552
if err != nil {
553
-
e.l.Warn("config cache check: narinfo probe failed; treating as absent", "path", storePath, "error", err)
553
+
e.l.Warn("toplevel check: narinfo probe failed; treating as absent", "path", storePath, "error", err)
554
554
return false
555
555
}
556
556
defer resp.Body.Close()
+23
spindle/engines/microvm/engine_test.go
+23
spindle/engines/microvm/engine_test.go
···
125
125
t.Fatalf("first step should be the activation step, got %+v", wf.Steps[0])
126
126
}
127
127
}
128
+
129
+
func TestInitWorkflowCachesRenameHint(t *testing.T) {
130
+
dir := t.TempDir()
131
+
writeTestImageSpec(t, dir, "alpine", validImageSpec())
132
+
133
+
e := testEngine(t, dir)
134
+
_, err := e.InitWorkflow(tangled.Pipeline_Workflow{
135
+
Raw: `
136
+
image: alpine
137
+
caches:
138
+
https://cache.nixos.org: ""
139
+
steps:
140
+
- name: hello
141
+
command: hello
142
+
`,
143
+
}, tangled.Pipeline{})
144
+
if err == nil {
145
+
t.Fatal("expected an error for the old caches key")
146
+
}
147
+
if !strings.Contains(err.Error(), "renamed to substituters") {
148
+
t.Fatalf("error should point at substituters, got: %v", err)
149
+
}
150
+
}
+6
-6
spindle/engines/microvm/models.go
+6
-6
spindle/engines/microvm/models.go
···
12
12
Dependencies []string `yaml:"dependencies"`
13
13
Registry map[string]any `yaml:"registry"`
14
14
Environment map[string]string `yaml:"environment"`
15
-
Caches map[string]string `yaml:"caches"`
15
+
Substituters map[string]string `yaml:"substituters"`
16
16
Steps []struct {
17
17
Name string `yaml:"name"`
18
18
Command string `yaml:"command"`
···
20
20
} `yaml:"steps"`
21
21
}
22
22
23
-
// flattens the caches map into sorted substituter URLs and trusted public keys
24
-
func workflowCaches(caches map[string]string) (urls []string, keys []string, err error) {
25
-
for cacheURL, key := range caches {
23
+
// sorted so the guest env is deterministic
24
+
func workflowSubstituters(substituters map[string]string) (urls []string, keys []string, err error) {
25
+
for cacheURL, key := range substituters {
26
26
urls = append(urls, cacheURL)
27
27
if key != "" {
28
28
keys = append(keys, key)
29
29
}
30
30
}
31
-
if _, err := parseCacheUpstreams(urls); err != nil {
32
-
return nil, nil, fmt.Errorf("caches: %w", err)
31
+
if _, err := parseSubstituterUpstreams(urls); err != nil {
32
+
return nil, nil, fmt.Errorf("substituters: %w", err)
33
33
}
34
34
slices.Sort(urls)
35
35
slices.Sort(keys)
+5
-5
spindle/engines/microvm/models_test.go
+5
-5
spindle/engines/microvm/models_test.go
···
5
5
"testing"
6
6
)
7
7
8
-
func TestWorkflowCaches(t *testing.T) {
9
-
urls, keys, err := workflowCaches(map[string]string{
8
+
func TestWorkflowSubstituters(t *testing.T) {
9
+
urls, keys, err := workflowSubstituters(map[string]string{
10
10
"https://hydra.nixos.org/": "hydra.nixos.org-1:CNHJZBh9K4tP3EKF6FkkgeVYsS3ohTl+oS0Qa8bezVs=",
11
11
"https://cache.garnix.io/": "cache.garnix.io:CTFPyKSLcx5RMJKfLo5EEPUObbA78b0YQ2DTCJXqr9g=",
12
12
"https://unsigned.example/": "",
···
32
32
}
33
33
}
34
34
35
-
func TestWorkflowCachesRejectsBadURLs(t *testing.T) {
35
+
func TestWorkflowSubstitutersRejectsBadURLs(t *testing.T) {
36
36
for _, bad := range []string{"ftp://cache.example/", "not a url"} {
37
-
if _, _, err := workflowCaches(map[string]string{bad: ""}); err == nil {
38
-
t.Errorf("workflowCaches(%q): expected error, got nil", bad)
37
+
if _, _, err := workflowSubstituters(map[string]string{bad: ""}); err == nil {
38
+
t.Errorf("workflowSubstituters(%q): expected error, got nil", bad)
39
39
}
40
40
}
41
41
}
+12
-12
spindle/engines/microvm/nixos_toplevel_cache.go
spindle/engines/microvm/nixos_toplevel.go
+12
-12
spindle/engines/microvm/nixos_toplevel_cache.go
spindle/engines/microvm/nixos_toplevel.go
···
12
12
"tangled.org/core/spindle/db"
13
13
)
14
14
15
-
const nixosToplevelCacheSchemaVersion = 1
15
+
const nixosToplevelSchemaVersion = 1
16
16
17
-
type nixosToplevelCacheRecord struct {
17
+
type nixosToplevelRecord struct {
18
18
ConfigKey string `json:"config_key"`
19
19
Toplevel string `json:"toplevel"`
20
20
UpdatedAt time.Time `json:"updated_at"`
21
21
}
22
22
23
-
type nixosToplevelCacheStore struct {
23
+
type nixosToplevelStore struct {
24
24
db *db.DB
25
25
}
26
26
27
-
func newNixOSToplevelCacheStore(d *db.DB) nixosToplevelCacheStore {
28
-
return nixosToplevelCacheStore{db: d}
27
+
func newNixOSToplevelStore(d *db.DB) nixosToplevelStore {
28
+
return nixosToplevelStore{db: d}
29
29
}
30
30
31
-
func (s nixosToplevelCacheStore) Lookup(configKey string) (nixosToplevelCacheRecord, bool, error) {
31
+
func (s nixosToplevelStore) Lookup(configKey string) (nixosToplevelRecord, bool, error) {
32
32
if s.db == nil {
33
-
return nixosToplevelCacheRecord{}, false, nil
33
+
return nixosToplevelRecord{}, false, nil
34
34
}
35
35
r, err := s.db.GetNixOSToplevelCacheRecord(configKey)
36
36
if err != nil {
37
37
if errors.Is(err, sql.ErrNoRows) {
38
-
return nixosToplevelCacheRecord{}, false, nil
38
+
return nixosToplevelRecord{}, false, nil
39
39
}
40
-
return nixosToplevelCacheRecord{}, false, err
40
+
return nixosToplevelRecord{}, false, err
41
41
}
42
-
return nixosToplevelCacheRecord{
42
+
return nixosToplevelRecord{
43
43
ConfigKey: r.ConfigKey,
44
44
Toplevel: r.Toplevel,
45
45
UpdatedAt: r.UpdatedAt,
46
46
}, true, nil
47
47
}
48
48
49
-
func (s nixosToplevelCacheStore) Commit(configKey, toplevel string) error {
49
+
func (s nixosToplevelStore) Commit(configKey, toplevel string) error {
50
50
if configKey == "" {
51
51
return fmt.Errorf("config key is empty")
52
52
}
···
82
82
BaseConfig string `json:"base_config"`
83
83
UserConfig string `json:"user_config"`
84
84
}{
85
-
Schema: nixosToplevelCacheSchemaVersion,
85
+
Schema: nixosToplevelSchemaVersion,
86
86
BaseConfig: baseHash,
87
87
UserConfig: userConfigHash(cfg),
88
88
}
+22
-22
spindle/engines/microvm/read_cache_proxy.go
spindle/engines/microvm/substituter_proxy.go
+22
-22
spindle/engines/microvm/read_cache_proxy.go
spindle/engines/microvm/substituter_proxy.go
···
21
21
)
22
22
23
23
const (
24
-
readCacheProxyPortMin = 20000
25
-
readCacheProxyPortMax = 60000
24
+
substituterProxyPortMin = 20000
25
+
substituterProxyPortMax = 60000
26
26
)
27
27
28
-
type ReadCacheProxy struct {
28
+
type SubstituterProxy struct {
29
29
port uint32
30
30
31
31
ln *vsock.Listener
32
32
server *http.Server
33
33
}
34
34
35
-
func StartReadCacheProxy(ctx context.Context, cid uint32, upstreams []CacheUpstream, logger *slog.Logger) (*ReadCacheProxy, error) {
35
+
func StartSubstituterProxy(ctx context.Context, cid uint32, upstreams []SubstituterUpstream, logger *slog.Logger) (*SubstituterProxy, error) {
36
36
if logger == nil {
37
37
logger = slog.Default()
38
38
}
···
47
47
return nil, err
48
48
}
49
49
50
-
proxy := &ReadCacheProxy{
50
+
proxy := &SubstituterProxy{
51
51
port: port,
52
52
ln: ln,
53
53
}
···
72
72
return proxy, nil
73
73
}
74
74
75
-
func (p *ReadCacheProxy) Port() uint32 {
75
+
func (p *SubstituterProxy) Port() uint32 {
76
76
if p == nil {
77
77
return 0
78
78
}
79
79
return p.port
80
80
}
81
81
82
-
func (p *ReadCacheProxy) Close() error {
82
+
func (p *SubstituterProxy) Close() error {
83
83
if p == nil {
84
84
return nil
85
85
}
···
121
121
}
122
122
}
123
123
124
-
func parseCacheUpstreams(raw []string) ([]*url.URL, error) {
124
+
func parseSubstituterUpstreams(raw []string) ([]*url.URL, error) {
125
125
upstreams := make([]*url.URL, 0, len(raw))
126
126
seen := make(map[string]struct{}, len(raw))
127
127
for _, value := range raw {
···
149
149
return upstreams, nil
150
150
}
151
151
152
-
type CacheUpstream struct {
152
+
type SubstituterUpstream struct {
153
153
url *url.URL
154
154
// guarded upstreams come from the workflow file
155
155
// requests to them are refused for special-purpose address ranges
156
156
guarded bool
157
157
}
158
158
159
-
func BuildCacheUpstreams(rawTrusted, rawGuarded []string) ([]CacheUpstream, error) {
160
-
trusted, err := parseCacheUpstreams(rawTrusted)
159
+
func BuildSubstituterUpstreams(rawTrusted, rawGuarded []string) ([]SubstituterUpstream, error) {
160
+
trusted, err := parseSubstituterUpstreams(rawTrusted)
161
161
if err != nil {
162
162
return nil, err
163
163
}
164
-
guarded, err := parseCacheUpstreams(rawGuarded)
164
+
guarded, err := parseSubstituterUpstreams(rawGuarded)
165
165
if err != nil {
166
166
return nil, err
167
167
}
168
-
return mergeCacheUpstreams(trusted, guarded), nil
168
+
return mergeSubstituterUpstreams(trusted, guarded), nil
169
169
}
170
170
171
-
func mergeCacheUpstreams(trusted, guarded []*url.URL) []CacheUpstream {
172
-
merged := make([]CacheUpstream, 0, len(trusted)+len(guarded))
171
+
func mergeSubstituterUpstreams(trusted, guarded []*url.URL) []SubstituterUpstream {
172
+
merged := make([]SubstituterUpstream, 0, len(trusted)+len(guarded))
173
173
seen := make(map[string]struct{}, len(trusted)+len(guarded))
174
174
for _, u := range trusted {
175
175
if _, ok := seen[u.String()]; ok {
176
176
continue
177
177
}
178
178
seen[u.String()] = struct{}{}
179
-
merged = append(merged, CacheUpstream{url: u})
179
+
merged = append(merged, SubstituterUpstream{url: u})
180
180
}
181
181
for _, u := range guarded {
182
182
if _, ok := seen[u.String()]; ok {
183
183
continue
184
184
}
185
185
seen[u.String()] = struct{}{}
186
-
merged = append(merged, CacheUpstream{url: u, guarded: true})
186
+
merged = append(merged, SubstituterUpstream{url: u, guarded: true})
187
187
}
188
188
return merged
189
189
}
···
215
215
if _, err := rand.Read(data[:]); err != nil {
216
216
return 0, fmt.Errorf("allocate read vsock port: %w", err)
217
217
}
218
-
span := uint32(readCacheProxyPortMax - readCacheProxyPortMin)
219
-
return readCacheProxyPortMin + binary.BigEndian.Uint32(data[:])%span, nil
218
+
span := uint32(substituterProxyPortMax - substituterProxyPortMin)
219
+
return substituterProxyPortMin + binary.BigEndian.Uint32(data[:])%span, nil
220
220
}
221
221
222
222
var proxyTransport = &http.Transport{
···
275
275
// merging)
276
276
const nixCacheInfo = "StoreDir: /nix/store\nWantMassQuery: 1\nPriority: 40\n"
277
277
278
-
func cacheProxyHandler(upstreams []CacheUpstream, logger *slog.Logger) http.Handler {
278
+
func cacheProxyHandler(upstreams []SubstituterUpstream, logger *slog.Logger) http.Handler {
279
279
proxy := &httputil.ReverseProxy{
280
280
// nothing to do here: the racing transport builds the full URL per
281
281
// upstream, it just needs the guest's path/query left intact
···
318
318
}
319
319
320
320
type parallelRacingTransport struct {
321
-
upstreams []CacheUpstream
321
+
upstreams []SubstituterUpstream
322
322
underlying http.RoundTripper
323
323
guardedUnderlying http.RoundTripper
324
324
logger *slog.Logger
···
341
341
ctx, cancel := context.WithCancel(req.Context())
342
342
cancels[i] = cancel
343
343
344
-
go func(idx int, target CacheUpstream, uCtx context.Context) {
344
+
go func(idx int, target SubstituterUpstream, uCtx context.Context) {
345
345
defer wg.Done()
346
346
347
347
raceReq := req.Clone(uCtx)
+18
-18
spindle/engines/microvm/read_cache_proxy_test.go
spindle/engines/microvm/substituter_proxy_test.go
+18
-18
spindle/engines/microvm/read_cache_proxy_test.go
spindle/engines/microvm/substituter_proxy_test.go
···
21
21
}))
22
22
defer second.Close()
23
23
24
-
upstreams, err := parseCacheUpstreams([]string{first.URL, second.URL})
24
+
upstreams, err := parseSubstituterUpstreams([]string{first.URL, second.URL})
25
25
if err != nil {
26
26
t.Fatal(err)
27
27
}
28
28
29
29
req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil)
30
30
rec := httptest.NewRecorder()
31
-
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
31
+
cacheProxyHandler(mergeSubstituterUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
32
32
33
33
if rec.Code != http.StatusOK {
34
34
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
···
44
44
}))
45
45
defer upstream.Close()
46
46
47
-
upstreams, err := parseCacheUpstreams([]string{upstream.URL})
47
+
upstreams, err := parseSubstituterUpstreams([]string{upstream.URL})
48
48
if err != nil {
49
49
t.Fatal(err)
50
50
}
51
51
52
52
req := httptest.NewRequest(http.MethodGet, "http://guest/nix-cache-info", nil)
53
53
rec := httptest.NewRecorder()
54
-
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
54
+
cacheProxyHandler(mergeSubstituterUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
55
55
56
56
if rec.Code != http.StatusOK {
57
57
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
···
72
72
}))
73
73
defer healthy.Close()
74
74
75
-
upstreams, err := parseCacheUpstreams([]string{erroring.URL, healthy.URL})
75
+
upstreams, err := parseSubstituterUpstreams([]string{erroring.URL, healthy.URL})
76
76
if err != nil {
77
77
t.Fatal(err)
78
78
}
79
79
80
80
req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil)
81
81
rec := httptest.NewRecorder()
82
-
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
82
+
cacheProxyHandler(mergeSubstituterUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
83
83
84
84
if rec.Code != http.StatusOK {
85
85
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
···
105
105
defer upstream.Close()
106
106
107
107
upstreamURL := "http://dawn:woof@" + strings.TrimPrefix(upstream.URL, "http://") + "/sub/cache/?token=s3cret"
108
-
upstreams, err := parseCacheUpstreams([]string{upstreamURL})
108
+
upstreams, err := parseSubstituterUpstreams([]string{upstreamURL})
109
109
if err != nil {
110
110
t.Fatal(err)
111
111
}
112
112
113
113
req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil)
114
114
rec := httptest.NewRecorder()
115
-
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
115
+
cacheProxyHandler(mergeSubstituterUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
116
116
117
117
if rec.Code != http.StatusOK {
118
118
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
···
122
122
}
123
123
}
124
124
125
-
func TestCacheProxyGuardAllowsPublicIPv4(t *testing.T) {
126
-
if err := refuseSpecialPurposeAddrs("tcp", "104.26.13.82:443", nil); err != nil {
127
-
t.Fatalf("public IPv4 address was blocked: %v", err)
128
-
}
129
-
}
130
-
131
125
func TestCacheProxyGuardedUpstreamCannotReachBlockedRanges(t *testing.T) {
132
126
// httptest listens on 127.0.0.1, which is in the blocked ranges; reaching
133
127
// it would mean a workflow-defined cache can hit the host's loopback
···
136
130
}))
137
131
defer upstream.Close()
138
132
139
-
upstreams, err := parseCacheUpstreams([]string{upstream.URL})
133
+
upstreams, err := parseSubstituterUpstreams([]string{upstream.URL})
140
134
if err != nil {
141
135
t.Fatal(err)
142
136
}
143
137
144
138
req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil)
145
139
rec := httptest.NewRecorder()
146
-
cacheProxyHandler(mergeCacheUpstreams(nil, upstreams), slog.Default()).ServeHTTP(rec, req)
140
+
cacheProxyHandler(mergeSubstituterUpstreams(nil, upstreams), slog.Default()).ServeHTTP(rec, req)
147
141
148
142
if rec.Code != http.StatusBadGateway {
149
143
t.Fatalf("status: got %d, want 502; body=%q", rec.Code, rec.Body.String())
···
161
155
defer upstream.Close()
162
156
upstreamHost = strings.TrimPrefix(upstream.URL, "http://")
163
157
164
-
upstreams, err := parseCacheUpstreams([]string{upstream.URL})
158
+
upstreams, err := parseSubstituterUpstreams([]string{upstream.URL})
165
159
if err != nil {
166
160
t.Fatal(err)
167
161
}
···
169
163
req := httptest.NewRequest(http.MethodGet, "http://127.0.0.1:10500/abc.narinfo", nil)
170
164
req.Host = "127.0.0.1:10500"
171
165
rec := httptest.NewRecorder()
172
-
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
166
+
cacheProxyHandler(mergeSubstituterUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
173
167
174
168
if rec.Code != http.StatusOK {
175
169
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
176
170
}
177
171
}
172
+
173
+
func TestCacheProxyGuardAllowsPublicIPv4(t *testing.T) {
174
+
if err := refuseSpecialPurposeAddrs("tcp", "104.26.13.82:443", nil); err != nil {
175
+
t.Fatalf("public IPv4 address was blocked: %v", err)
176
+
}
177
+
}
spindle/engines/microvm/start-test-cache.sh
spindle/engines/microvm/start-test-substituter.sh
spindle/engines/microvm/start-test-cache.sh
spindle/engines/microvm/start-test-substituter.sh
+1
-1
spindle/engines/microvm/test-spindle-microvm.sh
+1
-1
spindle/engines/microvm/test-spindle-microvm.sh
···
271
271
trap 'exit 143' TERM
272
272
273
273
CACHE_PORT=$(pick_free_port)
274
-
./spindle/engines/microvm/start-test-cache.sh "$TEMP_DIR" "$CACHE_PORT"
274
+
./spindle/engines/microvm/start-test-substituter.sh "$TEMP_DIR" "$CACHE_PORT"
275
275
source "$TEMP_DIR/env.sh"
276
276
277
277
run_vm() {
+5
-5
spindle/engines/microvm/upload_cache_http.go
spindle/engines/microvm/substituter_upload_http.go
+5
-5
spindle/engines/microvm/upload_cache_http.go
spindle/engines/microvm/substituter_upload_http.go
···
17
17
handler http.Handler
18
18
}
19
19
20
-
func newHTTPUploadProxyBackend(target *url.URL, readUpstreams []CacheUpstream, logger *slog.Logger) *httpUploadBackend {
20
+
func newHTTPUploadProxyBackend(target *url.URL, readUpstreams []SubstituterUpstream, logger *slog.Logger) *httpUploadBackend {
21
21
return &httpUploadBackend{handler: uploadProxyHandler(target, readUpstreams, logger)}
22
22
}
23
23
···
27
27
28
28
func (b *httpUploadBackend) Close() error { return nil }
29
29
30
-
func uploadProxyHandler(target *url.URL, readUpstreams []CacheUpstream, logger *slog.Logger) http.Handler {
30
+
func uploadProxyHandler(target *url.URL, readUpstreams []SubstituterUpstream, logger *slog.Logger) http.Handler {
31
31
rp := httputil.NewSingleHostReverseProxy(target)
32
32
rp.ErrorLog = slog.NewLogLogger(logger.Handler(), slog.LevelError)
33
33
···
46
46
47
47
// before uploading, nix copy asks the destination whether it already has each
48
48
// path by GET/HEAD-ing <hash>.narinfo and skips the ones it does. we answer
49
-
// that check across the upload target *and* the read caches: if any of them
49
+
// that check across the upload target *and* the read substituters: if any of them
50
50
// already serves the path there is no point uploading it (the guest would
51
51
// just substitute it from there anyway).
52
-
narinfoUpstreams := append([]CacheUpstream{{url: target}}, readUpstreams...)
52
+
narinfoUpstreams := append([]SubstituterUpstream{{url: target}}, readUpstreams...)
53
53
exists := newNarinfoExistenceTransport(narinfoUpstreams, logger)
54
54
55
55
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
···
61
61
})
62
62
}
63
63
64
-
func newNarinfoExistenceTransport(upstreams []CacheUpstream, logger *slog.Logger) http.RoundTripper {
64
+
func newNarinfoExistenceTransport(upstreams []SubstituterUpstream, logger *slog.Logger) http.RoundTripper {
65
65
return ¶llelRacingTransport{
66
66
upstreams: upstreams,
67
67
underlying: proxyTransport,
spindle/engines/microvm/upload_cache_narinfo.go
spindle/engines/microvm/narinfo.go
spindle/engines/microvm/upload_cache_narinfo.go
spindle/engines/microvm/narinfo.go
+2
-2
spindle/engines/microvm/upload_cache_nix_store.go
spindle/engines/microvm/substituter_upload_nix_store.go
+2
-2
spindle/engines/microvm/upload_cache_nix_store.go
spindle/engines/microvm/substituter_upload_nix_store.go
···
40
40
type NixStoreUploadBackend struct {
41
41
stagingDir string
42
42
targetStore string
43
-
readUpstreams []CacheUpstream
43
+
readUpstreams []SubstituterUpstream
44
44
logger *slog.Logger
45
45
runner CommandRunner
46
46
maxNarUploadSize int64
47
47
}
48
48
49
-
func newNixStoreUploadBackend(targetStore, stagingDir string, readUpstreams []CacheUpstream, logger *slog.Logger, runner CommandRunner) (*NixStoreUploadBackend, error) {
49
+
func newNixStoreUploadBackend(targetStore, stagingDir string, readUpstreams []SubstituterUpstream, logger *slog.Logger, runner CommandRunner) (*NixStoreUploadBackend, error) {
50
50
absStaging, err := filepath.Abs(stagingDir)
51
51
if err != nil {
52
52
return nil, fmt.Errorf("resolve staging dir %q: %w", stagingDir, err)
+6
-6
spindle/engines/microvm/upload_cache_nix_store_test.go
spindle/engines/microvm/substituter_upload_nix_store_test.go
+6
-6
spindle/engines/microvm/upload_cache_nix_store_test.go
spindle/engines/microvm/substituter_upload_nix_store_test.go
···
23
23
testStorePath = "/nix/store/" + testStoreHash + "-abc-output"
24
24
)
25
25
26
-
func TestUploadCacheBackendSchemeDispatch(t *testing.T) {
26
+
func TestSubstituterUploadBackendSchemeDispatch(t *testing.T) {
27
27
staging := t.TempDir()
28
28
logger := slog.Default()
29
29
···
44
44
45
45
for _, tc := range cases {
46
46
t.Run(tc.uploadURL, func(t *testing.T) {
47
-
backend, err := newUploadCacheBackend(tc.uploadURL, nil, staging, logger)
47
+
backend, err := newSubstituterUploadBackend(tc.uploadURL, nil, staging, logger)
48
48
if tc.wantErr {
49
49
if err == nil {
50
50
t.Fatalf("expected error for %q", tc.uploadURL)
···
62
62
}
63
63
}
64
64
65
-
func TestUploadCacheBackendEmptyURL(t *testing.T) {
66
-
backend, err := newUploadCacheBackend("", nil, t.TempDir(), slog.Default())
65
+
func TestSubstituterUploadBackendEmptyURL(t *testing.T) {
66
+
backend, err := newSubstituterUploadBackend("", nil, t.TempDir(), slog.Default())
67
67
if err != nil {
68
68
t.Fatalf("unexpected error: %v", err)
69
69
}
···
245
245
246
246
runner := &fakeRunner{}
247
247
staging := t.TempDir()
248
-
b, err := newNixStoreUploadBackend("ssh-ng://cache-host", staging, []CacheUpstream{{url: upURL}}, slog.Default(), runner)
248
+
b, err := newNixStoreUploadBackend("ssh-ng://cache-host", staging, []SubstituterUpstream{{url: upURL}}, slog.Default(), runner)
249
249
if err != nil {
250
250
t.Fatalf("newNixStoreUploadBackend: %v", err)
251
251
}
···
313
313
}
314
314
315
315
staging := t.TempDir()
316
-
b, err := newNixStoreUploadBackend("ssh://cache-host", staging, []CacheUpstream{{url: upURL}}, slog.Default(), nil)
316
+
b, err := newNixStoreUploadBackend("ssh://cache-host", staging, []SubstituterUpstream{{url: upURL}}, slog.Default(), nil)
317
317
if err != nil {
318
318
t.Fatalf("newNixStoreUploadBackend: %v", err)
319
319
}
+11
-11
spindle/engines/microvm/upload_cache_proxy.go
spindle/engines/microvm/substituter_upload_proxy.go
+11
-11
spindle/engines/microvm/upload_cache_proxy.go
spindle/engines/microvm/substituter_upload_proxy.go
···
14
14
"github.com/mdlayher/vsock"
15
15
)
16
16
17
-
type UploadCacheBackend interface {
17
+
type SubstituterUploadBackend interface {
18
18
http.Handler
19
19
Close() error
20
20
}
21
21
22
-
type UploadCacheProxy struct {
22
+
type SubstituterUploadProxy struct {
23
23
port uint32
24
24
25
25
ln *vsock.Listener
26
26
server *http.Server
27
-
backend UploadCacheBackend
27
+
backend SubstituterUploadBackend
28
28
}
29
29
30
-
func StartUploadCacheProxy(ctx context.Context, cid uint32, uploadURL string, readUpstreams []CacheUpstream, stagingDir string, logger *slog.Logger) (*UploadCacheProxy, error) {
30
+
func StartSubstituterUploadProxy(ctx context.Context, cid uint32, uploadURL string, readUpstreams []SubstituterUpstream, stagingDir string, logger *slog.Logger) (*SubstituterUploadProxy, error) {
31
31
if strings.TrimSpace(uploadURL) == "" {
32
32
return nil, nil
33
33
}
···
37
37
}
38
38
logger = logger.With("where", "upload_cache_proxy", "cid", cid, "uploadURL", uploadURL)
39
39
40
-
backend, err := newUploadCacheBackend(uploadURL, readUpstreams, stagingDir, logger)
40
+
backend, err := newSubstituterUploadBackend(uploadURL, readUpstreams, stagingDir, logger)
41
41
if err != nil {
42
42
return nil, err
43
43
}
···
47
47
return nil, fmt.Errorf("listen for cache upload proxy: %w", err)
48
48
}
49
49
50
-
proxy := &UploadCacheProxy{
50
+
proxy := &SubstituterUploadProxy{
51
51
port: port,
52
52
ln: ln,
53
53
backend: backend,
···
65
65
}
66
66
go func() {
67
67
if err := proxy.server.Serve(filtered); err != nil && !errors.Is(err, http.ErrServerClosed) && !errors.Is(err, net.ErrClosed) {
68
-
logger.Warn("upload cache proxy stopped", "port", port, "error", err)
68
+
logger.Warn("substituter upload proxy stopped", "port", port, "error", err)
69
69
}
70
70
}()
71
71
72
-
logger.Info("started upload cache proxy", "port", port, "target", uploadURL, "readUpstreams", len(readUpstreams))
72
+
logger.Info("started substituter upload proxy", "port", port, "target", uploadURL, "readUpstreams", len(readUpstreams))
73
73
return proxy, nil
74
74
}
75
75
76
-
func newUploadCacheBackend(uploadURL string, readUpstreams []CacheUpstream, stagingDir string, logger *slog.Logger) (UploadCacheBackend, error) {
76
+
func newSubstituterUploadBackend(uploadURL string, readUpstreams []SubstituterUpstream, stagingDir string, logger *slog.Logger) (SubstituterUploadBackend, error) {
77
77
if strings.TrimSpace(uploadURL) == "" {
78
78
return nil, nil
79
79
}
···
106
106
}
107
107
}
108
108
109
-
func (p *UploadCacheProxy) Port() uint32 {
109
+
func (p *SubstituterUploadProxy) Port() uint32 {
110
110
if p == nil {
111
111
return 0
112
112
}
113
113
return p.port
114
114
}
115
115
116
-
func (p *UploadCacheProxy) Close() error {
116
+
func (p *SubstituterUploadProxy) Close() error {
117
117
if p == nil {
118
118
return nil
119
119
}
+2
-2
spindle/engines/microvm/upload_cache_proxy_test.go
spindle/engines/microvm/substituter_upload_proxy_test.go
+2
-2
spindle/engines/microvm/upload_cache_proxy_test.go
spindle/engines/microvm/substituter_upload_proxy_test.go
···
69
69
70
70
handler := uploadProxyHandler(
71
71
mustParseURL(t, target.URL),
72
-
[]CacheUpstream{{url: mustParseURL(t, upstream.URL)}},
72
+
[]SubstituterUpstream{{url: mustParseURL(t, upstream.URL)}},
73
73
slog.Default(),
74
74
)
75
75
···
97
97
98
98
handler := uploadProxyHandler(
99
99
mustParseURL(t, target.URL),
100
-
[]CacheUpstream{{url: mustParseURL(t, upstream.URL)}},
100
+
[]SubstituterUpstream{{url: mustParseURL(t, upstream.URL)}},
101
101
slog.Default(),
102
102
)
103
103
+17
-17
spindle/engines/microvm/vm.go
+17
-17
spindle/engines/microvm/vm.go
···
148
148
}
149
149
150
150
type workflowState struct {
151
-
ImageSpec ImageSpec
152
-
ImageSpecPath string
153
-
Config manifestConfig
154
-
ConfigKey string
155
-
Image string
156
-
CacheReadURLs []string
157
-
CacheTrustedPublicKeys []string
158
-
VM VMHandle
159
-
Agent *AgentSession
160
-
ReadCache *ReadCacheProxy
161
-
UploadCache *UploadCacheProxy
162
-
DNSProxy *DNSProxy
163
-
WorkDir string
164
-
NixOSToplevelCache nixosToplevelCacheStore
151
+
ImageSpec ImageSpec
152
+
ImageSpecPath string
153
+
Config manifestConfig
154
+
ConfigKey string
155
+
Image string
156
+
SubstituterReadURLs []string
157
+
SubstituterTrustedPublicKeys []string
158
+
VM VMHandle
159
+
Agent *AgentSession
160
+
Substituter *SubstituterProxy
161
+
SubstituterUpload *SubstituterUploadProxy
162
+
DNSProxy *DNSProxy
163
+
WorkDir string
164
+
NixOSToplevels nixosToplevelStore
165
165
}
166
166
167
167
func (e *Engine) cleanupState(ctx context.Context, wid models.WorkflowId, state *workflowState) error {
···
178
178
}
179
179
err = errors.Join(err, e.shutdownVM(ctx, wid, state))
180
180
err = errors.Join(err, closeIO(&state.Agent))
181
-
err = errors.Join(err, closeIO(&state.ReadCache))
182
-
err = errors.Join(err, closeIO(&state.UploadCache))
181
+
err = errors.Join(err, closeIO(&state.Substituter))
182
+
err = errors.Join(err, closeIO(&state.SubstituterUpload))
183
183
err = errors.Join(err, closeIO(&state.DNSProxy))
184
184
err = errors.Join(err, removeWorkDir(state))
185
185
return err
···
190
190
return nil
191
191
}
192
192
193
-
drainCtx, cancel := context.WithTimeout(ctx, cacheDrainTimeout)
193
+
drainCtx, cancel := context.WithTimeout(ctx, substituterDrainTimeout)
194
194
defer cancel()
195
195
196
196
if state.Agent != nil {