amadan

carlos / platform Public

hostrelocate: a participant acknowledges, releases, installs and probes in the order its phase allows

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude 4879c1e571885bf1c8294cc1a9aed2eaffd91feb parent 032db0d
2 files changed, +638 −0
  • internal/hostrelocate/participant.go +246 −0
  • internal/hostrelocate/participant_test.go +392 −0
diff --git a/internal/hostrelocate/participant.go b/internal/hostrelocate/participant.go
new file mode 100644
index 000000000..8031b7012
--- /dev/null
+++ b/internal/hostrelocate/participant.go
@@ -0,0 +1,246 @@
+package hostrelocate
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "slices"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// Effects are a participant box's local effects, each under the host
+// guard and bound to the claim observation it is given. Plan 6b's box
+// assembly implements them (ClaimAssembly); this package only orders them
+// and records each result in the claim.
+type Effects interface {
+ // Acknowledge writes this box's pending controller state, refuses while
+ // root has an update begun (§5.10), takes the execution lease
+ // (hostexec.Open under the agent's lifetime, then Ready) and returns
+ // the acknowledgement to journal. After an agent restart it opens a
+ // fresh controller for the same participation; the journaled
+ // acknowledgement (the first lease, D6) is left as it is.
+ Acknowledge(context.Context, hostclaim.Observed) (hostclaim.ExecutionAcknowledgement, error)
+ // Ready reports whether this box's execution controller for the
+ // operation is open and active now: ErrNoController when none is
+ // retained (the agent restarted), hostexec's error when it was lost.
+ // Every effect after the acknowledgement runs only after Ready.
+ Ready(context.Context, hostclaim.Observed) error
+ // Release runs §5.8's release sequence for this box's current writer
+ // session: stop every writer (stopevidence.WritersStopped), flush the
+ // database to its final TXID, publish every pending file and wake-state
+ // mutation, close the replicator and publisher, and complete full stop
+ // evidence (ClaimStopCoordinator.FenceAndStop). It returns the journal
+ // entry; it is idempotent for one session.
+ Release(context.Context, hostclaim.Observed) (hostclaim.Release, error)
+ // Build builds the prepared generation P from the journaled pin with
+ // moves' shared install routine (L16), under the replica lease.
+ Build(context.Context, hostclaim.Observed) error
+ // InstallLocal makes this box's local install from P's files manifest
+ // and returns the receipt (§5.8 "The local install").
+ InstallLocal(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, error)
+ // EnsureLocal is called before P's first admission, each time: it
+ // verifies that the local install the journaled receipt names is
+ // complete on this box and, when it is not (a lost lease's scoped
+ // cleanup retires an unopened P's local install), restores it under
+ // install_stamp from P's files manifest. It is never called once P is
+ // opened: an opened P is never reinstalled (rev 15, round 14).
+ EnsureLocal(context.Context, hostclaim.Observed) error
+ // VerifyReceipt is called before every admission after P is opened:
+ // it reports whether this box's durable local install receipt is
+ // still the one the claim journaled (box, enrollment, install_stamp,
+ // journal digest). It checks the receipt and its binding only, never
+ // the files, env or timers, which the admitted writers have changed
+ // legitimately since. A missing or different receipt is false: the
+ // local state cannot be rebuilt from P, so the operation is stuck
+ // (§5.8, B2-5; rev 16, round 15).
+ VerifyReceipt(context.Context, hostclaim.Observed) (bool, error)
+ // Probe makes one cold probe admission (AdmitWriter, with coverage the
+ // caller names) and runs the probe. passed is false while it runs.
+ Probe(ctx context.Context, obs hostclaim.Observed, coverage string) (session uint64, passed bool, err error)
+ // Retained lists the participations whose execution controllers this
+ // box still holds; Finish closes one and releases its lifetime lock.
+ // A controller is finished once the claim no longer runs that
+ // operation with this box as a participant (Sweep), so the next
+ // relocation of the host can open its own (hostexec keeps the lock
+ // until Close). Retained also lists every participation journaled on
+ // disk whose durable record a restart still has to clear; an entry it
+ // cannot read is the error, returned beside every one it can.
+ Retained() ([]Participation, error)
+ Finish(ctx context.Context, host, op string) error
+}
+
+// Participation names one retained execution controller.
+type Participation struct {
+ Host, Op string
+}
+
+// Participant is one box's half of relocation. Step makes at most one
+// participant transition for one host, chosen from the claim alone.
+type Participant struct {
+ Claims release.Store
+ Box string
+ Enrollment string
+ Effects Effects
+ Now func() time.Time
+}
+
+// Step acts for host if this box has work in its relocation, and reports
+// ErrNotYet when it has none.
+func (p Participant) Step(ctx context.Context, host string) (hostclaim.Observed, error) {
+ if p.Claims == nil || p.Box == "" || p.Enrollment == "" || p.Effects == nil || p.Now == nil {
+ return hostclaim.Observed{}, ErrUnsupported
+ }
+ obs, err := hostclaim.Read(ctx, p.Claims, host)
+ if err != nil || obs.Absent {
+ return obs, err
+ }
+ c := obs.Claim
+ if c.Op.Kind != hostclaim.KindRelocating || c.Op.Journal["stuck"] != "" || !slices.Contains(c.Op.Participants, p.Box) || slices.Contains(c.Lost, p.Box) {
+ return obs, ErrNotYet
+ }
+ id := c.Op.ID
+ if c.Op.Journal["exec_ack."+p.Box] == "" {
+ ack, err := p.Effects.Acknowledge(ctx, obs)
+ if err != nil {
+ return obs, err
+ }
+ return p.update(ctx, obs, hostclaim.AcknowledgeExecution(p.Box, ack))
+ }
+ // The controller is checked before every effect, never inferred from
+ // the historical acknowledgement: a restarted agent reopens one (the
+ // claim keeps the first lease's record) and acts on the next pass.
+ if err := p.Effects.Ready(ctx, obs); errors.Is(err, ErrNoController) {
+ _, err := p.Effects.Acknowledge(ctx, obs)
+ return obs, errors.Join(ErrNotYet, err)
+ } else if err != nil {
+ return obs, err
+ }
+ executor := c.Op.Exec == p.Box
+ switch {
+ case executor && (c.Op.Phase == hostclaim.PhaseReleasing || c.Op.Phase == hostclaim.PhaseProbeFailed) && c.Op.Journal["release"] == "":
+ r, err := p.Effects.Release(ctx, obs)
+ if err != nil {
+ return obs, err
+ }
+ return p.update(ctx, obs, hostclaim.JournalRelease(id, r))
+ case executor && c.Op.Phase == hostclaim.PhaseInstalling:
+ return obs, errors.Join(ErrNotYet, p.Effects.Build(ctx, obs))
+ case executor && c.Op.Phase == hostclaim.PhaseInstalled:
+ if c.Op.Journal["local_install"] != "" {
+ // The receipt is a past fact; the files, env and timers it
+ // names are checked (and restored) before probing.
+ if err := p.Effects.EnsureLocal(ctx, obs); err != nil {
+ return obs, err
+ }
+ // The target enters probing itself (§5.8 phase table).
+ return p.update(ctx, obs, hostclaim.ToProbing(id, p.Now()))
+ }
+ l, err := p.Effects.InstallLocal(ctx, obs)
+ if err != nil {
+ return obs, err
+ }
+ if l.Box != p.Box || l.Enrollment != p.Enrollment {
+ return obs, ErrRefused
+ }
+ return p.update(ctx, obs, hostclaim.RecordLocalInstall(id, l))
+ case executor && c.Op.Phase == hostclaim.PhaseProbing && c.Op.Journal["probe"] == "":
+ if !hostclaim.PreparedOpened(c) {
+ // Before P's first admission the local install must be whole.
+ if err := p.Effects.EnsureLocal(ctx, obs); err != nil {
+ return obs, err
+ }
+ } else {
+ // After it, the install is never remade; its receipt must
+ // still be this box's, or the operation is stuck.
+ ok, err := p.Effects.VerifyReceipt(ctx, obs)
+ if err != nil {
+ return obs, err
+ }
+ if !ok {
+ return p.update(ctx, obs, hostclaim.MarkStuck(id, "the local install receipt is missing after P was admitted (§5.8, B2-5)"))
+ }
+ }
+ // The attempt is counted in the claim before it is made (§5.8:
+ // at most three); the fourth is probe-failed, by this CAS.
+ counted, err := hostclaim.Update(ctx, p.Claims, obs, hostclaim.BeginProbeAttempt(id, p.Box))
+ if errors.Is(err, hostclaim.ErrProbeAttemptsExhausted) {
+ return p.update(ctx, obs, hostclaim.ToProbeFailed(id, p.Now()))
+ }
+ if err != nil {
+ return obs, err
+ }
+ obs, c = counted, counted.Claim
+ session, passed, err := p.Effects.Probe(ctx, obs, coverageFor(c))
+ if err != nil || !passed {
+ return obs, errors.Join(ErrNotYet, err)
+ }
+ // The probe's admission wrote the claim: record the result on a
+ // fresh read, which RecordProbe binds to the admitted session.
+ fresh, err := hostclaim.Read(ctx, p.Claims, host)
+ if err != nil || fresh.Absent {
+ return obs, errors.Join(ErrNotYet, err)
+ }
+ return p.update(ctx, fresh, hostclaim.RecordProbe(id, hostclaim.Probe{Box: p.Box, Session: session}))
+ }
+ return obs, ErrNotYet
+}
+
+// Sweep finishes every retained controller whose participation is over:
+// the claim no longer runs that operation, or this box is no longer a
+// participant of it. It runs once per pass, for hosts the pass itself
+// skips (a completed relocation is a serving claim).
+func (p Participant) Sweep(ctx context.Context) error {
+ if p.Claims == nil || p.Effects == nil {
+ return ErrUnsupported
+ }
+ retained, err := p.Effects.Retained()
+ errs := []error{err} // reported, while every readable participation is still finished
+ for _, r := range retained {
+ obs, err := hostclaim.Read(ctx, p.Claims, r.Host)
+ if err != nil {
+ errs = append(errs, err)
+ continue
+ }
+ if !obs.Absent && obs.Claim.Op.Kind == hostclaim.KindRelocating && obs.Claim.Op.ID == r.Op && slices.Contains(obs.Claim.Op.Participants, p.Box) && !slices.Contains(obs.Claim.Lost, p.Box) {
+ continue
+ }
+ if err := p.Effects.Finish(ctx, r.Host, r.Op); err != nil {
+ errs = append(errs, err)
+ }
+ }
+ return errors.Join(errs...)
+}
+
+// coverageFor is the probe admission's coverage (§5.8, L9): `released`
+// only for the admission that opens P, when this operation journaled a
+// release whose box is the predecessor writer; "" otherwise (a failover,
+// a later attempt, or a host that never had a writer). AdmitWriter checks
+// the rest.
+func coverageFor(c *hostclaim.Claim) string {
+ var r hostclaim.Release
+ if json.Unmarshal([]byte(c.Op.Journal["release"]), &r) != nil || r.ReleasedBy == "" {
+ return ""
+ }
+ if hostclaim.PreparedOpened(c) {
+ return ""
+ }
+ pred := hostclaim.AdmissionRecord{Session: c.Holder.Session, Box: c.Holder.WriterBox, Generation: c.Holder.Generation}
+ if n := len(c.Holder.Admissions); n > 0 {
+ pred = c.Holder.Admissions[n-1]
+ }
+ if pred.Box != r.ReleasedBy || pred.Session != r.Session || pred.Generation != r.Generation {
+ return ""
+ }
+ return "released"
+}
+
+func (p Participant) update(ctx context.Context, obs hostclaim.Observed, tr func(*hostclaim.Claim) error) (hostclaim.Observed, error) {
+ next, err := hostclaim.Update(ctx, p.Claims, obs, tr)
+ if errors.Is(err, hostclaim.ErrAlreadyRecorded) || errors.Is(err, hostclaim.ErrAckAlreadyApplied) {
+ return obs, nil
+ }
+ return next, err
+}
diff --git a/internal/hostrelocate/participant_test.go b/internal/hostrelocate/participant_test.go
new file mode 100644
index 000000000..c64b3f716
--- /dev/null
+++ b/internal/hostrelocate/participant_test.go
@@ -0,0 +1,392 @@
+package hostrelocate
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+)
+
+// fakeBox is one box's local effects. Its writes to the claim stand in for
+// what the real effects leave in storage (P built, a probe admitted).
+type fakeBox struct {
+ f *fixture
+ box string
+ coverages []string
+ // opened counts controllers opened; restarted makes Ready report none
+ // until the next Acknowledge, as a restarted agent would.
+ opened int
+ restarted bool
+ effects []string
+ op string
+ finished bool
+ // failProbes makes every probe admission fail readiness.
+ failProbes bool
+ // retainedErr is a damaged journal entry Retained reports.
+ retainedErr error
+ // retired: a scoped cleanup removed the local install; restores,
+ // reinstalls and probedMissing count what EnsureLocal and Probe saw.
+ retired bool
+ restores, reinstalls, probedMissing int
+ // receiptLost: the box's durable local install receipt is gone.
+ receiptLost bool
+}
+
+func (b *fakeBox) Acknowledge(_ context.Context, o hostclaim.Observed) (hostclaim.ExecutionAcknowledgement, error) {
+ b.opened++
+ b.restarted, b.finished, b.op = false, false, o.Claim.Op.ID
+ return hostclaim.ExecutionAcknowledgement{Op: o.Claim.Op.ID, Enrollment: enrollment(b.box), BootID: "boot", ControllerID: "ctl-" + b.box, LeaseETag: "lease", Expires: b.f.now.Add(15 * 60e9)}, nil
+}
+func (b *fakeBox) Retained() ([]Participation, error) {
+ if b.opened == 0 || b.finished {
+ return nil, b.retainedErr
+ }
+ return []Participation{{Host: host, Op: b.op}}, b.retainedErr
+}
+func (b *fakeBox) Finish(_ context.Context, h, op string) error {
+ b.finished = true
+ b.effects = append(b.effects, "finish "+op)
+ return nil
+}
+func (b *fakeBox) Ready(context.Context, hostclaim.Observed) error {
+ b.effects = append(b.effects, "ready")
+ if b.restarted || b.opened == 0 {
+ return ErrNoController
+ }
+ return nil
+}
+func (b *fakeBox) Release(_ context.Context, o hostclaim.Observed) (hostclaim.Release, error) {
+ h := o.Claim.Holder
+ return hostclaim.Release{ReleasedBy: b.box, Enrollment: enrollment(b.box), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r-" + b.box, ReceiptSHA256: strings.Repeat("d", 64)}, nil
+}
+func (b *fakeBox) Build(context.Context, hostclaim.Observed) error {
+ b.effects = append(b.effects, "build")
+ b.f.participant(func(c *hostclaim.Claim) error { c.Op.Journal["test.built"] = c.Op.Journal["prepared"]; return nil })
+ return nil
+}
+func (b *fakeBox) InstallLocal(_ context.Context, o hostclaim.Observed) (hostclaim.LocalInstall, error) {
+ b.effects = append(b.effects, "install")
+ n, _ := strconv.ParseUint(o.Claim.Op.Journal["install_stamp"], 10, 64)
+ return hostclaim.LocalInstall{Box: b.box, Enrollment: enrollment(b.box), InstallStamp: n, FilesManifestVersion: 1, JournalSHA256: strings.Repeat("e", 64)}, nil
+}
+func (b *fakeBox) EnsureLocal(_ context.Context, o hostclaim.Observed) error {
+ b.effects = append(b.effects, "ensure-local")
+ if hostclaim.PreparedOpened(o.Claim) {
+ b.reinstalls++ // never: an opened P is not reinstalled
+ }
+ if b.retired {
+ b.retired, b.restores = false, b.restores+1
+ }
+ return nil
+}
+func (b *fakeBox) VerifyReceipt(context.Context, hostclaim.Observed) (bool, error) {
+ b.effects = append(b.effects, "verify-receipt")
+ return !b.receiptLost, nil
+}
+func (b *fakeBox) Probe(ctx context.Context, o hostclaim.Observed, coverage string) (uint64, bool, error) {
+ if b.retired {
+ b.probedMissing++ // a probe admission over a missing local install
+ }
+ b.coverages = append(b.coverages, coverage)
+ c := o.Claim
+ next, err := hostclaim.Update(ctx, b.f.s, o, hostclaim.AdmitWriter(hostclaim.Admission{ObservedOpID: c.Op.ID, ObservedPhase: c.Op.Phase, ObservedSession: c.Holder.Session, ObservedGeneration: c.Holder.Generation,
+ Box: b.box, Mode: "cold", NewGeneration: c.Op.Journal["prepared"], KeepStamp: true, Coverage: coverage, Region: "ie", PredecessorRegion: "ie"}))
+ if err != nil {
+ return 0, false, err
+ }
+ if b.failProbes {
+ return next.Claim.Holder.Session, false, nil
+ }
+ return next.Claim.Holder.Session, true, nil
+}
+
+func (f *fixture) clock() time.Time { return f.now }
+
+// run steps the worker and both participants until the host serves again.
+func run(t *testing.T, f *fixture, parts ...Participant) hostclaim.Observed {
+ t.Helper()
+ ctx := context.Background()
+ for i := 0; i < 60; i++ {
+ if o := f.claim(); o.Claim.Op.Kind == hostclaim.KindServing {
+ return o
+ }
+ if _, err := f.w.Step(ctx, host); err != nil && !errors.Is(err, ErrNotYet) {
+ t.Fatalf("worker: %v", err)
+ }
+ for _, p := range parts {
+ if _, err := p.Step(ctx, host); err != nil && !errors.Is(err, ErrNotYet) {
+ t.Fatalf("%s: %v", p.Box, err)
+ }
+ }
+ }
+ t.Fatalf("no progress: %+v", f.claim().Claim.Op)
+ return hostclaim.Observed{}
+}
+
+func TestADrainCompletesWithBothParticipants(t *testing.T) {
+ f := newFixture(t)
+ src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ o := run(t, f, Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}, Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock})
+ if o.Claim.Op.Exec != "dst" || len(dst.coverages) != 1 || dst.coverages[0] != "released" {
+ t.Fatalf("exec %s coverages %v", o.Claim.Op.Exec, dst.coverages)
+ }
+ last := o.Claim.Holder.Admissions[len(o.Claim.Holder.Admissions)-1]
+ if last.WritesLost != nil {
+ t.Fatalf("a drain declared a loss: %+v", last)
+ }
+}
+
+func TestAFailoverCompletesAndDeclaresTheSource(t *testing.T) {
+ f := newFixture(t)
+ f.terminate("src")
+ dst := &fakeBox{f: f, box: "dst"}
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonFailover, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ o := run(t, f, Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock})
+ last := o.Claim.Holder.Admissions[len(o.Claim.Holder.Admissions)-1]
+ if dst.coverages[0] != "" || last.WritesLost == nil || last.WritesLost.Box != "src" {
+ t.Fatalf("coverages %v last %+v", dst.coverages, last)
+ }
+}
+
+func TestALostParticipantDoesNothing(t *testing.T) {
+ f := newFixture(t)
+ f.terminate("src")
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonFailover, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ src := &fakeBox{f: f, box: "src"}
+ before := f.claim().ETag
+ if _, err := (Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}).Step(context.Background(), host); !errors.Is(err, ErrNotYet) || f.claim().ETag != before {
+ t.Fatalf("a fenced source acted: %v", err)
+ }
+}
+
+func TestAParticipantChecksItsControllerBeforeEveryEffectAndReopensAfterARestart(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ ps := Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}
+ pd := Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock}
+ for _, p := range []Participant{ps, pd} {
+ if _, err := p.Step(ctx, host); err != nil {
+ t.Fatal(err)
+ }
+ }
+ f.worker(nil) // releasing
+ // The target's agent restarts: its next step opens a controller and
+ // journals nothing; the one after acts.
+ dst.restarted = true
+ before := f.claim().ETag
+ if _, err := pd.Step(ctx, host); !errors.Is(err, ErrNotYet) || dst.opened != 2 || f.claim().ETag != before {
+ t.Fatalf("restart: %v opened=%d wrote=%v", err, dst.opened, f.claim().ETag != before)
+ }
+ if n := len(f.claim().Claim.Op.Journal["exec_ack.dst"]); n == 0 {
+ t.Fatal("the first acknowledgement was dropped")
+ }
+ // The source releases: Ready ran before the effect.
+ src.effects = nil
+ if _, err := ps.Step(ctx, host); err != nil {
+ t.Fatal(err)
+ }
+ if len(src.effects) == 0 || src.effects[0] != "ready" {
+ t.Fatalf("release ran without checking the controller: %v", src.effects)
+ }
+ // A lost controller refuses the effect and writes nothing.
+ o := run(t, f, ps, pd)
+ if o.Claim.Op.Exec != "dst" {
+ t.Fatalf("exec %s", o.Claim.Op.Exec)
+ }
+ for i, e := range dst.effects {
+ if e == "build" && (i == 0 || dst.effects[i-1] != "ready") {
+ t.Fatalf("build without a readiness check first: %v", dst.effects)
+ }
+ }
+}
+
+func TestAParticipantGivesUpAfterThreeProbeAttempts(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst", failProbes: true}
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ ps := Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}
+ pd := Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock}
+ for i := 0; i < 40; i++ {
+ o := f.claim()
+ if o.Claim.Op.Phase == hostclaim.PhaseProbeFailed {
+ if n := len(dst.coverages); n != hostclaim.MaxProbeAttempts {
+ t.Fatalf("%d probe admissions, want %d", n, hostclaim.MaxProbeAttempts)
+ }
+ if journal := o.Claim.Op.Journal; journal["probe_attempts.dst"] != "3" {
+ t.Fatalf("attempts journaled: %v", journal["probe_attempts.dst"])
+ }
+ return
+ }
+ f.w.Step(ctx, host)
+ ps.Step(ctx, host)
+ pd.Step(ctx, host)
+ }
+ t.Fatalf("never probe-failed: %+v", f.claim().Claim.Op)
+}
+
+func TestAParticipantFinishesItsControllerOnceTheOperationIsOver(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ ps := Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}
+ pd := Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock}
+ run(t, f, ps, pd)
+ if err := pd.Sweep(ctx); err != nil || !dst.finished {
+ t.Fatalf("the serving target kept its relocation controller: %v finished=%v", err, dst.finished)
+ }
+ // A damaged journal entry beside it is reported, and the readable
+ // participation is still finished.
+ src.retainedErr = errors.New("a damaged journal entry")
+ if err := ps.Sweep(ctx); !errors.Is(err, src.retainedErr) || !src.finished {
+ t.Fatalf("the former source kept its controller, or the damaged entry was hidden: %v finished=%v", err, src.finished)
+ }
+ src.retainedErr = nil
+ // A second relocation opens a fresh controller.
+ spa := &fakeBox{f: f, box: "spa"}
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "dst", Target: "spa"}); err != nil {
+ t.Fatal(err)
+ }
+ before := dst.opened
+ run(t, f, pd, Participant{Claims: f.s, Box: "spa", Enrollment: enrollment("spa"), Effects: spa, Now: f.clock})
+ if dst.opened != before+1 {
+ t.Fatalf("the second relocation did not open its own controller: %d -> %d", before, dst.opened)
+ }
+ // Mid-operation, Sweep keeps the controller.
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "spa", Target: "sp2"}); err != nil {
+ t.Fatal(err)
+ }
+ pspa := Participant{Claims: f.s, Box: "spa", Enrollment: enrollment("spa"), Effects: spa, Now: f.clock}
+ spa.finished = false
+ if _, err := pspa.Step(ctx, host); err != nil {
+ t.Fatal(err)
+ }
+ if err := pspa.Sweep(ctx); err != nil || spa.finished {
+ t.Fatalf("a running operation's controller was finished: %v", err)
+ }
+}
+
+// Round 14 finding 2: the local install is journaled, the lease is lost,
+// and the scoped cleanup retires the unopened P's local install. Before
+// probing, and before P's first admission, the participant has the
+// install verified and restored; once P is opened it is never touched.
+func TestAParticipantRestoresARetiredLocalInstallBeforeTheFirstAdmission(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ ps := Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}
+ pd := Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock}
+ for i := 0; i < 40 && f.claim().Claim.Op.Journal["local_install"] == ""; i++ {
+ _, _ = f.w.Step(ctx, host)
+ ps.Step(ctx, host)
+ pd.Step(ctx, host)
+ }
+ if f.claim().Claim.Op.Journal["local_install"] == "" {
+ t.Fatalf("never installed locally: %+v", f.claim().Claim.Op)
+ }
+ dst.retired = true // the lease was lost; the scoped cleanup retired it
+ run(t, f, ps, pd)
+ if dst.restores != 1 || dst.probedMissing != 0 || dst.reinstalls != 0 {
+ t.Fatalf("restores %d, probes over a missing install %d, reinstalls after P %d", dst.restores, dst.probedMissing, dst.reinstalls)
+ }
+}
+
+// Round 15 finding 1: after P's first admission the local install is
+// never remade, and a box that has lost its receipt cannot rebuild it from
+// P: the next admission is refused and the operation is stuck (§5.8,
+// B2-5), across a restart. With the receipt intact, the later attempt goes
+// ahead and nothing is reinstalled or reverted: the files, env and timers
+// the admitted writers changed are theirs.
+func TestAParticipantIsStuckWhenItsReceiptIsLostAfterPOpens(t *testing.T) {
+ for _, lost := range []bool{true, false} {
+ t.Run(fmt.Sprintf("receipt lost %v", lost), func(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst", failProbes: true}
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ ps := Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}
+ pd := Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock}
+ for i := 0; i < 60 && len(dst.coverages) == 0; i++ {
+ _, _ = f.w.Step(ctx, host)
+ ps.Step(ctx, host)
+ pd.Step(ctx, host)
+ }
+ if len(dst.coverages) != 1 || !hostclaim.PreparedOpened(f.claim().Claim) {
+ t.Fatalf("P was not opened: %d admissions", len(dst.coverages))
+ }
+ dst.receiptLost, dst.restarted = lost, true // and the agent restarts
+ dst.effects = nil
+ for i := 0; i < 6; i++ {
+ pd.Step(ctx, host)
+ }
+ c := f.claim().Claim
+ for _, e := range dst.effects {
+ if e == "ensure-local" || e == "install" {
+ t.Fatalf("the local install was touched after P opened: %v", dst.effects)
+ }
+ }
+ if lost {
+ if c.Op.Journal["stuck"] == "" || len(dst.coverages) != 1 {
+ t.Fatalf("a lost receipt after P opened: stuck %q, admissions %d", c.Op.Journal["stuck"], len(dst.coverages))
+ }
+ return
+ }
+ if c.Op.Journal["stuck"] != "" || len(dst.coverages) < 2 {
+ t.Fatalf("an intact receipt: stuck %q, admissions %d", c.Op.Journal["stuck"], len(dst.coverages))
+ }
+ })
+ }
+}
+
+// The scoped cleanup can also retire the unopened P's local install after
+// the target has entered probing and before P's first admission: the
+// probing step restores it before it counts or makes an attempt.
+func TestAParticipantRestoresALocalInstallRetiredAfterItEntersProbing(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ ps := Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}
+ pd := Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock}
+ for i := 0; i < 40 && f.claim().Claim.Op.Phase != hostclaim.PhaseProbing; i++ {
+ _, _ = f.w.Step(ctx, host)
+ ps.Step(ctx, host)
+ pd.Step(ctx, host)
+ }
+ if c := f.claim().Claim; c.Op.Phase != hostclaim.PhaseProbing || hostclaim.PreparedOpened(c) || len(dst.coverages) != 0 {
+ t.Fatalf("not probing before P's first admission: %s, %d admissions", c.Op.Phase, len(dst.coverages))
+ }
+ dst.retired = true // the lease was lost after probing began
+ run(t, f, ps, pd)
+ if dst.restores != 1 || dst.probedMissing != 0 || dst.reinstalls != 0 {
+ t.Fatalf("restores %d, probes over a missing install %d, reinstalls after P %d", dst.restores, dst.probedMissing, dst.reinstalls)
+ }
+}