bd37e91 test(recovery): complete acceptance verification · vev

M internal/app/run.go +7 -2


@@ -117,6 +117,11 @@ var (
117	117		version = "0.1.0-dev"
118	118		commit = "none"
119	119		date = "unknown"
	120	+
	121	+	openOrMigrate = persist.OpenOrMigrate
	122	+	listenDaemon = func(dir string, observer ports.SerializedRuntimeObserver) (ports.Listener, error) {
	123	+	return ipc.Listen(dir, ipc.WithRuntimeObserver(observer))
	124	+	}
120	125		)
121	126
122	127		// versionLine renders the --version output.

@@ -632,7 +637,7 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
632	637		noticeStore := noticefile.New(platform.StateDir())
633	638		daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticeStore))
634	639		storePath := persist.StorePath(platform.StateDir())
635		-	opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{
	640	+	opened, err := openOrMigrate(ctx, persist.OpenDeps{
636	641		StateDir: platform.StateDir(),
637	642		Random: rand.Reader,
638	643		SnapshotMigration: snapshotRepository,

@@ -676,7 +681,7 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
676	681		// publication. Phase 3 snapshot restoration remains asynchronous in Serve.
677	682		d, ln, err := constructDaemonBeforeSocketPublication(
678	683		func() *daemon.Daemon { return daemon.New(pty.NewFactory(), clk, log, daemonOpts...) },
679		-	func() (ports.Listener, error) { return ipc.Listen(ipc.SocketDir(), ipc.WithRuntimeObserver(observer)) },
	684	+	func() (ports.Listener, error) { return listenDaemon(ipc.SocketDir(), observer) },
680	685		)
681	686		if err != nil {
682	687		closeErr := opened.Catalogue.Close()

 	version = "0.1.0-dev"
 	commit  = "none"
 	date    = "unknown"
+	openOrMigrate = persist.OpenOrMigrate
+	listenDaemon  = func(dir string, observer ports.SerializedRuntimeObserver) (ports.Listener, error) {
+		return ipc.Listen(dir, ipc.WithRuntimeObserver(observer))
+	}
 )
 // versionLine renders the --version output.
 	noticeStore := noticefile.New(platform.StateDir())
 	daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticeStore))
 	storePath := persist.StorePath(platform.StateDir())
-	opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{
+	opened, err := openOrMigrate(ctx, persist.OpenDeps{
 		StateDir:          platform.StateDir(),
 		Random:            rand.Reader,
 		SnapshotMigration: snapshotRepository,
 	// publication. Phase 3 snapshot restoration remains asynchronous in Serve.
 	d, ln, err := constructDaemonBeforeSocketPublication(
 		func() *daemon.Daemon { return daemon.New(pty.NewFactory(), clk, log, daemonOpts...) },
-		func() (ports.Listener, error) { return ipc.Listen(ipc.SocketDir(), ipc.WithRuntimeObserver(observer)) },
+		func() (ports.Listener, error) { return listenDaemon(ipc.SocketDir(), observer) },
 	)
 	if err != nil {
 		closeErr := opened.Catalogue.Close()

M internal/app/run_test.go +39 -0


@@ -212,6 +212,45 @@ func TestOfflineCommandsPropagateLifecycleReleaseErrors(t *testing.T) {
212	212		}
213	213		}
214	214
	215	+	func TestCatalogueFailureDoesNotListen(t *testing.T) {
	216	+	originalOpenOrMigrate, originalListenDaemon := openOrMigrate, listenDaemon
	217	+	t.Cleanup(func() {
	218	+	openOrMigrate = originalOpenOrMigrate
	219	+	listenDaemon = originalListenDaemon
	220	+	})
	221	+
	222	+	for _, catalogueErr := range []error{errors.New("catalogue corrupt"), errors.New("catalogue unavailable")} {
	223	+	t.Run(catalogueErr.Error(), func(t *testing.T) {
	224	+	runtimeRoot, stateRoot := t.TempDir(), t.TempDir()
	225	+	runtimeDir := filepath.Join(runtimeRoot, "vev")
	226	+	stateDir := filepath.Join(stateRoot, "vev")
	227	+	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
	228	+	t.Setenv("XDG_STATE_HOME", stateRoot)
	229	+
	230	+	listenCalls := 0
	231	+	openOrMigrate = func(_ context.Context, deps persist.OpenDeps) (persist.OpenResult, error) {
	232	+	require.Equal(t, stateDir, deps.StateDir)
	233	+	return persist.OpenResult{}, catalogueErr
	234	+	}
	235	+	listenDaemon = func(string, ports.SerializedRuntimeObserver) (ports.Listener, error) {
	236	+	listenCalls++
	237	+	return nil, errors.New("IPC listen must not run after catalogue failure")
	238	+	}
	239	+
	240	+	err := runWithLifecycleOwner(context.Background(), runtimeDir, stateDir, func(ctx context.Context) error {
	241	+	return runDaemonOwnedWithLogger(ctx, slog.New(slog.NewTextHandler(io.Discard, nil)))
	242	+	})
	243	+	require.ErrorIs(t, err, catalogueErr)
	244	+	require.Zero(t, listenCalls)
	245	+	require.NoFileExists(t, filepath.Join(runtimeDir, "daemon.sock"))
	246	+
	247	+	owner, acquireErr := lifecycle.TryAcquire(runtimeDir)
	248	+	require.NoError(t, acquireErr, "catalogue failure must release lifecycle ownership")
	249	+	require.NoError(t, owner.Release())
	250	+	})
	251	+	}
	252	+	}
	253	+
215	254		func TestCatalogueRegistryConstructionPrecedesSocketPublication(t *testing.T) {
216	255		var events []string
217	256		_, _, err := constructDaemonBeforeSocketPublication(


@@ -212,6 +212,45 @@ func TestOfflineCommandsPropagateLifecycleReleaseErrors(t *testing.T) {
212	}	212		}
213	}	213		}
214		214
		215	+	func TestCatalogueFailureDoesNotListen(t *testing.T) {
		216	+	originalOpenOrMigrate, originalListenDaemon := openOrMigrate, listenDaemon
		217	+	t.Cleanup(func() {
		218	+	openOrMigrate = originalOpenOrMigrate
		219	+	listenDaemon = originalListenDaemon
		220	+	})
		221	+
		222	+	for _, catalogueErr := range []error{errors.New("catalogue corrupt"), errors.New("catalogue unavailable")} {
		223	+	t.Run(catalogueErr.Error(), func(t *testing.T) {
		224	+	runtimeRoot, stateRoot := t.TempDir(), t.TempDir()
		225	+	runtimeDir := filepath.Join(runtimeRoot, "vev")
		226	+	stateDir := filepath.Join(stateRoot, "vev")
		227	+	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
		228	+	t.Setenv("XDG_STATE_HOME", stateRoot)
		229	+
		230	+	listenCalls := 0
		231	+	openOrMigrate = func(_ context.Context, deps persist.OpenDeps) (persist.OpenResult, error) {
		232	+	require.Equal(t, stateDir, deps.StateDir)
		233	+	return persist.OpenResult{}, catalogueErr
		234	+	}
		235	+	listenDaemon = func(string, ports.SerializedRuntimeObserver) (ports.Listener, error) {
		236	+	listenCalls++
		237	+	return nil, errors.New("IPC listen must not run after catalogue failure")
		238	+	}
		239	+
		240	+	err := runWithLifecycleOwner(context.Background(), runtimeDir, stateDir, func(ctx context.Context) error {
		241	+	return runDaemonOwnedWithLogger(ctx, slog.New(slog.NewTextHandler(io.Discard, nil)))
		242	+	})
		243	+	require.ErrorIs(t, err, catalogueErr)
		244	+	require.Zero(t, listenCalls)
		245	+	require.NoFileExists(t, filepath.Join(runtimeDir, "daemon.sock"))
		246	+
		247	+	owner, acquireErr := lifecycle.TryAcquire(runtimeDir)
		248	+	require.NoError(t, acquireErr, "catalogue failure must release lifecycle ownership")
		249	+	require.NoError(t, owner.Release())
		250	+	})
		251	+	}
		252	+	}
		253	+
215	func TestCatalogueRegistryConstructionPrecedesSocketPublication(t *testing.T) {	254		func TestCatalogueRegistryConstructionPrecedesSocketPublication(t *testing.T) {
216	var events []string	255		var events []string
217	_, _, err := constructDaemonBeforeSocketPublication(	256		_, _, err := constructDaemonBeforeSocketPublication(

M internal/persist/migration.go +13 -10


@@ -563,7 +563,7 @@ func appendMigrationString(out []byte, value string) []byte {
563	563		out = binary.BigEndian.AppendUint32(out, uint32(len(value)))
564	564		return append(out, value...)
565	565		}
566		-	func atomicMigrationWrite(path string, data []byte) error {
	566	+	func atomicMigrationWrite(path string, data []byte) (retErr error) {
567	567		dir := filepath.Dir(path)
568	568		if err := safedir.EnsurePrivate(dir); err != nil {
569	569		return err

@@ -573,18 +573,21 @@ func atomicMigrationWrite(path string, data []byte) error {
573	573		return err
574	574		}
575	575		tmpName := tmp.Name()
576		-	defer os.Remove(tmpName)
	576	+	published := false
	577	+	defer func() {
	578	+	if !published {
	579	+	// A failed publication must not leave a replayable partial intent.
	580	+	retErr = errors.Join(retErr, os.Remove(tmpName))
	581	+	}
	582	+	}()
577	583		if err := tmp.Chmod(0o600); err != nil {
578		-	tmp.Close()
579		-	return err
	584	+	return errors.Join(err, tmp.Close())
580	585		}
581	586		if _, err := tmp.Write(data); err != nil {
582		-	tmp.Close()
583		-	return err
	587	+	return errors.Join(err, tmp.Close())
584	588		}
585	589		if err := tmp.Sync(); err != nil {
586		-	tmp.Close()
587		-	return err
	590	+	return errors.Join(err, tmp.Close())
588	591		}
589	592		if err := tmp.Close(); err != nil {
590	593		return err

@@ -592,6 +595,7 @@ func atomicMigrationWrite(path string, data []byte) error {
592	595		if err := os.Rename(tmpName, path); err != nil {
593	596		return err
594	597		}
	598	+	published = true
595	599		return syncMigrationDir(dir)
596	600		}
597	601		func syncMigrationDir(dir string) error {

@@ -599,8 +603,7 @@ func syncMigrationDir(dir string) error {
599	603		if err != nil {
600	604		return err
601	605		}
602		-	defer f.Close()
603		-	return f.Sync()
	606	+	return errors.Join(f.Sync(), f.Close())
604	607		}
605	608		func catalogueRecordsEqual(a, b []domain.CatalogueRecord) bool {
606	609		if len(a) != len(b) {

 	out = binary.BigEndian.AppendUint32(out, uint32(len(value)))
 	return append(out, value...)
 }
-func atomicMigrationWrite(path string, data []byte) error {
+func atomicMigrationWrite(path string, data []byte) (retErr error) {
 	dir := filepath.Dir(path)
 	if err := safedir.EnsurePrivate(dir); err != nil {
 		return err
 		return err
 	}
 	tmpName := tmp.Name()
-	defer os.Remove(tmpName)
+	published := false
+	defer func() {
+		if !published {
+			// A failed publication must not leave a replayable partial intent.
+			retErr = errors.Join(retErr, os.Remove(tmpName))
+		}
+	}()
 	if err := tmp.Chmod(0o600); err != nil {
-		tmp.Close()
+		return errors.Join(err, tmp.Close())
-		return err
 	}
 	if _, err := tmp.Write(data); err != nil {
-		tmp.Close()
+		return errors.Join(err, tmp.Close())
-		return err
 	}
 	if err := tmp.Sync(); err != nil {
-		tmp.Close()
+		return errors.Join(err, tmp.Close())
-		return err
 	}
 	if err := tmp.Close(); err != nil {
 		return err
 	if err := os.Rename(tmpName, path); err != nil {
 		return err
 	}
+	published = true
 	return syncMigrationDir(dir)
 }
 func syncMigrationDir(dir string) error {
 	if err != nil {
 		return err
 	}
-	defer f.Close()
+	return errors.Join(f.Sync(), f.Close())
-	return f.Sync()
 }
 func catalogueRecordsEqual(a, b []domain.CatalogueRecord) bool {
 	if len(a) != len(b) {

M internal/persist/migration_test.go +15 -0


@@ -278,6 +278,21 @@ func TestOpenOrMigrateLegacyProbe(t *testing.T) {
278	278		}
279	279		}
280	280
	281	+	func TestAtomicMigrationWriteRemovesTempAfterRenameFailure(t *testing.T) {
	282	+	dir := t.TempDir()
	283	+	target := filepath.Join(dir, "target")
	284	+	require.NoError(t, os.Mkdir(target, 0o700))
	285	+
	286	+	err := atomicMigrationWrite(target, []byte("intent"))
	287	+	require.Error(t, err)
	288	+
	289	+	entries, readErr := os.ReadDir(dir)
	290	+	require.NoError(t, readErr)
	291	+	for _, entry := range entries {
	292	+	require.NotContains(t, entry.Name(), ".migration-", "failed publication must remove its temporary intent")
	293	+	}
	294	+	}
	295	+
281	296		type countingReader struct{ next byte }
282	297
283	298		func (r *countingReader) Read(p []byte) (int, error) {


@@ -278,6 +278,21 @@ func TestOpenOrMigrateLegacyProbe(t *testing.T) {
278	}	278		}
279	}	279		}
280		280
		281	+	func TestAtomicMigrationWriteRemovesTempAfterRenameFailure(t *testing.T) {
		282	+	dir := t.TempDir()
		283	+	target := filepath.Join(dir, "target")
		284	+	require.NoError(t, os.Mkdir(target, 0o700))
		285	+
		286	+	err := atomicMigrationWrite(target, []byte("intent"))
		287	+	require.Error(t, err)
		288	+
		289	+	entries, readErr := os.ReadDir(dir)
		290	+	require.NoError(t, readErr)
		291	+	for _, entry := range entries {
		292	+	require.NotContains(t, entry.Name(), ".migration-", "failed publication must remove its temporary intent")
		293	+	}
		294	+	}
		295	+
281	type countingReader struct{ next byte }	296		type countingReader struct{ next byte }
282		297
283	func (r *countingReader) Read(p []byte) (int, error) {	298		func (r *countingReader) Read(p []byte) (int, error) {

M internal/usecase/daemon/performance_benchmark_test.go +8 -1


@@ -725,7 +725,14 @@ func snapshotGeneration(publication ports.SnapshotPublication) ports.SnapshotGen
725	725		for _, object := range publication.Objects {
726	726		objects[object.Digest] = object.Data
727	727		}
728		-	return ports.SnapshotGeneration{Name: publication.Name, Generation: publication.Generation, Manifest: publication.Manifest, Objects: objects}
	728	+	return ports.SnapshotGeneration{
	729	+	IncarnationID: publication.IncarnationID,
	730	+	Name: publication.Name,
	731	+	Generation: publication.Generation,
	732	+	ParentCheckpoint: publication.ParentCheckpoint,
	733	+	Manifest: publication.Manifest,
	734	+	Objects: objects,
	735	+	}
729	736		}
730	737
731	738		func snapshotObjectBytes(publication ports.SnapshotPublication) uint64 {

 	for _, object := range publication.Objects {
 		objects[object.Digest] = object.Data
 	}
-	return ports.SnapshotGeneration{Name: publication.Name, Generation: publication.Generation, Manifest: publication.Manifest, Objects: objects}
+	return ports.SnapshotGeneration{
+		IncarnationID:    publication.IncarnationID,
+		Name:             publication.Name,
+		Generation:       publication.Generation,
+		ParentCheckpoint: publication.ParentCheckpoint,
+		Manifest:         publication.Manifest,
+		Objects:          objects,
+	}
 }
 func snapshotObjectBytes(publication ports.SnapshotPublication) uint64 {