amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "encoding/json"
6 "os"
7 "path/filepath"
8 "testing"
9 "time"
10
11 "github.com/carlosframework/platform/internal/hostbox"
12 "github.com/carlosframework/platform/internal/hostclaim"
13 "github.com/carlosframework/platform/internal/hostguard"
14 "github.com/carlosframework/platform/internal/hostwrite"
15 "github.com/carlosframework/platform/internal/provision"
16 "github.com/carlosframework/platform/internal/release"
17 "github.com/carlosframework/platform/internal/rootparticipation"
18)
19
20type restartConsumptionFixture struct {
21 p rootRestartRequests
22 e *proofRootEnv
23 u *plan14RestartUnits
24 claim *hostclaim.Claim
25 marks string
26}
27
28func newRestartConsumptionFixture(t *testing.T) restartConsumptionFixture {
29 t.Helper()
30 e, rt, u, checks, _ := plan14AdoptionFixture(t)
31 rec, err := provision.GetRecord(t.Context(), e.store, rt.Account, rt.App, rt.Host)
32 if err != nil {
33 t.Fatal(err)
34 }
35 rec.Environment = "relay"
36 rec.CreatedAt = time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
37 plan14PutRecord(t, e.store, *rec)
38 inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
39 if err != nil {
40 t.Fatal(err)
41 }
42 c := &hostclaim.Claim{Host: rt.Host, Holder: &hostclaim.Holder{Key: provision.RecordKey(rt.Account, rt.App, rt.Host), Account: rt.Account, App: rt.App, Incarnation: inc, Stamp: 1, Session: 1, WriterBox: "box"}}
43 c.Op.Kind, c.Op.Exec = hostclaim.KindServing, "box"
44 state := &rollbackDisk{dir: t.TempDir()}
45 g := hostguard.Guard{DataDir: t.TempDir(), LockDir: t.TempDir(), Claims: rootViewClaimReader{obs: hostclaim.Observed{Host: rt.Host, ETag: "claim", Claim: c}}, Root: hostguard.NewRootAuthority(state)}
46 in := hostguard.RootInstallation{Host: rt.Host, Incarnation: inc, Stamp: 1, Account: rt.Account, App: rt.App, Record: c.Holder.Key, Box: "box", Enrollment: "en", Migration: 1, InventoryRevision: 1}
47 if err := g.EstablishRoot(t.Context(), in, func(*hostclaim.Claim) error { return nil }); err != nil {
48 t.Fatal(err)
49 }
50 writes := hostwrite.Config{Enabled: true, Selector: hostbox.WorkSource{Guard: g, Box: "box", Region: "ie", Now: time.Now,
51 Policy: func(_ context.Context, c *hostclaim.Claim, s hostbox.Selection, _ hostwrite.Identity) error {
52 return hostbox.Serving(c, s)
53 },
54 Local: func(context.Context, *hostclaim.Claim, hostbox.Selection) error { return nil },
55 }}
56 p := rootRestartRequests{journal: &release.RootRestarts{Store: &rollbackDisk{dir: t.TempDir()}},
57 source: rootRestartRequestSource{state: state, store: e.store, census: func() (rootparticipation.Census, error) {
58 return rootparticipation.Census{Schema: 1, Box: "box", Enrollment: "en", Migration: 1, Revision: 1, Complete: true}, nil
59 }},
60 writes: writes, sys: u, checks: checks, getenv: func(string) string { return "5s" }, optDir: u.opt,
61 }
62 return restartConsumptionFixture{p: p, e: e, u: u, claim: c, marks: restartMarksPath(t.TempDir())}
63}
64
65func (f restartConsumptionFixture) request(t *testing.T, kind, value string) {
66 t.Helper()
67 if _, err := release.AppendRestartScope(t.Context(), f.e.store, "bab", "console", release.RestartScope{Kind: kind, Value: value}); err != nil {
68 t.Fatal(err)
69 }
70}
71
72func (f restartConsumptionFixture) run(t *testing.T) error {
73 t.Helper()
74 return restartRequestPass(func() (bool, error) { return false, nil }, func() error {
75 routes, err := f.e.root.reg.List()
76 if err != nil {
77 return err
78 }
79 restartPass(t.Context(), routes, f.e.store, f.marks, func(host string) error { return f.u.TryRestart("console.service") })
80 return nil
81 }, func() error { return f.p.run(t.Context()) })
82}
83
84func TestPlan14RootRestartProductionConsumption(t *testing.T) {
85 t.Parallel()
86 for _, attack := range []string{"none", "delete-marks", "rename-parent", "forge-marks", "delete-row", "forge-selector", "forge-unit"} {
87 t.Run(attack, func(t *testing.T) {
88 f := newRestartConsumptionFixture(t)
89 if err := f.run(t); err != nil {
90 t.Fatal(err)
91 }
92 f.request(t, release.RestartScopeEnvironment, "relay")
93 if err := f.run(t); err != nil || len(f.u.effects) != 1 {
94 t.Fatalf("first request: %v %v", err, f.u.effects)
95 }
96 if err := os.WriteFile(f.marks, []byte(`{"schema":1,"apps":{},"scopes":{}}`), 0600); err != nil {
97 t.Fatal(err)
98 }
99 switch attack {
100 case "delete-marks":
101 if err := os.Remove(f.marks); err != nil {
102 t.Fatal(err)
103 }
104 case "rename-parent":
105 if err := os.Rename(filepath.Dir(f.marks), filepath.Dir(f.marks)+".old"); err != nil {
106 t.Fatal(err)
107 }
108 t.Cleanup(func() { _ = os.RemoveAll(filepath.Dir(f.marks) + ".old") })
109 case "forge-marks":
110 if err := os.WriteFile(f.marks, []byte(`{"schema":1,"apps":{"bab|console":"forged"},"scopes":{"bab|console":"forged"}}`), 0600); err != nil {
111 t.Fatal(err)
112 }
113 case "delete-row":
114 if err := f.e.root.reg.Delete(observedHost); err != nil {
115 t.Fatal(err)
116 }
117 case "forge-selector", "forge-unit":
118 r, err := f.e.root.reg.Get(observedHost)
119 if err != nil {
120 t.Fatal(err)
121 }
122 r.Environment, r.Unit, r.Hibernate = "forged", "unrelated.service", true
123 if err := f.e.root.reg.Upsert(*r); err != nil {
124 t.Fatal(err)
125 }
126 }
127 if err := f.run(t); err != nil || len(f.u.effects) != 1 {
128 t.Fatalf("replay: %v %v", err, f.u.effects)
129 }
130 f.request(t, release.RestartScopeEnvironment, "forged")
131 if err := f.run(t); err != nil || len(f.u.effects) != 1 {
132 t.Fatalf("widened: %v %v", err, f.u.effects)
133 }
134 f.request(t, release.RestartScopeHost, observedHost)
135 if err := f.run(t); err != nil || len(f.u.effects) != 2 {
136 t.Fatalf("fresh request: %v %v", err, f.u.effects)
137 }
138 })
139 }
140}
141
142func TestPlan14RootRestartProductionHeldRollover(t *testing.T) {
143 t.Parallel()
144 f := newRestartConsumptionFixture(t)
145 if err := f.run(t); err != nil {
146 t.Fatal(err)
147 }
148 f.claim.Op.Kind = hostclaim.KindLanding
149 f.request(t, release.RestartScopeHost, observedHost)
150 if err := f.e.root.reg.Delete(observedHost); err != nil {
151 t.Fatal(err)
152 }
153 if err := f.run(t); err == nil || len(f.u.effects) != 0 {
154 t.Fatalf("held: %v %v", err, f.u.effects)
155 }
156 for i := 0; i < 34; i++ {
157 f.request(t, release.RestartScopeHost, "other.example.com")
158 }
159 f.claim.Op.Kind = hostclaim.KindServing
160 if err := f.run(t); err != nil || len(f.u.effects) != 1 {
161 t.Fatalf("lost pending: %v %v", err, f.u.effects)
162 }
163 if err := f.run(t); err != nil || len(f.u.effects) != 1 {
164 t.Fatalf("replayed pending: %v %v", err, f.u.effects)
165 }
166}
167
168func TestPlan14RootRestartProductionBindingChange(t *testing.T) {
169 t.Parallel()
170 f := newRestartConsumptionFixture(t)
171 if err := f.run(t); err != nil {
172 t.Fatal(err)
173 }
174 f.claim.Op.Kind = hostclaim.KindLanding
175 f.request(t, release.RestartScopeHost, observedHost)
176 _ = f.run(t)
177 rec, err := provision.GetRecord(t.Context(), f.e.store, "bab", "console", observedHost)
178 if err != nil {
179 t.Fatal(err)
180 }
181 rec.Environment = "replacement"
182 plan14PutRecord(t, f.e.store, *rec)
183 f.claim.Op.Kind = hostclaim.KindServing
184 if err := f.run(t); err == nil || len(f.u.effects) != 0 {
185 t.Fatalf("changed binding: %v %v", err, f.u.effects)
186 }
187}
188
189func TestPlan14RootRestartProductionAdmissionAfterIntent(t *testing.T) {
190 t.Parallel()
191 f := newRestartConsumptionFixture(t)
192 if err := f.run(t); err != nil {
193 t.Fatal(err)
194 }
195 f.request(t, release.RestartScopeHost, observedHost)
196 disk := f.p.journal.Store.(*rollbackDisk)
197 disk.afterWrite = func(_ string, raw []byte) {
198 var r struct{ Pending *release.RestartIntent }
199 if json.Unmarshal(raw, &r) == nil && r.Pending != nil && r.Pending.Targets[0].Before != "" {
200 f.claim.Op.Kind = hostclaim.KindLanding
201 }
202 }
203 if err := f.run(t); err == nil || len(f.u.effects) != 0 {
204 t.Fatalf("move crossed dispatch intent: %v %v", err, f.u.effects)
205 }
206 disk.afterWrite = nil
207 f.claim.Op.Kind = hostclaim.KindServing
208 if err := f.run(t); err == nil || len(f.u.effects) != 0 {
209 t.Fatalf("ambiguous intent replayed: %v %v", err, f.u.effects)
210 }
211 // A verified successor completes the captured request without another call.
212 f.u.props.InvocationID += "successor"
213 f.u.props.MainPID++
214 if err := f.run(t); err != nil || len(f.u.effects) != 0 {
215 t.Fatalf("successor: %v %v", err, f.u.effects)
216 }
217}
218
219func TestPlan14RootRestartProductionUnknown(t *testing.T) {
220 t.Parallel()
221 for _, attack := range []string{"census-error", "census-incomplete", "root-corrupt", "record-missing", "retired-pending", "flag-off", "foreign-census", "inactive"} {
222 t.Run(attack, func(t *testing.T) {
223 f := newRestartConsumptionFixture(t)
224 if err := f.run(t); err != nil {
225 t.Fatal(err)
226 }
227 f.request(t, release.RestartScopeHost, observedHost)
228 switch attack {
229 case "census-error":
230 f.p.source.census = func() (rootparticipation.Census, error) { return rootparticipation.Census{}, os.ErrPermission }
231 case "census-incomplete":
232 f.p.source.census = func() (rootparticipation.Census, error) { return rootparticipation.Census{Box: "box"}, nil }
233 case "foreign-census":
234 f.p.source.census = func() (rootparticipation.Census, error) {
235 return rootparticipation.Census{Box: "other", Complete: true}, nil
236 }
237 case "root-corrupt":
238 if err := f.p.source.state.Write(observedHost+".json", []byte(`{}`)); err != nil {
239 t.Fatal(err)
240 }
241 case "record-missing":
242 if err := f.e.store.Delete(t.Context(), provision.RecordKey("bab", "console", observedHost)); err != nil {
243 t.Fatal(err)
244 }
245 case "retired-pending":
246 f.claim.Op.Kind = hostclaim.KindLanding
247 _ = f.run(t)
248 f.claim.Op.Kind = hostclaim.KindServing
249 if err := f.p.source.state.Remove(observedHost + ".json"); err != nil {
250 t.Fatal(err)
251 }
252 case "flag-off":
253 f.p.writes.Enabled = false
254 case "inactive":
255 f.u.props.ActiveState = "inactive"
256 }
257 if err := f.run(t); err == nil || len(f.u.effects) != 0 {
258 t.Fatalf("unknown granted: %v %v", err, f.u.effects)
259 }
260 })
261 }
262}
263
264func TestPlan14RootRestartRequestCensusGate(t *testing.T) {
265 for _, mode := range []string{"absent", "present", "unreadable"} {
266 t.Run(mode, func(t *testing.T) {
267 old, root := 0, 0
268 err := restartRequestPass(func() (bool, error) {
269 if mode == "unreadable" {
270 return false, os.ErrPermission
271 }
272 return mode == "absent", nil
273 }, func() error { old++; return nil }, func() error { root++; return nil })
274 if mode == "unreadable" {
275 if err == nil || old+root != 0 {
276 t.Fatal(err, old, root)
277 }
278 return
279 }
280 if err != nil || (mode == "absent" && (old != 1 || root != 0)) || (mode == "present" && (old != 0 || root != 1)) {
281 t.Fatal(err, old, root)
282 }
283 })
284 }
285}
286
287type restartRequestSyncFault struct {
288 *rollbackDisk
289 stage string
290}
291
292func (s *restartRequestSyncFault) Write(name string, raw []byte) error {
293 var r struct {
294 Pending *release.RestartIntent
295 ScopeID string `json:"scope_id"`
296 }
297 if err := json.Unmarshal(raw, &r); err != nil {
298 return err
299 }
300 stage := "baseline"
301 if r.Pending != nil {
302 stage = "pending"
303 if r.Pending.Targets[0].Before != "" {
304 stage = "intent"
305 }
306 if r.Pending.Targets[0].After != "" {
307 stage = "completed"
308 }
309 } else if r.ScopeID != release.RestartScopeNone {
310 stage = "consumed"
311 }
312 if stage != s.stage {
313 return s.rollbackDisk.Write(name, raw)
314 }
315 f, err := os.CreateTemp(s.dir, "unsynced-")
316 if err != nil {
317 return err
318 }
319 defer os.Remove(f.Name())
320 if _, err = f.Write(raw); err != nil {
321 f.Close()
322 return err
323 }
324 if err = f.Sync(); err != nil {
325 f.Close()
326 return err
327 }
328 if err = f.Close(); err != nil {
329 return err
330 }
331 if err = os.Rename(f.Name(), filepath.Join(s.dir, name)); err != nil {
332 return err
333 }
334 return os.ErrPermission
335}
336
337func TestPlan14RootRestartProductionSyncFailure(t *testing.T) {
338 t.Parallel()
339 for _, stage := range []string{"baseline", "pending", "intent", "completed", "consumed"} {
340 t.Run(stage, func(t *testing.T) {
341 f := newRestartConsumptionFixture(t)
342 if err := f.run(t); err != nil {
343 t.Fatal(err)
344 }
345 f.request(t, release.RestartScopeHost, observedHost)
346 fault := &restartRequestSyncFault{rollbackDisk: f.p.journal.Store.(*rollbackDisk), stage: stage}
347 f.p.journal.Store = fault
348 if err := f.run(t); err == nil {
349 t.Fatal("failed sync accepted")
350 }
351 want := 0
352 if stage == "completed" || stage == "consumed" {
353 want = 1
354 }
355 if len(f.u.effects) != want {
356 t.Fatal(f.u.effects)
357 }
358 // The read must repair durability before dispatch or completion.
359 _ = f.run(t)
360 if len(f.u.effects) != want {
361 t.Fatal("repeat under failed sync", f.u.effects)
362 }
363 fault.stage = ""
364 err := f.run(t)
365 if stage == "intent" {
366 if err == nil || len(f.u.effects) != 0 {
367 t.Fatalf("ambiguous repeat: %v %v", err, f.u.effects)
368 }
369 f.u.props.InvocationID += "verified"
370 f.u.props.MainPID++
371 if err = f.run(t); err != nil || len(f.u.effects) != 0 {
372 t.Fatalf("repair: %v %v", err, f.u.effects)
373 }
374 } else if err != nil || len(f.u.effects) != 1 {
375 t.Fatalf("repair: %v %v", err, f.u.effects)
376 }
377 })
378 }
379}
380
381func TestPlan14RootRestartProductionCanonicalSelection(t *testing.T) {
382 t.Parallel()
383 for _, kind := range []string{"hibernate", "exec", "sidecar"} {
384 t.Run(kind, func(t *testing.T) {
385 f := newRestartConsumptionFixture(t)
386 if kind == "sidecar" {
387 // A valid root sidecar installation cannot hide another app's units.
388 in, err := f.p.source.installation(t.Context(), observedHost)
389 if err != nil {
390 t.Fatal(err)
391 }
392 in.Host = provision.SidecarHost("worker", in.App, in.Account)
393 in.Record = provision.SidecarRecordKey(in.Account, in.App, "worker")
394 claim := *f.claim
395 holder := *claim.Holder
396 claim.Holder = &holder
397 claim.Host = in.Host
398 holder.Key = in.Record
399 g := hostguard.Guard{DataDir: t.TempDir(), LockDir: t.TempDir(), Root: hostguard.NewRootAuthority(f.p.source.state), Claims: rootViewClaimReader{obs: hostclaim.Observed{Host: in.Host, ETag: "sidecar", Claim: &claim}}}
400 if err = g.EstablishRoot(t.Context(), in, func(*hostclaim.Claim) error { return nil }); err != nil {
401 t.Fatal(err)
402 }
403 } else {
404 rec, err := provision.GetRecord(t.Context(), f.e.store, "bab", "console", observedHost)
405 if err != nil {
406 t.Fatal(err)
407 }
408 if kind == "exec" {
409 rec.Unit = ""
410 } else {
411 yes := true
412 rec.Hibernate = &yes
413 rec.Addr = ""
414 rec.Unit = "carlos-app@" + rec.Host + ".service"
415 }
416 plan14PutRecord(t, f.e.store, *rec)
417 }
418 if err := f.run(t); err != nil {
419 t.Fatal(err)
420 }
421 f.request(t, release.RestartScopeHost, observedHost)
422 if err := f.run(t); err != nil {
423 t.Fatal(err)
424 }
425 want := 0
426 if kind == "sidecar" {
427 want = 1
428 }
429 if len(f.u.effects) != want {
430 t.Fatal(kind, f.u.effects)
431 }
432 })
433 }
434}
435
436func TestPlan14RootRestartProductionReplacedInstallation(t *testing.T) {
437 t.Parallel()
438 f := newRestartConsumptionFixture(t)
439 if err := f.run(t); err != nil {
440 t.Fatal(err)
441 }
442 f.request(t, release.RestartScopeHost, observedHost)
443 f.claim.Op.Kind = hostclaim.KindLanding
444 _ = f.run(t)
445 in, err := f.p.source.installation(t.Context(), observedHost)
446 if err != nil {
447 t.Fatal(err)
448 }
449 f.claim.Holder.Stamp++
450 in.Stamp++
451 w := f.p.writes.Selector.(hostbox.WorkSource)
452 if err = w.Guard.EstablishRoot(t.Context(), in, func(*hostclaim.Claim) error { return nil }); err != nil {
453 t.Fatal(err)
454 }
455 f.claim.Op.Kind = hostclaim.KindServing
456 if err = f.run(t); err == nil || len(f.u.effects) != 0 {
457 t.Fatalf("captured request crossed installation: %v %v", err, f.u.effects)
458 }
459}
460
461type restartRequestErrorUnits struct {
462 *plan14RestartUnits
463 after bool
464}
465
466func (u restartRequestErrorUnits) TryRestart(unit string) error {
467 if u.after {
468 if err := u.plan14RestartUnits.TryRestart(unit); err != nil {
469 return err
470 }
471 }
472 return os.ErrPermission
473}
474
475func TestPlan14RootRestartProductionDispatchError(t *testing.T) {
476 t.Parallel()
477 for _, after := range []bool{false, true} {
478 t.Run(map[bool]string{false: "no-effect", true: "effect-then-error"}[after], func(t *testing.T) {
479 f := newRestartConsumptionFixture(t)
480 if err := f.run(t); err != nil {
481 t.Fatal(err)
482 }
483 f.request(t, release.RestartScopeHost, observedHost)
484 f.p.sys = restartRequestErrorUnits{f.u, after}
485 if err := f.run(t); err == nil {
486 t.Fatal("lost dispatch error")
487 }
488 b := release.RestartBinding{Box: "box", Account: "bab", App: "console"}
489 pending, err := f.p.journal.Observe(b, release.RestartSnapshot{})
490 if err != nil || pending == nil {
491 t.Fatalf("ambiguous consumed: %+v %v", pending, err)
492 }
493 f.p.sys = f.u
494 err = f.run(t)
495 if after {
496 if err != nil || len(f.u.effects) != 1 {
497 t.Fatal(err, f.u.effects)
498 }
499 } else {
500 if err == nil || len(f.u.effects) != 0 {
501 t.Fatal(err, f.u.effects)
502 }
503 }
504 })
505 }
506}
507
508func TestPlan14RootRestartProductionDefiniteFailure(t *testing.T) {
509 t.Parallel()
510 f := newRestartConsumptionFixture(t)
511 if err := f.run(t); err != nil {
512 t.Fatal(err)
513 }
514 f.request(t, release.RestartScopeHost, observedHost)
515 f.u.afterEffect = func() { f.u.props.ActiveState = "failed"; f.u.props.MainPID = 0 }
516 if err := f.run(t); err == nil || len(f.u.effects) != 1 {
517 t.Fatal(err, f.u.effects)
518 }
519 if err := f.run(t); err != nil || len(f.u.effects) != 1 {
520 t.Fatalf("failed request repeated: %v %v", err, f.u.effects)
521 }
522 f.u.afterEffect = nil
523 f.u.props.ActiveState = "active"
524 f.u.props.MainPID = 8
525 f.u.props.InvocationID += "recovered"
526 f.request(t, release.RestartScopeHost, observedHost)
527 if err := f.run(t); err != nil || len(f.u.effects) != 2 {
528 t.Fatalf("new request lost: %v %v", err, f.u.effects)
529 }
530}
531
532type restartRequestShowError struct{ *plan14RestartUnits }
533
534func (u restartRequestShowError) Show(unit string) (unitProps, error) {
535 if len(u.effects) > 0 {
536 return unitProps{}, os.ErrPermission
537 }
538 return u.plan14RestartUnits.Show(unit)
539}
540
541func TestPlan14RootRestartProductionObservationError(t *testing.T) {
542 t.Parallel()
543 f := newRestartConsumptionFixture(t)
544 if err := f.run(t); err != nil {
545 t.Fatal(err)
546 }
547 f.request(t, release.RestartScopeHost, observedHost)
548 f.p.sys = restartRequestShowError{f.u}
549 if err := f.run(t); err == nil || len(f.u.effects) != 1 {
550 t.Fatal(err, f.u.effects)
551 }
552 b := release.RestartBinding{Box: "box", Account: "bab", App: "console"}
553 pending, err := f.p.journal.Observe(b, release.RestartSnapshot{})
554 if err != nil || pending == nil {
555 t.Fatalf("observation failure consumed: %+v %v", pending, err)
556 }
557 f.p.sys = f.u
558 if err = f.run(t); err != nil || len(f.u.effects) != 1 {
559 t.Fatalf("observed successor replayed: %v %v", err, f.u.effects)
560 }
561}
562