b8794ea feat(persist): migrate catalogue records safely (#249) · vev

M docs/configuration.md +1 -1


@@ -89,7 +89,7 @@ Remote host commands, listing, and successful direct-attach learning are always
89	89
90	90		## Logs and durable state
91	91
92		-	Set `VEV_LOG=debug`, `VEV_LOG=warn`, or `VEV_LOG=error` to change verbosity; the default is `info`. JSON-line logs such as `vev-daemon.log` live in `$XDG_STATE_HOME/vev`, or `~/.local/state/vev` when unset. The same state directory contains the strict session catalogue, migration journal, notices, `hosts.json`, and `snapshots/`. The lifecycle lock and socket live in `$XDG_RUNTIME_DIR/vev` (with platform runtime fallbacks).
	92	+	Set `VEV_LOG=debug`, `VEV_LOG=warn`, or `VEV_LOG=error` to change verbosity; the default is `info`. JSON-line logs such as `vev-daemon.log` live in `$XDG_STATE_HOME/vev`, or `~/.local/state/vev` when unset. The same state directory contains the strict session catalogue, any private pre-migration catalogue backup, notices, `hosts.json`, and `snapshots/`. The lifecycle lock and socket live in `$XDG_RUNTIME_DIR/vev` (with platform runtime fallbacks).
93	93
94	94		Recovery events include `lifecycle_owner_wait`, `lifecycle_owner_acquired`, `lifecycle_owner_released`, `catalogue_validated`, `catalogue_compaction_recovery_complete`, `session_restore_complete`, `fallback_checkpoint_promoted`, `snapshot_head_repair_complete`, `session_degraded`, `snapshot_maintenance_progress`, `interrupted_transaction_recovery_complete`, and `daemon_startup_complete`.
95	95


@@ -89,7 +89,7 @@ Remote host commands, listing, and successful direct-attach learning are always
89			89
90		## Logs and durable state	90		## Logs and durable state
91			91
92	-	Set `VEV_LOG=debug`, `VEV_LOG=warn`, or `VEV_LOG=error` to change verbosity; the default is `info`. JSON-line logs such as `vev-daemon.log` live in `$XDG_STATE_HOME/vev`, or `~/.local/state/vev` when unset. The same state directory contains the strict session catalogue, migration journal, notices, `hosts.json`, and `snapshots/`. The lifecycle lock and socket live in `$XDG_RUNTIME_DIR/vev` (with platform runtime fallbacks).	92	+	Set `VEV_LOG=debug`, `VEV_LOG=warn`, or `VEV_LOG=error` to change verbosity; the default is `info`. JSON-line logs such as `vev-daemon.log` live in `$XDG_STATE_HOME/vev`, or `~/.local/state/vev` when unset. The same state directory contains the strict session catalogue, any private pre-migration catalogue backup, notices, `hosts.json`, and `snapshots/`. The lifecycle lock and socket live in `$XDG_RUNTIME_DIR/vev` (with platform runtime fallbacks).
93			93
94		Recovery events include `lifecycle_owner_wait`, `lifecycle_owner_acquired`, `lifecycle_owner_released`, `catalogue_validated`, `catalogue_compaction_recovery_complete`, `session_restore_complete`, `fallback_checkpoint_promoted`, `snapshot_head_repair_complete`, `session_degraded`, `snapshot_maintenance_progress`, `interrupted_transaction_recovery_complete`, and `daemon_startup_complete`.	94		Recovery events include `lifecycle_owner_wait`, `lifecycle_owner_acquired`, `lifecycle_owner_released`, `catalogue_validated`, `catalogue_compaction_recovery_complete`, `session_restore_complete`, `fallback_checkpoint_promoted`, `snapshot_head_repair_complete`, `session_degraded`, `snapshot_maintenance_progress`, `interrupted_transaction_recovery_complete`, and `daemon_startup_complete`.
95			95

M docs/durable-session-recovery.md +4 -4


@@ -12,11 +12,11 @@ vev opens named-session state only while holding `$XDG_RUNTIME_DIR/vev/lifecycle
12	12
13	13		Discard creates a new incarnation and retains the old record and snapshots under `snapshots/quarantine/` until an explicit later purge.
14	14
15		-	## Protocol upgrades
	15	+	## Catalogue format upgrades
16	16
17		-	Each catalogue record stores the protocol version that wrote it. At startup, a record from any other protocol—including a newer protocol after a downgrade—is atomically replaced with a fresh incarnation before the daemon publishes its socket. Only the session name is retained; the working directory, layout, tabs, terminal history, recovery transcript, process-recovery state, and checkpoints are discarded. Records created before protocol tracking are treated the same way.
	17	+	Catalogue format version 6 is independent from the live client/daemon protocol. At startup, vev losslessly converts supported version 3–5 records to version 6 before publishing its socket. The conversion preserves session identity, working directory, tabs, timestamps, checkpoint references, and degradation state; a wire-protocol update alone never resets durable sessions.
18	18
19		-	This reset is per named session and crash-safe. Once the fresh catalogue authority is committed, leftover snapshot objects are no longer authoritative and startup garbage collection can remove them. Catalogue corruption is not treated as a protocol mismatch and remains fail-closed.
	19	+	Before rewriting a legacy catalogue, vev creates and syncs a private `sessions.kv.pre-v6.bak` backup beside it. The catalogue is then replaced atomically. If the source is corrupt, from an unsupported future format, or conflicts with an existing backup, startup fails closed and leaves the catalogue untouched. Version 6 is not readable by older binaries; to roll back, stop vev and restore the backup before starting the older binary.
20	20
21	21		## Incompatible checkpoints
22	22

@@ -36,6 +36,6 @@ A `lifecycle_owner_wait` event normally means another daemon is initializing or
36	36
37	37		## Diagnostics
38	38
39		-	Durable state and recovery journals are under `$XDG_STATE_HOME/vev` (default `~/.local/state/vev`). Snapshot quarantine is under its `snapshots/quarantine/` tree. Runtime ownership is under `$XDG_RUNTIME_DIR/vev`; logs are JSON lines in the state directory.
	39	+	Durable state and catalogue migration backups are under `$XDG_STATE_HOME/vev` (default `~/.local/state/vev`). Snapshot quarantine is under its `snapshots/quarantine/` tree. Runtime ownership is under `$XDG_RUNTIME_DIR/vev`; logs are JSON lines in the state directory.
40	40
41	41		Use session names, incarnation IDs, generation numbers, reason codes, and cursors from recovery events when diagnosing a failure. Recovery logs never include terminal or snapshot object contents.


@@ -12,11 +12,11 @@ vev opens named-session state only while holding `$XDG_RUNTIME_DIR/vev/lifecycle
12			12
13		Discard creates a new incarnation and retains the old record and snapshots under `snapshots/quarantine/` until an explicit later purge.	13		Discard creates a new incarnation and retains the old record and snapshots under `snapshots/quarantine/` until an explicit later purge.
14			14
15	-	## Protocol upgrades	15	+	## Catalogue format upgrades
16			16
17	-	Each catalogue record stores the protocol version that wrote it. At startup, a record from any other protocol—including a newer protocol after a downgrade—is atomically replaced with a fresh incarnation before the daemon publishes its socket. Only the session name is retained; the working directory, layout, tabs, terminal history, recovery transcript, process-recovery state, and checkpoints are discarded. Records created before protocol tracking are treated the same way.	17	+	Catalogue format version 6 is independent from the live client/daemon protocol. At startup, vev losslessly converts supported version 3–5 records to version 6 before publishing its socket. The conversion preserves session identity, working directory, tabs, timestamps, checkpoint references, and degradation state; a wire-protocol update alone never resets durable sessions.
18			18
19	-	This reset is per named session and crash-safe. Once the fresh catalogue authority is committed, leftover snapshot objects are no longer authoritative and startup garbage collection can remove them. Catalogue corruption is not treated as a protocol mismatch and remains fail-closed.	19	+	Before rewriting a legacy catalogue, vev creates and syncs a private `sessions.kv.pre-v6.bak` backup beside it. The catalogue is then replaced atomically. If the source is corrupt, from an unsupported future format, or conflicts with an existing backup, startup fails closed and leaves the catalogue untouched. Version 6 is not readable by older binaries; to roll back, stop vev and restore the backup before starting the older binary.
20			20
21		## Incompatible checkpoints	21		## Incompatible checkpoints
22			22

@@ -36,6 +36,6 @@ A `lifecycle_owner_wait` event normally means another daemon is initializing or
36			36
37		## Diagnostics	37		## Diagnostics
38			38
39	-	Durable state and recovery journals are under `$XDG_STATE_HOME/vev` (default `~/.local/state/vev`). Snapshot quarantine is under its `snapshots/quarantine/` tree. Runtime ownership is under `$XDG_RUNTIME_DIR/vev`; logs are JSON lines in the state directory.	39	+	Durable state and catalogue migration backups are under `$XDG_STATE_HOME/vev` (default `~/.local/state/vev`). Snapshot quarantine is under its `snapshots/quarantine/` tree. Runtime ownership is under `$XDG_RUNTIME_DIR/vev`; logs are JSON lines in the state directory.
40			40
41		Use session names, incarnation IDs, generation numbers, reason codes, and cursors from recovery events when diagnosing a failure. Recovery logs never include terminal or snapshot object contents.	41		Use session names, incarnation IDs, generation numbers, reason codes, and cursors from recovery events when diagnosing a failure. Recovery logs never include terminal or snapshot object contents.

M docs/remote-resilience.md +7 -7


@@ -79,13 +79,13 @@ the destination's authoritative output is flushed.
79	79
80	80		## Durable record compatibility
81	81
82		-	Catalogue record format version 5 stores the protocol version alongside stable
83		-	tab identity metadata. At daemon startup, version 3 and 4 records and version 5
84		-	records written by another protocol are decoded only to retain their session
85		-	names; all other session state is reset before socket publication. Compatible
86		-	version 5 records restore normally. Older binaries reject version 5 records, so
87		-	a downgrade also resets sessions when the older binary can read the catalogue,
88		-	or fails closed when it cannot. Back up the vev state before a rollback.
	82	+	Catalogue record format version 6 stores durable session metadata independently
	83	+	from the live protocol version. At daemon startup, supported version 3–5 records
	84	+	are losslessly converted before socket publication and the original catalogue is
	85	+	preserved as a private `sessions.kv.pre-v6.bak` backup. A wire-protocol change
	86	+	does not reset session state. Unsupported or corrupt records fail closed without
	87	+	modifying the catalogue. Older binaries cannot read version 6, so restore the
	88	+	backup before a rollback.
89	89
90	90		Remote catalogues are bounded and versioned. The current schema is mandatory
91	91		and contains lifecycle IDs, ordered typed tab records, state, active tab,


@@ -79,13 +79,13 @@ the destination's authoritative output is flushed.
79			79
80		## Durable record compatibility	80		## Durable record compatibility
81			81
82	-	Catalogue record format version 5 stores the protocol version alongside stable	82	+	Catalogue record format version 6 stores durable session metadata independently
83	-	tab identity metadata. At daemon startup, version 3 and 4 records and version 5	83	+	from the live protocol version. At daemon startup, supported version 3–5 records
84	-	records written by another protocol are decoded only to retain their session	84	+	are losslessly converted before socket publication and the original catalogue is
85	-	names; all other session state is reset before socket publication. Compatible	85	+	preserved as a private `sessions.kv.pre-v6.bak` backup. A wire-protocol change
86	-	version 5 records restore normally. Older binaries reject version 5 records, so	86	+	does not reset session state. Unsupported or corrupt records fail closed without
87	-	a downgrade also resets sessions when the older binary can read the catalogue,	87	+	modifying the catalogue. Older binaries cannot read version 6, so restore the
88	-	or fails closed when it cannot. Back up the vev state before a rollback.	88	+	backup before a rollback.
89			89
90		Remote catalogues are bounded and versioned. The current schema is mandatory	90		Remote catalogues are bounded and versioned. The current schema is mandatory
91		and contains lifecycle IDs, ordered typed tab records, state, active tab,	91		and contains lifecycle IDs, ordered typed tab records, state, active tab,

M internal/app/run.go +3 -33


@@ -573,33 +573,6 @@ func runDaemon() (retErr error) {
573	573		}, lifecycleStartupDependencies(log))
574	574		}
575	575
576		-	func resetProtocolIncompatibleRecords(ctx context.Context, log slog.Logger, coordinator recovery.Coordinator, records, incompatible []domain.CatalogueRecord) ([]domain.CatalogueRecord, error) {
577		-	if len(incompatible) == 0 {
578		-	return records, nil
579		-	}
580		-	byName := make(map[string]int, len(records))
581		-	for i := range records {
582		-	byName[records[i].Name] = i
583		-	}
584		-	for _, old := range incompatible {
585		-	fresh, committed, err := coordinator.ResetProtocolIncompatible(ctx, old.Name, old.IncarnationID)
586		-	if !committed {
587		-	if err != nil {
588		-	return nil, fmt.Errorf("reset protocol-incompatible session %q: %w", old.Name, err)
589		-	}
590		-	return nil, fmt.Errorf("reset protocol-incompatible session %q: catalogue authority changed", old.Name)
591		-	}
592		-	if index, ok := byName[old.Name]; ok {
593		-	records[index] = fresh
594		-	}
595		-	log.Info("protocol_incompatible_session_reset", "session", old.Name, "stored_incarnation", old.IncarnationID.String())
596		-	if err != nil {
597		-	log.Warn("protocol_incompatible_snapshot_cleanup_failed", "session", old.Name, "err", err)
598		-	}
599		-	}
600		-	return records, nil
601		-	}
602		-
603	576		func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) {
604	577		log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode)
605	578		}

