amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package appconfig
2
3import (
4 "context"
5 "errors"
6 "os"
7 "path/filepath"
8 "strings"
9 "sync"
10 "sync/atomic"
11 "testing"
12
13 "github.com/carlosframework/platform/internal/hostwrite"
14 "github.com/carlosframework/platform/internal/provision"
15 "github.com/carlosframework/platform/internal/registry"
16 "github.com/carlosframework/platform/internal/release"
17)
18
19type pausedConfigWork struct {
20 valid atomic.Bool
21 once sync.Once
22 entered chan struct{}
23 resume chan struct{}
24}
25
26func newPausedConfigWork() *pausedConfigWork {
27 w := &pausedConfigWork{entered: make(chan struct{}), resume: make(chan struct{})}
28 w.valid.Store(true)
29 return w
30}
31
32func (w *pausedConfigWork) With(_ context.Context, id hostwrite.Identity, fn func() error) error {
33 if id.Config == "" {
34 return errors.New("missing rendered identity")
35 }
36 w.once.Do(func() {
37 close(w.entered)
38 <-w.resume
39 })
40 if !w.valid.Load() {
41 return errors.New("selection changed")
42 }
43 return fn()
44}
45
46func TestSyncExecChangedSelectionBeforeMutationWritesNothing(t *testing.T) {
47 dir := t.TempDir()
48 host := "app.example"
49 rt := registry.Route{Host: host, Account: "bab", App: "app"}
50 w := newPausedConfigWork()
51 bounces := atomic.Int64{}
52 done := make(chan error, 1)
53 go func() {
54 done <- syncExecHostClaimed(t.Context(), w, rt, &Manifest{Vars: map[string]Var{"A": {Kind: "env", Value: "one"}}}, Key("bab", "app"), nil, dir, func(string) error {
55 bounces.Add(1)
56 return nil
57 }, "", "", nil)
58 }()
59 <-w.entered
60 w.valid.Store(false)
61 close(w.resume)
62 if err := <-done; err == nil {
63 t.Fatal("stale config write succeeded")
64 }
65 if _, err := os.Stat(filepath.Join(dir, host+configEnvSuffix)); !os.IsNotExist(err) {
66 t.Fatalf("config file exists after refusal: %v", err)
67 }
68 if bounces.Load() != 0 {
69 t.Fatal("stale config bounced process")
70 }
71}
72
73func TestSyncExecChangedSelectionBeforeRemovalPreservesBytes(t *testing.T) {
74 dir := t.TempDir()
75 host := "app.example"
76 path := filepath.Join(dir, host+configEnvSuffix)
77 if err := os.WriteFile(path, []byte("KEEP=1\n"), 0o600); err != nil {
78 t.Fatal(err)
79 }
80 w := newPausedConfigWork()
81 done := make(chan error, 1)
82 go func() {
83 done <- syncExecHostClaimed(t.Context(), w, registry.Route{Host: host, Account: "bab", App: "app"}, nil, Key("bab", "app"), nil, dir, func(string) error { return nil }, "", "", nil)
84 }()
85 <-w.entered
86 w.valid.Store(false)
87 close(w.resume)
88 if err := <-done; err == nil {
89 t.Fatal("stale removal succeeded")
90 }
91 if got, err := os.ReadFile(path); err != nil || string(got) != "KEEP=1\n" {
92 t.Fatalf("preserved config = %q, %v", got, err)
93 }
94}
95
96func TestSyncEnabledMissingProviderRefusesBeforeConfigReads(t *testing.T) {
97 reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
98 if err != nil {
99 t.Fatal(err)
100 }
101 t.Cleanup(func() { reg.Close() })
102 if err := reg.Upsert(registry.Route{Host: "app.example", Kind: "instance", Account: "bab", App: "app", Backing: registry.BackingUnit, Status: "ready"}); err != nil {
103 t.Fatal(err)
104 }
105 err = Sync(t.Context(), reg, nil, nil, t.TempDir(), func(string) error { return nil }, nil, hostwrite.Config{Enabled: true})
106 if !errors.Is(err, hostwrite.ErrUnavailable) {
107 t.Fatalf("Sync missing provider = %v", err)
108 }
109}
110
111func TestSyncRejectsMultipleHostWriteConfigurations(t *testing.T) {
112 reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
113 if err != nil {
114 t.Fatal(err)
115 }
116 t.Cleanup(func() { reg.Close() })
117 if err := Sync(context.Background(), reg, nil, nil, t.TempDir(), nil, nil, hostwrite.Config{}, hostwrite.Config{}); !errors.Is(err, errHostWriteOptions) {
118 t.Fatalf("multiple configs = %v", err)
119 }
120}
121
122func TestClaimDisabledConfigErrorContinuesHostsAndReaps(t *testing.T) {
123 for _, exec := range []bool{false, true} {
124 t.Run(map[bool]string{false: "unit", true: "exec"}[exec], func(t *testing.T) {
125 reg := testRegistry(t)
126 defer reg.Close()
127 store := release.NewDirStore(t.TempDir())
128 dir := t.TempDir()
129 if err := SetEnv(t.Context(), store, "bab", "app", map[string]string{"VALUE": "new"}); err != nil {
130 t.Fatal(err)
131 }
132 suffix := ".env"
133 backing := registry.BackingUnit
134 if exec {
135 suffix = configEnvSuffix
136 backing = ""
137 }
138 for _, host := range []string{"a.example", "b.example"} {
139 if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Account: "bab", App: "app", Backing: backing, Status: "ready", Socket: "/unused.sock"}); err != nil {
140 t.Fatal(err)
141 }
142 }
143 // A directory at the temporary-file path produces a deterministic error even as root.
144 if err := os.Mkdir(filepath.Join(dir, "a.example"+suffix+".tmp"), 0700); err != nil {
145 t.Fatal(err)
146 }
147 stale := filepath.Join(dir, "gone.example"+suffix)
148 if err := os.WriteFile(stale, []byte("old"), 0600); err != nil {
149 t.Fatal(err)
150 }
151 calls := map[string]int{}
152 restart := func(host string) error { calls[host]++; return nil }
153 var err error
154 callbacks := Callbacks{Restart: restart,
155 SelectedRestart: func(context.Context, hostwrite.Selection) error { t.Fatal("disabled selected restart"); return nil },
156 SelectedAck: func(context.Context, hostwrite.Selection, ConfigAck) error {
157 t.Fatal("disabled selected ack")
158 return nil
159 },
160 }
161 if exec {
162 err = SyncExecWithCallbacks(t.Context(), reg, store, nil, dir, "", callbacks)
163 } else {
164 err = SyncWithCallbacks(t.Context(), reg, store, nil, dir, callbacks)
165 }
166 if err != nil {
167 t.Fatalf("legacy pass aborted: %v", err)
168 }
169 if _, err := os.Stat(filepath.Join(dir, "b.example"+suffix)); err != nil || calls["b.example"] != 1 {
170 t.Fatalf("later host: %v calls=%v", err, calls)
171 }
172 if _, err := os.Stat(stale); !os.IsNotExist(err) {
173 t.Fatalf("reap skipped: %v", err)
174 }
175 })
176 }
177}
178
179type configTestSelector struct{}
180
181func (configTestSelector) Select(context.Context, string, string) (hostwrite.Work, error) {
182 return legacyHostWork{}, nil
183}
184
185func TestClaimEnabledConfigReadAndRenderFailures(t *testing.T) {
186 for _, shape := range []string{"unit", "flat-exec", "nested-exec"} {
187 for _, failure := range []string{"read", "render"} {
188 t.Run(shape+"/"+failure, func(t *testing.T) {
189 reg := testRegistry(t)
190 defer reg.Close()
191 store := release.NewDirStore(t.TempDir())
192 dir := t.TempDir()
193 rt := registry.Route{Host: "app.example", Kind: "instance", Account: "bab", App: "app", Status: "ready", Socket: "/unused.sock"}
194 path := filepath.Join(dir, rt.Host+".env")
195 if shape == "unit" {
196 rt.Backing = registry.BackingUnit
197 } else {
198 if shape == "nested-exec" {
199 rt.DBPath = provision.NestedDBPath(dir, rt.Host)
200 rt.Socket = provision.NestedSocket(rt.Host)
201 }
202 path = ConfigEnvPath(dir, rt)
203 }
204 if err := reg.Upsert(rt); err != nil {
205 t.Fatal(err)
206 }
207 vars := map[string]string{"VALUE": "new"}
208 if err := SetEnv(t.Context(), store, "bab", "app", vars); err != nil {
209 t.Fatal(err)
210 }
211 if failure == "read" {
212 if err := os.MkdirAll(path, 0700); err != nil {
213 t.Fatal(err)
214 }
215 } else {
216 _, etag, err := store.Get(t.Context(), Key("bab", "app"))
217 if err != nil {
218 t.Fatal(err)
219 }
220 if _, err := store.Put(t.Context(), Key("bab", "app"), []byte(`{"schema":1,"vars":{"VALUE":{"kind":"secret"}}}`), etag); err != nil {
221 t.Fatal(err)
222 }
223 }
224 callbacks := 0
225 restart := func(string) error { callbacks++; return nil }
226 ack := func(ConfigAck) { callbacks++ }
227 writes := hostwrite.Config{Enabled: true, Selector: configTestSelector{}}
228 var err error
229 if shape == "unit" {
230 err = Sync(t.Context(), reg, store, nil, dir, restart, ack, writes)
231 } else {
232 err = SyncExec(t.Context(), reg, store, nil, dir, restart, "", ack, WithHostWrite(writes))
233 }
234 if err == nil {
235 t.Fatal("enabled pass reported success after config " + failure + " failure")
236 }
237 if callbacks != 0 {
238 t.Fatalf("callbacks after failure: %d", callbacks)
239 }
240 })
241 }
242 }
243}
244
245func TestClaimConfigRefusesHostnameOnlyCallbacks(t *testing.T) {
246 for _, shape := range []string{"unit", "exec", "exec-removal", "ack"} {
247 t.Run(shape, func(t *testing.T) {
248 reg := testRegistry(t)
249 defer reg.Close()
250 store := release.NewDirStore(t.TempDir())
251 dir := t.TempDir()
252 rt := registry.Route{Host: "app.example", Kind: "instance", Account: "bab", App: "app", Status: "ready", Socket: "/unused.sock"}
253 if shape == "unit" {
254 rt.Backing = registry.BackingUnit
255 }
256 if shape == "ack" {
257 rt.Socket = ""
258 rt.Addr = "127.0.0.1:9999"
259 }
260 if err := reg.Upsert(rt); err != nil {
261 t.Fatal(err)
262 }
263 if shape == "exec-removal" {
264 if err := os.WriteFile(ConfigEnvPath(dir, rt), []byte("OLD=1\n"), 0600); err != nil {
265 t.Fatal(err)
266 }
267 } else if err := SetEnv(t.Context(), store, "bab", "app", map[string]string{"VALUE": "new"}); err != nil {
268 t.Fatal(err)
269 }
270 if shape == "ack" {
271 if err := SetExecDelivery(t.Context(), store, "bab", "app", ExecDeliveryFile); err != nil {
272 t.Fatal(err)
273 }
274 }
275 calls := 0
276 restart := func(string) error { calls++; return nil }
277 ack := func(ConfigAck) { calls++ }
278 writes := hostwrite.Config{Enabled: true, Selector: configTestSelector{}}
279 var err error
280 if shape == "unit" {
281 err = Sync(t.Context(), reg, store, nil, dir, restart, ack, writes)
282 } else {
283 err = SyncExec(t.Context(), reg, store, nil, dir, restart, "", ack, WithHostWrite(writes))
284 }
285 if calls != 0 {
286 t.Fatalf("enabled callbacks lost selection: %d effects, err=%v", calls, err)
287 }
288 if !errors.Is(err, hostwrite.ErrUnavailable) {
289 t.Fatalf("missing selected callback: %v", err)
290 }
291 })
292 }
293}
294
295type callbackConfigWork struct {
296 valid bool
297 held bool
298}
299
300func (w *callbackConfigWork) With(_ context.Context, id hostwrite.Identity, fn func() error) error {
301 if !w.valid {
302 return errors.New("selection changed")
303 }
304 if w.held || id.Config == "" {
305 return errors.New("invalid callback guard")
306 }
307 w.held = true
308 defer func() { w.held = false }()
309 return fn()
310}
311
312type selectedConfigTestSelector struct {
313 work hostwrite.Work
314 calls int
315}
316
317func (s *selectedConfigTestSelector) Select(context.Context, string, string) (hostwrite.Work, error) {
318 s.calls++
319 return s.work, nil
320}
321
322func TestClaimConfigCallbacksRetainSelection(t *testing.T) {
323 for _, shape := range []string{"unit", "exec", "exec-removal"} {
324 for _, boundary := range []string{"restart", "ack", "current"} {
325 if shape == "exec-removal" && boundary == "ack" {
326 continue
327 }
328 t.Run(shape+"/"+boundary, func(t *testing.T) {
329 reg := testRegistry(t)
330 defer reg.Close()
331 store := release.NewDirStore(t.TempDir())
332 dir := t.TempDir()
333 rt := registry.Route{Host: "app.example", Kind: "instance", Account: "bab", App: "app", Status: "ready", Socket: "/unused.sock"}
334 if shape == "unit" {
335 rt.Backing = registry.BackingUnit
336 }
337 if err := reg.Upsert(rt); err != nil {
338 t.Fatal(err)
339 }
340 if shape == "exec-removal" {
341 if err := os.WriteFile(ConfigEnvPath(dir, rt), []byte("OLD=1\n"), 0600); err != nil {
342 t.Fatal(err)
343 }
344 } else if err := SetEnv(t.Context(), store, "bab", "app", map[string]string{"VALUE": "new"}); err != nil {
345 t.Fatal(err)
346 }
347 w := &callbackConfigWork{valid: true}
348 selector := &selectedConfigTestSelector{work: w}
349 restarts, acks := 0, 0
350 var identity hostwrite.Identity
351 check := func(s hostwrite.Selection) {
352 if w.held {
353 t.Fatal("callback/network wait under guard")
354 }
355 if s.Work != w || s.Host != rt.Host || s.Identity.Config == "" {
356 t.Fatalf("lost selection: %+v", s)
357 }
358 if identity.Config != "" && s.Identity != identity {
359 t.Fatal("identity changed between callbacks")
360 }
361 identity = s.Identity
362 }
363 callbacks := Callbacks{
364 Restart: func(string) error { t.Fatal("legacy restart"); return nil },
365 Ack: func(ConfigAck) { t.Fatal("legacy ack") },
366 SelectedRestart: func(ctx context.Context, s hostwrite.Selection) error {
367 check(s)
368 if boundary == "restart" {
369 w.valid = false
370 }
371 return s.With(ctx, func() error { restarts++; return nil })
372 },
373 SelectedAck: func(ctx context.Context, s hostwrite.Selection, a ConfigAck) error {
374 check(s)
375 if a.Host != rt.Host {
376 t.Fatal(a.Host)
377 }
378 if boundary == "ack" {
379 w.valid = false
380 }
381 return s.With(ctx, func() error { acks++; return nil })
382 },
383 }
384 writes := hostwrite.Config{Enabled: true, Selector: selector}
385 var err error
386 if shape == "unit" {
387 err = SyncWithCallbacks(t.Context(), reg, store, nil, dir, callbacks, writes)
388 } else {
389 err = SyncExecWithCallbacks(t.Context(), reg, store, nil, dir, "", callbacks, WithHostWrite(writes))
390 }
391 wantRestarts, wantAcks := 1, 1
392 if shape == "exec-removal" {
393 wantAcks = 0
394 }
395 if boundary == "restart" {
396 wantRestarts = 0
397 wantAcks = 0
398 }
399 if boundary == "ack" {
400 wantAcks = 0
401 }
402 if (err == nil) != (boundary == "current") || restarts != wantRestarts || acks != wantAcks || selector.calls != 1 {
403 t.Fatalf("err=%v restarts=%d acks=%d selections=%d", err, restarts, acks, selector.calls)
404 }
405 })
406 }
407 }
408}
409
410type fencedConfigTestSelector struct {
411 t *testing.T
412 held string
413 selected []string
414}
415
416func (s *fencedConfigTestSelector) Select(_ context.Context, host, _ string) (hostwrite.Work, error) {
417 if host == s.held {
418 s.t.Fatal("fenced host selected for guarded work")
419 }
420 s.selected = append(s.selected, host)
421 return legacyHostWork{}, nil
422}
423
424func TestSyncExecFencedHostWithHostWrites(t *testing.T) {
425 for _, mode := range []string{"default", "disabled", "enabled"} {
426 for _, api := range []string{"legacy", "callbacks"} {
427 for _, shape := range []string{"flat", "nested"} {
428 for _, operation := range []string{"write", "remove"} {
429 t.Run(mode+"/"+api+"/"+shape+"/"+operation, func(t *testing.T) {
430 reg := testRegistry(t)
431 defer reg.Close()
432 store := release.NewDirStore(t.TempDir())
433 dir := t.TempDir()
434 held, sibling := "held.example", "sibling.example"
435 for _, host := range []string{held, sibling} {
436 rt := registry.Route{Host: host, Kind: "instance", Account: "bab", App: strings.TrimSuffix(host, ".example"), Status: "ready", Socket: "/unused.sock"}
437 if host == held && shape == "nested" {
438 rt.DBPath = provision.NestedDBPath(dir, host)
439 rt.Socket = provision.NestedSocket(host)
440 }
441 if err := reg.Upsert(rt); err != nil {
442 t.Fatal(err)
443 }
444 if host == sibling || operation == "write" {
445 if err := SetEnv(t.Context(), store, "bab", rt.App, map[string]string{"VALUE": "new"}); err != nil {
446 t.Fatal(err)
447 }
448 }
449 }
450 paths := []string{filepath.Join(dir, held+configEnvSuffix), filepath.Join(provision.InstanceDir(dir, held), held+configEnvSuffix)}
451 if mode == "enabled" {
452 paths = append(paths, filepath.Join(dir, held+configEnvSuffix+".tmp"))
453 }
454 for _, path := range paths {
455 if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
456 t.Fatal(err)
457 }
458 if err := os.WriteFile(path, []byte("KEEP=1\n"), 0600); err != nil {
459 t.Fatal(err)
460 }
461 }
462 selector := &fencedConfigTestSelector{t: t, held: held}
463 opts := []SyncOption{SkipFenced(func(host string) bool { return host == held })}
464 if mode != "default" {
465 opts = append(opts, WithHostWrite(hostwrite.Config{Enabled: mode == "enabled", Selector: selector}))
466 }
467 restarts, acks := 0, 0
468 restart := func(host string) error {
469 if host != sibling {
470 t.Fatalf("restarted fenced host %s", host)
471 }
472 restarts++
473 return nil
474 }
475 ack := func(a ConfigAck) {
476 if a.Host != sibling {
477 t.Fatalf("acknowledged fenced host %s", a.Host)
478 }
479 acks++
480 }
481 var err error
482 if api == "legacy" {
483 err = SyncExec(t.Context(), reg, store, nil, dir, restart, "", ack, opts...)
484 } else {
485 callbacks := Callbacks{Restart: restart, Ack: ack,
486 SelectedRestart: func(_ context.Context, s hostwrite.Selection) error { return restart(s.Host) },
487 SelectedAck: func(_ context.Context, s hostwrite.Selection, a ConfigAck) error {
488 if s.Host != a.Host {
489 t.Fatal("ack selection lost")
490 }
491 ack(a)
492 return nil
493 },
494 }
495 err = SyncExecWithCallbacks(t.Context(), reg, store, nil, dir, "", callbacks, opts...)
496 }
497 wantCalls := 1
498 if api == "legacy" && mode == "enabled" {
499 if !errors.Is(err, hostwrite.ErrUnavailable) {
500 t.Fatalf("legacy callback refusal = %v", err)
501 }
502 wantCalls = 0
503 } else if err != nil {
504 t.Fatal(err)
505 }
506 if restarts != wantCalls || acks != wantCalls {
507 t.Fatalf("restarts=%d acks=%d, want %d each", restarts, acks, wantCalls)
508 }
509 for _, path := range paths {
510 if got, err := os.ReadFile(path); err != nil || string(got) != "KEEP=1\n" {
511 t.Fatalf("fenced file %s = %q, %v", path, got, err)
512 }
513 }
514 if got, err := os.ReadFile(filepath.Join(dir, sibling+configEnvSuffix)); err != nil || string(got) != "VALUE=new\n" {
515 t.Fatalf("sibling config = %q, %v", got, err)
516 }
517 wantSelections := 0
518 if mode == "enabled" {
519 wantSelections = 1
520 }
521 if len(selector.selected) != wantSelections {
522 t.Fatalf("selected=%v, want %d", selector.selected, wantSelections)
523 }
524 })
525 }
526 }
527 }
528 }
529}
530
531func TestSyncExecRejectsMultipleHostWriteConfigurations(t *testing.T) {
532 reg := testRegistry(t)
533 defer reg.Close()
534 opts := []SyncOption{WithHostWrite(hostwrite.Config{Enabled: true}), SkipFenced(func(string) bool { return true }), WithHostWrite(hostwrite.Config{})}
535 if err := SyncExec(t.Context(), reg, nil, nil, t.TempDir(), nil, "", nil, opts...); !errors.Is(err, errHostWriteOptions) {
536 t.Fatalf("multiple configs = %v", err)
537 }
538 if err := SyncExecWithCallbacks(t.Context(), reg, nil, nil, t.TempDir(), "", Callbacks{}, opts...); !errors.Is(err, errHostWriteOptions) {
539 t.Fatalf("multiple configs with callbacks = %v", err)
540 }
541}
542