amadan
moves: into-home witnesses through the claim runtime (home part 2, Task 4 fix round 1)
The production probe mode runs the claim runtime's real start order on either box: the production Stop of this box's own predecessor writer session, Admit, Prepare, Restore (the production cold restore, with the production claimArtifactRetirement so RetireSuperseded re-tags a retired installation's inventory), Reserve, Materialize, Failed on an error; stand-ins only for the launch and its readiness. The <db>.gen stand-in is gone. The source recoveries (clean rollback, dirty return, pre-flip revert, and TestRevertProbedAfterARealAdmission) go through the production recovery adapter into that runtime on ca-box; the hand-rolled writer admission is removed. The source is made what serving leaves first (sourceServed: installation record, inventory, src_gen's files manifest and origin). The clean rollback and the dirty return now fail two landing attempts; the second's restore retires the first's installation. Fixture identity: the recovery fixture's root stop inventory names the controller's own box, and its stop identity carries the home binding.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4 files changed,
+275
−241
internal/host/claimretryrecovery_test.go+7 −2internal/host/moveshome_target_test.go+125 −177internal/host/movesparks_test.go+126 −47internal/host/movesrevert_test.go+17 −15
CI cancelled — superseded by a newer push — run details
CI log
=== gate === cd launchctx && go vet ./... && go test ./... -count=1 ok amadan.net/carlos/platform.git/launchctx 0.011s ? amadan.net/carlos/platform.git/launchctx/controlname [no test files] ? amadan.net/carlos/platform.git/launchctx/wireid [no test files] go vet ./... gofmt -l .: clean go test ./... -count=1 -timeout 30m amadan: job cancelled — a newer push superseded it
diff --git a/internal/host/claimretryrecovery_test.go b/internal/host/claimretryrecovery_test.go| index 3226254bf..e849a7250 100644 |
| --- a/internal/host/claimretryrecovery_test.go |
| +++ b/internal/host/claimretryrecovery_test.go |
| @@ -142,6 +142,10 @@ type recoveryFixture struct { |
| // captures the writer. Unset, every admission's evidence passes. |
| homeRegion string |
| admissionEvidence func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error |
| + // fleetHomeRegion is the fleet's signed home designation the box's |
| + // actor carries; with homeRegion it is the controller's own stop |
| + // identity's home binding (park.VerifiedHome). "" for none. |
| + fleetHomeRegion string |
| } |
| // identity is the controller's box identity, the defaults filled in. |
| @@ -247,7 +251,8 @@ func (f *recoveryFixture) root(drain StopDrainer) ClaimStopCoordinator { |
| return StopInventory{}, err |
| } |
| h := o.Claim.Holder |
| - return StopInventory{Complete: true, Host: host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, Box: "box-1", Enrollment: "enrollment-1", DirectoryRevision: "7"}, nil |
| + box, enrollment, _, _ := f.identity() |
| + return StopInventory{Complete: true, Host: host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, Box: box, Enrollment: enrollment, DirectoryRevision: "7"}, nil |
| })} |
| } |
| @@ -260,7 +265,7 @@ func (f *recoveryFixture) controller(boot, controller string, pid int, start str |
| revision = 7 |
| } |
| box, enrollment, region, leaseRegion := f.identity() |
| - local := claimStopLocal{Actor: hostbox.Actor{Box: box, Enrollment: enrollment, Revision: revision, Region: region}, BootID: boot, ControllerID: controller, PID: pid, ProcessStart: start} |
| + local := claimStopLocal{Actor: hostbox.Actor{Box: box, Enrollment: enrollment, Revision: revision, Region: region, HomeRegion: f.homeRegion, FleetHomeRegion: f.fleetHomeRegion}, BootID: boot, ControllerID: controller, PID: pid, ProcessStart: start} |
| f.peer = StopPeer{UID: 1001, PID: pid, ProcessStart: start, BootID: boot, ControllerID: controller} |
| producers := &ClaimProducerSet{} |
| w := &ClaimWakeAssembly{LeaseRegion: leaseRegion, Inputs: recoveryInputs{f}, Manager: new(cgroup.Manager)} |
diff --git a/internal/host/moveshome_target_test.go b/internal/host/moveshome_target_test.go| index ee06c87f8..729bf8d99 100644 |
| --- a/internal/host/moveshome_target_test.go |
| +++ b/internal/host/moveshome_target_test.go |
| @@ -8,21 +8,19 @@ import ( |
| "os" |
| "path/filepath" |
| "sort" |
| + "strconv" |
| "strings" |
| "sync" |
| "testing" |
| "time" |
| "github.com/carlosframework/platform/internal/activator" |
| - "github.com/carlosframework/platform/internal/appconfig" |
| "github.com/carlosframework/platform/internal/genselect" |
| "github.com/carlosframework/platform/internal/hostactors" |
| "github.com/carlosframework/platform/internal/hostbox" |
| "github.com/carlosframework/platform/internal/hostclaim" |
| "github.com/carlosframework/platform/internal/hostfiles" |
| - "github.com/carlosframework/platform/internal/hostguard" |
| "github.com/carlosframework/platform/internal/hostops" |
| - "github.com/carlosframework/platform/internal/hostwrite" |
| "github.com/carlosframework/platform/internal/instancekey" |
| "github.com/carlosframework/platform/internal/lease" |
| "github.com/carlosframework/platform/internal/moves" |
| @@ -1409,135 +1407,6 @@ func intoHomeAbort(t *testing.T) ([]hostclaim.Claim, []string) { |
| return watch.since(0), leases.all() |
| } |
| -// prodRecovery is the named source's recovery wake through the production |
| -// recovery owners (movesRecoveryOwners, dispatched by movesProbeOwners on |
| -// the op's kind) over the harness's claim, records and buckets: the |
| -// activator's recovery probe entry as far as the claim runs on main. Each |
| -// wake selects the probe's inputs with the owners' Select and admits through |
| -// the claim's writer with the owners' AdmissionEvidence bound, as the claim |
| -// runtime's Admit does. The source is ca-box in ca, a named box: no home |
| -// alias, the fleet's designation only. The owners compare the launch |
| -// environment by the full local digest (localEnvDigest), the recovery's |
| -// same-box rule. |
| -type prodRecovery struct { |
| - h *phase1Harness |
| - owners *movesProbeOwners |
| - // observed, selected and admitted are each wake's claim before the |
| - // admission, its inputs and the claim after it. |
| - observed []hostclaim.Observed |
| - selected []ClaimLaunchInputs |
| - admitted []hostclaim.Observed |
| - // before, when set, runs after Select and before the admission. |
| - before func(o hostclaim.Observed, in ClaimLaunchInputs) |
| - // work, when set, is the start's selected work Materialize runs under. |
| - work hostwrite.Work |
| - seq int |
| -} |
| - |
| -func newProdRecovery(t *testing.T, h *phase1Harness, dataDir string, route func() registry.Route) *prodRecovery { |
| - t.Helper() |
| - key, err := moves.ReportKeyFromSecrets(h.cfg.SecretsKey) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - r := &prodRecovery{h: h} |
| - actor := h.roster[h.srcBox] |
| - filesDir := func(host string) (string, string) { |
| - return filepath.Join(dataDir, host), filepath.Join(dataDir, host+".env") |
| - } |
| - recorded := func() *provision.MoveInputs { |
| - m := h.read(t, p1Host).Move |
| - if m.Rollback != nil { |
| - return m.Rollback.RecoveryInputs |
| - } |
| - return m.RecoveryInputs |
| - } |
| - owners := &movesProbeOwners{HomeRegion: actor.HomeRegion, FleetHomeRegion: actor.FleetHomeRegion, Claims: h.s, Records: MovesRecordsFrom(h.s), |
| - ReadyPath: movesReadyPath(h.s), Data: h.vs, Bucket: h.srcBucket, Region: actor.Region, Box: actor.Box, DataDir: dataDir, FilesDir: filesDir, |
| - Markers: p1Markers{h.vs}, |
| - Route: func(host string) (*registry.Route, error) { rt := route(); return &rt, nil }, |
| - // What this box would launch: its own composition over its data |
| - // directory with its own region, the live release. |
| - Launch: func(_ context.Context, rt registry.Route) (ClaimServingLaunch, moves.EnvReport, error) { |
| - composed, supplied, _, err := (&activator.ExecBackend{DataDir: dataDir, Region: h.src}).LaunchLayers(rt) |
| - if err != nil { |
| - return ClaimServingLaunch{}, moves.EnvReport{}, err |
| - } |
| - delete(composed, "CARLOS_MOVE_PROBE") |
| - env := make([]string, 0, len(composed)) |
| - for k, v := range composed { |
| - env = append(env, k+"="+v) |
| - } |
| - sort.Strings(env) |
| - return ClaimServingLaunch{Executable: "/opt/carlos/live/" + p1Host, ExecutableSHA256: strings.Repeat("a", 64), Env: env, Version: "v7"}, moves.ReportLaunch(key, composed, supplied), nil |
| - }, |
| - EnvKey: key, |
| - Admitted: MovesRecordAdmittedAttempts, |
| - Works: func(context.Context, string, string, hostguard.Captured) (hostwrite.Work, error) { |
| - return p1Work{order: new([]string), name: "held"}, nil |
| - }, |
| - Render: func(_ context.Context, _ registry.Route, _ string, work func(appconfig.HeldRender) (hostwrite.Work, error)) (appconfig.HeldRender, error) { |
| - in := recorded() |
| - hr := appconfig.HeldRender{BundleETags: in.BundleETags, OverlayETag: in.OverlayETag} |
| - if _, err := work(hr); err != nil { |
| - return appconfig.HeldRender{}, err |
| - } |
| - return hr, nil |
| - }, |
| - Release: func(ctx context.Context, _ registry.Route, version string, work hostwrite.Work) error { |
| - return work.With(ctx, hostwrite.Identity{Release: version}, func() error { return nil }) |
| - }, |
| - NewID: func() string { r.seq++; return fmt.Sprintf("fresh%d", r.seq) }, |
| - Now: h.clock.Now} |
| - owners.Files = &movesFiles{Files: p1FilesStore{vs: h.vs, bucket: h.srcBucket}, Data: h.vs, Bucket: h.srcBucket, HomeRegion: actor.FleetHomeRegion, Region: actor.Region, Box: actor.Box, |
| - DataDir: dataDir, EnvFile: envFileOf(filesDir), Proof: ClaimOriginProof{Claims: h.s, HomeRegion: actor.FleetHomeRegion, Region: actor.Region}} |
| - if !owners.valid() { |
| - t.Fatal("fixture: the source's recovery owners are not valid") |
| - } |
| - r.owners = owners |
| - return r |
| -} |
| - |
| -// materializeWith makes each wake materialize the admitted inputs too, |
| -// under work, after the restore stand-in (the activator's restore rewrites |
| -// <dbPath>.gen to the admitted generation): the owners' held render and |
| -// release, the launch environment's full local digest, the start gate, the |
| -// C8 establishment and the start check. |
| -func (r *prodRecovery) materializeWith(work hostwrite.Work) { r.work = work } |
| - |
| -// wake is one recovery probe wake: Select, then the writer's admission. |
| -func (r *prodRecovery) wake(ctx context.Context, rt registry.Route, p activator.StartPolicy) error { |
| - h := r.h |
| - o, err := hostclaim.Read(ctx, h.s, p1Host) |
| - if err != nil { |
| - return err |
| - } |
| - in, err := r.owners.Select(ctx, rt, o, p) |
| - if err != nil { |
| - return err |
| - } |
| - r.observed, r.selected = append(r.observed, o), append(r.selected, in) |
| - if r.before != nil { |
| - r.before(o, in) |
| - } |
| - w := hostbox.Writer{Claims: h.s, Data: h.vs, Bucket: h.srcBucket, Region: r.owners.Region, Box: r.owners.Box, AdmissionEvidence: r.owners.AdmissionEvidence} |
| - admitted, err := w.Admit(ctx, o, in.Admission, in.Baseline) |
| - if err != nil { |
| - return err |
| - } |
| - r.admitted = append(r.admitted, admitted) |
| - if r.work == nil { |
| - return nil |
| - } |
| - gen := admitted.Claim.Holder.Generation |
| - if cur, err := hostbox.ReadGenerationMarker(rt.DBPath + ".gen"); err != nil || cur != gen { |
| - if err := hostbox.WriteGenerationMarker(rt.DBPath+".gen", gen); err != nil { |
| - return err |
| - } |
| - } |
| - return r.owners.Materialize(ctx, admitted, in, r.work) |
| -} |
| - |
| // TestMoveIntoHomeCleanRollback: the home target's probe, through the |
| // production probe owners, is admitted on home-box and fails before any |
| // write, until its attempts are spent. The rollback releases clean: |
| @@ -1564,26 +1433,26 @@ func intoHomeCleanRollback(t *testing.T) ([]hostclaim.Claim, []string) { |
| leases := &stagingLeaseKeys{} |
| leases.wrap(&h.cfg) |
| cleanHomeStart(t, firstMoving(t, watch.since(0))) |
| - // One attempt: the probe's prepared admission keeps the flip's stamp, |
| - // the stamp the target's installation is registered at. |
| + // Two attempts, each admitted through the production owners and the |
| + // claim runtime and failing before any write. The second is a fresh |
| + // generation with its own stamp: its restore retires the first's |
| + // installation and re-tags its inventory (RetireSuperseded). |
| if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, func(rec *provision.Record) error { |
| - rec.Move.CutAttempts = 1 |
| + rec.Move.CutAttempts = 2 |
| return nil |
| }); err != nil { |
| t.Fatal(err) |
| } |
| - // The probe through the production owners, admitted and failing before |
| - // any write. |
| p := h.productionProbe(t) |
| p.afterMaterialize = func(int) error { return errors.New("test: the probe failed") } |
| - for i := 0; i < 6 && h.read(t, p1Host).Move.State != provision.MoveRollingBack; i++ { |
| + for i := 0; i < 8 && h.read(t, p1Host).Move.State != provision.MoveRollingBack; i++ { |
| h.rollbackPass(t) |
| } |
| - if len(p.admissions) == 0 { |
| - t.Fatal("the probe was never admitted") |
| + if len(p.admissions) != 2 || p.stopped != 1 { |
| + t.Fatalf("the probe's attempts: admissions %+v, predecessors stopped %d", p.admissions, p.stopped) |
| } |
| failed := p.admissions[len(p.admissions)-1] |
| - if failed.Box != "home-box" || failed.Region != "ie" { |
| + if failed.Box != "home-box" || failed.Region != "ie" || failed.Generation == p1ID || failed.Predecessor != p1ID { |
| t.Fatalf("the home target's last admission: %+v", failed) |
| } |
| for i := 0; i < 4 && h.rb(t).TargetPhase != provision.RbTargetReleased; i++ { |
| @@ -1613,10 +1482,9 @@ func intoHomeCleanRollback(t *testing.T) ([]hostclaim.Claim, []string) { |
| // The source's recovery through the production recovery owners. Its |
| // bucket is versioned, as production's is: the harness stored the |
| // source's replica directly, so each object gets its version here. |
| - sourceMarker(t, h.phase1Harness) |
| srcGen := c.Op.Journal[hostclaim.JournalSrcGen] |
| - rec := newProdRecovery(t, h.phase1Harness, h.sdir, func() registry.Route { return h.srcRoute }) |
| - rec.before = func(o hostclaim.Observed, in ClaimLaunchInputs) { |
| + src := sourceRecovery(t, h) |
| + src.beforeAdmit = func(o hostclaim.Observed, in ClaimLaunchInputs) { |
| // Stage one: the clean ReturnToSource put the holder back on |
| // src_gen, and the owners restore from src_gen. |
| if o.Claim.Holder.Generation != srcGen || srcGen != "cur" || in.Baseline == nil || in.Baseline.Generation != srcGen || in.Prepared { |
| @@ -1626,18 +1494,21 @@ func intoHomeCleanRollback(t *testing.T) ([]hostclaim.Claim, []string) { |
| t.Fatalf("the selected admission: %+v", a) |
| } |
| } |
| - h.recoveryWake = rec.wake |
| h.srcPass(t) |
| - if len(rec.admitted) != 1 { |
| - t.Fatalf("the recovery admitted %d times", len(rec.admitted)) |
| + if src.wakes != 1 || src.materialized.Claim == nil { |
| + t.Fatalf("the recovery's starts: %d, materialized %v", src.wakes, src.materialized.Claim != nil) |
| } |
| - // Stage two: the holder is on the fresh generation; the admission is |
| - // clean-return, in ca, after the home target's last admission in ie. |
| - ah := rec.admitted[0].Claim.Holder |
| + // Stage two: the holder is on the fresh generation the runtime admitted |
| + // and restored; the admission is clean-return, in ca, after the home |
| + // target's last admission in ie. |
| + ah := src.admitted.Claim.Holder |
| last := ah.Admissions[len(ah.Admissions)-1] |
| - if ah.Generation != rec.selected[0].Admission.NewGeneration || ah.Generation == srcGen || ah.WriterBox != "ca-box" { |
| + if ah.Generation != src.materializedIn.Admission.NewGeneration || ah.Generation == srcGen || ah.WriterBox != "ca-box" { |
| t.Fatalf("the holder after the admission: %+v", ah) |
| } |
| + if gen, err := hostbox.ReadGenerationMarker(h.srcRoute.DBPath + ".gen"); err != nil || gen != ah.Generation { |
| + t.Fatalf("the restore's generation marker %q %v, want %s", gen, err, ah.Generation) |
| + } |
| if last.Coverage != "clean-return" || last.WritesLost != nil || len(last.AdditionalWritesLost) != 0 || last.Region != "ca" || last.Box != "ca-box" { |
| t.Fatalf("the clean return's admission: %+v", last) |
| } |
| @@ -1724,12 +1595,14 @@ func intoHomeDirtyReturn(t *testing.T) ([]hostclaim.Claim, []string) { |
| leases.wrap(&h.cfg) |
| cleanHomeStart(t, firstMoving(t, watch.since(0))) |
| if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, func(rec *provision.Record) error { |
| - rec.Move.CutAttempts = 1 |
| + rec.Move.CutAttempts = 2 |
| return nil |
| }); err != nil { |
| t.Fatal(err) |
| } |
| - // The probe is admitted, the app writes, and the probe fails. |
| + // Each attempt is admitted through the production owners and the claim |
| + // runtime, the app writes, and the attempt fails. The second's restore |
| + // retires the first's installation (RetireSuperseded). |
| p := h.productionProbe(t) |
| p.afterMaterialize = func(int) error { |
| h.targetChanged(t) |
| @@ -1746,8 +1619,8 @@ func intoHomeDirtyReturn(t *testing.T) ([]hostclaim.Claim, []string) { |
| for i := 0; i < 6 && h.read(t, p1Host).Move.State != provision.MoveRollingBack; i++ { |
| h.rollbackPass(t) |
| } |
| - if len(p.admissions) != 1 || p.admissions[0].Box != "home-box" || p.admissions[0].Region != "ie" { |
| - t.Fatalf("the home target's admission: %+v", p.admissions) |
| + if len(p.admissions) != 2 || p.stopped != 1 || p.admissions[1].Box != "home-box" || p.admissions[1].Region != "ie" || p.admissions[1].Predecessor != p1ID { |
| + t.Fatalf("the home target's admissions: %+v (predecessors stopped %d)", p.admissions, p.stopped) |
| } |
| before := len(leases.all()) |
| for i := 0; i < 6 && h.rb(t).TargetPhase != provision.RbTargetReturned; i++ { |
| @@ -1778,18 +1651,17 @@ func intoHomeDirtyReturn(t *testing.T) ([]hostclaim.Claim, []string) { |
| t.Fatalf("the return lane's staging leases: %v", got) |
| } |
| - // The source's recovery through the production recovery owners, the |
| - // real C8 provider establishing from the return's prepared manifest. |
| - sourceMarker(t, h.phase1Harness) |
| - rec := newProdRecovery(t, h.phase1Harness, h.sdir, func() registry.Route { return h.srcRoute }) |
| - rec.materializeWith(p1Work{order: new([]string), name: "spawn"}) |
| + // The source's recovery through the claim runtime on ca-box: the |
| + // production recovery owners, the real C8 provider establishing from the |
| + // return's prepared manifest. |
| + rec := sourceRecovery(t, h) |
| // Under CARLOS_HOST_CLAIM (the probe mode sets it) the source's |
| // configured step publishes the live release through the held release; |
| // the stand-in answers as that publication would. |
| h.releaseLive = func(context.Context, release.Store, provision.Record, string) moves.Check { |
| return moves.Check{OK: true} |
| } |
| - rec.before = func(o hostclaim.Observed, in ClaimLaunchInputs) { |
| + rec.beforeAdmit = func(o hostclaim.Observed, in ClaimLaunchInputs) { |
| residueQueued := false |
| for _, ob := range o.Claim.Obligations { |
| if ob.Kind == "park-generation" && ob.Region == "ca" && ob.Owner == "region:ca" { |
| @@ -1803,10 +1675,9 @@ func intoHomeDirtyReturn(t *testing.T) ([]hostclaim.Claim, []string) { |
| t.Fatalf("the dirty return's selected admission: %+v prepared %v", a, in.Prepared) |
| } |
| } |
| - h.recoveryWake = rec.wake |
| h.srcPass(t) |
| - if len(rec.admitted) != 1 { |
| - t.Fatalf("the recovery admitted %d times", len(rec.admitted)) |
| + if rec.wakes != 1 || rec.materialized.Claim == nil || rec.materializedIn.Admission.NewGeneration != rb.ID { |
| + t.Fatalf("the recovery's starts: %d, materialized %+v", rec.wakes, rec.materializedIn.Admission) |
| } |
| got := h.opened(t) |
| if got.ID != rb.ID { |
| @@ -1866,7 +1737,9 @@ func intoHomePreFlipRevert(t *testing.T) ([]hostclaim.Claim, []string) { |
| t.Helper() |
| ctx := context.Background() |
| var watch *claimWatch |
| - h := newRevertHarness(t, intoHome, realLaunchEnv, watchClaims(&watch)) |
| + // The cut harness under a landing harness: its source replica is a real |
| + // one, which the recovery's cold restore restores. |
| + h := newRevertHarnessOn(t, newLandingHarness(t, intoHome, realLaunchEnv, watchClaims(&watch)).cutHarness) |
| leases := &stagingLeaseKeys{} |
| leases.wrap(&h.cfg) |
| h.fromStopped(t) |
| @@ -1889,27 +1762,20 @@ func intoHomePreFlipRevert(t *testing.T) ([]hostclaim.Claim, []string) { |
| // The source's revert probe through the production recovery owners. |
| h.arm(t) |
| - sourceMarker(t, h.phase1Harness) |
| - rec := newProdRecovery(t, h.phase1Harness, h.cfg.DataDir, func() registry.Route { return h.route }) |
| srcGen := c.Holder.Generation |
| - rec.before = func(o hostclaim.Observed, in ClaimLaunchInputs) { |
| + rec := revertRecovery(t, h) |
| + rec.beforeAdmit = func(o hostclaim.Observed, in ClaimLaunchInputs) { |
| if a := in.Admission; a.Box != "ca-box" || a.Region != "ca" || a.PredecessorRegion != "ca" || a.Coverage != "" || a.Mode != "cold" || a.ObservedGeneration != srcGen || |
| in.Baseline == nil || in.Baseline.Generation != srcGen { |
| t.Fatalf("the revert's selected admission: %+v baseline %+v", a, in.Baseline) |
| } |
| } |
| - h.cfg.recoveryProbe = func(ctx context.Context, s release.Store, r provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) { |
| - h.probes = append(h.probes, req) |
| - w := h.w() |
| - a := recoveryProbeAdapter{Box: w.cfg.Box, Claims: true, Route: w.cfg.Route, Current: w.recoveryInputs, Wake: rec.wake} |
| - return a.probe(ctx, s, r, req) |
| - } |
| h.pass(t) |
| h.reopened(t) |
| - if len(rec.admitted) != 1 { |
| - t.Fatalf("the revert probe admitted %d times", len(rec.admitted)) |
| + if rec.wakes != 1 || rec.materialized.Claim == nil { |
| + t.Fatalf("the revert probe's starts: %d, materialized %v", rec.wakes, rec.materialized.Claim != nil) |
| } |
| - ah := rec.admitted[0].Claim.Holder |
| + ah := rec.admitted.Claim.Holder |
| last := ah.Admissions[len(ah.Admissions)-1] |
| if ah.Generation == srcGen || ah.WriterBox != "ca-box" || last.Region != "ca" || last.PredecessorRegion != "ca" || last.Predecessor != srcGen || last.Coverage != "" { |
| t.Fatalf("the revert's admission: holder %+v last %+v", ah, last) |
| @@ -1953,6 +1819,88 @@ func intoHomePreFlipRevert(t *testing.T) ([]hostclaim.Claim, []string) { |
| return watch.since(0), leases.all() |
| } |
| +// sourceRecovery binds the recover harness's source box (ca-box) to the |
| +// production mode: the recovery adapter's wake runs the claim runtime on |
| +// ca-box (newProdWake's start: Admit, Prepare, Restore, Reserve and |
| +// Materialize through the production recovery owners and the real C8). |
| +// The source is first made what a served source leaves (sourceServed). |
| +func sourceRecovery(t *testing.T, h *recoverHarness) *prodProbe { |
| + t.Helper() |
| + sourceServed(t, h.phase1Harness, h.sdir) |
| + p := newProdWake(t, h.phase1Harness, prodSide{box: h.srcBox, code: h.src, dataDir: h.sdir, bucket: h.srcBucket, aws: h.srcAWS, |
| + route: func() registry.Route { return h.srcRoute }, |
| + inputs: func() *provision.MoveInputs { return h.read(t, p1Host).Move.Rollback.RecoveryInputs }}) |
| + h.recoveryWake = p.recoveryWake |
| + return p |
| +} |
| + |
| +// revertRecovery binds a revert harness's source box (ca-box) to the |
| +// production mode, as sourceRecovery does for a rollback: the revert's |
| +// recovery adapter wakes the claim runtime on ca-box. |
| +func revertRecovery(t *testing.T, h *revertHarness) *prodProbe { |
| + t.Helper() |
| + sourceServed(t, h.phase1Harness, h.cfg.DataDir) |
| + p := newProdWake(t, h.phase1Harness, prodSide{box: h.srcBox, code: h.src, dataDir: h.cfg.DataDir, bucket: h.srcBucket, aws: h.srcAWS, |
| + route: func() registry.Route { return h.route }, |
| + inputs: func() *provision.MoveInputs { return h.read(t, p1Host).Move.RecoveryInputs }}) |
| + h.cfg.recoveryProbe = func(ctx context.Context, s release.Store, r provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) { |
| + h.probes = append(h.probes, req) |
| + w := h.w() |
| + a := recoveryProbeAdapter{Box: w.cfg.Box, Claims: true, Route: w.cfg.Route, Current: w.recoveryInputs, Wake: p.recoveryWake} |
| + return a.probe(ctx, s, r, req) |
| + } |
| + return p |
| +} |
| + |
| +// sourceServed makes the source box and its bucket what serving the |
| +// instance under claims leaves (the harness built them by hand): the |
| +// bucket versioned with a generation.json naming src_gen (sourceMarker); |
| +// the source's installation in the claim's inventory at the flip's source |
| +// stamp; src_gen's files manifest in its bucket, built from the source's |
| +// files and env; and src_gen's origin recorded, as an admission through the |
| +// real CAS (or the #250 backfill) records it. |
| +func sourceServed(t *testing.T, h *phase1Harness, dataDir string) { |
| + t.Helper() |
| + ctx := context.Background() |
| + sourceMarker(t, h) |
| + c := h.claim(t) |
| + var err error |
| + // The source's installation: the flip's journalled stamp and src_gen, |
| + // or, for a revert whose holder never left the source, the holder's. |
| + stamp, gen := c.Holder.Stamp, c.Holder.Generation |
| + if raw, ok := c.Op.Journal[hostclaim.JournalSourceStamp]; ok { |
| + if stamp, err = strconv.ParseUint(raw, 10, 64); err != nil || stamp == 0 { |
| + t.Fatalf("fixture: the claim's source stamp %q: %v", raw, err) |
| + } |
| + gen = c.Op.Journal[hostclaim.JournalSrcGen] |
| + } |
| + // The box's guard directories and its installation record at that |
| + // stamp, unless a step of the recovery already wrote it. |
| + for _, d := range []string{"hostguard", ".parked"} { |
| + if err := os.MkdirAll(filepath.Join(dataDir, d), 0o770); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + record := filepath.Join(dataDir, "hostguard", p1Host+".json") |
| + if _, err := os.Stat(record); errors.Is(err, os.ErrNotExist) { |
| + body := fmt.Sprintf(`{"v":1,"host":%q,"incarnation":%q,"active_stamp":%d,"retired_stamps":[]}`, p1Host, h.inc, stamp) |
| + if err := os.WriteFile(record, []byte(body), 0o660); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { |
| + if c.Inventory == nil { |
| + c.Inventory = map[string]hostclaim.InventoryEntry{} |
| + } |
| + c.Inventory[h.srcBox] = hostclaim.InventoryEntry{Layout: "flat", Paths: map[string]uint64{p1Host + ".db": stamp, p1Host + ".db.gen": stamp, p1Host: stamp, p1Host + ".env": stamp}} |
| + }) |
| + if _, err = buildPreparedManifest(ctx, h.vs, h.srcBucket, cutRoot, gen, p1ID, stamp, filepath.Join(dataDir, p1Host), filepath.Join(dataDir, p1Host+".env"), true, |
| + hostfiles.WakeState{V: 1, Host: p1Host, Timers: []hostfiles.WakeTimer{}}); err != nil { |
| + t.Fatalf("src_gen's files manifest: %v", err) |
| + } |
| + recordOrigins(t, h.s, p1Host, hostclaim.OriginTuple{Incarnation: h.inc, Location: cutRoot, Region: park.PlacementRegion(h.src), Generation: gen}) |
| +} |
| + |
| // firstMoving is the first claim written with the move's op: the start |
| // CAS's. |
| func firstMoving(t *testing.T, writes []hostclaim.Claim) *hostclaim.Claim { |
diff --git a/internal/host/movesparks_test.go b/internal/host/movesparks_test.go| index 3dcab785f..2af37cdd0 100644 |
| --- a/internal/host/movesparks_test.go |
| +++ b/internal/host/movesparks_test.go |
| @@ -529,10 +529,13 @@ func (p1FilesStore) Delete(context.Context, string) error { |
| return errors.New("test: not implemented") |
| } |
| -// prodProbe is the production probe mode's state. |
| +// prodProbe is the production mode's state on one box: the claim runtime |
| +// (claim Plan 9 Group D's ClaimWakeAssembly, the recovery fixture's |
| +// controller) with the moves owners bound, on the box the side names. |
| type prodProbe struct { |
| t *testing.T |
| - h *landingHarness |
| + h *phase1Harness |
| + side prodSide |
| ctl *recoveryController |
| owners *movesProbeOwners |
| // order is the held works, render and release, in the order they ran. |
| @@ -551,32 +554,74 @@ type prodProbe struct { |
| selected ClaimLaunchInputs |
| attempt hostclaim.WakeRetryAttempt |
| // admissions are the admissions the runtime made, one per attempt, and |
| - // admitted the claim each left, the last attempt's last. |
| + // admitted the claim each left, the last attempt's last; observed is |
| + // the claim each start read before its admission. |
| admissions []hostclaim.AdmissionRecord |
| admitted hostclaim.Observed |
| + observed []hostclaim.Observed |
| + // stopped counts the predecessor writer sessions this box's own wakes |
| + // stopped through the production Stop. |
| + stopped int |
| // materialized is the claim and the launch inputs the runtime last |
| // materialized: the admitted inputs, which for a fresh generation are |
| // not the selected ones (each Select mints its own generation). |
| materialized hostclaim.Observed |
| materializedIn ClaimLaunchInputs |
| // afterMaterialize, when set, runs in the n-th wake (from 1) after |
| - // Materialize and before readiness, while the attempt is live; an error |
| - // ends the wake there, as an agent that died there leaves it (home part |
| - // 2, Task 3). |
| + // Materialize and before the launch, while the attempt is live; an error |
| + // fails the attempt there (the runtime's Failed), as an attempt that |
| + // failed before readiness, or an agent that died there, leaves it (home |
| + // part 2, Tasks 3 and 4). |
| afterMaterialize func(n int) error |
| - wakes int |
| - last activator.ProbeResult |
| - live bool |
| - seq int |
| + // beforeAdmit, when set, sees each start's claim and the owners' |
| + // selection before the runtime's admission. |
| + beforeAdmit func(o hostclaim.Observed, in ClaimLaunchInputs) |
| + wakes int |
| + last activator.ProbeResult |
| + live bool |
| + seq int |
| +} |
| + |
| +// prodSide is the box a production mode runs on: its roster box, its |
| +// placement code (its composer's region), its data directory, data bucket |
| +// and the bucket's AWS region, its registry row, and the inputs its held |
| +// render renders (the move's pins for a landing, the recorded recovery |
| +// inputs for a recovery). |
| +type prodSide struct { |
| + box, code, dataDir, bucket, aws string |
| + route func() registry.Route |
| + inputs func() *provision.MoveInputs |
| } |
| // productionProbe switches the harness's probe to the production mode: the |
| // target's probe wakes through movesProbeOwners over the claim runtime, with |
| // the target box's roster identity. |
| func (h *landingHarness) productionProbe(t *testing.T) *prodProbe { |
| + t.Helper() |
| + p := newProdWake(t, h.phase1Harness, prodSide{box: h.dstBox, code: h.dst, dataDir: h.tdir, bucket: h.dstBucket, aws: h.dstAWS, |
| + route: func() registry.Route { return h.route }, |
| + inputs: func() *provision.MoveInputs { return h.read(t, p1Host).Move.Inputs }}) |
| + h.probe, h.probeLive = p.wake, p.liveProbe |
| + return p |
| +} |
| + |
| +// newProdWake builds the production mode on side's box. The owners are the |
| +// production movesProbeOwners (which dispatch a recovery op to their |
| +// recovery owners) with the real C8 provider, bound into the claim runtime |
| +// with the owners' admission evidence, the moves attempt journal and the |
| +// production artifact retirement (claimArtifactRetirement over park.Local, |
| +// as claimconstruction.go wires RetireArtifacts and RecoverArtifacts). |
| +// What stands in is what main cannot run here: the launch (every claim |
| +// start is refused on main until the isolated root launcher lands), its |
| +// readiness, the held works and render (the claim's move read access over |
| +// this box's registry), root's stop transport (the recovery fixture's |
| +// coordinator), the cross-box cold handover (AfterAdmission) and the |
| +// activator's live-probe entry. A Materialize that passes is the attempt |
| +// passing. |
| +func newProdWake(t *testing.T, h *phase1Harness, side prodSide) *prodProbe { |
| t.Helper() |
| if !h.realEnv { |
| - // The probe's launch is the target's own composition; a source that |
| + // The probe's launch is the box's own composition; a source that |
| // pinned the h.env fixture would never match it. |
| t.Fatal("fixture: the production probe mode needs realLaunchEnv") |
| } |
| @@ -584,34 +629,36 @@ func (h *landingHarness) productionProbe(t *testing.T) *prodProbe { |
| // under both switches; the real C8's proof reads that record. |
| t.Setenv("CARLOS_HOST_CLAIM", "1") |
| t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", "1") |
| - actor := h.roster[h.dstBox] |
| + actor := h.roster[side.box] |
| journal, err := hostbox.OpenLaunchJournal(filepath.Join(t.TempDir(), "launches.db")) |
| if err != nil { |
| t.Fatal(err) |
| } |
| t.Cleanup(func() { journal.Close() }) |
| - p := &prodProbe{t: t, h: h} |
| + p := &prodProbe{t: t, h: h, side: side} |
| + guard := hostguard.Guard{DataDir: side.dataDir, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: h.s}} |
| f := &recoveryFixture{t: t, claims: h.s, evidence: release.NewDirStore(t.TempDir()), signer: ed25519.NewKeyFromSeed(make([]byte, ed25519.SeedSize)), |
| - guard: hostguard.Guard{DataDir: h.tdir, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: h.s}}, |
| - journal: journal, kernel: &recoveryKernel{journal: journal}, now: h.clock.Now(), recordPrelaunch: true, |
| - data: h.vs, bucket: h.dstBucket, box: actor.Box, enrollment: actor.Enrollment, region: actor.Region, leaseRegion: h.dstAWS} |
| + guard: guard, journal: journal, kernel: &recoveryKernel{journal: journal}, now: h.clock.Now(), recordPrelaunch: true, |
| + data: h.vs, bucket: side.bucket, box: actor.Box, enrollment: actor.Enrollment, region: actor.Region, leaseRegion: side.aws} |
| key, err := moves.ReportKeyFromSecrets(h.cfg.SecretsKey) |
| if err != nil { |
| t.Fatal(err) |
| } |
| - filesDir := h.lcfg.FilesDir |
| + filesDir := func(host string) (string, string) { |
| + return filepath.Join(side.dataDir, host), filepath.Join(side.dataDir, host+".env") |
| + } |
| owners := &movesProbeOwners{HomeRegion: actor.HomeRegion, FleetHomeRegion: actor.FleetHomeRegion, Claims: h.s, Records: MovesRecordsFrom(h.s), |
| - ReadyPath: movesReadyPath(h.s), Data: h.vs, Bucket: h.dstBucket, Region: actor.Region, Box: actor.Box, DataDir: h.tdir, FilesDir: filesDir, |
| + ReadyPath: movesReadyPath(h.s), Data: h.vs, Bucket: side.bucket, Region: actor.Region, Box: actor.Box, DataDir: side.dataDir, FilesDir: filesDir, |
| Markers: p1Markers{h.vs}, |
| - Route: func(host string) (*registry.Route, error) { rt := h.route; return &rt, nil }, |
| + Route: func(host string) (*registry.Route, error) { rt := side.route(); return &rt, nil }, |
| // The held row's launch: what this box would launch, composed by |
| - // the production composer over the target's data directory with |
| - // the target's own region, as claimMoveLaunchSource composes it |
| + // the production composer over its data directory with its own |
| + // region, as claimMoveLaunchSource composes it |
| // (ExecBackend.LaunchLayers, CARLOS_MOVE_PROBE dropped), the live |
| // release, and the env report made from that same composition, whose |
| // digest is the launch's move-portable pin. |
| Launch: func(_ context.Context, rt registry.Route) (ClaimServingLaunch, moves.EnvReport, error) { |
| - composed, supplied, _, err := (&activator.ExecBackend{DataDir: h.tdir, Region: h.dst}).LaunchLayers(rt) |
| + composed, supplied, _, err := (&activator.ExecBackend{DataDir: side.dataDir, Region: side.code}).LaunchLayers(rt) |
| if err != nil { |
| return ClaimServingLaunch{}, moves.EnvReport{}, err |
| } |
| @@ -631,7 +678,7 @@ func (h *landingHarness) productionProbe(t *testing.T) *prodProbe { |
| }, |
| Render: func(ctx context.Context, _ registry.Route, _ string, work func(appconfig.HeldRender) (hostwrite.Work, error)) (appconfig.HeldRender, error) { |
| p.order = append(p.order, "render") |
| - in := h.read(t, p1Host).Move.Inputs |
| + in := side.inputs() |
| hr := appconfig.HeldRender{BundleETags: in.BundleETags, OverlayETag: in.OverlayETag} |
| if _, err := work(hr); err != nil { |
| return appconfig.HeldRender{}, err |
| @@ -644,21 +691,21 @@ func (h *landingHarness) productionProbe(t *testing.T) *prodProbe { |
| }, |
| NewID: func() string { p.seq++; return fmt.Sprintf("fresh%d", p.seq) }, |
| Now: h.clock.Now} |
| - // The C8 provider is the real one over the target's bucket, as the |
| - // claim assembly binds it, with Plan 15's proof over the claim store. |
| - owners.Files = &movesFiles{Files: p1FilesStore{vs: h.vs, bucket: h.dstBucket}, Data: h.vs, Bucket: h.dstBucket, HomeRegion: actor.FleetHomeRegion, Region: actor.Region, Box: actor.Box, |
| - DataDir: h.tdir, EnvFile: envFileOf(filesDir), Proof: ClaimOriginProof{Claims: h.s, HomeRegion: actor.FleetHomeRegion, Region: actor.Region}} |
| + // The C8 provider is the real one over the box's bucket, as the claim |
| + // assembly binds it, with Plan 15's proof over the claim store. |
| + owners.Files = &movesFiles{Files: p1FilesStore{vs: h.vs, bucket: side.bucket}, Data: h.vs, Bucket: side.bucket, HomeRegion: actor.FleetHomeRegion, Region: actor.Region, Box: actor.Box, |
| + DataDir: side.dataDir, EnvFile: envFileOf(filesDir), Proof: ClaimOriginProof{Claims: h.s, HomeRegion: actor.FleetHomeRegion, Region: actor.Region}} |
| if !owners.valid() { |
| t.Fatal("fixture: the production probe owners are not valid") |
| } |
| - attempts := &movesProbeAttempts{Claims: h.s, Records: MovesRecordsFrom(h.s), DataDir: h.tdir, NewID: owners.NewID, Now: h.clock.Now, Admitted: MovesRecordAdmittedAttempts} |
| + attempts := &movesProbeAttempts{Claims: h.s, Records: MovesRecordsFrom(h.s), DataDir: side.dataDir, NewID: owners.NewID, Now: h.clock.Now, Admitted: MovesRecordAdmittedAttempts} |
| bound := owners.MovesOwners(attempts) |
| dispatch := ClaimOwnerDispatch{Moves: &bound.ClaimKindOwners} |
| // The writer's admission evidence is the owner dispatch's, as the claim |
| // assembly binds it (claimassembly.go), bound before the runtime is |
| // built: Build captures the writer, so a later swap would leave the |
| // controller's allow-all evidence in force. |
| - f.homeRegion = actor.HomeRegion |
| + f.homeRegion, f.fleetHomeRegion = actor.HomeRegion, actor.FleetHomeRegion |
| f.admissionEvidence = func(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, plan *genselect.Plan) error { |
| p.evidence++ |
| if p.rejectEvidence != nil { |
| @@ -672,8 +719,13 @@ func (h *landingHarness) productionProbe(t *testing.T) *prodProbe { |
| c.wake.Continue = hostclaim.SessionContinues |
| c.wake.Operations = bound.Operations |
| c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil } |
| + // The production artifact retirement: a restore over a database of |
| + // another generation retires that installation through park.Local's |
| + // RetireSuperseded, which re-tags its inventory paths with the admitted |
| + // installation's stamp. |
| + retirement := claimArtifactRetirement{Local: park.Local{Guard: guard, Claims: h.s, DataDir: side.dataDir, Box: actor.Box}, Continue: hostclaim.SessionContinues} |
| + c.wake.RetireArtifacts, c.wake.RecoverArtifacts = retirement.Retire, retirement.Recover |
| p.ctl, p.owners = c, owners |
| - h.probe, h.probeLive = p.wake, p.liveProbe |
| return p |
| } |
| @@ -692,14 +744,43 @@ func (r p1RecordingInputs) Materialize(ctx context.Context, o hostclaim.Observed |
| // wake is one probe attempt through the claim runtime: the activator's |
| // ProbeWake with its start policy, as far as the claim runs on main. |
| func (p *prodProbe) wake(ctx context.Context, cfg activator.ProbeConfig) (activator.ProbeResult, error) { |
| + policy := activator.StartPolicy{MaxStarts: 5, Window: time.Duration(moves.ProbeSeconds * float64(time.Second)), Probe: true, |
| + Deadline: cfg.Deadline, Prepared: cfg.Prepared, Inputs: cfg.Inputs} |
| + return p.start(ctx, p.side.route(), policy) |
| +} |
| + |
| +// recoveryWake is the activator's recovery probe entry |
| +// (ClaimRecoveryProbeWake) on this box: one claim wake under the |
| +// adapter's policy. |
| +func (p *prodProbe) recoveryWake(ctx context.Context, rt registry.Route, policy activator.StartPolicy) error { |
| + _, err := p.start(ctx, rt, policy) |
| + return err |
| +} |
| + |
| +// start is one claim wake's first start, in the activator's order |
| +// (claimwake.go): Prerequisites' stop of this box's own predecessor writer |
| +// session (the production Stop, through the fixture's root coordinator), |
| +// Admit, Prepare, Restore (the production cold restore of the selected |
| +// baseline, retiring a superseded installation), then the attempt's |
| +// Reserve and Materialize; a failure from then on is the attempt's |
| +// (Failed). The launch, its readiness and Healthy stand in: a Materialize |
| +// that passes is the attempt passing. |
| +func (p *prodProbe) start(ctx context.Context, rt registry.Route, policy activator.StartPolicy) (activator.ProbeResult, error) { |
| p.live = false |
| h := p.h |
| o, err := hostclaim.Read(ctx, h.s, p1Host) |
| if err != nil { |
| return activator.ProbeResult{}, err |
| } |
| - policy := activator.StartPolicy{MaxStarts: 5, Window: time.Duration(moves.ProbeSeconds * float64(time.Second)), Probe: true, |
| - Deadline: cfg.Deadline, Prepared: cfg.Prepared, Inputs: cfg.Inputs} |
| + if prior := o.Claim.Holder; prior != nil && prior.WriterBox == p.side.box && prior.Session != 0 { |
| + if err := p.ctl.wake.Stop(ctx, o); err != nil { |
| + return activator.ProbeResult{}, fmt.Errorf("stop the predecessor session %d: %w", prior.Session, err) |
| + } |
| + p.stopped++ |
| + if o, err = hostclaim.Read(ctx, h.s, p1Host); err != nil { |
| + return activator.ProbeResult{}, err |
| + } |
| + } |
| live, cancel := context.WithCancel(context.Background()) |
| p.t.Cleanup(cancel) |
| done := make(chan struct{}) |
| @@ -707,10 +788,13 @@ func (p *prodProbe) wake(ctx context.Context, cfg activator.ProbeConfig) (activa |
| p.ctl.wake.mu.Lock() |
| p.ctl.wake.pending[p1Host] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done} |
| p.ctl.wake.mu.Unlock() |
| - rt := h.route |
| + p.observed = append(p.observed, o) |
| if p.selected, err = p.owners.Select(ctx, rt, o, policy); err != nil { |
| return activator.ProbeResult{}, err |
| } |
| + if p.beforeAdmit != nil { |
| + p.beforeAdmit(o, p.selected) |
| + } |
| admitted, err := p.ctl.runtime.Admit(ctx, rt, o, policy) |
| if err != nil { |
| return activator.ProbeResult{}, err |
| @@ -723,30 +807,25 @@ func (p *prodProbe) wake(ctx context.Context, cfg activator.ProbeConfig) (activa |
| if err != nil { |
| return activator.ProbeResult{}, err |
| } |
| - if s.OpenProbe != nil { |
| - if _, err := s.OpenProbe(ctx, h.clock.Now()); err != nil { |
| - return activator.ProbeResult{}, err |
| - } |
| + if err := s.Restore(ctx); err != nil { |
| + return activator.ProbeResult{}, fmt.Errorf("restore: %w", err) |
| } |
| if err := s.Reserve(ctx, 0); err != nil { |
| return activator.ProbeResult{}, err |
| } |
| p.attempt = s.CurrentAttempt() |
| - // The activator's cold restore from the selected baseline stands in |
| - // here: it rewrites <dbPath>.gen to the admitted generation. The |
| - // landing's install already wrote the move id, so only a restarted |
| - // attempt's fresh generation changes it (home part 2, Task 3). |
| - if gen, err := hostbox.ReadGenerationMarker(rt.DBPath + ".gen"); err != nil || gen != ah.Generation { |
| - if err := hostbox.WriteGenerationMarker(rt.DBPath+".gen", ah.Generation); err != nil { |
| + if err := s.Materialize(ctx); err != nil { |
| + return activator.ProbeResult{}, errors.Join(err, s.Failed(ctx, 0)) |
| + } |
| + if s.OpenProbe != nil { |
| + // At the exec (the start's beforeStart hook). |
| + if _, err := s.OpenProbe(ctx, h.clock.Now()); err != nil { |
| return activator.ProbeResult{}, err |
| } |
| } |
| - if err := s.Materialize(ctx); err != nil { |
| - return activator.ProbeResult{}, err |
| - } |
| if p.afterMaterialize != nil { |
| if err := p.afterMaterialize(p.wakes); err != nil { |
| - return activator.ProbeResult{}, err |
| + return activator.ProbeResult{}, errors.Join(err, s.Failed(ctx, 0)) |
| } |
| } |
| p.last = activator.ProbeResult{Session: ah.Session, Stamp: ah.Stamp, Generation: ah.Generation} |
diff --git a/internal/host/movesrevert_test.go b/internal/host/movesrevert_test.go| index 9f839a2ca..47242a196 100644 |
| --- a/internal/host/movesrevert_test.go |
| +++ b/internal/host/movesrevert_test.go |
| @@ -50,7 +50,14 @@ type revertHarness struct { |
| func newRevertHarness(t *testing.T, opts ...p1Option) *revertHarness { |
| t.Helper() |
| - h := &revertHarness{cutHarness: newCutHarness(t, opts...), tdir: t.TempDir(), tleases: &fakeParkLeases{}} |
| + return newRevertHarnessOn(t, newCutHarness(t, opts...)) |
| +} |
| + |
| +// newRevertHarnessOn is newRevertHarness over a cut harness the caller |
| +// built (a landing harness's, whose source replica is a real one). |
| +func newRevertHarnessOn(t *testing.T, ch *cutHarness) *revertHarness { |
| + t.Helper() |
| + h := &revertHarness{cutHarness: ch, tdir: t.TempDir(), tleases: &fakeParkLeases{}} |
| // The row follows the cut's hold, as holdMoving keeps it: held moving |
| // under the op, a row held by another op refused. |
| h.cfg.HoldMoving = func(host, op string) error { |
| @@ -652,28 +659,23 @@ func countString(xs []string, s string) int { |
| // the claim again; with the step's own read it lost the CAS every pass, and |
| // each pass's probe admitted one more fresh generation until the attempts |
| // were spent. Here, on the default named move, the probe runs through the |
| -// production adapter and recovery owners and admits once; the source then |
| -// reopens in the same pass. |
| +// production adapter and the claim runtime (Admit, Prepare, Restore, |
| +// Reserve, Materialize) with the production recovery owners, and admits |
| +// once; the source then reopens in the same pass. |
| func TestRevertProbedAfterARealAdmission(t *testing.T) { |
| - h := newRevertHarness(t, realLaunchEnv) |
| + h := newRevertHarnessOn(t, newLandingHarness(t, realLaunchEnv).cutHarness) |
| h.fromStopped(t) |
| h.arm(t) |
| - sourceMarker(t, h.phase1Harness) |
| - rec := newProdRecovery(t, h.phase1Harness, h.cfg.DataDir, func() registry.Route { return h.route }) |
| - h.cfg.recoveryProbe = func(ctx context.Context, s release.Store, r provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) { |
| - w := h.w() |
| - a := recoveryProbeAdapter{Box: w.cfg.Box, Claims: true, Route: w.cfg.Route, Current: w.recoveryInputs, Wake: rec.wake} |
| - return a.probe(ctx, s, r, req) |
| - } |
| + rec := revertRecovery(t, h) |
| h.pass(t) |
| h.reopened(t) |
| - if len(rec.admitted) != 1 { |
| - t.Fatalf("the revert probe admitted %d times", len(rec.admitted)) |
| + if rec.wakes != 1 || rec.materialized.Claim == nil { |
| + t.Fatalf("the revert probe's starts: %d, materialized %v", rec.wakes, rec.materialized.Claim != nil) |
| } |
| h.pass(t) |
| h.reopened(t) |
| - if len(rec.admitted) != 1 { |
| - t.Fatalf("a pass after the reopen probed again: %d admissions", len(rec.admitted)) |
| + if rec.wakes != 1 { |
| + t.Fatalf("a pass after the reopen probed again: %d starts", rec.wakes) |
| } |
| } |