amadan

carlos / platform Public

host: the relocation owner set: wake inputs over main's prepared publication, an op-keyed attempt journal that root reads with region HA on, the admission hooks and the probe attempt

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude f28a805dcfe30bca06dcc5b6fbe800ba548b8f3f parent ce6f8bb
9 files changed, +2047 −0
  • internal/host/claimlaunch_production.go +11 −0
  • internal/host/claimowners.go +12 −0
  • internal/host/relocationassembly.go +1 −0
  • internal/host/relocationattempts.go +362 −0
  • internal/host/relocationattempts_test.go +168 −0
  • internal/host/relocationowners.go +213 −0
  • internal/host/relocationowners_test.go +643 −0
  • internal/host/relocationwake.go +432 −0
  • internal/host/relocationwake_test.go +205 −0
diff --git a/internal/host/claimlaunch_production.go b/internal/host/claimlaunch_production.go
index f122c3dc6..913704088 100644
--- a/internal/host/claimlaunch_production.go
+++ b/internal/host/claimlaunch_production.go
@@ -3,12 +3,14 @@ package host
import (
"context"
"net/http"
+ "os"
"os/exec"
"path/filepath"
"strings"
"github.com/carlosframework/platform/internal/controlsock"
"github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
execlease "github.com/carlosframework/platform/internal/hostexec"
"github.com/carlosframework/platform/internal/instancectl"
)
@@ -80,10 +82,19 @@ func claimLaunchFor(root ClaimRootHTTP, actor func(context.Context) (hostbox.Act
}
launch := &ClaimLaunchRoot{ValidateExecutable: claimStagedExecutable(&instancectl.Server{OptDir: roots["opt"], HostApp: instanceRegistryHostApp(nil)}), DataRoot: roots["data"], Clock: execlease.SystemClock{}, Authenticate: root.Stop.Authenticate, Guard: coordinator.Guard, Actor: actor, Journal: root.Launches, Units: units,
Kernel: linuxClaimLaunchKernel{OptRoot: roots["opt"], CgroupRoot: "/sys/fs/cgroup", ProcRoot: "/proc"}}
+ if regionHAEnabled(claimLaunchEnv) {
+ // Claim Plan 11 "Region HA integration and rollout": the root-owned
+ // authoritative attempt reader for kind relocating (region HA ruling
+ // 48). Off, Attempts stays empty and the kind is refused as before.
+ launch.Attempts = map[string]claimLaunchAttemptReader{hostclaim.KindRelocating: relocationAttemptReader{DataDir: roots["data"]}}
+ }
root.Stop.Coordinator.Kernel = claimUnitStopKernel{Inner: coordinator.Kernel, Journal: root.Launches, Units: units}
return ClaimRootWithLaunch{Root: root, Launch: launch}, nil
}
+// claimLaunchEnv is the process environment; tests replace it.
+var claimLaunchEnv = os.Getenv
+
func claimLaunchAbs(p string) bool { return filepath.IsAbs(p) && filepath.Clean(p) == p }
func claimStagedExecutable(server *instancectl.Server) func(string, string, string) error {
diff --git a/internal/host/claimowners.go b/internal/host/claimowners.go
index 81f797b73..c93b8d0c7 100644
--- a/internal/host/claimowners.go
+++ b/internal/host/claimowners.go
@@ -490,6 +490,13 @@ func (s *claimServingOwners) PublishFenced(ctx context.Context, o hostclaim.Obse
if err := s.serving(o); err != nil {
return ClaimPublicationReceipt{}, err
}
+ return s.publishFresh(ctx, o, in, pub)
+}
+
+// publishFresh is PublishFenced after its kind check, unchanged: the one
+// fresh-generation marker writer, which region HA's later probe attempts
+// also use (rev 2, ruling 47).
+func (s *claimServingOwners) publishFresh(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, pub ClaimPublication) (ClaimPublicationReceipt, error) {
if pub.Sync == nil || pub.Check == nil {
return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
}
@@ -617,6 +624,11 @@ func (s *claimServingOwners) publicationEvidence(ctx context.Context, o hostclai
if err := s.serving(o); err != nil {
return err
}
+ return s.freshEvidence(ctx, o)
+}
+
+// freshEvidence is publicationEvidence after its kind check, unchanged.
+func (s *claimServingOwners) freshEvidence(ctx context.Context, o hostclaim.Observed) error {
h := o.Claim.Holder
s.mu.Lock()
receipt, ok := s.published[claimPublishedKey(o.Host, h.Session)]
diff --git a/internal/host/relocationassembly.go b/internal/host/relocationassembly.go
index 2278e3b0c..9db05306f 100644
--- a/internal/host/relocationassembly.go
+++ b/internal/host/relocationassembly.go
@@ -131,6 +131,7 @@ func (a ClaimAssembly) RelocationEffects(cfg RelocationAssemblyConfig, ack func(
Install: a.relocationInstall(cfg),
VerifyLocal: a.relocationVerifyLocal(cfg),
Receipt: a.relocationReceipt(cfg),
+ Admit: a.relocationAdmit(cfg),
}
}
diff --git a/internal/host/relocationattempts.go b/internal/host/relocationattempts.go
new file mode 100644
index 000000000..9f142057c
--- /dev/null
+++ b/internal/host/relocationattempts.go
@@ -0,0 +1,362 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strconv"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// relocationAttemptJournal is one operation's probe journal, at
+// <DataDir>/relocate/ops/<op>/attempts.json (ruling 41, keyed by op per
+// ruling 44), written whole and renamed. The claim's probe_attempts.<box>
+// is the record attempt k; the journal binds each admitted session to one
+// k, keeps each k's first app start, and each k's current start. Every
+// field has its own JSON name (finding 14).
+type relocationAttemptJournal struct {
+ V int `json:"v"`
+ Host string `json:"host"`
+ Op string `json:"op"`
+ Opened map[string]time.Time `json:"opened"` // "k" -> the attempt's first app start
+ Sessions map[string]int `json:"sessions"` // "session" -> k
+ Starts map[string]relocationStart `json:"starts"` // "k" -> the attempt's current start
+}
+
+// relocationStart is one start's reservation, bound to the admitted
+// session and generation root's reader validates (Plan 11: the exact
+// operation, phase, generation, stamp and session).
+type relocationStart struct {
+ Session uint64 `json:"session"`
+ Generation string `json:"generation"`
+ Stamp uint64 `json:"stamp"`
+ N int `json:"n"`
+ Attempt hostclaim.WakeRetryAttempt `json:"attempt"`
+ State string `json:"state"` // reserved, failed, healthy
+}
+
+func relocationAttemptsPath(dataDir, op string) (string, error) {
+ dir, err := relocationOpDir(dataDir, op)
+ if err != nil {
+ return "", err
+ }
+ return filepath.Join(dir, "attempts.json"), nil
+}
+
+func readRelocationAttempts(dataDir, host, op string) (relocationAttemptJournal, error) {
+ path, err := relocationAttemptsPath(dataDir, op)
+ if err != nil {
+ return relocationAttemptJournal{}, err
+ }
+ raw, err := os.ReadFile(path)
+ if errors.Is(err, os.ErrNotExist) {
+ return relocationAttemptJournal{V: 1, Host: host, Op: op, Opened: map[string]time.Time{}, Sessions: map[string]int{}, Starts: map[string]relocationStart{}}, nil
+ }
+ if err != nil {
+ return relocationAttemptJournal{}, err
+ }
+ var j relocationAttemptJournal
+ if err := json.Unmarshal(raw, &j); err != nil || j.V != 1 || j.Host != host || j.Op != op || j.Opened == nil || j.Sessions == nil || j.Starts == nil {
+ return relocationAttemptJournal{}, fmt.Errorf("%w: relocation attempt journal %s", hostbox.ErrChanged, op)
+ }
+ return j, nil
+}
+
+func (j relocationAttemptJournal) write(dataDir string) error {
+ path, err := relocationAttemptsPath(dataDir, j.Op)
+ if err != nil {
+ return err
+ }
+ raw, err := json.Marshal(j)
+ if err != nil {
+ return err
+ }
+ if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
+ return err
+ }
+ return writeFileAtomic(path, raw, 0o600)
+}
+
+// relocationProbeAttempts is region HA's OperationAttempts for a probe
+// admitted under relocating/probing (claim Plan 11 R1, D2), modelled on
+// movesProbeAttempts without the record. Every method refuses
+// (hostbox.ErrChanged) once the claim no longer names the admitted writer
+// at probing, unstuck, with k = probe_attempts.<box> in 1..MaxProbeAttempts.
+type relocationProbeAttempts struct {
+ Claims release.Store
+ DataDir string
+ Box string
+ NewID func() string
+ Now func() time.Time
+ Settle, Ready time.Duration
+ mu sync.Mutex
+ current map[string]boundAttempt // host\x00session
+}
+
+var _ OperationAttempts = (*relocationProbeAttempts)(nil)
+
+func (p *relocationProbeAttempts) bound(ctx context.Context, s OperationStart) (int, error) {
+ if p == nil || p.Claims == nil || p.DataDir == "" || p.Box == "" || p.NewID == nil || p.Now == nil {
+ return 0, hostwrite.ErrUnavailable
+ }
+ a := s.Admitted
+ if a.Claim == nil || a.Claim.Holder == nil || a.Claim.Op.Kind != hostclaim.KindRelocating || !s.Policy.Probe {
+ return 0, hostbox.ErrUnsupported
+ }
+ h := a.Claim.Holder
+ if s.Owner.Host != a.Host || s.Owner.Session != h.Session || s.Owner.Stamp != h.Stamp || s.Owner.Incarnation != h.Incarnation || s.Owner.Box != p.Box {
+ return 0, hostbox.ErrChanged
+ }
+ fresh, err := hostclaim.Read(ctx, p.Claims, a.Host)
+ if err != nil {
+ return 0, err
+ }
+ if !claimSameWriter(a, fresh, hostclaim.SessionContinues) || fresh.Claim.Op.Phase != hostclaim.PhaseRelocatingProbing || fresh.Claim.Op.Journal["stuck"] != "" || fresh.Claim.Op.Exec != p.Box {
+ return 0, hostbox.ErrChanged
+ }
+ k, err := strconv.Atoi(fresh.Claim.Op.Journal["probe_attempts."+p.Box])
+ if err != nil || k < 1 || k > hostclaim.MaxProbeAttempts {
+ return 0, hostbox.ErrChanged
+ }
+ return k, nil
+}
+
+func relocationSessionKey(s OperationStart) string {
+ return s.Admitted.Host + "\x00" + strconv.FormatUint(s.Owner.Session, 10)
+}
+
+// bind checks, and with write records, the session's record attempt.
+func (j relocationAttemptJournal) bind(session uint64, k int, write bool) error {
+ key := strconv.FormatUint(session, 10)
+ got, ok := j.Sessions[key]
+ switch {
+ case ok && got == k:
+ return nil
+ case ok:
+ return fmt.Errorf("%w: session %d is bound to attempt %d, not %d", hostbox.ErrChanged, session, got, k)
+ case !write:
+ return fmt.Errorf("%w: session %d is bound to no attempt", hostbox.ErrChanged, session)
+ }
+ j.Sessions[key] = k
+ return nil
+}
+
+func (p *relocationProbeAttempts) OpenProbe(ctx context.Context, s OperationStart, proposed time.Time) (time.Time, error) {
+ k, err := p.bound(ctx, s)
+ if err != nil {
+ return time.Time{}, err
+ }
+ if proposed.IsZero() {
+ return time.Time{}, hostwrite.ErrUnavailable
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ j, err := readRelocationAttempts(p.DataDir, s.Admitted.Host, s.Admitted.Claim.Op.ID)
+ if err != nil {
+ return time.Time{}, err
+ }
+ if err := j.bind(s.Owner.Session, k, true); err != nil {
+ return time.Time{}, err
+ }
+ key := strconv.Itoa(k)
+ opened, ok := j.Opened[key]
+ if !ok {
+ opened = proposed.UTC()
+ j.Opened[key] = opened
+ }
+ return opened, j.write(p.DataDir)
+}
+
+func (p *relocationProbeAttempts) Reserve(ctx context.Context, s OperationStart, n int) error {
+ k, err := p.bound(ctx, s)
+ if err != nil {
+ return err
+ }
+ digest, err := ClaimLaunchDigest(s.Inputs)
+ if err != nil {
+ return err
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ j, err := readRelocationAttempts(p.DataDir, s.Admitted.Host, s.Admitted.Claim.Op.ID)
+ if err != nil {
+ return err
+ }
+ if err := j.bind(s.Owner.Session, k, n == 0); err != nil {
+ return err
+ }
+ key := strconv.Itoa(k)
+ st, ok := j.Starts[key]
+ same := ok && st.Session == s.Owner.Session
+ h := s.Admitted.Claim.Holder
+ switch {
+ case n < 0:
+ return hostbox.ErrChanged
+ case n == 0 && same:
+ if st.N != 0 || st.State != probeReserved || !sameStartOwner(st.Attempt, s, digest) || st.Generation != h.Generation || st.Stamp != h.Stamp {
+ return hostbox.ErrChanged
+ }
+ case n == 0:
+ st = relocationStart{Session: s.Owner.Session, Generation: h.Generation, Stamp: h.Stamp, State: probeReserved}
+ default:
+ if !same || st.N != n-1 || st.State != probeFailed || st.Attempt.LaunchInputs != digest {
+ return hostbox.ErrChanged
+ }
+ st = relocationStart{Session: s.Owner.Session, Generation: h.Generation, Stamp: h.Stamp, N: n, State: probeReserved}
+ }
+ if st.Attempt.ID == "" {
+ o := s.Owner
+ st.Attempt = hostclaim.WakeRetryAttempt{ID: p.NewID(), Box: o.Box, Enrollment: o.Enrollment, BootID: o.BootID, ControllerID: o.ControllerID, Session: o.Session, InvocationID: p.NewID(), LaunchInputs: digest}
+ if st.Attempt.ID == "" || st.Attempt.InvocationID == "" || st.Attempt.ID == st.Attempt.InvocationID {
+ return hostwrite.ErrUnavailable
+ }
+ j.Starts[key] = st
+ }
+ if err := j.write(p.DataDir); err != nil {
+ return err
+ }
+ if p.current == nil {
+ p.current = map[string]boundAttempt{}
+ }
+ p.current[relocationSessionKey(s)] = boundAttempt{k: k, attempt: st.Attempt}
+ return nil
+}
+
+func (p *relocationProbeAttempts) CurrentAttempt(s OperationStart) hostclaim.WakeRetryAttempt {
+ if p == nil {
+ return hostclaim.WakeRetryAttempt{}
+ }
+ p.mu.Lock()
+ cur, ok := p.current[relocationSessionKey(s)]
+ p.mu.Unlock()
+ if !ok {
+ return hostclaim.WakeRetryAttempt{}
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ if k, err := p.bound(ctx, s); err != nil || k != cur.k {
+ return hostclaim.WakeRetryAttempt{}
+ }
+ return cur.attempt
+}
+
+func (p *relocationProbeAttempts) Failed(ctx context.Context, s OperationStart, n int) error {
+ recovery, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
+ defer cancel()
+ return p.settle(recovery, s, n, probeFailed)
+}
+
+func (p *relocationProbeAttempts) settle(ctx context.Context, s OperationStart, n int, state string) error {
+ k, err := p.bound(ctx, s)
+ if err != nil {
+ return err
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ j, err := readRelocationAttempts(p.DataDir, s.Admitted.Host, s.Admitted.Claim.Op.ID)
+ if err != nil {
+ return err
+ }
+ if err := j.bind(s.Owner.Session, k, false); err != nil {
+ return err
+ }
+ key := strconv.Itoa(k)
+ st, ok := j.Starts[key]
+ cur := p.current[relocationSessionKey(s)]
+ if !ok || st.Session != s.Owner.Session || st.N != n || st.State != probeReserved || cur.k != k || st.Attempt != cur.attempt {
+ return hostbox.ErrChanged
+ }
+ st.State = state
+ j.Starts[key] = st
+ return j.write(p.DataDir)
+}
+
+// Healthy is movesProbeAttempts.Healthy's settle and second request
+// (moves §7.7), which the relocation probe shares (spec §5.8).
+func (p *relocationProbeAttempts) Healthy(ctx context.Context, s OperationStart, n int, cmd *exec.Cmd, again func(context.Context) error) error {
+ if _, err := p.bound(ctx, s); err != nil {
+ return err
+ }
+ if again == nil || !probeAlive(cmd) {
+ return hostbox.ErrChanged
+ }
+ settle, ready := p.Settle, p.Ready
+ if settle <= 0 {
+ settle = time.Duration(moves.SettleSeconds * float64(time.Second))
+ }
+ if ready <= 0 {
+ ready = time.Duration(moves.ProbeSeconds * float64(time.Second))
+ }
+ timer := time.NewTimer(settle)
+ select {
+ case <-ctx.Done():
+ timer.Stop()
+ return ctx.Err()
+ case <-timer.C:
+ }
+ if !probeAlive(cmd) {
+ return fmt.Errorf("%w: the app exited during the settle", hostbox.ErrChanged)
+ }
+ asked, cancel := context.WithTimeout(ctx, ready)
+ err := again(asked)
+ cancel()
+ if err != nil {
+ return err
+ }
+ if !probeAlive(cmd) {
+ return fmt.Errorf("%w: the app exited after the second request", hostbox.ErrChanged)
+ }
+ return p.settle(ctx, s, n, probeHealthy)
+}
+
+// relocationAttemptReader is root's authoritative attempt reader for kind
+// relocating (ruling 48; claim Plan 11 "install and test root-owned
+// authoritative attempt readers"): it reads the operation's durable
+// journal and returns the reserved start only for exactly the observed
+// op at probing, its executor's attempt k, the holder's session,
+// generation and stamp (the journaled install stamp). Root's authorize
+// then binds the reservation, the launch-input digest and the peer.
+type relocationAttemptReader struct{ DataDir string }
+
+func (r relocationAttemptReader) ReadAttempt(_ context.Context, o hostclaim.Observed) (hostclaim.WakeRetryAttempt, error) {
+ c := o.Claim
+ if r.DataDir == "" || c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindRelocating || c.Op.Phase != hostclaim.PhaseRelocatingProbing || c.Op.Journal["stuck"] != "" {
+ return hostclaim.WakeRetryAttempt{}, hostbox.ErrChanged
+ }
+ h := c.Holder
+ // P's opening keeps the install stamp; a later cold attempt runs at
+ // its own admission's stamp (generation.go: !KeepStamp sets stamp to
+ // the session), which the journaled start must name (Astra round 2
+ // finding 4). Main's claimLaunchAuthority still pins every relocating
+ // launch to install_stamp until claim ask C11.
+ if h.Generation == c.Op.Journal["prepared"] && c.Op.Journal["install_stamp"] != strconv.FormatUint(h.Stamp, 10) {
+ return hostclaim.WakeRetryAttempt{}, hostbox.ErrChanged
+ }
+ k, err := strconv.Atoi(c.Op.Journal["probe_attempts."+c.Op.Exec])
+ if err != nil || k < 1 || k > hostclaim.MaxProbeAttempts {
+ return hostclaim.WakeRetryAttempt{}, hostbox.ErrChanged
+ }
+ j, err := readRelocationAttempts(r.DataDir, o.Host, c.Op.ID)
+ if err != nil {
+ return hostclaim.WakeRetryAttempt{}, err
+ }
+ if j.Sessions[strconv.FormatUint(h.Session, 10)] != k {
+ return hostclaim.WakeRetryAttempt{}, hostbox.ErrChanged
+ }
+ st, ok := j.Starts[strconv.Itoa(k)]
+ if !ok || st.State != probeReserved || st.Session != h.Session || st.Generation != h.Generation || st.Stamp != h.Stamp || st.Attempt.Session != h.Session || st.Attempt.Box != c.Op.Exec {
+ return hostclaim.WakeRetryAttempt{}, hostbox.ErrChanged
+ }
+ return st.Attempt, nil
+}
diff --git a/internal/host/relocationattempts_test.go b/internal/host/relocationattempts_test.go
new file mode 100644
index 000000000..698b70f8f
--- /dev/null
+++ b/internal/host/relocationattempts_test.go
@@ -0,0 +1,168 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "net"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// Finding 14: every journal field has its own JSON name and survives a
+// round trip; a journal for another host or op is refused.
+func TestRelocationAttemptJournalRoundTrips(t *testing.T) {
+ dd := t.TempDir()
+ j, err := readRelocationAttempts(dd, "shop.example", "op1")
+ if err != nil {
+ t.Fatal(err)
+ }
+ j.Opened["1"] = time.Date(2026, 10, 10, 9, 0, 0, 0, time.UTC)
+ j.Sessions["7"] = 1
+ j.Starts["1"] = relocationStart{Session: 7, Generation: "P1", Stamp: 9, State: probeReserved, Attempt: hostclaim.WakeRetryAttempt{ID: "a", InvocationID: "b", Session: 7, Box: "dst"}}
+ if err := j.write(dd); err != nil {
+ t.Fatal(err)
+ }
+ raw, _ := os.ReadFile(filepath.Join(dd, "relocate", "ops", "op1", "attempts.json"))
+ for _, key := range []string{`"host":"shop.example"`, `"op":"op1"`} {
+ if !strings.Contains(string(raw), key) {
+ t.Fatalf("%s missing from %s", key, raw)
+ }
+ }
+ got, err := readRelocationAttempts(dd, "shop.example", "op1")
+ if err != nil || got.Starts["1"] != j.Starts["1"] || got.Sessions["7"] != 1 || !got.Opened["1"].Equal(j.Opened["1"]) {
+ t.Fatalf("round trip: %+v %v", got, err)
+ }
+ if _, err := readRelocationAttempts(dd, "other.example", "op1"); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("another host's journal was read: %v", err)
+ }
+}
+
+// Finding 7: root's reader returns the reserved start only for the exact
+// op at probing, the executor's attempt k, session, generation and stamp.
+func TestRootReadsTheRelocationAttemptJournal(t *testing.T) {
+ dd := t.TempDir()
+ j, _ := readRelocationAttempts(dd, "shop.example", "op1")
+ j.Sessions["7"] = 1
+ attempt := hostclaim.WakeRetryAttempt{ID: "a", InvocationID: "b", Session: 7, Box: "dst"}
+ j.Starts["1"] = relocationStart{Session: 7, Generation: "P1", Stamp: 9, State: probeReserved, Attempt: attempt}
+ if err := j.write(dd); err != nil {
+ t.Fatal(err)
+ }
+ claim := func() *hostclaim.Claim {
+ return &hostclaim.Claim{Host: "shop.example", Holder: &hostclaim.Holder{Incarnation: "inc", Stamp: 9, Session: 7, Generation: "P1", WriterBox: "dst"},
+ Op: hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "op1", Phase: hostclaim.PhaseRelocatingProbing, Exec: "dst", Journal: map[string]string{"install_stamp": "9", "prepared": "P1", "probe_attempts.dst": "1"}}}
+ }
+ r := relocationAttemptReader{DataDir: dd}
+ got, err := r.ReadAttempt(context.Background(), hostclaim.Observed{Host: "shop.example", Claim: claim()})
+ if err != nil || got != attempt {
+ t.Fatalf("read %+v %v", got, err)
+ }
+ // Astra round 2 finding 4: a later cold attempt runs at its own
+ // admission's stamp; the reader takes the journaled start at that
+ // stamp (root's claimLaunchAuthority pin is claim ask C11).
+ j2, _ := readRelocationAttempts(dd, "shop.example", "op1")
+ j2.Sessions["14"] = 2
+ later := hostclaim.WakeRetryAttempt{ID: "c", InvocationID: "d", Session: 14, Box: "dst"}
+ j2.Starts["2"] = relocationStart{Session: 14, Generation: "g2", Stamp: 14, State: probeReserved, Attempt: later}
+ if err := j2.write(dd); err != nil {
+ t.Fatal(err)
+ }
+ second := claim()
+ second.Holder.Stamp, second.Holder.Session, second.Holder.Generation = 14, 14, "g2"
+ second.Op.Journal["probe_attempts.dst"] = "2"
+ second.Op.Journal["prepared_opened"] = "P1"
+ if got, err := r.ReadAttempt(context.Background(), hostclaim.Observed{Host: "shop.example", Claim: second}); err != nil || got != later {
+ t.Fatalf("second attempt: %+v %v", got, err)
+ }
+ for name, change := range map[string]func(*hostclaim.Claim){
+ "another session": func(c *hostclaim.Claim) { c.Holder.Session = 8 },
+ "another generation": func(c *hostclaim.Claim) { c.Holder.Generation = "g2" },
+ "another stamp": func(c *hostclaim.Claim) { c.Holder.Stamp = 10; c.Op.Journal["install_stamp"] = "10" },
+ "P at a stamp not the install stamp": func(c *hostclaim.Claim) { c.Op.Journal["install_stamp"] = "8" },
+ "a later attempt": func(c *hostclaim.Claim) { c.Op.Journal["probe_attempts.dst"] = "2" },
+ "installed": func(c *hostclaim.Claim) { c.Op.Phase = hostclaim.PhaseInstalled },
+ "stuck": func(c *hostclaim.Claim) { c.Op.Journal["stuck"] = "x" },
+ } {
+ c := claim()
+ change(c)
+ if _, err := r.ReadAttempt(context.Background(), hostclaim.Observed{Host: "shop.example", Claim: c}); err == nil {
+ t.Fatalf("%s: read", name)
+ }
+ }
+}
+
+// Finding 4: a relocation target's route is the nested layout, and the
+// files and env paths are the instance directory's.
+func TestRelocationRouteIsNestedAndPathsAgree(t *testing.T) {
+ dd := "/data"
+ rt := provision.RouteFor(provision.Record{Host: "shop.example", App: "shop", Account: "acct", Channel: "stable"}, dd, "", true)
+ if rt.DBPath != provision.NestedDBPath(dd, "shop.example") || rt.Socket != provision.NestedSocket("shop.example") {
+ t.Fatalf("not nested: %+v", rt)
+ }
+ p := relocationPathsFor(dd, rt)
+ dir := provision.InstanceDir(dd, "shop.example")
+ if p.FilesDir != filepath.Join(dir, "shop.example") || p.EnvPath != filepath.Join(dir, "shop.example.env") {
+ t.Fatalf("paths %+v", p)
+ }
+ if _, err := claimIsolatedPaths(map[string]string{"data": dd}, "shop.example", ClaimLaunchInputs{DBPath: rt.DBPath, Limit: cgroup.Limit{MemoryMax: cgroup.TierDefault.MemoryMax}}); err != nil {
+ t.Fatalf("the isolated launch refuses the minted route: %v", err)
+ }
+}
+
+// Finding 7, ruling 48: root installs the relocation attempt reader only
+// with region HA on; its launch capability answers for kind relocating
+// only then.
+func TestRootInstallsTheRelocationReaderOnlyWithRegionHA(t *testing.T) {
+ journal, err := hostbox.OpenLaunchJournal(filepath.Join(t.TempDir(), "launch.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer journal.Close()
+ inner := stopKernelFunc(func(context.Context, hostguard.FenceTarget, StopInventory, DrainAcknowledgement) (StopKernelEvidence, error) {
+ return StopKernelEvidence{}, nil
+ })
+ root := ClaimRootHTTP{Launches: journal, Stop: ClaimStopHTTP{Authenticate: func(context.Context, net.Conn) (StopPeer, error) { return StopPeer{}, nil },
+ Coordinator: ClaimStopCoordinator{Kernel: inner, Guard: hostguard.Guard{DataDir: "/data", Claims: &launchClaims{}}}}}
+ actor := func(context.Context) (hostbox.Actor, error) { return hostbox.Actor{}, nil }
+ saved := claimLaunchEnv
+ t.Cleanup(func() { claimLaunchEnv = saved })
+ q := hostbox.RoleLaunchRequest{V: hostbox.RoleLaunchV, Op: hostbox.RoleLaunchCapability, Kind: hostclaim.KindRelocating}
+ for _, on := range []bool{true, false} {
+ value := ""
+ if on {
+ value = "1"
+ }
+ claimLaunchEnv = func(k string) string {
+ if k == "CARLOS_REGION_HA" {
+ return value
+ }
+ return ""
+ }
+ wrapped, err := claimLaunchFor(root, actor, map[string]string{"opt": "/opt/carlos", "data": "/data"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ reader, installed := wrapped.Launch.Attempts[hostclaim.KindRelocating].(relocationAttemptReader)
+ if installed != on || (on && reader.DataDir != "/data") || (!on && wrapped.Launch.Attempts != nil) {
+ t.Fatalf("region HA %v: attempts %+v", on, wrapped.Launch.Attempts)
+ }
+ f := newLaunchFixture(t, servingOp())
+ f.root.Attempts = wrapped.Launch.Attempts
+ _, err = f.root.Serve(t.Context(), f.peer, q)
+ if on && err != nil {
+ t.Fatalf("the relocating capability with region HA on: %v", err)
+ }
+ if !on && !errors.Is(err, hostbox.ErrUnsupported) {
+ t.Fatalf("the relocating capability with region HA off: %v", err)
+ }
+ }
+}
diff --git a/internal/host/relocationowners.go b/internal/host/relocationowners.go
new file mode 100644
index 000000000..3656574a8
--- /dev/null
+++ b/internal/host/relocationowners.go
@@ -0,0 +1,213 @@
+package host
+
+import (
+ "context"
+ "crypto/rand"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "os"
+ "slices"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostactors"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostrelocate"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// RelocationSource is region HA's ClaimProductionRelocationSource (claim
+// Plan 11 R1, R8), registered once at agent startup (Task 10) and opened
+// by the production provider only with CARLOS_REGION_HA on. Open returns
+// the complete owner set and two refusals (rulings 37, 39), or, while
+// moves' C8 provider is not bound (dependency C8, ruling 53), the zero
+// owner set: relocating stays refused exactly as before Plan 11.
+type RelocationSource struct {
+ DataDir string
+ Directory hostactors.Directory
+ // FilesStore opens the data bucket as a release.Store (ruling 50);
+ // production is release.NewS3Store.
+ FilesStore func(ctx context.Context, bucket string) (release.Store, error)
+
+ mu sync.Mutex
+ serving ClaimServingSources
+}
+
+var _ ClaimProductionRelocationSource = (*RelocationSource)(nil)
+
+// SetServingSources hands the source the serving launch, markers and the
+// C8 provider before Open (claimproduction.go, inside the region HA gate).
+func (s *RelocationSource) SetServingSources(src ClaimServingSources) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.serving = src
+}
+
+func (s *RelocationSource) Open(ctx context.Context, store release.Store, data lease.VersionedStore, actor hostbox.Actor) (ClaimRelocationOwners, []hostbox.PolicyRefusal, error) {
+ if s == nil {
+ return ClaimRelocationOwners{}, nil, hostwrite.ErrUnavailable
+ }
+ s.mu.Lock()
+ src := s.serving
+ s.mu.Unlock()
+ if store == nil || data == nil || actor.Box == "" || actor.Bucket == "" || actor.Region == "" || s.DataDir == "" || s.FilesStore == nil {
+ return ClaimRelocationOwners{}, nil, hostwrite.ErrUnavailable
+ }
+ if src.Files == nil {
+ // C8 unbound (ruling 53): no owners, no refusals. The dispatcher
+ // keeps relocating refused and Task 10's pass stays off.
+ log.Printf("region HA: relocation owners not opened: the moves C8 files-manifest provider is not bound")
+ return ClaimRelocationOwners{}, nil, nil
+ }
+ if src.Launch == nil || src.Markers == nil {
+ return ClaimRelocationOwners{}, nil, hostwrite.ErrUnavailable
+ }
+ files, err := s.FilesStore(ctx, actor.Bucket)
+ if err != nil {
+ return ClaimRelocationOwners{}, nil, err
+ }
+ releases := StopReceiptReleases{Receipts: RelocationReleaseReceipts{Store: store, Directory: s.Directory}}
+ inputs := &RelocationWakeInputs{Launch: src.Launch, FilesManifest: src.Files, Markers: src.Markers, Claims: store, FilesStore: files,
+ Data: hostbox.RequestStore{Store: data, Check: func(context.Context) error { return nil }}, Bucket: actor.Bucket, Region: actor.Region, Box: actor.Box,
+ NewID: newClaimID, Now: time.Now, Releases: releases}
+ attempts := &relocationProbeAttempts{Claims: store, DataDir: s.DataDir, Box: actor.Box, NewID: newClaimID, Now: time.Now}
+ owners := ClaimRelocationOwners{
+ BeforeAdmission: relocationBeforeAdmission(actor.Box, releases),
+ AfterAdmission: relocationAfterAdmission(s.DataDir, actor.Box),
+ ClaimKindOwners: ClaimKindOwners{Inputs: inputs, AdmissionEvidence: inputs.AdmissionEvidence, PublicationEvidence: inputs.PublicationEvidence},
+ Operations: attempts,
+ Policy: relocationPolicy(actor.Box),
+ }
+ return owners, []hostbox.PolicyRefusal{refuseStaleStamp, refuseLostBox}, nil
+}
+
+func (s *RelocationSource) Close(context.Context) error { return nil }
+
+// newClaimID is the 16-byte hex id claimconstruction.go's newID makes.
+func newClaimID() string {
+ var b [16]byte
+ if _, err := rand.Read(b[:]); err != nil {
+ return ""
+ }
+ return hex.EncodeToString(b[:])
+}
+
+func refuseStaleStamp(_ context.Context, c *hostclaim.Claim, s hostbox.Selection, _ hostwrite.Identity) error {
+ if c != nil && c.Op.Kind == hostclaim.KindRelocating && (c.Holder == nil || s.Captured.Incarnation != c.Holder.Incarnation || s.Captured.Stamp != c.Holder.Stamp) {
+ return hostbox.ErrChanged
+ }
+ return nil
+}
+
+func refuseLostBox(_ context.Context, c *hostclaim.Claim, s hostbox.Selection, _ hostwrite.Identity) error {
+ if c != nil && slices.Contains(c.Lost, s.Box) {
+ return hostbox.ErrChanged
+ }
+ return nil
+}
+
+// relocationBeforeAdmission verifies the source's release (a drain's
+// journaled release at its receipt) or its fence (a failover) before the
+// target's first admission; a later attempt's predecessor is this box's
+// own session, which main's Stop ended.
+func relocationBeforeAdmission(box string, releases hostclaim.ReleaseAuthority) func(context.Context, hostclaim.Observed) error {
+ return func(ctx context.Context, prior hostclaim.Observed) error {
+ c := prior.Claim
+ if c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindRelocating || c.Op.Exec != box {
+ return hostbox.ErrChanged
+ }
+ if hostclaim.PreparedOpened(c) {
+ return nil
+ }
+ var rel hostclaim.Release
+ if json.Unmarshal([]byte(c.Op.Journal["release"]), &rel) == nil && rel.ReleasedBy != "" {
+ return releases.VerifyRelease(ctx, *c, rel)
+ }
+ src := c.Op.Intent["source_box"]
+ if slices.Contains(c.Lost, src) && c.Op.Journal["fence."+src] != "" {
+ return nil
+ }
+ return fmt.Errorf("%w: no release and no fence for %s", hostrelocate.ErrRefused, src)
+ }
+}
+
+// relocationAfterAdmission verifies the target's installation after the
+// admission: the local install receipt is this box's, as journaled, and
+// P's opening keeps install_stamp.
+func relocationAfterAdmission(dataDir, box string) func(context.Context, hostclaim.Observed, hostclaim.Observed) error {
+ return func(_ context.Context, prior, admitted hostclaim.Observed) error {
+ c, n := prior.Claim, admitted.Claim
+ if c == nil || n == nil || n.Holder == nil || n.Op.ID != c.Op.ID || n.Op.Exec != box || n.Holder.WriterBox != box {
+ return hostbox.ErrChanged
+ }
+ var l hostclaim.LocalInstall
+ if json.Unmarshal([]byte(n.Op.Journal["local_install"]), &l) != nil || l.Box != box || l.InstallStamp == 0 {
+ return hostbox.ErrChanged
+ }
+ path, err := relocationInstallPath(dataDir, n.Op.ID)
+ if err != nil {
+ return err
+ }
+ raw, err := os.ReadFile(path)
+ if err != nil {
+ return err
+ }
+ if sum := sha256.Sum256(raw); hex.EncodeToString(sum[:]) != l.JournalSHA256 {
+ return hostbox.ErrChanged
+ }
+ if n.Holder.Generation == n.Op.Journal["prepared"] && n.Holder.Stamp != l.InstallStamp {
+ return hostbox.ErrChanged
+ }
+ return nil
+ }
+}
+
+// relocationAdmit is one probe attempt: the activator's ProbeWake with the
+// probe deadline and, on P's prepared opening, P's install manifest.
+func (a ClaimAssembly) relocationAdmit(cfg RelocationAssemblyConfig) func(context.Context, hostclaim.Observed, bool) (uint64, bool, error) {
+ return func(ctx context.Context, o hostclaim.Observed, certified bool) (uint64, bool, error) {
+ if cfg.Activator == nil || cfg.Versions == nil || o.Claim == nil || o.Claim.Holder == nil {
+ return 0, false, ErrClaimStopUnavailable
+ }
+ c := o.Claim
+ if certified != hostrelocate.CertifiedFor(c) {
+ return 0, false, fmt.Errorf("%w: the participant and the owners disagree on certification", hostrelocate.ErrRefused)
+ }
+ since, err := time.Parse(time.RFC3339Nano, c.Op.Journal["probing_since"])
+ if err != nil {
+ return 0, false, err
+ }
+ pc := activator.ProbeConfig{Host: o.Host, Op: c.Op.ID, HoldOp: c.Op.ID, Deadline: since.Add(hostrelocate.ProbeDeadline)}
+ if !hostclaim.PreparedOpened(c) {
+ auth, err := preparedAuthority(c)
+ if err != nil {
+ return 0, false, err
+ }
+ if auth.InstallManifest != "" {
+ im, _, err := readInstallManifest(ctx, cfg.Versions, cfg.Bucket, moves.GenerationPrefix(c.Holder.Location, auth.Generation))
+ if err != nil {
+ return 0, false, err
+ }
+ pc.Prepared = &im
+ }
+ }
+ res, err := cfg.Activator.ProbeWake(ctx, pc)
+ switch {
+ case err == nil:
+ return res.Session, true, nil
+ case errors.Is(err, activator.ErrProbeFailed), errors.Is(err, activator.ErrAttemptOver):
+ return 0, false, nil
+ default:
+ return 0, false, err
+ }
+ }
+}
diff --git a/internal/host/relocationowners_test.go b/internal/host/relocationowners_test.go
new file mode 100644
index 000000000..3d24a6ba5
--- /dev/null
+++ b/internal/host/relocationowners_test.go
@@ -0,0 +1,643 @@
+package host
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "reflect"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostfiles"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostrelocate"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/superfly/ltx"
+)
+
+const (
+ roHost = "shop.example"
+ roOp = "rel-1"
+ roP = "P1"
+)
+
+var roProbe = activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}
+
+// roReleases is the release authority the owners verify a drain's release
+// with; it records every call.
+type roReleases struct {
+ calls int
+ err error
+}
+
+func (r *roReleases) VerifyRelease(context.Context, hostclaim.Claim, hostclaim.Release) error {
+ r.calls++
+ return r.err
+}
+
+// roFiles is moves' C8 provider: it records each Establish under the work
+// it is given.
+type roFiles struct{ established []string }
+
+func (f *roFiles) Establish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
+ return work.With(ctx, in.Identity, func() error {
+ f.established = append(f.established, fmt.Sprintf("%s@%d", o.Host, o.Claim.Holder.Session))
+ return nil
+ })
+}
+
+// roFixture is a relocation target at relocating/probing, P built and not
+// opened: the claim as the participant's ToProbing leaves it, P's snapshot,
+// anchor and install manifest in the versioned bucket, P's prepared files
+// manifest in the data bucket, and the wake inputs bound as RelocationSource
+// binds them.
+type roFixture struct {
+ t *testing.T
+ claims *release.DirStore
+ files *release.DirStore
+ bucket *leasetest.VersionedBucket
+ releases *roReleases
+ c8 *roFiles
+ inputs *RelocationWakeInputs
+ route registry.Route
+ manifest moves.InstallManifest
+ seq int
+ stamp uint64
+}
+
+type roOptions struct {
+ empty bool // an evidenced empty pin: no snapshot, no install manifest
+ failover bool // no journaled release; the source lost and fenced
+}
+
+func newROFixture(t *testing.T, o roOptions) *roFixture {
+ t.Helper()
+ ctx := t.Context()
+ f := &roFixture{t: t, claims: release.NewDirStore(t.TempDir()), files: release.NewDirStore(t.TempDir()), bucket: leasetest.NewVersionedBucket(), releases: &roReleases{}, c8: &roFiles{}}
+ f.bucket.AddBucket("data")
+ created := time.Date(2026, 10, 7, 1, 0, 0, 0, time.UTC)
+ enrollment := strings.Repeat("a", 32)
+ nh := hostclaim.NewHolder{Family: "instance", Key: "apps/acct/app/instances/" + roHost + ".json", Account: "acct", App: "app", Host: roHost, Region: "ie", RecordRegion: "ie", CreatedAt: created}
+ body := []byte(`{"host":"` + roHost + `","account":"acct","app":"app","region":"ie","created_at":"2026-10-07T01:00:00Z"}`)
+ obs, err := hostclaim.Read(ctx, f.claims, roHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, tr := range []func(*hostclaim.Claim) error{
+ hostclaim.BeginCreateOn(nh, body, "c1", "console", created, hostclaim.ExecutorSelection{Box: "src", Enrollment: enrollment}),
+ hostclaim.CompleteCreate("c1", "src"),
+ hostclaim.AdmitWriter(hostclaim.Admission{Box: "src", Mode: "cold", NewGeneration: "g1", Region: "ie", KeepStamp: true}),
+ } {
+ if obs, err = hostclaim.Update(ctx, f.claims, obs, tr); err != nil {
+ t.Fatal(err)
+ }
+ }
+ h := obs.Claim.Holder
+ prefix := moves.GenerationPrefix(h.Location, roP)
+ auth := hostclaim.PreparedAuthority{Generation: roP, SourceFilesSeq: 5}
+ ref := hostclaim.RecoveryPin{Key: "control/relocations/" + roHost + "/" + roOp + "/pin-1.json", SHA256: strings.Repeat("e", 64), Generation: "g1", FilesSeq: 5, Release: "v1"}
+ if o.empty {
+ ref.Empty = true
+ } else {
+ ref.MaxTXID = 42
+ snap, anchor := moves.SnapshotKey(prefix, 42), moves.AnchorKey(prefix, 42)
+ f.manifest = moves.InstallManifest{V: 1, Op: roOp, Host: roHost, N: 42, SnapshotKey: snap, SnapshotVersion: f.bucket.Put("data", snap, ltxHeader(t, 1, 42)), SnapshotCRC64: 7,
+ AnchorKey: anchor, AnchorVersion: f.bucket.Put("data", anchor, ltxHeader(t, 42, 42)), IntegrityCheck: "ok", DBSHA256: strings.Repeat("d", 64)}
+ raw, _ := json.Marshal(f.manifest)
+ f.bucket.Put("data", moves.InstallManifestKey(prefix), raw)
+ sum := sha256.Sum256(raw)
+ auth.InstallManifest, auth.SHA256, auth.SnapshotTXID = moves.InstallManifestKey(prefix), hex.EncodeToString(sum[:]), 42
+ }
+ pm, _ := json.Marshal(hostfiles.Manifest{V: hostfiles.ManifestVersion, Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}})
+ if _, err := f.files.Put(ctx, hostfiles.ManifestKey(h.Location+"/files/"+roP+"/"), pm, ""); err != nil {
+ t.Fatal(err)
+ }
+ pmSum := sha256.Sum256(pm)
+ auth.FilesManifestSHA256 = hex.EncodeToString(pmSum[:])
+ if obs, err = hostclaim.Update(ctx, f.claims, obs, func(c *hostclaim.Claim) error {
+ f.stamp = c.Gen + 1
+ c.Holder.Stamp = f.stamp
+ auth.InstallStamp = f.stamp
+ stamp := strconv.FormatUint(f.stamp, 10)
+ journal := map[string]string{"pin": string(mustJSON(t, ref)), "prepared": roP, "install_stamp": stamp, "install_release": "v1",
+ "prepared_authority": string(mustJSON(t, auth)), "probing_since": created.Add(time.Hour).Format(time.RFC3339Nano),
+ "exec_ack.src": `{"op":"rel-1"}`, "exec_ack.dst": `{"op":"rel-1"}`,
+ "local_install": string(mustJSON(t, hostclaim.LocalInstall{Box: "dst", Enrollment: enrollment, InstallStamp: f.stamp, JournalSHA256: strings.Repeat("c", 64)}))}
+ if !o.failover {
+ journal["release"] = string(mustJSON(t, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment, Generation: c.Holder.Generation, Session: c.Holder.Session, FinalTXID: ref.MaxTXID, ManifestSeq: 5, Receipt: "r", ReceiptSHA256: strings.Repeat("f", 64)}))
+ } else {
+ c.Lost = []string{"src"}
+ journal["fence.src"] = "control/hosts/" + roHost + "/fences/src.json"
+ }
+ c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: roOp, Phase: hostclaim.PhaseRelocatingProbing, Exec: "dst", Participants: []string{"src", "dst"},
+ Intent: map[string]string{"region": "ie", "source_box": "src", "target_box": "dst", "target_enrollment": enrollment}, Journal: journal}
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+ f.route = registry.Route{Host: roHost, Kind: "instance", App: "app", Account: "acct", Status: "relocating", HoldOp: roOp, DBPath: provision.NestedDBPath("/data", roHost), Socket: provision.NestedSocket(roHost)}
+ f.inputs = &RelocationWakeInputs{
+ Launch: func(context.Context, registry.Route) (ClaimServingLaunch, error) {
+ return ClaimServingLaunch{Identity: hostwrite.Identity{Config: "sha256:config", Release: "sha256:release"}, Executable: "/opt/carlos/live/" + roHost, ExecutableSHA256: strings.Repeat("b", 64), Version: "v1"}, nil
+ },
+ FilesManifest: f.c8, Markers: plan9D5Markers{f.bucket}, Claims: f.claims, FilesStore: f.files, Data: f.bucket, Bucket: "data", Region: "ie", Box: "dst",
+ NewID: func() string { f.seq++; return fmt.Sprintf("fresh%d", f.seq) }, Now: func() time.Time { return created.Add(2 * time.Hour) }, Releases: f.releases}
+ return f
+}
+
+func ltxHeader(t *testing.T, min, max uint64) []byte {
+ t.Helper()
+ body, err := (&ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, MinTXID: ltx.TXID(min), MaxTXID: ltx.TXID(max)}).MarshalBinary()
+ if err != nil {
+ t.Fatal(err)
+ }
+ return body
+}
+
+func (f *roFixture) read() hostclaim.Observed {
+ f.t.Helper()
+ o, err := hostclaim.Read(f.t.Context(), f.claims, roHost)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ return o
+}
+
+func (f *roFixture) update(tr func(*hostclaim.Claim) error) hostclaim.Observed {
+ f.t.Helper()
+ o, err := hostclaim.Update(f.t.Context(), f.claims, f.read(), tr)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ return o
+}
+
+// admit counts the attempt and applies the selected admission with the
+// claim's own CAS, as the participant and the wake do.
+func (f *roFixture) admit(o hostclaim.Observed, in ClaimLaunchInputs) hostclaim.Observed {
+ f.t.Helper()
+ counted, err := hostclaim.Update(f.t.Context(), f.claims, o, hostclaim.BeginProbeAttempt(roOp, "dst"))
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ in.Admission.ObservedSession, in.Admission.ObservedGeneration = counted.Claim.Holder.Session, counted.Claim.Holder.Generation
+ admitted, err := hostclaim.Update(f.t.Context(), f.claims, counted, hostclaim.AdmitWriter(in.Admission))
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ return admitted
+}
+
+// C4, ruling 47: P's first opening is cold into P, keeping the install
+// stamp, prepared from P's snapshot, and certified only by this
+// operation's verified release; a failover's opening is not certified; an
+// empty pin's opening is empty, with no baseline.
+func TestRelocationSelectCertifiesOnlyPsOpeningAfterItsRelease(t *testing.T) {
+ ctx := t.Context()
+ t.Run("a drain's release", func(t *testing.T) {
+ f := newROFixture(t, roOptions{})
+ o := f.read()
+ in, err := f.inputs.Select(ctx, f.route, o, roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ h := o.Claim.Holder
+ want := hostclaim.Admission{ObservedOpID: roOp, ObservedPhase: hostclaim.PhaseRelocatingProbing, ObservedSession: h.Session, ObservedGeneration: h.Generation,
+ Box: "dst", Mode: "cold", NewGeneration: roP, KeepStamp: true, PredecessorCertified: true, Region: "ie", PredecessorRegion: "ie", RemoteMaxTXID: 42}
+ if !reflect.DeepEqual(in.Admission, want) {
+ t.Fatalf("admission:\n got %+v\nwant %+v", in.Admission, want)
+ }
+ if !in.Prepared || in.Empty || in.Baseline == nil || in.Baseline.Generation != roP || in.Baseline.MaxTXID != 42 || f.releases.calls != 1 {
+ t.Fatalf("prepared=%v empty=%v baseline=%+v release checks=%d", in.Prepared, in.Empty, in.Baseline, f.releases.calls)
+ }
+ if in.DBPath != f.route.DBPath || in.Version != "v1" || len(in.Roles) != 1 || in.Roles[0] != "main" || in.ReadyPath != "/" {
+ t.Fatalf("launch inputs %+v", in)
+ }
+ f.releases.err = errors.New("the receipt does not verify")
+ if _, err := f.inputs.Select(ctx, f.route, o, roProbe); !errors.Is(err, f.releases.err) {
+ t.Fatalf("an unverified release certified: %v", err)
+ }
+ })
+ t.Run("a failover", func(t *testing.T) {
+ f := newROFixture(t, roOptions{failover: true})
+ in, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if in.Admission.PredecessorCertified || !in.Admission.KeepStamp || in.Admission.NewGeneration != roP || f.releases.calls != 0 {
+ t.Fatalf("failover admission %+v (release checks %d)", in.Admission, f.releases.calls)
+ }
+ })
+ t.Run("an empty pin", func(t *testing.T) {
+ f := newROFixture(t, roOptions{empty: true})
+ in, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !in.Empty || in.Prepared || in.Baseline != nil || in.Admission.NewGeneration != roP || !in.Admission.KeepStamp || in.Admission.RemoteMaxTXID != 0 {
+ t.Fatalf("empty opening %+v", in)
+ }
+ if err := f.inputs.EmptyAuthority(ctx, f.read(), in); err != nil {
+ t.Fatalf("the evidenced empty pin: %v", err)
+ }
+ })
+}
+
+// P is opened once. After its opening every attempt is a fresh generation
+// from the predecessor, keeping no stamp, never certified, and evidence for
+// P's opening is refused on the opened claim.
+func TestRelocationSelectNeverReopensP(t *testing.T) {
+ f := newROFixture(t, roOptions{})
+ ctx := t.Context()
+ first, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ opened := f.admit(f.read(), first)
+ if !hostclaim.PreparedOpened(opened.Claim) || opened.Claim.Holder.Stamp != f.stamp {
+ t.Fatalf("P's opening: %+v", opened.Claim.Holder)
+ }
+ again, err := f.inputs.Select(ctx, f.route, opened, roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ a := again.Admission
+ if a.NewGeneration != "fresh1" || a.KeepStamp || a.PredecessorCertified || again.Prepared || a.ObservedGeneration != roP || a.ObservedSession != opened.Claim.Holder.Session {
+ t.Fatalf("the later attempt %+v prepared=%v", a, again.Prepared)
+ }
+ if err := f.inputs.AdmissionEvidence(ctx, opened, again.Admission, again.Baseline); err != nil {
+ t.Fatalf("the later attempt's evidence: %v", err)
+ }
+ reopen := again.Admission
+ reopen.NewGeneration, reopen.KeepStamp = roP, true
+ if err := f.inputs.AdmissionEvidence(ctx, opened, reopen, first.Baseline); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("P reopened: %v", err)
+ }
+}
+
+// The inputs serve only this box's relocation probe, of the pinned
+// release, at a nested route.
+func TestRelocationSelectRefusesEverythingButThisBoxsProbe(t *testing.T) {
+ ctx := t.Context()
+ for name, tc := range map[string]struct {
+ claim func(*hostclaim.Claim) error
+ policy activator.StartPolicy
+ route func(*registry.Route)
+ launch string
+ want error
+ }{
+ "installed": {claim: func(c *hostclaim.Claim) error { c.Op.Phase = hostclaim.PhaseInstalled; return nil }, policy: roProbe, want: hostrelocate.ErrRefused},
+ "another executor": {claim: func(c *hostclaim.Claim) error { c.Op.Exec, c.Op.Intent["target_box"] = "src", "src"; return nil }, policy: roProbe, want: hostrelocate.ErrRefused},
+ "stuck": {claim: func(c *hostclaim.Claim) error { c.Op.Journal["stuck"] = "an operator"; return nil }, policy: roProbe, want: hostrelocate.ErrRefused},
+ "not a probe": {policy: activator.StartPolicy{MaxStarts: 5, Window: time.Minute}, want: hostrelocate.ErrRefused},
+ "another release": {policy: roProbe, launch: "v2", want: hostrelocate.ErrRefused},
+ "a relative database": {policy: roProbe, route: func(rt *registry.Route) { rt.DBPath = "shop.db" }, want: hostbox.ErrUnsupported},
+ "another host": {policy: roProbe, route: func(rt *registry.Route) { rt.Host = "other.example" }, want: hostbox.ErrUnsupported},
+ } {
+ t.Run(name, func(t *testing.T) {
+ f := newROFixture(t, roOptions{})
+ if tc.claim != nil {
+ f.update(tc.claim)
+ }
+ rt := f.route
+ if tc.route != nil {
+ tc.route(&rt)
+ }
+ if tc.launch != "" {
+ launch := f.inputs.Launch
+ f.inputs.Launch = func(ctx context.Context, rt registry.Route) (ClaimServingLaunch, error) {
+ l, err := launch(ctx, rt)
+ l.Version = tc.launch
+ return l, err
+ }
+ }
+ if _, err := f.inputs.Select(ctx, rt, f.read(), tc.policy); !errors.Is(err, tc.want) {
+ t.Fatalf("got %v, want %v", err, tc.want)
+ }
+ })
+ }
+}
+
+// Ruling 47: P's opening is bound to its observation, its certified
+// baseline (P's snapshot at the pin's N) and P's prepared files manifest by
+// the digest prepared_authority journals, read again.
+func TestRelocationAdmissionEvidenceBindsTheBaselineAndThePreparedManifest(t *testing.T) {
+ ctx := t.Context()
+ f := newROFixture(t, roOptions{})
+ o := f.read()
+ in, err := f.inputs.Select(ctx, f.route, o, roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := f.inputs.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); err != nil {
+ t.Fatal(err)
+ }
+ for name, mutate := range map[string]func(*hostclaim.Admission, *genselect.Plan){
+ "a stale baseline": func(a *hostclaim.Admission, p *genselect.Plan) { p.MaxTXID, a.RemoteMaxTXID = 41, 41 },
+ "another generation": func(_ *hostclaim.Admission, p *genselect.Plan) { p.Generation = "g1" },
+ "stamp not kept": func(a *hostclaim.Admission, _ *genselect.Plan) { a.KeepStamp = false },
+ "not certified": func(a *hostclaim.Admission, _ *genselect.Plan) { a.PredecessorCertified = false },
+ "another box": func(a *hostclaim.Admission, _ *genselect.Plan) { a.Box = "src" },
+ "sealed coverage": func(a *hostclaim.Admission, _ *genselect.Plan) { a.Coverage = "sealed" },
+ "another region": func(a *hostclaim.Admission, _ *genselect.Plan) { a.Region = "uk" },
+ "another session": func(a *hostclaim.Admission, _ *genselect.Plan) { a.ObservedSession++ },
+ "warm": func(a *hostclaim.Admission, _ *genselect.Plan) { a.Mode = "warm" },
+ "lost writes claimed": func(a *hostclaim.Admission, _ *genselect.Plan) { a.AdditionalWritesLost = []hostclaim.WritesLost{{}} },
+ } {
+ a, p := in.Admission, *in.Baseline
+ mutate(&a, &p)
+ if err := f.inputs.AdmissionEvidence(ctx, o, a, &p); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("%s: %v", name, err)
+ }
+ }
+ // P's files manifest replaced after the build: the digest no longer
+ // matches the authority.
+ key := hostfiles.ManifestKey(o.Claim.Holder.Location + "/files/" + roP + "/")
+ _, etag, err := f.files.Get(ctx, key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ other, _ := json.Marshal(hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: 0, Entries: map[string]hostfiles.Entry{"x": {Seq: 1, Deleted: true}}, Platform: map[string]hostfiles.Entry{}})
+ if _, err := f.files.Put(ctx, key, other, etag); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.inputs.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("another files manifest: %v", err)
+ }
+ // An empty pin's opening has no baseline; a non-empty one must.
+ e := newROFixture(t, roOptions{empty: true})
+ eo := e.read()
+ ein, err := e.inputs.Select(ctx, e.route, eo, roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := e.inputs.AdmissionEvidence(ctx, eo, ein.Admission, nil); err != nil {
+ t.Fatalf("the empty opening: %v", err)
+ }
+ if err := f.inputs.AdmissionEvidence(ctx, o, in.Admission, nil); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("a non-empty opening without its baseline: %v", err)
+ }
+}
+
+// Ruling 41: one admitted session is bound to one record attempt k (the
+// claim's probe_attempts.<box>); the attempt's first app start is kept
+// across restarts; a start is reserved, settled and re-reserved under the
+// same session; a session bound to k is refused once the record moves on.
+// The journal is keyed by op (ruling 44).
+func TestRelocationProbeAttemptsBindOneSessionPerRecordAttempt(t *testing.T) {
+ ctx := t.Context()
+ f := newROFixture(t, roOptions{})
+ in, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ admitted := f.admit(f.read(), in)
+ h := admitted.Claim.Holder
+ dataDir := t.TempDir()
+ now := time.Date(2026, 10, 7, 4, 0, 0, 0, time.UTC)
+ p := &relocationProbeAttempts{Claims: f.claims, DataDir: dataDir, Box: "dst", NewID: f.inputs.NewID, Now: func() time.Time { return now }}
+ owner := hostguard.FenceTarget{Region: "ie", Host: roHost, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Box: "dst", Enrollment: strings.Repeat("a", 32), BootID: "boot", ControllerID: "ctl"}
+ s := OperationStart{Admitted: admitted, Policy: roProbe, Inputs: in, Owner: owner}
+ opened, err := p.OpenProbe(ctx, s, now)
+ if err != nil || !opened.Equal(now) {
+ t.Fatalf("open %v %v", opened, err)
+ }
+ if again, err := p.OpenProbe(ctx, s, now.Add(time.Minute)); err != nil || !again.Equal(now) {
+ t.Fatalf("a restart moved the attempt's first start: %v %v", again, err)
+ }
+ if err := p.Reserve(ctx, s, 0); err != nil {
+ t.Fatal(err)
+ }
+ a := p.CurrentAttempt(s)
+ if a.ID == "" || a.Session != h.Session || a.Box != "dst" || a.ControllerID != "ctl" {
+ t.Fatalf("the reservation %+v", a)
+ }
+ if got, err := (relocationAttemptReader{DataDir: dataDir}).ReadAttempt(ctx, admitted); err != nil || got != a {
+ t.Fatalf("root reads %+v %v", got, err)
+ }
+ if err := p.Reserve(ctx, s, 0); err != nil || p.CurrentAttempt(s) != a {
+ t.Fatalf("a repeated reservation: %v", err)
+ }
+ if err := p.Reserve(ctx, s, 1); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("a restart before the first start settled: %v", err)
+ }
+ if err := p.Failed(ctx, s, 0); err != nil {
+ t.Fatal(err)
+ }
+ if err := p.Reserve(ctx, s, 1); err != nil {
+ t.Fatalf("the restart after a failed start: %v", err)
+ }
+ if b := p.CurrentAttempt(s); b.Session != h.Session || b.LaunchInputs != a.LaunchInputs {
+ t.Fatalf("the restart %+v", b)
+ }
+ path, _ := relocationAttemptsPath(dataDir, roOp)
+ if strings.Contains(path, roHost) || filepath.Base(filepath.Dir(path)) != roOp {
+ t.Fatalf("the journal path %s", path)
+ }
+ // The record attempt moves on with the same session admitted: the
+ // session is bound to attempt 1 and refused for attempt 2.
+ moved := f.update(func(c *hostclaim.Claim) error { c.Op.Journal["probe_attempts.dst"] = "2"; return nil })
+ s.Admitted = moved
+ if err := p.Reserve(ctx, s, 0); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("one session, two record attempts: %v", err)
+ }
+ // Another session than the admitted writer is refused before any journal.
+ other := OperationStart{Admitted: admitted, Policy: roProbe, Inputs: in, Owner: owner}
+ other.Owner.Session++
+ if _, err := p.OpenProbe(ctx, other, now); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("another session: %v", err)
+ }
+}
+
+// The admission hooks: before P's first admission the source's release is
+// verified at its receipt (a drain) or its fence is journaled with the
+// source lost (a failover); after it, this box's install journal is the
+// receipt's and P keeps the install stamp.
+func TestRelocationAdmissionHooksVerifyReleaseAndInstall(t *testing.T) {
+ ctx := t.Context()
+ releases := &roReleases{}
+ before := relocationBeforeAdmission("dst", releases)
+ f := newROFixture(t, roOptions{})
+ if err := before(ctx, f.read()); err != nil || releases.calls != 1 {
+ t.Fatalf("a drain's release: %v (%d checks)", err, releases.calls)
+ }
+ releases.err = errors.New("no receipt")
+ if err := before(ctx, f.read()); !errors.Is(err, releases.err) {
+ t.Fatalf("an unverified release: %v", err)
+ }
+ g := newROFixture(t, roOptions{failover: true})
+ if err := before(ctx, g.read()); err != nil {
+ t.Fatalf("a fenced failover: %v", err)
+ }
+ unfenced := g.update(func(c *hostclaim.Claim) error { delete(c.Op.Journal, "fence.src"); return nil })
+ if err := before(ctx, unfenced); !errors.Is(err, hostrelocate.ErrRefused) {
+ t.Fatalf("neither release nor fence: %v", err)
+ }
+ if err := relocationBeforeAdmission("other", releases)(ctx, f.read()); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("another box: %v", err)
+ }
+
+ // After: the install journal at the receipt's digest.
+ dataDir := t.TempDir()
+ after := relocationAfterAdmission(dataDir, "dst")
+ in, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ prior := f.read()
+ path, _ := relocationInstallPath(dataDir, roOp)
+ if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ journal := []byte(`{"v":1,"host":"shop.example","op":"rel-1"}`)
+ if err := os.WriteFile(path, journal, 0o600); err != nil {
+ t.Fatal(err)
+ }
+ sum := sha256.Sum256(journal)
+ f.update(func(c *hostclaim.Claim) error {
+ var l hostclaim.LocalInstall
+ _ = json.Unmarshal([]byte(c.Op.Journal["local_install"]), &l)
+ l.JournalSHA256 = hex.EncodeToString(sum[:])
+ c.Op.Journal["local_install"] = string(mustJSON(t, l))
+ return nil
+ })
+ admitted := f.admit(f.read(), in)
+ if err := after(ctx, prior, admitted); err != nil {
+ t.Fatalf("after P's opening: %v", err)
+ }
+ if err := os.WriteFile(path, []byte(`{"v":1}`), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if err := after(ctx, prior, admitted); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("another install journal: %v", err)
+ }
+ if err := relocationAfterAdmission(dataDir, "src")(ctx, prior, admitted); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("another box's admission: %v", err)
+ }
+}
+
+// relocationProbeActivator is the activator's probe and fence, recorded.
+type relocationProbeActivator struct {
+ probes []activator.ProbeConfig
+ result activator.ProbeResult
+ err error
+}
+
+func (a *relocationProbeActivator) ProbeWake(_ context.Context, pc activator.ProbeConfig) (activator.ProbeResult, error) {
+ a.probes = append(a.probes, pc)
+ return a.result, a.err
+}
+
+// The probe attempt never fences: a fence asked of this double fails the
+// test's call with an error naming it, and lifts nothing.
+func (a *relocationProbeActivator) FenceAndStopAs(context.Context, string, uint64, string) (*activator.StopEvidence, error) {
+ return nil, errors.New("the probe attempt asked for a fence")
+}
+
+func (a *relocationProbeActivator) LiftFenceAs(string, uint64, string) bool { return false }
+
+// One probe attempt is one ProbeWake, bounded by the probe deadline,
+// carrying P's install manifest on P's opening; a failed or overdue
+// attempt is not passed, and any other error is returned.
+func TestRelocationAdmitRunsOneProbeAttempt(t *testing.T) {
+ ctx := t.Context()
+ f := newROFixture(t, roOptions{})
+ act := &relocationProbeActivator{result: activator.ProbeResult{Session: 99}}
+ admit := ClaimAssembly{}.relocationAdmit(RelocationAssemblyConfig{Activator: act, Versions: f.bucket, Bucket: "data"})
+ o := f.read()
+ session, passed, err := admit(ctx, o, true)
+ if err != nil || !passed || session != 99 || len(act.probes) != 1 {
+ t.Fatalf("passed: %d %v %v", session, passed, err)
+ }
+ pc := act.probes[0]
+ since, _ := time.Parse(time.RFC3339Nano, o.Claim.Op.Journal["probing_since"])
+ if pc.Host != roHost || pc.Op != roOp || pc.HoldOp != roOp || !pc.Deadline.Equal(since.Add(hostrelocate.ProbeDeadline)) || pc.Prepared == nil || *pc.Prepared != f.manifest {
+ t.Fatalf("probe config %+v", pc)
+ }
+ for _, failed := range []error{activator.ErrProbeFailed, activator.ErrAttemptOver} {
+ act.err = failed
+ if _, passed, err := admit(ctx, o, true); passed || err != nil {
+ t.Fatalf("%v: passed=%v %v", failed, passed, err)
+ }
+ }
+ act.err = errors.New("the activator is closing")
+ if _, _, err := admit(ctx, o, true); !errors.Is(err, act.err) {
+ t.Fatalf("another error: %v", err)
+ }
+ if _, _, err := admit(ctx, o, false); !errors.Is(err, hostrelocate.ErrRefused) {
+ t.Fatalf("certification disagreed: %v", err)
+ }
+ if n := len(act.probes); n != 4 {
+ t.Fatalf("%d probes, want 4", n)
+ }
+}
+
+// Rulings 37 and 39: with moves' C8 provider bound, Open returns the
+// complete owner set and the two refusals; the relocating dispatch binds.
+// Without the region's bucket or a store constructor it is unavailable,
+// and a nil source is refused before it is locked.
+func TestRelocationSourceOpensACompleteOwnerSet(t *testing.T) {
+ ctx := t.Context()
+ var opened []string
+ src := &RelocationSource{DataDir: t.TempDir(), FilesStore: func(_ context.Context, bucket string) (release.Store, error) {
+ opened = append(opened, bucket)
+ return release.NewDirStore(t.TempDir()), nil
+ }}
+ src.SetServingSources(ClaimServingSources{
+ Launch: func(context.Context, registry.Route) (ClaimServingLaunch, error) { return ClaimServingLaunch{}, nil },
+ Markers: plan9D5Markers{leasetest.NewVersionedBucket()},
+ Files: &roFiles{},
+ })
+ actor := hostbox.Actor{Box: "dst", Bucket: "data-ie", Region: "ie"}
+ owners, refusals, err := src.Open(ctx, release.NewDirStore(t.TempDir()), &fakeVersioned{}, actor)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if owners.zero() || !owners.complete() || owners.Operations == nil || owners.Policy == nil || owners.BeforeAdmission == nil || owners.AfterAdmission == nil || len(refusals) != 2 {
+ t.Fatalf("owners %+v refusals %d", owners, len(refusals))
+ }
+ if len(opened) != 1 || opened[0] != "data-ie" {
+ t.Fatalf("the files store was opened for %v", opened)
+ }
+ if bound, err := claimRelocationC1(plan11Env("1", "1"), true, owners); !bound || err != nil {
+ t.Fatalf("the relocating dispatch did not bind: %v %v", bound, err)
+ }
+ if _, _, err := src.Open(ctx, release.NewDirStore(t.TempDir()), &fakeVersioned{}, hostbox.Actor{Box: "dst", Region: "ie"}); !errors.Is(err, hostwrite.ErrUnavailable) {
+ t.Fatalf("no bucket: %v", err)
+ }
+ var none *RelocationSource
+ if _, _, err := none.Open(ctx, release.NewDirStore(t.TempDir()), &fakeVersioned{}, actor); !errors.Is(err, hostwrite.ErrUnavailable) {
+ t.Fatalf("a nil source: %v", err)
+ }
+}
+
+// Finding 11 (C8): without moves' files-manifest provider the source opens
+// no owners, and the dispatcher keeps relocating refused.
+func TestRelocationStaysUnboundWithoutTheC8Provider(t *testing.T) {
+ src := &RelocationSource{DataDir: t.TempDir(), FilesStore: func(context.Context, string) (release.Store, error) { return release.NewDirStore(t.TempDir()), nil }}
+ src.SetServingSources(ClaimServingSources{Launch: func(context.Context, registry.Route) (ClaimServingLaunch, error) { return ClaimServingLaunch{}, nil }, Markers: nil})
+ owners, refusals, err := src.Open(context.Background(), release.NewDirStore(t.TempDir()), &fakeVersioned{}, hostbox.Actor{Box: "dst", Bucket: "data-ie", Region: "ie"})
+ if err != nil || !owners.zero() || refusals != nil {
+ t.Fatalf("opened without C8: %+v %v %v", owners, refusals, err)
+ }
+ bound, err := claimRelocationC1(plan11Env("1", "1"), true, owners)
+ if bound || err != nil {
+ t.Fatalf("relocating bound without C8: %v %v", bound, err)
+ }
+}
diff --git a/internal/host/relocationwake.go b/internal/host/relocationwake.go
new file mode 100644
index 000000000..d13df9d55
--- /dev/null
+++ b/internal/host/relocationwake.go
@@ -0,0 +1,432 @@
+package host
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "reflect"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostfiles"
+ "github.com/carlosframework/platform/internal/hostrelocate"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// RelocationWakeInputs is the relocating kind's ClaimWakeInputs (claim
+// Plan 11 R1). Field names are distinct (finding 14): FilesManifest is
+// moves' C8 provider, FilesStore the data bucket as a release.Store.
+type RelocationWakeInputs struct {
+ Launch func(context.Context, registry.Route) (ClaimServingLaunch, error)
+ FilesManifest ClaimFilesManifest
+ Markers ClaimMarkerStore
+ Claims release.Store
+ FilesStore release.Store
+ Data genselect.Reader
+ Bucket string
+ Region string
+ Box string
+ NewID func() string
+ Now func() time.Time
+ // Releases verifies a journaled release at its receipt (stage A's
+ // StopReceiptReleases over main's Complete).
+ Releases hostclaim.ReleaseAuthority
+
+ mu sync.Mutex
+ fences map[string]hostwrite.Work // host\x00session: the admitted spawn selection Materialize ran under
+ published map[string]ClaimPublicationReceipt // host\x00session: P's prepared marker
+ fresh *claimServingOwners // a fresh generation's marker: main's writer (ruling 47)
+}
+
+var (
+ _ ClaimWakeInputs = (*RelocationWakeInputs)(nil)
+ _ ClaimFencedPublication = (*RelocationWakeInputs)(nil)
+)
+
+func (r *RelocationWakeInputs) valid() bool {
+ return r != nil && r.Launch != nil && r.FilesManifest != nil && r.Markers != nil && r.Claims != nil && r.FilesStore != nil && r.Data != nil &&
+ r.Bucket != "" && r.Region != "" && r.Box != "" && r.NewID != nil && r.Now != nil && r.Releases != nil
+}
+
+func (r *RelocationWakeInputs) probe(o hostclaim.Observed, p activator.StartPolicy) error {
+ if !r.valid() {
+ return hostwrite.ErrUnavailable
+ }
+ c := o.Claim
+ if o.Absent || c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindRelocating || c.Op.Phase != hostclaim.PhaseRelocatingProbing || c.Op.Exec != r.Box || !p.Probe || c.Op.Journal["stuck"] != "" {
+ return fmt.Errorf("%w: not this box's relocation probe", hostrelocate.ErrRefused)
+ }
+ return nil
+}
+
+// freshOwners is main's serving publisher bound to this box and region,
+// used only through its kind-free halves (publishFresh, freshEvidence).
+func (r *RelocationWakeInputs) freshOwners() *claimServingOwners {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ if r.fresh == nil {
+ r.fresh = &claimServingOwners{Data: r.Data, Bucket: r.Bucket, Region: r.Region, Box: r.Box, NewID: r.NewID, Now: r.Now, Sources: ClaimServingSources{Markers: r.Markers}}
+ }
+ return r.fresh
+}
+
+func (r *RelocationWakeInputs) request(o hostclaim.Observed) genselect.Request {
+ return genselect.Request{Bucket: r.Bucket, Region: r.Region, Host: o.Host, Holder: *o.Claim.Holder}
+}
+
+func preparedAuthority(c *hostclaim.Claim) (hostclaim.PreparedAuthority, error) {
+ var auth hostclaim.PreparedAuthority
+ if json.Unmarshal([]byte(c.Op.Journal["prepared_authority"]), &auth) != nil || auth.Generation == "" || auth.Generation != c.Op.Journal["prepared"] {
+ return hostclaim.PreparedAuthority{}, hostbox.ErrChanged
+ }
+ return auth, nil
+}
+
+// Select (§5.8 "The probe's admission"): the first attempt opens P with
+// KeepStamp, certified only by this operation's verified release (C4);
+// it is a prepared admission (Prepared, main's pre-replication marker)
+// when P has a snapshot, and an empty one otherwise (ruling 47). Every
+// later attempt is cold into a fresh generation from the predecessor.
+func (r *RelocationWakeInputs) Select(ctx context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (ClaimLaunchInputs, error) {
+ if err := r.probe(o, p); err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ c, h := o.Claim, o.Claim.Holder
+ if rt.Host != o.Host || !strings.HasPrefix(rt.DBPath, "/") {
+ return ClaimLaunchInputs{}, hostbox.ErrUnsupported
+ }
+ l, err := r.Launch(ctx, rt)
+ if err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ if l.Identity.Config == "" || l.Identity.Release == "" || l.Executable == "" || len(l.ExecutableSHA256) != 64 || l.Version == "" {
+ return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
+ }
+ if l.Version != c.Op.Journal["install_release"] {
+ return ClaimLaunchInputs{}, fmt.Errorf("%w: the staged release %s is not the pinned %s", hostrelocate.ErrRefused, l.Version, c.Op.Journal["install_release"])
+ }
+ prepared := c.Op.Journal["prepared"]
+ in := ClaimLaunchInputs{Roles: []string{"main"}, Identity: l.Identity, Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: append([]string(nil), l.Env...),
+ DBPath: rt.DBPath, Version: l.Version, Limit: l.Limit, ReadyPath: "/"}
+ adm := hostclaim.Admission{ObservedOpID: c.Op.ID, ObservedPhase: hostclaim.PhaseRelocatingProbing, ObservedSession: h.Session, ObservedGeneration: h.Generation,
+ Box: r.Box, Mode: "cold", Region: r.Region, PredecessorRegion: r.Region}
+ if n := len(h.Admissions); n > 0 {
+ adm.PredecessorRegion = h.Admissions[n-1].Region
+ }
+ if !hostclaim.PreparedOpened(c) {
+ certified := hostrelocate.CertifiedFor(c)
+ if certified {
+ var rel hostclaim.Release
+ if json.Unmarshal([]byte(c.Op.Journal["release"]), &rel) != nil {
+ return ClaimLaunchInputs{}, hostbox.ErrChanged
+ }
+ if err := r.Releases.VerifyRelease(ctx, *c, rel); err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ }
+ auth, err := preparedAuthority(c)
+ if err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ adm.NewGeneration, adm.KeepStamp, adm.PredecessorCertified = prepared, true, certified
+ if auth.InstallManifest == "" {
+ // An evidenced empty pin opened as P: no snapshot, so no root
+ // until the first transaction; main's fenced publisher waits
+ // for it (ErrClaimRootPending).
+ in.Empty = true
+ in.Admission = adm
+ return in, nil
+ }
+ im, _, err := readInstallManifest(ctx, r.Data, r.Bucket, moves.GenerationPrefix(h.Location, prepared))
+ if err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ plan, err := genselect.SelectPrepared(ctx, r.Data, r.request(o), genselect.PreparedPin{Generation: prepared, SnapshotKey: im.SnapshotKey, SnapshotVersion: im.SnapshotVersion, SnapshotTXID: im.N, SnapshotCRC64: im.SnapshotCRC64})
+ if err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ in.Prepared = true
+ adm.RemoteMaxTXID, in.Baseline = plan.MaxTXID, plan
+ in.Admission = adm
+ return in, nil
+ }
+ adm.NewGeneration = r.NewID()
+ if !claimFreshGeneration(h, adm.NewGeneration) {
+ return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
+ }
+ plan, err := genselect.Select(ctx, r.Data, r.request(o))
+ switch {
+ case errors.Is(err, genselect.ErrEmpty):
+ in.Empty = true
+ case err != nil:
+ return ClaimLaunchInputs{}, err
+ default:
+ adm.RemoteMaxTXID, in.Baseline = plan.MaxTXID, plan
+ }
+ in.Admission = adm
+ return in, nil
+}
+
+// Materialize, under the admitted spawn selection: the held identities
+// proved, C8 establishes the files manifest (Establish publishes
+// <host>.env's exact bytes before it returns, O1), the selection is
+// recorded as the session's fence, and on P's prepared opening P's marker
+// is published under that fence (main's prepared-publication contract,
+// ruling 47). The held render and release ran at the local install.
+func (r *RelocationWakeInputs) Materialize(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
+ if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil {
+ return err
+ }
+ if work == nil || in.Identity.Config == "" || in.Identity.Release == "" {
+ return hostwrite.ErrUnavailable
+ }
+ fence := writerFence(work, in.Identity)
+ if err := fence(ctx); err != nil {
+ return err
+ }
+ if err := r.FilesManifest.Establish(ctx, o, in, work); err != nil {
+ return err
+ }
+ h := o.Claim.Holder
+ r.mu.Lock()
+ if r.fences == nil {
+ r.fences = map[string]hostwrite.Work{}
+ }
+ r.fences[claimPublishedKey(o.Host, h.Session)] = work
+ r.mu.Unlock()
+ if !in.Prepared {
+ return nil
+ }
+ receipt, err := r.preparedMarker(ctx, o, fence)
+ if err != nil {
+ return err
+ }
+ r.record(o.Host, h.Session, receipt)
+ return nil
+}
+
+func (r *RelocationWakeInputs) record(host string, session uint64, receipt ClaimPublicationReceipt) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ if r.published == nil {
+ r.published = map[string]ClaimPublicationReceipt{}
+ }
+ r.published[claimPublishedKey(host, session)] = receipt
+}
+
+// preparedMarker establishes P's verified marker under fence: install.json
+// is the journaled prepared authority's bytes, and the location's
+// generation.json names P with the build's pin (left as it is when it
+// already does) or supersedes the current marker by its exact ETag.
+func (r *RelocationWakeInputs) preparedMarker(ctx context.Context, o hostclaim.Observed, fence func(context.Context) error) (ClaimPublicationReceipt, error) {
+ c, h := o.Claim, o.Claim.Holder
+ auth, err := preparedAuthority(c)
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
+ defer cancel()
+ im, raw, err := readInstallManifest(bounded, r.Data, r.Bucket, moves.GenerationPrefix(h.Location, auth.Generation))
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ sum := sha256.Sum256(raw)
+ if hex.EncodeToString(sum[:]) != auth.SHA256 || im.Host != o.Host || im.N != auth.SnapshotTXID || h.Generation != auth.Generation {
+ return ClaimPublicationReceipt{}, fmt.Errorf("%w: install.json is not the prepared authority's", hostbox.ErrChanged)
+ }
+ root := genselect.File{Key: im.SnapshotKey, VersionID: im.SnapshotVersion, Level: 9, MinTXID: 1, MaxTXID: im.N}
+ current, err := (&movesProbeOwners{Data: r.Data, Bucket: r.Bucket}).currentMarker(bounded, h) // moves' reader, one implementation
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ expected := ""
+ if current != nil {
+ got, err := r.Data.GetVersion(bounded, r.Bucket, markerKey(h), current.ID)
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ var marker genselect.Marker
+ if err := json.Unmarshal(got, &marker); err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ if marker.Op == auth.Generation {
+ if marker.Incarnation != h.Incarnation || marker.Host != o.Host || marker.Box != r.Box || marker.ManifestSHA256 != auth.SHA256 {
+ return ClaimPublicationReceipt{}, hostbox.ErrChanged
+ }
+ return ClaimPublicationReceipt{MarkerVersion: current.ID, Root: root}, nil
+ }
+ expected = current.ETag
+ }
+ body, err := json.Marshal(genselect.Marker{Op: auth.Generation, Incarnation: h.Incarnation, Host: o.Host, Box: r.Box, InstalledAt: r.Now().UTC(), ManifestSHA256: auth.SHA256, Pin: manifestPin(im)})
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ if err := fence(bounded); err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ version, err := r.Markers.PutMarker(bounded, r.Bucket, markerKey(h), body, expected, fence)
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ if version == "" {
+ return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
+ }
+ return ClaimPublicationReceipt{MarkerVersion: version, Root: root}, nil
+}
+
+// EmptyAuthority: P's opening is empty only with an evidenced empty pin;
+// a later attempt over an empty predecessor is an ordinary empty cold
+// admission (a complete listing of the location holds nothing).
+func (r *RelocationWakeInputs) EmptyAuthority(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
+ if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil {
+ return err
+ }
+ if !in.Empty || in.Baseline != nil || in.Prepared {
+ return hostbox.ErrChanged
+ }
+ if in.Admission.NewGeneration == o.Claim.Op.Journal["prepared"] {
+ var ref hostclaim.RecoveryPin
+ if json.Unmarshal([]byte(o.Claim.Op.Journal["pin"]), &ref) != nil || !ref.Empty {
+ return fmt.Errorf("%w: no evidenced empty pin", hostrelocate.ErrRefused)
+ }
+ return nil
+ }
+ versions, err := r.Data.ListVersions(ctx, r.Bucket, strings.TrimSuffix(o.Claim.Holder.Location, "/")+"/")
+ if err != nil {
+ return err
+ }
+ if len(versions) != 0 {
+ return hostbox.ErrChanged
+ }
+ return nil
+}
+
+// Publish is main's pre-replication call for a prepared admission
+// (ClaimWakeAssembly s.Replicate when in.Prepared). It finds, or writes,
+// P's marker under the fence Materialize recorded for the admitted session
+// and records the receipt; without that fence it refuses. Any other call
+// is refused: a fresh generation is published fenced (PublishFenced).
+func (r *RelocationWakeInputs) Publish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
+ if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil {
+ return err
+ }
+ h := o.Claim.Holder
+ if !in.Prepared || in.Admission.NewGeneration != o.Claim.Op.Journal["prepared"] || h.Generation != in.Admission.NewGeneration {
+ return hostbox.ErrChanged
+ }
+ r.mu.Lock()
+ work := r.fences[claimPublishedKey(o.Host, h.Session)]
+ r.mu.Unlock()
+ if work == nil {
+ return fmt.Errorf("%w: no admitted writer's fence for session %d", hostbox.ErrChanged, h.Session)
+ }
+ receipt, err := r.preparedMarker(ctx, o, writerFence(work, in.Identity))
+ if err != nil {
+ return err
+ }
+ r.record(o.Host, h.Session, receipt)
+ return nil
+}
+
+// PublishFenced is main's fresh-generation marker writer, for a later
+// attempt's generation and an empty P's opening (ruling 47).
+func (r *RelocationWakeInputs) PublishFenced(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, pub ClaimPublication) (ClaimPublicationReceipt, error) {
+ if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ if in.Prepared {
+ return ClaimPublicationReceipt{}, hostbox.ErrChanged
+ }
+ return r.freshOwners().publishFresh(ctx, o, in, pub)
+}
+
+// AdmissionEvidence binds a relocation probe admission to its observation
+// and to the certified baseline, read again: on P's opening the exact
+// prepared pin and P's files manifest by digest (prepared_authority); on a
+// later attempt the predecessor's fresh baseline.
+func (r *RelocationWakeInputs) AdmissionEvidence(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, baseline *genselect.Plan) error {
+ if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil {
+ return err
+ }
+ c, h := o.Claim, o.Claim.Holder
+ if in.ObservedOpID != c.Op.ID || in.ObservedPhase != hostclaim.PhaseRelocatingProbing || in.ObservedSession != h.Session || in.ObservedGeneration != h.Generation || in.Box != r.Box || in.Mode != "cold" || in.Coverage != "" || len(in.AdditionalWritesLost) != 0 || in.Region != r.Region {
+ return hostbox.ErrChanged
+ }
+ prepared := c.Op.Journal["prepared"]
+ if in.NewGeneration == prepared {
+ if !in.KeepStamp || hostclaim.PreparedOpened(c) || in.PredecessorCertified != hostrelocate.CertifiedFor(c) {
+ return hostbox.ErrChanged
+ }
+ auth, err := preparedAuthority(c)
+ if err != nil {
+ return err
+ }
+ raw, _, err := r.FilesStore.Get(ctx, hostfiles.ManifestKey(h.Location+"/files/"+prepared+"/"))
+ if err != nil {
+ return err
+ }
+ if sum := sha256.Sum256(raw); hex.EncodeToString(sum[:]) != auth.FilesManifestSHA256 {
+ return fmt.Errorf("%w: P's files manifest is not the prepared authority's", hostbox.ErrChanged)
+ }
+ if baseline == nil {
+ var ref hostclaim.RecoveryPin
+ if json.Unmarshal([]byte(c.Op.Journal["pin"]), &ref) != nil || !ref.Empty || in.RemoteMaxTXID != 0 || auth.InstallManifest != "" {
+ return hostbox.ErrChanged
+ }
+ return nil
+ }
+ if baseline.Generation != prepared || baseline.MaxTXID != in.RemoteMaxTXID || baseline.MaxTXID != auth.SnapshotTXID {
+ return hostbox.ErrChanged
+ }
+ return nil
+ }
+ if in.KeepStamp || in.PredecessorCertified || !claimFreshGeneration(h, in.NewGeneration) {
+ return hostbox.ErrChanged
+ }
+ if baseline == nil {
+ return nil // an empty predecessor, proved by EmptyAuthority
+ }
+ fresh, err := genselect.Select(ctx, r.Data, r.request(o))
+ if err != nil {
+ return err
+ }
+ if fresh.Generation != baseline.Generation || fresh.MaxTXID != baseline.MaxTXID || in.RemoteMaxTXID != fresh.MaxTXID || !reflect.DeepEqual(fresh.Files, baseline.Files) {
+ return hostbox.ErrChanged
+ }
+ return nil
+}
+
+// PublicationEvidence: on P's prepared opening, this process's receipt for
+// the admitted session with the marker still exactly that version; on any
+// fresh generation, main's freshEvidence.
+func (r *RelocationWakeInputs) PublicationEvidence(ctx context.Context, o hostclaim.Observed) error {
+ if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil {
+ return err
+ }
+ h := o.Claim.Holder
+ r.mu.Lock()
+ receipt, prepared := r.published[claimPublishedKey(o.Host, h.Session)]
+ r.mu.Unlock()
+ if !prepared {
+ return r.freshOwners().freshEvidence(ctx, o)
+ }
+ current, err := (&movesProbeOwners{Data: r.Data, Bucket: r.Bucket}).currentMarker(ctx, h)
+ if err != nil {
+ return err
+ }
+ if current == nil || current.ID != receipt.MarkerVersion || h.Generation != o.Claim.Op.Journal["prepared"] {
+ return hostbox.ErrChanged
+ }
+ return nil
+}
diff --git a/internal/host/relocationwake_test.go b/internal/host/relocationwake_test.go
new file mode 100644
index 000000000..a18504b56
--- /dev/null
+++ b/internal/host/relocationwake_test.go
@@ -0,0 +1,205 @@
+package host
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "os/exec"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostfiles"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// Finding 6, ruling 47: P's first probe publishes through the shared
+// assembly. The claim's own runtime (the recovery fixture's controller)
+// admits P's opening through the relocation dispatch, binds region HA's
+// attempt journal to the probe start, and Materialize, under the admitted
+// spawn selection, runs moves' C8 Establish and writes P's marker: P as
+// the marker's op, the install manifest's pin, prepared_authority's
+// digest. Publish (the pre-replication call) finds that marker and
+// records the receipt; PublicationEvidence then passes. Publish without
+// Materialize's fence refuses.
+func TestRelocationFirstProbePublishesThroughTheAssembly(t *testing.T) {
+ f := newRecoveryFixture(t)
+ ctx := t.Context()
+ bucket := leasetest.NewVersionedBucket()
+ bucket.AddBucket("data")
+ files := release.NewDirStore(t.TempDir())
+ o := f.read()
+ h := o.Claim.Holder
+ prefix := moves.GenerationPrefix(h.Location, roP)
+ snap, anchor := moves.SnapshotKey(prefix, 42), moves.AnchorKey(prefix, 42)
+ im := moves.InstallManifest{V: 1, Op: roOp, Host: recoveryHost, N: 42, SnapshotKey: snap, SnapshotVersion: bucket.Put("data", snap, ltxHeader(t, 1, 42)), SnapshotCRC64: 7,
+ AnchorKey: anchor, AnchorVersion: bucket.Put("data", anchor, ltxHeader(t, 42, 42)), IntegrityCheck: "ok", DBSHA256: strings.Repeat("d", 64)}
+ imRaw := mustJSON(t, im)
+ bucket.Put("data", moves.InstallManifestKey(prefix), imRaw)
+ pm := mustJSON(t, hostfiles.Manifest{V: hostfiles.ManifestVersion, Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}})
+ if _, err := files.Put(ctx, hostfiles.ManifestKey(h.Location+"/files/"+roP+"/"), pm, ""); err != nil {
+ t.Fatal(err)
+ }
+ imSum, pmSum := sha256.Sum256(imRaw), sha256.Sum256(pm)
+ auth := hostclaim.PreparedAuthority{Generation: roP, InstallManifest: moves.InstallManifestKey(prefix), SHA256: hex.EncodeToString(imSum[:]), SnapshotTXID: 42, SourceFilesSeq: 5, FilesManifestSHA256: hex.EncodeToString(pmSum[:])}
+ ref := hostclaim.RecoveryPin{Key: "control/relocations/" + recoveryHost + "/" + roOp + "/pin-1.json", SHA256: strings.Repeat("e", 64), Generation: h.Generation, MaxTXID: 42, FilesSeq: 5, Release: "v1"}
+ o, err := hostclaim.Update(ctx, f.claims, o, func(c *hostclaim.Claim) error {
+ c.Holder.Stamp = c.Gen + 1
+ auth.InstallStamp = c.Holder.Stamp
+ c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: roOp, Phase: hostclaim.PhaseRelocatingProbing, Exec: "box-1", Participants: []string{"box-0", "box-1"},
+ Intent: map[string]string{"region": "ie", "source_box": "box-0", "target_box": "box-1", "target_enrollment": "enrollment-1"},
+ Journal: map[string]string{"pin": string(mustJSON(t, ref)), "prepared": roP, "install_stamp": strconv.FormatUint(c.Holder.Stamp, 10), "install_release": "v1",
+ "prepared_authority": string(mustJSON(t, auth)), "probing_since": f.now.Format(time.RFC3339Nano), "exec_ack.box-0": `{"op":"rel-1"}`, "exec_ack.box-1": `{"op":"rel-1"}`,
+ "local_install": string(mustJSON(t, hostclaim.LocalInstall{Box: "box-1", Enrollment: "enrollment-1", InstallStamp: c.Holder.Stamp, JournalSHA256: strings.Repeat("c", 64)}))}}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if o, err = hostclaim.Update(ctx, f.claims, o, hostclaim.BeginProbeAttempt(roOp, "box-1")); err != nil {
+ t.Fatal(err)
+ }
+ c8 := &roFiles{}
+ inputs := &RelocationWakeInputs{
+ Launch: func(context.Context, registry.Route) (ClaimServingLaunch, error) {
+ return ClaimServingLaunch{Identity: hostwrite.Identity{Config: "sha256:config", Release: "sha256:release"}, Executable: "/usr/bin/true", ExecutableSHA256: strings.Repeat("a", 64), Version: "v1"}, nil
+ },
+ FilesManifest: c8, Markers: plan9D5Markers{bucket}, Claims: f.claims, FilesStore: files, Data: bucket, Bucket: "data", Region: "ie", Box: "box-1",
+ NewID: f.newID, Now: func() time.Time { return f.now }, Releases: &roReleases{}}
+ attempts := &relocationProbeAttempts{Claims: f.claims, DataDir: t.TempDir(), Box: "box-1", NewID: f.newID, Now: func() time.Time { return f.now }}
+ c := f.controller("boot-1", "controller-1", 4101, "start-1")
+ c.wake.Inputs = ClaimOwnerDispatch{Relocation: &ClaimKindOwners{Inputs: inputs, AdmissionEvidence: inputs.AdmissionEvidence, PublicationEvidence: inputs.PublicationEvidence}}
+ c.wake.Continue = hostclaim.SessionContinues
+ // As claimGroupDWiring routes them once the relocation owners bind.
+ ready := func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }
+ c.wake.Operations = claimOperationRouter{hostclaim.KindRelocating: {Attempts: attempts, ProbeReady: ready}}
+ live, cancel := context.WithCancel(context.Background())
+ t.Cleanup(cancel)
+ done := make(chan struct{})
+ go func() { <-live.Done(); close(done) }()
+ c.wake.mu.Lock()
+ c.wake.pending[recoveryHost] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done}
+ c.wake.mu.Unlock()
+ route := registry.Route{Host: recoveryHost, Kind: "instance", App: "jam", Account: "bab", Status: "relocating", HoldOp: roOp, Socket: filepath.Join(f.guard.DataDir, "app.sock"), DBPath: filepath.Join(f.guard.DataDir, "state", "data.db")}
+
+ admitted, err := c.runtime.Admit(ctx, route, o, roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ah := admitted.Claim.Holder
+ if ah.Generation != roP || ah.Stamp != o.Claim.Holder.Stamp || ah.WriterBox != "box-1" || !hostclaim.PreparedOpened(admitted.Claim) {
+ t.Fatalf("P's opening: %+v", ah)
+ }
+ s, err := c.runtime.Prepare(ctx, route, admitted, roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if s.OpenProbe == nil {
+ t.Fatal("the probe start carries no attempt journal")
+ }
+ if _, err := s.OpenProbe(ctx, f.now); err != nil {
+ t.Fatal(err)
+ }
+ if err := s.Reserve(ctx, 0); err != nil {
+ t.Fatal(err)
+ }
+ if a := s.CurrentAttempt(); a.ID == "" || a.Session != ah.Session || a.ControllerID != "controller-1" {
+ t.Fatalf("the reservation: %+v", a)
+ }
+ if err := s.Materialize(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if len(c8.established) != 1 {
+ t.Fatalf("C8 established %v", c8.established)
+ }
+ markerKey := strings.TrimSuffix(h.Location, "/") + "/generation.json"
+ body, ok := bucket.Current("data", markerKey)
+ var m genselect.Marker
+ if !ok || json.Unmarshal(body, &m) != nil || m.Op != roP || m.Box != "box-1" || m.Host != recoveryHost || m.ManifestSHA256 != auth.SHA256 || m.Pin != manifestPin(im) {
+ t.Fatalf("P's marker: %s", body)
+ }
+ written := bucket.LatestID("data", markerKey)
+ // The pre-replication publish (ClaimWakeAssembly s.Replicate's first
+ // call for a prepared admission) finds the marker and records it.
+ in, err := inputs.Select(ctx, route, o, roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := inputs.Publish(ctx, admitted, in); err != nil {
+ t.Fatal(err)
+ }
+ if bucket.LatestID("data", markerKey) != written {
+ t.Fatal("Publish rewrote P's marker")
+ }
+ if err := inputs.PublicationEvidence(ctx, admitted); err != nil {
+ t.Fatalf("publication evidence: %v", err)
+ }
+ // A fresh wake inputs (another process) has no fence for the session.
+ other := &RelocationWakeInputs{Launch: inputs.Launch, FilesManifest: c8, Markers: inputs.Markers, Claims: f.claims, FilesStore: files, Data: bucket, Bucket: "data", Region: "ie", Box: "box-1",
+ NewID: f.newID, Now: inputs.Now, Releases: inputs.Releases}
+ if err := other.Publish(ctx, admitted, in); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("Publish without Materialize's fence: %v", err)
+ }
+}
+
+// Ruling 47: a later attempt's fresh generation, and an empty P's opening,
+// publish through main's fenced writer only after their first root.
+func TestRelocationLaterAttemptsPublishFenced(t *testing.T) {
+ ctx := t.Context()
+ f := newROFixture(t, roOptions{})
+ first, err := f.inputs.Select(ctx, f.route, f.read(), roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ opened := f.admit(f.read(), first)
+ again, err := f.inputs.Select(ctx, f.route, opened, roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ later := f.admit(opened, again)
+ gen := later.Claim.Holder.Generation
+ synced := 0
+ pub := ClaimPublication{Sync: func(context.Context) error { synced++; return nil }, Check: func(context.Context) error { return nil }}
+ if err := f.inputs.Publish(ctx, later, again); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("an unfenced marker for a fresh generation: %v", err)
+ }
+ if _, err := f.inputs.PublishFenced(ctx, later, again, pub); err == nil || synced != 1 {
+ t.Fatalf("a marker before the root: %v synced=%d", err, synced)
+ }
+ root := moves.SnapshotKey(moves.GenerationPrefix(later.Claim.Holder.Location, gen), 44)
+ f.bucket.Put("data", root, ltxHeader(t, 1, 44))
+ receipt, err := f.inputs.PublishFenced(ctx, later, again, pub)
+ if err != nil || receipt.Root.Key != root || receipt.MarkerVersion == "" {
+ t.Fatalf("publish: %+v %v", receipt, err)
+ }
+ if err := f.inputs.PublicationEvidence(ctx, later); err != nil {
+ t.Fatalf("fresh evidence: %v", err)
+ }
+ prepared := again
+ prepared.Prepared = true
+ if _, err := f.inputs.PublishFenced(ctx, later, prepared, pub); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("a prepared admission through the fenced writer: %v", err)
+ }
+
+ // An empty P: no snapshot, no root until the first transaction.
+ e := newROFixture(t, roOptions{empty: true})
+ ein, err := e.inputs.Select(ctx, e.route, e.read(), roProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ eopened := e.admit(e.read(), ein)
+ if _, err := e.inputs.PublishFenced(ctx, eopened, ein, pub); err == nil {
+ t.Fatal("an empty P published a marker with no root")
+ }
+}