snapshot_resume_publication_test.go Raw History Folder

1package daemon
2
3import (
4	"bytes"
5	"context"
6	"crypto/sha256"
7	"fmt"
8	"sync"
9	"testing"
10
11	"github.com/stretchr/testify/require"
12
13	"github.com/bnema/vev/internal/domain"
14	"github.com/bnema/vev/internal/ports"
15)
16
17type snapshotLifecycleRepository struct {
18	mu          sync.Mutex
19	current     map[domain.IncarnationID]ports.CheckpointRef
20	generations map[domain.IncarnationID]map[ports.CheckpointRef]ports.SnapshotGeneration
21	attempts    []ports.SnapshotPublication
22	errors      []error
23}
24
25func (r *snapshotLifecycleRepository) Publish(ctx context.Context, publication ports.SnapshotPublication) error {
26	r.mu.Lock()
27	defer r.mu.Unlock()
28
29	err := r.publish(ctx, publication)
30	r.attempts = append(r.attempts, publication)
31	r.errors = append(r.errors, err)
32	return err
33}
34
35func (r *snapshotLifecycleRepository) publish(ctx context.Context, publication ports.SnapshotPublication) error {
36	if err := ctx.Err(); err != nil {
37		return err
38	}
39	if publication.IncarnationID == (domain.IncarnationID{}) || publication.Name == "" || publication.Generation == 0 {
40		return fmt.Errorf("invalid snapshot publication identity")
41	}
42
43	current, exists := r.current[publication.IncarnationID]
44	if !exists {
45		if publication.Generation != 1 || publication.ParentCheckpoint != nil {
46			return fmt.Errorf("first snapshot generation must be 1 with no parent")
47		}
48	} else {
49		generation := r.generations[publication.IncarnationID][current]
50		if generation.Name != publication.Name {
51			return fmt.Errorf("snapshot name does not match current checkpoint")
52		}
53		if publication.Generation == current.Generation && bytes.Equal(publication.Manifest, generation.Manifest) {
54			return nil
55		}
56		if publication.Generation != current.Generation+1 || publication.ParentCheckpoint == nil || *publication.ParentCheckpoint != current {
57			return fmt.Errorf("snapshot generation or parent does not follow current checkpoint")
58		}
59	}
60
61	ref := ports.CheckpointRef{Generation: publication.Generation, ManifestDigest: sha256.Sum256(publication.Manifest)}
62	objects := make(map[ports.SnapshotDigest][]byte, len(publication.Objects))
63	for _, object := range publication.Objects {
64		objects[object.Digest] = append([]byte(nil), object.Data...)
65	}
66	generation := ports.SnapshotGeneration{
67		IncarnationID:    publication.IncarnationID,
68		Name:             publication.Name,
69		Generation:       publication.Generation,
70		ParentCheckpoint: cloneCheckpointRef(publication.ParentCheckpoint),
71		Manifest:         append([]byte(nil), publication.Manifest...),
72		Objects:          objects,
73	}
74	if r.generations[publication.IncarnationID] == nil {
75		r.generations[publication.IncarnationID] = make(map[ports.CheckpointRef]ports.SnapshotGeneration)
76	}
77	r.generations[publication.IncarnationID][ref] = generation
78	r.current[publication.IncarnationID] = ref
79	return nil
80}
81
82func (r *snapshotLifecycleRepository) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {
83	r.mu.Lock()
84	defer r.mu.Unlock()
85	if err := ctx.Err(); err != nil {
86		return ports.SnapshotGeneration{}, err
87	}
88	generation, ok := r.generations[id][ref]
89	if !ok || generation.Name != name {
90		return ports.SnapshotGeneration{}, fmt.Errorf("checkpoint not found")
91	}
92	return generation, nil
93}
94
95func (r *snapshotLifecycleRepository) ReconcileCheckpoint(ctx context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
96	r.mu.Lock()
97	defer r.mu.Unlock()
98	if err := ctx.Err(); err != nil {
99		return err
100	}
101	if _, ok := r.generations[id][ref]; !ok {
102		return fmt.Errorf("checkpoint not found")
103	}
104	r.current[id] = ref
105	return nil
106}
107
108func (r *snapshotLifecycleRepository) DeleteIncarnation(ctx context.Context, id domain.IncarnationID) error {
109	r.mu.Lock()
110	defer r.mu.Unlock()
111	if err := ctx.Err(); err != nil {
112		return err
113	}
114	delete(r.current, id)
115	delete(r.generations, id)
116	return nil
117}
118
119func (r *snapshotLifecycleRepository) CollectGarbage(ctx context.Context, _ map[domain.IncarnationID]domain.CheckpointRef) error {
120	return ctx.Err()
121}
122
123func (r *snapshotLifecycleRepository) results() ([]ports.SnapshotPublication, []error) {
124	r.mu.Lock()
125	defer r.mu.Unlock()
126	return append([]ports.SnapshotPublication(nil), r.attempts...), append([]error(nil), r.errors...)
127}
128
129func newSnapshotLifecycleRepository(testing.TB) *snapshotLifecycleRepository {
130	return &snapshotLifecycleRepository{
131		current:     make(map[domain.IncarnationID]ports.CheckpointRef),
132		generations: make(map[domain.IncarnationID]map[ports.CheckpointRef]ports.SnapshotGeneration),
133	}
134}
135
136func cloneCheckpointRef(ref *domain.CheckpointRef) *domain.CheckpointRef {
137	if ref == nil {
138		return nil
139	}
140	clone := *ref
141	return &clone
142}
143
144func TestStoppedNamedSessionResumePublishesNextCheckpointInSameDaemon(t *testing.T) {
145	keeperPTY, releaseKeeper := newBlockingPTY(t)
146	firstPTY, releaseFirst := newBlockingPTY(t)
147	resumedPTY, releaseResumed := newBlockingPTY(t)
148	repository := newSnapshotLifecycleRepository(t)
149	d := newTestDaemon(t, newFactorySeq(t, keeperPTY, firstPTY, resumedPTY), stubClock{})
150	WithSnapshotRepository(repository)(d)
151	store, _ := newMockStore(t)
152	WithStore(t, store)(d)
153	startSnapshotEncodeWorker(t, d)
154	t.Cleanup(func() {
155		d.snapsEnabled = false
156		releaseKeeper()
157		releaseFirst()
158		releaseResumed()
159		d.sessWg.Wait()
160	})
161
162	_, _, err := d.route(snapshotLifecycleHello(ports.IntentEphemeral, ""), &closeTrackingTransport{})
163	require.NoError(t, err, "keeper session setup failed")
164	sess, _, err := d.route(snapshotLifecycleHello(ports.IntentNew, "work"), &closeTrackingTransport{})
165	require.NoError(t, err)
166
167	testAttachmentTab(sess).focusedPane().screen.Write([]byte("first checkpoint"))
168	markSnapshotDirty(sess)
169	require.True(t, d.scheduleSnapshot(sess))
170	awaitSnapshotClean(t, sess)
171	firstRecord := snapshotLifecycleRecord(t, d, sess.name)
172	require.NotNil(t, firstRecord.Committed)
173	require.Equal(t, uint64(1), firstRecord.Committed.Generation)
174
175	require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, false))
176	d.mu.Lock()
177	stopped, retained := d.stopped[sess.name]
178	d.mu.Unlock()
179	require.True(t, retained, "non-purged named session must remain stopped")
180	require.False(t, stopped.purging)
181	closedRecord := snapshotLifecycleRecord(t, d, sess.name)
182	require.NotNil(t, closedRecord.Committed)
183	require.Equal(t, firstRecord.Committed.Generation+1, closedRecord.Committed.Generation,
184		"terminal checkpoint must be authoritative before same-daemon resume")
185
186	resumed, _, err := d.route(snapshotLifecycleHello(ports.IntentAttach, sess.name), &closeTrackingTransport{})
187	require.NoError(t, err)
188	require.NotSame(t, sess, resumed, "stopped-session attach must create a new live runtime")
189	require.Equal(t, sess.incarnation, resumed.incarnation, "resume must retain durable incarnation authority")
190
191	testAttachmentTab(resumed).focusedPane().screen.Write([]byte("checkpoint after resume"))
192	markSnapshotDirty(resumed)
193	require.True(t, d.scheduleSnapshot(resumed))
194	awaitSnapshotIdle(t, resumed)
195
196	resumedRecord := snapshotLifecycleRecord(t, d, resumed.name)
197	require.NotNil(t, resumedRecord.Committed)
198	attempts, publicationErrors := repository.results()
199	require.Len(t, attempts, 3, "the resumed checkpoint must reach the repository after the initial and terminal checkpoints")
200	require.NoError(t, publicationErrors[2])
201	require.False(t, resumed.snapDirty.Load(), "the published resumed checkpoint must clear dirty state")
202	d.snapshotNoticeMu.Lock()
203	failure := d.snapshotActiveFailureSignature
204	d.snapshotNoticeMu.Unlock()
205	require.Empty(t, failure, "successful resumed publication must not retain a failure signature")
206	require.Equal(t, closedRecord.Committed.Generation+1, resumedRecord.Committed.Generation,
207		"same-daemon resume must publish exactly committed generation+1")
208	generation, err := repository.LoadCheckpoint(context.Background(), resumed.incarnation, resumed.name, *resumedRecord.Committed)
209	require.NoError(t, err)
210	require.Equal(t, closedRecord.Committed, generation.ParentCheckpoint,
211		"same-daemon resume must use the committed terminal checkpoint as parent")
212}
213
214func snapshotLifecycleHello(intent uint8, name string) ports.Hello {
215	return ports.Hello{
216		Version: ports.ProtocolVersion,
217		Intent:  intent,
218		Name:    name,
219		Cwd:     "/tmp",
220		Size:    domain.Size{Cols: 80, Rows: 24},
221	}
222}
223
224func snapshotLifecycleRecord(t testing.TB, d *Daemon, name string) domain.CatalogueRecord {
225	t.Helper()
226	record, ok, err := d.catalogue.Record(name)
227	require.NoError(t, err)
228	require.True(t, ok, "catalogue record %q not found", name)
229	return record
230}
snapshot_resume_publication_test.go RawHistoryFolder

snapshot_resume_publication_test.go Raw History Folder
