detached 106b3de

test(daemon): cover incremental restore and import

commit
106b3dec59745bd84255b1cacbbd8cdc55acf6dc
author
brice · · 2026-07-21 23:36 UTC
parent
12d44e0
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
6 files · +518 -36
Split
UnifiedSplit
M internal/usecase/daemon/daemon.go +26 -21
@@ -126,8 +126,12 @@ type Daemon struct {
126126 // narrower checkpoint-only test seam.
127127 snapshotDeletion ports.SnapshotDeletionTombstone
128128 snapsEnabled bool
129- noticeStore ports.NoticeStore
130- snapshotMarshal func(snapcodec.Session) ([]byte, error)
129+ // legacyImportPending holds verified legacy blobs whose source deletion
130+ // failed, so a later import retries only deletion rather than publication.
131+ legacyImportMu sync.Mutex
132+ legacyImportPending map[string][]byte
133+ noticeStore ports.NoticeStore
134+ snapshotMarshal func(snapcodec.Session) ([]byte, error)
131135 // snapshotChunkCache contains only encoded immutable sealed chunks. It is
132136 // independent of pane state and is bounded to prevent checkpoint history
133137 // from becoming unbounded daemon memory.
@@ -374,25 +378,26 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
374378 shell = "/bin/sh"
375379 }
376380 d := &Daemon{
377- sessions: make(map[domain.SessionID]*session),
378- stopped: make(map[string]stoppedSession),
379- parked: make(map[uint64]*parkedAttachment),
380- ptys: ptys,
381- clock: clock,
382- log: log,
383- baseEnv: os.Environ(),
384- shell: shell,
385- persist: persist.New(nil),
386- dirOrHome: dirOrHome,
387- done: make(chan struct{}),
388- restoreDone: make(chan struct{}),
389- animWake: make(chan struct{}, 1),
390- snapshotMarshal: snapcodec.Marshal,
391- snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit),
392- snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity),
393- snapshotWake: make(chan struct{}, 1),
394- notices: newNoticeCenter(),
395- resumeParkGrace: defaultResumeParkGrace,
381+ sessions: make(map[domain.SessionID]*session),
382+ stopped: make(map[string]stoppedSession),
383+ parked: make(map[uint64]*parkedAttachment),
384+ ptys: ptys,
385+ clock: clock,
386+ log: log,
387+ baseEnv: os.Environ(),
388+ shell: shell,
389+ persist: persist.New(nil),
390+ dirOrHome: dirOrHome,
391+ done: make(chan struct{}),
392+ restoreDone: make(chan struct{}),
393+ animWake: make(chan struct{}, 1),
394+ snapshotMarshal: snapcodec.Marshal,
395+ snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit),
396+ legacyImportPending: make(map[string][]byte),
397+ snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity),
398+ snapshotWake: make(chan struct{}, 1),
399+ notices: newNoticeCenter(),
400+ resumeParkGrace: defaultResumeParkGrace,
396401 barScripts: &barScriptState{
397402 cfg: barConfigFromDomain(domain.Defaults().Bar),
398403 outputs: make(map[domain.SessionID]barScriptOutputs),
@@ -126,8 +126,12 @@ type Daemon struct {
126 // narrower checkpoint-only test seam.126 // narrower checkpoint-only test seam.
127 snapshotDeletion ports.SnapshotDeletionTombstone127 snapshotDeletion ports.SnapshotDeletionTombstone
128 snapsEnabled bool128 snapsEnabled bool
129- noticeStore ports.NoticeStore129+ // legacyImportPending holds verified legacy blobs whose source deletion
130- snapshotMarshal func(snapcodec.Session) ([]byte, error)130+ // failed, so a later import retries only deletion rather than publication.
131+ legacyImportMu sync.Mutex
132+ legacyImportPending map[string][]byte
133+ noticeStore ports.NoticeStore
134+ snapshotMarshal func(snapcodec.Session) ([]byte, error)
131 // snapshotChunkCache contains only encoded immutable sealed chunks. It is135 // snapshotChunkCache contains only encoded immutable sealed chunks. It is
132 // independent of pane state and is bounded to prevent checkpoint history136 // independent of pane state and is bounded to prevent checkpoint history
133 // from becoming unbounded daemon memory.137 // from becoming unbounded daemon memory.
@@ -374,25 +378,26 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
374 shell = "/bin/sh"378 shell = "/bin/sh"
375 }379 }
376 d := &Daemon{380 d := &Daemon{
377- sessions: make(map[domain.SessionID]*session),381+ sessions: make(map[domain.SessionID]*session),
378- stopped: make(map[string]stoppedSession),382+ stopped: make(map[string]stoppedSession),
379- parked: make(map[uint64]*parkedAttachment),383+ parked: make(map[uint64]*parkedAttachment),
380- ptys: ptys,384+ ptys: ptys,
381- clock: clock,385+ clock: clock,
382- log: log,386+ log: log,
383- baseEnv: os.Environ(),387+ baseEnv: os.Environ(),
384- shell: shell,388+ shell: shell,
385- persist: persist.New(nil),389+ persist: persist.New(nil),
386- dirOrHome: dirOrHome,390+ dirOrHome: dirOrHome,
387- done: make(chan struct{}),391+ done: make(chan struct{}),
388- restoreDone: make(chan struct{}),392+ restoreDone: make(chan struct{}),
389- animWake: make(chan struct{}, 1),393+ animWake: make(chan struct{}, 1),
390- snapshotMarshal: snapcodec.Marshal,394+ snapshotMarshal: snapcodec.Marshal,
391- snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit),395+ snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit),
392- snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity),396+ legacyImportPending: make(map[string][]byte),
393- snapshotWake: make(chan struct{}, 1),397+ snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity),
394- notices: newNoticeCenter(),398+ snapshotWake: make(chan struct{}, 1),
395- resumeParkGrace: defaultResumeParkGrace,399+ notices: newNoticeCenter(),
400+ resumeParkGrace: defaultResumeParkGrace,
396 barScripts: &barScriptState{401 barScripts: &barScriptState{
397 cfg: barConfigFromDomain(domain.Defaults().Bar),402 cfg: barConfigFromDomain(domain.Defaults().Bar),
398 outputs: make(map[domain.SessionID]barScriptOutputs),403 outputs: make(map[domain.SessionID]barScriptOutputs),
A internal/usecase/daemon/snapshot_import_test.go +172 -0
@@ -0,0 +1,172 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "errors"
6+ "testing"
7+
8+ "github.com/stretchr/testify/require"
9+
10+ "github.com/bnema/vev/internal/ports"
11+ snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
12+)
13+
14+type snapshotAcceptanceLegacySource struct {
15+ blobs []ports.LegacySnapshot
16+ loadCalls int
17+ deleteCalls []string
18+ deleteErr error
19+}
20+
21+func (s *snapshotAcceptanceLegacySource) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) {
22+ if err := ctx.Err(); err != nil {
23+ return nil, err
24+ }
25+ s.loadCalls++
26+ out := make([]ports.LegacySnapshot, len(s.blobs))
27+ for i, blob := range s.blobs {
28+ out[i] = ports.LegacySnapshot{Name: blob.Name, Data: append([]byte(nil), blob.Data...)}
29+ }
30+ return out, nil
31+}
32+
33+func (s *snapshotAcceptanceLegacySource) DeleteLegacy(ctx context.Context, name string) error {
34+ if err := ctx.Err(); err != nil {
35+ return err
36+ }
37+ s.deleteCalls = append(s.deleteCalls, name)
38+ if s.deleteErr != nil {
39+ return s.deleteErr
40+ }
41+ for i, blob := range s.blobs {
42+ if blob.Name == name {
43+ s.blobs = append(s.blobs[:i], s.blobs[i+1:]...)
44+ break
45+ }
46+ }
47+ return nil
48+}
49+
50+func legacyAcceptanceBlob(t *testing.T, snapshot snapcodec.Session) ports.LegacySnapshot {
51+ t.Helper()
52+ data, err := snapcodec.Marshal(snapshot)
53+ require.NoError(t, err)
54+ return ports.LegacySnapshot{Name: snapshot.Name, Data: data}
55+}
56+
57+func TestImportLegacyPublishesCompleteGenerationBeforeDeletingSource(t *testing.T) {
58+ snapshot := restoreAcceptanceSession(t, "legacy")
59+ repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}
60+ legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}}
61+ d := newTestDaemon(t, nil, stubClock{})
62+ WithSnapshotRepository(repository, legacy)(d)
63+ expected, verifyErr := legacyPublication(snapshot)
64+ require.NoError(t, verifyErr)
65+ _, verifyErr = snapcodec.Unmarshal(legacy.blobs[0].Data)
66+ require.NoError(t, verifyErr)
67+ require.NoError(t, verifyLegacyImportGeneration(acceptanceGeneration(t, snapshot, 1), expected))
68+
69+ d.importLegacySnapshots(context.Background())
70+
71+ require.Len(t, repository.publishes, 1)
72+ publication := repository.publishes[0]
73+ require.Equal(t, snapshot.Name, publication.Name)
74+ require.EqualValues(t, 1, publication.Generation)
75+ manifest, err := snapcodec.UnmarshalManifest(publication.Manifest)
76+ require.NoError(t, err)
77+ require.Equal(t, uint64(1), manifest.Generation)
78+ require.Len(t, manifest.Tabs[0].Panes[0].Sealed, 2)
79+ for _, object := range publication.Objects {
80+ kind, _, err := snapcodec.UnmarshalObject(object.Data)
81+ require.NoError(t, err)
82+ require.Contains(t, []snapcodec.ObjectKind{snapcodec.HistoryChunk, snapcodec.HistoryTail, snapcodec.Visible}, kind)
83+ require.Equal(t, object.Digest, snapcodec.ManifestDigest(object.Data))
84+ }
85+ require.Equal(t, []string{snapshot.Name}, legacy.deleteCalls)
86+ require.Empty(t, legacy.blobs)
87+}
88+
89+func TestImportLegacyExistingIncrementalSkipsLegacy(t *testing.T) {
90+ snapshot := restoreAcceptanceSession(t, "already-incremental")
91+ repository := &snapshotAcceptanceRepository{names: []string{snapshot.Name}, generations: make(map[string]ports.SnapshotGeneration)}
92+ legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}}
93+ d := newTestDaemon(t, nil, stubClock{})
94+ WithSnapshotRepository(repository, legacy)(d)
95+
96+ d.importLegacySnapshots(context.Background())
97+
98+ require.Empty(t, repository.publishes)
99+ require.Empty(t, legacy.deleteCalls)
100+ require.Len(t, legacy.blobs, 1)
101+}
102+
103+func TestImportLegacyRetainsSourceUntilExactReloadVerification(t *testing.T) {
104+ snapshot := restoreAcceptanceSession(t, "retain")
105+ for _, test := range []struct {
106+ name string
107+ publish error
108+ load error
109+ mutate func(ports.SnapshotGeneration, ports.SnapshotPublication) ports.SnapshotGeneration
110+ }{
111+ {"publish failure", errors.New("publish failed"), nil, nil},
112+ {"reload failure", nil, errors.New("reload failed"), nil},
113+ {"manifest metadata mismatch", nil, nil, func(g ports.SnapshotGeneration, _ ports.SnapshotPublication) ports.SnapshotGeneration {
114+ g.Generation++
115+ return g
116+ }},
117+ {"object bytes mismatch", nil, nil, func(g ports.SnapshotGeneration, _ ports.SnapshotPublication) ports.SnapshotGeneration {
118+ for digest, data := range g.Objects {
119+ data[len(data)-1] ^= 1
120+ g.Objects[digest] = data
121+ break
122+ }
123+ return g
124+ }},
125+ } {
126+ t.Run(test.name, func(t *testing.T) {
127+ repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration), publishErr: test.publish, loadErr: test.load, loadMutate: test.mutate}
128+ legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}}
129+ d := newTestDaemon(t, nil, stubClock{})
130+ WithSnapshotRepository(repository, legacy)(d)
131+ d.importLegacySnapshots(context.Background())
132+ require.Empty(t, legacy.deleteCalls)
133+ require.Len(t, legacy.blobs, 1)
134+ })
135+ }
136+}
137+
138+func TestImportLegacyDeleteFailureRetriesWithoutRepublishing(t *testing.T) {
139+ snapshot := restoreAcceptanceSession(t, "retry-delete")
140+ repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}
141+ legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}, deleteErr: errors.New("sync failed")}
142+ d := newTestDaemon(t, nil, stubClock{})
143+ WithSnapshotRepository(repository, legacy)(d)
144+
145+ d.importLegacySnapshots(context.Background())
146+ legacy.deleteErr = nil
147+ d.importLegacySnapshots(context.Background())
148+
149+ require.Len(t, repository.publishes, 1, "a retry must only retry source deletion")
150+ require.Equal(t, []string{snapshot.Name, snapshot.Name}, legacy.deleteCalls)
151+ require.Empty(t, legacy.blobs)
152+}
153+
154+func TestImportLegacyCancellationAndPerSessionContinuation(t *testing.T) {
155+ bad := restoreAcceptanceSession(t, "bad")
156+ good := restoreAcceptanceSession(t, "good")
157+ badBlob := legacyAcceptanceBlob(t, bad)
158+ badBlob.Data = []byte("not-a-v3-snapshot")
159+ repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}
160+ legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{badBlob, legacyAcceptanceBlob(t, good)}}
161+ d := newTestDaemon(t, nil, stubClock{})
162+ WithSnapshotRepository(repository, legacy)(d)
163+
164+ d.importLegacySnapshots(context.Background())
165+ require.Len(t, repository.publishes, 1, "a corrupt session must not block later sessions")
166+ require.Equal(t, []string{good.Name}, legacy.deleteCalls)
167+
168+ cancelled, cancel := context.WithCancel(context.Background())
169+ cancel()
170+ d.importLegacySnapshots(cancelled)
171+ require.Equal(t, 1, legacy.loadCalls, "a canceled import must not touch the legacy source")
172+}
@@ -0,0 +1,172 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "errors"
6+ "testing"
7+
8+ "github.com/stretchr/testify/require"
9+
10+ "github.com/bnema/vev/internal/ports"
11+ snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
12+)
13+
14+type snapshotAcceptanceLegacySource struct {
15+ blobs []ports.LegacySnapshot
16+ loadCalls int
17+ deleteCalls []string
18+ deleteErr error
19+}
20+
21+func (s *snapshotAcceptanceLegacySource) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) {
22+ if err := ctx.Err(); err != nil {
23+ return nil, err
24+ }
25+ s.loadCalls++
26+ out := make([]ports.LegacySnapshot, len(s.blobs))
27+ for i, blob := range s.blobs {
28+ out[i] = ports.LegacySnapshot{Name: blob.Name, Data: append([]byte(nil), blob.Data...)}
29+ }
30+ return out, nil
31+}
32+
33+func (s *snapshotAcceptanceLegacySource) DeleteLegacy(ctx context.Context, name string) error {
34+ if err := ctx.Err(); err != nil {
35+ return err
36+ }
37+ s.deleteCalls = append(s.deleteCalls, name)
38+ if s.deleteErr != nil {
39+ return s.deleteErr
40+ }
41+ for i, blob := range s.blobs {
42+ if blob.Name == name {
43+ s.blobs = append(s.blobs[:i], s.blobs[i+1:]...)
44+ break
45+ }
46+ }
47+ return nil
48+}
49+
50+func legacyAcceptanceBlob(t *testing.T, snapshot snapcodec.Session) ports.LegacySnapshot {
51+ t.Helper()
52+ data, err := snapcodec.Marshal(snapshot)
53+ require.NoError(t, err)
54+ return ports.LegacySnapshot{Name: snapshot.Name, Data: data}
55+}
56+
57+func TestImportLegacyPublishesCompleteGenerationBeforeDeletingSource(t *testing.T) {
58+ snapshot := restoreAcceptanceSession(t, "legacy")
59+ repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}
60+ legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}}
61+ d := newTestDaemon(t, nil, stubClock{})
62+ WithSnapshotRepository(repository, legacy)(d)
63+ expected, verifyErr := legacyPublication(snapshot)
64+ require.NoError(t, verifyErr)
65+ _, verifyErr = snapcodec.Unmarshal(legacy.blobs[0].Data)
66+ require.NoError(t, verifyErr)
67+ require.NoError(t, verifyLegacyImportGeneration(acceptanceGeneration(t, snapshot, 1), expected))
68+
69+ d.importLegacySnapshots(context.Background())
70+
71+ require.Len(t, repository.publishes, 1)
72+ publication := repository.publishes[0]
73+ require.Equal(t, snapshot.Name, publication.Name)
74+ require.EqualValues(t, 1, publication.Generation)
75+ manifest, err := snapcodec.UnmarshalManifest(publication.Manifest)
76+ require.NoError(t, err)
77+ require.Equal(t, uint64(1), manifest.Generation)
78+ require.Len(t, manifest.Tabs[0].Panes[0].Sealed, 2)
79+ for _, object := range publication.Objects {
80+ kind, _, err := snapcodec.UnmarshalObject(object.Data)
81+ require.NoError(t, err)
82+ require.Contains(t, []snapcodec.ObjectKind{snapcodec.HistoryChunk, snapcodec.HistoryTail, snapcodec.Visible}, kind)
83+ require.Equal(t, object.Digest, snapcodec.ManifestDigest(object.Data))
84+ }
85+ require.Equal(t, []string{snapshot.Name}, legacy.deleteCalls)
86+ require.Empty(t, legacy.blobs)
87+}
88+
89+func TestImportLegacyExistingIncrementalSkipsLegacy(t *testing.T) {
90+ snapshot := restoreAcceptanceSession(t, "already-incremental")
91+ repository := &snapshotAcceptanceRepository{names: []string{snapshot.Name}, generations: make(map[string]ports.SnapshotGeneration)}
92+ legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}}
93+ d := newTestDaemon(t, nil, stubClock{})
94+ WithSnapshotRepository(repository, legacy)(d)
95+
96+ d.importLegacySnapshots(context.Background())
97+
98+ require.Empty(t, repository.publishes)
99+ require.Empty(t, legacy.deleteCalls)
100+ require.Len(t, legacy.blobs, 1)
101+}
102+
103+func TestImportLegacyRetainsSourceUntilExactReloadVerification(t *testing.T) {
104+ snapshot := restoreAcceptanceSession(t, "retain")
105+ for _, test := range []struct {
106+ name string
107+ publish error
108+ load error
109+ mutate func(ports.SnapshotGeneration, ports.SnapshotPublication) ports.SnapshotGeneration
110+ }{
111+ {"publish failure", errors.New("publish failed"), nil, nil},
112+ {"reload failure", nil, errors.New("reload failed"), nil},
113+ {"manifest metadata mismatch", nil, nil, func(g ports.SnapshotGeneration, _ ports.SnapshotPublication) ports.SnapshotGeneration {
114+ g.Generation++
115+ return g
116+ }},
117+ {"object bytes mismatch", nil, nil, func(g ports.SnapshotGeneration, _ ports.SnapshotPublication) ports.SnapshotGeneration {
118+ for digest, data := range g.Objects {
119+ data[len(data)-1] ^= 1
120+ g.Objects[digest] = data
121+ break
122+ }
123+ return g
124+ }},
125+ } {
126+ t.Run(test.name, func(t *testing.T) {
127+ repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration), publishErr: test.publish, loadErr: test.load, loadMutate: test.mutate}
128+ legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}}
129+ d := newTestDaemon(t, nil, stubClock{})
130+ WithSnapshotRepository(repository, legacy)(d)
131+ d.importLegacySnapshots(context.Background())
132+ require.Empty(t, legacy.deleteCalls)
133+ require.Len(t, legacy.blobs, 1)
134+ })
135+ }
136+}
137+
138+func TestImportLegacyDeleteFailureRetriesWithoutRepublishing(t *testing.T) {
139+ snapshot := restoreAcceptanceSession(t, "retry-delete")
140+ repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}
141+ legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}, deleteErr: errors.New("sync failed")}
142+ d := newTestDaemon(t, nil, stubClock{})
143+ WithSnapshotRepository(repository, legacy)(d)
144+
145+ d.importLegacySnapshots(context.Background())
146+ legacy.deleteErr = nil
147+ d.importLegacySnapshots(context.Background())
148+
149+ require.Len(t, repository.publishes, 1, "a retry must only retry source deletion")
150+ require.Equal(t, []string{snapshot.Name, snapshot.Name}, legacy.deleteCalls)
151+ require.Empty(t, legacy.blobs)
152+}
153+
154+func TestImportLegacyCancellationAndPerSessionContinuation(t *testing.T) {
155+ bad := restoreAcceptanceSession(t, "bad")
156+ good := restoreAcceptanceSession(t, "good")
157+ badBlob := legacyAcceptanceBlob(t, bad)
158+ badBlob.Data = []byte("not-a-v3-snapshot")
159+ repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}
160+ legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{badBlob, legacyAcceptanceBlob(t, good)}}
161+ d := newTestDaemon(t, nil, stubClock{})
162+ WithSnapshotRepository(repository, legacy)(d)
163+
164+ d.importLegacySnapshots(context.Background())
165+ require.Len(t, repository.publishes, 1, "a corrupt session must not block later sessions")
166+ require.Equal(t, []string{good.Name}, legacy.deleteCalls)
167+
168+ cancelled, cancel := context.WithCancel(context.Background())
169+ cancel()
170+ d.importLegacySnapshots(cancelled)
171+ require.Equal(t, 1, legacy.loadCalls, "a canceled import must not touch the legacy source")
172+}
M internal/usecase/daemon/snapshot_legacy_import.go +0 -7
@@ -1,7 +1,6 @@
11 package daemon
22
33 import (
4- "bytes"
54 "fmt"
65
76 "github.com/bnema/vev/internal/ports"
@@ -43,9 +42,3 @@ func legacyPublication(snapshot snapcodec.Session) (ports.SnapshotPublication, e
4342 }
4443 return ports.SnapshotPublication{Name: snapshot.Name, Generation: 1, Manifest: encoded, Objects: objects}, nil
4544 }
46-
47-func sameSnapshotSession(left, right snapcodec.Session) bool {
48- leftBytes, leftErr := snapcodec.Marshal(left)
49- rightBytes, rightErr := snapcodec.Marshal(right)
50- return leftErr == nil && rightErr == nil && bytes.Equal(leftBytes, rightBytes)
51-}
@@ -1,7 +1,6 @@
1 package daemon1 package daemon
2 2
3 import (3 import (
4- "bytes"
5 "fmt"4 "fmt"
6 5
7 "github.com/bnema/vev/internal/ports"6 "github.com/bnema/vev/internal/ports"
@@ -43,9 +42,3 @@ func legacyPublication(snapshot snapcodec.Session) (ports.SnapshotPublication, e
43 }42 }
44 return ports.SnapshotPublication{Name: snapshot.Name, Generation: 1, Manifest: encoded, Objects: objects}, nil43 return ports.SnapshotPublication{Name: snapshot.Name, Generation: 1, Manifest: encoded, Objects: objects}, nil
45 }44 }
46-
47-func sameSnapshotSession(left, right snapcodec.Session) bool {
48- leftBytes, leftErr := snapcodec.Marshal(left)
49- rightBytes, rightErr := snapcodec.Marshal(right)
50- return leftErr == nil && rightErr == nil && bytes.Equal(leftBytes, rightBytes)
51-}
M internal/usecase/daemon/snapshot_restore_incremental.go +66 -7
@@ -1,7 +1,9 @@
11 package daemon
22
33 import (
4+ "bytes"
45 "context"
6+ "crypto/sha256"
57 "fmt"
68
79 "github.com/bnema/vev/internal/domain"
@@ -88,7 +90,7 @@ func generationObject(generation ports.SnapshotGeneration, ref snapcodec.ObjectR
8890 return nil, fmt.Errorf("snapshot: object kind mismatch")
8991 }
9092 data, ok := generation.Objects[ref.Digest]
91- if !ok || uint32(len(data)) != ref.Size {
93+ if !ok || uint32(len(data)) != ref.Size || sha256.Sum256(data) != ref.Digest {
9294 return nil, fmt.Errorf("snapshot: missing object")
9395 }
9496 gotKind, payload, err := snapcodec.UnmarshalObject(data)
@@ -102,7 +104,7 @@ func generationObject(generation ports.SnapshotGeneration, ref snapcodec.ObjectR
102104 // a legacy blob only after publication and a full repository reload verify the
103105 // exact converted session, leaving failures retryable on the next startup.
104106 func (d *Daemon) importLegacySnapshots(ctx context.Context) {
105- if d.legacySnapshots == nil || d.snapshotRepository == nil {
107+ if d.legacySnapshots == nil || d.snapshotRepository == nil || ctx.Err() != nil {
106108 return
107109 }
108110 existing, err := d.snapshotRepository.List(ctx)
@@ -120,7 +122,11 @@ func (d *Daemon) importLegacySnapshots(ctx context.Context) {
120122 return
121123 }
122124 for _, blob := range legacy {
125+ if ctx.Err() != nil {
126+ return
127+ }
123128 if _, ok := present[blob.Name]; ok {
129+ d.retryLegacyDelete(ctx, blob)
124130 continue
125131 }
126132 snapshot, err := snapcodec.Unmarshal(blob.Data)
@@ -136,21 +142,74 @@ func (d *Daemon) importLegacySnapshots(ctx context.Context) {
136142 var generation ports.SnapshotGeneration
137143 generation, err = d.snapshotRepository.Load(ctx, snapshot.Name)
138144 if err == nil {
139- var restored snapcodec.Session
140- restored, err = sessionFromGeneration(generation)
141- if err == nil && !sameSnapshotSession(snapshot, restored) {
142- err = fmt.Errorf("snapshot: legacy import verification mismatch")
143- }
145+ err = verifyLegacyImportGeneration(generation, publication)
144146 }
145147 }
146148 if err != nil {
147149 d.log.Warn("importing legacy snapshot failed", "session", blob.Name, "err", err)
148150 continue
149151 }
152+ d.rememberLegacyDelete(blob)
150153 if err := d.legacySnapshots.DeleteLegacy(ctx, blob.Name); err != nil {
151154 d.log.Warn("deleting imported legacy snapshot failed", "session", blob.Name, "err", err)
152155 continue
153156 }
157+ d.forgetLegacyDelete(blob.Name)
154158 present[blob.Name] = struct{}{}
155159 }
156160 }
161+
162+func verifyLegacyImportGeneration(generation ports.SnapshotGeneration, publication ports.SnapshotPublication) error {
163+ if generation.Name != publication.Name || generation.Generation != publication.Generation || !bytes.Equal(generation.Manifest, publication.Manifest) {
164+ return fmt.Errorf("snapshot: legacy import verification mismatch")
165+ }
166+ expected := make(map[ports.SnapshotDigest][]byte, len(publication.Objects))
167+ for _, object := range publication.Objects {
168+ if sha256.Sum256(object.Data) != object.Digest {
169+ return fmt.Errorf("snapshot: legacy import publication digest mismatch")
170+ }
171+ expected[object.Digest] = object.Data
172+ }
173+ if len(generation.Objects) != len(expected) {
174+ return fmt.Errorf("snapshot: legacy import verification mismatch")
175+ }
176+ for digest, expectedData := range expected {
177+ actual, ok := generation.Objects[digest]
178+ if !ok || !bytes.Equal(actual, expectedData) || sha256.Sum256(actual) != digest {
179+ return fmt.Errorf("snapshot: legacy import verification mismatch")
180+ }
181+ }
182+ return nil
183+}
184+
185+func (d *Daemon) rememberLegacyDelete(blob ports.LegacySnapshot) {
186+ d.legacyImportMu.Lock()
187+ if d.legacyImportPending == nil {
188+ d.legacyImportPending = make(map[string][]byte)
189+ }
190+ d.legacyImportPending[blob.Name] = append([]byte(nil), blob.Data...)
191+ d.legacyImportMu.Unlock()
192+}
193+
194+func (d *Daemon) forgetLegacyDelete(name string) {
195+ d.legacyImportMu.Lock()
196+ delete(d.legacyImportPending, name)
197+ d.legacyImportMu.Unlock()
198+}
199+
200+// retryLegacyDelete only acts on a blob which this daemon has already
201+// published and exactly reloaded. An unrelated pre-existing generation must
202+// continue to leave its legacy counterpart untouched.
203+func (d *Daemon) retryLegacyDelete(ctx context.Context, blob ports.LegacySnapshot) {
204+ d.legacyImportMu.Lock()
205+ pending, ok := d.legacyImportPending[blob.Name]
206+ d.legacyImportMu.Unlock()
207+ if !ok || !bytes.Equal(pending, blob.Data) {
208+ return
209+ }
210+ if err := d.legacySnapshots.DeleteLegacy(ctx, blob.Name); err != nil {
211+ d.log.Warn("deleting imported legacy snapshot failed", "session", blob.Name, "err", err)
212+ return
213+ }
214+ d.forgetLegacyDelete(blob.Name)
215+}
@@ -1,7 +1,9 @@
1 package daemon1 package daemon
2 2
3 import (3 import (
4+ "bytes"
4 "context"5 "context"
6+ "crypto/sha256"
5 "fmt"7 "fmt"
6 8
7 "github.com/bnema/vev/internal/domain"9 "github.com/bnema/vev/internal/domain"
@@ -88,7 +90,7 @@ func generationObject(generation ports.SnapshotGeneration, ref snapcodec.ObjectR
88 return nil, fmt.Errorf("snapshot: object kind mismatch")90 return nil, fmt.Errorf("snapshot: object kind mismatch")
89 }91 }
90 data, ok := generation.Objects[ref.Digest]92 data, ok := generation.Objects[ref.Digest]
91- if !ok || uint32(len(data)) != ref.Size {93+ if !ok || uint32(len(data)) != ref.Size || sha256.Sum256(data) != ref.Digest {
92 return nil, fmt.Errorf("snapshot: missing object")94 return nil, fmt.Errorf("snapshot: missing object")
93 }95 }
94 gotKind, payload, err := snapcodec.UnmarshalObject(data)96 gotKind, payload, err := snapcodec.UnmarshalObject(data)
@@ -102,7 +104,7 @@ func generationObject(generation ports.SnapshotGeneration, ref snapcodec.ObjectR
102 // a legacy blob only after publication and a full repository reload verify the104 // a legacy blob only after publication and a full repository reload verify the
103 // exact converted session, leaving failures retryable on the next startup.105 // exact converted session, leaving failures retryable on the next startup.
104 func (d *Daemon) importLegacySnapshots(ctx context.Context) {106 func (d *Daemon) importLegacySnapshots(ctx context.Context) {
105- if d.legacySnapshots == nil || d.snapshotRepository == nil {107+ if d.legacySnapshots == nil || d.snapshotRepository == nil || ctx.Err() != nil {
106 return108 return
107 }109 }
108 existing, err := d.snapshotRepository.List(ctx)110 existing, err := d.snapshotRepository.List(ctx)
@@ -120,7 +122,11 @@ func (d *Daemon) importLegacySnapshots(ctx context.Context) {
120 return122 return
121 }123 }
122 for _, blob := range legacy {124 for _, blob := range legacy {
125+ if ctx.Err() != nil {
126+ return
127+ }
123 if _, ok := present[blob.Name]; ok {128 if _, ok := present[blob.Name]; ok {
129+ d.retryLegacyDelete(ctx, blob)
124 continue130 continue
125 }131 }
126 snapshot, err := snapcodec.Unmarshal(blob.Data)132 snapshot, err := snapcodec.Unmarshal(blob.Data)
@@ -136,21 +142,74 @@ func (d *Daemon) importLegacySnapshots(ctx context.Context) {
136 var generation ports.SnapshotGeneration142 var generation ports.SnapshotGeneration
137 generation, err = d.snapshotRepository.Load(ctx, snapshot.Name)143 generation, err = d.snapshotRepository.Load(ctx, snapshot.Name)
138 if err == nil {144 if err == nil {
139- var restored snapcodec.Session145+ err = verifyLegacyImportGeneration(generation, publication)
140- restored, err = sessionFromGeneration(generation)
141- if err == nil && !sameSnapshotSession(snapshot, restored) {
142- err = fmt.Errorf("snapshot: legacy import verification mismatch")
143- }
144 }146 }
145 }147 }
146 if err != nil {148 if err != nil {
147 d.log.Warn("importing legacy snapshot failed", "session", blob.Name, "err", err)149 d.log.Warn("importing legacy snapshot failed", "session", blob.Name, "err", err)
148 continue150 continue
149 }151 }
152+ d.rememberLegacyDelete(blob)
150 if err := d.legacySnapshots.DeleteLegacy(ctx, blob.Name); err != nil {153 if err := d.legacySnapshots.DeleteLegacy(ctx, blob.Name); err != nil {
151 d.log.Warn("deleting imported legacy snapshot failed", "session", blob.Name, "err", err)154 d.log.Warn("deleting imported legacy snapshot failed", "session", blob.Name, "err", err)
152 continue155 continue
153 }156 }
157+ d.forgetLegacyDelete(blob.Name)
154 present[blob.Name] = struct{}{}158 present[blob.Name] = struct{}{}
155 }159 }
156 }160 }
161+
162+func verifyLegacyImportGeneration(generation ports.SnapshotGeneration, publication ports.SnapshotPublication) error {
163+ if generation.Name != publication.Name || generation.Generation != publication.Generation || !bytes.Equal(generation.Manifest, publication.Manifest) {
164+ return fmt.Errorf("snapshot: legacy import verification mismatch")
165+ }
166+ expected := make(map[ports.SnapshotDigest][]byte, len(publication.Objects))
167+ for _, object := range publication.Objects {
168+ if sha256.Sum256(object.Data) != object.Digest {
169+ return fmt.Errorf("snapshot: legacy import publication digest mismatch")
170+ }
171+ expected[object.Digest] = object.Data
172+ }
173+ if len(generation.Objects) != len(expected) {
174+ return fmt.Errorf("snapshot: legacy import verification mismatch")
175+ }
176+ for digest, expectedData := range expected {
177+ actual, ok := generation.Objects[digest]
178+ if !ok || !bytes.Equal(actual, expectedData) || sha256.Sum256(actual) != digest {
179+ return fmt.Errorf("snapshot: legacy import verification mismatch")
180+ }
181+ }
182+ return nil
183+}
184+
185+func (d *Daemon) rememberLegacyDelete(blob ports.LegacySnapshot) {
186+ d.legacyImportMu.Lock()
187+ if d.legacyImportPending == nil {
188+ d.legacyImportPending = make(map[string][]byte)
189+ }
190+ d.legacyImportPending[blob.Name] = append([]byte(nil), blob.Data...)
191+ d.legacyImportMu.Unlock()
192+}
193+
194+func (d *Daemon) forgetLegacyDelete(name string) {
195+ d.legacyImportMu.Lock()
196+ delete(d.legacyImportPending, name)
197+ d.legacyImportMu.Unlock()
198+}
199+
200+// retryLegacyDelete only acts on a blob which this daemon has already
201+// published and exactly reloaded. An unrelated pre-existing generation must
202+// continue to leave its legacy counterpart untouched.
203+func (d *Daemon) retryLegacyDelete(ctx context.Context, blob ports.LegacySnapshot) {
204+ d.legacyImportMu.Lock()
205+ pending, ok := d.legacyImportPending[blob.Name]
206+ d.legacyImportMu.Unlock()
207+ if !ok || !bytes.Equal(pending, blob.Data) {
208+ return
209+ }
210+ if err := d.legacySnapshots.DeleteLegacy(ctx, blob.Name); err != nil {
211+ d.log.Warn("deleting imported legacy snapshot failed", "session", blob.Name, "err", err)
212+ return
213+ }
214+ d.forgetLegacyDelete(blob.Name)
215+}
A internal/usecase/daemon/snapshot_restore_test.go +250 -0
@@ -0,0 +1,250 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "errors"
6+ "sync"
7+ "testing"
8+
9+ "github.com/stretchr/testify/require"
10+
11+ "github.com/bnema/vev/internal/domain"
12+ "github.com/bnema/vev/internal/ports"
13+ "github.com/bnema/vev/internal/usecase/layout"
14+ snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
15+ "github.com/bnema/vev/pkg/renderer"
16+ "github.com/bnema/vev/pkg/vt"
17+)
18+
19+type snapshotAcceptanceRepository struct {
20+ mu sync.Mutex
21+ names []string
22+ generations map[string]ports.SnapshotGeneration
23+ publishErr error
24+ listErr error
25+ loadErr error
26+ publishes []ports.SnapshotPublication
27+ loadMutate func(ports.SnapshotGeneration, ports.SnapshotPublication) ports.SnapshotGeneration
28+}
29+
30+func (r *snapshotAcceptanceRepository) Publish(ctx context.Context, p ports.SnapshotPublication) error {
31+ if err := ctx.Err(); err != nil {
32+ return err
33+ }
34+ r.mu.Lock()
35+ defer r.mu.Unlock()
36+ r.publishes = append(r.publishes, p)
37+ if r.publishErr != nil {
38+ return r.publishErr
39+ }
40+ objects := make(map[ports.SnapshotDigest][]byte, len(p.Objects))
41+ for _, object := range p.Objects {
42+ objects[object.Digest] = append([]byte(nil), object.Data...)
43+ }
44+ r.generations[p.Name] = ports.SnapshotGeneration{Name: p.Name, Generation: p.Generation, Manifest: append([]byte(nil), p.Manifest...), Objects: objects}
45+ found := false
46+ for _, name := range r.names {
47+ found = found || name == p.Name
48+ }
49+ if !found {
50+ r.names = append(r.names, p.Name)
51+ }
52+ return nil
53+}
54+
55+func (r *snapshotAcceptanceRepository) List(ctx context.Context) ([]string, error) {
56+ if err := ctx.Err(); err != nil {
57+ return nil, err
58+ }
59+ r.mu.Lock()
60+ defer r.mu.Unlock()
61+ if r.listErr != nil {
62+ return nil, r.listErr
63+ }
64+ return append([]string(nil), r.names...), nil
65+}
66+
67+func (r *snapshotAcceptanceRepository) Load(ctx context.Context, name string) (ports.SnapshotGeneration, error) {
68+ if err := ctx.Err(); err != nil {
69+ return ports.SnapshotGeneration{}, err
70+ }
71+ r.mu.Lock()
72+ defer r.mu.Unlock()
73+ if r.loadErr != nil {
74+ return ports.SnapshotGeneration{}, r.loadErr
75+ }
76+ generation, ok := r.generations[name]
77+ if !ok {
78+ return ports.SnapshotGeneration{}, errors.New("missing generation")
79+ }
80+ if r.loadMutate != nil && len(r.publishes) > 0 {
81+ generation = r.loadMutate(generation, r.publishes[len(r.publishes)-1])
82+ }
83+ return cloneAcceptanceGeneration(generation), nil
84+}
85+
86+func (*snapshotAcceptanceRepository) Delete(context.Context, string) error { return nil }
87+func (*snapshotAcceptanceRepository) Maintain(context.Context) error { return nil }
88+
89+func cloneAcceptanceGeneration(g ports.SnapshotGeneration) ports.SnapshotGeneration {
90+ out := g
91+ out.Manifest = append([]byte(nil), g.Manifest...)
92+ out.Objects = make(map[ports.SnapshotDigest][]byte, len(g.Objects))
93+ for digest, data := range g.Objects {
94+ out.Objects[digest] = append([]byte(nil), data...)
95+ }
96+ return out
97+}
98+
99+func restoreAcceptanceSession(t *testing.T, name string) snapcodec.Session {
100+ t.Helper()
101+ history := vt.NewHistory(vt.HistoryConfig{MaxRows: 8, MaxCells: 8 * 293, ChunkRows: 2})
102+ for _, ch := range []rune{'a', 'b', 'c', 'd', 'e'} {
103+ row := make([]renderer.Cell, 293)
104+ for i := range row {
105+ row[i] = renderer.Cell{Rune: ch}
106+ }
107+ require.NoError(t, history.Append(row))
108+ }
109+ view := history.SnapshotView()
110+ sealed := make([][]byte, view.ChunkCount())
111+ for i := range sealed {
112+ var err error
113+ sealed[i], err = vt.MarshalHistoryChunk(view.Chunk(i))
114+ require.NoError(t, err)
115+ }
116+ tail, err := vt.MarshalHistoryTail(view)
117+ require.NoError(t, err)
118+
119+ screen := vt.NewScreen(293, 4)
120+ screen.Write([]byte("primary-visible"))
121+ screen.Write([]byte("\x1b[?1049h"))
122+ screen.Write([]byte("alternate-active"))
123+ visible, err := screen.MarshalPrimaryVisible()
124+ require.NoError(t, err)
125+
126+ paneID := layout.PaneID("pane-7")
127+ return snapcodec.Session{
128+ Name: name, CreatedAt: 77, Active: 0,
129+ Tabs: []snapcodec.Tab{{
130+ StableID: "tab-stable", Cols: 293, Rows: 4, NextPaneID: 8, Focus: paneID,
131+ Tree: &layout.Tree{Root: layout.NewLeaf(paneID), Focus: paneID},
132+ Panes: []snapcodec.Pane{{
133+ ID: paneID, StableID: "pane-stable", Cwd: "/snapshot/cwd", SealedChunks: sealed, Tail: tail, Visible: visible,
134+ Process: &snapcodec.Process{Argv: []string{"sleep", "99"}, Strategy: "argv"},
135+ }},
136+ }},
137+ }
138+}
139+
140+func acceptanceGeneration(t *testing.T, snapshot snapcodec.Session, generation uint64) ports.SnapshotGeneration {
141+ t.Helper()
142+ publication, err := legacyPublication(snapshot)
143+ require.NoError(t, err)
144+ manifest, err := snapcodec.UnmarshalManifest(publication.Manifest)
145+ require.NoError(t, err)
146+ manifest.Generation = generation
147+ encoded, err := snapcodec.MarshalManifest(manifest)
148+ require.NoError(t, err)
149+ objects := make(map[ports.SnapshotDigest][]byte, len(publication.Objects))
150+ for _, object := range publication.Objects {
151+ objects[object.Digest] = append([]byte(nil), object.Data...)
152+ }
153+ return ports.SnapshotGeneration{Name: snapshot.Name, Generation: generation, Manifest: encoded, Objects: objects}
154+}
155+
156+func TestRestoreIncrementalGenerationAcceptance(t *testing.T) {
157+ snapshot := restoreAcceptanceSession(t, "restored")
158+ generation := acceptanceGeneration(t, snapshot, 9)
159+ repository := &snapshotAcceptanceRepository{names: []string{snapshot.Name}, generations: map[string]ports.SnapshotGeneration{snapshot.Name: generation}}
160+ pty, release := newBlockingPTY(t)
161+ d := newTestDaemon(t, newFactory(t, pty), stubClock{})
162+ store, _ := newMockStore(t)
163+ WithStore(store)(d)
164+ WithSnapshotRepository(repository, nil)(d)
165+ t.Cleanup(func() { release(); d.sessWg.Wait() })
166+
167+ d.restoreIncrementalSnapshots(context.Background())
168+
169+ d.mu.Lock()
170+ restored := d.findByNameLocked(snapshot.Name)
171+ d.mu.Unlock()
172+ require.NotNil(t, restored)
173+ require.False(t, restored.snapDirty.Load(), "a loaded generation must begin clean")
174+ require.Equal(t, 0, restored.active)
175+ require.Equal(t, "/snapshot/cwd", restored.cwd)
176+ require.Len(t, restored.tabs, 1)
177+ tab := restored.tabs[0]
178+ require.Equal(t, 293, tab.size.Cols)
179+ require.Equal(t, layout.PaneID("pane-7"), tab.tree.Focus)
180+ pane := tab.panes[layout.PaneID("pane-7")]
181+ pane.mu.Lock()
182+ require.Equal(t, "primary-visible", string([]rune{pane.screen.Frame.Row(0)[0].Rune, pane.screen.Frame.Row(0)[1].Rune, pane.screen.Frame.Row(0)[2].Rune, pane.screen.Frame.Row(0)[3].Rune, pane.screen.Frame.Row(0)[4].Rune, pane.screen.Frame.Row(0)[5].Rune, pane.screen.Frame.Row(0)[6].Rune, pane.screen.Frame.Row(0)[7].Rune, pane.screen.Frame.Row(0)[8].Rune, pane.screen.Frame.Row(0)[9].Rune, pane.screen.Frame.Row(0)[10].Rune, pane.screen.Frame.Row(0)[11].Rune, pane.screen.Frame.Row(0)[12].Rune, pane.screen.Frame.Row(0)[13].Rune, pane.screen.Frame.Row(0)[14].Rune}))
183+ require.Equal(t, 5, pane.history.Len())
184+ require.Equal(t, 5*293, pane.history.Cells())
185+ require.Equal(t, defaultScrollbackRows, pane.history.Cap())
186+ require.Equal(t, defaultScrollbackCells, pane.history.CellCap())
187+ pane.mu.Unlock()
188+}
189+
190+func TestRestoreIncrementalFallbackAndInvalidObjectMappings(t *testing.T) {
191+ snapshot := restoreAcceptanceSession(t, "fallback")
192+ valid := acceptanceGeneration(t, snapshot, 3)
193+ valid.Fallback = true
194+ pty, release := newBlockingPTY(t)
195+ d := newTestDaemon(t, newFactory(t, pty), stubClock{})
196+ store, _ := newMockStore(t)
197+ WithStore(store)(d)
198+ repository := &snapshotAcceptanceRepository{names: []string{snapshot.Name}, generations: map[string]ports.SnapshotGeneration{snapshot.Name: valid}}
199+ WithSnapshotRepository(repository, nil)(d)
200+ t.Cleanup(func() { release(); d.sessWg.Wait() })
201+ d.restoreIncrementalSnapshots(context.Background())
202+ require.Len(t, d.notices.history(), 1)
203+ require.Equal(t, domain.NoticeWarn, d.notices.history()[0].Severity)
204+
205+ for _, mutate := range []struct {
206+ name string
207+ fn func(*ports.SnapshotGeneration)
208+ }{
209+ {"missing", func(g *ports.SnapshotGeneration) {
210+ for digest := range g.Objects {
211+ delete(g.Objects, digest)
212+ break
213+ }
214+ }},
215+ {"wrong kind", func(g *ports.SnapshotGeneration) {
216+ for digest, data := range g.Objects {
217+ kind, payload, err := snapcodec.UnmarshalObject(data)
218+ require.NoError(t, err)
219+ if kind == snapcodec.HistoryTail {
220+ replacement, err := snapcodec.MarshalObject(snapcodec.Visible, payload)
221+ require.NoError(t, err)
222+ delete(g.Objects, digest)
223+ g.Objects[digest] = replacement.Data
224+ break
225+ }
226+ }
227+ }},
228+ {"wrong digest", func(g *ports.SnapshotGeneration) {
229+ for digest, data := range g.Objects {
230+ kind, payload, err := snapcodec.UnmarshalObject(data)
231+ require.NoError(t, err)
232+ payload = append([]byte(nil), payload...)
233+ payload[len(payload)-1] ^= 1
234+ replacement, err := snapcodec.MarshalObject(kind, payload)
235+ require.NoError(t, err)
236+ require.Equal(t, len(data), len(replacement.Data))
237+ g.Objects[digest] = replacement.Data
238+ break
239+ }
240+ }},
241+ } {
242+ t.Run(mutate.name, func(t *testing.T) {
243+ generation := cloneAcceptanceGeneration(valid)
244+ generation.Fallback = false
245+ mutate.fn(&generation)
246+ _, err := sessionFromGeneration(generation)
247+ require.Error(t, err)
248+ })
249+ }
250+}
@@ -0,0 +1,250 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "errors"
6+ "sync"
7+ "testing"
8+
9+ "github.com/stretchr/testify/require"
10+
11+ "github.com/bnema/vev/internal/domain"
12+ "github.com/bnema/vev/internal/ports"
13+ "github.com/bnema/vev/internal/usecase/layout"
14+ snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
15+ "github.com/bnema/vev/pkg/renderer"
16+ "github.com/bnema/vev/pkg/vt"
17+)
18+
19+type snapshotAcceptanceRepository struct {
20+ mu sync.Mutex
21+ names []string
22+ generations map[string]ports.SnapshotGeneration
23+ publishErr error
24+ listErr error
25+ loadErr error
26+ publishes []ports.SnapshotPublication
27+ loadMutate func(ports.SnapshotGeneration, ports.SnapshotPublication) ports.SnapshotGeneration
28+}
29+
30+func (r *snapshotAcceptanceRepository) Publish(ctx context.Context, p ports.SnapshotPublication) error {
31+ if err := ctx.Err(); err != nil {
32+ return err
33+ }
34+ r.mu.Lock()
35+ defer r.mu.Unlock()
36+ r.publishes = append(r.publishes, p)
37+ if r.publishErr != nil {
38+ return r.publishErr
39+ }
40+ objects := make(map[ports.SnapshotDigest][]byte, len(p.Objects))
41+ for _, object := range p.Objects {
42+ objects[object.Digest] = append([]byte(nil), object.Data...)
43+ }
44+ r.generations[p.Name] = ports.SnapshotGeneration{Name: p.Name, Generation: p.Generation, Manifest: append([]byte(nil), p.Manifest...), Objects: objects}
45+ found := false
46+ for _, name := range r.names {
47+ found = found || name == p.Name
48+ }
49+ if !found {
50+ r.names = append(r.names, p.Name)
51+ }
52+ return nil
53+}
54+
55+func (r *snapshotAcceptanceRepository) List(ctx context.Context) ([]string, error) {
56+ if err := ctx.Err(); err != nil {
57+ return nil, err
58+ }
59+ r.mu.Lock()
60+ defer r.mu.Unlock()
61+ if r.listErr != nil {
62+ return nil, r.listErr
63+ }
64+ return append([]string(nil), r.names...), nil
65+}
66+
67+func (r *snapshotAcceptanceRepository) Load(ctx context.Context, name string) (ports.SnapshotGeneration, error) {
68+ if err := ctx.Err(); err != nil {
69+ return ports.SnapshotGeneration{}, err
70+ }
71+ r.mu.Lock()
72+ defer r.mu.Unlock()
73+ if r.loadErr != nil {
74+ return ports.SnapshotGeneration{}, r.loadErr
75+ }
76+ generation, ok := r.generations[name]
77+ if !ok {
78+ return ports.SnapshotGeneration{}, errors.New("missing generation")
79+ }
80+ if r.loadMutate != nil && len(r.publishes) > 0 {
81+ generation = r.loadMutate(generation, r.publishes[len(r.publishes)-1])
82+ }
83+ return cloneAcceptanceGeneration(generation), nil
84+}
85+
86+func (*snapshotAcceptanceRepository) Delete(context.Context, string) error { return nil }
87+func (*snapshotAcceptanceRepository) Maintain(context.Context) error { return nil }
88+
89+func cloneAcceptanceGeneration(g ports.SnapshotGeneration) ports.SnapshotGeneration {
90+ out := g
91+ out.Manifest = append([]byte(nil), g.Manifest...)
92+ out.Objects = make(map[ports.SnapshotDigest][]byte, len(g.Objects))
93+ for digest, data := range g.Objects {
94+ out.Objects[digest] = append([]byte(nil), data...)
95+ }
96+ return out
97+}
98+
99+func restoreAcceptanceSession(t *testing.T, name string) snapcodec.Session {
100+ t.Helper()
101+ history := vt.NewHistory(vt.HistoryConfig{MaxRows: 8, MaxCells: 8 * 293, ChunkRows: 2})
102+ for _, ch := range []rune{'a', 'b', 'c', 'd', 'e'} {
103+ row := make([]renderer.Cell, 293)
104+ for i := range row {
105+ row[i] = renderer.Cell{Rune: ch}
106+ }
107+ require.NoError(t, history.Append(row))
108+ }
109+ view := history.SnapshotView()
110+ sealed := make([][]byte, view.ChunkCount())
111+ for i := range sealed {
112+ var err error
113+ sealed[i], err = vt.MarshalHistoryChunk(view.Chunk(i))
114+ require.NoError(t, err)
115+ }
116+ tail, err := vt.MarshalHistoryTail(view)
117+ require.NoError(t, err)
118+
119+ screen := vt.NewScreen(293, 4)
120+ screen.Write([]byte("primary-visible"))
121+ screen.Write([]byte("\x1b[?1049h"))
122+ screen.Write([]byte("alternate-active"))
123+ visible, err := screen.MarshalPrimaryVisible()
124+ require.NoError(t, err)
125+
126+ paneID := layout.PaneID("pane-7")
127+ return snapcodec.Session{
128+ Name: name, CreatedAt: 77, Active: 0,
129+ Tabs: []snapcodec.Tab{{
130+ StableID: "tab-stable", Cols: 293, Rows: 4, NextPaneID: 8, Focus: paneID,
131+ Tree: &layout.Tree{Root: layout.NewLeaf(paneID), Focus: paneID},
132+ Panes: []snapcodec.Pane{{
133+ ID: paneID, StableID: "pane-stable", Cwd: "/snapshot/cwd", SealedChunks: sealed, Tail: tail, Visible: visible,
134+ Process: &snapcodec.Process{Argv: []string{"sleep", "99"}, Strategy: "argv"},
135+ }},
136+ }},
137+ }
138+}
139+
140+func acceptanceGeneration(t *testing.T, snapshot snapcodec.Session, generation uint64) ports.SnapshotGeneration {
141+ t.Helper()
142+ publication, err := legacyPublication(snapshot)
143+ require.NoError(t, err)
144+ manifest, err := snapcodec.UnmarshalManifest(publication.Manifest)
145+ require.NoError(t, err)
146+ manifest.Generation = generation
147+ encoded, err := snapcodec.MarshalManifest(manifest)
148+ require.NoError(t, err)
149+ objects := make(map[ports.SnapshotDigest][]byte, len(publication.Objects))
150+ for _, object := range publication.Objects {
151+ objects[object.Digest] = append([]byte(nil), object.Data...)
152+ }
153+ return ports.SnapshotGeneration{Name: snapshot.Name, Generation: generation, Manifest: encoded, Objects: objects}
154+}
155+
156+func TestRestoreIncrementalGenerationAcceptance(t *testing.T) {
157+ snapshot := restoreAcceptanceSession(t, "restored")
158+ generation := acceptanceGeneration(t, snapshot, 9)
159+ repository := &snapshotAcceptanceRepository{names: []string{snapshot.Name}, generations: map[string]ports.SnapshotGeneration{snapshot.Name: generation}}
160+ pty, release := newBlockingPTY(t)
161+ d := newTestDaemon(t, newFactory(t, pty), stubClock{})
162+ store, _ := newMockStore(t)
163+ WithStore(store)(d)
164+ WithSnapshotRepository(repository, nil)(d)
165+ t.Cleanup(func() { release(); d.sessWg.Wait() })
166+
167+ d.restoreIncrementalSnapshots(context.Background())
168+
169+ d.mu.Lock()
170+ restored := d.findByNameLocked(snapshot.Name)
171+ d.mu.Unlock()
172+ require.NotNil(t, restored)
173+ require.False(t, restored.snapDirty.Load(), "a loaded generation must begin clean")
174+ require.Equal(t, 0, restored.active)
175+ require.Equal(t, "/snapshot/cwd", restored.cwd)
176+ require.Len(t, restored.tabs, 1)
177+ tab := restored.tabs[0]
178+ require.Equal(t, 293, tab.size.Cols)
179+ require.Equal(t, layout.PaneID("pane-7"), tab.tree.Focus)
180+ pane := tab.panes[layout.PaneID("pane-7")]
181+ pane.mu.Lock()
182+ require.Equal(t, "primary-visible", string([]rune{pane.screen.Frame.Row(0)[0].Rune, pane.screen.Frame.Row(0)[1].Rune, pane.screen.Frame.Row(0)[2].Rune, pane.screen.Frame.Row(0)[3].Rune, pane.screen.Frame.Row(0)[4].Rune, pane.screen.Frame.Row(0)[5].Rune, pane.screen.Frame.Row(0)[6].Rune, pane.screen.Frame.Row(0)[7].Rune, pane.screen.Frame.Row(0)[8].Rune, pane.screen.Frame.Row(0)[9].Rune, pane.screen.Frame.Row(0)[10].Rune, pane.screen.Frame.Row(0)[11].Rune, pane.screen.Frame.Row(0)[12].Rune, pane.screen.Frame.Row(0)[13].Rune, pane.screen.Frame.Row(0)[14].Rune}))
183+ require.Equal(t, 5, pane.history.Len())
184+ require.Equal(t, 5*293, pane.history.Cells())
185+ require.Equal(t, defaultScrollbackRows, pane.history.Cap())
186+ require.Equal(t, defaultScrollbackCells, pane.history.CellCap())
187+ pane.mu.Unlock()
188+}
189+
190+func TestRestoreIncrementalFallbackAndInvalidObjectMappings(t *testing.T) {
191+ snapshot := restoreAcceptanceSession(t, "fallback")
192+ valid := acceptanceGeneration(t, snapshot, 3)
193+ valid.Fallback = true
194+ pty, release := newBlockingPTY(t)
195+ d := newTestDaemon(t, newFactory(t, pty), stubClock{})
196+ store, _ := newMockStore(t)
197+ WithStore(store)(d)
198+ repository := &snapshotAcceptanceRepository{names: []string{snapshot.Name}, generations: map[string]ports.SnapshotGeneration{snapshot.Name: valid}}
199+ WithSnapshotRepository(repository, nil)(d)
200+ t.Cleanup(func() { release(); d.sessWg.Wait() })
201+ d.restoreIncrementalSnapshots(context.Background())
202+ require.Len(t, d.notices.history(), 1)
203+ require.Equal(t, domain.NoticeWarn, d.notices.history()[0].Severity)
204+
205+ for _, mutate := range []struct {
206+ name string
207+ fn func(*ports.SnapshotGeneration)
208+ }{
209+ {"missing", func(g *ports.SnapshotGeneration) {
210+ for digest := range g.Objects {
211+ delete(g.Objects, digest)
212+ break
213+ }
214+ }},
215+ {"wrong kind", func(g *ports.SnapshotGeneration) {
216+ for digest, data := range g.Objects {
217+ kind, payload, err := snapcodec.UnmarshalObject(data)
218+ require.NoError(t, err)
219+ if kind == snapcodec.HistoryTail {
220+ replacement, err := snapcodec.MarshalObject(snapcodec.Visible, payload)
221+ require.NoError(t, err)
222+ delete(g.Objects, digest)
223+ g.Objects[digest] = replacement.Data
224+ break
225+ }
226+ }
227+ }},
228+ {"wrong digest", func(g *ports.SnapshotGeneration) {
229+ for digest, data := range g.Objects {
230+ kind, payload, err := snapcodec.UnmarshalObject(data)
231+ require.NoError(t, err)
232+ payload = append([]byte(nil), payload...)
233+ payload[len(payload)-1] ^= 1
234+ replacement, err := snapcodec.MarshalObject(kind, payload)
235+ require.NoError(t, err)
236+ require.Equal(t, len(data), len(replacement.Data))
237+ g.Objects[digest] = replacement.Data
238+ break
239+ }
240+ }},
241+ } {
242+ t.Run(mutate.name, func(t *testing.T) {
243+ generation := cloneAcceptanceGeneration(valid)
244+ generation.Fallback = false
245+ mutate.fn(&generation)
246+ _, err := sessionFromGeneration(generation)
247+ require.Error(t, err)
248+ })
249+ }
250+}
M internal/usecase/snapshot/preflight.go +4 -1
@@ -214,7 +214,10 @@ func preflightPaneStructure(r *payloadReader, totals *vt.DecodeStats, blobs *uin
214214 if err != nil {
215215 return err
216216 }
217- if tail.Chunks != 0 || tail.Rows != 0 || tail.Cells != 0 {
217+ // A capture keeps the live mutable tail separate from immutable sealed
218+ // chunks. It may therefore contain one partial chunk (or the canonical
219+ // empty tail), but never a sequence of sealed chunks.
220+ if tail.Chunks > 1 {
218221 return fmt.Errorf("%w: tail blob role", ErrInvalidData)
219222 }
220223 visible, err := preflightBlob(r, totals, blobs, false, budget)
@@ -214,7 +214,10 @@ func preflightPaneStructure(r *payloadReader, totals *vt.DecodeStats, blobs *uin
214 if err != nil {214 if err != nil {
215 return err215 return err
216 }216 }
217- if tail.Chunks != 0 || tail.Rows != 0 || tail.Cells != 0 {217+ // A capture keeps the live mutable tail separate from immutable sealed
218+ // chunks. It may therefore contain one partial chunk (or the canonical
219+ // empty tail), but never a sequence of sealed chunks.
220+ if tail.Chunks > 1 {
218 return fmt.Errorf("%w: tail blob role", ErrInvalidData)221 return fmt.Errorf("%w: tail blob role", ErrInvalidData)
219 }222 }
220 visible, err := preflightBlob(r, totals, blobs, false, budget)223 visible, err := preflightBlob(r, totals, blobs, false, budget)

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close