amadan

carlos / platform Public

lifetimes: lazy migration marks the record by CAS, fenced in the write primitive, and never strands a marker it issued

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 9991e16f87372fdd68f535e17898eb9f15f9ffac parent 2a75caa
18 files changed, +1817 −53
  • cmd/console/main.go +11 −0
  • docs/superpowers/notes/census-proposal-self-serve-moves-1.md +2 −0
  • internal/console/claim_invariant_fixtures_test.go +14 −13
  • internal/console/claim_invariant_probes_worker_test.go +5 −4
  • internal/console/claim_workers.go +21 −20
  • internal/console/lifetimes_create.go +8 −1
  • internal/console/lifetimes_migrate.go +431 −0
  • internal/console/lifetimes_migrate_test.go +946 −0
  • internal/console/lifetimes_recovery.go +14 −3
  • internal/console/server.go +6 −0
  • internal/console/testdata/claim-routes.txt +1 −0
  • internal/provision/lifetime_migration.go +125 −0
  • internal/provision/plan14_root_reads_test.go +3 −2
  • internal/provision/provision.go +19 −3
  • internal/provision/record_sequence.go +6 −0
  • internal/provision/record_sequence_test.go +163 −0
  • internal/provision/self_serve_moves1_census_test.go +36 −2
  • internal/provision/testdata/plan14-root-reads-self-serve-moves-1.tsv +6 −5
