ed8ebfe refactor(snapshot): unify atomic file preparation · vev

M internal/adapters/snapshot/atomic.go +31 -65


@@ -110,56 +110,20 @@ func (r *Repository) writeImmutable(path string, data []byte, verifier func([]by
110	110		if err := r.ensurePrivateDirectoryPhase(dir, phase); err != nil {
111	111		return err
112	112		}
113		-	temp, err := r.createTemp(dir)
114		-	if err != nil {
115		-	return err
116		-	}
117		-	tempPath := temp.Name()
118		-	closed := false
119		-	cleanup := func(cause error) error {
120		-	if !closed {
121		-	if err := r.closeFile(temp); err != nil {
122		-	cause = errors.Join(cause, err)
	113	+	return r.withAtomicTemp(dir, data, func(tempPath string) (bool, error) {
	114	+	if err := r.installImmutable(tempPath, path); err != nil {
	115	+	if !errors.Is(err, os.ErrExist) {
	116	+	return true, err
123	117		}
124		-	closed = true
125		-	}
126		-	if err := r.remove(tempPath); err != nil && !errors.Is(err, os.ErrNotExist) {
127		-	cause = errors.Join(cause, err)
128		-	} else if err == nil {
129		-	cause = errors.Join(cause, r.syncDirectory(dir))
130		-	}
131		-	return cause
132		-	}
133		-	if err := temp.Chmod(0o600); err != nil {
134		-	return cleanup(err)
135		-	}
136		-	if err := r.writeFile(temp, data); err != nil {
137		-	return cleanup(err)
138		-	}
139		-	if err := r.syncFile(temp); err != nil {
140		-	return cleanup(err)
141		-	}
142		-	if err := r.closeFile(temp); err != nil {
143		-	closed = true
144		-	return cleanup(err)
145		-	}
146		-	closed = true
147		-	if err := r.installImmutable(tempPath, path); err != nil {
148		-	if errors.Is(err, os.ErrExist) {
149	118		existing, readErr := r.readBounded(path)
150	119		if readErr != nil {
151		-	return cleanup(readErr)
152		-	}
153		-	if verifyErr := verifier(existing); verifyErr != nil {
154		-	return cleanup(verifyErr)
	120	+	return true, readErr
155	121		}
156		-	return cleanup(nil)
	122	+	return true, verifier(existing)
157	123		}
158		-	return cleanup(err)
159		-	}
160		-	// Removing the linked temporary entry before the directory sync lets one
161		-	// sync persist both the immutable install and its consumed temporary file.
162		-	return cleanup(nil)
	124	+	// link retains the temporary name, so cleanup must remove and sync it.
	125	+	return true, nil
	126	+	})
163	127		}
164	128
165	129		// writeMutable is used only for HEAD, the sole authoritative pointer allowed

@@ -169,6 +133,24 @@ func (r *Repository) writeMutable(path string, data []byte) error {
169	133		if err := r.ensurePrivateDirectory(dir); err != nil {
170	134		return err
171	135		}
	136	+	return r.withAtomicTemp(dir, data, func(tempPath string) (bool, error) {
	137	+	if err := r.rename(tempPath, path); err != nil {
	138	+	return true, err
	139	+	}
	140	+	// rename consumed tempPath. Its directory sync persists both rename and
	141	+	// the absence of the temporary entry.
	142	+	if err := r.syncDirectory(dir); err != nil {
	143	+	return true, err
	144	+	}
	145	+	return false, nil
	146	+	})
	147	+	}
	148	+
	149	+	// withAtomicTemp prepares a private, durable temporary file, invokes publish,
	150	+	// then closes and removes the temporary entry when publish reports that its
	151	+	// operation did not consume it. Cleanup errors are joined after the operation
	152	+	// error in the same order as the former immutable and mutable write paths.
	153	+	func (r *Repository) withAtomicTemp(dir string, data []byte, publish func(string) (removeTemp bool, err error)) error {
172	154		temp, err := r.createTemp(dir)
173	155		if err != nil {
174	156		return err

@@ -203,15 +185,11 @@ func (r *Repository) writeMutable(path string, data []byte) error {
203	185		return cleanup(err)
204	186		}
205	187		closed = true
206		-	if err := r.rename(tempPath, path); err != nil {
207		-	return cleanup(err)
208		-	}
209		-	if err := r.syncDirectory(dir); err != nil {
210		-	return cleanup(err)
	188	+	removeTemp, err := publish(tempPath)
	189	+	if !removeTemp {
	190	+	return err
211	191		}
212		-	// rename consumed tempPath. Its directory sync above persists both rename
213		-	// and the absence of the temporary entry.
214		-	return nil
	192	+	return cleanup(err)
215	193		}
216	194
217	195		func (r Repository) createTemp(dir string) (os.File, error) {

@@ -412,15 +390,3 @@ func readBoundedFile(f *os.File) ([]byte, error) {
412	390		}
413	391		return data, nil
414	392		}
415		-
416		-	func equalBytes(a, b []byte) bool {
417		-	if len(a) != len(b) {
418		-	return false
419		-	}
420		-	for i := range a {
421		-	if a[i] != b[i] {
422		-	return false
423		-	}
424		-	}
425		-	return true
426		-	}

 	if err := r.ensurePrivateDirectoryPhase(dir, phase); err != nil {
 		return err
 	}
-	temp, err := r.createTemp(dir)
+	return r.withAtomicTemp(dir, data, func(tempPath string) (bool, error) {
-	if err != nil {
+		if err := r.installImmutable(tempPath, path); err != nil {
-		return err
+			if !errors.Is(err, os.ErrExist) {
-	}
+				return true, err
-	tempPath := temp.Name()
-	closed := false
-	cleanup := func(cause error) error {
-		if !closed {
-			if err := r.closeFile(temp); err != nil {
-				cause = errors.Join(cause, err)
 			}
-			closed = true
-		}
-		if err := r.remove(tempPath); err != nil && !errors.Is(err, os.ErrNotExist) {
-			cause = errors.Join(cause, err)
-		} else if err == nil {
-			cause = errors.Join(cause, r.syncDirectory(dir))
-		}
-		return cause
-	}
-	if err := temp.Chmod(0o600); err != nil {
-		return cleanup(err)
-	}
-	if err := r.writeFile(temp, data); err != nil {
-		return cleanup(err)
-	}
-	if err := r.syncFile(temp); err != nil {
-		return cleanup(err)
-	}
-	if err := r.closeFile(temp); err != nil {
-		closed = true
-		return cleanup(err)
-	}
-	closed = true
-	if err := r.installImmutable(tempPath, path); err != nil {
-		if errors.Is(err, os.ErrExist) {
 			existing, readErr := r.readBounded(path)
 			if readErr != nil {
-				return cleanup(readErr)
+				return true, readErr
-			}
-			if verifyErr := verifier(existing); verifyErr != nil {
-				return cleanup(verifyErr)
 			}
-			return cleanup(nil)
+			return true, verifier(existing)
 		}
-		return cleanup(err)
+		// link retains the temporary name, so cleanup must remove and sync it.
-	}
+		return true, nil
-	// Removing the linked temporary entry before the directory sync lets one
+	})
-	// sync persist both the immutable install and its consumed temporary file.
-	return cleanup(nil)
 }
 // writeMutable is used only for HEAD, the sole authoritative pointer allowed
 	if err := r.ensurePrivateDirectory(dir); err != nil {
 		return err
 	}
+	return r.withAtomicTemp(dir, data, func(tempPath string) (bool, error) {
+		if err := r.rename(tempPath, path); err != nil {
+			return true, err
+		}
+		// rename consumed tempPath. Its directory sync persists both rename and
+		// the absence of the temporary entry.
+		if err := r.syncDirectory(dir); err != nil {
+			return true, err
+		}
+		return false, nil
+	})
+}
+// withAtomicTemp prepares a private, durable temporary file, invokes publish,
+// then closes and removes the temporary entry when publish reports that its
+// operation did not consume it. Cleanup errors are joined after the operation
+// error in the same order as the former immutable and mutable write paths.
+func (r *Repository) withAtomicTemp(dir string, data []byte, publish func(string) (removeTemp bool, err error)) error {
 	temp, err := r.createTemp(dir)
 	if err != nil {
 		return err
 		return cleanup(err)
 	}
 	closed = true
-	if err := r.rename(tempPath, path); err != nil {
+	removeTemp, err := publish(tempPath)
-		return cleanup(err)
+	if !removeTemp {
-	}
+		return err
-	if err := r.syncDirectory(dir); err != nil {
-		return cleanup(err)
 	}
-	// rename consumed tempPath. Its directory sync above persists both rename
+	return cleanup(err)
-	// and the absence of the temporary entry.
-	return nil
 }
 func (r *Repository) createTemp(dir string) (*os.File, error) {
 	}
 	return data, nil
 }
-func equalBytes(a, b []byte) bool {
-	if len(a) != len(b) {
-		return false
-	}
-	for i := range a {
-		if a[i] != b[i] {
-			return false
-		}
-	}
-	return true
-}

M internal/adapters/snapshot/atomic_test.go +59 -1

@@ -1,7 +1,9 @@
1	1		package snapshot
2	2
3	3		import (
	4	+	"bytes"
4	5		"errors"
	6	+	"os"
5	7		"path/filepath"
6	8		"testing"
7	9		)

@@ -22,7 +24,7 @@ func TestWriteImmutableUsesOneFinalDirectorySync(t *testing.T) {
22	24		}
23	25		data := []byte("immutable")
24	26		if err := repo.writeImmutable(filepath.Join(dir, "object"), data, func(existing []byte) error {
25		-	if !equalBytes(existing, data) {
	27	+	if !bytes.Equal(existing, data) {
26	28		return errors.New("unexpected immutable data")
27	29		}
28	30		return nil

@@ -33,3 +35,59 @@ func TestWriteImmutableUsesOneFinalDirectorySync(t *testing.T) {
33	35		t.Fatalf("successful immutable directory syncs = %d, want 1", syncs)
34	36		}
35	37		}
	38	+
	39	+	func TestWithAtomicTempPreservesLifecycleOrder(t *testing.T) {
	40	+	repo := NewRepository(privateDir(t))
	41	+	dir := filepath.Join(repo.dir, "temporary")
	42	+	if err := repo.ensurePrivateDirectory(dir); err != nil {
	43	+	t.Fatal(err)
	44	+	}
	45	+
	46	+	var events []string
	47	+	repo.hooks.createTemp = func(string) error { events = append(events, "create"); return nil }
	48	+	repo.hooks.writeTemp = func(string) error { events = append(events, "write"); return nil }
	49	+	repo.hooks.syncFile = func(string) error { events = append(events, "sync file"); return nil }
	50	+	repo.hooks.closeFile = func(string) error { events = append(events, "close"); return nil }
	51	+	repo.hooks.remove = func(string) error { events = append(events, "remove"); return nil }
	52	+	repo.hooks.syncDirectory = func(string) error { events = append(events, "sync directory"); return nil }
	53	+
	54	+	if err := repo.withAtomicTemp(dir, []byte("data"), func(path string) (bool, error) {
	55	+	events = append(events, "publish")
	56	+	info, err := os.Stat(path)
	57	+	if err != nil {
	58	+	return true, err
	59	+	}
	60	+	if got := info.Mode().Perm(); got != 0o600 {
	61	+	t.Fatalf("temporary file mode = %o, want 600", got)
	62	+	}
	63	+	return true, nil
	64	+	}); err != nil {
	65	+	t.Fatal(err)
	66	+	}
	67	+	want := []string{"create", "write", "sync file", "close", "publish", "remove", "sync directory"}
	68	+	if len(events) != len(want) {
	69	+	t.Fatalf("lifecycle = %v, want %v", events, want)
	70	+	}
	71	+	for i := range want {
	72	+	if events[i] != want[i] {
	73	+	t.Fatalf("lifecycle = %v, want %v", events, want)
	74	+	}
	75	+	}
	76	+	}
	77	+
	78	+	func TestWriteMutableDoesNotRemoveConsumedTemporaryFile(t *testing.T) {
	79	+	repo := NewRepository(privateDir(t))
	80	+	path := filepath.Join(repo.dir, "HEAD")
	81	+	repo.hooks.remove = func(string) error { return errors.New("remove must not run") }
	82	+
	83	+	if err := repo.writeMutable(path, []byte("head")); err != nil {
	84	+	t.Fatalf("writeMutable error = %v", err)
	85	+	}
	86	+	got, err := os.ReadFile(path)
	87	+	if err != nil {
	88	+	t.Fatal(err)
	89	+	}
	90	+	if !bytes.Equal(got, []byte("head")) {
	91	+	t.Fatalf("HEAD = %q, want %q", got, "head")
	92	+	}
	93	+	}

@@ -1,7 +1,9 @@
1		package snapshot	1		package snapshot
2			2
3		import (	3		import (
			4	+	"bytes"
4		"errors"	5		"errors"
			6	+	"os"
5		"path/filepath"	7		"path/filepath"
6		"testing"	8		"testing"
7		)	9		)

@@ -22,7 +24,7 @@ func TestWriteImmutableUsesOneFinalDirectorySync(t *testing.T) {
22		}	24		}
23		data := []byte("immutable")	25		data := []byte("immutable")
24		if err := repo.writeImmutable(filepath.Join(dir, "object"), data, func(existing []byte) error {	26		if err := repo.writeImmutable(filepath.Join(dir, "object"), data, func(existing []byte) error {
25	-	if !equalBytes(existing, data) {	27	+	if !bytes.Equal(existing, data) {
26		return errors.New("unexpected immutable data")	28		return errors.New("unexpected immutable data")
27		}	29		}
28		return nil	30		return nil

@@ -33,3 +35,59 @@ func TestWriteImmutableUsesOneFinalDirectorySync(t *testing.T) {
33		t.Fatalf("successful immutable directory syncs = %d, want 1", syncs)	35		t.Fatalf("successful immutable directory syncs = %d, want 1", syncs)
34		}	36		}
35		}	37		}
			38	+
			39	+	func TestWithAtomicTempPreservesLifecycleOrder(t *testing.T) {
			40	+	repo := NewRepository(privateDir(t))
			41	+	dir := filepath.Join(repo.dir, "temporary")
			42	+	if err := repo.ensurePrivateDirectory(dir); err != nil {
			43	+	t.Fatal(err)
			44	+	}
			45	+
			46	+	var events []string
			47	+	repo.hooks.createTemp = func(string) error { events = append(events, "create"); return nil }
			48	+	repo.hooks.writeTemp = func(string) error { events = append(events, "write"); return nil }
			49	+	repo.hooks.syncFile = func(string) error { events = append(events, "sync file"); return nil }
			50	+	repo.hooks.closeFile = func(string) error { events = append(events, "close"); return nil }
			51	+	repo.hooks.remove = func(string) error { events = append(events, "remove"); return nil }
			52	+	repo.hooks.syncDirectory = func(string) error { events = append(events, "sync directory"); return nil }
			53	+
			54	+	if err := repo.withAtomicTemp(dir, []byte("data"), func(path string) (bool, error) {
			55	+	events = append(events, "publish")
			56	+	info, err := os.Stat(path)
			57	+	if err != nil {
			58	+	return true, err
			59	+	}
			60	+	if got := info.Mode().Perm(); got != 0o600 {
			61	+	t.Fatalf("temporary file mode = %o, want 600", got)
			62	+	}
			63	+	return true, nil
			64	+	}); err != nil {
			65	+	t.Fatal(err)
			66	+	}
			67	+	want := []string{"create", "write", "sync file", "close", "publish", "remove", "sync directory"}
			68	+	if len(events) != len(want) {
			69	+	t.Fatalf("lifecycle = %v, want %v", events, want)
			70	+	}
			71	+	for i := range want {
			72	+	if events[i] != want[i] {
			73	+	t.Fatalf("lifecycle = %v, want %v", events, want)
			74	+	}
			75	+	}
			76	+	}
			77	+
			78	+	func TestWriteMutableDoesNotRemoveConsumedTemporaryFile(t *testing.T) {
			79	+	repo := NewRepository(privateDir(t))
			80	+	path := filepath.Join(repo.dir, "HEAD")
			81	+	repo.hooks.remove = func(string) error { return errors.New("remove must not run") }
			82	+
			83	+	if err := repo.writeMutable(path, []byte("head")); err != nil {
			84	+	t.Fatalf("writeMutable error = %v", err)
			85	+	}
			86	+	got, err := os.ReadFile(path)
			87	+	if err != nil {
			88	+	t.Fatal(err)
			89	+	}
			90	+	if !bytes.Equal(got, []byte("head")) {
			91	+	t.Fatalf("HEAD = %q, want %q", got, "head")
			92	+	}
			93	+	}

M internal/adapters/snapshot/publish.go +6 -5

@@ -1,6 +1,7 @@
1	1		package snapshot
2	2
3	3		import (
	4	+	"bytes"
4	5		"context"
5	6		"crypto/sha256"
6	7		"fmt"

@@ -37,7 +38,7 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
37	38		// map. Checking the immutable bytes directly also keeps its descriptor
38	39		// reads beneath the pinned repository root.
39	40		if generation, _, err := r.readHead(key); err == nil && generation == publication.Generation {
40		-	if current, err := r.readBounded(r.manifestPath(key, generation)); err == nil && equalBytes(current, publication.Manifest) {
	41	+	if current, err := r.readBounded(r.manifestPath(key, generation)); err == nil && bytes.Equal(current, publication.Manifest) {
41	42		return nil
42	43		}
43	44		}

@@ -50,7 +51,7 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
50	51		return fmt.Errorf("snapshot generation %d, current %d: immutable conflict", publication.Generation, current)
51	52		}
52	53		if publication.Generation == current {
53		-	if !equalBytes(currentManifest, publication.Manifest) {
	54	+	if !bytes.Equal(currentManifest, publication.Manifest) {
54	55		return fmt.Errorf("snapshot generation %d: immutable conflict", publication.Generation)
55	56		}
56	57		return nil

@@ -111,7 +112,7 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
111	112		return err
112	113		}
113	114		if exists {
114		-	if !equalBytes(existing, publication.Manifest) {
	115	+	if !bytes.Equal(existing, publication.Manifest) {
115	116		return fmt.Errorf("manifest generation %d: immutable conflict", publication.Generation)
116	117		}
117	118		} else {

@@ -124,7 +125,7 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
124	125		return err
125	126		}
126	127		if err := r.writeImmutable(manifestPath, publication.Manifest, func(existing []byte) error {
127		-	if !equalBytes(existing, publication.Manifest) {
	128	+	if !bytes.Equal(existing, publication.Manifest) {
128	129		return fmt.Errorf("manifest generation %d: immutable conflict", publication.Generation)
129	130		}
130	131		return nil

@@ -251,7 +252,7 @@ func validateSuppliedObject(objects []ports.SnapshotObject, digest ports.Snapsho
251	252		if actual != candidate.Digest \|\| candidate.Digest != digest {
252	253		return ports.SnapshotObject{}, fmt.Errorf("object digest mismatch")
253	254		}
254		-	if !equalBytes(candidate.Data, object.Data) {
	255	+	if !bytes.Equal(candidate.Data, object.Data) {
255	256		return ports.SnapshotObject{}, fmt.Errorf("conflicting object")
256	257		}
257	258		}

 package snapshot
 import (
+	"bytes"
 	"context"
 	"crypto/sha256"
 	"fmt"
 	// map. Checking the immutable bytes directly also keeps its descriptor
 	// reads beneath the pinned repository root.
 	if generation, _, err := r.readHead(key); err == nil && generation == publication.Generation {
-		if current, err := r.readBounded(r.manifestPath(key, generation)); err == nil && equalBytes(current, publication.Manifest) {
+		if current, err := r.readBounded(r.manifestPath(key, generation)); err == nil && bytes.Equal(current, publication.Manifest) {
 			return nil
 		}
 	}
 		return fmt.Errorf("snapshot generation %d, current %d: immutable conflict", publication.Generation, current)
 	}
 	if publication.Generation == current {
-		if !equalBytes(currentManifest, publication.Manifest) {
+		if !bytes.Equal(currentManifest, publication.Manifest) {
 			return fmt.Errorf("snapshot generation %d: immutable conflict", publication.Generation)
 		}
 		return nil
 		return err
 	}
 	if exists {
-		if !equalBytes(existing, publication.Manifest) {
+		if !bytes.Equal(existing, publication.Manifest) {
 			return fmt.Errorf("manifest generation %d: immutable conflict", publication.Generation)
 		}
 	} else {
 			return err
 		}
 		if err := r.writeImmutable(manifestPath, publication.Manifest, func(existing []byte) error {
-			if !equalBytes(existing, publication.Manifest) {
+			if !bytes.Equal(existing, publication.Manifest) {
 				return fmt.Errorf("manifest generation %d: immutable conflict", publication.Generation)
 			}
 			return nil
 		if actual != candidate.Digest || candidate.Digest != digest {
 			return ports.SnapshotObject{}, fmt.Errorf("object digest mismatch")
 		}
-		if !equalBytes(candidate.Data, object.Data) {
+		if !bytes.Equal(candidate.Data, object.Data) {
 			return ports.SnapshotObject{}, fmt.Errorf("conflicting object")
 		}
 	}

M internal/adapters/snapshot/repository_durability_test.go +3 -2

@@ -1,6 +1,7 @@
1	1		package snapshot
2	2
3	3		import (
	4	+	"bytes"
4	5		"context"
5	6		"errors"
6	7		"os"

@@ -39,7 +40,7 @@ func prepareHeadStage(t testing.T, repo Repository, pub ports.SnapshotPublicat
39	40		key := sessionKey(pub.Name)
40	41		prepareObjects(t, repo, pub)
41	42		if err := repo.writeImmutable(repo.manifestPath(key, pub.Generation), pub.Manifest, func(data []byte) error {
42		-	if !equalBytes(data, pub.Manifest) {
	43	+	if !bytes.Equal(data, pub.Manifest) {
43	44		return errors.New("invalid manifest")
44	45		}
45	46		return nil

@@ -171,7 +172,7 @@ func prepareObjects(t testing.T, repo Repository, pub ports.SnapshotPublicatio
171	172		for _, object := range pub.Objects {
172	173		ref := manifestReference(t, pub.Manifest, object.Digest)
173	174		if err := repo.writeImmutable(repo.objectPath(key, object.Digest), object.Data, func(data []byte) error {
174		-	if !equalBytes(data, object.Data) \|\| !validObject(data, ref) {
	175	+	if !bytes.Equal(data, object.Data) \|\| !validObject(data, ref) {
175	176		return errors.New("invalid object")
176	177		}
177	178		return nil

 package snapshot
 import (
+	"bytes"
 	"context"
 	"errors"
 	"os"
 	key := sessionKey(pub.Name)
 	prepareObjects(t, repo, pub)
 	if err := repo.writeImmutable(repo.manifestPath(key, pub.Generation), pub.Manifest, func(data []byte) error {
-		if !equalBytes(data, pub.Manifest) {
+		if !bytes.Equal(data, pub.Manifest) {
 			return errors.New("invalid manifest")
 		}
 		return nil
 	for _, object := range pub.Objects {
 		ref := manifestReference(t, pub.Manifest, object.Digest)
 		if err := repo.writeImmutable(repo.objectPath(key, object.Digest), object.Data, func(data []byte) error {
-			if !equalBytes(data, object.Data) || !validObject(data, ref) {
+			if !bytes.Equal(data, object.Data) || !validObject(data, ref) {
 				return errors.New("invalid object")
 			}
 			return nil