amadan

carlos / platform Public

console: every claims-off record removal retires through a persisted retirement of a stated kind, with the caller's expectation and the owned slot retired before any absence shows

retireRecordWithLifetime (s.lockApp, then the hostname) and its Locked form
admit, observe, apply the caller's RetireExpectation (absent, an ETag, the
incarnation it made, or whatever is current), then begin a record_retirements
row of one kind and run one dispatcher at the first entry and at every
recovery: normal (N1 the exact bytes become their tombstone by CAS, N2 the
slot retires from its captured ETag, N3 the tombstone goes on its ETag),
orphan (O1, Task 3's) and cleanup (C1 the slot first, C2 the stale or corrupt
body on its ETag). A legacy record migrates first, bound to the caller's
read. A corrupt body is quarantined and the row converts to cleanup. A
retirement ends refused or void on a changed record only after fenceRecord,
whose own-tombstone test now needs the operation's incarnation; on a
successor slot it is void with no record fence. Recovery runs the normal and
cleanup kinds through the same dispatcher.

deleteUnderRetiredSlot observes afresh before every attempt and deletes only
while the slot is still the key's at N, retired: its retry used to re-read
the body alone, so a second console that finished the retirement and created
the hostname's next lifetime could have its record taken
(TestDelayedDeleteCannotEraseASuccessor).

The doors, admitted on: the delete door's import rule judges its read
(readDeletedRecord returns the Observation), retires an absent key's own
live slot, and removes a stale or corrupt body and an ordinary record
through the retirement bound to its read; the seed finish, the sibling undo
(bound to the lifetime the create made) and the claims-off retireHostRecord
go through it too; both detach doors retire with retireAny inside the two
locks they already hold. errLifetimeSettling and errRetireRefused answer
409, a pending retirement 502 (or 409 in progress for detach). A worker
iteration now carries its admitted unit in its context, so the seeds sweep's
finish dispatches by it.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 75161c9377fb9d88432fc359fe137f161348d7df parent 8d46150
13 files changed, +1905 −80
  • internal/console/apidomains.go +15 −2
  • internal/console/apiinstanceimport.go +78 −32
  • internal/console/apiinstances.go +17 −7
  • internal/console/claim_retire.go +5 −2
  • internal/console/domainattach.go +36 −3
  • internal/console/hostclaimops.go +5 −0
  • internal/console/lifetimes_create.go +25 −13
  • internal/console/lifetimes_create_test.go +21 −10
  • internal/console/lifetimes_cutover.go +3 −1
  • internal/console/lifetimes_recovery.go +12 −9
  • internal/console/lifetimes_retire.go +375 −0
  • internal/console/lifetimes_retire_test.go +1305 −0
  • internal/console/seedfinish.go +8 −1
