amadan

carlos / platform Public

hostfiles: immutable objects and one manifest decide which file content is current

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude 91de8c49c6a34cbf2efc73ab790c762c80dfe924 parent e9246bd
4 files changed, +1067 −0
  • internal/hostfiles/hostfiles_test.go +489 −0
  • internal/hostfiles/manifest.go +91 −0
  • internal/hostfiles/publisher.go +192 −0
  • internal/hostfiles/restore.go +295 −0
diff --git a/internal/hostfiles/hostfiles_test.go b/internal/hostfiles/hostfiles_test.go
new file mode 100644
index 000000000..ca603ac23
--- /dev/null
+++ b/internal/hostfiles/hostfiles_test.go
@@ -0,0 +1,489 @@
+package hostfiles
+
+import (
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "reflect"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const pfx = "instances/shop.test/abc/files/g1/"
+
+func open(t *testing.T, s release.Store) *Publisher {
+ t.Helper()
+ p, err := Open(context.Background(), s, pfx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return p
+}
+
+func stage(t *testing.T, p *Publisher, path, body string) {
+ t.Helper()
+ if err := p.Stage(context.Background(), path, []byte(body), 0o600, false); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func commit(t *testing.T, p *Publisher) int64 {
+ t.Helper()
+ seq, err := p.Commit(context.Background())
+ if err != nil {
+ t.Fatal(err)
+ }
+ return seq
+}
+
+func read(t *testing.T, dir, rel string) string {
+ t.Helper()
+ b, err := os.ReadFile(filepath.Join(dir, rel))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return string(b)
+}
+
+func TestPublishThenRestoreRoundTrips(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ stage(t, p, "instance.key", "k1")
+ stage(t, p, "uploads/a/b.txt", "hello")
+ if seq := commit(t, p); seq != 2 {
+ t.Fatalf("seq = %d, want 2", seq)
+ }
+ dir := t.TempDir()
+ m, st, err := Restore(ctx, s, pfx, dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if m.Seq != 2 || st.Written != 2 {
+ t.Fatalf("manifest seq %d, written %d", m.Seq, st.Written)
+ }
+ if read(t, dir, "instance.key") != "k1" || read(t, dir, "uploads/a/b.txt") != "hello" {
+ t.Fatal("content differs")
+ }
+ fi, _ := os.Stat(filepath.Join(dir, "instance.key"))
+ if fi.Mode().Perm() != 0o600 {
+ t.Fatalf("mode = %v, want 0600", fi.Mode().Perm())
+ }
+}
+
+func TestALaterWriteWinsAndAnUnchangedOneTakesNoSequence(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ stage(t, p, "instance.key", "old")
+ commit(t, p)
+ stage(t, p, "instance.key", "new")
+ stage(t, p, "instance.key", "new") // unchanged: no new sequence number
+ if seq := commit(t, p); seq != 2 {
+ t.Fatalf("seq = %d, want 2", seq)
+ }
+ dir := t.TempDir()
+ if _, _, err := Restore(ctx, s, pfx, dir); err != nil {
+ t.Fatal(err)
+ }
+ if got := read(t, dir, "instance.key"); got != "new" {
+ t.Fatalf("restored %q, want new", got)
+ }
+}
+
+// A1-4: a stalled upload of the OLD content that lands after the new
+// content was published must not come back.
+func TestALateUploadCannotResurrectOldContent(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ stage(t, p, "instance.key", "new")
+ commit(t, p)
+ // The late arrival: any object at all under objects/, with a higher number.
+ if _, err := s.Put(ctx, pfx+"objects/0000000000000099-deadbeef", []byte("old"), ""); err != nil {
+ t.Fatal(err)
+ }
+ dir := t.TempDir()
+ if _, _, err := Restore(ctx, s, pfx, dir); err != nil {
+ t.Fatal(err)
+ }
+ if got := read(t, dir, "instance.key"); got != "new" {
+ t.Fatalf("restored %q, want new", got)
+ }
+}
+
+func TestADeleteIsATombstoneAndRestoreRemovesTheFile(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ stage(t, p, "a.txt", "a")
+ stage(t, p, "b.txt", "b")
+ commit(t, p)
+ dir := t.TempDir()
+ if _, _, err := Restore(ctx, s, pfx, dir); err != nil {
+ t.Fatal(err)
+ }
+ if err := p.StageDelete("a.txt"); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+ _, st, err := Restore(ctx, s, pfx, dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := os.Stat(filepath.Join(dir, "a.txt")); !os.IsNotExist(err) {
+ t.Fatalf("a.txt still there: %v", err)
+ }
+ if st.Removed != 1 || read(t, dir, "b.txt") != "b" {
+ t.Fatalf("removed %d", st.Removed)
+ }
+ if got := p.Paths(); len(got) != 1 || got[0] != "b.txt" {
+ t.Fatalf("Paths = %v", got)
+ }
+}
+
+func TestASecondWriterIsSupersededNotMerged(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ a, b := open(t, s), open(t, s)
+ stage(t, a, "x", "from a")
+ commit(t, a)
+ stage(t, b, "x", "from b")
+ if _, err := b.Commit(context.Background()); !errors.Is(err, ErrSuperseded) {
+ t.Fatalf("err = %v, want ErrSuperseded", err)
+ }
+ dir := t.TempDir()
+ if _, _, err := Restore(context.Background(), s, pfx, dir); err != nil {
+ t.Fatal(err)
+ }
+ if got := read(t, dir, "x"); got != "from a" {
+ t.Fatalf("restored %q, want from a", got)
+ }
+}
+
+func TestAReopenedPublisherContinuesTheSequence(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ stage(t, p, "x", "1")
+ commit(t, p)
+ q := open(t, s)
+ stage(t, q, "y", "2")
+ if seq := commit(t, q); seq != 2 {
+ t.Fatalf("seq = %d, want 2", seq)
+ }
+}
+
+// Review Focus 1.
+func TestCleanPathRefusesEscapes(t *testing.T) {
+ for _, bad := range []string{"", "/etc/passwd", "../x", "a/../../x", "a/../b", "./a", "a//b", "a/", ".", "..", "a\x00b"} {
+ if _, err := CleanPath(bad); err == nil {
+ t.Errorf("CleanPath(%q) accepted", bad)
+ }
+ }
+ for _, ok := range []string{"a", "a/b.txt", ".hidden", "a b/c", `back\slash`} {
+ if got, err := CleanPath(ok); err != nil || got != ok {
+ t.Errorf("CleanPath(%q) = %q, %v", ok, got, err)
+ }
+ }
+}
+
+func TestStageAndRestoreRefuseAnEscapingPath(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ if err := p.Stage(ctx, "../evil", []byte("x"), 0o600, false); err == nil {
+ t.Fatal("Stage accepted ../evil")
+ }
+ // A manifest written by something else with a hostile path.
+ body := `{"v":1,"seq":1,"entries":{"../evil":{"object":"objects/0000000000000001-aa","sha256":"aa","seq":1}},"platform":{}}`
+ hostile := "instances/h/i/files/g/"
+ if _, err := s.Put(ctx, ManifestKey(hostile), []byte(body), ""); err != nil {
+ t.Fatal(err)
+ }
+ parent := t.TempDir()
+ dir := filepath.Join(parent, "files")
+ if err := os.Mkdir(dir, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := Restore(ctx, s, hostile, dir); err == nil {
+ t.Fatal("Restore accepted a manifest with ../evil")
+ }
+ if _, err := os.Stat(filepath.Join(parent, "evil")); !os.IsNotExist(err) {
+ t.Fatal("../evil was written")
+ }
+}
+
+func TestRestoreRefusesContentThatDoesNotMatchItsHash(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ stage(t, p, "x", "good")
+ commit(t, p)
+ m, _, err := Load(ctx, s, pfx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ key := pfx + m.Entries["x"].Object
+ _, etag, _ := s.Get(ctx, key)
+ if _, err := s.Put(ctx, key, []byte("tampered"), etag); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := Restore(ctx, s, pfx, t.TempDir()); err == nil || !strings.Contains(err.Error(), "sha256") {
+ t.Fatalf("err = %v, want a sha256 mismatch", err)
+ }
+}
+
+func TestAnAbsentManifestRestoresNothing(t *testing.T) {
+ m, st, err := Restore(context.Background(), release.NewDirStore(t.TempDir()), pfx, t.TempDir())
+ if err != nil || m.Seq != 0 || st.Written != 0 {
+ t.Fatalf("m=%+v st=%+v err=%v", m, st, err)
+ }
+}
+
+func TestPlatformItemsAreNotWrittenIntoTheFilesDirectory(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ if err := p.StagePlatform(ctx, "wake.json", []byte(`{"v":1}`)); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+ dir := t.TempDir()
+ m, _, err := Restore(ctx, s, pfx, dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if entries, _ := os.ReadDir(dir); len(entries) != 0 {
+ t.Fatalf("files directory has %d entries, want 0", len(entries))
+ }
+ got, err := ReadPlatform(ctx, s, pfx, m, "wake.json")
+ if err != nil || string(got) != `{"v":1}` {
+ t.Fatalf("ReadPlatform = %q, %v", got, err)
+ }
+ if _, err := ReadPlatform(ctx, s, pfx, m, "absent"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("absent item: err = %v, want os.ErrNotExist", err)
+ }
+}
+
+// Review Focus 3: a skipped file stays in the manifest, visibly.
+func TestASkippedFileIsRecordedAndReportedOnRestore(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ if err := p.StageSkipped("huge.bin", "too-large"); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+ _, st, err := Restore(ctx, s, pfx, t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(st.Skipped) != 1 || st.Skipped[0] != "huge.bin" {
+ t.Fatalf("Skipped = %v", st.Skipped)
+ }
+}
+
+// A destination that already holds a symlink pointing elsewhere must not
+// let a write or a delete reach through it.
+func TestRestoreNeverWritesOrDeletesThroughASymlink(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ stage(t, p, "sub/x", "ours")
+ stage(t, p, "gone/y", "temp")
+ commit(t, p)
+ if err := p.StageDelete("gone/y"); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+
+ outside, dir := t.TempDir(), t.TempDir()
+ if err := os.WriteFile(filepath.Join(outside, "y"), []byte("not ours"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ for _, link := range []string{"sub", "gone"} {
+ if err := os.Symlink(outside, filepath.Join(dir, link)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if _, _, err := Restore(ctx, s, pfx, dir); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := os.Stat(filepath.Join(outside, "x")); !os.IsNotExist(err) {
+ t.Fatal("a file was written through the symlink")
+ }
+ if got := read(t, outside, "y"); got != "not ours" {
+ t.Fatalf("a file outside was changed to %q", got)
+ }
+ if fi, err := os.Lstat(filepath.Join(dir, "sub")); err != nil || !fi.IsDir() {
+ t.Fatalf("sub is not a real directory: %v %v", fi, err)
+ }
+ if got := read(t, dir, "sub/x"); got != "ours" {
+ t.Fatalf("sub/x = %q", got)
+ }
+}
+
+func TestRestoreHandlesADirectoryBecomingAFileAndBack(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ stage(t, p, "a/b", "child")
+ commit(t, p)
+ prior := t.TempDir() // a copy restored while "a" was a directory
+ if _, _, err := Restore(ctx, s, pfx, prior); err != nil {
+ t.Fatal(err)
+ }
+
+ // "a" becomes a file.
+ if err := p.StageDelete("a/b"); err != nil {
+ t.Fatal(err)
+ }
+ stage(t, p, "a", "now a file")
+ commit(t, p)
+ for name, dir := range map[string]string{"an empty destination": t.TempDir(), "a previously restored one": prior} {
+ if _, _, err := Restore(ctx, s, pfx, dir); err != nil {
+ t.Fatalf("%s: %v", name, err)
+ }
+ if got := read(t, dir, "a"); got != "now a file" {
+ t.Fatalf("%s: a = %q", name, got)
+ }
+ }
+
+ // And back: "a" becomes a directory again.
+ if err := p.StageDelete("a"); err != nil {
+ t.Fatal(err)
+ }
+ stage(t, p, "a/c", "child again")
+ commit(t, p)
+ for name, dir := range map[string]string{"an empty destination": t.TempDir(), "a previously restored one": prior} {
+ if _, _, err := Restore(ctx, s, pfx, dir); err != nil {
+ t.Fatalf("%s: %v", name, err)
+ }
+ if got := read(t, dir, "a/c"); got != "child again" {
+ t.Fatalf("%s: a/c = %q", name, got)
+ }
+ }
+}
+
+func TestRestoreIntoAMissingDirectoryIsAnError(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ stage(t, p, "x", "x")
+ commit(t, p)
+ if _, _, err := Restore(ctx, s, pfx, filepath.Join(t.TempDir(), "absent")); err == nil {
+ t.Fatal("restored into a directory that does not exist")
+ }
+}
+
+// Plan round 2, finding 3: a restore that returns nil has made its names
+// durable, and one that could not must say so.
+func TestRestoreSyncsEveryDirectoryItChangedBeforeReturning(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ stage(t, p, "top.txt", "t")
+ stage(t, p, "a/b/deep.txt", "d")
+ stage(t, p, "gone/old.txt", "o")
+ commit(t, p)
+ dir := t.TempDir()
+ if _, _, err := Restore(ctx, s, pfx, dir); err != nil {
+ t.Fatal(err)
+ }
+ if err := p.StageDelete("gone/old.txt"); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+
+ var synced []string
+ orig := syncDir
+ t.Cleanup(func() { syncDir = orig })
+ syncDir = func(root *os.Root, rel string) error {
+ // Everything must already be in place when the first sync runs.
+ if len(synced) == 0 {
+ if _, err := os.Stat(filepath.Join(dir, "gone", "old.txt")); !os.IsNotExist(err) {
+ t.Error("a directory was synced before the tombstone was applied")
+ }
+ }
+ synced = append(synced, rel)
+ return orig(root, rel)
+ }
+ if _, _, err := Restore(ctx, s, pfx, dir); err != nil {
+ t.Fatal(err)
+ }
+ want := []string{"a/b", "a", "gone", "."}
+ if !reflect.DeepEqual(synced, want) {
+ t.Fatalf("synced %v, want %v (deepest first, the root last)", synced, want)
+ }
+
+ syncDir = func(*os.Root, string) error { return errors.New("disk says no") }
+ if _, _, err := Restore(ctx, s, pfx, dir); err == nil || !strings.Contains(err.Error(), "disk says no") {
+ t.Fatalf("err = %v; a failed directory sync must fail the restore", err)
+ }
+}
+
+// Plan round 3: a restore that failed at its directory sync has already
+// removed the file. The retry finds nothing to remove, and must still
+// make that removal durable before it reports success.
+func TestARetriedRestoreStillSyncsADeletionAnEarlierAttemptMade(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ stage(t, p, "gone/old.txt", "o")
+ commit(t, p)
+ dir := t.TempDir()
+ if _, _, err := Restore(ctx, s, pfx, dir); err != nil {
+ t.Fatal(err)
+ }
+ if err := p.StageDelete("gone/old.txt"); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+
+ orig := syncDir
+ t.Cleanup(func() { syncDir = orig })
+ syncDir = func(*os.Root, string) error { return errors.New("disk says no") }
+ if _, _, err := Restore(ctx, s, pfx, dir); err == nil {
+ t.Fatal("the first attempt must fail")
+ }
+ if _, err := os.Stat(filepath.Join(dir, "gone", "old.txt")); !os.IsNotExist(err) {
+ t.Fatal("the first attempt should have removed the file before failing")
+ }
+ var synced []string
+ syncDir = func(root *os.Root, rel string) error { synced = append(synced, rel); return orig(root, rel) }
+ _, st, err := Restore(ctx, s, pfx, dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if st.Removed != 0 {
+ t.Fatalf("Removed = %d; the retry had nothing left to remove", st.Removed)
+ }
+ if want := []string{"gone", "."}; !reflect.DeepEqual(synced, want) {
+ t.Fatalf("the retry synced %v, want %v", synced, want)
+ }
+}
+
+// A deeper directory replaced by a file: the directories that vanish
+// under the new file are not synced (they do not exist), and the restore
+// still succeeds and syncs what survives.
+func TestRestoreHandlesADeepDirectoryBecomingAFile(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ p := open(t, s)
+ stage(t, p, "a/b/c", "leaf")
+ commit(t, p)
+ dir := t.TempDir()
+ if _, _, err := Restore(ctx, s, pfx, dir); err != nil {
+ t.Fatal(err)
+ }
+ if err := p.StageDelete("a/b/c"); err != nil {
+ t.Fatal(err)
+ }
+ stage(t, p, "a", "now a file")
+ commit(t, p)
+
+ var synced []string
+ orig := syncDir
+ t.Cleanup(func() { syncDir = orig })
+ syncDir = func(root *os.Root, rel string) error { synced = append(synced, rel); return orig(root, rel) }
+ for name, d := range map[string]string{"a previously restored destination": dir, "an empty one": t.TempDir()} {
+ synced = nil
+ if _, _, err := Restore(ctx, s, pfx, d); err != nil {
+ t.Fatalf("%s: %v", name, err)
+ }
+ if got := read(t, d, "a"); got != "now a file" {
+ t.Fatalf("%s: a = %q", name, got)
+ }
+ if want := []string{"."}; !reflect.DeepEqual(synced, want) {
+ t.Fatalf("%s: synced %v, want %v: a/b and a are gone, the root records it", name, synced, want)
+ }
+ }
+}
diff --git a/internal/hostfiles/manifest.go b/internal/hostfiles/manifest.go
new file mode 100644
index 000000000..80111a6ee
--- /dev/null
+++ b/internal/hostfiles/manifest.go
@@ -0,0 +1,91 @@
+// Package hostfiles publishes an instance's files directory, and the small
+// platform-managed items that travel with it, to object storage in an order
+// that survives late and duplicate uploads (region HA spec 1 §5.6).
+//
+// Nothing is overwritten: each change is a new immutable object, and one
+// manifest, replaced by compare-and-swap, says which object is current.
+package hostfiles
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "path"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// MaxFileBytes is the largest file published as one object. release.Store
+// takes a whole body in memory, so a larger file is recorded as skipped
+// rather than read. Plan 1b lifts this with a streaming put.
+const MaxFileBytes = 64 << 20
+
+type Entry struct {
+ Object string `json:"object,omitempty"`
+ SHA256 string `json:"sha256,omitempty"`
+ Size int64 `json:"size,omitempty"`
+ Mode uint32 `json:"mode,omitempty"`
+ Seq int64 `json:"seq"`
+ Deleted bool `json:"deleted,omitempty"`
+ Torn bool `json:"torn,omitempty"`
+ Skipped string `json:"skipped,omitempty"`
+}
+
+type Manifest struct {
+ V int `json:"v"`
+ Seq int64 `json:"seq"`
+ Entries map[string]Entry `json:"entries"`
+ Platform map[string]Entry `json:"platform"`
+}
+
+func ManifestKey(prefix string) string { return prefix + "manifest.json" }
+
+func objectKey(seq int64, sha string) string { return fmt.Sprintf("objects/%016d-%s", seq, sha) }
+
+// CleanPath accepts only a relative, already-clean, slash-separated path
+// that stays inside its root. It never repairs a path: a manifest is
+// written by this package, so anything else is damage or an attack.
+func CleanPath(p string) (string, error) {
+ if p == "" || strings.ContainsRune(p, 0) || strings.HasPrefix(p, "/") {
+ return "", fmt.Errorf("hostfiles: path %q is not a relative path", p)
+ }
+ c := path.Clean(p)
+ if c != p || c == "." || c == ".." || strings.HasPrefix(c, "../") {
+ return "", fmt.Errorf("hostfiles: path %q is not clean or leaves its root", p)
+ }
+ return c, nil
+}
+
+// Load reads the manifest. The bool is false when there is none yet.
+func Load(ctx context.Context, s release.Store, prefix string) (Manifest, bool, error) {
+ m, _, found, err := load(ctx, s, prefix)
+ return m, found, err
+}
+
+func load(ctx context.Context, s release.Store, prefix string) (Manifest, string, bool, error) {
+ empty := Manifest{V: 1, Entries: map[string]Entry{}, Platform: map[string]Entry{}}
+ raw, etag, err := s.Get(ctx, ManifestKey(prefix))
+ if errors.Is(err, os.ErrNotExist) {
+ return empty, "", false, nil
+ }
+ if err != nil {
+ return Manifest{}, "", false, err
+ }
+ var m Manifest
+ if err := json.Unmarshal(raw, &m); err != nil {
+ return Manifest{}, "", false, fmt.Errorf("hostfiles: manifest %s: %w", prefix, err)
+ }
+ if m.V != 1 {
+ return Manifest{}, "", false, fmt.Errorf("hostfiles: manifest %s is version %d; this binary reads 1", prefix, m.V)
+ }
+ if m.Entries == nil {
+ m.Entries = map[string]Entry{}
+ }
+ if m.Platform == nil {
+ m.Platform = map[string]Entry{}
+ }
+ return m, etag, true, nil
+}
diff --git a/internal/hostfiles/publisher.go b/internal/hostfiles/publisher.go
new file mode 100644
index 000000000..2d9880b89
--- /dev/null
+++ b/internal/hostfiles/publisher.go
@@ -0,0 +1,192 @@
+package hostfiles
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "sort"
+ "strings"
+ "sync"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// ErrSuperseded means another writer replaced the manifest. The caller
+// stops publishing: it is no longer this host's executor, or two
+// publishers were started for one host.
+var ErrSuperseded = errors.New("hostfiles: another writer published this manifest")
+
+// Publisher stages changes and commits them as one manifest revision. It
+// is safe for concurrent use.
+type Publisher struct {
+ s release.Store
+ prefix string
+
+ mu sync.Mutex
+ man Manifest
+ etag string
+ next int64 // the last sequence number handed out; man.Seq is the last committed
+ dirty bool
+}
+
+func Open(ctx context.Context, s release.Store, prefix string) (*Publisher, error) {
+ if !strings.HasSuffix(prefix, "/") {
+ return nil, fmt.Errorf("hostfiles: prefix %q must end in /", prefix)
+ }
+ m, etag, _, err := load(ctx, s, prefix)
+ if err != nil {
+ return nil, err
+ }
+ return &Publisher{s: s, prefix: prefix, man: m, etag: etag, next: m.Seq}, nil
+}
+
+func sum(body []byte) string {
+ h := sha256.Sum256(body)
+ return hex.EncodeToString(h[:])
+}
+
+// put writes body as the next immutable object and returns its entry.
+// Callers hold p.mu.
+func (p *Publisher) put(ctx context.Context, body []byte, mode uint32, torn bool) (Entry, error) {
+ sha := sum(body)
+ seq := p.next + 1
+ key := objectKey(seq, sha)
+ if _, err := p.s.Put(ctx, p.prefix+key, body, ""); err != nil && !errors.Is(err, release.ErrCASConflict) {
+ // A conflict means this exact key exists: the same sequence number
+ // and the same hash, left by an attempt that crashed before its
+ // commit. The bytes are the ones we were about to write.
+ return Entry{}, err
+ }
+ p.next = seq
+ return Entry{Object: key, SHA256: sha, Size: int64(len(body)), Mode: mode & 0o777, Seq: seq, Torn: torn}, nil
+}
+
+// Stage records path's content. Content identical to what is already
+// staged or committed takes no sequence number and writes nothing.
+func (p *Publisher) Stage(ctx context.Context, path string, body []byte, mode uint32, torn bool) error {
+ clean, err := CleanPath(path)
+ if err != nil {
+ return err
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if cur, ok := p.man.Entries[clean]; ok && !cur.Deleted && cur.Skipped == "" &&
+ cur.SHA256 == sum(body) && cur.Mode == mode&0o777 && cur.Torn == torn {
+ return nil
+ }
+ e, err := p.put(ctx, body, mode, torn)
+ if err != nil {
+ return err
+ }
+ p.man.Entries[clean], p.dirty = e, true
+ return nil
+}
+
+// StageSkipped records that path exists but was not published, and why.
+func (p *Publisher) StageSkipped(path, why string) error {
+ clean, err := CleanPath(path)
+ if err != nil {
+ return err
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if cur, ok := p.man.Entries[clean]; ok && cur.Skipped == why {
+ return nil
+ }
+ p.next++
+ p.man.Entries[clean], p.dirty = Entry{Seq: p.next, Skipped: why}, true
+ return nil
+}
+
+// StageDelete records that path is gone. Deleting a path the manifest never
+// had, or already has as deleted, is a no-op.
+func (p *Publisher) StageDelete(path string) error {
+ clean, err := CleanPath(path)
+ if err != nil {
+ return err
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if cur, ok := p.man.Entries[clean]; !ok || cur.Deleted {
+ return nil
+ }
+ p.next++
+ p.man.Entries[clean], p.dirty = Entry{Seq: p.next, Deleted: true}, true
+ return nil
+}
+
+// StagePlatform records a platform-managed item (wake state, an env file).
+// It shares the manifest's ordering but is never written into the files
+// directory by Restore.
+func (p *Publisher) StagePlatform(ctx context.Context, name string, body []byte) error {
+ if name == "" || strings.ContainsAny(name, "/\x00") {
+ return fmt.Errorf("hostfiles: platform item %q is not a plain name", name)
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if cur, ok := p.man.Platform[name]; ok && cur.SHA256 == sum(body) {
+ return nil
+ }
+ e, err := p.put(ctx, body, 0o600, false)
+ if err != nil {
+ return err
+ }
+ p.man.Platform[name], p.dirty = e, true
+ return nil
+}
+
+// Commit publishes everything staged as one manifest revision and returns
+// its sequence number. With nothing staged it returns the current one. A
+// failed commit leaves the staged changes in place, so the caller retries;
+// ErrSuperseded is final.
+func (p *Publisher) Commit(ctx context.Context) (int64, error) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if !p.dirty {
+ return p.man.Seq, nil
+ }
+ out := p.man
+ out.Seq = p.next
+ body, err := json.Marshal(out)
+ if err != nil {
+ return 0, fmt.Errorf("hostfiles: marshal manifest: %w", err)
+ }
+ etag, err := p.s.Put(ctx, ManifestKey(p.prefix), body, p.etag)
+ if errors.Is(err, release.ErrCASConflict) {
+ return 0, fmt.Errorf("%w: %s", ErrSuperseded, p.prefix)
+ }
+ if err != nil {
+ return 0, err
+ }
+ p.man.Seq, p.etag, p.dirty = out.Seq, etag, false
+ return out.Seq, nil
+}
+
+func (p *Publisher) Dirty() bool { p.mu.Lock(); defer p.mu.Unlock(); return p.dirty }
+
+func (p *Publisher) Seq() int64 { p.mu.Lock(); defer p.mu.Unlock(); return p.man.Seq }
+
+// Paths lists every path the manifest holds as present: published or
+// skipped, not deleted.
+func (p *Publisher) Paths() []string {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ var out []string
+ for path, e := range p.man.Entries {
+ if !e.Deleted {
+ out = append(out, path)
+ }
+ }
+ sort.Strings(out)
+ return out
+}
+
+// pending is the last sequence number handed out, committed or not.
+func (p *Publisher) pending() int64 {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.next
+}
diff --git a/internal/hostfiles/restore.go b/internal/hostfiles/restore.go
new file mode 100644
index 000000000..37f44fb41
--- /dev/null
+++ b/internal/hostfiles/restore.go
@@ -0,0 +1,295 @@
+package hostfiles
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "errors"
+ "fmt"
+ "io/fs"
+ "os"
+ "path"
+ "path/filepath"
+ "sort"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// syncDir makes a directory's entries durable. A test seam.
+var syncDir = func(root *os.Root, rel string) error {
+ d, err := root.Open(filepath.FromSlash(rel))
+ if err != nil {
+ return err
+ }
+ defer d.Close()
+ return d.Sync()
+}
+
+type RestoreStats struct {
+ Written, Removed int
+ Skipped []string // paths the manifest holds as present but unpublished
+}
+
+// Restore makes dir match the manifest under prefix: it removes every
+// tombstoned entry, writes every published one, and reports the skipped
+// ones. It reads nothing but the manifest and the objects the manifest
+// names, and verifies each object against the hash the manifest records.
+//
+// Every path is validated before anything is touched, so a manifest with
+// one bad path changes nothing. Every file operation goes through an
+// os.Root opened on dir, so nothing is written or removed outside it even
+// if dir already holds a symlink pointing elsewhere.
+//
+// When Restore returns nil, what it did is durable: each file was synced
+// before its rename, and every directory in which a name was created,
+// replaced or removed has been synced since. A caller may record the
+// installation as complete on that basis, and on no weaker one.
+func Restore(ctx context.Context, s release.Store, prefix, dir string) (Manifest, RestoreStats, error) {
+ var st RestoreStats
+ m, _, _, err := load(ctx, s, prefix)
+ if err != nil {
+ return Manifest{}, st, err
+ }
+ var live, dead []string
+ for p, e := range m.Entries {
+ if _, err := CleanPath(p); err != nil {
+ return Manifest{}, st, err
+ }
+ if e.Deleted {
+ dead = append(dead, p)
+ } else {
+ live = append(live, p)
+ }
+ }
+ if len(live)+len(dead) == 0 {
+ return m, st, nil
+ }
+ root, err := os.OpenRoot(dir)
+ if err != nil {
+ return Manifest{}, st, fmt.Errorf("hostfiles: restore into %s: %w", dir, err)
+ }
+ defer root.Close()
+
+ // Tombstones first, deepest first, so a directory that became a file
+ // is emptied before the file is written over it.
+ // Every directory on the way to a path the manifest speaks of, live or
+ // tombstoned, is a directory whose entries this restore may have
+ // changed: now, or in an earlier attempt that removed or created the
+ // name and then failed before its sync. So the set does not depend on
+ // what this call happened to find already done.
+ touched := map[string]bool{".": true}
+ mark := func(p string) {
+ for d := path.Dir(p); d != "."; d = path.Dir(d) {
+ touched[d] = true
+ }
+ }
+ sort.Slice(dead, func(i, j int) bool {
+ di, dj := strings.Count(dead[i], "/"), strings.Count(dead[j], "/")
+ if di != dj {
+ return di > dj
+ }
+ return dead[i] < dead[j]
+ })
+ for _, p := range dead {
+ mark(p)
+ removed, err := removeInside(root, p)
+ if err != nil {
+ return Manifest{}, st, fmt.Errorf("hostfiles: remove %s: %w", p, err)
+ }
+ if removed {
+ st.Removed++
+ }
+ }
+ sort.Strings(live)
+ for _, p := range live {
+ e := m.Entries[p]
+ if e.Skipped != "" {
+ st.Skipped = append(st.Skipped, p)
+ continue
+ }
+ mark(p)
+ body, err := fetch(ctx, s, prefix, e)
+ if err != nil {
+ return Manifest{}, st, fmt.Errorf("hostfiles: %s: %w", p, err)
+ }
+ if err := writeInside(root, p, body, os.FileMode(e.Mode)); err != nil {
+ return Manifest{}, st, fmt.Errorf("hostfiles: write %s: %w", p, err)
+ }
+ st.Written++
+ }
+ // Sync the directories that survive, deepest first and the root last,
+ // so a directory is durable before the entry that names it is. One
+ // that no longer exists as a real directory (it was emptied and then
+ // replaced by a file, or it sat under a path that became a file) has
+ // nothing to sync: the directory above it records its going.
+ dirs := make([]string, 0, len(touched))
+ for d := range touched {
+ dirs = append(dirs, d)
+ }
+ sort.Slice(dirs, func(i, j int) bool {
+ if (dirs[i] == ".") != (dirs[j] == ".") {
+ return dirs[j] == "."
+ }
+ di, dj := strings.Count(dirs[i], "/"), strings.Count(dirs[j], "/")
+ if di != dj {
+ return di > dj
+ }
+ return dirs[i] < dirs[j]
+ })
+ for _, d := range dirs {
+ if d != "." {
+ if ok, err := realParents(root, d+"/x"); err != nil {
+ return Manifest{}, st, fmt.Errorf("hostfiles: sync directory %s: %w", d, err)
+ } else if !ok {
+ continue
+ }
+ }
+ if err := syncDir(root, d); err != nil {
+ return Manifest{}, st, fmt.Errorf("hostfiles: sync directory %s: %w", d, err)
+ }
+ }
+ return m, st, nil
+}
+
+// ReadPlatform returns a platform-managed item's bytes, verified. A name
+// the manifest does not hold is os.ErrNotExist.
+func ReadPlatform(ctx context.Context, s release.Store, prefix string, m Manifest, name string) ([]byte, error) {
+ e, ok := m.Platform[name]
+ if !ok {
+ return nil, fmt.Errorf("hostfiles: platform item %q: %w", name, os.ErrNotExist)
+ }
+ return fetch(ctx, s, prefix, e)
+}
+
+func fetch(ctx context.Context, s release.Store, prefix string, e Entry) ([]byte, error) {
+ body, _, err := s.Get(ctx, prefix+e.Object)
+ if err != nil {
+ return nil, err
+ }
+ if got := sum(body); got != e.SHA256 {
+ return nil, fmt.Errorf("object %s has sha256 %s, the manifest says %s", e.Object, got, e.SHA256)
+ }
+ return body, nil
+}
+
+// realParents reports whether every ancestor of rel inside root is a real
+// directory: not a symlink, not a file, not missing.
+func realParents(root *os.Root, rel string) (bool, error) {
+ parts := strings.Split(rel, "/")
+ cur := ""
+ for _, part := range parts[:len(parts)-1] {
+ cur = path.Join(cur, part)
+ fi, err := root.Lstat(filepath.FromSlash(cur))
+ if errors.Is(err, fs.ErrNotExist) {
+ return false, nil
+ }
+ if err != nil {
+ return false, err
+ }
+ if !fi.IsDir() {
+ return false, nil
+ }
+ }
+ return true, nil
+}
+
+// removeInside removes rel if it is a file of ours. A path whose ancestor
+// is a symlink, a file or missing is not ours to remove and is left alone:
+// that is how a tombstone for "a/b" behaves once "a" has become a file,
+// and why a symlinked ancestor never lets a delete reach outside root.
+func removeInside(root *os.Root, rel string) (bool, error) {
+ ok, err := realParents(root, rel)
+ if err != nil || !ok {
+ return false, err
+ }
+ name := filepath.FromSlash(rel)
+ fi, err := root.Lstat(name)
+ if errors.Is(err, fs.ErrNotExist) {
+ return false, nil
+ }
+ if err != nil {
+ return false, err
+ }
+ if fi.IsDir() {
+ // The path is a directory now: a later entry lives under it, and
+ // the tombstone is for a file that no longer exists.
+ return false, nil
+ }
+ if err := root.Remove(name); err != nil && !errors.Is(err, fs.ErrNotExist) {
+ return false, err
+ }
+ return true, nil
+}
+
+// writeInside writes rel atomically inside root. Anything in the way that
+// is not a real directory (a file where a directory must be, a symlink)
+// is removed first, and a directory where the file must be is removed
+// with its contents: the manifest is the truth about this tree.
+func writeInside(root *os.Root, rel string, body []byte, mode os.FileMode) error {
+ if mode == 0 {
+ mode = 0o600
+ }
+ parent := path.Dir(rel)
+ if parent != "." {
+ cur := ""
+ for _, part := range strings.Split(parent, "/") {
+ cur = path.Join(cur, part)
+ fi, err := root.Lstat(filepath.FromSlash(cur))
+ if errors.Is(err, fs.ErrNotExist) {
+ break
+ }
+ if err != nil {
+ return err
+ }
+ if !fi.IsDir() {
+ if err := root.Remove(filepath.FromSlash(cur)); err != nil {
+ return err
+ }
+ break
+ }
+ }
+ if err := root.MkdirAll(filepath.FromSlash(parent), 0o755); err != nil {
+ return err
+ }
+ }
+ name := filepath.FromSlash(rel)
+ if fi, err := root.Lstat(name); err == nil && fi.IsDir() {
+ if err := root.RemoveAll(name); err != nil {
+ return err
+ }
+ }
+ var rnd [8]byte
+ if _, err := rand.Read(rnd[:]); err != nil {
+ return err
+ }
+ tmp := filepath.FromSlash(path.Join(parent, ".carlos-restore-"+hex.EncodeToString(rnd[:])))
+ f, err := root.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
+ if err != nil {
+ return err
+ }
+ done := false
+ defer func() {
+ if !done {
+ f.Close()
+ root.Remove(tmp)
+ }
+ }()
+ if _, err := f.Write(body); err != nil {
+ return err
+ }
+ if err := f.Chmod(mode); err != nil {
+ return err
+ }
+ if err := f.Sync(); err != nil {
+ return err
+ }
+ if err := f.Close(); err != nil {
+ return err
+ }
+ if err := root.Rename(tmp, name); err != nil {
+ return err
+ }
+ done = true
+ return nil
+}