| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "os" |
| 8 | "os/exec" |
| 9 | "path/filepath" |
| 10 | "sync" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/hostwrite" |
| 15 | "github.com/carlosframework/platform/internal/provision" |
| 16 | "github.com/carlosframework/platform/internal/registry" |
| 17 | "github.com/carlosframework/platform/internal/release" |
| 18 | "github.com/carlosframework/platform/internal/rootstate" |
| 19 | ) |
| 20 | |
| 21 | type rollbackDisk struct { |
| 22 | rootstate.Store |
| 23 | dir string |
| 24 | afterWrite func(string, []byte) |
| 25 | beforeWrite func(string, []byte) |
| 26 | } |
| 27 | |
| 28 | func (s *rollbackDisk) Read(k string) ([]byte, error) { return os.ReadFile(filepath.Join(s.dir, k)) } |
| 29 | func (s *rollbackDisk) Write(k string, b []byte) error { |
| 30 | if s.beforeWrite != nil { |
| 31 | s.beforeWrite(k, b) |
| 32 | } |
| 33 | f, err := os.CreateTemp(s.dir, "pending-") |
| 34 | if err != nil { |
| 35 | return err |
| 36 | } |
| 37 | n := f.Name() |
| 38 | defer os.Remove(n) |
| 39 | if _, err = f.Write(b); err == nil { |
| 40 | err = f.Sync() |
| 41 | } |
| 42 | err = errors.Join(err, f.Close()) |
| 43 | if err != nil { |
| 44 | return err |
| 45 | } |
| 46 | if err = os.Rename(n, filepath.Join(s.dir, k)); err != nil { |
| 47 | return err |
| 48 | } |
| 49 | d, err := os.Open(s.dir) |
| 50 | if err != nil { |
| 51 | return err |
| 52 | } |
| 53 | err = errors.Join(d.Sync(), d.Close()) |
| 54 | if err != nil { |
| 55 | return err |
| 56 | } |
| 57 | if s.afterWrite != nil { |
| 58 | s.afterWrite(k, b) |
| 59 | } |
| 60 | return nil |
| 61 | } |
| 62 | func (s *rollbackDisk) Remove(k string) error { |
| 63 | err := os.Remove(filepath.Join(s.dir, k)) |
| 64 | if errors.Is(err, os.ErrNotExist) { |
| 65 | return nil |
| 66 | } |
| 67 | return err |
| 68 | } |
| 69 | func (s *rollbackDisk) Names() ([]string, error) { |
| 70 | es, err := os.ReadDir(s.dir) |
| 71 | var out []string |
| 72 | for _, e := range es { |
| 73 | out = append(out, e.Name()) |
| 74 | } |
| 75 | return out, err |
| 76 | } |
| 77 | func (s *rollbackDisk) Lock(context.Context, string) (func(), error) { return func() {}, nil } |
| 78 | func (s *rollbackDisk) Close() error { return nil } |
| 79 | |
| 80 | type rollbackProcessFixture struct { |
| 81 | Props unitProps |
| 82 | Exe, Opt, Bake, Store string |
| 83 | Effects []string |
| 84 | } |
| 85 | |
| 86 | func TestPlan14RollbackProcessCrash(t *testing.T) { |
| 87 | t.Parallel() |
| 88 | if boundary := os.Getenv("PLAN14_CRASH_BOUNDARY"); boundary != "" { |
| 89 | dir := os.Getenv("PLAN14_CRASH_DIR") |
| 90 | e, rt, u, c, pub := plan14AdoptionFixture(t) |
| 91 | storeDir := filepath.Join(dir, "deployment") |
| 92 | store := release.NewDirStore(storeDir) |
| 93 | rec, _, err := provision.FindRecordByHost(t.Context(), e.store, rt.Host) |
| 94 | if err != nil { |
| 95 | t.Fatal(err) |
| 96 | } |
| 97 | plan14PutRecord(t, store, *rec) |
| 98 | disk := &rollbackDisk{dir: filepath.Join(dir, "attempts")} |
| 99 | if err := os.Mkdir(disk.dir, 0700); err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | c.attempts = func() (rootstate.Store, error) { return disk, nil } |
| 103 | crash := func() { |
| 104 | b, err := json.Marshal(rollbackProcessFixture{Props: u.props, Exe: u.exe, Opt: u.opt, Bake: e.bake, Store: storeDir, Effects: u.effects}) |
| 105 | if err != nil { |
| 106 | t.Fatal(err) |
| 107 | } |
| 108 | if err := os.WriteFile(filepath.Join(dir, "fixture.json"), b, 0600); err != nil { |
| 109 | t.Fatal(err) |
| 110 | } |
| 111 | os.Exit(73) |
| 112 | } |
| 113 | disk.afterWrite = func(k string, b []byte) { |
| 114 | if k != rollbackJournalKey(rt.Host) { |
| 115 | return |
| 116 | } |
| 117 | var a rollbackAttempt |
| 118 | if json.Unmarshal(b, &a) != nil { |
| 119 | return |
| 120 | } |
| 121 | if boundary == "captured" && !a.Failed && !a.Recovering && !a.Settled { |
| 122 | crash() |
| 123 | } |
| 124 | if boundary == "failed journal" && a.Failed && !a.Recovering && !a.Settled { |
| 125 | crash() |
| 126 | } |
| 127 | if boundary == "rollback pending" && a.Recovering && !a.Settled { |
| 128 | crash() |
| 129 | } |
| 130 | if boundary == "settled" && a.Settled { |
| 131 | crash() |
| 132 | } |
| 133 | } |
| 134 | u.crashNew = boundary != "forward running" |
| 135 | u.afterEffect = func() { |
| 136 | if (boundary == "forward restart" || boundary == "forward running") && len(u.effects) == 1 { |
| 137 | crash() |
| 138 | } |
| 139 | if boundary == "rollback running" && len(u.effects) == 3 { |
| 140 | crash() |
| 141 | } |
| 142 | } |
| 143 | forward := rootRestartRouteWith(e.root.reg, u, u.opt, func(string) string { return "5s" }, nil, c) |
| 144 | _ = release.Pass(t.Context(), e.root.reg, e.store, pub, u.opt, forward, release.WithRestartFailurePropagation(), release.WithRollbackRestart(rollbackUnitRestart(e.root.reg, u, u.opt, func(string) string { return "5s" }, c))) |
| 145 | t.Fatal("crash boundary not reached") |
| 146 | } |
| 147 | for _, boundary := range []string{"captured", "forward running", "forward restart", "failed journal", "rollback pending", "rollback running", "settled"} { |
| 148 | t.Run(boundary, func(t *testing.T) { |
| 149 | dir := t.TempDir() |
| 150 | ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) |
| 151 | defer cancel() |
| 152 | cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestPlan14RollbackProcessCrash$", "-test.count=1") |
| 153 | cmd.Dir = dir |
| 154 | cmd.Env = append(os.Environ(), "TMPDIR="+dir, "PLAN14_CRASH_BOUNDARY="+boundary, "PLAN14_CRASH_DIR="+dir, "CARLOS_HOST_CLAIM=") |
| 155 | out, err := cmd.CombinedOutput() |
| 156 | var exit *exec.ExitError |
| 157 | if !errors.As(err, &exit) || exit.ExitCode() != 73 { |
| 158 | t.Fatalf("subprocess did not crash at %s: %v\n%s", boundary, err, out) |
| 159 | } |
| 160 | body, err := os.ReadFile(filepath.Join(dir, "fixture.json")) |
| 161 | if err != nil { |
| 162 | t.Fatal(err) |
| 163 | } |
| 164 | var f rollbackProcessFixture |
| 165 | if err := json.Unmarshal(body, &f); err != nil { |
| 166 | t.Fatal(err) |
| 167 | } |
| 168 | u := &plan14RestartUnits{props: f.Props, exe: f.Exe, opt: f.Opt, effects: f.Effects} |
| 169 | disk := &rollbackDisk{dir: filepath.Join(dir, "attempts")} |
| 170 | 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 { |
| 171 | for _, b := range release.ReadHealthy(f.Bake, acct, app) { |
| 172 | if b.Version == ver && b.SHA256 == digest { |
| 173 | return &b |
| 174 | } |
| 175 | } |
| 176 | return nil |
| 177 | }} |
| 178 | a, err := readRollbackAttempt(disk, observedHost) |
| 179 | if err != nil { |
| 180 | t.Fatal(err) |
| 181 | } |
| 182 | now := a.Created.Add(time.Second) |
| 183 | c := rootRestartChecks{targetFile: os.Open, started: &sync.Map{}, attempts: func() (rootstate.Store, error) { return disk, nil }, prepareRollback: source.prepareRollback, rollback: source.rollback, host: source.forHost, ready: func(registry.Route) func() error { return func() error { return nil } }, now: func() time.Time { return now }, sleep: func(d time.Duration) { now = now.Add(d) }} |
| 184 | before := len(u.effects) |
| 185 | err = recoverRootRestarts(t.Context(), u, u.opt, func(string) string { return "5s" }, c, hostwrite.Config{}) |
| 186 | if boundary == "captured" { |
| 187 | if err == nil || len(u.effects) != before { |
| 188 | t.Fatalf("unstarted update restarted: %v %v", err, u.effects) |
| 189 | } |
| 190 | } else { |
| 191 | if err != nil { |
| 192 | t.Fatal(err) |
| 193 | } |
| 194 | want := "v1" |
| 195 | if boundary == "forward running" { |
| 196 | want = "v2" |
| 197 | } |
| 198 | if u.exe != release.VersionPath(u.opt, "bab", want, "console") || u.props.MainPID <= 0 { |
| 199 | t.Fatalf("no recovered process: %+v", u) |
| 200 | } |
| 201 | if (boundary == "forward running" || boundary == "rollback running" || boundary == "settled") && len(u.effects) != before { |
| 202 | t.Fatalf("restarted twice: %v", u.effects) |
| 203 | } |
| 204 | } |
| 205 | a, err = readRollbackAttempt(disk, observedHost) |
| 206 | if err != nil || !a.Settled { |
| 207 | t.Fatalf("unsettled recovery: %+v %v", a, err) |
| 208 | } |
| 209 | if _, err := disk.Read(rollbackCapabilityKey(a.Host, a.ID)); !errors.Is(err, os.ErrNotExist) { |
| 210 | t.Fatal("capability survived recovery") |
| 211 | } |
| 212 | before = len(u.effects) |
| 213 | if err := recoverRootRestarts(t.Context(), u, u.opt, func(string) string { return "5s" }, c, hostwrite.Config{}); err != nil || len(u.effects) != before { |
| 214 | t.Fatalf("settled replay: %v %v", err, u.effects) |
| 215 | } |
| 216 | }) |
| 217 | } |
| 218 | } |
| 219 | |
amadan