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 "encoding/json"
5 "errors"
6 "os"
7 "os/exec"
8 "path/filepath"
9 "sync"
10 "testing"
11 "time"
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 "github.com/carlosframework/platform/internal/rootstate"
18)
19
20func TestPlan14HistoryRecoveryProcessCrash(t *testing.T) {
21 t.Parallel()
22 if boundary := os.Getenv("PLAN14_HISTORY_CRASH"); boundary != "" {
23 dir := os.Getenv("PLAN14_HISTORY_DIR")
24 e, rt, u, c, pub := plan14AdoptionFixture(t)
25 storeDir := filepath.Join(dir, "deployment")
26 rec, _, err := provision.FindRecordByHost(t.Context(), e.store, rt.Host)
27 if err != nil {
28 t.Fatal(err)
29 }
30 plan14PutRecord(t, release.NewDirStore(storeDir), *rec)
31 h := seedHostRootHistory(t, rt)
32 disk := &rollbackDisk{dir: filepath.Join(dir, "attempts")}
33 historyDisk := &rollbackDisk{dir: filepath.Join(dir, "history")}
34 for _, d := range []*rollbackDisk{disk, historyDisk} {
35 if err := os.Mkdir(d.dir, 0700); err != nil {
36 t.Fatal(err)
37 }
38 }
39 for k, raw := range h.Store.(*rollbackMemory).files {
40 if err := historyDisk.Write(k, raw); err != nil {
41 t.Fatal(err)
42 }
43 }
44 h.Store = historyDisk
45 deps := boundRootHistory(h, c.resolve, e.root.obs)
46 c.attempts = func() (rootstate.Store, error) { return disk, nil }
47 c.history = func() (*release.RootHistoryPass, error) { return deps, nil }
48 linkRootHistory(deps, c, u.opt, func(string) string { return "5s" })
49 crash := func() {
50 raw, err := json.Marshal(rollbackProcessFixture{Props: u.props, Exe: u.exe, Opt: u.opt, Bake: e.bake, Store: storeDir, Effects: u.effects})
51 if err != nil {
52 t.Fatal(err)
53 }
54 if err := os.WriteFile(filepath.Join(dir, "fixture.json"), raw, 0600); err != nil {
55 t.Fatal(err)
56 }
57 os.Exit(74)
58 }
59 disk.afterWrite = func(k string, raw []byte) {
60 if k != rollbackJournalKey(rt.Host) {
61 return
62 }
63 var a rollbackAttempt
64 if json.Unmarshal(raw, &a) != nil || a.History == nil {
65 return
66 }
67 if boundary == "linked" && !a.Failed && !a.Recovering && !a.Settled {
68 crash()
69 }
70 if boundary == "forward settled" && a.Settled {
71 crash()
72 }
73 if boundary == "rollback pending" && a.Recovering && !a.Settled {
74 crash()
75 }
76 }
77 historyDisk.afterWrite = func(k string, raw []byte) {
78 if boundary != "history settled" {
79 return
80 }
81 var r struct {
82 Binding release.HistoryBinding `json:"binding"`
83 Attempt *release.HistoryAttempt `json:"attempt"`
84 }
85 if json.Unmarshal(raw, &r) == nil && r.Binding.Host == rt.Host && r.Attempt == nil && len(u.effects) > 0 {
86 crash()
87 }
88 }
89 u.crashNew = boundary != "forward running" && boundary != "forward unhealthy" && boundary != "forward settled" && boundary != "history settled"
90 u.afterEffect = func() {
91 if (boundary == "forward failed" || boundary == "forward running" || boundary == "forward unhealthy" || boundary == "rollback fails") && len(u.effects) == 1 {
92 crash()
93 }
94 if boundary == "rollback running" && len(u.effects) == 3 {
95 crash()
96 }
97 }
98 forward := rootRestartRouteWith(e.root.reg, u, u.opt, func(string) string { return "5s" }, nil, c)
99 _ = release.Pass(t.Context(), e.root.reg, e.store, pub, u.opt, forward, release.WithRootHistory(deps), release.WithRestartFailurePropagation(), release.WithRollbackRestart(rollbackUnitRestart(e.root.reg, u, u.opt, func(string) string { return "5s" }, c)))
100 t.Fatal("crash boundary not reached")
101 }
102 for _, boundary := range []string{"linked", "forward failed", "forward running", "forward unhealthy", "forward settled", "rollback pending", "rollback running", "history settled", "rollback fails"} {
103 t.Run(boundary, func(t *testing.T) {
104 dir := t.TempDir()
105 cmd := exec.CommandContext(t.Context(), os.Args[0], "-test.run=^TestPlan14HistoryRecoveryProcessCrash$", "-test.count=1")
106 cmd.Dir = dir
107 cmd.Env = append(os.Environ(), "TMPDIR="+dir, "PLAN14_HISTORY_CRASH="+boundary, "PLAN14_HISTORY_DIR="+dir, "CARLOS_HOST_CLAIM=")
108 out, err := cmd.CombinedOutput()
109 var exit *exec.ExitError
110 if !errors.As(err, &exit) || exit.ExitCode() != 74 {
111 t.Fatalf("child: %v\n%s", err, out)
112 }
113 raw, err := os.ReadFile(filepath.Join(dir, "fixture.json"))
114 if err != nil {
115 t.Fatal(err)
116 }
117 var f rollbackProcessFixture
118 if err := json.Unmarshal(raw, &f); err != nil {
119 t.Fatal(err)
120 }
121 u := &plan14RestartUnits{props: f.Props, exe: f.Exe, opt: f.Opt, effects: f.Effects, badRollback: boundary == "rollback fails"}
122 disk := &rollbackDisk{dir: filepath.Join(dir, "attempts")}
123 source := &proofBindingSource{store: release.NewDirStore(f.Store), sys: u, optDir: u.opt, openFile: os.Open, unitDirs: []string{filepath.Dir(u.props.FragmentPath)}, healthy: func(acct, app, ver, digest string) *release.HealthyBuild {
124 for _, b := range release.ReadHealthy(f.Bake, acct, app) {
125 if b.Version == ver && b.SHA256 == digest {
126 return &b
127 }
128 }
129 return nil
130 }}
131 a, err := readRollbackAttempt(disk, observedHost)
132 if err != nil || a.History == nil {
133 t.Fatalf("missing linked intent: %+v %v", a, err)
134 }
135 now := a.Created.Add(time.Second)
136 ready := func(registry.Route) func() error {
137 return func() error {
138 if boundary == "forward unhealthy" && u.exe == release.VersionPath(u.opt, "bab", "v2", "console") {
139 return os.ErrPermission
140 }
141 return nil
142 }
143 }
144 h := &release.RootHistory{Store: &rollbackDisk{dir: filepath.Join(dir, "history")}}
145 obs := proofObserver{resolve: source.resolve, sys: u, optDir: u.opt, procExe: func(int) string { return u.exe }, now: time.Now, ready: ready}
146 deps := boundRootHistory(h, source.resolve, obs)
147 c := rootRestartChecks{history: func() (*release.RootHistoryPass, error) { return deps, nil }, started: &sync.Map{}, attempts: func() (rootstate.Store, error) { return disk, nil }, prepareRollback: source.prepareRollback, rollback: source.rollback, host: source.forHost, ready: ready, now: func() time.Time { return now }, sleep: func(d time.Duration) { now = now.Add(d) }}
148 before := len(u.effects)
149 err = recoverRootRestarts(t.Context(), u, u.opt, func(string) string { return "5s" }, c, hostwrite.Config{})
150 refs, refsErr := h.Refusals(a.HistoryBinding)
151 if boundary == "rollback fails" {
152 if err == nil || refsErr == nil {
153 t.Fatalf("failed rollback settled: %v %v", err, refsErr)
154 }
155 before := len(u.effects)
156 if err := recoverRootRestarts(t.Context(), u, u.opt, func(string) string { return "5s" }, c, hostwrite.Config{}); err != nil || len(u.effects) != before {
157 t.Fatalf("terminal rollback repeated: %v %v", err, u.effects)
158 }
159 if _, err := h.Refusals(a.HistoryBinding); err == nil {
160 t.Fatal("terminal failure erased history intent")
161 }
162 return
163 }
164 if err != nil || refsErr != nil {
165 t.Fatalf("resume: %v %v", err, refsErr)
166 }
167 forward := boundary == "forward running" || boundary == "forward settled" || boundary == "history settled"
168 if forward && (len(refs) != 0 || u.exe != release.VersionPath(u.opt, "bab", "v2", "console")) {
169 t.Fatalf("forward not recovered: %+v %s", refs, u.exe)
170 }
171 if !forward && (len(refs) != 1 || !refs[0].Recovered || refs[0].Version != "v2" || refs[0].RestoredVersion != "v1") {
172 t.Fatalf("rollback deny missing: %+v", refs)
173 }
174 if boundary != "forward failed" && boundary != "forward unhealthy" && boundary != "rollback pending" && len(u.effects) != before {
175 t.Fatal("recovery repeated an observed effect")
176 }
177 before = len(u.effects)
178 if err := recoverRootRestarts(t.Context(), u, u.opt, func(string) string { return "5s" }, c, hostwrite.Config{}); err != nil || len(u.effects) != before {
179 t.Fatalf("replay: %v", err)
180 }
181 })
182 }
183}
184