diff --git a/internal/console/apidomains.go b/internal/console/apidomains.go
index c5a961fa7..a17674e4a 100644
--- a/internal/console/apidomains.go
+++ b/internal/console/apidomains.go
@@ -993,8 +993,21 @@ func (s *Server) handleAPIDomainDetach(w http.ResponseWriter, r *http.Request) {
// another owner's, an alias) skips it silently and DetachClaim's own
// neutrality answers as before.
if cur, err := domains.GetClaim(ctx, s.cfg.Releases, host); err == nil &&
- cur.Account == acct.Sqid && cur.App == app && cur.Route == "" {
- if err := s.retireInstanceRecord(ctx, acct.Sqid, app, host, principal, cur.RecordIncarnation); err != nil {
+ cur.Account == acct.Sqid && cur.App == app && cur.Route == "" && lifetimesUnit(ctx) {
+ // Admitted on: the record retires through its tombstone inside the
+ // two locks held above (domainattach.go detachRetireLocked).
+ ref, err := s.detachRetireLocked(ctx, acct.Sqid, app, host)
+ if err != nil {
+ log.Printf("console: api domains detach %s/%s %s: record delete: %v", acct.Sqid, app, host, err)
+ jsonError(w, http.StatusBadGateway, "could not detach the hostname — try again")
+ return
+ }
+ if ref != nil {
+ jsonError(w, ref.Status, ref.Msg)
+ return
+ }
+ } else if err == nil && cur.Account == acct.Sqid && cur.App == app && cur.Route == "" {
+ if err := s.retireInstanceRecord(ctx, acct.Sqid, app, host, principal, cur.RecordIncarnation, retireAny); err != nil {
if accepted, ok := claimRetireAcceptedOf(err); ok {
s.audit(acct, app, principal, "domain-detach", "detach of hostname "+host+" accepted, operation "+accepted.operation+" (cli)")
s.bumpSerial(ctx)
diff --git a/internal/console/apiinstanceimport.go b/internal/console/apiinstanceimport.go
index 38be3f9fb..7906e1e1b 100644
--- a/internal/console/apiinstanceimport.go
+++ b/internal/console/apiinstanceimport.go
@@ -809,28 +809,33 @@ func (s *Server) finishImport(r *http.Request, sqid, app, host, id string) error
return s.finishSeed(ctx, sqid, app, host, id)
}
-// readDeletedRecord is the delete door's read of host's record for its
-// import rule, with the etag the ordinary delete is then bound to. Only
-// the store read can fail it (preflight M15, review I1): a body that will
-// not decode, or that disagrees with its own path, is no import this
-// console can finish, so it answers nil and the ordinary delete removes
-// it as it always has.
-func (s *Server) readDeletedRecord(ctx context.Context, sqid, app, host string) (*provision.Record, string, error) {
+// readDeletedRecord is the delete door's observation of host's record for
+// its import rule, with the etag the ordinary delete is then bound to.
+// Admitted on it is the judge's (s.judge.Observe: the record and its slot);
+// before the cutover the same read with no judge, which reads no slot. Only
+// an unreadable observation fails it (preflight M15, review I1): a body that
+// will not decode, or that disagrees with its own path, is no import this
+// console can finish, so it comes back corrupt and the ordinary delete
+// removes it as it always has.
+func (s *Server) readDeletedRecord(ctx context.Context, sqid, app, host string) (provision.Observation, error) {
ctx, cancel := context.WithTimeout(ctx, instanceStoreTimeout)
defer cancel()
- raw, etag, err := s.cfg.Releases.Get(ctx, provision.RecordKey(sqid, app, host))
- if err != nil {
- return nil, "", err
- }
- rec, err := provision.DecodeRecord(raw)
- if err == nil && (rec.Account != sqid || rec.App != app || rec.Host != host) {
- err = errors.New("the record disagrees with its own path")
- }
- if err != nil {
- log.Printf("console: api instances delete %s/%s/%s: the record could not be read as an import (%v), so it is deleted the ordinary way", sqid, app, host, err)
- return nil, etag, nil
+ var j *provision.LifetimeJudge
+ if lifetimesUnit(ctx) {
+ j = s.judge
+ }
+ obs := j.Observe(ctx, s.cfg.Releases, sqid, app, host)
+ switch obs.Verdict {
+ case provision.LifetimeUnreadable:
+ err := obs.Err
+ if err == nil {
+ err = provision.ErrLifetimeUnreadable
+ }
+ return obs, err
+ case provision.LifetimeCorrupt:
+ log.Printf("console: api instances delete %s/%s/%s: the record could not be read as an import (%v), so it is deleted the ordinary way", sqid, app, host, obs.Err)
}
- return &rec, etag, nil
+ return obs, nil
}
// handleAPIImportAbort abandons an import before it is committed (spec
@@ -942,20 +947,38 @@ func (s *Server) refuseImportDelete(w http.ResponseWriter, r *http.Request, acct
// only while it is the one the rule read (deleteRead).
func (s *Server) importDeleteRule(w http.ResponseWriter, r *http.Request, acct *Account, app, host string, again bool) (answered, finished bool) {
sqid := acct.Sqid
- rec, etag, err := s.readDeletedRecord(r.Context(), sqid, app, host)
- if errors.Is(err, os.ErrNotExist) {
+ obs, err := s.readDeletedRecord(r.Context(), sqid, app, host)
+ if err != nil {
+ log.Printf("console: api instances delete %s/%s/%s: read the record: %v", sqid, app, host, err)
+ jsonError(w, http.StatusBadGateway, importDeleteFailed)
+ return true, false
+ }
+ if !obs.Present {
// With host claims on, a record can be gone while its retirement
// is not (parking, with obligations left): the claim's own flow
// answers that, pending or done (re-review finding 10).
if s.cfg.HostClaim {
return false, false
}
+ // Admitted on, absence still retires the key's own live slot (an
+ // orphan); a record created since the read is someone else's and
+ // is left (lifetimes_retire.go).
+ if lifetimesUnit(r.Context()) {
+ ctx, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout)
+ err := s.retireRecordWithLifetime(ctx, sqid, app, host, retireExpectAbsent)
+ cancel()
+ if err != nil {
+ return s.deleteRefused(w, sqid, app, host, err)
+ }
+ }
return true, true
}
- if err != nil {
- log.Printf("console: api instances delete %s/%s/%s: read the record: %v", sqid, app, host, err)
- jsonError(w, http.StatusBadGateway, importDeleteFailed)
- return true, false
+ etag := obs.ETag
+ rec := obs.Rec
+ if obs.Verdict != provision.LifetimeValid {
+ // Stale under its slot, or corrupt: no import this console can
+ // finish; the ordinary delete, bound to this read.
+ rec = nil
}
if rec == nil || !rec.Seed.Import() || rec.Seed.State == seeds.StateCommitted {
return s.deleteRead(w, r, acct, app, host, etag, again)
@@ -1009,13 +1032,31 @@ func (s *Server) importDeleteRule(w http.ResponseWriter, r *http.Request, acct *
return true, false
}
if err := s.finishImport(r, sqid, app, host, id); err != nil {
- log.Printf("console: api instances delete %s/%s/%s: finish import seed %s: %v", sqid, app, host, id, err)
- jsonError(w, http.StatusBadGateway, importDeleteFailed)
- return true, false
+ return s.deleteRefused(w, sqid, app, host, fmt.Errorf("finish import seed %s: %w", id, err))
}
return true, true
}
+// deleteRefused answers a delete whose removal failed: a lifetime's own
+// refusal (settling, refused, pending) as itself, anything else main's
+// delete sentence.
+func (s *Server) deleteRefused(w http.ResponseWriter, sqid, app, host string, err error) (answered, finished bool) {
+ if ref, ok := lifetimeRetireRefusal(err, msgLifetimeSettling, msgLifetimeDeletePending, http.StatusBadGateway); ok {
+ if !errors.Is(err, errLifetimeSettling) {
+ log.Printf("console: api instances delete %s/%s/%s: %v", sqid, app, host, err)
+ }
+ jsonError(w, ref.Status, ref.Msg)
+ return true, false
+ }
+ log.Printf("console: api instances delete %s/%s/%s: %v", sqid, app, host, err)
+ status := http.StatusBadGateway
+ if errors.Is(err, errRetireExpectation) {
+ status = http.StatusInternalServerError // a programming error, not the bucket
+ }
+ jsonError(w, status, importDeleteFailed)
+ return true, false
+}
+
// deleteRead is the ordinary delete of host's record, conditional on the
// etag the import rule read it under (final review F7). A record gone
// since is deleted already. A record that changed since may now be an
@@ -1031,14 +1072,19 @@ func (s *Server) deleteRead(w http.ResponseWriter, r *http.Request, acct *Accoun
}
ctx, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout)
defer cancel()
- err := release.DeleteIfMatch(ctx, s.cfg.Releases, provision.RecordKey(acct.Sqid, app, host), etag)
+ var err error
+ if lifetimesUnit(ctx) {
+ // Admitted on: a retirement through the tombstone, bound to this
+ // read (lifetimes_retire.go).
+ err = s.retireRecordWithLifetime(ctx, acct.Sqid, app, host, retireExpectETag(etag))
+ } else {
+ err = release.DeleteIfMatch(ctx, s.cfg.Releases, provision.RecordKey(acct.Sqid, app, host), etag)
+ }
switch {
case err == nil:
return true, true
case errors.Is(err, release.ErrCASConflict) && again:
return s.importDeleteRule(w, r, acct, app, host, false)
}
- log.Printf("console: api instances delete %s/%s/%s: %v", acct.Sqid, app, host, err)
- jsonError(w, http.StatusBadGateway, importDeleteFailed)
- return true, false
+ return s.deleteRefused(w, acct.Sqid, app, host, err)
}
diff --git a/internal/console/apiinstances.go b/internal/console/apiinstances.go
index 2b8228f95..65efe8c61 100644
--- a/internal/console/apiinstances.go
+++ b/internal/console/apiinstances.go
@@ -1709,10 +1709,13 @@ func (s *Server) handleAPIInstanceCreate(w http.ResponseWriter, r *http.Request)
// re-check a rename makes (appsettings.go).
if siblingLabel != "" {
if live, trashed, err := s.store.AppNameState(acct.ID, siblingLabel); err == nil && (live || trashed) {
- // Today's chain in both modes. Task 4 binds it to the lifetime
- // this create made (RetireExpectation{Incarnation:
- // rec.Incarnation}) through retireRecordWithLifetime.
- if derr := s.retireInstanceRecord(ctx, acct.Sqid, app, host, principal, mustCreateIncarnation(acct.Sqid, app, host, rec.CreatedAt)); derr != nil {
+ // Admitted on, the undo retires only the lifetime this create
+ // made: a successor of the hostname since is left untouched.
+ undo := retireAny
+ if lifetimesUnit(ctx) {
+ undo = retireExpectIncarnation(rec.Incarnation)
+ }
+ if derr := s.retireInstanceRecord(ctx, acct.Sqid, app, host, principal, mustCreateIncarnation(acct.Sqid, app, host, rec.CreatedAt), undo); derr != nil {
log.Printf("console: api instances create %s/%s/%s: %s was claimed mid-create and UNDO FAILED: %v", acct.Sqid, app, host, siblingLabel, derr)
}
if privateBucket != "" {
@@ -2258,8 +2261,9 @@ func (s *Server) handleAPIInstanceDelete(w http.ResponseWriter, r *http.Request)
return
}
// Admitted on, a hostname whose create, retirement or migration is
- // still settling is not deleted under it (lifetimes.go). The delete
- // itself is today's until Task 4 makes it a retirement.
+ // still settling is refused up front (lifetimes.go); the delete itself
+ // is a retirement (importDeleteRule, lifetimes_retire.go), which admits
+ // again under the locks.
if lifetimesUnit(ctx) {
switch err := s.hostSettling(acct.Sqid, app, host); {
case errors.Is(err, errLifetimeSettling):
@@ -2287,7 +2291,13 @@ func (s *Server) handleAPIInstanceDelete(w http.ResponseWriter, r *http.Request)
}
return
}
- if err := s.retireInstanceRecord(ctx, acct.Sqid, app, host, principal, ""); err != nil {
+ // Reached on a claim console only: importDeleteRule answers every
+ // claims-off delete.
+ if err := s.retireInstanceRecord(ctx, acct.Sqid, app, host, principal, "", retireAny); err != nil {
+ if ref, ok := lifetimeRetireRefusal(err, msgLifetimeSettling, msgLifetimeDeletePending, http.StatusBadGateway); ok {
+ jsonError(w, ref.Status, ref.Msg)
+ return
+ }
// An accepted delete is this principal's durable request: audit it
// and announce it before answering 202 with the operation.
if accepted, ok := claimRetireAcceptedOf(err); ok {
diff --git a/internal/console/claim_retire.go b/internal/console/claim_retire.go
index 48714f96a..faf5f44d2 100644
--- a/internal/console/claim_retire.go
+++ b/internal/console/claim_retire.go
@@ -297,8 +297,11 @@ func (s *Server) releaseClaimRetireMarkers(ctx context.Context, h *hostops.Opera
return err
}
-func (s *Server) retireInstanceRecord(ctx context.Context, account, app, host, by, incarnation string) error {
- return s.retireHostRecord(ctx, provision.RetireRef{Family: "instance", Account: account, App: app, Host: host, By: by, Incarnation: incarnation})
+// retireInstanceRecord retires an instance record. expect is the claims-off
+// lifetime retirement's expectation (lifetimes_retire.go); the claims-on
+// chain and the legacy delete ignore it.
+func (s *Server) retireInstanceRecord(ctx context.Context, account, app, host, by, incarnation string, expect provision.RetireExpectation) error {
+ return s.retireHostRecord(ctx, provision.RetireRef{Family: "instance", Account: account, App: app, Host: host, By: by, Incarnation: incarnation, Expect: expect})
}
func mustCreateIncarnation(account, app, host string, created time.Time) string {
diff --git a/internal/console/domainattach.go b/internal/console/domainattach.go
index 8adac6a1c..6c8e2d536 100644
--- a/internal/console/domainattach.go
+++ b/internal/console/domainattach.go
@@ -351,11 +351,23 @@ func (s *Server) detachDomain(ctx context.Context, acct *Account, app, host, pri
if claim.Account != acct.Sqid || claim.App != app {
return notAttached
}
- if claim.Route == "" {
+ if claim.Route == "" && lifetimesUnit(ctx) {
+ // First-route, admitted on: the record retires through its
+ // tombstone inside the two locks held above, before the claim.
+ ref, err := s.detachRetireLocked(ctx, acct.Sqid, app, host)
+ if err != nil {
+ log.Printf("console: detach %s: record delete: %v", host, err)
+ return &doorRefusal{http.StatusInternalServerError,
+ "could not remove this domain's route — it is still attached, try again"}
+ }
+ if ref != nil {
+ return ref
+ }
+ } else if claim.Route == "" {
// First-route: the record goes before the claim, the mirror of
// attach. A record left behind by a released claim would keep a
// route alive for a host anyone may now claim.
- if err := s.retireInstanceRecord(ctx, acct.Sqid, app, host, "domain-detach", claim.RecordIncarnation); err != nil {
+ if err := s.retireInstanceRecord(ctx, acct.Sqid, app, host, "domain-detach", claim.RecordIncarnation, retireAny); err != nil {
// The record's delete operation is durable and still running
// (or another operation holds the app): the hostname stays
// claimed until a later detach finds the record gone.
@@ -395,6 +407,27 @@ func (s *Server) detachDomain(ctx context.Context, acct *Account, app, host, pri
return nil
}
+// detachRetireLocked is a claims-off detach's record removal admitted on:
+// retireRecordWithLifetimeLocked(..., retireAny) inside the app lock and the
+// hostname exclusion the caller holds. (nil, nil) means the record is gone
+// and the claim may be released. A refusal is the lifetime's answer (the
+// claim stays); an ordinary error each door answers with its own sentence.
+func (s *Server) detachRetireLocked(ctx context.Context, sqid, app, host string) (*doorRefusal, error) {
+ err := s.retireRecordWithLifetimeLocked(ctx, sqid, app, host, retireAny)
+ if err == nil {
+ return nil, nil
+ }
+ // Settling and pending both leave the route's removal to finish first:
+ // a later detach finds the record gone and releases the claim.
+ if ref, ok := lifetimeRetireRefusal(err, claimDetachInProgressMsg, claimDetachInProgressMsg, http.StatusConflict); ok {
+ if !errors.Is(err, errLifetimeSettling) {
+ log.Printf("console: detach %s: record retirement: %v", host, err)
+ }
+ return ref, nil
+ }
+ return nil, err
+}
+
// readyRouteHost answers "does this app already have a route to alias?" with
// the host of the first one, or "" for none. The selection is
// domains.firstReadyUpstream's, minus the upstream copy: only the HOST is
@@ -563,7 +596,7 @@ func (s *Server) rollbackAttach(ctx context.Context, acct *Account, app, host st
return
}
if dropRecord {
- if err := s.retireInstanceRecord(ctx, acct.Sqid, app, host, "attach-rollback", ""); err != nil {
+ if err := s.retireInstanceRecord(ctx, acct.Sqid, app, host, "attach-rollback", "", retireAny); err != nil {
log.Printf("console: attach %s: record rollback: %v", host, err)
}
}
diff --git a/internal/console/hostclaimops.go b/internal/console/hostclaimops.go
index 152aa34c4..65c31b8c1 100644
--- a/internal/console/hostclaimops.go
+++ b/internal/console/hostclaimops.go
@@ -16,6 +16,11 @@ func (s *Server) retireHostRecord(ctx context.Context, ref provision.RetireRef)
if s.cfg.HostClaim {
return s.retireClaimHostRecord(ctx, ref)
}
+ if ref.Family == "instance" && lifetimesUnit(ctx) {
+ // Claims-off, admitted on: a retirement through a tombstone, bound to
+ // the caller's own expectation (lifetimes_retire.go).
+ return s.retireRecordWithLifetime(ctx, ref.Account, ref.App, ref.Host, ref.Expect)
+ }
var claims provision.ClaimRetirer
if s.cfg.HostClaimOps != nil {
claims = s.cfg.HostClaimOps
diff --git a/internal/console/lifetimes_create.go b/internal/console/lifetimes_create.go
index cc1dfce25..e88aa2b58 100644
--- a/internal/console/lifetimes_create.go
+++ b/internal/console/lifetimes_create.go
@@ -419,21 +419,30 @@ func (s *Server) retireSlotLocked(ctx context.Context, r Retirement) (slotDispos
}
// deleteUnderRetiredSlot removes any body under r's key on its ETag: every
-// body under a retired slot is stale. A conflict means another body landed
-// meanwhile, stale all the same: with again it is re-read and deleted on its
-// ETag once more (a further conflict is an error), without it the conflict
-// is logged and left.
+// body under a retired slot is stale. Each attempt is a fresh observation
+// (the record, then the slot), and it deletes only while that slot is still
+// r's key at N, retired: a body read before such a slot cannot be a
+// successor's record, which is written only after the slot moves on. A slot
+// that has moved on leaves nothing under it for r to remove (nil). A
+// conflict means another body landed meanwhile, stale all the same: with
+// again it is observed and deleted once more (a further conflict is an
+// error), without it the conflict is logged and left.
func (s *Server) deleteUnderRetiredSlot(ctx context.Context, r Retirement, again bool) error {
key := provision.RecordKey(r.Account, r.App, r.Host)
for round := 0; ; round++ {
- _, etag, err := s.cfg.Releases.Get(ctx, key)
- if errors.Is(err, os.ErrNotExist) {
- return nil
- }
+ obs, err := s.classifyLocked(ctx, r.Account, r.App, r.Host)
if err != nil {
return err
}
- err = release.DeleteIfMatch(ctx, s.cfg.Releases, key, etag)
+ if !obs.Present {
+ return nil
+ }
+ slot := obs.Slot
+ if slot == nil || !slotOwnedBy(slot, r.Account, r.App) || slot.Incarnation != r.Incarnation || slot.State != provision.SlotRetired {
+ log.Printf("console: lifetimes: retirement %s of %s: the slot moved on (%+v); the body under %s is not this retirement's to remove", r.OpID, r.Host, slot, key)
+ return nil
+ }
+ err = release.DeleteIfMatch(ctx, s.cfg.Releases, key, obs.ETag)
switch {
case err == nil, !errors.Is(err, release.ErrCASConflict):
return err
@@ -449,10 +458,13 @@ func (s *Server) deleteUnderRetiredSlot(ctx context.Context, r Retirement, again
// ---- the fence ---------------------------------------------------------------
// fenceOwner is what this operation's own outstanding write would look like:
-// its tombstone, or a record marked with MarkerID.
+// its tombstone at Incarnation (a retirement's N; 0 for a legacy body's
+// tombstone), or a record marked with MarkerID. A tombstone of another
+// incarnation is not this operation's: the fence rewrites it like any body.
type fenceOwner struct {
- Tombstone bool
- MarkerID string
+ Tombstone bool
+ Incarnation uint64
+ MarkerID string
}
// fenceResult is fenceRecord's answer.
@@ -498,7 +510,7 @@ func (s *Server) fenceRecord(ctx context.Context, acct, app, host string, op fen
switch {
case !obs.Present:
return fenceAbsent, obs, nil
- case obs.Rec != nil && op.Tombstone && obs.Rec.Retired:
+ case obs.Rec != nil && op.Tombstone && obs.Rec.Retired && obs.Rec.Incarnation == op.Incarnation:
return ownWriteWon, obs, nil
case obs.Rec != nil && op.MarkerID != "" && obs.Rec.IncarnationMigrating == op.MarkerID:
return ownWriteWon, obs, nil
diff --git a/internal/console/lifetimes_create_test.go b/internal/console/lifetimes_create_test.go
index 21ed0aca1..c1cf95a24 100644
--- a/internal/console/lifetimes_create_test.go
+++ b/internal/console/lifetimes_create_test.go
@@ -1327,7 +1327,18 @@ func TestCreateChecksTheAppUnderTheLock(t *testing.T) {
msg string
}{
{"trashed", trashShop, http.StatusConflict, msgLifetimeAppTrashed},
+ {"trashed, the row update failed after the marker landed", func(t *testing.T, l ltOn) {
+ if err := release.TrashApp(context.Background(), l.fx.rel, "bab", ltApp, "paul@keymail.dev", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ }, http.StatusConflict, msgLifetimeAppTrashed},
{"recreated", recreateShop, http.StatusConflict, msgLifetimeAppChanged},
+ {"recreated by the temporary constructor", func(t *testing.T, l ltOn) {
+ purgeShop(t, l)
+ if _, err := l.fx.st.CreateTemporaryAppIn(OpsAccountID, ltApp, "paul@keymail.dev", time.Now(), time.Now().Add(time.Hour)); err != nil {
+ t.Fatal(err)
+ }
+ }, http.StatusConflict, msgLifetimeAppChanged},
{"purged", purgeShop, http.StatusNotFound, "404 page not found"},
}
for _, c := range cases {
@@ -1776,7 +1787,7 @@ func TestCompetingAttachCannotReleaseAnotherAttachsClaim(t *testing.T) {
if got := claimHolder(t, l.fx.rel, ltDomain); got != "bab/shop2" {
t.Fatalf("claim = %q, want shop2's", got)
}
- recDel := l.bucket.index("delete " + recKey(ltApp, ltDomain))
+ recDel := l.bucket.index("delete-if-match " + recKey(ltApp, ltDomain)) // the tombstone's removal
claimPut := l.bucket.index("put " + recKey("shop2", ltDomain))
if recDel < 0 || claimPut < recDel {
t.Fatalf("shop's record removal at %d, shop2's record at %d: want the removal first", recDel, claimPut)
@@ -1821,7 +1832,7 @@ func TestDetachCompletesThroughTheRealHandlers(t *testing.T) {
if raw, _ := rawOf(t, l.fx.rel, ltApp, ltDomain); raw != nil {
t.Fatal("the record outlived its claim")
}
- recDel := l.bucket.index("delete " + recKey(ltApp, ltDomain))
+ recDel := l.bucket.index("delete-if-match " + recKey(ltApp, ltDomain)) // the tombstone's removal
claimDel := l.bucket.index("delete-if-match " + domains.ClaimKey(ltDomain))
if claimDel < 0 {
claimDel = l.bucket.index("delete " + domains.ClaimKey(ltDomain))
@@ -2209,20 +2220,20 @@ func TestDispatchFollowsTheUnit(t *testing.T) {
}
a := oneAlloc(t, l.fx.st, ltFresh, lifetimeCommitted)
requireServing(t, l.fx.rel, ltApp, ltFresh, 1, a.OpID)
- // An on unit's delete and readDeletedRecord are still today's (Task 4
- // retires through a tombstone and judges the read): the record goes
- // unconditionally and the live slot stays, an orphan the next create
- // reclaims.
+ // An on unit's delete retires through the tombstone (Task 4): the slot
+ // retires before the key is gone, and a retirement row commits.
if code, _, body := rawReq(t, http.MethodDelete, instancesURL(l.fx, "bab", ltApp)+"/"+ltFresh, l.member, ""); code != http.StatusOK {
t.Fatalf("delete once on = %d %s", code, body)
}
- if slot, _ := slotOf(t, l.fx.rel, ltFresh); slot == nil || slot.State != provision.SlotLive || len(retireRows(t, l.fx.st, ltFresh)) != 0 {
- t.Fatalf("slot = %+v: the on delete is still the legacy one until Task 4", slot)
+ if slot, _ := slotOf(t, l.fx.rel, ltFresh); slot == nil || slot.State != provision.SlotRetired || len(retireRows(t, l.fx.st, ltFresh)) != 1 {
+ t.Fatalf("slot = %+v: the on delete retires through the tombstone", slot)
}
+ // And an on unit's readDeletedRecord is the judge's: a body under a
+ // retired slot is stale.
retiredSlot(t, l.fx.rel, ltOld)
legacyRecord(t, l.fx.rel, ltOld, "stable")
- if rec, _, err := l.fx.srv.readDeletedRecord(onCtx(t, l.fx.srv), "bab", ltApp, ltOld); err != nil || rec == nil {
- t.Fatalf("readDeletedRecord of a stale body = %+v %v: still the raw read until Task 4", rec, err)
+ if obs, err := l.fx.srv.readDeletedRecord(onCtx(t, l.fx.srv), "bab", ltApp, ltOld); err != nil || !obs.Present || obs.Verdict != provision.LifetimeStale {
+ t.Fatalf("readDeletedRecord of a stale body = %+v %v: want the judged stale observation", obs, err)
}
}
diff --git a/internal/console/lifetimes_cutover.go b/internal/console/lifetimes_cutover.go
index 44aa078f1..45a7a64fc 100644
--- a/internal/console/lifetimes_cutover.go
+++ b/internal/console/lifetimes_cutover.go
@@ -379,7 +379,9 @@ func (s *Server) lifetimesIteration(ctx context.Context, name string, mutating,
if onlyOn && u.mode != cutoverOn {
return false
}
- rctx := s.recordCtx(ctx, u)
+ // The unit rides in the context as a request's does, so the work an
+ // iteration reaches dispatches by it (lifetimesUnit).
+ rctx := context.WithValue(s.recordCtx(ctx, u), admittedUnitKey{}, u)
if lifetimesIterationHook != nil {
lifetimesIterationHook(name, rctx, u)
}
diff --git a/internal/console/lifetimes_recovery.go b/internal/console/lifetimes_recovery.go
index cedd14323..f03737ce5 100644
--- a/internal/console/lifetimes_recovery.go
+++ b/internal/console/lifetimes_recovery.go
@@ -46,10 +46,10 @@ func (s *Server) lifetimeRecoveryIteration(ctx context.Context) bool {
})
}
-// recoverLifetimes is one pass: the orphan retirements (those cancelling an
-// allocation first), then the allocations, then the migrations. Task 4 adds
-// the other retirement kinds. An error is a listing that failed; a row that
-// could not be settled stays pending for the next pass.
+// recoverLifetimes is one pass: the retirements of every kind (those
+// cancelling an allocation first), then the allocations, then the
+// migrations. An error is a listing that failed; a row that could not be
+// settled stays pending for the next pass.
func (s *Server) recoverLifetimes(ctx context.Context) error {
ctx = provision.WithLifetimes(ctx, s.judge)
rets, err := s.store.PendingRetirementsCancellingFirst()
@@ -57,13 +57,16 @@ func (s *Server) recoverLifetimes(ctx context.Context) error {
return fmt.Errorf("pending retirements: %w", err)
}
for _, r := range rets {
- if r.Kind != retireOrphan {
- s.lifetimeLogHourly("retirement-kind "+r.OpID, "console: lifetime recovery: retirement %s of %s is of kind %s, which this console does not recover yet; it stays pending", r.OpID, r.Host, r.Kind)
- continue
- }
r := r
_ = s.withAppHost(r.Account, r.App, r.Host, func() error {
- s.recoverOrphanLocked(ctx, r)
+ switch r.Kind {
+ case retireOrphan:
+ s.recoverOrphanLocked(ctx, r)
+ case retireNormal, retireCleanup:
+ s.recoverRetirementLocked(ctx, r) // lifetimes_retire.go
+ default:
+ s.lifetimeLogHourly("retirement-kind "+r.OpID, "console: lifetime recovery: retirement %s of %s is of kind %s, which this console does not know; it stays pending", r.OpID, r.Host, r.Kind)
+ }
return nil
})
}
diff --git a/internal/console/lifetimes_retire.go b/internal/console/lifetimes_retire.go
new file mode 100644
index 000000000..169400474
--- /dev/null
+++ b/internal/console/lifetimes_retire.go
@@ -0,0 +1,375 @@
+package console
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "net/http"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// Retirement through a tombstone (self-serve moves plan 1, Task 4; spec
+// §6.4). Every claims-off removal of an instance record admitted with
+// lifetimes on goes through a persisted retirement (record_retirements) of
+// one kind:
+//
+// - normal: a valid record at N under its live slot {N, this key}: N1 the
+// record becomes its tombstone (WriteRecord against the exact bytes read,
+// the sequence on), N2 the slot retires by CAS from the captured ETag, N3
+// the tombstone is deleted on its ETag;
+// - orphan: an owned live slot with no record: O1 the slot retires (Task 3);
+// - cleanup: a stale or corrupt body beside an owned live slot: C1 the slot
+// retires first, C2 the body is deleted on its ETag.
+//
+// The slot always retires before the key is absent, so no create-only write
+// of an earlier incarnation can land on an empty key under a live slot. Once
+// the row is durable no failure returns an ordinary error: the outcome is
+// errLifetimePending (recovery owns the row), a persisted terminal state, or
+// success. A retirement is terminated refused or void on evidence that the
+// record changed only after fenceRecord rewrote the current body; on slot
+// evidence (a successor) it is void with no record fence.
+
+const (
+ msgRetireRefused = "the instance changed while it was being deleted. check it and try again"
+ msgLifetimeDeletePending = "could not confirm the instance was deleted. check again in a minute"
+)
+
+// errRetireRefused: the record this retirement captured was changed within
+// its own lifetime (another ETag at the same incarnation). The record is
+// fenced and stands; the row is refused.
+var errRetireRefused = errors.New("console: the record changed in its lifetime; the retirement is refused")
+
+// errRetireExpectation is a caller stating more than one expectation: a
+// programming error, answered 500 by every door.
+var errRetireExpectation = errors.New("console: a retirement states more than one expectation")
+
+// The named expectations (provision.RetireExpectation).
+var (
+ // retireExpectAbsent: the caller saw no record; a record present now is
+ // left untouched.
+ retireExpectAbsent = provision.RetireExpectation{Absent: true}
+ // retireAny: retire whatever is current (detach).
+ retireAny = provision.RetireExpectation{}
+)
+
+// retireExpectETag binds the retirement to the caller's read. An empty ETag
+// is a programming error.
+func retireExpectETag(etag string) provision.RetireExpectation {
+ if etag == "" {
+ panic("console: retireExpectETag of an empty ETag")
+ }
+ return provision.RetireExpectation{ETag: etag}
+}
+
+// retireExpectIncarnation binds the retirement to the lifetime the caller
+// made. Incarnation 0 names no lifetime and is a programming error.
+func retireExpectIncarnation(n uint64) provision.RetireExpectation {
+ if n == 0 {
+ panic("console: retireExpectIncarnation of incarnation 0")
+ }
+ return provision.RetireExpectation{Incarnation: n}
+}
+
+// retireRounds bounds the retirement dispatcher of one entry.
+const retireRounds = 10
+
+// retireRecordWithLifetime retires acct/app/host's record as expect states,
+// under s.lockApp then the hostname lock (the lock table). See
+// retireRecordWithLifetimeLocked for the answers.
+func (s *Server) retireRecordWithLifetime(ctx context.Context, acct, app, host string, expect provision.RetireExpectation) error {
+ if expect.Set() > 1 {
+ return errRetireExpectation
+ }
+ return s.withAppHost(acct, app, host, func() error {
+ return s.retireRecordWithLifetimeLocked(ctx, acct, app, host, expect)
+ })
+}
+
+// retireRecordWithLifetimeLocked is the retirement under both locks, which
+// the caller holds (detach's and the purge's form). nil means the key is
+// absent (or was left untouched on purpose: an expectation the current
+// record does not meet) and the slot, if any, is retired or someone else's.
+// Otherwise: release.ErrCASConflict (an ETag expectation not met, or a
+// legacy record that moved before its migration), errLifetimeSettling
+// (another operation of the hostname is pending), errRetireRefused,
+// errLifetimePending, errRetireExpectation, or a read error with nothing
+// persisted.
+func (s *Server) retireRecordWithLifetimeLocked(ctx context.Context, acct, app, host string, expect provision.RetireExpectation) error {
+ if expect.Set() > 1 {
+ return errRetireExpectation
+ }
+ ctx = provision.WithLifetimes(ctx, s.judge)
+ key := provision.RecordKey(acct, app, host)
+ if err := s.admit(host, ""); err != nil {
+ return err
+ }
+ obs, err := s.classifyLocked(ctx, acct, app, host)
+ if err != nil {
+ return err
+ }
+ slot := obs.Slot
+ owned := slotOwnedBy(slot, acct, app)
+ // The caller's expectation.
+ switch {
+ case expect.Absent && obs.Present:
+ return nil
+ case expect.ETag != "" && (!obs.Present || obs.ETag != expect.ETag):
+ return fmt.Errorf("console: lifetimes: %s changed since it was read: %w", key, release.ErrCASConflict)
+ case expect.Incarnation != 0 && !(obs.Present && obs.Rec != nil && obs.Rec.Incarnation == expect.Incarnation &&
+ owned && slot.Incarnation == expect.Incarnation):
+ return nil
+ }
+ liveOwned := owned && slot.State == provision.SlotLive
+ switch {
+ case !obs.Present && slot == nil: // (a)
+ return nil
+ case !obs.Present && liveOwned: // (b): an orphan
+ if err := s.retireSlotOrphanLocked(ctx, host, slot, obs.SlotETag, ""); !errors.Is(err, errLifetimeSettling) {
+ return err
+ }
+ return nil // a successor took the slot: void
+ case !obs.Present: // (c): another key's slot, or retired
+ return nil
+ case obs.Verdict == provision.LifetimeStale || obs.Verdict == provision.LifetimeCorrupt: // (d), (g)
+ return s.retireStaleLocked(ctx, acct, app, host, obs, liveOwned)
+ }
+ rec, etag, slotETag := obs.Rec, obs.ETag, obs.SlotETag
+ if rec.Incarnation == 0 && slot == nil { // (e): legacy, bound to this read
+ mrec, _, metag, state, err := s.migrateHostLifetimeLocked(ctx, acct, app, host, obs.ETag)
+ switch {
+ case errors.Is(err, release.ErrCASConflict):
+ return err
+ case state != lifetimeCommitted:
+ if err != nil && !errors.Is(err, errLifetimePending) && !errors.Is(err, errLifetimeSettling) {
+ log.Printf("console: lifetimes: retire %s: migrate it first: %v", key, err)
+ }
+ return errLifetimeSettling
+ }
+ if slot, slotETag, err = provision.ReadHostIncarnation(ctx, s.cfg.Releases, host); err != nil {
+ return fmt.Errorf("console: lifetimes: retire %s: read the migrated slot: %w", key, err)
+ }
+ rec, etag = mrec, metag
+ liveOwned = slotOwnedBy(slot, acct, app) && slot.State == provision.SlotLive
+ }
+ // (f)
+ if !liveOwned || rec == nil || rec.Retired || rec.Incarnation != slot.Incarnation {
+ log.Printf("console: lifetimes: retire %s: the record (%v) and the slot (%+v) are no lifetime this console can retire", key, rec != nil, slot)
+ return errLifetimeSettling
+ }
+ // The dispatcher re-reads the exact bytes it acts on (N1 is bound to
+ // RecordETag).
+ opID, err := newLifetimeOpID()
+ if err != nil {
+ return err
+ }
+ r := Retirement{OpID: opID, Kind: retireNormal, Account: acct, App: app, Host: host, Incarnation: rec.Incarnation,
+ RecordETag: etag, SlotETag: slotETag, SlotIntent: slot.Intent, State: lifetimePending}
+ if err := s.store.BeginRetirement(r); err != nil {
+ return fmt.Errorf("console: lifetimes: begin the retirement of %s: %w", host, err)
+ }
+ // Durable from here: pending, terminal or done.
+ return s.retirementStepsLocked(ctx, r)
+}
+
+// retireStaleLocked is cases (d) and (g): a body the slot does not vouch for.
+// A corrupt body is copied aside first. Beside a live slot of this key it is
+// a cleanup retirement (the slot first); otherwise the body is removed on its
+// ETag with no intent, and a conflict is another writer still settling.
+func (s *Server) retireStaleLocked(ctx context.Context, acct, app, host string, obs provision.Observation, liveOwned bool) error {
+ key := provision.RecordKey(acct, app, host)
+ if obs.Verdict == provision.LifetimeCorrupt {
+ if err := s.quarantineCopy(ctx, acct, app, host, obs.Raw, obs.ETag); err != nil {
+ return err
+ }
+ }
+ if liveOwned {
+ opID, err := newLifetimeOpID()
+ if err != nil {
+ return err
+ }
+ r := Retirement{OpID: opID, Kind: retireCleanup, Account: acct, App: app, Host: host, Incarnation: obs.Slot.Incarnation,
+ RecordETag: obs.ETag, SlotETag: obs.SlotETag, SlotIntent: obs.Slot.Intent, State: lifetimePending}
+ if err := s.store.BeginRetirement(r); err != nil {
+ return fmt.Errorf("console: lifetimes: begin the cleanup of %s: %w", host, err)
+ }
+ return s.retirementStepsLocked(ctx, r)
+ }
+ switch err := release.DeleteIfMatch(ctx, s.cfg.Releases, key, obs.ETag); {
+ case err == nil:
+ return nil
+ case errors.Is(err, release.ErrCASConflict):
+ return errLifetimeSettling
+ default:
+ return fmt.Errorf("console: lifetimes: remove the stale body %s: %w", key, err)
+ }
+}
+
+// retirementStepsLocked is the retirement dispatcher, the one way into the
+// steps at the first entry and at every recovery: classify, choose from the
+// observation and the row alone, act, and classify again. With owned,
+// succ = a later incarnation or another key, pred = the captured live slot,
+// retired = this key's slot at N retired:
+//
+// - succ: void (slot evidence; no record fence);
+// - retired: any body under the key is stale and is deleted on its ETag
+// (re-read once on a conflict), then committed;
+// - neither succ, pred nor retired: pending, logged once an hour;
+// - pred, orphan: O1 (Task 3's steps);
+// - pred, cleanup: C1, then the retired branch (C2);
+// - pred, normal: absent → N2; corrupt → quarantine, convert to cleanup;
+// the tombstone at N → N2; another incarnation → fence, void; the
+// captured bytes → N1; another ETag at N → fence, refused.
+//
+// A failed write or re-read is pending, never a terminal branch.
+func (s *Server) retirementStepsLocked(ctx context.Context, r Retirement) error {
+ key := provision.RecordKey(r.Account, r.App, r.Host)
+ logf := func(format string, args ...any) {
+ log.Printf("console: lifetimes: retirement %s (%s) of %s: "+format, append([]any{r.OpID, r.Kind, r.Host}, args...)...)
+ }
+ ctx = provision.WithLifetimes(ctx, s.judge)
+ for round := 0; round < retireRounds; round++ {
+ obs, err := s.classifyLocked(ctx, r.Account, r.App, r.Host)
+ if err != nil {
+ logf("%v (pending)", err)
+ return errLifetimePending
+ }
+ slot := obs.Slot
+ owned := slotOwnedBy(slot, r.Account, r.App)
+ succ := slot != nil && (slot.Incarnation > r.Incarnation || !owned)
+ pred := slot != nil && owned && slot.Incarnation == r.Incarnation && slot.State == provision.SlotLive && slot.Intent == r.SlotIntent
+ retired := slot != nil && owned && slot.Incarnation == r.Incarnation && slot.State == provision.SlotRetired
+ switch {
+ case succ:
+ return s.settleRetirement(r, lifetimeVoid, nil)
+ case retired:
+ if obs.Present {
+ if err := s.deleteUnderRetiredSlot(ctx, r, true); err != nil {
+ logf("remove the body under the retired slot: %v (pending)", err)
+ return errLifetimePending
+ }
+ }
+ return s.settleRetirement(r, lifetimeCommitted, nil)
+ case !pred:
+ s.lifetimeLogHourly("retirement "+r.OpID, "console: lifetimes: retirement %s of %s: the slot is neither its own nor a successor (%+v); it stays pending", r.OpID, r.Host, slot)
+ return errLifetimePending
+ }
+ switch r.Kind {
+ case retireOrphan:
+ return s.orphanStepsLocked(ctx, r)
+ case retireCleanup:
+ if done, err := s.retireSlotStep(ctx, r, logf); done {
+ return err
+ }
+ continue // C2 from the retired branch
+ }
+ // normal, under its own live slot.
+ switch rec := obs.Rec; {
+ case !obs.Present:
+ // The record is gone beneath the live slot (an older delayed
+ // deletion): N2, then committed with no tombstone to remove.
+ if done, err := s.retireSlotStep(ctx, r, logf); done {
+ return err
+ }
+ case obs.Verdict == provision.LifetimeCorrupt:
+ if err := s.quarantineCopy(ctx, r.Account, r.App, r.Host, obs.Raw, obs.ETag); err != nil {
+ logf("quarantine the corrupt body: %v (pending)", err)
+ return errLifetimePending
+ }
+ if err := s.store.ConvertRetirementKind(r.OpID, retireCleanup, obs.ETag); err != nil {
+ logf("convert to a cleanup: %v (pending)", err)
+ return errLifetimePending
+ }
+ r.Kind, r.RecordETag = retireCleanup, obs.ETag
+ case rec.Retired && rec.Incarnation == r.Incarnation:
+ if done, err := s.retireSlotStep(ctx, r, logf); done { // N2, then N3 from the retired branch
+ return err
+ }
+ case rec.Incarnation != r.Incarnation:
+ _, redispatch, err := s.terminateFenced(ctx, r.Account, r.App, r.Host, fenceOwner{Tombstone: true, Incarnation: r.Incarnation}, func() error {
+ return s.store.SetRetirementState(r.OpID, lifetimeVoid)
+ })
+ switch {
+ case err != nil:
+ return errLifetimePending
+ case redispatch:
+ continue
+ }
+ return nil
+ case obs.ETag == r.RecordETag:
+ // N1: the exact bytes captured become their tombstone.
+ tomb, err := provision.TombstoneBody(obs.Raw)
+ if err != nil {
+ logf("the tombstone of %s: %v (pending)", key, err)
+ return errLifetimePending
+ }
+ _, _, err = provision.WriteRecord(provision.WithRecordSequence(ctx), s.cfg.Releases, key, obs.Raw, tomb, obs.ETag)
+ if err != nil && !errors.Is(err, release.ErrCASConflict) {
+ logf("write the tombstone: %v (pending)", err)
+ return errLifetimePending
+ }
+ // Landed or lost: on from what is there now.
+ default:
+ // The same lifetime under another ETag: changed in its lifetime.
+ _, redispatch, err := s.terminateFenced(ctx, r.Account, r.App, r.Host, fenceOwner{Tombstone: true, Incarnation: r.Incarnation}, func() error {
+ return s.store.SetRetirementState(r.OpID, lifetimeRefused)
+ })
+ switch {
+ case err != nil:
+ return errLifetimePending
+ case redispatch:
+ continue
+ }
+ return errRetireRefused
+ }
+ }
+ logf("no settled step after %d rounds (pending)", retireRounds)
+ return errLifetimePending
+}
+
+// retireSlotStep is N2, O1 and C1's slot write. done reports that the caller
+// returns err; otherwise the slot is retired and the dispatcher goes on.
+func (s *Server) retireSlotStep(ctx context.Context, r Retirement, logf func(string, ...any)) (bool, error) {
+ switch disp, err := s.retireSlotLocked(ctx, r); disp {
+ case slotSuperseded:
+ return true, s.settleRetirement(r, lifetimeVoid, nil)
+ case slotUnreadable:
+ logf("retire the slot: %v (pending)", err)
+ return true, errLifetimePending
+ }
+ return false, nil
+}
+
+// recoverRetirementLocked is recovery of a normal or cleanup retirement:
+// admitted against every other pending operation of the hostname, then the
+// dispatcher from a fresh classification. The caller holds s.lockApp then
+// the hostname lock.
+func (s *Server) recoverRetirementLocked(ctx context.Context, r Retirement) {
+ logKey := "retirement " + r.OpID
+ if err := s.admit(r.Host, r.OpID); err != nil {
+ s.lifetimeLogHourly(logKey, "console: lifetime recovery: retirement %s of %s waits: %v", r.OpID, r.Host, err)
+ return
+ }
+ if err := s.retirementStepsLocked(ctx, r); errors.Is(err, errLifetimePending) {
+ s.lifetimeLogHourly(logKey, "console: lifetime recovery: retirement %s of %s stays pending", r.OpID, r.Host)
+ }
+}
+
+// lifetimeRetireRefusal maps a retirement's error onto a door's answer:
+// settling and refused are 409s, pending is pendingStatus with the door's
+// own sentence. ok is false for release.ErrCASConflict and ordinary errors
+// (errRetireExpectation among them), which each door answers as it always
+// has.
+func lifetimeRetireRefusal(err error, settlingMsg, pendingMsg string, pendingStatus int) (*doorRefusal, bool) {
+ switch {
+ case errors.Is(err, errLifetimeSettling):
+ return &doorRefusal{http.StatusConflict, settlingMsg}, true
+ case errors.Is(err, errRetireRefused):
+ return &doorRefusal{http.StatusConflict, msgRetireRefused}, true
+ case errors.Is(err, errLifetimePending):
+ return &doorRefusal{pendingStatus, pendingMsg}, true
+ }
+ return nil, false
+}
diff --git a/internal/console/lifetimes_retire_test.go b/internal/console/lifetimes_retire_test.go
new file mode 100644
index 000000000..9661d77bf
--- /dev/null
+++ b/internal/console/lifetimes_retire_test.go
@@ -0,0 +1,1305 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "net/http"
+ "os"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/seeds"
+)
+
+// Retirement through a tombstone (lifetimes_retire.go), driven through the
+// real doors, server, store and SQLite of a claims-off console with
+// lifetimes on, over the ltStore bucket that can fail, pause or record any
+// call (lifetimes_create_test.go).
+
+// ---- fixtures -------------------------------------------------------------
+
+// del is the real DELETE door for host.
+func (l ltOn) del(t *testing.T, host string) (int, string) {
+ t.Helper()
+ code, _, body := rawReq(t, http.MethodDelete, instancesURL(l.fx, "bab", ltApp)+"/"+host, l.member, "")
+ return code, body
+}
+
+// mustCreate creates host through the real door and returns the record.
+func (l ltOn) mustCreate(t *testing.T, host string) provision.Record {
+ t.Helper()
+ code, body := l.create(t, host)
+ if code != http.StatusCreated {
+ t.Fatalf("create %s = %d %s", host, code, body)
+ }
+ raw, _ := rawOf(t, l.fx.rel, ltApp, host)
+ return decodeRec(t, raw)
+}
+
+// requireRetired: no body under bab/app/host, and the slot {n, retired,
+// bab/app}.
+func requireRetired(t *testing.T, st release.Store, app, host string, n uint64) {
+ t.Helper()
+ if raw, _ := rawOf(t, st, app, host); raw != nil {
+ t.Fatalf("a body is still under %s: %s", host, raw)
+ }
+ slot, _ := slotOf(t, st, host)
+ if slot == nil || slot.Incarnation != n || slot.State != provision.SlotRetired || slot.Account != "bab" || slot.App != app {
+ t.Fatalf("slot of %s = %+v, want {%d, retired, bab/%s}", host, slot, n, app)
+ }
+}
+
+// lastRetirement is host's newest retirement row.
+func lastRetirement(t *testing.T, st *Store, host string) retireRow {
+ t.Helper()
+ rows := retireRows(t, st, host)
+ if len(rows) == 0 {
+ t.Fatalf("no retirement of %s", host)
+ }
+ return rows[len(rows)-1]
+}
+
+// onNth runs fn before the nth call of op on key, then lets the call go on.
+func onNth(op, key string, nth int32, fn func()) func(string, string) error {
+ var n atomic.Int32
+ return func(o, k string) error {
+ if o == op && k == key && n.Add(1) == nth {
+ fn()
+ }
+ return nil
+ }
+}
+
+// bump is an ordinary update of bab/app/host's record straight in st: the
+// channel flipped through WriteRecord, so record_seq advances.
+func bump(t *testing.T, st release.Store, app, host string) string {
+ t.Helper()
+ raw, etag := rawOf(t, st, app, host)
+ rec := decodeRec(t, raw)
+ rec.Channel = map[bool]string{true: "edge", false: "stable"}[rec.Channel != "edge"]
+ body, _ := json.Marshal(rec)
+ e, _, err := provision.WriteRecord(context.Background(), st, recKey(app, host), raw, body, etag)
+ if err != nil {
+ t.Fatalf("bump %s: %v", host, err)
+ }
+ return e
+}
+
+// sqlFailRetirement makes every move of record_retirements to state fail
+// until the returned func drops the trigger.
+func sqlFailRetirement(t *testing.T, st *Store, name, when string) func() {
+ t.Helper()
+ if _, err := st.db.Exec(`CREATE TRIGGER ` + name + ` BEFORE UPDATE ON record_retirements WHEN ` + when + `
+BEGIN SELECT RAISE(ABORT, 'test: the retirement update is refused'); END`); err != nil {
+ t.Fatal(err)
+ }
+ var once sync.Once
+ return func() {
+ once.Do(func() {
+ if _, err := st.db.Exec(`DROP TRIGGER ` + name); err != nil {
+ t.Fatal(err)
+ }
+ })
+ }
+}
+
+// retireOn runs the retirement core on srv in a unit admitted on.
+func retireOn(t *testing.T, srv *Server, host string, expect provision.RetireExpectation) error {
+ t.Helper()
+ return srv.retireRecordWithLifetime(onCtx(t, srv), "bab", ltApp, host, expect)
+}
+
+// quarantined is every quarantine copy of bab/shop/host.
+func quarantined(t *testing.T, st release.Store, host string) [][]byte {
+ t.Helper()
+ keys, err := st.List(context.Background(), "control/quarantine/bab/"+ltApp+"/")
+ if err != nil {
+ t.Fatal(err)
+ }
+ var out [][]byte
+ for _, k := range keys {
+ if strings.HasPrefix(k, "control/quarantine/bab/"+ltApp+"/"+host+".") {
+ b, _, err := st.Get(context.Background(), k)
+ if err != nil {
+ t.Fatal(err)
+ }
+ out = append(out, b)
+ }
+ }
+ return out
+}
+
+// slotRetiredWhenKeyLeaves records, at the moment the record key's body is
+// removed, whether the slot was already retired. It fails the test if the
+// key is removed under a live slot.
+func slotRetiredWhenKeyLeaves(t *testing.T, l ltOn, host string) func(string, string) error {
+ t.Helper()
+ return func(op, key string) error {
+ if (op == "delete-if-match" || op == "delete") && key == recKey(ltApp, host) {
+ slot, _, err := provision.ReadHostIncarnation(context.Background(), l.fx.rel, host)
+ if err == nil && slot != nil && slot.State == provision.SlotLive && slotOwnedBy(slot, "bab", ltApp) {
+ t.Errorf("the body of %s was removed under the live slot %+v", host, *slot)
+ }
+ }
+ return nil
+ }
+}
+
+// ---- the delete door --------------------------------------------------------
+
+func TestDeleteDoorRetiresThroughTheTombstone(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ if code, body := l.del(t, ltOne); code != http.StatusOK || !strings.Contains(body, "deleted") {
+ t.Fatalf("delete = %d %s", code, body)
+ }
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ if r := lastRetirement(t, l.fx.st, ltOne); r.Kind != retireNormal || r.State != lifetimeCommitted || r.Incarnation != 1 {
+ t.Fatalf("retirement = %+v, want one committed normal at 1", r)
+ }
+ ops := l.bucket.opsOf(recKey(ltApp, ltOne))
+ // The create's PUT, the tombstone PUT, then the tombstone's conditional
+ // delete; never a plain delete of the key.
+ var puts, deletes int
+ for _, op := range ops {
+ switch op {
+ case "put":
+ puts++
+ case "delete-if-match":
+ deletes++
+ case "delete":
+ t.Fatalf("ops on the key = %v: a plain delete", ops)
+ }
+ }
+ if puts != 2 || deletes != 1 {
+ t.Fatalf("ops on the key = %v, want the create, the tombstone and its conditional delete", ops)
+ }
+}
+
+func TestSlotRetiresBeforeTheKeyIsAbsent(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ var tombstoned atomic.Bool
+ l.bucket.setHook(chain(slotRetiredWhenKeyLeaves(t, l, ltOne), func(op, key string) error {
+ if op == "put" && key == provision.HostIncarnationKey(ltOne) {
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ tombstoned.Store(raw != nil && decodeRec(t, raw).Retired)
+ }
+ return nil
+ }))
+ if code, body := l.del(t, ltOne); code != http.StatusOK {
+ t.Fatalf("delete = %d %s", code, body)
+ }
+ if !tombstoned.Load() {
+ t.Fatal("the slot was retired before the tombstone landed")
+ }
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+}
+
+func TestDeleteDoorAbsentPathAdmitsAndRetiresOrphans(t *testing.T) {
+ t.Run("initial absence", func(t *testing.T) {
+ l := newLtOn(t)
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: "bab", App: ltApp, Intent: "create lost"})
+ if code, body := l.del(t, ltOne); code != http.StatusOK || !strings.Contains(body, "deleted") {
+ t.Fatalf("delete = %d %s", code, body)
+ }
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ if r := lastRetirement(t, l.fx.st, ltOne); r.Kind != retireOrphan || r.State != lifetimeCommitted {
+ t.Fatalf("retirement = %+v, want a committed orphan", r)
+ }
+ })
+ t.Run("the absent path admits", func(t *testing.T) {
+ l := newLtOn(t)
+ if err := l.fx.st.BeginRetirement(Retirement{OpID: "other", Kind: retireOrphan, Account: "bab", App: "web", Host: ltOne, Incarnation: 1}); err != nil {
+ t.Fatal(err)
+ }
+ if code, body := l.del(t, ltOne); code != http.StatusConflict || ltErr(body) != msgLifetimeSettling {
+ t.Fatalf("delete = %d %s, want 409 settling", code, body)
+ }
+ })
+ t.Run("absence found on the judge's re-read", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ // A delayed deletion takes the record after the door's first read:
+ // its ETag expectation conflicts, and the re-decision finds absence.
+ var once sync.Once
+ l.bucket.setAfter(func(op, key string, err error) {
+ if op == "get" && key == recKey(ltApp, ltOne) && err == nil {
+ once.Do(func() {
+ if err := l.fx.rel.Delete(context.Background(), recKey(ltApp, ltOne)); err != nil {
+ t.Error(err)
+ }
+ })
+ }
+ })
+ if code, body := l.del(t, ltOne); code != http.StatusOK || !strings.Contains(body, "deleted") {
+ t.Fatalf("delete = %d %s", code, body)
+ }
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ if r := lastRetirement(t, l.fx.st, ltOne); r.Kind != retireOrphan || r.State != lifetimeCommitted {
+ t.Fatalf("retirement = %+v, want the owned slot retired as an orphan", r)
+ }
+ })
+ t.Run("a record created since is untouched", func(t *testing.T) {
+ l := newLtOn(t)
+ var once sync.Once
+ var created string
+ l.bucket.setAfter(func(op, key string, err error) {
+ if op == "get" && key == recKey(ltApp, ltOne) && errors.Is(err, os.ErrNotExist) {
+ once.Do(func() { created = legacyBody(t, l.fx.rel, ltApp, ltOne, "stable") })
+ }
+ })
+ if code, body := l.del(t, ltOne); code != http.StatusOK || !strings.Contains(body, "deleted") {
+ t.Fatalf("delete = %d %s", code, body)
+ }
+ if created == "" {
+ t.Fatal("the record was never created behind the door's read")
+ }
+ if _, etag := rawOf(t, l.fx.rel, ltApp, ltOne); etag != created {
+ t.Fatalf("the record created since = %q, want it untouched (%q)", etag, created)
+ }
+ if rows := retireRows(t, l.fx.st, ltOne); len(rows) != 0 {
+ t.Fatalf("retirements = %+v, want none", rows)
+ }
+ })
+}
+
+func TestStaleCleanupRetiresALiveOwnedSlotFirst(t *testing.T) {
+ for _, body := range []string{"fieldless", "corrupt"} {
+ t.Run(body, func(t *testing.T) {
+ l := newLtOn(t)
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: "bab", App: ltApp, Intent: "create late"})
+ if body == "fieldless" {
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ } else if _, err := l.fx.rel.Put(context.Background(), recKey(ltApp, ltOne), []byte(`{"schema":`), ""); err != nil {
+ t.Fatal(err)
+ }
+ // A blocked incarnation-1 create-only write, released after the
+ // retirement.
+ late, _ := json.Marshal(provision.Record{Schema: 1, Host: ltOne, App: ltApp, Account: "bab", Channel: "stable", Incarnation: 1, CreatedAt: time.Now().UTC()})
+ l.bucket.setHook(slotRetiredWhenKeyLeaves(t, l, ltOne))
+ if code, resp := l.del(t, ltOne); code != http.StatusOK {
+ t.Fatalf("delete = %d %s", code, resp)
+ }
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ if r := lastRetirement(t, l.fx.st, ltOne); r.Kind != retireCleanup || r.State != lifetimeCommitted {
+ t.Fatalf("retirement = %+v, want a committed cleanup", r)
+ }
+ if body == "corrupt" {
+ if q := quarantined(t, l.fx.rel, ltOne); len(q) != 1 || string(q[0]) != `{"schema":` {
+ t.Fatalf("quarantine = %q, want the corrupt bytes", q)
+ }
+ }
+ if _, _, err := provision.WriteRecord(provision.WithRecordSequence(context.Background()), l.fx.rel, recKey(ltApp, ltOne), []byte(`{}`), late, ""); err != nil {
+ t.Fatalf("the late create-only write = %v", err)
+ }
+ if obs := l.fx.srv.judge.Observe(context.Background(), l.fx.rel, "bab", ltApp, ltOne); obs.Verdict != provision.LifetimeStale {
+ t.Fatalf("the late body is %v, want stale under the retired slot", obs.Verdict)
+ }
+ })
+ }
+}
+
+func TestDeleteDoorRemovesStaleAndCorruptRecordsConditionally(t *testing.T) {
+ slots := map[string]provision.HostIncarnation{
+ "a retired slot": {Incarnation: 1, State: provision.SlotRetired, Account: "bab", App: ltApp, Intent: "retire old"},
+ "a foreign slot": {Incarnation: 1, State: provision.SlotLive, Account: "bab", App: "web", Intent: "create web"},
+ }
+ for name, slot := range slots {
+ for _, body := range []string{"fieldless", "corrupt"} {
+ t.Run(name+", "+body, func(t *testing.T) {
+ l := newLtOn(t)
+ slotETag := writeSlot(t, l.fx.rel, ltOne, slot)
+ if body == "fieldless" {
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ } else if _, err := l.fx.rel.Put(context.Background(), recKey(ltApp, ltOne), []byte(`[1,2`), ""); err != nil {
+ t.Fatal(err)
+ }
+ if code, resp := l.del(t, ltOne); code != http.StatusOK {
+ t.Fatalf("delete = %d %s", code, resp)
+ }
+ if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); raw != nil {
+ t.Fatalf("the body survived: %s", raw)
+ }
+ if _, e := slotOf(t, l.fx.rel, ltOne); e != slotETag {
+ t.Fatal("the slot was written: it is not this key's to retire")
+ }
+ if rows := retireRows(t, l.fx.st, ltOne); len(rows) != 0 {
+ t.Fatalf("retirements = %+v, want a plain conditional delete", rows)
+ }
+ for _, op := range l.bucket.opsOf(recKey(ltApp, ltOne)) {
+ if op == "delete" {
+ t.Fatal("a plain delete of the key")
+ }
+ }
+ if body == "corrupt" && len(quarantined(t, l.fx.rel, ltOne)) != 1 {
+ t.Fatal("the corrupt body was not quarantined first")
+ }
+ })
+ }
+ }
+ t.Run("a body that changes under the conditional delete is settling", func(t *testing.T) {
+ l := newLtOn(t)
+ writeSlot(t, l.fx.rel, ltOne, slots["a retired slot"])
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ l.bucket.setHook(onNth("delete-if-match", recKey(ltApp, ltOne), 1, func() {
+ raw, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if _, err := l.fx.rel.Put(context.Background(), recKey(ltApp, ltOne), bytes.Replace(raw, []byte(`"stable"`), []byte(`"edge"`), 1), etag); err != nil {
+ t.Error(err)
+ }
+ }))
+ code, resp := l.del(t, ltOne)
+ if code == http.StatusOK {
+ t.Fatalf("delete = %d %s, want a refusal: the body changed under it", code, resp)
+ }
+ if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); raw == nil {
+ t.Fatal("the changed body was deleted")
+ }
+ })
+}
+
+func TestSeedFinishRetiresThroughTheTombstone(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ l.fx.srv.cfg.SeedData = release.NewDirStore(t.TempDir())
+ id, err := seeds.NewID()
+ if err != nil {
+ t.Fatal(err)
+ }
+ // The record carries a rolled-back import, as a box leaves it.
+ raw, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ rec := decodeRec(t, raw)
+ rec.Seed = &provision.SeedSource{ID: id, Kind: seeds.KindImport, State: seeds.StateRolledBack, By: "alice@keymail.dev", At: time.Now().UTC()}
+ body, _ := json.Marshal(rec)
+ if _, _, err := provision.WriteRecord(context.Background(), l.fx.rel, recKey(ltApp, ltOne), raw, body, etag); err != nil {
+ t.Fatal(err)
+ }
+ if err := l.fx.srv.finishSeed(onCtx(t, l.fx.srv), "bab", ltApp, ltOne, id); err != nil {
+ t.Fatalf("finish = %v", err)
+ }
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ if r := lastRetirement(t, l.fx.st, ltOne); r.Kind != retireNormal || r.State != lifetimeCommitted {
+ t.Fatalf("retirement = %+v, want a committed normal", r)
+ }
+ var tombstone bool
+ for _, op := range l.bucket.opsOf(recKey(ltApp, ltOne)) {
+ tombstone = tombstone || op == "put"
+ if op == "delete" {
+ t.Fatal("a plain delete of the key")
+ }
+ }
+ if !tombstone {
+ t.Fatal("no tombstone was written")
+ }
+}
+
+// ---- the retirement's own branches -----------------------------------------
+
+func TestRetireRefusesWhenTheRecordChangedInItsLifetime(t *testing.T) {
+ t.Run("the core", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ // An ordinary update lands between the retirement's read and its
+ // tombstone CAS.
+ var changed string
+ l.bucket.setHook(onNth("put", recKey(ltApp, ltOne), 1, func() { changed = bump(t, l.fx.rel, ltApp, ltOne) }))
+ err := retireOn(t, l.fx.srv, ltOne, retireAny)
+ if !errors.Is(err, errRetireRefused) {
+ t.Fatalf("retire = %v, want errRetireRefused", err)
+ }
+ r := lastRetirement(t, l.fx.st, ltOne)
+ if r.State != lifetimeRefused {
+ t.Fatalf("retirement = %+v, want refused", r)
+ }
+ raw, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ rec := decodeRec(t, raw)
+ if rec.Retired || rec.Channel != "edge" || etag == changed {
+ t.Fatalf("record = %s, want the update intact and fenced", raw)
+ }
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot == nil || slot.State != provision.SlotLive {
+ t.Fatalf("slot = %+v, want it live", slot)
+ }
+ })
+ t.Run("the door", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ l.bucket.setHook(onNth("put", recKey(ltApp, ltOne), 1, func() { bump(t, l.fx.rel, ltApp, ltOne) }))
+ if code, body := l.del(t, ltOne); code != http.StatusConflict || ltErr(body) != msgRetireRefused {
+ t.Fatalf("delete = %d %s, want 409 %q", code, body, msgRetireRefused)
+ }
+ })
+}
+
+func TestRetireVoidsAgainstASuccessor(t *testing.T) {
+ t.Run("a record of another incarnation", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ l.bucket.setHook(onNth("put", recKey(ltApp, ltOne), 1, func() {
+ raw, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ rec := decodeRec(t, raw)
+ rec.Incarnation = 7
+ body, _ := json.Marshal(rec)
+ if _, _, err := provision.WriteRecord(context.Background(), l.fx.rel, recKey(ltApp, ltOne), raw, body, etag); err != nil {
+ t.Error(err)
+ }
+ }))
+ if err := retireOn(t, l.fx.srv, ltOne, retireAny); err != nil {
+ t.Fatalf("retire = %v, want nil (void)", err)
+ }
+ if r := lastRetirement(t, l.fx.st, ltOne); r.State != lifetimeVoid {
+ t.Fatalf("retirement = %+v, want void", r)
+ }
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if rec := decodeRec(t, raw); rec.Incarnation != 7 || rec.Retired || rec.RecordSeq < 3 {
+ t.Fatalf("record = %s, want the other incarnation intact and fenced", raw)
+ }
+ })
+ t.Run("a successor slot", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ var succ string
+ l.bucket.setHook(onNth("put", recKey(ltApp, ltOne), 1, func() {
+ succ = writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 2, State: provision.SlotLive, Account: "bab", App: "web", Intent: "create web"})
+ }))
+ if err := retireOn(t, l.fx.srv, ltOne, retireAny); err != nil {
+ t.Fatalf("retire = %v, want nil (void)", err)
+ }
+ if r := lastRetirement(t, l.fx.st, ltOne); r.State != lifetimeVoid {
+ t.Fatalf("retirement = %+v, want void", r)
+ }
+ if _, e := slotOf(t, l.fx.rel, ltOne); e != succ {
+ t.Fatal("the successor's slot was written")
+ }
+ })
+}
+
+func TestRetireNeverMovesTheSlotBackwards(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ // The retirement's tombstone write fails: the row is pending at N=1.
+ l.bucket.setHook(failNth("put", recKey(ltApp, ltOne), 1, false))
+ if err := retireOn(t, l.fx.srv, ltOne, retireAny); !errors.Is(err, errLifetimePending) {
+ t.Fatalf("retire = %v, want pending", err)
+ }
+ l.bucket.setHook(nil)
+ r := lastRetirement(t, l.fx.st, ltOne)
+ // The hostname moved on to incarnation 2 meanwhile (outside this
+ // console's admission).
+ succ := writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 2, State: provision.SlotLive, Account: "bab", App: ltApp, Intent: "create two"})
+ rows, _ := l.fx.st.PendingRetirements()
+ disp, _ := l.fx.srv.retireSlotLocked(context.Background(), rows[0])
+ if disp != slotSuperseded {
+ t.Fatalf("the late slot write = %v, want superseded", disp)
+ }
+ recoverOn(t, l.restart(t))
+ if got := lastRetirement(t, l.fx.st, ltOne); got.OpID != r.OpID || got.State != lifetimeVoid {
+ t.Fatalf("retirement = %+v, want void", got)
+ }
+ if slot, e := slotOf(t, l.fx.rel, ltOne); e != succ || slot.Incarnation != 2 {
+ t.Fatalf("slot = %+v, want {2, live} untouched", slot)
+ }
+}
+
+func TestTombstoneConflictWithAbsentRecordStillRetiresTheSlot(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ // An older delayed deletion takes the record before the tombstone CAS.
+ l.bucket.setHook(onNth("put", recKey(ltApp, ltOne), 1, func() {
+ if err := l.fx.rel.Delete(context.Background(), recKey(ltApp, ltOne)); err != nil {
+ t.Error(err)
+ }
+ }))
+ if err := retireOn(t, l.fx.srv, ltOne, retireAny); err != nil {
+ t.Fatalf("retire = %v", err)
+ }
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ if r := lastRetirement(t, l.fx.st, ltOne); r.State != lifetimeCommitted {
+ t.Fatalf("retirement = %+v, want committed", r)
+ }
+ for _, op := range l.bucket.opsOf(recKey(ltApp, ltOne)) {
+ if op == "delete-if-match" {
+ t.Fatal("a tombstone delete ran with no tombstone to remove")
+ }
+ }
+}
+
+func TestOrphanSlotIsRetired(t *testing.T) {
+ l := newLtOn(t)
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 3, State: provision.SlotLive, Account: "bab", App: ltApp, Intent: "create lost"})
+ if err := retireOn(t, l.fx.srv, ltOne, retireAny); err != nil {
+ t.Fatalf("retire = %v", err)
+ }
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 3)
+ if r := lastRetirement(t, l.fx.st, ltOne); r.Kind != retireOrphan || r.State != lifetimeCommitted {
+ t.Fatalf("retirement = %+v", r)
+ }
+}
+
+func TestOrphanRetirementSurvivesARestoredRecord(t *testing.T) {
+ l := newLtOn(t)
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: "bab", App: ltApp, Intent: "create lost"})
+ // O1's slot write fails: the orphan retirement is pending.
+ l.bucket.setHook(failNth("put", provision.HostIncarnationKey(ltOne), 1, false))
+ if err := retireOn(t, l.fx.srv, ltOne, retireAny); !errors.Is(err, errLifetimePending) {
+ t.Fatalf("retire = %v, want pending", err)
+ }
+ l.bucket.setHook(nil)
+ // The record comes back under the key (a late write of the lost create).
+ 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)
+ }
+ recoverOn(t, l.restart(t))
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ if r := lastRetirement(t, l.fx.st, ltOne); r.Kind != retireOrphan || r.State != lifetimeCommitted {
+ t.Fatalf("retirement = %+v, want committed", r)
+ }
+}
+
+func TestExpectationsGateRetirement(t *testing.T) {
+ t.Run("an incarnation the record no longer has", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ if code, body := l.del(t, ltOne); code != http.StatusOK {
+ t.Fatalf("delete = %d %s", code, body)
+ }
+ if rec := l.mustCreate(t, ltOne); rec.Incarnation != 2 {
+ t.Fatalf("recreated at %d, want 2", rec.Incarnation)
+ }
+ _, before := rawOf(t, l.fx.rel, ltApp, ltOne)
+ n := len(retireRows(t, l.fx.st, ltOne))
+ if err := retireOn(t, l.fx.srv, ltOne, retireExpectIncarnation(1)); err != nil {
+ t.Fatalf("retire = %v, want nil", err)
+ }
+ if _, after := rawOf(t, l.fx.rel, ltApp, ltOne); after != before || len(retireRows(t, l.fx.st, ltOne)) != n {
+ t.Fatal("the record at 2 was touched")
+ }
+ })
+ t.Run("retireAny retires what is current", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ if err := retireOn(t, l.fx.srv, ltOne, retireAny); err != nil {
+ t.Fatal(err)
+ }
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ })
+ t.Run("an ETag that moved is a conflict", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ _, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ bump(t, l.fx.rel, ltApp, ltOne)
+ if err := retireOn(t, l.fx.srv, ltOne, retireExpectETag(etag)); !errors.Is(err, release.ErrCASConflict) {
+ t.Fatalf("retire = %v, want a CAS conflict", err)
+ }
+ if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); raw == nil || len(retireRows(t, l.fx.st, ltOne)) != 0 {
+ t.Fatal("a conflicting retirement touched the record")
+ }
+ })
+ t.Run("absent when present leaves it", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ if err := retireOn(t, l.fx.srv, ltOne, retireExpectAbsent); err != nil {
+ t.Fatal(err)
+ }
+ if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); raw == nil {
+ t.Fatal("an absent expectation removed a present record")
+ }
+ })
+ t.Run("two expectations are a programming error", func(t *testing.T) {
+ l := newLtOn(t)
+ err := retireOn(t, l.fx.srv, ltOne, provision.RetireExpectation{Absent: true, Incarnation: 1})
+ if !errors.Is(err, errRetireExpectation) {
+ t.Fatalf("retire = %v, want errRetireExpectation", err)
+ }
+ })
+}
+
+func TestLegacyRetirementIsBoundToTheCallersRead(t *testing.T) {
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ _, read := rawOf(t, l.fx.rel, ltApp, ltOne)
+ // An ordinary update lands between the caller's read and the marker CAS
+ // (the first PUT of the key is the migration's marker).
+ l.bucket.setHook(onNth("put", recKey(ltApp, ltOne), 1, func() { bump(t, l.fx.rel, ltApp, ltOne) }))
+ err := retireOn(t, l.fx.srv, ltOne, retireExpectETag(read))
+ if !errors.Is(err, release.ErrCASConflict) {
+ t.Fatalf("retire = %v, want a CAS conflict", err)
+ }
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if rec := decodeRec(t, raw); rec.Channel != "edge" || rec.Retired || rec.IncarnationMigrating != "" {
+ t.Fatalf("record = %s, want the update intact and unmarked", raw)
+ }
+ var state string
+ if err := l.fx.st.db.QueryRow(`SELECT state FROM record_migrations WHERE host = ?`, ltOne).Scan(&state); err != nil || state != lifetimeSuperseded {
+ t.Fatalf("migration = %q %v, want superseded", state, err)
+ }
+ if rows := retireRows(t, l.fx.st, ltOne); len(rows) != 0 {
+ t.Fatalf("retirements = %+v, want none", rows)
+ }
+}
+
+func TestDeleteDoorRefusesAnImportThatAppearedAfterItsFirstRead(t *testing.T) {
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ id, _ := seeds.NewID()
+ var once sync.Once
+ l.bucket.setAfter(func(op, key string, err error) {
+ if op == "get" && key == recKey(ltApp, ltOne) && err == nil {
+ once.Do(func() {
+ raw, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ rec := decodeRec(t, raw)
+ rec.Seed = &provision.SeedSource{ID: id, Kind: seeds.KindImport, State: seeds.StateReady, By: "alice@keymail.dev", At: time.Now().UTC()}
+ body, _ := json.Marshal(rec)
+ if _, err := l.fx.rel.Put(context.Background(), recKey(ltApp, ltOne), body, etag); err != nil {
+ t.Error(err)
+ }
+ })
+ }
+ })
+ code, body := l.del(t, ltOne)
+ if code != http.StatusConflict || !strings.Contains(ltErr(body), "Abort the import instead") {
+ t.Fatalf("delete = %d %s, want the import rule's 409", code, body)
+ }
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if raw == nil || decodeRec(t, raw).Seed == nil {
+ t.Fatal("the import was deleted")
+ }
+}
+
+func TestDelayedDeleteCannotEraseASuccessor(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ // The first console's tombstone delete (N3) is held in flight.
+ gate, hook := pauseNth(t, "delete-if-match", recKey(ltApp, ltOne), 1)
+ l.bucket.setHook(hook)
+ answered := make(chan int, 1)
+ go func() { code, _ := l.del(t, ltOne); answered <- code }()
+ waitFor(t, gate.entered, "the tombstone delete")
+ // A second console finishes the retirement and the hostname's next
+ // lifetime is created.
+ second := l.restart(t)
+ l.bucket.setHook(nil)
+ recoverOn(t, second)
+ life, _ := l.fx.st.AppLife(OpsAccountID, ltApp)
+ rec := provision.Record{Schema: 1, Host: ltOne, App: ltApp, Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC()}
+ if _, err := second.createRecordWithLifetime(onCtx(t, second), rec, life.Lifetime); err != nil {
+ t.Fatalf("the successor's create = %v", err)
+ }
+ _, succ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ gate.open()
+ <-answered
+ raw, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if raw == nil || etag != succ || decodeRec(t, raw).Incarnation != 2 {
+ t.Fatalf("record = %s, want the successor at 2 intact", raw)
+ }
+}
+
+// ---- recovery ---------------------------------------------------------------
+
+// pendingRetire starts a retirement of ltOne's record through the core and
+// makes it stop at the step named, pending: "begin" (the tombstone write
+// fails), "n1" (the slot write fails after the tombstone), "n2" (the
+// tombstone delete fails after the slot), "n3" (the commit fails).
+func (l ltOn) pendingRetire(t *testing.T, step string) retireRow {
+ t.Helper()
+ var drop func()
+ switch step {
+ case "begin":
+ l.bucket.setHook(failNth("put", recKey(ltApp, ltOne), 1, false))
+ case "n1":
+ l.bucket.setHook(failNth("put", provision.HostIncarnationKey(ltOne), 1, false))
+ case "n2":
+ l.bucket.setHook(failNth("delete-if-match", recKey(ltApp, ltOne), 1, false))
+ case "n3":
+ drop = sqlFailRetirement(t, l.fx.st, "lt_fail_retire_commit", "NEW.state = 'committed'")
+ default:
+ t.Fatalf("no step %q", step)
+ }
+ err := retireOn(t, l.fx.srv, ltOne, retireAny)
+ l.bucket.setHook(nil)
+ if drop != nil {
+ drop()
+ }
+ if !errors.Is(err, errLifetimePending) {
+ t.Fatalf("retire stopping at %s = %v, want pending", step, err)
+ }
+ r := lastRetirement(t, l.fx.st, ltOne)
+ if r.State != lifetimePending {
+ t.Fatalf("retirement = %+v, want pending", r)
+ }
+ return r
+}
+
+// recoveredTo runs one recovery pass on a restarted console and requires the
+// retirement's state.
+func (l ltOn) recoveredTo(t *testing.T, op, state string) {
+ t.Helper()
+ recoverOn(t, l.restart(t))
+ for _, r := range retireRows(t, l.fx.st, ltOne) {
+ if r.OpID == op {
+ if r.State != state {
+ t.Fatalf("retirement %s = %+v, want %s", op, r, state)
+ }
+ return
+ }
+ }
+ t.Fatalf("no retirement %s", op)
+}
+
+func TestRetirementRecoveryBranches(t *testing.T) {
+ ctx := context.Background()
+ for _, step := range []string{"begin", "n1", "n2", "n3"} {
+ t.Run("normal, crashed at "+step, func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, step)
+ l.recoveredTo(t, r.OpID, lifetimeCommitted)
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ })
+ }
+ t.Run("normal, a second crash during recovery", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, "begin")
+ second := l.restart(t)
+ second.cfg.Releases.(*ltStore).setHook(failNth("put", provision.HostIncarnationKey(ltOne), 1, false))
+ recoverOn(t, second)
+ l.bucket.setHook(nil)
+ if got := lastRetirement(t, l.fx.st, ltOne); got.State != lifetimePending {
+ t.Fatalf("after the second crash = %+v, want pending", got)
+ }
+ l.recoveredTo(t, r.OpID, lifetimeCommitted)
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ })
+ t.Run("normal, a changed body is refused after the fence", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, "begin")
+ bump(t, l.fx.rel, ltApp, ltOne)
+ l.recoveredTo(t, r.OpID, lifetimeRefused)
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if rec := decodeRec(t, raw); rec.Retired || rec.Channel != "edge" {
+ t.Fatalf("record = %s, want the changed body intact", raw)
+ }
+ })
+ t.Run("normal, another incarnation is void after the fence", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, "begin")
+ raw, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ rec := decodeRec(t, raw)
+ rec.Incarnation = 5
+ body, _ := json.Marshal(rec)
+ if _, _, err := provision.WriteRecord(ctx, l.fx.rel, recKey(ltApp, ltOne), raw, body, etag); err != nil {
+ t.Fatal(err)
+ }
+ l.recoveredTo(t, r.OpID, lifetimeVoid)
+ if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); raw == nil || decodeRec(t, raw).Incarnation != 5 {
+ t.Fatalf("record = %s, want the other incarnation intact", raw)
+ }
+ })
+ t.Run("any kind, a successor slot is void", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, "begin")
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 2, State: provision.SlotLive, Account: "bab", App: "web", Intent: "create web"})
+ l.recoveredTo(t, r.OpID, lifetimeVoid)
+ })
+ t.Run("normal, a corrupt body converts to cleanup", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, "begin")
+ _, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if _, err := l.fx.rel.Put(ctx, recKey(ltApp, ltOne), []byte(`{"host":`), etag); err != nil {
+ t.Fatal(err)
+ }
+ l.bucket.setHook(slotRetiredWhenKeyLeaves(t, l, ltOne))
+ l.recoveredTo(t, r.OpID, lifetimeCommitted)
+ if got := lastRetirement(t, l.fx.st, ltOne); got.Kind != retireCleanup {
+ t.Fatalf("retirement = %+v, want converted to cleanup", got)
+ }
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ if len(quarantined(t, l.fx.rel, ltOne)) == 0 {
+ t.Fatal("the corrupt body was not quarantined")
+ }
+ })
+ t.Run("normal, a delayed slot write that landed", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ // The slot write lands but its answer is lost.
+ var n atomic.Int32
+ l.bucket.setAfter(func(op, key string, err error) {})
+ l.bucket.setHook(func(op, key string) error {
+ if op == "put" && key == provision.HostIncarnationKey(ltOne) && n.Add(1) == 1 {
+ return errBucketDown
+ }
+ return nil
+ })
+ // Write the retired slot behind the failed answer, as the bucket did.
+ r := func() retireRow {
+ err := retireOn(t, l.fx.srv, ltOne, retireAny)
+ if !errors.Is(err, errLifetimePending) {
+ t.Fatalf("retire = %v, want pending", err)
+ }
+ return lastRetirement(t, l.fx.st, ltOne)
+ }()
+ l.bucket.setHook(nil)
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 1, State: provision.SlotRetired, Account: "bab", App: ltApp, Intent: "retire " + r.OpID})
+ l.recoveredTo(t, r.OpID, lifetimeCommitted)
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ })
+ t.Run("normal, the tombstone lost to a deletion: committed without N3", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, "begin")
+ if err := l.fx.rel.Delete(ctx, recKey(ltApp, ltOne)); err != nil {
+ t.Fatal(err)
+ }
+ before := len(l.bucket.opsOf(recKey(ltApp, ltOne)))
+ l.recoveredTo(t, r.OpID, lifetimeCommitted)
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ for _, op := range l.bucket.opsOf(recKey(ltApp, ltOne))[before:] {
+ if op == "delete-if-match" || op == "put" {
+ t.Fatalf("recovery wrote the absent key: %s", op)
+ }
+ }
+ })
+ t.Run("an unreadable slot leaves it pending", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, "begin")
+ second := l.restart(t)
+ l.bucket.setHook(failNth("get", provision.HostIncarnationKey(ltOne), 1, true))
+ recoverOn(t, second)
+ l.bucket.setHook(nil)
+ if got := lastRetirement(t, l.fx.st, ltOne); got.OpID != r.OpID || got.State != lifetimePending {
+ t.Fatalf("retirement = %+v, want pending", got)
+ }
+ if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); raw == nil || decodeRec(t, raw).Retired {
+ t.Fatal("an unreadable observation acted")
+ }
+ })
+ t.Run("a slot that is neither its own nor a successor stays pending", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, "begin")
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: "bab", App: ltApp, Intent: "create someone-else"})
+ l.recoveredTo(t, r.OpID, lifetimePending)
+ })
+ for _, step := range []string{"c1", "c2"} {
+ t.Run("cleanup, crashed at "+step, func(t *testing.T) {
+ l := newLtOn(t)
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: "bab", App: ltApp, Intent: "create late"})
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ if step == "c1" {
+ l.bucket.setHook(failNth("put", provision.HostIncarnationKey(ltOne), 1, false))
+ } else {
+ l.bucket.setHook(failNth("delete-if-match", recKey(ltApp, ltOne), 1, false))
+ }
+ _, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if err := retireOn(t, l.fx.srv, ltOne, retireExpectETag(etag)); !errors.Is(err, errLifetimePending) {
+ t.Fatalf("retire = %v, want pending", err)
+ }
+ l.bucket.setHook(slotRetiredWhenKeyLeaves(t, l, ltOne))
+ r := lastRetirement(t, l.fx.st, ltOne)
+ if r.Kind != retireCleanup {
+ t.Fatalf("retirement = %+v, want cleanup", r)
+ }
+ l.recoveredTo(t, r.OpID, lifetimeCommitted)
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ })
+ }
+}
+
+func TestCorruptBodyIsQuarantinedSlotFirst(t *testing.T) {
+ ctx := context.Background()
+ const malformed = `{"schema":1,"host":`
+ // corruptBeforeTombstone makes the body malformed JSON between the
+ // observation and the tombstone CAS.
+ corruptBeforeTombstone := func(t *testing.T, l ltOn) func(string, string) error {
+ return onNth("put", recKey(ltApp, ltOne), 1, func() {
+ _, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if _, err := l.fx.rel.Put(ctx, recKey(ltApp, ltOne), []byte(malformed), etag); err != nil {
+ t.Error(err)
+ }
+ })
+ }
+ // late is a delayed incarnation-1 create-only write of the original
+ // create: it can land only on an absent key.
+ late := func(t *testing.T, l ltOn) error {
+ body, _ := json.Marshal(provision.Record{Schema: 1, Host: ltOne, App: ltApp, Account: "bab", Channel: "stable", Incarnation: 1, CreatedAt: time.Now().UTC()})
+ _, _, err := provision.WriteRecord(provision.WithRecordSequence(ctx), l.fx.rel, recKey(ltApp, ltOne), []byte(`{}`), body, "")
+ return err
+ }
+ finished := func(t *testing.T, l ltOn) {
+ t.Helper()
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ r := lastRetirement(t, l.fx.st, ltOne)
+ if r.Kind != retireCleanup || r.State != lifetimeCommitted {
+ t.Fatalf("retirement = %+v, want a committed cleanup", r)
+ }
+ q := quarantined(t, l.fx.rel, ltOne)
+ if len(q) == 0 || string(q[0]) != malformed {
+ t.Fatalf("quarantine = %q, want the malformed bytes", q)
+ }
+ if err := late(t, l); err != nil {
+ t.Fatalf("the delayed create after the retirement = %v", err)
+ }
+ if obs := l.fx.srv.judge.Observe(ctx, l.fx.rel, "bab", ltApp, ltOne); obs.Verdict != provision.LifetimeStale {
+ t.Fatalf("the restored record is %v, want stale", obs.Verdict)
+ }
+ }
+ t.Run("in one call", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ l.bucket.setHook(chain(corruptBeforeTombstone(t, l), slotRetiredWhenKeyLeaves(t, l, ltOne)))
+ if err := retireOn(t, l.fx.srv, ltOne, retireAny); err != nil {
+ t.Fatalf("retire = %v", err)
+ }
+ finished(t, l)
+ })
+ crashes := map[string]func(t *testing.T, l ltOn) (func(string, string) error, func()){
+ "after the copy": func(t *testing.T, l ltOn) (func(string, string) error, func()) {
+ return nil, sqlFailRetirement(t, l.fx.st, "lt_fail_convert", "NEW.kind = 'cleanup'")
+ },
+ "after the conversion": func(t *testing.T, l ltOn) (func(string, string) error, func()) {
+ return failNth("put", provision.HostIncarnationKey(ltOne), 1, false), nil
+ },
+ "after C1": func(t *testing.T, l ltOn) (func(string, string) error, func()) {
+ return failNth("delete-if-match", recKey(ltApp, ltOne), 1, false), nil
+ },
+ }
+ for name, crash := range crashes {
+ t.Run("crashed "+name, func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ fail, drop := crash(t, l)
+ hooks := []func(string, string) error{corruptBeforeTombstone(t, l), slotRetiredWhenKeyLeaves(t, l, ltOne)}
+ if fail != nil {
+ hooks = append(hooks, fail)
+ }
+ l.bucket.setHook(chain(hooks...))
+ if err := retireOn(t, l.fx.srv, ltOne, retireAny); !errors.Is(err, errLifetimePending) {
+ t.Fatalf("retire = %v, want pending", err)
+ }
+ if drop != nil {
+ drop()
+ }
+ l.bucket.setHook(slotRetiredWhenKeyLeaves(t, l, ltOne))
+ // The delayed original create is released in the interval: the
+ // key is present (corrupt) or the slot already retired, so it
+ // never lands under a live slot.
+ if err := late(t, l); err == nil {
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot.State == provision.SlotLive {
+ t.Fatal("the delayed create landed under the live slot")
+ }
+ }
+ r := lastRetirement(t, l.fx.st, ltOne)
+ l.recoveredTo(t, r.OpID, lifetimeCommitted)
+ if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); raw != nil {
+ t.Fatalf("a body is left: %s", raw)
+ }
+ finished(t, l)
+ })
+ }
+ t.Run("the quarantine delete loses to a repaired body", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ repaired, _ := json.Marshal(provision.Record{Schema: 1, Host: ltOne, App: ltApp, Account: "bab", Channel: "edge", Incarnation: 1, CreatedAt: time.Now().UTC()})
+ l.bucket.setHook(chain(corruptBeforeTombstone(t, l), onNth("delete-if-match", recKey(ltApp, ltOne), 1, func() {
+ _, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if _, err := l.fx.rel.Put(ctx, recKey(ltApp, ltOne), repaired, etag); err != nil {
+ t.Error(err)
+ }
+ })))
+ if err := retireOn(t, l.fx.srv, ltOne, retireAny); err != nil {
+ t.Fatalf("retire = %v", err)
+ }
+ finished(t, l)
+ })
+}
+
+func TestTerminationFencesFirst(t *testing.T) {
+ seq := func(t *testing.T, l ltOn) uint64 {
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ return decodeRec(t, raw).RecordSeq
+ }
+ t.Run("refused fences the changed record", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, "begin")
+ bump(t, l.fx.rel, ltApp, ltOne)
+ before := seq(t, l)
+ l.recoveredTo(t, r.OpID, lifetimeRefused)
+ if after := seq(t, l); after != before+1 {
+ t.Fatalf("record_seq %d -> %d, want the fence to advance it", before, after)
+ }
+ })
+ t.Run("void on another incarnation fences it", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, "begin")
+ raw, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ rec := decodeRec(t, raw)
+ rec.Incarnation = 9
+ body, _ := json.Marshal(rec)
+ if _, _, err := provision.WriteRecord(context.Background(), l.fx.rel, recKey(ltApp, ltOne), raw, body, etag); err != nil {
+ t.Fatal(err)
+ }
+ before := seq(t, l)
+ l.recoveredTo(t, r.OpID, lifetimeVoid)
+ if after := seq(t, l); after != before+1 {
+ t.Fatalf("record_seq %d -> %d, want the fence to advance it", before, after)
+ }
+ })
+ t.Run("a fence that cannot be written leaves it pending", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, "begin")
+ bump(t, l.fx.rel, ltApp, ltOne)
+ before := seq(t, l)
+ second := l.restart(t)
+ l.bucket.setHook(failNth("put", recKey(ltApp, ltOne), 1, true))
+ recoverOn(t, second)
+ l.bucket.setHook(nil)
+ if got := lastRetirement(t, l.fx.st, ltOne); got.OpID != r.OpID || got.State != lifetimePending {
+ t.Fatalf("retirement = %+v, want pending", got)
+ }
+ if after := seq(t, l); after != before {
+ t.Fatal("the record changed")
+ }
+ })
+ t.Run("a successor slot terminates on slot evidence alone", func(t *testing.T) {
+ l := newLtOn(t)
+ l.mustCreate(t, ltOne)
+ r := l.pendingRetire(t, "begin")
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 2, State: provision.SlotLive, Account: "bab", App: "web", Intent: "create web"})
+ before := len(l.bucket.opsOf(recKey(ltApp, ltOne)))
+ l.recoveredTo(t, r.OpID, lifetimeVoid)
+ for _, op := range l.bucket.opsOf(recKey(ltApp, ltOne))[before:] {
+ if op == "put" {
+ t.Fatal("a record fence on slot evidence")
+ }
+ }
+ })
+}
+
+// ---- detach, the sibling undo, the orphan rule ---------------------------------
+
+func TestDetachMapsLifetimeErrors(t *testing.T) {
+ attach := func(t *testing.T, l ltOn) {
+ t.Helper()
+ code, _, body := rawReq(t, http.MethodPost, domainsURL(l.fx, "bab", ltApp), l.member,
+ `{"host":"`+ltDomain+`","kind":"static","channel":"stable"}`)
+ if code != http.StatusOK {
+ t.Fatalf("attach = %d %s", code, body)
+ }
+ }
+ cases := []struct {
+ name string
+ arrange func(t *testing.T, l ltOn)
+ msg string
+ }{
+ {"another operation is settling", func(t *testing.T, l ltOn) {
+ if err := l.fx.st.BeginRetirement(Retirement{OpID: "other", Kind: retireOrphan, Account: "bab", App: "web", Host: ltDomain, Incarnation: 1}); err != nil {
+ t.Fatal(err)
+ }
+ }, claimDetachInProgressMsg},
+ {"the record changed in its lifetime", func(t *testing.T, l ltOn) {
+ l.bucket.setHook(onNth("put", recKey(ltApp, ltDomain), 1, func() { bump(t, l.fx.rel, ltApp, ltDomain) }))
+ }, msgRetireRefused},
+ {"the retirement is pending", func(t *testing.T, l ltOn) {
+ l.bucket.setHook(failNth("put", recKey(ltApp, ltDomain), 1, false))
+ }, claimDetachInProgressMsg},
+ }
+ for _, c := range cases {
+ for _, door := range []string{"api", "browser"} {
+ t.Run(c.name+", "+door, func(t *testing.T) {
+ l := newLtOn(t)
+ attach(t, l)
+ c.arrange(t, l)
+ if door == "api" {
+ code, _, body := rawReq(t, http.MethodDelete, domainsURL(l.fx, "bab", ltApp)+"/"+ltDomain, l.member, "")
+ if code != http.StatusConflict || ltErr(body) != c.msg {
+ t.Fatalf("detach = %d %s, want 409 %q", code, body, c.msg)
+ }
+ } else {
+ ref := l.fx.srv.detachDomain(onCtx(t, l.fx.srv), l.acct(t), ltApp, ltDomain, "alice@keymail.dev")
+ if ref == nil || ref.Status != http.StatusConflict || ref.Msg != c.msg {
+ t.Fatalf("detach = %+v, want 409 %q", ref, c.msg)
+ }
+ }
+ l.bucket.setHook(nil)
+ if got := claimHolder(t, l.fx.rel, ltDomain); got == "" {
+ t.Fatal("the claim was released over a refused or pending record removal")
+ }
+ })
+ }
+ }
+}
+
+func TestSiblingUndoRetiresOnlyTheLifetimeItCreated(t *testing.T) {
+ const host = "web.bab.oncarlos.test"
+ setup := func(t *testing.T) ltOn {
+ l := newLtOn(t)
+ l.fx.srv.cfg.AppsDomain = "oncarlos.test"
+ enableInstances(t, l.fx, ltApp, provision.Manifest{Schema: 1, App: ltApp, Domain: "bab.oncarlos.test", Channel: "stable"})
+ return l
+ }
+ claimSibling := func(t *testing.T, l ltOn, then func()) {
+ var once sync.Once
+ l.bucket.setAfter(func(op, key string, err error) {
+ if op == "put" && key == recKey(ltApp, host) && err == nil {
+ once.Do(func() {
+ if _, err := l.fx.st.CreateAppIn(OpsAccountID, "web", "paul@keymail.dev", time.Now()); err != nil {
+ t.Error(err)
+ }
+ if then != nil {
+ then()
+ }
+ })
+ }
+ })
+ }
+ t.Run("its own lifetime is retired", func(t *testing.T) {
+ l := setup(t)
+ claimSibling(t, l, nil)
+ if code, body := l.create(t, host); code != http.StatusConflict || !strings.Contains(body, "platform apex") {
+ t.Fatalf("create = %d %s, want the sibling refusal", code, body)
+ }
+ requireRetired(t, l.fx.rel, ltApp, host, 1)
+ })
+ t.Run("a successor lifetime is left", func(t *testing.T) {
+ l := setup(t)
+ claimSibling(t, l, func() {
+ // Another lifetime of the hostname replaced this one before the
+ // undo ran.
+ _, slotETag := slotOf(t, l.fx.rel, host)
+ if _, err := provision.WriteHostIncarnation(context.Background(), l.fx.rel, host,
+ provision.HostIncarnation{Incarnation: 2, State: provision.SlotLive, Account: "bab", App: ltApp, Intent: "create other"}, slotETag); err != nil {
+ t.Error(err)
+ }
+ raw, etag := rawOf(t, l.fx.rel, ltApp, host)
+ rec := decodeRec(t, raw)
+ rec.Incarnation = 2
+ body, _ := json.Marshal(rec)
+ if _, _, err := provision.WriteRecord(context.Background(), l.fx.rel, recKey(ltApp, host), raw, body, etag); err != nil {
+ t.Error(err)
+ }
+ })
+ if code, body := l.create(t, host); code != http.StatusConflict {
+ t.Fatalf("create = %d %s, want the sibling refusal", code, body)
+ }
+ raw, _ := rawOf(t, l.fx.rel, ltApp, host)
+ if raw == nil || decodeRec(t, raw).Incarnation != 2 {
+ t.Fatalf("record = %s, want the successor at 2 untouched", raw)
+ }
+ })
+}
+
+func TestOrphanSlotReclaimedByTheNextCreate(t *testing.T) {
+ t.Run("a live slot with no record and no pending op", func(t *testing.T) {
+ l := newLtOn(t)
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: "bab", App: ltApp, Intent: "create lost"})
+ if rec := l.mustCreate(t, ltOne); rec.Incarnation != 2 {
+ t.Fatalf("created at %d, want 2", rec.Incarnation)
+ }
+ if r := lastRetirement(t, l.fx.st, ltOne); r.Kind != retireOrphan || r.State != lifetimeCommitted {
+ t.Fatalf("retirement = %+v, want a committed orphan", r)
+ }
+ })
+ t.Run("with a pending op it is refused", func(t *testing.T) {
+ l := newLtOn(t)
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: "bab", App: "web", Intent: "create web"})
+ if err := l.fx.st.BeginRetirement(Retirement{OpID: "held", Kind: retireNormal, Account: "bab", App: "web", Host: ltOne, Incarnation: 1}); err != nil {
+ t.Fatal(err)
+ }
+ if code, body := l.create(t, ltOne); code != http.StatusConflict {
+ t.Fatalf("create = %d %s, want 409", code, body)
+ }
+ if rows := allocRows(t, l.fx.st, ltOne); len(rows) != 0 {
+ t.Fatalf("allocations = %+v, want none", rows)
+ }
+ })
+ t.Run("a stale body under the owner key is removed by a create under another key", func(t *testing.T) {
+ l := newLtOn(t)
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 1, State: provision.SlotRetired, Account: "bab", App: "web", Intent: "retire web"})
+ if _, err := l.fx.rel.Put(context.Background(), recKey("web", ltOne), mustJSON(t, provision.Record{Schema: 1, Host: ltOne, App: "web", Account: "bab", Channel: "stable", Incarnation: 1, CreatedAt: time.Now().UTC()}), ""); err != nil {
+ t.Fatal(err)
+ }
+ l.mustCreate(t, ltOne)
+ if raw, _ := rawOf(t, l.fx.rel, "web", ltOne); raw != nil {
+ t.Fatalf("the owner's stale body survived: %s", raw)
+ }
+ })
+}
+
+func mustJSON(t *testing.T, v any) []byte {
+ t.Helper()
+ b, err := json.Marshal(v)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return b
+}
+
+func TestCancellationIsOneTransaction(t *testing.T) {
+ t.Run("a crash after the handover leaves one pending row", func(t *testing.T) {
+ l := newLtOn(t)
+ a := l.pendingCreate(t, ltOne, "record")
+ trashShop(t, l)
+ second := l.restart(t)
+ // O1's slot write fails after the void and the retirement land.
+ l.bucket.setHook(failNth("put", provision.HostIncarnationKey(ltOne), 1, false))
+ recoverOn(t, second)
+ l.bucket.setHook(nil)
+ pending := 0
+ for _, r := range allocRows(t, l.fx.st, ltOne) {
+ if r.State == lifetimePending {
+ pending++
+ }
+ }
+ rets := retireRows(t, l.fx.st, ltOne)
+ for _, r := range rets {
+ if r.State == lifetimePending {
+ pending++
+ }
+ }
+ if pending != 1 || len(rets) != 1 || rets[0].Cancels != a.OpID || rets[0].State != lifetimePending {
+ t.Fatalf("allocations %+v, retirements %+v: want exactly the pending retirement", allocRows(t, l.fx.st, ltOne), rets)
+ }
+ oneAlloc(t, l.fx.st, ltOne, lifetimeVoid)
+ recoverOn(t, l.restart(t))
+ if r := lastRetirement(t, l.fx.st, ltOne); r.State != lifetimeCommitted {
+ t.Fatalf("retirement = %+v, want committed", r)
+ }
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ })
+ t.Run("a crash inside the handover changes nothing", func(t *testing.T) {
+ l := newLtOn(t)
+ l.pendingCreate(t, ltOne, "record")
+ trashShop(t, l)
+ if _, err := l.fx.st.db.Exec(`CREATE TRIGGER lt_fail_handover BEFORE INSERT ON record_retirements
+BEGIN SELECT RAISE(ABORT, 'test: the handover is refused'); END`); err != nil {
+ t.Fatal(err)
+ }
+ recoverOn(t, l.restart(t))
+ oneAlloc(t, l.fx.st, ltOne, lifetimePending)
+ if rows := retireRows(t, l.fx.st, ltOne); len(rows) != 0 {
+ t.Fatalf("retirements = %+v, want none", rows)
+ }
+ if _, err := l.fx.st.db.Exec(`DROP TRIGGER lt_fail_handover`); err != nil {
+ t.Fatal(err)
+ }
+ recoverOn(t, l.restart(t))
+ oneAlloc(t, l.fx.st, ltOne, lifetimeVoid)
+ requireRetired(t, l.fx.rel, ltApp, ltOne, 1)
+ })
+}
diff --git a/internal/console/seedfinish.go b/internal/console/seedfinish.go
index f79a4f18c..811e3b6f6 100644
--- a/internal/console/seedfinish.go
+++ b/internal/console/seedfinish.go
@@ -105,7 +105,14 @@ func (s *Server) finishSeedOnce(ctx context.Context, sqid, app, host, seedID str
return nil // another finish deleted the record
}
}
- if err := release.DeleteIfMatch(ctx, s.cfg.Releases, provision.RecordKey(sqid, app, host), etag); err != nil && !errors.Is(err, os.ErrNotExist) {
+ if lifetimesUnit(ctx) {
+ // Admitted on: the record retires through its tombstone, bound to
+ // the last read that saw this seed (lifetimes_retire.go). A CAS
+ // conflict is finishSeed's one re-read.
+ if err := s.retireRecordWithLifetime(ctx, sqid, app, host, retireExpectETag(etag)); err != nil {
+ return fmt.Errorf("retire the record of %s: %w", host, err)
+ }
+ } else if err := release.DeleteIfMatch(ctx, s.cfg.Releases, provision.RecordKey(sqid, app, host), etag); err != nil && !errors.Is(err, os.ErrNotExist) {
return fmt.Errorf("delete the record of %s: %w", host, err)
}
log.Printf("console: finished seed %s of %s/%s/%s", seedID, sqid, app, host)