amadan

carlos / platform Public

moves plan 2: polish (parked review minors)

P2-T1-1: MoveRecordSource replay test pins the defensive copy of Files.
P2-T4-1: staging-preservation test compares full listings, not counts.
P2-T7-1: journal Write fsyncs the parent of every directory it creates.
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
claude 590a44393b0c9a3f5854d0c3aab3e8dbefc6cbaa parent 0431d30
4 files changed, +124 −2
  • internal/activator/backend_exec_test.go +10 −2
  • internal/moves/journal.go +35 −0
  • internal/moves/journal_test.go +60 −0
  • internal/provision/move_test.go +19 −0
diff --git a/internal/activator/backend_exec_test.go b/internal/activator/backend_exec_test.go
index f9a13b19b..8bd1e88b6 100644
--- a/internal/activator/backend_exec_test.go
+++ b/internal/activator/backend_exec_test.go
@@ -7,6 +7,7 @@ import (
"os"
"os/exec"
"path/filepath"
+ "reflect"
"runtime"
"strings"
"sync"
@@ -1258,6 +1259,13 @@ func TestRestoreDoesNotPromoteStagingUnderPhase1(t *testing.T) {
f.put(b.Cfg.Bucket, "moves/"+host+"/generations/a/snapshot", "e1")
f.put(b.Cfg.Bucket, "moves/"+host+"/mv_abcdefghijklmnopqrstuvwxyz/g-cur/0000/0000000000000001-0000000000000002.ltx", "e2")
before, _ := f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/")
+ wantStaging := map[string]lease.MirrorObject{
+ "moves/" + host + "/generations/a/snapshot": {ETag: "e1"},
+ "moves/" + host + "/mv_abcdefghijklmnopqrstuvwxyz/g-cur/0000/0000000000000001-0000000000000002.ltx": {ETag: "e2"},
+ }
+ if !reflect.DeepEqual(before, wantStaging) {
+ t.Fatalf("staging fixture: %v, want %v", before, wantStaging)
+ }
fresh, err := b.Restore(ctx, rt)
if err != nil || !fresh {
@@ -1267,7 +1275,7 @@ func TestRestoreDoesNotPromoteStagingUnderPhase1(t *testing.T) {
t.Fatalf("staging was promoted into the own prefix: %v", own)
}
after, _ := f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/")
- if len(after) != len(before) || len(before) != 2 {
+ if !reflect.DeepEqual(after, before) {
t.Fatalf("staging changed: before %v after %v", before, after)
}
@@ -1278,7 +1286,7 @@ func TestRestoreDoesNotPromoteStagingUnderPhase1(t *testing.T) {
if err != nil || fresh {
t.Fatalf("own lineage: fresh=%v err=%v, want a restore", fresh, err)
}
- if after, _ = f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/"); len(after) != 2 {
+ if after, _ = f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/"); !reflect.DeepEqual(after, before) {
t.Fatalf("staging touched by an own-lineage restore: %v", after)
}
}
diff --git a/internal/moves/journal.go b/internal/moves/journal.go
index 4fc2a7e05..d83fe8fef 100644
--- a/internal/moves/journal.go
+++ b/internal/moves/journal.go
@@ -111,9 +111,21 @@ func (j *Journal) Write(dataDir string) error {
return err
}
dir := JournalDir(dataDir, j.Host, j.Op)
+ created, err := missingDirs(dir)
+ if err != nil {
+ return err
+ }
if err := os.MkdirAll(dir, 0o700); err != nil {
return err
}
+ // A directory this call created is only durable once its parent's
+ // entry is, so sync the parent of each one, or a crash could lose the
+ // whole chain with the journal inside it.
+ for _, c := range created {
+ if err := syncDir(filepath.Dir(c)); err != nil {
+ return err
+ }
+ }
tmp, err := os.CreateTemp(dir, journalFile+".tmp-*")
if err != nil {
return err
@@ -138,6 +150,29 @@ func (j *Journal) Write(dataDir string) error {
os.Remove(tmpName)
return err
}
+ return syncDir(dir)
+}
+
+// missingDirs lists the directories at and above dir that do not exist yet,
+// deepest first: the ones MkdirAll is about to create.
+func missingDirs(dir string) ([]string, error) {
+ var out []string
+ for p := filepath.Clean(dir); ; p = filepath.Dir(p) {
+ _, err := os.Lstat(p)
+ if err == nil {
+ return out, nil
+ }
+ if !errors.Is(err, fs.ErrNotExist) {
+ return nil, err
+ }
+ out = append(out, p)
+ if filepath.Dir(p) == p {
+ return out, nil
+ }
+ }
+}
+
+func syncDir(dir string) error {
d, err := os.Open(dir)
if err != nil {
return err
diff --git a/internal/moves/journal_test.go b/internal/moves/journal_test.go
index 33ad2c5ca..3e6ddf3e3 100644
--- a/internal/moves/journal_test.go
+++ b/internal/moves/journal_test.go
@@ -151,3 +151,63 @@ func TestJournalMarkOnZeroValue(t *testing.T) {
t.Fatal("unexpected")
}
}
+
+func TestMissingDirsListsWhatMkdirAllWillCreate(t *testing.T) {
+ dir := t.TempDir()
+ deep := filepath.Join(dir, "a", "b", "c")
+ got, err := missingDirs(deep)
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := []string{deep, filepath.Join(dir, "a", "b"), filepath.Join(dir, "a")}
+ if strings.Join(got, "|") != strings.Join(want, "|") {
+ t.Fatalf("missingDirs = %v, want %v", got, want)
+ }
+ if err := os.MkdirAll(deep, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if got, err = missingDirs(deep); err != nil || len(got) != 0 {
+ t.Fatalf("missingDirs of an existing dir = %v, %v", got, err)
+ }
+}
+
+// A first write creates .moves/<host>/<op>/ (every component new); a second
+// write to another op reuses .moves/<host>/. Both write and read back.
+func TestJournalWriteCreatesTheChainAndReadsBack(t *testing.T) {
+ dir := filepath.Join(t.TempDir(), "data") // dataDir itself is new too
+ for _, op := range []string{"op1", "op2"} {
+ j, err := ReadJournal(dir, "h", op)
+ if err != nil {
+ t.Fatal(err)
+ }
+ j.Mark("a", op)
+ if err := j.Write(dir); err != nil {
+ t.Fatalf("Write %s: %v", op, err)
+ }
+ back, err := ReadJournal(dir, "h", op)
+ if err != nil || back.Steps["a"] != op {
+ t.Fatalf("read back %s: %+v %v", op, back, err)
+ }
+ }
+}
+
+func TestJournalWriteUnwritableParentErrorsCleanly(t *testing.T) {
+ if os.Geteuid() == 0 {
+ t.Skip("root ignores directory permissions")
+ }
+ dir := t.TempDir()
+ if err := os.MkdirAll(filepath.Join(dir, ".moves"), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chmod(filepath.Join(dir, ".moves"), 0o500); err != nil {
+ t.Fatal(err)
+ }
+ defer os.Chmod(filepath.Join(dir, ".moves"), 0o700)
+ j := &Journal{Host: "h", Op: "op"}
+ if err := j.Write(dir); err == nil {
+ t.Fatal("Write under an unwritable parent succeeded")
+ }
+ if _, err := os.Stat(JournalDir(dir, "h", "op")); err == nil {
+ t.Fatal("a failed Write left the journal directory")
+ }
+}
diff --git a/internal/provision/move_test.go b/internal/provision/move_test.go
index 20d29cc2d..e55036c30 100644
--- a/internal/provision/move_test.go
+++ b/internal/provision/move_test.go
@@ -298,6 +298,25 @@ func TestCutPhasesAdvanceInOrderUnderTheGuard(t *testing.T) {
if err := MoveRecordSource(g, MoveSource{FinalTXID: 42, TimersSHA256: "sha256:t"})(rec); err != nil || rec.Move.Source.FinalTXID != 42 {
t.Fatalf("source: %v", err)
}
+ // A replay of the same transition succeeds, and the record keeps its
+ // own copy of the files digest: the caller mutating its value after
+ // the call does not reach the record (P2-T1-1).
+ files := &MoveFilesDigest{Plain: "sha256:p", Sealed: "sha256:s", Bytes: 7, Entries: 2}
+ src := MoveSource{FinalTXID: 42, TimersSHA256: "sha256:t", Files: files}
+ for i := 0; i < 2; i++ {
+ if err := MoveRecordSource(g, src)(rec); err != nil || rec.Move.Source.FinalTXID != 42 {
+ t.Fatalf("source replay %d: %v", i, err)
+ }
+ }
+ if rec.Move.Source.Files == nil || rec.Move.Source.Files == files {
+ t.Fatalf("the record shares the caller's files digest: %p vs %p", rec.Move.Source.Files, files)
+ }
+ want := MoveFilesDigest{Plain: "sha256:p", Sealed: "sha256:s", Bytes: 7, Entries: 2}
+ files.Plain, files.Sealed, files.Bytes, files.Entries = "x", "y", 99, 99
+ src.Files = nil
+ if got := *rec.Move.Source.Files; got != want {
+ t.Fatalf("the record's files digest followed the caller's: %+v, want %+v", got, want)
+ }
for _, step := range [][2]string{{CutStopped, CutStaged}, {CutStaged, CutArchived}} {
if err := MoveCutPhase(g, step[0], step[1], nil)(rec); err != nil {
t.Fatal(err)