test(snapshot): cover repository durability faults

commit: a8652f510fd3f474b0ee4bff41f661ca0a0ab2c6
author: brice · 3mo ago · 2026-07-21 20:15 UTC
parent: 2d19589
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +172 -14

M internal/adapters/snapshot/repository.go +27 -14


@@ -449,14 +449,17 @@ func verifyObjectFile(path string, digest ports.SnapshotDigest, ref codec.Object
449	449		}
450	450
451	451		func (r *Repository) ensureSession(key string) error {
452		-	for _, dir := range []string{
453		-	r.dir,
454		-	filepath.Join(r.dir, repositorySessionsDir),
455		-	r.sessionPath(key),
456		-	filepath.Join(r.sessionPath(key), repositoryObjectsDir),
457		-	filepath.Join(r.sessionPath(key), repositoryGenerations),
	452	+	for _, directory := range []struct {
	453	+	path string
	454	+	phase string
	455	+	}{
	456	+	{r.dir, "repository"},
	457	+	{filepath.Join(r.dir, repositorySessionsDir), "sessions"},
	458	+	{r.sessionPath(key), "session"},
	459	+	{filepath.Join(r.sessionPath(key), repositoryObjectsDir), "objects"},
	460	+	{filepath.Join(r.sessionPath(key), repositoryGenerations), "generations"},
458	461		} {
459		-	if err := r.ensurePrivateDirectory(dir); err != nil {
	462	+	if err := r.ensurePrivateDirectoryPhase(directory.path, directory.phase); err != nil {
460	463		return fmt.Errorf("create snapshot repository directory: %w", err)
461	464		}
462	465		}

@@ -464,6 +467,10 @@ func (r *Repository) ensureSession(key string) error {
464	467		}
465	468
466	469		func (r *Repository) ensurePrivateDirectory(dir string) error {
	470	+	return r.ensurePrivateDirectoryPhase(dir, "snapshot directory")
	471	+	}
	472	+
	473	+	func (r *Repository) ensurePrivateDirectoryPhase(dir, phase string) error {
467	474		_, err := os.Lstat(dir)
468	475		created := errors.Is(err, os.ErrNotExist)
469	476		if err != nil && !created {

@@ -472,7 +479,7 @@ func (r *Repository) ensurePrivateDirectory(dir string) error {
472	479		if created {
473	480		parent := filepath.Dir(dir)
474	481		if _, err := os.Lstat(parent); errors.Is(err, os.ErrNotExist) {
475		-	if err := r.ensurePrivateDirectory(parent); err != nil {
	482	+	if err := r.ensurePrivateDirectoryPhase(parent, "snapshot directory"); err != nil {
476	483		return err
477	484		}
478	485		} else if err != nil {

@@ -486,7 +493,9 @@ func (r *Repository) ensurePrivateDirectory(dir string) error {
486	493		return err
487	494		}
488	495		if created {
489		-	return r.syncDirectory(filepath.Dir(dir))
	496	+	if err := r.syncDirectory(filepath.Dir(dir)); err != nil {
	497	+	return fmt.Errorf("%s parent directory sync: %w", phase, err)
	498	+	}
490	499		}
491	500		return nil
492	501		}

@@ -597,7 +606,11 @@ func (r *Repository) readHead(key string) (uint64, ports.SnapshotDigest, error)
597	606		// after verifier reads it through the same secure descriptor path as Load.
598	607		func (r *Repository) writeImmutable(path string, data []byte, verifier func([]byte) error) error {
599	608		dir := filepath.Dir(path)
600		-	if err := r.ensurePrivateDirectory(dir); err != nil {
	609	+	phase := "object shard"
	610	+	if filepath.Base(dir) == repositoryGenerations {
	611	+	phase = "generation"
	612	+	}
	613	+	if err := r.ensurePrivateDirectoryPhase(dir, phase); err != nil {
601	614		return err
602	615		}
603	616		temp, err := r.createTemp(dir)

@@ -608,8 +621,8 @@ func (r *Repository) writeImmutable(path string, data []byte, verifier func([]by
608	621		closed := false
609	622		cleanup := func(cause error) error {
610	623		if !closed {
611		-	if err := r.closeFile(temp); err != nil && cause == nil {
612		-	cause = err
	624	+	if err := r.closeFile(temp); err != nil {
	625	+	cause = errors.Join(cause, err)
613	626		}
614	627		closed = true
615	628		}

@@ -668,8 +681,8 @@ func (r *Repository) writeMutable(path string, data []byte) error {
668	681		closed := false
669	682		cleanup := func(cause error) error {
670	683		if !closed {
671		-	if err := r.closeFile(temp); err != nil && cause == nil {
672		-	cause = err
	684	+	if err := r.closeFile(temp); err != nil {
	685	+	cause = errors.Join(cause, err)
673	686		}
674	687		closed = true
675	688		}

 }
 func (r *Repository) ensureSession(key string) error {
-	for _, dir := range []string{
+	for _, directory := range []struct {
-		r.dir,
+		path  string
-		filepath.Join(r.dir, repositorySessionsDir),
+		phase string
-		r.sessionPath(key),
+	}{
-		filepath.Join(r.sessionPath(key), repositoryObjectsDir),
+		{r.dir, "repository"},
-		filepath.Join(r.sessionPath(key), repositoryGenerations),
+		{filepath.Join(r.dir, repositorySessionsDir), "sessions"},
+		{r.sessionPath(key), "session"},
+		{filepath.Join(r.sessionPath(key), repositoryObjectsDir), "objects"},
+		{filepath.Join(r.sessionPath(key), repositoryGenerations), "generations"},
 	} {
-		if err := r.ensurePrivateDirectory(dir); err != nil {
+		if err := r.ensurePrivateDirectoryPhase(directory.path, directory.phase); err != nil {
 			return fmt.Errorf("create snapshot repository directory: %w", err)
 		}
 	}
 }
 func (r *Repository) ensurePrivateDirectory(dir string) error {
+	return r.ensurePrivateDirectoryPhase(dir, "snapshot directory")
+}
+func (r *Repository) ensurePrivateDirectoryPhase(dir, phase string) error {
 	_, err := os.Lstat(dir)
 	created := errors.Is(err, os.ErrNotExist)
 	if err != nil && !created {
 	if created {
 		parent := filepath.Dir(dir)
 		if _, err := os.Lstat(parent); errors.Is(err, os.ErrNotExist) {
-			if err := r.ensurePrivateDirectory(parent); err != nil {
+			if err := r.ensurePrivateDirectoryPhase(parent, "snapshot directory"); err != nil {
 				return err
 			}
 		} else if err != nil {
 		return err
 	}
 	if created {
-		return r.syncDirectory(filepath.Dir(dir))
+		if err := r.syncDirectory(filepath.Dir(dir)); err != nil {
+			return fmt.Errorf("%s parent directory sync: %w", phase, err)
+		}
 	}
 	return nil
 }
 // after verifier reads it through the same secure descriptor path as Load.
 func (r *Repository) writeImmutable(path string, data []byte, verifier func([]byte) error) error {
 	dir := filepath.Dir(path)
-	if err := r.ensurePrivateDirectory(dir); err != nil {
+	phase := "object shard"
+	if filepath.Base(dir) == repositoryGenerations {
+		phase = "generation"
+	}
+	if err := r.ensurePrivateDirectoryPhase(dir, phase); err != nil {
 		return err
 	}
 	temp, err := r.createTemp(dir)
 	closed := false
 	cleanup := func(cause error) error {
 		if !closed {
-			if err := r.closeFile(temp); err != nil && cause == nil {
+			if err := r.closeFile(temp); err != nil {
-				cause = err
+				cause = errors.Join(cause, err)
 			}
 			closed = true
 		}
 	closed := false
 	cleanup := func(cause error) error {
 		if !closed {
-			if err := r.closeFile(temp); err != nil && cause == nil {
+			if err := r.closeFile(temp); err != nil {
-				cause = err
+				cause = errors.Join(cause, err)
 			}
 			closed = true
 		}

M internal/adapters/snapshot/repository_fault_test.go +145 -0


@@ -4,6 +4,8 @@ import (
4	4		"context"
5	5		"errors"
6	6		"os"
	7	+	"path/filepath"
	8	+	"strings"
7	9		"testing"
8	10
9	11		codec "github.com/bnema/vev/internal/usecase/snapshot"

@@ -31,6 +33,149 @@ func TestRepositoryHeadFailureKeepsOldCompleteGeneration(t *testing.T) {
31	33		}
32	34		}
33	35
	36	+	func TestRepositoryCleanupFaultsSurfaceForImmutableAndMutablePublication(t *testing.T) {
	37	+	primary := errors.New("primary write failure")
	38	+	cleanup := errors.New("cleanup failure")
	39	+	cases := []struct {
	40	+	name string
	41	+	location string
	42	+	cleanup string
	43	+	}{
	44	+	{name: "immutable/close temporary", location: "manifest", cleanup: "close"},
	45	+	{name: "immutable/remove temporary", location: "manifest", cleanup: "remove"},
	46	+	{name: "immutable/sync after temporary removal", location: "manifest", cleanup: "sync-after-remove"},
	47	+	{name: "mutable/close temporary", location: "HEAD", cleanup: "close"},
	48	+	{name: "mutable/remove temporary", location: "HEAD", cleanup: "remove"},
	49	+	{name: "mutable/sync after temporary removal", location: "HEAD", cleanup: "sync-after-remove"},
	50	+	}
	51	+	for _, tc := range cases {
	52	+	t.Run(tc.name, func(t *testing.T) {
	53	+	repo := NewRepository(privateDir(t))
	54	+	first := repositoryPublication(t, "named", 1, []byte("one"))
	55	+	if err := repo.Publish(context.Background(), first); err != nil {
	56	+	t.Fatal(err)
	57	+	}
	58	+	second := repositoryPublication(t, "named", 2, []byte("two"))
	59	+	key := sessionKey(second.Name)
	60	+	if tc.location == "manifest" {
	61	+	prepareObjects(t, repo, second)
	62	+	} else {
	63	+	prepareHeadStage(t, repo, second)
	64	+	}
	65	+	dir := filepath.Dir(repo.manifestPath(key, second.Generation))
	66	+	phase := "write manifest"
	67	+	if tc.location == "HEAD" {
	68	+	dir = filepath.Dir(repo.headPath(key))
	69	+	phase = "write HEAD"
	70	+	}
	71	+	repo.hooks.writeTemp = func(path string) error {
	72	+	if filepath.Dir(path) == dir {
	73	+	return primary
	74	+	}
	75	+	return nil
	76	+	}
	77	+	repo.hooks.closeFile = func(path string) error {
	78	+	if tc.cleanup == "close" && filepath.Dir(path) == dir {
	79	+	return cleanup
	80	+	}
	81	+	return nil
	82	+	}
	83	+	repo.hooks.remove = func(string) error {
	84	+	if tc.cleanup == "remove" {
	85	+	return cleanup
	86	+	}
	87	+	return nil
	88	+	}
	89	+	repo.hooks.syncDirectory = func(path string) error {
	90	+	if tc.cleanup == "sync-after-remove" && path == dir {
	91	+	return cleanup
	92	+	}
	93	+	return nil
	94	+	}
	95	+
	96	+	err := repo.Publish(context.Background(), second)
	97	+	if !errors.Is(err, primary) \|\| !errors.Is(err, cleanup) \|\| !strings.Contains(err.Error(), phase) {
	98	+	t.Fatalf("Publish error = %v, want %q and both primary and cleanup failures", err, phase)
	99	+	}
	100	+	if tc.cleanup != "remove" {
	101	+	matches, globErr := filepath.Glob(filepath.Join(dir, ".tmp-*"))
	102	+	if globErr != nil \|\| len(matches) != 0 {
	103	+	t.Fatalf("temporary files after successful cleanup = %v, glob error = %v", matches, globErr)
	104	+	}
	105	+	}
	106	+	repo.hooks = repositoryHooks{}
	107	+	assertRepositoryCompleteGeneration(t, repo, "named", 1, 2)
	108	+	})
	109	+	}
	110	+	}
	111	+
	112	+	func TestRepositoryNewDirectorySyncFaultsAreIndependent(t *testing.T) {
	113	+	cases := []struct {
	114	+	name string
	115	+	phase string
	116	+	occurrence int
	117	+	directory func(*Repository, string) string
	118	+	}{
	119	+	{name: "repository", phase: "repository", occurrence: 1, directory: func(_ *Repository, root string) string { return filepath.Dir(root) }},
	120	+	{name: "sessions", phase: "sessions", occurrence: 1, directory: func(repo *Repository, _ string) string { return repo.dir }},
	121	+	{name: "session", phase: "session", occurrence: 1, directory: func(repo *Repository, _ string) string { return filepath.Join(repo.dir, repositorySessionsDir) }},
	122	+	{name: "objects", phase: "objects", occurrence: 1, directory: func(repo *Repository, _ string) string { return repo.sessionPath(sessionKey("named")) }},
	123	+	{name: "generations", phase: "generations", occurrence: 2, directory: func(repo *Repository, _ string) string { return repo.sessionPath(sessionKey("named")) }},
	124	+	{name: "object shard", phase: "object shard", occurrence: 1, directory: func(repo *Repository, _ string) string {
	125	+	return filepath.Join(repo.sessionPath(sessionKey("named")), repositoryObjectsDir)
	126	+	}},
	127	+	}
	128	+	for _, tc := range cases {
	129	+	t.Run(tc.name, func(t *testing.T) {
	130	+	parent := privateDir(t)
	131	+	root := filepath.Join(parent, "repository")
	132	+	repo := NewRepository(root)
	133	+	wantDirectory := tc.directory(repo, root)
	134	+	calls := 0
	135	+	injected := errors.New("new directory sync failure")
	136	+	repo.hooks.syncDirectory = func(directory string) error {
	137	+	if directory == wantDirectory {
	138	+	calls++
	139	+	if calls == tc.occurrence {
	140	+	return injected
	141	+	}
	142	+	}
	143	+	return nil
	144	+	}
	145	+
	146	+	publication := repositoryPublication(t, "named", 1, []byte("one"))
	147	+	err := repo.Publish(context.Background(), publication)
	148	+	if !errors.Is(err, injected) \|\| !strings.Contains(err.Error(), tc.phase) {
	149	+	t.Fatalf("Publish error = %v, want %q-labelled directory sync failure", err, tc.phase)
	150	+	}
	151	+	repo.hooks = repositoryHooks{}
	152	+	// A first-generation failure cannot have an older complete generation;
	153	+	// it must not expose an incomplete one as if it were loadable.
	154	+	if _, loadErr := repo.Load(context.Background(), "named"); loadErr == nil \|\| !strings.Contains(loadErr.Error(), "no complete snapshot generation") {
	155	+	t.Fatalf("Load error after incomplete first generation = %v, want no complete generation", loadErr)
	156	+	}
	157	+	if err := repo.Publish(context.Background(), publication); err != nil {
	158	+	t.Fatal(err)
	159	+	}
	160	+	assertRepositoryCompleteGeneration(t, repo, "named", 1)
	161	+	})
	162	+	}
	163	+	}
	164	+
	165	+	func assertRepositoryCompleteGeneration(t testing.T, repo Repository, name string, generations ...uint64) {
	166	+	t.Helper()
	167	+	got, err := repo.Load(context.Background(), name)
	168	+	if err != nil {
	169	+	t.Fatal(err)
	170	+	}
	171	+	for _, generation := range generations {
	172	+	if got.Generation == generation && len(got.Manifest) != 0 && len(got.Objects) != 0 {
	173	+	return
	174	+	}
	175	+	}
	176	+	t.Fatalf("Load = incomplete or mixed generation %+v, want one of %v", got, generations)
	177	+	}
	178	+
34	179		func TestRepositoryLoadFallsBackFromIncompleteNewestGeneration(t *testing.T) {
35	180		dir := privateDir(t)
36	181		repo := NewRepository(dir)


@@ -4,6 +4,8 @@ import (
4	"context"	4		"context"
5	"errors"	5		"errors"
6	"os"	6		"os"
		7	+	"path/filepath"
		8	+	"strings"
7	"testing"	9		"testing"
8		10
9	codec "github.com/bnema/vev/internal/usecase/snapshot"	11		codec "github.com/bnema/vev/internal/usecase/snapshot"

@@ -31,6 +33,149 @@ func TestRepositoryHeadFailureKeepsOldCompleteGeneration(t *testing.T) {
31	}	33		}
32	}	34		}
33		35
		36	+	func TestRepositoryCleanupFaultsSurfaceForImmutableAndMutablePublication(t *testing.T) {
		37	+	primary := errors.New("primary write failure")
		38	+	cleanup := errors.New("cleanup failure")
		39	+	cases := []struct {
		40	+	name string
		41	+	location string
		42	+	cleanup string
		43	+	}{
		44	+	{name: "immutable/close temporary", location: "manifest", cleanup: "close"},
		45	+	{name: "immutable/remove temporary", location: "manifest", cleanup: "remove"},
		46	+	{name: "immutable/sync after temporary removal", location: "manifest", cleanup: "sync-after-remove"},
		47	+	{name: "mutable/close temporary", location: "HEAD", cleanup: "close"},
		48	+	{name: "mutable/remove temporary", location: "HEAD", cleanup: "remove"},
		49	+	{name: "mutable/sync after temporary removal", location: "HEAD", cleanup: "sync-after-remove"},
		50	+	}
		51	+	for _, tc := range cases {
		52	+	t.Run(tc.name, func(t *testing.T) {
		53	+	repo := NewRepository(privateDir(t))
		54	+	first := repositoryPublication(t, "named", 1, []byte("one"))
		55	+	if err := repo.Publish(context.Background(), first); err != nil {
		56	+	t.Fatal(err)
		57	+	}
		58	+	second := repositoryPublication(t, "named", 2, []byte("two"))
		59	+	key := sessionKey(second.Name)
		60	+	if tc.location == "manifest" {
		61	+	prepareObjects(t, repo, second)
		62	+	} else {
		63	+	prepareHeadStage(t, repo, second)
		64	+	}
		65	+	dir := filepath.Dir(repo.manifestPath(key, second.Generation))
		66	+	phase := "write manifest"
		67	+	if tc.location == "HEAD" {
		68	+	dir = filepath.Dir(repo.headPath(key))
		69	+	phase = "write HEAD"
		70	+	}
		71	+	repo.hooks.writeTemp = func(path string) error {
		72	+	if filepath.Dir(path) == dir {
		73	+	return primary
		74	+	}
		75	+	return nil
		76	+	}
		77	+	repo.hooks.closeFile = func(path string) error {
		78	+	if tc.cleanup == "close" && filepath.Dir(path) == dir {
		79	+	return cleanup
		80	+	}
		81	+	return nil
		82	+	}
		83	+	repo.hooks.remove = func(string) error {
		84	+	if tc.cleanup == "remove" {
		85	+	return cleanup
		86	+	}
		87	+	return nil
		88	+	}
		89	+	repo.hooks.syncDirectory = func(path string) error {
		90	+	if tc.cleanup == "sync-after-remove" && path == dir {
		91	+	return cleanup
		92	+	}
		93	+	return nil
		94	+	}
		95	+
		96	+	err := repo.Publish(context.Background(), second)
		97	+	if !errors.Is(err, primary) \|\| !errors.Is(err, cleanup) \|\| !strings.Contains(err.Error(), phase) {
		98	+	t.Fatalf("Publish error = %v, want %q and both primary and cleanup failures", err, phase)
		99	+	}
		100	+	if tc.cleanup != "remove" {
		101	+	matches, globErr := filepath.Glob(filepath.Join(dir, ".tmp-*"))
		102	+	if globErr != nil \|\| len(matches) != 0 {
		103	+	t.Fatalf("temporary files after successful cleanup = %v, glob error = %v", matches, globErr)
		104	+	}
		105	+	}
		106	+	repo.hooks = repositoryHooks{}
		107	+	assertRepositoryCompleteGeneration(t, repo, "named", 1, 2)
		108	+	})
		109	+	}
		110	+	}
		111	+
		112	+	func TestRepositoryNewDirectorySyncFaultsAreIndependent(t *testing.T) {
		113	+	cases := []struct {
		114	+	name string
		115	+	phase string
		116	+	occurrence int
		117	+	directory func(*Repository, string) string
		118	+	}{
		119	+	{name: "repository", phase: "repository", occurrence: 1, directory: func(_ *Repository, root string) string { return filepath.Dir(root) }},
		120	+	{name: "sessions", phase: "sessions", occurrence: 1, directory: func(repo *Repository, _ string) string { return repo.dir }},
		121	+	{name: "session", phase: "session", occurrence: 1, directory: func(repo *Repository, _ string) string { return filepath.Join(repo.dir, repositorySessionsDir) }},
		122	+	{name: "objects", phase: "objects", occurrence: 1, directory: func(repo *Repository, _ string) string { return repo.sessionPath(sessionKey("named")) }},
		123	+	{name: "generations", phase: "generations", occurrence: 2, directory: func(repo *Repository, _ string) string { return repo.sessionPath(sessionKey("named")) }},
		124	+	{name: "object shard", phase: "object shard", occurrence: 1, directory: func(repo *Repository, _ string) string {
		125	+	return filepath.Join(repo.sessionPath(sessionKey("named")), repositoryObjectsDir)
		126	+	}},
		127	+	}
		128	+	for _, tc := range cases {
		129	+	t.Run(tc.name, func(t *testing.T) {
		130	+	parent := privateDir(t)
		131	+	root := filepath.Join(parent, "repository")
		132	+	repo := NewRepository(root)
		133	+	wantDirectory := tc.directory(repo, root)
		134	+	calls := 0
		135	+	injected := errors.New("new directory sync failure")
		136	+	repo.hooks.syncDirectory = func(directory string) error {
		137	+	if directory == wantDirectory {
		138	+	calls++
		139	+	if calls == tc.occurrence {
		140	+	return injected
		141	+	}
		142	+	}
		143	+	return nil
		144	+	}
		145	+
		146	+	publication := repositoryPublication(t, "named", 1, []byte("one"))
		147	+	err := repo.Publish(context.Background(), publication)
		148	+	if !errors.Is(err, injected) \|\| !strings.Contains(err.Error(), tc.phase) {
		149	+	t.Fatalf("Publish error = %v, want %q-labelled directory sync failure", err, tc.phase)
		150	+	}
		151	+	repo.hooks = repositoryHooks{}
		152	+	// A first-generation failure cannot have an older complete generation;
		153	+	// it must not expose an incomplete one as if it were loadable.
		154	+	if _, loadErr := repo.Load(context.Background(), "named"); loadErr == nil \|\| !strings.Contains(loadErr.Error(), "no complete snapshot generation") {
		155	+	t.Fatalf("Load error after incomplete first generation = %v, want no complete generation", loadErr)
		156	+	}
		157	+	if err := repo.Publish(context.Background(), publication); err != nil {
		158	+	t.Fatal(err)
		159	+	}
		160	+	assertRepositoryCompleteGeneration(t, repo, "named", 1)
		161	+	})
		162	+	}
		163	+	}
		164	+
		165	+	func assertRepositoryCompleteGeneration(t testing.T, repo Repository, name string, generations ...uint64) {
		166	+	t.Helper()
		167	+	got, err := repo.Load(context.Background(), name)
		168	+	if err != nil {
		169	+	t.Fatal(err)
		170	+	}
		171	+	for _, generation := range generations {
		172	+	if got.Generation == generation && len(got.Manifest) != 0 && len(got.Objects) != 0 {
		173	+	return
		174	+	}
		175	+	}
		176	+	t.Fatalf("Load = incomplete or mixed generation %+v, want one of %v", got, generations)
		177	+	}
		178	+
34	func TestRepositoryLoadFallsBackFromIncompleteNewestGeneration(t *testing.T) {	179		func TestRepositoryLoadFallsBackFromIncompleteNewestGeneration(t *testing.T) {
35	dir := privateDir(t)	180		dir := privateDir(t)
36	repo := NewRepository(dir)	181		repo := NewRepository(dir)