amadan

carlos / platform Public

moves.Journal: the box-local operation journal (§9.1, §9.2, F12)

JournalDir, ReadJournal (absent is empty, corrupt fails closed), Write
(temp file, fsync, rename, directory fsync), Done and Mark.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 0a6296e06081a545f52268fbba9cb38323d81662 parent 45a3af2
2 files changed, +315 −0
  • internal/moves/journal.go +162 −0
  • internal/moves/journal_test.go +153 −0
diff --git a/internal/moves/journal.go b/internal/moves/journal.go
new file mode 100644
index 000000000..4fc2a7e05
--- /dev/null
+++ b/internal/moves/journal.go
@@ -0,0 +1,162 @@
+package moves
+
+import (
+ "bytes"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io/fs"
+ "os"
+ "path/filepath"
+ "strings"
+ "time"
+)
+
+// journalVersion is the only journal format this code reads or writes. A
+// file with any other version is refused, never reinterpreted.
+const journalVersion = 1
+
+const journalFile = "journal.json"
+
+// Journal is the box-local record of how far one move operation has got on
+// this box (§9.1, §9.2). A step that is Done is never redone after a
+// restart; the value a step was marked with is whatever that step needs to
+// resume (a digest, an id, a timestamp). Values are internal state, not
+// operator copy.
+type Journal struct {
+ V int `json:"v"`
+ Host string `json:"host"`
+ Op string `json:"op"`
+ Phase string `json:"phase,omitempty"`
+ Steps map[string]string `json:"steps,omitempty"`
+ Attempt int `json:"attempt,omitempty"`
+ UpdatedAt time.Time `json:"updated_at"`
+}
+
+// JournalDir is where one operation's journal lives on the box:
+// <dataDir>/.moves/<host>/<op>/.
+func JournalDir(dataDir, host, op string) string {
+ return filepath.Join(dataDir, ".moves", host, op)
+}
+
+// journalName refuses a host or operation id that is empty or could leave
+// its directory. Both come from console-issued records, but the journal
+// path is built from them, so they are checked here as well.
+func journalName(kind, s string) error {
+ if s == "" || s == "." || s == ".." || strings.ContainsAny(s, "/\\\x00") {
+ return fmt.Errorf("moves journal: unusable %s %q", kind, s)
+ }
+ return nil
+}
+
+func journalNames(host, op string) error {
+ if err := journalName("host", host); err != nil {
+ return err
+ }
+ return journalName("op", op)
+}
+
+// ReadJournal loads the journal for (host, op). A journal that does not
+// exist yet is an empty one. A file that exists but cannot be trusted
+// (unreadable, not valid JSON, an unknown version, or naming a different
+// host or op) is an error and never an empty journal: starting over could
+// redo a step that already ran.
+func ReadJournal(dataDir, host, op string) (*Journal, error) {
+ if err := journalNames(host, op); err != nil {
+ return nil, err
+ }
+ b, err := os.ReadFile(filepath.Join(JournalDir(dataDir, host, op), journalFile))
+ if errors.Is(err, fs.ErrNotExist) {
+ return &Journal{V: journalVersion, Host: host, Op: op, Steps: map[string]string{}}, nil
+ }
+ if err != nil {
+ return nil, fmt.Errorf("moves journal %s/%s: %w", host, op, err)
+ }
+ var j Journal
+ dec := json.NewDecoder(bytes.NewReader(b))
+ if err := dec.Decode(&j); err != nil {
+ return nil, fmt.Errorf("moves journal %s/%s is corrupt: %w", host, op, err)
+ }
+ if dec.More() {
+ return nil, fmt.Errorf("moves journal %s/%s is corrupt: trailing data", host, op)
+ }
+ if j.V != journalVersion {
+ return nil, fmt.Errorf("moves journal %s/%s: unknown version %d", host, op, j.V)
+ }
+ if j.Host != host || j.Op != op {
+ return nil, fmt.Errorf("moves journal %s/%s names %s/%s", host, op, j.Host, j.Op)
+ }
+ if j.Steps == nil {
+ j.Steps = map[string]string{}
+ }
+ return &j, nil
+}
+
+// Write stamps UpdatedAt and saves the journal: a temp file in the same
+// directory, fsync, rename over the old file, then fsync of the directory,
+// so a crash leaves either the old journal or the new one.
+func (j *Journal) Write(dataDir string) error {
+ if err := journalNames(j.Host, j.Op); err != nil {
+ return err
+ }
+ if j.V == 0 {
+ j.V = journalVersion
+ }
+ if j.V != journalVersion {
+ return fmt.Errorf("moves journal %s/%s: unknown version %d", j.Host, j.Op, j.V)
+ }
+ j.UpdatedAt = time.Now().UTC()
+ b, err := json.Marshal(j)
+ if err != nil {
+ return err
+ }
+ dir := JournalDir(dataDir, j.Host, j.Op)
+ if err := os.MkdirAll(dir, 0o700); err != nil {
+ return err
+ }
+ tmp, err := os.CreateTemp(dir, journalFile+".tmp-*")
+ if err != nil {
+ return err
+ }
+ tmpName := tmp.Name()
+ fail := func(err error) error {
+ tmp.Close()
+ os.Remove(tmpName)
+ return err
+ }
+ if _, err := tmp.Write(b); err != nil {
+ return fail(err)
+ }
+ if err := tmp.Sync(); err != nil {
+ return fail(err)
+ }
+ if err := tmp.Close(); err != nil {
+ os.Remove(tmpName)
+ return err
+ }
+ if err := os.Rename(tmpName, filepath.Join(dir, journalFile)); err != nil {
+ os.Remove(tmpName)
+ return err
+ }
+ d, err := os.Open(dir)
+ if err != nil {
+ return err
+ }
+ defer d.Close()
+ return d.Sync()
+}
+
+// Done reports whether step has been marked.
+func (j *Journal) Done(step string) bool {
+ _, ok := j.Steps[step]
+ return ok
+}
+
+// Mark records step as done with value. It changes the in-memory journal
+// only; Write makes it durable.
+func (j *Journal) Mark(step, value string) {
+ if j.Steps == nil {
+ j.Steps = map[string]string{}
+ }
+ j.Steps[step] = value
+}
diff --git a/internal/moves/journal_test.go b/internal/moves/journal_test.go
new file mode 100644
index 000000000..33ad2c5ca
--- /dev/null
+++ b/internal/moves/journal_test.go
@@ -0,0 +1,153 @@
+package moves
+
+import (
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestJournalDirLayout(t *testing.T) {
+ got := JournalDir("/data", "app.example.com", "op1")
+ want := filepath.Join("/data", ".moves", "app.example.com", "op1")
+ if got != want {
+ t.Fatalf("JournalDir = %q, want %q", got, want)
+ }
+}
+
+func TestReadJournalAbsentIsEmpty(t *testing.T) {
+ dir := t.TempDir()
+ j, err := ReadJournal(dir, "h", "op")
+ if err != nil {
+ t.Fatalf("absent journal: %v", err)
+ }
+ if j == nil || j.Host != "h" || j.Op != "op" || j.Phase != "" || j.Attempt != 0 || len(j.Steps) != 0 {
+ t.Fatalf("absent journal not empty: %+v", j)
+ }
+ if j.Done("cut") {
+ t.Fatal("empty journal reports a step done")
+ }
+ // Reading must not create anything.
+ if _, err := os.Stat(filepath.Join(dir, ".moves")); !os.IsNotExist(err) {
+ t.Fatalf("ReadJournal created state: %v", err)
+ }
+}
+
+func TestJournalRoundTripAndDone(t *testing.T) {
+ dir := t.TempDir()
+ j, err := ReadJournal(dir, "h", "op")
+ if err != nil {
+ t.Fatal(err)
+ }
+ j.Phase = "cut"
+ j.Attempt = 2
+ if j.Done("seal") {
+ t.Fatal("Done before Mark")
+ }
+ j.Mark("seal", "sha256:abc")
+ j.Mark("flip", "")
+ if !j.Done("seal") || !j.Done("flip") || j.Done("land") {
+ t.Fatalf("Done does not reflect Mark: %+v", j.Steps)
+ }
+ if err := j.Write(dir); err != nil {
+ t.Fatal(err)
+ }
+ if j.UpdatedAt.IsZero() || time.Since(j.UpdatedAt) > time.Minute {
+ t.Fatalf("UpdatedAt not stamped: %v", j.UpdatedAt)
+ }
+ back, err := ReadJournal(dir, "h", "op")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if back.V != j.V || back.Host != "h" || back.Op != "op" || back.Phase != "cut" || back.Attempt != 2 {
+ t.Fatalf("round trip: %+v", back)
+ }
+ if back.Steps["seal"] != "sha256:abc" || !back.Done("flip") || back.Done("land") {
+ t.Fatalf("steps lost: %+v", back.Steps)
+ }
+ if !back.UpdatedAt.Equal(j.UpdatedAt) {
+ t.Fatalf("UpdatedAt %v != %v", back.UpdatedAt, j.UpdatedAt)
+ }
+}
+
+func TestJournalWriteLeavesOnlyTheJournal(t *testing.T) {
+ dir := t.TempDir()
+ j, _ := ReadJournal(dir, "h", "op")
+ j.Mark("a", "1")
+ for i := 0; i < 3; i++ {
+ j.Attempt = i
+ if err := j.Write(dir); err != nil {
+ t.Fatal(err)
+ }
+ }
+ ents, err := os.ReadDir(JournalDir(dir, "h", "op"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(ents) != 1 || ents[0].Name() != "journal.json" {
+ t.Fatalf("directory holds %v, want only journal.json", ents)
+ }
+ fi, err := os.Stat(JournalDir(dir, "h", "op"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if fi.Mode().Perm() != 0o700 {
+ t.Fatalf("journal dir mode %v, want 0700", fi.Mode().Perm())
+ }
+}
+
+func TestJournalCorruptFailsClosed(t *testing.T) {
+ cases := map[string]string{
+ "garbage": "{not json",
+ "empty": "",
+ "truncated": `{"v":1,"host":"h","op":"op","steps":{"a"`,
+ "wrong host": `{"v":1,"host":"other","op":"op"}`,
+ "wrong op": `{"v":1,"host":"h","op":"other"}`,
+ "future version": `{"v":99,"host":"h","op":"op"}`,
+ "zero version": `{"host":"h","op":"op"}`,
+ }
+ for name, body := range cases {
+ t.Run(name, func(t *testing.T) {
+ dir := t.TempDir()
+ jd := JournalDir(dir, "h", "op")
+ if err := os.MkdirAll(jd, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(jd, "journal.json"), []byte(body), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ j, err := ReadJournal(dir, "h", "op")
+ if err == nil {
+ t.Fatalf("corrupt journal read as %+v", j)
+ }
+ })
+ }
+}
+
+func TestJournalRejectsUnsafeNames(t *testing.T) {
+ dir := t.TempDir()
+ for _, c := range [][2]string{{"", "op"}, {"h", ""}, {"../x", "op"}, {"h", "a/b"}, {".", "op"}, {"h", ".."}} {
+ if _, err := ReadJournal(dir, c[0], c[1]); err == nil {
+ t.Errorf("ReadJournal(%q,%q) accepted", c[0], c[1])
+ }
+ j := &Journal{V: 1, Host: c[0], Op: c[1]}
+ if err := j.Write(dir); err == nil {
+ t.Errorf("Write(%q,%q) accepted", c[0], c[1])
+ }
+ }
+ if ents, _ := os.ReadDir(dir); len(ents) != 0 {
+ t.Fatalf("rejected names left files: %v", ents)
+ }
+}
+
+func TestJournalMarkOnZeroValue(t *testing.T) {
+ var j Journal
+ j.Mark("x", "y")
+ if !j.Done("x") || j.Steps["x"] != "y" {
+ t.Fatalf("%+v", j)
+ }
+ if strings.Contains(JournalDir("d", "h", "o"), "..") {
+ t.Fatal("unexpected")
+ }
+}