amadan

carlos / platform Public

host: split the landing into orchestration, park adapters and install helpers (no behaviour change)

movesland.go keeps landingPass and the phases; moveslandpark.go holds the
park engine assembly, the stand-in view and authorizer, the generation
classifier and the own-location lease; moveslandinstall.go holds the
archive extraction, the sealed timers, the database install, the rebuild
and the fsync helpers. Every block moved verbatim.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude ba73b152860d88ebc798feb867f38433dfdb1aad parent c8bfef4
4 files changed, +499 −451
  • approved_moves_copy_test.go +2 −0
  • internal/host/movesland.go +0 −451
  • internal/host/moveslandinstall.go +321 −0
  • internal/host/moveslandpark.go +176 −0
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index a035ae96d..8bf6ace78 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -558,6 +558,8 @@ var movesScanFiles = []string{
"internal/host/movesheldlock.go",
"internal/host/moveshold.go",
"internal/host/movesland.go",
+ "internal/host/moveslandinstall.go",
+ "internal/host/moveslandpark.go",
"internal/host/movesmeasure.go",
"internal/host/movesmeasure_linux.go",
"internal/host/movesmeasure_other.go",
diff --git a/internal/host/movesland.go b/internal/host/movesland.go
index bce15a9d7..746de3755 100644
--- a/internal/host/movesland.go
+++ b/internal/host/movesland.go
@@ -2,11 +2,9 @@ package host
import (
"context"
- "crypto/ecdh"
"encoding/json"
"errors"
"fmt"
- "io"
"log"
"os"
"path/filepath"
@@ -15,7 +13,6 @@ import (
"strings"
"time"
- "github.com/carlosframework/platform/internal/appconfig"
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/hostguard"
"github.com/carlosframework/platform/internal/hostwrite"
@@ -105,12 +102,6 @@ const (
// C8 (the prepared-manifest build, region HA spec 1 §5.8) is not on main.
const landingFilesManifestSkipped = "skipped: the prepared files manifest needs C8 (files.BuildPrepared)"
-// filesGenMarker is the .files-gen body (§5.1): the op and the incarnation.
-type filesGenMarker struct {
- Op string `json:"op"`
- Incarnation string `json:"incarnation"`
-}
-
// landingRun is one pass's state for one landing record.
type landingRun struct {
w *movesWorker
@@ -363,87 +354,6 @@ func (l *landingRun) landingCheck(phase string) func(*hostclaim.Claim) error {
}
}
-// authorizeChore is park.Local.Authorize's stand-in until Plan 5: a chore is
-// run only when it is this landing's own and the claim still names this
-// landing with this box executing.
-func (l *landingRun) authorizeChore(_ context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
- if c == nil || o.Op != l.g.ID || c.Op.ID != l.g.ID || c.Op.Kind != hostclaim.KindLanding || c.Op.Exec != l.cfg().Box {
- return fmt.Errorf("%w: not this landing's chore", park.ErrUnsafe)
- }
- return nil
-}
-
-// authorizeObligation is park.Engine.Authorize's stand-in (ruling G17): the
-// same rule over an observation.
-func (l *landingRun) authorizeObligation(ctx context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
- if obs.Absent {
- return park.ErrUnsafe
- }
- return l.authorizeChore(ctx, obs.Claim, o)
-}
-
-// landingView is park.Engine.View's stand-in (ruling G17): the location's
-// objects classified by key, and the source's region authoritative, which
-// §5.1's table says it is for a landing before the target's probe is
-// admitted (every phase here). Plan 5's provider replaces it; a test may set
-// cfg.ParkView to the view a later state has.
-func (l *landingRun) landingView(_ context.Context, obs hostclaim.Observed) (park.View, error) {
- if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil {
- return park.View{}, park.ErrUnsafe
- }
- location := strings.TrimSuffix(obs.Claim.Holder.Location, "/")
- return park.View{ClaimETag: obs.ETag, Region: l.cfg().Region, AuthoritativeRegion: l.rec.Move.From,
- GenerationOf: func(r park.Ref) (string, bool) { return landingGenerationOf(location, r.Key) }}, nil
-}
-
-// landingMarkerGeneration is what landingGenerationOf answers for the
-// location's generation.json. The marker is the one key every generation
-// reuses (§5.1): the probe's admission supersedes an earlier stay's marker
-// with If-Match rather than deleting it, and a park-generation chore whose
-// deletion set named the marker key could never be acknowledged once a new
-// marker stands there (the engine skips the marker key and then refuses a
-// recorded delete at it). So the marker belongs to no generation's chore:
-// parking never queues one for it and the engine passes it over. "@" keeps
-// it apart from any generation name, as hostclaim.RootGeneration does.
-const landingMarkerGeneration = "@marker"
-
-// landingGenerationOf classifies an object under a holder location by
-// generation, as the view does: `g-<gen>/...` and `files/g-<gen>/...` are
-// gen's; a level directory (four hex digits) at the location is the
-// origin, ""; generation.json is landingMarkerGeneration; anything else is
-// unknown, and a caller fails closed on it. The naming rule is
-// moves.GenerationPrefix's twin.
-func landingGenerationOf(location, key string) (string, bool) {
- rel, ok := strings.CutPrefix(key, location+"/")
- if !ok || rel == "" {
- return "", false
- }
- for _, p := range []string{"g-", "files/g-"} {
- if rest, ok := strings.CutPrefix(rel, p); ok {
- if gen, _, ok := strings.Cut(rest, "/"); ok && gen != "" {
- return gen, true
- }
- return "", false
- }
- }
- if rel == "generation.json" {
- return landingMarkerGeneration, true
- }
- if first, _, ok := strings.Cut(rel, "/"); ok && len(first) == 4 && isHexString(first) {
- return "", true
- }
- return "", false
-}
-
-func isHexString(s string) bool {
- for _, c := range s {
- if (c < '0' || c > '9') && (c < 'a' || c > 'f') && (c < 'A' || c > 'F') {
- return false
- }
- }
- return s != ""
-}
-
// route is the host's registry row on this box.
func (l *landingRun) route() (registry.Route, error) {
rt, err := l.cfg().Route(l.rec.Host)
@@ -743,80 +653,6 @@ func (l *landingRun) parking(ctx context.Context) (bool, error) {
return l.advance(ctx, provision.LandParking, provision.LandBuilding, nil)
}
-// engine is the park engine over this box's own bucket: the wired leases,
-// and the view and authorizer Plan 5 supplies, or the stand-ins.
-func (l *landingRun) engine(vs lease.VersionedStore) park.Engine {
- cfg := l.cfg()
- view, auth := cfg.ParkView, cfg.ParkAuthorize
- if view == nil {
- view = l.landingView
- }
- if auth == nil {
- auth = l.authorizeObligation
- }
- return park.Engine{Store: vs, Claims: l.s, Bucket: cfg.OwnBucket, Region: cfg.Region, Leases: cfg.ParkLeases,
- Now: cfg.Now, NewID: obligationID, View: view, Authorize: auth, After: cfg.parkFault}
-}
-
-// lease is the own-location lease the build holds (§7.6 checking: "hold it
-// through published"), kept in the worker's per-host map across passes and
-// re-acquired after a restart. waiting is true when another holder has it.
-func (l *landingRun) lease(ctx context.Context, holder hostclaim.Holder) (held park.HeldLease, waiting bool, err error) {
- cfg := l.cfg()
- host := l.rec.Host
- l.w.mu.Lock()
- held = l.w.landLeases[host]
- l.w.mu.Unlock()
- if held != nil {
- if err := held.Check(ctx); err == nil {
- return held, false, nil
- }
- l.releaseLease()
- }
- held, err = cfg.OwnLease(ctx, cfg.OwnBucket, strings.TrimSuffix(holder.Location, "/"))
- if err != nil {
- var he *lease.HeldError
- if errors.As(err, &he) {
- log.Printf("moves: %s: another holder has the location's lease; the build waits", host)
- return nil, true, nil
- }
- return nil, false, fmt.Errorf("the location's lease: %w", err)
- }
- l.w.mu.Lock()
- if l.w.landLeases == nil {
- l.w.landLeases = map[string]park.HeldLease{}
- }
- l.w.landLeases[host] = held
- l.w.mu.Unlock()
- return held, false, nil
-}
-
-// releaseLease releases the host's own-location lease, if held.
-func (l *landingRun) releaseLease() {
- host := l.rec.Host
- l.w.mu.Lock()
- held := l.w.landLeases[host]
- delete(l.w.landLeases, host)
- l.w.mu.Unlock()
- if held == nil {
- return
- }
- rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
- defer cancel()
- if err := held.Release(rctx); err != nil {
- log.Printf("moves: %s: release the location's lease: %v", host, err)
- }
-}
-
-// settleLease releases the lease once the record is past published, marked
-// for a rollback, or no longer landing: the lease is the build's alone.
-func (l *landingRun) settleLease() {
- m := l.rec.Move
- if m == nil || m.State != provision.MoveLanding || m.Reason != "" || !landingBefore(m.LandingPhase, provision.LandLocal) {
- l.releaseLease()
- }
-}
-
// building (§7.6, §5.1 step 2): the generation built from the seal's pinned
// replica versions under the location's lease (moves.BuildGeneration: the
// intent, the snapshot, the anchor, the completed install manifest, every
@@ -1095,116 +931,6 @@ func (l *landingRun) local_(ctx context.Context) (bool, error) {
return l.advance(ctx, provision.LandLocal, provision.LandConfigured, nil)
}
-// extractArchive reads the seal's archive and manifest by exact version from
-// this box's own bucket, verifies the archive against the manifest and the
-// record's digests (archive.Verify: the sealed digest, every entry), and
-// extracts it into dir and envPath (archive.Open under the box's secrets
-// key, then archive.Extract, which admits nothing it did not read whole).
-func (l *landingRun) extractArchive(ctx context.Context, seal moves.Seal, want provision.MoveFilesDigest, dir, envPath string) error {
- cfg := l.cfg()
- var sealedObj, manifestObj *moves.SealedObject
- for i := range seal.Files {
- switch o := &seal.Files[i]; {
- case strings.HasSuffix(o.Key, ".sealed"):
- sealedObj = o
- case strings.HasSuffix(o.Key, ".json"):
- manifestObj = o
- }
- }
- if sealedObj == nil || manifestObj == nil {
- return errors.New("the seal does not name the archive and its manifest")
- }
- vs, err := l.versioned(ctx)
- if err != nil {
- return err
- }
- raw, err := vs.GetVersion(ctx, cfg.OwnBucket, manifestObj.Key, manifestObj.VersionID)
- if err != nil {
- return fmt.Errorf("read the archive manifest: %w", err)
- }
- var m archive.Manifest
- if err := json.Unmarshal(raw, &m); err != nil {
- return fmt.Errorf("the archive manifest: %w", err)
- }
- if m.Op != l.g.ID || m.Incarnation != l.rec.Move.Incarnation || m.Plain != want.Plain || m.Sealed != want.Sealed {
- return fmt.Errorf("the archive manifest is for %s/%s (%s, %s), not this move's %s (%s, %s)", m.Op, m.Incarnation, m.Plain, m.Sealed, l.g.ID, want.Plain, want.Sealed)
- }
- if err := os.MkdirAll(l.journalDir(), 0o700); err != nil {
- return err
- }
- sealedPath := filepath.Join(l.journalDir(), "archive.sealed")
- if err := os.Remove(sealedPath); err != nil && !errors.Is(err, os.ErrNotExist) {
- return err
- }
- defer os.Remove(sealedPath)
- if err := downloadVersion(ctx, vs, cfg.OwnBucket, sealedObj.Key, sealedObj.VersionID, sealedPath); err != nil {
- return fmt.Errorf("read the archive: %w", err)
- }
- f, err := os.Open(sealedPath)
- if err != nil {
- return err
- }
- defer f.Close()
- bind := appconfig.Binding{Account: l.rec.Account, App: l.rec.App}
- if err := archive.Verify(f, m, cfg.SecretsPriv, bind); err != nil {
- return fmt.Errorf("verify the archive: %w", err)
- }
- return openAndExtract(f, m, cfg.SecretsPriv, bind, dir, envPath)
-}
-
-// openAndExtract pipes the opened archive into Extract, as carriedEnv does.
-func openAndExtract(f *os.File, m archive.Manifest, priv *ecdh.PrivateKey, bind appconfig.Binding, dir, envPath string) error {
- if _, err := f.Seek(0, io.SeekStart); err != nil {
- return err
- }
- pr, pw := io.Pipe()
- done := make(chan error, 1)
- go func() {
- err := archive.Open(f, m.SealedDataKey, priv, bind, pw)
- pw.CloseWithError(err)
- done <- err
- }()
- xerr := archive.Extract(pr, dir, envPath)
- pr.CloseWithError(errors.New("extract stopped reading"))
- if oerr := <-done; oerr != nil && xerr == nil {
- xerr = oerr
- }
- return xerr
-}
-
-// sealedTimers reads the sealed timers snapshot by exact version and checks
-// its digest against the record's; every row is this host's.
-func (l *landingRun) sealedTimers(ctx context.Context, seal moves.Seal, wantDigest string) ([]registry.Timer, error) {
- cfg := l.cfg()
- if len(seal.Timers) != 1 {
- return nil, fmt.Errorf("the seal names %d timers objects, want 1", len(seal.Timers))
- }
- vs, err := l.versioned(ctx)
- if err != nil {
- return nil, err
- }
- raw, err := vs.GetVersion(ctx, cfg.OwnBucket, seal.Timers[0].Key, seal.Timers[0].VersionID)
- if err != nil {
- return nil, fmt.Errorf("read the timers snapshot: %w", err)
- }
- snap, err := moves.DecodeTimers(raw)
- if err != nil {
- return nil, err
- }
- if snap.Host != l.rec.Host {
- return nil, fmt.Errorf("the timers snapshot is %s's, not %s's", snap.Host, l.rec.Host)
- }
- if got := moves.TimersDigest(snap.Rows); got != wantDigest {
- return nil, fmt.Errorf("the timers snapshot digests to %s, the record says %s", got, wantDigest)
- }
- for _, r := range snap.Rows {
- if r.Host != l.rec.Host {
- return nil, fmt.Errorf("timer %q is %s's, not %s's", r.Name, r.Host, l.rec.Host)
- }
- }
- return snap.Rows, nil
-}
-
// configured (§7.6, §9.1, §9.2): the held row rendered (RenderHeld, the
// move-bound ack), its bundle and overlay etags required to be the ones
// pinned at readiness (else the rollback marker, "configuration changed
@@ -1265,180 +991,3 @@ func (l *landingRun) configured(ctx context.Context) (bool, error) {
m.Landing.InstalledDBSHA256 = sha
})
}
-
-// sameBundle compares two bundle etag maps.
-func sameBundle(a, b map[string]string) bool {
- if len(a) != len(b) {
- return false
- }
- for k, v := range a {
- if bv, ok := b[k]; !ok || bv != v {
- return false
- }
- }
- return true
-}
-
-// restore is the install of the built database (§7.6 restored, R6-3):
-// its sha256 must be the manifest's and CheckDB must pass, else the
-// rollback marker; then the rename into DBPath and the generation marker.
-// installed is false when the phase ended with the marker written. The
-// returned sha carries the record's "sha256:" prefix, as Source.DBSHA256
-// does.
-func (l *landingRun) restore(ctx context.Context, obs hostclaim.Observed) (sha string, installed bool, err error) {
- cfg := l.cfg()
- host, id := l.rec.Host, l.g.ID
- m, err := l.installManifest(ctx, obs)
- if err != nil {
- return "", false, err
- }
- want := "sha256:" + m.DBSHA256
- dbPath := l.dbPath()
- present, err := regularFile(dbPath)
- if err != nil {
- return "", false, err
- }
- if present {
- got, err := fileSHA256(dbPath)
- if err != nil {
- return "", false, err
- }
- if got != want {
- return "", false, fmt.Errorf("%s holds a database that is not this landing's build (%s, want %s)", dbPath, got, want)
- }
- } else {
- built := filepath.Join(l.buildDir(), host+".db")
- if ok, err := regularFile(built); err != nil {
- return "", false, err
- } else if !ok {
- // The scratch was lost (a rebuilt data directory): build again
- // from the pinned set, which reproduces the same bytes.
- if _, err := l.rebuild(ctx, obs); err != nil {
- return "", false, err
- }
- }
- got, err := fileSHA256(built)
- if err != nil {
- return "", false, err
- }
- if got != want {
- log.Printf("moves: %s: the built database is %s, the manifest says %s", host, got, want)
- ok, err := l.markRollback(ctx, provision.RollbackInstallMismatch)
- return "", ok, err
- }
- if err := lease.CheckDB(built); err != nil {
- log.Printf("moves: %s: the built database does not pass its check: %v", host, err)
- ok, err := l.markRollback(ctx, provision.RollbackInstallMismatch)
- return "", ok, err
- }
- for _, side := range []string{"-wal", "-shm"} {
- os.Remove(built + side)
- }
- if err := os.Rename(built, dbPath); err != nil {
- return "", false, fmt.Errorf("install the database: %w", err)
- }
- if err := syncPath(filepath.Dir(dbPath)); err != nil {
- return "", false, err
- }
- }
- if err := l.fault("after-db-rename"); err != nil {
- return "", false, err
- }
- if err := cfg.WriteGen(dbPath+".gen", id); err != nil {
- return "", false, fmt.Errorf("write the generation marker: %w", err)
- }
- if err := syncPath(filepath.Dir(dbPath)); err != nil {
- return "", false, err
- }
- log.Printf("moves: %s: database installed at %s (%s), generation %s", host, dbPath, want, id)
- return want, true, nil
-}
-
-// installManifest is the journal's, or the build's again when the journal
-// lost it.
-func (l *landingRun) installManifest(ctx context.Context, obs hostclaim.Observed) (moves.InstallManifest, error) {
- j, err := l.journal()
- if err != nil {
- return moves.InstallManifest{}, err
- }
- var m moves.InstallManifest
- if raw, ok := j.Steps[landStepInstall]; ok && json.Unmarshal([]byte(raw), &m) == nil && m.Op == l.g.ID && m.Host == l.rec.Host && m.DBSHA256 != "" {
- return m, nil
- }
- return l.rebuild(ctx, obs)
-}
-
-// rebuild runs BuildGeneration again over the seal: a replay that reuses
-// every object it finds and restores the same bytes into the scratch.
-func (l *landingRun) rebuild(ctx context.Context, obs hostclaim.Observed) (moves.InstallManifest, error) {
- cfg := l.cfg()
- seal, err := journalledSeal(obs.Claim)
- if err != nil {
- return moves.InstallManifest{}, err
- }
- vs, err := l.versioned(ctx)
- if err != nil {
- return moves.InstallManifest{}, err
- }
- holder := *obs.Claim.Holder
- m, err := moves.BuildGeneration(ctx, moves.BuildInputs{Store: vs, Bucket: cfg.OwnBucket, Source: seal.Replica, N: seal.N,
- GenerationPrefix: moves.GenerationPrefix(holder.Location, l.g.ID), Op: l.g.ID, Host: l.rec.Host, WorkDir: l.buildDir(), Now: cfg.Now})
- if err != nil {
- return moves.InstallManifest{}, fmt.Errorf("rebuild the generation: %w", err)
- }
- j, err := l.journal()
- if err != nil {
- return m, err
- }
- body, err := json.Marshal(m)
- if err != nil {
- return m, err
- }
- j.Mark(landStepInstall, string(body))
- return m, j.Write(cfg.DataDir)
-}
-
-// writeFileAtomic writes body to path through a temporary file in the same
-// directory, synced and renamed, then syncs the directory.
-func writeFileAtomic(path string, body []byte, mode os.FileMode) error {
- dir := filepath.Dir(path)
- f, err := os.CreateTemp(dir, "."+filepath.Base(path)+".*")
- if err != nil {
- return err
- }
- tmp := f.Name()
- if _, err := f.Write(body); err != nil {
- f.Close()
- os.Remove(tmp)
- return err
- }
- if err := f.Chmod(mode); err != nil {
- f.Close()
- os.Remove(tmp)
- return err
- }
- if err := f.Sync(); err != nil {
- f.Close()
- os.Remove(tmp)
- return err
- }
- if err := f.Close(); err != nil {
- os.Remove(tmp)
- return err
- }
- if err := os.Rename(tmp, path); err != nil {
- os.Remove(tmp)
- return err
- }
- return syncPath(dir)
-}
-
-// syncPath fsyncs a file or directory.
-func syncPath(path string) error {
- f, err := os.Open(path)
- if err != nil {
- return err
- }
- defer f.Close()
- return f.Sync()
-}
diff --git a/internal/host/moveslandinstall.go b/internal/host/moveslandinstall.go
new file mode 100644
index 000000000..509343b9c
--- /dev/null
+++ b/internal/host/moveslandinstall.go
@@ -0,0 +1,321 @@
+package host
+
+import (
+ "context"
+ "crypto/ecdh"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log"
+ "os"
+ "path/filepath"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "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"
+)
+
+// The landing's install helpers (moves spec §7.6 local, restored): the
+// sealed archive read by exact version, verified and extracted; the sealed
+// timers; the database install from the build scratch with its marker; the
+// build again when the journal lost it; and the fsync and atomic-write
+// helpers. landingPass and the phases are in movesland.go.
+
+// filesGenMarker is the .files-gen body (§5.1): the op and the incarnation.
+type filesGenMarker struct {
+ Op string `json:"op"`
+ Incarnation string `json:"incarnation"`
+}
+
+// extractArchive reads the seal's archive and manifest by exact version from
+// this box's own bucket, verifies the archive against the manifest and the
+// record's digests (archive.Verify: the sealed digest, every entry), and
+// extracts it into dir and envPath (archive.Open under the box's secrets
+// key, then archive.Extract, which admits nothing it did not read whole).
+func (l *landingRun) extractArchive(ctx context.Context, seal moves.Seal, want provision.MoveFilesDigest, dir, envPath string) error {
+ cfg := l.cfg()
+ var sealedObj, manifestObj *moves.SealedObject
+ for i := range seal.Files {
+ switch o := &seal.Files[i]; {
+ case strings.HasSuffix(o.Key, ".sealed"):
+ sealedObj = o
+ case strings.HasSuffix(o.Key, ".json"):
+ manifestObj = o
+ }
+ }
+ if sealedObj == nil || manifestObj == nil {
+ return errors.New("the seal does not name the archive and its manifest")
+ }
+ vs, err := l.versioned(ctx)
+ if err != nil {
+ return err
+ }
+ raw, err := vs.GetVersion(ctx, cfg.OwnBucket, manifestObj.Key, manifestObj.VersionID)
+ if err != nil {
+ return fmt.Errorf("read the archive manifest: %w", err)
+ }
+ var m archive.Manifest
+ if err := json.Unmarshal(raw, &m); err != nil {
+ return fmt.Errorf("the archive manifest: %w", err)
+ }
+ if m.Op != l.g.ID || m.Incarnation != l.rec.Move.Incarnation || m.Plain != want.Plain || m.Sealed != want.Sealed {
+ return fmt.Errorf("the archive manifest is for %s/%s (%s, %s), not this move's %s (%s, %s)", m.Op, m.Incarnation, m.Plain, m.Sealed, l.g.ID, want.Plain, want.Sealed)
+ }
+ if err := os.MkdirAll(l.journalDir(), 0o700); err != nil {
+ return err
+ }
+ sealedPath := filepath.Join(l.journalDir(), "archive.sealed")
+ if err := os.Remove(sealedPath); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ defer os.Remove(sealedPath)
+ if err := downloadVersion(ctx, vs, cfg.OwnBucket, sealedObj.Key, sealedObj.VersionID, sealedPath); err != nil {
+ return fmt.Errorf("read the archive: %w", err)
+ }
+ f, err := os.Open(sealedPath)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ bind := appconfig.Binding{Account: l.rec.Account, App: l.rec.App}
+ if err := archive.Verify(f, m, cfg.SecretsPriv, bind); err != nil {
+ return fmt.Errorf("verify the archive: %w", err)
+ }
+ return openAndExtract(f, m, cfg.SecretsPriv, bind, dir, envPath)
+}
+
+// openAndExtract pipes the opened archive into Extract, as carriedEnv does.
+func openAndExtract(f *os.File, m archive.Manifest, priv *ecdh.PrivateKey, bind appconfig.Binding, dir, envPath string) error {
+ if _, err := f.Seek(0, io.SeekStart); err != nil {
+ return err
+ }
+ pr, pw := io.Pipe()
+ done := make(chan error, 1)
+ go func() {
+ err := archive.Open(f, m.SealedDataKey, priv, bind, pw)
+ pw.CloseWithError(err)
+ done <- err
+ }()
+ xerr := archive.Extract(pr, dir, envPath)
+ pr.CloseWithError(errors.New("extract stopped reading"))
+ if oerr := <-done; oerr != nil && xerr == nil {
+ xerr = oerr
+ }
+ return xerr
+}
+
+// sealedTimers reads the sealed timers snapshot by exact version and checks
+// its digest against the record's; every row is this host's.
+func (l *landingRun) sealedTimers(ctx context.Context, seal moves.Seal, wantDigest string) ([]registry.Timer, error) {
+ cfg := l.cfg()
+ if len(seal.Timers) != 1 {
+ return nil, fmt.Errorf("the seal names %d timers objects, want 1", len(seal.Timers))
+ }
+ vs, err := l.versioned(ctx)
+ if err != nil {
+ return nil, err
+ }
+ raw, err := vs.GetVersion(ctx, cfg.OwnBucket, seal.Timers[0].Key, seal.Timers[0].VersionID)
+ if err != nil {
+ return nil, fmt.Errorf("read the timers snapshot: %w", err)
+ }
+ snap, err := moves.DecodeTimers(raw)
+ if err != nil {
+ return nil, err
+ }
+ if snap.Host != l.rec.Host {
+ return nil, fmt.Errorf("the timers snapshot is %s's, not %s's", snap.Host, l.rec.Host)
+ }
+ if got := moves.TimersDigest(snap.Rows); got != wantDigest {
+ return nil, fmt.Errorf("the timers snapshot digests to %s, the record says %s", got, wantDigest)
+ }
+ for _, r := range snap.Rows {
+ if r.Host != l.rec.Host {
+ return nil, fmt.Errorf("timer %q is %s's, not %s's", r.Name, r.Host, l.rec.Host)
+ }
+ }
+ return snap.Rows, nil
+}
+
+// sameBundle compares two bundle etag maps.
+func sameBundle(a, b map[string]string) bool {
+ if len(a) != len(b) {
+ return false
+ }
+ for k, v := range a {
+ if bv, ok := b[k]; !ok || bv != v {
+ return false
+ }
+ }
+ return true
+}
+
+// restore is the install of the built database (§7.6 restored, R6-3):
+// its sha256 must be the manifest's and CheckDB must pass, else the
+// rollback marker; then the rename into DBPath and the generation marker.
+// installed is false when the phase ended with the marker written. The
+// returned sha carries the record's "sha256:" prefix, as Source.DBSHA256
+// does.
+func (l *landingRun) restore(ctx context.Context, obs hostclaim.Observed) (sha string, installed bool, err error) {
+ cfg := l.cfg()
+ host, id := l.rec.Host, l.g.ID
+ m, err := l.installManifest(ctx, obs)
+ if err != nil {
+ return "", false, err
+ }
+ want := "sha256:" + m.DBSHA256
+ dbPath := l.dbPath()
+ present, err := regularFile(dbPath)
+ if err != nil {
+ return "", false, err
+ }
+ if present {
+ got, err := fileSHA256(dbPath)
+ if err != nil {
+ return "", false, err
+ }
+ if got != want {
+ return "", false, fmt.Errorf("%s holds a database that is not this landing's build (%s, want %s)", dbPath, got, want)
+ }
+ } else {
+ built := filepath.Join(l.buildDir(), host+".db")
+ if ok, err := regularFile(built); err != nil {
+ return "", false, err
+ } else if !ok {
+ // The scratch was lost (a rebuilt data directory): build again
+ // from the pinned set, which reproduces the same bytes.
+ if _, err := l.rebuild(ctx, obs); err != nil {
+ return "", false, err
+ }
+ }
+ got, err := fileSHA256(built)
+ if err != nil {
+ return "", false, err
+ }
+ if got != want {
+ log.Printf("moves: %s: the built database is %s, the manifest says %s", host, got, want)
+ ok, err := l.markRollback(ctx, provision.RollbackInstallMismatch)
+ return "", ok, err
+ }
+ if err := lease.CheckDB(built); err != nil {
+ log.Printf("moves: %s: the built database does not pass its check: %v", host, err)
+ ok, err := l.markRollback(ctx, provision.RollbackInstallMismatch)
+ return "", ok, err
+ }
+ for _, side := range []string{"-wal", "-shm"} {
+ os.Remove(built + side)
+ }
+ if err := os.Rename(built, dbPath); err != nil {
+ return "", false, fmt.Errorf("install the database: %w", err)
+ }
+ if err := syncPath(filepath.Dir(dbPath)); err != nil {
+ return "", false, err
+ }
+ }
+ if err := l.fault("after-db-rename"); err != nil {
+ return "", false, err
+ }
+ if err := cfg.WriteGen(dbPath+".gen", id); err != nil {
+ return "", false, fmt.Errorf("write the generation marker: %w", err)
+ }
+ if err := syncPath(filepath.Dir(dbPath)); err != nil {
+ return "", false, err
+ }
+ log.Printf("moves: %s: database installed at %s (%s), generation %s", host, dbPath, want, id)
+ return want, true, nil
+}
+
+// installManifest is the journal's, or the build's again when the journal
+// lost it.
+func (l *landingRun) installManifest(ctx context.Context, obs hostclaim.Observed) (moves.InstallManifest, error) {
+ j, err := l.journal()
+ if err != nil {
+ return moves.InstallManifest{}, err
+ }
+ var m moves.InstallManifest
+ if raw, ok := j.Steps[landStepInstall]; ok && json.Unmarshal([]byte(raw), &m) == nil && m.Op == l.g.ID && m.Host == l.rec.Host && m.DBSHA256 != "" {
+ return m, nil
+ }
+ return l.rebuild(ctx, obs)
+}
+
+// rebuild runs BuildGeneration again over the seal: a replay that reuses
+// every object it finds and restores the same bytes into the scratch.
+func (l *landingRun) rebuild(ctx context.Context, obs hostclaim.Observed) (moves.InstallManifest, error) {
+ cfg := l.cfg()
+ seal, err := journalledSeal(obs.Claim)
+ if err != nil {
+ return moves.InstallManifest{}, err
+ }
+ vs, err := l.versioned(ctx)
+ if err != nil {
+ return moves.InstallManifest{}, err
+ }
+ holder := *obs.Claim.Holder
+ m, err := moves.BuildGeneration(ctx, moves.BuildInputs{Store: vs, Bucket: cfg.OwnBucket, Source: seal.Replica, N: seal.N,
+ GenerationPrefix: moves.GenerationPrefix(holder.Location, l.g.ID), Op: l.g.ID, Host: l.rec.Host, WorkDir: l.buildDir(), Now: cfg.Now})
+ if err != nil {
+ return moves.InstallManifest{}, fmt.Errorf("rebuild the generation: %w", err)
+ }
+ j, err := l.journal()
+ if err != nil {
+ return m, err
+ }
+ body, err := json.Marshal(m)
+ if err != nil {
+ return m, err
+ }
+ j.Mark(landStepInstall, string(body))
+ return m, j.Write(cfg.DataDir)
+}
+
+// writeFileAtomic writes body to path through a temporary file in the same
+// directory, synced and renamed, then syncs the directory.
+func writeFileAtomic(path string, body []byte, mode os.FileMode) error {
+ dir := filepath.Dir(path)
+ f, err := os.CreateTemp(dir, "."+filepath.Base(path)+".*")
+ if err != nil {
+ return err
+ }
+ tmp := f.Name()
+ if _, err := f.Write(body); err != nil {
+ f.Close()
+ os.Remove(tmp)
+ return err
+ }
+ if err := f.Chmod(mode); err != nil {
+ f.Close()
+ os.Remove(tmp)
+ return err
+ }
+ if err := f.Sync(); err != nil {
+ f.Close()
+ os.Remove(tmp)
+ return err
+ }
+ if err := f.Close(); err != nil {
+ os.Remove(tmp)
+ return err
+ }
+ if err := os.Rename(tmp, path); err != nil {
+ os.Remove(tmp)
+ return err
+ }
+ return syncPath(dir)
+}
+
+// syncPath fsyncs a file or directory.
+func syncPath(path string) error {
+ f, err := os.Open(path)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ return f.Sync()
+}
diff --git a/internal/host/moveslandpark.go b/internal/host/moveslandpark.go
new file mode 100644
index 000000000..eeb262e27
--- /dev/null
+++ b/internal/host/moveslandpark.go
@@ -0,0 +1,176 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// The landing's park and lease adapters (moves spec §6.4, §7.6; ruling
+// G17): the park engine over this box's own bucket for the residue parks,
+// the stand-in view and authorizer the tests use until Plan 5's providers,
+// the generation classifier they share, and the own-location lease the
+// build holds. landingPass and the phases are in movesland.go.
+
+// authorizeChore is park.Local.Authorize's stand-in until Plan 5: a chore is
+// run only when it is this landing's own and the claim still names this
+// landing with this box executing.
+func (l *landingRun) authorizeChore(_ context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
+ if c == nil || o.Op != l.g.ID || c.Op.ID != l.g.ID || c.Op.Kind != hostclaim.KindLanding || c.Op.Exec != l.cfg().Box {
+ return fmt.Errorf("%w: not this landing's chore", park.ErrUnsafe)
+ }
+ return nil
+}
+
+// authorizeObligation is park.Engine.Authorize's stand-in (ruling G17): the
+// same rule over an observation.
+func (l *landingRun) authorizeObligation(ctx context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
+ if obs.Absent {
+ return park.ErrUnsafe
+ }
+ return l.authorizeChore(ctx, obs.Claim, o)
+}
+
+// landingView is park.Engine.View's stand-in (ruling G17): the location's
+// objects classified by key, and the source's region authoritative, which
+// §5.1's table says it is for a landing before the target's probe is
+// admitted (every phase here). Plan 5's provider replaces it; a test may set
+// cfg.ParkView to the view a later state has.
+func (l *landingRun) landingView(_ context.Context, obs hostclaim.Observed) (park.View, error) {
+ if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil {
+ return park.View{}, park.ErrUnsafe
+ }
+ location := strings.TrimSuffix(obs.Claim.Holder.Location, "/")
+ return park.View{ClaimETag: obs.ETag, Region: l.cfg().Region, AuthoritativeRegion: l.rec.Move.From,
+ GenerationOf: func(r park.Ref) (string, bool) { return landingGenerationOf(location, r.Key) }}, nil
+}
+
+// landingMarkerGeneration is what landingGenerationOf answers for the
+// location's generation.json. The marker is the one key every generation
+// reuses (§5.1): the probe's admission supersedes an earlier stay's marker
+// with If-Match rather than deleting it, and a park-generation chore whose
+// deletion set named the marker key could never be acknowledged once a new
+// marker stands there (the engine skips the marker key and then refuses a
+// recorded delete at it). So the marker belongs to no generation's chore:
+// parking never queues one for it and the engine passes it over. "@" keeps
+// it apart from any generation name, as hostclaim.RootGeneration does.
+const landingMarkerGeneration = "@marker"
+
+// landingGenerationOf classifies an object under a holder location by
+// generation, as the view does: `g-<gen>/...` and `files/g-<gen>/...` are
+// gen's; a level directory (four hex digits) at the location is the
+// origin, ""; generation.json is landingMarkerGeneration; anything else is
+// unknown, and a caller fails closed on it. The naming rule is
+// moves.GenerationPrefix's twin.
+func landingGenerationOf(location, key string) (string, bool) {
+ rel, ok := strings.CutPrefix(key, location+"/")
+ if !ok || rel == "" {
+ return "", false
+ }
+ for _, p := range []string{"g-", "files/g-"} {
+ if rest, ok := strings.CutPrefix(rel, p); ok {
+ if gen, _, ok := strings.Cut(rest, "/"); ok && gen != "" {
+ return gen, true
+ }
+ return "", false
+ }
+ }
+ if rel == "generation.json" {
+ return landingMarkerGeneration, true
+ }
+ if first, _, ok := strings.Cut(rel, "/"); ok && len(first) == 4 && isHexString(first) {
+ return "", true
+ }
+ return "", false
+}
+
+func isHexString(s string) bool {
+ for _, c := range s {
+ if (c < '0' || c > '9') && (c < 'a' || c > 'f') && (c < 'A' || c > 'F') {
+ return false
+ }
+ }
+ return s != ""
+}
+
+// engine is the park engine over this box's own bucket: the wired leases,
+// and the view and authorizer Plan 5 supplies, or the stand-ins.
+func (l *landingRun) engine(vs lease.VersionedStore) park.Engine {
+ cfg := l.cfg()
+ view, auth := cfg.ParkView, cfg.ParkAuthorize
+ if view == nil {
+ view = l.landingView
+ }
+ if auth == nil {
+ auth = l.authorizeObligation
+ }
+ return park.Engine{Store: vs, Claims: l.s, Bucket: cfg.OwnBucket, Region: cfg.Region, Leases: cfg.ParkLeases,
+ Now: cfg.Now, NewID: obligationID, View: view, Authorize: auth, After: cfg.parkFault}
+}
+
+// lease is the own-location lease the build holds (§7.6 checking: "hold it
+// through published"), kept in the worker's per-host map across passes and
+// re-acquired after a restart. waiting is true when another holder has it.
+func (l *landingRun) lease(ctx context.Context, holder hostclaim.Holder) (held park.HeldLease, waiting bool, err error) {
+ cfg := l.cfg()
+ host := l.rec.Host
+ l.w.mu.Lock()
+ held = l.w.landLeases[host]
+ l.w.mu.Unlock()
+ if held != nil {
+ if err := held.Check(ctx); err == nil {
+ return held, false, nil
+ }
+ l.releaseLease()
+ }
+ held, err = cfg.OwnLease(ctx, cfg.OwnBucket, strings.TrimSuffix(holder.Location, "/"))
+ if err != nil {
+ var he *lease.HeldError
+ if errors.As(err, &he) {
+ log.Printf("moves: %s: another holder has the location's lease; the build waits", host)
+ return nil, true, nil
+ }
+ return nil, false, fmt.Errorf("the location's lease: %w", err)
+ }
+ l.w.mu.Lock()
+ if l.w.landLeases == nil {
+ l.w.landLeases = map[string]park.HeldLease{}
+ }
+ l.w.landLeases[host] = held
+ l.w.mu.Unlock()
+ return held, false, nil
+}
+
+// releaseLease releases the host's own-location lease, if held.
+func (l *landingRun) releaseLease() {
+ host := l.rec.Host
+ l.w.mu.Lock()
+ held := l.w.landLeases[host]
+ delete(l.w.landLeases, host)
+ l.w.mu.Unlock()
+ if held == nil {
+ return
+ }
+ rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ if err := held.Release(rctx); err != nil {
+ log.Printf("moves: %s: release the location's lease: %v", host, err)
+ }
+}
+
+// settleLease releases the lease once the record is past published, marked
+// for a rollback, or no longer landing: the lease is the build's alone.
+func (l *landingRun) settleLease() {
+ m := l.rec.Move
+ if m == nil || m.State != provision.MoveLanding || m.Reason != "" || !landingBefore(m.LandingPhase, provision.LandLocal) {
+ l.releaseLease()
+ }
+}