fix(persistence): reset sessions across protocol versions
- commit
- 09015bc9b5f421390972600151fb81522dd2d0a5
- author
- brice · · 2026-08-11 12:47 UTC
- committer
- brice · · 2026-08-11 12:56 UTC
- parent
- b8a37b3
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
9 files · +251 -34
M README.md +3 -0
| @@ -41,6 +41,9 @@ vev kill <name>|--all kill sessions | |||
| 41 | 41 | ||
| 42 | 42 | The daemon starts on first use and exits after the last session. Numbered sessions survive detach; named sessions also survive daemon restarts. See [durable-session recovery](docs/durable-session-recovery.md). | |
| 43 | 43 | ||
| 44 | + | > [!WARNING] | |
| 45 | + | > A release that bumps vev's protocol version resets named sessions saved by an older protocol. Session names are retained, but layouts, tabs, terminal history, recovery transcripts, and process-recovery state are discarded. Each local or remote daemon applies this reset to its own sessions when it first starts with the new protocol. | |
| 46 | + | ||
| 44 | 47 | ## Keys and palette | |
| 45 | 48 | ||
| 46 | 49 | No prefix key; bindings use Alt. | |
| @@ -41,6 +41,9 @@ vev kill <name>|--all kill sessions | |||||
| 41 | 41 | ||||
| 42 | The daemon starts on first use and exits after the last session. Numbered sessions survive detach; named sessions also survive daemon restarts. See [durable-session recovery](docs/durable-session-recovery.md). | 42 | The daemon starts on first use and exits after the last session. Numbered sessions survive detach; named sessions also survive daemon restarts. See [durable-session recovery](docs/durable-session-recovery.md). | ||
| 43 | 43 | ||||
| 44 | + | > [!WARNING] | |||
| 45 | + | > A release that bumps vev's protocol version resets named sessions saved by an older protocol. Session names are retained, but layouts, tabs, terminal history, recovery transcripts, and process-recovery state are discarded. Each local or remote daemon applies this reset to its own sessions when it first starts with the new protocol. | |||
| 46 | + | ||||
| 44 | ## Keys and palette | 47 | ## Keys and palette | ||
| 45 | 48 | ||||
| 46 | No prefix key; bindings use Alt. | 49 | No prefix key; bindings use Alt. | ||
M docs/durable-session-recovery.md +7 -1
| @@ -12,9 +12,15 @@ 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 | |
| 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. | |
| 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. | |
| 20 | + | ||
| 15 | 21 | ## Incompatible checkpoints | |
| 16 | 22 | ||
| 17 | - | After verifying a VEVM manifest's digest, vev treats any VEVM version mismatch as an incompatible healthy checkpoint. It atomically replaces only that named session's exact healthy checkpoint with a fresh incarnation. The replacement retains the session name and working directory, but has no checkpoint, tabs, terminal history, or recovery transcript. | |
| 23 | + | After verifying a VEVM manifest's digest, vev treats any VEVM version mismatch as an incompatible healthy checkpoint. It atomically replaces only that named session's exact healthy checkpoint with a fresh incarnation. Unlike a protocol reset, this replacement retains the session name and working directory, but has no checkpoint, tabs, terminal history, or recovery transcript. | |
| 18 | 24 | ||
| 19 | 25 | Digest mismatches, corruption, validation failures, I/O errors, and ambiguous failures are never reset or purged. The session remains degraded for explicit recovery. | |
| 20 | 26 | ||
| @@ -12,9 +12,15 @@ 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 | |||
| 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. | |||
| 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. | |||
| 20 | + | ||||
| 15 | ## Incompatible checkpoints | 21 | ## Incompatible checkpoints | ||
| 16 | 22 | ||||
| 17 | - | After verifying a VEVM manifest's digest, vev treats any VEVM version mismatch as an incompatible healthy checkpoint. It atomically replaces only that named session's exact healthy checkpoint with a fresh incarnation. The replacement retains the session name and working directory, but has no checkpoint, tabs, terminal history, or recovery transcript. | 23 | + | After verifying a VEVM manifest's digest, vev treats any VEVM version mismatch as an incompatible healthy checkpoint. It atomically replaces only that named session's exact healthy checkpoint with a fresh incarnation. Unlike a protocol reset, this replacement retains the session name and working directory, but has no checkpoint, tabs, terminal history, or recovery transcript. |
| 18 | 24 | ||||
| 19 | Digest mismatches, corruption, validation failures, I/O errors, and ambiguous failures are never reset or purged. The session remains degraded for explicit recovery. | 25 | Digest mismatches, corruption, validation failures, I/O errors, and ambiguous failures are never reset or purged. The session remains degraded for explicit recovery. | ||
| 20 | 26 | ||||
M docs/remote-resilience.md +10 -8
| @@ -37,17 +37,19 @@ shared between clients. | |||
| 37 | 37 | A picker-selected remote target is a direct handoff with one input pump. Exact | |
| 38 | 38 | lifecycle/name identities are carried separately from display labels, and a | |
| 39 | 39 | successful daemon-local switch publishes a new committed identity before the | |
| 40 | - | client republishes its snapshot. These navigation actions and the strict | |
| 41 | - | handshake layout use protocol v27. Attach handshakes may carry an exact target, | |
| 42 | - | and successful Welcomes can return the daemon's committed identity. | |
| 40 | + | client republishes its snapshot. These navigation actions use the strict, | |
| 41 | + | exact-match protocol version. Attach handshakes may carry an exact target, and | |
| 42 | + | successful Welcomes can return the daemon's committed identity. | |
| 43 | 43 | ||
| 44 | 44 | ## Durable record compatibility | |
| 45 | 45 | ||
| 46 | - | The catalogue record format is currently version 4 because records now retain | |
| 47 | - | stable tab identity metadata. Readers accept version 3 for one-way upgrades, but | |
| 48 | - | older binaries reject records written at version 4. Back up the vev state | |
| 49 | - | before a rollback; downgrading after an upgrade can leave newer catalogue | |
| 50 | - | records unreadable and may require recreating those sessions. | |
| 46 | + | Catalogue record format version 5 stores the protocol version alongside stable | |
| 47 | + | tab identity metadata. At daemon startup, version 3 and 4 records and version 5 | |
| 48 | + | records written by another protocol are decoded only to retain their session | |
| 49 | + | names; all other session state is reset before socket publication. Compatible | |
| 50 | + | version 5 records restore normally. Older binaries reject version 5 records, so | |
| 51 | + | a downgrade also resets sessions when the older binary can read the catalogue, | |
| 52 | + | or fails closed when it cannot. Back up the vev state before a rollback. | |
| 51 | 53 | ||
| 52 | 54 | Remote catalogues are bounded and versioned. Complete catalogues contain ordered | |
| 53 | 55 | tab IDs, names, state, active tab, attachment state, and MRU sequence. Stopped | |
| @@ -37,17 +37,19 @@ shared between clients. | |||||
| 37 | A picker-selected remote target is a direct handoff with one input pump. Exact | 37 | A picker-selected remote target is a direct handoff with one input pump. Exact | ||
| 38 | lifecycle/name identities are carried separately from display labels, and a | 38 | lifecycle/name identities are carried separately from display labels, and a | ||
| 39 | successful daemon-local switch publishes a new committed identity before the | 39 | successful daemon-local switch publishes a new committed identity before the | ||
| 40 | - | client republishes its snapshot. These navigation actions and the strict | 40 | + | client republishes its snapshot. These navigation actions use the strict, |
| 41 | - | handshake layout use protocol v27. Attach handshakes may carry an exact target, | 41 | + | exact-match protocol version. Attach handshakes may carry an exact target, and |
| 42 | - | and successful Welcomes can return the daemon's committed identity. | 42 | + | successful Welcomes can return the daemon's committed identity. |
| 43 | 43 | ||||
| 44 | ## Durable record compatibility | 44 | ## Durable record compatibility | ||
| 45 | 45 | ||||
| 46 | - | The catalogue record format is currently version 4 because records now retain | 46 | + | Catalogue record format version 5 stores the protocol version alongside stable |
| 47 | - | stable tab identity metadata. Readers accept version 3 for one-way upgrades, but | 47 | + | tab identity metadata. At daemon startup, version 3 and 4 records and version 5 |
| 48 | - | older binaries reject records written at version 4. Back up the vev state | 48 | + | records written by another protocol are decoded only to retain their session |
| 49 | - | before a rollback; downgrading after an upgrade can leave newer catalogue | 49 | + | names; all other session state is reset before socket publication. Compatible |
| 50 | - | records unreadable and may require recreating those sessions. | 50 | + | version 5 records restore normally. Older binaries reject version 5 records, so |
| 51 | + | a downgrade also resets sessions when the older binary can read the catalogue, | |||
| 52 | + | or fails closed when it cannot. Back up the vev state before a rollback. | |||
| 51 | 53 | ||||
| 52 | Remote catalogues are bounded and versioned. Complete catalogues contain ordered | 54 | Remote catalogues are bounded and versioned. Complete catalogues contain ordered | ||
| 53 | tab IDs, names, state, active tab, attachment state, and MRU sequence. Stopped | 55 | tab IDs, names, state, active tab, attachment state, and MRU sequence. Stopped | ||
M internal/app/run.go +35 -1
| @@ -568,6 +568,33 @@ func runDaemon() (retErr error) { | |||
| 568 | 568 | }, lifecycleStartupDependencies(log)) | |
| 569 | 569 | } | |
| 570 | 570 | ||
| 571 | + | func resetProtocolIncompatibleRecords(ctx context.Context, log *slog.Logger, coordinator *recovery.Coordinator, records, incompatible []domain.CatalogueRecord) ([]domain.CatalogueRecord, error) { | |
| 572 | + | if len(incompatible) == 0 { | |
| 573 | + | return records, nil | |
| 574 | + | } | |
| 575 | + | byName := make(map[string]int, len(records)) | |
| 576 | + | for i := range records { | |
| 577 | + | byName[records[i].Name] = i | |
| 578 | + | } | |
| 579 | + | for _, old := range incompatible { | |
| 580 | + | fresh, committed, err := coordinator.ResetProtocolIncompatible(ctx, old.Name, old.IncarnationID) | |
| 581 | + | if !committed { | |
| 582 | + | if err != nil { | |
| 583 | + | return nil, fmt.Errorf("reset protocol-incompatible session %q: %w", old.Name, err) | |
| 584 | + | } | |
| 585 | + | return nil, fmt.Errorf("reset protocol-incompatible session %q: catalogue authority changed", old.Name) | |
| 586 | + | } | |
| 587 | + | if index, ok := byName[old.Name]; ok { | |
| 588 | + | records[index] = fresh | |
| 589 | + | } | |
| 590 | + | log.Info("protocol_incompatible_session_reset", "session", old.Name, "stored_incarnation", old.IncarnationID.String()) | |
| 591 | + | if err != nil { | |
| 592 | + | log.Warn("protocol_incompatible_snapshot_cleanup_failed", "session", old.Name, "err", err) | |
| 593 | + | } | |
| 594 | + | } | |
| 595 | + | return records, nil | |
| 596 | + | } | |
| 597 | + | ||
| 571 | 598 | func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) { | |
| 572 | 599 | log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode) | |
| 573 | 600 | } | |
| @@ -666,8 +693,15 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err | |||
| 666 | 693 | if opened.NewInstall { | |
| 667 | 694 | recoveryMode = "new-install" | |
| 668 | 695 | } | |
| 669 | - | logCatalogueRecovery(log, opened.Records, recoveryMode) | |
| 670 | 696 | coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader) | |
| 697 | + | opened.Records, err = resetProtocolIncompatibleRecords(ctx, log, coordinator, opened.Records, opened.IncompatibleRecords) | |
| 698 | + | if err != nil { | |
| 699 | + | return errors.Join(fmt.Errorf("vev: reset protocol-incompatible durable sessions: %w", err), opened.Catalogue.Close()) | |
| 700 | + | } | |
| 701 | + | if len(opened.IncompatibleRecords) != 0 { | |
| 702 | + | recoveryMode = "protocol-reset" | |
| 703 | + | } | |
| 704 | + | logCatalogueRecovery(log, opened.Records, recoveryMode) | |
| 671 | 705 | daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator)) | |
| 672 | 706 | daemonOpts = append(daemonOpts, daemon.WithDurableMaintenance(opened.Catalogue, snapshotRepository)) | |
| 673 | 707 | log.Info("session persistence enabled", "path", storePath) | |
| @@ -568,6 +568,33 @@ func runDaemon() (retErr error) { | |||||
| 568 | }, lifecycleStartupDependencies(log)) | 568 | }, lifecycleStartupDependencies(log)) | ||
| 569 | } | 569 | } | ||
| 570 | 570 | ||||
| 571 | + | func resetProtocolIncompatibleRecords(ctx context.Context, log *slog.Logger, coordinator *recovery.Coordinator, records, incompatible []domain.CatalogueRecord) ([]domain.CatalogueRecord, error) { | |||
| 572 | + | if len(incompatible) == 0 { | |||
| 573 | + | return records, nil | |||
| 574 | + | } | |||
| 575 | + | byName := make(map[string]int, len(records)) | |||
| 576 | + | for i := range records { | |||
| 577 | + | byName[records[i].Name] = i | |||
| 578 | + | } | |||
| 579 | + | for _, old := range incompatible { | |||
| 580 | + | fresh, committed, err := coordinator.ResetProtocolIncompatible(ctx, old.Name, old.IncarnationID) | |||
| 581 | + | if !committed { | |||
| 582 | + | if err != nil { | |||
| 583 | + | return nil, fmt.Errorf("reset protocol-incompatible session %q: %w", old.Name, err) | |||
| 584 | + | } | |||
| 585 | + | return nil, fmt.Errorf("reset protocol-incompatible session %q: catalogue authority changed", old.Name) | |||
| 586 | + | } | |||
| 587 | + | if index, ok := byName[old.Name]; ok { | |||
| 588 | + | records[index] = fresh | |||
| 589 | + | } | |||
| 590 | + | log.Info("protocol_incompatible_session_reset", "session", old.Name, "stored_incarnation", old.IncarnationID.String()) | |||
| 591 | + | if err != nil { | |||
| 592 | + | log.Warn("protocol_incompatible_snapshot_cleanup_failed", "session", old.Name, "err", err) | |||
| 593 | + | } | |||
| 594 | + | } | |||
| 595 | + | return records, nil | |||
| 596 | + | } | |||
| 597 | + | ||||
| 571 | func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) { | 598 | func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) { | ||
| 572 | log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode) | 599 | log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode) | ||
| 573 | } | 600 | } | ||
| @@ -666,8 +693,15 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err | |||||
| 666 | if opened.NewInstall { | 693 | if opened.NewInstall { | ||
| 667 | recoveryMode = "new-install" | 694 | recoveryMode = "new-install" | ||
| 668 | } | 695 | } | ||
| 669 | - | logCatalogueRecovery(log, opened.Records, recoveryMode) | |||
| 670 | coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader) | 696 | coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader) | ||
| 697 | + | opened.Records, err = resetProtocolIncompatibleRecords(ctx, log, coordinator, opened.Records, opened.IncompatibleRecords) | |||
| 698 | + | if err != nil { | |||
| 699 | + | return errors.Join(fmt.Errorf("vev: reset protocol-incompatible durable sessions: %w", err), opened.Catalogue.Close()) | |||
| 700 | + | } | |||
| 701 | + | if len(opened.IncompatibleRecords) != 0 { | |||
| 702 | + | recoveryMode = "protocol-reset" | |||
| 703 | + | } | |||
| 704 | + | logCatalogueRecovery(log, opened.Records, recoveryMode) | |||
| 671 | daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator)) | 705 | daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator)) | ||
| 672 | daemonOpts = append(daemonOpts, daemon.WithDurableMaintenance(opened.Catalogue, snapshotRepository)) | 706 | daemonOpts = append(daemonOpts, daemon.WithDurableMaintenance(opened.Catalogue, snapshotRepository)) | ||
| 673 | log.Info("session persistence enabled", "path", storePath) | 707 | log.Info("session persistence enabled", "path", storePath) | ||
M internal/persist/catalogue_test.go +48 -1
| @@ -1,6 +1,7 @@ | |||
| 1 | 1 | package persist | |
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | + | "encoding/binary" | |
| 4 | 5 | "os" | |
| 5 | 6 | "path/filepath" | |
| 6 | 7 | "testing" | |
| @@ -8,6 +9,7 @@ import ( | |||
| 8 | 9 | "github.com/stretchr/testify/require" | |
| 9 | 10 | ||
| 10 | 11 | "github.com/bnema/vev/internal/domain" | |
| 12 | + | "github.com/bnema/vev/internal/ports" | |
| 11 | 13 | "github.com/bnema/vev/pkg/kv" | |
| 12 | 14 | "github.com/bnema/vev/pkg/safedir" | |
| 13 | 15 | ) | |
| @@ -48,6 +50,51 @@ func testCatalogueRecordRoundTrip(t *testing.T) { | |||
| 48 | 50 | } | |
| 49 | 51 | } | |
| 50 | 52 | ||
| 53 | + | func TestOpenOrCreateReportsRecordsFromOlderProtocol(t *testing.T) { | |
| 54 | + | record := domain.CatalogueRecord{ | |
| 55 | + | Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/workspace", | |
| 56 | + | CreatedAt: 11, UpdatedAt: 22, LastUsedSeq: 33, | |
| 57 | + | TabNames: []string{"shell", "logs"}, | |
| 58 | + | Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}}, | |
| 59 | + | } | |
| 60 | + | olderProtocol, err := encodeRecordValueForProtocol(record, ports.ProtocolVersion-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 | + | for _, tt := range []struct { | |
| 70 | + | name string | |
| 71 | + | encoded []byte | |
| 72 | + | incompatible bool | |
| 73 | + | }{ | |
| 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}, | |
| 77 | + | } { | |
| 78 | + | t.Run(tt.name, func(t *testing.T) { | |
| 79 | + | dir := privateDir(t) | |
| 80 | + | store, err := kv.Open(StorePath(dir)) | |
| 81 | + | require.NoError(t, err) | |
| 82 | + | require.NoError(t, store.Set([]byte(record.Name), tt.encoded)) | |
| 83 | + | require.NoError(t, store.Close()) | |
| 84 | + | ||
| 85 | + | opened, err := OpenOrCreate(dir) | |
| 86 | + | require.NoError(t, err) | |
| 87 | + | defer func() { require.NoError(t, opened.Catalogue.Close()) }() | |
| 88 | + | 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 | + | } | |
| 94 | + | }) | |
| 95 | + | } | |
| 96 | + | } | |
| 97 | + | ||
| 51 | 98 | func TestDecodeRecordValueRejectsMalformedCheckpointMarker(t *testing.T) { | |
| 52 | 99 | encoded, err := encodeRecordValue(validRecord("work", 1)) | |
| 53 | 100 | require.NoError(t, err) | |
| @@ -60,7 +107,7 @@ func TestDecodeRecordValueRejectsMalformedCheckpointMarker(t *testing.T) { | |||
| 60 | 107 | func checkpointMarkerOffset(t *testing.T, encoded []byte) int { | |
| 61 | 108 | t.Helper() | |
| 62 | 109 | r := valueReader{data: encoded} | |
| 63 | - | _, ok := r.take(len(catalogueMagic) + 2 + len(domain.IncarnationID{})) | |
| 110 | + | _, ok := r.take(len(catalogueMagic) + 2 + 2 + len(domain.IncarnationID{})) | |
| 64 | 111 | require.True(t, ok) | |
| 65 | 112 | _, ok = r.str() | |
| 66 | 113 | require.True(t, ok) | |
| @@ -1,6 +1,7 @@ | |||||
| 1 | package persist | 1 | package persist | ||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | + | "encoding/binary" | |||
| 4 | "os" | 5 | "os" | ||
| 5 | "path/filepath" | 6 | "path/filepath" | ||
| 6 | "testing" | 7 | "testing" | ||
| @@ -8,6 +9,7 @@ import ( | |||||
| 8 | "github.com/stretchr/testify/require" | 9 | "github.com/stretchr/testify/require" | ||
| 9 | 10 | ||||
| 10 | "github.com/bnema/vev/internal/domain" | 11 | "github.com/bnema/vev/internal/domain" | ||
| 12 | + | "github.com/bnema/vev/internal/ports" | |||
| 11 | "github.com/bnema/vev/pkg/kv" | 13 | "github.com/bnema/vev/pkg/kv" | ||
| 12 | "github.com/bnema/vev/pkg/safedir" | 14 | "github.com/bnema/vev/pkg/safedir" | ||
| 13 | ) | 15 | ) | ||
| @@ -48,6 +50,51 @@ func testCatalogueRecordRoundTrip(t *testing.T) { | |||||
| 48 | } | 50 | } | ||
| 49 | } | 51 | } | ||
| 50 | 52 | ||||
| 53 | + | func TestOpenOrCreateReportsRecordsFromOlderProtocol(t *testing.T) { | |||
| 54 | + | record := domain.CatalogueRecord{ | |||
| 55 | + | Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/workspace", | |||
| 56 | + | CreatedAt: 11, UpdatedAt: 22, LastUsedSeq: 33, | |||
| 57 | + | TabNames: []string{"shell", "logs"}, | |||
| 58 | + | Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}}, | |||
| 59 | + | } | |||
| 60 | + | olderProtocol, err := encodeRecordValueForProtocol(record, ports.ProtocolVersion-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 | + | for _, tt := range []struct { | |||
| 70 | + | name string | |||
| 71 | + | encoded []byte | |||
| 72 | + | incompatible bool | |||
| 73 | + | }{ | |||
| 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}, | |||
| 77 | + | } { | |||
| 78 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 79 | + | dir := privateDir(t) | |||
| 80 | + | store, err := kv.Open(StorePath(dir)) | |||
| 81 | + | require.NoError(t, err) | |||
| 82 | + | require.NoError(t, store.Set([]byte(record.Name), tt.encoded)) | |||
| 83 | + | require.NoError(t, store.Close()) | |||
| 84 | + | ||||
| 85 | + | opened, err := OpenOrCreate(dir) | |||
| 86 | + | require.NoError(t, err) | |||
| 87 | + | defer func() { require.NoError(t, opened.Catalogue.Close()) }() | |||
| 88 | + | 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 | + | } | |||
| 94 | + | }) | |||
| 95 | + | } | |||
| 96 | + | } | |||
| 97 | + | ||||
| 51 | func TestDecodeRecordValueRejectsMalformedCheckpointMarker(t *testing.T) { | 98 | func TestDecodeRecordValueRejectsMalformedCheckpointMarker(t *testing.T) { | ||
| 52 | encoded, err := encodeRecordValue(validRecord("work", 1)) | 99 | encoded, err := encodeRecordValue(validRecord("work", 1)) | ||
| 53 | require.NoError(t, err) | 100 | require.NoError(t, err) | ||
| @@ -60,7 +107,7 @@ func TestDecodeRecordValueRejectsMalformedCheckpointMarker(t *testing.T) { | |||||
| 60 | func checkpointMarkerOffset(t *testing.T, encoded []byte) int { | 107 | func checkpointMarkerOffset(t *testing.T, encoded []byte) int { | ||
| 61 | t.Helper() | 108 | t.Helper() | ||
| 62 | r := valueReader{data: encoded} | 109 | r := valueReader{data: encoded} | ||
| 63 | - | _, ok := r.take(len(catalogueMagic) + 2 + len(domain.IncarnationID{})) | 110 | + | _, ok := r.take(len(catalogueMagic) + 2 + 2 + len(domain.IncarnationID{})) |
| 64 | require.True(t, ok) | 111 | require.True(t, ok) | ||
| 65 | _, ok = r.str() | 112 | _, ok = r.str() | ||
| 66 | require.True(t, ok) | 113 | require.True(t, ok) | ||
M internal/persist/persist.go +36 -4
| @@ -28,9 +28,10 @@ var ErrCatalogueUnreadable = errors.New("persist: catalogue unreadable") | |||
| 28 | 28 | ||
| 29 | 29 | // OpenResult is the outcome of opening durable session state at startup. | |
| 30 | 30 | type OpenResult struct { | |
| 31 | - | Catalogue *Persister | |
| 32 | - | Records []domain.CatalogueRecord | |
| 33 | - | NewInstall bool | |
| 31 | + | Catalogue *Persister | |
| 32 | + | Records []domain.CatalogueRecord | |
| 33 | + | IncompatibleRecords []domain.CatalogueRecord | |
| 34 | + | NewInstall bool | |
| 34 | 35 | } | |
| 35 | 36 | ||
| 36 | 37 | func StorePath(dir string) string { return filepath.Join(dir, filename) } | |
| @@ -87,7 +88,11 @@ func OpenOrCreate(dir string) (OpenResult, error) { | |||
| 87 | 88 | if err != nil { | |
| 88 | 89 | return OpenResult{}, fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err) | |
| 89 | 90 | } | |
| 90 | - | return OpenResult{Catalogue: catalogue, Records: records, NewInstall: !existed}, nil | |
| 91 | + | incompatible, err := catalogue.loadIncompatibleRecords() | |
| 92 | + | if err != nil { | |
| 93 | + | return OpenResult{}, errors.Join(fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err), catalogue.Close()) | |
| 94 | + | } | |
| 95 | + | return OpenResult{Catalogue: catalogue, Records: records, IncompatibleRecords: incompatible, NewInstall: !existed}, nil | |
| 91 | 96 | } | |
| 92 | 97 | ||
| 93 | 98 | func New(store ports.Store) *Persister { return &Persister{store: store} } | |
| @@ -431,6 +436,33 @@ func (p *Persister) Delete(name string) error { | |||
| 431 | 436 | return p.Apply(map[string]*domain.CatalogueRecord{name: nil}) | |
| 432 | 437 | } | |
| 433 | 438 | func (p *Persister) LoadAll() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() } | |
| 439 | + | ||
| 440 | + | func (p *Persister) loadIncompatibleRecords() ([]domain.CatalogueRecord, error) { | |
| 441 | + | if p == nil || p.store == nil { | |
| 442 | + | return nil, errPersistenceUnavailable | |
| 443 | + | } | |
| 444 | + | p.mu.Lock() | |
| 445 | + | defer p.mu.Unlock() | |
| 446 | + | if err := p.terminalLocked(); err != nil { | |
| 447 | + | return nil, err | |
| 448 | + | } | |
| 449 | + | var records []domain.CatalogueRecord | |
| 450 | + | var decodeErr error | |
| 451 | + | p.store.Range(func(key, value []byte) bool { | |
| 452 | + | stored, err := decodeStoredRecordValue(string(key), value) | |
| 453 | + | if err != nil { | |
| 454 | + | decodeErr = err | |
| 455 | + | return false | |
| 456 | + | } | |
| 457 | + | if stored.protocolVersion != ports.ProtocolVersion { | |
| 458 | + | records = append(records, stored.record) | |
| 459 | + | } | |
| 460 | + | return true | |
| 461 | + | }) | |
| 462 | + | sort.Slice(records, func(i, j int) bool { return records[i].Name < records[j].Name }) | |
| 463 | + | return records, decodeErr | |
| 464 | + | } | |
| 465 | + | ||
| 434 | 466 | func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) { | |
| 435 | 467 | if p == nil { | |
| 436 | 468 | return []domain.CatalogueRecord{}, nil | |
| @@ -28,9 +28,10 @@ var ErrCatalogueUnreadable = errors.New("persist: catalogue unreadable") | |||||
| 28 | 28 | ||||
| 29 | // OpenResult is the outcome of opening durable session state at startup. | 29 | // OpenResult is the outcome of opening durable session state at startup. | ||
| 30 | type OpenResult struct { | 30 | type OpenResult struct { | ||
| 31 | - | Catalogue *Persister | 31 | + | Catalogue *Persister |
| 32 | - | Records []domain.CatalogueRecord | 32 | + | Records []domain.CatalogueRecord |
| 33 | - | NewInstall bool | 33 | + | IncompatibleRecords []domain.CatalogueRecord |
| 34 | + | NewInstall bool | |||
| 34 | } | 35 | } | ||
| 35 | 36 | ||||
| 36 | func StorePath(dir string) string { return filepath.Join(dir, filename) } | 37 | func StorePath(dir string) string { return filepath.Join(dir, filename) } | ||
| @@ -87,7 +88,11 @@ func OpenOrCreate(dir string) (OpenResult, error) { | |||||
| 87 | if err != nil { | 88 | if err != nil { | ||
| 88 | return OpenResult{}, fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err) | 89 | return OpenResult{}, fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err) | ||
| 89 | } | 90 | } | ||
| 90 | - | return OpenResult{Catalogue: catalogue, Records: records, NewInstall: !existed}, nil | 91 | + | incompatible, err := catalogue.loadIncompatibleRecords() |
| 92 | + | if err != nil { | |||
| 93 | + | return OpenResult{}, errors.Join(fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err), catalogue.Close()) | |||
| 94 | + | } | |||
| 95 | + | return OpenResult{Catalogue: catalogue, Records: records, IncompatibleRecords: incompatible, NewInstall: !existed}, nil | |||
| 91 | } | 96 | } | ||
| 92 | 97 | ||||
| 93 | func New(store ports.Store) *Persister { return &Persister{store: store} } | 98 | func New(store ports.Store) *Persister { return &Persister{store: store} } | ||
| @@ -431,6 +436,33 @@ func (p *Persister) Delete(name string) error { | |||||
| 431 | return p.Apply(map[string]*domain.CatalogueRecord{name: nil}) | 436 | return p.Apply(map[string]*domain.CatalogueRecord{name: nil}) | ||
| 432 | } | 437 | } | ||
| 433 | func (p *Persister) LoadAll() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() } | 438 | func (p *Persister) LoadAll() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() } | ||
| 439 | + | ||||
| 440 | + | func (p *Persister) loadIncompatibleRecords() ([]domain.CatalogueRecord, error) { | |||
| 441 | + | if p == nil || p.store == nil { | |||
| 442 | + | return nil, errPersistenceUnavailable | |||
| 443 | + | } | |||
| 444 | + | p.mu.Lock() | |||
| 445 | + | defer p.mu.Unlock() | |||
| 446 | + | if err := p.terminalLocked(); err != nil { | |||
| 447 | + | return nil, err | |||
| 448 | + | } | |||
| 449 | + | var records []domain.CatalogueRecord | |||
| 450 | + | var decodeErr error | |||
| 451 | + | p.store.Range(func(key, value []byte) bool { | |||
| 452 | + | stored, err := decodeStoredRecordValue(string(key), value) | |||
| 453 | + | if err != nil { | |||
| 454 | + | decodeErr = err | |||
| 455 | + | return false | |||
| 456 | + | } | |||
| 457 | + | if stored.protocolVersion != ports.ProtocolVersion { | |||
| 458 | + | records = append(records, stored.record) | |||
| 459 | + | } | |||
| 460 | + | return true | |||
| 461 | + | }) | |||
| 462 | + | sort.Slice(records, func(i, j int) bool { return records[i].Name < records[j].Name }) | |||
| 463 | + | return records, decodeErr | |||
| 464 | + | } | |||
| 465 | + | ||||
| 434 | func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) { | 466 | func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) { | ||
| 435 | if p == nil { | 467 | if p == nil { | ||
| 436 | return []domain.CatalogueRecord{}, nil | 468 | return []domain.CatalogueRecord{}, nil | ||
M internal/persist/record.go +43 -19
| @@ -7,11 +7,13 @@ import ( | |||
| 7 | 7 | "math" | |
| 8 | 8 | ||
| 9 | 9 | "github.com/bnema/vev/internal/domain" | |
| 10 | + | "github.com/bnema/vev/internal/ports" | |
| 10 | 11 | ) | |
| 11 | 12 | ||
| 12 | 13 | const ( | |
| 13 | - | catalogueRecordVersion uint16 = 4 | |
| 14 | - | legacyCatalogueRecordVersion uint16 = 3 | |
| 14 | + | catalogueRecordVersion uint16 = 5 | |
| 15 | + | protocolLessCatalogueRecordVersion uint16 = 4 | |
| 16 | + | legacyCatalogueRecordVersion uint16 = 3 | |
| 15 | 17 | ) | |
| 16 | 18 | ||
| 17 | 19 | var errMalformedRecord = errors.New("persist: malformed catalogue record") | |
| @@ -22,12 +24,17 @@ var catalogueMagic = [4]byte{'V', 'E', 'V', 'C'} | |||
| 22 | 24 | type Record = domain.CatalogueRecord | |
| 23 | 25 | ||
| 24 | 26 | func encodeRecordValue(record domain.CatalogueRecord) ([]byte, error) { | |
| 27 | + | return encodeRecordValueForProtocol(record, ports.ProtocolVersion) | |
| 28 | + | } | |
| 29 | + | ||
| 30 | + | func encodeRecordValueForProtocol(record domain.CatalogueRecord, protocolVersion uint16) ([]byte, error) { | |
| 25 | 31 | if err := record.Validate(); err != nil { | |
| 26 | 32 | return nil, fmt.Errorf("persist: invalid catalogue record: %w", err) | |
| 27 | 33 | } | |
| 28 | 34 | buf := make([]byte, 0, 128) | |
| 29 | 35 | buf = append(buf, catalogueMagic[:]...) | |
| 30 | 36 | buf = binary.BigEndian.AppendUint16(buf, catalogueRecordVersion) | |
| 37 | + | buf = binary.BigEndian.AppendUint16(buf, protocolVersion) | |
| 31 | 38 | buf = append(buf, record.IncarnationID[:]...) | |
| 32 | 39 | var err error | |
| 33 | 40 | buf, err = appendCheckedString(buf, record.Cwd) | |
| @@ -66,45 +73,62 @@ func encodeRecordValue(record domain.CatalogueRecord) ([]byte, error) { | |||
| 66 | 73 | return buf, err | |
| 67 | 74 | } | |
| 68 | 75 | ||
| 76 | + | type storedRecord struct { | |
| 77 | + | record domain.CatalogueRecord | |
| 78 | + | protocolVersion uint16 | |
| 79 | + | } | |
| 80 | + | ||
| 69 | 81 | func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error) { | |
| 82 | + | stored, err := decodeStoredRecordValue(name, value) | |
| 83 | + | return stored.record, err | |
| 84 | + | } | |
| 85 | + | ||
| 86 | + | func decodeStoredRecordValue(name string, value []byte) (storedRecord, error) { | |
| 70 | 87 | if err := domain.ValidateSessionName(name); err != nil { | |
| 71 | - | return domain.CatalogueRecord{}, err | |
| 88 | + | return storedRecord{}, err | |
| 72 | 89 | } | |
| 73 | 90 | reader := valueReader{data: value} | |
| 74 | 91 | magic, ok := reader.take(4) | |
| 75 | 92 | if !ok || string(magic) != string(catalogueMagic[:]) { | |
| 76 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 93 | + | return storedRecord{}, errMalformedRecord | |
| 77 | 94 | } | |
| 78 | 95 | version, ok := reader.u16() | |
| 79 | - | if !ok || (version != catalogueRecordVersion && version != legacyCatalogueRecordVersion) { | |
| 80 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 96 | + | if !ok || (version != catalogueRecordVersion && version != protocolLessCatalogueRecordVersion && version != legacyCatalogueRecordVersion) { | |
| 97 | + | return storedRecord{}, errMalformedRecord | |
| 98 | + | } | |
| 99 | + | var protocolVersion uint16 | |
| 100 | + | if version == catalogueRecordVersion { | |
| 101 | + | protocolVersion, ok = reader.u16() | |
| 102 | + | if !ok { | |
| 103 | + | return storedRecord{}, errMalformedRecord | |
| 104 | + | } | |
| 81 | 105 | } | |
| 82 | 106 | id, ok := reader.take(16) | |
| 83 | 107 | if !ok { | |
| 84 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 108 | + | return storedRecord{}, errMalformedRecord | |
| 85 | 109 | } | |
| 86 | 110 | record := domain.CatalogueRecord{Name: name} | |
| 87 | 111 | copy(record.IncarnationID[:], id) | |
| 88 | 112 | var valid bool | |
| 89 | 113 | if record.Cwd, valid = reader.str(); !valid { | |
| 90 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 114 | + | return storedRecord{}, errMalformedRecord | |
| 91 | 115 | } | |
| 92 | 116 | created, ok := reader.u64() | |
| 93 | 117 | if !ok { | |
| 94 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 118 | + | return storedRecord{}, errMalformedRecord | |
| 95 | 119 | } | |
| 96 | 120 | record.CreatedAt = int64(created) | |
| 97 | 121 | updated, ok := reader.u64() | |
| 98 | 122 | if !ok { | |
| 99 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 123 | + | return storedRecord{}, errMalformedRecord | |
| 100 | 124 | } | |
| 101 | 125 | record.UpdatedAt = int64(updated) | |
| 102 | 126 | if record.LastUsedSeq, ok = reader.u64(); !ok { | |
| 103 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 127 | + | return storedRecord{}, errMalformedRecord | |
| 104 | 128 | } | |
| 105 | 129 | count, ok := reader.u32() | |
| 106 | 130 | if !ok || uint64(count) > uint64(reader.remaining())/4 { | |
| 107 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 131 | + | return storedRecord{}, errMalformedRecord | |
| 108 | 132 | } | |
| 109 | 133 | if count > 0 { | |
| 110 | 134 | record.TabNames = make([]string, 0, int(count)) | |
| @@ -112,7 +136,7 @@ func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error | |||
| 112 | 136 | for range count { | |
| 113 | 137 | tab, ok := reader.str() | |
| 114 | 138 | if !ok { | |
| 115 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 139 | + | return storedRecord{}, errMalformedRecord | |
| 116 | 140 | } | |
| 117 | 141 | record.TabNames = append(record.TabNames, tab) | |
| 118 | 142 | } | |
| @@ -126,7 +150,7 @@ func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error | |||
| 126 | 150 | } else { | |
| 127 | 151 | tabRecordCount, ok := reader.u32() | |
| 128 | 152 | if !ok || uint64(tabRecordCount) > uint64(reader.remaining())/8 { | |
| 129 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 153 | + | return storedRecord{}, errMalformedRecord | |
| 130 | 154 | } | |
| 131 | 155 | if tabRecordCount > 0 { | |
| 132 | 156 | record.TabRecords = make([]domain.CatalogueTabRecord, 0, int(tabRecordCount)) | |
| @@ -135,21 +159,21 @@ func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error | |||
| 135 | 159 | stableID, stableOK := reader.str() | |
| 136 | 160 | name, nameOK := reader.str() | |
| 137 | 161 | if !stableOK || !nameOK { | |
| 138 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 162 | + | return storedRecord{}, errMalformedRecord | |
| 139 | 163 | } | |
| 140 | 164 | record.TabRecords = append(record.TabRecords, domain.CatalogueTabRecord{StableID: domain.TabStableID(stableID), Name: name}) | |
| 141 | 165 | } | |
| 142 | 166 | } | |
| 143 | 167 | if record.Committed, ok = reader.ref(); !ok { | |
| 144 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 168 | + | return storedRecord{}, errMalformedRecord | |
| 145 | 169 | } | |
| 146 | 170 | if record.DegradedReason, ok = reader.str(); !ok || reader.remaining() != 0 { | |
| 147 | - | return domain.CatalogueRecord{}, errMalformedRecord | |
| 171 | + | return storedRecord{}, errMalformedRecord | |
| 148 | 172 | } | |
| 149 | 173 | if err := record.Validate(); err != nil { | |
| 150 | - | return domain.CatalogueRecord{}, errors.Join(errMalformedRecord, fmt.Errorf("persist: invalid catalogue record: %w", err)) | |
| 174 | + | return storedRecord{}, errors.Join(errMalformedRecord, fmt.Errorf("persist: invalid catalogue record: %w", err)) | |
| 151 | 175 | } | |
| 152 | - | return record, nil | |
| 176 | + | return storedRecord{record: record, protocolVersion: protocolVersion}, nil | |
| 153 | 177 | } | |
| 154 | 178 | ||
| 155 | 179 | func appendCheckedString(buf []byte, value string) ([]byte, error) { | |
| @@ -7,11 +7,13 @@ import ( | |||||
| 7 | "math" | 7 | "math" | ||
| 8 | 8 | ||||
| 9 | "github.com/bnema/vev/internal/domain" | 9 | "github.com/bnema/vev/internal/domain" | ||
| 10 | + | "github.com/bnema/vev/internal/ports" | |||
| 10 | ) | 11 | ) | ||
| 11 | 12 | ||||
| 12 | const ( | 13 | const ( | ||
| 13 | - | catalogueRecordVersion uint16 = 4 | 14 | + | catalogueRecordVersion uint16 = 5 |
| 14 | - | legacyCatalogueRecordVersion uint16 = 3 | 15 | + | protocolLessCatalogueRecordVersion uint16 = 4 |
| 16 | + | legacyCatalogueRecordVersion uint16 = 3 | |||
| 15 | ) | 17 | ) | ||
| 16 | 18 | ||||
| 17 | var errMalformedRecord = errors.New("persist: malformed catalogue record") | 19 | var errMalformedRecord = errors.New("persist: malformed catalogue record") | ||
| @@ -22,12 +24,17 @@ var catalogueMagic = [4]byte{'V', 'E', 'V', 'C'} | |||||
| 22 | type Record = domain.CatalogueRecord | 24 | type Record = domain.CatalogueRecord | ||
| 23 | 25 | ||||
| 24 | func encodeRecordValue(record domain.CatalogueRecord) ([]byte, error) { | 26 | func encodeRecordValue(record domain.CatalogueRecord) ([]byte, error) { | ||
| 27 | + | return encodeRecordValueForProtocol(record, ports.ProtocolVersion) | |||
| 28 | + | } | |||
| 29 | + | ||||
| 30 | + | func encodeRecordValueForProtocol(record domain.CatalogueRecord, protocolVersion uint16) ([]byte, error) { | |||
| 25 | if err := record.Validate(); err != nil { | 31 | if err := record.Validate(); err != nil { | ||
| 26 | return nil, fmt.Errorf("persist: invalid catalogue record: %w", err) | 32 | return nil, fmt.Errorf("persist: invalid catalogue record: %w", err) | ||
| 27 | } | 33 | } | ||
| 28 | buf := make([]byte, 0, 128) | 34 | buf := make([]byte, 0, 128) | ||
| 29 | buf = append(buf, catalogueMagic[:]...) | 35 | buf = append(buf, catalogueMagic[:]...) | ||
| 30 | buf = binary.BigEndian.AppendUint16(buf, catalogueRecordVersion) | 36 | buf = binary.BigEndian.AppendUint16(buf, catalogueRecordVersion) | ||
| 37 | + | buf = binary.BigEndian.AppendUint16(buf, protocolVersion) | |||
| 31 | buf = append(buf, record.IncarnationID[:]...) | 38 | buf = append(buf, record.IncarnationID[:]...) | ||
| 32 | var err error | 39 | var err error | ||
| 33 | buf, err = appendCheckedString(buf, record.Cwd) | 40 | buf, err = appendCheckedString(buf, record.Cwd) | ||
| @@ -66,45 +73,62 @@ func encodeRecordValue(record domain.CatalogueRecord) ([]byte, error) { | |||||
| 66 | return buf, err | 73 | return buf, err | ||
| 67 | } | 74 | } | ||
| 68 | 75 | ||||
| 76 | + | type storedRecord struct { | |||
| 77 | + | record domain.CatalogueRecord | |||
| 78 | + | protocolVersion uint16 | |||
| 79 | + | } | |||
| 80 | + | ||||
| 69 | func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error) { | 81 | func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error) { | ||
| 82 | + | stored, err := decodeStoredRecordValue(name, value) | |||
| 83 | + | return stored.record, err | |||
| 84 | + | } | |||
| 85 | + | ||||
| 86 | + | func decodeStoredRecordValue(name string, value []byte) (storedRecord, error) { | |||
| 70 | if err := domain.ValidateSessionName(name); err != nil { | 87 | if err := domain.ValidateSessionName(name); err != nil { | ||
| 71 | - | return domain.CatalogueRecord{}, err | 88 | + | return storedRecord{}, err |
| 72 | } | 89 | } | ||
| 73 | reader := valueReader{data: value} | 90 | reader := valueReader{data: value} | ||
| 74 | magic, ok := reader.take(4) | 91 | magic, ok := reader.take(4) | ||
| 75 | if !ok || string(magic) != string(catalogueMagic[:]) { | 92 | if !ok || string(magic) != string(catalogueMagic[:]) { | ||
| 76 | - | return domain.CatalogueRecord{}, errMalformedRecord | 93 | + | return storedRecord{}, errMalformedRecord |
| 77 | } | 94 | } | ||
| 78 | version, ok := reader.u16() | 95 | version, ok := reader.u16() | ||
| 79 | - | if !ok || (version != catalogueRecordVersion && version != legacyCatalogueRecordVersion) { | 96 | + | if !ok || (version != catalogueRecordVersion && version != protocolLessCatalogueRecordVersion && version != legacyCatalogueRecordVersion) { |
| 80 | - | return domain.CatalogueRecord{}, errMalformedRecord | 97 | + | return storedRecord{}, errMalformedRecord |
| 98 | + | } | |||
| 99 | + | var protocolVersion uint16 | |||
| 100 | + | if version == catalogueRecordVersion { | |||
| 101 | + | protocolVersion, ok = reader.u16() | |||
| 102 | + | if !ok { | |||
| 103 | + | return storedRecord{}, errMalformedRecord | |||
| 104 | + | } | |||
| 81 | } | 105 | } | ||
| 82 | id, ok := reader.take(16) | 106 | id, ok := reader.take(16) | ||
| 83 | if !ok { | 107 | if !ok { | ||
| 84 | - | return domain.CatalogueRecord{}, errMalformedRecord | 108 | + | return storedRecord{}, errMalformedRecord |
| 85 | } | 109 | } | ||
| 86 | record := domain.CatalogueRecord{Name: name} | 110 | record := domain.CatalogueRecord{Name: name} | ||
| 87 | copy(record.IncarnationID[:], id) | 111 | copy(record.IncarnationID[:], id) | ||
| 88 | var valid bool | 112 | var valid bool | ||
| 89 | if record.Cwd, valid = reader.str(); !valid { | 113 | if record.Cwd, valid = reader.str(); !valid { | ||
| 90 | - | return domain.CatalogueRecord{}, errMalformedRecord | 114 | + | return storedRecord{}, errMalformedRecord |
| 91 | } | 115 | } | ||
| 92 | created, ok := reader.u64() | 116 | created, ok := reader.u64() | ||
| 93 | if !ok { | 117 | if !ok { | ||
| 94 | - | return domain.CatalogueRecord{}, errMalformedRecord | 118 | + | return storedRecord{}, errMalformedRecord |
| 95 | } | 119 | } | ||
| 96 | record.CreatedAt = int64(created) | 120 | record.CreatedAt = int64(created) | ||
| 97 | updated, ok := reader.u64() | 121 | updated, ok := reader.u64() | ||
| 98 | if !ok { | 122 | if !ok { | ||
| 99 | - | return domain.CatalogueRecord{}, errMalformedRecord | 123 | + | return storedRecord{}, errMalformedRecord |
| 100 | } | 124 | } | ||
| 101 | record.UpdatedAt = int64(updated) | 125 | record.UpdatedAt = int64(updated) | ||
| 102 | if record.LastUsedSeq, ok = reader.u64(); !ok { | 126 | if record.LastUsedSeq, ok = reader.u64(); !ok { | ||
| 103 | - | return domain.CatalogueRecord{}, errMalformedRecord | 127 | + | return storedRecord{}, errMalformedRecord |
| 104 | } | 128 | } | ||
| 105 | count, ok := reader.u32() | 129 | count, ok := reader.u32() | ||
| 106 | if !ok || uint64(count) > uint64(reader.remaining())/4 { | 130 | if !ok || uint64(count) > uint64(reader.remaining())/4 { | ||
| 107 | - | return domain.CatalogueRecord{}, errMalformedRecord | 131 | + | return storedRecord{}, errMalformedRecord |
| 108 | } | 132 | } | ||
| 109 | if count > 0 { | 133 | if count > 0 { | ||
| 110 | record.TabNames = make([]string, 0, int(count)) | 134 | record.TabNames = make([]string, 0, int(count)) | ||
| @@ -112,7 +136,7 @@ func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error | |||||
| 112 | for range count { | 136 | for range count { | ||
| 113 | tab, ok := reader.str() | 137 | tab, ok := reader.str() | ||
| 114 | if !ok { | 138 | if !ok { | ||
| 115 | - | return domain.CatalogueRecord{}, errMalformedRecord | 139 | + | return storedRecord{}, errMalformedRecord |
| 116 | } | 140 | } | ||
| 117 | record.TabNames = append(record.TabNames, tab) | 141 | record.TabNames = append(record.TabNames, tab) | ||
| 118 | } | 142 | } | ||
| @@ -126,7 +150,7 @@ func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error | |||||
| 126 | } else { | 150 | } else { | ||
| 127 | tabRecordCount, ok := reader.u32() | 151 | tabRecordCount, ok := reader.u32() | ||
| 128 | if !ok || uint64(tabRecordCount) > uint64(reader.remaining())/8 { | 152 | if !ok || uint64(tabRecordCount) > uint64(reader.remaining())/8 { | ||
| 129 | - | return domain.CatalogueRecord{}, errMalformedRecord | 153 | + | return storedRecord{}, errMalformedRecord |
| 130 | } | 154 | } | ||
| 131 | if tabRecordCount > 0 { | 155 | if tabRecordCount > 0 { | ||
| 132 | record.TabRecords = make([]domain.CatalogueTabRecord, 0, int(tabRecordCount)) | 156 | record.TabRecords = make([]domain.CatalogueTabRecord, 0, int(tabRecordCount)) | ||
| @@ -135,21 +159,21 @@ func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error | |||||
| 135 | stableID, stableOK := reader.str() | 159 | stableID, stableOK := reader.str() | ||
| 136 | name, nameOK := reader.str() | 160 | name, nameOK := reader.str() | ||
| 137 | if !stableOK || !nameOK { | 161 | if !stableOK || !nameOK { | ||
| 138 | - | return domain.CatalogueRecord{}, errMalformedRecord | 162 | + | return storedRecord{}, errMalformedRecord |
| 139 | } | 163 | } | ||
| 140 | record.TabRecords = append(record.TabRecords, domain.CatalogueTabRecord{StableID: domain.TabStableID(stableID), Name: name}) | 164 | record.TabRecords = append(record.TabRecords, domain.CatalogueTabRecord{StableID: domain.TabStableID(stableID), Name: name}) | ||
| 141 | } | 165 | } | ||
| 142 | } | 166 | } | ||
| 143 | if record.Committed, ok = reader.ref(); !ok { | 167 | if record.Committed, ok = reader.ref(); !ok { | ||
| 144 | - | return domain.CatalogueRecord{}, errMalformedRecord | 168 | + | return storedRecord{}, errMalformedRecord |
| 145 | } | 169 | } | ||
| 146 | if record.DegradedReason, ok = reader.str(); !ok || reader.remaining() != 0 { | 170 | if record.DegradedReason, ok = reader.str(); !ok || reader.remaining() != 0 { | ||
| 147 | - | return domain.CatalogueRecord{}, errMalformedRecord | 171 | + | return storedRecord{}, errMalformedRecord |
| 148 | } | 172 | } | ||
| 149 | if err := record.Validate(); err != nil { | 173 | if err := record.Validate(); err != nil { | ||
| 150 | - | return domain.CatalogueRecord{}, errors.Join(errMalformedRecord, fmt.Errorf("persist: invalid catalogue record: %w", err)) | 174 | + | return storedRecord{}, errors.Join(errMalformedRecord, fmt.Errorf("persist: invalid catalogue record: %w", err)) |
| 151 | } | 175 | } | ||
| 152 | - | return record, nil | 176 | + | return storedRecord{record: record, protocolVersion: protocolVersion}, nil |
| 153 | } | 177 | } | ||
| 154 | 178 | ||||
| 155 | func appendCheckedString(buf []byte, value string) ([]byte, error) { | 179 | func appendCheckedString(buf []byte, value string) ([]byte, error) { | ||
M internal/usecase/recovery/reset.go +39 -0
| @@ -8,6 +8,45 @@ 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 | + | ||
| 11 | 50 | // ResetIncompatible replaces an exact healthy checkpoint authority with a | |
| 12 | 51 | // fresh incarnation. Stale observations are harmless no-ops. | |
| 13 | 52 | func (c *Coordinator) ResetIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID, expectedCheckpoint domain.CheckpointRef) (domain.CatalogueRecord, bool, error) { | |
| @@ -8,6 +8,45 @@ 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 | + | ||||
| 11 | // ResetIncompatible replaces an exact healthy checkpoint authority with a | 50 | // ResetIncompatible replaces an exact healthy checkpoint authority with a | ||
| 12 | // fresh incarnation. Stale observations are harmless no-ops. | 51 | // fresh incarnation. Stale observations are harmless no-ops. | ||
| 13 | func (c *Coordinator) ResetIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID, expectedCheckpoint domain.CheckpointRef) (domain.CatalogueRecord, bool, error) { | 52 | 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 +30 -0
| @@ -235,6 +235,36 @@ 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 | + | ||
| 238 | 268 | func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) { | |
| 239 | 269 | t.Parallel() | |
| 240 | 270 | old := healthyTransactionRecord() | |
| @@ -235,6 +235,36 @@ 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 | + | ||||
| 238 | func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) { | 268 | func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) { | ||
| 239 | t.Parallel() | 269 | t.Parallel() | ||
| 240 | old := healthyTransactionRecord() | 270 | old := healthyTransactionRecord() | ||