amadan

carlos / platform Public

host: a release is proved by its full stop receipt; it stops the writers, flushes files, wake state and env, records its positions and only then drains

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude 1ee322d6002e42259c3c30e9b8edd4552a3d9f7e parent 4879c1e
10 files changed, +3087 −1
  • internal/host/relocationcontrollers_linux_test.go +357 −0
  • internal/host/relocationeffects.go +869 −0
  • internal/host/relocationeffects_test.go +726 −0
  • internal/host/relocationflush.go +179 −0
  • internal/host/relocationlifecycle_linux_test.go +788 −0
  • internal/host/relocationreleases.go +53 −0
  • internal/host/relocationreleases_test.go +62 −0
  • internal/hostexec/readlocal.go +17 −0
  • internal/hostfiles/platform.go +33 −0
  • internal/hostrelocate/participant.go +3 −1
diff --git a/internal/host/relocationcontrollers_linux_test.go b/internal/host/relocationcontrollers_linux_test.go
new file mode 100644
index 000000000..89028d43f
--- /dev/null
+++ b/internal/host/relocationcontrollers_linux_test.go
@@ -0,0 +1,357 @@
+//go:build linux
+
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ execctl "github.com/carlosframework/platform/internal/hostexec"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostquarantine"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type activeDirectory struct{}
+
+func (activeDirectory) Lookup(context.Context, string) (hostquarantine.Actor, error) {
+ return hostquarantine.Actor{Enrollment: strings.Repeat("a", 32), State: "active"}, nil
+}
+
+// A real hostexec controller on disk: after the relocation completes,
+// Finish closes it and clears its durable record, so root's §5.10 check
+// (hostexecParticipation) lets an executable update through. While the
+// operation still runs, the clear refuses and the record stays. A cleanup
+// that failed keeps the record however often Close is called afterwards
+// (hostexec returns the failure once), and across a restart, until a
+// repeated cleanup succeeds.
+func TestFinishClearsTheDurableRecordSoUpdatesAreEligible(t *testing.T) {
+ ctx := context.Background()
+ dataDir := t.TempDir()
+ for _, name := range []string{"hostguard", ".hostexec", "locks", "exec-locks"} {
+ if err := os.Mkdir(filepath.Join(dataDir, name), 0o770); err != nil {
+ t.Fatal(err)
+ }
+ }
+ store := release.NewDirStore(t.TempDir())
+ const host = "shop.bab.oncarlos.com"
+ created := time.Date(2026, 10, 4, 1, 0, 0, 0, time.UTC)
+ nh := hostclaim.NewHolder{Family: "instance", Key: "apps/bab/jam/instances/" + host + ".json", Account: "bab", App: "jam", Host: host, Region: "ie", RecordRegion: "ie", CreatedAt: created}
+ body := []byte(`{"host":"` + host + `","account":"bab","app":"jam","region":"ie","created_at":"2026-10-04T01:00:00Z"}`)
+ obs, err := hostclaim.Read(ctx, store, host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ step := func(o hostclaim.Observed, tr func(*hostclaim.Claim) error) hostclaim.Observed {
+ t.Helper()
+ next, err := hostclaim.Update(ctx, store, o, tr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return next
+ }
+ obs = step(obs, hostclaim.BeginCreateOn(nh, body, "c1", "console", created, hostclaim.ExecutorSelection{Box: "src", Enrollment: strings.Repeat("a", 32)}))
+ obs = step(obs, hostclaim.CompleteCreate("c1", "src"))
+ obs = step(obs, hostclaim.AdmitWriter(hostclaim.Admission{Box: "src", Mode: "cold", NewGeneration: "g1", Region: "ie", KeepStamp: true}))
+ obs = step(obs, func(c *hostclaim.Claim) error {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "rel-1", Phase: hostclaim.PhaseStarting, Exec: "src", Participants: []string{"src", "dst"},
+ Intent: map[string]string{"region": "ie", "source_box": "src", "target_box": "dst"}}
+ return nil
+ })
+ failing := false
+ scoped := func(context.Context) error {
+ if failing {
+ return errors.New("cleanup: the partial install would not go")
+ }
+ return nil
+ }
+ cfg := execctl.Config{Store: store, Guard: hostguard.Guard{DataDir: dataDir, LockDir: filepath.Join(dataDir, "locks"), Claims: hostguard.StoreReader{Store: store}},
+ Directory: activeDirectory{}, Clock: execctl.SystemClock{}, LockDir: filepath.Join(dataDir, "exec-locks"),
+ CloseAdmission: func() {}, CancelWork: func() {}, Cleanup: scoped}
+ journal := filepath.Join(dataDir, "relocate", "participations")
+ controllers := func() *RelocationControllers {
+ return &RelocationControllers{Journal: journal, DataDir: dataDir,
+ Cleanup: func(ctx context.Context, _ execctl.Binding) error { return scoped(ctx) },
+ Clear: func(ctx context.Context, b execctl.Binding) error {
+ c := cfg
+ c.Binding = b
+ return execctl.ClearStale(ctx, c)
+ }}
+ }
+ e, _, _, _ := effectsFixture(t)
+ e.Controllers = controllers()
+ e.Identity = func(context.Context) (RelocationIdentity, error) {
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: strings.Repeat("a", 32), Revision: 4}, BootID: "boot", ControllerID: "ctl"}, nil
+ }
+ e.Controller = func(ctx context.Context, b execctl.Binding) (RelocationController, error) {
+ c := cfg
+ c.Binding = b
+ return OpenRelocationController(ctx, c)
+ }
+ if _, err := e.Acknowledge(ctx, obs); err != nil {
+ t.Fatal(err)
+ }
+ if err := hostexecParticipation(dataDir); err == nil {
+ t.Fatal("a participant's record did not block an update")
+ }
+ // The controller's cleanup fails: Close reports it, the record stays.
+ failing = true
+ if err := e.Finish(ctx, host, "rel-1"); err == nil {
+ t.Fatal("a failed cleanup was not reported")
+ }
+ if err := hostexecParticipation(dataDir); err == nil || len(kept(e)) != 1 {
+ t.Fatalf("a failed cleanup cleared the record: %v retained=%v", err, kept(e))
+ }
+ // Still relocating: the clear refuses, the record and the entry stay.
+ failing = false
+ if err := e.Finish(ctx, host, "rel-1"); !errors.Is(err, execctl.ErrInactive) {
+ t.Fatalf("finish during the operation: %v", err)
+ }
+ if err := hostexecParticipation(dataDir); err == nil || len(kept(e)) != 1 {
+ t.Fatalf("the record was cleared while the operation ran: %v retained=%v", err, kept(e))
+ }
+ // Served on the target. The cleanup fails again: hostexec's Close now
+ // returns nil, and the next sweep must still not clear.
+ step(obs, func(c *hostclaim.Claim) error {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "dst", Participants: []string{"dst"}}
+ return nil
+ })
+ failing = true
+ if err := e.Finish(ctx, host, "rel-1"); err == nil {
+ t.Fatal("a sweep after a failed cleanup reported success")
+ }
+ if err := hostexecParticipation(dataDir); err == nil || len(kept(e)) != 1 {
+ t.Fatalf("a nil Close cleared the record: %v retained=%v", err, kept(e))
+ }
+ // The agent restarts with the cleanup still failing: nothing clears.
+ restarted := controllers()
+ if err := restarted.Finish(ctx, host, "rel-1"); err == nil {
+ t.Fatal("a restart cleared without a successful cleanup")
+ }
+ if err := hostexecParticipation(dataDir); err == nil || len(kept(restarted)) != 1 {
+ t.Fatalf("a restart cleared the record: %v retained=%v", err, kept(restarted))
+ }
+ // The cleanup succeeds: the retry clears the record and the journal.
+ failing = false
+ if err := restarted.Finish(ctx, host, "rel-1"); err != nil {
+ t.Fatal(err)
+ }
+ if err := hostexecParticipation(dataDir); err != nil {
+ t.Fatalf("after the relocation an update is still refused: %v", err)
+ }
+ if len(kept(restarted)) != 0 {
+ t.Fatalf("retained: %v", kept(restarted))
+ }
+}
+
+// Round 12 finding 1: relocation A completes and B starts before this box
+// sweeps A. A's real controller still holds the host's lifetime lock;
+// Acknowledge for B retires it first (hostexec refuses to clear A's record
+// while B lists this box, so B's Open replaces it), under the pass's
+// bounded context, and B's controller opens. A lock held by a controller
+// this box does not retain is waited for no longer than the pass.
+func TestASuccessorRelocationRetiresItsPredecessorsController(t *testing.T) {
+ ctx := context.Background()
+ dataDir := t.TempDir()
+ for _, name := range []string{"hostguard", ".hostexec", "locks", "exec-locks"} {
+ if err := os.Mkdir(filepath.Join(dataDir, name), 0o770); err != nil {
+ t.Fatal(err)
+ }
+ }
+ store := release.NewDirStore(t.TempDir())
+ const host = "shop.bab.oncarlos.com"
+ created := time.Date(2026, 10, 4, 1, 0, 0, 0, time.UTC)
+ nh := hostclaim.NewHolder{Family: "instance", Key: "apps/bab/jam/instances/" + host + ".json", Account: "bab", App: "jam", Host: host, Region: "ie", RecordRegion: "ie", CreatedAt: created}
+ body := []byte(`{"host":"` + host + `","account":"bab","app":"jam","region":"ie","created_at":"2026-10-04T01:00:00Z"}`)
+ obs, err := hostclaim.Read(ctx, store, host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ step := func(o hostclaim.Observed, tr func(*hostclaim.Claim) error) hostclaim.Observed {
+ t.Helper()
+ next, err := hostclaim.Update(ctx, store, o, tr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return next
+ }
+ relocating := func(id, from, to string) func(*hostclaim.Claim) error {
+ return func(c *hostclaim.Claim) error {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: id, Phase: hostclaim.PhaseStarting, Exec: from, Participants: []string{from, to},
+ Intent: map[string]string{"region": "ie", "source_box": from, "target_box": to}}
+ return nil
+ }
+ }
+ obs = step(obs, hostclaim.BeginCreateOn(nh, body, "c1", "console", created, hostclaim.ExecutorSelection{Box: "src", Enrollment: strings.Repeat("a", 32)}))
+ obs = step(obs, hostclaim.CompleteCreate("c1", "src"))
+ obs = step(obs, hostclaim.AdmitWriter(hostclaim.Admission{Box: "src", Mode: "cold", NewGeneration: "g1", Region: "ie", KeepStamp: true}))
+ obs = step(obs, relocating("rel-1", "src", "dst"))
+ cfg := execctl.Config{Store: store, Guard: hostguard.Guard{DataDir: dataDir, LockDir: filepath.Join(dataDir, "locks"), Claims: hostguard.StoreReader{Store: store}},
+ Directory: activeDirectory{}, Clock: execctl.SystemClock{}, LockDir: filepath.Join(dataDir, "exec-locks"),
+ CloseAdmission: func() {}, CancelWork: func() {}, Cleanup: func(context.Context) error { return nil }}
+ lifetime, stop := context.WithCancel(ctx)
+ t.Cleanup(stop)
+ e, _, _, _ := effectsFixture(t)
+ e.Lifetime = lifetime
+ e.Controllers = &RelocationControllers{Journal: filepath.Join(dataDir, "relocate", "participations"), DataDir: dataDir,
+ Cleanup: func(context.Context, execctl.Binding) error { return nil },
+ Clear: func(ctx context.Context, b execctl.Binding) error {
+ c := cfg
+ c.Binding = b
+ return execctl.ClearStale(ctx, c)
+ }}
+ e.Identity = func(context.Context) (RelocationIdentity, error) {
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: strings.Repeat("a", 32), Revision: 4}, BootID: "boot", ControllerID: "ctl"}, nil
+ }
+ e.Controller = func(ctx context.Context, b execctl.Binding) (RelocationController, error) {
+ c := cfg
+ c.Binding = b
+ return OpenRelocationController(ctx, c)
+ }
+ if _, err := e.Acknowledge(ctx, obs); err != nil {
+ t.Fatal(err)
+ }
+ // A completes on dst, and B (dst back to src) starts before this box's
+ // sweep has finished A.
+ obs = step(obs, func(c *hostclaim.Claim) error {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "dst", Participants: []string{"dst"}}
+ return nil
+ })
+ obs = step(obs, relocating("rel-2", "dst", "src"))
+ pass, cancel := context.WithTimeout(ctx, 10*time.Second)
+ defer cancel()
+ ack, err := e.Acknowledge(pass, obs)
+ if err != nil {
+ t.Fatalf("the successor's acknowledgement: %v", err)
+ }
+ if ack.Op != "rel-2" {
+ t.Fatalf("ack: %+v", ack)
+ }
+ // A's journal entry stays until a sweep verifies the record is B's
+ // (rev 14), then it goes; B's stays.
+ if err := e.Finish(pass, host, "rel-1"); err != nil {
+ t.Fatalf("sweeping A after the handover: %v", err)
+ }
+ got, err := e.Retained()
+ if err != nil || len(got) != 1 || got[0].Op != "rel-2" {
+ t.Fatalf("retained after the handover: %+v %v", got, err)
+ }
+ if err := hostexecParticipation(dataDir); err == nil {
+ t.Fatal("B's record did not block an update")
+ }
+
+ // A lock held by a controller this box does not retain (here, another
+ // operation's opened directly): the next open waits no longer than the
+ // pass, and opens nothing.
+ if err := e.Controllers.Close(ctx); err != nil {
+ t.Fatal(err)
+ }
+ obs = step(obs, relocating("rel-3", "dst", "src"))
+ foreign := cfg
+ foreign.Binding = execctl.Binding{Box: "src", Enrollment: strings.Repeat("a", 32), Host: host, Incarnation: obs.Claim.Holder.Incarnation, Op: "rel-3", BootID: "boot"}
+ held, err := execctl.Open(lifetime, foreign)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer held.Close(ctx)
+ other := *e
+ other.Controllers = &RelocationControllers{Journal: filepath.Join(dataDir, "relocate", "other"), Cleanup: func(context.Context, execctl.Binding) error { return nil }}
+ short, cancelShort := context.WithTimeout(ctx, 300*time.Millisecond)
+ defer cancelShort()
+ began := time.Now()
+ if _, err := other.Acknowledge(short, obs); err == nil {
+ t.Fatal("opened while another controller held the host")
+ }
+ if waited := time.Since(began); waited > 5*time.Second {
+ t.Fatalf("the open waited %v, past the pass's bound", waited)
+ }
+ if got, _ := other.Controllers.Retained(); len(got) != 1 || got[0].Op != "rel-3" {
+ t.Fatalf("the bounded open retained a controller: %+v", got) // only the journal entry
+ }
+ other.Controllers.mu.Lock()
+ n := len(other.Controllers.open)
+ other.Controllers.mu.Unlock()
+ if n != 0 {
+ t.Fatal("a controller was kept from a cancelled open")
+ }
+}
+
+// Round 14: a failed clear forgets a participation only on hostexec's own
+// validated reading of the record. A parseable record missing its
+// enrollment or incarnation is one hostexec refuses, so it is never
+// "another binding": the participation is kept, across a restart, until
+// the record is repaired and a cleanup and clear succeed.
+func TestARecordHostexecRefusesKeepsItsParticipation(t *testing.T) {
+ ctx := context.Background()
+ const host = "shop.bab.oncarlos.com"
+ b := execctl.Binding{Box: "src", Enrollment: strings.Repeat("a", 32), Host: host, Incarnation: "inc", Op: "rel-1", BootID: "boot", ControllerID: "ctl"}
+ valid := execctl.State{V: 1, Binding: b, Expires: time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC), SafeUntil: 1, State: "lost"}
+ for _, missing := range []string{"enrollment", "incarnation"} {
+ t.Run(missing, func(t *testing.T) {
+ dataDir := t.TempDir()
+ if err := os.Mkdir(filepath.Join(dataDir, ".hostexec"), 0o770); err != nil {
+ t.Fatal(err)
+ }
+ path := filepath.Join(dataDir, ".hostexec", host+".json")
+ write := func(st execctl.State, drop string) {
+ t.Helper()
+ raw, _ := json.Marshal(st)
+ var m map[string]any
+ _ = json.Unmarshal(raw, &m)
+ delete(m, drop)
+ raw, _ = json.Marshal(m)
+ if err := os.WriteFile(path, raw, 0o660); err != nil {
+ t.Fatal(err)
+ }
+ }
+ write(valid, missing)
+ cleared, cleanups := 0, 0
+ clearErr := execctl.ErrState // ClearStale refuses the invalid record
+ controllers := func() *RelocationControllers {
+ return &RelocationControllers{Journal: filepath.Join(dataDir, "relocate", "participations"), DataDir: dataDir,
+ Cleanup: func(context.Context, execctl.Binding) error { cleanups++; return nil },
+ Clear: func(context.Context, execctl.Binding) error {
+ if clearErr != nil {
+ return clearErr
+ }
+ cleared++
+ return os.Remove(path)
+ }}
+ }
+ r := controllers()
+ if err := r.record(host, "rel-1", b); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.Finish(ctx, host, "rel-1"); err == nil {
+ t.Fatal("a record hostexec refuses was taken as another binding")
+ }
+ if got := kept(&RelocationEffects{Controllers: r}); len(got) != 1 {
+ t.Fatalf("the participation was forgotten: %+v", got)
+ }
+ // A restart: the journal still names it, and still nothing clears.
+ r = controllers()
+ if err := r.Finish(ctx, host, "rel-1"); err == nil || len(kept(&RelocationEffects{Controllers: r})) != 1 {
+ t.Fatalf("after a restart: %v", err)
+ }
+ // Repaired: the record is valid and this binding's; the cleanup
+ // and the clear run, and only then is the participation gone.
+ write(valid, "")
+ clearErr = nil
+ before := cleanups
+ if err := r.Finish(ctx, host, "rel-1"); err != nil || cleared != 1 || cleanups != before+1 {
+ t.Fatalf("after the repair: %v cleared=%d cleanups=%d", err, cleared, cleanups-before)
+ }
+ if got := kept(&RelocationEffects{Controllers: r}); len(got) != 0 {
+ t.Fatalf("retained after the clear: %+v", got)
+ }
+ })
+ }
+}
diff --git a/internal/host/relocationeffects.go b/internal/host/relocationeffects.go
new file mode 100644
index 000000000..b7500aa92
--- /dev/null
+++ b/internal/host/relocationeffects.go
@@ -0,0 +1,869 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+ "sync"
+
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ execctl "github.com/carlosframework/platform/internal/hostexec"
+ "github.com/carlosframework/platform/internal/hostrelocate"
+ "github.com/carlosframework/platform/internal/stopevidence"
+)
+
+// RelocationIdentity is this box's verified actor and stop identity, as
+// Plan 6b's assembly reports it (claimStopLocal).
+type RelocationIdentity struct {
+ Actor hostbox.Actor
+ BootID, ControllerID string
+}
+
+// RelocationController is the execution controller as this package uses
+// it: hostexec.Controller's Ready and Close, and Clear, which removes the
+// controller's durable local record once its cleanup is done
+// (hostexec.ClearStale with the controller's config): Close leaves the
+// record in "lost", and under §5.10 any record blocks an executable
+// update until it is cleared.
+type RelocationController interface {
+ Ready(context.Context) (execctl.State, error)
+ Close(context.Context) error
+ Clear(context.Context) error
+ // Lost is closed once the controller has lost its lease for good
+ // (hostexec latches the first failure: a failed renewal, a lost
+ // authority, a clock fault).
+ Lost() <-chan struct{}
+}
+
+// participation is one journaled participation whose controller may have
+// left a durable record: written before the controller is opened
+// (Acknowledge), removed only after the record is cleared (Finish). A
+// crash anywhere in between leaves the entry, so a restarted agent finds
+// every record it may have left, including one from a controller that
+// never reached its first Finish or whose Open failed after writing.
+type participation struct {
+ V int `json:"v"`
+ Host string `json:"host"`
+ Op string `json:"op"`
+ Binding execctl.Binding `json:"binding"`
+}
+
+// RelocationControllers retains the execution controllers this agent has
+// opened for its participations, one per {host, op}, for the agent's
+// lifetime. A controller is never bound to a pass's context (claim D6:
+// the lease lives as long as the participation). A restarted agent has
+// none and opens a fresh one; the claim keeps the first acknowledgement.
+type RelocationControllers struct {
+ mu sync.Mutex
+ open map[string]RelocationController
+ // Journal is where participations are recorded before their
+ // controller opens (<DataDir>/relocate/participations); Finish
+ // retries a journaled one after a restart, when no controller object
+ // exists, through the binding it recorded. Required: Acknowledge
+ // opens nothing without it.
+ Journal string
+ // Cleanup is the participation's scoped cleanup for one binding
+ // (Task 8a's relocationCleanup: it retires only that operation's own
+ // partial installation of that host and incarnation, after checking
+ // the binding's box and enrollment are this box's verified actor, and
+ // does nothing once this box has admitted a writer under the
+ // operation or serves the host). Finish repeats it, and needs it to
+ // succeed, in the same call as every clear: hostexec.Controller.Close
+ // returns a cleanup failure once and nil on every later call, and a
+ // restarted agent has no controller at all, so neither a nil Close
+ // nor a journal entry proves the cleanup ran. Nil clears nothing.
+ Cleanup func(context.Context, execctl.Binding) error
+ // Clear removes a journaled binding's durable record after a restart
+ // (hostexec.ClearStale with the assembly's config for the binding).
+ Clear func(context.Context, execctl.Binding) error
+ // DataDir is hostexec's data directory (Guard.DataDir), whose
+ // .hostexec/<host>.json is the durable record. When a clear fails,
+ // the participation is forgotten only if that record is verifiably
+ // not this binding's: absent, or naming another binding (rev 14).
+ // Empty: never verifiable, so a failed clear always keeps it.
+ DataDir string
+ // bindings remembers each open controller's binding.
+ bindings map[string]execctl.Binding
+}
+
+func controllerKey(host, op string) string { return host + "\x00" + op }
+
+func (r *RelocationControllers) get(host, op string) (RelocationController, bool) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ c, ok := r.open[controllerKey(host, op)]
+ return c, ok
+}
+
+func (r *RelocationControllers) put(host, op string, b execctl.Binding, c RelocationController) (RelocationController, bool) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ if r.open == nil {
+ r.open, r.bindings = map[string]RelocationController{}, map[string]execctl.Binding{}
+ }
+ if prior, ok := r.open[controllerKey(host, op)]; ok {
+ return prior, false
+ }
+ r.open[controllerKey(host, op)], r.bindings[controllerKey(host, op)] = c, b
+ return c, true
+}
+
+const journalSuffix = ".json"
+
+func (r *RelocationControllers) journalPath(host, op string) string {
+ return filepath.Join(r.Journal, host+"~"+op+journalSuffix)
+}
+
+// record journals a participation before its controller opens.
+func (r *RelocationControllers) record(host, op string, b execctl.Binding) error {
+ if r.Journal == "" {
+ return ErrClaimStopUnavailable
+ }
+ raw, err := json.Marshal(participation{V: 1, Host: host, Op: op, Binding: b})
+ if err != nil {
+ return err
+ }
+ if err := os.MkdirAll(r.Journal, 0o700); err != nil {
+ return err
+ }
+ return writeFileSynced(r.journalPath(host, op), raw, 0o600)
+}
+
+// journaled lists the committed participations on disk. Only names
+// ending in .json are committed records: writeFileSynced's .json.tmp is
+// an uncommitted write a crash may have truncated, and is ignored (its
+// participation opened nothing, because the record is written before the
+// controller opens). A committed record that cannot be read is reported,
+// joined into the error, and every readable one is still returned, so
+// one damaged entry never hides the others' cleanup.
+func (r *RelocationControllers) journaled() ([]participation, error) {
+ if r.Journal == "" {
+ return nil, nil
+ }
+ entries, err := os.ReadDir(r.Journal)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ var out []participation
+ var errs []error
+ for _, e := range entries {
+ if e.IsDir() || !strings.HasSuffix(e.Name(), journalSuffix) {
+ continue
+ }
+ raw, err := os.ReadFile(filepath.Join(r.Journal, e.Name()))
+ if err != nil {
+ errs = append(errs, err)
+ continue
+ }
+ var f participation
+ if err := json.Unmarshal(raw, &f); err != nil || f.V != 1 || f.Host == "" || f.Op == "" {
+ errs = append(errs, fmt.Errorf("%w: participation journal %s", ErrClaimStopUnavailable, e.Name()))
+ continue
+ }
+ out = append(out, f)
+ }
+ return out, errors.Join(errs...)
+}
+
+// Retained lists the participations whose controllers are open, and the
+// journaled ones whose durable record may still need clearing. A damaged
+// journal entry is returned as the error beside every readable one.
+func (r *RelocationControllers) Retained() ([]hostrelocate.Participation, error) {
+ r.mu.Lock()
+ seen := map[string]bool{}
+ var out []hostrelocate.Participation
+ for key := range r.open {
+ host, op, _ := strings.Cut(key, "\x00")
+ seen[key] = true
+ out = append(out, hostrelocate.Participation{Host: host, Op: op})
+ }
+ r.mu.Unlock()
+ journaled, err := r.journaled()
+ for _, f := range journaled {
+ if !seen[controllerKey(f.Host, f.Op)] {
+ seen[controllerKey(f.Host, f.Op)] = true
+ out = append(out, hostrelocate.Participation{Host: f.Host, Op: f.Op})
+ }
+ }
+ sort.Slice(out, func(i, j int) bool {
+ return out[i].Host < out[j].Host || (out[i].Host == out[j].Host && out[i].Op < out[j].Op)
+ })
+ return out, err
+}
+
+// Finish closes and forgets one participation's controller, releasing the
+// host's lifetime lock for the next operation, and clears its durable
+// record. Closing runs hostexec's Cleanup hook: the relocation's
+// controller is opened with a cleanup scoped to the relocation's own
+// partial work (Task 8a), never the serving writer's session, so
+// finishing after a successful relocation stops nothing that serves.
+//
+// The entry stays until Close succeeds: hostexec retains the lifetime
+// lock on a timed-out join and needs another Close, so a controller whose
+// close the pass's context cut short is found by the next sweep. Then the
+// scoped cleanup is repeated (Cleanup) and the record is cleared only if
+// that cleanup succeeded in this call: a nil Close after an earlier
+// failed one, or a restart with no controller, proves nothing. The
+// journal entry goes only once the record is gone, so a restart in
+// between retries the cleanup and the clear through the recorded binding.
+func (r *RelocationControllers) Finish(ctx context.Context, host, op string) error {
+ return r.finish(ctx, host, op, false)
+}
+
+// errSuperseded is finish's answer, for a successor's handover, when the
+// predecessor's controller is closed and cleaned up but hostexec refuses
+// to clear its record because the claim's running operation lists this
+// box (hostexec.ClearStale, controller.go: "a fresh successor operation
+// may exclude this box"; here it includes it). The lifetime lock is free,
+// and the successor's Open replaces the record under it.
+var errSuperseded = errors.New("relocation: the predecessor's record passes to its successor")
+
+func (r *RelocationControllers) finish(ctx context.Context, host, op string, successor bool) error {
+ key := controllerKey(host, op)
+ r.mu.Lock()
+ c, ok := r.open[key]
+ b := r.bindings[key]
+ r.mu.Unlock()
+ if !ok {
+ // No controller object: a restart, or a crash before the first
+ // Finish. Retry through the journaled binding.
+ journaled, err := r.journaled()
+ found := false
+ for _, f := range journaled {
+ if f.Host == host && f.Op == op {
+ found, b = true, f.Binding
+ }
+ }
+ if !found {
+ return err // nil: nothing journaled; else the damaged entry may be this one
+ }
+ }
+ if r.Cleanup == nil || (!ok && r.Clear == nil) {
+ return ErrClaimStopUnavailable
+ }
+ if ok {
+ if err := c.Close(ctx); err != nil {
+ return err // cleanup failed or was cut short: the record stays, and so does the entry
+ }
+ }
+ if err := r.Cleanup(ctx, b); err != nil {
+ return err // the record stays: no clear without a cleanup that succeeded now
+ }
+ clear := func() error { return r.Clear(ctx, b) }
+ if ok {
+ clear = func() error { return c.Clear(ctx) }
+ }
+ if !ok {
+ // No controller object: when the record is verifiably not this
+ // binding's (gone, or a successor's that holds the host's lock),
+ // there is nothing to clear, and ClearStale would only wait out
+ // the pass on the successor's lock.
+ if mine, verr := r.recordIsMine(host, b); verr == nil && !mine {
+ return r.forget(host, op)
+ }
+ }
+ superseded := false
+ if err := clear(); err != nil {
+ // Whatever the error, the participation goes only if the record
+ // itself shows it is not this binding's (round 13 finding 2:
+ // ClearStale's ErrState also means an absent claim or an invalid
+ // local record, and those leave this binding's record in place).
+ mine, verr := r.recordIsMine(host, b)
+ switch {
+ case verr != nil:
+ return errors.Join(err, verr)
+ case !mine:
+ // Verified: no record, or another binding's (a successor
+ // opened over it).
+ case successor && errors.Is(err, execctl.ErrInactive):
+ superseded = true // the lock is free; the journal stays until the successor opens
+ default:
+ return err
+ }
+ }
+ r.mu.Lock()
+ if ok && r.open[key] == c {
+ delete(r.open, key)
+ delete(r.bindings, key)
+ }
+ r.mu.Unlock()
+ if superseded {
+ return errSuperseded
+ }
+ return r.forget(host, op)
+}
+
+// recordIsMine reads hostexec's durable record for host through
+// hostexec's own validated reader (ReadLocal) and reports whether it is
+// b's binding (box, enrollment, host, incarnation, operation: what
+// ClearStale matches). Only an absent record, or a valid record of another
+// binding, is "not mine"; a record hostexec would refuse (one missing its
+// enrollment or incarnation, say) is an error, and the participation is
+// kept (round 14).
+func (r *RelocationControllers) recordIsMine(host string, b execctl.Binding) (bool, error) {
+ if r.DataDir == "" {
+ return false, fmt.Errorf("%w: no hostexec data directory to verify the record", ErrClaimStopUnavailable)
+ }
+ st, err := execctl.ReadLocal(r.DataDir, host)
+ if errors.Is(err, os.ErrNotExist) {
+ return false, nil
+ }
+ if err != nil {
+ return false, fmt.Errorf("hostexec record for %s: %w", host, err)
+ }
+ return st.Box == b.Box && st.Enrollment == b.Enrollment && st.Incarnation == b.Incarnation && st.Op == b.Op, nil
+}
+
+// live returns the open controller for {host, op}, retiring it first if it
+// has lost its lease (round 13 finding 1): hostexec latches the first
+// failure, so a lost controller never becomes ready again. Retiring joins
+// it (Close, bounded by ctx), repeats the scoped cleanup, and forgets the
+// object but keeps the journal: the record on disk stays "lost" (durable
+// denial) until a fresh controller, opened with fresh authority and its
+// own lease, replaces it. ok false: none open (or just retired).
+func (r *RelocationControllers) live(ctx context.Context, host, op string) (RelocationController, bool, error) {
+ key := controllerKey(host, op)
+ r.mu.Lock()
+ c, ok := r.open[key]
+ b := r.bindings[key]
+ r.mu.Unlock()
+ if !ok {
+ return nil, false, nil
+ }
+ select {
+ case <-c.Lost():
+ default:
+ return c, true, nil
+ }
+ if r.Cleanup == nil {
+ return nil, false, ErrClaimStopUnavailable
+ }
+ if err := c.Close(ctx); err != nil {
+ return nil, false, err // joined later; the entry stays
+ }
+ if err := r.Cleanup(ctx, b); err != nil {
+ return nil, false, err // the entry stays: the next pass repeats both
+ }
+ r.mu.Lock()
+ if r.open[key] == c {
+ delete(r.open, key)
+ delete(r.bindings, key)
+ }
+ r.mu.Unlock()
+ return nil, false, nil
+}
+
+// forget removes one participation's journal entry.
+func (r *RelocationControllers) forget(host, op string) error {
+ if r.Journal != "" {
+ if err := os.Remove(r.journalPath(host, op)); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ }
+ return nil
+}
+
+// handover retires every other participation of host this box still holds
+// before op's controller opens (round 12 finding 1): hostexec keeps one
+// controller per host, and a predecessor whose sweep has not yet run still
+// holds the host's lifetime lock, so the successor's Open would wait on it
+// for the whole pass. Each predecessor is closed and its scoped cleanup
+// repeated under the pass's context (bounded). One whose record hostexec
+// will not clear because the successor lists this box (errSuperseded)
+// keeps its journal entry: the successor's Open replaces the record under
+// the freed lock, and a later sweep forgets the entry once the record is
+// verifiably the successor's (rev 14).
+func (r *RelocationControllers) handover(ctx context.Context, host, op string) error {
+ retained, _ := r.Retained() // a damaged entry is Sweep's to report
+ for _, p := range retained {
+ if p.Host != host || p.Op == op {
+ continue
+ }
+ if err := r.finish(ctx, host, p.Op, true); err != nil && !errors.Is(err, errSuperseded) {
+ return fmt.Errorf("relocation: retiring %s's controller for %s before %s: %w", p.Op, host, op, err)
+ }
+ }
+ return nil
+}
+
+// writeFileSynced writes path whole: a synced temporary file renamed into
+// place, then the directory synced, so a committed entry survives a crash.
+func writeFileSynced(path string, body []byte, mode os.FileMode) error {
+ tmp := path + ".tmp"
+ if err := os.WriteFile(tmp, body, mode); err != nil {
+ return err
+ }
+ f, err := os.Open(tmp)
+ if err != nil {
+ return err
+ }
+ if err := f.Sync(); err != nil {
+ _ = f.Close()
+ return err
+ }
+ _ = f.Close()
+ if err := os.Rename(tmp, path); err != nil {
+ return err
+ }
+ dir, err := os.Open(filepath.Dir(path))
+ if err != nil {
+ return err
+ }
+ defer dir.Close()
+ return dir.Sync()
+}
+
+// Close closes every retained controller (agent shutdown).
+func (r *RelocationControllers) Close(ctx context.Context) error {
+ r.mu.Lock()
+ open := r.open
+ r.open = nil
+ r.mu.Unlock()
+ var first error
+ for _, c := range open {
+ if err := c.Close(ctx); err != nil && first == nil {
+ first = err
+ }
+ }
+ return first
+}
+
+// ReleasePositions keeps the flushed positions of one release durably,
+// bound to the exact stop request (host, incarnation, stamp, session,
+// operation, phase, box, enrollment, boot, controller), from before root
+// completes its receipt. A replayed receipt (root's durable fence after a
+// lost response or a crash before JournalRelease) is paired with them;
+// without a record for that exact request the release is refused rather
+// than guessed.
+// Recorded matches the stop's installation and session (host, incarnation,
+// stamp, session, operation, phase, obligation, box, enrollment), not its
+// controller identity: a restarted agent asks with a new boot and
+// controller id, and gets back the request the flush was made under, so
+// it can ask root for that stop's durable receipt.
+type ReleasePositions interface {
+ Record(ctx context.Context, r StopRequest, txid uint64, seq int64) error
+ Recorded(ctx context.Context, r StopRequest) (recorded StopRequest, txid uint64, seq int64, ok bool, err error)
+}
+
+// RelocationEffects is hostrelocate.Effects over providers Plan 6b's box
+// assembly supplies. It owns only the order of §5.8's release sequence and
+// the shape of each record; every provider acts under the host guard.
+type RelocationEffects struct {
+ // Lifetime is the agent's context: controllers are opened under it,
+ // never under a pass's.
+ Lifetime context.Context
+ Controllers *RelocationControllers
+ Identity func(context.Context) (RelocationIdentity, error)
+ // Controller is Plan 8's execution controller for one binding
+ // (hostexec.Open with the assembly's CloseAdmission, CancelWork and
+ // Cleanup). It writes pending state, checks root's update-begun marker,
+ // and takes the execution lease (§5.10, claim D6).
+ Controller func(context.Context, execctl.Binding) (RelocationController, error)
+ // Producers is the admitted session's drain (ClaimWakeAssembly.Producers).
+ Producers func(context.Context, hostclaim.Observed) (StopDrainer, error)
+ // Collect asks root for full stop evidence (Plan 6b's CollectClaimStop
+ // over the instance-control socket).
+ Collect func(context.Context, StopRequest, StopDrainer) (StopResult, error)
+ // StopWriters is release step 1's action (§5.8, A4-1): close admission,
+ // settle spawn attempts, drain in-flight timer deliveries, stop the
+ // host's process tree and wait for it to empty, keeping the replicator
+ // and the files publisher open. Writers reports what it left.
+ StopWriters func(context.Context, hostclaim.Observed) error
+ // Writers reports the drained facts and the replicator's descriptors
+ // for release step 1's observation; Observe makes it. The complete
+ // descriptor scan reads every process's table, which the agent (uid
+ // carlos) cannot: production's Observe asks root over the
+ // instance-control socket (R3); an agent-side scan fails closed on
+ // the first unreadable table.
+ //
+ // Root's scan (stopevidence.WritersStopped over /proc) must run in the
+ // host PID namespace with CAP_SYS_PTRACE: kcmp(2) takes pids in the
+ // caller's own namespace and needs ptrace read access to every task,
+ // and without either the scan is ErrIncomplete. The agent and root
+ // must also see the database path the same way in /proc (one mount
+ // namespace view): the allow-list compares DBPaths and the
+ // replicator's descriptors as /proc shows them to each reader, so a
+ // different view is ErrForeignHolder. Either mismatch fails every
+ // release closed, never open.
+ Writers func(context.Context, hostclaim.Observed) (stopevidence.WritersInputs, error)
+ Observe func(context.Context, stopevidence.WritersInputs) (stopevidence.WritersEvidence, error)
+ // FlushDB is release step 2: final checkpoint and SyncAndWait, then the
+ // stopped database's final TXID N, returned only once the newest
+ // level-0 file's MaxTXID is N. A sync error is returned, never swallowed.
+ FlushDB func(context.Context, hostclaim.Observed) (uint64, error)
+ // FlushFiles is release step 3: publish every pending file, the wake
+ // state and the env file into the session's generation, whatever
+ // files_durability says (ReleaseFlushFiles), and return the manifest's
+ // final sequence.
+ FlushFiles func(context.Context, hostclaim.Observed) (int64, error)
+ // Positions is the durable record of the flushed positions.
+ Positions ReleasePositions
+ // BuildP, Install and Admit are the target's (§5.8): moves' shared
+ // install routine for P, the local install from P's files manifest,
+ // and one cold probe admission followed by the probe.
+ BuildP func(context.Context, hostclaim.Observed) error
+ Install func(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, error)
+ // VerifyLocal reports whether the local install the receipt names is
+ // complete on this box now: the files directory as P's prepared
+ // manifest lists it, the env file and the wake state, under the
+ // receipt's install_stamp (Task 8c's production body).
+ VerifyLocal func(context.Context, hostclaim.Observed, hostclaim.LocalInstall) (bool, error)
+ // Receipt reads this box's durable local install receipt for the
+ // claim's host (Task 8c: the install journal written under its
+ // install_stamp); false when there is none.
+ Receipt func(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, bool, error)
+ Admit func(context.Context, hostclaim.Observed, string) (uint64, bool, error)
+}
+
+var _ hostrelocate.Effects = (*RelocationEffects)(nil)
+
+// Acknowledge opens this box's execution controller for the operation
+// under the agent's lifetime, retains it, and returns the lease it took.
+// A controller already open for the participation is reused.
+func (e *RelocationEffects) Acknowledge(ctx context.Context, obs hostclaim.Observed) (hostclaim.ExecutionAcknowledgement, error) {
+ if e.Identity == nil || e.Controller == nil || e.Controllers == nil || e.Lifetime == nil || obs.Claim == nil || obs.Claim.Holder == nil {
+ return hostclaim.ExecutionAcknowledgement{}, ErrClaimStopUnavailable
+ }
+ c := obs.Claim
+ ctl, ok, err := e.Controllers.live(ctx, c.Host, c.Op.ID)
+ if err != nil {
+ return hostclaim.ExecutionAcknowledgement{}, err
+ }
+ if !ok {
+ id, err := e.Identity(ctx)
+ if err != nil {
+ return hostclaim.ExecutionAcknowledgement{}, err
+ }
+ b := execctl.Binding{Box: id.Actor.Box, Enrollment: id.Actor.Enrollment, Host: c.Host, Incarnation: c.Holder.Incarnation, Op: c.Op.ID, BootID: id.BootID}
+ // Journaled before the open: hostexec writes its durable record
+ // inside Open, so a crash or a failed Open leaves a record only
+ // the journal lets a restarted agent find and clear.
+ if err := e.Controllers.record(c.Host, c.Op.ID, b); err != nil {
+ return hostclaim.ExecutionAcknowledgement{}, err
+ }
+ // A predecessor this box has not swept yet still holds the host's
+ // lifetime lock: retire it first, under this pass's context.
+ if err := e.Controllers.handover(ctx, c.Host, c.Op.ID); err != nil {
+ return hostclaim.ExecutionAcknowledgement{}, err
+ }
+ opened, err := openBounded(ctx, e.Lifetime, func(octx context.Context) (RelocationController, error) { return e.Controller(octx, b) })
+ if err != nil {
+ return hostclaim.ExecutionAcknowledgement{}, err
+ }
+ if ctl, ok = e.Controllers.put(c.Host, c.Op.ID, b, opened); !ok {
+ _ = opened.Close(ctx) // another pass opened one first
+ }
+ }
+ st, err := ctl.Ready(ctx)
+ if err != nil {
+ return hostclaim.ExecutionAcknowledgement{}, err
+ }
+ return hostclaim.ExecutionAcknowledgement{Op: c.Op.ID, Enrollment: st.Enrollment, BootID: st.BootID, ControllerID: st.ControllerID, LeaseETag: st.LeaseETag, Expires: st.Expires}, nil
+}
+
+// openBounded opens a controller that lives under lifetime (claim D6) but
+// waits for the host's lifetime lock no longer than the pass's ctx:
+// hostexec.Open takes one context for both. If ctx ends first, the open is
+// cancelled (Open returns, releasing what it took); a controller that
+// opened in the same instant is closed, never kept with a cancelled life.
+func openBounded(ctx, lifetime context.Context, open func(context.Context) (RelocationController, error)) (RelocationController, error) {
+ octx, cancel := context.WithCancel(lifetime)
+ type result struct {
+ c RelocationController
+ err error
+ }
+ ch := make(chan result, 1)
+ // hostexec.Open waits on its Cleanup hook without a context, so the
+ // open runs apart and the pass waits for it no longer than ctx.
+ go func() {
+ c, err := open(octx)
+ ch <- result{c, err}
+ }()
+ select {
+ case res := <-ch:
+ if res.err != nil {
+ cancel()
+ return nil, res.err
+ }
+ context.AfterFunc(lifetime, cancel) // octx ends only with lifetime
+ return res.c, nil
+ case <-ctx.Done():
+ cancel()
+ go func() {
+ if res := <-ch; res.c != nil {
+ _ = res.c.Close(context.Background()) // opened too late: never kept
+ }
+ }()
+ return nil, ctx.Err()
+ }
+}
+
+// Ready checks the retained controller for the operation is active now.
+func (e *RelocationEffects) Ready(ctx context.Context, obs hostclaim.Observed) error {
+ if e.Controllers == nil || obs.Claim == nil {
+ return ErrClaimStopUnavailable
+ }
+ ctl, ok, err := e.Controllers.live(ctx, obs.Claim.Host, obs.Claim.Op.ID)
+ if err != nil {
+ return err
+ }
+ if !ok {
+ return hostrelocate.ErrNoController // the participant reopens one (Acknowledge)
+ }
+ _, err = ctl.Ready(ctx)
+ return err
+}
+
+// Release runs the release sequence inside root's stop collection: root
+// records its durable fence and asks the agent to drain; the drain does
+// steps 1 to 3 before it closes the producers and acknowledges (§5.8,
+// claim P6-D1). The receipt is step 4. The flushed positions are durable
+// before the producers close, so a receipt root replays later still
+// carries them.
+func (e *RelocationEffects) Release(ctx context.Context, obs hostclaim.Observed) (hostclaim.Release, error) {
+ if e.Identity == nil || e.Producers == nil || e.Collect == nil || e.StopWriters == nil || e.Writers == nil || e.Observe == nil || e.FlushDB == nil || e.FlushFiles == nil || e.Positions == nil || obs.Claim == nil || obs.Claim.Holder == nil {
+ return hostclaim.Release{}, ErrClaimStopUnavailable
+ }
+ id, err := e.Identity(ctx)
+ if err != nil {
+ return hostclaim.Release{}, err
+ }
+ c, h := obs.Claim, obs.Claim.Holder
+ req := StopRequest{Region: id.Actor.Region, Host: c.Host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, WriterSession: h.Session,
+ OperationID: c.Op.ID, Phase: c.Op.Phase, ObligationID: RelocationReleaseObligation, Box: id.Actor.Box, Enrollment: id.Actor.Enrollment,
+ BootID: id.BootID, ControllerID: id.ControllerID, ExpectedClaimETag: obs.ETag, ExpectedDirectoryRevision: fmt.Sprint(id.Actor.Revision)}
+ // A flush this box already made for this stop, by any controller of
+ // this installation: after a restart the request is made under the
+ // identity that flushed, so root finds that stop's durable receipt.
+ recorded, txid, seq, flushed, err := e.Positions.Recorded(ctx, req)
+ if err != nil {
+ return hostclaim.Release{}, err
+ }
+ if flushed {
+ // The stop is the recorded one: root's fence and its inventory
+ // check name the directory revision the flush was made under, so
+ // a retry keeps it even when the roster has moved on since
+ // (round 8 finding 2), whether or not this is a recovery.
+ req.ExpectedDirectoryRevision = recorded.ExpectedDirectoryRevision
+ }
+ var d StopDrainer = RecoveryReleaseDrain{}
+ drain := &releaseDrain{e: e, obs: obs}
+ if flushed && (recorded.BootID != req.BootID || recorded.ControllerID != req.ControllerID) {
+ // The flushing process is gone with its producers: only root's
+ // durable receipt for the recorded identity can finish this
+ // release. The request names the recorded owner (boot, controller,
+ // directory revision) and the drain refuses; the Collect provider
+ // marks it a recovery request (6b's authenticated Recover path,
+ // Task 8a), under which root accepts another tuple for this box's
+ // session and replays the receipt.
+ req.BootID, req.ControllerID = recorded.BootID, recorded.ControllerID
+ } else {
+ if drain.inner, err = e.Producers(ctx, obs); err != nil {
+ return hostclaim.Release{}, err
+ }
+ if flushed {
+ // This process already flushed and recorded this stop, and
+ // root has no durable receipt yet (it failed after the drain
+ // acknowledged, before CompleteFence: claimstop.go). Steps 1
+ // to 3 are done and the replicator may be closed: resume at
+ // the producers' idempotent drain with the recorded positions,
+ // and let root verify and persist the receipt.
+ drain.resumed, drain.txid, drain.seq = true, txid, seq
+ }
+ d = drain
+ }
+ result, err := e.Collect(ctx, req, d)
+ if err != nil {
+ return hostclaim.Release{}, err
+ }
+ if drain.done {
+ txid, seq = drain.txid, drain.seq
+ } else if !flushed {
+ // Root replayed a durable receipt and nothing recorded a flush for
+ // this stop: never guess.
+ return hostclaim.Release{}, fmt.Errorf("%w: a replayed receipt has no recorded flushed positions for this stop", ErrClaimStopReceipt)
+ }
+ return hostclaim.Release{ReleasedBy: id.Actor.Box, Enrollment: id.Actor.Enrollment, Generation: h.Generation, Session: h.Session,
+ FinalTXID: txid, ManifestSeq: seq, Receipt: result.Key, ReceiptSHA256: result.Digest}, nil
+}
+
+// releaseDrain is the agent's drain for a release: step 1's action, its
+// observation, steps 2 and 3, the durable positions, and only then the
+// producers' own drain and acknowledgement.
+type releaseDrain struct {
+ e *RelocationEffects
+ obs hostclaim.Observed
+ inner StopDrainer
+ txid uint64
+ seq int64
+ done bool
+ // resumed: this stop's positions are already durable (an earlier
+ // attempt by this controller flushed and recorded them), so the
+ // drain goes straight to the producers' own drain.
+ resumed bool
+}
+
+// RecoveryReleaseDrain refuses: a restarted agent's producers are not the
+// stop's, and root must replay the durable receipt or nothing. The Collect
+// provider recognises it and submits the request as a recovery (6b). It is
+// main's recoveryStopDrain (claimwakestop.go), the one drain a recovered
+// stop is offered, under the name Task 8a's Collect matches.
+type RecoveryReleaseDrain = recoveryStopDrain
+
+func (d *releaseDrain) DrainClaim(ctx context.Context, r StopRequest, fence string) (DrainAcknowledgement, error) {
+ if d.resumed {
+ ack, err := d.inner.DrainClaim(ctx, r, fence)
+ if err != nil {
+ return DrainAcknowledgement{}, err
+ }
+ d.done = true
+ return ack, nil
+ }
+ if err := d.e.StopWriters(ctx, d.obs); err != nil {
+ return DrainAcknowledgement{}, err
+ }
+ in, err := d.e.Writers(ctx, d.obs)
+ if err != nil {
+ return DrainAcknowledgement{}, err
+ }
+ // Root's complete descriptor scan (see Writers): host PID namespace,
+ // CAP_SYS_PTRACE, and the agent's view of DBPaths, or it refuses.
+ if _, err := d.e.Observe(ctx, in); err != nil {
+ return DrainAcknowledgement{}, err
+ }
+ if d.txid, err = d.e.FlushDB(ctx, d.obs); err != nil {
+ return DrainAcknowledgement{}, err
+ }
+ if d.seq, err = d.e.FlushFiles(ctx, d.obs); err != nil {
+ return DrainAcknowledgement{}, err
+ }
+ if err := d.e.Positions.Record(ctx, r, d.txid, d.seq); err != nil {
+ return DrainAcknowledgement{}, err
+ }
+ ack, err := d.inner.DrainClaim(ctx, r, fence)
+ if err != nil {
+ return DrainAcknowledgement{}, err
+ }
+ d.done = true
+ return ack, nil
+}
+
+func (e *RelocationEffects) Retained() ([]hostrelocate.Participation, error) {
+ if e.Controllers == nil {
+ return nil, nil
+ }
+ return e.Controllers.Retained()
+}
+
+func (e *RelocationEffects) Finish(ctx context.Context, host, op string) error {
+ if e.Controllers == nil {
+ return ErrClaimStopUnavailable
+ }
+ return e.Controllers.Finish(ctx, host, op)
+}
+
+func (e *RelocationEffects) Build(ctx context.Context, obs hostclaim.Observed) error {
+ if e.BuildP == nil {
+ return ErrClaimStopUnavailable
+ }
+ return e.BuildP(ctx, obs)
+}
+
+func (e *RelocationEffects) InstallLocal(ctx context.Context, obs hostclaim.Observed) (hostclaim.LocalInstall, error) {
+ if e.Install == nil {
+ return hostclaim.LocalInstall{}, ErrClaimStopUnavailable
+ }
+ return e.Install(ctx, obs)
+}
+
+// EnsureLocal is hostrelocate.Effects' check before P's first admission:
+// the journaled receipt's install is verified and, if a lost lease's scoped
+// cleanup retired it, restored by the same Install provider under the same
+// install_stamp, then verified again. An opened P is never reinstalled.
+func (e *RelocationEffects) EnsureLocal(ctx context.Context, obs hostclaim.Observed) error {
+ if e.VerifyLocal == nil || e.Install == nil || obs.Claim == nil {
+ return ErrClaimStopUnavailable
+ }
+ c := obs.Claim
+ if hostclaim.PreparedOpened(c) {
+ return nil
+ }
+ var l hostclaim.LocalInstall
+ if err := json.Unmarshal([]byte(c.Op.Journal["local_install"]), &l); err != nil || l.InstallStamp == 0 {
+ return fmt.Errorf("%w: no local install receipt to verify", ErrClaimStopReceipt)
+ }
+ if ok, err := e.VerifyLocal(ctx, obs, l); err != nil || ok {
+ return err
+ }
+ got, err := e.Install(ctx, obs)
+ if err != nil {
+ return err
+ }
+ if got.Box != l.Box || got.Enrollment != l.Enrollment || got.InstallStamp != l.InstallStamp {
+ return fmt.Errorf("%w: the restored install is not the receipt's", ErrClaimStopReceipt)
+ }
+ ok, err := e.VerifyLocal(ctx, obs, l)
+ if err == nil && !ok {
+ err = fmt.Errorf("%w: the restored install did not verify", ErrClaimStopReceipt)
+ }
+ return err
+}
+
+// VerifyReceipt is hostrelocate.Effects' check before every admission
+// after P is opened: the box's durable receipt must be the one the claim
+// journaled, exactly. It reads the receipt alone: the files, env and
+// timers are the admitted writers' now, and nothing is restored.
+func (e *RelocationEffects) VerifyReceipt(ctx context.Context, obs hostclaim.Observed) (bool, error) {
+ if e.Receipt == nil || obs.Claim == nil {
+ return false, ErrClaimStopUnavailable
+ }
+ var want hostclaim.LocalInstall
+ if err := json.Unmarshal([]byte(obs.Claim.Op.Journal["local_install"]), &want); err != nil || want.InstallStamp == 0 {
+ return false, nil // no journaled receipt to match: nothing vouches for the local state
+ }
+ got, ok, err := e.Receipt(ctx, obs)
+ if err != nil {
+ return false, err
+ }
+ return ok && got == want, nil
+}
+
+func (e *RelocationEffects) Probe(ctx context.Context, obs hostclaim.Observed, coverage string) (uint64, bool, error) {
+ if e.Admit == nil {
+ return 0, false, ErrClaimStopUnavailable
+ }
+ return e.Admit(ctx, obs, coverage)
+}
+
+// execController is hostexec.Controller as a RelocationController: Clear
+// is hostexec.ClearStale with the controller's own config, the durable
+// record's removal once the claim no longer runs the operation with this
+// box (CC9). It is the Controller provider's production value (Task 8a
+// fills the assembly's hooks into cfg).
+type execController struct {
+ *execctl.Controller
+ cfg execctl.Config
+}
+
+func (c *execController) Clear(ctx context.Context) error { return execctl.ClearStale(ctx, c.cfg) }
+
+func (c *execController) Lost() <-chan struct{} { return c.Controller.Lost() }
+
+// OpenRelocationController opens hostexec's controller for cfg and wraps
+// it with its durable clear.
+func OpenRelocationController(ctx context.Context, cfg execctl.Config) (RelocationController, error) {
+ c, err := execctl.Open(ctx, cfg)
+ if err != nil {
+ return nil, err
+ }
+ return &execController{Controller: c, cfg: cfg}, nil // a pointer: the retained set compares controllers by identity
+}
diff --git a/internal/host/relocationeffects_test.go b/internal/host/relocationeffects_test.go
new file mode 100644
index 000000000..5408488f3
--- /dev/null
+++ b/internal/host/relocationeffects_test.go
@@ -0,0 +1,726 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "path/filepath"
+ "slices"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ execctl "github.com/carlosframework/platform/internal/hostexec"
+ "github.com/carlosframework/platform/internal/hostfiles"
+ "github.com/carlosframework/platform/internal/hostrelocate"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/stopevidence"
+)
+
+type emptyGroup struct{}
+
+func (emptyGroup) Populated() (bool, error) { return false, nil }
+func (emptyGroup) Members() ([]int, error) { return nil, nil }
+
+type recordingDrain struct{ calls *[]string }
+
+func (d recordingDrain) DrainClaim(context.Context, StopRequest, string) (DrainAcknowledgement, error) {
+ *d.calls = append(*d.calls, "producers")
+ return DrainAcknowledgement{FenceID: "fence"}, nil
+}
+
+// readyController is one opened controller: it reports the context it was
+// opened under, so a test can see it was the agent's, not a pass's.
+type readyController struct {
+ st execctl.State
+ opened context.Context
+ closed bool
+ cleared bool
+ lost error
+ clearErr error
+ closeErr error
+}
+
+func (c *readyController) Clear(context.Context) error {
+ if c.clearErr != nil {
+ return c.clearErr
+ }
+ c.cleared = true
+ return nil
+}
+
+func (c *readyController) Lost() <-chan struct{} {
+ ch := make(chan struct{})
+ if c.lost != nil {
+ close(ch)
+ }
+ return ch
+}
+
+func (c *readyController) Ready(context.Context) (execctl.State, error) {
+ if c.lost != nil {
+ return execctl.State{}, c.lost
+ }
+ return c.st, nil
+}
+func (c *readyController) Close(ctx context.Context) error {
+ if err := ctx.Err(); err != nil {
+ return err // hostexec retains the lifetime lock on a cut-short join
+ }
+ if c.closed {
+ return nil // hostexec: every Close after the first returns nil
+ }
+ c.closed = true
+ return c.closeErr // hostexec: a cleanup failure is returned once
+}
+
+// testControllers is a RelocationControllers with a journal on disk and a
+// scoped cleanup that succeeds.
+func testControllers(t *testing.T) *RelocationControllers {
+ return &RelocationControllers{Journal: filepath.Join(t.TempDir(), "participations"), Cleanup: func(context.Context, execctl.Binding) error { return nil }}
+}
+
+// kept is Retained's list, for tests whose journal holds nothing damaged.
+func kept(r interface {
+ Retained() ([]hostrelocate.Participation, error)
+}) []hostrelocate.Participation {
+ got, _ := r.Retained()
+ return got
+}
+
+type memPositions struct {
+ calls *[]string
+ req StopRequest
+ txid uint64
+ seq int64
+ ok bool
+}
+
+func (m *memPositions) Record(_ context.Context, r StopRequest, txid uint64, seq int64) error {
+ *m.calls = append(*m.calls, "record")
+ m.req, m.txid, m.seq, m.ok = r, txid, seq, true
+ return nil
+}
+func (m *memPositions) Recorded(_ context.Context, r StopRequest) (StopRequest, uint64, int64, bool, error) {
+ if !m.ok || !sameStop(m.req, r) {
+ return StopRequest{}, 0, 0, false, nil
+ }
+ return m.req, m.txid, m.seq, true, nil
+}
+
+func effectsFixture(t *testing.T) (*RelocationEffects, hostclaim.Observed, *[]string, *StopRequest) {
+ t.Helper()
+ calls := &[]string{}
+ seen := &StopRequest{}
+ proc := t.TempDir() // no process holds anything
+ exp := time.Date(2026, 10, 7, 10, 0, 0, 0, time.UTC)
+ e := &RelocationEffects{
+ Lifetime: context.Background(),
+ Controllers: testControllers(t),
+ Identity: func(context.Context) (RelocationIdentity, error) {
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 4}, BootID: "boot", ControllerID: "ctl"}, nil
+ },
+ Controller: func(ctx context.Context, b execctl.Binding) (RelocationController, error) {
+ return &readyController{opened: ctx, st: execctl.State{Binding: execctl.Binding{Enrollment: "enr", BootID: "boot", ControllerID: "c9"}, LeaseETag: "lease-1", Expires: exp}}, nil
+ },
+ Producers: func(context.Context, hostclaim.Observed) (StopDrainer, error) { return recordingDrain{calls}, nil },
+ StopWriters: func(context.Context, hostclaim.Observed) error {
+ *calls = append(*calls, "stop-writers")
+ return nil
+ },
+ Writers: func(context.Context, hostclaim.Observed) (stopevidence.WritersInputs, error) {
+ *calls = append(*calls, "writers")
+ return stopevidence.WritersInputs{Host: "shop.example", FenceClosed: true, SpawnsSettled: true, WorkDrained: true, Cgroup: emptyGroup{},
+ DBPaths: []string{"/data/shop.db"}, ReplicatorPID: 9, ReplicatorStart: "1", ProcRoot: proc}, nil
+ },
+ FlushDB: func(context.Context, hostclaim.Observed) (uint64, error) {
+ *calls = append(*calls, "db")
+ return 41, nil
+ },
+ FlushFiles: func(context.Context, hostclaim.Observed) (int64, error) {
+ *calls = append(*calls, "files")
+ return 8, nil
+ },
+ Positions: &memPositions{calls: calls},
+ Observe: stopevidence.WritersStopped,
+ Collect: func(ctx context.Context, r StopRequest, d StopDrainer) (StopResult, error) {
+ *seen = r
+ if _, err := d.DrainClaim(ctx, r, "fence"); err != nil {
+ return StopResult{}, err
+ }
+ *calls = append(*calls, "receipt")
+ return StopResult{Key: "control/hosts/shop.example/evidence/rel-1/src/r.json", Digest: strings.Repeat("a", 64)}, nil
+ },
+ }
+ obs := hostclaim.Observed{Host: "shop.example", ETag: "etag-7", Claim: &hostclaim.Claim{Host: "shop.example",
+ Holder: &hostclaim.Holder{Incarnation: "inc", Stamp: 5, Session: 6, Generation: "g6", WriterBox: "src"},
+ Op: hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "rel-1", Phase: hostclaim.PhaseReleasing, Exec: "src"}}}
+ return e, obs, calls, seen
+}
+
+func TestRelocationReleaseRunsStepsOneToThreeInsideTheDrain(t *testing.T) {
+ e, obs, calls, seen := effectsFixture(t)
+ r, err := e.Release(context.Background(), obs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if want := []string{"stop-writers", "writers", "db", "files", "record", "producers", "receipt"}; !slices.Equal(*calls, want) {
+ t.Fatalf("order %v, want %v", *calls, want)
+ }
+ if r.FinalTXID != 41 || r.ManifestSeq != 8 || r.Session != 6 || r.Generation != "g6" || r.ReleasedBy != "src" || r.Receipt == "" {
+ t.Fatalf("release: %+v", r)
+ }
+ if seen.ObligationID != RelocationReleaseObligation || seen.Phase != hostclaim.PhaseReleasing || seen.ExpectedClaimETag != "etag-7" || seen.ExpectedDirectoryRevision != "4" {
+ t.Fatalf("stop request: %+v", *seen)
+ }
+}
+
+func TestRelocationReleaseStopsAtTheFirstFailedStep(t *testing.T) {
+ failure := errors.New("sync failed")
+ for _, tc := range []struct {
+ name string
+ change func(*RelocationEffects)
+ want []string
+ }{
+ {"the writers cannot be stopped", func(e *RelocationEffects) {
+ e.StopWriters = func(context.Context, hostclaim.Observed) error { return failure }
+ }, nil},
+ {"writers still running", func(e *RelocationEffects) {
+ w := e.Writers
+ e.Writers = func(ctx context.Context, o hostclaim.Observed) (stopevidence.WritersInputs, error) {
+ in, err := w(ctx, o)
+ in.WorkDrained = false
+ return in, err
+ }
+ }, []string{"stop-writers", "writers"}},
+ {"the database flush", func(e *RelocationEffects) {
+ e.FlushDB = func(context.Context, hostclaim.Observed) (uint64, error) { return 0, failure }
+ }, []string{"stop-writers", "writers"}},
+ {"the files flush", func(e *RelocationEffects) {
+ e.FlushFiles = func(context.Context, hostclaim.Observed) (int64, error) { return 0, failure }
+ }, []string{"stop-writers", "writers", "db"}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ e, obs, calls, _ := effectsFixture(t)
+ tc.change(e)
+ if _, err := e.Release(context.Background(), obs); err == nil {
+ t.Fatal("released")
+ }
+ if !slices.Equal(*calls, tc.want) {
+ t.Fatalf("ran %v, want %v: nothing after a failed step", *calls, tc.want)
+ }
+ })
+ }
+}
+
+func TestRelocationReleaseReplaysOnlyRecordedPositions(t *testing.T) {
+ replayed := func(context.Context, StopRequest, StopDrainer) (StopResult, error) {
+ return StopResult{Key: "k", Digest: strings.Repeat("a", 64)}, nil // root replayed; no drain ran here
+ }
+ t.Run("no record", func(t *testing.T) {
+ e, obs, _, _ := effectsFixture(t)
+ e.Collect = replayed
+ if _, err := e.Release(context.Background(), obs); !errors.Is(err, ErrClaimStopReceipt) {
+ t.Fatalf("got %v", err)
+ }
+ })
+ t.Run("the lost response", func(t *testing.T) {
+ // The first call flushed and recorded; root completed its receipt;
+ // the response was lost. The retry pairs the replayed receipt with
+ // the recorded positions.
+ e, obs, _, _ := effectsFixture(t)
+ collect := e.Collect
+ e.Collect = func(ctx context.Context, r StopRequest, d StopDrainer) (StopResult, error) {
+ _, err := collect(ctx, r, d)
+ return StopResult{}, errors.Join(err, errors.New("connection reset"))
+ }
+ if _, err := e.Release(context.Background(), obs); err == nil {
+ t.Fatal("the lost response released")
+ }
+ e.Collect = replayed
+ r, err := e.Release(context.Background(), obs)
+ if err != nil || r.FinalTXID != 41 || r.ManifestSeq != 8 {
+ t.Fatalf("replay: %+v %v", r, err)
+ }
+ })
+ t.Run("root fails after the producers closed", func(t *testing.T) {
+ // Round 7: the drain flushed, recorded and closed the producers;
+ // root then failed before its receipt was durable (kernel
+ // verification or CompleteFence). The retry, same controller, no
+ // restart, resumes at the producers' drain with the recorded
+ // positions: no step of the release runs again.
+ e, obs, calls, seen := effectsFixture(t)
+ collect := e.Collect
+ e.Collect = func(ctx context.Context, r StopRequest, d StopDrainer) (StopResult, error) {
+ if _, err := d.DrainClaim(ctx, r, "fence"); err != nil {
+ return StopResult{}, err
+ }
+ return StopResult{}, errors.New("verify stopped: cgroup not yet empty") // no receipt persisted
+ }
+ if _, err := e.Release(context.Background(), obs); err == nil {
+ t.Fatal("a release without a receipt succeeded")
+ }
+ *calls = nil
+ e.Collect = collect
+ // Round 8 finding 2: the roster moved on between the attempts
+ // (same boot, same controller); the retry still names the
+ // revision the stop was flushed under.
+ e.Identity = func(context.Context) (RelocationIdentity, error) {
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 5}, BootID: "boot", ControllerID: "ctl"}, nil
+ }
+ r, err := e.Release(context.Background(), obs)
+ if err != nil || r.FinalTXID != 41 || r.ManifestSeq != 8 {
+ t.Fatalf("retry: %+v %v", r, err)
+ }
+ if want := []string{"producers", "receipt"}; !slices.Equal(*calls, want) {
+ t.Fatalf("the retry ran %v, want %v (no flush, no record, no writers stop again)", *calls, want)
+ }
+ if seen.ExpectedDirectoryRevision != "4" || seen.BootID != "boot" || seen.ControllerID != "ctl" {
+ t.Fatalf("the retry asked root under another stop: %+v", *seen)
+ }
+ })
+ t.Run("after a restart the recorded identity asks root", func(t *testing.T) {
+ // The agent flushed and recorded under controller "ctl", root
+ // completed its receipt, the agent restarted. The new process asks
+ // root under the recorded boot and controller (so root finds that
+ // stop's durable receipt), offers no drain of its own, and pairs
+ // the replayed receipt with the recorded positions.
+ e, obs, _, _ := effectsFixture(t)
+ if _, err := e.Release(context.Background(), obs); err != nil {
+ t.Fatal(err)
+ }
+ e.Identity = func(context.Context) (RelocationIdentity, error) {
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 5}, BootID: "boot-2", ControllerID: "ctl-2"}, nil
+ }
+ var asked StopRequest
+ recovery := false
+ e.Collect = func(_ context.Context, r StopRequest, d StopDrainer) (StopResult, error) {
+ asked = r
+ _, recovery = d.(RecoveryReleaseDrain) // Task 8a sets req.Recover from this
+ if _, err := d.DrainClaim(context.Background(), r, "fence"); !errors.Is(err, ErrClaimStopIdentity) {
+ return StopResult{}, errors.New("a restarted agent offered a drain for another process's stop")
+ }
+ return StopResult{Key: "k", Digest: strings.Repeat("a", 64)}, nil
+ }
+ r, err := e.Release(context.Background(), obs)
+ if err != nil || r.FinalTXID != 41 || r.ManifestSeq != 8 {
+ t.Fatalf("after restart: %+v %v", r, err)
+ }
+ if asked.BootID != "boot" || asked.ControllerID != "ctl" || asked.ExpectedDirectoryRevision != "4" || !recovery {
+ t.Fatalf("root was not asked as the recorded owner's recovery: %+v recovery=%v", asked, recovery)
+ }
+ })
+ t.Run("a record of another session", func(t *testing.T) {
+ e, obs, _, _ := effectsFixture(t)
+ if _, err := e.Release(context.Background(), obs); err != nil {
+ t.Fatal(err)
+ }
+ e.Collect = replayed
+ obs.Claim.Holder.Session = 7
+ if _, err := e.Release(context.Background(), obs); !errors.Is(err, ErrClaimStopReceipt) {
+ t.Fatalf("another session's positions were used: %v", err)
+ }
+ })
+}
+
+func TestReleasePositionsFileBindsTheExactStop(t *testing.T) {
+ ctx := context.Background()
+ f := ReleasePositionsFile{Dir: filepath.Join(t.TempDir(), "relocate")}
+ r := StopRequest{Host: "shop.example", Incarnation: "inc", InstallationStamp: 5, WriterSession: 6, OperationID: "rel-1", Phase: "releasing", ObligationID: RelocationReleaseObligation,
+ Box: "src", Enrollment: "enr", BootID: "boot", ControllerID: "ctl", ExpectedClaimETag: "e1", ExpectedDirectoryRevision: "4"}
+ if _, _, _, ok, err := f.Recorded(ctx, r); ok || err != nil {
+ t.Fatalf("before any record: %v %v", ok, err)
+ }
+ if err := f.Record(ctx, r, 41, 8); err != nil {
+ t.Fatal(err)
+ }
+ moved := r
+ moved.ExpectedClaimETag, moved.ExpectedDirectoryRevision = "e2", "5" // a target replacement moved the claim
+ if rec, txid, seq, ok, err := f.Recorded(ctx, moved); !ok || err != nil || txid != 41 || seq != 8 || rec.ControllerID != "ctl" {
+ t.Fatalf("recorded: %d %d %v %v", txid, seq, ok, err)
+ }
+ // A restarted controller of the same installation finds the record,
+ // and gets the identity the flush was made under.
+ restarted := r
+ restarted.BootID, restarted.ControllerID = "boot-2", "ctl-2"
+ if rec, _, _, ok, err := f.Recorded(ctx, restarted); !ok || err != nil || rec.BootID != "boot" || rec.ControllerID != "ctl" {
+ t.Fatalf("after a restart: %+v %v %v", rec, ok, err)
+ }
+ for name, change := range map[string]func(*StopRequest){
+ "session": func(r *StopRequest) { r.WriterSession = 7 },
+ "stamp": func(r *StopRequest) { r.InstallationStamp = 6 },
+ "operation": func(r *StopRequest) { r.OperationID = "rel-2" },
+ "phase": func(r *StopRequest) { r.Phase = "probe-failed" },
+ "box": func(r *StopRequest) { r.Box = "dst" },
+ } {
+ other := r
+ change(&other)
+ if _, _, _, ok, _ := f.Recorded(ctx, other); ok {
+ t.Fatalf("%s: another stop's positions were returned", name)
+ }
+ }
+ if err := os.WriteFile(filepath.Join(f.Dir, "shop.example.json"), []byte("{"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, _, _, err := f.Recorded(ctx, r); !errors.Is(err, ErrClaimStopReceipt) {
+ t.Fatalf("a damaged record: %v", err)
+ }
+}
+
+func TestRelocationAcknowledgeRetainsTheControllerUnderTheAgentLifetime(t *testing.T) {
+ e, obs, _, _ := effectsFixture(t)
+ var opened []*readyController
+ exp := time.Date(2026, 10, 7, 10, 0, 0, 0, time.UTC)
+ var bound execctl.Binding
+ e.Controller = func(ctx context.Context, b execctl.Binding) (RelocationController, error) {
+ bound = b
+ c := &readyController{opened: ctx, st: execctl.State{Binding: execctl.Binding{Enrollment: "enr", BootID: "boot", ControllerID: "c9"}, LeaseETag: "lease-1", Expires: exp}}
+ opened = append(opened, c)
+ return c, nil
+ }
+ pass, cancel := context.WithCancel(context.Background())
+ ack, err := e.Acknowledge(pass, obs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ cancel() // the pass ends; the controller must not
+ if bound.Op != "rel-1" || bound.Host != "shop.example" || bound.Incarnation != "inc" || ack.LeaseETag != "lease-1" || ack.ControllerID != "c9" || !ack.Expires.Equal(exp) {
+ t.Fatalf("binding %+v ack %+v", bound, ack)
+ }
+ if opened[0].opened.Err() != nil {
+ t.Fatal("the controller was opened under the pass's context")
+ }
+ // A second pass reuses the retained controller; Ready sees it.
+ if _, err := e.Acknowledge(context.Background(), obs); err != nil || len(opened) != 1 {
+ t.Fatalf("second pass: %v opened=%d", err, len(opened))
+ }
+ if err := e.Ready(context.Background(), obs); err != nil {
+ t.Fatal(err)
+ }
+ // A lost lease is recovered without a restart (round 13 finding 1):
+ // Ready joins the lost controller and repeats the scoped cleanup,
+ // reports no controller, and the participant's Acknowledge opens a
+ // fresh one; a cleanup that fails keeps the lost one for the next pass.
+ opened[0].lost = execctl.ErrLost
+ cleanupErr := errors.New("cleanup: a process would not stop")
+ e.Controllers.Cleanup = func(context.Context, execctl.Binding) error { return cleanupErr }
+ if err := e.Ready(context.Background(), obs); !errors.Is(err, cleanupErr) || !opened[0].closed {
+ t.Fatalf("lost controller, cleanup failing: %v closed=%v", err, opened[0].closed)
+ }
+ e.Controllers.Cleanup = func(context.Context, execctl.Binding) error { return nil }
+ if err := e.Ready(context.Background(), obs); !errors.Is(err, hostrelocate.ErrNoController) {
+ t.Fatalf("lost controller: %v", err)
+ }
+ if _, err := e.Acknowledge(context.Background(), obs); err != nil || len(opened) != 2 {
+ t.Fatalf("recovery: %v opened=%d", err, len(opened))
+ }
+ if err := e.Ready(context.Background(), obs); err != nil {
+ t.Fatalf("after recovery: %v", err)
+ }
+ // An agent restart has no retained controller: Ready says so, and
+ // Acknowledge opens a fresh one.
+ restarted := *e
+ restarted.Controllers = testControllers(t)
+ if err := restarted.Ready(context.Background(), obs); !errors.Is(err, hostrelocate.ErrNoController) {
+ t.Fatalf("after a restart: %v", err)
+ }
+ if _, err := restarted.Acknowledge(context.Background(), obs); err != nil || len(opened) != 3 {
+ t.Fatalf("reopen: %v opened=%d", err, len(opened))
+ }
+ // Finishing a participation closes its controller and forgets it; a
+ // later Acknowledge for the same host opens a new one.
+ if got := kept(&restarted); len(got) != 1 || got[0] != (hostrelocate.Participation{Host: "shop.example", Op: "rel-1"}) {
+ t.Fatalf("retained: %+v", got)
+ }
+ // A close cut short by the pass's context keeps the entry, so a later
+ // sweep can finish it; then it is forgotten.
+ cut, cancel := context.WithCancel(context.Background())
+ cancel()
+ if err := restarted.Finish(cut, "shop.example", "rel-1"); err == nil || opened[2].closed || len(kept(&restarted)) != 1 {
+ t.Fatalf("a cut-short close forgot the controller: %v closed=%v retained=%v", err, opened[2].closed, kept(&restarted))
+ }
+ if err := restarted.Finish(context.Background(), "shop.example", "rel-1"); err != nil || !opened[2].closed || !opened[2].cleared || len(kept(&restarted)) != 0 {
+ t.Fatalf("finish: %v closed=%v cleared=%v retained=%v", err, opened[2].closed, opened[2].cleared, kept(&restarted))
+ }
+ if err := restarted.Ready(context.Background(), obs); !errors.Is(err, hostrelocate.ErrNoController) {
+ t.Fatalf("after finish: %v", err)
+ }
+ obs.Claim.Op.ID = "rel-2"
+ if _, err := restarted.Acknowledge(context.Background(), obs); err != nil || len(opened) != 4 {
+ t.Fatalf("the next relocation: %v opened=%d", err, len(opened))
+ }
+ if err := restarted.Controllers.Close(context.Background()); err != nil || !opened[3].closed {
+ t.Fatalf("close: %v closed=%v", err, opened[3].closed)
+ }
+}
+
+func TestReleaseFlushFilesCarriesTheLastWakeStateAndEnv(t *testing.T) {
+ ctx := context.Background()
+ data := release.NewDirStore(t.TempDir())
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ const host = "shop.example"
+ if err := reg.Upsert(registry.Route{Host: host, Kind: "service", Addr: "10.0.0.1:3000", Status: "ready", Hibernate: true}); err != nil {
+ t.Fatal(err)
+ }
+ dir, envDir := t.TempDir(), t.TempDir()
+ if err := os.WriteFile(filepath.Join(dir, "a.txt"), []byte("hello"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(envDir, host+".env"), []byte("KEY=v\n"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ f := ReleaseFlushFiles{Data: data, Registry: reg, FilesDir: func(string) string { return dir }, EnvPath: func(h string) string { return filepath.Join(envDir, h+".env") }}
+ obs := hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Host: host, Holder: &hostclaim.Holder{Location: "instances/" + host + "/inc", Generation: "g6"}}}
+ first, err := f.Flush(ctx, obs)
+ if err != nil || first == 0 {
+ t.Fatalf("first flush: %d %v", first, err)
+ }
+ // A timer registered immediately before the release, and a new file.
+ if err := reg.ReplaceTimers(host, []registry.Timer{{Host: host, Name: "late", Kind: "schedule", Spec: "0 3 * * *", Path: "/n", NextAt: 5000, CreatedAt: 10, DueAt: 5000, Registration: "reg-late"}}); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.SetNextDue(host, 7777); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(dir, "b.txt"), []byte("later"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ seq, err := f.Flush(ctx, obs)
+ if err != nil || seq <= first {
+ t.Fatalf("second flush: %d %v", seq, err)
+ }
+ prefix := "instances/" + host + "/inc/files/g6/"
+ m, ok, err := hostfiles.Load(ctx, data, prefix)
+ if err != nil || !ok || m.Seq != seq {
+ t.Fatalf("manifest: %+v %v %v", m, ok, err)
+ }
+ if _, ok := m.Entries["b.txt"]; !ok {
+ t.Fatal("the file written before the release is not in the manifest")
+ }
+ ws, ok, err := hostfiles.LoadWake(ctx, data, prefix, m, nil)
+ if err != nil || !ok || len(ws.Timers) != 1 || ws.Timers[0].Name != "late" || ws.NextDue != 7777 {
+ t.Fatalf("wake state: %+v %v %v", ws, ok, err)
+ }
+ env, err := hostfiles.ReadPlatform(ctx, data, prefix, m, hostfiles.EnvItem, nil)
+ if err != nil || string(env) != "KEY=v\n" {
+ t.Fatalf("env: %q %v", env, err)
+ }
+ // Nothing changed: the same sequence, still the final state.
+ again, err := f.Flush(ctx, obs)
+ if err != nil || again != seq {
+ t.Fatalf("repeat: %d %v", again, err)
+ }
+ // The env file is removed before a later release: the flush stages its
+ // deletion, so the final state carries no env, not the old KEY=v.
+ if err := os.Remove(filepath.Join(envDir, host+".env")); err != nil {
+ t.Fatal(err)
+ }
+ gone, err := f.Flush(ctx, obs)
+ if err != nil || gone <= seq {
+ t.Fatalf("flush after the env file was removed: %d %v", gone, err)
+ }
+ m, ok, err = hostfiles.Load(ctx, data, prefix)
+ if err != nil || !ok || m.Seq != gone {
+ t.Fatalf("manifest after removal: %+v %v %v", m, ok, err)
+ }
+ if env, err := hostfiles.ReadPlatform(ctx, data, prefix, m, hostfiles.EnvItem, nil); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("a removed env file was still published: %q %v", env, err)
+ }
+ if again, err := f.Flush(ctx, obs); err != nil || again != gone {
+ t.Fatalf("repeat after removal: %d %v", again, err)
+ }
+ var raw map[string]any
+ body, _, _ := data.Get(ctx, hostfiles.ManifestKey(prefix))
+ _ = json.Unmarshal(body, &raw)
+ if raw["v"] != float64(hostfiles.ManifestVersion) {
+ t.Fatalf("manifest version: %v", raw["v"])
+ }
+}
+
+func TestFinishClearsTheDurableRecordAndRetriesAcrossARestart(t *testing.T) {
+ journal := filepath.Join(t.TempDir(), "participations")
+ e, obs, _, _ := effectsFixture(t)
+ cleanups := 0
+ var cleanupErr error
+ cleanup := func(_ context.Context, b execctl.Binding) error {
+ if b.Op != "rel-1" || b.Box != "src" || b.Host != "shop.example" || b.Incarnation != "inc" {
+ t.Fatalf("cleanup for another binding: %+v", b)
+ }
+ cleanups++
+ return cleanupErr
+ }
+ e.Controllers = &RelocationControllers{Journal: journal, Cleanup: cleanup}
+ var ctl *readyController
+ exp := time.Date(2026, 10, 7, 10, 0, 0, 0, time.UTC)
+ e.Controller = func(ctx context.Context, b execctl.Binding) (RelocationController, error) {
+ ctl = &readyController{opened: ctx, st: execctl.State{Binding: b, LeaseETag: "lease-1", Expires: exp}}
+ return ctl, nil
+ }
+ // Without a journal nothing opens: a record no restart could find.
+ unjournaled := *e
+ unjournaled.Controllers = &RelocationControllers{Cleanup: cleanup}
+ if _, err := unjournaled.Acknowledge(context.Background(), obs); !errors.Is(err, ErrClaimStopUnavailable) || ctl != nil {
+ t.Fatalf("a controller opened without a journal: %v", err)
+ }
+ if _, err := e.Acknowledge(context.Background(), obs); err != nil {
+ t.Fatal(err)
+ }
+ // The controller's cleanup fails. hostexec's Close reports it once.
+ ctl.closeErr = errors.New("cleanup: a process would not stop")
+ cleanupErr = ctl.closeErr
+ if err := e.Finish(context.Background(), "shop.example", "rel-1"); err == nil || !ctl.closed || ctl.cleared || len(kept(e)) != 1 {
+ t.Fatalf("a failed cleanup: %v closed=%v cleared=%v retained=%v", err, ctl.closed, ctl.cleared, kept(e))
+ }
+ // The next sweep: Close now returns nil, which proves nothing. The
+ // repeated cleanup still fails, so the record stays.
+ if err := e.Finish(context.Background(), "shop.example", "rel-1"); err == nil || ctl.cleared || len(kept(e)) != 1 {
+ t.Fatalf("a nil Close after a failed cleanup cleared the record: %v cleared=%v retained=%v", err, ctl.cleared, kept(e))
+ }
+ // The clear itself fails: the entry and the journal stay.
+ cleanupErr = nil
+ ctl.clearErr = errors.New("claim read failed")
+ if err := e.Finish(context.Background(), "shop.example", "rel-1"); err == nil || ctl.cleared || len(kept(e)) != 1 {
+ t.Fatalf("a failed clear: %v cleared=%v retained=%v", err, ctl.cleared, kept(e))
+ }
+ // The agent restarts: no controller object, the journal remains. The
+ // cleanup is repeated first; while it fails nothing is cleared.
+ cleared := []execctl.Binding{}
+ clear := func(_ context.Context, b execctl.Binding) error { cleared = append(cleared, b); return nil }
+ again := &RelocationControllers{Journal: journal, Cleanup: cleanup, Clear: clear}
+ if got := kept(again); len(got) != 1 || got[0].Op != "rel-1" {
+ t.Fatalf("after a restart the journaled participation is forgotten: %+v", got)
+ }
+ cleanupErr = errors.New("cleanup: still failing")
+ if err := again.Finish(context.Background(), "shop.example", "rel-1"); err == nil || len(cleared) != 0 || len(kept(again)) != 1 {
+ t.Fatalf("a restart cleared without a successful cleanup: %v cleared=%+v", err, cleared)
+ }
+ cleanupErr = nil
+ before := cleanups
+ if err := again.Finish(context.Background(), "shop.example", "rel-1"); err != nil || cleanups != before+1 || len(cleared) != 1 || cleared[0].Op != "rel-1" || cleared[0].Box != "src" {
+ t.Fatalf("retry after restart: %v cleanups=%d cleared=%+v", err, cleanups-before, cleared)
+ }
+ if got := kept(again); len(got) != 0 {
+ t.Fatalf("the journal entry outlived the clear: %+v", got)
+ }
+}
+
+// A crash before the first Finish (or a failed Open after hostexec wrote
+// its record) leaves no controller and no finishing state, only the
+// participation journaled before the open: a restart finds it, and an
+// uncommitted, truncated journal write beside it hides nothing.
+func TestARestartFindsAParticipationThatNeverReachedFinish(t *testing.T) {
+ journal := filepath.Join(t.TempDir(), "participations")
+ e, obs, _, _ := effectsFixture(t)
+ e.Controllers = &RelocationControllers{Journal: journal, Cleanup: func(context.Context, execctl.Binding) error { return nil }}
+ if _, err := e.Acknowledge(context.Background(), obs); err != nil {
+ t.Fatal(err)
+ }
+ // The crash: the controller object is gone. Another participation's
+ // journal write was cut short, leaving a truncated temporary file.
+ if err := os.WriteFile(filepath.Join(journal, "spa.example~rel-9.json.tmp"), []byte(`{"v":1,"ho`), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ cleaned, cleared := []execctl.Binding{}, []execctl.Binding{}
+ again := &RelocationControllers{Journal: journal,
+ Cleanup: func(_ context.Context, b execctl.Binding) error { cleaned = append(cleaned, b); return nil },
+ Clear: func(_ context.Context, b execctl.Binding) error { cleared = append(cleared, b); return nil }}
+ got, err := again.Retained()
+ if err != nil || len(got) != 1 || got[0] != (hostrelocate.Participation{Host: "shop.example", Op: "rel-1"}) {
+ t.Fatalf("restart with a truncated temporary file: %+v %v", got, err)
+ }
+ // A committed entry that cannot be read is reported, never hiding the
+ // readable one.
+ if err := os.WriteFile(filepath.Join(journal, "bad.example~rel-3.json"), []byte("{"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ got, err = again.Retained()
+ if !errors.Is(err, ErrClaimStopUnavailable) || len(got) != 1 || got[0].Host != "shop.example" {
+ t.Fatalf("a damaged committed entry: %+v %v", got, err)
+ }
+ if err := again.Finish(context.Background(), "shop.example", "rel-1"); err != nil || len(cleaned) != 1 || len(cleared) != 1 || cleared[0].Host != "shop.example" || cleared[0].Incarnation != "inc" {
+ t.Fatalf("finish after the crash: %v cleaned=%+v cleared=%+v", err, cleaned, cleared)
+ }
+ if got, _ := again.Retained(); len(got) != 0 {
+ t.Fatalf("the participation outlived its clear: %+v", got)
+ }
+}
+
+// Round 14 finding 2: EnsureLocal verifies the receipt's install, restores
+// a retired one under the same stamp and verifies it again, refuses a
+// restore that is not the receipt's, and never touches an opened P.
+func TestEnsureLocalRestoresARetiredInstallOnlyBeforePOpens(t *testing.T) {
+ ctx := context.Background()
+ e, obs, _, _ := effectsFixture(t)
+ receipt := hostclaim.LocalInstall{Box: "dst", Enrollment: "enr-d", InstallStamp: 9, FilesManifestVersion: 1, JournalSHA256: strings.Repeat("e", 64)}
+ raw, _ := json.Marshal(receipt)
+ obs.Claim.Op.Phase = hostclaim.PhaseProbing
+ obs.Claim.Op.Journal = map[string]string{"local_install": string(raw), "prepared": "P1", "install_stamp": "9"}
+ present, installs := true, 0
+ e.VerifyLocal = func(context.Context, hostclaim.Observed, hostclaim.LocalInstall) (bool, error) { return present, nil }
+ restored := receipt
+ e.Install = func(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, error) {
+ installs++
+ present = true
+ return restored, nil
+ }
+ if err := e.EnsureLocal(ctx, obs); err != nil || installs != 0 {
+ t.Fatalf("a complete install: %v installs=%d", err, installs)
+ }
+ present = false
+ if err := e.EnsureLocal(ctx, obs); err != nil || installs != 1 || !present {
+ t.Fatalf("a retired install: %v installs=%d", err, installs)
+ }
+ present, restored.InstallStamp = false, 10
+ if err := e.EnsureLocal(ctx, obs); !errors.Is(err, ErrClaimStopReceipt) {
+ t.Fatalf("a restore under another stamp: %v", err)
+ }
+ present, installs = false, 0
+ obs.Claim.Op.Journal["prepared_opened"] = "P1"
+ if err := e.EnsureLocal(ctx, obs); err != nil || installs != 0 {
+ t.Fatalf("an opened P was reinstalled: %v installs=%d", err, installs)
+ }
+}
+
+// Round 15 finding 1: after P opens, VerifyReceipt compares the box's
+// durable receipt with the journaled one and nothing else: a missing or
+// different receipt is false (the participant marks the operation stuck);
+// changed files never trigger a restore.
+func TestVerifyReceiptChecksTheReceiptOnly(t *testing.T) {
+ ctx := context.Background()
+ e, obs, _, _ := effectsFixture(t)
+ receipt := hostclaim.LocalInstall{Box: "dst", Enrollment: "enr-d", InstallStamp: 9, FilesManifestVersion: 1, JournalSHA256: strings.Repeat("e", 64)}
+ raw, _ := json.Marshal(receipt)
+ obs.Claim.Op.Journal = map[string]string{"local_install": string(raw), "prepared": "P1", "prepared_opened": "P1", "install_stamp": "9"}
+ stored, present := receipt, true
+ e.Receipt = func(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, bool, error) {
+ return stored, present, nil
+ }
+ touched := 0
+ e.VerifyLocal = func(context.Context, hostclaim.Observed, hostclaim.LocalInstall) (bool, error) {
+ touched++
+ return false, nil
+ }
+ e.Install = func(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, error) {
+ touched++
+ return receipt, nil
+ }
+ if ok, err := e.VerifyReceipt(ctx, obs); !ok || err != nil || touched != 0 {
+ t.Fatalf("an intact receipt over changed files: %v %v touched=%d", ok, err, touched)
+ }
+ present = false
+ if ok, err := e.VerifyReceipt(ctx, obs); ok || err != nil {
+ t.Fatalf("a missing receipt: %v %v", ok, err)
+ }
+ present, stored.InstallStamp = true, 8
+ if ok, err := e.VerifyReceipt(ctx, obs); ok || err != nil {
+ t.Fatalf("another installation's receipt: %v %v", ok, err)
+ }
+}
diff --git a/internal/host/relocationflush.go b/internal/host/relocationflush.go
new file mode 100644
index 000000000..fe7770d44
--- /dev/null
+++ b/internal/host/relocationflush.go
@@ -0,0 +1,179 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+
+ "github.com/carlosframework/platform/internal/controlname"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostfiles"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// ReleaseFlushFiles is release step 3 (§5.8, §5.6 "Release flush"): one
+// complete publish of the host's files directory, its wake state (timer
+// rows, NextDue, Replication) and its env file into the session's
+// generation, as one manifest revision, whatever the app's
+// files_durability. hostfiles.Publish leaves platform items out, so this
+// is the Open, StagePlatform, SyncDir, Commit sequence the contracts note
+// §2 names. It runs after StopWriters, so nothing changes under it.
+type ReleaseFlushFiles struct {
+ Data release.Store
+ Registry *registry.Registry
+ FilesDir func(host string) string
+ EnvPath func(host string) string
+ // Skip is the files publisher's own exclusion (nil publishes every
+ // regular file).
+ Skip func(rel string, isDir bool) bool
+}
+
+// Flush publishes and returns the committed manifest sequence. The env
+// file is staged when it exists; when it does not, its deletion is staged
+// (StagePlatformDelete), so a host whose env file was removed after an
+// earlier flush publishes none rather than the old bytes. A
+// manifest revision is committed even when nothing changed (Commit
+// returns the current sequence), so the release's manifest_seq is always
+// the sequence that holds this box's final state.
+func (f ReleaseFlushFiles) Flush(ctx context.Context, obs hostclaim.Observed) (int64, error) {
+ if f.Data == nil || f.Registry == nil || f.FilesDir == nil || f.EnvPath == nil || obs.Claim == nil || obs.Claim.Holder == nil {
+ return 0, ErrClaimStopUnavailable
+ }
+ h := obs.Claim.Holder
+ if h.Generation == "" || h.Location == "" || !controlname.ValidHostShape(obs.Host) {
+ return 0, ErrClaimStopUnavailable
+ }
+ p, err := hostfiles.Open(ctx, f.Data, h.Location+"/files/"+h.Generation+"/")
+ if err != nil {
+ return 0, err
+ }
+ ws, err := hostfiles.ExportWake(f.Registry, obs.Host)
+ if err != nil {
+ return 0, err
+ }
+ if err := hostfiles.StageWake(ctx, p, ws); err != nil {
+ return 0, err
+ }
+ env, err := os.ReadFile(f.EnvPath(obs.Host))
+ switch {
+ case err == nil:
+ if err := p.StagePlatform(ctx, hostfiles.EnvItem, env); err != nil {
+ return 0, err
+ }
+ case errors.Is(err, os.ErrNotExist):
+ // The env file is gone: the release's final state has none, so a
+ // previously published one is deleted, never carried forward.
+ if err := p.StagePlatformDelete(hostfiles.EnvItem); err != nil {
+ return 0, err
+ }
+ default:
+ return 0, err
+ }
+ st, err := hostfiles.SyncDir(ctx, p, f.FilesDir(obs.Host), f.Skip)
+ if err != nil {
+ return 0, err
+ }
+ if len(st.Unreadable) > 0 || len(st.Retry) > 0 || len(st.Torn) > 0 {
+ // Writers are stopped (step 1), so an unstable file is a fault,
+ // not a race: the release does not vouch for an image it could
+ // not read whole.
+ return 0, fmt.Errorf("hostfiles: release flush of %s: unreadable %q replaced %q torn %q", obs.Host, st.Unreadable, st.Retry, st.Torn)
+ }
+ return p.Commit(ctx)
+}
+
+// releasePosition is one recorded release: the exact stop request and
+// the positions its flush reached.
+type releasePosition struct {
+ V int `json:"v"`
+ Request StopRequest `json:"request"`
+ TXID uint64 `json:"txid"`
+ Seq int64 `json:"seq"`
+}
+
+// ReleasePositionsFile is ReleasePositions on disk: one file per host under
+// Dir, written whole and renamed into place, read back against the exact
+// stop request. The claim's ETag and the directory revision are not part
+// of the match: a target replacement moves both while the source's stop
+// stays the same stop.
+type ReleasePositionsFile struct {
+ Dir string
+}
+
+func (f ReleasePositionsFile) path(host string) (string, error) {
+ if f.Dir == "" || !controlname.ValidHostShape(host) {
+ return "", ErrClaimStopUnavailable
+ }
+ return filepath.Join(f.Dir, host+".json"), nil
+}
+
+func (f ReleasePositionsFile) Record(_ context.Context, r StopRequest, txid uint64, seq int64) error {
+ path, err := f.path(r.Host)
+ if err != nil {
+ return err
+ }
+ if err := os.MkdirAll(f.Dir, 0o700); err != nil {
+ return err
+ }
+ raw, err := json.Marshal(releasePosition{V: 1, Request: r, TXID: txid, Seq: seq})
+ if err != nil {
+ return err
+ }
+ tmp := path + ".tmp"
+ w, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
+ if err != nil {
+ return err
+ }
+ if _, err := w.Write(raw); err != nil {
+ _ = w.Close()
+ return err
+ }
+ if err := w.Sync(); err != nil {
+ _ = w.Close()
+ return err
+ }
+ if err := w.Close(); err != nil {
+ return err
+ }
+ if err := os.Rename(tmp, path); err != nil {
+ return err
+ }
+ dir, err := os.Open(f.Dir)
+ if err != nil {
+ return err
+ }
+ defer dir.Close()
+ return dir.Sync()
+}
+
+// sameStop is the installation and session, not the controller.
+func sameStop(a, b StopRequest) bool {
+ return a.Host == b.Host && a.Incarnation == b.Incarnation && a.InstallationStamp == b.InstallationStamp && a.WriterSession == b.WriterSession &&
+ a.OperationID == b.OperationID && a.Phase == b.Phase && a.ObligationID == b.ObligationID && a.Box == b.Box && a.Enrollment == b.Enrollment
+}
+
+func (f ReleasePositionsFile) Recorded(_ context.Context, r StopRequest) (StopRequest, uint64, int64, bool, error) {
+ path, err := f.path(r.Host)
+ if err != nil {
+ return StopRequest{}, 0, 0, false, err
+ }
+ raw, err := os.ReadFile(path)
+ if errors.Is(err, os.ErrNotExist) {
+ return StopRequest{}, 0, 0, false, nil
+ }
+ if err != nil {
+ return StopRequest{}, 0, 0, false, err
+ }
+ var p releasePosition
+ if err := json.Unmarshal(raw, &p); err != nil || p.V != 1 {
+ return StopRequest{}, 0, 0, false, fmt.Errorf("%w: release positions for %s", ErrClaimStopReceipt, r.Host)
+ }
+ if !sameStop(p.Request, r) {
+ return StopRequest{}, 0, 0, false, nil
+ }
+ return p.Request, p.TXID, p.Seq, true, nil
+}
diff --git a/internal/host/relocationlifecycle_linux_test.go b/internal/host/relocationlifecycle_linux_test.go
new file mode 100644
index 000000000..2ad97642c
--- /dev/null
+++ b/internal/host/relocationlifecycle_linux_test.go
@@ -0,0 +1,788 @@
+//go:build linux
+
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "math/rand"
+ "os"
+ "path/filepath"
+ "slices"
+ "strconv"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ execctl "github.com/carlosframework/platform/internal/hostexec"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostrelocate"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// lifeSeeds is the property test's seed count (rev 14: 200 by default).
+// Run it niced; it is not a sweep.
+const lifeSeeds = 200
+
+// lifeStore is the bucket with faults: lease puts that fail (a renewal
+// that cannot reach the store) and claim reads that fail or find nothing.
+type lifeStore struct {
+ release.Store
+ mu sync.Mutex
+ failLease bool
+ claimErr error
+ claimKey string
+}
+
+func (s *lifeStore) Put(ctx context.Context, key string, b []byte, etag string) (string, error) {
+ s.mu.Lock()
+ fail := s.failLease && strings.Contains(key, "/exec/")
+ s.mu.Unlock()
+ if fail {
+ return "", errors.New("lease put: store unreachable")
+ }
+ return s.Store.Put(ctx, key, b, etag)
+}
+
+func (s *lifeStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ s.mu.Lock()
+ err := s.claimErr
+ s.mu.Unlock()
+ if err != nil && key == s.claimKey {
+ return nil, "", err
+ }
+ return s.Store.Get(ctx, key)
+}
+
+func (s *lifeStore) DeleteIfMatch(ctx context.Context, key, etag string) error {
+ return release.DeleteIfMatch(ctx, s.Store, key, etag)
+}
+
+// lifeClock is hostexec's clock under the test's hand: every After fires
+// on tick, so a renewal happens exactly when a step says so.
+type lifeClock struct {
+ mu sync.Mutex
+ wall time.Time
+ boot int64
+ waiters []chan time.Time
+}
+
+func (c *lifeClock) Now() (time.Time, int64, error) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return c.wall, c.boot, nil
+}
+
+func (c *lifeClock) After(time.Duration) <-chan time.Time {
+ ch := make(chan time.Time, 1)
+ c.mu.Lock()
+ c.waiters = append(c.waiters, ch)
+ c.mu.Unlock()
+ return ch
+}
+
+func (c *lifeClock) tick() {
+ c.mu.Lock()
+ c.wall, c.boot = c.wall.Add(time.Second), c.boot+int64(time.Second)
+ ws := c.waiters
+ c.waiters = nil
+ now := c.wall
+ c.mu.Unlock()
+ for _, w := range ws {
+ w <- now
+ }
+}
+
+// lifeWorld is one box ("src") taking part in a host's relocations over
+// real hostexec controllers, with the faults a seed injects.
+type lifeWorld struct {
+ t *testing.T
+ ctx context.Context
+ inner release.Store
+ store *lifeStore
+ clock *lifeClock
+ dataDir string
+ host string
+ cfg execctl.Config
+ e *RelocationEffects
+ p hostrelocate.Participant
+ ops int
+ mu sync.Mutex
+ cleanErr bool // both cleanups fail
+ block chan struct{} // non-nil: both cleanups wait for it (a close that times out)
+ cleanOK map[string]bool
+ // installed is this box's local install per operation (rev 15): the
+ // Install provider makes it, a scoped cleanup of an operation whose P
+ // is unopened retires it, VerifyLocal reads it.
+ installed map[string]bool
+ // gen is the agent process's generation: a crash starts the next one.
+ // A dead generation's hooks never complete (rev 16).
+ gen int
+ // clears and opens count durable records cleared and controllers
+ // opened in the current step, for invariant (h).
+ clears, opens int
+ // the record as the previous step left it.
+ prevExists, prevReadable bool
+ prevOp string
+ saved []byte // a record the seed corrupted, to restore on recovery
+ violated []string
+}
+
+const lifeDeadline = 400 * time.Millisecond
+
+func newLifeWorld(t *testing.T) *lifeWorld {
+ t.Helper()
+ ctx := context.Background()
+ w := &lifeWorld{t: t, ctx: ctx, host: "shop.bab.oncarlos.com", clock: &lifeClock{wall: time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC), boot: int64(time.Hour)}, cleanOK: map[string]bool{}, installed: map[string]bool{}}
+ w.dataDir = t.TempDir()
+ for _, name := range []string{"hostguard", ".hostexec", "locks", "exec-locks"} {
+ if err := os.Mkdir(filepath.Join(w.dataDir, name), 0o770); err != nil {
+ t.Fatal(err)
+ }
+ }
+ w.inner = release.NewDirStore(t.TempDir())
+ key, _ := hostclaim.Key(w.host)
+ w.store = &lifeStore{Store: w.inner, claimKey: key}
+ created := time.Date(2026, 10, 4, 1, 0, 0, 0, time.UTC)
+ nh := hostclaim.NewHolder{Family: "instance", Key: "apps/bab/jam/instances/" + w.host + ".json", Account: "bab", App: "jam", Host: w.host, Region: "ie", RecordRegion: "ie", CreatedAt: created}
+ body := []byte(`{"host":"` + w.host + `","account":"bab","app":"jam","region":"ie","created_at":"2026-10-04T01:00:00Z"}`)
+ w.claim(hostclaim.BeginCreateOn(nh, body, "c1", "console", created, hostclaim.ExecutorSelection{Box: "src", Enrollment: strings.Repeat("a", 32)}))
+ w.claim(hostclaim.CompleteCreate("c1", "src"))
+ w.claim(hostclaim.AdmitWriter(hostclaim.Admission{Box: "src", Mode: "cold", NewGeneration: "g1", Region: "ie", KeepStamp: true}))
+ w.cfg = execctl.Config{Store: w.store, Guard: hostguard.Guard{DataDir: w.dataLink(0), LockDir: filepath.Join(w.dataDir, "locks"), Claims: hostguard.StoreReader{Store: w.store}},
+ Directory: activeDirectory{}, Clock: w.clock, LockDir: filepath.Join(w.dataDir, "exec-locks"),
+ CloseAdmission: func() {}, CancelWork: func() {}}
+ lifetime, stop := context.WithCancel(ctx)
+ t.Cleanup(stop)
+ w.e = &RelocationEffects{Lifetime: lifetime,
+ Identity: func(context.Context) (RelocationIdentity, error) {
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: strings.Repeat("a", 32), Revision: 4}, BootID: "boot", ControllerID: "ctl"}, nil
+ },
+ Controller: func(ctx context.Context, b execctl.Binding) (RelocationController, error) {
+ c := w.cfg
+ c.Binding = b
+ // hostexec's own Cleanup hook is the same scoped cleanup,
+ // bound to the binding (Task 8a); it does not count for (a).
+ if cur := w.current(); cur.Op.ID == b.Op && !slices.Contains(cur.Op.Participants, "src") {
+ w.violate(fmt.Sprintf("(i) %s opened again after this box was excluded from it", b.Op))
+ }
+ gen := w.generation()
+ c.Cleanup = func(context.Context) error { return w.cleanupScoped(context.Background(), b.Op, false, gen) }
+ opened, err := OpenRelocationController(ctx, c)
+ if err != nil {
+ return nil, err
+ }
+ w.mu.Lock()
+ w.opens++
+ w.mu.Unlock()
+ return &lifeController{RelocationController: opened, w: w, op: b.Op}, nil
+ },
+ Install: func(_ context.Context, o hostclaim.Observed) (hostclaim.LocalInstall, error) {
+ c := w.current()
+ if hostclaim.PreparedOpened(c) {
+ w.violate("(g) the local install was remade after P was opened")
+ }
+ w.mu.Lock()
+ w.installed[c.Op.ID] = true
+ w.mu.Unlock()
+ n, _ := strconv.ParseUint(c.Op.Journal["install_stamp"], 10, 64)
+ return hostclaim.LocalInstall{Box: "src", Enrollment: strings.Repeat("a", 32), InstallStamp: n, FilesManifestVersion: 1, JournalSHA256: strings.Repeat("e", 64)}, nil
+ },
+ Receipt: func(_ context.Context, o hostclaim.Observed) (hostclaim.LocalInstall, bool, error) {
+ var l hostclaim.LocalInstall
+ _ = json.Unmarshal([]byte(o.Claim.Op.Journal["local_install"]), &l)
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ return l, w.installed[o.Claim.Op.ID], nil
+ },
+ VerifyLocal: func(_ context.Context, o hostclaim.Observed, _ hostclaim.LocalInstall) (bool, error) {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ return w.installed[o.Claim.Op.ID], nil
+ },
+ Admit: func(_ context.Context, o hostclaim.Observed, _ string) (uint64, bool, error) {
+ c := w.current()
+ w.mu.Lock()
+ present := w.installed[c.Op.ID]
+ w.mu.Unlock()
+ if !hostclaim.PreparedOpened(c) && !present {
+ w.violate("(f) P's first admission over a missing local install")
+ }
+ // The admission opens P (the claim journals it).
+ if _, err := hostclaim.Update(context.Background(), w.inner, o, func(c *hostclaim.Claim) error {
+ c.Op.Journal["prepared_opened"] = c.Op.Journal["prepared"]
+ return nil
+ }); err != nil {
+ return 0, false, err
+ }
+ return 0, false, nil // the probe is still running
+ }}
+ w.e.Controllers = w.controllers()
+ w.p = hostrelocate.Participant{Claims: w.store, Box: "src", Enrollment: strings.Repeat("a", 32), Effects: w.e, Now: func() time.Time { return w.clock.wall }}
+ return w
+}
+
+// lifeController checks invariant (a) at every clear of a live
+// controller's record.
+type lifeController struct {
+ RelocationController
+ w *lifeWorld
+ op string
+}
+
+func (c *lifeController) Clear(ctx context.Context) error {
+ c.w.clearing(c.op)
+ err := c.RelocationController.Clear(ctx)
+ c.w.cleared(err)
+ return err
+}
+
+func (w *lifeWorld) cleared(err error) {
+ if err == nil {
+ w.mu.Lock()
+ w.clears++
+ w.mu.Unlock()
+ }
+}
+
+func (w *lifeWorld) generation() int {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ return w.gen
+}
+
+func (w *lifeWorld) controllers() *RelocationControllers {
+ gen := w.generation()
+ return &RelocationControllers{Journal: filepath.Join(w.dataDir, "relocate", "participations"), DataDir: w.dataDir,
+ Cleanup: func(ctx context.Context, b execctl.Binding) error { return w.cleanupScoped(ctx, b.Op, true, gen) },
+ Clear: func(ctx context.Context, b execctl.Binding) error {
+ w.clearing(b.Op)
+ c := w.cfg
+ c.Binding = b
+ err := execctl.ClearStale(ctx, c)
+ w.cleared(err)
+ return err
+ }}
+}
+
+// cleanupScoped is the scoped cleanup (Task 8a's relocationCleanup):
+// it retires the operation's local install unless the claim has opened
+// that operation's P (it never stops what serves). counts says whether it
+// is RelocationControllers' own call, the one invariant (a) asks for.
+func (w *lifeWorld) cleanupScoped(ctx context.Context, op string, counts bool, gen int) error {
+ w.mu.Lock()
+ block, fail, dead := w.block, w.cleanErr, gen != w.gen
+ w.mu.Unlock()
+ if dead {
+ select {} // a crashed process's hooks never complete
+ }
+ w.mu.Lock()
+ if counts {
+ w.cleanOK[op] = false
+ }
+ w.mu.Unlock()
+ if block != nil {
+ select {
+ case <-block:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ }
+ if gen != w.generation() {
+ select {} // the process died while its cleanup waited
+ }
+ if fail {
+ return errors.New("cleanup: a partial install would not go")
+ }
+ c, err := hostclaim.Read(context.Background(), w.inner, w.host)
+ opened := err == nil && !c.Absent && c.Claim.Op.ID == op && hostclaim.PreparedOpened(c.Claim)
+ w.mu.Lock()
+ if !opened {
+ w.installed[op] = false
+ }
+ if counts {
+ w.cleanOK[op] = true
+ }
+ w.mu.Unlock()
+ return nil
+}
+
+// clearing is invariant (a): a record is cleared only straight after a
+// scoped cleanup of the same participation succeeded, in the same call.
+func (w *lifeWorld) clearing(op string) {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ if !w.cleanOK[op] {
+ w.violated = append(w.violated, "(a) clear of "+op+" without a successful scoped cleanup first")
+ }
+ w.cleanOK[op] = false
+}
+
+func (w *lifeWorld) claim(tr func(*hostclaim.Claim) error) {
+ w.t.Helper()
+ o, err := hostclaim.Read(w.ctx, w.inner, w.host)
+ if err != nil {
+ w.t.Fatal(err)
+ }
+ if _, err := hostclaim.Update(w.ctx, w.inner, o, tr); err != nil {
+ w.t.Fatal(err)
+ }
+}
+
+func (w *lifeWorld) current() *hostclaim.Claim {
+ w.t.Helper()
+ o, err := hostclaim.Read(w.ctx, w.inner, w.host)
+ if err != nil {
+ w.t.Fatal(err)
+ }
+ return o.Claim
+}
+
+// startRelocation starts the next relocation, src taking part (as source
+// or as target), whatever the claim was doing.
+func (w *lifeWorld) startRelocation() {
+ w.ops++
+ id := fmt.Sprintf("rel-%d", w.ops)
+ from, to := "src", "dst"
+ if exec := w.current().Op.Exec; exec != "" && exec != "src" && exec != "none" {
+ from, to = exec, "src"
+ }
+ w.claim(func(c *hostclaim.Claim) error {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: id, Phase: hostclaim.PhaseStarting, Exec: from, Participants: []string{from, to},
+ Intent: map[string]string{"region": "ie", "source_box": from, "target_box": to, "target_enrollment": map[string]string{"src": strings.Repeat("a", 32), "dst": strings.Repeat("d", 32), "sp2": strings.Repeat("e", 32)}[to]}}
+ return nil
+ })
+}
+
+// replaced is the worker replacing this box, the target, within the same
+// operation (rev 19, round 18 finding 1): after its release in
+// probe-failed, or at any target phase, the relocation goes on with sp2
+// and this box is no longer a participant. Its controller, record and
+// journal must go, and nothing for this operation may open again here.
+func (w *lifeWorld) replaced() bool {
+ c := w.current()
+ if c.Op.Kind != hostclaim.KindRelocating || c.Op.Intent["target_box"] != "src" {
+ return false
+ }
+ w.claim(func(c *hostclaim.Claim) error {
+ if c.Op.Journal == nil {
+ c.Op.Journal = map[string]string{}
+ }
+ c.Op.Participants = append(slices.DeleteFunc(slices.Clone(c.Op.Participants), func(b string) bool { return b == "src" }), "sp2")
+ c.Op.Intent["target_box"], c.Op.Intent["target_enrollment"] = "sp2", strings.Repeat("e", 32)
+ if c.Op.Phase != hostclaim.PhaseStarting && c.Op.Phase != hostclaim.PhaseReleasing {
+ c.Op.Exec, c.Op.Phase = "none", hostclaim.PhasePinning
+ for _, k := range []string{"local_install", "probe", "prepared", "prepared_opened", "install_stamp", "release", "release.phase", "probing_since", "probe_failed_since"} {
+ delete(c.Op.Journal, k)
+ }
+ }
+ c.Op.Journal["excluded_targets"] = `["src"]`
+ c.Op.Journal["exec_ack.sp2"] = "{}"
+ return nil
+ })
+ return true
+}
+
+// probeFailedAndReleased moves this box, the target, to probe-failed with
+// its release journaled (the worker's and its own half, in short).
+func (w *lifeWorld) probeFailedAndReleased() {
+ w.claim(func(c *hostclaim.Claim) error {
+ if c.Op.Journal == nil {
+ c.Op.Journal = map[string]string{}
+ }
+ c.Op.Phase = hostclaim.PhaseProbeFailed
+ c.Op.Journal["probe_failed_since"] = w.clock.wall.Format(time.RFC3339Nano)
+ c.Op.Journal["release"] = `{"released_by":"src"}`
+ c.Op.Journal["release.phase"] = hostclaim.PhaseProbeFailed
+ return nil
+ })
+}
+
+// install is the worker's half while this box is the target: P built and
+// the operation handed to it (installed), its local install still to make.
+func (w *lifeWorld) install() {
+ c := w.current()
+ if c.Op.Kind != hostclaim.KindRelocating || c.Op.Intent["target_box"] != "src" || (c.Op.Phase != hostclaim.PhaseStarting && c.Op.Phase != hostclaim.PhaseReleasing) {
+ return
+ }
+ w.claim(func(c *hostclaim.Claim) error {
+ c.Op.Phase, c.Op.Exec = hostclaim.PhaseInstalled, "src"
+ if c.Op.Journal == nil {
+ c.Op.Journal = map[string]string{}
+ }
+ c.Op.Journal["install_stamp"] = strconv.FormatUint(c.Gen+1, 10)
+ c.Op.Journal["prepared"] = "P-" + c.Op.ID
+ if c.Op.Journal["exec_ack.dst"] == "" {
+ c.Op.Journal["exec_ack.dst"] = "{}"
+ }
+ return nil
+ })
+}
+
+func (w *lifeWorld) complete() {
+ if c := w.current(); c.Op.Kind == hostclaim.KindRelocating {
+ box := c.Op.Intent["target_box"]
+ w.claim(func(c *hostclaim.Claim) error {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: box, Participants: []string{box}}
+ return nil
+ })
+ }
+}
+
+// pass is one participant pass, Step then Sweep as the agent runs them,
+// under a deadline; (d) is that it returns within it.
+func (w *lifeWorld) pass() {
+ w.mu.Lock()
+ w.cleanOK = map[string]bool{} // "the same call": nothing carries over from an earlier pass
+ w.mu.Unlock()
+ done := make(chan struct{})
+ start := time.Now()
+ go func() {
+ defer close(done)
+ ctx, cancel := context.WithTimeout(w.ctx, lifeDeadline)
+ defer cancel()
+ _, _ = w.p.Step(ctx, w.host)
+ _ = w.p.Sweep(ctx)
+ }()
+ select {
+ case <-done:
+ if took := time.Since(start); took > lifeDeadline+time.Second {
+ w.violate(fmt.Sprintf("(d) a pass took %v past its %v deadline", took, lifeDeadline))
+ }
+ case <-time.After(lifeDeadline + 3*time.Second):
+ w.t.Fatalf("(d) a pass never returned within %v", lifeDeadline+3*time.Second)
+ }
+}
+
+func (w *lifeWorld) violate(s string) {
+ w.mu.Lock()
+ w.violated = append(w.violated, s)
+ w.mu.Unlock()
+}
+
+// renew fires hostexec's timers so every controller renews (or loses).
+func (w *lifeWorld) renew() {
+ w.clock.tick()
+ time.Sleep(20 * time.Millisecond)
+}
+
+func (w *lifeWorld) record() (execctl.State, bool, bool) {
+ raw, err := os.ReadFile(filepath.Join(w.dataDir, ".hostexec", w.host+".json"))
+ if errors.Is(err, os.ErrNotExist) {
+ return execctl.State{}, false, true
+ }
+ var st execctl.State
+ if err != nil || json.Unmarshal(raw, &st) != nil {
+ return st, true, false
+ }
+ return st, true, true
+}
+
+func (w *lifeWorld) open() []RelocationController {
+ r := w.e.Controllers
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ var out []RelocationController
+ for _, c := range r.open {
+ out = append(out, c)
+ }
+ return out
+}
+
+// check is invariants (a) to (c) after a step.
+func (w *lifeWorld) check(step string) {
+ st, exists, readable := w.record()
+ if exists && readable {
+ retained, _ := w.e.Controllers.Retained()
+ if !slices.ContainsFunc(retained, func(p hostrelocate.Participation) bool { return p.Host == w.host && p.Op == st.Op }) {
+ w.violate(fmt.Sprintf("(b) after %s: %s's record exists but its participation is forgotten (retained %+v)", step, st.Op, retained))
+ }
+ }
+ if n := len(w.open()); n > 1 {
+ w.violate(fmt.Sprintf("(c) after %s: %d controllers for one host", step, n))
+ }
+ // (h): a record goes only by a clear or by a successor's open over it,
+ // never by a crash or any other step; while one exists, an executable
+ // update is refused (root's §5.10 check).
+ w.mu.Lock()
+ clears, opens := w.clears, w.opens
+ w.clears, w.opens = 0, 0
+ w.mu.Unlock()
+ if w.prevExists && w.prevReadable && (!exists || (readable && st.Op != w.prevOp)) && clears == 0 && opens == 0 {
+ w.violate(fmt.Sprintf("(h) after %s: %s's record went with no clear and no successor's open", step, w.prevOp))
+ }
+ if exists && hostexecParticipation(w.dataDir) == nil {
+ w.violate(fmt.Sprintf("(h) after %s: a record exists but an executable update is not refused", step))
+ }
+ w.prevExists, w.prevReadable, w.prevOp = exists, readable, st.Op
+}
+
+// dataLink is generation gen's path to the data directory: a symlink, so
+// a crash can cut a dead generation off from it. A dead process writes
+// nothing; in-process, a dead controller's goroutines are still running
+// (hostexec's save loop retries a "lost" write until it lands), so the
+// link it opens through is removed.
+func (w *lifeWorld) dataLink(gen int) string {
+ link := filepath.Join(filepath.Dir(w.dataDir), fmt.Sprintf("data-%d", gen))
+ if err := os.Symlink(w.dataDir, link); err != nil && !os.IsExist(err) {
+ w.t.Fatal(err)
+ }
+ return link
+}
+
+// crash is the agent process dying abruptly (rev 16, round 15): no
+// controller is closed and none of its hooks run or complete; its durable
+// records stay, and every external fault (a blocked or failing cleanup, a
+// store that refuses) stays as it was. The kernel releases a dead
+// process's locks: the next generation uses fresh lock files, and the dead
+// generation's clock never fires again, so its controllers never renew,
+// lose or clean up. Recovery is a separate, later event.
+func (w *lifeWorld) crash() {
+ w.mu.Lock()
+ w.gen++
+ gen := w.gen
+ w.mu.Unlock()
+ now, boot, _ := w.clock.Now()
+ w.clock = &lifeClock{wall: now, boot: boot}
+ for _, d := range []string{fmt.Sprintf("exec-locks-%d", gen), fmt.Sprintf("locks-%d", gen)} {
+ if err := os.MkdirAll(filepath.Join(w.dataDir, d), 0o770); err != nil {
+ w.t.Fatal(err)
+ }
+ }
+ w.cfg.Clock = w.clock
+ w.cfg.LockDir = filepath.Join(w.dataDir, fmt.Sprintf("exec-locks-%d", gen))
+ w.cfg.Guard.LockDir = filepath.Join(w.dataDir, fmt.Sprintf("locks-%d", gen))
+ if err := os.Remove(w.cfg.Guard.DataDir); err != nil {
+ w.t.Fatal(err)
+ }
+ w.cfg.Guard.DataDir = w.dataLink(gen)
+ w.e.Controllers = w.controllers()
+}
+
+func (w *lifeWorld) recoverCleanup() {
+ w.mu.Lock()
+ if w.block != nil {
+ close(w.block)
+ w.block = nil
+ }
+ w.cleanErr = false
+ w.mu.Unlock()
+}
+
+func (w *lifeWorld) recoverAll() {
+ w.recoverCleanup()
+ w.store.mu.Lock()
+ w.store.failLease, w.store.claimErr = false, nil
+ w.store.mu.Unlock()
+ if w.saved != nil {
+ path := filepath.Join(w.dataDir, ".hostexec", w.host+".json")
+ if raw, err := os.ReadFile(path); err == nil && string(raw) == "{" {
+ _ = os.WriteFile(path, w.saved, 0o660)
+ // hostexec retries a "lost" write every 100ms through a
+ // directory it already holds open, so a controller that met
+ // the invalid record (a dead generation's included: in-process
+ // its goroutine outlives the crash) lands its write now, as
+ // part of the repair, never after a later clear.
+ time.Sleep(300 * time.Millisecond)
+ }
+ w.saved = nil
+ }
+}
+
+func (w *lifeWorld) event(r *rand.Rand) string {
+ switch n := r.Intn(16); n {
+ case 15:
+ if w.replaced() {
+ return "this box replaced in the same operation"
+ }
+ return "no replacement (not the target)"
+ case 14:
+ w.install()
+ return "the target's installation (P built)"
+ case 0, 1, 2, 3:
+ w.pass()
+ return "pass"
+ case 4:
+ w.renew()
+ return "renew"
+ case 5:
+ w.store.mu.Lock()
+ w.store.failLease = true
+ w.store.mu.Unlock()
+ w.renew()
+ return "lease renewal fails"
+ case 6:
+ w.mu.Lock()
+ if w.block == nil {
+ w.block = make(chan struct{})
+ }
+ w.mu.Unlock()
+ return "cleanup blocks (close times out)"
+ case 7:
+ w.mu.Lock()
+ w.cleanErr = true
+ w.mu.Unlock()
+ return "cleanup fails"
+ case 8:
+ w.store.mu.Lock()
+ if r.Intn(2) == 0 {
+ w.store.claimErr = errors.New("claim read: store unreachable")
+ } else {
+ w.store.claimErr = os.ErrNotExist // ClearStale: ErrState for an absent claim
+ }
+ w.store.mu.Unlock()
+ return "claim reads fail"
+ case 9:
+ path := filepath.Join(w.dataDir, ".hostexec", w.host+".json")
+ if raw, err := os.ReadFile(path); err == nil && w.saved == nil {
+ w.saved = raw
+ _ = os.WriteFile(path, []byte("{"), 0o660) // ClearStale: an invalid local record
+ }
+ return "local record invalid"
+ case 10:
+ w.crash()
+ return "agent crash (no hooks run, faults persist)"
+ case 11:
+ w.complete()
+ w.startRelocation() // the successor starts before this box's sweep
+ return "completion, successor before the sweep"
+ case 12:
+ w.complete()
+ return "completion"
+ default:
+ w.recoverAll()
+ return "storage recovers"
+ }
+}
+
+// TestRelocationControllerLifecycleProperties drives seeded random
+// sequences of lease-renewal failures, close timeouts, cleanup failures,
+// ClearStale failures (claim unreadable, claim absent, invalid local
+// record, a successor that lists this box), agent restarts, successors
+// before the sweep, completions and recovery, over real hostexec
+// controllers, and checks after every step: (a) no clear without a
+// successful scoped cleanup in the same call; (b) no participation
+// forgotten while its record exists; (c) at most one controller per host;
+// (d) no pass past its deadline; and (e) once faults stop, within eight
+// passes the current relocation is ready again, then after completion
+// every record is cleared and no controller remains, and a new relocation
+// opens.
+func TestRelocationControllerLifecycleProperties(t *testing.T) {
+ seeds := lifeSeeds
+ if testing.Short() {
+ seeds = 20
+ }
+ for seed := 1; seed <= seeds; seed++ {
+ t.Run(fmt.Sprint(seed), func(t *testing.T) {
+ r := rand.New(rand.NewSource(int64(seed)))
+ w := newLifeWorld(t)
+ defer w.recoverAll()
+ w.startRelocation()
+ var trace []string
+ if seed%2 == 0 {
+ // Every other seed starts as the target with its local
+ // install journaled (rev 15), so the random events fall on
+ // an installation that a lost lease's cleanup can retire
+ // before P's first admission.
+ w.complete()
+ w.startRelocation() // dst back to src: this box is the target
+ w.pass()
+ w.install()
+ w.pass()
+ trace = append(trace, "target, installed, local install journaled")
+ if w.current().Op.Journal["local_install"] == "" {
+ t.Fatalf("seed %d: the local install was not journaled: %+v", seed, w.current().Op)
+ }
+ if seed%4 == 2 {
+ // Round 18's trace: the target's probe fails, it
+ // releases, it is replaced while the relocation goes
+ // on elsewhere, its cleanup blocks, the agent crashes.
+ w.pass()
+ w.probeFailedAndReleased()
+ if !w.replaced() {
+ t.Fatalf("seed %d: this box was not the target to replace: %+v", seed, w.current().Op)
+ }
+ w.mu.Lock()
+ w.block = make(chan struct{})
+ w.mu.Unlock()
+ w.pass()
+ w.check("replaced with its cleanup blocked")
+ w.crash()
+ w.check("agent crash after the replacement")
+ trace = append(trace, "probe failed, released, replaced, cleanup blocked, crash")
+ }
+ }
+ for i := 0; i < 30; i++ {
+ ev := w.event(r)
+ trace = append(trace, ev)
+ w.check(ev)
+ if len(w.violated) > 0 {
+ t.Fatalf("seed %d: %v\ntrace: %v", seed, w.violated, trace)
+ }
+ }
+ // (e) Faults stop.
+ w.recoverAll()
+ trace = append(trace, "faults stop")
+ // A participating box is ready again; an excluded one (rev 19)
+ // has finished its cleanup: no controller, no record, no
+ // journaled participation, while the relocation goes on.
+ ready := func() bool {
+ c := w.current()
+ if c.Op.Kind != hostclaim.KindRelocating {
+ return true
+ }
+ if !slices.Contains(c.Op.Participants, "src") {
+ _, exists, _ := w.record()
+ retained, err := w.e.Controllers.Retained()
+ return !exists && err == nil && len(retained) == 0 && len(w.open()) == 0
+ }
+ if c.Op.Journal["exec_ack.src"] == "" {
+ return false
+ }
+ obs, err := hostclaim.Read(w.ctx, w.inner, w.host)
+ return err == nil && w.e.Ready(w.ctx, obs) == nil
+ }
+ ok := false
+ for i := 0; i < 8 && !ok; i++ {
+ w.renew()
+ w.pass()
+ w.check("recovery pass")
+ ok = ready()
+ }
+ if !ok || len(w.violated) > 0 {
+ t.Fatalf("seed %d: the running relocation did not recover within 8 passes: %v\ntrace: %v", seed, w.violated, trace)
+ }
+ w.complete()
+ clean := false
+ for i := 0; i < 8 && !clean; i++ {
+ w.renew()
+ w.pass()
+ w.check("sweep pass")
+ _, exists, _ := w.record()
+ retained, err := w.e.Controllers.Retained()
+ clean = !exists && len(retained) == 0 && err == nil && len(w.open()) == 0
+ }
+ if !clean || len(w.violated) > 0 {
+ st, exists, _ := w.record()
+ retained, _ := w.e.Controllers.Retained()
+ t.Fatalf("seed %d: not clean within 8 passes after completion: record %v %+v retained %+v open %d %v\ntrace: %v", seed, exists, st, retained, len(w.open()), w.violated, trace)
+ }
+ w.startRelocation()
+ for i := 0; i < 3; i++ {
+ w.pass()
+ }
+ w.check("new relocation")
+ if !ready() || len(w.open()) != 1 || len(w.violated) > 0 {
+ t.Fatalf("seed %d: a new relocation did not open: open %d %v\ntrace: %v", seed, len(w.open()), w.violated, trace)
+ }
+ for _, c := range w.open() {
+ _ = c.Close(context.Background())
+ }
+ })
+ }
+}
diff --git a/internal/host/relocationreleases.go b/internal/host/relocationreleases.go
new file mode 100644
index 000000000..038f7c708
--- /dev/null
+++ b/internal/host/relocationreleases.go
@@ -0,0 +1,53 @@
+package host
+
+import (
+ "bytes"
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// RelocationReleaseObligation is the stop request's obligation component
+// for a relocation's release (§5.8 release step 4). A release retires
+// nothing, so it names no park-local chore; root's FenceAndStop requires
+// the component, and the receipt carries it so a retirement stop can never
+// be offered as a release, nor a release as a retirement.
+const RelocationReleaseObligation = "relocation-release"
+
+// StopReceiptReleases is hostclaim.ReleaseAuthority over this package's
+// stop receipts (ClaimStopCoordinator.Publish): a relocation's journaled
+// release counts only when the immutable receipt it names hashes to the
+// journaled digest and is the full stop of exactly that box, enrollment,
+// writer session, host, incarnation, operation and phase (region HA spec 1
+// §5.8 release step 4, claim §6.1).
+type StopReceiptReleases struct {
+ Evidence release.Store
+}
+
+var _ hostclaim.ReleaseAuthority = StopReceiptReleases{}
+
+func (a StopReceiptReleases) VerifyRelease(ctx context.Context, c hostclaim.Claim, r hostclaim.Release) error {
+ if a.Evidence == nil || c.Holder == nil {
+ return ErrClaimStopUnavailable
+ }
+ raw, _, err := a.Evidence.Get(ctx, r.Receipt)
+ if err != nil {
+ return err
+ }
+ sum := sha256.Sum256(raw)
+ var receipt StopReceipt
+ if hex.EncodeToString(sum[:]) != r.ReceiptSHA256 || json.Unmarshal(raw, &receipt) != nil || !bytes.Equal(raw, mustMarshalStopReceipt(receipt)) {
+ return ErrClaimStopReceipt
+ }
+ key := stopEvidenceKey(receipt.Host, receipt.OperationID, receipt.Box, receipt.ID)
+ if receipt.V != 1 || receipt.Kind != "stopped" || receipt.OperationID == "" || r.Receipt != key || receipt.Host != c.Host || receipt.Incarnation != c.Holder.Incarnation ||
+ receipt.OperationID != c.Op.ID || receipt.Phase != c.Op.Phase || receipt.ObligationID != RelocationReleaseObligation || receipt.Box != r.ReleasedBy || receipt.Enrollment != r.Enrollment ||
+ receipt.WriterSession != r.Session || receipt.EvidenceAt.IsZero() {
+ return ErrClaimStopReceipt
+ }
+ return nil
+}
diff --git a/internal/host/relocationreleases_test.go b/internal/host/relocationreleases_test.go
new file mode 100644
index 000000000..df07964c0
--- /dev/null
+++ b/internal/host/relocationreleases_test.go
@@ -0,0 +1,62 @@
+package host
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestStopReceiptReleasesBindsTheReceiptToTheRelease(t *testing.T) {
+ ctx := context.Background()
+ inc := strings.Repeat("a", 64)
+ c := hostclaim.Claim{Host: "shop.example", Holder: &hostclaim.Holder{Incarnation: inc}, Op: hostclaim.Op{ID: "rel-1", Kind: hostclaim.KindRelocating, Phase: hostclaim.PhaseReleasing}}
+ receipt := StopReceipt{V: 1, ID: "r1", Host: "shop.example", Incarnation: inc, WriterSession: 12, OperationID: "rel-1", Phase: hostclaim.PhaseReleasing, ObligationID: RelocationReleaseObligation,
+ Box: "src", Enrollment: "enr", Kind: "stopped", EvidenceAt: time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC)}
+ put := func(s release.Store, r StopReceipt) hostclaim.Release {
+ body := mustMarshalStopReceipt(r)
+ key := "control/hosts/shop.example/evidence/" + r.OperationID + "/" + r.Box + "/" + r.ID + ".json"
+ if _, err := s.Put(ctx, key, body, ""); err != nil {
+ t.Fatal(err)
+ }
+ sum := sha256.Sum256(body)
+ return hostclaim.Release{ReleasedBy: "src", Enrollment: "enr", Session: 12, Receipt: key, ReceiptSHA256: hex.EncodeToString(sum[:])}
+ }
+ s := release.NewDirStore(t.TempDir())
+ a := StopReceiptReleases{Evidence: s}
+ if err := a.VerifyRelease(ctx, c, put(s, receipt)); err != nil {
+ t.Fatal(err)
+ }
+ for _, tc := range []struct {
+ name string
+ mutate func(*StopReceipt, *hostclaim.Release)
+ }{
+ {"another session", func(r *StopReceipt, _ *hostclaim.Release) { r.WriterSession = 11 }},
+ {"another box", func(r *StopReceipt, _ *hostclaim.Release) { r.Box = "dst" }},
+ {"another operation", func(r *StopReceipt, _ *hostclaim.Release) { r.OperationID = "rel-0" }},
+ {"a retirement's stop", func(r *StopReceipt, _ *hostclaim.Release) { r.ObligationID = "park-1" }},
+ {"another phase", func(r *StopReceipt, _ *hostclaim.Release) { r.Phase = hostclaim.PhaseProbeFailed }},
+ {"another incarnation", func(r *StopReceipt, _ *hostclaim.Release) { r.Incarnation = strings.Repeat("b", 64) }},
+ {"a digest that is not the bytes", func(_ *StopReceipt, rel *hostclaim.Release) { rel.ReceiptSHA256 = strings.Repeat("0", 64) }},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ r := receipt
+ var mutated hostclaim.Release
+ tc.mutate(&r, &mutated)
+ rel := put(s, r)
+ if mutated.ReceiptSHA256 != "" {
+ rel.ReceiptSHA256 = mutated.ReceiptSHA256
+ }
+ if err := (StopReceiptReleases{Evidence: s}).VerifyRelease(ctx, c, rel); !errors.Is(err, ErrClaimStopReceipt) {
+ t.Fatalf("accepted: %v", err)
+ }
+ })
+ }
+}
diff --git a/internal/hostexec/readlocal.go b/internal/hostexec/readlocal.go
new file mode 100644
index 000000000..778bce234
--- /dev/null
+++ b/internal/hostexec/readlocal.go
@@ -0,0 +1,17 @@
+package hostexec
+
+// ReadLocal reads host's durable local record (<dataDir>/.hostexec/<host>.json)
+// with the validation Open, CheckLocal and ClearStale apply to it: a
+// regular file, a complete binding (box, enrollment, host, incarnation,
+// op, boot and controller ids), a known state. An absent record is
+// os.ErrNotExist; anything it would refuse is an error, never a State.
+// It takes no lock and grants nothing: a caller deciding whether a record
+// is still its own uses it, and keeps its own state on any error.
+func ReadLocal(dataDir, host string) (State, error) {
+ d, err := openLocal(dataDir)
+ if err != nil {
+ return State{}, err
+ }
+ defer d.close()
+ return d.read(host)
+}
diff --git a/internal/hostfiles/platform.go b/internal/hostfiles/platform.go
new file mode 100644
index 000000000..18dbbe48e
--- /dev/null
+++ b/internal/hostfiles/platform.go
@@ -0,0 +1,33 @@
+package hostfiles
+
+import (
+ "fmt"
+ "strings"
+)
+
+// EnvItem is the host's env file (<host>.env) among the manifest's
+// platform items. A release flush stages it beside the wake state so a
+// relocation's prepared manifest carries it (region HA spec 1 §5.6:
+// always continuous, whatever the setting); the local install restores it
+// with ReadPlatform.
+const EnvItem = "host.env"
+
+// StagePlatformDelete records that a platform item is gone: the next
+// Commit publishes a revision without it, at a new sequence, so a reader
+// of that revision gets os.ErrNotExist from ReadPlatform instead of the
+// last bytes it published. Deleting an item the manifest does not hold
+// is a no-op, like StageDelete.
+func (p *Publisher) StagePlatformDelete(name string) error {
+ if name == "" || strings.ContainsAny(name, "/\x00") {
+ return fmt.Errorf("hostfiles: platform item %q is not a plain name", name)
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if _, ok := p.man.Platform[name]; !ok {
+ return nil
+ }
+ p.next++
+ delete(p.man.Platform, name)
+ p.dirty = true
+ return nil
+}
diff --git a/internal/hostrelocate/participant.go b/internal/hostrelocate/participant.go
index 8031b7012..c44ccabfd 100644
--- a/internal/hostrelocate/participant.go
+++ b/internal/hostrelocate/participant.go
@@ -25,7 +25,9 @@ type Effects interface {
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.
+ // retained (the agent restarted, or a controller that lost its lease
+ // was joined and its scoped cleanup repeated), and the join's or the
+ // cleanup's error while a lost one cannot be retired yet.
// 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