@@ -699,12 +672,9 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
699	672		recoveryMode = "new-install"
700	673		}
701	674		coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader)
702		-	opened.Records, err = resetProtocolIncompatibleRecords(ctx, log, coordinator, opened.Records, opened.IncompatibleRecords)
703		-	if err != nil {
704		-	return errors.Join(fmt.Errorf("vev: reset protocol-incompatible durable sessions: %w", err), opened.Catalogue.Close())
705		-	}
706		-	if len(opened.IncompatibleRecords) != 0 {
707		-	recoveryMode = "protocol-reset"
	675	+	if opened.Migration.Performed {
	676	+	recoveryMode = "catalogue-migrated"
	677	+	log.Info("catalogue_migrated", "source_formats", opened.Migration.SourceFormats, "target_format", opened.Migration.TargetFormat, "records", opened.Migration.RecordCount, "backup", opened.Migration.BackupPath)
708	678		}
709	679		logCatalogueRecovery(log, opened.Records, recoveryMode)
710	680		daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))


@@ -573,33 +573,6 @@ func runDaemon() (retErr error) {
573		}, lifecycleStartupDependencies(log))	573		}, lifecycleStartupDependencies(log))
574		}	574		}
575			575
576	-	func resetProtocolIncompatibleRecords(ctx context.Context, log slog.Logger, coordinator recovery.Coordinator, records, incompatible []domain.CatalogueRecord) ([]domain.CatalogueRecord, error) {
577	-	if len(incompatible) == 0 {
578	-	return records, nil
579	-	}
580	-	byName := make(map[string]int, len(records))
581	-	for i := range records {
582	-	byName[records[i].Name] = i
583	-	}
584	-	for _, old := range incompatible {
585	-	fresh, committed, err := coordinator.ResetProtocolIncompatible(ctx, old.Name, old.IncarnationID)
586	-	if !committed {
587	-	if err != nil {
588	-	return nil, fmt.Errorf("reset protocol-incompatible session %q: %w", old.Name, err)
589	-	}
590	-	return nil, fmt.Errorf("reset protocol-incompatible session %q: catalogue authority changed", old.Name)
591	-	}
592	-	if index, ok := byName[old.Name]; ok {
593	-	records[index] = fresh
594	-	}
595	-	log.Info("protocol_incompatible_session_reset", "session", old.Name, "stored_incarnation", old.IncarnationID.String())
596	-	if err != nil {
597	-	log.Warn("protocol_incompatible_snapshot_cleanup_failed", "session", old.Name, "err", err)
598	-	}
599	-	}
600	-	return records, nil
601	-	}
602	-
603		func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) {	576		func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) {
604		log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode)	577		log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode)
605		}	578		}

@@ -699,12 +672,9 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
699		recoveryMode = "new-install"	672		recoveryMode = "new-install"
700		}	673		}
701		coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader)	674		coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader)
702	-	opened.Records, err = resetProtocolIncompatibleRecords(ctx, log, coordinator, opened.Records, opened.IncompatibleRecords)	675	+	if opened.Migration.Performed {
703	-	if err != nil {	676	+	recoveryMode = "catalogue-migrated"
704	-	return errors.Join(fmt.Errorf("vev: reset protocol-incompatible durable sessions: %w", err), opened.Catalogue.Close())	677	+	log.Info("catalogue_migrated", "source_formats", opened.Migration.SourceFormats, "target_format", opened.Migration.TargetFormat, "records", opened.Migration.RecordCount, "backup", opened.Migration.BackupPath)
705	-	}
706	-	if len(opened.IncompatibleRecords) != 0 {
707	-	recoveryMode = "protocol-reset"
708		}	678		}
709		logCatalogueRecovery(log, opened.Records, recoveryMode)	679		logCatalogueRecovery(log, opened.Records, recoveryMode)
710		daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))	680		daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))

M internal/persist/catalogue_test.go +43 -25


@@ -50,47 +50,46 @@ func testCatalogueRecordRoundTrip(t *testing.T) {
50	50		}
51	51		}
52	52
53		-	func TestOpenOrCreateReportsRecordsFromOlderProtocol(t *testing.T) {
	53	+	func TestOpenOrCreateMigratesLegacyRecordsWithoutProtocolReset(t *testing.T) {
54	54		record := domain.CatalogueRecord{
55	55		Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/workspace",
56	56		CreatedAt: 11, UpdatedAt: 22, LastUsedSeq: 33,
57		-	TabNames: []string{"shell", "logs"},
	57	+	TabNames: []string{"shell", "logs"}, TabRecords: []domain.CatalogueTabRecord{{StableID: "shell", Name: "shell"}, {StableID: "logs", Name: "logs"}},
58	58		Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
59	59		}
60		-	olderProtocol, err := encodeRecordValueForProtocol(record, protocol.Version-1)
61		-	require.NoError(t, err)
62		-	protocolLess, err := encodeRecordValue(record)
63		-	require.NoError(t, err)
64		-	binary.BigEndian.PutUint16(protocolLess[len(catalogueMagic):], protocolLessCatalogueRecordVersion)
65		-	protocolLess = append(protocolLess[:len(catalogueMagic)+2], protocolLess[len(catalogueMagic)+4:]...)
66		-
67		-	currentProtocol, err := encodeRecordValue(record)
68		-	require.NoError(t, err)
69	60		for _, tt := range []struct {
70		-	name string
71		-	encoded []byte
72		-	incompatible bool
	61	+	name string
	62	+	format uint16
	63	+	wire uint16
73	64		}{
74		-	{name: "current protocol", encoded: currentProtocol},
75		-	{name: "recorded older protocol", encoded: olderProtocol, incompatible: true},
76		-	{name: "protocol-less v4 record", encoded: protocolLess, incompatible: true},
	65	+	{name: "v5 older protocol", format: protocolRecordVersion, wire: protocol.Version - 1},
	66	+	{name: "v5 current protocol", format: protocolRecordVersion, wire: protocol.Version},
	67	+	{name: "v5 newer protocol", format: protocolRecordVersion, wire: protocol.Version + 1},
	68	+	{name: "v4", format: protocolLessRecordVersion},
77	69		} {
78	70		t.Run(tt.name, func(t *testing.T) {
	71	+	encoded, err := encodeRecordValueForFormat(record, tt.format, tt.wire)
	72	+	require.NoError(t, err)
79	73		dir := privateDir(t)
80	74		store, err := kv.Open(StorePath(dir))
81	75		require.NoError(t, err)
82		-	require.NoError(t, store.Set([]byte(record.Name), tt.encoded))
	76	+	require.NoError(t, store.Set([]byte(record.Name), encoded))
83	77		require.NoError(t, store.Close())
84	78
85	79		opened, err := OpenOrCreate(dir)
86	80		require.NoError(t, err)
87		-	defer func() { require.NoError(t, opened.Catalogue.Close()) }()
88	81		require.Equal(t, []domain.CatalogueRecord{record}, opened.Records)
89		-	if tt.incompatible {
90		-	require.Equal(t, []domain.CatalogueRecord{record}, opened.IncompatibleRecords)
91		-	} else {
92		-	require.Empty(t, opened.IncompatibleRecords)
93		-	}
	82	+	require.True(t, opened.Migration.Performed)
	83	+	require.Equal(t, []uint16{tt.format}, opened.Migration.SourceFormats)
	84	+	require.NoError(t, opened.Catalogue.Close())
	85	+	migrated, err := kv.Open(StorePath(dir))
	86	+	require.NoError(t, err)
	87	+	value, ok := migrated.Get([]byte(record.Name))
	88	+	require.True(t, ok)
	89	+	require.NoError(t, migrated.Close())
	90	+	stored, err := decodeStoredRecordValue(record.Name, value)
	91	+	require.NoError(t, err)
	92	+	require.Equal(t, catalogueRecordVersion, stored.formatVersion)
94	93		})
95	94		}
96	95		}

@@ -104,10 +103,29 @@ func TestDecodeRecordValueRejectsMalformedCheckpointMarker(t *testing.T) {
104	103		require.ErrorIs(t, err, errMalformedRecord)
105	104		}
106	105
	106	+	func TestDecodeRecordValueRejectsUnsupportedFormat(t *testing.T) {
	107	+	encoded, err := encodeRecordValue(validRecord("work", 1))
	108	+	require.NoError(t, err)
	109	+	binary.BigEndian.PutUint16(encoded[len(catalogueMagic):], catalogueRecordVersion+1)
	110	+
	111	+	_, err = decodeRecordValue("work", encoded)
	112	+	var unsupported *UnsupportedCatalogueRecordFormatError
	113	+	require.ErrorAs(t, err, &unsupported)
	114	+	require.Equal(t, catalogueRecordVersion+1, unsupported.Got)
	115	+	}
	116	+
107	117		func checkpointMarkerOffset(t *testing.T, encoded []byte) int {
108	118		t.Helper()
109	119		r := valueReader{data: encoded}
110		-	_, ok := r.take(len(catalogueMagic) + 2 + 2 + len(domain.IncarnationID{}))
	120	+	_, ok := r.take(len(catalogueMagic))
	121	+	require.True(t, ok)
	122	+	version, ok := r.u16()
	123	+	require.True(t, ok)
	124	+	if version == protocolRecordVersion {
	125	+	_, ok = r.u16()
	126	+	require.True(t, ok)
	127	+	}
	128	+	_, ok = r.take(len(domain.IncarnationID{}))
111	129		require.True(t, ok)
112	130		_, ok = r.str()
113	131		require.True(t, ok)

 	}
 }
