| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "os" |
| 8 | "slices" |
| 9 | "sort" |
| 10 | "strings" |
| 11 | "sync" |
| 12 | "sync/atomic" |
| 13 | "testing" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/appconfig" |
| 17 | "github.com/carlosframework/platform/internal/domains" |
| 18 | "github.com/carlosframework/platform/internal/provision" |
| 19 | "github.com/carlosframework/platform/internal/registry" |
| 20 | "github.com/carlosframework/platform/internal/release" |
| 21 | "github.com/carlosframework/platform/internal/telemetry" |
| 22 | ) |
| 23 | |
| 24 | // ciCountingStore is a DirStore with the change index on that records every |
| 25 | // read by method and key. Embedding the concrete store forwards every |
| 26 | // optional extension (ConditionalGetter and ETagLister included), so the |
| 27 | // view and the watch see the store production gives them. |
| 28 | type ciCountingStore struct { |
| 29 | *release.DirStore |
| 30 | mu sync.Mutex |
| 31 | calls []string |
| 32 | failList string // List and ListETags of exactly this prefix fail |
| 33 | // stall, when set, holds every GetIfChanged of the change index until |
| 34 | // the caller's ctx ends or stall is closed: a hung index GET while the |
| 35 | // rest of the bucket answers. |
| 36 | stall chan struct{} |
| 37 | } |
| 38 | |
| 39 | // stallIndex hangs the index GET until the returned func is called. |
| 40 | func (s *ciCountingStore) stallIndex() (release func()) { |
| 41 | ch := make(chan struct{}) |
| 42 | s.mu.Lock() |
| 43 | s.stall = ch |
| 44 | s.mu.Unlock() |
| 45 | return func() { |
| 46 | s.mu.Lock() |
| 47 | s.stall = nil |
| 48 | s.mu.Unlock() |
| 49 | close(ch) |
| 50 | } |
| 51 | } |
| 52 | |
| 53 | func (s *ciCountingStore) setFailList(p string) { s.mu.Lock(); s.failList = p; s.mu.Unlock() } |
| 54 | |
| 55 | func (s *ciCountingStore) listFails(prefix string) bool { |
| 56 | s.mu.Lock() |
| 57 | defer s.mu.Unlock() |
| 58 | return s.failList != "" && s.failList == prefix |
| 59 | } |
| 60 | |
| 61 | func newCICountingStore(t *testing.T) *ciCountingStore { |
| 62 | t.Helper() |
| 63 | d := release.NewDirStore(t.TempDir()) |
| 64 | d.EnableChangeIndex() |
| 65 | return &ciCountingStore{DirStore: d} |
| 66 | } |
| 67 | |
| 68 | func (s *ciCountingStore) note(method, key string) { |
| 69 | s.mu.Lock() |
| 70 | s.calls = append(s.calls, method+" "+key) |
| 71 | s.mu.Unlock() |
| 72 | } |
| 73 | |
| 74 | func (s *ciCountingStore) reset() { s.mu.Lock(); s.calls = nil; s.mu.Unlock() } |
| 75 | |
| 76 | func (s *ciCountingStore) got() []string { |
| 77 | s.mu.Lock() |
| 78 | defer s.mu.Unlock() |
| 79 | out := slices.Clone(s.calls) |
| 80 | sort.Strings(out) |
| 81 | return out |
| 82 | } |
| 83 | |
| 84 | // count is how many recorded calls start with method+" " and name a key |
| 85 | // under prefix. |
| 86 | func (s *ciCountingStore) count(method, prefix string) int { |
| 87 | n := 0 |
| 88 | for _, c := range s.got() { |
| 89 | if strings.HasPrefix(c, method+" "+prefix) { |
| 90 | n++ |
| 91 | } |
| 92 | } |
| 93 | return n |
| 94 | } |
| 95 | |
| 96 | func (s *ciCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 97 | s.note("Get", key) |
| 98 | return s.DirStore.Get(ctx, key) |
| 99 | } |
| 100 | |
| 101 | func (s *ciCountingStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) { |
| 102 | s.note("GetIfChanged", key) |
| 103 | s.mu.Lock() |
| 104 | stall := s.stall |
| 105 | s.mu.Unlock() |
| 106 | if stall != nil && key == release.ChangeIndexKey { |
| 107 | select { |
| 108 | case <-ctx.Done(): |
| 109 | return nil, "", false, ctx.Err() |
| 110 | case <-stall: |
| 111 | } |
| 112 | } |
| 113 | return s.DirStore.GetIfChanged(ctx, key, etag) |
| 114 | } |
| 115 | |
| 116 | func (s *ciCountingStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 117 | s.note("List", prefix) |
| 118 | if s.listFails(prefix) { |
| 119 | return nil, errors.New("list refused") |
| 120 | } |
| 121 | return s.DirStore.List(ctx, prefix) |
| 122 | } |
| 123 | |
| 124 | func (s *ciCountingStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) { |
| 125 | s.note("ListETags", prefix) |
| 126 | if s.listFails(prefix) { |
| 127 | return nil, errors.New("list refused") |
| 128 | } |
| 129 | return s.DirStore.ListETags(ctx, prefix) |
| 130 | } |
| 131 | |
| 132 | func (s *ciCountingStore) ListDirs(ctx context.Context, prefix string) ([]string, error) { |
| 133 | s.note("ListDirs", prefix) |
| 134 | return s.DirStore.ListDirs(ctx, prefix) |
| 135 | } |
| 136 | |
| 137 | func (s *ciCountingStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) { |
| 138 | s.note("ListMod", prefix) |
| 139 | return s.DirStore.ListMod(ctx, prefix) |
| 140 | } |
| 141 | |
| 142 | // ciWorld is a deployment of one account with two apps, shop and cafe, each |
| 143 | // with one instance record and a config bundle, written through the indexed |
| 144 | // store. |
| 145 | type ciWorld struct { |
| 146 | live *ciCountingStore |
| 147 | acct string |
| 148 | now time.Time |
| 149 | } |
| 150 | |
| 151 | func newCIWorld(t *testing.T) *ciWorld { |
| 152 | t.Helper() |
| 153 | t.Setenv("CARLOS_HOST_CLAIM", "") // the appconfig and provision guards read the process env |
| 154 | ctx := context.Background() |
| 155 | w := &ciWorld{live: newCICountingStore(t), acct: domains.Sqid(2), now: time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)} |
| 156 | manifest, err := json.Marshal(map[string]any{"id": 2, "sqid": w.acct, "name": "fixture", "created_at": w.now}) |
| 157 | if err != nil { |
| 158 | t.Fatal(err) |
| 159 | } |
| 160 | if _, err := w.live.Put(ctx, "accounts/"+w.acct+"/account.json", manifest, ""); err != nil { |
| 161 | t.Fatal(err) |
| 162 | } |
| 163 | for _, app := range []string{"shop", "cafe"} { |
| 164 | if err := provision.PutManifest(ctx, w.live, w.acct, app, provision.Manifest{Schema: 1, App: app, Domain: "example.com"}); err != nil { |
| 165 | t.Fatal(err) |
| 166 | } |
| 167 | w.record(t, app, app+".example.com") |
| 168 | if err := appconfig.SetEnv(ctx, w.live, w.acct, app, map[string]string{"APP": app}); err != nil { |
| 169 | t.Fatal(err) |
| 170 | } |
| 171 | } |
| 172 | w.live.reset() |
| 173 | return w |
| 174 | } |
| 175 | |
| 176 | func (w *ciWorld) record(t *testing.T, app, host string) { |
| 177 | t.Helper() |
| 178 | rec := provision.Record{Schema: 1, Host: host, Slug: strings.SplitN(host, ".", 2)[0], Account: w.acct, App: app, Channel: "stable", CreatedAt: w.now} |
| 179 | if err := provision.CreateRecord(context.Background(), w.live, rec); err != nil { |
| 180 | t.Fatal(err) |
| 181 | } |
| 182 | } |
| 183 | |
| 184 | func (w *ciWorld) scope(app string) string { return "apps/" + w.acct + "/" + app } |
| 185 | |
| 186 | // armView arms the change index over live for one test, as RunAgent does, |
| 187 | // and disarms it afterwards. Tests that call it stay serial. |
| 188 | func armView(t *testing.T, live release.Store) *provision.RecordView { |
| 189 | t.Helper() |
| 190 | watch := release.NewChangeWatch(live) |
| 191 | v := provision.NewRecordView(live, watch) |
| 192 | armedChangeWatch.Store(watch) |
| 193 | armedRecordView.Store(v) |
| 194 | t.Cleanup(func() { |
| 195 | armedChangeWatch.Store(nil) |
| 196 | armedRecordView.Store(nil) |
| 197 | }) |
| 198 | return v |
| 199 | } |
| 200 | |
| 201 | // disarmView makes sure no earlier test left the view armed. |
| 202 | func disarmView(t *testing.T) { |
| 203 | t.Helper() |
| 204 | armedChangeWatch.Store(nil) |
| 205 | armedRecordView.Store(nil) |
| 206 | } |
| 207 | |
| 208 | func ciReconcileConfig(t *testing.T) provision.ReconcileConfig { |
| 209 | return provision.ReconcileConfig{DataDir: t.TempDir(), Layout: flatLayout} |
| 210 | } |
| 211 | |
| 212 | // recordFamilyLists counts listings under the record families. |
| 213 | func recordFamilyLists(s *ciCountingStore) int { |
| 214 | n := 0 |
| 215 | for _, m := range []string{"List", "ListETags", "ListDirs"} { |
| 216 | n += s.count(m, "apps/") + s.count(m, "accounts/") |
| 217 | } |
| 218 | return n |
| 219 | } |
| 220 | |
| 221 | func TestChangeIndexArmedOnlyForRead(t *testing.T) { |
| 222 | t.Parallel() |
| 223 | for v, want := range map[string]bool{"": false, "read": true, "READ": false, "write": false, "1": false, " read": false} { |
| 224 | if got := changeIndexArmed(func(k string) string { |
| 225 | if k == "CARLOS_CHANGE_INDEX" { |
| 226 | return v |
| 227 | } |
| 228 | return "" |
| 229 | }); got != want { |
| 230 | t.Errorf("CARLOS_CHANGE_INDEX=%q armed = %v, want %v", v, got, want) |
| 231 | } |
| 232 | } |
| 233 | } |
| 234 | |
| 235 | // A fleet box stays disarmed with the variable set: its store door has no |
| 236 | // conditional read, so every Refresh would walk anyway (spec §3.4). |
| 237 | func TestChangeIndexStaysDisarmedOnFleetBox(t *testing.T) { |
| 238 | disarmView(t) |
| 239 | t.Cleanup(func() { disarmView(t) }) |
| 240 | opened := 0 |
| 241 | open := func(context.Context) (release.Store, error) { opened++; return newCICountingStore(t), nil } |
| 242 | env := func(m map[string]string) func(string) string { return func(k string) string { return m[k] } } |
| 243 | |
| 244 | if v := armChangeIndex(context.Background(), env(map[string]string{"CARLOS_CHANGE_INDEX": "read", "CARLOS_FLEET": "pi"}), open); v != nil || armedRecordView.Load() != nil || armedChangeWatch.Load() != nil { |
| 245 | t.Fatal("a fleet box armed the change index") |
| 246 | } |
| 247 | if v := armChangeIndex(context.Background(), env(nil), open); v != nil || armedRecordView.Load() != nil { |
| 248 | t.Fatal("an unset CARLOS_CHANGE_INDEX armed the change index") |
| 249 | } |
| 250 | if opened != 0 { |
| 251 | t.Fatalf("a disarmed box opened the deployment store %d time(s)", opened) |
| 252 | } |
| 253 | v := armChangeIndex(context.Background(), env(map[string]string{"CARLOS_CHANGE_INDEX": "read"}), open) |
| 254 | if v == nil || armedRecordView.Load() != v || armedChangeWatch.Load() == nil { |
| 255 | t.Fatal("a platform box with CARLOS_CHANGE_INDEX=read did not arm") |
| 256 | } |
| 257 | } |
| 258 | |
| 259 | // Armed, the record walks of a reconcile pass (the offline keep view's |
| 260 | // listing, Reconcile and ReconcileSidecars) read the view: once it is built, |
| 261 | // two passes with no writes list nothing on the bucket. |
| 262 | func TestReconcileUsesViewWhenArmed(t *testing.T) { |
| 263 | w := newCIWorld(t) |
| 264 | v := armView(t, w.live) |
| 265 | if err := v.Rebuild(context.Background()); err != nil { |
| 266 | t.Fatal(err) |
| 267 | } |
| 268 | reg := restartTestRegistry(t) |
| 269 | // A capable worker whose keep view has read nothing yet: the first pass |
| 270 | // lists records for it (refreshOfflineView), from the bucket, never the |
| 271 | // view (final review R1, finding 1: everything a move acts on is live). |
| 272 | mw := newMovesWorker(moveCfg{OfflineCapable: true}) |
| 273 | mw.cfg.Offline = newOfflineView(mw.cfg, nil) |
| 274 | cfg := ciReconcileConfig(t) |
| 275 | w.live.reset() |
| 276 | if err := reconcileSerializedWith(context.Background(), w.live, reg, cfg, mw); err != nil { |
| 277 | t.Fatal(err) |
| 278 | } |
| 279 | if !mw.cfg.Offline.built() || mw.cfg.Offline.failed() { |
| 280 | t.Fatal("the offline keep view was not read") |
| 281 | } |
| 282 | if n := w.live.count("List", "accounts/"); n != 1 { |
| 283 | t.Fatalf("the unbuilt offline keep view listed accounts/ %d time(s), want once, from the bucket: %v", n, w.live.got()) |
| 284 | } |
| 285 | w.live.reset() |
| 286 | for range 2 { |
| 287 | if err := reconcileSerializedWith(context.Background(), w.live, reg, cfg, mw); err != nil { |
| 288 | t.Fatal(err) |
| 289 | } |
| 290 | } |
| 291 | if n := w.live.count("List", ""); n != 0 { |
| 292 | t.Fatalf("two armed passes with no writes listed the bucket %d time(s): %v", n, w.live.got()) |
| 293 | } |
| 294 | if n := recordFamilyLists(w.live); n != 0 { |
| 295 | t.Fatalf("two armed passes listed the record families %d time(s): %v", n, w.live.got()) |
| 296 | } |
| 297 | for _, h := range []string{"shop.example.com", "cafe.example.com"} { |
| 298 | if rt, err := reg.Get(h); err != nil || rt == nil { |
| 299 | t.Fatalf("no route for %s from the view's records: %v", h, err) |
| 300 | } |
| 301 | } |
| 302 | } |
| 303 | |
| 304 | // Disarmed, the same passes walk the bucket exactly as before the index. |
| 305 | func TestReconcileWalksWhenDisarmed(t *testing.T) { |
| 306 | disarmView(t) |
| 307 | w := newCIWorld(t) |
| 308 | reg := restartTestRegistry(t) |
| 309 | cfg := ciReconcileConfig(t) |
| 310 | for range 2 { |
| 311 | if err := reconcileSerializedWith(context.Background(), w.live, reg, cfg, nil); err != nil { |
| 312 | t.Fatal(err) |
| 313 | } |
| 314 | } |
| 315 | if n := w.live.count("List", "accounts/"); n < 2 { |
| 316 | t.Fatalf("two disarmed passes listed accounts/ %d time(s), want one walk each: %v", n, w.live.got()) |
| 317 | } |
| 318 | if n := w.live.count("GetIfChanged", ""); n != 0 { |
| 319 | t.Fatalf("a disarmed pass read the change index: %v", w.live.got()) |
| 320 | } |
| 321 | } |
| 322 | |
| 323 | // instancesLoop's floor is reconcileSerialized over its own store: armed, |
| 324 | // with nothing changed, it costs one conditional GET of the index and no |
| 325 | // listing. |
| 326 | func TestInstancesLoopFloorIsRefreshWhenArmed(t *testing.T) { |
| 327 | w := newCIWorld(t) |
| 328 | v := armView(t, w.live) |
| 329 | if err := v.Rebuild(context.Background()); err != nil { |
| 330 | t.Fatal(err) |
| 331 | } |
| 332 | reg := restartTestRegistry(t) |
| 333 | cfg := ciReconcileConfig(t) |
| 334 | if err := reconcileSerialized(context.Background(), w.live, reg, cfg); err != nil { |
| 335 | t.Fatal(err) |
| 336 | } |
| 337 | w.live.reset() |
| 338 | if err := reconcileSerialized(context.Background(), w.live, reg, cfg); err != nil { |
| 339 | t.Fatal(err) |
| 340 | } |
| 341 | if n := w.live.count("GetIfChanged", release.ChangeIndexKey); n != 1 { |
| 342 | t.Fatalf("a quiet floor read the index %d time(s), want 1: %v", n, w.live.got()) |
| 343 | } |
| 344 | if n := w.live.count("List", "") + w.live.count("ListETags", "") + w.live.count("ListDirs", ""); n != 0 { |
| 345 | t.Fatalf("a quiet floor listed the bucket: %v", w.live.got()) |
| 346 | } |
| 347 | } |
| 348 | |
| 349 | // A view that cannot vouch (its Refresh failed, so there is no Complete |
| 350 | // snapshot) is never used: the caller walks live, as before the index. |
| 351 | func TestViewStoreFallsBackWhenTheViewCannotVouch(t *testing.T) { |
| 352 | disarmView(t) |
| 353 | w := newCIWorld(t) |
| 354 | if s, _, ok := viewStore(context.Background(), w.live); ok || s != release.Store(w.live) { |
| 355 | t.Fatal("a disarmed viewStore did not hand back live") |
| 356 | } |
| 357 | armView(t, w.live) |
| 358 | w.live.setFailList("accounts/") |
| 359 | if s, _, ok := viewStore(context.Background(), w.live); ok || s != release.Store(w.live) { |
| 360 | t.Fatal("viewStore used the view after a failed Refresh") |
| 361 | } |
| 362 | w.live.setFailList("") |
| 363 | s, epoch, ok := viewStore(context.Background(), w.live) |
| 364 | if !ok || s == release.Store(w.live) || epoch == 0 { |
| 365 | t.Fatalf("viewStore after a good Refresh: ok %v epoch %d", ok, epoch) |
| 366 | } |
| 367 | } |
| 368 | |
| 369 | // A hung index GET while the rest of the bucket answers (amadan #336 final |
| 370 | // review, finding 3): the reconcile pass gives up on the index after |
| 371 | // ChangeIndexPollTimeout, walks, and mints what the walk found; the watch |
| 372 | // is unknown meanwhile, so a gated read marked before the stall is due |
| 373 | // again; and the next good poll brings the watch back. |
| 374 | func TestStalledIndexGetKeepsReconcileWalking(t *testing.T) { |
| 375 | w := newCIWorld(t) |
| 376 | v := armView(t, w.live) |
| 377 | if err := v.Rebuild(context.Background()); err != nil { |
| 378 | t.Fatal(err) |
| 379 | } |
| 380 | reg := restartTestRegistry(t) |
| 381 | cfg := ciReconcileConfig(t) |
| 382 | if err := reconcileSerialized(context.Background(), w.live, reg, cfg); err != nil { |
| 383 | t.Fatal(err) |
| 384 | } |
| 385 | gate := newChangeGate() |
| 386 | clock := newGateClock() |
| 387 | gen, _ := gate.Due(w.scope("shop"), clock.now()) |
| 388 | gate.Mark(w.scope("shop"), gen, clock.now(), true) |
| 389 | if _, due := gate.Due(w.scope("shop"), clock.now()); due { |
| 390 | t.Fatal("fixture: a marked scope is due before the stall") |
| 391 | } |
| 392 | |
| 393 | w.record(t, "shop", "two.shop.example.com") |
| 394 | unstall := w.live.stallIndex() |
| 395 | defer func() { |
| 396 | if unstall != nil { |
| 397 | unstall() |
| 398 | } |
| 399 | }() |
| 400 | done := make(chan error, 1) |
| 401 | go func() { done <- reconcileSerialized(context.Background(), w.live, reg, cfg) }() |
| 402 | select { |
| 403 | case err := <-done: |
| 404 | if err != nil { |
| 405 | t.Fatal(err) |
| 406 | } |
| 407 | case <-time.After(4 * release.ChangeIndexPollTimeout): |
| 408 | t.Fatal("a reconcile pass hung behind a stalled index GET") |
| 409 | } |
| 410 | if rt, err := reg.Get("two.shop.example.com"); err != nil || rt == nil { |
| 411 | t.Fatalf("the pass over a stalled index did not walk to the new record: %v", err) |
| 412 | } |
| 413 | if armedChangeWatch.Load().Usable() { |
| 414 | t.Fatal("the watch vouches after its poll gave up") |
| 415 | } |
| 416 | if _, due := gate.Due(w.scope("shop"), clock.now()); !due { |
| 417 | t.Fatal("a gated read trusted the frozen observation") |
| 418 | } |
| 419 | |
| 420 | unstall() |
| 421 | unstall = nil |
| 422 | if err := reconcileSerialized(context.Background(), w.live, reg, cfg); err != nil { |
| 423 | t.Fatal(err) |
| 424 | } |
| 425 | if !armedChangeWatch.Load().Usable() { |
| 426 | t.Fatal("the watch did not recover on the next good poll") |
| 427 | } |
| 428 | } |
| 429 | |
| 430 | // ---- moves ----------------------------------------------------------------- |
| 431 | |
| 432 | // The moves pass decides whether it has work from its own refresh of the |
| 433 | // view inside w.pass, never from the snapshot the reconcile leg of its pass |
| 434 | // set read (spec §4.3, codex 1): pass A is paused after its reconcile, a |
| 435 | // move is requested, and A's moves leg sees it and walks the bucket. |
| 436 | func TestMovesPassDecidesFromItsOwnRefresh(t *testing.T) { |
| 437 | w := newCIWorld(t) |
| 438 | armView(t, w.live) |
| 439 | mw := newMovesWorker(moveCfg{}) |
| 440 | armedMovesWorker.Store(mw) |
| 441 | t.Cleanup(func() { armedMovesWorker.Store(nil) }) |
| 442 | reg := restartTestRegistry(t) |
| 443 | rcfg := ciReconcileConfig(t) |
| 444 | legs := &passLegs{store: w.live, reg: reg, optDir: t.TempDir(), dd: t.TempDir(), |
| 445 | cfg: func() provision.ReconcileConfig { return rcfg }} |
| 446 | legs.full(context.Background()) // walks once: nothing to move |
| 447 | if !mw.walkedIdle { |
| 448 | t.Fatal("fixture: a walk with no move left the worker busy") |
| 449 | } |
| 450 | |
| 451 | paused, resume := make(chan struct{}), make(chan struct{}) |
| 452 | var first atomic.Bool |
| 453 | passAfterReconcileHook = func() { |
| 454 | if first.CompareAndSwap(false, true) { |
| 455 | paused <- struct{}{} |
| 456 | <-resume |
| 457 | } |
| 458 | } |
| 459 | t.Cleanup(func() { passAfterReconcileHook = nil }) |
| 460 | done := make(chan struct{}) |
| 461 | go func() { defer close(done); legs.full(context.Background()) }() // pass A |
| 462 | <-paused |
| 463 | if _, err := provision.SetRecordMove(context.Background(), w.live, w.acct, "shop", "shop.example.com", provision.MoveRequest("uk", w.now)); err != nil { |
| 464 | t.Fatal(err) |
| 465 | } |
| 466 | w.live.reset() |
| 467 | close(resume) |
| 468 | <-done |
| 469 | if n := w.live.count("List", "accounts/"); n == 0 { |
| 470 | t.Fatalf("pass A's moves leg did not walk for a move requested after its reconcile: %v", w.live.got()) |
| 471 | } |
| 472 | if mw.walkedIdle { |
| 473 | t.Fatal("a walk that found a move left the worker idle") |
| 474 | } |
| 475 | } |
| 476 | |
| 477 | // With nothing to move, the armed moves pass costs no walk: the record view |
| 478 | // vouches that no record carries a move, and the worker holds no move state. |
| 479 | // Anything the worker holds (here, a cut's wake fence) makes it walk. |
| 480 | func TestMovesPassSkipsTheWalkOnlyWithNoMoveWork(t *testing.T) { |
| 481 | w := newCIWorld(t) |
| 482 | armView(t, w.live) |
| 483 | mw := newMovesWorker(moveCfg{}) |
| 484 | if err := movesPass(context.Background(), w.live, mw); err != nil { |
| 485 | t.Fatal(err) |
| 486 | } |
| 487 | w.live.reset() |
| 488 | if err := movesPass(context.Background(), w.live, mw); err != nil { |
| 489 | t.Fatal(err) |
| 490 | } |
| 491 | if n := recordFamilyLists(w.live); n != 0 { |
| 492 | t.Fatalf("an idle armed moves pass listed the record families: %v", w.live.got()) |
| 493 | } |
| 494 | mw.fenced.Store("shop.example.com", uint64(1)) |
| 495 | w.live.reset() |
| 496 | if err := movesPass(context.Background(), w.live, mw); err != nil { |
| 497 | t.Fatal(err) |
| 498 | } |
| 499 | if n := w.live.count("List", "accounts/"); n == 0 { |
| 500 | t.Fatalf("a worker holding a wake fence skipped its walk: %v", w.live.got()) |
| 501 | } |
| 502 | } |
| 503 | |
| 504 | // Final review R1 finding 1 (amadan #336): the moves pass acts on the |
| 505 | // bucket, never on a snapshot older than what a pass already saw there. |
| 506 | // An abort's completion misses its bump, so the view still holds the move |
| 507 | // aborting; an unrelated scope then fails to re-read and a pass walks the |
| 508 | // bucket instead (the live fallback); a new move of the same host is |
| 509 | // requested (its bump missed too) and staging fills; the unrelated scope |
| 510 | // recovers, so the view is Complete again with the old aborting state. |
| 511 | // Acting on that snapshot would delete the new move's staging. The pass |
| 512 | // walks the bucket, sees the new move preparing, and leaves staging alone. |
| 513 | func TestMovesPassNeverActsOnAnOlderSnapshot(t *testing.T) { |
| 514 | h := newMoveHarness(t) |
| 515 | root := t.TempDir() |
| 516 | indexed := release.NewDirStore(root) |
| 517 | indexed.EnableChangeIndex() |
| 518 | live := &ciCountingStore{DirStore: indexed} |
| 519 | bypass := release.NewDirStore(root) // writes whose bump failed |
| 520 | ctx := context.Background() |
| 521 | edges, _, err := h.s.Get(ctx, "control/edges/uk-box.json") |
| 522 | if err != nil { |
| 523 | t.Fatal(err) |
| 524 | } |
| 525 | if _, err := live.Put(ctx, "control/edges/uk-box.json", edges, ""); err != nil { |
| 526 | t.Fatal(err) |
| 527 | } |
| 528 | h.s = live |
| 529 | host := "a.go.example.com" |
| 530 | h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MoveAbort()) |
| 531 | other := "b.go.example.com" |
| 532 | if err := provision.CreateRecord(ctx, live, provision.Record{Schema: 1, Host: other, Slug: "b", App: "other", Account: "bab", |
| 533 | Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()}); err != nil { |
| 534 | t.Fatal(err) |
| 535 | } |
| 536 | v := armView(t, live) |
| 537 | if err := v.Rebuild(ctx); err != nil { |
| 538 | t.Fatal(err) |
| 539 | } |
| 540 | if recs, err := provision.ListAllRecords(ctx, v.Snapshot().Store(live)); err != nil || len(recs) != 2 { |
| 541 | t.Fatalf("fixture: the view holds %d record(s), %v; want both", len(recs), err) |
| 542 | } |
| 543 | // The abort completes, and its bump fails. |
| 544 | if _, err := provision.SetRecordMove(ctx, bypass, "bab", "app", host, provision.MoveAbortComplete()); err != nil { |
| 545 | t.Fatal(err) |
| 546 | } |
| 547 | // An unrelated scope moves and will not re-read: the view cannot vouch, |
| 548 | // and the pass walks the bucket, where the abort is complete. |
| 549 | third := provision.Record{Schema: 1, Host: "c.go.example.com", Slug: "c", App: "other", Account: "bab", |
| 550 | Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()} |
| 551 | if err := provision.CreateRecord(ctx, live, third); err != nil { |
| 552 | t.Fatal(err) |
| 553 | } |
| 554 | live.setFailList(provision.RecordsPrefix("bab", "other")) |
| 555 | if _, _, ok := viewStore(ctx, live); ok { |
| 556 | t.Fatal("fixture: the view vouched with a scope failing to re-read") |
| 557 | } |
| 558 | wk := h.worker() |
| 559 | if err := movesPass(ctx, live, wk); err != nil { |
| 560 | t.Fatal(err) |
| 561 | } |
| 562 | // The host is moved again, the bump failing again, and staging fills. |
| 563 | if _, err := provision.SetRecordMove(ctx, bypass, "bab", "app", host, provision.MoveRequest("uk", time.Now())); err != nil { |
| 564 | t.Fatal(err) |
| 565 | } |
| 566 | h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1") |
| 567 | h.f.put("data-uk", "moves/"+host+"/generations/g/snapshot", "e1") |
| 568 | // The unrelated scope recovers: Complete again, with the old abort. |
| 569 | live.setFailList("") |
| 570 | if _, _, ok := viewStore(ctx, live); !ok { |
| 571 | t.Fatal("fixture: the view did not recover") |
| 572 | } |
| 573 | snapRec, err := provision.GetRecord(ctx, v.Snapshot().Store(live), "bab", "app", host) |
| 574 | if err != nil || snapRec.Move == nil || snapRec.Move.State != provision.MoveAborting { |
| 575 | t.Fatalf("fixture: the recovered snapshot should hold the old abort: %+v %v", snapRec, err) |
| 576 | } |
| 577 | |
| 578 | if err := movesPass(ctx, live, wk); err != nil { |
| 579 | t.Fatal(err) |
| 580 | } |
| 581 | for _, op := range h.f.ops { |
| 582 | if strings.HasPrefix(op, "delete data-uk/moves/"+host+"/") { |
| 583 | t.Fatalf("the pass acted on the snapshot's old abort and deleted the new move's staging: %v", h.f.ops) |
| 584 | } |
| 585 | } |
| 586 | if staged := h.f.bodies("data-uk", "moves/"+host+"/"); len(staged) != 1 { |
| 587 | t.Fatalf("the new move's staging after the pass: %v", staged) |
| 588 | } |
| 589 | if got := h.read(t, host); got.Move == nil || got.Move.State == provision.MoveAborting { |
| 590 | t.Fatalf("the record after the pass: %+v", got.Move) |
| 591 | } |
| 592 | } |
| 593 | |
| 594 | // ---- fastTick -------------------------------------------------------------- |
| 595 | |
| 596 | // ciFastDeps is a fastDeps over an indexed store with counting hooks for |
| 597 | // both pass shapes. |
| 598 | func ciFastDeps(t *testing.T) (*fastDeps, *release.DirStore, *[]string, *[][]string) { |
| 599 | t.Helper() |
| 600 | env := newFastFixture(t) |
| 601 | ds := release.NewDirStore(t.TempDir()) |
| 602 | ds.EnableChangeIndex() |
| 603 | env.store = ds |
| 604 | var full []string |
| 605 | var scoped [][]string |
| 606 | d := env.deps(func() { full = append(full, "full") }) |
| 607 | d.watch = release.NewChangeWatch(ds) |
| 608 | d.scopedPass = func(_ context.Context, moved []string) { scoped = append(scoped, moved) } |
| 609 | return d, ds, &full, &scoped |
| 610 | } |
| 611 | |
| 612 | func TestFastTickIndexOnlyRunsScopedPass(t *testing.T) { |
| 613 | t.Parallel() |
| 614 | ctx := context.Background() |
| 615 | d, ds, full, scoped := ciFastDeps(t) |
| 616 | now := time.Unix(1754640000, 0) |
| 617 | fastTick(ctx, d, now) // seeds the serial and the index: no pass |
| 618 | if len(*full)+len(*scoped) != 0 { |
| 619 | t.Fatalf("the seeding tick ran a pass: full %v scoped %v", *full, *scoped) |
| 620 | } |
| 621 | if _, err := ds.Put(ctx, "apps/bab/shop/instances/one.shop.example.com.json", []byte(`{}`), ""); err != nil { |
| 622 | t.Fatal(err) |
| 623 | } |
| 624 | fastTick(ctx, d, now.Add(10*time.Second)) |
| 625 | if len(*full) != 0 { |
| 626 | t.Fatalf("an index-only change ran the full pass: %v", *full) |
| 627 | } |
| 628 | if want := [][]string{{"apps/bab/shop"}}; !slices.EqualFunc(*scoped, want, slices.Equal[[]string]) { |
| 629 | t.Fatalf("scoped passes %v, want %v", *scoped, want) |
| 630 | } |
| 631 | // Inside the clamp the change waits, and is acted on at the next tick |
| 632 | // past it. |
| 633 | if _, err := ds.Put(ctx, "apps/bab/cafe/instances/one.cafe.example.com.json", []byte(`{}`), ""); err != nil { |
| 634 | t.Fatal(err) |
| 635 | } |
| 636 | fastTick(ctx, d, now.Add(12*time.Second)) |
| 637 | if len(*scoped) != 1 { |
| 638 | t.Fatalf("a change inside the clamp ran at once: %v", *scoped) |
| 639 | } |
| 640 | fastTick(ctx, d, now.Add(16*time.Second)) |
| 641 | if want := [][]string{{"apps/bab/shop"}, {"apps/bab/cafe"}}; !slices.EqualFunc(*scoped, want, slices.Equal[[]string]) { |
| 642 | t.Fatalf("scoped passes %v, want %v", *scoped, want) |
| 643 | } |
| 644 | fastTick(ctx, d, now.Add(30*time.Second)) |
| 645 | if len(*scoped) != 2 { |
| 646 | t.Fatalf("an unchanged index ran the scoped pass again: %v", *scoped) |
| 647 | } |
| 648 | // Scopes outside the record view's (control/edges) run no pass. |
| 649 | if _, err := ds.Put(ctx, "control/edges/one.json", []byte(`{}`), ""); err != nil { |
| 650 | t.Fatal(err) |
| 651 | } |
| 652 | fastTick(ctx, d, now.Add(40*time.Second)) |
| 653 | if len(*scoped) != 2 || len(*full) != 0 { |
| 654 | t.Fatalf("a control/ scope ran a pass: full %v scoped %v", *full, *scoped) |
| 655 | } |
| 656 | } |
| 657 | |
| 658 | // A shared watch: another poller (the record view) taking Poll's moved list |
| 659 | // first does not hide the movement from fastTick. |
| 660 | func TestFastTickActsOnObservedNotPollReturn(t *testing.T) { |
| 661 | t.Parallel() |
| 662 | ctx := context.Background() |
| 663 | d, ds, _, scoped := ciFastDeps(t) |
| 664 | now := time.Unix(1754640000, 0) |
| 665 | fastTick(ctx, d, now) |
| 666 | if _, err := ds.Put(ctx, "apps/bab/shop/instances/one.shop.example.com.json", []byte(`{}`), ""); err != nil { |
| 667 | t.Fatal(err) |
| 668 | } |
| 669 | if moved, err := d.watch.Poll(ctx); err != nil || len(moved) == 0 { |
| 670 | t.Fatalf("the sibling poll saw %v, %v", moved, err) |
| 671 | } |
| 672 | fastTick(ctx, d, now.Add(10*time.Second)) |
| 673 | if want := [][]string{{"apps/bab/shop"}}; !slices.EqualFunc(*scoped, want, slices.Equal[[]string]) { |
| 674 | t.Fatalf("scoped passes %v, want %v: a sibling's Poll swallowed the movement", *scoped, want) |
| 675 | } |
| 676 | } |
| 677 | |
| 678 | func TestSerialChangeStillRunsFullPass(t *testing.T) { |
| 679 | t.Parallel() |
| 680 | ctx := context.Background() |
| 681 | d, ds, full, scoped := ciFastDeps(t) |
| 682 | now := time.Unix(1754640000, 0) |
| 683 | fastTick(ctx, d, now) |
| 684 | if _, err := ds.Put(ctx, "apps/bab/shop/instances/one.shop.example.com.json", []byte(`{}`), ""); err != nil { |
| 685 | t.Fatal(err) |
| 686 | } |
| 687 | if err := release.BumpSerial(ctx, ds, now); err != nil { |
| 688 | t.Fatal(err) |
| 689 | } |
| 690 | fastTick(ctx, d, now.Add(10*time.Second)) |
| 691 | if len(*full) != 1 || len(*scoped) != 0 { |
| 692 | t.Fatalf("a serial change with the index moved: full %v scoped %v, want one full pass only", *full, *scoped) |
| 693 | } |
| 694 | // The full pass covered the movement: the next tick runs nothing. |
| 695 | fastTick(ctx, d, now.Add(20*time.Second)) |
| 696 | if len(*full) != 1 || len(*scoped) != 0 { |
| 697 | t.Fatalf("the movement the full pass covered ran again: full %v scoped %v", *full, *scoped) |
| 698 | } |
| 699 | } |
| 700 | |
| 701 | // The backstop asks for the unrestricted full pass after its rebuild; the |
| 702 | // loop's own tick runs it, once. |
| 703 | func TestFastTickRunsTheBackstopsFullPass(t *testing.T) { |
| 704 | t.Parallel() |
| 705 | ctx := context.Background() |
| 706 | d, ds, full, scoped := ciFastDeps(t) |
| 707 | var due atomic.Bool |
| 708 | d.fullDue = &due |
| 709 | now := time.Unix(1754640000, 0) |
| 710 | fastTick(ctx, d, now) |
| 711 | v := provision.NewRecordView(ds, d.watch) |
| 712 | recordViewBackstop(ctx, v, &due) |
| 713 | if v.Snapshot() == nil { |
| 714 | t.Fatal("the backstop did not rebuild the view") |
| 715 | } |
| 716 | fastTick(ctx, d, now.Add(10*time.Second)) |
| 717 | if len(*full) != 1 || len(*scoped) != 0 || due.Load() { |
| 718 | t.Fatalf("after the backstop: full %v scoped %v due %v, want one full pass", *full, *scoped, due.Load()) |
| 719 | } |
| 720 | fastTick(ctx, d, now.Add(20*time.Second)) |
| 721 | if len(*full) != 1 { |
| 722 | t.Fatalf("the backstop's full pass ran again: %v", *full) |
| 723 | } |
| 724 | } |
| 725 | |
| 726 | func TestBackstopEveryIsJitteredAroundTheBackstop(t *testing.T) { |
| 727 | t.Parallel() |
| 728 | lo, hi := release.ChangeIndexBackstop*9/10, release.ChangeIndexBackstop*11/10 |
| 729 | for range 200 { |
| 730 | if d := backstopEvery(); d < lo || d >= hi { |
| 731 | t.Fatalf("backstopEvery() = %v, want in [%v, %v)", d, lo, hi) |
| 732 | } |
| 733 | } |
| 734 | } |
| 735 | |
| 736 | // ---- the scoped pass ------------------------------------------------------- |
| 737 | |
| 738 | // Codex 10's witness: on a box hosting shop and cafe, an index bump for shop |
| 739 | // runs a scoped pass that reads only shop's keys (its app prefix and its |
| 740 | // build state), plus the change index, while shop's new config is |
| 741 | // delivered. Nothing under cafe's prefix, nothing for cafe's build state, |
| 742 | // and nothing box-wide (edges files, domain claims) is read. |
| 743 | func TestScopedPassTouchesOnlyMovedApp(t *testing.T) { |
| 744 | w := newCIWorld(t) |
| 745 | armView(t, w.live) |
| 746 | armedMovesWorker.Store(newMovesWorker(moveCfg{})) |
| 747 | t.Cleanup(func() { armedMovesWorker.Store(nil) }) |
| 748 | // A sink, so the telemetry leg runs (telemetryPass returns early without |
| 749 | // one): its reads are gated too, so the scoped pass reads only shop's. |
| 750 | var pushed []map[string]bool |
| 751 | setAnalyticsSink(func(on map[string]bool) { pushed = append(pushed, on) }) |
| 752 | t.Cleanup(func() { analyticsSink.Store(nil) }) |
| 753 | raw, _, err := appconfig.GenerateKey() |
| 754 | if err != nil { |
| 755 | t.Fatal(err) |
| 756 | } |
| 757 | pk, err := appconfig.ParsePrivateKey(raw, "test") |
| 758 | if err != nil { |
| 759 | t.Fatal(err) |
| 760 | } |
| 761 | reg := restartTestRegistry(t) |
| 762 | dd := t.TempDir() |
| 763 | // Build states live in the same counting store, under inst/<host>/. |
| 764 | bs := &buildStates{hibBucket: "hib", prefix: func(h string) string { return "inst/" + h }, |
| 765 | storeFor: storesOf(map[string]release.Store{"hib": w.live}), reg: reg, optDir: t.TempDir(), bakeDir: t.TempDir(), now: time.Now} |
| 766 | legs := &passLegs{store: w.live, reg: reg, optDir: t.TempDir(), priv: pk, dd: dd, bs: bs, |
| 767 | cfg: func() provision.ReconcileConfig { return provision.ReconcileConfig{DataDir: dd, Layout: flatLayout} }} |
| 768 | w.live.reset() |
| 769 | legs.full(context.Background()) |
| 770 | // The fixture exercises every box-wide leg on the full pass. |
| 771 | for _, want := range []string{"inst/cafe.example.com/", "domains/", "control/edges", |
| 772 | telemetry.AccountKey(w.acct), telemetry.AppKey(w.acct, "cafe")} { |
| 773 | if !slices.ContainsFunc(w.live.got(), func(c string) bool { _, k, _ := strings.Cut(c, " "); return strings.HasPrefix(k, want) }) { |
| 774 | t.Fatalf("fixture: the full pass read nothing under %s: %v", want, w.live.got()) |
| 775 | } |
| 776 | } |
| 777 | for _, h := range []string{"shop.example.com", "cafe.example.com"} { |
| 778 | rt, err := reg.Get(h) |
| 779 | if err != nil || rt == nil { |
| 780 | t.Fatalf("fixture: no route for %s: %v", h, err) |
| 781 | } |
| 782 | if !strings.Contains(readConfigEnv(t, dd, *rt), "APP=") { |
| 783 | t.Fatalf("fixture: %s got no config from the full pass", h) |
| 784 | } |
| 785 | } |
| 786 | |
| 787 | if err := appconfig.SetEnv(context.Background(), w.live, w.acct, "shop", map[string]string{"APP": "shop2"}); err != nil { |
| 788 | t.Fatal(err) |
| 789 | } |
| 790 | w.live.reset() |
| 791 | legs.scoped(context.Background(), []string{w.scope("shop")}) |
| 792 | |
| 793 | if len(pushed) != 2 || len(pushed[1]) != 0 { |
| 794 | t.Fatalf("switch maps pushed %v, want one per pass, both all off", pushed) |
| 795 | } |
| 796 | shop := "apps/" + w.acct + "/shop/" |
| 797 | var shopReads int |
| 798 | for _, c := range w.live.got() { |
| 799 | _, key, _ := strings.Cut(c, " ") |
| 800 | switch { |
| 801 | case strings.HasPrefix(key, shop): |
| 802 | shopReads++ |
| 803 | case key == release.ChangeIndexKey, strings.HasPrefix(key, "inst/shop.example.com/"): |
| 804 | case strings.HasPrefix(key, "control/config-ack/") && strings.HasSuffix(key, "/shop.example.com.json"): |
| 805 | // shop's config acknowledgement, written for the delivery |
| 806 | default: |
| 807 | t.Errorf("the scoped pass for shop read %s, which is not shop's", c) |
| 808 | } |
| 809 | } |
| 810 | if shopReads == 0 { |
| 811 | t.Fatalf("the scoped pass read nothing under shop: %v", w.live.got()) |
| 812 | } |
| 813 | rt, _ := reg.Get("shop.example.com") |
| 814 | if got := readConfigEnv(t, dd, *rt); !strings.Contains(got, "APP=shop2") { |
| 815 | t.Fatalf("shop's config after the scoped pass = %q, want APP=shop2", got) |
| 816 | } |
| 817 | } |
| 818 | |
| 819 | func readConfigEnv(t *testing.T, dd string, rt registry.Route) string { |
| 820 | t.Helper() |
| 821 | b, err := os.ReadFile(appconfig.ConfigEnvPath(dd, rt)) |
| 822 | if err != nil && !os.IsNotExist(err) { |
| 823 | t.Fatal(err) |
| 824 | } |
| 825 | return string(b) |
| 826 | } |
| 827 | |
amadan