amadan

carlos / platform Public

offline-move: landing, open and the target's revert

The target box's step of the offline operator move (plan Task 5).

The landing is a list of ensures judged from disk, registry and bucket
against the immutable seal.json (its sha256 checked against the record's
evidence): row, warm-cache, key-adoption, lease, db, files, config, env,
env-pin, timers, next-due, replication, release. A matching item is left
alone, an absent one landed, a different one parked into target-before and
landed again, so a pass at any point (after any item, after the CAS, after
a restart) neither re-parks a correct item nor fails on its own extraction.
The row stays held landing; nothing starts or takes the target's lease.

Open re-verifies every landing item against the seal and only then releases
the hold under reconcileMu and the hold lock; ordinary adoption stages the
build and the first wake promotes moves/<host>/. The archive prefix is
deleted and the box reports opened.

The revert deletes the held row (DeleteIfHold, timers included) under
reconcileMu and parks the landed files, env and config.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude c1f27ca2fcffbeb712d68781188a3ed8d49666f0 parent 2639808
2 files changed, +2452 −16
  • internal/host/movesoffline_land.go +1186 −16
  • internal/host/movesoffline_land_test.go +1266 −0
diff --git a/internal/host/movesoffline_land.go b/internal/host/movesoffline_land.go
index d3ea217f5..ba6a11da2 100644
--- a/internal/host/movesoffline_land.go
+++ b/internal/host/movesoffline_land.go
@@ -1,30 +1,1200 @@
package host
import (
+ "bytes"
"context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log"
+ "os"
+ "path"
+ "path/filepath"
+ "sort"
+ "strings"
+ "time"
+ "github.com/superfly/ltx"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/instancekey"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
-// offlineTargetStep is the target box's part of an offline move: the
-// landing, open and the target's revert
-// (docs/superpowers/plans/2026-10-20-offline-operator-move.md, Task 5, which
-// owns this file). offlineRound calls it on a box whose placement equals
-// rec.Move.To, in every offline state, after the source and sealer steps.
-// rec carries an offline move (rec.Move.Offline()) and the box is capable
-// (moveCfg.OfflineCapable). Reconcile mints the row held "landing" from the
-// record alone (offlineHold) and keeps it until the move is cleared
-// (offlineView.keep); ending that hold (open, revert) must take
-// provision.WithHostHoldLock around the release. Every step must be an
-// ensure; an error is logged by the pass, throttled per host, and the next
-// pass runs the step again.
+// The target box's part of an offline move
+// (docs/superpowers/plans/2026-10-20-offline-operator-move.md, Task 5): the
+// landing, open on the target, and the target's revert. offlineRound calls
+// offlineTargetStep on a box whose placement equals rec.Move.To, in every
+// offline state, after the source and sealer steps.
//
-// The landing compares against the immutable seal record
-// (moves.OfflineSealKey(host, id), added on Task 6's branch), read with
-// s.Get; never against the sealer's per-pass report, which keeps changing.
+// The landing is a list of ensures (Fable's B2). Every item is judged from
+// what is on disk, in the registry and in the bucket, compared with the
+// immutable seal (seal.json, moves.OfflineSealKey), never with the sealer's
+// living report: an item that matches is OK and left alone, an item that is
+// absent is landed, and an item that is present and different is renamed into
+// the move's target-before park directory and landed again. So a pass may run
+// at any point, after any item, after the CAS or after an agent restart, and
+// it neither parks a correct item nor fails on its own earlier extraction.
//
-// Task 3 leaves it a no-op.
+// Nothing here starts the instance or takes its lease. The row stays held
+// "landing" until open; open re-verifies every item against the seal and only
+// then releases the hold, and the build arrives the way every new instance's
+// does, through ordinary adoption and root's existing stage pass. The first
+// request promotes the staged replica (ExecBackend.restorePlain).
+
+// offlineTargetStep is the target step of one offline state. Every step is an
+// ensure: an error is logged by the pass, throttled per host, and the next
+// pass runs the step again.
func (w *movesWorker) offlineTargetStep(ctx context.Context, s release.Store, rec provision.Record) error {
+ if w.cfg.Box == "" {
+ return errors.New("offline move: this box has no label to report under")
+ }
+ switch rec.Move.State {
+ case provision.OfflineLanding, provision.OfflineLanded:
+ return w.offlineLand(ctx, s, rec)
+ case provision.OfflineOpen:
+ return w.offlineTargetOpen(ctx, s, rec)
+ case provision.OfflineReverting:
+ return w.offlineTargetRevert(ctx, s, rec)
+ }
+ return nil
+}
+
+// offlineLand is the landing pass in offline-landing and offline-landed:
+// ensure every item, rewrite the target report, and, once every item is OK
+// in offline-landing, move the record to offline-landed.
+func (w *movesWorker) offlineLand(ctx context.Context, s release.Store, rec provision.Record) error {
+ m := rec.Move
+ if rec.Region != m.To {
+ return nil // the flip has not reached this record yet
+ }
+ r, ok, err := w.landReport(ctx, s, rec)
+ if err != nil {
+ return err
+ }
+ if err := writeOfflineReport(ctx, s, r); err != nil {
+ return err
+ }
+ if err := w.landCrash("report"); err != nil {
+ return err
+ }
+ if !ok {
+ _, bad := r.ItemsOK(moves.OfflineOwnerLanding)
+ w.notice(rec.Host+"#offline-land", "moves: %s: offline move %s: landing, not yet OK: %s", rec.Host, m.ID, strings.Join(bad, ", "))
+ return nil
+ }
+ if m.State != provision.OfflineLanding {
+ return nil
+ }
+ if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.OfflineLandedBy(m.ID, m.To)); err != nil {
+ if errors.Is(err, provision.ErrMoveConflict) {
+ log.Printf("moves: %s: offline move %s: not marked landed: %v", rec.Host, m.ID, err)
+ return nil
+ }
+ return err
+ }
+ log.Printf("moves: %s: offline move %s: landed; every item is OK", rec.Host, m.ID)
+ return w.landCrash("landed")
+}
+
+// landCrash is the landing's test seam (moveCfg.landFault, the landing's
+// crash points): an error stands for a crash at point, after the item of
+// that name was ensured ("report" after the report is written, "landed"
+// after the CAS). Nil in production.
+func (w *movesWorker) landCrash(point string) error {
+ if w.cfg.landFault == nil {
+ return nil
+ }
+ return w.cfg.landFault("offline-" + point)
+}
+
+// offlineTargetOpen is the target step at offline-open. While the row is
+// still held by the move, every landing item is ensured and verified against
+// the seal once more, and only when all are OK is the hold released (under
+// reconcileMu and the host's hold lock, as the source's revert releases its
+// own). The row then reads ready: ordinary adoption stages the channel's
+// build, and the first request promotes moves/<host>/. Then the move's
+// archive prefix is deleted from this box's bucket and the box reports
+// opened. A pass after the release finds the row ready and does not verify
+// again (the staging may already be promoted and gone).
+func (w *movesWorker) offlineTargetOpen(ctx context.Context, s release.Store, rec provision.Record) error {
+ cfg := w.cfg
+ m, host := rec.Move, rec.Host
+ rt, err := w.offlineRow(host)
+ if err != nil {
+ return err
+ }
+ prev := w.prevTargetReport(ctx, s, host, m.ID)
+ items := map[string]moves.Check(nil)
+ if prev != nil {
+ items = prev.Items
+ }
+ switch {
+ case rt == nil:
+ return fmt.Errorf("offline move %s: %s: no row on the target at open", m.ID, host)
+ case rt.HoldOp == m.ID && registry.IsHold(rt.Status):
+ r, ok, err := w.landReport(ctx, s, rec)
+ if err != nil {
+ return err
+ }
+ if !ok {
+ if err := writeOfflineReport(ctx, s, r); err != nil {
+ return err
+ }
+ _, bad := r.ItemsOK(moves.OfflineOwnerLanding)
+ return fmt.Errorf("offline move %s: %s: not opening: %s not OK on this box", m.ID, host, strings.Join(bad, ", "))
+ }
+ items = r.Items
+ if cfg.ReleaseHold == nil {
+ return errors.New("offline move: no registry to release the hold through")
+ }
+ var relErr error
+ reconcileMu.Lock()
+ provision.WithHostHoldLock(host, func() {
+ _, relErr = cfg.ReleaseHold(host, m.ID)
+ })
+ reconcileMu.Unlock()
+ if relErr != nil {
+ return relErr
+ }
+ if rt, err = w.offlineRow(host); err != nil {
+ return err
+ }
+ if rt == nil || registry.IsHold(rt.Status) {
+ return fmt.Errorf("offline move %s: %s: the hold was not released", m.ID, host)
+ }
+ log.Printf("moves: %s: offline move %s: opened; the row is ready and adoption stages its build", host, m.ID)
+ case rt.HoldOp != "" || registry.IsHold(rt.Status):
+ return fmt.Errorf("offline move %s: %s is held %s by %q, not by this move", m.ID, host, rt.Status, rt.HoldOp)
+ case prev == nil:
+ // A ready row this box never landed (reconcile mints a missing
+ // row ready once the record is open): this pass did not release
+ // it and will not vouch for it.
+ return fmt.Errorf("offline move %s: %s is ready on this box, which has no landing report for the move", m.ID, host)
+ }
+ if prev == nil || prev.Phase != moves.OfflinePhaseOpened {
+ st, err := w.landStore(ctx)
+ if err != nil {
+ return err
+ }
+ if err := lease.DeletePrefix(ctx, st, cfg.OwnBucket, moves.OfflineArchivePrefix(host, m.ID)); err != nil {
+ return fmt.Errorf("offline move %s: delete the archive: %w", m.ID, err)
+ }
+ }
+ return writeOfflineReport(ctx, s, moves.OfflineReport{
+ Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: moves.OfflinePhaseOpened, At: cfg.now().UTC(),
+ Items: landingItemsOnly(items), RowPresent: rt != nil, Files: isDir(filepath.Join(cfg.DataDir, host)), Env: lexists(filepath.Join(cfg.DataDir, host+".env")),
+ })
+}
+
+// offlineTargetRevert is the target step at offline-reverting: when this box
+// holds the row for the move, delete it (DeleteIfHold, which takes its
+// timers with it) under reconcileMu and the hold lock, so a reconcile pass
+// that listed the record at landing cannot mint it held again after the
+// delete; then rename what the landing extracted and rendered (the files
+// directory, <host>.env and <host>.config.env) into the move's target park
+// directory, and report reverted. The source is untouched, so the revert
+// loses nothing. A box that never landed anything (no row held by the move
+// and no report of its own) does nothing.
+func (w *movesWorker) offlineTargetRevert(ctx context.Context, s release.Store, rec provision.Record) error {
+ cfg := w.cfg
+ m, host := rec.Move, rec.Host
+ rt, err := w.offlineRow(host)
+ if err != nil {
+ return err
+ }
+ prev := w.prevTargetReport(ctx, s, host, m.ID)
+ held := rt != nil && rt.HoldOp == m.ID && registry.IsHold(rt.Status)
+ if rt != nil && !held {
+ // Not this move's row (the move holds its row until open, and an
+ // open move is never aborted): leave it and what it serves.
+ return fmt.Errorf("offline move %s: %s is %q held by %q on this box, not by the move; the revert leaves it", m.ID, host, rt.Status, rt.HoldOp)
+ }
+ if !held && prev == nil {
+ return nil
+ }
+ if held {
+ if cfg.DeleteIfHold == nil {
+ return errors.New("offline move: no registry to delete the held row through")
+ }
+ var delErr error
+ reconcileMu.Lock()
+ provision.WithHostHoldLock(host, func() {
+ _, delErr = cfg.DeleteIfHold(host, m.ID)
+ })
+ reconcileMu.Unlock()
+ if delErr != nil {
+ return delErr
+ }
+ if rt, err = w.offlineRow(host); err != nil {
+ return err
+ }
+ if rt != nil && rt.HoldOp == m.ID {
+ return fmt.Errorf("offline move %s: %s is still held by the move", m.ID, host)
+ }
+ log.Printf("moves: %s: offline move %s: the target's held row is deleted", host, m.ID)
+ }
+ if prev != nil {
+ parkDir := moves.OfflineParkDir(cfg.DataDir, host, m.ID, moves.OfflineRoleTarget)
+ if parkDir == "" {
+ return fmt.Errorf("offline move %s: no park directory for %s under %q", m.ID, host, cfg.DataDir)
+ }
+ for _, p := range []string{filepath.Join(cfg.DataDir, host), filepath.Join(cfg.DataDir, host+".env"), filepath.Join(cfg.DataDir, host+".config.env")} {
+ if err := parkInto(p, parkDir); err != nil {
+ return fmt.Errorf("offline move %s: park %s: %w", m.ID, p, err)
+ }
+ }
+ }
+ return writeOfflineReport(ctx, s, moves.OfflineReport{
+ Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC(),
+ RowPresent: rt != nil, Files: isDir(filepath.Join(cfg.DataDir, host)), Env: lexists(filepath.Join(cfg.DataDir, host+".env")),
+ })
+}
+
+// parkInto renames p into dir under its own name (parkName numbers it when
+// the name is taken). An absent p is nothing to do.
+func parkInto(p, dir string) error {
+ if !lexists(p) {
+ return nil
+ }
+ if err := os.MkdirAll(dir, 0o700); err != nil {
+ return err
+ }
+ dst := parkName(dir, filepath.Base(p))
+ if err := os.Rename(p, dst); err != nil {
+ return err
+ }
+ log.Printf("moves: parked %s at %s", p, dst)
return nil
}
+
+// landingItemsOnly keeps the items a target report may carry.
+func landingItemsOnly(in map[string]moves.Check) map[string]moves.Check {
+ if len(in) == 0 {
+ return nil
+ }
+ out := map[string]moves.Check{}
+ for id, c := range in {
+ if owner, ok := moves.OfflineItemOwner(id); ok && owner == moves.OfflineOwnerLanding {
+ out[id] = c
+ }
+ }
+ return out
+}
+
+// prevTargetReport is this box's last target report for the move, nil when
+// there is none or it cannot be trusted.
+func (w *movesWorker) prevTargetReport(ctx context.Context, s release.Store, host, op string) *moves.OfflineReport {
+ key := moves.OfflineReportKey(host, op, moves.OfflineRoleTarget, w.cfg.Box)
+ if key == "" {
+ return nil
+ }
+ body, _, err := s.Get(ctx, key)
+ if err != nil {
+ return nil
+ }
+ var r moves.OfflineReport
+ if json.Unmarshal(body, &r) != nil || moves.ValidateOfflineReport(r) != nil || r.Host != host || r.Op != op || r.Role != moves.OfflineRoleTarget {
+ return nil
+ }
+ return &r
+}
+
+// landStore is this box's own data bucket (moveCfg.Mirror for its own
+// region): the staged replica, the archive and the timers snapshot.
+func (w *movesWorker) landStore(ctx context.Context) (lease.MirrorStore, error) {
+ if w.cfg.Mirror == nil || w.cfg.OwnBucket == "" {
+ return nil, errors.New("offline move: no data bucket on this box")
+ }
+ return w.mirror(ctx, w.cfg.OwnBucketRegion)
+}
+
+// landRun is one landing pass's state.
+type landRun struct {
+ w *movesWorker
+ s release.Store
+ rec provision.Record
+ m *provision.MoveState
+ host string
+ seal moves.OfflineReport
+ rt *registry.Route
+ prev *moves.OfflineReport
+
+ items map[string]moves.Check
+ hard bool // an item failed in a way the next pass will not clear by itself
+
+ st lease.MirrorStore
+ arch *offlineArchive
+ archErr error
+ archHard bool
+}
+
+// offlineArchive is the move's archive in this box's bucket, read once per
+// pass and only when an item needs it.
+type offlineArchive struct {
+ manifest archive.Manifest
+ sealed []byte
+ timers *moves.TimersSnapshot
+}
+
+func (r *landRun) cfg() moveCfg { return r.w.cfg }
+
+func (r *landRun) ok(id string) { r.items[id] = moves.Check{OK: true} }
+
+func (r *landRun) okBecause(id, why string) { r.items[id] = moves.Check{OK: true, Reason: why} }
+
+// wait records an item not yet OK that a later pass may make OK.
+func (r *landRun) wait(id, format string, args ...any) {
+ r.items[id] = moves.Check{Reason: fmt.Sprintf(format, args...)}
+}
+
+// fail records a mismatch: the report's phase becomes failed.
+func (r *landRun) fail(id, format string, args ...any) {
+ r.items[id] = moves.Check{Reason: fmt.Sprintf(format, args...)}
+ r.hard = true
+}
+
+func (r *landRun) isOK(id string) bool { return r.items[id].OK }
+
+func (r *landRun) filesDir() string { return filepath.Join(r.cfg().DataDir, r.host) }
+func (r *landRun) envPath() string { return filepath.Join(r.cfg().DataDir, r.host+".env") }
+func (r *landRun) configPath() string { return filepath.Join(r.cfg().DataDir, r.host+".config.env") }
+func (r *landRun) beforeDir() string {
+ return moves.OfflineParkDir(r.cfg().DataDir, r.host, r.m.ID, moves.OfflineParkTargetBefore)
+}
+
+// landReport ensures every landing item and returns the target report with
+// its phase (landed, or failed on a mismatch), and whether every item is OK.
+// Items wait, never fail, for what a later pass can mend (an S3 error, a row
+// not minted yet, a render that could not run); they fail on a mismatch with
+// the seal.
+func (w *movesWorker) landReport(ctx context.Context, s release.Store, rec provision.Record) (moves.OfflineReport, bool, error) {
+ cfg := w.cfg
+ r := &landRun{w: w, s: s, rec: rec, m: rec.Move, host: rec.Host, items: map[string]moves.Check{}}
+ r.prev = w.prevTargetReport(ctx, s, r.host, r.m.ID)
+ rt, err := w.offlineRow(r.host)
+ if err != nil {
+ return moves.OfflineReport{}, false, err
+ }
+ r.rt = rt
+ r.row()
+ if r.isOK("row") && r.readSeal(ctx) {
+ // In this order: files before config and the env check (the key is
+ // read from both files), config before the env pin.
+ for _, step := range []struct {
+ id string
+ run func()
+ }{
+ {"warm-cache", r.warmCache},
+ {"key-adoption", r.keyAdoption},
+ {"lease", func() { r.lease(ctx) }},
+ {"db", func() { r.db(ctx) }},
+ {"files", func() { r.files(ctx) }},
+ {"config", func() { r.config(ctx) }},
+ {"env", r.env},
+ {"env-pin", r.envPin},
+ {"timers", func() { r.timers(ctx) }},
+ {"next-due", r.nextDue},
+ {"replication", r.replication},
+ {"release", func() { r.release(ctx) }},
+ } {
+ step.run()
+ if err := w.landCrash(step.id); err != nil {
+ return moves.OfflineReport{}, false, err
+ }
+ }
+ }
+ for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) {
+ if _, done := r.items[it.ID]; !done {
+ r.wait(it.ID, "not checked: %s", r.blocked())
+ }
+ }
+ phase := moves.OfflinePhaseLanded
+ if r.hard {
+ phase = moves.OfflinePhaseFailed
+ }
+ out := moves.OfflineReport{
+ Op: r.m.ID, Host: r.host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: phase, At: cfg.now().UTC(), Items: r.items,
+ Held: r.rt != nil && r.rt.HoldOp == r.m.ID && registry.IsHold(r.rt.Status), RowPresent: r.rt != nil,
+ LocalDB: r.anyWarm(), Files: isDir(r.filesDir()), Env: lexists(r.envPath()),
+ }
+ if r.rt != nil {
+ out.NextDue, out.Replication = r.rt.NextDue, r.rt.Replication
+ }
+ ok, _ := out.ItemsOK(moves.OfflineOwnerLanding)
+ return out, ok, nil
+}
+
+// blocked says why the items after a failed row or seal were not checked.
+func (r *landRun) blocked() string {
+ if !r.isOK("row") {
+ return "the row is not held for this move"
+ }
+ return "the seal could not be read"
+}
+
+// row: the reconciled row exists, held "landing" by this move, at flat paths.
+func (r *landRun) row() {
+ rt := r.rt
+ switch {
+ case rt == nil:
+ r.wait("row", "no row yet: reconcile mints it held")
+ case rt.HoldOp != "" && rt.HoldOp != r.m.ID:
+ r.fail("row", "held %s by %s, not by this move", rt.Status, rt.HoldOp)
+ case rt.HoldOp != r.m.ID || rt.Status != "landing":
+ r.wait("row", "the row is %q, not held landing by this move yet", rt.Status)
+ case provision.NestedPaths(r.cfg().DataDir, *rt):
+ r.fail("row", "the row's paths are nested; an offline move lands flat rows only")
+ default:
+ r.ok("row")
+ }
+}
+
+// readSeal reads seal.json and checks it against the record's evidence:
+// the key and the sha256 of the exact bytes the sealer wrote, once.
+func (r *landRun) readSeal(ctx context.Context) bool {
+ src := r.m.Source
+ if src == nil || src.Evidence == "" {
+ r.failAll("the record carries no seal evidence")
+ return false
+ }
+ key := moves.OfflineSealKey(r.host, r.m.ID)
+ raw, _, err := r.s.Get(ctx, key)
+ if errors.Is(err, os.ErrNotExist) {
+ r.failAll("the seal %s is missing", key)
+ return false
+ }
+ if err != nil {
+ r.waitAll("the seal could not be read: %v", err)
+ return false
+ }
+ if got := moves.OfflineSealEvidence(r.host, r.m.ID, raw); got != src.Evidence {
+ r.failAll("the seal's bytes are not the ones the record names (%s, recorded %s)", got, src.Evidence)
+ return false
+ }
+ var seal moves.OfflineReport
+ if err := json.Unmarshal(raw, &seal); err != nil {
+ r.failAll("the seal does not decode: %v", err)
+ return false
+ }
+ if err := moves.ValidateOfflineSeal(seal, r.host, r.m.ID); err != nil {
+ r.failAll("%v", err)
+ return false
+ }
+ r.seal = seal
+ return true
+}
+
+// failAll fails every item but row for a seal that cannot be used.
+func (r *landRun) failAll(format string, args ...any) {
+ for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) {
+ if it.ID != "row" {
+ r.fail(it.ID, format, args...)
+ }
+ }
+}
+
+func (r *landRun) waitAll(format string, args ...any) {
+ for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) {
+ if it.ID != "row" {
+ r.wait(it.ID, format, args...)
+ }
+ }
+}
+
+// warmPaths are the local database artifacts a box could trust as a warm
+// cache for the host: the row's database and the default flat one, each
+// with its WAL, SHM, both litestream meta directories and its seed marker.
+func (r *landRun) warmPaths() []string {
+ dbs := []string{provision.DefaultDBPath(r.cfg().DataDir, r.host)}
+ if r.rt != nil && r.rt.DBPath != "" && r.rt.DBPath != dbs[0] {
+ dbs = append(dbs, r.rt.DBPath)
+ }
+ var out []string
+ for _, db := range dbs {
+ out = append(out, db, db+"-wal", db+"-shm", db+"-litestream", activatorMetaDir(db), db+activator.SeedMarkerSuffix)
+ }
+ return out
+}
+
+func (r *landRun) anyWarm() bool {
+ for _, p := range r.warmPaths() {
+ if lexists(p) {
+ return true
+ }
+ }
+ return false
+}
+
+// ensureAbsent renames every present path into target-before and reports
+// the item from what is left.
+func (r *landRun) ensureAbsent(id string, paths []string) {
+ dir := r.beforeDir()
+ for _, p := range paths {
+ if !lexists(p) {
+ continue
+ }
+ if err := parkInto(p, dir); err != nil {
+ r.wait(id, "%s could not be parked: %v", p, err)
+ return
+ }
+ }
+ for _, p := range paths {
+ if lexists(p) {
+ r.wait(id, "%s is still present", p)
+ return
+ }
+ }
+ r.ok(id)
+}
+
+// warmCache: no local database or meta directory for the host. The target
+// never woke it, so anything here predates the move and would win over the
+// staged replica on the first wake.
+func (r *landRun) warmCache() { r.ensureAbsent("warm-cache", r.warmPaths()) }
+
+// keyAdoption: no instance-key adoption record (written only for claim
+// launches; never carried).
+func (r *landRun) keyAdoption() {
+ r.ensureAbsent("key-adoption", []string{filepath.Join(r.cfg().DataDir, r.host+instancekey.AdoptionSuffix)})
+}
+
+// lease: the target bucket's lock for the host is absent, a lifted
+// tombstone, or an ordinary lease that has expired. A tombstone of another
+// move is lifted. Nothing here ever takes the lease.
+func (r *landRun) lease(ctx context.Context) {
+ cfg := r.cfg()
+ loc := activator.HibernatePrefix(cfg.OwnPrefix, r.host)
+ _, owner, expires, err := lease.LockState(ctx, cfg.OwnBucket, loc, cfg.OwnBucketRegion)
+ if err != nil {
+ r.wait("lease", "the target's lock could not be read: %v", err)
+ return
+ }
+ switch {
+ case owner == "" && expires.IsZero():
+ r.ok("lease")
+ return
+ case lease.IsLiftedTombstone(owner, expires):
+ r.ok("lease")
+ return
+ case owner == moves.TombstoneOwner(r.m.ID):
+ r.fail("lease", "the target's lock is this move's own tombstone")
+ return
+ case !strings.HasPrefix(owner, lease.TombstonePrefix):
+ if time.Now().After(expires) {
+ r.ok("lease")
+ return
+ }
+ r.wait("lease", "a writer holds the target's lock: %s until %s", owner, expires.UTC().Format(time.RFC3339))
+ return
+ }
+ stale := owner
+ if _, err := lease.LiftTombstone(ctx, cfg.OwnBucket, loc, cfg.OwnBucketRegion, func(o string) bool { return o == stale }); err != nil {
+ r.wait("lease", "the stale tombstone %s could not be lifted: %v", stale, err)
+ return
+ }
+ _, owner, expires, err = lease.LockState(ctx, cfg.OwnBucket, loc, cfg.OwnBucketRegion)
+ if err != nil {
+ r.wait("lease", "the target's lock could not be read after the lift: %v", err)
+ return
+ }
+ if (owner == "" && expires.IsZero()) || lease.IsLiftedTombstone(owner, expires) {
+ log.Printf("moves: %s: offline move %s: lifted the stale tombstone %s in the target's bucket", r.host, r.m.ID, stale)
+ r.ok("lease")
+ return
+ }
+ r.wait("lease", "the target's lock is %s after the lift", owner)
+}
+
+func (r *landRun) store(ctx context.Context) (lease.MirrorStore, error) {
+ if r.st != nil {
+ return r.st, nil
+ }
+ st, err := r.w.landStore(ctx)
+ if err != nil {
+ return nil, err
+ }
+ r.st = st
+ return st, nil
+}
+
+// stagedFingerprint names one listing of the staged replica under one seal:
+// the db item is verified once per fingerprint.
+func stagedFingerprint(evidence string, staged map[string]moves.StagedObject) string {
+ keys := make([]string, 0, len(staged))
+ for k := range staged {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ h := sha256.New()
+ fmt.Fprintf(h, "%s\n", evidence)
+ for _, k := range keys {
+ fmt.Fprintf(h, "%s %d %s\n", k, staged[k].Size, staged[k].ETag)
+ }
+ return "sha256:" + hex.EncodeToString(h.Sum(nil))
+}
+
+// stagedDiff describes how got differs from want, "" when equal.
+func stagedDiff(want, got map[string]moves.StagedObject) string {
+ var missing, extra, changed []string
+ for k, w := range want {
+ g, ok := got[k]
+ switch {
+ case !ok:
+ missing = append(missing, k)
+ case g.Size != w.Size || g.ETag != w.ETag:
+ changed = append(changed, k)
+ }
+ }
+ for k := range got {
+ if _, ok := want[k]; !ok {
+ extra = append(extra, k)
+ }
+ }
+ var parts []string
+ for _, p := range []struct {
+ what string
+ keys []string
+ }{{"missing", missing}, {"extra", extra}, {"changed", changed}} {
+ if len(p.keys) == 0 {
+ continue
+ }
+ sort.Strings(p.keys)
+ more := ""
+ if len(p.keys) > 3 {
+ more = fmt.Sprintf(" and %d more", len(p.keys)-3)
+ p.keys = p.keys[:3]
+ }
+ parts = append(parts, p.what+" "+strings.Join(p.keys, ", ")+more)
+ }
+ return strings.Join(parts, "; ")
+}
+
+const dbVerifiedPrefix = "verified "
+
+// db: the staged replica under moves/<host>/ is exactly what the seal
+// staged (nothing consumes it before open), its newest TXID is the seal's
+// FinalTXID, and it restores clean (lease.VerifyReplica into a scratch
+// directory under the move's verify park directory, removed after). The
+// restore runs once per seal and listing: the fingerprint is remembered in
+// the item's reason, and a pass whose listing is unchanged does not restore
+// again.
+func (r *landRun) db(ctx context.Context) {
+ cfg := r.cfg()
+ st, err := r.store(ctx)
+ if err != nil {
+ r.wait("db", "%v", err)
+ return
+ }
+ prefix := activator.MovesPrefix(r.host)
+ objs, err := st.List(ctx, cfg.OwnBucket, prefix)
+ if err != nil {
+ r.wait("db", "the staged replica could not be listed: %v", err)
+ return
+ }
+ got := make(map[string]moves.StagedObject, len(objs))
+ var newest ltx.TXID
+ for k, o := range objs {
+ got[strings.TrimPrefix(k, prefix)] = moves.StagedObject{Size: o.Size, ETag: o.ETag}
+ if _, maxTX, err := ltx.ParseFilename(path.Base(k)); err == nil && maxTX > newest {
+ newest = maxTX
+ }
+ }
+ if len(r.seal.Staged) == 0 {
+ r.fail("db", "the seal staged nothing")
+ return
+ }
+ if diff := stagedDiff(r.seal.Staged, got); diff != "" {
+ r.fail("db", "the staged replica is not the one the seal staged: %s", diff)
+ return
+ }
+ if uint64(newest) != r.seal.FinalTXID {
+ r.fail("db", "the staged replica ends at TXID %d, the seal at %d", newest, r.seal.FinalTXID)
+ return
+ }
+ verified := dbVerifiedPrefix + stagedFingerprint(r.m.Source.Evidence, got)
+ if p := r.prev; p != nil {
+ if c, ok := p.Items["db"]; ok && c.OK && c.Reason == verified {
+ r.okBecause("db", verified)
+ return
+ }
+ }
+ dir := moves.OfflineParkDir(cfg.DataDir, r.host, r.m.ID, moves.OfflineParkVerify)
+ if dir == "" {
+ r.fail("db", "no verify directory for %s", r.host)
+ return
+ }
+ if err := os.MkdirAll(dir, 0o700); err != nil {
+ r.wait("db", "the verify directory: %v", err)
+ return
+ }
+ defer os.RemoveAll(dir)
+ if _, err := lease.VerifyReplica(ctx, cfg.OwnBucket, strings.TrimSuffix(prefix, "/"), cfg.OwnBucketRegion, dir); err != nil {
+ r.fail("db", "the staged replica does not restore clean: %v", err)
+ return
+ }
+ r.okBecause("db", verified)
+}
+
+// readArchive fetches the move's archive, its manifest and the timers
+// snapshot from this box's bucket, once per pass. The names are found by
+// listing the move's archive prefix: the one sealed object (".sealed"), the
+// manifest that names the seal's digests, and the snapshot of this host.
+func (r *landRun) readArchive(ctx context.Context) (*offlineArchive, bool, error) {
+ if r.arch != nil || r.archErr != nil {
+ return r.arch, r.archHard, r.archErr
+ }
+ r.arch, r.archHard, r.archErr = r.fetchArchive(ctx)
+ return r.arch, r.archHard, r.archErr
+}
+
+func (r *landRun) fetchArchive(ctx context.Context) (*offlineArchive, bool, error) {
+ cfg := r.cfg()
+ st, err := r.store(ctx)
+ if err != nil {
+ return nil, false, err
+ }
+ prefix := moves.OfflineArchivePrefix(r.host, r.m.ID)
+ objs, err := st.List(ctx, cfg.OwnBucket, prefix)
+ if err != nil {
+ return nil, false, fmt.Errorf("list the archive: %w", err)
+ }
+ keys := make([]string, 0, len(objs))
+ for k := range objs {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ out := &offlineArchive{}
+ var sealedKey string
+ var manifests int
+ for _, k := range keys {
+ switch {
+ case strings.HasSuffix(k, ".sealed"):
+ if sealedKey != "" {
+ return nil, true, fmt.Errorf("the archive prefix holds more than one sealed object (%s, %s)", sealedKey, k)
+ }
+ sealedKey = k
+ case strings.HasSuffix(k, ".json"):
+ body, err := st.Get(ctx, cfg.OwnBucket, k)
+ if err != nil {
+ return nil, false, fmt.Errorf("read %s: %w", k, err)
+ }
+ var am archive.Manifest
+ if json.Unmarshal(body, &am) == nil && am.Plain != "" && am.SealedDataKey != "" {
+ if am.Plain != r.seal.FilesPlain || (r.seal.FilesSealed != "" && am.Sealed != r.seal.FilesSealed) || am.Op != r.m.ID {
+ return nil, true, fmt.Errorf("the archive manifest %s is not the sealed one (plain %s, sealed %s, op %s)", k, am.Plain, am.Sealed, am.Op)
+ }
+ out.manifest = am
+ manifests++
+ continue
+ }
+ if snap, err := moves.DecodeTimers(body); err == nil && snap.Host == r.host {
+ if out.timers != nil {
+ return nil, true, fmt.Errorf("the archive prefix holds more than one timers snapshot")
+ }
+ out.timers = &snap
+ }
+ }
+ }
+ if sealedKey == "" || manifests != 1 {
+ return nil, true, fmt.Errorf("the archive prefix %s holds %d sealed object(s) and %d manifest(s), want one of each", prefix, boolInt(sealedKey != ""), manifests)
+ }
+ if out.sealed, err = st.Get(ctx, cfg.OwnBucket, sealedKey); err != nil {
+ return nil, false, fmt.Errorf("read %s: %w", sealedKey, err)
+ }
+ return out, false, nil
+}
+
+func boolInt(b bool) int {
+ if b {
+ return 1
+ }
+ return 0
+}
+
+// filesMatch reports whether the flat files directory and <host>.env are
+// the seal's, by the archive's own plaintext digest over the pair.
+func (r *landRun) filesMatch() (bool, string) {
+ if !isDir(r.filesDir()) {
+ return false, "the files directory is absent"
+ }
+ plain, _, _, err := archive.PlainDigest(r.filesDir(), r.envPath())
+ if err != nil {
+ return false, err.Error()
+ }
+ if plain != r.seal.FilesPlain {
+ return false, "digest " + plain
+ }
+ return true, ""
+}
+
+// files: <DataDir>/<host>/ and <host>.env carry the seal's plaintext digest.
+// OK: left alone. Absent: fetched, verified and extracted. Present and
+// different: the files directory (when not empty) and an env file that is
+// not the seal's are renamed into target-before first, then extracted.
+func (r *landRun) files(ctx context.Context) {
+ if ok, _ := r.filesMatch(); ok {
+ r.ok("files")
+ return
+ }
+ cfg := r.cfg()
+ if cfg.ConfigPriv == nil {
+ r.wait("files", "this box has no CARLOS_SECRETS_KEY to open the archive with")
+ return
+ }
+ a, hard, err := r.readArchive(ctx)
+ if err != nil {
+ if hard {
+ r.fail("files", "%v", err)
+ } else {
+ r.wait("files", "%v", err)
+ }
+ return
+ }
+ bind := appconfig.Binding{Account: r.rec.Account, App: r.rec.App}
+ if err := archive.Verify(bytes.NewReader(a.sealed), a.manifest, cfg.ConfigPriv, bind); err != nil {
+ r.fail("files", "the archive does not verify: %v", err)
+ return
+ }
+ if nonEmptyOrNotDir(r.filesDir()) {
+ if err := parkInto(r.filesDir(), r.beforeDir()); err != nil {
+ r.wait("files", "the files directory could not be parked: %v", err)
+ return
+ }
+ }
+ if lexists(r.envPath()) && !sameSHA(r.seal.EnvSHA256, hexFileSHA256(r.envPath())) {
+ if err := parkInto(r.envPath(), r.beforeDir()); err != nil {
+ r.wait("files", "the env file could not be parked: %v", err)
+ return
+ }
+ }
+ if err := extractSealed(a, cfg, bind, r.filesDir(), r.envPath()); err != nil {
+ r.wait("files", "the archive could not be extracted: %v", err)
+ return
+ }
+ log.Printf("moves: %s: offline move %s: extracted the files and env", r.host, r.m.ID)
+ if ok, why := r.filesMatch(); !ok {
+ r.fail("files", "after extraction the files are not the seal's (%s, sealed %s)", why, r.seal.FilesPlain)
+ return
+ }
+ r.ok("files")
+}
+
+// extractSealed opens the sealed archive and extracts it, as the Phase 1
+// install does (openAndExtract): Extract admits nothing it did not read
+// whole.
+func extractSealed(a *offlineArchive, cfg moveCfg, bind appconfig.Binding, filesDir, envPath string) error {
+ pr, pw := io.Pipe()
+ done := make(chan error, 1)
+ go func() {
+ err := archive.Open(bytes.NewReader(a.sealed), a.manifest.SealedDataKey, cfg.ConfigPriv, bind, pw)
+ pw.CloseWithError(err)
+ done <- err
+ }()
+ xerr := archive.Extract(pr, filesDir, envPath)
+ pr.CloseWithError(errors.New("extract stopped reading"))
+ if oerr := <-done; oerr != nil && xerr == nil {
+ xerr = oerr
+ }
+ return xerr
+}
+
+// nonEmptyOrNotDir reports a path Extract would refuse as its files
+// directory: present and not an empty directory.
+func nonEmptyOrNotDir(p string) bool {
+ st, err := os.Lstat(p)
+ if err != nil {
+ return false
+ }
+ if !st.IsDir() {
+ return true
+ }
+ ents, err := os.ReadDir(p)
+ return err != nil || len(ents) > 0
+}
+
+// hexFileSHA256 is the hex sha256 of a file's bytes, "" when it cannot be
+// read.
+func hexFileSHA256(p string) string {
+ b, err := os.ReadFile(p)
+ if err != nil {
+ return ""
+ }
+ sum := sha256.Sum256(b)
+ return hex.EncodeToString(sum[:])
+}
+
+// sameSHA reports whether a sealed sha256 names gotHex, the bare hex form a
+// box computes. The seal may carry it bare or as "sha256:<hex>"; "" names
+// "" only (nothing there).
+func sameSHA(want, gotHex string) bool {
+ if gotHex == "" {
+ return want == ""
+ }
+ return want == gotHex || want == "sha256:"+gotHex
+}
+
+// config: the held row's <host>.config.env rendered from the bucket bundle
+// (RenderHeld, which admits the row held by this move and writes a
+// move-bound ack). OK when the file holds the rendered bytes and this box's
+// ack names this move and those bytes. Rendered every pass, so a bundle
+// changed since the seal reaches the file and fails the env pin.
+func (r *landRun) config(ctx context.Context) {
+ cfg := r.cfg()
+ if cfg.RenderHeld == nil {
+ r.wait("config", "this box has no held render")
+ return
+ }
+ hr, err := cfg.RenderHeld(ctx, r.s, *r.rt, r.m.ID, nil, func(string) error { return nil })
+ if err != nil {
+ r.wait("config", "the config could not be rendered: %v", err)
+ return
+ }
+ if hr.Path != r.configPath() {
+ r.fail("config", "rendered at %s, not %s", hr.Path, r.configPath())
+ return
+ }
+ if hexFileSHA256(hr.Path) != hr.SHA256 {
+ r.wait("config", "%s does not hold the rendered bytes", hr.Path)
+ return
+ }
+ body, _, err := r.s.Get(ctx, appconfig.ConfigAckKey(cfg.Box, r.host))
+ if err != nil {
+ r.wait("config", "the config ack could not be read: %v", err)
+ return
+ }
+ var ack appconfig.ConfigAck
+ if err := json.Unmarshal(body, &ack); err != nil || ack.MoveOp != r.m.ID || ack.FileSHA256 != hr.SHA256 {
+ r.wait("config", "the config ack does not name this move and the rendered file")
+ return
+ }
+ r.ok("config")
+}
+
+// env: <host>.env is the source's byte for byte (the seal's sha256), and
+// the instance key the app would be handed, from the env file and the
+// rendered config as the activator merges them (instancekey.Effective), is
+// the source's (the seal's key sha256). Both travel in the archive, so the
+// files item lands them; this is the check.
+func (r *landRun) env() {
+ if !r.isOK("files") {
+ r.wait("env", "waits for files")
+ return
+ }
+ own, err := os.ReadFile(r.envPath())
+ switch {
+ case errors.Is(err, os.ErrNotExist) && r.seal.EnvSHA256 == "":
+ case err != nil:
+ r.fail("env", "%s: %v", r.envPath(), err)
+ return
+ default:
+ if sum := sha256.Sum256(own); !sameSHA(r.seal.EnvSHA256, hex.EncodeToString(sum[:])) {
+ r.fail("env", "%s is not the source's (sha256 %s, sealed %s)", r.envPath(), hex.EncodeToString(sum[:]), r.seal.EnvSHA256)
+ return
+ }
+ }
+ if !r.isOK("config") {
+ r.wait("env", "waits for config: the instance key is read from both files")
+ return
+ }
+ cfgEnv, err := os.ReadFile(r.configPath())
+ if err != nil && !errors.Is(err, os.ErrNotExist) {
+ r.wait("env", "%s: %v", r.configPath(), err)
+ return
+ }
+ key, _ := instancekey.Effective(instancekey.ParseEnv(cfgEnv), instancekey.ParseEnv(own))
+ got := ""
+ if key != "" {
+ got = instancekey.Digest(key)
+ }
+ if !sameSHA(r.seal.KeySHA256, got) {
+ r.fail("env", "the instance key is not the source's (sha256 %q, sealed %q)", got, r.seal.KeySHA256)
+ return
+ }
+ r.ok("env")
+}
+
+// envPin: the composed launch environment of the held row
+// (ExecBackend.LaunchLayers) has the source's portable digest. App settings
+// must be equal; box settings are left out (moves.Pinned).
+func (r *landRun) envPin() {
+ cfg := r.cfg()
+ if !r.isOK("files") || !r.isOK("config") {
+ r.wait("env-pin", "waits for files and config")
+ return
+ }
+ if r.m.Source.EnvDigest != "" && r.m.Source.EnvDigest != r.seal.EnvDigest {
+ r.fail("env-pin", "the record's env digest and the seal's disagree")
+ return
+ }
+ if cfg.LaunchLayers == nil {
+ r.wait("env-pin", "this box has no exec backend to compose the launch environment")
+ return
+ }
+ key, err := moves.ReportKeyFromSecrets(cfg.SecretsKey)
+ if err != nil {
+ r.wait("env-pin", "no report key: %v", err)
+ return
+ }
+ env, inst, _, err := cfg.LaunchLayers(*r.rt)
+ if err != nil {
+ r.wait("env-pin", "the launch environment could not be composed: %v", err)
+ return
+ }
+ if got := moves.PortableDigest(key, env, inst); got != r.seal.EnvDigest {
+ // The seal carries the source's digest only, not its per-variable
+ // report, so CompareEnv cannot name the variables here.
+ r.fail("env-pin", "the launch environment differs from the source's (%s, sealed %s)", got, r.seal.EnvDigest)
+ return
+ }
+ r.ok("env-pin")
+}
+
+// timers: the row's timer rows digest to the seal's. Otherwise they are
+// replaced from the snapshot beside the archive (its digest checked first)
+// and read back.
+func (r *landRun) timers(ctx context.Context) {
+ reg := r.cfg().Registry
+ if reg == nil {
+ r.wait("timers", "no registry on this box")
+ return
+ }
+ if r.m.Source.TimersSHA256 != "" && r.m.Source.TimersSHA256 != r.seal.TimersDigest {
+ r.fail("timers", "the record's timers digest and the seal's disagree")
+ return
+ }
+ rows, err := reg.TimersFor(r.host)
+ if err != nil {
+ r.wait("timers", "%v", err)
+ return
+ }
+ if moves.TimersDigest(rows) == r.seal.TimersDigest {
+ r.ok("timers")
+ return
+ }
+ a, hard, err := r.readArchive(ctx)
+ switch {
+ case err != nil && hard:
+ r.fail("timers", "%v", err)
+ return
+ case err != nil:
+ r.wait("timers", "%v", err)
+ return
+ case a.timers == nil:
+ r.fail("timers", "the archive prefix holds no timers snapshot for %s", r.host)
+ return
+ }
+ if got := moves.TimersDigest(a.timers.Rows); got != r.seal.TimersDigest {
+ r.fail("timers", "the timers snapshot digests to %s, the seal %s", got, r.seal.TimersDigest)
+ return
+ }
+ if err := reg.ReplaceTimers(r.host, a.timers.Rows); err != nil {
+ r.wait("timers", "replace the timers: %v", err)
+ return
+ }
+ if rows, err = reg.TimersFor(r.host); err != nil {
+ r.wait("timers", "%v", err)
+ return
+ }
+ if got := moves.TimersDigest(rows); got != r.seal.TimersDigest {
+ r.fail("timers", "the timers read back as %s, the seal %s", got, r.seal.TimersDigest)
+ return
+ }
+ r.ok("timers")
+}
+
+// reread refreshes the row after a column write.
+func (r *landRun) reread(id string) bool {
+ rt, err := r.w.offlineRow(r.host)
+ if err != nil || rt == nil {
+ r.wait(id, "the row could not be read back: %v", err)
+ return false
+ }
+ r.rt = rt
+ return true
+}
+
+// nextDue: the row's NextDue is the seal's; else set and read back.
+func (r *landRun) nextDue() {
+ if r.rt.NextDue == r.seal.NextDue {
+ r.ok("next-due")
+ return
+ }
+ reg := r.cfg().Registry
+ if reg == nil {
+ r.wait("next-due", "no registry on this box")
+ return
+ }
+ if err := reg.SetNextDue(r.host, r.seal.NextDue); err != nil {
+ r.wait("next-due", "%v", err)
+ return
+ }
+ if !r.reread("next-due") {
+ return
+ }
+ if r.rt.NextDue != r.seal.NextDue {
+ r.fail("next-due", "reads back %d, the seal %d", r.rt.NextDue, r.seal.NextDue)
+ return
+ }
+ r.ok("next-due")
+}
+
+// replication: the row's replication rung is the seal's; else set and read
+// back.
+func (r *landRun) replication() {
+ if r.rt.Replication == r.seal.Replication {
+ r.ok("replication")
+ return
+ }
+ reg := r.cfg().Registry
+ if reg == nil {
+ r.wait("replication", "no registry on this box")
+ return
+ }
+ if err := reg.SetReplication(r.host, r.seal.Replication); err != nil {
+ r.fail("replication", "%v", err)
+ return
+ }
+ if !r.reread("replication") {
+ return
+ }
+ if r.rt.Replication != r.seal.Replication {
+ r.fail("replication", "reads back %q, the seal %q", r.rt.Replication, r.seal.Replication)
+ return
+ }
+ r.ok("replication")
+}
+
+// release: the app's channel resolves to the build the seal recorded. A
+// promote during the move fails it; the build itself arrives after open
+// through ordinary adoption.
+func (r *landRun) release(ctx context.Context) {
+ cfg := r.cfg()
+ if r.m.Source.Release != "" && r.m.Source.Release != r.seal.Release {
+ r.fail("release", "the record's release and the seal's disagree")
+ return
+ }
+ if cfg.ChannelVersion == nil {
+ r.wait("release", "this box cannot resolve channels")
+ return
+ }
+ v, err := cfg.ChannelVersion(ctx, r.s, *r.rt)
+ if err != nil {
+ r.wait("release", "the channel could not be resolved: %v", err)
+ return
+ }
+ if v != r.seal.Release {
+ r.fail("release", "the channel points at %q, the seal recorded %q", v, r.seal.Release)
+ return
+ }
+ r.ok("release")
+}
diff --git a/internal/host/movesoffline_land_test.go b/internal/host/movesoffline_land_test.go
new file mode 100644
index 000000000..51593c609
--- /dev/null
+++ b/internal/host/movesoffline_land_test.go
@@ -0,0 +1,1266 @@
+package host
+
+import (
+ "bytes"
+ "context"
+ "crypto/ecdh"
+ "crypto/ed25519"
+ "crypto/md5"
+ "crypto/sha256"
+ "database/sql"
+ "encoding/hex"
+ "encoding/json"
+ "encoding/xml"
+ "errors"
+ "fmt"
+ "io"
+ "io/fs"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path"
+ "path/filepath"
+ "reflect"
+ "sort"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/superfly/ltx"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/instancekey"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The offline move's landing, open on the target and the target's revert
+// (Task 5 of docs/superpowers/plans/2026-10-20-offline-operator-move.md),
+// tested through the functions the agent calls (newMovesWorker, armOffline,
+// reconcileSerializedWith, the edges-file writer and movesPass), over one
+// DirStore for the deployment bucket, leasetest's S3 model for the data
+// buckets data-ie and data-uk, real registries, a real litestream replica
+// written by lease.ReplicateWithLease, the real archive, the real held
+// render and the exec backend's real launch composition.
+//
+// The seal is this file's fixture (landWorld.sealFixture): Task 4's sealer
+// runs in parallel and is not on this branch. It writes what the sealer's
+// brief says the sealer writes, through the same functions: the staged
+// replica by lease.MirrorPrefix, the archive by archive.Build and Seal, the
+// timers snapshot, the env pin from LaunchLayers, and seal.json create-only
+// with the record's evidence.
+
+// landS3 is the data buckets of the land tests: leasetest's S3 model and its
+// wire server, with three calls the server does not speak yet answered from
+// the same model: ListObjectsV2 (litestream's replica client and
+// lease.NewS3Mirror list that way; Task 4 is adding a minimal one to
+// leasetest), DeleteObjects and HEAD. Every other request is
+// leasetest.NewS3Server's.
+type landS3 struct {
+ f *leasetest.VersionedBucket
+ srv *httptest.Server
+}
+
+func newLandS3(t *testing.T, buckets ...string) *landS3 {
+ t.Helper()
+ f := leasetest.NewVersionedBucket()
+ for _, b := range buckets {
+ f.AddBucket(b)
+ }
+ inner, _ := leasetest.NewS3Server(f)
+ s := &landS3{f: f}
+ s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { // 127.0.0.1 only
+ bucket, rawKey, _ := strings.Cut(strings.TrimPrefix(r.URL.EscapedPath(), "/"), "/")
+ q := r.URL.Query()
+ switch {
+ case r.Method == http.MethodGet && rawKey == "" && q.Get("list-type") == "2":
+ s.listV2(w, r.Context(), bucket, q.Get("prefix"), q.Get("delimiter"))
+ case r.Method == http.MethodPost && rawKey == "" && q.Has("delete"):
+ s.deleteObjects(w, r, bucket)
+ case r.Method == http.MethodHead && rawKey != "":
+ key := strings.ReplaceAll(rawKey, "%2F", "/")
+ if body, ok := f.Current(bucket, key); ok {
+ sum := md5.Sum(body)
+ w.Header().Set("Content-Length", strconv.Itoa(len(body)))
+ w.Header().Set("ETag", `"`+hex.EncodeToString(sum[:])+`"`)
+ w.WriteHeader(http.StatusOK)
+ return
+ }
+ w.WriteHeader(http.StatusNotFound)
+ default:
+ inner.ServeHTTP(w, r)
+ }
+ }))
+ t.Cleanup(s.srv.Close)
+ for k, v := range map[string]string{
+ "CARLOS_S3_ENDPOINT": s.srv.URL, "AWS_ACCESS_KEY_ID": "test-key", "AWS_SECRET_ACCESS_KEY": "test-secret",
+ "AWS_SESSION_TOKEN": "", "AWS_PROFILE": "", "AWS_REGION": "", "AWS_EC2_METADATA_DISABLED": "true",
+ "AWS_CONFIG_FILE": filepath.Join(t.TempDir(), "absent"), "AWS_SHARED_CREDENTIALS_FILE": filepath.Join(t.TempDir(), "absent"),
+ } {
+ t.Setenv(k, v)
+ }
+ return s
+}
+
+type v2Object struct {
+ Key string `xml:"Key"`
+ LastModified string `xml:"LastModified"`
+ ETag string `xml:"ETag"`
+ Size int64 `xml:"Size"`
+ StorageClass string `xml:"StorageClass"`
+}
+
+type v2Prefix struct {
+ Prefix string `xml:"Prefix"`
+}
+
+type v2Result struct {
+ XMLName xml.Name `xml:"ListBucketResult"`
+ Name string `xml:"Name"`
+ Prefix string `xml:"Prefix"`
+ Delimiter string `xml:"Delimiter,omitempty"`
+ KeyCount int `xml:"KeyCount"`
+ MaxKeys int `xml:"MaxKeys"`
+ IsTruncated bool `xml:"IsTruncated"`
+ Contents []v2Object `xml:"Contents"`
+ CommonPrefixes []v2Prefix `xml:"CommonPrefixes"`
+}
+
+// listV2 answers ListObjectsV2 with the current objects under prefix, in one
+// page, rolling keys up at the delimiter as S3 does.
+func (s *landS3) listV2(w http.ResponseWriter, ctx context.Context, bucket, prefix, delim string) {
+ vers, err := s.f.ListVersions(ctx, bucket, prefix)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusNotFound)
+ return
+ }
+ res := v2Result{Name: bucket, Prefix: prefix, Delimiter: delim, MaxKeys: 1000}
+ seen := map[string]bool{}
+ for _, v := range vers {
+ if !v.IsLatest || v.DeleteMarker {
+ continue
+ }
+ if delim != "" {
+ if i := strings.Index(v.Key[len(prefix):], delim); i >= 0 {
+ cp := v.Key[:len(prefix)+i+len(delim)]
+ if !seen[cp] {
+ seen[cp] = true
+ res.CommonPrefixes = append(res.CommonPrefixes, v2Prefix{cp})
+ }
+ continue
+ }
+ }
+ res.Contents = append(res.Contents, v2Object{Key: v.Key, LastModified: v.LastModified.UTC().Format(time.RFC3339), ETag: `"` + v.ETag + `"`, Size: v.Size, StorageClass: "STANDARD"})
+ }
+ sort.Slice(res.Contents, func(i, j int) bool { return res.Contents[i].Key < res.Contents[j].Key })
+ res.KeyCount = len(res.Contents) + len(res.CommonPrefixes)
+ w.Header().Set("Content-Type", "application/xml")
+ xml.NewEncoder(w).Encode(res)
+}
+
+func (s *landS3) deleteObjects(w http.ResponseWriter, r *http.Request, bucket string) {
+ var req struct {
+ Objects []struct {
+ Key string `xml:"Key"`
+ } `xml:"Object"`
+ }
+ body, _ := io.ReadAll(r.Body)
+ if err := xml.Unmarshal(body, &req); err != nil {
+ http.Error(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ type deleted struct {
+ Key string `xml:"Key"`
+ }
+ out := struct {
+ XMLName xml.Name `xml:"DeleteResult"`
+ Deleted []deleted `xml:"Deleted"`
+ }{}
+ for _, o := range req.Objects {
+ if _, ok := s.f.Current(bucket, o.Key); ok {
+ s.f.BareDelete(bucket, o.Key)
+ }
+ out.Deleted = append(out.Deleted, deleted{o.Key})
+ }
+ w.Header().Set("Content-Type", "application/xml")
+ xml.NewEncoder(w).Encode(out)
+}
+
+// current is the current objects under prefix in bucket.
+func (s *landS3) current(t *testing.T, bucket, prefix string) map[string][]byte {
+ t.Helper()
+ vers, err := s.f.ListVersions(context.Background(), bucket, prefix)
+ if err != nil {
+ t.Fatal(err)
+ }
+ out := map[string][]byte{}
+ for _, v := range vers {
+ if v.IsLatest && !v.DeleteMarker {
+ body, _ := s.f.Current(bucket, v.Key)
+ out[v.Key] = body
+ }
+ }
+ return out
+}
+
+// --- the world ---------------------------------------------------------------
+
+const (
+ landKeyValue = "ik_0123456789abcdef0123456789abcdef"
+ landStore = "shop-store-ie"
+ landSecret = "store-secret-value"
+)
+
+// landWorld is offHost at home (ie-a, data-ie) moving to uk (uk-a,
+// data-uk), with a real database replica, a files directory, an env file
+// carrying the instance key, an app config bundle carrying the store
+// binding, timers, a NextDue, a replication rung and a promoted release.
+type landWorld struct {
+ t *testing.T
+ wd *offWorld
+ s3 *landS3
+ ie, uk *offBox
+ ieEB *activator.ExecBackend
+ ukEB *activator.ExecBackend
+
+ secrets string
+ priv *ecdh.PrivateKey
+ relPriv ed25519.PrivateKey
+
+ seal moves.OfflineReport
+ sealRaw []byte
+ src provision.MoveSource
+
+ // configKey: the app's configuration sets CARLOS_INSTANCE_KEY to it and
+ // <host>.env carries none, so the key the app is handed comes from the
+ // rendered <host>.config.env (instancekey.LayerConfig). "" is the usual
+ // shape: the key minted into <host>.env.
+ configKey string
+}
+
+func newLandWorld(t *testing.T) *landWorld { return newLandWorldWith(t, "") }
+
+// newLandWorldWith is newLandWorld with the instance key set by the app's
+// configuration (configKey), when not "".
+func newLandWorldWith(t *testing.T, configKey string) *landWorld {
+ t.Helper()
+ ctx := context.Background()
+ lw := &landWorld{t: t, wd: newOffWorld(t), configKey: configKey}
+ lw.s3 = newLandS3(t, "data-ie", "data-uk")
+ var err error
+ privB64, _, err := appconfig.GenerateKey()
+ if err != nil {
+ t.Fatal(err)
+ }
+ lw.secrets = privB64
+ if lw.priv, err = appconfig.ParsePrivateKey(privB64, "test"); err != nil {
+ t.Fatal(err)
+ }
+ seed, pub, err := release.GenerateKey()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if lw.relPriv, err = release.ParseSigningKey(seed, "test"); err != nil {
+ t.Fatal(err)
+ }
+ s := lw.wd.s
+ acct := lw.wd.acct
+ shipVersion(t, s, acct, "app", "v1")
+ shipVersion(t, s, acct, "app", "v2")
+ lw.promote("v1")
+ lw.bundle(landStore)
+
+ env := func(bucket, region string) map[string]string {
+ return map[string]string{"CARLOS_HIBERNATE_BUCKET": bucket, "CARLOS_HIBERNATE_REGION": region, "CARLOS_SECRETS_KEY": privB64, "CARLOS_RELEASE_PUBKEY": pub}
+ }
+ lw.ie = lw.wd.box("ie-a", "", env("data-ie", "eu-west-1"))
+ lw.uk = lw.wd.box("uk-a", "uk", env("data-uk", "eu-west-2"))
+ lw.ieEB = &activator.ExecBackend{DataDir: lw.ie.dd, OptDir: lw.ie.opt}
+ lw.ukEB = &activator.ExecBackend{DataDir: lw.uk.dd, OptDir: lw.uk.opt, Region: "uk"}
+ lw.arm(lw.ie)
+ lw.arm(lw.uk)
+
+ // The source at rest: its row, then everything the instance holds.
+ lw.ie.tick()
+ lw.uk.tick()
+ rt := lw.ie.row(offHost)
+ if rt == nil {
+ t.Fatal("the source has no row")
+ }
+ dd := lw.ie.dd
+ writeFile(t, filepath.Join(dd, offHost, "notes.txt"), "the app's own file\n")
+ writeFile(t, filepath.Join(dd, offHost, "uploads", "a.bin"), "\x00\x01binary\x02")
+ ownKey := "CARLOS_INSTANCE_KEY=" + landKeyValue + "\n"
+ if configKey != "" {
+ ownKey = ""
+ }
+ writeFile(t, filepath.Join(dd, offHost+".env"), ownKey+"CARLOS_ADMIN_TOKEN=admin-token\nAPP_OVERRIDE=1\n")
+ if err := lw.ie.reg.ReplaceTimers(offHost, []registry.Timer{
+ {Host: offHost, Name: "digest", Kind: "cron", Spec: "0 * * * *", Path: "/cron/digest", NextAt: 1_800_000_000, CreatedAt: 1_700_000_000, DueAt: 1_800_000_000},
+ {Host: offHost, Name: "reminder", Kind: "once", Spec: "", Path: "/once/reminder", NextAt: 1_800_000_500, CreatedAt: 1_700_000_100, DueAt: 1_800_000_500},
+ }); err != nil {
+ t.Fatal(err)
+ }
+ if err := lw.ie.reg.SetNextDue(offHost, 1_800_000_000); err != nil {
+ t.Fatal(err)
+ }
+ if err := lw.ie.reg.SetReplication(offHost, registry.ReplicationBatched); err != nil {
+ t.Fatal(err)
+ }
+ lw.replicate(rt.DBPath, "one", "two", "three")
+
+ // start, the source holds and reports, its config is rendered held
+ // (as SyncExec would have rendered it at rest), seal, sealed.
+ lw.wd.start()
+ lw.ie.tick()
+ if _, err := lw.ie.w.cfg.RenderHeld(ctx, s, *lw.ie.row(offHost), offID, nil, func(string) error { return nil }); err != nil {
+ t.Fatal(err)
+ }
+ lw.wd.seal()
+ lw.sealFixture()
+ lw.wd.move(provision.OfflineSealedBy(offID, "ie-a", lw.src))
+ return lw
+}
+
+// arm wires a box as armMovesPass would, over its own data bucket.
+func (lw *landWorld) arm(b *offBox) {
+ bucket, region := b.getenv("CARLOS_HIBERNATE_BUCKET"), b.getenv("CARLOS_HIBERNATE_REGION")
+ eb := lw.ieEB
+ if b == lw.uk {
+ eb = lw.ukEB
+ }
+ b.svc = activator.New(activator.Config{Bucket: bucket, Idle: time.Hour}, &fakeWakeBackend{}, b.reg)
+ cfg := moveCfg{Region: b.region, Box: b.name, DataDir: b.dd, OptDir: b.opt, Quiesced: b.svc.Quiesced, Route: b.reg.Get, Now: b.wd.clock,
+ OwnBucket: bucket, OwnBucketRegion: region, OwnPrefix: "instances",
+ Mirror: func(ctx context.Context, dst string) (lease.MirrorStore, error) {
+ return lease.NewS3Mirror(ctx, region, dst)
+ }}
+ armOffline(&cfg, b.dd, b.reg, b.svc, nil, hostwrite.Config{}, b.getenv)
+ cfg.Offline = newOfflineView(cfg, nil)
+ cfg.ReleaseHold = releaseHeld(b.reg, nil)
+ cfg.Procs = func(host string) (int, error) { return cgroup.HostProcs(b.cg, host, "") }
+ cfg.LaunchLayers = eb.LaunchLayers
+ b.w = newMovesWorker(cfg)
+}
+
+// restart is an agent restart on the target: a fresh activator and worker
+// over the same registry, data directory and buckets.
+func (lw *landWorld) restart() { lw.arm(lw.uk) }
+
+func (lw *landWorld) promote(v string) {
+ lw.t.Helper()
+ if err := release.Promote(context.Background(), lw.wd.s, lw.relPriv, lw.wd.acct, "app", "stable", v); err != nil {
+ lw.t.Fatal(err)
+ }
+}
+
+// bundle writes the app's config: the store binding, its secret and a
+// setting.
+func (lw *landWorld) bundle(store string) {
+ lw.t.Helper()
+ env, err := appconfig.SealBound(lw.priv.PublicKey(), appconfig.Binding{Account: lw.wd.acct, App: "app"}, []byte(landSecret))
+ if err != nil {
+ lw.t.Fatal(err)
+ }
+ m := appconfig.Manifest{Schema: 1, App: "app", Vars: map[string]appconfig.Var{
+ "STORE_BUCKET": {Kind: "env", Value: store},
+ "STORE_ENDPOINT": {Kind: "env", Value: "https://store.example.com"},
+ "STORE_SECRET_KEY": {Kind: "secret", Envelope: env},
+ "FEATURE": {Kind: "env", Value: "on"},
+ }}
+ if lw.configKey != "" {
+ m.Vars[instancekey.Var] = appconfig.Var{Kind: "env", Value: lw.configKey}
+ }
+ if err := moves.PutReplace(context.Background(), lw.wd.s, appconfig.Key(lw.wd.acct, "app"), mustJSON(lw.t, m)); err != nil {
+ lw.t.Fatal(err)
+ }
+}
+
+// replicate writes a real SQLite database at dbPath with rows, replicated
+// to data-ie under the host's own prefix by litestream, and closed.
+func (lw *landWorld) replicate(dbPath string, rows ...string) {
+ t := lw.t
+ t.Helper()
+ ctx := context.Background()
+ db, err := sql.Open("sqlite", dbPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ if _, err := db.Exec(`CREATE TABLE notes(id INTEGER PRIMARY KEY, body TEXT)`); err != nil {
+ t.Fatal(err)
+ }
+ rep, err := lease.ReplicateWithLease(ctx, dbPath, "data-ie", activator.HibernatePrefix("instances", offHost), "eu-west-1")
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, r := range rows {
+ if _, err := db.Exec(`INSERT INTO notes(body) VALUES (?)`, r); err != nil {
+ t.Fatal(err)
+ }
+ if err := rep.SyncAndWait(ctx); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := rep.Close(ctx); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// sealFixture writes what Task 4's sealer writes at seal, from the source as
+// it stands: the staged replica in data-uk, the archive, its manifest and the
+// timers snapshot under the move's archive prefix, and seal.json; and sets
+// lw.src, the record's MoveSource with the seal's evidence.
+func (lw *landWorld) sealFixture() {
+ t := lw.t
+ t.Helper()
+ ctx := context.Background()
+ ie := lw.ie
+ rt := ie.row(offHost)
+ m, err := lease.NewS3Mirror(ctx, "eu-west-1", "eu-west-2")
+ if err != nil {
+ t.Fatal(err)
+ }
+ round, err := lease.MirrorPrefix(ctx, m, "data-ie", activator.HibernatePrefix("instances", offHost)+"/", "data-uk", activator.MovesPrefix(offHost), lease.Receipts{})
+ if err != nil || round.Deferred != 0 {
+ t.Fatalf("mirror: %+v %v", round, err)
+ }
+ staged := map[string]moves.StagedObject{}
+ var final ltx.TXID
+ objs, err := m.List(ctx, "data-uk", activator.MovesPrefix(offHost))
+ if err != nil {
+ t.Fatal(err)
+ }
+ for k, o := range objs {
+ staged[strings.TrimPrefix(k, activator.MovesPrefix(offHost))] = moves.StagedObject{Size: o.Size, ETag: o.ETag}
+ if _, maxTX, err := ltx.ParseFilename(path.Base(k)); err == nil && maxTX > final {
+ final = maxTX
+ }
+ }
+ if final == 0 {
+ t.Fatal("nothing staged")
+ }
+
+ filesDir, envPath := filepath.Join(ie.dd, offHost), filepath.Join(ie.dd, offHost+".env")
+ var plainBuf, sealedBuf bytes.Buffer
+ plain, entries, n, err := archive.Build(filesDir, envPath, &plainBuf)
+ if err != nil {
+ t.Fatal(err)
+ }
+ dk, sealedSHA, err := archive.Seal(&plainBuf, lw.priv.PublicKey(), appconfig.Binding{Account: lw.wd.acct, App: "app"}, &sealedBuf)
+ if err != nil {
+ t.Fatal(err)
+ }
+ man := archive.Manifest{V: archive.ManifestVersion, Op: offID, Plain: plain, Sealed: sealedSHA, Bytes: n, Entries: entries, SealedDataKey: dk, Chunk: archive.ChunkSize}
+ prefix := moves.OfflineArchivePrefix(offHost, offID)
+ lw.s3.f.Put("data-uk", prefix+"files.sealed", sealedBuf.Bytes())
+ lw.s3.f.Put("data-uk", prefix+"files.json", mustJSON(t, man))
+ timers, err := ie.reg.TimersFor(offHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ active := moves.ActiveTimers(timers)
+ snap, err := moves.TimersSnapshot{Host: offHost, Rows: active, TakenAt: lw.wd.clock()}.Encode()
+ if err != nil {
+ t.Fatal(err)
+ }
+ lw.s3.f.Put("data-uk", prefix+"timers.json", snap)
+
+ envBytes, err := os.ReadFile(envPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ cfgBytes, _ := os.ReadFile(filepath.Join(ie.dd, offHost+".config.env"))
+ key, _ := instancekey.Effective(instancekey.ParseEnv(cfgBytes), instancekey.ParseEnv(envBytes))
+ rk, err := moves.ReportKeyFromSecrets(lw.secrets)
+ if err != nil {
+ t.Fatal(err)
+ }
+ launch, inst, _, err := lw.ieEB.LaunchLayers(*rt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ envSum := sha256.Sum256(envBytes)
+ items := map[string]moves.Check{}
+ for _, it := range moves.OfflineSourced() {
+ items[it.ID] = moves.Check{OK: true}
+ }
+ lw.seal = moves.OfflineReport{
+ V: moves.OfflineReportVersion, Op: offID, Host: offHost, Box: "ie-a", Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseSealed,
+ At: lw.wd.clock(), TombstoneOwner: moves.TombstoneOwner(offID), Items: items,
+ FinalTXID: uint64(final), Staged: staged, FilesPlain: plain, FilesSealed: sealedSHA, FilesBytes: n, FilesEntries: len(entries),
+ EnvSHA256: hex.EncodeToString(envSum[:]), KeySHA256: instancekey.Digest(key), EnvDigest: moves.PortableDigest(rk, launch, inst),
+ Release: "v1", TimersDigest: moves.TimersDigest(active), Replication: rt.Replication, NextDue: rt.NextDue,
+ }
+ if err := moves.ValidateOfflineSeal(lw.seal, offHost, offID); err != nil {
+ t.Fatal(err)
+ }
+ lw.sealRaw = mustJSON(t, lw.seal)
+ if _, created, err := putOfflineCreateOnly(ctx, lw.wd.s, moves.OfflineSealKey(offHost, offID), lw.sealRaw); err != nil || !created {
+ t.Fatalf("seal.json: created %v, %v", created, err)
+ }
+ lw.src = provision.MoveSource{
+ FinalTXID: uint64(final), Files: &provision.MoveFilesDigest{Plain: plain, Sealed: sealedSHA, Bytes: n, Entries: len(entries)},
+ TimersSHA256: lw.seal.TimersDigest, Release: "v1", EnvDigest: lw.seal.EnvDigest, Replication: rt.Replication, NextDue: rt.NextDue,
+ Evidence: moves.OfflineSealEvidence(offHost, offID, lw.sealRaw),
+ }
+}
+
+// flipAndLand flips the record and runs the target box's ticks.
+func (lw *landWorld) flipAndLand(ticks int) {
+ lw.wd.flip()
+ for range ticks {
+ lw.uk.tick()
+ }
+}
+
+func (lw *landWorld) report() moves.OfflineReport {
+ lw.t.Helper()
+ return lw.wd.report(moves.OfflineRoleTarget, "uk-a")
+}
+
+func (lw *landWorld) state() string {
+ lw.t.Helper()
+ rec := lw.wd.record()
+ if rec.Move == nil {
+ return ""
+ }
+ return rec.Move.State
+}
+
+// requireLanded is the happy landing's state: offline-landed, every landing
+// item OK in a landed report, the row held landing by the move, the edges
+// file not listing the host, and nothing ever taking the target's lease.
+func (lw *landWorld) requireLanded() {
+ t := lw.t
+ t.Helper()
+ if st := lw.state(); st != provision.OfflineLanded {
+ t.Fatalf("state %q, want %q; report %+v", st, provision.OfflineLanded, lw.report().Items)
+ }
+ r := lw.report()
+ if ok, bad := r.ItemsOK(moves.OfflineOwnerLanding); !ok || r.Phase != moves.OfflinePhaseLanded {
+ t.Fatalf("phase %s, not OK: %v (%+v)", r.Phase, bad, r.Items)
+ }
+ if len(r.Items) != len(moves.OfflineOwned(moves.OfflineOwnerLanding)) {
+ t.Fatalf("report carries %d items, want every landing item", len(r.Items))
+ }
+ rt := lw.uk.row(offHost)
+ if rt == nil || rt.Status != "landing" || rt.HoldOp != offID {
+ t.Fatalf("target row %+v", rt)
+ }
+ lw.uk.edges()
+ if lw.uk.lists(offHost) {
+ t.Fatal("the edges file lists the host before open")
+ }
+ lw.requireNoTargetLease()
+}
+
+// requireNoTargetLease: nothing has written under data-uk instances/ but,
+// at most, a lifted tombstone (a stale one the landing lifted), and the
+// host's lock is free.
+func (lw *landWorld) requireNoTargetLease() {
+ lw.t.Helper()
+ loc := activator.HibernatePrefix("instances", offHost)
+ for k := range lw.s3.current(lw.t, "data-uk", "instances/") {
+ if k != loc+"/lock.json" {
+ lw.t.Fatalf("data-uk holds %s before the first wake", k)
+ }
+ }
+ _, owner, expires, err := lease.LockState(context.Background(), "data-uk", loc, "eu-west-2")
+ if err != nil || !(owner == "" || lease.IsLiftedTombstone(owner, expires)) {
+ lw.t.Fatalf("the target's lock is %q until %v (%v)", owner, expires, err)
+ }
+}
+
+func keysOf[V any](m map[string]V) []string {
+ out := make([]string, 0, len(m))
+ for k := range m {
+ out = append(out, k)
+ }
+ sort.Strings(out)
+ return out
+}
+
+// tree is every regular file and symlink under dir with its content's
+// sha256, every directory with its mode: what a pass may not change. The
+// registry is left out (its file changes on every reconcile).
+func tree(t *testing.T, dir string) map[string]string {
+ t.Helper()
+ out := map[string]string{}
+ err := filepath.WalkDir(dir, func(p string, d fs.DirEntry, err error) error {
+ if err != nil {
+ return err
+ }
+ rel, _ := filepath.Rel(dir, p)
+ if strings.HasPrefix(rel, "registry.db") {
+ return nil
+ }
+ info, err := d.Info()
+ if err != nil {
+ return err
+ }
+ switch {
+ case d.IsDir():
+ out[rel+"/"] = info.Mode().String()
+ case info.Mode().IsRegular():
+ b, err := os.ReadFile(p)
+ if err != nil {
+ return err
+ }
+ sum := sha256.Sum256(b)
+ out[rel] = hex.EncodeToString(sum[:]) + " " + info.Mode().String()
+ default:
+ out[rel] = info.Mode().String()
+ }
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ return out
+}
+
+// landed is what the target landed, by name: the files directory, the env
+// file and the config file, as the source has them.
+func (lw *landWorld) requireSourceBytes() {
+ t := lw.t
+ t.Helper()
+ src, dst := tree(t, filepath.Join(lw.ie.dd, offHost)), tree(t, filepath.Join(lw.uk.dd, offHost))
+ if !reflect.DeepEqual(src, dst) {
+ t.Fatalf("files differ:\nsource %v\ntarget %v", src, dst)
+ }
+ for _, name := range []string{offHost + ".env", offHost + ".config.env"} {
+ a, err := os.ReadFile(filepath.Join(lw.ie.dd, name))
+ if err != nil {
+ t.Fatal(err)
+ }
+ b, err := os.ReadFile(filepath.Join(lw.uk.dd, name))
+ if err != nil {
+ t.Fatalf("%s on the target: %v", name, err)
+ }
+ if !bytes.Equal(a, b) {
+ t.Fatalf("%s differs:\nsource %q\ntarget %q", name, a, b)
+ }
+ }
+}
+
+func itemsSummary(r moves.OfflineReport) map[string]moves.Check { return r.Items }
+
+func (lw *landWorld) parked(role string) map[string]string {
+ dir := moves.OfflineParkDir(lw.uk.dd, offHost, offID, role)
+ if !isDir(dir) {
+ return nil
+ }
+ return tree(lw.t, dir)
+}
+
+// --- (a) and (f): the landing --------------------------------------------------
+
+// The full landing reaches offline-landed with every landing item OK, the
+// row held, nothing served and nothing taking the target's lease; and the
+// lesson of the 2026-10-02 outage: the target's <host>.env, and so its
+// instance key, is the source's byte for byte, and its rendered
+// <host>.config.env carries the app's store binding, its secret included.
+func TestOfflineLandHappyPath(t *testing.T) {
+ lw := newLandWorld(t)
+ lw.wd.flip()
+ lw.uk.reconcile()
+ if rt := lw.uk.row(offHost); rt == nil || rt.Status != "landing" || rt.HoldOp != offID {
+ t.Fatalf("the target row before the landing: %+v", rt)
+ }
+ lw.uk.edges()
+ lw.uk.pass()
+ lw.requireLanded()
+ lw.requireSourceBytes()
+
+ // The instance key the app would be handed, byte for byte.
+ read := func(b *offBox, name string) map[string]string {
+ body, err := os.ReadFile(filepath.Join(b.dd, name))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return instancekey.ParseEnv(body)
+ }
+ srcKey, _ := instancekey.Effective(read(lw.ie, offHost+".config.env"), read(lw.ie, offHost+".env"))
+ dstKey, layer := instancekey.Effective(read(lw.uk, offHost+".config.env"), read(lw.uk, offHost+".env"))
+ if srcKey != landKeyValue || dstKey != srcKey || layer != instancekey.LayerOwn {
+ t.Fatalf("instance key: source %q, target %q (layer %v)", srcKey, dstKey, layer)
+ }
+ // The store binding, as the target will launch it.
+ env, _, _, err := lw.ukEB.LaunchLayers(*lw.uk.row(offHost))
+ if err != nil {
+ t.Fatal(err)
+ }
+ for k, want := range map[string]string{"STORE_BUCKET": landStore, "STORE_ENDPOINT": "https://store.example.com", "STORE_SECRET_KEY": landSecret, "CARLOS_INSTANCE_KEY": landKeyValue, "CARLOS_ADMIN_TOKEN": "admin-token", "APP_OVERRIDE": "1"} {
+ if env[k] != want {
+ t.Fatalf("the target launches with %s=%q, want %q", k, env[k], want)
+ }
+ }
+ // The ack names the move; the staged replica is untouched.
+ ack := appconfig.ConfigAck{}
+ body, _, err := lw.wd.s.Get(context.Background(), appconfig.ConfigAckKey("uk-a", offHost))
+ if err != nil || json.Unmarshal(body, &ack) != nil || ack.MoveOp != offID {
+ t.Fatalf("config ack %+v %v", ack, err)
+ }
+ staged := lw.s3.current(t, "data-uk", activator.MovesPrefix(offHost))
+ if len(staged) != len(lw.seal.Staged) {
+ t.Fatalf("staged %v, sealed %v", keysOf(staged), keysOf(lw.seal.Staged))
+ }
+ // Timers, NextDue and replication arrived.
+ rows, err := lw.uk.reg.TimersFor(offHost)
+ if err != nil || moves.TimersDigest(rows) != lw.seal.TimersDigest || len(rows) != 2 {
+ t.Fatalf("timers %+v %v", rows, err)
+ }
+ if rt := lw.uk.row(offHost); rt.NextDue != 1_800_000_000 || rt.Replication != registry.ReplicationBatched {
+ t.Fatalf("row columns %+v", rt)
+ }
+ if lw.parked(moves.OfflineParkTargetBefore) != nil || lw.parked(moves.OfflineParkVerify) != nil {
+ t.Fatal("a clean landing parked something or left its verify scratch")
+ }
+}
+
+// --- (c) B2: every pass is an ensure ---------------------------------------------
+
+// A second, third and post-CAS pass, and a pass after an agent restart,
+// change nothing on disk, write nothing to the staged replica or the
+// archive, and report the same items.
+func TestOfflineLandIsAnEnsure(t *testing.T) {
+ lw := newLandWorld(t)
+ lw.flipAndLand(1)
+ lw.requireLanded()
+ disk := tree(t, lw.uk.dd)
+ items := itemsSummary(lw.report())
+ data := lw.s3.current(t, "data-uk", "moves/")
+ for i := range 4 {
+ if i == 2 {
+ lw.restart()
+ }
+ lw.uk.tick()
+ lw.requireLanded()
+ if got := tree(t, lw.uk.dd); !reflect.DeepEqual(got, disk) {
+ t.Fatalf("pass %d changed the disk:\nbefore %v\nafter %v", i, disk, got)
+ }
+ if got := itemsSummary(lw.report()); !reflect.DeepEqual(got, items) {
+ t.Fatalf("pass %d reports %+v, the first %+v", i, got, items)
+ }
+ if got := lw.s3.current(t, "data-uk", "moves/"); !reflect.DeepEqual(got, data) {
+ t.Fatalf("pass %d changed data-uk moves/", i)
+ }
+ }
+ // The replica was restored once: the remembered fingerprint stands.
+ if c := lw.report().Items["db"]; !strings.HasPrefix(c.Reason, dbVerifiedPrefix) {
+ t.Fatalf("db %+v", c)
+ }
+}
+
+// A crash at every point of the landing (after each item, after the report,
+// after the CAS), followed by an agent restart, converges: the next passes
+// land everything, park nothing, and leave the disk as a clean landing does.
+func TestOfflineLandConvergesAfterACrashAtEveryPoint(t *testing.T) {
+ var points []string
+ for _, id := range []string{"warm-cache", "key-adoption", "lease", "db", "files", "config", "env", "env-pin", "timers", "next-due", "replication", "release", "report", "landed"} {
+ points = append(points, "offline-"+id)
+ }
+ for _, point := range points {
+ t.Run(point, func(t *testing.T) {
+ lw := newLandWorld(t)
+ crashed := 0
+ lw.uk.w.cfg.landFault = func(p string) error {
+ if p == point && crashed == 0 {
+ crashed++
+ return errors.New("crash")
+ }
+ return nil
+ }
+ lw.flipAndLand(1)
+ if crashed != 1 {
+ t.Fatalf("the crash point %s was never reached", point)
+ }
+ lw.restart()
+ lw.uk.tick()
+ lw.uk.tick()
+ lw.requireLanded()
+ lw.requireSourceBytes()
+ if p := lw.parked(moves.OfflineParkTargetBefore); p != nil {
+ t.Fatalf("parked its own landing: %v", p)
+ }
+ })
+ }
+}
+
+// --- (b) missing or corrupted items: mended -----------------------------------------
+
+// Each item the landing carries, removed or damaged after the landing, is
+// put back by the next pass: a missing one landed, a different one parked
+// into target-before and landed again. The report stays landed and the
+// record offline-landed.
+func TestOfflineLandMendsWhatIsMissing(t *testing.T) {
+ cases := []struct {
+ name string
+ damage func(lw *landWorld)
+ parked []string // names that must appear in target-before
+ }{
+ {"files directory removed", func(lw *landWorld) { os.RemoveAll(filepath.Join(lw.uk.dd, offHost)) }, nil},
+ {"a file changed", func(lw *landWorld) {
+ writeFile(lw.t, filepath.Join(lw.uk.dd, offHost, "notes.txt"), "tampered\n")
+ }, []string{offHost + "/notes.txt"}},
+ {"a file added", func(lw *landWorld) {
+ writeFile(lw.t, filepath.Join(lw.uk.dd, offHost, "stray"), "x")
+ }, []string{offHost + "/stray"}},
+ {"env removed", func(lw *landWorld) { os.Remove(filepath.Join(lw.uk.dd, offHost+".env")) }, []string{offHost + "/notes.txt"}},
+ {"env changed: a new instance key", func(lw *landWorld) {
+ writeFile(lw.t, filepath.Join(lw.uk.dd, offHost+".env"), "CARLOS_INSTANCE_KEY=minted-on-the-target\n")
+ }, []string{offHost + ".env", offHost + "/notes.txt"}},
+ {"config removed", func(lw *landWorld) { os.Remove(filepath.Join(lw.uk.dd, offHost+".config.env")) }, nil},
+ {"config changed", func(lw *landWorld) {
+ writeFile(lw.t, filepath.Join(lw.uk.dd, offHost+".config.env"), "STORE_BUCKET=elsewhere\n")
+ }, nil},
+ {"a timer removed", func(lw *landWorld) {
+ if err := lw.uk.reg.ReplaceTimers(offHost, nil); err != nil {
+ lw.t.Fatal(err)
+ }
+ }, nil},
+ {"next due changed", func(lw *landWorld) { lw.uk.reg.SetNextDue(offHost, 5) }, nil},
+ {"replication changed", func(lw *landWorld) { lw.uk.reg.SetReplication(offHost, registry.ReplicationDaily) }, nil},
+ {"a local database appeared", func(lw *landWorld) {
+ writeFile(lw.t, provision.DefaultDBPath(lw.uk.dd, offHost), "stale")
+ writeFile(lw.t, filepath.Join(activatorMetaDir(provision.DefaultDBPath(lw.uk.dd, offHost)), "ltx", "0", "x"), "meta")
+ }, []string{offHost + ".db", ".activator-" + offHost + ".db-litestream/ltx/0/x"}},
+ {"an adoption record appeared", func(lw *landWorld) {
+ writeFile(lw.t, filepath.Join(lw.uk.dd, offHost+instancekey.AdoptionSuffix), "{}")
+ }, []string{offHost + instancekey.AdoptionSuffix}},
+ }
+ for _, c := range cases {
+ t.Run(c.name, func(t *testing.T) {
+ lw := newLandWorld(t)
+ lw.flipAndLand(1)
+ lw.requireLanded()
+ c.damage(lw)
+ lw.uk.tick()
+ lw.requireLanded()
+ lw.requireSourceBytes()
+ if rows, _ := lw.uk.reg.TimersFor(offHost); moves.TimersDigest(rows) != lw.seal.TimersDigest {
+ t.Fatal("timers not mended")
+ }
+ if rt := lw.uk.row(offHost); rt.NextDue != lw.seal.NextDue || rt.Replication != lw.seal.Replication {
+ t.Fatalf("row columns not mended: %+v", rt)
+ }
+ parked := lw.parked(moves.OfflineParkTargetBefore)
+ for _, name := range c.parked {
+ if _, ok := parked[name]; !ok {
+ t.Fatalf("%s not parked; target-before holds %v", name, keysOf(parked))
+ }
+ }
+ if len(c.parked) == 0 && parked != nil {
+ t.Fatalf("parked %v", keysOf(parked))
+ }
+ // And it stays mended: the next pass parks nothing more.
+ before := tree(t, lw.uk.dd)
+ lw.uk.tick()
+ if got := tree(t, lw.uk.dd); !reflect.DeepEqual(got, before) {
+ t.Fatal("the pass after the mend changed the disk")
+ }
+ })
+ }
+}
+
+// What a target holds for the host before the landing (an old residence's
+// database, meta dir, adoption record, files and env) is parked, never
+// trusted; a stale tombstone of another move in the target's bucket is
+// lifted; and the landing completes.
+func TestOfflineLandParksAndLiftsWhatItFinds(t *testing.T) {
+ lw := newLandWorld(t)
+ ctx := context.Background()
+ dd := lw.uk.dd
+ db := provision.DefaultDBPath(dd, offHost)
+ writeFile(t, db, "old residence")
+ writeFile(t, db+"-wal", "wal")
+ writeFile(t, filepath.Join(activatorMetaDir(db), "ltx", "0", "0000000000000009-0000000000000009.ltx"), "meta")
+ writeFile(t, filepath.Join(dd, offHost+instancekey.AdoptionSuffix), "{}")
+ writeFile(t, filepath.Join(dd, offHost, "old.txt"), "old residence's file")
+ writeFile(t, filepath.Join(dd, offHost+".env"), "CARLOS_INSTANCE_KEY=old-residence-key\n")
+ loc := activator.HibernatePrefix("instances", offHost)
+ if err := lease.Tombstone(ctx, "data-uk", loc, "eu-west-2", moves.TombstoneOwner(offID2), time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ lw.flipAndLand(1)
+ lw.requireLanded()
+ lw.requireSourceBytes()
+ parked := lw.parked(moves.OfflineParkTargetBefore)
+ for _, name := range []string{offHost + ".db", offHost + ".db-wal", ".activator-" + offHost + ".db-litestream/ltx/0/0000000000000009-0000000000000009.ltx",
+ offHost + instancekey.AdoptionSuffix, offHost + "/old.txt", offHost + ".env"} {
+ if _, ok := parked[name]; !ok {
+ t.Fatalf("%s not parked: %v", name, keysOf(parked))
+ }
+ }
+ _, owner, expires, err := lease.LockState(ctx, "data-uk", loc, "eu-west-2")
+ if err != nil || owner != moves.TombstoneOwner(offID2) || !lease.IsLiftedTombstone(owner, expires) {
+ t.Fatalf("the stale tombstone: %q %v %v", owner, expires, err)
+ }
+ // Lifted, not taken: Acquire would now succeed, and nothing else
+ // appeared under instances/.
+ if got := keysOf(lw.s3.current(t, "data-uk", "instances/")); len(got) != 1 || got[0] != loc+"/lock.json" {
+ t.Fatalf("data-uk instances/: %v", got)
+ }
+}
+
+// --- (b) mismatches: failed, named --------------------------------------------------
+
+// Every mismatch with the seal fails its own item in a failed report, and
+// the record stays offline-landing: an extra staged object, a changed one,
+// a changed archive, a changed bundle, a promoted channel, a seal that is
+// not the record's, a changed timers snapshot, a live writer's lease, and a
+// tombstone of this very move in the target's bucket.
+func TestOfflineLandFailsEachMismatch(t *testing.T) {
+ prefix := moves.OfflineArchivePrefix(offHost, offID)
+ cases := []struct {
+ name string
+ item string
+ hard bool
+ damage func(lw *landWorld)
+ }{
+ {"an extra staged object", "db", true, func(lw *landWorld) {
+ lw.s3.f.Put("data-uk", activator.MovesPrefix(offHost)+"0000/0000000000000009-0000000000000009.ltx", []byte("x"))
+ }},
+ {"a changed staged object", "db", true, func(lw *landWorld) {
+ k := keysOf(lw.seal.Staged)[0]
+ lw.s3.f.Put("data-uk", activator.MovesPrefix(offHost)+k, []byte("not the sealed bytes"))
+ }},
+ {"a staged object gone", "db", true, func(lw *landWorld) {
+ lw.s3.f.BareDelete("data-uk", activator.MovesPrefix(offHost)+keysOf(lw.seal.Staged)[0])
+ }},
+ {"a changed archive", "files", true, func(lw *landWorld) {
+ b, _ := lw.s3.f.Current("data-uk", prefix+"files.sealed")
+ b = append([]byte(nil), b...)
+ b[len(b)/2] ^= 0xff
+ lw.s3.f.Put("data-uk", prefix+"files.sealed", b)
+ }},
+ {"a changed bundle", "env-pin", true, func(lw *landWorld) { lw.bundle("shop-store-somewhere-else") }},
+ {"a promoted channel", "release", true, func(lw *landWorld) { lw.promote("v2") }},
+ {"a changed timers snapshot", "timers", true, func(lw *landWorld) {
+ snap, _ := moves.TimersSnapshot{Host: offHost, TakenAt: lw.wd.clock()}.Encode()
+ lw.s3.f.Put("data-uk", prefix+"timers.json", snap)
+ }},
+ {"a live writer holds the target's lease", "lease", false, func(lw *landWorld) {
+ held, err := lease.Acquire(context.Background(), "data-uk", activator.HibernatePrefix("instances", offHost), "eu-west-2")
+ if err != nil {
+ lw.t.Fatal(err)
+ }
+ lw.t.Cleanup(func() { held.Release(context.Background()) })
+ }},
+ {"this move's own tombstone in the target's bucket", "lease", true, func(lw *landWorld) {
+ if err := lease.Tombstone(context.Background(), "data-uk", activator.HibernatePrefix("instances", offHost), "eu-west-2", moves.TombstoneOwner(offID), time.Hour); err != nil {
+ lw.t.Fatal(err)
+ }
+ }},
+ }
+ for _, c := range cases {
+ t.Run(c.name, func(t *testing.T) {
+ lw := newLandWorld(t)
+ c.damage(lw)
+ lw.flipAndLand(2)
+ if st := lw.state(); st != provision.OfflineLanding {
+ t.Fatalf("state %s", st)
+ }
+ r := lw.report()
+ if r.Items[c.item].OK || r.Items[c.item].Reason == "" {
+ t.Fatalf("%s: %+v", c.item, r.Items[c.item])
+ }
+ if want := map[bool]string{true: moves.OfflinePhaseFailed, false: moves.OfflinePhaseLanded}[c.hard]; r.Phase != want {
+ t.Fatalf("phase %s, want %s (%+v)", r.Phase, want, r.Items)
+ }
+ if rt := lw.uk.row(offHost); rt == nil || rt.HoldOp != offID {
+ t.Fatalf("row %+v", rt)
+ }
+ lw.uk.edges()
+ if lw.uk.lists(offHost) {
+ t.Fatal("listed")
+ }
+ })
+ }
+
+ // A seal whose bytes are not the ones the record names fails every item
+ // after the row; nothing is landed.
+ t.Run("a seal that is not the record's", func(t *testing.T) {
+ lw := newLandWorld(t)
+ if err := moves.PutReplace(context.Background(), lw.wd.s, moves.OfflineSealKey(offHost, offID), append(lw.sealRaw, ' ')); err != nil {
+ t.Fatal(err)
+ }
+ lw.flipAndLand(1)
+ r := lw.report()
+ if r.Phase != moves.OfflinePhaseFailed || !r.Items["row"].OK {
+ t.Fatalf("%+v", r)
+ }
+ for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) {
+ if it.ID != "row" && r.Items[it.ID].OK {
+ t.Fatalf("%s OK under a seal that is not the record's", it.ID)
+ }
+ }
+ if isDir(filepath.Join(lw.uk.dd, offHost)) || lexists(filepath.Join(lw.uk.dd, offHost+".env")) {
+ t.Fatal("landed files under a seal that is not the record's")
+ }
+ })
+
+ // A mismatch that clears (the channel promoted back) lands on a later
+ // pass: the report is recomputed from what is there every pass.
+ t.Run("a mismatch that clears", func(t *testing.T) {
+ lw := newLandWorld(t)
+ lw.promote("v2")
+ lw.flipAndLand(1)
+ if r := lw.report(); r.Phase != moves.OfflinePhaseFailed || r.Items["release"].OK {
+ t.Fatalf("%+v", r.Items["release"])
+ }
+ lw.promote("v1")
+ lw.uk.tick()
+ lw.requireLanded()
+ })
+}
+
+// --- (d) open ---------------------------------------------------------------------
+
+// open releases the hold only once every landing item verifies again
+// against the seal: a promote after the landing keeps the row held; once
+// it is undone the hold is released, the row is ready and listed, the
+// archive is deleted from the target's bucket, and the report says opened.
+// A following ExecBackend.Restore promotes the staged replica and its
+// database passes CheckDB with every row; until that first wake nothing has
+// taken the target's lease.
+func TestOfflineLandOpenReleasesOnlyWhenEveryItemVerifies(t *testing.T) {
+ lw := newLandWorld(t)
+ ctx := context.Background()
+ lw.flipAndLand(1)
+ lw.requireLanded()
+ lw.wd.open()
+
+ // An item that no longer verifies: not opened.
+ lw.promote("v2")
+ lw.uk.tick()
+ if rt := lw.uk.row(offHost); rt == nil || rt.HoldOp != offID || rt.Status != "landing" {
+ t.Fatalf("released with the release item failing: %+v", rt)
+ }
+ if r := lw.report(); r.Phase == moves.OfflinePhaseOpened || r.Items["release"].OK {
+ t.Fatalf("report at open: %s %+v", r.Phase, r.Items["release"])
+ }
+ lw.requireNoTargetLease()
+
+ // An item the open's ensure mends (a timer lost) does not block it.
+ lw.promote("v1")
+ if err := lw.uk.reg.ReplaceTimers(offHost, nil); err != nil {
+ t.Fatal(err)
+ }
+ lw.uk.tick()
+ rt := lw.uk.row(offHost)
+ if rt == nil || rt.Status != "ready" || rt.HoldOp != "" {
+ t.Fatalf("not opened: %+v (%+v)", rt, lw.report().Items)
+ }
+ if rows, _ := lw.uk.reg.TimersFor(offHost); moves.TimersDigest(rows) != lw.seal.TimersDigest {
+ t.Fatal("timers not mended before the release")
+ }
+ r := lw.report()
+ if r.Phase != moves.OfflinePhaseOpened {
+ t.Fatalf("phase %s", r.Phase)
+ }
+ if ok, bad := r.ItemsOK(moves.OfflineOwnerLanding); !ok {
+ t.Fatalf("opened report: %v", bad)
+ }
+ lw.uk.edges()
+ if !lw.uk.lists(offHost) {
+ t.Fatal("the edges file does not list the opened host")
+ }
+ if got := lw.s3.current(t, "data-uk", moves.OfflineArchivePrefix(offHost, offID)); len(got) != 0 {
+ t.Fatalf("the archive is still there: %v", keysOf(got))
+ }
+ lw.requireNoTargetLease()
+ lw.requireSourceBytes()
+
+ // The first wake's restore: staging promoted into data-uk's own prefix,
+ // the database restored and checked.
+ m, err := lease.NewS3Mirror(ctx, "eu-west-2", "eu-west-2")
+ if err != nil {
+ t.Fatal(err)
+ }
+ eb := &activator.ExecBackend{Cfg: activator.Config{Bucket: "data-uk", Region: "eu-west-2", Prefix: "instances"}, DataDir: lw.uk.dd, OptDir: lw.uk.opt, Mirror: m}
+ fresh, err := eb.Restore(ctx, *rt)
+ if err != nil || fresh {
+ t.Fatalf("restore: fresh %v, %v", fresh, err)
+ }
+ if err := lease.CheckDB(rt.DBPath); err != nil {
+ t.Fatal(err)
+ }
+ db, err := sql.Open("sqlite", rt.DBPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var bodies []string
+ rows, err := db.Query(`SELECT body FROM notes ORDER BY id`)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for rows.Next() {
+ var b string
+ rows.Scan(&b)
+ bodies = append(bodies, b)
+ }
+ rows.Close()
+ db.Close()
+ if fmt.Sprint(bodies) != "[one two three]" {
+ t.Fatalf("restored rows %v", bodies)
+ }
+ if got := lw.s3.current(t, "data-uk", activator.MovesPrefix(offHost)); len(got) != 0 {
+ t.Fatalf("staging left after the promotion: %v", keysOf(got))
+ }
+
+ // Later passes at open keep the report and do not re-verify the
+ // staging the wake consumed.
+ lw.uk.tick()
+ lw.restart()
+ lw.uk.tick()
+ if r := lw.report(); r.Phase != moves.OfflinePhaseOpened {
+ t.Fatalf("phase after the wake: %s", r.Phase)
+ }
+ if rt := lw.uk.row(offHost); rt.Status != "ready" {
+ t.Fatalf("row %+v", rt)
+ }
+}
+
+// --- (e) the target's revert ----------------------------------------------------------
+
+// An abort after the flip parks what the target landed (files, env, config)
+// under the move's target park directory, deletes the held row and its
+// timers, reports reverted, and leaves data-uk instances/<host>/ untouched;
+// a later pass changes nothing.
+func TestOfflineLandTargetRevert(t *testing.T) {
+ lw := newLandWorld(t)
+ lw.flipAndLand(1)
+ lw.requireLanded()
+ landed := tree(t, filepath.Join(lw.uk.dd, offHost))
+ lw.wd.abort()
+ lw.uk.tick()
+ if rt := lw.uk.row(offHost); rt != nil {
+ t.Fatalf("the row is still there: %+v", rt)
+ }
+ if rows, _ := lw.uk.reg.TimersFor(offHost); len(rows) != 0 {
+ t.Fatalf("timers left: %+v", rows)
+ }
+ for _, name := range []string{offHost, offHost + ".env", offHost + ".config.env"} {
+ if lexists(filepath.Join(lw.uk.dd, name)) {
+ t.Fatalf("%s left in place", name)
+ }
+ }
+ parkDir := moves.OfflineParkDir(lw.uk.dd, offHost, offID, moves.OfflineRoleTarget)
+ if got := tree(t, filepath.Join(parkDir, offHost)); !reflect.DeepEqual(got, landed) {
+ t.Fatalf("parked files %v, landed %v", got, landed)
+ }
+ for _, name := range []string{offHost + ".env", offHost + ".config.env"} {
+ if !lexists(filepath.Join(parkDir, name)) {
+ t.Fatalf("%s not parked", name)
+ }
+ }
+ if r := lw.report(); r.Phase != moves.OfflinePhaseReverted || r.RowPresent {
+ t.Fatalf("report %+v", r)
+ }
+ lw.requireNoTargetLease()
+ if lw.state() != provision.OfflineReverting {
+ t.Fatalf("state %s", lw.state())
+ }
+ disk := tree(t, lw.uk.dd)
+ lw.restart()
+ lw.uk.tick()
+ lw.uk.tick()
+ if rt := lw.uk.row(offHost); rt != nil {
+ t.Fatalf("the row came back: %+v", rt)
+ }
+ if got := tree(t, lw.uk.dd); !reflect.DeepEqual(got, disk) {
+ t.Fatal("a later revert pass changed the disk")
+ }
+ lw.uk.edges()
+ if lw.uk.lists(offHost) {
+ t.Fatal("listed after the revert")
+ }
+}
+
+// An abort after the flip on a target that never landed: the held row
+// reconcile minted is deleted, nothing is parked, and a target that never
+// had a row does nothing at all.
+func TestOfflineLandTargetRevertBeforeTheLanding(t *testing.T) {
+ lw := newLandWorld(t)
+ // The target mints its row held and runs its moves pass, which crashes
+ // before anything is landed or reported.
+ lw.uk.w.cfg.landFault = func(p string) error {
+ if p == "offline-warm-cache" {
+ return errors.New("crash")
+ }
+ return nil
+ }
+ lw.flipAndLand(1)
+ if rt := lw.uk.row(offHost); rt == nil || rt.HoldOp != offID {
+ t.Fatalf("row %+v", rt)
+ }
+ if _, ok := lw.wd.reportIf(moves.OfflineRoleTarget, "uk-a"); ok {
+ t.Fatal("reported before the crash point")
+ }
+ lw.wd.abort()
+ lw.restart()
+ lw.uk.tick()
+ if rt := lw.uk.row(offHost); rt != nil {
+ t.Fatalf("row %+v", rt)
+ }
+ if isDir(moves.OfflineParkDir(lw.uk.dd, offHost, offID, moves.OfflineRoleTarget)) {
+ t.Fatal("parked something it never landed")
+ }
+ if r := lw.report(); r.Phase != moves.OfflinePhaseReverted {
+ t.Fatalf("report %+v", r)
+ }
+
+ lw2 := newLandWorld(t)
+ lw2.wd.flip()
+ lw2.wd.abort()
+ lw2.uk.tick()
+ if rt := lw2.uk.row(offHost); rt != nil {
+ t.Fatalf("row %+v", rt)
+ }
+ if _, ok := lw2.wd.reportIf(moves.OfflineRoleTarget, "uk-a"); ok {
+ t.Fatal("a target that never had a row reported")
+ }
+}
+
+// The instance key the app is handed may come from the app's configuration
+// rather than <host>.env (instancekey.Effective). The env item judges the key
+// the target would hand over, from both files as the activator merges them:
+// a configured key lands; a configured key changed after the seal fails the
+// env item by name (and the env pin), so the target never hands the app a key
+// that is not the source's.
+func TestOfflineLandConfiguredInstanceKey(t *testing.T) {
+ lw := newLandWorldWith(t, landKeyValue)
+ lw.flipAndLand(1)
+ lw.requireLanded()
+ lw.requireSourceBytes()
+ env, _, _, err := lw.ukEB.LaunchLayers(*lw.uk.row(offHost))
+ if err != nil || env[instancekey.Var] != landKeyValue {
+ t.Fatalf("the target hands over %q (%v)", env[instancekey.Var], err)
+ }
+
+ lw2 := newLandWorldWith(t, landKeyValue)
+ lw2.configKey = "ik_ffffffffffffffffffffffffffffffff"
+ lw2.bundle(landStore)
+ lw2.flipAndLand(2)
+ if st := lw2.state(); st != provision.OfflineLanding {
+ t.Fatalf("state %s", st)
+ }
+ r := lw2.report()
+ if c := r.Items["env"]; c.OK || !strings.Contains(c.Reason, "instance key") || r.Phase != moves.OfflinePhaseFailed {
+ t.Fatalf("env %+v, phase %s", c, r.Phase)
+ }
+ if r.Items["env-pin"].OK {
+ t.Fatal("the env pin passed with another key")
+ }
+}