-func TestOpenOrCreateReportsRecordsFromOlderProtocol(t *testing.T) {
+func TestOpenOrCreateMigratesLegacyRecordsWithoutProtocolReset(t *testing.T) {
 	record := domain.CatalogueRecord{
 		Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/workspace",
 		CreatedAt: 11, UpdatedAt: 22, LastUsedSeq: 33,
-		TabNames:  []string{"shell", "logs"},
+		TabNames: []string{"shell", "logs"}, TabRecords: []domain.CatalogueTabRecord{{StableID: "shell", Name: "shell"}, {StableID: "logs", Name: "logs"}},
 		Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
 	}
-	olderProtocol, err := encodeRecordValueForProtocol(record, protocol.Version-1)
-	require.NoError(t, err)
-	protocolLess, err := encodeRecordValue(record)
-	require.NoError(t, err)
-	binary.BigEndian.PutUint16(protocolLess[len(catalogueMagic):], protocolLessCatalogueRecordVersion)
-	protocolLess = append(protocolLess[:len(catalogueMagic)+2], protocolLess[len(catalogueMagic)+4:]...)
-	currentProtocol, err := encodeRecordValue(record)
-	require.NoError(t, err)
 	for _, tt := range []struct {
-		name         string
+		name   string
-		encoded      []byte
+		format uint16
-		incompatible bool
+		wire   uint16
 	}{
-		{name: "current protocol", encoded: currentProtocol},
+		{name: "v5 older protocol", format: protocolRecordVersion, wire: protocol.Version - 1},
-		{name: "recorded older protocol", encoded: olderProtocol, incompatible: true},
+		{name: "v5 current protocol", format: protocolRecordVersion, wire: protocol.Version},
-		{name: "protocol-less v4 record", encoded: protocolLess, incompatible: true},
+		{name: "v5 newer protocol", format: protocolRecordVersion, wire: protocol.Version + 1},
+		{name: "v4", format: protocolLessRecordVersion},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
+			encoded, err := encodeRecordValueForFormat(record, tt.format, tt.wire)
+			require.NoError(t, err)
 			dir := privateDir(t)
 			store, err := kv.Open(StorePath(dir))
 			require.NoError(t, err)
-			require.NoError(t, store.Set([]byte(record.Name), tt.encoded))
+			require.NoError(t, store.Set([]byte(record.Name), encoded))
 			require.NoError(t, store.Close())
 			opened, err := OpenOrCreate(dir)
 			require.NoError(t, err)
-			defer func() { require.NoError(t, opened.Catalogue.Close()) }()
 			require.Equal(t, []domain.CatalogueRecord{record}, opened.Records)
-			if tt.incompatible {
+			require.True(t, opened.Migration.Performed)
-				require.Equal(t, []domain.CatalogueRecord{record}, opened.IncompatibleRecords)
+			require.Equal(t, []uint16{tt.format}, opened.Migration.SourceFormats)
-			} else {
+			require.NoError(t, opened.Catalogue.Close())
-				require.Empty(t, opened.IncompatibleRecords)
+			migrated, err := kv.Open(StorePath(dir))
-			}
+			require.NoError(t, err)
+			value, ok := migrated.Get([]byte(record.Name))
+			require.True(t, ok)
+			require.NoError(t, migrated.Close())
+			stored, err := decodeStoredRecordValue(record.Name, value)
+			require.NoError(t, err)
+			require.Equal(t, catalogueRecordVersion, stored.formatVersion)
 		})
 	}
 }
 	require.ErrorIs(t, err, errMalformedRecord)
 }
+func TestDecodeRecordValueRejectsUnsupportedFormat(t *testing.T) {
+	encoded, err := encodeRecordValue(validRecord("work", 1))
+	require.NoError(t, err)
+	binary.BigEndian.PutUint16(encoded[len(catalogueMagic):], catalogueRecordVersion+1)
+	_, err = decodeRecordValue("work", encoded)
+	var unsupported *UnsupportedCatalogueRecordFormatError
+	require.ErrorAs(t, err, &unsupported)
+	require.Equal(t, catalogueRecordVersion+1, unsupported.Got)
+}
 func checkpointMarkerOffset(t *testing.T, encoded []byte) int {
 	t.Helper()
 	r := valueReader{data: encoded}
-	_, ok := r.take(len(catalogueMagic) + 2 + 2 + len(domain.IncarnationID{}))
+	_, ok := r.take(len(catalogueMagic))
+	require.True(t, ok)
+	version, ok := r.u16()
+	require.True(t, ok)
+	if version == protocolRecordVersion {
+		_, ok = r.u16()
+		require.True(t, ok)
+	}
+	_, ok = r.take(len(domain.IncarnationID{}))
 	require.True(t, ok)
 	_, ok = r.str()
 	require.True(t, ok)

A internal/persist/migration.go +180 -0

@@ -0,0 +1,180 @@
	1	+	package persist
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"errors"
	6	+	"fmt"
	7	+	"io"
	8	+	"os"
	9	+	"path/filepath"
	10	+	"sort"
	11	+
	12	+	"github.com/bnema/vev/internal/domain"
	13	+	)
	14	+
	15	+	const legacyCatalogueBackupSuffix = ".pre-v6.bak"
	16	+
	17	+	// Migration describes an automatic, lossless catalogue format migration.
	18	+	type Migration struct {
	19	+	Performed bool
	20	+	SourceFormats []uint16
	21	+	TargetFormat uint16
	22	+	RecordCount int
	23	+	BackupPath string
	24	+	}
	25	+
	26	+	func (p *Persister) migrateLegacyRecords(path string, records []domain.CatalogueRecord) (Migration, error) {
	27	+	if p == nil \|\| p.store == nil {
	28	+	return Migration{}, errPersistenceUnavailable
	29	+	}
	30	+	formats, err := p.recordFormats()
	31	+	if err != nil {
	32	+	return Migration{}, err
	33	+	}
	34	+	legacy := false
	35	+	for _, format := range formats {
	36	+	if format != catalogueRecordVersion {
	37	+	legacy = true
	38	+	break
	39	+	}
	40	+	}
	41	+	if !legacy {
	42	+	return Migration{}, nil
	43	+	}
	44	+
	45	+	raw, err := os.ReadFile(path)
	46	+	if err != nil {
	47	+	return Migration{}, fmt.Errorf("read catalogue backup source: %w", err)
	48	+	}
	49	+	backupPath := path + legacyCatalogueBackupSuffix
	50	+	if err := createMatchingBackup(backupPath, raw); err != nil {
	51	+	return Migration{}, err
	52	+	}
	53	+
	54	+	updates := make(map[string]*domain.CatalogueRecord, len(records))
	55	+	for i := range records {
	56	+	updates[records[i].Name] = &records[i]
	57	+	}
	58	+	if err := p.Apply(updates); err != nil {
	59	+	return Migration{}, fmt.Errorf("write migrated catalogue: %w", err)
	60	+	}
	61	+	return Migration{
	62	+	Performed: true,
	63	+	SourceFormats: formats,
	64	+	TargetFormat: catalogueRecordVersion,
	65	+	RecordCount: len(records),
	66	+	BackupPath: backupPath,
	67	+	}, nil
	68	+	}
	69	+
	70	+	func (p *Persister) recordFormats() ([]uint16, error) {
	71	+	formats := make(map[uint16]struct{})
	72	+	var decodeErr error
	73	+	p.store.Range(func(key, value []byte) bool {
	74	+	stored, err := decodeStoredRecordValue(string(key), value)
	75	+	if err != nil {
	76	+	decodeErr = err
	77	+	return false
	78	+	}
	79	+	formats[stored.formatVersion] = struct{}{}
	80	+	return true
	81	+	})
	82	+	if decodeErr != nil {
	83	+	return nil, decodeErr
	84	+	}
	85	+	out := make([]uint16, 0, len(formats))
	86	+	for format := range formats {
	87	+	out = append(out, format)
	88	+	}
	89	+	sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
	90	+	return out, nil
	91	+	}
	92	+
	93	+	func createMatchingBackup(path string, data []byte) error {
	94	+	matched, err := existingBackupMatches(path, data)
	95	+	if err != nil \|\| matched {
	96	+	return err
	97	+	}
	98	+
	99	+	file, err := os.CreateTemp(filepath.Dir(path), filepath.Base(path)+".tmp-*")
	100	+	if err != nil {
	101	+	return fmt.Errorf("create catalogue backup temporary file: %w", err)
	102	+	}
	103	+	tmp := file.Name()
	104	+	cleanup := func(err error) error {
	105	+	return errors.Join(err, file.Close(), os.Remove(tmp))
	106	+	}
	107	+	if err := file.Chmod(0o600); err != nil {
	108	+	return cleanup(fmt.Errorf("set catalogue backup permissions: %w", err))
	109	+	}
	110	+	if err := writeBackup(file, data); err != nil {
	111	+	return cleanup(fmt.Errorf("write catalogue backup: %w", err))
	112	+	}
	113	+	if err := file.Sync(); err != nil {
	114	+	return cleanup(fmt.Errorf("sync catalogue backup: %w", err))
	115	+	}
	116	+	if err := file.Close(); err != nil {
	117	+	_ = os.Remove(tmp)
	118	+	return fmt.Errorf("close catalogue backup: %w", err)
	119	+	}
	120	+	if err := os.Link(tmp, path); err != nil {
	121	+	_ = os.Remove(tmp)
	122	+	if errors.Is(err, os.ErrExist) {
	123	+	matched, matchErr := existingBackupMatches(path, data)
	124	+	if matchErr != nil {
	125	+	return matchErr
	126	+	}
	127	+	if !matched {
	128	+	return fmt.Errorf("persist: catalogue backup %q disappeared during migration", path)
	129	+	}
	130	+	} else {
	131	+	return fmt.Errorf("publish catalogue backup: %w", err)
	132	+	}
	133	+	} else if err := os.Remove(tmp); err != nil {
	134	+	return fmt.Errorf("remove catalogue backup temporary file: %w", err)
	135	+	}
	136	+	dir, err := os.Open(filepath.Dir(path))
	137	+	if err != nil {
	138	+	return fmt.Errorf("open catalogue backup directory: %w", err)
	139	+	}
	140	+	defer dir.Close()
	141	+	if err := dir.Sync(); err != nil {
	142	+	return fmt.Errorf("sync catalogue backup directory: %w", err)
	143	+	}
	144	+	return nil
	145	+	}
	146	+
	147	+	func existingBackupMatches(path string, data []byte) (bool, error) {
	148	+	info, err := os.Lstat(path)
	149	+	if errors.Is(err, os.ErrNotExist) {
	150	+	return false, nil
	151	+	}
	152	+	if err != nil {
	153	+	return false, fmt.Errorf("stat existing catalogue backup: %w", err)
	154	+	}
	155	+	if !info.Mode().IsRegular() \|\| info.Mode().Perm() != 0o600 {
	156	+	return false, fmt.Errorf("persist: catalogue backup %q is not a private regular file", path)
	157	+	}
	158	+	existing, err := os.ReadFile(path)
	159	+	if err != nil {
	160	+	return false, fmt.Errorf("read existing catalogue backup: %w", err)
	161	+	}
	162	+	if !bytes.Equal(existing, data) {
	163	+	return false, fmt.Errorf("persist: catalogue backup %q does not match migration source", path)
	164	+	}
	165	+	return true, nil
	166	+	}
	167	+
	168	+	func writeBackup(file *os.File, data []byte) error {
	169	+	for len(data) > 0 {
	170	+	n, err := file.Write(data)
	171	+	if err != nil {
	172	+	return err
	173	+	}
	174	+	if n == 0 {
	175	+	return io.ErrShortWrite
	176	+	}
	177	+	data = data[n:]
	178	+	}
	179	+	return nil
	180	+	}

