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

snapshot_resume_publication_test.go Raw History Folder
