The marker records which object the publish is placing

cmd/flux-op/publish.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"syscall"
 )
 
 // publish moves the result into place.
@@ -30,6 +31,23 @@ import (
 // without something else already being wrong - and a non-atomic publish is
 // exactly what the caller was promised would not occur.
 func publish(staging, destination, root, id string) error {
+	// Which object is about to be published, read before anything moves. A
+	// sweep that finds a destination occupied cannot otherwise tell what is
+	// sitting there: the object this publish placed, or one the app owner put
+	// at that path themselves while the destination stood empty. Recording the
+	// answer costs one lstat and replaces a guess with a comparison.
+	//
+	// This also means a staging path that is not there fails before the
+	// destination is displaced rather than after, which leaves nothing to sweep.
+	staged, err := os.Lstat(staging)
+	if err != nil {
+		return err
+	}
+	stagedIdentity, err := identity(staged)
+	if err != nil {
+		return err
+	}
+
 	// Lstat, not Stat: a dangling symlink at the destination is an entry that
 	// has to be moved aside, and a check that followed it would treat the
 	// destination as empty and rename over the link.
@@ -47,7 +65,8 @@ func publish(staging, destination, root, id string) error {
 	// crash between the two renames below leaves the caller's previous data
 	// under `old` with its own path empty, and without this the sweep has no way
 	// to know where to put it back - it would delete the only copy.
-	if err := os.WriteFile(marker, []byte(markerContents(destination, root)+"\n"), 0o644); err != nil {
+	record := markerContents(destination, root) + "\n" + stagedIdentity + "\n"
+	if err := os.WriteFile(marker, []byte(record), 0o644); err != nil {
 		return fmt.Errorf("could not record where %s belongs: %w", destination, err)
 	}
 
@@ -75,3 +94,27 @@ const (
 	swapPrefix   = ".flux-old-"
 	markerSuffix = ".dest"
 )
+
+// identity is what a sweep compares to decide whether a publish finished: the
+// inode number of the object being placed, and its modification time in
+// nanoseconds.
+//
+// Both survive rename, which is what makes them usable at all - ctime does not,
+// since rename updates it, and a recorded ctime would mismatch the moment the
+// publish succeeded.
+//
+// The inode number alone is not enough. Filesystems reuse them, so an entry the
+// app owner creates at the destination after a publish can carry the number
+// recorded here and be taken for the published object. An mtime to the
+// nanosecond does not collide by accident.
+//
+// It does not have to resist being forged. The app owner can read this file
+// through the file browser and can set an mtime, but a marker they match only
+// makes the sweep delete the data it was holding for them - their own.
+func identity(fi os.FileInfo) (string, error) {
+	stat, ok := fi.Sys().(*syscall.Stat_t)
+	if !ok {
+		return "", fmt.Errorf("cannot read the identity of %s on this platform", fi.Name())
+	}
+	return fmt.Sprintf("%d %d", stat.Ino, fi.ModTime().UnixNano()), nil
+}

cmd/flux-op/publish_test.go

@@ -1,8 +1,10 @@
 package main
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
+	"syscall"
 	"testing"
 )
 
@@ -157,27 +159,98 @@ func TestPublishTreatsADanglingSymlinkAsAnExistingEntry(t *testing.T) {
 // aside and the replacement never arrived. Reproduced by publishing a staging
 // path that does not exist, so the second rename fails exactly where a crash
 // would land.
+// What a sweep compares, derived independently of the code under test so the
+// format is pinned rather than echoed back.
+func identityOf(t *testing.T, path string) string {
+	t.Helper()
+	info, err := os.Lstat(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	stat, ok := info.Sys().(*syscall.Stat_t)
+	if !ok {
+		t.Fatalf("no stat for %s", path)
+	}
+	return fmt.Sprintf("%d %d", stat.Ino, info.ModTime().UnixNano())
+}
+
 func TestAnInterruptedPublishLeavesTheDataAndAMarkerThatPlacesIt(t *testing.T) {
 	root := t.TempDir()
-	staging := filepath.Join(root, ".flux-op-"+testID)
-	destination := filepath.Join(root, "nested", "deeper", "dest")
-	write(t, destination, "precious")
+
+	// A directory published over its own parent, which is a move a user can
+	// ask for. The first rename carries the staging path away inside the
+	// destination, so the second finds nothing at it and the publish stops
+	// between the two - the state a crash in that window leaves, reached
+	// without one.
+	destination := filepath.Join(root, "photos")
+	staging := filepath.Join(destination, "2024")
+	write(t, staging, "precious")
 
 	if err := publish(staging, destination, root, testID); err == nil {
-		t.Fatal("publishing a staging path that does not exist succeeded")
+		t.Fatal("publishing a directory over its own parent succeeded")
 	}
 
 	displaced := filepath.Join(root, swapPrefix+testID)
-	if got := read(t, displaced); got != "precious" {
+	if got := read(t, filepath.Join(displaced, "2024")); got != "precious" {
 		t.Errorf("displaced data holds %q, want precious", got)
 	}
+	if _, err := os.Lstat(destination); !os.IsNotExist(err) {
+		t.Error("the destination is still there, so this is not the interrupted state")
+	}
 
 	// Relative, so nothing that reads it can be sent off the volume by
 	// following an absolute path. At the volume root, which is the one
 	// directory the sweep reads - not beside the destination, wherever the
 	// caller happened to keep it.
-	if got := read(t, displaced+markerSuffix); got != "nested/deeper/dest\n" {
-		t.Errorf("marker holds %q, want nested/deeper/dest", got)
+	//
+	// The identity is read from the displaced copy: rename preserved it, which
+	// is the property the whole record depends on.
+	want := "photos\n" + identityOf(t, filepath.Join(displaced, "2024")) + "\n"
+	if got := read(t, displaced+markerSuffix); got != want {
+		t.Errorf("marker holds %q, want %q", got, want)
+	}
+}
+
+// The record has to name the object being placed, not the one being displaced -
+// the sweep compares it against whatever occupies the destination afterwards.
+func TestTheMarkerRecordsTheIdentityOfWhatIsBeingPublished(t *testing.T) {
+	root := t.TempDir()
+	destination := filepath.Join(root, "photos")
+	staging := filepath.Join(destination, "2024")
+	write(t, staging, "precious")
+	displacedIdentity := identityOf(t, destination)
+
+	if err := publish(staging, destination, root, testID); err == nil {
+		t.Fatal("publishing a directory over its own parent succeeded")
+	}
+
+	marker := read(t, filepath.Join(root, swapPrefix+testID)+markerSuffix)
+	if got := identityOf(t, filepath.Join(root, swapPrefix+testID, "2024")); marker != "photos\n"+got+"\n" {
+		t.Errorf("marker holds %q, want the identity of the published object %q", marker, got)
+	}
+	if marker == "photos\n"+displacedIdentity+"\n" {
+		t.Error("marker records the displaced entry rather than what is being published")
+	}
+}
+
+// Nothing is displaced over an operation that cannot be carried out, so there is
+// nothing for a sweep to put back afterwards.
+func TestAMissingStagingPathFailsBeforeAnythingMoves(t *testing.T) {
+	root := t.TempDir()
+	staging := filepath.Join(root, ".flux-op-"+testID)
+	destination := filepath.Join(root, "dest")
+	write(t, destination, "precious")
+
+	if err := publish(staging, destination, root, testID); err == nil {
+		t.Fatal("publishing a staging path that does not exist succeeded")
+	}
+
+	if got := read(t, destination); got != "precious" {
+		t.Errorf("destination holds %q, want precious", got)
+	}
+	entries, _ := os.ReadDir(root)
+	if len(entries) != 1 {
+		t.Errorf("volume holds %d entries, want only the untouched destination: %v", len(entries), entries)
 	}
 }
 

test/container/container_test.go

@@ -156,6 +156,33 @@ func contents(t *testing.T, volume, name string) string {
 	return strings.TrimRight(string(data), "\n")
 }
 
+// What a sweep compares, derived independently of the code under test so the
+// format is pinned rather than echoed back.
+//
+// Read from inside a container, because an inode number belongs to the
+// filesystem that issued it and a bind mount does not always carry it across
+// unchanged: Docker Desktop synthesises its own on macOS, where a node's Linux
+// bind mount hands back the same number the host sees. flux-op recorded what it
+// saw from in there, so the comparison is made from the same side.
+func identityOf(t *testing.T, volume, name string) string {
+	t.Helper()
+	result := inContainer(t, volume, "", `stat -c '%i %.9Y' "$@"`, "/work/"+name)
+	if result.exit != 0 {
+		t.Fatalf("could not stat %s (exit %d):\n%s", name, result.exit, result.output)
+	}
+	// The last line, not the whole output: running a foreign architecture under
+	// emulation puts a platform warning ahead of it, which is the same noise the
+	// exit code is read past below.
+	lines := strings.Split(strings.TrimSpace(result.output), "\n")
+	fields := strings.Fields(lines[len(lines)-1])
+	if len(fields) != 2 {
+		t.Fatalf("stat of %s returned %q", name, result.output)
+	}
+	// Seconds and nanoseconds arrive as one decimal number, padded to nine
+	// places, so removing the point is the whole conversion.
+	return fields[0] + " " + strings.Replace(fields[1], ".", "", 1)
+}
+
 func exists(volume, name string) bool {
 	_, err := os.Lstat(filepath.Join(volume, name))
 	return err == nil

test/container/publish_test.go

@@ -191,30 +191,63 @@ func TestAFailureNeverDiscardsAnOperandTheCallerOwns(t *testing.T) {
 // would land.
 func TestAnInterruptedPublishLeavesTheDataAndAMarkerThatPlacesIt(t *testing.T) {
 	volume := volumeDir(t)
-	seed(t, volume, `mkdir -p /work/a/b /work/x/y && echo precious > /work/x/y/out`)
+	seed(t, volume, `mkdir -p /work/x/y/out && echo precious > /work/x/y/out/2024`)
 
-	result := fluxOp(t, volume, "", baseArgs("/work/a/b/photos", "/work/x/y/out", "--")...)
+	// A directory published over its own parent, which is a move a user can ask
+	// for. The first rename carries the staging path away inside the
+	// destination, so the second finds nothing at it and the publish stops
+	// between the two - the state a crash in that window leaves, without having
+	// to kill the process to reach it.
+	result := fluxOp(t, volume, "", baseArgs("/work/x/y/out/2024", "/work/x/y/out", "--")...)
 
 	if result.exit == 0 {
-		t.Fatalf("publishing a staging path that does not exist succeeded:\n%s", result.output)
+		t.Fatalf("publishing a directory over its own parent succeeded:\n%s", result.output)
 	}
 
 	displaced := ".flux-old-" + operationID
-	if got := contents(t, volume, displaced); got != "precious" {
+	if got := contents(t, volume, displaced+"/2024"); got != "precious" {
 		t.Errorf("displaced data holds %q, want precious\n%s", got, tree(volume))
 	}
+	if exists(volume, "x/y/out") {
+		t.Error("the destination is still there, so this is not the interrupted state")
+	}
 
 	// Relative, so nothing that reads it can be sent off the volume by following
 	// an absolute path. At the volume root, which is the one directory the sweep
 	// reads - not beside the destination, wherever the caller kept it.
-	if got := contents(t, volume, displaced+".dest"); got != "x/y/out" {
-		t.Errorf("marker holds %q, want x/y/out", got)
+	marker := contents(t, volume, displaced+".dest")
+	lines := strings.Split(marker, "\n")
+	if len(lines) != 2 {
+		t.Fatalf("marker holds %q, want a destination and an identity", marker)
 	}
-	if exists(volume, "x/y/"+displaced) {
-		t.Error("the artefacts were left beside the destination rather than at the volume root")
+	if lines[0] != "x/y/out" {
+		t.Errorf("marker names %q, want x/y/out", lines[0])
 	}
-	if exists(volume, "x/y/out") {
-		t.Error("the destination is still there, so this is not the interrupted state")
+
+	// The identity is read from the displaced copy: rename preserved it, which
+	// is the property that lets a sweep tell the published object from one the
+	// app owner put at the same path.
+	if want := identityOf(t, volume, displaced+"/2024"); lines[1] != want {
+		t.Errorf("marker identity is %q, want %q\n%s", lines[1], want, tree(volume))
+	}
+}
+
+// Nothing is displaced over an operation that cannot be carried out, so there is
+// nothing for a sweep to put back afterwards.
+func TestAMissingStagingPathFailsBeforeAnythingMoves(t *testing.T) {
+	volume := volumeDir(t)
+	seed(t, volume, `mkdir -p /work/x/y && echo precious > /work/x/y/out`)
+
+	result := fluxOp(t, volume, "", baseArgs("/work/a/b/photos", "/work/x/y/out", "--")...)
+
+	if result.exit == 0 {
+		t.Fatalf("publishing a staging path that does not exist succeeded:\n%s", result.output)
+	}
+	if got := contents(t, volume, "x/y/out"); got != "precious" {
+		t.Errorf("destination holds %q, want precious\n%s", got, tree(volume))
+	}
+	if exists(volume, ".flux-old-"+operationID) || exists(volume, ".flux-old-"+operationID+".dest") {
+		t.Errorf("an operation that never started left artefacts behind\n%s", tree(volume))
 	}
 }