@@ -0,0 +1,180 @@
	1	+	package persist
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"errors"
	6	+	"fmt"
	7	+	"io"
	8	+	"os"
	9	+	"path/filepath"
	10	+	"sort"
	11	+
	12	+	"github.com/bnema/vev/internal/domain"
	13	+	)
	14	+
	15	+	const legacyCatalogueBackupSuffix = ".pre-v6.bak"
	16	+
	17	+	// Migration describes an automatic, lossless catalogue format migration.
	18	+	type Migration struct {
	19	+	Performed bool
	20	+	SourceFormats []uint16
	21	+	TargetFormat uint16
	22	+	RecordCount int
	23	+	BackupPath string
	24	+	}
	25	+
	26	+	func (p *Persister) migrateLegacyRecords(path string, records []domain.CatalogueRecord) (Migration, error) {
	27	+	if p == nil \|\| p.store == nil {
	28	+	return Migration{}, errPersistenceUnavailable
	29	+	}
	30	+	formats, err := p.recordFormats()
	31	+	if err != nil {
	32	+	return Migration{}, err
	33	+	}
	34	+	legacy := false
	35	+	for _, format := range formats {
	36	+	if format != catalogueRecordVersion {
	37	+	legacy = true
	38	+	break
	39	+	}
	40	+	}
	41	+	if !legacy {
	42	+	return Migration{}, nil
	43	+	}
	44	+
	45	+	raw, err := os.ReadFile(path)
	46	+	if err != nil {
	47	+	return Migration{}, fmt.Errorf("read catalogue backup source: %w", err)
	48	+	}
	49	+	backupPath := path + legacyCatalogueBackupSuffix
	50	+	if err := createMatchingBackup(backupPath, raw); err != nil {
	51	+	return Migration{}, err
	52	+	}
	53	+
	54	+	updates := make(map[string]*domain.CatalogueRecord, len(records))
	55	+	for i := range records {
	56	+	updates[records[i].Name] = &records[i]
	57	+	}
	58	+	if err := p.Apply(updates); err != nil {
	59	+	return Migration{}, fmt.Errorf("write migrated catalogue: %w", err)
	60	+	}
	61	+	return Migration{
	62	+	Performed: true,
	63	+	SourceFormats: formats,
	64	+	TargetFormat: catalogueRecordVersion,
	65	+	RecordCount: len(records),
	66	+	BackupPath: backupPath,
	67	+	}, nil
	68	+	}
	69	+
	70	+	func (p *Persister) recordFormats() ([]uint16, error) {
	71	+	formats := make(map[uint16]struct{})
	72	+	var decodeErr error
	73	+	p.store.Range(func(key, value []byte) bool {
	74	+	stored, err := decodeStoredRecordValue(string(key), value)
	75	+	if err != nil {
	76	+	decodeErr = err
	77	+	return false
	78	+	}
	79	+	formats[stored.formatVersion] = struct{}{}
	80	+	return true
	81	+	})
	82	+	if decodeErr != nil {
	83	+	return nil, decodeErr
	84	+	}
	85	+	out := make([]uint16, 0, len(formats))
	86	+	for format := range formats {
	87	+	out = append(out, format)
	88	+	}
	89	+	sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
	90	+	return out, nil
	91	+	}
	92	+
	93	+	func createMatchingBackup(path string, data []byte) error {
	94	+	matched, err := existingBackupMatches(path, data)
	95	+	if err != nil \|\| matched {
	96	+	return err
	97	+	}
	98	+
	99	+	file, err := os.CreateTemp(filepath.Dir(path), filepath.Base(path)+".tmp-*")
	100	+	if err != nil {
	101	+	return fmt.Errorf("create catalogue backup temporary file: %w", err)
	102	+	}
	103	+	tmp := file.Name()
	104	+	cleanup := func(err error) error {
	105	+	return errors.Join(err, file.Close(), os.Remove(tmp))
	106	+	}
	107	+	if err := file.Chmod(0o600); err != nil {
	108	+	return cleanup(fmt.Errorf("set catalogue backup permissions: %w", err))
	109	+	}
	110	+	if err := writeBackup(file, data); err != nil {
	111	+	return cleanup(fmt.Errorf("write catalogue backup: %w", err))
	112	+	}
	113	+	if err := file.Sync(); err != nil {
	114	+	return cleanup(fmt.Errorf("sync catalogue backup: %w", err))
	115	+	}
	116	+	if err := file.Close(); err != nil {
	117	+	_ = os.Remove(tmp)
	118	+	return fmt.Errorf("close catalogue backup: %w", err)
	119	+	}
	120	+	if err := os.Link(tmp, path); err != nil {
	121	+	_ = os.Remove(tmp)
	122	+	if errors.Is(err, os.ErrExist) {
	123	+	matched, matchErr := existingBackupMatches(path, data)
	124	+	if matchErr != nil {
	125	+	return matchErr
	126	+	}
	127	+	if !matched {
	128	+	return fmt.Errorf("persist: catalogue backup %q disappeared during migration", path)
	129	+	}
	130	+	} else {
	131	+	return fmt.Errorf("publish catalogue backup: %w", err)
	132	+	}
	133	+	} else if err := os.Remove(tmp); err != nil {
	134	+	return fmt.Errorf("remove catalogue backup temporary file: %w", err)
	135	+	}
	136	+	dir, err := os.Open(filepath.Dir(path))
	137	+	if err != nil {
	138	+	return fmt.Errorf("open catalogue backup directory: %w", err)
	139	+	}
	140	+	defer dir.Close()
	141	+	if err := dir.Sync(); err != nil {
	142	+	return fmt.Errorf("sync catalogue backup directory: %w", err)
	143	+	}
	144	+	return nil
	145	+	}
	146	+
	147	+	func existingBackupMatches(path string, data []byte) (bool, error) {
	148	+	info, err := os.Lstat(path)
	149	+	if errors.Is(err, os.ErrNotExist) {
	150	+	return false, nil
	151	+	}
	152	+	if err != nil {
	153	+	return false, fmt.Errorf("stat existing catalogue backup: %w", err)
	154	+	}
	155	+	if !info.Mode().IsRegular() \|\| info.Mode().Perm() != 0o600 {
	156	+	return false, fmt.Errorf("persist: catalogue backup %q is not a private regular file", path)
	157	+	}
	158	+	existing, err := os.ReadFile(path)
	159	+	if err != nil {
	160	+	return false, fmt.Errorf("read existing catalogue backup: %w", err)
	161	+	}
	162	+	if !bytes.Equal(existing, data) {
	163	+	return false, fmt.Errorf("persist: catalogue backup %q does not match migration source", path)
	164	+	}
	165	+	return true, nil
	166	+	}
	167	+
	168	+	func writeBackup(file *os.File, data []byte) error {
	169	+	for len(data) > 0 {
	170	+	n, err := file.Write(data)
	171	+	if err != nil {
	172	+	return err
	173	+	}
	174	+	if n == 0 {
	175	+	return io.ErrShortWrite
	176	+	}
	177	+	data = data[n:]
	178	+	}
	179	+	return nil
	180	+	}

