| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "errors" |
| 5 | "os" |
| 6 | "path/filepath" |
| 7 | "testing" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/registry" |
| 11 | "github.com/carlosframework/platform/internal/release" |
| 12 | ) |
| 13 | |
| 14 | type fakeLister struct { |
| 15 | routes []registry.Route |
| 16 | err error |
| 17 | } |
| 18 | |
| 19 | func (f fakeLister) List() ([]registry.Route, error) { return f.routes, f.err } |
| 20 | |
| 21 | func stageSweepVersion(t *testing.T, optDir, acct, ver, app string, age time.Duration) { |
| 22 | t.Helper() |
| 23 | p := release.VersionPath(optDir, acct, ver, app) |
| 24 | if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { |
| 25 | t.Fatal(err) |
| 26 | } |
| 27 | if err := os.WriteFile(p, []byte("binary"), 0o755); err != nil { |
| 28 | t.Fatal(err) |
| 29 | } |
| 30 | when := time.Now().Add(-age) |
| 31 | if err := os.Chtimes(p, when, when); err != nil { |
| 32 | t.Fatal(err) |
| 33 | } |
| 34 | } |
| 35 | |
| 36 | // noGenerations is a generations file path with neither file present: no |
| 37 | // generations, every box before its agent first published. |
| 38 | func noGenerations(t *testing.T) string { |
| 39 | return release.GenerationsPath(t.TempDir()) |
| 40 | } |
| 41 | |
| 42 | func staged(t *testing.T, optDir, ver string) bool { |
| 43 | t.Helper() |
| 44 | _, err := os.Stat(filepath.Join(optDir, "versions", ver)) |
| 45 | return err == nil |
| 46 | } |
| 47 | |
| 48 | // PrevVersion is a pin with no grace: a rollback is wanted precisely when the |
| 49 | // new build has been running long enough for the grace to have expired. |
| 50 | func TestSweepStagedVersionsPinsPrevVersion(t *testing.T) { |
| 51 | opt := t.TempDir() |
| 52 | stageSweepVersion(t, opt, "acct1", "v3", "blog", 48*time.Hour) |
| 53 | stageSweepVersion(t, opt, "acct1", "v2", "blog", 48*time.Hour) |
| 54 | stageSweepVersion(t, opt, "acct1", "v1", "blog", 48*time.Hour) |
| 55 | |
| 56 | reg := fakeLister{routes: []registry.Route{{Host: "blog.example.com", Version: "v3", PrevVersion: "v2"}}} |
| 57 | sweepStagedVersions(reg, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| 58 | |
| 59 | if !staged(t, opt, "v3") { |
| 60 | t.Error("v3 (Version) was removed") |
| 61 | } |
| 62 | if !staged(t, opt, "v2") { |
| 63 | t.Error("v2 (PrevVersion) was removed — the rollback target is a pin") |
| 64 | } |
| 65 | if staged(t, opt, "v1") { |
| 66 | t.Error("v1 survived; it is pinned by nothing and older than the grace") |
| 67 | } |
| 68 | } |
| 69 | |
| 70 | // A route with no version at all must not pin the empty string into |
| 71 | // something the sweep then matches. |
| 72 | func TestSweepStagedVersionsIgnoresEmptyVersions(t *testing.T) { |
| 73 | opt := t.TempDir() |
| 74 | stageSweepVersion(t, opt, "acct1", "v1", "blog", 48*time.Hour) |
| 75 | reg := fakeLister{routes: []registry.Route{{Host: "pending.example.com"}}} |
| 76 | sweepStagedVersions(reg, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| 77 | if staged(t, opt, "v1") { |
| 78 | t.Error("v1 survived a pin set that should be empty") |
| 79 | } |
| 80 | } |
| 81 | |
| 82 | // An unlistable registry means an incomplete pin set, and an incomplete pin |
| 83 | // set is how a running app loses its binary. Delete nothing. |
| 84 | func TestSweepStagedVersionsDeletesNothingWhenRegistryFails(t *testing.T) { |
| 85 | opt := t.TempDir() |
| 86 | stageSweepVersion(t, opt, "acct1", "v1", "blog", 48*time.Hour) |
| 87 | sweepStagedVersions(fakeLister{err: errors.New("database is locked")}, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| 88 | if !staged(t, opt, "v1") { |
| 89 | t.Error("swept with an unreadable registry") |
| 90 | } |
| 91 | } |
| 92 | |
| 93 | // Likewise for an unreadable live dir. |
| 94 | func TestSweepStagedVersionsDeletesNothingWhenLivePinsFail(t *testing.T) { |
| 95 | opt := t.TempDir() |
| 96 | stageSweepVersion(t, opt, "acct1", "v1", "blog", 48*time.Hour) |
| 97 | if err := os.MkdirAll(filepath.Join(opt, "live"), 0o755); err != nil { |
| 98 | t.Fatal(err) |
| 99 | } |
| 100 | if err := os.WriteFile(filepath.Join(opt, "live", "stray"), []byte("not a link"), 0o644); err != nil { |
| 101 | t.Fatal(err) |
| 102 | } |
| 103 | sweepStagedVersions(fakeLister{}, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| 104 | if !staged(t, opt, "v1") { |
| 105 | t.Error("swept with an unreadable live dir") |
| 106 | } |
| 107 | } |
| 108 | |
| 109 | // The live link is believed over the registry: PointLive succeeding and |
| 110 | // SetVersions then failing on a full /data is the incident's own state, and |
| 111 | // the link is the only record of what is running. |
| 112 | func TestSweepStagedVersionsBelievesLiveLinkOverRegistry(t *testing.T) { |
| 113 | opt := t.TempDir() |
| 114 | stageSweepVersion(t, opt, release.OpsAccount, "v2", "edge", 48*time.Hour) |
| 115 | if err := release.PointLive(opt, "edge.example.com", release.OpsAccount, "v2", "edge"); err != nil { |
| 116 | t.Fatal(err) |
| 117 | } |
| 118 | // The registry still says v1 — the version write never landed. |
| 119 | reg := fakeLister{routes: []registry.Route{{Host: "edge.example.com", Version: "v1"}}} |
| 120 | sweepStagedVersions(reg, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| 121 | if !staged(t, opt, "v2") { |
| 122 | t.Fatal("the running version was removed because the registry had not caught up") |
| 123 | } |
| 124 | } |
| 125 | |
| 126 | // Ruling R19 item 1 (Astra r3 F1(a)): a worker's captured build that only |
| 127 | // the generations file still names (by version, or by staged path alone) |
| 128 | // survives the sweep past its grace. What nothing names is still removed. |
| 129 | func TestSweepStagedVersionsKeepsBuildsTheGenerationsFileNames(t *testing.T) { |
| 130 | opt := t.TempDir() |
| 131 | for _, v := range []string{"v0", "v1", "v2", "v3"} { |
| 132 | stageSweepVersion(t, opt, "acct1", v, "bot", 48*time.Hour) |
| 133 | } |
| 134 | gen := noGenerations(t) |
| 135 | if err := release.WriteGenerations(gen, release.Generations{ |
| 136 | Hosts: map[string][]string{"bot.example.com": {"v1"}}, |
| 137 | Paths: map[string][]string{"chained.example.com": {release.VersionPath(opt, "acct1", "v2", "bot")}}, |
| 138 | }); err != nil { |
| 139 | t.Fatal(err) |
| 140 | } |
| 141 | reg := fakeLister{routes: []registry.Route{{Host: "bot.example.com", Version: "v3"}}} |
| 142 | sweepStagedVersions(reg, opt, gen, 6*time.Hour, time.Now()) |
| 143 | if !staged(t, opt, "v1") { |
| 144 | t.Error("v1 was removed; the generations file names it as a running worker's build") |
| 145 | } |
| 146 | if !staged(t, opt, "v2") { |
| 147 | t.Error("v2 was removed; the generations file names its staged path") |
| 148 | } |
| 149 | if !staged(t, opt, "v3") { |
| 150 | t.Error("v3 (Version) was removed") |
| 151 | } |
| 152 | if staged(t, opt, "v0") { |
| 153 | t.Error("v0 survived; nothing names it and it is older than the grace") |
| 154 | } |
| 155 | } |
| 156 | |
| 157 | // Ruling R19 item 1: a generations file that does not read, or whose |
| 158 | // publication did not finish (the intent ahead of the file), is an incomplete |
| 159 | // pin set, and the sweep removes nothing that pass. |
| 160 | func TestSweepStagedVersionsDeletesNothingWhenTheGenerationsFileIsIncomplete(t *testing.T) { |
| 161 | for name, files := range map[string][2]string{ |
| 162 | "unreadable": {"", `{"schema":1,"hosts":{"bot.exa`}, |
| 163 | "intent ahead": {"2", `{"schema":1,"seq":1,"complete":true,"hosts":{}}`}, |
| 164 | } { |
| 165 | t.Run(name, func(t *testing.T) { |
| 166 | opt := t.TempDir() |
| 167 | stageSweepVersion(t, opt, "acct1", "v1", "bot", 48*time.Hour) |
| 168 | gen := noGenerations(t) |
| 169 | if err := os.MkdirAll(filepath.Dir(gen), 0o755); err != nil { |
| 170 | t.Fatal(err) |
| 171 | } |
| 172 | if files[0] != "" { |
| 173 | if err := os.WriteFile(release.GenerationsSeqPath(gen), []byte(files[0]), 0o644); err != nil { |
| 174 | t.Fatal(err) |
| 175 | } |
| 176 | } |
| 177 | if err := os.WriteFile(gen, []byte(files[1]), 0o644); err != nil { |
| 178 | t.Fatal(err) |
| 179 | } |
| 180 | sweepStagedVersions(fakeLister{}, opt, gen, 6*time.Hour, time.Now()) |
| 181 | if !staged(t, opt, "v1") { |
| 182 | t.Error("swept with an incomplete generations file") |
| 183 | } |
| 184 | }) |
| 185 | } |
| 186 | } |
| 187 | |
| 188 | func TestVersionSweepGrace(t *testing.T) { |
| 189 | env := func(v string) func(string) string { |
| 190 | return func(k string) string { |
| 191 | if k == versionSweepGraceEnv { |
| 192 | return v |
| 193 | } |
| 194 | return "" |
| 195 | } |
| 196 | } |
| 197 | if got := versionSweepGrace(env("")); got != release.DefaultSweepGrace { |
| 198 | t.Errorf("unset = %s, want %s", got, release.DefaultSweepGrace) |
| 199 | } |
| 200 | if got := versionSweepGrace(env("30m")); got != 30*time.Minute { |
| 201 | t.Errorf("30m = %s", got) |
| 202 | } |
| 203 | // Garbage and negatives fall back rather than sweeping with a zero grace. |
| 204 | for _, bad := range []string{"soon", "-1h"} { |
| 205 | if got := versionSweepGrace(env(bad)); got != release.DefaultSweepGrace { |
| 206 | t.Errorf("%q = %s, want the default %s", bad, got, release.DefaultSweepGrace) |
| 207 | } |
| 208 | } |
| 209 | } |
| 210 | |
| 211 | // The pass lock is what makes the sweep safe against a concurrent |
| 212 | // carlos-stage.service, so it has to actually exclude. |
| 213 | func TestLockAdoptPassExcludes(t *testing.T) { |
| 214 | t.Parallel() |
| 215 | dd := t.TempDir() |
| 216 | unlock, err := lockAdoptPass(dd) |
| 217 | if err != nil { |
| 218 | t.Fatal(err) |
| 219 | } |
| 220 | held := make(chan struct{}) |
| 221 | go func() { |
| 222 | second, err := lockAdoptPass(dd) |
| 223 | if err == nil { |
| 224 | second() |
| 225 | } |
| 226 | close(held) |
| 227 | }() |
| 228 | select { |
| 229 | case <-held: |
| 230 | t.Fatal("a second pass took the lock while the first held it") |
| 231 | case <-time.After(100 * time.Millisecond): |
| 232 | } |
| 233 | unlock() |
| 234 | select { |
| 235 | case <-held: |
| 236 | case <-time.After(5 * time.Second): |
| 237 | t.Fatal("the second pass never got the lock after the first released it") |
| 238 | } |
| 239 | } |
| 240 | |
| 241 | // The adopt lock and the restart marks lock are different files: RunAdopt |
| 242 | // holds the first for the whole pass and restartPass takes the second inside |
| 243 | // it, so sharing an inode would deadlock every pass that restarts anything. |
| 244 | func TestAdoptAndRestartLocksAreDistinct(t *testing.T) { |
| 245 | dd := t.TempDir() |
| 246 | unlock, err := lockAdoptPass(dd) |
| 247 | if err != nil { |
| 248 | t.Fatal(err) |
| 249 | } |
| 250 | defer unlock() |
| 251 | inner, err := lockRestartMarks(restartMarksPath(dd)) |
| 252 | if err != nil { |
| 253 | t.Fatalf("restart marks lock blocked by the pass lock: %v", err) |
| 254 | } |
| 255 | inner() |
| 256 | } |
| 257 | |
| 258 | // Spec §8.5: the good build is a pin, with no grace, like PrevVersion: it is |
| 259 | // the fallback a failed first wake boots, however long ago it last served. |
| 260 | func TestSweepStagedVersionsKeepsTheGoodBuild(t *testing.T) { |
| 261 | opt := t.TempDir() |
| 262 | for _, v := range []string{"good", "gone"} { |
| 263 | stageSweepVersion(t, opt, "acct1", v, "km", 48*time.Hour) |
| 264 | } |
| 265 | reg := fakeLister{routes: []registry.Route{{Host: "a.example.com", Kind: "instance", App: "km", Account: "acct1", Version: "x", GoodVersion: "good"}}} |
| 266 | sweepStagedVersions(reg, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| 267 | if !staged(t, opt, "good") { |
| 268 | t.Error("the good build was removed") |
| 269 | } |
| 270 | if staged(t, opt, "gone") { |
| 271 | t.Error("gone survived; nothing names it and it is older than the grace") |
| 272 | } |
| 273 | } |
| 274 | |
| 275 | // settingsRow is an activator-managed instance with the given build-state |
| 276 | // copy and update-settings columns. |
| 277 | func settingsRow(host, buildState, settings string) registry.Route { |
| 278 | return registry.Route{Host: host, Kind: "instance", App: "blog", Account: "acct1", Hibernate: true, Socket: "/run/carlos/" + host + ".sock", |
| 279 | Version: "v9", BuildState: buildState, UpdateSettings: settings} |
| 280 | } |
| 281 | |
| 282 | // Ruling R-T9-STORE: the box-local sweep keeps every build a build state |
| 283 | // names (last_executed and a resolved request's T) and every pin, so a held |
| 284 | // instance's staged recovery copy (EvidenceStaged) is never evicted. |
| 285 | func TestSweepStagedVersionsKeepsBuildStateBuildsAndPins(t *testing.T) { |
| 286 | opt := t.TempDir() |
| 287 | for _, v := range []string{"v1", "v2", "v3", "v4", "v9"} { |
| 288 | stageSweepVersion(t, opt, "acct1", v, "blog", 48*time.Hour) |
| 289 | } |
| 290 | copyCol := release.BuildStateCopy{Known: true, Replica: true, |
| 291 | State: &release.BuildState{LastExecuted: "v1", Resolved: &release.ResolvedRequest{ID: 2, Target: "v2"}}}.Column() |
| 292 | reg := fakeLister{routes: []registry.Route{ |
| 293 | settingsRow("a.example.com", copyCol, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable"}.Column()), |
| 294 | settingsRow("b.example.com", release.BuildStateCopy{Known: true}.Column(), release.UpdateSettings{Updates: release.UpdatesAuto, Channel: "stable", Pin: "v3"}.Column()), |
| 295 | }} |
| 296 | sweepStagedVersions(reg, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| 297 | for _, v := range []string{"v1", "v2", "v3", "v9"} { |
| 298 | if !staged(t, opt, v) { |
| 299 | t.Errorf("%s was removed", v) |
| 300 | } |
| 301 | } |
| 302 | if staged(t, opt, "v4") { |
| 303 | t.Error("v4 survived; nothing names it") |
| 304 | } |
| 305 | } |
| 306 | |
| 307 | // A copy that keeps its last known state after a failed refresh still pins |
| 308 | // it: the build it names did not stop being the instance's. |
| 309 | func TestSweepStagedVersionsKeepsALastKnownState(t *testing.T) { |
| 310 | opt := t.TempDir() |
| 311 | stageSweepVersion(t, opt, "acct1", "v1", "blog", 48*time.Hour) |
| 312 | col := release.BuildStateCopy{Known: true, ReadErr: "timeout", State: &release.BuildState{LastExecuted: "v1"}}.Column() |
| 313 | sweepStagedVersions(fakeLister{routes: []registry.Route{settingsRow("a.example.com", col, "")}}, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| 314 | if !staged(t, opt, "v1") { |
| 315 | t.Error("the last known last_executed was removed after a failed refresh") |
| 316 | } |
| 317 | } |
| 318 | |
| 319 | // Astra round 1 finding 6: a held instance's copy that kept A after a failed |
| 320 | // refresh is not a complete pin set. The durable state may name B since (a |
| 321 | // recovery onto B on another box, then a return), and B's staged copy here is |
| 322 | // its only recovery copy, named by no live link, column or generation. The |
| 323 | // sweep deletes nothing, and so keeps A too. A default-settings row's failed |
| 324 | // refresh holds nothing on its last_executed, so it does not stop the sweep. |
| 325 | func TestSweepStagedVersionsRefusesAHeldCopyWithAReadError(t *testing.T) { |
| 326 | opt := t.TempDir() |
| 327 | for _, v := range []string{"A", "B", "v9"} { |
| 328 | stageSweepVersion(t, opt, "acct1", v, "blog", 48*time.Hour) |
| 329 | } |
| 330 | manual := release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable"}.Column() |
| 331 | col := release.BuildStateCopy{Known: true, Replica: true, ReadErr: "s3: timeout", State: &release.BuildState{LastExecuted: "A"}}.Column() |
| 332 | sweepStagedVersions(fakeLister{routes: []registry.Route{settingsRow("a.example.com", col, manual)}}, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| 333 | for _, v := range []string{"A", "B"} { |
| 334 | if !staged(t, opt, v) { |
| 335 | t.Errorf("%s was removed: the sweep ran on a held copy whose last refresh failed", v) |
| 336 | } |
| 337 | } |
| 338 | opt = t.TempDir() |
| 339 | for _, v := range []string{"A", "B", "v9"} { |
| 340 | stageSweepVersion(t, opt, "acct1", v, "blog", 48*time.Hour) |
| 341 | } |
| 342 | sweepStagedVersions(fakeLister{routes: []registry.Route{settingsRow("a.example.com", col, "")}}, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| 343 | if !staged(t, opt, "A") || staged(t, opt, "B") { |
| 344 | t.Errorf("default settings: A staged %v, B staged %v; want A kept (named) and B swept", staged(t, opt, "A"), staged(t, opt, "B")) |
| 345 | } |
| 346 | } |
| 347 | |
| 348 | // Any column the sweep cannot read is an incomplete pin set: an unparseable |
| 349 | // build-state copy or update-settings column, or a held instance whose build |
| 350 | // state is not read yet (its last_executed is unknown). Delete nothing. |
| 351 | func TestSweepStagedVersionsDeletesNothingWhenSettingsPinsAreUnknown(t *testing.T) { |
| 352 | manual := release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable"}.Column() |
| 353 | cases := map[string]registry.Route{ |
| 354 | "unparseable build state": settingsRow("a.example.com", "{", ""), |
| 355 | "unparseable update settings": settingsRow("a.example.com", "", "{"), |
| 356 | "held, build state unread": settingsRow("a.example.com", "", manual), |
| 357 | "pinned, build state unread": settingsRow("a.example.com", "", release.UpdateSettings{Updates: release.UpdatesAuto, Channel: "stable", Pin: "v1"}.Column()), |
| 358 | } |
| 359 | for name, rt := range cases { |
| 360 | t.Run(name, func(t *testing.T) { |
| 361 | opt := t.TempDir() |
| 362 | stageSweepVersion(t, opt, "acct1", "v1", "blog", 48*time.Hour) |
| 363 | sweepStagedVersions(fakeLister{routes: []registry.Route{rt}}, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| 364 | if !staged(t, opt, "v1") { |
| 365 | t.Error("swept with a pin set it could not complete") |
| 366 | } |
| 367 | }) |
| 368 | } |
| 369 | // Rows whose unread copy says nothing the sweep needs: default settings |
| 370 | // (nothing is held on its last_executed), and rows keepsBuildState never |
| 371 | // refreshes, a named unit or the platform's own row, whatever settings |
| 372 | // an app policy gives them (their copy stays unread forever). The sweep |
| 373 | // goes on. |
| 374 | unit := settingsRow("u.example.com", "", manual) |
| 375 | unit.Unit = "blog.service" |
| 376 | platform := settingsRow("p.example.com", "", manual) |
| 377 | platform.Platform = true |
| 378 | worker := settingsRow("w.example.com", "", manual) |
| 379 | worker.Kind, worker.Hibernate = "sidecar", false |
| 380 | goesOn := map[string]registry.Route{ |
| 381 | "worker, settings set": worker, |
| 382 | "default settings": settingsRow("a.example.com", "", ""), |
| 383 | "named unit, held": unit, |
| 384 | "platform row, held": platform, |
| 385 | } |
| 386 | for name, rt := range goesOn { |
| 387 | t.Run(name, func(t *testing.T) { |
| 388 | opt := t.TempDir() |
| 389 | stageSweepVersion(t, opt, "acct1", "v1", "blog", 48*time.Hour) |
| 390 | sweepStagedVersions(fakeLister{routes: []registry.Route{rt}}, opt, noGenerations(t), 6*time.Hour, time.Now()) |
| 391 | if staged(t, opt, "v1") { |
| 392 | t.Error("a row whose unread copy pins nothing stopped the sweep") |
| 393 | } |
| 394 | }) |
| 395 | } |
| 396 | } |
| 397 | |
amadan