| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "reflect" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/domains" |
| 13 | "github.com/carlosframework/platform/internal/provision" |
| 14 | "github.com/carlosframework/platform/internal/registry" |
| 15 | "github.com/carlosframework/platform/internal/release" |
| 16 | "github.com/carlosframework/platform/internal/seeds" |
| 17 | ) |
| 18 | |
| 19 | // The import rollback (seeded-instances spec §5.6) against this package's |
| 20 | // real Service: a failed wake of the import's row must not leave the |
| 21 | // rollback stuck once its stop is known (seeded pre-land review, P2). |
| 22 | // |
| 23 | // The rollback's fence hook is internal/host's fenceForRollback, which |
| 24 | // this package cannot import (host imports it). rollbackFence below is |
| 25 | // that rule, step for step, with the startup sweep taken as done. |
| 26 | |
| 27 | type importRollback struct { |
| 28 | a *Service |
| 29 | b *fakeBackend |
| 30 | reg *registry.Registry |
| 31 | store *release.DirStore |
| 32 | cfg provision.ReconcileConfig |
| 33 | rec provision.Record |
| 34 | original registry.Route |
| 35 | |
| 36 | mu sync.Mutex |
| 37 | removed int // RemoveLocal calls: the import's local database |
| 38 | replicas int // DeleteReplica calls: the host's replica |
| 39 | // beforeFence, when set, runs once as the rollback's fence hook is |
| 40 | // entered: after step 1 closed the gate, before the fence. |
| 41 | beforeFence func() |
| 42 | } |
| 43 | |
| 44 | func newImportRollback(t *testing.T) *importRollback { |
| 45 | t.Helper() |
| 46 | ctx := context.Background() |
| 47 | a, b, _, reg := testActivator(t, time.Hour) |
| 48 | a.Routes = reg |
| 49 | x := &importRollback{a: a, b: b, reg: reg, store: release.NewDirStore(t.TempDir())} |
| 50 | |
| 51 | sqid := domains.Sqid(2) |
| 52 | manifest := `{"id":2,"sqid":"` + sqid + `","name":"Shop","created_at":"2026-08-02T09:00:00Z"}` |
| 53 | if _, err := x.store.Put(ctx, "accounts/"+sqid+"/account.json", []byte(manifest), ""); err != nil { |
| 54 | t.Fatal(err) |
| 55 | } |
| 56 | if err := provision.PutManifest(ctx, x.store, sqid, "shop", provision.Manifest{Schema: 1, App: "shop", Domain: "shop.example.com"}); err != nil { |
| 57 | t.Fatal(err) |
| 58 | } |
| 59 | id, err := seeds.NewID() |
| 60 | if err != nil { |
| 61 | t.Fatal(err) |
| 62 | } |
| 63 | x.rec = provision.Record{Schema: 1, Host: "a.shop.example.com", Slug: "a", App: "shop", Account: sqid, |
| 64 | Region: "ie", Channel: "stable", CreatedAt: time.Now(), |
| 65 | Seed: &provision.SeedSource{ID: id, Kind: seeds.KindImport, State: seeds.StateReady, |
| 66 | SHA256: strings.Repeat("a", 64), Bytes: 10, Replaces: "127.0.0.1:8401", |
| 67 | VerifyHash: strings.Repeat("b", 64), By: "t", At: time.Now()}} |
| 68 | if err := provision.CreateRecord(ctx, x.store, x.rec); err != nil { |
| 69 | t.Fatal(err) |
| 70 | } |
| 71 | // The legacy row the import replaces: an addr route to the app's own |
| 72 | // process, which the rollback must put back exactly. |
| 73 | if err := reg.Upsert(registry.Route{Host: x.rec.Host, Kind: "service", Addr: "127.0.0.1:8401", App: "shop", |
| 74 | Account: sqid, Region: "ie", Channel: "stable", Status: "ready", Unit: "shop@a.service"}); err != nil { |
| 75 | t.Fatal(err) |
| 76 | } |
| 77 | x.original = *mustRoute(t, reg, x.rec.Host) |
| 78 | |
| 79 | data := release.NewDirStore(t.TempDir()) |
| 80 | if _, err := data.Put(ctx, seeds.SnapshotKey(x.rec.Host, id), []byte("db"), ""); err != nil { |
| 81 | t.Fatal(err) |
| 82 | } |
| 83 | var gens uint64 |
| 84 | x.cfg = provision.ReconcileConfig{Region: "ie", DataDir: t.TempDir(), |
| 85 | Layout: func(context.Context, release.Store) (string, error) { return release.LayoutModeNested, nil }, |
| 86 | Seeds: &provision.SeedHooks{ |
| 87 | Data: data, |
| 88 | LegacyOwners: func(context.Context) (*seeds.LegacyOwners, error) { |
| 89 | return &seeds.LegacyOwners{V: 1}, nil |
| 90 | }, |
| 91 | EnvBefore: func(string, string) (string, error) { return registry.EnvBeforeAbsent, nil }, |
| 92 | FenceForRollback: x.rollbackFence, |
| 93 | LiftFence: func(host string, gen uint64) { a.LiftFence(host, gen) }, |
| 94 | RemoveLocal: func(registry.Route) error { |
| 95 | x.mu.Lock() |
| 96 | x.removed++ |
| 97 | x.mu.Unlock() |
| 98 | return nil |
| 99 | }, |
| 100 | DeleteReplica: func(context.Context, string) error { |
| 101 | x.mu.Lock() |
| 102 | x.replicas++ |
| 103 | x.mu.Unlock() |
| 104 | return nil |
| 105 | }, |
| 106 | NextFenceGen: func() (uint64, error) { |
| 107 | x.mu.Lock() |
| 108 | defer x.mu.Unlock() |
| 109 | gens++ |
| 110 | return gens, nil |
| 111 | }, |
| 112 | LocalSeedID: func(string) (string, error) { return id, nil }, |
| 113 | RetireKey: func(string, string, string) error { return nil }, |
| 114 | }} |
| 115 | x.pass(t) // the swap |
| 116 | if row := x.row(t); !registry.IsImportRow(*row, registry.ImportJournal{Host: x.rec.Host, SeedID: id}) { |
| 117 | t.Fatalf("the swap did not happen: %+v", row) |
| 118 | } |
| 119 | return x |
| 120 | } |
| 121 | |
| 122 | // rollbackFence is internal/host's fenceForRollback over this Service. |
| 123 | func (x *importRollback) rollbackFence(ctx context.Context, host string, gen uint64) (bool, error) { |
| 124 | if f := x.beforeFence; f != nil { |
| 125 | x.beforeFence = nil |
| 126 | f() |
| 127 | } |
| 128 | if gen == 0 { |
| 129 | return false, ErrFenceGeneration |
| 130 | } |
| 131 | if x.a.FenceGeneration(host) == gen { |
| 132 | return true, nil |
| 133 | } |
| 134 | fctx, cancel := context.WithTimeout(ctx, 5*time.Second) |
| 135 | defer cancel() |
| 136 | _, err := x.a.FenceAndStopIfUnfenced(fctx, host, gen) |
| 137 | switch { |
| 138 | case err == nil: |
| 139 | return true, nil |
| 140 | case errors.Is(err, ErrFenceHeld): |
| 141 | return false, err |
| 142 | case errors.Is(err, ErrStopUnproven) && x.a.Quiesced(host) && !x.a.StopUncertain(host): |
| 143 | return true, nil |
| 144 | } |
| 145 | x.a.LiftFence(host, gen) |
| 146 | return false, err |
| 147 | } |
| 148 | |
| 149 | func (x *importRollback) pass(t *testing.T) { |
| 150 | t.Helper() |
| 151 | if err := provision.Reconcile(context.Background(), x.store, x.reg, x.cfg); err != nil { |
| 152 | t.Fatal(err) |
| 153 | } |
| 154 | } |
| 155 | |
| 156 | func (x *importRollback) row(t *testing.T) *registry.Route { |
| 157 | t.Helper() |
| 158 | return mustRoute(t, x.reg, x.rec.Host) |
| 159 | } |
| 160 | |
| 161 | func (x *importRollback) abort(t *testing.T) { |
| 162 | t.Helper() |
| 163 | if _, err := provision.SetRecordSeed(context.Background(), x.store, x.rec.Account, x.rec.App, x.rec.Host, |
| 164 | provision.SeedAbort(x.rec.Seed.ID)); err != nil { |
| 165 | t.Fatal(err) |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | // wake starts a request's wake of the import's row in the background and |
| 170 | // returns its outcome. |
| 171 | func (x *importRollback) wake(t *testing.T) <-chan error { |
| 172 | t.Helper() |
| 173 | rt := *x.row(t) |
| 174 | done := make(chan error, 1) |
| 175 | go func() { |
| 176 | release, err := x.a.beginRequest(context.Background(), rt) |
| 177 | if err == nil { |
| 178 | release() |
| 179 | } |
| 180 | done <- err |
| 181 | }() |
| 182 | return done |
| 183 | } |
| 184 | |
| 185 | // wantRestored: the original route back exactly, the import's local |
| 186 | // database and replica removed once each, and no fence left. |
| 187 | func (x *importRollback) wantRestored(t *testing.T) { |
| 188 | t.Helper() |
| 189 | if got := x.row(t); !reflect.DeepEqual(*got, x.original) { |
| 190 | t.Fatalf("row = %+v, want the original %+v", got, x.original) |
| 191 | } |
| 192 | if j, err := x.reg.ImportJournal(x.rec.Host); err != nil || (j != nil && j.State != registry.JournalCleanupPending) { |
| 193 | t.Fatalf("journal = %+v, %v", j, err) |
| 194 | } |
| 195 | x.mu.Lock() |
| 196 | removed := x.removed |
| 197 | x.mu.Unlock() |
| 198 | if removed != 1 { |
| 199 | t.Fatalf("the import's local database was removed %d times, want once", removed) |
| 200 | } |
| 201 | if x.a.Fenced(x.rec.Host) { |
| 202 | t.Fatal("the rollback left its fence") |
| 203 | } |
| 204 | } |
| 205 | |
| 206 | // wantBlocked: nothing deleted, the import's row still in place with its |
| 207 | // gate closed, and the journal still rolling back. |
| 208 | func (x *importRollback) wantBlocked(t *testing.T) { |
| 209 | t.Helper() |
| 210 | row := x.row(t) |
| 211 | if row.ImportGate != registry.ImportGateClosed { |
| 212 | t.Fatalf("row = %+v, want the import's row with its gate closed", row) |
| 213 | } |
| 214 | if j, err := x.reg.ImportJournal(x.rec.Host); err != nil || j == nil || j.State != registry.JournalRollingBack { |
| 215 | t.Fatalf("journal = %+v, %v; want rolling-back", j, err) |
| 216 | } |
| 217 | x.mu.Lock() |
| 218 | removed, replicas := x.removed, x.replicas |
| 219 | x.mu.Unlock() |
| 220 | if removed != 0 || replicas != 0 { |
| 221 | t.Fatalf("deleted under an unproven stop: local %d, replica %d", removed, replicas) |
| 222 | } |
| 223 | } |
| 224 | |
| 225 | func (x *importRollback) retained() bool { |
| 226 | x.a.mu.Lock() |
| 227 | defer x.a.mu.Unlock() |
| 228 | _, ok := x.a.unprovenWakes[x.rec.Host] |
| 229 | return ok |
| 230 | } |
| 231 | |
| 232 | // reap runs the sweep's retry of retained failed wakes once and waits |
| 233 | // for it to finish. |
| 234 | func (x *importRollback) reap(t *testing.T) { |
| 235 | t.Helper() |
| 236 | x.a.mu.Lock() |
| 237 | u := x.a.unprovenWakes[x.rec.Host] |
| 238 | x.a.reapUnprovenWakesLocked() |
| 239 | x.a.mu.Unlock() |
| 240 | waitFor(t, "the reaper's retry to finish", func() bool { |
| 241 | x.a.mu.Lock() |
| 242 | defer x.a.mu.Unlock() |
| 243 | return u == nil || !u.busy || x.a.unprovenWakes[x.rec.Host] != u |
| 244 | }) |
| 245 | } |
| 246 | |
| 247 | // queueStart makes the next run-unit start one that systemd has queued: |
| 248 | // it blocks until its context ends and comes back marked as a start root |
| 249 | // may still carry out (#292). |
| 250 | func (x *importRollback) queueStart(t *testing.T) { |
| 251 | t.Helper() |
| 252 | queued := make(chan struct{}) |
| 253 | x.b.mu.Lock() |
| 254 | x.b.startQueued = queued |
| 255 | x.b.mu.Unlock() |
| 256 | t.Cleanup(func() { close(queued) }) |
| 257 | } |
| 258 | |
| 259 | func mustRoute(t *testing.T, reg *registry.Registry, host string) *registry.Route { |
| 260 | t.Helper() |
| 261 | rt, err := reg.Get(host) |
| 262 | if err != nil || rt == nil { |
| 263 | t.Fatalf("route %s: %+v, %v", host, rt, err) |
| 264 | } |
| 265 | return rt |
| 266 | } |
| 267 | |
| 268 | // An import aborted while its verify wake's run-unit start is queued: |
| 269 | // the rollback's fence cancels the start, #292's proof stops the unit at |
| 270 | // once, and the same pass restores the original route. |
| 271 | func TestImportRollbackAfterAnAbortedStartProvenAtOnce(t *testing.T) { |
| 272 | t.Parallel() |
| 273 | x := newImportRollback(t) |
| 274 | x.queueStart(t) |
| 275 | woke := x.wake(t) |
| 276 | waitFor(t, "the verify wake to reach its queued start", func() bool { return x.b.startCount() == 1 }) |
| 277 | |
| 278 | x.abort(t) |
| 279 | x.pass(t) |
| 280 | x.wantRestored(t) |
| 281 | if err := <-woke; err == nil { |
| 282 | t.Fatal("the aborted wake was served") |
| 283 | } |
| 284 | x.b.mu.Lock() |
| 285 | stops := x.b.legacyStops |
| 286 | x.b.mu.Unlock() |
| 287 | if stops == 0 { |
| 288 | t.Fatal("the queued start's unit was never stopped") |
| 289 | } |
| 290 | } |
| 291 | |
| 292 | // The same, with #292's proof failing at first: the rollback waits, and |
| 293 | // once the sweep's retry proves the unit gone, the next pass restores the |
| 294 | // original route, with no agent restart in between. |
| 295 | func TestImportRollbackAfterAnAbortedStartProvenLater(t *testing.T) { |
| 296 | t.Parallel() |
| 297 | x := newImportRollback(t) |
| 298 | x.b.mu.Lock() |
| 299 | x.b.legacyStopErr = errors.New("stop carlos-instance@a.shop.example.com: instance control socket: connection refused") |
| 300 | x.b.mu.Unlock() |
| 301 | x.queueStart(t) |
| 302 | woke := x.wake(t) |
| 303 | waitFor(t, "the verify wake to reach its queued start", func() bool { return x.b.startCount() == 1 }) |
| 304 | |
| 305 | x.abort(t) |
| 306 | x.pass(t) |
| 307 | if err := <-woke; err == nil { |
| 308 | t.Fatal("the aborted wake was served") |
| 309 | } |
| 310 | if !x.retained() { |
| 311 | t.Fatal("the unproven start was not retained") |
| 312 | } |
| 313 | x.wantBlocked(t) |
| 314 | |
| 315 | x.b.mu.Lock() |
| 316 | x.b.legacyStopErr = nil |
| 317 | x.b.mu.Unlock() |
| 318 | x.reap(t) |
| 319 | if x.retained() { |
| 320 | t.Fatal("the proven start is still retained") |
| 321 | } |
| 322 | x.pass(t) |
| 323 | x.wantRestored(t) |
| 324 | } |
| 325 | |
| 326 | // While #292's proof keeps failing, the rollback deletes nothing, however |
| 327 | // many passes and retries run. |
| 328 | func TestImportRollbackKeepsEverythingWhileTheStopIsUnproven(t *testing.T) { |
| 329 | t.Parallel() |
| 330 | x := newImportRollback(t) |
| 331 | x.b.mu.Lock() |
| 332 | x.b.legacyStopErr = errors.New("stop carlos-instance@a.shop.example.com: unit still active") |
| 333 | x.b.mu.Unlock() |
| 334 | x.queueStart(t) |
| 335 | woke := x.wake(t) |
| 336 | waitFor(t, "the verify wake to reach its queued start", func() bool { return x.b.startCount() == 1 }) |
| 337 | |
| 338 | x.abort(t) |
| 339 | x.pass(t) |
| 340 | <-woke |
| 341 | for range 3 { |
| 342 | x.reap(t) |
| 343 | x.pass(t) |
| 344 | } |
| 345 | if !x.retained() { |
| 346 | t.Fatal("an unproven start was let go") |
| 347 | } |
| 348 | x.wantBlocked(t) |
| 349 | if !x.a.StopUncertain(x.rec.Host) { |
| 350 | t.Fatal("the unproven stop reads as certain") |
| 351 | } |
| 352 | } |
| 353 | |
| 354 | // The rollback's fence lands while the verify wake's build record is |
| 355 | // being written (amadan #272): the recheck after the write refuses the |
| 356 | // spawn before root is asked, so nothing started, and the same pass |
| 357 | // restores the original route. |
| 358 | func TestImportRollbackFencedDuringThePendingRecord(t *testing.T) { |
| 359 | x := newImportRollback(t) |
| 360 | x.b.setIdentity("v1") |
| 361 | x.abort(t) |
| 362 | rolled := make(chan error, 1) |
| 363 | var once sync.Once |
| 364 | x.a.BeforeExec = func(ctx context.Context, host, build string) error { |
| 365 | once.Do(func() { |
| 366 | // The rollback's pass runs now, while this write is in flight, |
| 367 | // and fences the host before the write returns. |
| 368 | go func() { rolled <- provision.Reconcile(context.Background(), x.store, x.reg, x.cfg) }() |
| 369 | for deadline := time.Now().Add(5 * time.Second); !x.a.Fenced(host) && time.Now().Before(deadline); { |
| 370 | time.Sleep(time.Millisecond) |
| 371 | } |
| 372 | }) |
| 373 | return nil |
| 374 | } |
| 375 | woke := x.wake(t) |
| 376 | if err := <-woke; !errors.Is(err, ErrFenced) { |
| 377 | t.Fatalf("wake = %v, want ErrFenced", err) |
| 378 | } |
| 379 | if err := <-rolled; err != nil { |
| 380 | t.Fatal(err) |
| 381 | } |
| 382 | if x.b.startCount() != 0 { |
| 383 | t.Fatalf("started %d after the fence", x.b.startCount()) |
| 384 | } |
| 385 | x.wantRestored(t) |
| 386 | } |
| 387 | |
| 388 | // A launch its pre-exec gate refused is known not started; a start that |
| 389 | // may have run is not, and an unrelated earlier uncertainty is kept. |
| 390 | func TestALaunchRefusedBeforeExecIsNotUncertain(t *testing.T) { |
| 391 | refused := runExecGate(withExecGate(context.Background(), func() error { return ErrFenced })) |
| 392 | if !errors.Is(refused, ErrFenced) || !launchNotStarted(refused) { |
| 393 | t.Fatalf("a gate refusal = %v, not started %v", refused, launchNotStarted(refused)) |
| 394 | } |
| 395 | if launchNotStarted(markLegacyStartUncertain(refused)) || launchNotStarted(markStartUncertain(refused)) { |
| 396 | t.Fatal("a start that may have run reads as not started") |
| 397 | } |
| 398 | if launchNotStarted(errors.New("start unit: root did not answer")) { |
| 399 | t.Fatal("an unmarked failure reads as not started") |
| 400 | } |
| 401 | |
| 402 | a, _, _, reg := testActivator(t, time.Hour) |
| 403 | defer reg.Close() |
| 404 | var known stopObservation |
| 405 | known.notStarted() |
| 406 | a.mu.Lock() |
| 407 | a.rememberStopLocked("fresh.example.com", &known) |
| 408 | a.rememberStopLocked("other.example.com", &stopObservation{uncertain: true}) |
| 409 | a.rememberStopLocked("other.example.com", &known) |
| 410 | a.mu.Unlock() |
| 411 | if a.StopUncertain("fresh.example.com") { |
| 412 | t.Fatal("a launch known not started reads as uncertain") |
| 413 | } |
| 414 | if !a.StopUncertain("other.example.com") { |
| 415 | t.Fatal("a later known stop erased an unrelated uncertainty") |
| 416 | } |
| 417 | } |
| 418 | |
| 419 | // A retained start's proof discharges only its own uncertainty: one that |
| 420 | // another source added to stays. |
| 421 | func TestTheLegacyProofDischargesOnlyItsOwnUncertainty(t *testing.T) { |
| 422 | t.Parallel() |
| 423 | for _, other := range []bool{false, true} { |
| 424 | a, b, _, reg := testActivator(t, time.Hour) |
| 425 | host := "own.example.com" |
| 426 | u := &unprovenWake{rt: testRoute(host), legacy: true} |
| 427 | a.mu.Lock() |
| 428 | if other { |
| 429 | a.rememberStopLocked(host, &stopObservation{uncertain: true}) |
| 430 | } |
| 431 | a.rememberStopLocked(host, &stopObservation{uncertain: true, owedBy: u}) |
| 432 | var refused stopObservation |
| 433 | refused.notStarted() // a wake refused before it started anything |
| 434 | a.rememberStopLocked(host, &refused) |
| 435 | a.unprovenWakes = map[string]*unprovenWake{host: u} |
| 436 | a.mu.Unlock() |
| 437 | a.reapUnprovenWake(host, u) |
| 438 | b.mu.Lock() |
| 439 | stops := b.legacyStops |
| 440 | b.mu.Unlock() |
| 441 | if stops != 1 { |
| 442 | t.Fatalf("other %v: %d proofs, want 1", other, stops) |
| 443 | } |
| 444 | if got := a.StopUncertain(host); got != other { |
| 445 | t.Fatalf("other %v: uncertain after the proof = %v, want %v", other, got, other) |
| 446 | } |
| 447 | reg.Close() |
| 448 | } |
| 449 | } |
| 450 | |
| 451 | // The not-started record is only for a wake that started nothing: one |
| 452 | // that holds a process, a replica or a sidecar of its own does not get it. |
| 453 | func TestOnlyAWakeThatStartedNothingIsRecordedNotStarted(t *testing.T) { |
| 454 | if !wakeStartedNothing(&wokenInstance{}) { |
| 455 | t.Fatal("an empty wake reads as having started something") |
| 456 | } |
| 457 | for name, w := range map[string]*wokenInstance{ |
| 458 | "process": {proc: &maintenanceProc{}}, |
| 459 | "replica": {rep: &maintenanceRep{}}, |
| 460 | "sidecar": {sidecar: &maintenanceProc{}}, |
| 461 | } { |
| 462 | if wakeStartedNothing(w) { |
| 463 | t.Fatalf("a wake holding a %s reads as having started nothing", name) |
| 464 | } |
| 465 | } |
| 466 | } |
| 467 | |
| 468 | // A refusal at a gate later in the wake than the main spawn's, with the |
| 469 | // main process and its replica already running: those stages still go |
| 470 | // through the checked stop and are never marked held by the refusal. |
| 471 | // The sidecar's start (step 6) is the wake's last gate; a replica that |
| 472 | // opens and then fails is the stage before it. |
| 473 | func TestALateRefusalMarksNothingHeld(t *testing.T) { |
| 474 | t.Parallel() |
| 475 | refused := launchRefused{ErrFenced} |
| 476 | |
| 477 | t.Run("the sidecar's gate", func(t *testing.T) { |
| 478 | a, base, _, reg := testActivator(t, time.Hour) |
| 479 | defer reg.Close() |
| 480 | base.sidecarStaged = true |
| 481 | main, partial, rep := &maintenanceProc{}, &maintenanceProc{}, &maintenanceRep{} |
| 482 | a.backend = &maintenanceBackend{fakeBackend: base, main: main, side: partial, rep: rep, sideFailure: refused} |
| 483 | rt := testRoute("late-sidecar.example") |
| 484 | release, err := a.beginRequest(t.Context(), rt) |
| 485 | if err != nil { |
| 486 | t.Fatal(err) |
| 487 | } |
| 488 | release() |
| 489 | if !a.StopUncertain(rt.Host) { |
| 490 | t.Fatal("a running wake reads as stopped") |
| 491 | } |
| 492 | if main.calls.Load() != 0 || rep.calls.Load() != 0 { |
| 493 | t.Fatal("the refusal stopped the main process or closed the replica") |
| 494 | } |
| 495 | if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), rt.Host, 1); err != ErrStopUnproven || proof != nil { |
| 496 | t.Fatal("a refused sidecar's returned handle yielded trusted evidence", err) |
| 497 | } |
| 498 | if main.calls.Load() != 1 || rep.calls.Load() != 1 || partial.calls.Load() != 1 { |
| 499 | t.Fatalf("the stop skipped a stage: main %d, replica %d, sidecar %d", main.calls.Load(), rep.calls.Load(), partial.calls.Load()) |
| 500 | } |
| 501 | }) |
| 502 | |
| 503 | t.Run("after the replica opened", func(t *testing.T) { |
| 504 | a, base, _, reg := testActivator(t, time.Hour) |
| 505 | defer reg.Close() |
| 506 | main, rep := &maintenanceProc{}, &maintenanceRep{} |
| 507 | a.backend = &maintenanceBackend{fakeBackend: base, main: main, rep: rep, replicateFailure: refused} |
| 508 | rt := testRoute("late-replica.example") |
| 509 | if _, err := a.beginRequest(t.Context(), rt); !errors.Is(err, ErrFenced) { |
| 510 | t.Fatalf("wake = %v, want the refusal", err) |
| 511 | } |
| 512 | if main.calls.Load() != 1 || !main.Exited() || rep.calls.Load() != 1 { |
| 513 | t.Fatalf("the failed wake's cleanup skipped a stage: main %d, replica %d", main.calls.Load(), rep.calls.Load()) |
| 514 | } |
| 515 | if !a.StopUncertain(rt.Host) { |
| 516 | t.Fatal("a refusal after the spawn marked the wake's stages held") |
| 517 | } |
| 518 | if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), rt.Host, 1); err != ErrStopUnproven || proof != nil { |
| 519 | t.Fatal("a late refusal yielded trusted evidence", err) |
| 520 | } |
| 521 | }) |
| 522 | } |
| 523 | |
| 524 | // Astra's re-review, P2: the rollback has closed the import's gate but not |
| 525 | // yet fenced the host, and a verify request the edge admitted on its |
| 526 | // cached row reaches the activator. The wake re-reads the row, finds the |
| 527 | // gate changed and refuses before its lease. That refusal started nothing: |
| 528 | // it must leave the host's stop history as it was, and the rollback |
| 529 | // restores the original route in the same pass, with no restart. |
| 530 | func TestImportRollbackAfterACachedRowWakeRefusedBeforeItsLease(t *testing.T) { |
| 531 | t.Parallel() |
| 532 | for _, history := range []string{"none", "a clean stop"} { |
| 533 | t.Run(history, func(t *testing.T) { |
| 534 | x := newImportRollback(t) |
| 535 | cached := *x.row(t) // the edge's copy, gated by the verify hash |
| 536 | if history == "a clean stop" { |
| 537 | arrangeAwake(t, x.a, cached) |
| 538 | x.a.mu.Lock() |
| 539 | e := x.a.entries[cached.Host] |
| 540 | x.a.beginStopLocked(e, StopIdle) |
| 541 | stopped := e.stopped |
| 542 | x.a.mu.Unlock() |
| 543 | <-stopped |
| 544 | if x.a.StopUncertain(cached.Host) { |
| 545 | t.Fatal("the verify wake's clean stop reads as uncertain") |
| 546 | } |
| 547 | } |
| 548 | starts := x.b.startCount() |
| 549 | x.abort(t) |
| 550 | var refused error |
| 551 | x.beforeFence = func() { |
| 552 | release, err := x.a.beginRequest(context.Background(), cached) |
| 553 | if err == nil { |
| 554 | release() |
| 555 | } |
| 556 | refused = err |
| 557 | } |
| 558 | x.pass(t) |
| 559 | if refused == nil { |
| 560 | t.Fatal("a wake admitted on the cached row was served after the gate closed") |
| 561 | } |
| 562 | if x.b.startCount() != starts { |
| 563 | t.Fatal("the refused wake started the instance") |
| 564 | } |
| 565 | x.wantRestored(t) |
| 566 | }) |
| 567 | } |
| 568 | } |
| 569 | |
| 570 | // A refusal before anything of a wake started leaves every earlier record |
| 571 | // exactly as it was: an uncertain one with its owner, and an incomplete |
| 572 | // one. |
| 573 | func TestAWakeThatStartedNothingLeavesTheHistoryAsItWas(t *testing.T) { |
| 574 | a, _, _, reg := testActivator(t, time.Hour) |
| 575 | defer reg.Close() |
| 576 | u := &unprovenWake{legacy: true} |
| 577 | priors := map[string]*stopObservation{ |
| 578 | "owed.example.com": {uncertain: true, owedBy: u}, |
| 579 | "incomplete.example.com": {mainStop: true, mainExited: true, sidecarStop: true}, |
| 580 | } |
| 581 | var refused stopObservation |
| 582 | refused.notStarted() |
| 583 | a.mu.Lock() |
| 584 | for host, prior := range priors { |
| 585 | a.rememberStopLocked(host, prior) |
| 586 | a.rememberStopLocked(host, &refused) |
| 587 | } |
| 588 | a.rememberStopLocked("new.example.com", &refused) |
| 589 | for host, prior := range priors { |
| 590 | if a.stopHistory[host] != prior { |
| 591 | t.Fatalf("%s: the refusal replaced %+v with %+v", host, prior, a.stopHistory[host]) |
| 592 | } |
| 593 | } |
| 594 | a.mu.Unlock() |
| 595 | if a.StopUncertain("new.example.com") { |
| 596 | t.Fatal("a host whose only wake started nothing reads as uncertain") |
| 597 | } |
| 598 | } |
| 599 | |
amadan