A internal/persist/migration_test.go +131 -0

@@ -0,0 +1,131 @@
	1	+	package persist
	2	+
	3	+	import (
	4	+	"os"
	5	+	"testing"
	6	+
	7	+	"github.com/stretchr/testify/require"
	8	+
	9	+	"github.com/bnema/vev/internal/domain"
	10	+	"github.com/bnema/vev/pkg/kv"
	11	+	)
	12	+
	13	+	const legacyRecordWireVersion = 5
	14	+
	15	+	func TestOpenOrCreateMigrationPreservesBackupAndIsIdempotent(t *testing.T) {
	16	+	t.Parallel()
	17	+	record := domain.CatalogueRecord{
	18	+	Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/workspace",
	19	+	CreatedAt: 11, UpdatedAt: 22, LastUsedSeq: 33,
	20	+	TabNames: []string{"shell"}, TabRecords: []domain.CatalogueTabRecord{{StableID: "tab-1", Name: "shell"}},
	21	+	Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
	22	+	}
	23	+	legacy, err := encodeRecordValueForFormat(record, protocolRecordVersion, legacyRecordWireVersion)
	24	+	require.NoError(t, err)
	25	+	dir := privateDir(t)
	26	+	store, err := kv.Open(StorePath(dir))
	27	+	require.NoError(t, err)
	28	+	require.NoError(t, store.Set([]byte(record.Name), legacy))
	29	+	require.NoError(t, store.Close())
	30	+	original, err := os.ReadFile(StorePath(dir))
	31	+	require.NoError(t, err)
	32	+	require.NoError(t, os.WriteFile(StorePath(dir)+legacyCatalogueBackupSuffix+".tmp", []byte("stale"), 0o600))
	33	+
	34	+	opened, err := OpenOrCreate(dir)
	35	+	require.NoError(t, err)
	36	+	require.Equal(t, []domain.CatalogueRecord{record}, opened.Records)
	37	+	require.True(t, opened.Migration.Performed)
	38	+	require.Equal(t, []uint16{protocolRecordVersion}, opened.Migration.SourceFormats)
	39	+	require.Equal(t, catalogueRecordVersion, opened.Migration.TargetFormat)
	40	+	require.Equal(t, 1, opened.Migration.RecordCount)
	41	+	require.NoError(t, opened.Catalogue.Close())
	42	+
	43	+	backup, err := os.ReadFile(StorePath(dir) + legacyCatalogueBackupSuffix)
	44	+	require.NoError(t, err)
	45	+	require.Equal(t, original, backup)
	46	+	info, err := os.Stat(StorePath(dir) + legacyCatalogueBackupSuffix)
	47	+	require.NoError(t, err)
	48	+	require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
	49	+	migrated, err := os.ReadFile(StorePath(dir))
	50	+	require.NoError(t, err)
	51	+
	52	+	reopened, err := OpenOrCreate(dir)
	53	+	require.NoError(t, err)
	54	+	require.False(t, reopened.Migration.Performed)
	55	+	require.Equal(t, []domain.CatalogueRecord{record}, reopened.Records)
	56	+	require.NoError(t, reopened.Catalogue.Close())
	57	+	after, err := os.ReadFile(StorePath(dir))
	58	+	require.NoError(t, err)
	59	+	require.Equal(t, migrated, after)
	60	+	}
	61	+
	62	+	func TestOpenOrCreateMigratesV3Records(t *testing.T) {
	63	+	t.Parallel()
	64	+	legacyRecord := domain.CatalogueRecord{
	65	+	Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/workspace",
	66	+	CreatedAt: 11, UpdatedAt: 22, LastUsedSeq: 33, TabNames: []string{"shell"},
	67	+	Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
	68	+	}
	69	+	legacy, err := encodeRecordValueForFormat(legacyRecord, legacyCatalogueRecordVersion, 0)
	70	+	require.NoError(t, err)
	71	+	dir := privateDir(t)
	72	+	store, err := kv.Open(StorePath(dir))
	73	+	require.NoError(t, err)
	74	+	require.NoError(t, store.Set([]byte(legacyRecord.Name), legacy))
	75	+	require.NoError(t, store.Close())
	76	+	original, err := os.ReadFile(StorePath(dir))
	77	+	require.NoError(t, err)
	78	+
	79	+	opened, err := OpenOrCreate(dir)
	80	+	require.NoError(t, err)
	81	+	want := legacyRecord
	82	+	want.TabRecords = []domain.CatalogueTabRecord{{Name: "shell"}}
	83	+	require.Equal(t, []domain.CatalogueRecord{want}, opened.Records)
	84	+	require.Equal(t, []uint16{legacyCatalogueRecordVersion}, opened.Migration.SourceFormats)
	85	+	require.NoError(t, opened.Catalogue.Close())
	86	+	backup, err := os.ReadFile(StorePath(dir) + legacyCatalogueBackupSuffix)
	87	+	require.NoError(t, err)
	88	+	require.Equal(t, original, backup)
	89	+	}
	90	+
	91	+	func TestOpenOrCreateMigrationRejectsConflictingBackupWithoutChangingCatalogue(t *testing.T) {
	92	+	t.Parallel()
	93	+	record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}}
	94	+	legacy, err := encodeRecordValueForFormat(record, protocolRecordVersion, legacyRecordWireVersion)
	95	+	require.NoError(t, err)
	96	+	dir := privateDir(t)
	97	+	store, err := kv.Open(StorePath(dir))
	98	+	require.NoError(t, err)
	99	+	require.NoError(t, store.Set([]byte(record.Name), legacy))
	100	+	require.NoError(t, store.Close())
	101	+	original, err := os.ReadFile(StorePath(dir))
	102	+	require.NoError(t, err)
	103	+	require.NoError(t, os.WriteFile(StorePath(dir)+legacyCatalogueBackupSuffix, []byte("different"), 0o600))
	104	+
	105	+	_, err = OpenOrCreate(dir)
	106	+	require.Error(t, err)
	107	+	after, readErr := os.ReadFile(StorePath(dir))
	108	+	require.NoError(t, readErr)
	109	+	require.Equal(t, original, after)
	110	+	}
	111	+
	112	+	func TestOpenOrCreateMigrationRejectsSymlinkBackupWithoutChangingCatalogue(t *testing.T) {
	113	+	t.Parallel()
	114	+	record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}}
	115	+	legacy, err := encodeRecordValueForFormat(record, protocolRecordVersion, legacyRecordWireVersion)
	116	+	require.NoError(t, err)
	117	+	dir := privateDir(t)
	118	+	store, err := kv.Open(StorePath(dir))
	119	+	require.NoError(t, err)
	120	+	require.NoError(t, store.Set([]byte(record.Name), legacy))
	121	+	require.NoError(t, store.Close())
	122	+	original, err := os.ReadFile(StorePath(dir))
	123	+	require.NoError(t, err)
	124	+	require.NoError(t, os.Symlink(StorePath(dir), StorePath(dir)+legacyCatalogueBackupSuffix))
	125	+
	126	+	_, err = OpenOrCreate(dir)
	127	+	require.Error(t, err)
	128	+	after, readErr := os.ReadFile(StorePath(dir))
	129	+	require.NoError(t, readErr)
	130	+	require.Equal(t, original, after)
	131	+	}