diff --git a/cmd/console/main.go b/cmd/console/main.go
index ef3141377..6de4364ba 100644
--- a/cmd/console/main.go
+++ b/cmd/console/main.go
@@ -51,6 +51,10 @@
// CARLOS_TEMPORARY_APPS=1 turns on temporary apps (carlos deploy
// --temporary); off by default, see the temporary static sites rollout.
//
+// CARLOS_LIFETIMES_MIGRATE=1 turns on the incarnation migration pass, which
+// moves fieldless instance records into slot-governed lifetimes; off by
+// default until every box of the deployment runs this release.
+//
// Tenant email (task 17) reads CARLOS_MAIL_OPERATORS (comma-separated
// addresses; narrows the fleet-operator bit further for the pause/cap
// door only — see mailOperators' doc comment), CARLOS_MAIL_REGION,
@@ -410,6 +414,7 @@ func serve() {
StoreOperators: storeOperators(),
PlanCatalog: catalog,
TemporaryApps: os.Getenv("CARLOS_TEMPORARY_APPS") == "1",
+ LifetimesMigrate: os.Getenv("CARLOS_LIFETIMES_MIGRATE") == "1",
// CARLOS_ACME_DELEGATION_ZONE — the same variable the edge reads
// (internal/edge/wildcard.go's newWildcardFromEnv) for the same
// delegated-ACME story; NewServer normalizes it (lowercase, trim).
@@ -545,6 +550,12 @@ func serve() {
// a crash or a failed write left pending. It does work only once
// lifetimes are on, and is off on a host-claim console.
go srv.RunClaimWorker(context.Background(), "StartLifetimeRecoveryPass")
+ // The incarnation migration pass (internal/console/lifetimes_migrate.go):
+ // every 60 seconds, moves up to 50 fieldless instance records into
+ // slot-governed lifetimes, oldest first. Off unless
+ // CARLOS_LIFETIMES_MIGRATE=1, it does work only once lifetimes are on,
+ // and is off on a host-claim console.
+ go srv.RunClaimWorker(context.Background(), "StartIncarnationMigrationPass")
// The relay's writer lease and its converging reconcile
// (internal/console/relayops.go). Same placement rule as every pass
// above, and it returns immediately when CARLOS_RELAY_REGIONS is empty —
diff --git a/docs/superpowers/notes/census-proposal-self-serve-moves-1.md b/docs/superpowers/notes/census-proposal-self-serve-moves-1.md
index 8519cf5e0..e829c08aa 100644
--- a/docs/superpowers/notes/census-proposal-self-serve-moves-1.md
+++ b/docs/superpowers/notes/census-proposal-self-serve-moves-1.md
@@ -15,3 +15,5 @@ Task 0 adds the lifetimes cutover: the operator door `POST /api/cli/system/lifet
Task 3 adds the claims-off create protocol and its recovery (`internal/console/lifetimes.go`, `lifetimes_create.go`, `lifetimes_recovery.go`) and routes the create, seed, import and first-route attach doors through it when the request was admitted with lifetimes on. The record-write census reported two changed sites, two stale entries and five unlisted sites. `Server.handleAPIInstanceCreate` and `Server.attachDomainAs` remain record-writers through `CreateRecord` on the legacy path; admitted on, they leave for `createRecordWithLifetime` and `attachFirstRouteWithLifetime`. `Server.beginFrom` and `Server.beginImport` no longer write directly: both call `Server.createSeedRecord`, a record-writer that is either the lifetime create or today's create-only `CreateRecord`. `Server.allocRecordLocked` is a record-writer: one create-only `WriteRecord` of the allocation's persisted body from `{}` with the record sequence on, so the record lands with `record_seq: 1`. `Server.fenceRecord` is a record-writer: it rewrites the exact bytes it observed against their own ETag with the sequence on, and writes no other body. `Server.allocCleanLocked` and `Server.deleteUnderRetiredSlot` are key-sites: each removes a stale body only with `DeleteIfMatch` on an ETag it holds, never a plain delete, and neither Puts. The slot is written only through `provision.WriteHostIncarnation`, which is already nonrecord. The root-read census reported two changed accepted-legacy units, one unclassified unit and two envelopes. `serve` is re-pinned in Task 0's rows, since it now also starts the lifetime recovery pass. `Server.hostHolder` keeps its root-route disposition; its instance-record holder read is now judged under a request admitted on (a nil judge is `GetRecord`). `newLifetimeOpID` is classified like `mintLifetime`: a crypto/rand read for an operation id that grants no root authority. The `cmd/console` and `internal/console` envelopes are re-pinned. Three new fingerprints move the seal. No class or task is added, and the registry fields stand. Everything here remains `PROPOSED-self-serve-moves-1`, awaiting Paul's census ruling.
Task 3 fix round 1 changes only door answers. `Server.handleAPIDomainAttach` stays a read-only key-site, but now answers attach refusals through `writeDoorRefusalJSON`. The `internal/console` envelope is re-pinned. No write site, root read, class or seal changes. This remains `PROPOSED-self-serve-moves-1`.
+
+Task 5 adds the lazy migration: `internal/provision/lifetime_migration.go` (`ErrRecordMigrating`, `WithMigration`, the marker and final bodies, the updaters' wait) and `internal/console/lifetimes_migrate.go` (the protocol, its dispatcher, recovery and the pass behind `CARLOS_LIFETIMES_MIGRATE=1`). The record-write census reported three changed sites and five unlisted ones. `WriteRecord` stays the one record-primitive: it now refuses a body read with another migration's marker, a refusal only, and a create-only write is unaffected. `updateRecordSeeding` stays a record-writer and reads a marked record again every second for up to 30 seconds. `Server.fenceRecord` stays a record-writer; over a marker of a migration that admission proves is not pending, it rewrites the exact bytes under that marker's id. `Server.migrationMarkLocked` (step 1) and `Server.migrationFinishLocked` (step 3) are record-writers, each one `WriteRecord` by CAS under its own marker id with the sequence on. `Server.migrationStepsLocked` and `Server.migrateFieldlessRecords` are key-sites; the only bucket write in either is a `DeleteIfMatch` of a corrupt body on its observed ETag, after the copy. `Server.quarantineCopy` is nonrecord: one create-only Put under `control/quarantine/`. The root-read overlay re-pins `serve`, which also starts the migration pass, and the `internal/provision`, `internal/console` and `cmd/console` envelopes. The changed `serve` fingerprint moves the seal. No root read, class or registry field is added. This remains `PROPOSED-self-serve-moves-1`.
diff --git a/internal/console/claim_invariant_fixtures_test.go b/internal/console/claim_invariant_fixtures_test.go
index 9113a7443..99e4abf25 100644
--- a/internal/console/claim_invariant_fixtures_test.go
+++ b/internal/console/claim_invariant_fixtures_test.go
@@ -1852,19 +1852,20 @@ var claimFixtures = map[string]claimFixture{
}},
// ---- workers ----
- "worker LogPlacementDiscrepancies": claimRead("logs region drift between the registry and instance records; writes nothing"),
- "worker PublishMovesConsole": claimNonApp("the moves console record (control/moves-console.json): the platform's moves protocol and key fingerprint; no app, and nothing with the moves switch off"),
- "worker ReportApexSquats": claimRead("logs apex squats found in the registry; writes nothing"),
- "worker StartRunSweep": claimRead("off on a host-claim console, where every run route refuses: it logs that and writes nothing"),
- "worker StartSeedsPass": claimRead("off on a host-claim console, where every seed door and finishSeed refuse: it logs that and writes nothing"),
- "worker StartLifetimesCutoverWatch": claimRead("off on a host-claim console, where lifetimes are not for it: it logs that and reads nothing"),
- "worker StartLifetimeRecoveryPass": claimRead("off on a host-claim console, where lifetimes are not for it: it logs that and reads nothing"),
- "worker StartAlertWatcher": claimNonApp("box alert state under alerts/boxes and operator mail"),
- "worker StartEconomicsPass": claimNonApp("daily usage census, statements and bills: deployment and account economics"),
- "worker StartGeoipJobs": claimNonApp("GeoIP orphan sweep, D1 expiry and feed status under control/geoip/: the platform database, no app"),
- "worker StartHostReports": claimNonApp("this console's census report (status/consoles) and roster activation release"),
- "worker StartInfraPass": claimNonApp("the deployment's infrastructure inventory snapshot"),
- "worker StartPlanLimitsPass": claimNonApp("account plan rows and their limits records under accounts/<sqid>/limits.json; names no app"),
+ "worker LogPlacementDiscrepancies": claimRead("logs region drift between the registry and instance records; writes nothing"),
+ "worker PublishMovesConsole": claimNonApp("the moves console record (control/moves-console.json): the platform's moves protocol and key fingerprint; no app, and nothing with the moves switch off"),
+ "worker ReportApexSquats": claimRead("logs apex squats found in the registry; writes nothing"),
+ "worker StartRunSweep": claimRead("off on a host-claim console, where every run route refuses: it logs that and writes nothing"),
+ "worker StartSeedsPass": claimRead("off on a host-claim console, where every seed door and finishSeed refuse: it logs that and writes nothing"),
+ "worker StartLifetimesCutoverWatch": claimRead("off on a host-claim console, where lifetimes are not for it: it logs that and reads nothing"),
+ "worker StartLifetimeRecoveryPass": claimRead("off on a host-claim console, where lifetimes are not for it: it logs that and reads nothing"),
+ "worker StartIncarnationMigrationPass": claimRead("off on a host-claim console, where lifetimes are not for it: it logs that and reads nothing"),
+ "worker StartAlertWatcher": claimNonApp("box alert state under alerts/boxes and operator mail"),
+ "worker StartEconomicsPass": claimNonApp("daily usage census, statements and bills: deployment and account economics"),
+ "worker StartGeoipJobs": claimNonApp("GeoIP orphan sweep, D1 expiry and feed status under control/geoip/: the platform database, no app"),
+ "worker StartHostReports": claimNonApp("this console's census report (status/consoles) and roster activation release"),
+ "worker StartInfraPass": claimNonApp("the deployment's infrastructure inventory snapshot"),
+ "worker StartPlanLimitsPass": claimNonApp("account plan rows and their limits records under accounts/<sqid>/limits.json; names no app"),
"worker StartClaimRecoveryPass": {door: &claimDoor{
apps: claimRefs("bab/titogo"),
setup: func(t *testing.T, cx *claimCase) { cx.crashedRestore(t) },
diff --git a/internal/console/claim_invariant_probes_worker_test.go b/internal/console/claim_invariant_probes_worker_test.go
index 97d974162..19e51823d 100644
--- a/internal/console/claim_invariant_probes_worker_test.go
+++ b/internal/console/claim_invariant_probes_worker_test.go
@@ -77,10 +77,11 @@ var claimWorkerProbes = map[string]claimProbe{
cx.addRoute(t, registry.Route{Host: "titogo.bab.oncarlos.com", Kind: "instance", Status: "ready",
App: "kass", Account: "bab", Addr: "127.0.0.1:9301"})
}, "apex squat", "titogo.bab.oncarlos.com"),
- "worker StartRunSweep": claimLogWorkerProbe("StartRunSweep", nil, "run sweep", "host-claim console"),
- "worker StartSeedsPass": claimLogWorkerProbe("StartSeedsPass", nil, "seeds pass", "host-claim console"),
- "worker StartLifetimesCutoverWatch": claimLogWorkerProbe("StartLifetimesCutoverWatch", nil, "lifetimes cutover watch", "host-claim console"),
- "worker StartLifetimeRecoveryPass": claimLogWorkerProbe("StartLifetimeRecoveryPass", nil, "lifetime recovery pass", "host-claim console"),
+ "worker StartRunSweep": claimLogWorkerProbe("StartRunSweep", nil, "run sweep", "host-claim console"),
+ "worker StartSeedsPass": claimLogWorkerProbe("StartSeedsPass", nil, "seeds pass", "host-claim console"),
+ "worker StartLifetimesCutoverWatch": claimLogWorkerProbe("StartLifetimesCutoverWatch", nil, "lifetimes cutover watch", "host-claim console"),
+ "worker StartLifetimeRecoveryPass": claimLogWorkerProbe("StartLifetimeRecoveryPass", nil, "lifetime recovery pass", "host-claim console"),
+ "worker StartIncarnationMigrationPass": claimLogWorkerProbe("StartIncarnationMigrationPass", nil, "incarnation migration pass", "host-claim console"),
"worker LogPlacementDiscrepancies": claimLogWorkerProbe("LogPlacementDiscrepancies", claimRegionalApp,
msgPlacementDiscrepancy("drift.go.tito.io", "blog", "uk")),
diff --git a/internal/console/claim_workers.go b/internal/console/claim_workers.go
index 84cfef248..5eb6eb8b3 100644
--- a/internal/console/claim_workers.go
+++ b/internal/console/claim_workers.go
@@ -5,26 +5,27 @@ import "context"
func (s *Server) registerClaimWorkers() {
s.claimWorkers = map[string]func(context.Context){}
for name, h := range map[string]func(context.Context){
- "StartHostReports": s.StartHostReports,
- "StartAlertWatcher": s.StartAlertWatcher,
- "StartGeoipJobs": s.StartGeoipJobs,
- "StartClaimRecoveryPass": s.StartClaimRecoveryPass,
- "StartEconomicsPass": s.StartEconomicsPass,
- "StartInfraPass": s.StartInfraPass,
- "StartScanPass": s.StartScanPass,
- "StartScanDrain": s.StartScanDrain,
- "StartExpiryPass": s.StartExpiryPass,
- "StartPlanLimitsPass": s.StartPlanLimitsPass,
- "StartRunSweep": s.StartRunSweep,
- "StartSeedsPass": s.StartSeedsPass,
- "StartLifetimesCutoverWatch": s.StartLifetimesCutoverWatch,
- "StartLifetimeRecoveryPass": s.StartLifetimeRecoveryPass,
- "StartRelay": s.StartRelay,
- "StartMailEventPush": s.StartMailEventPush,
- "StartDoorKeyResign": s.StartDoorKeyResign,
- "ReportApexSquats": s.ReportApexSquats,
- "LogPlacementDiscrepancies": s.LogPlacementDiscrepancies,
- "PublishMovesConsole": s.publishMovesConsoleWorker,
+ "StartHostReports": s.StartHostReports,
+ "StartAlertWatcher": s.StartAlertWatcher,
+ "StartGeoipJobs": s.StartGeoipJobs,
+ "StartClaimRecoveryPass": s.StartClaimRecoveryPass,
+ "StartEconomicsPass": s.StartEconomicsPass,
+ "StartInfraPass": s.StartInfraPass,
+ "StartScanPass": s.StartScanPass,
+ "StartScanDrain": s.StartScanDrain,
+ "StartExpiryPass": s.StartExpiryPass,
+ "StartPlanLimitsPass": s.StartPlanLimitsPass,
+ "StartRunSweep": s.StartRunSweep,
+ "StartSeedsPass": s.StartSeedsPass,
+ "StartLifetimesCutoverWatch": s.StartLifetimesCutoverWatch,
+ "StartLifetimeRecoveryPass": s.StartLifetimeRecoveryPass,
+ "StartIncarnationMigrationPass": s.StartIncarnationMigrationPass,
+ "StartRelay": s.StartRelay,
+ "StartMailEventPush": s.StartMailEventPush,
+ "StartDoorKeyResign": s.StartDoorKeyResign,
+ "ReportApexSquats": s.ReportApexSquats,
+ "LogPlacementDiscrepancies": s.LogPlacementDiscrepancies,
+ "PublishMovesConsole": s.publishMovesConsoleWorker,
} {
e := claimEntry{Kind: "worker", Pattern: name, Handler: claimHandlerName(h)}
s.claimEntries.add(e)
diff --git a/internal/console/lifetimes_create.go b/internal/console/lifetimes_create.go
index 9b7a4d367..7cc418b97 100644
--- a/internal/console/lifetimes_create.go
+++ b/internal/console/lifetimes_create.go
@@ -506,7 +506,14 @@ func (s *Server) fenceRecord(ctx context.Context, acct, app, host string, op fen
if lifetimesFenceDisabled {
return fenced, obs, nil
}
- etag, _, err := provision.WriteRecord(ctx, s.cfg.Releases, key, obs.Raw, obs.Raw, obs.ETag)
+ // A marker of another migration of this hostname: the caller was
+ // admitted under the hostname lock, so that migration is not pending
+ // and the fence may rewrite its exact bytes (Task 5).
+ wctx := ctx
+ if obs.Rec != nil && obs.Rec.IncarnationMigrating != "" {
+ wctx = provision.WithMigration(ctx, obs.Rec.IncarnationMigrating)
+ }
+ etag, _, err := provision.WriteRecord(wctx, s.cfg.Releases, key, obs.Raw, obs.Raw, obs.ETag)
if err == nil {
obs.ETag = etag
return fenced, obs, nil
diff --git a/internal/console/lifetimes_migrate.go b/internal/console/lifetimes_migrate.go
new file mode 100644
index 000000000..d7612ad60
--- /dev/null
+++ b/internal/console/lifetimes_migrate.go
@@ -0,0 +1,431 @@
+package console
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "sort"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The lazy migration (self-serve moves plan 1, Task 5): a fieldless record
+// whose hostname has no slot moves into a slot-governed lifetime at
+// incarnation 1 through a persisted intent (record_migrations):
+//
+// (1) the record marked by CAS: incarnation_migrating = the operation id;
+// from here WriteRecord refuses every writer but this migration;
+// (2) the slot {1, live, this key, "migration " + op} bound to the marked
+// ETag, create-only (If-Match over this migration's own slot);
+// (3) the record rewritten at incarnation 1 with the marker removed.
+//
+// The judge vouches for the record at every step: marked with no slot,
+// marked under its own migration slot bound to its ETag, and at incarnation 1
+// under that slot. migrationDispatch is the only way into the steps: at the
+// first entry, after every CAS conflict and at every recovery, so a marker
+// that lands late (a crashed console's PUT) is always recognised as this
+// operation's and finished, never stranded.
+
+const (
+ lifetimesMigrateInterval = 60 * time.Second
+ // migrationPassLimit is how many records one pass migrates at most.
+ migrationPassLimit = 50
+ // migrationMarkRounds bounds an unbound migration's marker CAS retries
+ // within one entry; past it the row stays pending.
+ migrationMarkRounds = 5
+ // migrationRounds bounds the dispatcher loop of one entry.
+ migrationRounds = 12
+)
+
+// migrationIntent is the slot intent of a migration.
+func migrationIntent(m Migration) string { return "migration " + m.OpID }
+
+// migrationSlot is the slot a migration writes, bound to the marked ETag.
+func migrationSlot(m Migration, markedETag string) provision.HostIncarnation {
+ return provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: m.Account, App: m.App,
+ Intent: migrationIntent(m), RecordETag: markedETag}
+}
+
+// migrationOwnsSlot reports whether slot is this migration's, whatever body
+// it is bound to.
+func migrationOwnsSlot(slot *provision.HostIncarnation, m Migration) bool {
+ return slot != nil && slot.Incarnation == 1 && slot.State == provision.SlotLive &&
+ slotOwnedBy(slot, m.Account, m.App) && slot.Intent == migrationIntent(m)
+}
+
+// fieldless reports whether rec is a legacy record: no incarnation, no
+// marker, not a tombstone.
+func fieldless(rec *provision.Record) bool {
+ return rec != nil && rec.Incarnation == 0 && rec.IncarnationMigrating == "" && !rec.Retired
+}
+
+// migrationAction is what the dispatcher chose.
+type migrationAction int
+
+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)
+ 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
+ migrateSupersede // anything else: fence a present record, superseded
+)
+
+// migrationDispatch chooses m's next step from obs alone.
+func migrationDispatch(obs provision.Observation, m Migration) migrationAction {
+ switch {
+ case obs.Verdict == provision.LifetimeUnreadable:
+ return migrateWait
+ case obs.Verdict == provision.LifetimeCorrupt && obs.Slot == nil:
+ return migrateQuarantine
+ case !obs.Present || obs.Rec == nil:
+ return migrateSupersede
+ }
+ rec := obs.Rec
+ switch {
+ case !rec.Retired && rec.IncarnationMigrating == m.OpID:
+ return migrateOwnMarker
+ case !rec.Retired && rec.IncarnationMigrating == "" && rec.Incarnation == 1 && migrationOwnsSlot(obs.Slot, m):
+ return migrateDone
+ case fieldless(rec) && obs.Slot == nil:
+ if m.ExpectETag != "" && obs.ETag != m.ExpectETag {
+ return migrateMoved
+ }
+ return migrateMark
+ }
+ return migrateSupersede
+}
+
+// migrateHostLifetimeLocked migrates acct/app/host's fieldless record. The
+// caller holds s.lockApp then the hostname lock (the lock table).
+// expectETag == "" is unbound: the migration retries from whatever record is
+// current. A real ETag binds the marker CAS to that read: a record that moved
+// since is release.ErrCASConflict with the migration superseded.
+//
+// It returns the record as written (state committed); or state superseded
+// (another lifetime, a gone or corrupt body; nil error); or errLifetimePending
+// (state pending: recovery owns the row); or, before the row is durable,
+// errLifetimeSettling or an ordinary error with nothing persisted.
+func (s *Server) migrateHostLifetimeLocked(ctx context.Context, acct, app, host, expectETag string) (*provision.Record, []byte, string, string, error) {
+ if err := s.admit(host, ""); err != nil {
+ return nil, nil, "", "", err
+ }
+ opID, err := newLifetimeOpID()
+ if err != nil {
+ return nil, nil, "", "", err
+ }
+ m := Migration{OpID: opID, Account: acct, App: app, Host: host, ExpectETag: expectETag, State: lifetimePending}
+ if err := s.store.BeginMigration(m); err != nil {
+ return nil, nil, "", "", fmt.Errorf("console: lifetimes: begin the migration of %s: %w", host, err)
+ }
+ // Durable from here: pending, terminal or the record.
+ return s.migrationStepsLocked(ctx, m)
+}
+
+// migrationPending is the answer for a row left pending.
+func migrationPending() (*provision.Record, []byte, string, string, error) {
+ return nil, nil, "", lifetimePending, errLifetimePending
+}
+
+// settleMigration moves m to a terminal state: done when written,
+// errLifetimePending when not.
+func (s *Server) settleMigration(m Migration, state string, done error) (*provision.Record, []byte, string, string, error) {
+ if err := s.store.SetMigrationState(m.OpID, state); err != nil {
+ log.Printf("console: lifetimes: migration %s of %s: mark %s: %v (pending)", m.OpID, m.Host, state, err)
+ return migrationPending()
+ }
+ return nil, nil, "", state, done
+}
+
+// migrationStepsLocked is the dispatcher loop: classify, dispatch, act, and
+// classify again after every step that may have changed what is there.
+func (s *Server) migrationStepsLocked(ctx context.Context, m Migration) (*provision.Record, []byte, string, string, error) {
+ key := provision.RecordKey(m.Account, m.App, m.Host)
+ logf := func(format string, args ...any) {
+ log.Printf("console: lifetimes: migration %s of %s: "+format, append([]any{m.OpID, m.Host}, args...)...)
+ }
+ marks := 0
+ for round := 0; round < migrationRounds; round++ {
+ obs, err := s.classifyLocked(ctx, m.Account, m.App, m.Host)
+ if err != nil {
+ logf("%v (pending)", err)
+ return migrationPending()
+ }
+ switch migrationDispatch(obs, m) {
+ case migrateWait:
+ logf("unreadable (pending)")
+ return migrationPending()
+ case migrateDone:
+ if err := s.store.SetMigrationState(m.OpID, lifetimeCommitted); err != nil {
+ logf("commit: %v (pending; the record stands)", err)
+ return migrationPending()
+ }
+ return obs.Rec, obs.Raw, obs.ETag, lifetimeCommitted, nil
+ case migrateOwnMarker:
+ return s.migrationFinishLocked(ctx, m, obs)
+ case migrateMark:
+ if marks == migrationMarkRounds {
+ logf("the record kept changing under %d marker writes (pending)", marks)
+ return migrationPending()
+ }
+ marks++
+ marked, err := s.migrationMarkLocked(ctx, m, obs)
+ if errors.Is(err, release.ErrCASConflict) {
+ continue // dispatch again from what is there now
+ }
+ if err != nil {
+ logf("mark the record: %v (pending)", err)
+ return migrationPending()
+ }
+ if err := s.store.SetMigrationMarkedETag(m.OpID, marked); err != nil {
+ logf("persist the marked ETag: %v (pending)", err)
+ return migrationPending()
+ }
+ m.MarkedETag = marked
+ case migrateMoved:
+ _, redispatch, err := s.terminateFenced(ctx, m.Account, m.App, m.Host, fenceOwner{MarkerID: m.OpID}, func() error {
+ return s.store.SetMigrationState(m.OpID, lifetimeSuperseded)
+ })
+ switch {
+ case err != nil:
+ return migrationPending()
+ case redispatch:
+ continue
+ }
+ return nil, nil, "", lifetimeSuperseded,
+ fmt.Errorf("console: lifetimes: %s changed since it was read: %w", key, release.ErrCASConflict)
+ case migrateQuarantine:
+ if err := s.quarantineCopy(ctx, m.Account, m.App, m.Host, obs.Raw, obs.ETag); err != nil {
+ logf("quarantine the corrupt body: %v (pending)", err)
+ return migrationPending()
+ }
+ err := release.DeleteIfMatch(ctx, s.cfg.Releases, key, obs.ETag)
+ if errors.Is(err, release.ErrCASConflict) {
+ continue
+ }
+ if err != nil {
+ logf("remove the corrupt body: %v (pending)", err)
+ return migrationPending()
+ }
+ return s.settleMigration(m, lifetimeSuperseded, nil)
+ case migrateSupersede:
+ if !obs.Present {
+ return s.settleMigration(m, lifetimeSuperseded, nil)
+ }
+ _, redispatch, err := s.terminateFenced(ctx, m.Account, m.App, m.Host, fenceOwner{MarkerID: m.OpID}, func() error {
+ return s.store.SetMigrationState(m.OpID, lifetimeSuperseded)
+ })
+ switch {
+ case err != nil:
+ return migrationPending()
+ case redispatch:
+ continue
+ }
+ return nil, nil, "", lifetimeSuperseded, nil
+ }
+ }
+ logf("no settled step after %d rounds (pending)", migrationRounds)
+ 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.
+func (s *Server) migrationMarkLocked(ctx context.Context, m Migration, obs provision.Observation) (string, error) {
+ marked, err := provision.MarkMigrating(obs.Raw, m.OpID)
+ if err != nil {
+ return "", err
+ }
+ wctx := provision.WithMigration(provision.WithRecordSequence(ctx), m.OpID)
+ etag, _, err := provision.WriteRecord(wctx, s.cfg.Releases, provision.RecordKey(m.Account, m.App, m.Host), obs.Raw, marked, obs.ETag)
+ return etag, err
+}
+
+// migrationFinishLocked continues from a record carrying m's marker: the
+// marked ETag persisted when the row lacks it, (2) unless the slot is already
+// this migration's bound to that ETag, then (3) and committed.
+func (s *Server) migrationFinishLocked(ctx context.Context, m Migration, obs provision.Observation) (*provision.Record, []byte, string, string, error) {
+ key := provision.RecordKey(m.Account, m.App, m.Host)
+ logf := func(format string, args ...any) {
+ log.Printf("console: lifetimes: migration %s of %s: "+format, append([]any{m.OpID, m.Host}, args...)...)
+ }
+ if m.MarkedETag != obs.ETag {
+ if err := s.store.SetMigrationMarkedETag(m.OpID, obs.ETag); err != nil {
+ logf("persist the marked ETag: %v (pending)", err)
+ return migrationPending()
+ }
+ m.MarkedETag = obs.ETag
+ }
+ // (2)
+ if !migrationOwnsSlot(obs.Slot, m) || obs.Slot.RecordETag != obs.ETag {
+ ifMatch := ""
+ if migrationOwnsSlot(obs.Slot, m) {
+ ifMatch = obs.SlotETag
+ }
+ _, err := provision.WriteHostIncarnation(ctx, s.cfg.Releases, m.Host, migrationSlot(m, obs.ETag), ifMatch)
+ if errors.Is(err, release.ErrCASConflict) {
+ slot, _, rerr := provision.ReadHostIncarnation(ctx, s.cfg.Releases, m.Host)
+ switch {
+ case rerr != nil:
+ logf("re-read the slot: %v (pending)", rerr)
+ return migrationPending()
+ case !migrationOwnsSlot(slot, m):
+ // Another lifetime took the hostname: under its slot the
+ // marked record is stale, and no write of ours is left.
+ logf("the slot is %+v, not this migration's (superseded)", slot)
+ return s.settleMigration(m, lifetimeSuperseded, nil)
+ }
+ } else if err != nil {
+ logf("write the slot: %v (pending)", err)
+ return migrationPending()
+ }
+ }
+ // (3)
+ final, err := provision.FinishMigration(obs.Raw)
+ if err != nil {
+ logf("the final body: %v (pending)", err)
+ return migrationPending()
+ }
+ wctx := provision.WithMigration(provision.WithRecordSequence(ctx), m.OpID)
+ etag, _, err := provision.WriteRecord(wctx, s.cfg.Releases, key, obs.Raw, final, obs.ETag)
+ if errors.Is(err, release.ErrCASConflict) {
+ again, rerr := s.classifyLocked(ctx, m.Account, m.App, m.Host)
+ if rerr != nil || migrationDispatch(again, m) != migrateDone {
+ logf("the final write lost (%v; pending)", rerr)
+ return migrationPending()
+ }
+ obs, etag = again, again.ETag
+ final = nil
+ } else if err != nil {
+ logf("write the final record: %v (pending)", err)
+ return migrationPending()
+ }
+ rec, raw := obs.Rec, obs.Raw
+ if final != nil {
+ // The bytes WriteRecord put: the final body with record_seq advanced.
+ if raw, err = provision.AdvanceRecordSequence(obs.Raw, final); err == nil {
+ var decoded provision.Record
+ decoded, err = provision.DecodeRecord(raw)
+ rec = &decoded
+ }
+ if err != nil {
+ logf("decode the final record: %v (pending; the record stands)", err)
+ return migrationPending()
+ }
+ }
+ if err := s.store.SetMigrationState(m.OpID, lifetimeCommitted); err != nil {
+ logf("commit: %v (pending; the record stands)", err)
+ return migrationPending()
+ }
+ return rec, raw, etag, lifetimeCommitted, nil
+}
+
+// quarantineCopy copies a corrupt record body to
+// control/quarantine/<account>/<app>/<host>.<unix nanos>.json, create-only,
+// and does not touch the record key.
+func (s *Server) quarantineCopy(ctx context.Context, acct, app, host string, raw []byte, etag string) error {
+ key := fmt.Sprintf("control/quarantine/%s/%s/%s.%d.json", acct, app, host, time.Now().UnixNano())
+ if _, err := s.cfg.Releases.Put(ctx, key, raw, ""); err != nil {
+ return fmt.Errorf("console: lifetimes: quarantine %s/%s/%s: %w", acct, app, host, err)
+ }
+ log.Printf("console: lifetimes: the corrupt record %s/%s/%s (ETag %s) is copied to %s", acct, app, host, etag, key)
+ return nil
+}
+
+// recoverMigrationLocked is the migration table: admitted against every other
+// pending operation of the hostname, then the dispatcher from a fresh
+// classification.
+func (s *Server) recoverMigrationLocked(ctx context.Context, m Migration) {
+ logKey := "migration " + m.OpID
+ if err := s.admit(m.Host, m.OpID); err != nil {
+ s.lifetimeLogHourly(logKey, "console: lifetime recovery: migration %s of %s waits: %v", m.OpID, m.Host, err)
+ return
+ }
+ if _, _, _, state, _ := s.migrationStepsLocked(ctx, m); state == lifetimePending {
+ s.lifetimeLogHourly(logKey, "console: lifetime recovery: migration %s of %s stays pending", m.OpID, m.Host)
+ }
+}
+
+// ---- the pass ----------------------------------------------------------------
+
+// StartIncarnationMigrationPass migrates fieldless records now and every 60
+// seconds for the life of the process, behind CARLOS_LIFETIMES_MIGRATE=1.
+// Each iteration is a unit of record work admitted as a writer, and it does
+// work only when admitted on. A host-claim console has no lifetimes.
+func (s *Server) StartIncarnationMigrationPass(ctx context.Context) {
+ if s.cfg.HostClaim {
+ log.Printf("console: incarnation migration pass: off on a host-claim console, where lifetimes are not for it")
+ return
+ }
+ if !s.cfg.LifetimesMigrate {
+ log.Printf("console: incarnation migration pass: off (CARLOS_LIFETIMES_MIGRATE is not 1)")
+ return
+ }
+ if !s.lifetimesGoverned() {
+ return
+ }
+ lifetimesWorker(ctx, lifetimesMigrateInterval, func(ctx context.Context) { s.incarnationMigrationIteration(ctx) })
+}
+
+// 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 {
+ if !s.cfg.LifetimesMigrate {
+ return false
+ }
+ return s.lifetimesIteration(ctx, "incarnation migration", true, true, func(ctx context.Context) {
+ if err := s.migrateFieldlessRecords(ctx); err != nil {
+ log.Printf("console: incarnation migration pass: %v", err)
+ }
+ })
+}
+
+// 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 {
+ recs, _, err := provision.ListAllRecordsTolerantJudged(ctx, s.cfg.Releases, s.judge)
+ if err != nil {
+ return fmt.Errorf("list the records: %w", err)
+ }
+ var candidates []provision.Record
+ for i := range recs {
+ if fieldless(&recs[i]) {
+ candidates = append(candidates, recs[i])
+ }
+ }
+ sort.SliceStable(candidates, func(i, j int) bool {
+ a, b := candidates[i], candidates[j]
+ if !a.CreatedAt.Equal(b.CreatedAt) {
+ return a.CreatedAt.Before(b.CreatedAt)
+ }
+ return provision.RecordKey(a.Account, a.App, a.Host) < provision.RecordKey(b.Account, b.App, b.Host)
+ })
+ migrated := 0
+ for _, rec := range candidates {
+ if migrated == migrationPassLimit {
+ break
+ }
+ _ = s.withAppHost(rec.Account, rec.App, rec.Host, func() error {
+ if s.admit(rec.Host, "") != nil {
+ return nil
+ }
+ obs, err := s.classifyLocked(ctx, rec.Account, rec.App, rec.Host)
+ if err != nil || !obs.Present || !fieldless(obs.Rec) || obs.Slot != nil {
+ return nil
+ }
+ migrated++
+ _, _, _, 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)
+ }
+ return nil
+ })
+ }
+ return nil
+}
diff --git a/internal/console/lifetimes_migrate_test.go b/internal/console/lifetimes_migrate_test.go
new file mode 100644
index 000000000..98e70b7eb
--- /dev/null
+++ b/internal/console/lifetimes_migrate_test.go
@@ -0,0 +1,946 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The lazy migration (lifetimes_migrate.go), driven through the real server,
+// store and provision.WriteRecord of a claims-off console with lifetimes on,
+// over a bucket that can fail, pause or record any call.
+
+// ---- fixtures -------------------------------------------------------------
+
+type migRow struct {
+ OpID, App, Expect, Marked, State string
+}
+
+func migRows(t *testing.T, st *Store, host string) []migRow {
+ t.Helper()
+ rows, err := st.db.Query(`SELECT op_id, app, expect_etag, marked_etag, state FROM record_migrations WHERE host = ? ORDER BY rowid`, host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer rows.Close()
+ var out []migRow
+ for rows.Next() {
+ var m migRow
+ if err := rows.Scan(&m.OpID, &m.App, &m.Expect, &m.Marked, &m.State); err != nil {
+ t.Fatal(err)
+ }
+ out = append(out, m)
+ }
+ return out
+}
+
+// oneMig is the host's only migration row, required in state.
+func oneMig(t *testing.T, st *Store, host, state string) migRow {
+ t.Helper()
+ rows := migRows(t, st, host)
+ if len(rows) != 1 || rows[0].State != state {
+ t.Fatalf("migrations of %s = %+v, want one %s", host, rows, state)
+ }
+ return rows[0]
+}
+
+type migResult struct {
+ rec *provision.Record
+ raw []byte
+ etag string
+ state string
+ err error
+}
+
+// migrateUnder runs migrateHostLifetimeLocked under both locks, as its
+// callers do, with ctx a unit admitted on.
+func migrateUnder(ctx context.Context, srv *Server, host, expect string) migResult {
+ var r migResult
+ _ = srv.withAppHost("bab", ltApp, host, func() error {
+ r.rec, r.raw, r.etag, r.state, r.err = srv.migrateHostLifetimeLocked(ctx, "bab", ltApp, host, expect)
+ return nil
+ })
+ return r
+}
+
+func migrateOn(t *testing.T, srv *Server, host, expect string) migResult {
+ t.Helper()
+ return migrateUnder(onCtx(t, srv), srv, host, expect)
+}
+
+// migrateAsync starts migrateUnder in a goroutine.
+func migrateAsync(t *testing.T, srv *Server, host, expect string) <-chan migResult {
+ t.Helper()
+ ctx := onCtx(t, srv)
+ out := make(chan migResult, 1)
+ go func() { out <- migrateUnder(ctx, srv, host, expect) }()
+ return out
+}
+
+func waitMig(t *testing.T, ch <-chan migResult) migResult {
+ t.Helper()
+ select {
+ case r := <-ch:
+ return r
+ case <-time.After(10 * time.Second):
+ t.Fatal("timed out waiting for the migration")
+ return migResult{}
+ }
+}
+
+// lateMarker issues op's step (1) marker PUT against etag, exactly as the
+// migration writes it, held by gate before it reaches the bucket: the write
+// a console that crashed mid-migration left outstanding.
+func lateMarker(t *testing.T, l ltOn, gate *pauseGate, op string, raw []byte, etag string) <-chan error {
+ t.Helper()
+ marked, err := provision.MarkMigrating(raw, op)
+ if err != nil {
+ t.Fatal(err)
+ }
+ out := make(chan error, 1)
+ go func() {
+ ctx := provision.WithMigration(provision.WithRecordSequence(context.Background()), op)
+ _, _, err := provision.WriteRecord(ctx, l.bucket, recKey(ltApp, ltOne), raw, marked, etag)
+ out <- err
+ }()
+ waitFor(t, gate.entered, "the outstanding marker write")
+ return out
+}
+
+// beginMig persists a pending migration as BeginMigration does at the start
+// of the protocol: the state a crash right after it leaves.
+func beginMig(t *testing.T, l ltOn, op, expect string) {
+ t.Helper()
+ if err := l.fx.st.BeginMigration(Migration{OpID: op, Account: "bab", App: ltApp, Host: ltOne, ExpectETag: expect, State: lifetimePending}); err != nil {
+ t.Fatal(err)
+ }
+}
+
+const migOp = "0123456789abcdef0123456789abcdef"
+
+// requireMigrated: the record of shop at host at incarnation 1 with no marker
+// and a record_seq, under the live slot {1, this key, "migration " + op}
+// bound to a marked ETag, and the judge vouching for it.
+func requireMigrated(t *testing.T, st release.Store, host, op string) provision.Record {
+ t.Helper()
+ raw, etag := rawOf(t, st, ltApp, host)
+ if raw == nil {
+ t.Fatalf("no record of %s", host)
+ }
+ rec := decodeRec(t, raw)
+ if rec.Incarnation != 1 || rec.IncarnationMigrating != "" || rec.RecordSeq == 0 || rec.Retired {
+ t.Fatalf("record = %s, want incarnation 1, no marker, a record_seq", raw)
+ }
+ slot, _ := slotOf(t, st, host)
+ want := provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: "bab", App: ltApp, Intent: "migration " + op}
+ if slot == nil || slot.RecordETag == "" {
+ t.Fatalf("slot = %+v, want %+v bound to the marked record", slot, want)
+ }
+ got := *slot
+ got.RecordETag = ""
+ if got != want {
+ t.Fatalf("slot = %+v, want %+v", *slot, want)
+ }
+ obs := provision.NewLifetimeJudge().Observe(context.Background(), st, "bab", ltApp, host)
+ if obs.Verdict != provision.LifetimeValid || obs.ETag != etag {
+ t.Fatalf("the judge says %v at %q, want valid at %q", obs.Verdict, obs.ETag, etag)
+ }
+ return rec
+}
+
+// recordPuts is how many PUTs reached key since the log began.
+func recordPuts(l ltOn, key string) int {
+ n := 0
+ for _, op := range l.bucket.opsOf(key) {
+ if op == "put" {
+ n++
+ }
+ }
+ return n
+}
+
+// sqlFailMigration refuses every update of record_migrations that sets col
+// (marked_etag, or state to committed) until the returned func drops it.
+func sqlFailMigration(t *testing.T, st *Store, when string) func() {
+ t.Helper()
+ if _, err := st.db.Exec(`CREATE TRIGGER lt_fail_migration BEFORE UPDATE ON record_migrations WHEN ` + when + `
+BEGIN SELECT RAISE(ABORT, 'test: the migration update is refused'); END`); err != nil {
+ t.Fatal(err)
+ }
+ var once sync.Once
+ drop := func() {
+ once.Do(func() {
+ if _, err := st.db.Exec(`DROP TRIGGER lt_fail_migration`); err != nil {
+ t.Fatal(err)
+ }
+ })
+ }
+ t.Cleanup(drop)
+ return drop
+}
+
+func channelBeta(raw []byte) []byte {
+ return bytes.Replace(raw, []byte(`"channel":"stable"`), []byte(`"channel":"beta"`), 1)
+}
+
+// updateFirst is a hook that lands an ordinary update of the record straight
+// in the bucket the first time a PUT of key is about to run: an update that
+// won between the migration's classification and its marker CAS.
+func updateFirst(t *testing.T, l ltOn, key string) func(string, string) error {
+ var once sync.Once
+ return func(op, k string) error {
+ if op == "put" && k == key {
+ once.Do(func() {
+ raw, e, err := l.fx.rel.Get(context.Background(), key)
+ if err == nil {
+ _, err = l.fx.rel.Put(context.Background(), key, channelBeta(raw), e)
+ }
+ if err != nil {
+ t.Errorf("the winning update: %v", err)
+ }
+ })
+ }
+ return nil
+ }
+}
+
+// ---- the protocol -------------------------------------------------------------
+
+func TestMigrationMarksThenSlotsThenSets(t *testing.T) {
+ l := newLtOn(t)
+ e := legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ key, slotKey := recKey(ltApp, ltOne), provision.HostIncarnationKey(ltOne)
+ r := migrateOn(t, l.fx.srv, ltOne, "")
+ if r.err != nil || r.state != lifetimeCommitted {
+ t.Fatalf("migrate = %q %v, want committed", r.state, r.err)
+ }
+ row := oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ rec := requireMigrated(t, l.fx.rel, ltOne, row.OpID)
+ if rec.RecordSeq != 2 || rec.Channel != "stable" {
+ t.Fatalf("record = %+v, want record_seq 2 (marker, final) and the rest kept", rec)
+ }
+ slot, _ := slotOf(t, l.fx.rel, ltOne)
+ if row.Expect != "" || row.Marked == "" || row.Marked == e || slot.RecordETag != row.Marked {
+ t.Fatalf("row = %+v, slot bound to %q: want unbound, the marked ETag persisted and the slot bound to it", row, slot.RecordETag)
+ }
+ raw, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if !bytes.Equal(r.raw, raw) || r.etag != etag || r.rec == nil || r.rec.Incarnation != 1 || r.rec.RecordSeq != 2 {
+ t.Fatalf("migrate returned %s at %q, want the record as written", r.raw, r.etag)
+ }
+ // The order: the marker, then the slot, then the final record.
+ var puts []string
+ l.bucket.mu.Lock()
+ for _, op := range l.bucket.ops {
+ switch op {
+ case "put " + key:
+ puts = append(puts, "record")
+ case "put " + slotKey:
+ puts = append(puts, "slot")
+ }
+ }
+ l.bucket.mu.Unlock()
+ if strings.Join(puts, ",") != "record,slot,record" {
+ t.Fatalf("puts = %v, want the marker, the slot, the final record", puts)
+ }
+}
+
+func TestMigrationRestartsFromAnUpdateThatWon(t *testing.T) {
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ key := recKey(ltApp, ltOne)
+ l.bucket.setHook(updateFirst(t, l, key))
+ r := migrateOn(t, l.fx.srv, ltOne, "")
+ if r.err != nil || r.state != lifetimeCommitted {
+ t.Fatalf("migrate = %q %v, want committed from the updated record", r.state, r.err)
+ }
+ row := oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ if rec := requireMigrated(t, l.fx.rel, ltOne, row.OpID); rec.Channel != "beta" {
+ t.Fatalf("record = %+v, want the update that won kept", rec)
+ }
+ if n := recordPuts(l, key); n != 3 {
+ t.Fatalf("record puts = %d, want the lost marker, the marker, the final record", n)
+ }
+}
+
+func TestBoundMigrationReturnsConflictToItsCaller(t *testing.T) {
+ l := newLtOn(t)
+ e := legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ l.bucket.setHook(updateFirst(t, l, recKey(ltApp, ltOne)))
+ r := migrateOn(t, l.fx.srv, ltOne, e)
+ if !errors.Is(r.err, release.ErrCASConflict) || r.state != lifetimeSuperseded {
+ t.Fatalf("migrate = %q %v, want superseded and the caller's CAS conflict", r.state, r.err)
+ }
+ oneMig(t, l.fx.st, ltOne, lifetimeSuperseded)
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if rec := decodeRec(t, raw); rec.Channel != "beta" || rec.IncarnationMigrating != "" || rec.Incarnation != 0 || rec.RecordSeq != 1 {
+ t.Fatalf("record = %s, want the update intact, unmarked, fenced (record_seq 1)", raw)
+ }
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot != nil {
+ t.Fatalf("slot = %+v, want none", slot)
+ }
+}
+
+// ---- the writers --------------------------------------------------------------
+
+func TestAnIssuedWriteLosesToTheMarkerAndRetries(t *testing.T) {
+ t.Cleanup(provision.SetRecordMigrationWaitForTest(10*time.Second, 10*time.Millisecond))
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ key := recKey(ltApp, ltOne)
+ // The writer read the unmarked record and issued its PUT (record put 1);
+ // the migration's final write (record put 3) is held so the marker
+ // (record put 2) stands while the writer retries.
+ wgate, whook := pauseNth(t, "put", key, 1)
+ fgate, fhook := pauseNth(t, "put", key, 3)
+ l.bucket.setHook(chain(fhook, whook)) // fhook first: it counts the writer's PUT while whook holds it
+ wctx := onCtx(t, l.fx.srv)
+ werr := make(chan error, 1)
+ go func() { werr <- provision.SetRecordRegion(wctx, l.bucket, "bab", ltApp, ltOne, "uk") }()
+ waitFor(t, wgate.entered, "the writer's PUT")
+ mig := migrateAsync(t, l.fx.srv, ltOne, "")
+ waitFor(t, fgate.entered, "the migration's final write")
+ gets := func() int {
+ n := 0
+ for _, op := range l.bucket.opsOf(key) {
+ if op == "get" {
+ n++
+ }
+ }
+ return n
+ }
+ before := gets()
+ wgate.open()
+ deadline := time.Now().Add(10 * time.Second)
+ for gets() < before+3 {
+ if time.Now().After(deadline) {
+ t.Fatal("the writer never re-read the marked record")
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ select {
+ case err := <-werr:
+ t.Fatalf("the writer returned %v while the marker stood; want it waiting", err)
+ default:
+ }
+ if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); decodeRec(t, raw).IncarnationMigrating == "" {
+ t.Fatalf("record = %s, want the marker standing", raw)
+ }
+ fgate.open()
+ if r := waitMig(t, mig); r.err != nil || r.state != lifetimeCommitted {
+ t.Fatalf("migrate = %q %v, want committed", r.state, r.err)
+ }
+ if err := <-werr; err != nil {
+ t.Fatalf("the writer = %v, want it to land on the migrated record", err)
+ }
+ row := oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ if rec := requireMigrated(t, l.fx.rel, ltOne, row.OpID); rec.Region != "uk" || rec.RecordSeq != 3 {
+ t.Fatalf("record = %+v, want the writer's region on the migrated record, record_seq 3", rec)
+ }
+}
+
+func TestWriterGivesUpAfterTheBound(t *testing.T) {
+ const wait = 200 * time.Millisecond
+ t.Cleanup(provision.SetRecordMigrationWaitForTest(wait, 20*time.Millisecond))
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ key := recKey(ltApp, ltOne)
+ fgate, fhook := pauseNth(t, "put", key, 2)
+ l.bucket.setHook(fhook)
+ mig := migrateAsync(t, l.fx.srv, ltOne, "")
+ waitFor(t, fgate.entered, "the migration's final write")
+ marked, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ start := time.Now()
+ err := provision.SetRecordRegion(onCtx(t, l.fx.srv), l.bucket, "bab", ltApp, ltOne, "uk")
+ if !errors.Is(err, provision.ErrRecordMigrating) || !errors.Is(err, release.ErrCASConflict) {
+ t.Fatalf("the writer = %v, want ErrRecordMigrating, a CAS conflict", err)
+ }
+ if waited := time.Since(start); waited < wait {
+ t.Fatalf("the writer gave up after %v, before the bound %v", waited, wait)
+ }
+ if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); !bytes.Equal(raw, marked) {
+ t.Fatalf("record = %s, want the marked record untouched", raw)
+ }
+ fgate.open()
+ if r := waitMig(t, mig); r.err != nil || r.state != lifetimeCommitted {
+ t.Fatalf("migrate = %q %v, want committed", r.state, r.err)
+ }
+ row := oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ if rec := requireMigrated(t, l.fx.rel, ltOne, row.OpID); rec.Region != "" {
+ t.Fatalf("record = %+v, want the writer's change absent", rec)
+ }
+}
+
+// ---- the late marker -----------------------------------------------------------
+
+func TestLateMarkerWriteIsCompletedNotStranded(t *testing.T) {
+ points := []string{"before the classification", "between the classification and the dispatch", "between the dispatch and the CAS", "after the CAS"}
+ for _, bound := range []bool{false, true} {
+ for _, point := range points {
+ name := "unbound, " + point
+ if bound {
+ name = "bound, " + point
+ }
+ t.Run(name, func(t *testing.T) {
+ l := newLtOn(t)
+ e := legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ key, slotKey := recKey(ltApp, ltOne), provision.HostIncarnationKey(ltOne)
+ expect := ""
+ if bound {
+ expect = e
+ }
+ beginMig(t, l, migOp, expect)
+ gate, hook := pauseNth(t, "put", key, 1)
+ l.bucket.setHook(hook)
+ late := lateMarker(t, l, gate, migOp, raw, e)
+ // The console that issued it is gone; a second one recovers.
+ srv := l.restart(t)
+ var lateErr error
+ released := false
+ releaseLate := func() {
+ gate.open()
+ lateErr, released = <-late, true
+ }
+ var fired atomic.Bool
+ switch point {
+ case "before the classification":
+ releaseLate()
+ case "between the classification and the dispatch":
+ l.bucket.setAfter(func(op, k string, _ error) {
+ if op == "get" && k == slotKey && fired.CompareAndSwap(false, true) {
+ releaseLate()
+ }
+ })
+ case "between the dispatch and the CAS":
+ l.bucket.setHook(chain(hook, func(op, k string) error {
+ if op == "put" && k == key && fired.CompareAndSwap(false, true) {
+ releaseLate()
+ }
+ return nil
+ }))
+ case "after the CAS":
+ l.bucket.setAfter(func(op, k string, err error) {
+ if op == "put" && k == key && err == nil && fired.CompareAndSwap(false, true) {
+ releaseLate()
+ }
+ })
+ }
+ recoverOn(t, srv)
+ if !released {
+ t.Fatal("the late marker was never released")
+ }
+ if point == "after the CAS" {
+ if !errors.Is(lateErr, release.ErrCASConflict) {
+ t.Fatalf("the late marker = %v, want it to lose to recovery's", lateErr)
+ }
+ } else if lateErr != nil {
+ t.Fatalf("the late marker = %v, want it landed", lateErr)
+ }
+ oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, ltOne, migOp)
+ })
+ }
+ }
+}
+
+func TestBoundMigrationFencesBeforeSuperseding(t *testing.T) {
+ ctx := context.Background()
+ // setup: a pending migration bound to A at E with its marker PUT against
+ // E outstanding; the record is B; recovery's classification sees B, and
+ // A (with E) comes back before recovery acts on it.
+ setup := func(t *testing.T) (ltOn, *Server, string, <-chan error, *pauseGate) {
+ l := newLtOn(t)
+ e := legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ a, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ key := recKey(ltApp, ltOne)
+ beginMig(t, l, migOp, e)
+ gate, hook := pauseNth(t, "put", key, 1)
+ l.bucket.setHook(hook)
+ late := lateMarker(t, l, gate, migOp, a, e)
+ eb, err := l.fx.rel.Put(ctx, key, channelBeta(a), e)
+ if err != nil {
+ t.Fatal(err)
+ }
+ srv := l.restart(t)
+ var fired atomic.Bool
+ l.bucket.setAfter(func(op, k string, _ error) {
+ if op == "get" && k == provision.HostIncarnationKey(ltOne) && fired.CompareAndSwap(false, true) {
+ if again, err := l.fx.rel.Put(ctx, key, a, eb); err != nil || again != e {
+ t.Errorf("restoring A = %q %v, want the ETag %q again", again, err, e)
+ }
+ }
+ })
+ return l, srv, e, late, gate
+ }
+ t.Run("the fence closes it", func(t *testing.T) {
+ l, srv, e, late, gate := setup(t)
+ recoverOn(t, srv)
+ oneMig(t, l.fx.st, ltOne, lifetimeSuperseded)
+ _, etag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if etag == e {
+ t.Fatal("superseded without a fence: the record still has the ETag the marker PUT names")
+ }
+ gate.open()
+ if err := <-late; !errors.Is(err, release.ErrCASConflict) {
+ t.Fatalf("the late marker = %v, want a CAS conflict", err)
+ }
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if rec := decodeRec(t, raw); rec.IncarnationMigrating != "" || rec.RecordSeq != 1 || rec.Channel != "stable" {
+ t.Fatalf("record = %s, want A fenced (record_seq 1) and no marker", raw)
+ }
+ })
+ t.Run("without the fence the marker is stranded", func(t *testing.T) {
+ l, srv, _, late, gate := setup(t)
+ lifetimesFenceDisabled = true
+ t.Cleanup(func() { lifetimesFenceDisabled = false })
+ recoverOn(t, srv)
+ oneMig(t, l.fx.st, ltOne, lifetimeSuperseded)
+ gate.open()
+ if err := <-late; err != nil {
+ t.Fatalf("the late marker = %v; with the fence off it lands (the defect)", err)
+ }
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if rec := decodeRec(t, raw); rec.IncarnationMigrating != migOp {
+ t.Fatalf("record = %s, want the stranded marker (the defect the fence closes)", raw)
+ }
+ })
+}
+
+// ---- recovery --------------------------------------------------------------
+
+func TestMigrationRecoveryBranches(t *testing.T) {
+ ctx := context.Background()
+ key, slotKey := recKey(ltApp, ltOne), provision.HostIncarnationKey(ltOne)
+ // pendingAt runs a real migration that stops pending at a failure.
+ pendingAt := func(t *testing.T, l ltOn, fail func()) migRow {
+ t.Helper()
+ fail()
+ r := migrateOn(t, l.fx.srv, ltOne, "")
+ if !errors.Is(r.err, errLifetimePending) || r.state != lifetimePending {
+ t.Fatalf("migrate = %q %v, want pending", r.state, r.err)
+ }
+ l.bucket.setHook(nil)
+ return oneMig(t, l.fx.st, ltOne, lifetimePending)
+ }
+ recoverAgain := func(t *testing.T, l ltOn) {
+ t.Helper()
+ recoverOn(t, l.restart(t))
+ }
+ t.Run("own marker, its ETag never persisted", func(t *testing.T) {
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ var drop func()
+ row := pendingAt(t, l, func() { drop = sqlFailMigration(t, l.fx.st, `NEW.marked_etag <> OLD.marked_etag`) })
+ drop()
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if row.Marked != "" || decodeRec(t, raw).IncarnationMigrating != row.OpID {
+ t.Fatalf("row %+v, record %s: want the marker landed and its ETag not persisted", row, raw)
+ }
+ recoverAgain(t, l)
+ oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, ltOne, row.OpID)
+ })
+ t.Run("own marker, no slot", func(t *testing.T) {
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ row := pendingAt(t, l, func() { l.bucket.setHook(failNth("put", slotKey, 1, false)) })
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot != nil || row.Marked == "" {
+ t.Fatalf("slot %+v, row %+v: want no slot and the marked ETag persisted", slot, row)
+ }
+ recoverAgain(t, l)
+ oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, ltOne, row.OpID)
+ })
+ t.Run("own marker under its slot", func(t *testing.T) {
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ row := pendingAt(t, l, func() { l.bucket.setHook(failNth("put", key, 2, false)) })
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot == nil || slot.RecordETag != row.Marked {
+ t.Fatalf("slot = %+v, want it bound to the marked record", slot)
+ }
+ recoverAgain(t, l)
+ oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, ltOne, row.OpID)
+ })
+ t.Run("own slot and done", func(t *testing.T) {
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ var drop func()
+ row := pendingAt(t, l, func() { drop = sqlFailMigration(t, l.fx.st, `NEW.state = 'committed'`) })
+ drop()
+ requireMigrated(t, l.fx.rel, ltOne, row.OpID)
+ puts, slotPuts := recordPuts(l, key), recordPuts(l, slotKey)
+ recoverAgain(t, l)
+ oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ if recordPuts(l, key) != puts || recordPuts(l, slotKey) != slotPuts {
+ t.Fatal("recovery of a finished migration wrote again")
+ }
+ })
+ for _, bound := range []bool{false, true} {
+ name := "eligible, unbound"
+ if bound {
+ name = "eligible, bound"
+ }
+ t.Run(name, func(t *testing.T) {
+ l := newLtOn(t)
+ e := legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ expect := ""
+ if bound {
+ expect = e
+ }
+ beginMig(t, l, migOp, expect)
+ recoverAgain(t, l)
+ oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, ltOne, migOp)
+ })
+ }
+ t.Run("bound and moved", func(t *testing.T) {
+ l := newLtOn(t)
+ e := legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ beginMig(t, l, migOp, e)
+ a, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if _, err := l.fx.rel.Put(ctx, key, channelBeta(a), e); err != nil {
+ t.Fatal(err)
+ }
+ recoverAgain(t, l)
+ oneMig(t, l.fx.st, ltOne, lifetimeSuperseded)
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if rec := decodeRec(t, raw); rec.Channel != "beta" || rec.RecordSeq != 1 || rec.IncarnationMigrating != "" {
+ t.Fatalf("record = %s, want the moved record fenced and unmarked", raw)
+ }
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot != nil {
+ t.Fatalf("slot = %+v, want none", slot)
+ }
+ })
+ t.Run("unreadable", func(t *testing.T) {
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ beginMig(t, l, migOp, "")
+ srv := l.restart(t)
+ l.bucket.setHook(failNth("get", slotKey, 1, true))
+ before := recordPuts(l, key)
+ recoverOn(t, srv)
+ oneMig(t, l.fx.st, ltOne, lifetimePending)
+ if recordPuts(l, key) != before {
+ t.Fatal("an unreadable observation wrote the record")
+ }
+ l.bucket.setHook(nil)
+ recoverOn(t, srv)
+ oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, ltOne, migOp)
+ })
+ t.Run("a corrupt body", func(t *testing.T) {
+ l := newLtOn(t)
+ e := legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ beginMig(t, l, migOp, "")
+ if _, err := l.fx.rel.Put(ctx, key, []byte("broken"), e); err != nil {
+ t.Fatal(err)
+ }
+ recoverAgain(t, l)
+ oneMig(t, l.fx.st, ltOne, lifetimeSuperseded)
+ if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); raw != nil {
+ t.Fatalf("record = %s, want the corrupt body removed", raw)
+ }
+ keys, err := l.fx.rel.List(ctx, "control/quarantine/bab/shop/")
+ if err != nil || len(keys) != 1 || !strings.HasPrefix(keys[0], "control/quarantine/bab/shop/"+ltOne+".") || !strings.HasSuffix(keys[0], ".json") {
+ t.Fatalf("quarantine = %v %v, want one copy", keys, err)
+ }
+ if copied, _, _ := l.fx.rel.Get(ctx, keys[0]); string(copied) != "broken" {
+ t.Fatalf("quarantined bytes = %q, want the corrupt body", copied)
+ }
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot != nil {
+ t.Fatalf("slot = %+v, want none", slot)
+ }
+ })
+ t.Run("gone", func(t *testing.T) {
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ beginMig(t, l, migOp, "")
+ if err := l.fx.rel.Delete(ctx, key); err != nil {
+ t.Fatal(err)
+ }
+ recoverAgain(t, l)
+ oneMig(t, l.fx.st, ltOne, lifetimeSuperseded)
+ if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); raw != nil {
+ t.Fatalf("record = %s, want none", raw)
+ }
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot != nil {
+ t.Fatalf("slot = %+v, want none", slot)
+ }
+ })
+ t.Run("another lifetime", 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)
+ rec := decodeRec(t, a)
+ rec.Incarnation = 2
+ body, _ := json.Marshal(rec)
+ if _, err := l.fx.rel.Put(ctx, key, body, e); err != nil {
+ t.Fatal(err)
+ }
+ want := provision.HostIncarnation{Incarnation: 2, State: provision.SlotLive, Account: "bab", App: ltApp, 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.Incarnation != 2 || got.RecordSeq != 1 {
+ t.Fatalf("record = %s, want the other lifetime's record fenced", raw)
+ }
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot == nil || *slot != want {
+ t.Fatalf("slot = %+v, want %+v untouched", slot, want)
+ }
+ })
+ t.Run("another operation's marker", 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)
+ }
+ 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)
+ }
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot != nil {
+ t.Fatalf("slot = %+v, want none", slot)
+ }
+ })
+ t.Run("a slot not of this migration", func(t *testing.T) {
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ beginMig(t, l, migOp, "")
+ 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 != "" || got.Incarnation != 0 || got.RecordSeq != 1 {
+ t.Fatalf("record = %s, want the stale record fenced and unmarked", raw)
+ }
+ if slot, _ := slotOf(t, l.fx.rel, ltOne); slot == nil || *slot != want {
+ t.Fatalf("slot = %+v, want %+v untouched", slot, want)
+ }
+ })
+ t.Run("a second crash during recovery", func(t *testing.T) {
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ beginMig(t, l, migOp, "")
+ srv := l.restart(t)
+ l.bucket.setHook(failNth("put", slotKey, 1, false))
+ recoverOn(t, srv)
+ oneMig(t, l.fx.st, ltOne, lifetimePending)
+ raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if decodeRec(t, raw).IncarnationMigrating != migOp {
+ t.Fatalf("record = %s, want marked by the first recovery", raw)
+ }
+ l.bucket.setHook(nil)
+ recoverAgain(t, l)
+ oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, ltOne, migOp)
+ })
+}
+
+// ---- the pass -------------------------------------------------------------------
+
+func TestMigrationPassIsOffWithoutTheFlag(t *testing.T) {
+ l := newLtOn(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ legacyBody(t, l.fx.rel, ltApp, ltOld, "stable")
+ one, oneTag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ done := make(chan struct{})
+ go func() {
+ l.fx.srv.StartIncarnationMigrationPass(context.Background())
+ close(done)
+ }()
+ waitFor(t, done, "the pass to return with CARLOS_LIFETIMES_MIGRATE unset")
+ if raw, tag := rawOf(t, l.fx.rel, ltApp, ltOne); !bytes.Equal(raw, one) || tag != oneTag {
+ t.Fatalf("record = %s, want it untouched with the pass off", raw)
+ }
+ if rows := append(migRows(t, l.fx.st, ltOne), migRows(t, l.fx.st, ltOld)...); len(rows) != 0 {
+ t.Fatalf("migrations = %+v, want none with the pass off", rows)
+ }
+ // With the flag the pass runs its iterations (one here: the context is
+ // already done when the first ends).
+ l.fx.srv.cfg.LifetimesMigrate = true
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ l.fx.srv.StartIncarnationMigrationPass(ctx)
+ for _, host := range []string{ltOne, ltOld} {
+ row := oneMig(t, l.fx.st, host, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, host, row.OpID)
+ }
+}
+
+func TestMigrationPassRunsOnlyWithLifetimesOn(t *testing.T) {
+ l := newLifetimesFx(t)
+ legacyRecord(t, l.fx.rel, ltOne, "stable")
+ l.fx.srv.cfg.LifetimesMigrate = true
+ if l.fx.srv.incarnationMigrationIteration(context.Background()) {
+ t.Fatal("the migration pass ran before the cutover")
+ }
+ if rows := migRows(t, l.fx.st, ltOne); len(rows) != 0 {
+ t.Fatalf("migrations = %+v, want none before the cutover", rows)
+ }
+}
+
+func TestMigrationSkipsAFieldlessRecordBesideAnUnrelatedSlot(t *testing.T) {
+ l := newLtOn(t)
+ l.fx.srv.cfg.LifetimesMigrate = true
+ // one: a fieldless record beside another app's live slot.
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ other := provision.HostIncarnation{Incarnation: 1, State: provision.SlotLive, Account: "bab", App: "blog", Intent: "create other"}
+ writeSlot(t, l.fx.rel, ltOne, other)
+ // fresh: a fieldless record beside a retired slot of its own key.
+ legacyBody(t, l.fx.rel, ltApp, ltFresh, "stable")
+ retired := provision.HostIncarnation{Incarnation: 3, State: provision.SlotRetired, Account: "bab", App: ltApp, Intent: "retire earlier"}
+ writeSlot(t, l.fx.rel, ltFresh, retired)
+ // old: a legacy record with no slot, which the pass migrates.
+ legacyBody(t, l.fx.rel, ltApp, ltOld, "stable")
+ one, oneTag := rawOf(t, l.fx.rel, ltApp, ltOne)
+ fresh, freshTag := rawOf(t, l.fx.rel, ltApp, ltFresh)
+ if !l.fx.srv.incarnationMigrationIteration(context.Background()) {
+ t.Fatal("the migration pass did not run with lifetimes on")
+ }
+ for host, want := range map[string]struct {
+ raw []byte
+ tag string
+ slot provision.HostIncarnation
+ }{ltOne: {one, oneTag, other}, ltFresh: {fresh, freshTag, retired}} {
+ if rows := migRows(t, l.fx.st, host); len(rows) != 0 {
+ t.Fatalf("%s: migrations = %+v, want none", host, rows)
+ }
+ if raw, tag := rawOf(t, l.fx.rel, ltApp, host); !bytes.Equal(raw, want.raw) || tag != want.tag {
+ t.Fatalf("%s: record = %s, want it untouched (not even fenced)", host, raw)
+ }
+ if slot, _ := slotOf(t, l.fx.rel, host); slot == nil || *slot != want.slot {
+ t.Fatalf("%s: slot = %+v, want %+v untouched", host, slot, want.slot)
+ }
+ }
+ row := oneMig(t, l.fx.st, ltOld, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, ltOld, row.OpID)
+}
+
+// ---- the judge agrees -----------------------------------------------------------
+
+func TestJudgeKeepsAMarkedRecordUnderItsMigrationSlot(t *testing.T) {
+ l := newLtOn(t)
+ 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 {
+ t.Helper()
+ obs := judge.Observe(context.Background(), l.fx.rel, "bab", ltApp, ltOne)
+ if obs.Verdict != want {
+ t.Fatalf("%s: the judge says %v (%v), want %v", what, obs.Verdict, obs.Err, want)
+ }
+ return obs
+ }
+ sgate, shook := pauseNth(t, "put", slotKey, 1)
+ fgate, fhook := pauseNth(t, "put", key, 2)
+ l.bucket.setHook(chain(shook, fhook))
+ mig := migrateAsync(t, l.fx.srv, ltOne, "")
+
+ 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)
+ }
+ op := obs.Rec.IncarnationMigrating
+
+ sgate.open()
+ waitFor(t, fgate.entered, "the final write")
+ obs = observe(provision.LifetimeValid, "marked, under its own migration slot")
+ if obs.Slot == nil || obs.Slot.Intent != "migration "+op || obs.Slot.RecordETag != obs.ETag {
+ t.Fatalf("observation = %+v, want the migration's slot bound to the marked record", obs)
+ }
+ // The same marked record under a slot of another operation, or under
+ // its own slot bound to another body, is stale.
+ another := *obs.Slot
+ another.Intent = "migration other"
+ if v := provision.LifetimeOf(obs.Rec, obs.ETag, &another); v != provision.LifetimeStale {
+ t.Fatalf("marked under another operation's slot: %v, want stale", v)
+ }
+ elsewhere := *obs.Slot
+ elsewhere.RecordETag = "another-body"
+ if v := provision.LifetimeOf(obs.Rec, obs.ETag, &elsewhere); v != provision.LifetimeStale {
+ t.Fatalf("marked under its slot bound to another body: %v, want stale", v)
+ }
+
+ fgate.open()
+ if r := waitMig(t, mig); r.err != nil || r.state != lifetimeCommitted {
+ t.Fatalf("migrate = %q %v, want committed", r.state, r.err)
+ }
+ obs = observe(provision.LifetimeValid, "the final record under the slot")
+ if obs.Rec.Incarnation != 1 || obs.Rec.IncarnationMigrating != "" {
+ t.Fatalf("observation = %+v, want incarnation 1 and no marker", obs)
+ }
+}
+
+func TestMigrationTakesTheAppLockThenTheHostname(t *testing.T) {
+ l := newLtOn(t)
+ l.fx.srv.cfg.LifetimesMigrate = true
+ c := installLockChecker(t)
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ if !l.fx.srv.incarnationMigrationIteration(context.Background()) {
+ t.Fatal("the migration pass did not run with lifetimes on")
+ }
+ if got := c.taken(""); len(got) < 2 || got[0] != "app bab/"+ltApp || got[1] != "host "+ltOne {
+ t.Fatalf("the pass took %v, want the app lock then the hostname", got)
+ }
+ oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ legacyBody(t, l.fx.rel, ltApp, ltOld, "stable")
+ if err := l.fx.st.BeginMigration(Migration{OpID: migOp, Account: "bab", App: ltApp, Host: ltOld, State: lifetimePending}); err != nil {
+ t.Fatal(err)
+ }
+ recoverOn(t, l.fx.srv)
+ if got := c.taken("host " + ltOld); len(got) != 1 {
+ t.Fatalf("recovery took the hostname %v, want once", got)
+ }
+ oneMig(t, l.fx.st, ltOld, lifetimeCommitted)
+ c.check(t)
+}
+
+func TestMigrationWaitsForAnotherPendingOperation(t *testing.T) {
+ l := newLtOn(t)
+ l.fx.srv.cfg.LifetimesMigrate = true
+ legacyBody(t, l.fx.rel, ltApp, ltOne, "stable")
+ one, _ := rawOf(t, l.fx.rel, ltApp, ltOne)
+ if err := l.fx.st.BeginRetirement(Retirement{OpID: "other", Kind: retireNormal, Account: "bab", App: ltApp, Host: ltOne, Incarnation: 1}); err != nil {
+ t.Fatal(err)
+ }
+ // The caller's form answers settling and persists nothing.
+ if r := migrateOn(t, l.fx.srv, ltOne, ""); !errors.Is(r.err, errLifetimeSettling) || r.state != "" {
+ t.Fatalf("migrate = %q %v, want settling", r.state, r.err)
+ }
+ // The pass leaves the hostname alone; a pending migration waits.
+ l.fx.srv.incarnationMigrationIteration(context.Background())
+ if rows := migRows(t, l.fx.st, ltOne); len(rows) != 0 {
+ t.Fatalf("migrations = %+v, want none while another operation is pending", rows)
+ }
+ beginMig(t, l, migOp, "")
+ recoverOn(t, l.fx.srv)
+ oneMig(t, l.fx.st, ltOne, lifetimePending)
+ if raw, _ := rawOf(t, l.fx.rel, ltApp, ltOne); !bytes.Equal(raw, one) {
+ t.Fatalf("record = %s, want it untouched while another operation is pending", raw)
+ }
+ if err := l.fx.st.SetRetirementState("other", lifetimeVoid); err != nil {
+ t.Fatal(err)
+ }
+ recoverOn(t, l.fx.srv)
+ oneMig(t, l.fx.st, ltOne, lifetimeCommitted)
+ requireMigrated(t, l.fx.rel, ltOne, migOp)
+}
diff --git a/internal/console/lifetimes_recovery.go b/internal/console/lifetimes_recovery.go
index 72c076941..cedd14323 100644
--- a/internal/console/lifetimes_recovery.go
+++ b/internal/console/lifetimes_recovery.go
@@ -47,9 +47,9 @@ func (s *Server) lifetimeRecoveryIteration(ctx context.Context) bool {
}
// recoverLifetimes is one pass: the orphan retirements (those cancelling an
-// allocation first), then the allocations. Task 4 adds the other retirement
-// kinds and Task 5 the migrations, in that order. An error is a listing that
-// failed; a row that could not be settled stays pending for the next pass.
+// 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.
func (s *Server) recoverLifetimes(ctx context.Context) error {
ctx = provision.WithLifetimes(ctx, s.judge)
rets, err := s.store.PendingRetirementsCancellingFirst()
@@ -78,6 +78,17 @@ func (s *Server) recoverLifetimes(ctx context.Context) error {
return nil
})
}
+ migs, err := s.store.PendingMigrations()
+ if err != nil {
+ return fmt.Errorf("pending migrations: %w", err)
+ }
+ for _, m := range migs {
+ m := m
+ _ = s.withAppHost(m.Account, m.App, m.Host, func() error {
+ s.recoverMigrationLocked(ctx, m)
+ return nil
+ })
+ }
return nil
}
diff --git a/internal/console/server.go b/internal/console/server.go
index f8e94f6b3..e24151ad9 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -733,6 +733,12 @@ type Config struct {
// every registry has been scanned clean (the rollout steps). The expiry
// pass runs either way, so apps created while it was on still expire.
TemporaryApps bool
+ // LifetimesMigrate turns on the incarnation migration pass
+ // (CARLOS_LIFETIMES_MIGRATE=1, lifetimes_migrate.go): fieldless records
+ // move into slot-governed lifetimes. Off by default, and switched on
+ // only once every box of the deployment runs a binary that refuses to
+ // write over a migration's marker (boxes roll before the console).
+ LifetimesMigrate bool
// StoreOperators are the addresses allowed to drive the GRANT door, and
// only that one (CARLOS_STORE_OPERATORS). Its own list rather than
// console ownership on purpose: granting CREATES backend estate and
diff --git a/internal/console/testdata/claim-routes.txt b/internal/console/testdata/claim-routes.txt
index 65b2dc9cb..729a4bb0c 100644
--- a/internal/console/testdata/claim-routes.txt
+++ b/internal/console/testdata/claim-routes.txt
@@ -292,6 +292,7 @@ worker StartEconomicsPass
worker StartExpiryPass
worker StartGeoipJobs
worker StartHostReports
+worker StartIncarnationMigrationPass
worker StartInfraPass
worker StartLifetimeRecoveryPass
worker StartLifetimesCutoverWatch
diff --git a/internal/provision/lifetime_migration.go b/internal/provision/lifetime_migration.go
new file mode 100644
index 000000000..30f0fca01
--- /dev/null
+++ b/internal/provision/lifetime_migration.go
@@ -0,0 +1,125 @@
+package provision
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "strconv"
+ "strings"
+ "sync/atomic"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The lazy migration of a fieldless record into a slot-governed lifetime
+// (self-serve moves plan 1, Task 5). The console marks the record by CAS
+// (incarnation_migrating = the operation id), writes the migration slot
+// bound to the marked ETag, then rewrites the record at incarnation 1 with
+// the marker removed. While the marker stands, WriteRecord refuses every
+// writer that does not carry the marker's own id, so no update can land
+// between the marker and the final write and unbind the slot from the body
+// it vouches for.
+
+// ErrRecordMigrating is WriteRecord's refusal of a marked record to a writer
+// that is not its migration. It is a CAS conflict: the caller reads again.
+var ErrRecordMigrating = fmt.Errorf("%w: the record is being migrated; read it again", release.ErrCASConflict)
+
+type recordMigrationKey struct{}
+
+// WithMigration lets WriteRecord rewrite a record marked with id. Only the
+// migration that wrote the marker, and a fence that rewrites the exact bytes
+// it read, carry it.
+func WithMigration(ctx context.Context, id string) context.Context {
+ return context.WithValue(ctx, recordMigrationKey{}, id)
+}
+
+func migrationFrom(ctx context.Context) string {
+ id, _ := ctx.Value(recordMigrationKey{}).(string)
+ return id
+}
+
+// checkRecordMarker refuses a write over before's fields when they carry a
+// marker other than the one ctx's migration owns. Every spelling of the
+// field counts, since DecodeRecord matches names without case.
+func checkRecordMarker(ctx context.Context, fields map[string]json.RawMessage) error {
+ own := migrationFrom(ctx)
+ for name, value := range fields {
+ if !strings.EqualFold(name, "incarnation_migrating") {
+ continue
+ }
+ var marker *string
+ if err := json.Unmarshal(value, &marker); err != nil {
+ return ErrRecordSequence
+ }
+ if marker != nil && *marker != "" && *marker != own {
+ return ErrRecordMigrating
+ }
+ }
+ return nil
+}
+
+// MarkMigrating is raw with incarnation_migrating set to id, every other
+// field kept as read. Write it with WriteRecord against raw.
+func MarkMigrating(raw []byte, id string) ([]byte, error) {
+ if id == "" {
+ return nil, errors.New("provision: a migration marker needs an id")
+ }
+ fields, _, err := RecordSequence(raw)
+ if err != nil {
+ return nil, err
+ }
+ value, err := json.Marshal(id)
+ if err != nil {
+ return nil, err
+ }
+ fields["incarnation_migrating"] = value
+ return json.Marshal(fields)
+}
+
+// FinishMigration is a marked record's body at incarnation 1 with the marker
+// removed, every other field kept as read. Write it with WriteRecord (under
+// WithMigration) against marked.
+func FinishMigration(marked []byte) ([]byte, error) {
+ fields, _, err := RecordSequence(marked)
+ if err != nil {
+ return nil, err
+ }
+ delete(fields, "incarnation_migrating")
+ fields["incarnation"] = json.RawMessage(strconv.FormatUint(1, 10))
+ return json.Marshal(fields)
+}
+
+// recordMigrationWait bounds how long an ordinary update waits for a marked
+// record's migration to finish; recordMigrationPoll is how often it reads
+// the record again meanwhile.
+const (
+ recordMigrationWait = 30 * time.Second
+ recordMigrationPoll = time.Second
+)
+
+// migrationWaitForTest and migrationPollForTest, when positive, replace them.
+var migrationWaitForTest, migrationPollForTest atomic.Int64
+
+func migrationWait() (wait, poll time.Duration) {
+ wait, poll = recordMigrationWait, recordMigrationPoll
+ if v := migrationWaitForTest.Load(); v > 0 {
+ wait = time.Duration(v)
+ }
+ if v := migrationPollForTest.Load(); v > 0 {
+ poll = time.Duration(v)
+ }
+ return wait, poll
+}
+
+// SetRecordMigrationWaitForTest shortens the updaters' wait for a marked
+// record and returns the restore. A TEST SEAM.
+func SetRecordMigrationWaitForTest(wait, poll time.Duration) (restore func()) {
+ oldWait := migrationWaitForTest.Swap(int64(wait))
+ oldPoll := migrationPollForTest.Swap(int64(poll))
+ return func() {
+ migrationWaitForTest.Store(oldWait)
+ migrationPollForTest.Store(oldPoll)
+ }
+}
diff --git a/internal/provision/plan14_root_reads_test.go b/internal/provision/plan14_root_reads_test.go
index f4b46b3e6..078120ee4 100644
--- a/internal/provision/plan14_root_reads_test.go
+++ b/internal/provision/plan14_root_reads_test.go
@@ -1269,8 +1269,9 @@ func TestPlan14RootReadCensus(t *testing.T) {
// Task 0 (the lifetimes cutover) re-pins four accepted-legacy console
// units, which moves the seal; still rev3=PROPOSED-self-serve-moves-1.
// Task 3 (the create protocol) re-pins serve and Server.hostHolder and
- // adds newLifetimeOpID, which moves it again.
- seal = "a11aea6699d2f113c315e8cacb6d2a6708395db567db08f74f215191f8958cd0"
+ // adds newLifetimeOpID, which moves it again. Task 5 (the lazy migration)
+ // re-pins serve, which starts the migration pass.
+ seal = "059b3cb0a336b314981b205d529d0205849751e7e0f1b1c4c10ede3430417797"
errs := append(compareRootReads(got, inv), checkRootReadLegacy(inv, seal)...)
errs = append(errs, checkRootReadDeclarations(got, inv)...)
if len(errs) > 0 {
diff --git a/internal/provision/provision.go b/internal/provision/provision.go
index 7629d53c9..a6fbc0e26 100644
--- a/internal/provision/provision.go
+++ b/internal/provision/provision.go
@@ -1029,11 +1029,27 @@ func updateRecordSeeding(ctx context.Context, s release.Store, acct, app, host s
rec.RecordSeq = seq
return &rec, nil
}
+ // One retry on a lost race; a second loss surfaces. A record a migration
+ // has marked refuses the write until the migration's final write: it is
+ // read again every poll, up to the wait, without spending that retry.
+ wait, poll := migrationWait()
+ deadline := time.Now().Add(wait)
+ retried := false
rec, err := update()
- if errors.Is(err, release.ErrCASConflict) {
- return update() // one retry on a lost race; a second loss surfaces
+ for {
+ switch {
+ case errors.Is(err, ErrRecordMigrating):
+ if !time.Now().Before(deadline) {
+ return rec, err
+ }
+ time.Sleep(poll)
+ case errors.Is(err, release.ErrCASConflict) && !retried:
+ retried = true
+ default:
+ return rec, err
+ }
+ rec, err = update()
}
- return rec, err
}
// MoveState is the record's move journal (move-any-app design §7.2). The
diff --git a/internal/provision/record_sequence.go b/internal/provision/record_sequence.go
index c2366bc2f..b859cfc2e 100644
--- a/internal/provision/record_sequence.go
+++ b/internal/provision/record_sequence.go
@@ -114,6 +114,9 @@ type preparedRecord struct {
// If-Match read ({} for create). body preserves that read's counter. Never
// call it with a synthesized before to reset a live record's sequence.
// Legacy absent counters and no claim context retain the exact body bytes.
+// A before marked by a migration (incarnation_migrating) is refused with
+// ErrRecordMigrating unless ctx carries WithMigration of that marker; a
+// create-only write reads no before and is unaffected.
func WriteRecord(ctx context.Context, s release.Store, key string, before, body []byte, etag string) (string, uint64, error) {
if !IsRecordKey(key) {
return "", 0, ErrRecordSequence
@@ -122,6 +125,9 @@ func WriteRecord(ctx context.Context, s release.Store, key string, before, body
if err != nil {
return "", 0, err
}
+ if err := checkRecordMarker(ctx, fields); err != nil {
+ return "", 0, err
+ }
_, candidate, err := RecordSequence(body)
if err != nil {
return "", 0, err
diff --git a/internal/provision/record_sequence_test.go b/internal/provision/record_sequence_test.go
new file mode 100644
index 000000000..b69e952d9
--- /dev/null
+++ b/internal/provision/record_sequence_test.go
@@ -0,0 +1,163 @@
+package provision
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The migration marker in the write primitive (self-serve moves plan 1,
+// Task 5): a record carrying incarnation_migrating refuses every writer that
+// does not carry the marker's own id.
+
+const markHost = "shop.example.com"
+
+// markedFixture is a legacy record of bab/shop at shop.example.com marked by
+// migration m1 through WriteRecord, as the console's step (1) writes it.
+func markedFixture(t *testing.T) (release.Store, []byte, string) {
+ t.Helper()
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ rec := Record{Schema: 1, Host: markHost, Slug: "shop", App: "shop", Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC()}
+ if err := CreateRecord(ctx, s, rec); err != nil {
+ t.Fatal(err)
+ }
+ key := RecordKey("bab", "shop", markHost)
+ raw, etag, err := s.Get(ctx, key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ marked, err := MarkMigrating(raw, "m1")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := WriteRecord(WithMigration(WithRecordSequence(ctx), "m1"), s, key, raw, marked, etag); err != nil {
+ t.Fatalf("the marker write: %v", err)
+ }
+ raw, etag, err = s.Get(ctx, key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got := mustDecode(t, raw); got.IncarnationMigrating != "m1" || got.RecordSeq != 1 {
+ t.Fatalf("marked record = %s, want marker m1 with record_seq 1", raw)
+ }
+ return s, raw, etag
+}
+
+func mustDecode(t *testing.T, raw []byte) Record {
+ t.Helper()
+ rec, err := DecodeRecord(raw)
+ if err != nil {
+ t.Fatalf("decode %s: %v", raw, err)
+ }
+ return rec
+}
+
+// shortMigrationWait bounds updateRecordSeeding's wait for a test.
+func shortMigrationWait(t *testing.T, wait time.Duration) {
+ t.Helper()
+ t.Cleanup(SetRecordMigrationWaitForTest(wait, wait/10))
+}
+
+func TestWriteRecordRefusesAMarkedRecordWithoutTheMigrationContext(t *testing.T) {
+ s, raw, etag := markedFixture(t)
+ ctx := context.Background()
+ key := RecordKey("bab", "shop", markHost)
+ body := bytes.Replace(raw, []byte(`"channel":"stable"`), []byte(`"channel":"beta"`), 1)
+ for name, wctx := range map[string]context.Context{
+ "no context": ctx,
+ "the sequence alone": WithRecordSequence(ctx),
+ "another migration": WithMigration(WithRecordSequence(ctx), "m2"),
+ "an empty migration": WithMigration(ctx, ""),
+ "under a judge": WithLifetimes(ctx, NewLifetimeJudge()),
+ "a near id": WithMigration(ctx, "m1-not"),
+ } {
+ _, _, err := WriteRecord(wctx, s, key, raw, body, etag)
+ if !errors.Is(err, ErrRecordMigrating) || !errors.Is(err, release.ErrCASConflict) {
+ t.Fatalf("%s: WriteRecord = %v, want ErrRecordMigrating, which is a CAS conflict", name, err)
+ }
+ }
+ // The ordinary updaters wait for the marker, then give up with it.
+ shortMigrationWait(t, 50*time.Millisecond)
+ start := time.Now()
+ err := SetRecordRegion(WithLifetimes(ctx, NewLifetimeJudge()), s, "bab", "shop", markHost, "uk")
+ if !errors.Is(err, ErrRecordMigrating) {
+ t.Fatalf("SetRecordRegion = %v, want ErrRecordMigrating after the wait", err)
+ }
+ if waited := time.Since(start); waited < 50*time.Millisecond {
+ t.Fatalf("SetRecordRegion gave up after %v, before the wait", waited)
+ }
+ if now, tag, _ := s.Get(ctx, key); !bytes.Equal(now, raw) || tag != etag {
+ t.Fatalf("record = %s, want the marked record untouched", now)
+ }
+}
+
+func TestWriteRecordAllowsTheMigrationsOwnWrite(t *testing.T) {
+ s, raw, etag := markedFixture(t)
+ ctx := WithMigration(WithRecordSequence(context.Background()), "m1")
+ key := RecordKey("bab", "shop", markHost)
+ final, err := FinishMigration(raw)
+ if err != nil {
+ t.Fatal(err)
+ }
+ tag, seq, err := WriteRecord(ctx, s, key, raw, final, etag)
+ if err != nil || seq != 2 {
+ t.Fatalf("the migration's own write = %d %v, want record_seq 2", seq, err)
+ }
+ now, cur, _ := s.Get(context.Background(), key)
+ got := mustDecode(t, now)
+ if cur != tag || got.Incarnation != 1 || got.IncarnationMigrating != "" || got.RecordSeq != 2 || got.Channel != "stable" {
+ t.Fatalf("record = %s, want incarnation 1, no marker, record_seq 2, the rest kept", now)
+ }
+ // Once the marker is gone every writer is admitted again.
+ if err := SetRecordRegion(context.Background(), s, "bab", "shop", markHost, "uk"); err != nil {
+ t.Fatalf("an update after the migration = %v", err)
+ }
+}
+
+func TestCreateOnlyWriteIgnoresTheMarker(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ key := RecordKey("bab", "shop", markHost)
+ rec := Record{Schema: 1, Host: markHost, Slug: "shop", App: "shop", Account: "bab", Channel: "stable", IncarnationMigrating: "m1", CreatedAt: time.Now().UTC()}
+ body, _ := json.Marshal(rec)
+ if _, _, err := WriteRecord(ctx, s, key, []byte(`{}`), body, ""); err != nil {
+ t.Fatalf("a create-only write of a marked body = %v, want it written", err)
+ }
+ // A create over a marked record is the plain create conflict, not the
+ // marker's refusal.
+ _, _, err := WriteRecord(ctx, s, key, []byte(`{}`), body, "")
+ if !errors.Is(err, release.ErrCASConflict) || errors.Is(err, ErrRecordMigrating) {
+ t.Fatalf("a create over a marked record = %v, want the plain CAS conflict", err)
+ }
+}
+
+func TestClaimRawUpdateRecordSurfacesTheMarker(t *testing.T) {
+ for _, judged := range []bool{false, true} {
+ s, raw, etag := markedFixture(t)
+ ctx := context.Background()
+ if judged {
+ ctx = WithLifetimes(ctx, NewLifetimeJudge())
+ }
+ // A long wait: the whole-record update must not take it.
+ shortMigrationWait(t, time.Minute)
+ rec := mustDecode(t, raw)
+ rec.Channel = "beta"
+ start := time.Now()
+ err := UpdateRecord(ctx, s, rec, etag)
+ if !errors.Is(err, ErrRecordMigrating) || !errors.Is(err, release.ErrCASConflict) {
+ t.Fatalf("judged %v: UpdateRecord = %v, want ErrRecordMigrating as it comes", judged, err)
+ }
+ if waited := time.Since(start); waited > 5*time.Second {
+ t.Fatalf("judged %v: UpdateRecord waited %v; it returns the marker at once", judged, waited)
+ }
+ if now, _, _ := s.Get(context.Background(), RecordKey("bab", "shop", markHost)); !bytes.Equal(now, raw) {
+ t.Fatalf("judged %v: record = %s, want it untouched", judged, now)
+ }
+ }
+}
diff --git a/internal/provision/self_serve_moves1_census_test.go b/internal/provision/self_serve_moves1_census_test.go
index 3c00fb2d6..bfc2b9d36 100644
--- a/internal/provision/self_serve_moves1_census_test.go
+++ b/internal/provision/self_serve_moves1_census_test.go
@@ -78,6 +78,33 @@ package provision
// before; a refusal from the attach core is answered through
// writeDoorRefusalJSON, so the app gone under the lifetime create's lock
// is the family's http.NotFound. No write site.
+//
+// Task 5 (the lazy migration), reviewed dispositions (all PROPOSED):
+// - record_sequence.go:WriteRecord: record-primitive as before; it now
+// refuses a before marked with incarnation_migrating (ErrRecordMigrating,
+// a CAS conflict) unless ctx carries WithMigration of that marker. A
+// refusal only: no Put added, a create-only write is unaffected.
+// - provision.go:updateRecordSeeding: record-writer as before through
+// WriteRecord; an ErrRecordMigrating answer is read again every second up
+// to 30 s, without spending the one CAS retry. No write site added.
+// - lifetimes_create.go:Server.fenceRecord: record-writer as before; over a
+// marker of another migration (which admission proves is not pending) the
+// exact bytes are rewritten under WithMigration of that marker.
+// - lifetimes_migrate.go:Server.migrationMarkLocked: record-writer; step
+// (1): the observed bytes with incarnation_migrating set, one WriteRecord
+// by CAS from the observed ETag, sequence on, under its own marker id.
+// - lifetimes_migrate.go:Server.migrationFinishLocked: record-writer; step
+// (3): the marked bytes at incarnation 1 with the marker removed, one
+// WriteRecord by CAS from the marked ETag, under its own marker id. Step
+// (2) is WriteHostIncarnation, nonrecord.
+// - lifetimes_migrate.go:Server.migrationStepsLocked: key-site; the
+// dispatcher; its only bucket write is DeleteIfMatch of a corrupt body on
+// the observed ETag after quarantineCopy, never a plain delete; no Put.
+// - lifetimes_migrate.go:Server.migrateFieldlessRecords: key-site; the pass
+// lists judged records and orders them; it writes nothing itself.
+// - lifetimes_migrate.go:Server.quarantineCopy: nonrecord; one create-only
+// Put of control/quarantine/<account>/<app>/<host>.<unix nanos>.json; the
+// record key is not touched.
func plan14SelfServeMoves1WriteCensusInventory() []censusEntry {
removed := []struct{ file, unit string }{
{"internal/provision/provision.go", "GetRecordETag"},
@@ -94,9 +121,10 @@ func plan14SelfServeMoves1WriteCensusInventory() []censusEntry {
{"internal/console/apiinstanceimport.go", "Server.beginImport"},
{"internal/console/apiinstanceseed.go", "Server.beginFrom"},
{"internal/console/apidomains.go", "Server.handleAPIDomainAttach"},
+ {"internal/provision/record_sequence.go", "WriteRecord"},
}
added := []censusEntry{
- {"internal/provision/provision.go", "updateRecordSeeding", "record-writer", "key:RecordKey,key:WriteRecord", "20e3b0498fb8d410c3d10bc7c788b0eec137cf58aa4dfa2afb42da837ebd71ea"},
+ {"internal/provision/provision.go", "updateRecordSeeding", "record-writer", "key:RecordKey,key:WriteRecord", "382b752cd9455453d2cab5bd3eed75f3ce87c393e64d993e34113d8dcecc037e"},
{"internal/provision/provision.go", "claimRawUpdateRecord", "record-writer", "key:RecordKey,key:WriteRecord,key:claimRawUpdateRecord", "782c1020ceea3661d972715f2ce2500119ebd56eb8e15f89c7a28d79d83758ba"},
{"internal/provision/rootrecord.go", "IndexRecords", "key-site", "key:ParseRecordKey,lit:instances", "e44ac431865bf38baa4105f66c9cb23014cc3c7ffcd41d5385244cc76fc4c0b9"},
{"internal/provision/lifetime_judge.go", "ListRecordsCompleteJudged", "key-site", "key:RecordsPrefix", "d851c7d8d04be82fbda88e257bc02825aff67c41b8a1f029f1a8c49f6fe4516c"},
@@ -114,10 +142,16 @@ func plan14SelfServeMoves1WriteCensusInventory() []censusEntry {
{"internal/console/domainattach.go", "Server.attachDomainAs", "record-writer", "key:CreateRecord", "205df195585b26df17f51b5d9ac5006a308104612df794d91208a6a040bfaf59"},
{"internal/console/apiinstanceseed.go", "Server.createSeedRecord", "record-writer", "key:CreateRecord", "a6fabf99b13e7e2302dbf2c30fb7c782f76f787141a5cc12580fde59805dfd4a"},
{"internal/console/lifetimes_create.go", "Server.allocRecordLocked", "record-writer", "key:RecordKey,key:WriteRecord", "e04927e2e52a0362da469199c04735fa3bfd926c1c898e63b4370a8d79f46daa"},
- {"internal/console/lifetimes_create.go", "Server.fenceRecord", "record-writer", "key:RecordKey,key:WriteRecord", "33d320aac193a0878b668054ea3b0d6872665c86851efa01d9e0d2532a902162"},
+ {"internal/console/lifetimes_create.go", "Server.fenceRecord", "record-writer", "key:RecordKey,key:WriteRecord", "748c966a35296b1c53be72b2fdaf7e02ae0a16b1ff24fa6aea53bfa22d4720b5"},
{"internal/console/lifetimes_create.go", "Server.allocCleanLocked", "key-site", "key:RecordKey", "035cf56d8f69517b66dc8ea6f9121d463e482d7dd05daf2b0c494aef1255f522"},
{"internal/console/lifetimes_create.go", "Server.deleteUnderRetiredSlot", "key-site", "key:RecordKey", "d0c8412a40e6f00d5db66e301fd108a7d5cddbfbe6dd80a93f02ac31f7cd24a0"},
{"internal/console/apidomains.go", "Server.handleAPIDomainAttach", "key-site", "lit:instances", "44016e4665cafc4d6d36ca8b4b9d01213978b8ca5041a2e83d9b93a397866b3d"},
+ {"internal/provision/record_sequence.go", "WriteRecord", "record-primitive", "key:IsRecordKey,key:WriteRecord,key:preparedRecordKey,sel:Put", "c55b4e8d27701ec80aeb32a8bd6b7e0b66cc9d8b9c14b252c32a80b977501998"},
+ {"internal/console/lifetimes_migrate.go", "Server.migrationMarkLocked", "record-writer", "key:RecordKey,key:WriteRecord", "9e39d73d7b51017fe0499eaf6ea3c8a7f9bbe77e2d76cebe3d6a3c72da449274"},
+ {"internal/console/lifetimes_migrate.go", "Server.migrationFinishLocked", "record-writer", "key:RecordKey,key:WriteRecord", "6729c5e00f700ccce6432586baa32859da0d3d03079b9fff4c0ddf874fdf6fbb"},
+ {"internal/console/lifetimes_migrate.go", "Server.migrationStepsLocked", "key-site", "key:RecordKey", "42fb3ed91659f4c38d1ea1fddd2c05fa165387d205fef578c83c6a64858b0b79"},
+ {"internal/console/lifetimes_migrate.go", "Server.migrateFieldlessRecords", "key-site", "key:RecordKey", "ede10c618c1a7030c5e054767d3fd4f1c35a4cb3db1e0c621883888b85215405"},
+ {"internal/console/lifetimes_migrate.go", "Server.quarantineCopy", "nonrecord", "sel:Put", "d4b7f4a8729b3e51789b5a53d8dd6cbc735f3a072314eafb02f92969a9818afb"},
}
entries := plan14ClaimRosterBindingWriteCensusInventory()
for _, r := range removed {
diff --git a/internal/provision/testdata/plan14-root-reads-self-serve-moves-1.tsv b/internal/provision/testdata/plan14-root-reads-self-serve-moves-1.tsv
index 2e9f39dc0..48d1e2c24 100644
--- a/internal/provision/testdata/plan14-root-reads-self-serve-moves-1.tsv
+++ b/internal/provision/testdata/plan14-root-reads-self-serve-moves-1.tsv
@@ -1,5 +1,5 @@
# Census overlay self-serve-moves-1, marked rev3=PROPOSED-self-serve-moves-1 (not an accepted ruling; docs/superpowers/notes/census-proposal-self-serve-moves-1.md). Task 1 adds the lifetime fields to provision.Record (incarnation, incarnation_migrating, move_seq, retired) and internal/provision/lifetime.go (the host incarnation slot, the lifetime verdict, the tombstone body, GetRecordRaw); GetRecordETag now hides a tombstone. Envelope only: no new root read, class or task, and the registry schema (@fields) is unchanged because Record is not a registry struct. internal/provision is re-pinned over the new code. The seal stands.
-@coverage internal/provision 5763793801381eae1dca8368722fd98acbe7bf95857a7bf2a70bad3d76f520e1
+@coverage internal/provision 051888ac12e2de70a70e2d044c85b263e54ea18332e818de99e1fc40195f617a
# Task 2 and fix round 1: rev3=PROPOSED-self-serve-moves-1 throughout; all dispositions below await the human census ruling. Console SQL and randomness run as carlos-console under console.service; no root decision is added. Existing legacy rows are re-frozen at this boundary, and the two constructors lose their own read sites to insertAppRow.
# The H3 removal syntax requires the census-ruling- prefix; PROPOSED in those tokens is deliberate and does not claim approval.
@legacy 296610cb0201ccb67b21152a50191b8b819f41e1d39fd43ca8b8d35cd75a2601 internal/console/lifetimes_store.go Store.BeginAllocation legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=console.service user carlos-console, J4 root-input: inserts one pending allocation in the console SQLite store, carrying the account, app and app lifetime; no bucket or root action file:Exec@132 7c7cb202595de014e2a549ebb001ae2c0afd800fc7959a6b5e145f5c1bc9c8ef
@@ -45,13 +45,13 @@ internal/console/store_gate.go claimSQLCensus legacy accepted-legacy accepted-le
@remove internal/console/tempstore.go Store.CreateTemporaryAppInRegion
@removed internal/console/tempstore.go Store.CreateTemporaryAppInRegion H3 census-ruling-PROPOSED-self-serve-moves-1;rev3=PROPOSED-self-serve-moves-1;basis=constructor remains declared but has no direct read; app insertion and lifetime minting moved to insertAppRow, classified above; proposed removal only
# Task 2 envelopes, rev3=PROPOSED-self-serve-moves-1: internal/console covers the lifetime store, app constructors, lifetime projection and table-scoped gate labels; internal/provision (above) adds only ErrLifetimeExhausted beyond Task 1. Registry fields are unchanged.
-@coverage internal/console 2bdf734eea83e7eac6706466754f6d84145ae04545315c6c785448a833a0906d
+@coverage internal/console 8e6ee043534c908f442d94e46a18628c440a3802c456660ea8177861874c983e
# Task 6a: rev3=PROPOSED-self-serve-moves-1; reviewed provision envelope now includes bounded record-then-slot observations, judged read/list/update gates and signed-counter exhaustion at slot read. No removal-loop, host or console change; registry fields and the legacy seal stand.
internal/provision/rootrecord.go IndexRecords authority B3,C rev3=PROPOSED-self-serve-moves-1;basis=retain every discovered candidate owner key per host for fresh lifetime judgement at Find; partial discovery still retries and no verdict or body is cached e44ac431865bf38baa4105f66c9cb23014cc3c7ffcd41d5385244cc76fc4c0b9
internal/provision/rootrecord.go RecordIndex.Find authority B3,C rev3=PROPOSED-self-serve-moves-1;basis=observe each candidate through the context judge, refuse corruption and read failures, require one valid owner and validate its schema and shape; nil-judge duplicates retain permission refusal 760eed59cbd6a8ddab1098ce3a5f7a4272ecda81c17cc079dc135aba3a85a8f8
# Task 0 (the lifetimes cutover), rev3=PROPOSED-self-serve-moves-1: four changed accepted-legacy console units re-pinned with reviewed bases (serve starts the cutover watch; NewServer adds the judge, the gate and its first marker read and moves claims-off app materialisation to the watch; the instance listing reads judged under an admitted-on unit; runFrom ends its handed-over unit); system.go runSystem and systemVerb re-pinned for the lifetimes verb at their existing authority; envelopes for ., cmd/console and internal/console (refreshed above) re-pinned. Four new legacy fingerprints need a new seal, checked after the preceding one. No root read, class or task is added; registry fields stand.
-@legacy db1053faf4844420750a6ae9ada80386f31cb1065f0960175212941bbd908229 cmd/console/main.go serve legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=the console (cmd/console serve under console.service, user carlos-console): Task 0 starts the lifetimes cutover watch and Task 3 the lifetime recovery pass beside the other claim workers; the same three data path literals move; no privileged reader is added data-literal@4077,data-literal@5178,data-literal@5415 115729fb838b72ec095cdbdb0ceedacb8f7021ac1f0bbca3f89f32ff5092a887
-cmd/console/main.go serve legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=the console (cmd/console serve under console.service, user carlos-console): Task 0 starts the lifetimes cutover watch and Task 3 the lifetime recovery pass beside the other claim workers; the same three data path literals move; no privileged reader is added data-literal@4077,data-literal@5178,data-literal@5415 115729fb838b72ec095cdbdb0ceedacb8f7021ac1f0bbca3f89f32ff5092a887
+@legacy 2a48fe52718c5debe01b7ac64bd72764ad2e33121ad5a56296f9e5f1964c77b9 cmd/console/main.go serve legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=the console (cmd/console serve under console.service, user carlos-console): Task 0 starts the lifetimes cutover watch, Task 3 the lifetime recovery pass and Task 5 the incarnation migration pass (off unless CARLOS_LIFETIMES_MIGRATE=1) beside the other claim workers; the same three data path literals move; no privileged reader is added data-literal@4077,data-literal@5178,data-literal@5415 6f914eeb97eebdd0b29486caa68496ee4f14ecc38c86b69419b5956c47f5a191
+cmd/console/main.go serve legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=the console (cmd/console serve under console.service, user carlos-console): Task 0 starts the lifetimes cutover watch, Task 3 the lifetime recovery pass and Task 5 the incarnation migration pass (off unless CARLOS_LIFETIMES_MIGRATE=1) beside the other claim workers; the same three data path literals move; no privileged reader is added data-literal@4077,data-literal@5178,data-literal@5415 6f914eeb97eebdd0b29486caa68496ee4f14ecc38c86b69419b5956c47f5a191
@legacy 0b23119c935fb39171bb2787d8725ca0eb7736c067eab08b07236df0a23c6596 internal/console/apiinstances.go Server.handleAPIInstanceList legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=API listing reads through ListRecordsCompleteJudged with the request judge, which is nil before the lifetimes cutover; the same SQLite query moves; no root settings decision occurs here file:Query@1830 ac702aaa6da4c5a6ed7bb69427ba03ec3f2ee7753cce57c58349a978bbdcfadf
internal/console/apiinstances.go Server.handleAPIInstanceList legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=API listing reads through ListRecordsCompleteJudged with the request judge, which is nil before the lifetimes cutover; the same SQLite query moves; no root settings decision occurs here file:Query@1830 ac702aaa6da4c5a6ed7bb69427ba03ec3f2ee7753cce57c58349a978bbdcfadf
@legacy 992541f409e55b7643b6d49ab5e583b87d7bb4ab23c7d86c60e89ebfa07287d6 internal/console/apiinstanceseed.go Server.runFrom legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=the detached half of the --from door, unchanged, now ends the request unit it was handed after its last write; the same scratch Stat moves; every failure still moves the record to aborted file:Stat@3142 0f0ed34b5a5c99cc7f8245f2e8d6284c300b4c7314b312af69f96d6d49f0d4c0
@@ -61,9 +61,10 @@ internal/console/server.go NewServer legacy accepted-legacy accepted-legacy;rev3
system.go runSystem authority B,C,D,F,G,J4 rev3=PROPOSED-self-serve-moves-1;identity=explicit operator CLI uid 0;basis=adds the lifetimes family to the system dispatch (carlos system lifetimes cutover, a console door call as the operator); every existing command branch unchanged 8276403c56bc08639e62dee3a348573707c6f5b6909c8b1423bdc21e900390c1
system.go systemVerb authority B,C,D,F,G,J4 rev3=PROPOSED-self-serve-moves-1;identity=explicit operator CLI uid 0;basis=adds the lifetimes family to the system dispatch (carlos system lifetimes cutover, a console door call as the operator); every existing command branch unchanged 5952f7a80803ed5279dea42b3b2166e2d2d93ddd9fbe2e13ac8b1c00044bdb70
@coverage . fc604d5a74c58e761759747a2a4b8f166c51576eb4c457c55370fa2d678f4bd1
-@coverage cmd/console dc88a1b2acc2da4c812cb9cc748409ae89798931bc59f5391edcb693b5ed9a52
+@coverage cmd/console 80fd8201d72d03f1e79579b2d29fa6354ba4b15d619d17be443ad650263c668c
# Task 3 (the create protocol and its recovery), rev3=PROPOSED-self-serve-moves-1: serve re-pinned in Task 0's rows above (it now also starts the lifetime recovery pass); Server.hostHolder re-pinned at its root-route disposition (the holder read is judged under a unit admitted on); newLifetimeOpID classified like mintLifetime (crypto/rand for an operation id). The cmd/console and internal/console envelopes (above) are re-pinned over the create protocol, its recovery, the door routing and the attach/detach exclusion. Three new legacy fingerprints move the seal. Fix round 1 re-pins the internal/console envelope over the per-door pending sentences, the exhausted-counter sentence, the neutral attach refusal and the attach door's 404. Registry fields stand.
@legacy 2230ae1fd590a6e3d76cf38fe3e8c308ed9b9ec8bf5c0875ca13db23d1ac6776 internal/console/apex.go Server.hostHolder legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=root-route: the registry route read is unchanged; the instance-record holder read under each other app is judged when the request was admitted with lifetimes on (a nil judge is GetRecord), so a body its slot no longer vouches for holds nothing registry:Account@363,registry:Account@383,registry:Account@454,registry:Account@474,registry:Account@524,registry:App@344,registry:App@440,whole:Route@330 5fb4c31472674d0f1d4232877d6dd860980381c9842ec0fea9084517034037f7
internal/console/apex.go Server.hostHolder legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=root-route: the registry route read is unchanged; the instance-record holder read under each other app is judged when the request was admitted with lifetimes on (a nil judge is GetRecord), so a body its slot no longer vouches for holds nothing registry:Account@363,registry:Account@383,registry:Account@454,registry:Account@474,registry:Account@524,registry:App@344,registry:App@440,whole:Route@330 5fb4c31472674d0f1d4232877d6dd860980381c9842ec0fea9084517034037f7
@legacy 44be6376317c5819fc902e851b039330aff78a8f2ddc5338e95054d863151c02 internal/console/lifetimes.go newLifetimeOpID legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=console.service user carlos-console, J4 root-input: reads crypto/rand for an independent 32-hex lifetime operation id (allocations and retirements); randomness grants no root authority file:Read@71 34364098aef13cf16819336874c65fe5aa38350ec6557cb0ab67fbc3596a0df3
internal/console/lifetimes.go newLifetimeOpID legacy accepted-legacy accepted-legacy;rev3=PROPOSED-self-serve-moves-1;basis=console.service user carlos-console, J4 root-input: reads crypto/rand for an independent 32-hex lifetime operation id (allocations and retirements); randomness grants no root authority file:Read@71 34364098aef13cf16819336874c65fe5aa38350ec6557cb0ab67fbc3596a0df3
+# Task 5 (the lazy migration), rev3=PROPOSED-self-serve-moves-1: serve re-pinned in Task 0's rows above (it also starts the incarnation migration pass, behind CARLOS_LIFETIMES_MIGRATE=1); the internal/provision (the marker refusal in WriteRecord, the updaters' bounded wait, the marker and final bodies), internal/console (the migration protocol, its recovery and pass, the fence over another migration's marker, the flag) and cmd/console envelopes re-pinned. One changed legacy fingerprint moves the seal. No root read, class or registry field is added.