amadan
console: the purge retires every lifetime of the app first, the expiry pass owns the store sweep, and restore evidence judges instance records three ways
purgeAppWithLifetimes runs release.PurgeAppGuards first (a refusal retires nothing), then per instance key and per pending allocation of the app, under the hostname lock: this app's pending retirements and allocations are settled by recovery's own branches (the app is trashed, so an allocation is cancelled), a present record is retired bound to the observation made in the same hold (a valid numbered record by its incarnation, anything else by its ETag), an absent key beside this app's live slot retires the slot as an orphan, and a foreign slot is never the purge's; any pending step stops the purge with the marker kept. Then release.PurgeAppExcept sweeps every other key, the marker last. The browser purge door and the expiry pass's early purge take it admitted on, under the app lock they hold; the hourly store sweep (release.ExpiredTrashApps, run after the pass's own work whether or not it has rows) purges headless apps past the cooloff under each app's lock with the marker re-read there. A step-up resume is now admitted as the held POST's own unit: the callback path is exempt from the gate, and the resumed purge used to reach the mux with no unit and take the legacy purge with lifetimes on. settleApp's restore evidence, admitted on, keeps every ordinary object's value and judges instance-record keys alone (RecordEvidenceJudged): a valid record lives, a stale body counts for nothing, an unreadable or corrupt one leaves the row trashed and logs once an hour. The claims-off materialisation admitted on settles each row under its own app lock with that evidence.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
9 files changed,
+761
−7
internal/console/expirypass.go+12 −1internal/console/lifetimes_create_test.go+7 −2internal/console/lifetimes_cutover.go+9 −1internal/console/lifetimes_purge.go+296 −0internal/console/lifetimes_purge_test.go+410 −0internal/console/ownership.go+9 −1internal/console/server.go+2 −0internal/console/settle.go+12 −1internal/console/stepup.go+4 −1
diff --git a/internal/console/expirypass.go b/internal/console/expirypass.go| index adb53b0d2..2839d36ef 100644 |
| --- a/internal/console/expirypass.go |
| +++ b/internal/console/expirypass.go |
| @@ -43,6 +43,11 @@ func (s *Server) StartExpiryPass(ctx context.Context) { |
| // retries early purges: PurgeAppIfTrashedAt acts only on a marker whose clock |
| // matches the row, and a wrong, empty bucket has no such marker. |
| func (s *Server) expiryPass(ctx context.Context, now time.Time) { |
| + // The hourly store sweep runs after the pass's own work, whether or not |
| + // the pass has rows: on a claims-off deployment it is the only purge of |
| + // an app past its trash cooloff, headless ones included, now that boxes |
| + // purge nothing (lifetimes_purge.go). |
| + defer s.trashSweep(ctx, now) |
| rows, err := s.store.TemporaryAppRows() |
| if err != nil { |
| log.Printf("console: expiry pass: list temporary apps: %v", err) |
| @@ -206,7 +211,13 @@ func (s *Server) purgeOne(ctx context.Context, r AppRow, now time.Time, ev evide |
| s.audit(acct, r.App, expiryActor, "purge-refused", "did not delete "+r.App+": it has email set up. Delete it from the Trash page") |
| return |
| } |
| - err = release.PurgeAppIfTrashedAt(ctx, s.cfg.Releases, s.cfg.ReleasePub, r.Sqid, r.App, time.Unix(life.PurgeFor, 0), now) |
| + if lifetimesUnit(ctx) { |
| + // Every lifetime of the app retired first, under the lock this |
| + // pass holds (lifetimes_purge.go). |
| + err = s.purgeAppWithLifetimes(ctx, r.Sqid, r.App, time.Unix(life.PurgeFor, 0), now) |
| + } else { |
| + err = release.PurgeAppIfTrashedAt(ctx, s.cfg.Releases, s.cfg.ReleasePub, r.Sqid, r.App, time.Unix(life.PurgeFor, 0), now) |
| + } |
| switch { |
| case err == nil: |
| if err := s.store.PurgeAppRow(r.AccountID, r.App); err != nil { |
diff --git a/internal/console/lifetimes_create_test.go b/internal/console/lifetimes_create_test.go| index c1cf95a24..5f5e37fdf 100644 |
| --- a/internal/console/lifetimes_create_test.go |
| +++ b/internal/console/lifetimes_create_test.go |
| @@ -2387,14 +2387,19 @@ func testLockOrderOfTheDoors(t *testing.T) { |
| if rows := retireRows(t, l.fx.st, "three.shop.example.com"); len(rows) != 1 || rows[0].State != lifetimeCommitted { |
| t.Fatalf("orphan retirement = %+v", rows) |
| } |
| + // The purge with lifetimes (Task 4): the app lock, then each hostname. |
| + if err := purgeCore(t, l.fx.srv, onCtx(t, l.fx.srv), time.Time{}); err != nil { |
| + t.Fatalf("purge = %v", err) |
| + } |
| c.check(t) |
| } |
| func TestRecoveryWaitsForAnotherPendingOperation(t *testing.T) { |
| l := newLtOn(t) |
| l.pendingCreate(t, ltOne, "record") |
| - // Another operation's intent for the hostname (a kind this pass does |
| - // not recover yet): the allocation is not touched while it is pending. |
| + // Another operation's intent for the hostname (a retirement with no |
| + // captured slot, which recovery cannot settle): the allocation is not |
| + // touched while it is pending. |
| if err := l.fx.st.BeginRetirement(Retirement{OpID: "other", Kind: retireNormal, Account: "bab", App: ltApp, Host: ltOne, Incarnation: 1}); err != nil { |
| t.Fatal(err) |
| } |
diff --git a/internal/console/lifetimes_cutover.go b/internal/console/lifetimes_cutover.go| index 45a7a64fc..d9734cf35 100644 |
| --- a/internal/console/lifetimes_cutover.go |
| +++ b/internal/console/lifetimes_cutover.go |
| @@ -479,7 +479,15 @@ func (s *Server) materializeAppsOnce(ctx context.Context) { |
| } |
| s.lifetimesIteration(ctx, "materialize apps", true, false, func(ctx context.Context) { |
| s.appsMaterialized.Store(true) |
| - if err := MaterializeApps(ctx, s.cfg.Releases, s.store); err != nil { |
| + // Admitted on, each candidate is settled under its own app lock |
| + // with the judged restore evidence (lifetimes_purge.go). |
| + var err error |
| + if lifetimesUnit(ctx) { |
| + err = s.materializeAppsJudged(ctx) |
| + } else { |
| + err = MaterializeApps(ctx, s.cfg.Releases, s.store) |
| + } |
| + if err != nil { |
| log.Printf("console: could not converge app rows with the bucket: %v", err) |
| } |
| }) |
diff --git a/internal/console/lifetimes_purge.go b/internal/console/lifetimes_purge.go| new file mode 100644 |
| index 000000000..c4eae65f0 |
| --- /dev/null |
| +++ b/internal/console/lifetimes_purge.go |
| @@ -0,0 +1,296 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "sort" |
| + "sync" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The purge with lifetimes (self-serve moves plan 1, Task 4). A claims-off |
| +// box purges nothing from this release on: the console's purge door, the |
| +// expiry pass's early purge and its hourly store sweep are the only purges, |
| +// and with lifetimes on each retires every lifetime of the app before the |
| +// app's name is released: |
| +// |
| +// 1. the guards (release.PurgeAppGuards): a refusal retires nothing; |
| +// 2. per instance key under the app prefix, and per pending allocation of |
| +// the app, under the hostname lock: the allocation is cancelled (the |
| +// app is trashed, so its lifetime check fails); a present record is |
| +// retired bound to the observation made under the same lock (a valid |
| +// numbered record by its incarnation, anything else by its ETag); an |
| +// absent key beside a live slot of this app retires the slot as an |
| +// orphan; a foreign slot is never the purge's; |
| +// 3. the sweep of every other key of the app, the trash marker last. |
| +// |
| +// Every caller holds s.lockApp(acct, app), the lock a create, a recovery and |
| +// a restore of the app take first, so none of them runs while the purge does. |
| + |
| +// purgeAppWithLifetimes is the purge of a trashed app with lifetimes on. want |
| +// is the marker clock the decision was made about (zero checks none). The |
| +// caller holds s.lockApp(sqid, app) and releases the apps row afterwards. Any |
| +// error leaves the marker: errLifetimePending and errLifetimeSettling stop |
| +// the purge for a later run, and every step is idempotent. |
| +func (s *Server) purgeAppWithLifetimes(ctx context.Context, sqid, app string, want, now time.Time) error { |
| + if err := release.PurgeAppGuards(ctx, s.cfg.Releases, s.cfg.ReleasePub, sqid, app, want, now); err != nil { |
| + return err |
| + } |
| + ctx = provision.WithLifetimes(ctx, s.judge) |
| + hosts, err := provision.ListRecordHosts(ctx, s.cfg.Releases, sqid, app) |
| + if err != nil { |
| + return fmt.Errorf("console: purge %s/%s: list the instance records: %w", sqid, app, err) |
| + } |
| + allocs, err := s.store.PendingLifetimesForApp(sqid, app) |
| + if err != nil { |
| + return fmt.Errorf("console: purge %s/%s: pending allocations: %w", sqid, app, err) |
| + } |
| + rets, err := s.store.PendingRetirementsCancellingFirst() |
| + if err != nil { |
| + return fmt.Errorf("console: purge %s/%s: pending retirements: %w", sqid, app, err) |
| + } |
| + byHost := map[string][]Allocation{} |
| + for _, a := range allocs { |
| + byHost[a.Host] = append(byHost[a.Host], a) |
| + } |
| + retsByHost := map[string][]Retirement{} |
| + for _, r := range rets { |
| + if r.Account == sqid && r.App == app { |
| + retsByHost[r.Host] = append(retsByHost[r.Host], r) |
| + } |
| + } |
| + seen := map[string]bool{} |
| + for _, h := range hosts { |
| + seen[h] = true |
| + } |
| + for h := range byHost { |
| + if !seen[h] { |
| + hosts, seen[h] = append(hosts, h), true |
| + } |
| + } |
| + for h := range retsByHost { |
| + if !seen[h] { |
| + hosts, seen[h] = append(hosts, h), true |
| + } |
| + } |
| + sort.Strings(hosts) |
| + for _, host := range hosts { |
| + err := s.withHostname(host, func() error { |
| + return s.purgeHostLocked(ctx, sqid, app, host, byHost[host], retsByHost[host]) |
| + }) |
| + if err != nil { |
| + return fmt.Errorf("console: purge %s/%s: %s: %w", sqid, app, host, err) |
| + } |
| + } |
| + return release.PurgeAppExcept(ctx, s.cfg.Releases, s.cfg.ReleasePub, sqid, app, want, now, func(key string) bool { |
| + return provision.IsInstanceRecordKeyOf(key, sqid, app) |
| + }) |
| +} |
| + |
| +// purgeHostLocked is step 2 for one hostname, under its lock: this app's |
| +// pending retirements and allocations settled first (recovery's own |
| +// branches; the app is trashed, so an allocation is cancelled), then the key |
| +// observed and retired in the same hold. |
| +func (s *Server) purgeHostLocked(ctx context.Context, sqid, app, host string, allocs []Allocation, rets []Retirement) error { |
| + for _, r := range rets { |
| + if r.Kind == retireOrphan { |
| + s.recoverOrphanLocked(ctx, r) |
| + } else { |
| + s.recoverRetirementLocked(ctx, r) |
| + } |
| + } |
| + for _, a := range allocs { |
| + s.recoverAllocationLocked(ctx, a) |
| + } |
| + switch n, err := s.store.PendingForHost(host, ""); { |
| + case err != nil: |
| + return err |
| + case n > 0: |
| + return errLifetimePending |
| + } |
| + obs, err := s.classifyLocked(ctx, sqid, app, host) |
| + if err != nil { |
| + return err |
| + } |
| + switch slot := obs.Slot; { |
| + case obs.Present: |
| + expect := retireExpectETag(obs.ETag) |
| + if obs.Verdict == provision.LifetimeValid && obs.Rec != nil && obs.Rec.Incarnation != 0 { |
| + expect = retireExpectIncarnation(obs.Rec.Incarnation) |
| + } |
| + if err := s.retireRecordWithLifetimeLocked(ctx, sqid, app, host, expect); err != nil { |
| + return err |
| + } |
| + case slotOwnedBy(slot, sqid, app) && slot.State == provision.SlotLive: |
| + if err := s.retireSlotOrphanLocked(ctx, host, slot, obs.SlotETag, ""); err != nil && !errors.Is(err, errLifetimeSettling) { |
| + return err |
| + } |
| + } |
| + // What is left under the key may be a body no slot vouches for; a |
| + // record still valid here is a lifetime the purge did not retire. |
| + after, err := s.classifyLocked(ctx, sqid, app, host) |
| + if err != nil { |
| + return err |
| + } |
| + if validRecord(after) { |
| + return fmt.Errorf("a record of this app still stands under %s: %w", host, errLifetimePending) |
| + } |
| + return nil |
| +} |
| + |
| +// ---- the hourly store sweep --------------------------------------------------- |
| + |
| +// trashSweepEvery is the store sweep's cadence inside the expiry pass. |
| +const trashSweepEvery = time.Hour |
| + |
| +// trashSweep is the store-driven half of the trash sweep on a claims-off |
| +// console: every app whose marker is past release.TrashCooloff, whether or |
| +// not this console has a row for it (a headless app is purged here once |
| +// boxes stop). At most once per trashSweepEvery by now. Each app is purged |
| +// under its lock with the marker re-read there; with lifetimes off it is the |
| +// legacy purge the boxes ran. |
| +func (s *Server) trashSweep(ctx context.Context, now time.Time) { |
| + if s.cfg.HostClaim || s.cfg.Releases == nil { |
| + return |
| + } |
| + s.trashSweepMu.Lock() |
| + due := s.trashSweptAt.IsZero() || now.Sub(s.trashSweptAt) >= trashSweepEvery |
| + if due { |
| + s.trashSweptAt = now |
| + } |
| + s.trashSweepMu.Unlock() |
| + if !due { |
| + return |
| + } |
| + apps, err := release.ExpiredTrashApps(ctx, s.cfg.Releases, now) |
| + if err != nil { |
| + log.Printf("console: trash sweep: list the trash: %v", err) |
| + return |
| + } |
| + for _, a := range apps { |
| + s.sweepTrashedApp(ctx, a, now) |
| + } |
| +} |
| + |
| +// trashSweepState is the sweep's clock, kept on the Server. |
| +type trashSweepState struct { |
| + trashSweepMu sync.Mutex |
| + trashSweptAt time.Time |
| +} |
| + |
| +func (s *Server) sweepTrashedApp(ctx context.Context, a release.TrashedApp, now time.Time) { |
| + defer s.lockApp(a.Account, a.App)() |
| + tomb, err := release.ReadAppTombstone(ctx, s.cfg.Releases, a.Account, a.App) |
| + if err != nil || tomb == nil || now.Sub(tomb.DeletedAt) < release.TrashCooloff { |
| + if err != nil { |
| + log.Printf("console: trash sweep: %s/%s: %v", a.Account, a.App, err) |
| + } |
| + return // restored, purged or re-trashed since the listing |
| + } |
| + acct, err := s.store.AccountBySqid(a.Account) |
| + if err != nil { |
| + log.Printf("console: trash sweep: %s/%s: %v", a.Account, a.App, err) |
| + return |
| + } |
| + var life *AppLife |
| + if acct != nil { |
| + if life, err = s.store.AppLife(acct.ID, a.App); err != nil { |
| + log.Printf("console: trash sweep: %s/%s: %v", a.Account, a.App, err) |
| + return |
| + } |
| + if life != nil && life.ControlPlane { |
| + log.Printf("console: trash sweep: %s/%s is marked as the control plane; not purging it", a.Account, a.App) |
| + return |
| + } |
| + } |
| + if lifetimesUnit(ctx) { |
| + err = s.purgeAppWithLifetimes(ctx, a.Account, a.App, tomb.DeletedAt, now) |
| + } else { |
| + err = release.PurgeAppIfTrashedAt(ctx, s.cfg.Releases, s.cfg.ReleasePub, a.Account, a.App, tomb.DeletedAt, now) |
| + } |
| + if err != nil { |
| + log.Printf("console: trash sweep: purge %s/%s: %v", a.Account, a.App, err) |
| + return |
| + } |
| + log.Printf("console: trash sweep: purged %s/%s, trashed %s, past the %s cooloff", |
| + a.Account, a.App, tomb.DeletedAt.UTC().Format(time.RFC3339), release.TrashCooloff) |
| + if life != nil && life.Trashed() { |
| + if err := s.store.PurgeAppRow(acct.ID, a.App); err != nil { |
| + log.Printf("console: trash sweep: release the name %s/%s: %v", a.Account, a.App, err) |
| + } |
| + } |
| +} |
| + |
| +// ---- restore evidence ---------------------------------------------------------- |
| + |
| +// recordEvidenceMax bounds how many instance keys restore evidence judges; |
| +// more is unknown. |
| +const recordEvidenceMax = 5000 |
| + |
| +// restoreEvidence answers settlement's "does the app still live?" from the |
| +// keys under its prefix. Before the cutover any key counts, as it always |
| +// has. With lifetimes on, an ordinary object (releases, channels, |
| +// configuration, ledgers, promotion records) keeps that value; instance |
| +// record keys alone are judged, three-valued: a valid record lives, a stale |
| +// or absent body counts for nothing, and an unreadable or corrupt one, or |
| +// too many, is unknown. |
| +func (s *Server) restoreEvidence(ctx context.Context, sqid, app string, keys []string) (lives, unknown bool, err error) { |
| + if !lifetimesUnit(ctx) || s.judge == nil { |
| + return len(keys) > 0, false, nil |
| + } |
| + ordinary, instance := false, false |
| + for _, k := range keys { |
| + if provision.IsInstanceRecordKeyOf(k, sqid, app) { |
| + instance = true |
| + } else { |
| + ordinary = true |
| + } |
| + } |
| + if ordinary { |
| + return true, false, nil |
| + } |
| + if !instance { |
| + return false, false, nil |
| + } |
| + recs, unknown, err := provision.RecordEvidenceJudged(ctx, s.cfg.Releases, s.judge, sqid, app, recordEvidenceMax) |
| + if err != nil { |
| + return false, false, err |
| + } |
| + if len(recs) > 0 { |
| + return true, false, nil |
| + } |
| + return false, unknown, nil |
| +} |
| + |
| +// materializeAppsJudged is the claims-off app materialisation admitted on: |
| +// every row settled under its own s.lockApp, the row and the marker re-read |
| +// there (settleApp), with the judged restore evidence. The listing taken |
| +// first is a hint and the empty-bucket tripwire, as in MaterializeApps. |
| +func (s *Server) materializeAppsJudged(ctx context.Context) error { |
| + if s.cfg.Releases == nil { |
| + return nil |
| + } |
| + rows, err := s.store.AppRows() |
| + if err != nil || len(rows) == 0 { |
| + return err |
| + } |
| + keys, err := s.cfg.Releases.List(ctx, "apps/") |
| + if err != nil { |
| + return err |
| + } |
| + evidence := bucketEvidence(keys) |
| + for _, r := range rows { |
| + func() { |
| + defer s.lockApp(r.Sqid, r.App)() |
| + if err := s.settleApp(ctx, r.AccountID, r.Sqid, r.App, time.Now(), func(context.Context) bool { return evidence }); err != nil { |
| + log.Printf("console: materialise %s/%s: %v", r.Sqid, r.App, err) |
| + } |
| + }() |
| + } |
| + return nil |
| +} |
diff --git a/internal/console/lifetimes_purge_test.go b/internal/console/lifetimes_purge_test.go| new file mode 100644 |
| index 000000000..86c6d1402 |
| --- /dev/null |
| +++ b/internal/console/lifetimes_purge_test.go |
| @@ -0,0 +1,410 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "net/http" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The purge with lifetimes, restore evidence and the judged materialisation |
| +// (lifetimes_purge.go), through the real doors and passes of a claims-off |
| +// console with lifetimes on (fixtures in lifetimes_create_test.go and |
| +// lifetimes_retire_test.go). |
| + |
| +// ---- the purge -------------------------------------------------------------- |
| + |
| +const ltFour = "four.shop.example.com" |
| + |
| +// purgeWorld is shop with every shape of lifetime the purge meets: a |
| +// numbered record (ltOne), a legacy record with no slot (ltOld), a pending |
| +// allocation with its live slot and no record (ltFresh) and one that never |
| +// wrote its slot (ltFour). Then shop is trashed. |
| +func purgeWorld(t *testing.T, l ltOn) (fresh, four allocRow) { |
| + t.Helper() |
| + l.mustCreate(t, ltOne) |
| + legacyBody(t, l.fx.rel, ltApp, ltOld, "stable") |
| + fresh = l.pendingCreate(t, ltFresh, "record") |
| + four = l.pendingCreate(t, ltFour, "slot") |
| + trashShop(t, l) |
| + return fresh, four |
| +} |
| + |
| +// requirePurged: every lifetime of purgeWorld retired, the marker and the |
| +// row gone. |
| +func requirePurged(t *testing.T, l ltOn, rowGone bool) { |
| + t.Helper() |
| + requireRetired(t, l.fx.rel, ltApp, ltOne, 1) |
| + requireRetired(t, l.fx.rel, ltApp, ltOld, 1) // migrated at 1, then retired |
| + requireRetired(t, l.fx.rel, ltApp, ltFresh, 1) |
| + oneAlloc(t, l.fx.st, ltFresh, lifetimeVoid) |
| + oneAlloc(t, l.fx.st, ltFour, lifetimeVoid) |
| + if slot, _ := slotOf(t, l.fx.rel, ltFour); slot != nil { |
| + t.Fatalf("slot of %s = %+v, want none", ltFour, slot) |
| + } |
| + if tomb, err := release.ReadAppTombstone(context.Background(), l.fx.rel, "bab", ltApp); err != nil || tomb != nil { |
| + t.Fatalf("marker = %+v %v, want gone", tomb, err) |
| + } |
| + if keys := appKeys(t, l.fx, ltApp); len(keys) != 0 { |
| + t.Fatalf("the bucket still holds %v", keys) |
| + } |
| + if life, _ := l.fx.st.AppLife(OpsAccountID, ltApp); rowGone && life != nil { |
| + t.Fatalf("row = %+v, want it gone", life) |
| + } |
| +} |
| + |
| +// purgeCore is the purge as every caller runs it: under shop's app lock. |
| +func purgeCore(t *testing.T, srv *Server, ctx context.Context, want time.Time) error { |
| + t.Helper() |
| + defer srv.lockApp("bab", ltApp)() |
| + return srv.purgeAppWithLifetimes(ctx, "bab", ltApp, want, time.Now()) |
| +} |
| + |
| +func TestConsolePurgeRetiresEveryLifetime(t *testing.T) { |
| + ctx := context.Background() |
| + t.Run("the browser purge", func(t *testing.T) { |
| + l := newLtOn(t) |
| + purgeWorld(t, l) |
| + c := newBrowser(t) |
| + signUp(t, l.fx, c, "code-lt-purge") |
| + csrf := sessionToken(t, l.fx, c) |
| + status, loc, _ := postForm(t, c, l.fx.u("/trash/delete"), urlValues("csrf", csrf, "app", ltApp, "confirm", ltApp)) |
| + if status != http.StatusSeeOther { |
| + t.Fatalf("purge = %d", status) |
| + } |
| + if status, _, body := completeStepUp(t, l.fx, c, loc); status != http.StatusSeeOther { |
| + t.Fatalf("resumed purge = %d %s", status, body) |
| + } |
| + requirePurged(t, l, true) |
| + }) |
| + t.Run("the expiry pass's row purge", func(t *testing.T) { |
| + l := newLtOn(t) |
| + purgeWorld(t, l) |
| + tomb, _ := release.ReadAppTombstone(ctx, l.fx.rel, "bab", ltApp) |
| + if _, err := l.fx.st.db.Exec(`UPDATE apps SET purge_for = ?, purge_after = ? WHERE app = ?`, tomb.DeletedAt.Unix(), tomb.DeletedAt.Unix()+1, ltApp); err != nil { |
| + t.Fatal(err) |
| + } |
| + now := tomb.DeletedAt.Add(time.Hour) |
| + if !l.fx.srv.lifetimesIteration(ctx, "expiry pass", true, false, func(ctx context.Context) { l.fx.srv.expiryPass(ctx, now) }) { |
| + t.Fatal("the expiry pass did not run") |
| + } |
| + requirePurged(t, l, true) |
| + }) |
| + t.Run("the hourly store sweep of an app with no console row", func(t *testing.T) { |
| + l := newLtOn(t) |
| + l.mustCreate(t, ltOne) |
| + old := time.Now().Add(-31 * 24 * time.Hour) |
| + if err := release.TrashApp(ctx, l.fx.rel, "bab", ltApp, "paul@keymail.dev", old); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.fx.st.TrashAppRow(OpsAccountID, ltApp, old); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.fx.st.PurgeAppRow(OpsAccountID, ltApp); err != nil { |
| + t.Fatal(err) |
| + } |
| + if !l.fx.srv.lifetimesIteration(ctx, "expiry pass", true, false, func(ctx context.Context) { l.fx.srv.expiryPass(ctx, time.Now()) }) { |
| + t.Fatal("the expiry pass did not run") |
| + } |
| + requireRetired(t, l.fx.rel, ltApp, ltOne, 1) |
| + if keys := appKeys(t, l.fx, ltApp); len(keys) != 0 { |
| + t.Fatalf("the bucket still holds %v", keys) |
| + } |
| + }) |
| + t.Run("the guards run first and a refused purge retires nothing", func(t *testing.T) { |
| + l := newLtOn(t) |
| + purgeWorld(t, l) |
| + err := purgeCore(t, l.fx.srv, onCtx(t, l.fx.srv), time.Unix(1, 0)) |
| + if !errors.Is(err, release.ErrTrashClockMismatch) { |
| + t.Fatalf("purge = %v, want the clock refusal", err) |
| + } |
| + for _, host := range []string{ltOne, ltOld, ltFresh, ltFour} { |
| + if rows := retireRows(t, l.fx.st, host); len(rows) != 0 { |
| + t.Fatalf("%s: retirements %+v after a refused purge", host, rows) |
| + } |
| + } |
| + oneAlloc(t, l.fx.st, ltFresh, lifetimePending) |
| + if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); raw == nil { |
| + t.Fatal("a refused purge removed a record") |
| + } |
| + }) |
| + t.Run("a pending step stops the purge, which the re-run completes", func(t *testing.T) { |
| + l := newLtOn(t) |
| + purgeWorld(t, l) |
| + l.bucket.setHook(failNth("put", recKey(ltApp, ltOne), 1, false)) // ltOne's tombstone |
| + if err := purgeCore(t, l.fx.srv, onCtx(t, l.fx.srv), time.Time{}); !errors.Is(err, errLifetimePending) { |
| + t.Fatalf("purge = %v, want pending", err) |
| + } |
| + l.bucket.setHook(nil) |
| + if tomb, _ := release.ReadAppTombstone(ctx, l.fx.rel, "bab", ltApp); tomb == nil { |
| + t.Fatal("the marker went with a lifetime still pending") |
| + } |
| + if err := purgeCore(t, l.fx.srv, onCtx(t, l.fx.srv), time.Time{}); err != nil { |
| + t.Fatalf("the re-run = %v", err) |
| + } |
| + if err := l.fx.st.PurgeAppRow(OpsAccountID, ltApp); err != nil { |
| + t.Fatal(err) |
| + } |
| + requirePurged(t, l, true) |
| + }) |
| + t.Run("a create blocked on the app lock during the purge refuses", func(t *testing.T) { |
| + l := newLtOn(t) |
| + l.mustCreate(t, ltOne) |
| + // The create's door read the app's lifetime while it was live. |
| + life, _ := l.fx.st.AppLife(OpsAccountID, ltApp) |
| + trashShop(t, l) |
| + gate, hook := pauseNth(t, "put", recKey(ltApp, ltOne), 1) // the purge's tombstone |
| + l.bucket.setHook(hook) |
| + var appWaits atomic.Int32 |
| + h := func(kind, key, event string) { |
| + if kind == "app" && key == "bab/"+ltApp && event == lockWaiting { |
| + appWaits.Add(1) |
| + } |
| + } |
| + lifetimeLockHook.Store(&h) |
| + t.Cleanup(func() { lifetimeLockHook.Store(nil) }) |
| + purged := make(chan error, 1) |
| + go func() { |
| + purged <- func() error { |
| + defer l.fx.srv.lockApp("bab", ltApp)() |
| + if err := l.fx.srv.purgeAppWithLifetimes(onCtx(t, l.fx.srv), "bab", ltApp, time.Time{}, time.Now()); err != nil { |
| + return err |
| + } |
| + return l.fx.st.PurgeAppRow(OpsAccountID, ltApp) |
| + }() |
| + }() |
| + waitFor(t, gate.entered, "the purge's tombstone") |
| + created := make(chan error, 1) |
| + go func() { |
| + rec := provision.Record{Schema: 1, Host: ltFresh, App: ltApp, Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC()} |
| + _, err := l.fx.srv.createRecordWithLifetime(onCtx(t, l.fx.srv), rec, life.Lifetime) |
| + created <- err |
| + }() |
| + waitCount(t, &appWaits, 2, "the create to wait on the app lock") |
| + gate.open() |
| + if err := <-purged; err != nil { |
| + t.Fatalf("purge = %v", err) |
| + } |
| + err := <-created |
| + if ref, ok := lifetimeCreateRefusal(ltFresh, err, msgLifetimeCreatePending); !ok || ref.Status != http.StatusNotFound { |
| + t.Fatalf("the blocked create = %v, want the app gone", err) |
| + } |
| + if rows := allocRows(t, l.fx.st, ltFresh); len(rows) != 0 { |
| + t.Fatalf("allocations = %+v, want none", rows) |
| + } |
| + }) |
| + t.Run("a recovery blocked on the app lock voids the allocation", func(t *testing.T) { |
| + l := newLtOn(t) |
| + l.pendingCreate(t, ltFresh, "record") |
| + second := l.restart(t) |
| + unlock := l.fx.srv.lockApp("bab", ltApp) |
| + done := make(chan struct{}) |
| + go func() { recoverOn(t, second); close(done) }() |
| + trashShop(t, l) |
| + unlock() |
| + <-done |
| + oneAlloc(t, l.fx.st, ltFresh, lifetimeVoid) |
| + }) |
| + t.Run("the hostname is then created by another app at N+1", func(t *testing.T) { |
| + l := newLtOn(t) |
| + l.mustCreate(t, ltOne) |
| + trashShop(t, l) |
| + if err := purgeCore(t, l.fx.srv, onCtx(t, l.fx.srv), time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := l.fx.st.CreateAppIn(OpsAccountID, "web", "paul@keymail.dev", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + enableInstances(t, l.fx, "web", provision.Manifest{Schema: 1, App: "web", Domain: "shop.example.com", Channel: "stable"}) |
| + code, _, body := rawReq(t, http.MethodPost, instancesURL(l.fx, "bab", "web"), l.member, `{"host":"`+ltOne+`"}`) |
| + if code != http.StatusCreated { |
| + t.Fatalf("web's create = %d %s", code, body) |
| + } |
| + requireServing(t, l.fx.rel, "web", ltOne, 2, "") |
| + }) |
| +} |
| + |
| +func urlValues(kv ...string) map[string][]string { |
| + out := map[string][]string{} |
| + for i := 0; i+1 < len(kv); i += 2 { |
| + out[kv[i]] = []string{kv[i+1]} |
| + } |
| + return out |
| +} |
| + |
| +// ---- restore evidence ------------------------------------------------------- |
| + |
| +// onlyInstanceKeys removes every object of shop that is not an instance |
| +// record key, so the evidence is the records alone. |
| +func onlyInstanceKeys(t *testing.T, l ltOn) { |
| + t.Helper() |
| + for _, k := range appKeys(t, l.fx, ltApp) { |
| + if !provision.IsInstanceRecordKeyOf(k, "bab", ltApp) { |
| + if err := l.fx.rel.Delete(context.Background(), k); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + } |
| +} |
| + |
| +// restoreFailed is a restore whose SQLite update failed: the marker gone, |
| +// the row still trashed. |
| +func restoreFailed(t *testing.T, l ltOn) { |
| + t.Helper() |
| + trashShop(t, l) |
| + if err := l.fx.rel.Delete(context.Background(), release.AppTombstoneKey("bab", ltApp)); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +func settleShop(t *testing.T, l ltOn) *AppLife { |
| + t.Helper() |
| + ctx := onCtx(t, l.fx.srv) |
| + defer l.fx.srv.lockApp("bab", ltApp)() |
| + if err := l.fx.srv.settleApp(ctx, OpsAccountID, "bab", ltApp, time.Now(), func(context.Context) bool { return true }); err != nil { |
| + t.Fatalf("settle = %v", err) |
| + } |
| + life, err := l.fx.st.AppLife(OpsAccountID, ltApp) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return life |
| +} |
| + |
| +func TestRestoreEvidenceKeepsOrdinaryObjects(t *testing.T) { |
| + t.Run("releases and channels, no instance records", func(t *testing.T) { |
| + l := newLtOn(t) |
| + for _, k := range []string{"apps/bab/shop/channels/stable.json", "apps/bab/shop/releases/1.0.0/manifest.json"} { |
| + if _, err := l.fx.rel.Put(context.Background(), k, []byte(`{}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + restoreFailed(t, l) |
| + if life := settleShop(t, l); life == nil || life.Trashed() { |
| + t.Fatalf("row = %+v, want restored", life) |
| + } |
| + }) |
| + t.Run("only a stale instance body", func(t *testing.T) { |
| + l := newLtOn(t) |
| + retiredSlot(t, l.fx.rel, ltOne) |
| + legacyBody(t, l.fx.rel, ltApp, ltOne, "stable") |
| + onlyInstanceKeys(t, l) |
| + restoreFailed(t, l) |
| + if life := settleShop(t, l); life != nil { |
| + t.Fatalf("row = %+v, want the name released", life) |
| + } |
| + }) |
| +} |
| + |
| +func TestRestoreEvidenceIsThreeValued(t *testing.T) { |
| + t.Run("a valid record is evidence", func(t *testing.T) { |
| + l := newLtOn(t) |
| + l.mustCreate(t, ltOne) |
| + onlyInstanceKeys(t, l) |
| + restoreFailed(t, l) |
| + if life := settleShop(t, l); life == nil || life.Trashed() { |
| + t.Fatalf("row = %+v, want restored", life) |
| + } |
| + }) |
| + t.Run("a corrupt body is unknown", func(t *testing.T) { |
| + l := newLtOn(t) |
| + if _, err := l.fx.rel.Put(context.Background(), recKey(ltApp, ltOne), []byte(`{`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + onlyInstanceKeys(t, l) |
| + restoreFailed(t, l) |
| + if life := settleShop(t, l); life == nil || !life.Trashed() { |
| + t.Fatalf("row = %+v, want it kept trashed", life) |
| + } |
| + }) |
| + t.Run("an unreadable key is unknown", func(t *testing.T) { |
| + l := newLtOn(t) |
| + l.mustCreate(t, ltOne) |
| + onlyInstanceKeys(t, l) |
| + restoreFailed(t, l) |
| + l.bucket.setHook(failNth("get", recKey(ltApp, ltOne), 1, true)) |
| + life := settleShop(t, l) |
| + l.bucket.setHook(nil) |
| + if life == nil || !life.Trashed() { |
| + t.Fatalf("row = %+v, want it kept trashed", life) |
| + } |
| + }) |
| +} |
| + |
| +func TestStaleBodyIsNotRestoreEvidence(t *testing.T) { |
| + l := newLtOn(t) |
| + l.mustCreate(t, ltOne) |
| + trashShop(t, l) |
| + // The purge ran; its PurgeAppRow write failed. |
| + if err := purgeCore(t, l.fx.srv, onCtx(t, l.fx.srv), time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + // A late write lands a body under the retired slot. |
| + late, _ := json.Marshal(provision.Record{Schema: 1, Host: ltOne, App: ltApp, Account: "bab", Channel: "stable", Incarnation: 1, CreatedAt: time.Now().UTC()}) |
| + if _, err := l.fx.rel.Put(context.Background(), recKey(ltApp, ltOne), late, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + if life := settleShop(t, l); life != nil { |
| + t.Fatalf("row = %+v, want the name released, not restored", life) |
| + } |
| +} |
| + |
| +func TestMaterializationHoldsTheAppLock(t *testing.T) { |
| + run := func(t *testing.T, l ltOn, meanwhile func()) { |
| + t.Helper() |
| + var appWaits atomic.Int32 |
| + h := func(kind, key, event string) { |
| + if kind == "app" && key == "bab/"+ltApp && event == lockWaiting { |
| + appWaits.Add(1) |
| + } |
| + } |
| + lifetimeLockHook.Store(&h) |
| + t.Cleanup(func() { lifetimeLockHook.Store(nil) }) |
| + unlock := l.fx.srv.lockApp("bab", ltApp) |
| + done := make(chan error, 1) |
| + go func() { done <- l.fx.srv.materializeAppsJudged(onCtx(t, l.fx.srv)) }() |
| + waitCount(t, &appWaits, 2, "materialisation to wait on the app lock") |
| + meanwhile() |
| + unlock() |
| + if err := <-done; err != nil { |
| + t.Fatalf("materialise = %v", err) |
| + } |
| + } |
| + t.Run("settled and recreated as a new lifetime", func(t *testing.T) { |
| + l := newLtOn(t) |
| + onlyInstanceKeys(t, l) |
| + restoreFailed(t, l) // trashed, no marker, no keys: a purge to finish |
| + var lifetime string |
| + run(t, l, func() { |
| + if err := l.fx.st.PurgeAppRow(OpsAccountID, ltApp); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := l.fx.st.CreateAppIn(OpsAccountID, ltApp, "paul@keymail.dev", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + life, _ := l.fx.st.AppLife(OpsAccountID, ltApp) |
| + lifetime = life.Lifetime |
| + }) |
| + if life, _ := l.fx.st.AppLife(OpsAccountID, ltApp); life == nil || life.Trashed() || life.Lifetime != lifetime { |
| + t.Fatalf("row = %+v, want the new lifetime untouched", life) |
| + } |
| + }) |
| + t.Run("re-trashed while it waited", func(t *testing.T) { |
| + l := newLtOn(t) |
| + restoreFailed(t, l) // trashed, no marker, objects: a restore to finish |
| + run(t, l, func() { |
| + if err := release.TrashApp(context.Background(), l.fx.rel, "bab", ltApp, "paul@keymail.dev", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + }) |
| + if life, _ := l.fx.st.AppLife(OpsAccountID, ltApp); life == nil || !life.Trashed() { |
| + t.Fatalf("row = %+v, want it still in the trash", life) |
| + } |
| + }) |
| +} |
diff --git a/internal/console/ownership.go b/internal/console/ownership.go| index 72cc1eb9f..f8df27cdf 100644 |
| --- a/internal/console/ownership.go |
| +++ b/internal/console/ownership.go |
| @@ -511,7 +511,15 @@ func (s *Server) handlePurgeApp(w http.ResponseWriter, r *http.Request) { |
| s.audit(acct, app, addr, "email-teardown", "tore down the email estate (credentials, IAM principal, domain claims) during permanent deletion") |
| } |
| } |
| - if err := release.PurgeApp(r.Context(), s.cfg.Releases, s.cfg.ReleasePub, acct.Sqid, app, time.Now()); err != nil { |
| + var purgeErr error |
| + if lifetimesUnit(r.Context()) { |
| + // Admitted on: every lifetime of the app is retired first, under |
| + // the app lock this door holds (lifetimes_purge.go). |
| + purgeErr = s.purgeAppWithLifetimes(r.Context(), acct.Sqid, app, time.Time{}, time.Now()) |
| + } else { |
| + purgeErr = release.PurgeApp(r.Context(), s.cfg.Releases, s.cfg.ReleasePub, acct.Sqid, app, time.Now()) |
| + } |
| + if err := purgeErr; err != nil { |
| log.Printf("console: purge %s/%s: %v", acct.Sqid, app, err) |
| fail(purgeAdvice(err, app)) |
| return |
diff --git a/internal/console/server.go b/internal/console/server.go| index e24151ad9..1c36862d1 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -1155,6 +1155,8 @@ type Server struct { |
| // written. Zero values are ready. |
| hostLocks sync.Map |
| lifetimeLogged sync.Map |
| + // The trash sweep's clock (lifetimes_purge.go). |
| + trashSweepState |
| // relayNow is the relay paths' clock seam: nil means time.Now. Tests |
| // drive lease expiry with it; nothing in production sets it. |
| relayNow func() time.Time |
diff --git a/internal/console/settle.go b/internal/console/settle.go| index 611722f0d..a708b4081 100644 |
| --- a/internal/console/settle.go |
| +++ b/internal/console/settle.go |
| @@ -107,7 +107,18 @@ func (s *Server) settleApp(ctx context.Context, accountID int64, sqid, app strin |
| if err != nil { |
| return err |
| } |
| - if len(keys) > 0 { |
| + // Admitted on, instance-record keys are judged three ways and every |
| + // other object keeps its value (lifetimes_purge.go restoreEvidence). |
| + lives, unknown, err := s.restoreEvidence(ctx, sqid, app, keys) |
| + if err != nil { |
| + return err |
| + } |
| + if unknown { |
| + s.lifetimeLogHourly("restore-evidence "+sqid+"/"+app, "console: %s/%s has no trash marker and an instance record that could not be judged; "+ |
| + "leaving the row trashed until it can", sqid, app) |
| + return nil |
| + } |
| + if lives { |
| if err := s.store.FinishRestoreRow(accountID, app, now); err != nil { |
| return err |
| } |
diff --git a/internal/console/stepup.go b/internal/console/stepup.go| index b8869d37e..39395198d 100644 |
| --- a/internal/console/stepup.go |
| +++ b/internal/console/stepup.go |
| @@ -430,5 +430,8 @@ func (s *Server) resume(w http.ResponseWriter, r *http.Request, p *pendingAction |
| } |
| req.Host = r.Host |
| req.RemoteAddr = r.RemoteAddr |
| - s.mux.ServeHTTP(w, req) |
| + // Admitted as the held POST's own unit of record work (the callback |
| + // path is exempt from the gate; the action it resumes is not), so a |
| + // resumed purge or teardown dispatches by the unit like any other. |
| + s.serveAdmitted(w, req) |
| } |