amadan

carlos / platform Public

hostrelocate: a regional worker starts a relocation and replays it one claim CAS at a time, reading fences only

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude 1553b7bf3088851bcee023f5285d0b605c75454a parent e13f694
5 files changed, +2260 −0
  • internal/hostfiles/manifestcheck.go +23 −0
  • internal/hostrelocate/pin.go +372 −0
  • internal/hostrelocate/pin_test.go +538 −0
  • internal/hostrelocate/worker.go +561 −0
  • internal/hostrelocate/worker_test.go +766 −0
diff --git a/internal/hostfiles/manifestcheck.go b/internal/hostfiles/manifestcheck.go
new file mode 100644
index 000000000..fbdd29d79
--- /dev/null
+++ b/internal/hostfiles/manifestcheck.go
@@ -0,0 +1,23 @@
+package hostfiles
+
+import "fmt"
+
+// ValidateManifest checks every entry and platform item of m against the
+// admitted origins, as Restore and CopyManifest do before they touch
+// anything (contracts note §1). A caller holding a manifest's exact bytes
+// from somewhere other than its prefix (a relocation's retained copy)
+// validates them with it before writing them under a new generation.
+func ValidateManifest(m Manifest, origins []string) error {
+ if m.V != ManifestVersion {
+ return fmt.Errorf("hostfiles: manifest is version %d; this binary reads %d", m.V, ManifestVersion)
+ }
+ if len(origins) == 0 {
+ return ErrOrigin
+ }
+ for _, o := range origins {
+ if err := checkPrefix(o); err != nil {
+ return err
+ }
+ }
+ return checkManifest(m, origins)
+}
diff --git a/internal/hostrelocate/pin.go b/internal/hostrelocate/pin.go
new file mode 100644
index 000000000..0bbd7492d
--- /dev/null
+++ b/internal/hostrelocate/pin.go
@@ -0,0 +1,372 @@
+package hostrelocate
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostfiles"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// PinObject is the immutable pinned recovery set (§5.8): the database
+// plan genselect selected, with every file's exact version, and the files
+// pin, the newest manifest in the admitted chain, by digest (spec §18: the
+// store has no version-id read, so content is identified by SHA-256).
+// ClaimGen is the claim generation the set was selected against; the
+// object's key names it (hostclaim.RecoveryPinKey), so a pin whose journal
+// CAS lost is never reused. DB is nil for an evidenced empty bootstrap.
+type PinObject struct {
+ V int `json:"v"`
+ Host string `json:"host"`
+ Op string `json:"op"`
+ ClaimGen uint64 `json:"claim_gen"`
+ DB *genselect.Plan `json:"db,omitempty"`
+ Empty string `json:"empty,omitempty"` // "release" | "session-0": the evidence
+ Files FilesPin `json:"files"`
+ PinnedAt time.Time `json:"pinned_at"`
+}
+
+// FilesPin names one generation's files manifest by prefix, sequence and
+// the digest of its exact bytes, and the origins its entries may name:
+// every generation prefix in the admitted chain as observed at the pin,
+// the manifest's own included (contracts note §1).
+type FilesPin struct {
+ Prefix string `json:"prefix"`
+ Seq int64 `json:"seq"`
+ SHA256 string `json:"sha256"`
+ Origins []string `json:"origins"`
+ // Retained is the pinner's immutable copy of the exact manifest bytes
+ // (RetainedManifestKey), written create-only before the pin object. A
+ // late revision of the source manifest changes nothing the build
+ // reads: P is written from this copy, never from the source again.
+ // Empty for a host that never published files.
+ Retained string `json:"retained,omitempty"`
+}
+
+// RetainedManifestKey is where the pinner keeps the pinned manifest's
+// bytes: control/relocations/<host>/<op>/manifest-<sha256>.json.
+func RetainedManifestKey(host, op, sha string) string {
+ return "control/relocations/" + host + "/" + op + "/manifest-" + sha + ".json"
+}
+
+// PinKey is hostclaim.RecoveryPinKey.
+func PinKey(host, op string, gen uint64) string { return hostclaim.RecoveryPinKey(host, op, gen) }
+
+// StorePinner selects the pinned recovery set against the claim
+// observation it is given (§5.8 "The pinned recovery set") and writes it
+// create-only at the key for that observation's claim generation. It
+// checks every bound JournalPin will check before it writes anything, so
+// a doomed pin is never published; a retry at the same generation reuses
+// its own object unchanged. The caller holds the location lease (claim §4:
+// "New pins must be established under the required leases").
+type StorePinner struct {
+ Claims release.Store // control bucket: pin objects
+ Data release.Store // the region's data bucket: files manifests
+ SelectDB func(context.Context, genselect.Request) (*genselect.Plan, error)
+ Bucket string // the region's data bucket name, for genselect
+ Region string
+}
+
+func (p StorePinner) Pin(ctx context.Context, obs hostclaim.Observed, now time.Time) (hostclaim.RecoveryPin, error) {
+ if p.Claims == nil || p.Data == nil || p.SelectDB == nil || p.Bucket == "" || p.Region == "" || obs.Claim == nil || obs.Claim.Holder == nil {
+ return hostclaim.RecoveryPin{}, ErrUnsupported
+ }
+ c := obs.Claim
+ key := PinKey(c.Host, c.Op.ID, c.Gen)
+ if raw, _, err := p.Claims.Get(ctx, key); err == nil {
+ return p.reuse(c, key, raw)
+ } else if !errors.Is(err, os.ErrNotExist) {
+ return hostclaim.RecoveryPin{}, err
+ }
+ var rel hostclaim.Release
+ released := json.Unmarshal([]byte(c.Op.Journal["release"]), &rel) == nil && rel.ReleasedBy != ""
+ for k, at := range c.Op.Journal {
+ if len(k) > len(".observed_at") && k[len(k)-len(".observed_at"):] == ".observed_at" {
+ t, err := time.Parse(time.RFC3339Nano, at)
+ if err != nil {
+ return hostclaim.RecoveryPin{}, err
+ }
+ if now.Before(t.Add(hostclaim.FailoverPinDelay)) {
+ return hostclaim.RecoveryPin{}, ErrNotYet
+ }
+ }
+ }
+ obj := PinObject{V: 1, Host: c.Host, Op: c.Op.ID, ClaimGen: c.Gen, PinnedAt: now.UTC()}
+ plan, err := p.SelectDB(ctx, genselect.Request{Bucket: p.Bucket, Region: p.Region, Host: c.Host, Holder: *c.Holder})
+ switch {
+ case err == nil:
+ obj.DB = plan
+ case errors.Is(err, genselect.ErrEmpty) && released && rel.FinalTXID == 0:
+ obj.Empty = "release" // the stopped writer flushed and reports position 0
+ case errors.Is(err, genselect.ErrEmpty) && !released && c.Holder.Session == 0:
+ obj.Empty = "session-0" // no writer was ever admitted, so nothing was written
+ default:
+ // A listing that finds nothing after a writer ran is not evidence
+ // of an empty database (§5.8, claim §6.4a): refuse, never bootstrap.
+ return hostclaim.RecoveryPin{}, err
+ }
+ if released && obj.DB != nil && obj.DB.MaxTXID < rel.FinalTXID {
+ return hostclaim.RecoveryPin{}, fmt.Errorf("%w: the listing is at %d, the release at %d", ErrNotYet, obj.DB.MaxTXID, rel.FinalTXID)
+ }
+ // A nonempty database written by an admitted writer is sealed with
+ // that writer's instance.key: the pinned manifest must carry it (rev
+ // 18, round 17 finding 1).
+ needKey := c.Holder.Session != 0 && obj.DB != nil && obj.DB.MaxTXID > 0
+ files, retained, err := p.filesPin(ctx, c.Holder, released && rel.ManifestSeq == 0, needKey)
+ if err != nil {
+ return hostclaim.RecoveryPin{}, err
+ }
+ obj.Files = files
+ if released && obj.Files.Seq < rel.ManifestSeq {
+ return hostclaim.RecoveryPin{}, fmt.Errorf("%w: the files manifest is at %d, the release at %d", ErrNotYet, obj.Files.Seq, rel.ManifestSeq)
+ }
+ if retained != nil {
+ // The exact bytes, kept where nothing rewrites them, before the
+ // pin that names them exists.
+ obj.Files.Retained = RetainedManifestKey(c.Host, c.Op.ID, files.SHA256)
+ if err := release.CreateExact(ctx, p.Claims, obj.Files.Retained, retained); err != nil && !errors.Is(err, release.ErrCASConflict) {
+ return hostclaim.RecoveryPin{}, err
+ }
+ }
+ raw, err := json.Marshal(obj)
+ if err != nil {
+ return hostclaim.RecoveryPin{}, err
+ }
+ if err := release.CreateExact(ctx, p.Claims, key, raw); err != nil {
+ if !errors.Is(err, release.ErrCASConflict) {
+ return hostclaim.RecoveryPin{}, err
+ }
+ got, _, gerr := p.Claims.Get(ctx, key) // another worker pinned this generation first
+ if gerr != nil {
+ return hostclaim.RecoveryPin{}, gerr
+ }
+ return p.reuse(c, key, got)
+ }
+ return pinRef(key, raw, obj), nil
+}
+
+func (p StorePinner) reuse(c *hostclaim.Claim, key string, raw []byte) (hostclaim.RecoveryPin, error) {
+ var obj PinObject
+ if err := json.Unmarshal(raw, &obj); err != nil || obj.V != 1 || obj.Host != c.Host || obj.Op != c.Op.ID || obj.ClaimGen != c.Gen || (obj.DB == nil) == (obj.Empty == "") {
+ return hostclaim.RecoveryPin{}, fmt.Errorf("%w: pin object %s is not this operation's at this generation", ErrRefused, key)
+ }
+ return pinRef(key, raw, obj), nil
+}
+
+func pinRef(key string, raw []byte, obj PinObject) hostclaim.RecoveryPin {
+ sum := sha256.Sum256(raw)
+ ref := hostclaim.RecoveryPin{Key: key, SHA256: hex.EncodeToString(sum[:]), FilesManifest: hostfiles.ManifestKey(obj.Files.Prefix), FilesSeq: obj.Files.Seq, PinnedAt: obj.PinnedAt, Empty: obj.DB == nil}
+ if obj.DB != nil {
+ ref.Generation, ref.MaxTXID = obj.DB.Generation, obj.DB.MaxTXID
+ }
+ return ref
+}
+
+// InstanceKeyPath is H6's always-continuous instance.key, at the root of
+// the host's files directory.
+const InstanceKeyPath = "instance.key"
+
+func liveEntry(m hostfiles.Manifest, path string) bool {
+ e, ok := m.Entries[path]
+ return ok && !e.Deleted && e.Skipped == "" && e.Object != ""
+}
+
+// ErrFilesUnevidenced is a pin refused because its files set cannot be
+// shown to hold the host's platform files: a writer was admitted and no
+// files manifest exists in the admitted chain (rev 17), or the database
+// is nonempty and the pinned manifest has no instance.key (rev 18). The
+// worker marks the operation stuck.
+var ErrFilesUnevidenced = errors.New("hostrelocate: the files evidence does not cover the host's platform files")
+
+// filesPin is §5.6's files pin: the manifest of holder.generation as
+// observed or, if that generation has none, of the nearest predecessor in
+// holder.admissions that has one. It is chosen independently of the
+// database baseline (B2-4). A read error refuses; no manifest anywhere in
+// the chain is a host that never published files, pinned at sequence 0.
+// Origins is the retained chain plus every generation of this incarnation
+// whose objects the pinned manifest references (rev 12, round 11): the
+// claim's §6.2a files live set is the objects the live manifests
+// reference, independent of holder.admissions, which AdmitWriter trims to
+// eight. An unchanged file published at g1 and carried by every later copy
+// stays restorable at g10. Every referenced object must lie under this
+// holder's location (its incarnation) in this region's data bucket, at
+// files/<generation>/objects/; any other reference refuses the pin.
+func (p StorePinner) filesPin(ctx context.Context, h *hostclaim.Holder, releasedEmpty, needKey bool) (FilesPin, []byte, error) {
+ chain := []string{h.Generation}
+ for i := len(h.Admissions) - 1; i >= 0; i-- {
+ chain = append(chain, h.Admissions[i].Generation, h.Admissions[i].Predecessor)
+ }
+ seen := map[string]bool{}
+ var origins []string
+ for _, gen := range chain {
+ if gen == "" || gen == hostclaim.RootGeneration || seen[gen] {
+ continue
+ }
+ seen[gen] = true
+ origins = append(origins, h.Location+"/files/"+gen+"/")
+ }
+ for _, prefix := range origins {
+ raw, _, err := p.Data.Get(ctx, hostfiles.ManifestKey(prefix))
+ if errors.Is(err, os.ErrNotExist) {
+ continue
+ }
+ if err != nil {
+ return FilesPin{}, nil, err
+ }
+ var m hostfiles.Manifest // decoded from the bytes the digest names
+ if err := json.Unmarshal(raw, &m); err != nil || m.V != hostfiles.ManifestVersion {
+ return FilesPin{}, nil, errors.Join(fmt.Errorf("%w: files manifest %s", ErrRefused, prefix), err)
+ }
+ if needKey && !liveEntry(m, InstanceKeyPath) {
+ // The manifest was copied before the app started and the
+ // key it minted afterwards never published (§5.6's
+ // continuous-sync window, L21): installing this set would mint
+ // a new key over data sealed with the old one.
+ return FilesPin{}, nil, fmt.Errorf("%w: %s's pinned manifest has no %s for a nonempty database", ErrFilesUnevidenced, prefix, InstanceKeyPath)
+ }
+ if origins, err = referencedOrigins(h.Location, origins, m); err != nil {
+ return FilesPin{}, nil, err
+ }
+ sum := sha256.Sum256(raw)
+ return FilesPin{Prefix: prefix, Seq: m.Seq, SHA256: hex.EncodeToString(sum[:]), Origins: origins}, raw, nil
+ }
+ // No manifest anywhere in the chain. An empty files pin needs the
+ // same kind of evidence an empty database pin does (ruling 15): no
+ // writer was ever admitted on this holder (session 0), so no app ever
+ // ran to mint instance.key, report.token or an env (H6); or the
+ // stopped writer's release flush, which publishes every file whatever
+ // the setting, reported nothing (manifest_seq 0). Any other admitted
+ // writer without a manifest (an adopted host whose files never
+ // published, a first manifest that never landed, a trimmed chain)
+ // refuses: the worker marks the operation stuck for an operator (rev
+ // 17, finding F2), never an install that mints a fresh key over data
+ // sealed with the old one.
+ if h.Session != 0 && !releasedEmpty {
+ return FilesPin{}, nil, fmt.Errorf("%w: %s has admitted a writer but no files manifest exists in its chain", ErrFilesUnevidenced, h.Location)
+ }
+ gen := h.Generation
+ if gen == "" {
+ gen = hostclaim.RootGeneration
+ }
+ prefix := h.Location + "/files/" + gen + "/"
+ return FilesPin{Prefix: prefix, Origins: []string{prefix}}, nil, nil
+}
+
+// referencedOrigins adds to origins the files prefix of every generation
+// whose objects m references, each of which must be an object this
+// incarnation's location published: <location>/files/<generation>/objects/.
+func referencedOrigins(location string, origins []string, m hostfiles.Manifest) ([]string, error) {
+ base := location + "/files/"
+ have := make(map[string]bool, len(origins))
+ for _, o := range origins {
+ have[o] = true
+ }
+ var extra []string
+ check := func(name string, e hostfiles.Entry) error {
+ if e.Object == "" {
+ return nil // a tombstone or a skipped entry names no object
+ }
+ gen, rest, ok := strings.Cut(strings.TrimPrefix(e.Object, base), "/")
+ if !strings.HasPrefix(e.Object, base) || !ok || gen == "" || gen == "." || gen == ".." || !strings.HasPrefix(rest, "objects/") {
+ return fmt.Errorf("%w: %s references %s, outside %s: %w", ErrRefused, name, e.Object, base, hostfiles.ErrOrigin)
+ }
+ if o := base + gen + "/"; !have[o] {
+ have[o] = true
+ extra = append(extra, o)
+ }
+ return nil
+ }
+ for name, e := range m.Entries {
+ if err := check(name, e); err != nil {
+ return nil, err
+ }
+ }
+ for name, e := range m.Platform {
+ if err := check(name, e); err != nil {
+ return nil, err
+ }
+ }
+ sort.Strings(extra)
+ return append(origins, extra...), nil
+}
+
+// ReadPin reads and verifies the pin object a journaled RecoveryPin names:
+// the exact bytes hash to the journaled digest (ruling 5). Every reader of
+// a pin goes through it.
+func ReadPin(ctx context.Context, claims release.Store, ref hostclaim.RecoveryPin) (PinObject, error) {
+ raw, _, err := claims.Get(ctx, ref.Key)
+ if err != nil {
+ return PinObject{}, err
+ }
+ if sum := sha256.Sum256(raw); hex.EncodeToString(sum[:]) != ref.SHA256 {
+ return PinObject{}, fmt.Errorf("%w: pin object %s does not hash to its journaled digest", ErrRefused, ref.Key)
+ }
+ var obj PinObject
+ if err := json.Unmarshal(raw, &obj); err != nil || obj.V != 1 {
+ return PinObject{}, fmt.Errorf("%w: pin object %s", ErrRefused, ref.Key)
+ }
+ return obj, nil
+}
+
+// CopyPinnedManifest writes the pinned files manifest into toPrefix
+// (§5.8 "write P's files manifest, create-only, as the pinned manifest's
+// entries"), from the pinner's retained copy of its exact bytes (claims:
+// the control bucket), never from the source prefix again: a late
+// revision of the source manifest cannot move the pin (B2-4). The bytes
+// must hash to the pin's digest and sit at the pinned sequence; every
+// entry is validated against the pinned origins (L13); the write is
+// create-only, and a retry finding the same bytes at toPrefix succeeds.
+// A host that never published files gets an empty manifest.
+func CopyPinnedManifest(ctx context.Context, claims, data release.Store, pin FilesPin, toPrefix string) (hostfiles.Manifest, error) {
+ if pin.Prefix == "" || toPrefix == "" || len(pin.Origins) == 0 || (pin.SHA256 == "") != (pin.Retained == "") {
+ return hostfiles.Manifest{}, ErrUnsupported
+ }
+ var pinned hostfiles.Manifest
+ if pin.Retained == "" {
+ // A host that never published files: P starts an empty manifest.
+ pinned = hostfiles.Manifest{V: hostfiles.ManifestVersion, Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}}
+ } else {
+ raw, _, err := claims.Get(ctx, pin.Retained)
+ if err != nil {
+ return hostfiles.Manifest{}, err
+ }
+ if sum := sha256.Sum256(raw); hex.EncodeToString(sum[:]) != pin.SHA256 {
+ return hostfiles.Manifest{}, fmt.Errorf("%w: retained manifest %s is not the pinned bytes", ErrRefused, pin.Retained)
+ }
+ if err := json.Unmarshal(raw, &pinned); err != nil || pinned.Seq != pin.Seq {
+ return hostfiles.Manifest{}, fmt.Errorf("%w: retained manifest %s", ErrRefused, pin.Retained)
+ }
+ if err := hostfiles.ValidateManifest(pinned, pin.Origins); err != nil {
+ return hostfiles.Manifest{}, err
+ }
+ }
+ // P's prepared manifest (claim rev 6.30 R18-1, moves rev 7.22 §7.6):
+ // the pinned entries, by their exact content-addressed objects, at
+ // sequence 0. The source's sequence is provenance (the pin keeps it),
+ // never P's: the session publisher starts at 1 after the first
+ // admission. The binding {location, P, op, install stamp} is the
+ // prefix (location, P) and the claim's prepared_authority (op, stamp,
+ // this manifest's digest).
+ pinned.Seq = 0
+ body, err := json.Marshal(pinned)
+ if err != nil {
+ return hostfiles.Manifest{}, err
+ }
+ if err := release.CreateExact(ctx, data, hostfiles.ManifestKey(toPrefix), body); err != nil {
+ if errors.Is(err, release.ErrCASConflict) {
+ return hostfiles.Manifest{}, hostfiles.ErrManifestExists
+ }
+ return hostfiles.Manifest{}, err
+ }
+ return pinned, nil
+}
diff --git a/internal/hostrelocate/pin_test.go b/internal/hostrelocate/pin_test.go
new file mode 100644
index 000000000..ad3afdcf2
--- /dev/null
+++ b/internal/hostrelocate/pin_test.go
@@ -0,0 +1,538 @@
+package hostrelocate
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostfiles"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const pinLocation = "instances/" + host + "/inc"
+
+func pinClaim() hostclaim.Observed {
+ return hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Host: host, Gen: 12, Op: hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "rel-1", Phase: hostclaim.PhasePinning, Journal: map[string]string{}},
+ Holder: &hostclaim.Holder{Location: pinLocation, Generation: "g3", Session: 3,
+ Admissions: []hostclaim.AdmissionRecord{{Generation: "g2", Predecessor: "g1"}, {Generation: "g3", Predecessor: "g2"}}}}}
+}
+
+// keyed adds the host's instance.key (H6), as every manifest of a host
+// whose database is nonempty must carry (rev 18).
+func keyed(m hostfiles.Manifest) hostfiles.Manifest {
+ sha := strings.Repeat("b", 64)
+ m.Entries["instance.key"] = hostfiles.Entry{Object: pinLocation + "/files/g1/objects/0000000000000001-" + sha, SHA256: sha, Size: 32, Mode: 0o600, Seq: 1}
+ return m
+}
+
+func putManifest(t *testing.T, data release.Store, gen string, m hostfiles.Manifest) (string, []byte) {
+ t.Helper()
+ raw, _ := json.Marshal(m)
+ key := hostfiles.ManifestKey(pinLocation + "/files/" + gen + "/")
+ if _, err := data.Put(context.Background(), key, raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ return key, raw
+}
+
+func selectAt(txid uint64) func(context.Context, genselect.Request) (*genselect.Plan, error) {
+ return func(_ context.Context, req genselect.Request) (*genselect.Plan, error) {
+ return &genselect.Plan{Generation: req.Holder.Generation, MaxTXID: txid}, nil
+ }
+}
+
+func TestStorePinnerPinsTheNewestFilesManifestInTheChain(t *testing.T) {
+ ctx := context.Background()
+ claims, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ // g3 (holder.generation) never published files; g2 did, then g1 earlier.
+ for gen, seq := range map[string]int64{"g2": 5, "g1": 9} {
+ putManifest(t, data, gen, keyed(hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: seq, Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}}))
+ }
+ selects := 0
+ p := StorePinner{Claims: claims, Data: data, Bucket: "data", Region: "ie",
+ SelectDB: func(ctx context.Context, req genselect.Request) (*genselect.Plan, error) {
+ selects++
+ return selectAt(41)(ctx, req)
+ }}
+ at := time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC)
+ pin, err := p.Pin(ctx, pinClaim(), at)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if pin.FilesManifest != hostfiles.ManifestKey(pinLocation+"/files/g2/") || pin.FilesSeq != 5 || pin.MaxTXID != 41 || pin.Key != PinKey(host, "rel-1", 12) || pin.Empty {
+ t.Fatalf("pin: %+v", pin)
+ }
+ obj, err := ReadPin(ctx, claims, pin)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if want := []string{pinLocation + "/files/g3/", pinLocation + "/files/g2/", pinLocation + "/files/g1/"}; !equalStrings(obj.Files.Origins, want) || obj.ClaimGen != 12 {
+ t.Fatalf("origins %v gen %d", obj.Files.Origins, obj.ClaimGen)
+ }
+ // The exact bytes are retained in the control bucket before the pin.
+ kept, _, kerr := claims.Get(ctx, obj.Files.Retained)
+ if kerr != nil || obj.Files.Retained != RetainedManifestKey(host, "rel-1", obj.Files.SHA256) {
+ t.Fatalf("retained copy: %s %v", obj.Files.Retained, kerr)
+ }
+ if sum := sha256.Sum256(kept); hex.EncodeToString(sum[:]) != obj.Files.SHA256 {
+ t.Fatal("the retained copy is not the pinned bytes")
+ }
+ // A retry at the same observation reuses the object unchanged.
+ again, err := p.Pin(ctx, pinClaim(), at.Add(time.Minute))
+ if err != nil || again != pin || selects != 1 {
+ t.Fatalf("replay: %+v %v selects=%d", again, err, selects)
+ }
+ // The claim moved (the journal CAS lost): the next pin is a fresh
+ // selection at the new generation; the old object is never reused.
+ moved := pinClaim()
+ moved.Claim.Gen = 13
+ next, err := p.Pin(ctx, moved, at.Add(2*time.Minute))
+ if err != nil || next.Key != PinKey(host, "rel-1", 13) || selects != 2 || next.PinnedAt.Equal(pin.PinnedAt) {
+ t.Fatalf("after the claim moved: %+v %v selects=%d", next, err, selects)
+ }
+ // An object at the key that is not this operation's is refused.
+ if _, err := claims.Put(ctx, PinKey(host, "rel-2", 12), []byte(`{"v":1,"host":"other","op":"rel-2","claim_gen":12,"db":{}}`), ""); err != nil {
+ t.Fatal(err)
+ }
+ other := pinClaim()
+ other.Claim.Op.ID = "rel-2"
+ if _, err := p.Pin(ctx, other, at); !errors.Is(err, ErrRefused) {
+ t.Fatalf("foreign pin object: %v", err)
+ }
+ // A rewritten object is caught by the journaled digest.
+ tampered := pin
+ tampered.SHA256 = strings.Repeat("0", 64)
+ if _, err := ReadPin(ctx, claims, tampered); !errors.Is(err, ErrRefused) {
+ t.Fatalf("a pin that does not hash to its digest was read: %v", err)
+ }
+}
+
+func equalStrings(a, b []string) bool {
+ if len(a) != len(b) {
+ return false
+ }
+ for i := range a {
+ if a[i] != b[i] {
+ return false
+ }
+ }
+ return true
+}
+
+func TestStorePinnerChecksTheBoundsBeforeItWrites(t *testing.T) {
+ ctx := context.Background()
+ at := time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC)
+ release := func(txid uint64, seq int64) string {
+ raw, _ := json.Marshal(hostclaim.Release{ReleasedBy: "src", Enrollment: "e", Generation: "g3", Session: 3, FinalTXID: txid, ManifestSeq: seq, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)})
+ return string(raw)
+ }
+ for _, tc := range []struct {
+ name string
+ journal map[string]string
+ txid uint64
+ seq int64
+ now time.Time
+ }{
+ {"the listing is short of the release", map[string]string{"release": release(45, 5)}, 41, 5, at},
+ {"the files manifest is short of the release", map[string]string{"release": release(41, 6)}, 41, 5, at},
+ {"inside the failover window", map[string]string{"fence.src.observed_at": at.Format(time.RFC3339Nano)}, 41, 5, at.Add(29 * time.Second)},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ claims, data := release_NewDirStore(t), release_NewDirStore(t)
+ putManifest(t, data, "g3", keyed(hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: tc.seq, Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}}))
+ o := pinClaim()
+ o.Claim.Op.Journal = tc.journal
+ p := StorePinner{Claims: claims, Data: data, Bucket: "data", Region: "ie", SelectDB: selectAt(tc.txid)}
+ if _, err := p.Pin(ctx, o, tc.now); !errors.Is(err, ErrNotYet) {
+ t.Fatalf("got %v, want ErrNotYet", err)
+ }
+ if keys, _ := claims.List(ctx, "control/relocations/"); len(keys) != 0 {
+ t.Fatalf("a pin that cannot be journaled was written: %v", keys)
+ }
+ })
+ }
+}
+
+func release_NewDirStore(t *testing.T) release.Store { return release.NewDirStore(t.TempDir()) }
+
+func TestStorePinnerPinsEmptyOnlyWithEvidence(t *testing.T) {
+ ctx := context.Background()
+ at := time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC)
+ empty := func(context.Context, genselect.Request) (*genselect.Plan, error) { return nil, genselect.ErrEmpty }
+ released := func(txid uint64) string {
+ raw, _ := json.Marshal(hostclaim.Release{ReleasedBy: "src", Enrollment: "e", Generation: "g3", Session: 3, FinalTXID: txid, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)})
+ return string(raw)
+ }
+ for _, tc := range []struct {
+ name string
+ session uint64
+ journal map[string]string
+ want string
+ }{
+ {"a release at txid 0", 3, map[string]string{"release": released(0)}, "release"},
+ {"a never-woken host", 0, map[string]string{}, "session-0"},
+ {"a release at txid 40 and an empty listing", 3, map[string]string{"release": released(40)}, ""},
+ {"a failover of a host that had a writer", 3, map[string]string{}, ""},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ claims, data := release_NewDirStore(t), release_NewDirStore(t)
+ o := pinClaim()
+ o.Claim.Holder.Session, o.Claim.Op.Journal = tc.session, tc.journal
+ p := StorePinner{Claims: claims, Data: data, Bucket: "data", Region: "ie", SelectDB: empty}
+ pin, err := p.Pin(ctx, o, at)
+ if tc.want == "" {
+ if !errors.Is(err, genselect.ErrEmpty) {
+ t.Fatalf("an unevidenced empty listing was pinned: %+v %v", pin, err)
+ }
+ return
+ }
+ if err != nil || !pin.Empty || pin.MaxTXID != 0 || pin.Generation != "" {
+ t.Fatalf("empty pin: %+v %v", pin, err)
+ }
+ obj, err := ReadPin(ctx, claims, pin)
+ if err != nil || obj.Empty != tc.want || obj.DB != nil {
+ t.Fatalf("object: %+v %v", obj, err)
+ }
+ })
+ }
+}
+
+func TestStorePinnerRefusesAnUnreadableManifest(t *testing.T) {
+ ctx := context.Background()
+ claims, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ if _, err := data.Put(ctx, hostfiles.ManifestKey(pinLocation+"/files/g3/"), []byte(`{"v":1}`), ""); err != nil {
+ t.Fatal(err)
+ }
+ p := StorePinner{Claims: claims, Data: data, Bucket: "data", Region: "ie", SelectDB: selectAt(41)}
+ if _, err := p.Pin(ctx, pinClaim(), time.Now()); !errors.Is(err, ErrRefused) {
+ t.Fatalf("a version 1 manifest was pinned: %v", err)
+ }
+}
+
+// pinnedManifest publishes gen's manifest whose one entry names an object
+// under origin (an earlier generation's prefix), and returns its pin.
+func pinnedManifest(t *testing.T, claims, data release.Store, gen, origin string, seq int64) (FilesPin, hostfiles.Manifest) {
+ t.Helper()
+ sum := sha256.Sum256([]byte("hello"))
+ sha := hex.EncodeToString(sum[:])
+ object := origin + "objects/" + strings.Repeat("0", 15) + "1-" + sha
+ if _, err := data.Put(context.Background(), object, []byte("hello"), ""); err != nil {
+ t.Fatal(err)
+ }
+ m := hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: seq, Entries: map[string]hostfiles.Entry{"a.txt": {Object: object, SHA256: sha, Size: 5, Mode: 0o644, Seq: 1}}, Platform: map[string]hostfiles.Entry{}}
+ _, raw := putManifest(t, data, gen, m)
+ rawSum := sha256.Sum256(raw)
+ manifestSHA := hex.EncodeToString(rawSum[:])
+ retained := RetainedManifestKey(host, "rel-1", manifestSHA)
+ if _, err := claims.Put(context.Background(), retained, raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ prefix := pinLocation + "/files/" + gen + "/"
+ return FilesPin{Prefix: prefix, Seq: seq, SHA256: manifestSHA, Origins: []string{prefix, origin}, Retained: retained}, m
+}
+
+func TestCopyPinnedManifestEnforcesThePin(t *testing.T) {
+ ctx := context.Background()
+ t.Run("a second relocation keeps objects of an earlier generation", func(t *testing.T) {
+ claims, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ pin, want := pinnedManifest(t, claims, data, "P1", pinLocation+"/files/g1/", 3)
+ got, err := CopyPinnedManifest(ctx, claims, data, pin, pinLocation+"/files/P2/")
+ if err != nil || got.Seq != 0 || want.Seq != 3 || got.Entries["a.txt"] != want.Entries["a.txt"] {
+ t.Fatalf("copy: %+v %v", got, err)
+ }
+ // A replay is a success; the copy's entries still name g1's object.
+ if _, err := CopyPinnedManifest(ctx, claims, data, pin, pinLocation+"/files/P2/"); err != nil {
+ t.Fatalf("replay: %v", err)
+ }
+ // Without the earlier generation among the origins the copy refuses.
+ narrow := pin
+ narrow.Origins = []string{pin.Prefix}
+ if _, err := CopyPinnedManifest(ctx, claims, data, narrow, pinLocation+"/files/P3/"); !errors.Is(err, hostfiles.ErrOrigin) {
+ t.Fatalf("an inherited object outside the origins: %v", err)
+ }
+ })
+ t.Run("a source manifest that moved after the pin is not read", func(t *testing.T) {
+ // A late PUT to the source changes nothing: P is written from the
+ // retained bytes, so an admitted pin stays usable (B2-4).
+ claims, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ pin, m := pinnedManifest(t, claims, data, "P1", pinLocation+"/files/g1/", 3)
+ m.Seq = 4
+ m.Entries["late.txt"] = m.Entries["a.txt"]
+ raw, _ := json.Marshal(m)
+ _, etag, _ := data.Get(ctx, hostfiles.ManifestKey(pin.Prefix))
+ if _, err := data.Put(ctx, hostfiles.ManifestKey(pin.Prefix), raw, etag); err != nil {
+ t.Fatal(err)
+ }
+ got, err := CopyPinnedManifest(ctx, claims, data, pin, pinLocation+"/files/P2/")
+ if _, late := got.Entries["late.txt"]; err != nil || got.Seq != 0 || late || len(got.Entries) != 1 {
+ t.Fatalf("the pinned revision: %+v %v", got, err)
+ }
+ })
+ t.Run("a retained copy that is not the pinned bytes", func(t *testing.T) {
+ claims, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ pin, m := pinnedManifest(t, claims, data, "P1", pinLocation+"/files/g1/", 3)
+ e := m.Entries["a.txt"]
+ e.Size = 6
+ m.Entries["a.txt"] = e
+ raw, _ := json.Marshal(m)
+ tampered := release.NewDirStore(t.TempDir())
+ if _, err := tampered.Put(ctx, pin.Retained, raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := CopyPinnedManifest(ctx, tampered, data, pin, pinLocation+"/files/P2/"); !errors.Is(err, ErrRefused) {
+ t.Fatalf("bytes the pin did not vouch for were copied: %v", err)
+ }
+ if _, _, err := data.Get(ctx, hostfiles.ManifestKey(pinLocation+"/files/P2/")); err == nil {
+ t.Fatal("P's manifest was written from bytes the pin did not admit")
+ }
+ })
+ t.Run("a host that never published files", func(t *testing.T) {
+ claims, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ pin := FilesPin{Prefix: pinLocation + "/files/g3/", Origins: []string{pinLocation + "/files/g3/"}}
+ m, err := CopyPinnedManifest(ctx, claims, data, pin, pinLocation+"/files/P1/")
+ if err != nil || m.Seq != 0 || len(m.Entries) != 0 {
+ t.Fatalf("empty manifest: %+v %v", m, err)
+ }
+ if _, err := CopyPinnedManifest(ctx, claims, data, pin, pinLocation+"/files/P1/"); err != nil {
+ t.Fatalf("replay: %v", err)
+ }
+ })
+}
+
+// Round 11: holder.admissions keeps eight entries, but an unchanged file
+// published at g1 is still referenced by every later generation's
+// manifest (each copies the last, and unchanged content is never
+// rewritten). Ten real admissions and publications later, with g1 trimmed
+// from the admission list, the pin still admits g1's object and the copy
+// succeeds; a reference outside this incarnation's files refuses.
+func TestPinKeepsAFileOriginTheAdmissionListTrimmed(t *testing.T) {
+ ctx := context.Background()
+ claims, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ created := time.Date(2026, 10, 4, 1, 0, 0, 0, time.UTC)
+ nh := hostclaim.NewHolder{Family: "instance", Key: "apps/bab/jam/instances/" + host + ".json", Account: "bab", App: "jam", Host: host, Region: "ie", RecordRegion: "ie", CreatedAt: created}
+ body := []byte(`{"host":"` + host + `","account":"bab","app":"jam","region":"ie","created_at":"2026-10-04T01:00:00Z"}`)
+ step := func(o hostclaim.Observed, tr func(*hostclaim.Claim) error) hostclaim.Observed {
+ t.Helper()
+ next, err := hostclaim.Update(ctx, claims, o, tr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return next
+ }
+ o, err := hostclaim.Read(ctx, claims, host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ o = step(o, hostclaim.BeginCreate(nh, body, "c", "console", created))
+ o = step(o, hostclaim.CompleteCreate("c", "src"))
+ loc := o.Claim.Holder.Location
+ n := 0
+ newID := func() string { n++; return fmt.Sprintf("id%d", n) }
+ var gens []string
+ prev, prevSeq := "", int64(0)
+ for i := 1; i <= 10; i++ {
+ gen := fmt.Sprintf("g%d", i)
+ o = step(o, hostclaim.AdmitWriter(hostclaim.Admission{Region: "ie", PredecessorRegion: "ie", ObservedSession: o.Claim.Holder.Session, ObservedGeneration: o.Claim.Holder.Generation,
+ Box: "src", Mode: "cold", NewGeneration: gen, PredecessorCertified: true}))
+ prefix := loc + "/files/" + gen + "/"
+ gens = append(gens, prefix)
+ if prev != "" {
+ if _, err := hostfiles.CopyManifest(ctx, data, prev, prefix, prevSeq, gens); err != nil {
+ t.Fatal(err)
+ }
+ }
+ p, err := hostfiles.Open(ctx, data, prefix)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // kept.txt and instance.key never change after g1; changed.txt
+ // changes every time.
+ if err := p.Stage(ctx, "kept.txt", []byte("written once"), 0o644, false); err != nil {
+ t.Fatal(err)
+ }
+ if err := p.Stage(ctx, "instance.key", []byte("the host's key"), 0o600, false); err != nil {
+ t.Fatal(err)
+ }
+ if err := p.Stage(ctx, "changed.txt", []byte(gen), 0o644, false); err != nil {
+ t.Fatal(err)
+ }
+ if prevSeq, err = p.Commit(ctx); err != nil {
+ t.Fatal(err)
+ }
+ prev = prefix
+ o = step(o, hostclaim.PublishGeneration("", "", o.Claim.Holder.Session, gen, "ie", newID))
+ }
+ h := o.Claim.Holder
+ if len(h.Admissions) != 8 {
+ t.Fatalf("admissions not trimmed to eight: %d", len(h.Admissions))
+ }
+ for _, a := range h.Admissions {
+ if a.Generation == "g1" || a.Predecessor == "g1" {
+ t.Fatalf("g1 is still in the admission list: %+v", h.Admissions)
+ }
+ }
+ m, _, err := hostfiles.Load(ctx, data, loc+"/files/g10/")
+ if err != nil || !strings.HasPrefix(m.Entries["kept.txt"].Object, loc+"/files/g1/objects/") {
+ t.Fatalf("the unchanged file is not g1's object: %+v %v", m.Entries["kept.txt"], err)
+ }
+ obs := o
+ obs.Claim.Op.ID = "rel-1"
+ p := StorePinner{Claims: claims, Data: data, Bucket: "data", Region: "ie", SelectDB: selectAt(41)}
+ pin, err := p.Pin(ctx, obs, time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC))
+ if err != nil {
+ t.Fatal(err)
+ }
+ obj, err := ReadPin(ctx, claims, pin)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, err := CopyPinnedManifest(ctx, claims, data, obj.Files, loc+"/files/P1/")
+ if err != nil || got.Entries["kept.txt"] != m.Entries["kept.txt"] {
+ t.Fatalf("copy with g1 trimmed from the admissions: %+v %v", got.Entries, err)
+ }
+ // A reference outside this incarnation's files is never admitted.
+ foreign := m
+ foreign.Entries = map[string]hostfiles.Entry{"instance.key": m.Entries["instance.key"], "x.txt": {Object: "instances/" + host + "/other-inc/files/g1/objects/" + strings.Repeat("a", 64), Seq: 1, SHA256: strings.Repeat("a", 64)}}
+ raw, _ := json.Marshal(foreign)
+ _, etag, _ := data.Get(ctx, hostfiles.ManifestKey(loc+"/files/g10/"))
+ if _, err := data.Put(ctx, hostfiles.ManifestKey(loc+"/files/g10/"), raw, etag); err != nil {
+ t.Fatal(err)
+ }
+ obs.Claim.Gen++
+ if _, err := p.Pin(ctx, obs, time.Date(2026, 10, 7, 9, 1, 0, 0, time.UTC)); !errors.Is(err, hostfiles.ErrOrigin) {
+ t.Fatalf("a foreign reference was pinned: %v", err)
+ }
+}
+
+// Round 14 finding 1: the source's files manifest is at sequence 8; P's
+// prepared manifest is a create-only copy of its entries at sequence 0
+// (claim rev 6.30 R18-1, moves rev 7.22 §7.6); the session publisher that
+// starts after the first admission takes it as its baseline and publishes
+// sequence 1, the inherited entries intact.
+func TestPreparedManifestIsSequenceZeroAndTheSessionPublishesOne(t *testing.T) {
+ ctx := context.Background()
+ claims, data := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ src := pinLocation + "/files/g3/"
+ p, err := hostfiles.Open(ctx, data, src)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for i := 1; i <= 7; i++ {
+ if i == 7 {
+ // The seventh revision also publishes the host's key (H6).
+ if err := p.Stage(ctx, "instance.key", []byte("the host's key"), 0o600, false); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := p.Stage(ctx, "a.txt", []byte(fmt.Sprintf("revision %d", i)), 0o644, false); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := p.Commit(ctx); err != nil {
+ t.Fatal(err)
+ }
+ }
+ pinner := StorePinner{Claims: claims, Data: data, Bucket: "data", Region: "ie", SelectDB: selectAt(41)}
+ ref, err := pinner.Pin(ctx, pinClaim(), time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC))
+ if err != nil || ref.FilesSeq != 8 { // seven revisions of a.txt and the key: eight objects
+ t.Fatalf("pin: %+v %v", ref, err)
+ }
+ obj, err := ReadPin(ctx, claims, ref)
+ if err != nil {
+ t.Fatal(err)
+ }
+ prepared := pinLocation + "/files/P1/"
+ m, err := CopyPinnedManifest(ctx, claims, data, obj.Files, prepared)
+ if err != nil || m.Seq != 0 {
+ t.Fatalf("prepared manifest: %+v %v", m, err)
+ }
+ stored, _, err := hostfiles.Load(ctx, data, prepared)
+ if err != nil || stored.Seq != 0 || stored.Entries["a.txt"].Object == "" || !strings.HasPrefix(stored.Entries["a.txt"].Object, src+"objects/") {
+ t.Fatalf("stored prepared manifest: %+v %v", stored, err)
+ }
+ session, err := hostfiles.Open(ctx, data, prepared)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := session.Stage(ctx, "b.txt", []byte("first session write"), 0o644, false); err != nil {
+ t.Fatal(err)
+ }
+ seq, err := session.Commit(ctx)
+ if err != nil || seq != 1 {
+ t.Fatalf("the first session publication: %d %v", seq, err)
+ }
+ after, _, err := hostfiles.Load(ctx, data, prepared)
+ if err != nil || after.Seq != 1 || after.Entries["a.txt"] != stored.Entries["a.txt"] || after.Entries["b.txt"].Object == "" {
+ t.Fatalf("after the first publication: %+v %v", after, err)
+ }
+}
+
+// Two-box model finding F2: no files manifest anywhere in the chain, and
+// a writer was admitted (session 3): the pin refuses rather than pin an
+// empty set an install would mint a fresh instance.key over. A holder
+// that never admitted a writer pins empty.
+func TestStorePinnerRefusesAnEmptyFilesPinAfterAWriter(t *testing.T) {
+ ctx := context.Background()
+ at := time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC)
+ for _, tc := range []struct {
+ session uint64
+ want error
+ }{{3, ErrFilesUnevidenced}, {0, nil}} {
+ claims, data := release_NewDirStore(t), release_NewDirStore(t)
+ o := pinClaim()
+ o.Claim.Holder.Session = tc.session
+ p := StorePinner{Claims: claims, Data: data, Bucket: "data", Region: "ie", SelectDB: selectAt(41)}
+ pin, err := p.Pin(ctx, o, at)
+ if !errors.Is(err, tc.want) {
+ t.Fatalf("session %d: %+v %v", tc.session, pin, err)
+ }
+ }
+}
+
+// Round 17 finding 1 (P1): a manifest exists, copied before the app
+// started, without the instance.key the app minted afterwards; the writer
+// was admitted and the database is nonempty. The pin refuses (the worker
+// marks the operation stuck); with the key in the manifest, or an empty
+// database, it pins.
+func TestStorePinnerRefusesAKeylessManifestForANonemptyDatabase(t *testing.T) {
+ ctx := context.Background()
+ at := time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC)
+ empty := func(context.Context, genselect.Request) (*genselect.Plan, error) { return nil, genselect.ErrEmpty }
+ released0 := func() string {
+ raw, _ := json.Marshal(hostclaim.Release{ReleasedBy: "src", Enrollment: "e", Generation: "g3", Session: 3, FinalTXID: 0, ManifestSeq: 4, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)})
+ return string(raw)
+ }
+ for _, tc := range []struct {
+ name string
+ key bool
+ db func(context.Context, genselect.Request) (*genselect.Plan, error)
+ journal map[string]string
+ want error
+ }{
+ {"keyless, nonempty database", false, selectAt(41), map[string]string{}, ErrFilesUnevidenced},
+ {"keyed, nonempty database", true, selectAt(41), map[string]string{}, nil},
+ {"keyless, empty database at a release of 0", false, empty, map[string]string{"release": released0()}, nil},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ claims, data := release_NewDirStore(t), release_NewDirStore(t)
+ m := hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: 4, Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}}
+ if tc.key {
+ m = keyed(m)
+ }
+ putManifest(t, data, "g3", m)
+ o := pinClaim()
+ o.Claim.Op.Journal = tc.journal
+ p := StorePinner{Claims: claims, Data: data, Bucket: "data", Region: "ie", SelectDB: tc.db}
+ if _, err := p.Pin(ctx, o, at); !errors.Is(err, tc.want) {
+ t.Fatalf("got %v, want %v", err, tc.want)
+ }
+ })
+ }
+}
diff --git a/internal/hostrelocate/worker.go b/internal/hostrelocate/worker.go
new file mode 100644
index 000000000..07615a530
--- /dev/null
+++ b/internal/hostrelocate/worker.go
@@ -0,0 +1,561 @@
+// Package hostrelocate is the regional worker's half of relocation (region
+// HA spec 1 §5.8, claim §4 L1): starting a relocation, and replaying every
+// worker CAS from the claim alone. Any active enrolled box of the region
+// may act (claim P6-D5); correctness never depends on which, because every
+// step is a claim CAS conditional on the operation, its phase and the
+// claim's ETag. The participants' halves (acknowledgement, release,
+// install, probe) are not here: they run on the boxes, under the host
+// guard, through Plan 6b's box assembly.
+//
+// This package only READS termination fences. hostfence.TerminationWorker
+// writes them (control/box-fences/<enrollment>.json).
+package hostrelocate
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "slices"
+ "strconv"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostactors"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostfence"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/termfence"
+)
+
+const (
+ // ProbeDeadline is §5.8's probe deadline: a live target whose probe has
+ // not passed this long after entering probing goes to probe-failed.
+ ProbeDeadline = 10 * time.Minute
+ // ReleaseDeadline is this plan's stopping policy for a release that
+ // never completes (§5.8: "a flush that never reaches N goes to stuck";
+ // "a probe-failed target that cannot finish its release sequence also
+ // goes to stuck"). The spec names no number; this one matches the
+ // probe deadline. A live box whose release is not journaled this long
+ // after entering releasing or probe-failed leaves the operation stuck
+ // for the operator: nothing moves away from a live box without its
+ // stop evidence.
+ ReleaseDeadline = 10 * time.Minute
+)
+
+var (
+ ErrUnsupported = errors.New("hostrelocate: a required provider is missing")
+ ErrNotWorker = errors.New("hostrelocate: this box is not an active, unfenced worker of the region")
+ ErrNotYet = errors.New("hostrelocate: waiting for a participant or a time bound")
+ ErrRefused = errors.New("hostrelocate: relocation refused")
+ ErrNoController = errors.New("hostrelocate: no execution controller is open for this participation")
+)
+
+// Lease is the replica lease on holder.location (claim C3), as
+// lease.Acquire returns it. The worker holds it from selecting a pin until
+// the pin is journaled, so a park worker (which takes the same lease to
+// retire) cannot retire the versions in between (claim §4: "New pins must
+// be established under the required leases and fresh claim authority
+// before conflicting retirement can proceed").
+type Lease interface {
+ Lost() <-chan struct{}
+ Release(context.Context) error
+}
+
+// Request asks for one relocation. Target is required; Source must be the
+// claim's executor. Reason is drain, rebalance or failover.
+type Request struct {
+ Reason, Source, Target string
+}
+
+// Owner is the app-admission registration owner of a box-started
+// operation (claim §5.1, "A regional worker starting relocating").
+type Owner struct {
+ Box, Enrollment, Controller string
+}
+
+// Admission is claim §5.1's app-admission registration for a box-started
+// relocation: Register binds the saved claim observation and holder
+// incarnation while admission is open; Recheck is the final check before
+// the single start CAS. Claim Plan 5's app gate implements it.
+type Admission interface {
+ Register(context.Context, hostclaim.Observed, Owner, string) (string, error)
+ Recheck(context.Context, string) error
+}
+
+// Pinner selects, writes and returns one pinned recovery set (§5.8).
+type Pinner interface {
+ Pin(context.Context, hostclaim.Observed, time.Time) (hostclaim.RecoveryPin, error)
+}
+
+// Prepared verifies the remote build of the operation's prepared
+// generation P (§5.8 "Building the prepared generation"): done is true only
+// when a fresh read verifies P's snapshot, level-0 anchor and completed
+// install manifest (or, for a host with nothing replicated, P's files
+// manifest alone). Moves' shared install routine (L16) writes them.
+type Prepared interface {
+ Verify(context.Context, hostclaim.Observed) (a hostclaim.PreparedAuthority, done bool, err error)
+}
+
+// Worker drives relocation for one region. Every field is required.
+type Worker struct {
+ Claims release.Store // the deployment (control) bucket
+ Directory hostactors.Directory
+ Fences hostclaim.FenceAuthority // hostfence.ClaimFenceAuthority
+ Releases hostclaim.ReleaseAuthority // Plan 6b's stop receipt reader
+ Admission Admission
+ // Census is the box-started start census (claim D7 with L5 and L20):
+ // hostactors.Directory.CensusBoxRelocation with the deployment's
+ // evidence and floor.
+ Census func(ctx context.Context, starter string, participants []string) error
+ // Choose names a target in the operation's region, never one in
+ // exclude (hostops.ExecutorChooser behind an adapter).
+ Choose func(ctx context.Context, obs hostclaim.Observed, exclude []string) (hostclaim.Target, error)
+ // Eligible refuses a named target that the complete eligible set
+ // (§2, boxchoice.Eligible over the roster and the edges files) does
+ // not hold: stale edges, a missing capability or secrets key, a
+ // draining box (hostops.ExecutorChooser.Permits).
+ Eligible func(ctx context.Context, obs hostclaim.Observed, box string) error
+ // Relocatable refuses a host whose authoritative record is not a
+ // relocatable shape: a unit- or socket-backed route, a tunnel, a
+ // sidecar (the same shapes Plan 6b's wake refuses).
+ Relocatable func(ctx context.Context, obs hostclaim.Observed) error
+ // Leases acquires the replica lease on the host's location in the
+ // region's data bucket (lease.Acquire).
+ Leases func(ctx context.Context, location string) (Lease, error)
+ Pins Pinner
+ Prepared Prepared
+ // Discharge judges a fenced box's volume records (boxvolumes.Discharge).
+ // It never gates anything: nil, or an unqualified verdict, leaves the
+ // box's park-local chores open for an operator (H19).
+ Discharge func(context.Context, hostactors.Box, termfence.Record) *hostclaim.LocalDischarge
+ // Notify bumps the control serial after an executor CAS (§5.3,
+ // "Prompt changes"). A failed bump costs cadence, never correctness.
+ Notify func(context.Context)
+ Self termfence.WrittenBy
+ Region string
+ Controller string
+ NewID func() string
+ Now func() time.Time
+}
+
+func (w Worker) ready() error {
+ if w.Claims == nil || w.Directory.Store == nil || w.Fences == nil || w.Releases == nil || w.Admission == nil || w.Census == nil ||
+ w.Choose == nil || w.Eligible == nil || w.Relocatable == nil || w.Leases == nil || w.Pins == nil || w.Prepared == nil || w.Discharge == nil || w.Notify == nil || w.Self.Box == "" || w.Self.Enrollment == "" ||
+ w.Region == "" || w.Controller == "" || w.NewID == nil || w.Now == nil {
+ return ErrUnsupported
+ }
+ return nil
+}
+
+// self checks, on every call, that this box is an active, unfenced worker
+// enrolled in the region (§5.8 "Each worker first checks its own roster
+// entry is active in this region").
+func (w Worker) self(ctx context.Context) (hostactors.Observed, error) {
+ o, err := w.Directory.Read(ctx)
+ if err != nil {
+ return hostactors.Observed{}, err
+ }
+ b, ok := o.Roster.Boxes[w.Self.Box]
+ if !ok || b.State != "active" || b.Enrollment != w.Self.Enrollment || b.Region != w.Region {
+ return hostactors.Observed{}, ErrNotWorker
+ }
+ fenced, err := w.Directory.TerminationFence(ctx, o, w.Self.Box, w.Now())
+ if err != nil || fenced {
+ return hostactors.Observed{}, errors.Join(ErrNotWorker, err)
+ }
+ return o, nil
+}
+
+// fence reads box's termination fence, if any, as the exact bytes a claim
+// CAS will cite. Absence is never evidence: no fence is (nil, nil); an
+// unreadable or invalid one is an error, never "not fenced".
+func (w Worker) fence(ctx context.Context, o hostactors.Observed, box string) (*hostclaim.FenceRef, termfence.Record, error) {
+ b, ok := o.Roster.Boxes[box]
+ if !ok {
+ return nil, termfence.Record{}, fmt.Errorf("%w: %s is not in the roster", ErrRefused, box)
+ }
+ key := hostfence.FenceKey(b.Enrollment)
+ raw, _, err := w.Directory.Store.Get(ctx, key)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, termfence.Record{}, nil
+ }
+ if err != nil {
+ return nil, termfence.Record{}, err
+ }
+ var f termfence.Record
+ if err := json.Unmarshal(raw, &f); err != nil {
+ return nil, termfence.Record{}, err
+ }
+ if err := hostactors.ValidateTerminationFence(o, box, f, w.Now()); err != nil {
+ return nil, termfence.Record{}, err
+ }
+ sum := sha256.Sum256(raw)
+ return &hostclaim.FenceRef{Enrollment: b.Enrollment, Key: key, SHA256: hex.EncodeToString(sum[:])}, f, nil
+}
+
+// Start writes the start CAS for req (§5.8 "The start CAS", §5.10). It
+// refuses unless the claim serves on req.Source; the roster shows the
+// starter, the target and (for a drain) the source active, unfenced and
+// with no executable update reserved (H14); the census passes; and the
+// app admits the registration. A failover requires the source's fence.
+func (w Worker) Start(ctx context.Context, host string, req Request) (hostclaim.Observed, error) {
+ if err := w.ready(); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ o, err := w.self(ctx)
+ if err != nil {
+ return hostclaim.Observed{}, err
+ }
+ obs, err := hostclaim.Read(ctx, w.Claims, host)
+ if err != nil {
+ return hostclaim.Observed{}, err
+ }
+ if obs.Absent || obs.Claim.Op.Kind != hostclaim.KindServing || obs.Claim.Op.Exec != req.Source || req.Target == req.Source {
+ return hostclaim.Observed{}, fmt.Errorf("%w: %s does not serve on %s", ErrRefused, host, req.Source)
+ }
+ if obs.Claim.Holder.Family != "instance" {
+ return hostclaim.Observed{}, fmt.Errorf("%w: a %s host is not relocatable", ErrRefused, obs.Claim.Holder.Family)
+ }
+ if err := w.Relocatable(ctx, obs); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ if err := w.Eligible(ctx, obs, req.Target); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ source, sok := o.Roster.Boxes[req.Source]
+ target, tok := o.Roster.Boxes[req.Target]
+ if !sok || !tok || source.Region != w.Region || target.Region != w.Region {
+ return hostclaim.Observed{}, fmt.Errorf("%w: source and target must be enrolled in %s", ErrRefused, w.Region)
+ }
+ ref, record, err := w.fence(ctx, o, req.Source)
+ if err != nil {
+ return hostclaim.Observed{}, err
+ }
+ refs := []string{w.Self.Box, req.Target}
+ switch req.Reason {
+ case hostclaim.ReasonFailover:
+ if ref == nil {
+ return hostclaim.Observed{}, fmt.Errorf("%w: a failover needs the source's termination fence", ErrRefused)
+ }
+ case hostclaim.ReasonDrain, hostclaim.ReasonRebalance:
+ if ref != nil {
+ return hostclaim.Observed{}, fmt.Errorf("%w: the source is fenced; start a failover", ErrRefused)
+ }
+ if obs.Claim.Holder.Session == 0 {
+ // Ruling 11 (rev 2): a never-woken source has no writer session
+ // to stop, and root refuses stop evidence for session 0 until
+ // claim Plan 5's no-process witness lands. Until then only a
+ // failover relocates such a host.
+ return hostclaim.Observed{}, fmt.Errorf("%w: %s never admitted a writer on %s; a drain has nothing to release", ErrRefused, host, req.Source)
+ }
+ refs = append(refs, req.Source)
+ default:
+ return hostclaim.Observed{}, fmt.Errorf("%w: reason %q", ErrRefused, req.Reason)
+ }
+ refs = slices.Compact(slices.Sorted(slices.Values(refs)))
+ revision, err := w.Directory.CheckBoxStart(ctx, refs...)
+ if err != nil {
+ return hostclaim.Observed{}, err
+ }
+ if err := w.Census(ctx, w.Self.Box, refs); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ id := w.NewID()
+ admission, err := w.Admission.Register(ctx, obs, Owner{Box: w.Self.Box, Enrollment: w.Self.Enrollment, Controller: w.Controller}, id)
+ if err != nil {
+ return hostclaim.Observed{}, err
+ }
+ if err := w.Admission.Recheck(ctx, admission); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ in := hostclaim.RelocationStart{OpID: id, By: "box:" + w.Self.Box, Now: w.Now(), Region: w.Region, Reason: req.Reason,
+ SourceBox: req.Source, SourceEnrollment: source.Enrollment, TargetBox: req.Target, TargetEnrollment: target.Enrollment,
+ DirectoryRevision: revision, AdmissionID: admission, Fence: ref}
+ if ref != nil {
+ in.Discharge = w.Discharge(ctx, source, record)
+ }
+ next, err := hostclaim.StartRelocation(ctx, w.Claims, obs, in, w.Fences)
+ if err == nil && ref != nil {
+ w.Notify(ctx)
+ }
+ return next, err
+}
+
+// Step makes at most one worker CAS for host's relocation, the one its
+// phase and evidence call for, and returns the claim it left. ErrNotYet
+// means a participant or a time bound is awaited; nothing was written.
+// A host that is not relocating, or whose operation is stuck, is left
+// alone.
+func (w Worker) Step(ctx context.Context, host string) (hostclaim.Observed, error) {
+ if err := w.ready(); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ o, err := w.self(ctx)
+ if err != nil {
+ return hostclaim.Observed{}, err
+ }
+ obs, err := hostclaim.Read(ctx, w.Claims, host)
+ if err != nil || obs.Absent {
+ return obs, err
+ }
+ c := obs.Claim
+ if c.Op.Kind != hostclaim.KindRelocating || c.Op.Intent["region"] != w.Region || c.Op.Journal["stuck"] != "" {
+ return obs, nil
+ }
+ id := c.Op.ID
+ // Evidence first: a fenced participant changes what every phase may do.
+ if src := c.Op.Intent["source_box"]; !slices.Contains(c.Lost, src) {
+ ref, record, err := w.fence(ctx, o, src)
+ if err != nil {
+ return obs, err
+ }
+ if ref != nil {
+ next, err := hostclaim.FenceSource(ctx, w.Claims, obs, id, *ref, w.Discharge(ctx, o.Roster.Boxes[src], record), w.Fences)
+ if err == nil {
+ w.Notify(ctx)
+ }
+ return next, err
+ }
+ }
+ if dst := c.Op.Intent["target_box"]; dst != "" && !slices.Contains(c.Lost, dst) {
+ ref, record, err := w.fence(ctx, o, dst)
+ if err != nil {
+ return obs, err
+ }
+ if ref != nil {
+ return w.replace(ctx, obs, *ref, w.Discharge(ctx, o.Roster.Boxes[dst], record), record.ObservedAt)
+ }
+ }
+ switch c.Op.Phase {
+ case hostclaim.PhaseStarting:
+ if !acked(c) {
+ return obs, ErrNotYet
+ }
+ if c.Op.Exec == "none" {
+ return w.update(ctx, obs, hostclaim.PinFailover(id))
+ }
+ return w.update(ctx, obs, hostclaim.ToReleasing(id, w.Now()))
+ case hostclaim.PhaseReleasing:
+ if c.Op.Journal["release"] == "" {
+ return w.releaseOverdue(ctx, obs, "releasing_since")
+ }
+ return hostclaim.PinAfterRelease(ctx, w.Claims, obs, id, w.Releases, nil)
+ case hostclaim.PhaseProbeFailed:
+ if c.Op.Journal["release"] == "" {
+ return w.releaseOverdue(ctx, obs, "probe_failed_since")
+ }
+ if exhausted(c) {
+ return w.stuckOn(ctx, obs, hostclaim.ErrReplacementsExceeded)
+ }
+ next, err := w.next(ctx, obs)
+ if err != nil {
+ // No eligible box yet; a later pass may find one, but not for
+ // ever: in a two-box region the source is excluded and there
+ // may be none (rev 17, finding F1). ReleaseDeadline after the
+ // target failed, the operation is stuck for an operator.
+ since, perr := time.Parse(time.RFC3339Nano, c.Op.Journal["probe_failed_since"])
+ if perr == nil && w.Now().Sub(since) > ReleaseDeadline {
+ return w.update(ctx, obs, hostclaim.MarkStuck(id, fmt.Sprintf("no eligible box for the nested relocation after %s: %v", ReleaseDeadline, err)))
+ }
+ return obs, err
+ }
+ out, err := hostclaim.PinAfterRelease(ctx, w.Claims, obs, id, w.Releases, &next)
+ if err != nil {
+ // A live target's replacement counts too: the third is stuck.
+ return w.stuckOn(ctx, obs, err)
+ }
+ return out, nil
+ case hostclaim.PhasePinning:
+ return w.pinning(ctx, obs)
+ case hostclaim.PhaseProbing:
+ since, err := time.Parse(time.RFC3339Nano, c.Op.Journal["probing_since"])
+ if err != nil {
+ return obs, err
+ }
+ if c.Op.Journal["probe"] != "" {
+ next, err := w.update(ctx, obs, hostclaim.CompleteRelocation(id, w.NewID))
+ if err == nil {
+ w.Notify(ctx)
+ }
+ return next, err
+ }
+ if w.Now().Sub(since) > ProbeDeadline {
+ return w.update(ctx, obs, hostclaim.ToProbeFailed(id, w.Now()))
+ }
+ return obs, ErrNotYet
+ case hostclaim.PhaseInstalling:
+ a, done, err := w.Prepared.Verify(ctx, obs)
+ if err != nil || !done || !acked(c) {
+ return obs, errors.Join(ErrNotYet, err)
+ }
+ return w.update(ctx, obs, hostclaim.ToInstalled(id, a))
+ }
+ // installed is the target's: its local install and its probe (Plan 6b).
+ return obs, ErrNotYet
+}
+
+func (w Worker) pinning(ctx context.Context, obs hostclaim.Observed) (hostclaim.Observed, error) {
+ c := obs.Claim
+ id := c.Op.ID
+ if c.Op.Intent["target_box"] == "" {
+ next, err := w.next(ctx, obs)
+ if err != nil {
+ return obs, err
+ }
+ return w.update(ctx, obs, hostclaim.NameTarget(id, next))
+ }
+ if c.Op.Journal["pin"] == "" {
+ for key, at := range c.Op.Journal {
+ if len(key) > len(".observed_at") && key[len(key)-len(".observed_at"):] == ".observed_at" {
+ t, err := time.Parse(time.RFC3339Nano, at)
+ if err != nil {
+ return obs, err
+ }
+ if w.Now().Before(t.Add(hostclaim.FailoverPinDelay)) {
+ return obs, ErrNotYet
+ }
+ }
+ }
+ return w.pinUnderLease(ctx, obs)
+ }
+ t := hostclaim.Target{Box: c.Op.Intent["target_box"], Enrollment: c.Op.Intent["target_enrollment"]}
+ if !acked(c) {
+ return obs, ErrNotYet
+ }
+ next, err := w.update(ctx, obs, hostclaim.ToInstalling(id, t, w.NewID()))
+ if err == nil {
+ w.Notify(ctx)
+ }
+ return next, err
+}
+
+// pinUnderLease selects and journals the pin while holding the replica
+// lease on holder.location, so no park can retire the pinned versions
+// between the selection and the journal CAS (claim §4). A lease lost
+// before the CAS refuses: the next pass selects afresh.
+func (w Worker) pinUnderLease(ctx context.Context, obs hostclaim.Observed) (hostclaim.Observed, error) {
+ held, err := w.Leases(ctx, obs.Claim.Holder.Location)
+ if err != nil {
+ return obs, err
+ }
+ defer func() { _ = held.Release(ctx) }()
+ pin, err := w.Pins.Pin(ctx, obs, w.Now())
+ if errors.Is(err, ErrFilesUnevidenced) {
+ // H6: instance.key, report.token and the env are continuous; a pin
+ // that cannot show them never existed waits for an operator.
+ return w.update(ctx, obs, hostclaim.MarkStuck(obs.Claim.Op.ID, err.Error()))
+ }
+ if err != nil {
+ return obs, err
+ }
+ select {
+ case <-held.Lost():
+ return obs, fmt.Errorf("%w: the location lease was lost before the pin was journaled", ErrNotYet)
+ default:
+ }
+ return w.update(ctx, obs, hostclaim.JournalPin(obs.Claim.Op.ID, pin))
+}
+
+// releaseOverdue marks the operation stuck when a live box has not
+// journaled its release ReleaseDeadline after entering the phase whose
+// start key names; before that it waits.
+func (w Worker) releaseOverdue(ctx context.Context, obs hostclaim.Observed, sinceKey string) (hostclaim.Observed, error) {
+ since, err := time.Parse(time.RFC3339Nano, obs.Claim.Op.Journal[sinceKey])
+ if err != nil {
+ return obs, err
+ }
+ if w.Now().Sub(since) <= ReleaseDeadline {
+ return obs, ErrNotYet
+ }
+ return w.update(ctx, obs, hostclaim.MarkStuck(obs.Claim.Op.ID, fmt.Sprintf("%s has not released %s after %s", obs.Claim.Op.Exec, ReleaseDeadline, obs.Claim.Op.Phase)))
+}
+
+// next names the operation's next target: never the source, a lost box,
+// a box this operation already replaced, or a box that owes a park-local
+// chore for the host (spec 1 §2; ReplaceTarget refuses one in its CAS).
+func (w Worker) next(ctx context.Context, obs hostclaim.Observed) (hostclaim.Target, error) {
+ c := obs.Claim
+ exclude := append([]string{c.Op.Intent["source_box"]}, c.Lost...)
+ exclude = append(exclude, hostclaim.LocalChoreBoxes(c)...)
+ if t := c.Op.Intent["target_box"]; t != "" {
+ exclude = append(exclude, t)
+ }
+ var replaced []string
+ if raw := c.Op.Journal["excluded_targets"]; raw != "" {
+ if err := json.Unmarshal([]byte(raw), &replaced); err != nil {
+ return hostclaim.Target{}, err
+ }
+ }
+ return w.Choose(ctx, obs, append(exclude, replaced...))
+}
+
+func (w Worker) replace(ctx context.Context, obs hostclaim.Observed, ref hostclaim.FenceRef, d *hostclaim.LocalDischarge, observedAt string) (hostclaim.Observed, error) {
+ // Exhaustion is checked before a box is chosen (§5.8 "The bound"):
+ // with the replacements used, no spare is looked for; the dead
+ // target's fence is consumed into a stuck pinning (the loss retained,
+ // its installation's authority revoked, no probe to wait for).
+ if exhausted(obs.Claim) {
+ out, err := hostclaim.FenceExhaustedTarget(ctx, w.Claims, obs, obs.Claim.Op.ID, ref, d, w.Fences)
+ if err == nil {
+ w.Notify(ctx)
+ }
+ return out, err
+ }
+ next, err := w.next(ctx, obs)
+ if err != nil {
+ // No eligible box to replace the dead target. A later pass may
+ // find one, but not for ever (rev 18, round 17 finding 2):
+ // ReleaseDeadline after the fence's observation the fence is
+ // consumed and the operation is stuck for an operator.
+ at, perr := time.Parse(time.RFC3339Nano, observedAt)
+ if perr == nil && w.Now().Sub(at) > ReleaseDeadline {
+ out, ferr := hostclaim.FenceTargetWithoutSpare(ctx, w.Claims, obs, obs.Claim.Op.ID, ref, d, w.Fences,
+ fmt.Sprintf("%s terminated and no eligible box replaced it after %s: %v", obs.Claim.Op.Intent["target_box"], ReleaseDeadline, err))
+ if ferr == nil {
+ w.Notify(ctx)
+ }
+ return out, ferr
+ }
+ return obs, err
+ }
+ out, err := hostclaim.ReplaceTarget(ctx, w.Claims, obs, obs.Claim.Op.ID, ref, d, next, w.NewID(), w.Fences)
+ if errors.Is(err, hostclaim.ErrReplacementsExceeded) {
+ return w.stuckOn(ctx, obs, err)
+ }
+ if err == nil {
+ w.Notify(ctx)
+ }
+ return out, err
+}
+
+// stuckOn marks the operation stuck when it has used its replacements;
+// any other error is returned for a later pass to retry.
+func (w Worker) stuckOn(ctx context.Context, obs hostclaim.Observed, cause error) (hostclaim.Observed, error) {
+ if !errors.Is(cause, hostclaim.ErrReplacementsExceeded) {
+ return obs, cause
+ }
+ return w.update(ctx, obs, hostclaim.MarkStuck(obs.Claim.Op.ID, cause.Error()))
+}
+
+func (w Worker) update(ctx context.Context, obs hostclaim.Observed, tr func(*hostclaim.Claim) error) (hostclaim.Observed, error) {
+ return hostclaim.Update(ctx, w.Claims, obs, tr)
+}
+
+// exhausted reports whether the operation has used its target replacements.
+func exhausted(c *hostclaim.Claim) bool {
+ n, _ := strconv.ParseUint(c.Op.Journal["replacements"], 10, 64)
+ return n >= hostclaim.MaxTargetReplacements
+}
+
+// acked is the D6 barrier as the worker reads it before a CAS that needs it.
+func acked(c *hostclaim.Claim) bool {
+ for _, box := range c.Op.Participants {
+ if c.Op.Journal["exec_ack."+box] == "" {
+ return false
+ }
+ }
+ return true
+}
diff --git a/internal/hostrelocate/worker_test.go b/internal/hostrelocate/worker_test.go
new file mode 100644
index 000000000..59f4e414d
--- /dev/null
+++ b/internal/hostrelocate/worker_test.go
@@ -0,0 +1,766 @@
+package hostrelocate
+
+import (
+ "context"
+ "crypto/ed25519"
+ "crypto/rand"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "slices"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostactors"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostfence"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/termfence"
+)
+
+const host = "shop.bab.oncarlos.com"
+
+var boxes = map[string]string{"wrk": "a", "src": "b", "dst": "c", "spa": "d", "sp2": "e", "sp3": "f"}
+
+func enrollment(box string) string { return strings.Repeat(boxes[box], 32) }
+
+type fixture struct {
+ pinErr error
+ t *testing.T
+ s release.Store
+ w Worker
+ now time.Time
+ notified int
+ sign release.Signer
+ roster hostactors.Roster
+ // leases records the lease provider's calls in order with the pinner's.
+ leases []string
+ loseLease bool
+ ineligible map[string]bool
+}
+
+// fakeLease is one held replica lease; lost is closed when the test takes
+// it away before the pin is journaled.
+type fakeLease struct {
+ f *fixture
+ lost chan struct{}
+}
+
+func (l *fakeLease) Lost() <-chan struct{} { return l.lost }
+func (l *fakeLease) Release(context.Context) error {
+ l.f.leases = append(l.f.leases, "release")
+ return nil
+}
+
+type okReleases struct{}
+
+func (okReleases) VerifyRelease(context.Context, hostclaim.Claim, hostclaim.Release) error {
+ return nil
+}
+
+type admission struct{ registered []string }
+
+func (a *admission) Register(_ context.Context, _ hostclaim.Observed, o Owner, id string) (string, error) {
+ if o.Box != "wrk" || o.Enrollment != enrollment("wrk") {
+ return "", errors.New("wrong owner")
+ }
+ a.registered = append(a.registered, id)
+ return "adm-" + id, nil
+}
+func (a *admission) Recheck(context.Context, string) error { return nil }
+
+// built reports P built once the target has said so (the participant's
+// half), as moves' install routine would leave it.
+type built struct{}
+
+func (built) Verify(_ context.Context, o hostclaim.Observed) (hostclaim.PreparedAuthority, bool, error) {
+ c := o.Claim
+ if c.Op.Journal["test.built"] != c.Op.Journal["prepared"] {
+ return hostclaim.PreparedAuthority{}, false, nil
+ }
+ var pin hostclaim.RecoveryPin
+ _ = json.Unmarshal([]byte(c.Op.Journal["pin"]), &pin)
+ return hostclaim.PreparedAuthority{Generation: c.Op.Journal["prepared"], InstallManifest: "m", SHA256: strings.Repeat("c", 64), SnapshotTXID: pin.MaxTXID,
+ SourceFilesSeq: pin.FilesSeq, FilesManifestSHA256: strings.Repeat("f", 64)}, true, nil
+}
+
+type pins struct{ f *fixture }
+
+func (p pins) Pin(_ context.Context, o hostclaim.Observed, now time.Time) (hostclaim.RecoveryPin, error) {
+ p.f.leases = append(p.f.leases, "pin")
+ if p.f.pinErr != nil {
+ return hostclaim.RecoveryPin{}, p.f.pinErr
+ }
+ var r hostclaim.Release
+ _ = json.Unmarshal([]byte(o.Claim.Op.Journal["release"]), &r)
+ return hostclaim.RecoveryPin{Key: PinKey(o.Claim.Host, o.Claim.Op.ID, o.Claim.Gen), SHA256: strings.Repeat("a", 64), Generation: o.Claim.Holder.Generation,
+ MaxTXID: max(r.FinalTXID, 40), FilesManifest: "m", FilesSeq: max(r.ManifestSeq, 7), PinnedAt: now}, nil
+}
+
+func newFixture(t *testing.T) *fixture {
+ t.Helper()
+ ctx := context.Background()
+ f := &fixture{t: t, s: release.NewDirStore(t.TempDir()), now: time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC)}
+ pub, key, err := ed25519.GenerateKey(rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ f.sign = release.RootSigner(key)
+ f.roster = hostactors.Roster{V: 1, DeploymentBucket: "bucket", Revision: 1, Regions: map[string]hostactors.Region{"ie": {Account: "123", AWSRegion: "eu-west-1"}},
+ Boxes: map[string]hostactors.Box{}, Consoles: map[string]hostactors.Console{}, Holds: map[string]hostactors.Reservation{}, Activations: map[string]hostactors.Reservation{}}
+ for ref := range boxes {
+ f.roster.Boxes[ref] = hostactors.Box{Enrollment: enrollment(ref), Namespace: "platform", BoxID: ref, Label: ref, Region: "ie", Pool: "default",
+ Cloud: hostactors.Cloud{Provider: "aws", Account: "123", Region: "eu-west-1", InstanceID: "i-" + ref}, EvidenceSHA256: strings.Repeat("c", 64), State: "active"}
+ }
+ f.writeRoster()
+ d := hostactors.Directory{Store: f.s, PublicKey: pub, Bucket: "bucket"}
+ clock := func() time.Time { return f.now }
+ n := 0
+ f.w = Worker{Claims: f.s, Directory: d, Fences: hostfence.ClaimFenceAuthority{Directory: d, Now: clock}, Releases: okReleases{}, Admission: &admission{},
+ Census: func(context.Context, string, []string) error { return nil },
+ Choose: func(_ context.Context, _ hostclaim.Observed, exclude []string) (hostclaim.Target, error) {
+ for _, b := range []string{"dst", "spa", "sp2", "sp3"} {
+ if !slices.Contains(exclude, b) {
+ return hostclaim.Target{Box: b, Enrollment: enrollment(b)}, nil
+ }
+ }
+ return hostclaim.Target{}, errors.New("no eligible box")
+ },
+ Eligible: func(_ context.Context, _ hostclaim.Observed, box string) error {
+ if f.ineligible[box] {
+ return errors.New("not eligible: " + box)
+ }
+ return nil
+ },
+ Relocatable: func(context.Context, hostclaim.Observed) error { return nil },
+ Leases: func(context.Context, string) (Lease, error) {
+ f.leases = append(f.leases, "acquire")
+ l := &fakeLease{f: f, lost: make(chan struct{})}
+ if f.loseLease {
+ close(l.lost)
+ }
+ return l, nil
+ },
+ Pins: pins{f},
+ Prepared: built{},
+ Discharge: func(context.Context, hostactors.Box, termfence.Record) *hostclaim.LocalDischarge { return nil },
+ Notify: func(context.Context) { f.notified++ },
+ Self: termfence.WrittenBy{Box: "wrk", Enrollment: enrollment("wrk")}, Region: "ie", Controller: "ctl",
+ NewID: func() string { n++; return "id" + string(rune('0'+n)) }, Now: clock}
+ f.ineligible = map[string]bool{}
+ // A host serving on src, whose source admitted one cold session.
+ created := time.Date(2026, 10, 4, 1, 0, 0, 0, time.UTC)
+ nh := hostclaim.NewHolder{Family: "instance", Key: "apps/bab/jam/instances/" + host + ".json", Account: "bab", App: "jam", Host: host, Region: "ie", RecordRegion: "ie", CreatedAt: created}
+ body := []byte(`{"host":"` + host + `","account":"bab","app":"jam","region":"ie","created_at":"2026-10-04T01:00:00Z"}`)
+ o, err := hostclaim.Read(ctx, f.s, host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ o = f.step(o, hostclaim.BeginCreateOn(nh, body, "c1", "console", created, hostclaim.ExecutorSelection{Box: "src", Enrollment: enrollment("src")}))
+ o = f.step(o, hostclaim.CompleteCreate("c1", "src"))
+ f.step(o, hostclaim.AdmitWriter(hostclaim.Admission{Box: "src", Mode: "cold", NewGeneration: "g1", Region: "ie", KeepStamp: true}))
+ return f
+}
+
+func (f *fixture) writeRoster() {
+ f.t.Helper()
+ raw, err := release.SignRootEnvelope(f.sign, f.roster)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ _, etag, err := f.s.Get(context.Background(), hostactors.Key)
+ if err != nil {
+ etag = ""
+ }
+ if _, err := f.s.Put(context.Background(), hostactors.Key, raw, etag); err != nil {
+ f.t.Fatal(err)
+ }
+}
+
+func (f *fixture) step(o hostclaim.Observed, tr func(*hostclaim.Claim) error) hostclaim.Observed {
+ f.t.Helper()
+ next, err := hostclaim.Update(context.Background(), f.s, o, tr)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ return next
+}
+
+func (f *fixture) claim() hostclaim.Observed {
+ f.t.Helper()
+ o, err := hostclaim.Read(context.Background(), f.s, host)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ return o
+}
+
+// participant applies a box-side transition (Plan 6b's half) to the claim.
+func (f *fixture) participant(tr func(*hostclaim.Claim) error) {
+ f.t.Helper()
+ f.step(f.claim(), tr)
+}
+
+func (f *fixture) ack(box string) {
+ f.participant(hostclaim.AcknowledgeExecution(box, hostclaim.ExecutionAcknowledgement{Op: f.claim().Claim.Op.ID, Enrollment: enrollment(box), BootID: "boot", ControllerID: "ctl", LeaseETag: "lease", Expires: f.now.Add(time.Hour)}))
+}
+
+// terminate writes box's termination fence as hostfence.TerminationWorker
+// would, observed a minute ago by wrk.
+func (f *fixture) terminate(box string) {
+ f.t.Helper()
+ b := f.roster.Boxes[box]
+ rec := termfence.Record{V: 1, Box: box, Enrollment: b.Enrollment, Cloud: b.Cloud, State: "terminated", ObservedAt: f.now.Add(-time.Minute).Format(time.RFC3339Nano),
+ Via: termfence.ViaDescribe, RequestID: "req", WrittenBy: termfence.WrittenBy{Box: "wrk", Enrollment: enrollment("wrk")}}
+ raw, _ := json.Marshal(rec)
+ if _, err := f.s.Put(context.Background(), hostfence.FenceKey(b.Enrollment), raw, ""); err != nil {
+ f.t.Fatal(err)
+ }
+}
+
+func (f *fixture) worker(want error) hostclaim.Observed {
+ f.t.Helper()
+ o, err := f.w.Step(context.Background(), host)
+ if !errors.Is(err, want) {
+ f.t.Fatalf("step: %v, want %v", err, want)
+ }
+ return o
+}
+
+func (f *fixture) installAndProbe(target string) {
+ f.t.Helper()
+ c := f.claim().Claim
+ id, p := c.Op.ID, c.Op.Journal["prepared"]
+ f.worker(ErrNotYet) // P is not built yet
+ f.participant(func(c *hostclaim.Claim) error { c.Op.Journal["test.built"] = p; return nil })
+ if o := f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseInstalled {
+ f.t.Fatalf("the worker records installed once P verifies: %s", o.Claim.Op.Phase)
+ }
+ stamp := f.claim().Claim.Op.Journal["install_stamp"]
+ var n uint64
+ _ = json.Unmarshal([]byte(stamp), &n)
+ f.participant(hostclaim.RecordLocalInstall(id, hostclaim.LocalInstall{Box: target, Enrollment: enrollment(target), InstallStamp: n, FilesManifestVersion: 1, JournalSHA256: strings.Repeat("d", 64)}))
+ f.participant(hostclaim.ToProbing(id, f.now))
+ o := f.claim()
+ f.participant(hostclaim.AdmitWriter(hostclaim.Admission{ObservedOpID: id, ObservedPhase: hostclaim.PhaseProbing, ObservedSession: o.Claim.Holder.Session, ObservedGeneration: o.Claim.Holder.Generation,
+ Box: target, Mode: "cold", NewGeneration: p, KeepStamp: true, Region: "ie", PredecessorRegion: "ie"}))
+}
+
+func TestWorkerDrivesADrain(t *testing.T) {
+ f := newFixture(t)
+ o, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if o.Claim.Op.Intent["admission_id"] == "" || o.Claim.Op.Intent["directory_revision"] != "1" {
+ t.Fatalf("intent: %+v", o.Claim.Op.Intent)
+ }
+ f.worker(ErrNotYet) // no acknowledgements yet
+ f.ack("src")
+ f.ack("dst")
+ if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseReleasing {
+ t.Fatalf("phase %s", o.Claim.Op.Phase)
+ }
+ f.worker(ErrNotYet) // the source has not released
+ h := f.claim().Claim.Holder
+ f.participant(hostclaim.JournalRelease(o.Claim.Op.ID, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
+ if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhasePinning || o.Claim.Op.Exec != "none" {
+ t.Fatalf("pinning: %+v", o.Claim.Op)
+ }
+ if o = f.worker(nil); o.Claim.Op.Journal["pin"] == "" {
+ t.Fatal("no pin journaled")
+ }
+ if want := []string{"acquire", "pin", "release"}; !slices.Equal(f.leases, want) {
+ t.Fatalf("the pin was not selected and journaled under the location lease: %v", f.leases)
+ }
+ before := f.notified
+ if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseInstalling || o.Claim.Op.Exec != "dst" || f.notified != before+1 {
+ t.Fatalf("installing: %+v notified %d", o.Claim.Op, f.notified-before)
+ }
+ f.installAndProbe("dst")
+ f.worker(ErrNotYet) // probing, no probe result, within the deadline
+ o = f.claim()
+ f.participant(hostclaim.RecordProbe(o.Claim.Op.ID, hostclaim.Probe{Box: "dst", Session: o.Claim.Holder.Session}))
+ if o = f.worker(nil); o.Claim.Op.Kind != hostclaim.KindServing || o.Claim.Op.Exec != "dst" {
+ t.Fatalf("serving: %+v", o.Claim.Op)
+ }
+}
+
+func TestWorkerFailoverWaitsForThePinWindow(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonFailover, Source: "src", Target: "dst"}); !errors.Is(err, ErrRefused) {
+ t.Fatalf("failover with no fence: %v", err)
+ }
+ f.terminate("src")
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); !errors.Is(err, ErrRefused) {
+ t.Fatalf("drain of a fenced source: %v", err)
+ }
+ o, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonFailover, Source: "src", Target: "dst"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if o.Claim.Op.Exec != "none" || !slices.Contains(o.Claim.Lost, "src") {
+ t.Fatalf("failover start: %+v", o.Claim.Op)
+ }
+ f.ack("dst")
+ if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhasePinning {
+ t.Fatalf("phase %s", o.Claim.Op.Phase)
+ }
+ // The fence was observed a minute ago, so the 30-second window is open.
+ if o = f.worker(nil); o.Claim.Op.Journal["pin"] == "" {
+ t.Fatal("not pinned")
+ }
+}
+
+func TestWorkerPinsNoEarlierThanThirtySecondsAfterTheFence(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ f.terminate("src")
+ f.now = f.now.Add(-time.Minute + 10*time.Second) // ten seconds after the observation
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonFailover, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ f.ack("dst")
+ f.worker(nil) // pinning
+ f.worker(ErrNotYet)
+ f.now = f.now.Add(20 * time.Second)
+ if o := f.worker(nil); o.Claim.Op.Journal["pin"] == "" {
+ t.Fatal("not pinned at observation plus thirty seconds")
+ }
+}
+
+func TestWorkerConvertsADrainWhoseSourceIsTerminated(t *testing.T) {
+ f := newFixture(t)
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ f.ack("dst")
+ f.terminate("src")
+ o := f.worker(nil)
+ if o.Claim.Op.Phase != hostclaim.PhaseStarting || o.Claim.Op.Exec != "none" || slices.Contains(o.Claim.Op.Participants, "src") || !slices.Contains(o.Claim.Lost, "src") {
+ t.Fatalf("conversion: %+v lost %v", o.Claim.Op, o.Claim.Lost)
+ }
+ if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhasePinning {
+ t.Fatalf("the barrier now needs only the target: %s", o.Claim.Op.Phase)
+ }
+}
+
+func TestWorkerReplacesATerminatedTargetTwiceThenStops(t *testing.T) {
+ f := newFixture(t)
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ for _, tc := range []struct{ dead, next string }{{"dst", "spa"}, {"spa", "sp2"}} {
+ f.terminate(tc.dead)
+ o := f.worker(nil)
+ if o.Claim.Op.Intent["target_box"] != tc.next || !slices.Contains(o.Claim.Lost, tc.dead) {
+ t.Fatalf("replacement of %s: %+v", tc.dead, o.Claim.Op.Intent)
+ }
+ }
+ f.terminate("sp2")
+ o := f.worker(nil)
+ if o.Claim.Op.Journal["stuck"] == "" {
+ t.Fatalf("a third replacement must stop the operation: %+v", o.Claim.Op.Journal)
+ }
+ before := o.Claim.Gen
+ if o = f.worker(nil); o.Claim.Gen != before {
+ t.Fatal("a worker moved a stuck operation")
+ }
+}
+
+func TestWorkerFailsAProbePastItsDeadline(t *testing.T) {
+ f := newFixture(t)
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ f.ack("src")
+ f.ack("dst")
+ f.worker(nil)
+ h := f.claim().Claim.Holder
+ f.participant(hostclaim.JournalRelease(f.claim().Claim.Op.ID, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
+ f.worker(nil)
+ f.worker(nil)
+ f.worker(nil)
+ f.installAndProbe("dst")
+ f.now = f.now.Add(ProbeDeadline + time.Second)
+ if o := f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseProbeFailed {
+ t.Fatalf("phase %s", o.Claim.Op.Phase)
+ }
+ f.worker(ErrNotYet) // the failed target has not released
+}
+
+func TestWorkerRefusesWhenItIsNotAnActiveWorker(t *testing.T) {
+ f := newFixture(t)
+ b := f.roster.Boxes["wrk"]
+ b.Region = "uk"
+ f.roster.Boxes["wrk"] = b
+ f.roster.Revision++
+ f.writeRoster()
+ before := f.claim().ETag
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); !errors.Is(err, ErrNotWorker) {
+ t.Fatalf("start: %v", err)
+ }
+ if f.claim().ETag != before {
+ t.Fatal("a refused start wrote the claim")
+ }
+}
+
+func TestWorkerStartRefusesAnUpdateReservedForAParticipant(t *testing.T) {
+ f := newFixture(t)
+ f.roster.Activations["upd-1"] = hostactors.Reservation{ID: "upd-1", FloorSHA256: strings.Repeat("f", 64),
+ Actors: map[string]string{"box:dst": enrollment("dst")}, Releases: map[string]hostactors.Report{"box:dst": {Enrollment: enrollment("dst")}}}
+ f.roster.Revision++
+ f.writeRoster()
+ before := f.claim().ETag
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err == nil {
+ t.Fatal("started while an executable update is reserved for the target (H14)")
+ }
+ if f.claim().ETag != before {
+ t.Fatal("a refused start wrote the claim")
+ }
+}
+
+func TestWorkerTreatsAnUnreadableFenceAsAnErrorNeverAsNoFence(t *testing.T) {
+ f := newFixture(t)
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ f.ack("src")
+ f.ack("dst")
+ if _, err := f.s.Put(context.Background(), hostfence.FenceKey(enrollment("src")), []byte("{"), ""); err != nil {
+ t.Fatal(err)
+ }
+ before := f.claim().ETag
+ if _, err := f.w.Step(context.Background(), host); err == nil || errors.Is(err, ErrNotYet) {
+ t.Fatalf("an unreadable fence let the drain proceed: %v", err)
+ }
+ if f.claim().ETag != before {
+ t.Fatal("the claim moved")
+ }
+}
+
+func TestWorkerRefusesToJournalAPinAfterTheLeaseIsLost(t *testing.T) {
+ f := newFixture(t)
+ f.terminate("src")
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonFailover, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ f.ack("dst")
+ f.worker(nil) // pinning
+ f.loseLease = true
+ o := f.worker(ErrNotYet)
+ if o.Claim.Op.Journal["pin"] != "" {
+ t.Fatal("a pin was journaled after its lease was lost: a park could have retired its versions")
+ }
+}
+
+func TestWorkerStartRefusesAnIneligibleTargetAndANonRelocatableHost(t *testing.T) {
+ f := newFixture(t)
+ before := f.claim().ETag
+ f.ineligible["dst"] = true
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err == nil {
+ t.Fatal("started on a target the eligible set does not hold")
+ }
+ f.ineligible["dst"] = false
+ shape := errors.New("unit-backed route")
+ f.w.Relocatable = func(context.Context, hostclaim.Observed) error { return shape }
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); !errors.Is(err, shape) {
+ t.Fatalf("started a non-relocatable host: %v", err)
+ }
+ if f.claim().ETag != before {
+ t.Fatal("a refused start wrote the claim")
+ }
+}
+
+func TestWorkerRefusesToDrainAHostThatNeverHadAWriter(t *testing.T) {
+ f := newFixture(t)
+ // A second host created on src and never woken: session 0.
+ ctx := context.Background()
+ const quiet = "quiet.bab.oncarlos.com"
+ nh := hostclaim.NewHolder{Family: "instance", Key: "apps/bab/jam/instances/" + quiet + ".json", Account: "bab", App: "jam", Host: quiet, Region: "ie", RecordRegion: "ie", CreatedAt: time.Date(2026, 10, 4, 1, 0, 0, 0, time.UTC)}
+ body := []byte(`{"host":"` + quiet + `","account":"bab","app":"jam","region":"ie","created_at":"2026-10-04T01:00:00Z"}`)
+ o, err := hostclaim.Read(ctx, f.s, quiet)
+ if err != nil {
+ t.Fatal(err)
+ }
+ o = f.step(o, hostclaim.BeginCreateOn(nh, body, "c2", "console", f.now, hostclaim.ExecutorSelection{Box: "src", Enrollment: enrollment("src")}))
+ f.step(o, hostclaim.CompleteCreate("c2", "src"))
+ if _, err := f.w.Start(ctx, quiet, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); !errors.Is(err, ErrRefused) {
+ t.Fatalf("drained a never-woken host: %v", err)
+ }
+ f.terminate("src")
+ if _, err := f.w.Start(ctx, quiet, Request{Reason: hostclaim.ReasonFailover, Source: "src", Target: "dst"}); err != nil {
+ t.Fatalf("a failover of a never-woken host: %v", err)
+ }
+}
+
+func TestWorkerMarksALiveTargetsThirdReplacementStuck(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ // Two terminated targets use the replacements; the third target is
+ // alive but fails its probe, and its release names a fourth box: stuck.
+ for _, dead := range []string{"dst", "spa"} {
+ f.terminate(dead)
+ f.worker(nil)
+ }
+ f.ack("src")
+ f.ack("sp2")
+ f.worker(nil) // releasing
+ h := f.claim().Claim.Holder
+ id := f.claim().Claim.Op.ID
+ f.participant(hostclaim.JournalRelease(id, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
+ f.worker(nil) // pinning
+ f.worker(nil) // pin
+ f.worker(nil) // installing on sp2
+ f.installAndProbe("sp2")
+ f.now = f.now.Add(ProbeDeadline + time.Second)
+ if o := f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseProbeFailed {
+ t.Fatalf("phase %s", o.Claim.Op.Phase)
+ }
+ h = f.claim().Claim.Holder
+ f.participant(hostclaim.JournalRelease(id, hostclaim.Release{ReleasedBy: "sp2", Enrollment: enrollment("sp2"), Generation: h.Generation, Session: h.Session, FinalTXID: 41, ManifestSeq: 7, Receipt: "r2", ReceiptSHA256: strings.Repeat("d", 64)}))
+ if o := f.worker(nil); o.Claim.Op.Journal["stuck"] == "" || o.Claim.Op.Phase != hostclaim.PhaseProbeFailed {
+ t.Fatalf("a live target's third replacement must stop the operation: %+v", o.Claim.Op)
+ }
+}
+
+func TestWorkerMarksExhaustionStuckBeforeLookingForASpare(t *testing.T) {
+ // Two replacements used; no box is eligible for a third. The
+ // operation is stuck at once, not retried forever on "no eligible box".
+ f := newFixture(t)
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ for _, dead := range []string{"dst", "spa"} {
+ f.terminate(dead)
+ f.worker(nil)
+ }
+ f.w.Choose = func(context.Context, hostclaim.Observed, []string) (hostclaim.Target, error) {
+ return hostclaim.Target{}, errors.New("no eligible box")
+ }
+ f.terminate("sp2")
+ if o := f.worker(nil); o.Claim.Op.Journal["stuck"] == "" {
+ t.Fatalf("exhaustion with no spare was not stuck: %+v", o.Claim.Op.Journal)
+ }
+}
+
+func TestWorkerFencesAnExhaustedTargetThatDiedAfterItsLocalInstall(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ for _, dead := range []string{"dst", "spa"} {
+ f.terminate(dead)
+ f.worker(nil)
+ }
+ f.ack("src")
+ f.ack("sp2")
+ f.worker(nil) // releasing
+ h := f.claim().Claim.Holder
+ id := f.claim().Claim.Op.ID
+ f.participant(hostclaim.JournalRelease(id, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
+ f.worker(nil) // pinning
+ f.worker(nil) // pin
+ f.worker(nil) // installing on sp2
+ c := f.claim().Claim
+ f.participant(func(c *hostclaim.Claim) error { c.Op.Journal["test.built"] = c.Op.Journal["prepared"]; return nil })
+ f.worker(nil) // installed
+ var n uint64
+ _ = json.Unmarshal([]byte(f.claim().Claim.Op.Journal["install_stamp"]), &n)
+ f.participant(hostclaim.RecordLocalInstall(c.Op.ID, hostclaim.LocalInstall{Box: "sp2", Enrollment: enrollment("sp2"), InstallStamp: n, FilesManifestVersion: 1, JournalSHA256: strings.Repeat("d", 64)}))
+ f.terminate("sp2") // dead between installed and its probe
+ o := f.worker(nil)
+ if o.Claim.Op.Journal["stuck"] == "" || o.Claim.Op.Phase != hostclaim.PhasePinning || o.Claim.Op.Exec != "none" || !slices.Contains(o.Claim.Lost, "sp2") {
+ t.Fatalf("an exhausted dead target was not fenced into a stuck pinning: %+v", o.Claim.Op)
+ }
+ before := o.Claim.Gen
+ if o = f.worker(nil); o.Claim.Gen != before {
+ t.Fatal("a worker moved a stuck operation")
+ }
+}
+
+func TestWorkerMarksAnOverdueReleaseStuck(t *testing.T) {
+ f := newFixture(t)
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ f.ack("src")
+ f.ack("dst")
+ f.worker(nil) // releasing
+ f.now = f.now.Add(ReleaseDeadline)
+ f.worker(ErrNotYet) // at the deadline: still waiting
+ f.now = f.now.Add(time.Second)
+ o := f.worker(nil)
+ if o.Claim.Op.Journal["stuck"] == "" || o.Claim.Op.Phase != hostclaim.PhaseReleasing || o.Claim.Op.Exec != "src" {
+ t.Fatalf("a source that never released must leave the operation stuck on the source: %+v", o.Claim.Op)
+ }
+}
+
+// Round 5 finding 1: a spare that owes an older local retirement for the
+// host is never offered to the chooser; the worker replaces the dead
+// target with a clean spare (the claim refuses the other in its CAS).
+func TestWorkerNeverReplacesATargetWithABoxOwingALocalRetirement(t *testing.T) {
+ f := newFixture(t)
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ f.participant(func(c *hostclaim.Claim) error {
+ c.Obligations = append(c.Obligations, hostclaim.Obligation{ID: "old", Op: "old-op", Kind: "park-local", Box: "spa", Retirement: "r", Owner: "box:spa"})
+ return nil
+ })
+ f.terminate("dst")
+ o := f.worker(nil)
+ if o.Claim.Op.Intent["target_box"] != "sp2" || slices.Contains(o.Claim.Op.Participants, "spa") {
+ t.Fatalf("replacement: %+v participants %v", o.Claim.Op.Intent, o.Claim.Op.Participants)
+ }
+}
+
+// Two-box model finding F2: a pin that cannot show the platform files
+// never existed (StorePinner's ErrFilesUnevidenced) stops the operation
+// for an operator instead of retrying or pinning empty.
+func TestWorkerMarksAnUnevidencedFilesPinStuck(t *testing.T) {
+ f := newFixture(t)
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ f.ack("src")
+ f.ack("dst")
+ f.worker(nil) // releasing
+ h := f.claim().Claim.Holder
+ f.participant(hostclaim.JournalRelease(f.claim().Claim.Op.ID, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
+ f.worker(nil) // pinning
+ f.pinErr = fmt.Errorf("%w: test", ErrFilesUnevidenced)
+ if o := f.worker(nil); o.Claim.Op.Journal["stuck"] == "" || o.Claim.Op.Journal["pin"] != "" {
+ t.Fatalf("an unevidenced files pin: %+v", o.Claim.Op)
+ }
+}
+
+// Two-box model finding F1: a drain's target fails its probe and
+// releases; with no spare (two boxes: the source is excluded) the worker
+// waits, then, ReleaseDeadline after the target failed, marks the
+// operation stuck for an operator.
+func TestWorkerBoundsAProbeFailedWithNoSpare(t *testing.T) {
+ for _, terminated := range []bool{false, true} {
+ t.Run(fmt.Sprintf("target terminated after its release %v", terminated), func(t *testing.T) { boundsAProbeFailed(t, terminated) })
+ }
+}
+
+func boundsAProbeFailed(t *testing.T, terminated bool) {
+ f := newFixture(t)
+ ctx := context.Background()
+ if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ f.ack("src")
+ f.ack("dst")
+ f.worker(nil) // releasing
+ h := f.claim().Claim.Holder
+ id := f.claim().Claim.Op.ID
+ f.participant(hostclaim.JournalRelease(id, hostclaim.Release{ReleasedBy: "src", Enrollment: enrollment("src"), Generation: h.Generation, Session: h.Session, FinalTXID: 40, ManifestSeq: 7, Receipt: "r", ReceiptSHA256: strings.Repeat("d", 64)}))
+ f.worker(nil) // pinning
+ f.worker(nil) // pin
+ f.worker(nil) // installing on dst
+ f.installAndProbe("dst")
+ f.now = f.now.Add(ProbeDeadline + time.Second)
+ if o := f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseProbeFailed {
+ t.Fatalf("phase %s", o.Claim.Op.Phase)
+ }
+ h = f.claim().Claim.Holder
+ f.participant(hostclaim.JournalRelease(id, hostclaim.Release{ReleasedBy: "dst", Enrollment: enrollment("dst"), Generation: h.Generation, Session: h.Session, FinalTXID: 41, ManifestSeq: 7, Receipt: "r2", ReceiptSHA256: strings.Repeat("d", 64)}))
+ noSpare := errors.New("no eligible box")
+ f.w.Choose = func(context.Context, hostclaim.Observed, []string) (hostclaim.Target, error) {
+ return hostclaim.Target{}, noSpare
+ }
+ if _, err := f.w.Step(ctx, host); !errors.Is(err, noSpare) || f.claim().Claim.Op.Journal["stuck"] != "" {
+ t.Fatalf("before the deadline: %v %+v", err, f.claim().Claim.Op.Journal)
+ }
+ if terminated {
+ // Round 17 finding 2: the released target is terminated. Its
+ // fence goes to the replacement path first; with no spare that
+ // path is bounded too, and consumes the fence as it stops.
+ f.terminate("dst")
+ if _, err := f.w.Step(ctx, host); !errors.Is(err, noSpare) || f.claim().Claim.Op.Journal["stuck"] != "" {
+ t.Fatalf("a dead target, before the deadline: %v %+v", err, f.claim().Claim.Op.Journal)
+ }
+ }
+ f.now = f.now.Add(ReleaseDeadline + time.Second)
+ o := f.worker(nil)
+ if o.Claim.Op.Journal["stuck"] == "" {
+ t.Fatalf("no spare after the deadline: %+v", o.Claim.Op)
+ }
+ if terminated && (!slices.Contains(o.Claim.Lost, "dst") || o.Claim.Op.Journal["fence.dst"] == "" || o.Claim.Op.Exec != "none") {
+ t.Fatalf("the dead target's fence was not consumed: %+v lost %v", o.Claim.Op, o.Claim.Lost)
+ }
+ if !terminated && o.Claim.Op.Phase != hostclaim.PhaseProbeFailed {
+ t.Fatalf("a live failed target: %+v", o.Claim.Op)
+ }
+}
+
+// preFlip is the moves door's durable pre-flip decision as
+// ConvertMovingToRelocating reads it: the record is still in ie.
+type preFlip struct{}
+
+func (preFlip) CheckMovingBeforeFlip(context.Context, hostclaim.Observed, string, hostclaim.FenceRef) (string, error) {
+ return "ie", nil
+}
+
+// Task 1's fix: a move converted to relocating binds fence.<source> with
+// no observation time, and JournalPin refuses until FenceSource has
+// consumed it. The worker reads the converted source as not yet lost, so
+// its first CAS is FenceSource; the pin then waits FailoverPinDelay past
+// the fence's observed_at, under the location lease.
+func TestWorkerFencesAConvertedMovesSourceBeforeItPins(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ far := strings.Repeat("9", 32) // the move's target, enrolled in uk
+ o := f.step(f.claim(), hostclaim.BeginAdmittedMove("mv1", "member", "src", "far", "uk", f.now,
+ hostclaim.MoveAdmission{AppID: "app", AdmissionID: "mv1", SourceEnrollment: enrollment("src"), TargetEnrollment: far}))
+ // The move's barrier, as the move door's boxes left it.
+ f.ack("src")
+ f.participant(hostclaim.AcknowledgeExecution("far", hostclaim.ExecutionAcknowledgement{Op: "mv1", Enrollment: far, BootID: "boot", ControllerID: "ctl", LeaseETag: "lease", Expires: f.now.Add(time.Hour)}))
+ f.terminate("src")
+ f.now = f.now.Add(-time.Minute + 10*time.Second) // ten seconds after the observation
+ ro, err := f.w.Directory.Read(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ref, _, err := f.w.fence(ctx, ro, "src")
+ if err != nil || ref == nil {
+ t.Fatalf("fence: %v %v", ref, err)
+ }
+ if o, err = hostclaim.ConvertMovingToRelocating(ctx, f.s, f.claim(), "mv1", *ref, f.w.Fences, preFlip{}); err != nil {
+ t.Fatal(err)
+ }
+ if o.Claim.Op.Journal["fence.src"] == "" || o.Claim.Op.Journal["fence.src.observed_at"] != "" || slices.Contains(o.Claim.Lost, "src") {
+ t.Fatalf("not the converted shape this test is about: %+v lost %v", o.Claim.Op.Journal, o.Claim.Lost)
+ }
+ o = f.worker(nil)
+ if o.Claim.Op.Journal["fence.src.observed_at"] == "" || !slices.Contains(o.Claim.Lost, "src") || o.Claim.Op.Phase != hostclaim.PhasePinning || o.Claim.Op.Journal["pin"] != "" {
+ t.Fatalf("the worker's first CAS on a converted move is FenceSource: %+v lost %v", o.Claim.Op, o.Claim.Lost)
+ }
+ if o = f.worker(nil); o.Claim.Op.Intent["target_box"] != "dst" {
+ t.Fatalf("the converted move names its same-region target: %+v", o.Claim.Op.Intent)
+ }
+ f.worker(ErrNotYet) // inside the 30-second hold
+ if len(f.leases) != 0 {
+ t.Fatalf("a lease was taken inside the hold: %v", f.leases)
+ }
+ f.now = f.now.Add(20 * time.Second)
+ if o = f.worker(nil); o.Claim.Op.Journal["pin"] == "" {
+ t.Fatal("not pinned at observation plus thirty seconds")
+ }
+ if want := []string{"acquire", "pin", "release"}; !slices.Equal(f.leases, want) {
+ t.Fatalf("the pin was not selected and journaled under the location lease: %v", f.leases)
+ }
+ f.worker(ErrNotYet) // the named target has not acknowledged
+ f.ack("dst")
+ if o = f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseInstalling || o.Claim.Op.Exec != "dst" {
+ t.Fatalf("installing: %+v", o.Claim.Op)
+ }
+}