@@ -0,0 +1,131 @@
	1	+	package persist
	2	+
	3	+	import (
	4	+	"os"
	5	+	"testing"
	6	+
	7	+	"github.com/stretchr/testify/require"
	8	+
	9	+	"github.com/bnema/vev/internal/domain"
	10	+	"github.com/bnema/vev/pkg/kv"
	11	+	)
	12	+
	13	+	const legacyRecordWireVersion = 5
	14	+
	15	+	func TestOpenOrCreateMigrationPreservesBackupAndIsIdempotent(t *testing.T) {
	16	+	t.Parallel()
	17	+	record := domain.CatalogueRecord{
	18	+	Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/workspace",
	19	+	CreatedAt: 11, UpdatedAt: 22, LastUsedSeq: 33,
	20	+	TabNames: []string{"shell"}, TabRecords: []domain.CatalogueTabRecord{{StableID: "tab-1", Name: "shell"}},
	21	+	Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
	22	+	}
	23	+	legacy, err := encodeRecordValueForFormat(record, protocolRecordVersion, legacyRecordWireVersion)
	24	+	require.NoError(t, err)
	25	+	dir := privateDir(t)
	26	+	store, err := kv.Open(StorePath(dir))
	27	+	require.NoError(t, err)
	28	+	require.NoError(t, store.Set([]byte(record.Name), legacy))
	29	+	require.NoError(t, store.Close())
	30	+	original, err := os.ReadFile(StorePath(dir))
	31	+	require.NoError(t, err)
	32	+	require.NoError(t, os.WriteFile(StorePath(dir)+legacyCatalogueBackupSuffix+".tmp", []byte("stale"), 0o600))
	33	+
	34	+	opened, err := OpenOrCreate(dir)
	35	+	require.NoError(t, err)
	36	+	require.Equal(t, []domain.CatalogueRecord{record}, opened.Records)
	37	+	require.True(t, opened.Migration.Performed)
	38	+	require.Equal(t, []uint16{protocolRecordVersion}, opened.Migration.SourceFormats)
	39	+	require.Equal(t, catalogueRecordVersion, opened.Migration.TargetFormat)
	40	+	require.Equal(t, 1, opened.Migration.RecordCount)
	41	+	require.NoError(t, opened.Catalogue.Close())
	42	+
	43	+	backup, err := os.ReadFile(StorePath(dir) + legacyCatalogueBackupSuffix)
	44	+	require.NoError(t, err)
	45	+	require.Equal(t, original, backup)
	46	+	info, err := os.Stat(StorePath(dir) + legacyCatalogueBackupSuffix)
	47	+	require.NoError(t, err)
	48	+	require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
	49	+	migrated, err := os.ReadFile(StorePath(dir))
	50	+	require.NoError(t, err)
	51	+
	52	+	reopened, err := OpenOrCreate(dir)
	53	+	require.NoError(t, err)
	54	+	require.False(t, reopened.Migration.Performed)
	55	+	require.Equal(t, []domain.CatalogueRecord{record}, reopened.Records)
	56	+	require.NoError(t, reopened.Catalogue.Close())
	57	+	after, err := os.ReadFile(StorePath(dir))
	58	+	require.NoError(t, err)
	59	+	require.Equal(t, migrated, after)
	60	+	}
	61	+
	62	+	func TestOpenOrCreateMigratesV3Records(t *testing.T) {
	63	+	t.Parallel()
	64	+	legacyRecord := domain.CatalogueRecord{
	65	+	Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/workspace",
	66	+	CreatedAt: 11, UpdatedAt: 22, LastUsedSeq: 33, TabNames: []string{"shell"},
	67	+	Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
	68	+	}
	69	+	legacy, err := encodeRecordValueForFormat(legacyRecord, legacyCatalogueRecordVersion, 0)
	70	+	require.NoError(t, err)
	71	+	dir := privateDir(t)
	72	+	store, err := kv.Open(StorePath(dir))
	73	+	require.NoError(t, err)
	74	+	require.NoError(t, store.Set([]byte(legacyRecord.Name), legacy))
	75	+	require.NoError(t, store.Close())
	76	+	original, err := os.ReadFile(StorePath(dir))
	77	+	require.NoError(t, err)
	78	+
	79	+	opened, err := OpenOrCreate(dir)
	80	+	require.NoError(t, err)
	81	+	want := legacyRecord
	82	+	want.TabRecords = []domain.CatalogueTabRecord{{Name: "shell"}}
	83	+	require.Equal(t, []domain.CatalogueRecord{want}, opened.Records)
	84	+	require.Equal(t, []uint16{legacyCatalogueRecordVersion}, opened.Migration.SourceFormats)
	85	+	require.NoError(t, opened.Catalogue.Close())
	86	+	backup, err := os.ReadFile(StorePath(dir) + legacyCatalogueBackupSuffix)
	87	+	require.NoError(t, err)
	88	+	require.Equal(t, original, backup)
	89	+	}
	90	+
	91	+	func TestOpenOrCreateMigrationRejectsConflictingBackupWithoutChangingCatalogue(t *testing.T) {
	92	+	t.Parallel()
	93	+	record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}}
	94	+	legacy, err := encodeRecordValueForFormat(record, protocolRecordVersion, legacyRecordWireVersion)
	95	+	require.NoError(t, err)
	96	+	dir := privateDir(t)
	97	+	store, err := kv.Open(StorePath(dir))
	98	+	require.NoError(t, err)
	99	+	require.NoError(t, store.Set([]byte(record.Name), legacy))
	100	+	require.NoError(t, store.Close())
	101	+	original, err := os.ReadFile(StorePath(dir))
	102	+	require.NoError(t, err)
	103	+	require.NoError(t, os.WriteFile(StorePath(dir)+legacyCatalogueBackupSuffix, []byte("different"), 0o600))
	104	+
	105	+	_, err = OpenOrCreate(dir)
	106	+	require.Error(t, err)
	107	+	after, readErr := os.ReadFile(StorePath(dir))
	108	+	require.NoError(t, readErr)
	109	+	require.Equal(t, original, after)
	110	+	}
	111	+
	112	+	func TestOpenOrCreateMigrationRejectsSymlinkBackupWithoutChangingCatalogue(t *testing.T) {
	113	+	t.Parallel()
	114	+	record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}}
	115	+	legacy, err := encodeRecordValueForFormat(record, protocolRecordVersion, legacyRecordWireVersion)
	116	+	require.NoError(t, err)
	117	+	dir := privateDir(t)
	118	+	store, err := kv.Open(StorePath(dir))
	119	+	require.NoError(t, err)
	120	+	require.NoError(t, store.Set([]byte(record.Name), legacy))
	121	+	require.NoError(t, store.Close())
	122	+	original, err := os.ReadFile(StorePath(dir))
	123	+	require.NoError(t, err)
	124	+	require.NoError(t, os.Symlink(StorePath(dir), StorePath(dir)+legacyCatalogueBackupSuffix))
	125	+
	126	+	_, err = OpenOrCreate(dir)
	127	+	require.Error(t, err)
	128	+	after, readErr := os.ReadFile(StorePath(dir))
	129	+	require.NoError(t, readErr)
	130	+	require.Equal(t, original, after)
	131	+	}

M internal/persist/persist.go +6 -33


@@ -10,7 +10,6 @@ import (
10	10
11	11		"github.com/bnema/vev/internal/domain"
12	12		"github.com/bnema/vev/internal/ports"
13		-	"github.com/bnema/vev/internal/protocol"
14	13		"github.com/bnema/vev/pkg/kv"
15	14		)
16	15

@@ -29,10 +28,10 @@ var ErrCatalogueUnreadable = errors.New("persist: catalogue unreadable")
29	28
30	29		// OpenResult is the outcome of opening durable session state at startup.
31	30		type OpenResult struct {
32		-	Catalogue *Persister
33		-	Records []domain.CatalogueRecord
34		-	IncompatibleRecords []domain.CatalogueRecord
35		-	NewInstall bool
	31	+	Catalogue *Persister
	32	+	Records []domain.CatalogueRecord
	33	+	Migration Migration
	34	+	NewInstall bool
36	35		}
37	36
38	37		func StorePath(dir string) string { return filepath.Join(dir, filename) }

@@ -89,11 +88,11 @@ func OpenOrCreate(dir string) (OpenResult, error) {
89	88		if err != nil {
90	89		return OpenResult{}, fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err)
91	90		}
92		-	incompatible, err := catalogue.loadIncompatibleRecords()
	91	+	migration, err := catalogue.migrateLegacyRecords(path, records)
93	92		if err != nil {
94	93		return OpenResult{}, errors.Join(fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err), catalogue.Close())
95	94		}
96		-	return OpenResult{Catalogue: catalogue, Records: records, IncompatibleRecords: incompatible, NewInstall: !existed}, nil
	95	+	return OpenResult{Catalogue: catalogue, Records: records, Migration: migration, NewInstall: !existed}, nil
97	96		}
98	97
99	98		func New(store ports.Store) *Persister { return &Persister{store: store} }

@@ -438,32 +437,6 @@ func (p *Persister) Delete(name string) error {
438	437		}
439	438		func (p *Persister) LoadAll() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }
440	439
441		-	func (p *Persister) loadIncompatibleRecords() ([]domain.CatalogueRecord, error) {
442		-	if p == nil \|\| p.store == nil {
443		-	return nil, errPersistenceUnavailable
444		-	}
445		-	p.mu.Lock()
446		-	defer p.mu.Unlock()
447		-	if err := p.terminalLocked(); err != nil {
448		-	return nil, err
449		-	}
450		-	var records []domain.CatalogueRecord
451		-	var decodeErr error
452		-	p.store.Range(func(key, value []byte) bool {
453		-	stored, err := decodeStoredRecordValue(string(key), value)
454		-	if err != nil {
455		-	decodeErr = err
456		-	return false
457		-	}
458		-	if stored.protocolVersion != protocol.Version {
459		-	records = append(records, stored.record)
460		-	}
461		-	return true
462		-	})
463		-	sort.Slice(records, func(i, j int) bool { return records[i].Name < records[j].Name })
464		-	return records, decodeErr
465		-	}
466		-
467	440		func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) {
468	441		if p == nil {
469	442		return []domain.CatalogueRecord{}, nil

 	"github.com/bnema/vev/internal/domain"
 	"github.com/bnema/vev/internal/ports"
-	"github.com/bnema/vev/internal/protocol"
 	"github.com/bnema/vev/pkg/kv"
 )
 // OpenResult is the outcome of opening durable session state at startup.
 type OpenResult struct {
-	Catalogue           *Persister
+	Catalogue  *Persister
-	Records             []domain.CatalogueRecord
+	Records    []domain.CatalogueRecord
-	IncompatibleRecords []domain.CatalogueRecord
+	Migration  Migration
-	NewInstall          bool
+	NewInstall bool
 }
 func StorePath(dir string) string { return filepath.Join(dir, filename) }
 	if err != nil {
 		return OpenResult{}, fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err)
 	}
-	incompatible, err := catalogue.loadIncompatibleRecords()
+	migration, err := catalogue.migrateLegacyRecords(path, records)
 	if err != nil {
 		return OpenResult{}, errors.Join(fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err), catalogue.Close())
 	}
-	return OpenResult{Catalogue: catalogue, Records: records, IncompatibleRecords: incompatible, NewInstall: !existed}, nil
+	return OpenResult{Catalogue: catalogue, Records: records, Migration: migration, NewInstall: !existed}, nil
 }
 func New(store ports.Store) *Persister { return &Persister{store: store} }
 }
 func (p *Persister) LoadAll() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }
