amadan

carlos / platform Public

console: a migration adopts a marker whose operation is gone, the pass selects such records, and an idle pass waits ten minutes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude c0e5eba3ec9af28d295999b2fe7bab1581f23b9c parent 99ffb42
2 files changed, +235 −30
  • internal/console/lifetimes_migrate.go +82 −23
  • internal/console/lifetimes_migrate_test.go +153 −7
diff --git a/internal/console/lifetimes_migrate.go b/internal/console/lifetimes_migrate.go
index d7612ad60..f300f6151 100644
--- a/internal/console/lifetimes_migrate.go
+++ b/internal/console/lifetimes_migrate.go
@@ -30,7 +30,10 @@ import (
// operation's and finished, never stranded.
const (
- lifetimesMigrateInterval = 60 * time.Second
+ // lifetimesMigrateInterval is the pass's cadence while it finds work;
+ // lifetimesMigrateIdleInterval after a pass that found no candidate.
+ lifetimesMigrateInterval = 60 * time.Second
+ lifetimesMigrateIdleInterval = 10 * time.Minute
// migrationPassLimit is how many records one pass migrates at most.
migrationPassLimit = 50
// migrationMarkRounds bounds an unbound migration's marker CAS retries
@@ -62,6 +65,14 @@ func fieldless(rec *provision.Record) bool {
return rec != nil && rec.Incarnation == 0 && rec.IncarnationMigrating == "" && !rec.Retired
}
+// preLifetime reports whether rec has no lifetime yet: no incarnation, not a
+// tombstone, marked or not. The pass selects these; a marker on one it is
+// admitted for belongs to an operation whose row is gone, and the migration
+// adopts it.
+func preLifetime(rec *provision.Record) bool {
+ return rec != nil && rec.Incarnation == 0 && !rec.Retired
+}
+
// migrationAction is what the dispatcher chose.
type migrationAction int
@@ -69,6 +80,7 @@ const (
migrateOwnMarker migrationAction = iota + 1 // the record carries this migration's marker: (2), (3)
migrateDone // incarnation 1 under this migration's slot: committed
migrateMark // eligible: (1)
+ migrateAdopt // another operation's marker and no slot: (1) over it
migrateMoved // bound, eligible at another ETag: fence, superseded, a CAS conflict
migrateWait // unreadable: pending
migrateQuarantine // a corrupt body and no slot: copy it aside, delete it, superseded
@@ -91,10 +103,16 @@ func migrationDispatch(obs provision.Observation, m Migration) migrationAction {
return migrateOwnMarker
case !rec.Retired && rec.IncarnationMigrating == "" && rec.Incarnation == 1 && migrationOwnsSlot(obs.Slot, m):
return migrateDone
- case fieldless(rec) && obs.Slot == nil:
+ case preLifetime(rec) && obs.Slot == nil:
+ // Unmarked, or marked by another operation: admitted under the
+ // hostname lock, no other operation of the hostname is pending, so
+ // that marker's row is gone and nothing will finish it but us.
if m.ExpectETag != "" && obs.ETag != m.ExpectETag {
return migrateMoved
}
+ if rec.IncarnationMigrating != "" {
+ return migrateAdopt
+ }
return migrateMark
}
return migrateSupersede
@@ -167,7 +185,7 @@ func (s *Server) migrationStepsLocked(ctx context.Context, m Migration) (*provis
return obs.Rec, obs.Raw, obs.ETag, lifetimeCommitted, nil
case migrateOwnMarker:
return s.migrationFinishLocked(ctx, m, obs)
- case migrateMark:
+ case migrateMark, migrateAdopt:
if marks == migrationMarkRounds {
logf("the record kept changing under %d marker writes (pending)", marks)
return migrationPending()
@@ -216,6 +234,9 @@ func (s *Server) migrationStepsLocked(ctx context.Context, m Migration) (*provis
if !obs.Present {
return s.settleMigration(m, lifetimeSuperseded, nil)
}
+ if obs.Verdict == provision.LifetimeCorrupt {
+ s.lifetimeLogHourly("migration-corrupt "+m.OpID, "console: lifetimes: migration %s: the body at %s is corrupt under a slot and needs operator attention", m.OpID, key)
+ }
_, redispatch, err := s.terminateFenced(ctx, m.Account, m.App, m.Host, fenceOwner{MarkerID: m.OpID}, func() error {
return s.store.SetMigrationState(m.OpID, lifetimeSuperseded)
})
@@ -232,15 +253,23 @@ func (s *Server) migrationStepsLocked(ctx context.Context, m Migration) (*provis
return migrationPending()
}
-// migrationMarkLocked is step (1): the observed bytes with the marker set,
-// by CAS from the observed ETag, the record sequence on. It returns the
-// marked ETag.
+// migrationMarkLocked is step (1): the observed bytes with this migration's
+// marker set, by CAS from the observed ETag, the record sequence on. Over
+// another operation's marker (an adoption) the write carries that marker's
+// id, the one writer the marker admits. It returns the marked ETag.
func (s *Server) migrationMarkLocked(ctx context.Context, m Migration, obs provision.Observation) (string, error) {
+ writer, err := provision.RecordMarker(obs.Raw)
+ if err != nil {
+ return "", err
+ }
+ if writer == "" {
+ writer = m.OpID
+ }
marked, err := provision.MarkMigrating(obs.Raw, m.OpID)
if err != nil {
return "", err
}
- wctx := provision.WithMigration(provision.WithRecordSequence(ctx), m.OpID)
+ wctx := provision.WithMigration(provision.WithRecordSequence(ctx), writer)
etag, _, err := provision.WriteRecord(wctx, s.cfg.Releases, provision.RecordKey(m.Account, m.App, m.Host), obs.Raw, marked, obs.ETag)
return etag, err
}
@@ -368,34 +397,64 @@ func (s *Server) StartIncarnationMigrationPass(ctx context.Context) {
if !s.lifetimesGoverned() {
return
}
- lifetimesWorker(ctx, lifetimesMigrateInterval, func(ctx context.Context) { s.incarnationMigrationIteration(ctx) })
+ // Task 0's worker lifecycle, paced: it never returns because of the
+ // gate's state, and each iteration admits for itself.
+ for {
+ ran, candidates := s.incarnationMigrationPass(ctx)
+ t := time.NewTimer(migrationPassInterval(ran, candidates))
+ select {
+ case <-ctx.Done():
+ t.Stop()
+ return
+ case <-t.C:
+ }
+ }
+}
+
+// migrationPassInterval is the wait before the next pass: ten minutes after
+// a pass that ran and found no candidate, 60 seconds otherwise.
+func migrationPassInterval(ran bool, candidates int) time.Duration {
+ if ran && candidates == 0 {
+ return lifetimesMigrateIdleInterval
+ }
+ return lifetimesMigrateInterval
}
// incarnationMigrationIteration is one pass as a unit: skipped unless the
// flag is set and the unit is admitted on. It reports whether the pass ran.
func (s *Server) incarnationMigrationIteration(ctx context.Context) bool {
+ ran, _ := s.incarnationMigrationPass(ctx)
+ return ran
+}
+
+// incarnationMigrationPass is incarnationMigrationIteration with the number
+// of candidates the pass found.
+func (s *Server) incarnationMigrationPass(ctx context.Context) (ran bool, candidates int) {
if !s.cfg.LifetimesMigrate {
- return false
+ return false, 0
}
- return s.lifetimesIteration(ctx, "incarnation migration", true, true, func(ctx context.Context) {
- if err := s.migrateFieldlessRecords(ctx); err != nil {
+ ran = s.lifetimesIteration(ctx, "incarnation migration", true, true, func(ctx context.Context) {
+ var err error
+ if candidates, err = s.migratePreLifetimeRecords(ctx); err != nil {
log.Printf("console: incarnation migration pass: %v", err)
}
})
+ return ran, candidates
}
-// migrateFieldlessRecords is one pass: at most migrationPassLimit fieldless
-// records, oldest first, each under s.lockApp then the hostname lock,
-// admitted, and migrated unbound only while it is still fieldless and
-// unmarked with no slot for its hostname.
-func (s *Server) migrateFieldlessRecords(ctx context.Context) error {
+// migratePreLifetimeRecords is one pass: at most migrationPassLimit records
+// with no lifetime yet (fieldless, or carrying a marker whose operation is
+// gone), oldest first, each under s.lockApp then the hostname lock,
+// admitted, and migrated unbound only while it still has no lifetime and no
+// slot for its hostname. It returns how many candidates it found.
+func (s *Server) migratePreLifetimeRecords(ctx context.Context) (int, error) {
recs, _, err := provision.ListAllRecordsTolerantJudged(ctx, s.cfg.Releases, s.judge)
if err != nil {
- return fmt.Errorf("list the records: %w", err)
+ return 0, fmt.Errorf("list the records: %w", err)
}
var candidates []provision.Record
for i := range recs {
- if fieldless(&recs[i]) {
+ if preLifetime(&recs[i]) {
candidates = append(candidates, recs[i])
}
}
@@ -406,9 +465,9 @@ func (s *Server) migrateFieldlessRecords(ctx context.Context) error {
}
return provision.RecordKey(a.Account, a.App, a.Host) < provision.RecordKey(b.Account, b.App, b.Host)
})
- migrated := 0
+ attempted := 0
for _, rec := range candidates {
- if migrated == migrationPassLimit {
+ if attempted == migrationPassLimit {
break
}
_ = s.withAppHost(rec.Account, rec.App, rec.Host, func() error {
@@ -416,10 +475,10 @@ func (s *Server) migrateFieldlessRecords(ctx context.Context) error {
return nil
}
obs, err := s.classifyLocked(ctx, rec.Account, rec.App, rec.Host)
- if err != nil || !obs.Present || !fieldless(obs.Rec) || obs.Slot != nil {
+ if err != nil || !obs.Present || !preLifetime(obs.Rec) || obs.Slot != nil {
return nil
}
- migrated++
+ attempted++
_, _, _, state, err := s.migrateHostLifetimeLocked(ctx, rec.Account, rec.App, rec.Host, "")
if err != nil || state != lifetimeCommitted {
log.Printf("console: incarnation migration pass: %s/%s/%s: %s %v", rec.Account, rec.App, rec.Host, state, err)
@@ -427,5 +486,5 @@ func (s *Server) migrateFieldlessRecords(ctx context.Context) error {
return nil
})
}
- return nil
+ return len(candidates), nil
}
diff --git a/internal/console/lifetimes_migrate_test.go b/internal/console/lifetimes_migrate_test.go
index 98e70b7eb..3c0c7d1c0 100644
--- a/internal/console/lifetimes_migrate_test.go
+++ b/internal/console/lifetimes_migrate_test.go
@@ -700,7 +700,7 @@ func TestMigrationRecoveryBranches(t *testing.T) {
t.Fatalf("slot = %+v, want %+v untouched", slot, want)
}
})
- t.Run("another operation's marker", func(t *testing.T) {
+ t.Run("another operation's marker, no slot: adopted", func(t *testing.T) {
l := newLtOn(t)
e := legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
beginMig(t, l, migOp, "")
@@ -709,14 +709,58 @@ func TestMigrationRecoveryBranches(t *testing.T) {
if _, err := l.fx.rel.Put(ctx, key, other, e); err != nil {
t.Fatal(err)
}
+ // The first recovery adopts the marker, then stops at the slot.
+ srv := l.restart(t)
+ l.bucket.setHook(failNth("put", slotKey, 1, false))
+ recoverOn(t, srv)
+ l.bucket.setHook(nil)
+ row := oneMig(t, l.fx.st, ltOne, lifetimePending)
+ raw, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if got := decodeRec(t, raw); got.IncarnationMigrating != migOp || got.RecordSeq != 1 || row.Marked != etag {
+ t.Fatalf("record = %s, row %+v: want this migration's marker over the other, record_seq 1, the marked ETag persisted", raw, row)
+ }
+ recoverAgain(t, l)
+ oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, ltOne, migOp)
+ })
+ t.Run("another operation's marker under a foreign slot", func(t *testing.T) {
+ l := newLtOn(t)
+ e := legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ beginMig(t, l, migOp, "")
+ a, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ other, _ := provision.MarkMigrating(a, "other")
+ if _, err := l.fx.rel.Put(ctx, key, other, e); err != nil {
+ t.Fatal(err)
+ }
+ want := provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: "bab", App: "blog", Intent: "create other"}
+ writeSlot(t, l.fx.rel, ltOne, want)
recoverAgain(t, l)
oneMig(t, l.fx.st, ltOne, lifetimeSuperseded)
raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
if got := decodeRec(t, raw); got.IncarnationMigrating != "other" || got.RecordSeq != 1 {
- t.Fatalf("record = %s, want the other marker kept and the record fenced", raw)
+ t.Fatalf("record = %s, want the other marker left and the record fenced", raw)
}
- if slot, _ := slotOf(t, l.fx.rel, ltOne); slot != nil {
- t.Fatalf("slot = %+v, want none", slot)
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot == nil || *slot != want {
+ t.Fatalf("slot = %+v, want %+v untouched", slot, want)
+ }
+ })
+ t.Run("own marker under its slot bound to another body", func(t *testing.T) {
+ l := newLtOn(t)
+ e := legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ beginMig(t, l, migOp, "")
+ a, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ marked, _ := provision.MarkMigrating(a, migOp)
+ me, err := l.fx.rel.Put(ctx, key, marked, e)
+ if err != nil {
+ t.Fatal(err)
+ }
+ writeSlot(t, l.fx.rel, ltOne, provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: "bab", App: ltApp,
+ Intent: "migration " + migOp, RecordETag: "another-body"})
+ recoverAgain(t, l)
+ row := oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, ltOne, migOp)
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot.RecordETag != me || row.Marked != me {
+ t.Fatalf("slot bound to %q, row %+v: want the slot rebound to the marked record %q", slot.RecordETag, row, me)
}
})
t.Run("a slot not of this migration", func(t *testing.T) {
@@ -837,8 +881,17 @@ func TestMigrationSkipsAFieldlessRecordBesideAnUnrelatedSlot(t *testing.T) {
// ---- the judge agrees -----------------------------------------------------------
func TestJudgeKeepsAMarkedRecordUnderItsMigrationSlot(t *testing.T) {
+ for _, start := range []string{"a fieldless record", "a marker whose operation is gone"} {
+ t.Run(start, func(t *testing.T) { testJudgeAlongTheMigration(t, start != "a fieldless record") })
+ }
+}
+
+// testJudgeAlongTheMigration runs one migration with its slot write and its
+// final write held and asks the judge at every step; adopt starts from a
+// record marked by an operation that has no row.
+func testJudgeAlongTheMigration(t *testing.T, adopt bool) {
l := newLtOn(t)
- legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ e := legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
key, slotKey := recKey(ltApp, ltOne), provision.HostIncarnationKey(ltOne)
judge := provision.NewLifetimeJudge()
observe := func(want provision.LifetimeVerdict, what string) provision.Observation {
@@ -849,6 +902,16 @@ func TestJudgeKeepsAMarkedRecordUnderItsMigrationSlot(t *testing.T) {
}
return obs
}
+ if adopt {
+ a, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ lost, _ := provision.MarkMigrating(a, "lost")
+ if _, err := l.fx.rel.Put(context.Background(), key, lost, e); err != nil {
+ t.Fatal(err)
+ }
+ if obs := observe(provision.LifetimeValid, "marked by a lost operation, no slot"); obs.Rec.IncarnationMigrating != "lost" {
+ t.Fatalf("observation = %+v, want the lost marker", obs)
+ }
+ }
sgate, shook := pauseNth(t, "put", slotKey, 1)
fgate, fhook := pauseNth(t, "put", key, 2)
l.bucket.setHook(chain(shook, fhook))
@@ -856,8 +919,8 @@ func TestJudgeKeepsAMarkedRecordUnderItsMigrationSlot(t *testing.T) {
waitFor(t, sgate.entered, "the slot write")
obs := observe(provision.LifetimeValid, "marked, no slot")
- if obs.Rec.IncarnationMigrating == "" || obs.Slot != nil {
- t.Fatalf("observation = %+v, want the marked record and no slot", obs)
+ if obs.Rec.IncarnationMigrating == "" || obs.Rec.IncarnationMigrating == "lost" || obs.Slot != nil {
+ t.Fatalf("observation = %+v, want this migration's marked record and no slot", obs)
}
op := obs.Rec.IncarnationMigrating
@@ -944,3 +1007,86 @@ func TestMigrationWaitsForAnotherPendingOperation(t *testing.T) {
oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
requireMigrated(t, l.fx.rel, ltOne, migOp)
}
+
+func TestMigrationPassAdoptsAMarkerWhoseOperationIsGone(t *testing.T) {
+ l := newLtOn(t)
+ l.fx.srv.cfg.LifetimesMigrate = true
+ e := legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ a, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ lost, _ := provision.MarkMigrating(a, "lost")
+ if _, err := l.fx.rel.Put(context.Background(), recKey(ltApp, ltOne), lost, e); err != nil {
+ t.Fatal(err)
+ }
+ // No row for "lost" and no slot: an ordinary writer can only stall on it.
+ t.Cleanup(provision.SetRecordMigrationWaitForTest(30*time.Millisecond, 10*time.Millisecond))
+ if err := provision.SetRecordRegion(onCtx(t, l.fx.srv), l.bucket, "bab", ltApp, ltOne, "uk"); !errors.Is(err, provision.ErrRecordMigrating) {
+ t.Fatalf("a writer over the stranded marker = %v, want ErrRecordMigrating", err)
+ }
+ ran, candidates := l.fx.srv.incarnationMigrationPass(context.Background())
+ if !ran || candidates != 1 {
+ t.Fatalf("the pass ran %v with %d candidates, want it to select the marked record", ran, candidates)
+ }
+ row := oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, ltOne, row.OpID)
+ if err := provision.SetRecordRegion(onCtx(t, l.fx.srv), l.bucket, "bab", ltApp, ltOne, "uk"); err != nil {
+ t.Fatalf("a writer after the adoption = %v", err)
+ }
+}
+
+func TestMigrationPassSlowsDownWhenIdle(t *testing.T) {
+ for _, c := range []struct {
+ ran bool
+ candidates int
+ want time.Duration
+ }{
+ {true, 0, 10 * time.Minute},
+ {true, 1, time.Minute},
+ {true, 120, time.Minute},
+ {false, 0, time.Minute},
+ } {
+ if got := migrationPassInterval(c.ran, c.candidates); got != c.want {
+ t.Fatalf("migrationPassInterval(%v, %d) = %v, want %v", c.ran, c.candidates, got, c.want)
+ }
+ }
+ // The pass reports what it found: one candidate, then none.
+ l := newLtOn(t)
+ l.fx.srv.cfg.LifetimesMigrate = true
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ if ran, n := l.fx.srv.incarnationMigrationPass(context.Background()); !ran || n != 1 {
+ t.Fatalf("first pass = %v %d, want one candidate", ran, n)
+ }
+ if ran, n := l.fx.srv.incarnationMigrationPass(context.Background()); !ran || n != 0 || migrationPassInterval(ran, n) != lifetimesMigrateIdleInterval {
+ t.Fatalf("second pass = %v %d, want none and the idle cadence", ran, n)
+ }
+}
+
+func TestMigrationFenceReadsOneMarkerAcrossSpellings(t *testing.T) {
+ l := newLtOn(t)
+ ctx := context.Background()
+ rec := provision.Record{Schema: 1, Host: ltOne, Slug: "one", App: ltApp, Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC()}
+ body, _ := json.Marshal(rec)
+ // Two spellings with different values; DecodeRecord reads the last.
+ body = bytes.Replace(body, []byte(`{`), []byte(`{"Incarnation_Migrating":"first","incarnation_migrating":"second",`), 1)
+ if _, err := l.fx.rel.Put(ctx, recKey(ltApp, ltOne), body, ""); err != nil {
+ t.Fatal(err)
+ }
+ if got, err := provision.RecordMarker(body); err != nil || got != "second" || decodeRec(t, body).IncarnationMigrating != "second" {
+ t.Fatalf("RecordMarker = %q %v, want the marker DecodeRecord reads", got, err)
+ }
+ var res fenceResult
+ _ = l.fx.srv.withHostname(ltOne, func() error {
+ var err error
+ res, _, err = l.fx.srv.fenceRecord(ctx, "bab", ltApp, ltOne, fenceOwner{Tombstone: true})
+ if err != nil {
+ t.Fatalf("the fence = %v, want it written", err)
+ }
+ return nil
+ })
+ if res != fenced {
+ t.Fatalf("the fence = %v, want fenced", res)
+ }
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if got := decodeRec(t, raw); got.RecordSeq != 1 || got.IncarnationMigrating != "second" {
+ t.Fatalf("record = %s, want the same body fenced", raw)
+ }
+}