amadan

carlos / platform Public

host: the source's finalization, fix round 1: the journal sweep, durable rename replay, env files, owners required

- A local sweep parks an op journal that reached the stamp step, whoever's
  record no longer routes here: a closed move proves the stamp landed
  (MoveClose refuses without source_parked_at). The stamp step is
  journalled pending before its CAS.
- The rename helper syncs the source's parent and every destination
  directory on a replay that finds the rename done, as on a fresh one, and
  offers an interruption point between the rename and its syncs.
- <host>.config.env and <host>-sidecar.env join the ownership-tracked
  retirement.
- The finalization is armed only with the park engine's leases, view and
  authorizer; nothing stands in for a missing owner, and the tests inject
  theirs. The rename helper moves to movesretire.go.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 6c5591c34a1aaa82bf8e540dc0ef1fb9a10fd4ea parent 1a9442f
5 files changed, +651 −349
  • approved_moves_copy_test.go +1 −0
  • internal/host/movesfinal.go +123 −334
  • internal/host/movesfinal_test.go +185 −15
  • internal/host/movespass.go +5 −0
  • internal/host/movesretire.go +337 −0

CI passed — run details

diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index 97371d3f9..bb7ec9b31 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -586,6 +586,7 @@ var movesScanFiles = []string{
"internal/host/movespolicy.go",
"internal/host/movesprobe.go",
"internal/host/movesreport.go",
+ "internal/host/movesretire.go",
"internal/host/movesstage.go",
"internal/host/movesstore.go",
"internal/host/refusalwatch.go",
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go
index 562786496..e02647890 100644
--- a/internal/host/movesfinal.go
+++ b/internal/host/movesfinal.go
@@ -2,7 +2,6 @@ package host
import (
"context"
- "encoding/json"
"errors"
"fmt"
"log"
@@ -77,7 +76,7 @@ const (
finalStepLocal = "source-park-local" // parkLocalArtifacts' step: the retirement id
finalStepRowDeleted = "row-deleted" // "pending" before the delete, "done" after it
finalStepConfigAck = "config-ack-source" // the config ack is copied and deleted
- finalStepParked = "source-parked" // source_parked_at is stamped
+ finalStepParked = "source-parked" // "pending" before the stamp's CAS, "done" after it
finalStepJournal = "source-park-journal" // where the journal is going, written just before it goes
)
@@ -132,10 +131,6 @@ func moveJournalPresent(dataDir, host, op string) bool {
return err == nil && fi.IsDir()
}
-func journalComponent(s string) bool {
- return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00")
-}
-
// participantRound is a post-flip participant's step for a record the
// region rule did not admit (movesPass). Only the source's finalization of a
// committed move is built; the rest is Plan 4's.
@@ -147,10 +142,14 @@ func (w *movesWorker) participantRound(ctx context.Context, s release.Store, rec
}
// sourceFinalArmed reports whether this box can run the finalization: every
-// hook it calls is wired. In production ParkLeases is not (ruling G17).
+// hook it calls is wired, the park engine's leases and Plan 5's view and
+// authorizer included. Nothing stands in for a missing owner in production
+// (fix round 1): the tests inject their own. In production none of the
+// three is wired yet (ruling G17).
func sourceFinalArmed(cfg moveCfg) bool {
return cfg.Box != "" && cfg.DataDir != "" && cfg.Guard.DataDir != "" && cfg.Local.DataDir != "" &&
- cfg.ParkLeases != nil && cfg.Versioned != nil && cfg.Route != nil && cfg.DeleteHeld != nil
+ cfg.ParkLeases != nil && cfg.ParkView != nil && cfg.ParkAuthorize != nil &&
+ cfg.Versioned != nil && cfg.Route != nil && cfg.DeleteHeld != nil
}
// sourceLocationParkScope is the source region's location park the commit
@@ -249,8 +248,8 @@ func (f *sourceFinal) run(ctx context.Context) error {
// is only the journal's own park is left, and the claim may already
// have moved on to the close.
if f.rec.Move.SourceParkedAt != nil {
- if !j.Done(finalStepParked) {
- if err := f.mark(finalStepParked, ""); err != nil {
+ if j.Steps[finalStepParked] != "done" {
+ if err := f.mark(finalStepParked, "done"); err != nil {
return err
}
}
@@ -298,11 +297,23 @@ func (f *sourceFinal) run(ctx context.Context) error {
return err
}
}
- if !j.Done(finalStepParked) {
+ if j.Steps[finalStepParked] != "done" {
+ // "pending" is the stamp step reached: from here the journal is
+ // the local sweep's to finish if the close overtakes this box
+ // (sweepFinalJournals).
+ if j.Steps[finalStepParked] != "pending" {
+ j.Mark(finalStepParked, "pending")
+ if err := j.Write(cfg.DataDir); err != nil {
+ return err
+ }
+ if err := f.fault("after-source-parked-pending"); err != nil {
+ return err
+ }
+ }
if err := f.stampParked(ctx); err != nil {
return err
}
- if err := f.mark(finalStepParked, ""); err != nil {
+ if err := f.mark(finalStepParked, "done"); err != nil {
return err
}
}
@@ -311,7 +322,7 @@ func (f *sourceFinal) run(ctx context.Context) error {
// sourceOf is the claim this finalization runs under: this move's landing,
// committed, this box its source with its evidence journalled, and the
-// record's incarnation. The stand-in authorizers use the same rule.
+// record's incarnation.
func (f *sourceFinal) sourceOf(obs hostclaim.Observed) error {
cfg := f.cfg()
if obs.Absent || obs.Claim == nil {
@@ -333,59 +344,12 @@ func (f *sourceFinal) sourceOf(obs hostclaim.Observed) error {
return nil
}
-// authorizeChore is park.Local.Authorize's stand-in for the source until
-// Plan 5: this move's own chore, for this box or this region, under sourceOf.
-func (f *sourceFinal) authorizeChore(_ context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
- if err := f.sourceOf(hostclaim.Observed{Host: f.rec.Host, Claim: c}); err != nil {
- return err
- }
- cfg := f.cfg()
- switch {
- case o.Op != f.g.ID:
- case o.Kind == "park-local" && o.Box == cfg.Box:
- return nil
- case o.Kind == "park-replica" && o.Region == cfg.Region:
- return nil
- }
- return fmt.Errorf("%w: not this source's chore", park.ErrUnsafe)
-}
-
-// authorizeObligation is park.Engine.Authorize's stand-in: the same rule
-// over an observation.
-func (f *sourceFinal) authorizeObligation(ctx context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
- if obs.Absent {
- return park.ErrUnsafe
- }
- return f.authorizeChore(ctx, obs.Claim, o)
-}
-
-// view is park.Engine.View's stand-in for the source's location park: the
-// target's region is authoritative after the commit, so this region's whole
-// location is retired and every version under it is selected (D6, H24: in
-// Phase 1 the scope is the root, instances/<host>, with every incarnation's
-// residue under it).
-func (f *sourceFinal) view(_ 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: f.cfg().Region, AuthoritativeRegion: f.rec.Move.To,
- GenerationOf: func(r park.Ref) (string, bool) { return landingGenerationOf(location, r.Key) }}, 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.
+// engine is the park engine over this box's own bucket, with the wired
+// leases, view and authorizer (sourceFinalArmed requires all three).
func (f *sourceFinal) engine(vs lease.VersionedStore) park.Engine {
cfg := f.cfg()
- view, auth := cfg.ParkView, cfg.ParkAuthorize
- if view == nil {
- view = f.view
- }
- if auth == nil {
- auth = f.authorizeObligation
- }
return park.Engine{Store: vs, Claims: f.s, Bucket: cfg.OwnBucket, Region: cfg.Region, Leases: cfg.ParkLeases,
- Now: f.now, NewID: obligationID, View: view, Authorize: auth, After: cfg.parkFault}
+ Now: f.now, NewID: obligationID, View: cfg.ParkView, Authorize: cfg.ParkAuthorize, After: cfg.parkFault}
}
// remote is step 1. done is true once the source region's location park has
@@ -483,7 +447,12 @@ func (f *sourceFinal) installation(ctx context.Context) (string, error) {
local.Box = cfg.Box
}
if local.Authorize == nil {
- local.Authorize = f.authorizeChore
+ // park.Local asks the same owner the engine does, over the claim
+ // it read under the guard.
+ authorize := cfg.ParkAuthorize
+ local.Authorize = func(ctx context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
+ return authorize(ctx, hostclaim.Observed{Host: host, Claim: c}, o)
+ }
}
if err := local.RetireInstallation(ctx, host, hostguard.Captured{Incarnation: captured.Incarnation, Stamp: stamp}, chore.ID, id); err != nil {
return "", fmt.Errorf("retire the source's installation: %w", err)
@@ -506,7 +475,8 @@ func (f *sourceFinal) installation(ctx context.Context) (string, error) {
// quarantine and litestream shadow, the files directory and <host>.env.
func sourceArtifactPaths(host string) []string {
db := host + ".db"
- return []string{db, db + ".gen", db + "-wal", db + "-shm", db + ".corrupt", ".activator-" + db + "-litestream", host, host + ".env"}
+ return []string{db, db + ".gen", db + "-wal", db + "-shm", db + ".corrupt", ".activator-" + db + "-litestream", host, host + ".env",
+ host + ".config.env", host + "-sidecar.env"}
}
// deleteRow is step 3's row and timers delete, under the host's hold lock
@@ -625,311 +595,130 @@ func (f *sourceFinal) stampParked(ctx context.Context) error {
// parkJournal renames the op's journal directory into the local retirement,
// after source_parked_at is stamped: until then it is what brings a replay
// back here. A crash before the rename leaves the journal, and the next pass
-// renames it; after it nothing of the move is left under .moves/.
+// (or the local sweep) renames it; after it nothing of the move is left
+// under .moves/.
func (f *sourceFinal) parkJournal() error {
- cfg := f.cfg()
- host, id := f.rec.Host, f.g.ID
- retirement := f.j.Steps[finalStepLocal]
+ return parkFinalJournal(f.cfg().DataDir, f.rec.Host, f.g.ID, f.j, f.mark)
+}
+
+// parkFinalJournal parks a finalized op's journal into the retirement its
+// own local park recorded, journalling the destination just before it goes.
+func parkFinalJournal(dataDir, host, op string, j *moves.Journal, mark func(step, value string) error) error {
+ retirement := j.Steps[finalStepLocal]
if retirement == "" {
- return fmt.Errorf("move %s: the journal records no local retirement for its own park", id)
+ return fmt.Errorf("move %s: the journal records no local retirement for its own park", op)
}
- plan, err := journalledLocalPlan(f.j, finalStepLocal)
+ plan, err := journalledLocalPlan(j, finalStepLocal)
if err != nil {
return err
}
- dst := filepath.Join(parkedArtifactsDir(cfg.DataDir, plan.Incarnation, host, retirement), ".moves", host, id)
- if f.j.Steps[finalStepJournal] != dst {
- if err := f.mark(finalStepJournal, dst); err != nil {
+ dst := filepath.Join(parkedArtifactsDir(dataDir, plan.Incarnation, host, retirement), ".moves", host, op)
+ if j.Steps[finalStepJournal] != dst {
+ if err := mark(finalStepJournal, dst); err != nil {
return err
}
}
- if err := parkOpJournal(cfg.DataDir, plan.Incarnation, host, id, retirement); err != nil {
+ if err := parkOpJournal(dataDir, plan.Incarnation, host, op, retirement); err != nil {
return fmt.Errorf("park the op journal: %w", err)
}
- log.Printf("moves: %s: the source's finalization of %s is complete", host, id)
+ log.Printf("moves: %s: the source's finalization of %s is complete", host, op)
return nil
}
-// ---- the moves-owned local rename (ruling H5(b)) ----
-
-// localArtifactPark is one moves-owned local retirement (ruling H5(b), the
-// pattern of Plan 2's extraction staging): artifacts the claim's inventory
-// does not list, so park.Local cannot retire them, renamed into the park
-// engine's local layout, .parked/<incarnation>/<host>/<retirement>/artifacts/,
-// keeping their paths. Plan 4's rollback reuses it.
-type localArtifactPark struct {
- DataDir, Incarnation, Host, Op string
- // Paths are the candidates, relative to DataDir. Absent ones are recorded
- // as absent and never touched.
- Paths []string
- // Step is the journal step that records the retirement id once every
- // rename is durable; Step+"-plan" records the plan before the first one.
- Step string
- NewID func() string
- // Fault is the test seam, after the plan and after each rename.
- Fault func(point string) error
-}
-
-// localParkPlan is what is journalled before the first rename: the
-// retirement id and every candidate's identity ("dev:ino"; "" when absent).
-type localParkPlan struct {
- Retirement string `json:"retirement"`
- Incarnation string `json:"incarnation"`
- Items []localParkItem `json:"items"`
-}
-
-type localParkItem struct {
- Path string `json:"path"`
- ID string `json:"id,omitempty"`
-}
-
-// localParkReceipt sits beside artifacts/ in the retirement, so a later
-// reader can find a moves-owned retirement by its op (D6, I57).
-type localParkReceipt struct {
- V int `json:"v"`
- Op string `json:"op"`
- Host string `json:"host"`
- Incarnation string `json:"incarnation"`
- Retirement string `json:"retirement"`
- Paths []string `json:"paths"`
-}
-
-const localParkReceiptName = "moves-retirement.json"
-
-// parkedEventDir and parkedArtifactsDir are the engine's local layout
-// (park.Local's parkLocal): one directory per retirement event.
-func parkedEventDir(dataDir, incarnation, host, retirement string) string {
- return filepath.Join(dataDir, ".parked", incarnation, host, retirement)
-}
-
-func parkedArtifactsDir(dataDir, incarnation, host, retirement string) string {
- return filepath.Join(parkedEventDir(dataDir, incarnation, host, retirement), "artifacts")
-}
-
-func journalledLocalPlan(j *moves.Journal, step string) (localParkPlan, error) {
- var plan localParkPlan
- raw, ok := j.Steps[step+"-plan"]
- if !ok {
- return plan, fmt.Errorf("the journal has no %s plan", step)
- }
- if err := json.Unmarshal([]byte(raw), &plan); err != nil {
- return plan, fmt.Errorf("the journal's %s plan: %w", step, err)
- }
- if !journalComponent(plan.Retirement) || !journalComponent(plan.Incarnation) {
- return plan, fmt.Errorf("the journal's %s plan names an unusable retirement", step)
- }
- return plan, nil
-}
-
-// artifactPathOK is a relative path under the data directory that is not
-// the guard's or the park's own.
-func artifactPathOK(p string) bool {
- if p == "" || filepath.IsAbs(p) || filepath.Clean(p) != p || strings.ContainsAny(p, "\\\x00") {
- return false
- }
- for _, c := range strings.Split(p, "/") {
- if !journalComponent(c) {
- return false
- }
- }
- first, _, _ := strings.Cut(p, "/")
- return first != ".parked" && first != "hostguard"
-}
-
-// parkLocalArtifacts runs one moves-owned local retirement and returns its
-// retirement id. In order: the plan (a fresh retirement id and each
-// candidate's identity) is journalled before anything moves; the retirement's
-// directories and its receipt are made; each recorded artifact is renamed to
-// the same path under artifacts/ and both parents synced; then Step records
-// the retirement id. A replay resumes from the journal:
-// - an artifact already at its destination with its identity is done;
-// - one still at its live path with its identity is renamed;
-// - one recorded absent, or whose live path now holds something with
-// another identity, is not this park's: it is left where it is, never
-// moved or removed;
-// - a destination that holds anything else is an error, and nothing more
-// is moved.
-func parkLocalArtifacts(ctx context.Context, j *moves.Journal, mark func(step, value string) error, p localArtifactPark) (string, error) {
- if r, ok := j.Steps[p.Step]; ok {
- return r, nil
- }
- if p.DataDir == "" || !journalComponent(p.Incarnation) || !journalComponent(p.Host) || p.Op == "" || p.Step == "" || p.NewID == nil {
- return "", errors.New("a local park names its data directory, incarnation, host, op and step")
- }
- for _, path := range p.Paths {
- if !artifactPathOK(path) {
- return "", fmt.Errorf("a local park cannot retire %q", path)
- }
- }
- fault := func(point string) error {
- if p.Fault != nil {
- return p.Fault(point)
- }
- return nil
- }
- plan, err := journalledLocalPlan(j, p.Step)
+// sweepFinalJournals is the local sweep of this box's op journals (fix round
+// 1, ruling 1), independent of any active move: a crash after the stamp and
+// before the journal's park, then the close, leaves a journal no record
+// routes back here (movesParticipantRecord answers nothing for a record with
+// no move). Each journal under <DataDir>/.moves/<host>/<op>/ that has reached
+// the stamp step is finished here:
+// - the stamp journalled "done": the journal is parked;
+// - "pending", and the listed record still carries this op: with
+// source_parked_at set it is journalled done and parked; without, it is
+// left to the finalization's own pass;
+// - "pending", and the listed record no longer carries this op (closed, or
+// another move since): the stamp landed, because the only transition
+// that clears a committed move is MoveClose, which refuses unless
+// source_parked_at is set (the aborts end at cutting/requested); so the
+// stamp is journalled done and the journal parked;
+// - a host the listing does not name (its record deleted, or unreadable)
+// is left as it is.
+//
+// A journal that has not reached the stamp step (no source-parked step: a
+// landing's, a cut's, a finalization still under way) is never touched, and
+// nothing is ever deleted: a journal is only renamed into its retirement.
+func (w *movesWorker) sweepFinalJournals(recs []provision.Record) {
+ dd := w.cfg.DataDir
+ if dd == "" {
+ return
+ }
+ hosts, err := os.ReadDir(filepath.Join(dd, ".moves"))
if err != nil {
- if _, planned := j.Steps[p.Step+"-plan"]; planned {
- return "", err
- }
- if plan, err = p.plan(); err != nil {
- return "", err
- }
- body, err := json.Marshal(plan)
- if err != nil {
- return "", err
- }
- if err := mark(p.Step+"-plan", string(body)); err != nil {
- return "", err
+ if !errors.Is(err, os.ErrNotExist) {
+ w.notice("#final-sweep", "moves: the op journal sweep cannot list %s: %v", filepath.Join(dd, ".moves"), err)
}
+ return
}
- if plan.Incarnation != p.Incarnation || len(plan.Items) != len(p.Paths) {
- return "", fmt.Errorf("the journal's %s plan is not this park's", p.Step)
+ byHost := map[string][]provision.Record{}
+ for _, rec := range recs {
+ byHost[rec.Host] = append(byHost[rec.Host], rec)
}
- for i, it := range plan.Items {
- if it.Path != p.Paths[i] {
- return "", fmt.Errorf("the journal's %s plan is not this park's", p.Step)
- }
- }
- dest := parkedArtifactsDir(p.DataDir, p.Incarnation, p.Host, plan.Retirement)
- if err := mkdirSynced(p.DataDir, dest); err != nil {
- return "", err
- }
- receipt, err := json.Marshal(localParkReceipt{V: 1, Op: p.Op, Host: p.Host, Incarnation: p.Incarnation, Retirement: plan.Retirement, Paths: p.Paths})
- if err != nil {
- return "", err
- }
- event := parkedEventDir(p.DataDir, p.Incarnation, p.Host, plan.Retirement)
- if err := writeFileAtomic(filepath.Join(event, localParkReceiptName), receipt, 0o640); err != nil {
- return "", err
- }
- for _, it := range plan.Items {
- if err := ctx.Err(); err != nil {
- return "", err
- }
- src, dst := filepath.Join(p.DataDir, it.Path), filepath.Join(dest, it.Path)
- if it.ID == "" {
- if _, err := os.Lstat(src); err == nil {
- log.Printf("moves: %s: %s appeared after the park recorded it absent; it is not this park's and is left where it is", p.Host, it.Path)
- }
- continue
- }
- if done, err := sameIdentity(dst, it.ID); err != nil {
- return "", err
- } else if done {
+ for _, hd := range hosts {
+ host := hd.Name()
+ if !hd.IsDir() || !journalComponent(host) {
continue
}
- if _, err := os.Lstat(dst); err == nil {
- return "", fmt.Errorf("%s holds what this park did not put there", dst)
- } else if !errors.Is(err, os.ErrNotExist) {
- return "", err
- }
- ours, err := sameIdentity(src, it.ID)
+ ops, err := os.ReadDir(filepath.Join(dd, ".moves", host))
if err != nil {
- return "", err
- }
- if !ours {
- log.Printf("moves: %s: %s is no longer the artifact the park recorded; what is there is left where it is", p.Host, it.Path)
+ w.notice(host+"#final-sweep", "moves: %s: the op journal sweep cannot list its journals: %v", host, err)
continue
}
- if err := mkdirSynced(p.DataDir, filepath.Dir(dst)); err != nil {
- return "", err
- }
- if err := os.Rename(src, dst); err != nil {
- return "", fmt.Errorf("park %s: %w", it.Path, err)
- }
- if err := syncPath(filepath.Dir(src)); err != nil {
- return "", err
- }
- if err := syncPath(filepath.Dir(dst)); err != nil {
- return "", err
- }
- if err := fault("after-rename:" + it.Path); err != nil {
- return "", err
+ for _, od := range ops {
+ op := od.Name()
+ if !od.IsDir() || !journalComponent(op) {
+ continue
+ }
+ if err := w.sweepFinalJournal(dd, host, op, byHost[host]); err != nil {
+ w.logTrouble(host, fmt.Errorf("the op journal sweep, %s: %w", op, err))
+ }
}
}
- if err := syncPath(dest); err != nil {
- return "", err
- }
- if err := mark(p.Step, plan.Retirement); err != nil {
- return "", err
- }
- log.Printf("moves: %s: the local artifacts of %s are parked in %s", p.Host, p.Op, event)
- return plan.Retirement, nil
}
-// plan records a fresh retirement id and each candidate's identity. The
-// retirement's directory must not exist yet: the id is this park's alone.
-func (p localArtifactPark) plan() (localParkPlan, error) {
- plan := localParkPlan{Retirement: p.NewID(), Incarnation: p.Incarnation}
- if !journalComponent(plan.Retirement) {
- return plan, errors.New("no retirement id")
+// sweepFinalJournal is one journal of sweepFinalJournals.
+func (w *movesWorker) sweepFinalJournal(dataDir, host, op string, recs []provision.Record) error {
+ j, err := moves.ReadJournal(dataDir, host, op)
+ if err != nil {
+ return err
}
- if _, err := os.Lstat(parkedEventDir(p.DataDir, p.Incarnation, p.Host, plan.Retirement)); err == nil {
- return plan, fmt.Errorf("retirement %s already exists", plan.Retirement)
- } else if !errors.Is(err, os.ErrNotExist) {
- return plan, err
+ stamp, reached := j.Steps[finalStepParked]
+ if !reached {
+ return nil
}
- for _, path := range p.Paths {
- id, err := pathIdentity(filepath.Join(p.DataDir, path))
- if errors.Is(err, os.ErrNotExist) {
- id, err = "", nil
+ if stamp != "done" {
+ if len(recs) == 0 {
+ w.notice(host+"#final-sweep-"+op, "moves: %s: the journal of %s reached its stamp, and no listed record names the host; it is left", host, op)
+ return nil
}
- if err != nil {
- return plan, err
+ for _, rec := range recs {
+ if m := rec.Move; m != nil && m.ID == op && m.SourceParkedAt == nil {
+ return nil // the finalization's own pass stamps it
+ }
}
- plan.Items = append(plan.Items, localParkItem{Path: path, ID: id})
}
- return plan, nil
-}
-
-// mkdirSynced makes dir (under root) and syncs the parent of every directory
-// it made, so a rename into it survives a crash.
-func mkdirSynced(root, dir string) error {
- var made []string
- for d := dir; d != root && d != filepath.Dir(d); d = filepath.Dir(d) {
- if _, err := os.Lstat(d); err == nil {
- break
- } else if !errors.Is(err, os.ErrNotExist) {
+ mark := func(step, value string) error {
+ j.Mark(step, value)
+ if err := j.Write(dataDir); err != nil {
return err
}
- made = append(made, d)
- }
- if err := os.MkdirAll(dir, 0o770); err != nil {
- return err
- }
- for _, d := range made {
- if err := syncPath(filepath.Dir(d)); err != nil {
- return err
+ if fn := w.cfg.finalFault; fn != nil {
+ return fn("after-" + step)
}
- }
- return nil
-}
-
-// parkOpJournal renames the op's journal directory into the retirement's
-// artifacts, at .moves/<host>/<op>. The directory is the op's own (its name
-// is the op id). Absent is done; a destination already there is refused.
-func parkOpJournal(dataDir, incarnation, host, op, retirement string) error {
- src := moves.JournalDir(dataDir, host, op)
- dst := filepath.Join(parkedArtifactsDir(dataDir, incarnation, host, retirement), ".moves", host, op)
- if _, err := os.Lstat(src); errors.Is(err, os.ErrNotExist) {
return nil
- } else if err != nil {
- return err
}
- if _, err := os.Lstat(dst); err == nil {
- return fmt.Errorf("%s already holds a journal", dst)
- } else if !errors.Is(err, os.ErrNotExist) {
- return err
- }
- if err := mkdirSynced(dataDir, filepath.Dir(dst)); err != nil {
- return err
- }
- if err := os.Rename(src, dst); err != nil {
- return err
- }
- if err := syncPath(filepath.Dir(src)); err != nil {
- return err
+ if stamp != "done" {
+ if err := mark(finalStepParked, "done"); err != nil {
+ return err
+ }
}
- return syncPath(filepath.Dir(dst))
+ return parkFinalJournal(dataDir, host, op, j, mark)
}
diff --git a/internal/host/movesfinal_test.go b/internal/host/movesfinal_test.go
index 168ffc9a2..236dc2510 100644
--- a/internal/host/movesfinal_test.go
+++ b/internal/host/movesfinal_test.go
@@ -69,7 +69,8 @@ func newFinalHarness(t *testing.T) *finalHarness {
}
}
dd := h.cfg.DataDir
- for name, body := range map[string]string{p1Host + ".db.gen": "cur", p1Host + ".db.corrupt": "quarantined"} {
+ for name, body := range map[string]string{p1Host + ".db.gen": "cur", p1Host + ".db.corrupt": "quarantined",
+ p1Host + ".config.env": "SECRET=1\n", p1Host + "-sidecar.env": "SIDE=1\n"} {
if err := os.WriteFile(filepath.Join(dd, name), []byte(body), 0o600); err != nil {
t.Fatal(err)
}
@@ -92,6 +93,7 @@ func newFinalHarness(t *testing.T) *finalHarness {
h.cfg.Guard = hostguard.Guard{DataDir: dd, LockDir: t.TempDir()}
h.cfg.Local = park.Local{DataDir: dd, Box: "ie-box"}
h.cfg.ParkLeases = h.leases
+ h.cfg.ParkView, h.cfg.ParkAuthorize = sourceTestView, sourceTestAuthorize
h.cfg.parkFault = h.lcfg.parkFault
h.cfg.finalFault = func(point string) error {
h.finalHits = append(h.finalHits, point)
@@ -189,7 +191,7 @@ func (h *finalHarness) sourceJournal(t *testing.T) *moves.Journal {
func (h *finalHarness) live(t *testing.T) []string {
t.Helper()
var out []string
- for _, p := range append(sourceArtifactPaths(p1Host), filepath.Join(".moves", p1Host, p1ID)) {
+ for _, p := range append(testSourcePaths(), filepath.Join(".moves", p1Host, p1ID)) {
if _, err := os.Lstat(filepath.Join(h.cfg.DataDir, p)); err == nil {
out = append(out, p)
} else if !errors.Is(err, os.ErrNotExist) {
@@ -219,7 +221,7 @@ func (h *finalHarness) parked(t *testing.T) []string {
// untouched asserts that nothing of the source's was finalized.
func (h *finalHarness) untouched(t *testing.T, why string) {
t.Helper()
- if got, want := h.live(t), []string{p1Host + ".db", p1Host + ".db.gen", p1Host + ".db.corrupt", p1Host, p1Host + ".env", filepath.Join(".moves", p1Host, p1ID)}; strings.Join(got, " ") != strings.Join(want, " ") {
+ if got, want := h.live(t), []string{p1Host + ".db", p1Host + ".db.gen", p1Host + ".db.corrupt", p1Host, p1Host + ".env", p1Host + ".config.env", p1Host + "-sidecar.env", filepath.Join(".moves", p1Host, p1ID)}; strings.Join(got, " ") != strings.Join(want, " ") {
t.Fatalf("%s: live artifacts %v, want %v", why, got, want)
}
if got := h.parked(t); len(got) != 0 {
@@ -272,7 +274,8 @@ func (h *finalHarness) finalized(t *testing.T) string {
rest = append(rest, tail)
}
jpath := "artifacts/.moves/" + p1Host + "/" + p1ID + "/journal.json"
- want := []string{"artifacts/" + p1Host + "/upload.txt", "artifacts/" + p1Host + ".db", "artifacts/" + p1Host + ".db.corrupt", "artifacts/" + p1Host + ".db.gen", "artifacts/" + p1Host + ".env", localParkReceiptName}
+ want := []string{"artifacts/" + p1Host + "/upload.txt", "artifacts/" + p1Host + ".db", "artifacts/" + p1Host + ".db.corrupt", "artifacts/" + p1Host + ".db.gen", "artifacts/" + p1Host + ".env",
+ "artifacts/" + p1Host + ".config.env", "artifacts/" + p1Host + "-sidecar.env", localParkReceiptName}
var got []string
sawJournal := false
for _, r := range rest {
@@ -382,7 +385,7 @@ func TestSourceParkWaitsForCommit(t *testing.T) {
t.Fatal(err)
}
h.sourcePass(t)
- if got := h.live(t); len(got) != 6 || len(h.parked(t)) != 0 || h.read(t, p1Host).Move.SourceParkedAt != nil {
+ if got := h.live(t); len(got) != 8 || len(h.parked(t)) != 0 || h.read(t, p1Host).Move.SourceParkedAt != nil {
t.Fatalf("finalized without the row: live %v parked %v", got, h.parked(t))
}
})
@@ -447,7 +450,7 @@ func TestSourceParkWaitsForTheRemoteHalf(t *testing.T) {
stuck := true
h.vs.skipDelete = func(key string) bool { return stuck && strings.HasPrefix(key, cutRoot+"/") }
h.sourcePass(t)
- if got := h.live(t); len(got) != 6 {
+ if got := h.live(t); len(got) != 8 {
t.Fatalf("the local half ran before the remote: live %v", got)
}
if j := h.sourceJournal(t); j.Done(finalStepRemote) || j.Done(finalStepLocal+"-plan") {
@@ -513,10 +516,12 @@ func TestSourceParkReplayIsIdempotent(t *testing.T) {
t.Fatalf("deleted %d times", h.deletes)
}
want := []string{"after-" + finalStepRemote, "after-" + finalStepInstallation, "after-" + finalStepLocal + "-plan",
- "after-rename:" + p1Host + ".db", "after-rename:" + p1Host + ".db.gen", "after-rename:" + p1Host + ".db.corrupt",
- "after-rename:" + p1Host, "after-rename:" + p1Host + ".env", "after-" + finalStepLocal,
+ "after-" + finalStepLocal,
"after-row-pending", "after-row-delete", "after-" + finalStepRowDeleted, "after-config-ack-copy", "after-" + finalStepConfigAck,
- "after-source-parked-record", "after-" + finalStepParked, "after-" + finalStepJournal}
+ "after-source-parked-pending", "after-source-parked-record", "after-" + finalStepParked, "after-" + finalStepJournal}
+ for _, p := range []string{p1Host + ".db", p1Host + ".db.gen", p1Host + ".db.corrupt", p1Host, p1Host + ".env", p1Host + ".config.env", p1Host + "-sidecar.env"} {
+ want = append(want, "after-rename-unsynced:"+p, "after-rename:"+p)
+ }
for _, p := range want {
if !h.crashAll[p] {
t.Errorf("never crashed at %s (crashed at %v)", p, h.crashAll)
@@ -745,12 +750,20 @@ func TestMovesParticipantRecord(t *testing.T) {
func TestSourceParkWaitsUnarmedInProduction(t *testing.T) {
h := newFinalHarness(t)
h.commit(t)
- h.cfg.ParkLeases = nil
- if sourceFinalArmed(h.cfg) {
- t.Fatal("armed without park leases")
+ for name, unarm := range map[string]func(*moveCfg){
+ "no park leases": func(c *moveCfg) { c.ParkLeases = nil },
+ "no park view": func(c *moveCfg) { c.ParkView = nil },
+ "no park authorizer": func(c *moveCfg) { c.ParkAuthorize = nil },
+ } {
+ saved := h.cfg
+ unarm(&h.cfg)
+ if sourceFinalArmed(h.cfg) {
+ t.Fatalf("%s: armed", name)
+ }
+ h.sourcePass(t)
+ h.untouched(t, name)
+ h.cfg = saved
}
- h.sourcePass(t)
- h.untouched(t, "unarmed")
var prod moveCfg
reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
if err != nil {
@@ -763,7 +776,7 @@ func TestSourceParkWaitsUnarmedInProduction(t *testing.T) {
t.Fatalf("production wiring: %+v", prod)
}
prod.Route = func(string) (*registry.Route, error) { return nil, nil }
- if prod.ParkLeases != nil || sourceFinalArmed(prod) {
+ if prod.ParkLeases != nil || prod.ParkView != nil || prod.ParkAuthorize != nil || sourceFinalArmed(prod) {
t.Fatal("production armed the finalization")
}
}
@@ -826,3 +839,160 @@ func TestParkLocalArtifactsRefusals(t *testing.T) {
t.Fatal("a plan for other paths was accepted")
}
}
+
+// testSourcePaths is what the tests expect the source to keep at its flat
+// live paths and park: the database with its marker, journal, quarantine
+// and litestream shadow, the files directory, <host>.env, and the two
+// rendered env files (fix round 1, ruling 3).
+func testSourcePaths() []string {
+ db := p1Host + ".db"
+ return []string{db, db + ".gen", db + "-wal", db + "-shm", db + ".corrupt", ".activator-" + db + "-litestream", p1Host, p1Host + ".env",
+ p1Host + ".config.env", p1Host + "-sidecar.env"}
+}
+
+// sourceTestView and sourceTestAuthorize are the park view and authorizer
+// the tests inject for the source's location park, in place of Plan 5's
+// owners (fix round 1, ruling 4: production never substitutes stand-ins).
+// The target's region is authoritative after the commit, so every version
+// under the location is selected; a chore is authorized only when it is
+// this move's, for ie-box or ie, under the move's committed landing with
+// ie-box its source.
+func sourceTestView(_ 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: "ie", AuthoritativeRegion: "uk",
+ GenerationOf: func(r park.Ref) (string, bool) { return landingGenerationOf(location, r.Key) }}, nil
+}
+
+func sourceTestAuthorize(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
+ c := obs.Claim
+ if obs.Absent || c == nil || c.Op.ID != p1ID || c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted ||
+ len(c.Op.Participants) != 2 || c.Op.Participants[0] != "ie-box" || o.Op != p1ID {
+ return park.ErrUnsafe
+ }
+ if (o.Kind == "park-local" && o.Box == "ie-box") || (o.Kind == "park-replica" && o.Region == "ie") {
+ return nil
+ }
+ return park.ErrUnsafe
+}
+
+// Fix round 1, ruling 1: a crash after source_parked_at is stamped and
+// before the journal is parked, then the close (which clears the move, so
+// no record routes here any more), then a fresh source pass: the local
+// sweep parks the journal, because the close proves the stamp landed.
+func TestSourceParkJournalSurvivesTheClose(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ h.finalAt = "after-source-parked-record"
+ h.crashSource(t)
+ if j := h.sourceJournal(t); j.Done(finalStepParked) && j.Steps[finalStepParked] == "done" {
+ t.Fatalf("fixture: the stamp is journalled done: %v", j.Steps)
+ }
+ ctx := context.Background()
+ g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
+ for _, mut := range []func(*provision.Record) error{provision.MoveOpened(g, h.clock.Now()), provision.MoveClose(g, h.clock.Now())} {
+ if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, mut); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if rec := h.read(t, p1Host); rec.Move != nil || rec.LastMove == nil || rec.LastMove.ID != p1ID {
+ t.Fatalf("fixture: not closed: %+v", rec)
+ }
+ if movesParticipantRecord(h.cfg, h.read(t, p1Host)) != roleNone {
+ t.Fatal("fixture: a closed record still routes here")
+ }
+ h.freshWorker()
+ h.sourcePass(t)
+ if got := h.live(t); len(got) != 0 {
+ t.Fatalf("still live after the close: %v", got)
+ }
+ j := h.parkedJournal(t)
+ if j.Steps[finalStepParked] != "done" {
+ t.Fatalf("the parked journal's stamp: %v", j.Steps)
+ }
+}
+
+// The sweep never touches a journal that has not reached the stamp, nor one
+// whose stamp is pending while the record still carries the op unstamped: a
+// journal of another op (a landing's, say) under a record with no move, and
+// this op's journal stopped right after it journalled the stamp pending,
+// before the stamp's CAS.
+func TestSourceParkSweepLeavesUnstampedJournals(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ // Another op's journal that would park cleanly if the sweep took it: a
+ // complete local plan and retirement, everything but the stamp step.
+ plan, err := json.Marshal(localParkPlan{Retirement: "r0", Incarnation: h.inc})
+ if err != nil {
+ t.Fatal(err)
+ }
+ otherSteps := map[string]string{finalStepLocal + "-plan": string(plan), finalStepLocal: "r0", finalStepRowDeleted: "done"}
+ other := &moves.Journal{Host: p1Host, Op: p1ID2, Steps: otherSteps}
+ if err := other.Write(h.cfg.DataDir); err != nil {
+ t.Fatal(err)
+ }
+ h.finalAt = "after-source-parked-pending"
+ h.crashSource(t)
+ if j := h.sourceJournal(t); j.Steps[finalStepParked] != "pending" || h.read(t, p1Host).Move.SourceParkedAt != nil {
+ t.Fatalf("fixture: %v", j.Steps)
+ }
+ h.cfg.ParkLeases = nil // unarmed: only the sweep runs
+ h.freshWorker()
+ h.sourcePass(t)
+ for _, op := range []string{p1ID, p1ID2} {
+ if !moveJournalPresent(h.cfg.DataDir, p1Host, op) {
+ t.Fatalf("the sweep took %s's journal before its stamp", op)
+ }
+ }
+ if j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID2); err != nil || len(j.Steps) != len(otherSteps) {
+ t.Fatalf("the sweep wrote into another op's journal: %v %v", j.Steps, err)
+ }
+}
+
+// Fix round 1, ruling 2: a replay that finds an artifact already at its
+// destination (a stop after the rename, before either sync) syncs the
+// source's parent and every destination parent before it goes on; and the
+// rename offers an interruption point between the rename and the syncs.
+func TestParkLocalArtifactsSyncsOnReplay(t *testing.T) {
+ dd := t.TempDir()
+ if err := os.MkdirAll(filepath.Join(dd, "d", "e"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(dd, "d", "e", "a"), []byte("a"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ j := &moves.Journal{Host: p1Host, Op: p1ID, Steps: map[string]string{}}
+ mark := func(step, value string) error { j.Mark(step, value); return nil }
+ var points []string
+ p := localArtifactPark{DataDir: dd, Incarnation: "inc", Host: p1Host, Op: p1ID, Step: "s", Paths: []string{"d/e/a"}, NewID: func() string { return "r1" },
+ Fault: func(point string) error {
+ points = append(points, point)
+ if point == "after-rename-unsynced:d/e/a" {
+ return errCrash
+ }
+ return nil
+ }}
+ if _, err := parkLocalArtifacts(context.Background(), j, mark, p); !errors.Is(err, errCrash) {
+ t.Fatalf("no interruption between the rename and its syncs: %v (points %v)", err, points)
+ }
+ var synced []string
+ saved := retireSync
+ retireSync = func(path string) error { synced = append(synced, path); return saved(path) }
+ t.Cleanup(func() { retireSync = saved })
+ p.Fault = nil
+ if r, err := parkLocalArtifacts(context.Background(), j, mark, p); err != nil || r != "r1" {
+ t.Fatalf("replay: %q %v", r, err)
+ }
+ art := parkedArtifactsDir(dd, "inc", p1Host, "r1")
+ for _, want := range []string{filepath.Join(dd, "d", "e"), filepath.Join(art, "d", "e"), filepath.Join(art, "d"), art, filepath.Dir(art), filepath.Join(dd, ".parked")} {
+ found := false
+ for _, s := range synced {
+ found = found || s == want
+ }
+ if !found {
+ t.Errorf("the replay did not sync %s (synced %v)", want, synced)
+ }
+ }
+}
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index dc823a37f..12f4ba48b 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -704,6 +704,11 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {
w.evictArchives(nil) // no move into this box: no verdict to keep
w.retireHeldRefusals(nil)
}
+ if cfg.Enabled {
+ // The source's op journals that reached the stamp, whoever's record
+ // no longer routes here (movesfinal.go, fix round 1).
+ w.sweepFinalJournals(recs)
+ }
w.liftFences(recs)
return nil
}
diff --git a/internal/host/movesretire.go b/internal/host/movesretire.go
new file mode 100644
index 000000000..9dd91bf26
--- /dev/null
+++ b/internal/host/movesretire.go
@@ -0,0 +1,337 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "os"
+ "path/filepath"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/moves"
+)
+
+// The moves-owned local retirement (ruling H5(b), the pattern of Plan 2's
+// extraction staging): artifacts the claim's inventory does not list, so
+// park.Local cannot retire them, renamed into the park engine's local
+// layout under a journalled plan, and an op's own journal directory parked
+// after them. The source's finalization (movesfinal.go) uses it; Plan 4's
+// rollback reuses it.
+
+func journalComponent(s string) bool {
+ return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00")
+}
+
+// localArtifactPark is one moves-owned local retirement (ruling H5(b), the
+// pattern of Plan 2's extraction staging): artifacts the claim's inventory
+// does not list, so park.Local cannot retire them, renamed into the park
+// engine's local layout, .parked/<incarnation>/<host>/<retirement>/artifacts/,
+// keeping their paths. Plan 4's rollback reuses it.
+type localArtifactPark struct {
+ DataDir, Incarnation, Host, Op string
+ // Paths are the candidates, relative to DataDir. Absent ones are recorded
+ // as absent and never touched.
+ Paths []string
+ // Step is the journal step that records the retirement id once every
+ // rename is durable; Step+"-plan" records the plan before the first one.
+ Step string
+ NewID func() string
+ // Fault is the test seam, after the plan and after each rename.
+ Fault func(point string) error
+}
+
+// localParkPlan is what is journalled before the first rename: the
+// retirement id and every candidate's identity ("dev:ino"; "" when absent).
+type localParkPlan struct {
+ Retirement string `json:"retirement"`
+ Incarnation string `json:"incarnation"`
+ Items []localParkItem `json:"items"`
+}
+
+type localParkItem struct {
+ Path string `json:"path"`
+ ID string `json:"id,omitempty"`
+}
+
+// localParkReceipt sits beside artifacts/ in the retirement, so a later
+// reader can find a moves-owned retirement by its op (D6, I57).
+type localParkReceipt struct {
+ V int `json:"v"`
+ Op string `json:"op"`
+ Host string `json:"host"`
+ Incarnation string `json:"incarnation"`
+ Retirement string `json:"retirement"`
+ Paths []string `json:"paths"`
+}
+
+const localParkReceiptName = "moves-retirement.json"
+
+// retireSync is the fsync every moves-owned retirement makes durable with
+// (syncPath); a variable so a test can see which directories were synced.
+var retireSync = syncPath
+
+// parkedEventDir and parkedArtifactsDir are the engine's local layout
+// (park.Local's parkLocal): one directory per retirement event.
+func parkedEventDir(dataDir, incarnation, host, retirement string) string {
+ return filepath.Join(dataDir, ".parked", incarnation, host, retirement)
+}
+
+func parkedArtifactsDir(dataDir, incarnation, host, retirement string) string {
+ return filepath.Join(parkedEventDir(dataDir, incarnation, host, retirement), "artifacts")
+}
+
+func journalledLocalPlan(j *moves.Journal, step string) (localParkPlan, error) {
+ var plan localParkPlan
+ raw, ok := j.Steps[step+"-plan"]
+ if !ok {
+ return plan, fmt.Errorf("the journal has no %s plan", step)
+ }
+ if err := json.Unmarshal([]byte(raw), &plan); err != nil {
+ return plan, fmt.Errorf("the journal's %s plan: %w", step, err)
+ }
+ if !journalComponent(plan.Retirement) || !journalComponent(plan.Incarnation) {
+ return plan, fmt.Errorf("the journal's %s plan names an unusable retirement", step)
+ }
+ return plan, nil
+}
+
+// artifactPathOK is a relative path under the data directory that is not
+// the guard's or the park's own.
+func artifactPathOK(p string) bool {
+ if p == "" || filepath.IsAbs(p) || filepath.Clean(p) != p || strings.ContainsAny(p, "\\\x00") {
+ return false
+ }
+ for _, c := range strings.Split(p, "/") {
+ if !journalComponent(c) {
+ return false
+ }
+ }
+ first, _, _ := strings.Cut(p, "/")
+ return first != ".parked" && first != "hostguard"
+}
+
+// parkLocalArtifacts runs one moves-owned local retirement and returns its
+// retirement id. In order: the plan (a fresh retirement id and each
+// candidate's identity) is journalled before anything moves; the retirement's
+// directories and its receipt are made; each recorded artifact is renamed to
+// the same path under artifacts/ and its source's parent and every
+// destination directory up to the data directory synced; then Step records
+// the retirement id. A replay resumes from the journal:
+// - an artifact already at its destination with its identity is synced as
+// a fresh rename is (a stop between the rename and its syncs leaves the
+// rename undurable, fix round 1), and is then done;
+// - one still at its live path with its identity is renamed;
+// - one recorded absent, or whose live path now holds something with
+// another identity, is not this park's: it is left where it is, never
+// moved or removed;
+// - a destination that holds anything else is an error, and nothing more
+// is moved.
+func parkLocalArtifacts(ctx context.Context, j *moves.Journal, mark func(step, value string) error, p localArtifactPark) (string, error) {
+ if r, ok := j.Steps[p.Step]; ok {
+ return r, nil
+ }
+ if p.DataDir == "" || !journalComponent(p.Incarnation) || !journalComponent(p.Host) || p.Op == "" || p.Step == "" || p.NewID == nil {
+ return "", errors.New("a local park names its data directory, incarnation, host, op and step")
+ }
+ for _, path := range p.Paths {
+ if !artifactPathOK(path) {
+ return "", fmt.Errorf("a local park cannot retire %q", path)
+ }
+ }
+ fault := func(point string) error {
+ if p.Fault != nil {
+ return p.Fault(point)
+ }
+ return nil
+ }
+ plan, err := journalledLocalPlan(j, p.Step)
+ if err != nil {
+ if _, planned := j.Steps[p.Step+"-plan"]; planned {
+ return "", err
+ }
+ if plan, err = p.plan(); err != nil {
+ return "", err
+ }
+ body, err := json.Marshal(plan)
+ if err != nil {
+ return "", err
+ }
+ if err := mark(p.Step+"-plan", string(body)); err != nil {
+ return "", err
+ }
+ }
+ if plan.Incarnation != p.Incarnation || len(plan.Items) != len(p.Paths) {
+ return "", fmt.Errorf("the journal's %s plan is not this park's", p.Step)
+ }
+ for i, it := range plan.Items {
+ if it.Path != p.Paths[i] {
+ return "", fmt.Errorf("the journal's %s plan is not this park's", p.Step)
+ }
+ }
+ dest := parkedArtifactsDir(p.DataDir, p.Incarnation, p.Host, plan.Retirement)
+ if err := mkdirSynced(p.DataDir, dest); err != nil {
+ return "", err
+ }
+ receipt, err := json.Marshal(localParkReceipt{V: 1, Op: p.Op, Host: p.Host, Incarnation: p.Incarnation, Retirement: plan.Retirement, Paths: p.Paths})
+ if err != nil {
+ return "", err
+ }
+ event := parkedEventDir(p.DataDir, p.Incarnation, p.Host, plan.Retirement)
+ if err := writeFileAtomic(filepath.Join(event, localParkReceiptName), receipt, 0o640); err != nil {
+ return "", err
+ }
+ for _, it := range plan.Items {
+ if err := ctx.Err(); err != nil {
+ return "", err
+ }
+ src, dst := filepath.Join(p.DataDir, it.Path), filepath.Join(dest, it.Path)
+ if it.ID == "" {
+ if _, err := os.Lstat(src); err == nil {
+ log.Printf("moves: %s: %s appeared after the park recorded it absent; it is not this park's and is left where it is", p.Host, it.Path)
+ }
+ continue
+ }
+ if done, err := sameIdentity(dst, it.ID); err != nil {
+ return "", err
+ } else if done {
+ if err := syncRename(p.DataDir, src, dst); err != nil {
+ return "", err
+ }
+ if err := fault("after-rename:" + it.Path); err != nil {
+ return "", err
+ }
+ continue
+ }
+ if _, err := os.Lstat(dst); err == nil {
+ return "", fmt.Errorf("%s holds what this park did not put there", dst)
+ } else if !errors.Is(err, os.ErrNotExist) {
+ return "", err
+ }
+ ours, err := sameIdentity(src, it.ID)
+ if err != nil {
+ return "", err
+ }
+ if !ours {
+ log.Printf("moves: %s: %s is no longer the artifact the park recorded; what is there is left where it is", p.Host, it.Path)
+ continue
+ }
+ if err := mkdirSynced(p.DataDir, filepath.Dir(dst)); err != nil {
+ return "", err
+ }
+ if err := os.Rename(src, dst); err != nil {
+ return "", fmt.Errorf("park %s: %w", it.Path, err)
+ }
+ if err := fault("after-rename-unsynced:" + it.Path); err != nil {
+ return "", err
+ }
+ if err := syncRename(p.DataDir, src, dst); err != nil {
+ return "", err
+ }
+ if err := fault("after-rename:" + it.Path); err != nil {
+ return "", err
+ }
+ }
+ if err := retireSync(dest); err != nil {
+ return "", err
+ }
+ if err := mark(p.Step, plan.Retirement); err != nil {
+ return "", err
+ }
+ log.Printf("moves: %s: the local artifacts of %s are parked in %s", p.Host, p.Op, event)
+ return plan.Retirement, nil
+}
+
+// plan records a fresh retirement id and each candidate's identity. The
+// retirement's directory must not exist yet: the id is this park's alone.
+func (p localArtifactPark) plan() (localParkPlan, error) {
+ plan := localParkPlan{Retirement: p.NewID(), Incarnation: p.Incarnation}
+ if !journalComponent(plan.Retirement) {
+ return plan, errors.New("no retirement id")
+ }
+ if _, err := os.Lstat(parkedEventDir(p.DataDir, p.Incarnation, p.Host, plan.Retirement)); err == nil {
+ return plan, fmt.Errorf("retirement %s already exists", plan.Retirement)
+ } else if !errors.Is(err, os.ErrNotExist) {
+ return plan, err
+ }
+ for _, path := range p.Paths {
+ id, err := pathIdentity(filepath.Join(p.DataDir, path))
+ if errors.Is(err, os.ErrNotExist) {
+ id, err = "", nil
+ }
+ if err != nil {
+ return plan, err
+ }
+ plan.Items = append(plan.Items, localParkItem{Path: path, ID: id})
+ }
+ return plan, nil
+}
+
+// mkdirSynced makes dir (under root) and syncs it and every directory above
+// it up to root. Every call syncs the whole chain, not only the directories
+// it made: a replay after a stop between MkdirAll and the syncs finds the
+// directories there and must still make their entries durable.
+func mkdirSynced(root, dir string) error {
+ if err := os.MkdirAll(dir, 0o770); err != nil {
+ return err
+ }
+ return syncChain(root, dir)
+}
+
+// syncChain syncs dir and every directory above it up to and including root.
+// dir must be root or under it.
+func syncChain(root, dir string) error {
+ root, dir = filepath.Clean(root), filepath.Clean(dir)
+ if rel, err := filepath.Rel(root, dir); err != nil || rel == ".." || strings.HasPrefix(rel, "../") {
+ return fmt.Errorf("%s is not under %s", dir, root)
+ }
+ for d := dir; ; d = filepath.Dir(d) {
+ if err := retireSync(d); err != nil {
+ return err
+ }
+ if d == root || d == filepath.Dir(d) {
+ return nil
+ }
+ }
+}
+
+// syncRename makes a rename from src to dst durable: the source's parent,
+// and every destination directory up to root.
+func syncRename(root, src, dst string) error {
+ if err := retireSync(filepath.Dir(src)); err != nil {
+ return err
+ }
+ return syncChain(root, filepath.Dir(dst))
+}
+
+// parkOpJournal renames the op's journal directory into the retirement's
+// artifacts, at .moves/<host>/<op>. The directory is the op's own (its name
+// is the op id). Absent with the destination there is a replay after the
+// rename, which is synced again and done; absent with no destination is
+// nothing to park; a destination already there beside the journal is
+// refused.
+func parkOpJournal(dataDir, incarnation, host, op, retirement string) error {
+ src := moves.JournalDir(dataDir, host, op)
+ dst := filepath.Join(parkedArtifactsDir(dataDir, incarnation, host, retirement), ".moves", host, op)
+ if _, err := os.Lstat(src); errors.Is(err, os.ErrNotExist) {
+ if _, err := os.Lstat(dst); err == nil {
+ return syncRename(dataDir, src, dst)
+ }
+ return nil
+ } else if err != nil {
+ return err
+ }
+ if _, err := os.Lstat(dst); err == nil {
+ return fmt.Errorf("%s already holds a journal", dst)
+ } else if !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ if err := mkdirSynced(dataDir, filepath.Dir(dst)); err != nil {
+ return err
+ }
+ if err := os.Rename(src, dst); err != nil {
+ return err
+ }
+ return syncRename(dataDir, src, dst)
+}