-func (p *Persister) loadIncompatibleRecords() ([]domain.CatalogueRecord, error) {
-	if p == nil || p.store == nil {
-		return nil, errPersistenceUnavailable
-	}
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	if err := p.terminalLocked(); err != nil {
-		return nil, err
-	}
-	var records []domain.CatalogueRecord
-	var decodeErr error
-	p.store.Range(func(key, value []byte) bool {
-		stored, err := decodeStoredRecordValue(string(key), value)
-		if err != nil {
-			decodeErr = err
-			return false
-		}
-		if stored.protocolVersion != protocol.Version {
-			records = append(records, stored.record)
-		}
-		return true
-	})
-	sort.Slice(records, func(i, j int) bool { return records[i].Name < records[j].Name })
-	return records, decodeErr
-}
 func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) {
 	if p == nil {
 		return []domain.CatalogueRecord{}, nil

M internal/persist/record.go +47 -29


@@ -7,34 +7,46 @@ import (
7	7		"math"
8	8
9	9		"github.com/bnema/vev/internal/domain"
10		-	"github.com/bnema/vev/internal/protocol"
11	10		)
12	11
13	12		const (
14		-	catalogueRecordVersion uint16 = 5
15		-	protocolLessCatalogueRecordVersion uint16 = 4
16		-	legacyCatalogueRecordVersion uint16 = 3
	13	+	catalogueRecordVersion uint16 = 6
	14	+	legacyCatalogueRecordVersion uint16 = 3
	15	+	protocolLessRecordVersion uint16 = 4
	16	+	protocolRecordVersion uint16 = 5
17	17		)
18	18
19	19		var errMalformedRecord = errors.New("persist: malformed catalogue record")
20	20
	21	+	// UnsupportedCatalogueRecordFormatError reports a well-formed record header
	22	+	// whose layout this binary cannot safely decode.
	23	+	type UnsupportedCatalogueRecordFormatError struct{ Got uint16 }
	24	+
	25	+	func (e *UnsupportedCatalogueRecordFormatError) Error() string {
	26	+	return fmt.Sprintf("persist: unsupported catalogue record format %d", e.Got)
	27	+	}
	28	+
21	29		var catalogueMagic = [4]byte{'V', 'E', 'V', 'C'}
22	30
23	31		// Record is the durable catalogue record used by daemon code.
24	32		type Record = domain.CatalogueRecord
25	33
26	34		func encodeRecordValue(record domain.CatalogueRecord) ([]byte, error) {
27		-	return encodeRecordValueForProtocol(record, protocol.Version)
	35	+	return encodeRecordValueForFormat(record, catalogueRecordVersion, 0)
28	36		}
29	37
30		-	func encodeRecordValueForProtocol(record domain.CatalogueRecord, protocolVersion uint16) ([]byte, error) {
	38	+	// encodeRecordValueForFormat exists to create strict historical fixtures. New
	39	+	// records must always use catalogueRecordVersion.
	40	+	func encodeRecordValueForFormat(record domain.CatalogueRecord, format, legacyProtocolVersion uint16) ([]byte, error) {
31	41		if err := record.Validate(); err != nil {
32	42		return nil, fmt.Errorf("persist: invalid catalogue record: %w", err)
33	43		}
34	44		buf := make([]byte, 0, 128)
35	45		buf = append(buf, catalogueMagic[:]...)
36		-	buf = binary.BigEndian.AppendUint16(buf, catalogueRecordVersion)
37		-	buf = binary.BigEndian.AppendUint16(buf, protocolVersion)
	46	+	buf = binary.BigEndian.AppendUint16(buf, format)
	47	+	if format == protocolRecordVersion {
	48	+	buf = binary.BigEndian.AppendUint16(buf, legacyProtocolVersion)
	49	+	}
38	50		buf = append(buf, record.IncarnationID[:]...)
39	51		var err error
40	52		buf, err = appendCheckedString(buf, record.Cwd)

@@ -54,18 +66,20 @@ func encodeRecordValueForProtocol(record domain.CatalogueRecord, protocolVersion
54	66		return nil, err
55	67		}
56	68		}
57		-	if uint64(len(record.TabRecords)) > math.MaxUint32 {
58		-	return nil, errors.New("persist: too many tab records")
59		-	}
60		-	buf = binary.BigEndian.AppendUint32(buf, uint32(len(record.TabRecords)))
61		-	for _, tab := range record.TabRecords {
62		-	buf, err = appendCheckedString(buf, string(tab.StableID))
63		-	if err != nil {
64		-	return nil, err
	69	+	if format != legacyCatalogueRecordVersion {
	70	+	if uint64(len(record.TabRecords)) > math.MaxUint32 {
	71	+	return nil, errors.New("persist: too many tab records")
65	72		}
66		-	buf, err = appendCheckedString(buf, tab.Name)
67		-	if err != nil {
68		-	return nil, err
	73	+	buf = binary.BigEndian.AppendUint32(buf, uint32(len(record.TabRecords)))
	74	+	for _, tab := range record.TabRecords {
	75	+	buf, err = appendCheckedString(buf, string(tab.StableID))
	76	+	if err != nil {
	77	+	return nil, err
	78	+	}
	79	+	buf, err = appendCheckedString(buf, tab.Name)
	80	+	if err != nil {
	81	+	return nil, err
	82	+	}
69	83		}
70	84		}
71	85		buf = appendRef(buf, record.Committed)

@@ -74,8 +88,8 @@ func encodeRecordValueForProtocol(record domain.CatalogueRecord, protocolVersion
74	88		}
75	89
76	90		type storedRecord struct {
77		-	record domain.CatalogueRecord
78		-	protocolVersion uint16
	91	+	record domain.CatalogueRecord
	92	+	formatVersion uint16
79	93		}
80	94
81	95		func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error) {

@@ -93,13 +107,16 @@ func decodeStoredRecordValue(name string, value []byte) (storedRecord, error) {
93	107		return storedRecord{}, errMalformedRecord
94	108		}
95	109		version, ok := reader.u16()
96		-	if !ok \|\| (version != catalogueRecordVersion && version != protocolLessCatalogueRecordVersion && version != legacyCatalogueRecordVersion) {
	110	+	if !ok {
97	111		return storedRecord{}, errMalformedRecord
98	112		}
99		-	var protocolVersion uint16
100		-	if version == catalogueRecordVersion {
101		-	protocolVersion, ok = reader.u16()
102		-	if !ok {
	113	+	switch version {
	114	+	case catalogueRecordVersion, protocolRecordVersion, protocolLessRecordVersion, legacyCatalogueRecordVersion:
	115	+	default:
	116	+	return storedRecord{}, &UnsupportedCatalogueRecordFormatError{Got: version}
	117	+	}
	118	+	if version == protocolRecordVersion {
	119	+	if _, ok := reader.u16(); !ok { // Legacy v5 wire protocol; deliberately ignored.
103	120		return storedRecord{}, errMalformedRecord
104	121		}
105	122		}

@@ -143,8 +160,8 @@ func decodeStoredRecordValue(name string, value []byte) (storedRecord, error) {
143	160		if version == legacyCatalogueRecordVersion {
144	161		if len(record.TabNames) != 0 {
145	162		record.TabRecords = make([]domain.CatalogueTabRecord, len(record.TabNames))
146		-	for i, name := range record.TabNames {
147		-	record.TabRecords[i] = domain.CatalogueTabRecord{Name: name}
	163	+	for i, tabName := range record.TabNames {
	164	+	record.TabRecords[i] = domain.CatalogueTabRecord{Name: tabName}
148	165		}
149	166		}
150	167		} else {

@@ -173,7 +190,7 @@ func decodeStoredRecordValue(name string, value []byte) (storedRecord, error) {
173	190		if err := record.Validate(); err != nil {
174	191		return storedRecord{}, errors.Join(errMalformedRecord, fmt.Errorf("persist: invalid catalogue record: %w", err))
175	192		}
176		-	return storedRecord{record: record, protocolVersion: protocolVersion}, nil
	193	+	return storedRecord{record: record, formatVersion: version}, nil
177	194		}
178	195
179	196		func appendCheckedString(buf []byte, value string) ([]byte, error) {

@@ -183,6 +200,7 @@ func appendCheckedString(buf []byte, value string) ([]byte, error) {
183	200		buf = binary.BigEndian.AppendUint32(buf, uint32(len(value)))
184	201		return append(buf, value...), nil
185	202		}
	203	+
186	204		func appendRef(buf []byte, ref *domain.CheckpointRef) []byte {
187	205		if ref == nil {
188	206		return append(buf, 0)

 	"math"
 	"github.com/bnema/vev/internal/domain"
-	"github.com/bnema/vev/internal/protocol"
 )
 const (
-	catalogueRecordVersion             uint16 = 5
+	catalogueRecordVersion       uint16 = 6
-	protocolLessCatalogueRecordVersion uint16 = 4
+	legacyCatalogueRecordVersion uint16 = 3
-	legacyCatalogueRecordVersion       uint16 = 3
+	protocolLessRecordVersion    uint16 = 4
+	protocolRecordVersion        uint16 = 5
 )
 var errMalformedRecord = errors.New("persist: malformed catalogue record")
+// UnsupportedCatalogueRecordFormatError reports a well-formed record header
+// whose layout this binary cannot safely decode.
+type UnsupportedCatalogueRecordFormatError struct{ Got uint16 }
+func (e *UnsupportedCatalogueRecordFormatError) Error() string {
+	return fmt.Sprintf("persist: unsupported catalogue record format %d", e.Got)
+}
 var catalogueMagic = [4]byte{'V', 'E', 'V', 'C'}
 // Record is the durable catalogue record used by daemon code.
 type Record = domain.CatalogueRecord
 func encodeRecordValue(record domain.CatalogueRecord) ([]byte, error) {
-	return encodeRecordValueForProtocol(record, protocol.Version)
+	return encodeRecordValueForFormat(record, catalogueRecordVersion, 0)
 }
-func encodeRecordValueForProtocol(record domain.CatalogueRecord, protocolVersion uint16) ([]byte, error) {
+// encodeRecordValueForFormat exists to create strict historical fixtures. New
+// records must always use catalogueRecordVersion.
+func encodeRecordValueForFormat(record domain.CatalogueRecord, format, legacyProtocolVersion uint16) ([]byte, error) {
 	if err := record.Validate(); err != nil {
 		return nil, fmt.Errorf("persist: invalid catalogue record: %w", err)
 	}
 	buf := make([]byte, 0, 128)
 	buf = append(buf, catalogueMagic[:]...)
-	buf = binary.BigEndian.AppendUint16(buf, catalogueRecordVersion)
+	buf = binary.BigEndian.AppendUint16(buf, format)
-	buf = binary.BigEndian.AppendUint16(buf, protocolVersion)
+	if format == protocolRecordVersion {
+		buf = binary.BigEndian.AppendUint16(buf, legacyProtocolVersion)
+	}
 	buf = append(buf, record.IncarnationID[:]...)
 	var err error
 	buf, err = appendCheckedString(buf, record.Cwd)
 			return nil, err
 		}
 	}
-	if uint64(len(record.TabRecords)) > math.MaxUint32 {
+	if format != legacyCatalogueRecordVersion {
-		return nil, errors.New("persist: too many tab records")
+		if uint64(len(record.TabRecords)) > math.MaxUint32 {
-	}
+			return nil, errors.New("persist: too many tab records")
-	buf = binary.BigEndian.AppendUint32(buf, uint32(len(record.TabRecords)))
-	for _, tab := range record.TabRecords {
-		buf, err = appendCheckedString(buf, string(tab.StableID))
-		if err != nil {
-			return nil, err
 		}
-		buf, err = appendCheckedString(buf, tab.Name)
+		buf = binary.BigEndian.AppendUint32(buf, uint32(len(record.TabRecords)))
-		if err != nil {
+		for _, tab := range record.TabRecords {
-			return nil, err
+			buf, err = appendCheckedString(buf, string(tab.StableID))
+			if err != nil {
+				return nil, err
+			}
+			buf, err = appendCheckedString(buf, tab.Name)
+			if err != nil {
+				return nil, err
+			}
 		}
 	}
 	buf = appendRef(buf, record.Committed)
 }
 type storedRecord struct {
-	record          domain.CatalogueRecord
+	record        domain.CatalogueRecord
-	protocolVersion uint16
+	formatVersion uint16
 }
 func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error) {
 		return storedRecord{}, errMalformedRecord
 	}
 	version, ok := reader.u16()
-	if !ok || (version != catalogueRecordVersion && version != protocolLessCatalogueRecordVersion && version != legacyCatalogueRecordVersion) {
+	if !ok {
 		return storedRecord{}, errMalformedRecord
 	}
-	var protocolVersion uint16
+	switch version {
-	if version == catalogueRecordVersion {
+	case catalogueRecordVersion, protocolRecordVersion, protocolLessRecordVersion, legacyCatalogueRecordVersion:
-		protocolVersion, ok = reader.u16()
+	default:
-		if !ok {
+		return storedRecord{}, &UnsupportedCatalogueRecordFormatError{Got: version}
+	}
+	if version == protocolRecordVersion {
+		if _, ok := reader.u16(); !ok { // Legacy v5 wire protocol; deliberately ignored.
 			return storedRecord{}, errMalformedRecord
 		}
 	}
 	if version == legacyCatalogueRecordVersion {
 		if len(record.TabNames) != 0 {
 			record.TabRecords = make([]domain.CatalogueTabRecord, len(record.TabNames))
-			for i, name := range record.TabNames {
+			for i, tabName := range record.TabNames {
-				record.TabRecords[i] = domain.CatalogueTabRecord{Name: name}
+				record.TabRecords[i] = domain.CatalogueTabRecord{Name: tabName}
 			}
 		}
 	} else {
 	if err := record.Validate(); err != nil {
 		return storedRecord{}, errors.Join(errMalformedRecord, fmt.Errorf("persist: invalid catalogue record: %w", err))
 	}
-	return storedRecord{record: record, protocolVersion: protocolVersion}, nil
+	return storedRecord{record: record, formatVersion: version}, nil
 }
 func appendCheckedString(buf []byte, value string) ([]byte, error) {
 	buf = binary.BigEndian.AppendUint32(buf, uint32(len(value)))
 	return append(buf, value...), nil
 }
 func appendRef(buf []byte, ref *domain.CheckpointRef) []byte {
 	if ref == nil {
 		return append(buf, 0)

M internal/usecase/recovery/reset.go +0 -39


@@ -8,45 +8,6 @@ import (
8	8		"github.com/bnema/vev/internal/domain"
9	9		)
10	10
11		-	// ResetProtocolIncompatible replaces one exact record with a fresh incarnation
12		-	// that retains only its name. A protocol boundary never restores metadata or
13		-	// checkpoints whose semantics may have changed.
14		-	func (c *Coordinator) ResetProtocolIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID) (domain.CatalogueRecord, bool, error) {
15		-	if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.locks == nil \|\| c.random == nil {
16		-	return domain.CatalogueRecord{}, false, errors.New("recovery: incomplete protocol-reset dependencies")
17		-	}
18		-	if err := ctx.Err(); err != nil {
19		-	return domain.CatalogueRecord{}, false, err
20		-	}
21		-	c.mutationMu.Lock()
22		-	defer c.mutationMu.Unlock()
23		-	unlock := c.locks.Lock([]string{name})
24		-	defer unlock()
25		-
26		-	record, ok, err := c.catalogue.Record(name)
27		-	if err != nil {
28		-	return domain.CatalogueRecord{}, false, err
29		-	}
30		-	if !ok \|\| record.IncarnationID != expectedIncarnation {
31		-	return domain.CatalogueRecord{}, false, nil
32		-	}
33		-	next, err := domain.NewIncarnationID(c.random)
34		-	if err != nil {
35		-	return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: generate protocol-reset incarnation: %w", err)
36		-	}
37		-	fresh := domain.CatalogueRecord{Name: record.Name, IncarnationID: next}
38		-	if err := fresh.Validate(); err != nil {
39		-	return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: invalid protocol-reset replacement: %w", err)
40		-	}
41		-	if err := c.catalogue.Replace(record.Name, fresh); err != nil {
42		-	return domain.CatalogueRecord{}, false, err
43		-	}
44		-	if err := c.repository.DeleteIncarnation(ctx, record.IncarnationID); err != nil {
45		-	return fresh, true, err
46		-	}
47		-	return fresh, true, nil
48		-	}
49		-
50	11		// ResetIncompatible replaces an exact healthy checkpoint authority with a
51	12		// fresh incarnation. Stale observations are harmless no-ops.
52	13		func (c *Coordinator) ResetIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID, expectedCheckpoint domain.CheckpointRef) (domain.CatalogueRecord, bool, error) {


@@ -8,45 +8,6 @@ import (
8		"github.com/bnema/vev/internal/domain"	8	"github.com/bnema/vev/internal/domain"
9		)	9	)
10			10
11	-	// ResetProtocolIncompatible replaces one exact record with a fresh incarnation
12	-	// that retains only its name. A protocol boundary never restores metadata or
13	-	// checkpoints whose semantics may have changed.
14	-	func (c *Coordinator) ResetProtocolIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID) (domain.CatalogueRecord, bool, error) {
15	-	if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.locks == nil \|\| c.random == nil {
16	-	return domain.CatalogueRecord{}, false, errors.New("recovery: incomplete protocol-reset dependencies")
17	-	}
18	-	if err := ctx.Err(); err != nil {
19	-	return domain.CatalogueRecord{}, false, err
20	-	}
21	-	c.mutationMu.Lock()
22	-	defer c.mutationMu.Unlock()
23	-	unlock := c.locks.Lock([]string{name})
24	-	defer unlock()
25	-
26	-	record, ok, err := c.catalogue.Record(name)
27	-	if err != nil {
28	-	return domain.CatalogueRecord{}, false, err
29	-	}
30	-	if !ok \|\| record.IncarnationID != expectedIncarnation {
31	-	return domain.CatalogueRecord{}, false, nil
32	-	}
33	-	next, err := domain.NewIncarnationID(c.random)
34	-	if err != nil {
35	-	return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: generate protocol-reset incarnation: %w", err)
36	-	}
37	-	fresh := domain.CatalogueRecord{Name: record.Name, IncarnationID: next}
38	-	if err := fresh.Validate(); err != nil {
39	-	return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: invalid protocol-reset replacement: %w", err)
40	-	}
41	-	if err := c.catalogue.Replace(record.Name, fresh); err != nil {
42	-	return domain.CatalogueRecord{}, false, err
43	-	}
44	-	if err := c.repository.DeleteIncarnation(ctx, record.IncarnationID); err != nil {
45	-	return fresh, true, err
46	-	}
47	-	return fresh, true, nil
48	-	}
49	-
50		// ResetIncompatible replaces an exact healthy checkpoint authority with a	11	// ResetIncompatible replaces an exact healthy checkpoint authority with a
51		// fresh incarnation. Stale observations are harmless no-ops.	12	// fresh incarnation. Stale observations are harmless no-ops.
52		func (c *Coordinator) ResetIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID, expectedCheckpoint domain.CheckpointRef) (domain.CatalogueRecord, bool, error) {	13	func (c *Coordinator) ResetIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID, expectedCheckpoint domain.CheckpointRef) (domain.CatalogueRecord, bool, error) {

M internal/usecase/recovery/transactions_test.go +0 -30


@@ -235,36 +235,6 @@ func healthyTransactionRecord() domain.CatalogueRecord {
235	235		}
236	236		}
237	237
238		-	func TestResetProtocolIncompatibleKeepsOnlySessionName(t *testing.T) {
239		-	t.Parallel()
240		-	old := healthyTransactionRecord()
241		-	old.TabRecords = []domain.CatalogueTabRecord{{StableID: "tab-1", Name: "shell"}}
242		-	old.DegradedReason = ""
243		-	catalogue := newTransactionCatalogue(old)
244		-	repository := &transactionRepository{}
245		-	coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
246		-
247		-	fresh, committed, err := coordinator.ResetProtocolIncompatible(context.Background(), old.Name, old.IncarnationID)
248		-
249		-	require.NoError(t, err)
250		-	require.True(t, committed)
251		-	require.Equal(t, old.Name, fresh.Name)
252		-	require.Equal(t, domain.IncarnationID(bytes.Repeat([]byte{2}, 16)), fresh.IncarnationID)
253		-	require.Empty(t, fresh.Cwd)
254		-	require.Zero(t, fresh.CreatedAt)
255		-	require.Zero(t, fresh.UpdatedAt)
256		-	require.Zero(t, fresh.LastUsedSeq)
257		-	require.Empty(t, fresh.TabNames)
258		-	require.Empty(t, fresh.TabRecords)
259		-	require.Nil(t, fresh.Committed)
260		-	require.Empty(t, fresh.DegradedReason)
261		-	stored, ok, recordErr := catalogue.Record(old.Name)
262		-	require.NoError(t, recordErr)
263		-	require.True(t, ok)
264		-	require.True(t, fresh.Equal(stored))
265		-	require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls)
266		-	}
267		-
268	238		func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) {
269	239		t.Parallel()
270	240		old := healthyTransactionRecord()


@@ -235,36 +235,6 @@ func healthyTransactionRecord() domain.CatalogueRecord {
235		}	235	}
236		}	236	}
237			237
238	-	func TestResetProtocolIncompatibleKeepsOnlySessionName(t *testing.T) {
239	-	t.Parallel()
240	-	old := healthyTransactionRecord()
241	-	old.TabRecords = []domain.CatalogueTabRecord{{StableID: "tab-1", Name: "shell"}}
242	-	old.DegradedReason = ""
243	-	catalogue := newTransactionCatalogue(old)
244	-	repository := &transactionRepository{}
245	-	coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
246	-
247	-	fresh, committed, err := coordinator.ResetProtocolIncompatible(context.Background(), old.Name, old.IncarnationID)
248	-
249	-	require.NoError(t, err)
250	-	require.True(t, committed)
251	-	require.Equal(t, old.Name, fresh.Name)
252	-	require.Equal(t, domain.IncarnationID(bytes.Repeat([]byte{2}, 16)), fresh.IncarnationID)
253	-	require.Empty(t, fresh.Cwd)
254	-	require.Zero(t, fresh.CreatedAt)
255	-	require.Zero(t, fresh.UpdatedAt)
256	-	require.Zero(t, fresh.LastUsedSeq)
257	-	require.Empty(t, fresh.TabNames)
258	-	require.Empty(t, fresh.TabRecords)
259	-	require.Nil(t, fresh.Committed)
260	-	require.Empty(t, fresh.DegradedReason)
261	-	stored, ok, recordErr := catalogue.Record(old.Name)
262	-	require.NoError(t, recordErr)
263	-	require.True(t, ok)
264	-	require.True(t, fresh.Equal(stored))
265	-	require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls)
266	-	}
267	-
268		func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) {	238	func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) {
269		t.Parallel()	239	t.Parallel()
270		old := healthyTransactionRecord()	240	old := healthyTransactionRecord()