amadan

carlos / platform Public

provision: reconcile observes each record through the judge, and the removal loop judges a row by its own record key

The decode loop is one sequence per key: the prefetched read, ObserveWith,
then the shape, channel and scope checks on the observation's record. A
stale record is recorded and left out of recs; an unreadable lifetime
protects the host; a listed key whose body is gone stays a read failure.

Claims-off, the removal loop observes the row's own key when the pass holds
no record for it under that key: a valid in-scope record defeats only the
record-gone reason, an out-of-scope one follows today's rules, an absent or
stale one is record gone and bypasses KeepParticipant, a held row going
through DeleteIfHold first. holdExisting re-reads through Observe; the import
journal's record readers take the judge. A nil judge is today's pass.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 7c8046dc656f256d0c2a3c77aa6115c4c8fd834f parent 52421ae
6 files changed, +801 −24
  • internal/provision/reconcile.go +146 −18
  • internal/provision/reconcile_lifetime_test.go +534 −0
  • internal/provision/seedjournal_test.go +1 −1
  • internal/provision/seedreconcile.go +6 −3
  • internal/provision/seedreconcile_lifetime_test.go +82 −0
  • internal/provision/seedrollback.go +32 −2
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go
index 6c9f1276c..73dc7ed9a 100644
--- a/internal/provision/reconcile.go
+++ b/internal/provision/reconcile.go
@@ -136,6 +136,19 @@ type ReconcileConfig struct {
// Seeds is the seeded-instances wiring (seedreconcile.go). Nil: every
// seeded record is left alone.
Seeds *SeedHooks
+ // Judge is the box's lifetime judge (self-serve moves plan 1, Task 6):
+ // set on a claims-off box, nil on a claims-on one. With a judge every
+ // listed record is observed, the record then its host incarnation
+ // slot, and only a record the slot vouches for is acted on; the removal
+ // phase observes a row's own record key when the pass holds no record
+ // for the row under that key. Nil reads no slot: today's pass exactly.
+ Judge *LifetimeJudge
+ // DeleteIfHold deletes a row only while op holds it, with the move's
+ // timers (registry.DeleteIfHoldWithTimers). The removal phase uses it,
+ // under a judge, for a held row whose own record is established gone
+ // or stale: the record carries no op, so the hold cannot be its own.
+ // Nil: such a row is removed as any other (claimRetire).
+ DeleteIfHold func(host, op string) (bool, error)
// beforeHoldWrite is the tests' pause under the host's hold lock, just
// before holdExisting's conditional write (TestReconcileHoldWrite...).
@@ -851,31 +864,77 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
recordKeys = append(recordKeys, key)
}
}
+ // stale holds hosts whose listed record the judge found its slot no
+ // longer vouches for (a fieldless record beside a slot, a lifetime
+ // behind its slot, a tombstone). Such a record is not in recs and not
+ // protected: to this pass it is gone. The removal phase observes a
+ // row's own key afresh before it acts on that.
+ stale := map[string]bool{}
for _, part := range chunks(recordKeys) {
bodies := getEach(ctx, s, part)
- for _, key := range part {
+ // One sequence per key: the prefetched read, then the judge's
+ // observation of it (the slot read after the record, in bounded
+ // rounds), then the checks below on the observation's record. A nil
+ // judge decodes and path-checks the prefetched body only. The slot
+ // reads run as the prefetch does, a handful at a time.
+ observations := make(map[string]Observation, len(part))
+ var obsMu sync.Mutex
+ observe := func(key string) {
acct, app, host, _ := ParseRecordKey(key, known)
- raw, err := bodies[key].body, bodies[key].err
- if err != nil {
+ b := bodies[key]
+ obs := cfg.Judge.ObserveWith(ctx, s, acct, app, host, b.body, b.etag, b.err)
+ obsMu.Lock()
+ observations[key] = obs
+ obsMu.Unlock()
+ }
+ if cfg.Judge == nil {
+ for _, key := range part {
+ observe(key)
+ }
+ } else {
+ forEach(part, observe)
+ }
+ for _, key := range part {
+ _, _, host, _ := ParseRecordKey(key, known)
+ obs := observations[key]
+ if !obs.Present {
+ err := bodies[key].err
+ if err == nil {
+ err = obs.Err
+ }
+ if err == nil {
+ err = os.ErrNotExist
+ }
log.Printf("provision: skipping record %s: read failed: %v", key, err)
// The record's absence from recs below must not read as "the
// bucket retired it" — a transient Get failure disables removal
// for the whole pass, same as an unreadable account manifest.
+ // A listed key whose body is gone by the time it is read is
+ // the same failure, judge or no judge: absence is not
+ // staleness.
readFailed = true
continue
}
- rec, err := DecodeRecord(raw)
- if err != nil {
- log.Printf("provision: skipping record %s: malformed: %v", key, err)
+ if obs.Verdict == LifetimeCorrupt {
+ log.Printf("provision: skipping record %s: malformed: %v", key, obs.Err)
protect[host] = true
continue
}
- if rec.Host != host || rec.Account != acct || rec.App != app {
- log.Printf("provision: skipping record %s: disagrees with its own path (host=%q account=%q app=%q)",
- key, rec.Host, rec.Account, rec.App)
+ if obs.Verdict == LifetimeUnreadable {
+ // The record read, but its lifetime could not be judged this
+ // pass (the slot would not read or is malformed, the rounds
+ // ran out, a counter is out of range): protected like a
+ // corrupt body, never taken for retired.
+ log.Printf("provision: skipping record %s: its lifetime could not be judged: %v", key, obs.Err)
protect[host] = true
continue
}
+ if obs.Verdict == LifetimeStale {
+ log.Printf("provision: skipping record %s: its host incarnation slot no longer vouches for it", key)
+ stale[host] = true
+ continue
+ }
+ rec := *obs.Rec
// The record IS the authorization, and this pass turns its host into
// a socket path and a database path (DefaultSocket/DefaultDBPath)
// and a registry primary key. The console's create door enforces
@@ -1531,14 +1590,55 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
// copies them from the record — so a row missing either is a
// shape this pass did not mint, and not one it will retire.
why := "record gone"
- if _, ok := recs[rt.Host]; ok {
+ // gone is the judge's evidence that the row's OWN record is
+ // absent or stale (claims-off only, below): it also outranks a
+ // keep view, which keys by hostname and cannot vouch for a row
+ // whose own record is gone.
+ gone := false
+ rec, have := recs[rt.Host]
+ ownKey := have && RecordKey(rec.Account, rec.App, rec.Host) == RecordKey(rt.Account, rt.App, rt.Host)
+ switch {
+ case journaled[rt.Host]:
+ continue // the import journal owns the row (seedreconcile.go)
+ case ownKey || (have && cfg.Judge == nil):
+ if rt.Account == "" || rt.App == "" || !trashFor(rt.Account, rt.App).removable {
+ continue
+ }
+ why = "app in the trash"
+ case cfg.Judge != nil && rt.Account != "" && rt.App != "":
+ // The pass holds no record for this row under the row's own
+ // key: none for the host, or another owner's (a legacy
+ // collision, or a successor lifetime minted after this row's
+ // owner retired). Validity and placement are separate
+ // questions, so the row's own key is observed afresh.
+ obs := cfg.Judge.Observe(ctx, s, rt.Account, rt.App, rt.Host)
+ switch {
+ case obs.Verdict == LifetimeUnreadable || obs.Verdict == LifetimeCorrupt:
+ log.Printf("provision: not removing %s: its record's lifetime could not be judged this pass: %v", rt.Host, obs.Err)
+ continue
+ case obs.Present && obs.Verdict == LifetimeValid && !obs.Rec.Retired:
+ if cfg.inScope(obs.Rec.Region, obs.Rec.Pool, everywhereClass(*obs.Rec)) {
+ // A valid record in this pass's scope defeats the
+ // record-gone reason only: the row still meets the
+ // app-trash rule and every exemption below.
+ if !trashFor(rt.Account, rt.App).removable {
+ continue
+ }
+ why = "app in the trash"
+ }
+ // Out of scope (another region after a flip, another
+ // pool): today's rules exactly, the keep view included.
+ default:
+ gone = true
+ if stale[rt.Host] || obs.Present {
+ why = "record gone (its lifetime is retired)"
+ }
+ }
+ case have:
if rt.Account == "" || rt.App == "" || !trashFor(rt.Account, rt.App).removable {
continue
}
why = "app in the trash"
- }
- if journaled[rt.Host] {
- continue // the import journal owns the row (seedreconcile.go)
}
if protect[rt.Host] {
log.Printf("provision: not removing %s: its record exists but could not be read cleanly this pass", rt.Host)
@@ -1548,7 +1648,7 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
// the record's region says (moves spec §9.6, claim spec §8): a
// move's source after the flip, its target during a rollback.
// The operation deletes the row at its own finalization.
- if cfg.KeepParticipant != nil && cfg.KeepParticipant(rt.Host) {
+ if !gone && cfg.KeepParticipant != nil && cfg.KeepParticipant(rt.Host) {
log.Printf("provision: not removing %s (%s): participant of a running operation", rt.Host, why)
continue
}
@@ -1556,6 +1656,22 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
log.Printf("provision: dry-run: would remove retired %s route %s (%s)", rt.Kind, rt.Host, why)
continue
}
+ // A held row whose own record is gone is held by an op that
+ // record does not carry: it is deleted only while that op still
+ // holds it, with the move's timers, so a row a new hold took
+ // since the listing is left to the next pass.
+ if gone && rt.HoldOp != "" && cfg.DeleteIfHold != nil {
+ deleted, err := cfg.DeleteIfHold(rt.Host, rt.HoldOp)
+ switch {
+ case err != nil:
+ log.Printf("provision: remove retired %s route %s held by %s: %v", rt.Kind, rt.Host, rt.HoldOp, err)
+ case !deleted:
+ log.Printf("provision: not removing %s (%s): its hold changed since this pass listed it; the next pass judges it afresh", rt.Host, why)
+ default:
+ log.Printf("provision: removed retired %s route %s held by %s (%s)", rt.Kind, rt.Host, rt.HoldOp, why)
+ }
+ continue
+ }
if err := cfg.claimRetire(ctx, reg, rt); err != nil {
log.Printf("provision: remove retired %s route %s: %v", rt.Kind, rt.Host, err)
continue
@@ -1709,7 +1825,17 @@ func holdExisting(ctx context.Context, s release.Store, reg *registry.Registry,
return
}
WithHostHoldLock(rec.Host, func() {
- fresh, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
+ // Observed, so under a judge a record its slot no longer vouches for
+ // asks for no hold; a nil judge reads it as GetRecord does.
+ obs := cfg.Judge.Observe(ctx, s, rec.Account, rec.App, rec.Host)
+ fresh, err := obs.Rec, obs.Err
+ switch {
+ case err != nil:
+ case !obs.Present:
+ err = os.ErrNotExist
+ case obs.Verdict == LifetimeStale || fresh.Retired:
+ err = fmt.Errorf("provision: record %s/%s/%s: its lifetime is retired: %w", rec.Account, rec.App, rec.Host, os.ErrNotExist)
+ }
if err != nil {
log.Printf("provision: %s: not holding this pass: re-read the record: %v", rec.Host, err)
return
@@ -2092,9 +2218,11 @@ func chunks(keys []string) [][]string {
return out
}
-// fetched is one prefetched object: its body, or why it could not be read.
+// fetched is one prefetched object: its body and version, or why it could
+// not be read.
type fetched struct {
body []byte
+ etag string
err error
}
@@ -2122,9 +2250,9 @@ func getEach(ctx context.Context, s release.Store, keys []string) map[string]fet
}
var mu sync.Mutex
forEach(keys, func(k string) {
- body, _, err := s.Get(ctx, k)
+ body, etag, err := s.Get(ctx, k)
mu.Lock()
- out[k] = fetched{body: body, err: err}
+ out[k] = fetched{body: body, etag: etag, err: err}
mu.Unlock()
})
return out
diff --git a/internal/provision/reconcile_lifetime_test.go b/internal/provision/reconcile_lifetime_test.go
new file mode 100644
index 000000000..45244cbb7
--- /dev/null
+++ b/internal/provision/reconcile_lifetime_test.go
@@ -0,0 +1,534 @@
+package provision
+
+import (
+ "context"
+ "errors"
+ "os"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/seeds"
+)
+
+// The reconcile pass under a lifetime judge (self-serve moves plan 1, Task
+// 6): the decode loop judges every listed record through ObserveWith, and
+// the removal loop observes a row's own record key when the pass holds no
+// record for it under that key.
+
+// lifeHost is the one hostname these scenes are about.
+const lifeHost = "a.shop.example.com"
+
+// lifeScene is a claims-off box: a bucket with the shop account and app (and
+// a second app, cart, for the cross-owner scenes), a registry, and a pass
+// configuration with the box's judge.
+type lifeScene struct {
+ s *release.DirStore
+ sqid string
+ reg *registry.Registry
+ cfg ReconcileConfig
+}
+
+func newLifeScene(t *testing.T) *lifeScene {
+ t.Helper()
+ s, sqid := shopSetup(t)
+ if err := PutManifest(ctx, s, sqid, "cart", Manifest{Schema: 1, App: "cart", Domain: "shop.example.com"}); err != nil {
+ t.Fatal(err)
+ }
+ return &lifeScene{s: s, sqid: sqid, reg: regWith(t),
+ cfg: ReconcileConfig{Region: "ie", DataDir: "/data", Judge: NewLifetimeJudge()}}
+}
+
+// record is the shop (or another app's) record for lifeHost at incarnation
+// inc; 0 is a fieldless legacy record.
+func (x *lifeScene) record(app string, inc uint64) Record {
+ rec := shopRecord(x.sqid)
+ rec.App, rec.Incarnation = app, inc
+ return rec
+}
+
+// put writes value under key whatever is there, and returns the etag.
+func (x *lifeScene) put(t *testing.T, key string, value any) string {
+ t.Helper()
+ _, tag := judgePut(t, x.s, key, value)
+ return tag
+}
+
+func (x *lifeScene) putRecord(t *testing.T, rec Record) string {
+ t.Helper()
+ return x.put(t, RecordKey(rec.Account, rec.App, rec.Host), rec)
+}
+
+func (x *lifeScene) putSlot(t *testing.T, app string, inc uint64, state string) {
+ t.Helper()
+ x.put(t, HostIncarnationKey(lifeHost), HostIncarnation{Account: x.sqid, App: app, Incarnation: inc, State: state})
+}
+
+func (x *lifeScene) deleteRecord(t *testing.T, app string) {
+ t.Helper()
+ if err := x.s.Delete(ctx, RecordKey(x.sqid, app, lifeHost)); err != nil && !errors.Is(err, os.ErrNotExist) {
+ t.Fatal(err)
+ }
+}
+
+func (x *lifeScene) pass(t *testing.T) {
+ t.Helper()
+ if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (x *lifeScene) row(t *testing.T) *registry.Route {
+ t.Helper()
+ rt, err := x.reg.Get(lifeHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return rt
+}
+
+// minted runs a pass that must mint the row for app and returns it.
+func (x *lifeScene) minted(t *testing.T, app string) registry.Route {
+ t.Helper()
+ x.pass(t)
+ rt := x.row(t)
+ if rt == nil || rt.App != app || !rt.Provisioned {
+ t.Fatalf("row after the minting pass = %+v, want %s's", rt, app)
+ }
+ return *rt
+}
+
+func TestReconcileRemovesARecordBehindItsSlot(t *testing.T) {
+ for _, c := range []struct {
+ name string
+ inc uint64 // the record's incarnation; 0 is fieldless
+ after func(t *testing.T, x *lifeScene)
+ }{
+ {name: "fieldless record beside a retired slot", inc: 0, after: func(t *testing.T, x *lifeScene) {
+ x.putSlot(t, "shop", 1, SlotRetired)
+ }},
+ {name: "incarnation 1 behind a retired slot", inc: 1, after: func(t *testing.T, x *lifeScene) {
+ x.putSlot(t, "shop", 1, SlotRetired)
+ }},
+ {name: "incarnation 1 behind incarnation 2", inc: 1, after: func(t *testing.T, x *lifeScene) {
+ x.putSlot(t, "shop", 2, SlotLive)
+ }},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ x := newLifeScene(t)
+ x.putRecord(t, x.record("shop", c.inc))
+ if c.inc > 0 {
+ x.putSlot(t, "shop", c.inc, SlotLive)
+ }
+ x.minted(t, "shop")
+ c.after(t, x)
+
+ // Claims on, the judge is nil and the pass reads no slot: the
+ // row stays, exactly as today.
+ claimsOn := x.cfg
+ claimsOn.Judge = nil
+ if err := Reconcile(ctx, x.s, x.reg, claimsOn); err != nil {
+ t.Fatal(err)
+ }
+ if x.row(t) == nil {
+ t.Fatal("a nil judge removed the row")
+ }
+
+ x.pass(t)
+ if rt := x.row(t); rt != nil {
+ t.Fatalf("the row of a record behind its slot survived: %+v", rt)
+ }
+ // And nothing is minted back on the next pass.
+ x.pass(t)
+ if rt := x.row(t); rt != nil {
+ t.Fatalf("a stale record was minted: %+v", rt)
+ }
+ })
+ }
+}
+
+func TestReconcileRemovesATombstone(t *testing.T) {
+ x := newLifeScene(t)
+ tag := x.putRecord(t, x.record("shop", 1))
+ x.putSlot(t, "shop", 1, SlotLive)
+ x.minted(t, "shop")
+ key := RecordKey(x.sqid, "shop", lifeHost)
+ raw, _, err := x.s.Get(ctx, key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ tomb, err := TombstoneBody(raw)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := x.s.Put(ctx, key, tomb, tag); err != nil {
+ t.Fatal(err)
+ }
+ x.putSlot(t, "shop", 1, SlotRetired)
+
+ // Today a tombstone is a body this pass cannot use, so it protects the
+ // row. A nil judge keeps that.
+ claimsOn := x.cfg
+ claimsOn.Judge = nil
+ if err := Reconcile(ctx, x.s, x.reg, claimsOn); err != nil {
+ t.Fatal(err)
+ }
+ if x.row(t) == nil {
+ t.Fatal("a nil judge removed the row of a tombstone")
+ }
+ x.pass(t)
+ if rt := x.row(t); rt != nil {
+ t.Fatalf("the row of a tombstone survived: %+v", rt)
+ }
+}
+
+func TestReconcileKeepsALegacyRecordWithNoSlot(t *testing.T) {
+ x := newLifeScene(t)
+ x.putRecord(t, x.record("shop", 0))
+ x.minted(t, "shop")
+ x.pass(t)
+ if rt := x.row(t); rt == nil || rt.App != "shop" {
+ t.Fatalf("a legacy record with no slot lost its row: %+v", rt)
+ }
+}
+
+func TestReconcileKeepsAMarkedRecordUnderItsMigrationSlot(t *testing.T) {
+ x := newLifeScene(t)
+ rec := x.record("shop", 0)
+ rec.IncarnationMigrating = "mig_1"
+ tag := x.putRecord(t, rec)
+ x.put(t, HostIncarnationKey(lifeHost), HostIncarnation{Account: x.sqid, App: "shop", Incarnation: 1, State: SlotLive,
+ Intent: "migration mig_1", RecordETag: tag})
+ x.minted(t, "shop")
+ x.pass(t)
+ if rt := x.row(t); rt == nil {
+ t.Fatal("a marked record under its migration slot lost its row")
+ }
+}
+
+// slotFailStore fails every Get of the host's slot once armed.
+type slotFailStore struct {
+ release.Store
+ mu sync.Mutex
+ reads int
+ after int // the slot reads that succeed before the failures start
+ fail bool
+}
+
+func (s *slotFailStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if strings.HasPrefix(key, "control/host-incarnations/") {
+ s.mu.Lock()
+ s.reads++
+ failing := s.fail && s.reads > s.after
+ s.mu.Unlock()
+ if failing {
+ return nil, "", errors.New("slot read failed")
+ }
+ }
+ return s.Store.Get(ctx, key)
+}
+
+func TestReconcileProtectsWhenTheSlotIsUnreadable(t *testing.T) {
+ t.Run("the decode loop", func(t *testing.T) {
+ x := newLifeScene(t)
+ x.putRecord(t, x.record("shop", 1))
+ x.putSlot(t, "shop", 1, SlotLive)
+ x.minted(t, "shop")
+ // The record is now behind its slot, but the slot cannot be read:
+ // the row is protected, not removed.
+ x.putSlot(t, "shop", 2, SlotLive)
+ failing := &slotFailStore{Store: x.s, fail: true}
+ if err := Reconcile(ctx, failing, x.reg, x.cfg); err != nil {
+ t.Fatal(err)
+ }
+ if x.row(t) == nil {
+ t.Fatal("an unreadable slot removed the row")
+ }
+ })
+ t.Run("a malformed slot", func(t *testing.T) {
+ x := newLifeScene(t)
+ x.putRecord(t, x.record("shop", 1))
+ x.putSlot(t, "shop", 1, SlotLive)
+ x.minted(t, "shop")
+ x.put(t, HostIncarnationKey(lifeHost), map[string]any{"incarnation": 0})
+ x.pass(t)
+ if x.row(t) == nil {
+ t.Fatal("a malformed slot removed the row")
+ }
+ })
+ t.Run("the removal loop's own observation", func(t *testing.T) {
+ // The row is shop's, shop's record is gone and cart holds the
+ // hostname: the removal loop observes shop's key itself, and when
+ // that observation cannot read the slot the row is protected.
+ x := newLifeScene(t)
+ x.putRecord(t, x.record("shop", 0))
+ x.minted(t, "shop")
+ x.deleteRecord(t, "shop")
+ x.putRecord(t, x.record("cart", 0))
+ failing := &slotFailStore{Store: x.s, fail: true, after: 1} // the decode loop's read of cart's slot
+ if err := Reconcile(ctx, failing, x.reg, x.cfg); err != nil {
+ t.Fatal(err)
+ }
+ if rt := x.row(t); rt == nil || rt.App != "shop" {
+ t.Fatalf("an unreadable observation removed the row: %+v", rt)
+ }
+ // Readable, the same evidence retires it.
+ x.pass(t)
+ if rt := x.row(t); rt != nil && rt.App == "shop" {
+ t.Fatalf("shop's row survived its record: %+v", rt)
+ }
+ })
+}
+
+// swapOnFirstRead replaces the record and slot right after the first read of
+// the record key: the prefetched body is the predecessor, the slot read
+// after it already names the successor.
+type swapOnFirstRead struct {
+ release.Store
+ key string
+ once sync.Once
+ swap func()
+}
+
+func (s *swapOnFirstRead) Get(ctx context.Context, key string) ([]byte, string, error) {
+ raw, tag, err := s.Store.Get(ctx, key)
+ if key == s.key {
+ s.once.Do(s.swap)
+ }
+ return raw, tag, err
+}
+
+func TestReconcileUsesTheObservation(t *testing.T) {
+ x := newLifeScene(t)
+ old := x.record("shop", 1)
+ old.Channel = "stable"
+ x.putRecord(t, old)
+ x.putSlot(t, "shop", 1, SlotLive)
+ next := x.record("shop", 2)
+ next.Channel = "beta"
+ s := &swapOnFirstRead{Store: x.s, key: RecordKey(x.sqid, "shop", lifeHost), swap: func() {
+ x.putSlot(t, "shop", 2, SlotLive)
+ x.putRecord(t, next)
+ }}
+ if err := Reconcile(ctx, s, x.reg, x.cfg); err != nil {
+ t.Fatal(err)
+ }
+ rt := x.row(t)
+ if rt == nil || rt.Channel != "beta" {
+ t.Fatalf("row = %+v, want the observation's record (channel beta), not the prefetched one", rt)
+ }
+}
+
+func TestStaleOverridesAPopulatedKeepView(t *testing.T) {
+ for _, held := range []bool{false, true} {
+ name := "unheld"
+ if held {
+ name = "held"
+ }
+ t.Run(name, func(t *testing.T) {
+ x := newLifeScene(t)
+ x.putRecord(t, x.record("shop", 0))
+ x.minted(t, "shop")
+ if held {
+ if ok, err := x.reg.SetHoldIf(lifeHost, "moving", "mv_1", ""); err != nil || !ok {
+ t.Fatalf("hold = %v, %v", ok, err)
+ }
+ }
+ x.putSlot(t, "shop", 1, SlotRetired)
+ var deleted []string
+ x.cfg.KeepParticipant = func(string) bool { return true }
+ x.cfg.DeleteIfHold = func(host, op string) (bool, error) {
+ deleted = append(deleted, host+" "+op)
+ return x.reg.DeleteIfHoldWithTimers(host, op)
+ }
+ x.pass(t)
+ if rt := x.row(t); rt != nil {
+ t.Fatalf("a keep view kept the row of a stale record: %+v", rt)
+ }
+ if held && (len(deleted) != 1 || deleted[0] != lifeHost+" mv_1") {
+ t.Fatalf("held row deletes = %q, want one DeleteIfHold by its op", deleted)
+ }
+ if !held && len(deleted) != 0 {
+ t.Fatalf("an unheld row went through DeleteIfHold: %q", deleted)
+ }
+ })
+ }
+ t.Run("a nil judge keeps today's keep", func(t *testing.T) {
+ x := newLifeScene(t)
+ x.putRecord(t, x.record("shop", 0))
+ x.minted(t, "shop")
+ x.putSlot(t, "shop", 1, SlotRetired)
+ x.deleteRecord(t, "shop")
+ x.cfg.Judge = nil
+ x.cfg.KeepParticipant = func(string) bool { return true }
+ x.pass(t)
+ if x.row(t) == nil {
+ t.Fatal("a nil judge bypassed KeepParticipant")
+ }
+ })
+}
+
+func TestCrossOwnerRecreationWithdrawsThePredecessorRow(t *testing.T) {
+ type variant struct {
+ name string
+ absent bool // shop's record physically gone (else present behind the slot)
+ held bool
+ keepAll bool
+ }
+ var variants []variant
+ for _, absent := range []bool{true, false} {
+ for _, held := range []bool{false, true} {
+ for _, keep := range []bool{false, true} {
+ n := "stale"
+ if absent {
+ n = "absent"
+ }
+ if held {
+ n += ", held"
+ }
+ if keep {
+ n += ", kept by the view"
+ }
+ variants = append(variants, variant{n, absent, held, keep})
+ }
+ }
+ }
+ for _, v := range variants {
+ t.Run(v.name, func(t *testing.T) {
+ // Box B minted shop's row at incarnation 1 and missed its
+ // retirement; cart has since recreated the hostname at 2.
+ x := newLifeScene(t)
+ x.putRecord(t, x.record("shop", 1))
+ x.putSlot(t, "shop", 1, SlotLive)
+ x.minted(t, "shop")
+ if v.held {
+ if ok, err := x.reg.SetHoldIf(lifeHost, "moving", "mv_1", ""); err != nil || !ok {
+ t.Fatalf("hold = %v, %v", ok, err)
+ }
+ }
+ if v.absent {
+ x.deleteRecord(t, "shop")
+ }
+ x.putSlot(t, "cart", 2, SlotLive)
+ x.putRecord(t, x.record("cart", 2))
+ x.cfg.DeleteIfHold = x.reg.DeleteIfHoldWithTimers
+ if v.keepAll {
+ x.cfg.KeepParticipant = func(string) bool { return true }
+ }
+
+ x.pass(t)
+ if rt := x.row(t); rt != nil {
+ t.Fatalf("pass 1 left %+v, want the predecessor's row retired", rt)
+ }
+ x.pass(t)
+ if rt := x.row(t); rt == nil || rt.App != "cart" {
+ t.Fatalf("pass 2 row = %+v, want cart's minted", rt)
+ }
+ })
+ }
+ t.Run("two valid legacy owners: the non-selected key's row is kept", func(t *testing.T) {
+ x := newLifeScene(t)
+ x.putRecord(t, x.record("shop", 0))
+ x.minted(t, "shop")
+ // cart sorts first, so the pass selects cart's record for the host.
+ x.putRecord(t, x.record("cart", 0))
+ x.pass(t)
+ x.pass(t)
+ if rt := x.row(t); rt == nil || rt.App != "shop" {
+ t.Fatalf("row = %+v, want shop's kept as today", rt)
+ }
+ })
+ t.Run("a journaled host is untouched", func(t *testing.T) {
+ // An import this box swapped, on a box that cannot act on imports:
+ // the journal owns the row, whatever the judge says of its record.
+ j := journaledImport(t, seeds.StateSwapped)
+ j.cfg.Seeds = nil
+ j.cfg.Judge = NewLifetimeJudge()
+ if err := PutManifest(ctx, j.s, j.rec.Account, "cart", Manifest{Schema: 1, App: "cart", Domain: "shop.example.com"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := j.s.Delete(ctx, RecordKey(j.rec.Account, j.rec.App, j.rec.Host)); err != nil {
+ t.Fatal(err)
+ }
+ cart := shopRecord(j.rec.Account)
+ cart.App, cart.Incarnation = "cart", 2
+ judgePut(t, j.s, RecordKey(cart.Account, cart.App, cart.Host), cart)
+ judgePut(t, j.s, HostIncarnationKey(cart.Host), HostIncarnation{Account: cart.Account, App: "cart", Incarnation: 2, State: SlotLive})
+ for range 2 {
+ if err := Reconcile(ctx, j.s, j.reg, j.cfg); err != nil {
+ t.Fatal(err)
+ }
+ }
+ j.untouched(t)
+ })
+ t.Run("a hand-managed row is untouched", func(t *testing.T) {
+ x := newLifeScene(t)
+ hand := RouteFor(x.record("shop", 0), "/data", "", false, nil)
+ hand.Provisioned = false
+ if err := x.reg.Upsert(hand); err != nil {
+ t.Fatal(err)
+ }
+ x.putSlot(t, "cart", 2, SlotLive)
+ x.putRecord(t, x.record("cart", 2))
+ x.pass(t)
+ x.pass(t)
+ if rt := x.row(t); rt == nil || rt.App != "shop" || rt.Provisioned {
+ t.Fatalf("a hand-managed row was touched: %+v", rt)
+ }
+ })
+}
+
+func TestTrashedOwnerCollisionIsRemoved(t *testing.T) {
+ x := newLifeScene(t)
+ x.putRecord(t, x.record("shop", 0))
+ x.minted(t, "shop")
+ x.putRecord(t, x.record("cart", 0)) // cart sorts first: recs selects it
+ if err := release.TrashApp(ctx, x.s, x.sqid, "shop", "op@example.com", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ x.pass(t)
+ if rt := x.row(t); rt != nil && rt.App == "shop" {
+ t.Fatalf("a trashed owner's colliding row survived: %+v", rt)
+ }
+}
+
+func TestOutOfScopeValidRecordFollowsTodaysRemoval(t *testing.T) {
+ for _, c := range []struct {
+ name string
+ move func(*Record)
+ keep bool
+ trash bool
+ kept bool
+ }{
+ {name: "another region", move: func(r *Record) { r.Region = "uk" }},
+ {name: "another pool", move: func(r *Record) { r.Pool = "other" }},
+ {name: "another region, kept by the view", move: func(r *Record) { r.Region = "uk" }, keep: true, kept: true},
+ {name: "another pool, kept by the view", move: func(r *Record) { r.Pool = "other" }, keep: true, kept: true},
+ {name: "another region, the app trashed", move: func(r *Record) { r.Region = "uk" }, trash: true},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ x := newLifeScene(t)
+ rec := x.record("shop", 1)
+ x.putRecord(t, rec)
+ x.putSlot(t, "shop", 1, SlotLive)
+ x.minted(t, "shop")
+ c.move(&rec)
+ x.putRecord(t, rec)
+ if c.keep {
+ x.cfg.KeepParticipant = func(string) bool { return true }
+ }
+ if c.trash {
+ if err := release.TrashApp(ctx, x.s, x.sqid, "shop", "op@example.com", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ }
+ x.pass(t)
+ if got := x.row(t) != nil; got != c.kept {
+ t.Fatalf("row kept = %v, want %v", got, c.kept)
+ }
+ })
+ }
+}
diff --git a/internal/provision/seedjournal_test.go b/internal/provision/seedjournal_test.go
index 9744cc7c2..b56e71a0b 100644
--- a/internal/provision/seedjournal_test.go
+++ b/internal/provision/seedjournal_test.go
@@ -467,7 +467,7 @@ func TestJournalRecordFailsClosed(t *testing.T) {
t.Fatal(err)
}
}
- got, pass, known := journalRecord(ctx, c.store, c.j, c.recs, c.readFailed, c.protect)
+ got, pass, known := journalRecord(ctx, c.store, c.j, c.recs, c.readFailed, c.protect, nil)
if known != c.known || pass != c.pass || (known && (got == nil) != c.gone) {
t.Fatalf("record %v, in pass %v, known %v; want known %v, gone %v, in pass %v", got, pass, known, c.known, c.gone, c.pass)
}
diff --git a/internal/provision/seedreconcile.go b/internal/provision/seedreconcile.go
index 4d723ead8..d9af28c3e 100644
--- a/internal/provision/seedreconcile.go
+++ b/internal/provision/seedreconcile.go
@@ -369,7 +369,7 @@ func journalPass(ctx context.Context, s release.Store, reg *registry.Registry, c
log.Printf("provision: %s: import %s is journaled at %s and claim control is on for this box; its row is protected and nothing else is done", j.Host, j.SeedID, j.State)
continue
}
- rec, inPass, known := journalRecord(ctx, s, j, recs, readFailed, protect)
+ rec, inPass, known := journalRecord(ctx, s, j, recs, readFailed, protect, cfg.Judge)
if !known {
log.Printf("provision: %s: import %s left alone this pass: its record could not be read", j.Host, j.SeedID)
continue
@@ -448,7 +448,7 @@ func startRollback(ctx context.Context, s release.Store, reg *registry.Registry,
// key all decide nothing this pass: a committed import must never be
// rolled back on a guess.
func journalRecord(ctx context.Context, s release.Store, j registry.ImportJournal, recs map[string]Record,
- readFailed bool, protect map[string]bool) (rec *Record, inPass, known bool) {
+ readFailed bool, protect map[string]bool, judge *LifetimeJudge) (rec *Record, inPass, known bool) {
if protect[j.Host] || j.RecordKey == "" {
return nil, false, false
}
@@ -458,7 +458,10 @@ func journalRecord(ctx context.Context, s release.Store, j registry.ImportJourna
if readFailed {
return nil, false, false
}
- r, err := journalRecordByKey(ctx, s, j)
+ // Under a judge a stale record is gone (nil, known): the journal's rules
+ // for an absent record decide it. An observation that cannot be judged
+ // leaves known false, as a failed read does.
+ r, err := journalRecordByKey(ctx, s, j, judge)
if err != nil {
log.Printf("provision: %s: the import's record %s is not one this pass can use (%v)", j.Host, j.RecordKey, err)
return nil, false, false
diff --git a/internal/provision/seedreconcile_lifetime_test.go b/internal/provision/seedreconcile_lifetime_test.go
new file mode 100644
index 000000000..35d9f4381
--- /dev/null
+++ b/internal/provision/seedreconcile_lifetime_test.go
@@ -0,0 +1,82 @@
+package provision
+
+import (
+ "reflect"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/seeds"
+)
+
+// A journal row at swapped decides by its own record (seedreconcile.go). A
+// record its slot no longer vouches for is gone to the journal's rules,
+// which roll the import back; it never opens the gate or redoes the swap's
+// record CAS. A record whose lifetime cannot be judged leaves the import
+// alone, as an unreadable record does today.
+func TestImportJournalIgnoresAStaleRecord(t *testing.T) {
+ for _, state := range []string{seeds.StateCommitted, seeds.StateReady} {
+ t.Run("stale "+state, func(t *testing.T) {
+ x := journaledImport(t, state)
+ x.cfg.Judge = NewLifetimeJudge()
+ // The fieldless record beside a retired slot: an old console's
+ // late write after the lifetime ended.
+ judgePut(t, x.s, HostIncarnationKey(x.rec.Host), HostIncarnation{Account: x.rec.Account, App: x.rec.App, Incarnation: 1, State: SlotRetired})
+ b := x.box(t)
+ before, _ := GetRecord(ctx, x.s, x.rec.Account, x.rec.App, x.rec.Host)
+ if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
+ t.Fatal(err)
+ }
+ // No swap replay and no commit: the record is exactly as it was
+ // (a committed import would have had its gate cleared and the
+ // journal dropped, a ready one its record CAS redone to swapped).
+ after, _ := GetRecord(ctx, x.s, x.rec.Account, x.rec.App, x.rec.Host)
+ if !reflect.DeepEqual(after, before) {
+ t.Fatalf("record changed: %+v -> %+v", before, after)
+ }
+ // The journal's rule for a record that is gone: the rollback.
+ x.rolledBack(t, b, true)
+ })
+ }
+ t.Run("without a judge the committed import commits as today", func(t *testing.T) {
+ x := journaledImport(t, seeds.StateCommitted)
+ judgePut(t, x.s, HostIncarnationKey(x.rec.Host), HostIncarnation{Account: x.rec.Account, App: x.rec.App, Incarnation: 1, State: SlotRetired})
+ if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
+ t.Fatal(err)
+ }
+ if j, _ := x.reg.ImportJournal(x.rec.Host); j != nil {
+ t.Fatalf("journal = %+v, want it dropped by the commit", j)
+ }
+ if got := mustGet(t, x.reg, x.rec.Host); got.ImportGate != "" {
+ t.Fatalf("gate = %q, want it cleared", got.ImportGate)
+ }
+ })
+ t.Run("an unreadable observation leaves the import alone", func(t *testing.T) {
+ x := journaledImport(t, seeds.StateCommitted)
+ x.cfg.Judge = NewLifetimeJudge()
+ judgePut(t, x.s, HostIncarnationKey(x.rec.Host), map[string]any{"incarnation": 0})
+ b := x.box(t)
+ if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
+ t.Fatal(err)
+ }
+ x.untouched(t)
+ if len(b.steps) != 0 {
+ t.Fatalf("a rollback started: %q", b.steps)
+ }
+ })
+ t.Run("the cleanup's re-read is judged", func(t *testing.T) {
+ // A rollback already under way meets the stale record at its
+ // cleanup step: the record is gone to it, so step 8 marks nothing.
+ x := journaledImport(t, seeds.StateAborted)
+ x.cfg.Judge = NewLifetimeJudge()
+ judgePut(t, x.s, HostIncarnationKey(x.rec.Host), HostIncarnation{Account: x.rec.Account, App: x.rec.App, Incarnation: 1, State: SlotRetired})
+ b := x.box(t)
+ before, _ := GetRecord(ctx, x.s, x.rec.Account, x.rec.App, x.rec.Host)
+ if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil {
+ t.Fatal(err)
+ }
+ x.rolledBack(t, b, true)
+ after, _ := GetRecord(ctx, x.s, x.rec.Account, x.rec.App, x.rec.Host)
+ if !reflect.DeepEqual(after, before) {
+ t.Fatalf("a stale record was marked rolled back: %+v -> %+v", before, after)
+ }
+ })
+}
diff --git a/internal/provision/seedrollback.go b/internal/provision/seedrollback.go
index 48fbd7c15..f9d0266b9 100644
--- a/internal/provision/seedrollback.go
+++ b/internal/provision/seedrollback.go
@@ -6,6 +6,7 @@ import (
"fmt"
"log"
"os"
+ "strings"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
@@ -193,7 +194,7 @@ func cleanUpImport(ctx context.Context, s release.Store, reg *registry.Registry,
// The replica is keyed by host, not by seed, so the record is read
// again right before it: the pass's copy may be stale.
- rec, err := journalRecordByKey(ctx, s, j)
+ rec, err := journalRecordByKey(ctx, s, j, cfg.Judge)
if err != nil {
log.Printf("provision: %s: import %s cleanup deferred: its record cannot be read: %v", j.Host, j.SeedID, err)
return
@@ -324,7 +325,16 @@ func replicaIsTheImports(ctx context.Context, s release.Store, j registry.Import
// journalRecordByKey reads the record under the journal's own key: nil when
// it is gone, an error when it cannot be read, does not decode, or is not
// the journal's (another host or key in its body).
-func journalRecordByKey(ctx context.Context, s release.Store, j registry.ImportJournal) (*Record, error) {
+//
+// Under a judge (a claims-off box) the key is observed: a record its slot no
+// longer vouches for is gone, nil with no error, which the journal's own
+// rules already handle; a corrupt body is its decode error as today, and an
+// observation that cannot be judged is its error. A nil judge reads the key
+// exactly as before.
+func journalRecordByKey(ctx context.Context, s release.Store, j registry.ImportJournal, judge *LifetimeJudge) (*Record, error) {
+ if judge != nil {
+ return journalRecordJudged(ctx, s, j, judge)
+ }
raw, _, err := s.Get(ctx, j.RecordKey)
if errors.Is(err, os.ErrNotExist) {
return nil, nil
@@ -341,3 +351,23 @@ func journalRecordByKey(ctx context.Context, s release.Store, j registry.ImportJ
}
return &r, nil
}
+
+// journalRecordJudged is journalRecordByKey under a judge.
+func journalRecordJudged(ctx context.Context, s release.Store, j registry.ImportJournal, judge *LifetimeJudge) (*Record, error) {
+ acct, _, _ := strings.Cut(strings.TrimPrefix(j.RecordKey, "apps/"), "/")
+ acct, app, host, ok := ParseRecordKey(j.RecordKey, map[string]bool{acct: true})
+ if !ok {
+ return nil, fmt.Errorf("the journal's record key %q is not an instance record key", j.RecordKey)
+ }
+ if host != j.Host {
+ return nil, fmt.Errorf("the record under %s is not %s's", j.RecordKey, j.Host)
+ }
+ obs := judge.Observe(ctx, s, acct, app, host)
+ switch {
+ case obs.Verdict == LifetimeUnreadable || obs.Verdict == LifetimeCorrupt:
+ return nil, obs.Err
+ case !obs.Present || obs.Verdict == LifetimeStale:
+ return nil, nil
+ }
+ return obs.Rec, nil
+}