Merge pull request #139 from bnema/fix/snapshot-resume-publication
fix(snapshot): preserve lineage when resuming sessions
2 files · +294 -0
M internal/usecase/daemon/session.go +5 -0
| @@ -319,6 +319,11 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz | |||
| 319 | 319 | return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err) | |
| 320 | 320 | } | |
| 321 | 321 | sess.incarnation = record.IncarnationID | |
| 322 | + | if resuming && authoritativeExists && authoritative.Committed != nil && authoritative.IncarnationID == sess.incarnation { | |
| 323 | + | checkpoint := *authoritative.Committed | |
| 324 | + | sess.snapshotPublishedGeneration = checkpoint.Generation | |
| 325 | + | sess.snapshotPublishedCheckpoint = &checkpoint | |
| 326 | + | } | |
| 322 | 327 | } | |
| 323 | 328 | delete(d.stopped, name) | |
| 324 | 329 | } | |
| @@ -319,6 +319,11 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz | |||||
| 319 | return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err) | 319 | return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err) | ||
| 320 | } | 320 | } | ||
| 321 | sess.incarnation = record.IncarnationID | 321 | sess.incarnation = record.IncarnationID | ||
| 322 | + | if resuming && authoritativeExists && authoritative.Committed != nil && authoritative.IncarnationID == sess.incarnation { | |||
| 323 | + | checkpoint := *authoritative.Committed | |||
| 324 | + | sess.snapshotPublishedGeneration = checkpoint.Generation | |||
| 325 | + | sess.snapshotPublishedCheckpoint = &checkpoint | |||
| 326 | + | } | |||
| 322 | } | 327 | } | ||
| 323 | delete(d.stopped, name) | 328 | delete(d.stopped, name) | ||
| 324 | } | 329 | } | ||
A internal/usecase/daemon/snapshot_resume_publication_test.go +289 -0
| @@ -0,0 +1,289 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 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 | + | ||
| 17 | + | type 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 | + | ||
| 25 | + | func (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 | + | ||
| 35 | + | func (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 | + | ||
| 82 | + | func (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 | + | ||
| 95 | + | func (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 | + | ||
| 108 | + | func (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 | + | ||
| 119 | + | func (r *snapshotLifecycleRepository) CollectGarbage(ctx context.Context, _ map[domain.IncarnationID]domain.CheckpointRef) error { | |
| 120 | + | return ctx.Err() | |
| 121 | + | } | |
| 122 | + | ||
| 123 | + | func (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 | + | ||
| 129 | + | func 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 | + | ||
| 136 | + | func cloneCheckpointRef(ref *domain.CheckpointRef) *domain.CheckpointRef { | |
| 137 | + | if ref == nil { | |
| 138 | + | return nil | |
| 139 | + | } | |
| 140 | + | clone := *ref | |
| 141 | + | return &clone | |
| 142 | + | } | |
| 143 | + | ||
| 144 | + | func TestNamedSessionPublishesNextCheckpointAfterClientSnatch(t *testing.T) { | |
| 145 | + | pty, release := newBlockingPTY(t) | |
| 146 | + | repository := newSnapshotLifecycleRepository(t) | |
| 147 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 148 | + | WithSnapshotRepository(repository)(d) | |
| 149 | + | store, _ := newMockStore(t) | |
| 150 | + | WithStore(t, store)(d) | |
| 151 | + | startSnapshotEncodeWorker(t, d) | |
| 152 | + | t.Cleanup(func() { | |
| 153 | + | d.snapsEnabled = false | |
| 154 | + | release() | |
| 155 | + | d.sessWg.Wait() | |
| 156 | + | }) | |
| 157 | + | ||
| 158 | + | firstTransport := &closeTrackingTransport{} | |
| 159 | + | sess, firstClient, err := d.route(snapshotLifecycleHello(ports.IntentNew, "work"), firstTransport) | |
| 160 | + | require.NoError(t, err) | |
| 161 | + | ||
| 162 | + | sess.activeTab().focusedPane().screen.Write([]byte("first checkpoint")) | |
| 163 | + | markSnapshotDirty(sess) | |
| 164 | + | require.True(t, d.scheduleSnapshot(sess)) | |
| 165 | + | awaitSnapshotIdle(t, sess) | |
| 166 | + | d.snapshotNoticeMu.Lock() | |
| 167 | + | firstFailure := d.snapshotActiveFailureSignature | |
| 168 | + | d.snapshotNoticeMu.Unlock() | |
| 169 | + | attempts, publicationErrors := repository.results() | |
| 170 | + | require.Len(t, attempts, 1) | |
| 171 | + | require.NoError(t, publicationErrors[0]) | |
| 172 | + | require.Empty(t, firstFailure) | |
| 173 | + | require.False(t, sess.snapDirty.Load()) | |
| 174 | + | firstRecord := snapshotLifecycleRecord(t, d, sess.name) | |
| 175 | + | require.NotNil(t, firstRecord.Committed) | |
| 176 | + | require.Equal(t, uint64(1), firstRecord.Committed.Generation) | |
| 177 | + | ||
| 178 | + | secondTransport := &closeTrackingTransport{} | |
| 179 | + | snatchedSession, secondClient, err := d.route(snapshotLifecycleHello(ports.IntentAttach, "work"), secondTransport) | |
| 180 | + | require.NoError(t, err) | |
| 181 | + | require.Same(t, sess, snatchedSession, "client snatch must retain the live session") | |
| 182 | + | require.NotSame(t, firstClient, secondClient, "client B must displace client A") | |
| 183 | + | ||
| 184 | + | sess.activeTab().focusedPane().screen.Write([]byte("second checkpoint")) | |
| 185 | + | markSnapshotDirty(sess) | |
| 186 | + | require.True(t, d.scheduleFinalSnapshot(sess)) | |
| 187 | + | awaitSnapshotIdle(t, sess) | |
| 188 | + | attempts, publicationErrors = repository.results() | |
| 189 | + | require.Len(t, attempts, 2) | |
| 190 | + | require.NoError(t, publicationErrors[1]) | |
| 191 | + | require.False(t, sess.snapDirty.Load()) | |
| 192 | + | ||
| 193 | + | secondRecord := snapshotLifecycleRecord(t, d, sess.name) | |
| 194 | + | require.NotNil(t, secondRecord.Committed) | |
| 195 | + | require.Equal(t, firstRecord.Committed.Generation+1, secondRecord.Committed.Generation, | |
| 196 | + | "client snatch must not reset or conflict with the repository generation") | |
| 197 | + | generation, err := repository.LoadCheckpoint(context.Background(), sess.incarnation, sess.name, *secondRecord.Committed) | |
| 198 | + | require.NoError(t, err) | |
| 199 | + | require.Equal(t, firstRecord.Committed, generation.ParentCheckpoint, | |
| 200 | + | "the post-snatch checkpoint must retain the committed parent") | |
| 201 | + | } | |
| 202 | + | ||
| 203 | + | func TestStoppedNamedSessionResumePublishesNextCheckpointInSameDaemon(t *testing.T) { | |
| 204 | + | keeperPTY, releaseKeeper := newBlockingPTY(t) | |
| 205 | + | firstPTY, releaseFirst := newBlockingPTY(t) | |
| 206 | + | resumedPTY, releaseResumed := newBlockingPTY(t) | |
| 207 | + | repository := newSnapshotLifecycleRepository(t) | |
| 208 | + | d := newTestDaemon(t, newFactorySeq(t, keeperPTY, firstPTY, resumedPTY), stubClock{}) | |
| 209 | + | WithSnapshotRepository(repository)(d) | |
| 210 | + | store, _ := newMockStore(t) | |
| 211 | + | WithStore(t, store)(d) | |
| 212 | + | startSnapshotEncodeWorker(t, d) | |
| 213 | + | t.Cleanup(func() { | |
| 214 | + | d.snapsEnabled = false | |
| 215 | + | releaseKeeper() | |
| 216 | + | releaseFirst() | |
| 217 | + | releaseResumed() | |
| 218 | + | d.sessWg.Wait() | |
| 219 | + | }) | |
| 220 | + | ||
| 221 | + | _, _, err := d.route(snapshotLifecycleHello(ports.IntentEphemeral, ""), &closeTrackingTransport{}) | |
| 222 | + | require.NoError(t, err, "keeper session setup failed") | |
| 223 | + | sess, _, err := d.route(snapshotLifecycleHello(ports.IntentNew, "work"), &closeTrackingTransport{}) | |
| 224 | + | require.NoError(t, err) | |
| 225 | + | ||
| 226 | + | sess.activeTab().focusedPane().screen.Write([]byte("first checkpoint")) | |
| 227 | + | markSnapshotDirty(sess) | |
| 228 | + | require.True(t, d.scheduleSnapshot(sess)) | |
| 229 | + | awaitSnapshotClean(t, sess) | |
| 230 | + | firstRecord := snapshotLifecycleRecord(t, d, sess.name) | |
| 231 | + | require.NotNil(t, firstRecord.Committed) | |
| 232 | + | require.Equal(t, uint64(1), firstRecord.Committed.Generation) | |
| 233 | + | ||
| 234 | + | require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, false)) | |
| 235 | + | d.mu.Lock() | |
| 236 | + | stopped, retained := d.stopped[sess.name] | |
| 237 | + | d.mu.Unlock() | |
| 238 | + | require.True(t, retained, "non-purged named session must remain stopped") | |
| 239 | + | require.False(t, stopped.purging) | |
| 240 | + | closedRecord := snapshotLifecycleRecord(t, d, sess.name) | |
| 241 | + | require.NotNil(t, closedRecord.Committed) | |
| 242 | + | require.Equal(t, firstRecord.Committed.Generation+1, closedRecord.Committed.Generation, | |
| 243 | + | "terminal checkpoint must be authoritative before same-daemon resume") | |
| 244 | + | ||
| 245 | + | resumed, _, err := d.route(snapshotLifecycleHello(ports.IntentAttach, sess.name), &closeTrackingTransport{}) | |
| 246 | + | require.NoError(t, err) | |
| 247 | + | require.NotSame(t, sess, resumed, "stopped-session attach must create a new live runtime") | |
| 248 | + | require.Equal(t, sess.incarnation, resumed.incarnation, "resume must retain durable incarnation authority") | |
| 249 | + | ||
| 250 | + | resumed.activeTab().focusedPane().screen.Write([]byte("checkpoint after resume")) | |
| 251 | + | markSnapshotDirty(resumed) | |
| 252 | + | require.True(t, d.scheduleSnapshot(resumed)) | |
| 253 | + | awaitSnapshotIdle(t, resumed) | |
| 254 | + | ||
| 255 | + | resumedRecord := snapshotLifecycleRecord(t, d, resumed.name) | |
| 256 | + | require.NotNil(t, resumedRecord.Committed) | |
| 257 | + | attempts, publicationErrors := repository.results() | |
| 258 | + | require.Len(t, attempts, 3, "the resumed checkpoint must reach the repository after the initial and terminal checkpoints") | |
| 259 | + | require.NoError(t, publicationErrors[2]) | |
| 260 | + | require.False(t, resumed.snapDirty.Load(), "the published resumed checkpoint must clear dirty state") | |
| 261 | + | d.snapshotNoticeMu.Lock() | |
| 262 | + | failure := d.snapshotActiveFailureSignature | |
| 263 | + | d.snapshotNoticeMu.Unlock() | |
| 264 | + | require.Empty(t, failure, "successful resumed publication must not retain a failure signature") | |
| 265 | + | require.Equal(t, closedRecord.Committed.Generation+1, resumedRecord.Committed.Generation, | |
| 266 | + | "same-daemon resume must publish exactly committed generation+1") | |
| 267 | + | generation, err := repository.LoadCheckpoint(context.Background(), resumed.incarnation, resumed.name, *resumedRecord.Committed) | |
| 268 | + | require.NoError(t, err) | |
| 269 | + | require.Equal(t, closedRecord.Committed, generation.ParentCheckpoint, | |
| 270 | + | "same-daemon resume must use the committed terminal checkpoint as parent") | |
| 271 | + | } | |
| 272 | + | ||
| 273 | + | func snapshotLifecycleHello(intent uint8, name string) ports.Hello { | |
| 274 | + | return ports.Hello{ | |
| 275 | + | Version: ports.ProtocolVersion, | |
| 276 | + | Intent: intent, | |
| 277 | + | Name: name, | |
| 278 | + | Cwd: "/tmp", | |
| 279 | + | Size: domain.Size{Cols: 80, Rows: 24}, | |
| 280 | + | } | |
| 281 | + | } | |
| 282 | + | ||
| 283 | + | func snapshotLifecycleRecord(t testing.TB, d *Daemon, name string) domain.CatalogueRecord { | |
| 284 | + | t.Helper() | |
| 285 | + | record, ok, err := d.catalogue.Record(name) | |
| 286 | + | require.NoError(t, err) | |
| 287 | + | require.True(t, ok, "catalogue record %q not found", name) | |
| 288 | + | return record | |
| 289 | + | } | |
| @@ -0,0 +1,289 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 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 | + | ||||
| 17 | + | type 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 | + | ||||
| 25 | + | func (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 | + | ||||
| 35 | + | func (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 | + | ||||
| 82 | + | func (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 | + | ||||
| 95 | + | func (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 | + | ||||
| 108 | + | func (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 | + | ||||
| 119 | + | func (r *snapshotLifecycleRepository) CollectGarbage(ctx context.Context, _ map[domain.IncarnationID]domain.CheckpointRef) error { | |||
| 120 | + | return ctx.Err() | |||
| 121 | + | } | |||
| 122 | + | ||||
| 123 | + | func (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 | + | ||||
| 129 | + | func 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 | + | ||||
| 136 | + | func cloneCheckpointRef(ref *domain.CheckpointRef) *domain.CheckpointRef { | |||
| 137 | + | if ref == nil { | |||
| 138 | + | return nil | |||
| 139 | + | } | |||
| 140 | + | clone := *ref | |||
| 141 | + | return &clone | |||
| 142 | + | } | |||
| 143 | + | ||||
| 144 | + | func TestNamedSessionPublishesNextCheckpointAfterClientSnatch(t *testing.T) { | |||
| 145 | + | pty, release := newBlockingPTY(t) | |||
| 146 | + | repository := newSnapshotLifecycleRepository(t) | |||
| 147 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 148 | + | WithSnapshotRepository(repository)(d) | |||
| 149 | + | store, _ := newMockStore(t) | |||
| 150 | + | WithStore(t, store)(d) | |||
| 151 | + | startSnapshotEncodeWorker(t, d) | |||
| 152 | + | t.Cleanup(func() { | |||
| 153 | + | d.snapsEnabled = false | |||
| 154 | + | release() | |||
| 155 | + | d.sessWg.Wait() | |||
| 156 | + | }) | |||
| 157 | + | ||||
| 158 | + | firstTransport := &closeTrackingTransport{} | |||
| 159 | + | sess, firstClient, err := d.route(snapshotLifecycleHello(ports.IntentNew, "work"), firstTransport) | |||
| 160 | + | require.NoError(t, err) | |||
| 161 | + | ||||
| 162 | + | sess.activeTab().focusedPane().screen.Write([]byte("first checkpoint")) | |||
| 163 | + | markSnapshotDirty(sess) | |||
| 164 | + | require.True(t, d.scheduleSnapshot(sess)) | |||
| 165 | + | awaitSnapshotIdle(t, sess) | |||
| 166 | + | d.snapshotNoticeMu.Lock() | |||
| 167 | + | firstFailure := d.snapshotActiveFailureSignature | |||
| 168 | + | d.snapshotNoticeMu.Unlock() | |||
| 169 | + | attempts, publicationErrors := repository.results() | |||
| 170 | + | require.Len(t, attempts, 1) | |||
| 171 | + | require.NoError(t, publicationErrors[0]) | |||
| 172 | + | require.Empty(t, firstFailure) | |||
| 173 | + | require.False(t, sess.snapDirty.Load()) | |||
| 174 | + | firstRecord := snapshotLifecycleRecord(t, d, sess.name) | |||
| 175 | + | require.NotNil(t, firstRecord.Committed) | |||
| 176 | + | require.Equal(t, uint64(1), firstRecord.Committed.Generation) | |||
| 177 | + | ||||
| 178 | + | secondTransport := &closeTrackingTransport{} | |||
| 179 | + | snatchedSession, secondClient, err := d.route(snapshotLifecycleHello(ports.IntentAttach, "work"), secondTransport) | |||
| 180 | + | require.NoError(t, err) | |||
| 181 | + | require.Same(t, sess, snatchedSession, "client snatch must retain the live session") | |||
| 182 | + | require.NotSame(t, firstClient, secondClient, "client B must displace client A") | |||
| 183 | + | ||||
| 184 | + | sess.activeTab().focusedPane().screen.Write([]byte("second checkpoint")) | |||
| 185 | + | markSnapshotDirty(sess) | |||
| 186 | + | require.True(t, d.scheduleFinalSnapshot(sess)) | |||
| 187 | + | awaitSnapshotIdle(t, sess) | |||
| 188 | + | attempts, publicationErrors = repository.results() | |||
| 189 | + | require.Len(t, attempts, 2) | |||
| 190 | + | require.NoError(t, publicationErrors[1]) | |||
| 191 | + | require.False(t, sess.snapDirty.Load()) | |||
| 192 | + | ||||
| 193 | + | secondRecord := snapshotLifecycleRecord(t, d, sess.name) | |||
| 194 | + | require.NotNil(t, secondRecord.Committed) | |||
| 195 | + | require.Equal(t, firstRecord.Committed.Generation+1, secondRecord.Committed.Generation, | |||
| 196 | + | "client snatch must not reset or conflict with the repository generation") | |||
| 197 | + | generation, err := repository.LoadCheckpoint(context.Background(), sess.incarnation, sess.name, *secondRecord.Committed) | |||
| 198 | + | require.NoError(t, err) | |||
| 199 | + | require.Equal(t, firstRecord.Committed, generation.ParentCheckpoint, | |||
| 200 | + | "the post-snatch checkpoint must retain the committed parent") | |||
| 201 | + | } | |||
| 202 | + | ||||
| 203 | + | func TestStoppedNamedSessionResumePublishesNextCheckpointInSameDaemon(t *testing.T) { | |||
| 204 | + | keeperPTY, releaseKeeper := newBlockingPTY(t) | |||
| 205 | + | firstPTY, releaseFirst := newBlockingPTY(t) | |||
| 206 | + | resumedPTY, releaseResumed := newBlockingPTY(t) | |||
| 207 | + | repository := newSnapshotLifecycleRepository(t) | |||
| 208 | + | d := newTestDaemon(t, newFactorySeq(t, keeperPTY, firstPTY, resumedPTY), stubClock{}) | |||
| 209 | + | WithSnapshotRepository(repository)(d) | |||
| 210 | + | store, _ := newMockStore(t) | |||
| 211 | + | WithStore(t, store)(d) | |||
| 212 | + | startSnapshotEncodeWorker(t, d) | |||
| 213 | + | t.Cleanup(func() { | |||
| 214 | + | d.snapsEnabled = false | |||
| 215 | + | releaseKeeper() | |||
| 216 | + | releaseFirst() | |||
| 217 | + | releaseResumed() | |||
| 218 | + | d.sessWg.Wait() | |||
| 219 | + | }) | |||
| 220 | + | ||||
| 221 | + | _, _, err := d.route(snapshotLifecycleHello(ports.IntentEphemeral, ""), &closeTrackingTransport{}) | |||
| 222 | + | require.NoError(t, err, "keeper session setup failed") | |||
| 223 | + | sess, _, err := d.route(snapshotLifecycleHello(ports.IntentNew, "work"), &closeTrackingTransport{}) | |||
| 224 | + | require.NoError(t, err) | |||
| 225 | + | ||||
| 226 | + | sess.activeTab().focusedPane().screen.Write([]byte("first checkpoint")) | |||
| 227 | + | markSnapshotDirty(sess) | |||
| 228 | + | require.True(t, d.scheduleSnapshot(sess)) | |||
| 229 | + | awaitSnapshotClean(t, sess) | |||
| 230 | + | firstRecord := snapshotLifecycleRecord(t, d, sess.name) | |||
| 231 | + | require.NotNil(t, firstRecord.Committed) | |||
| 232 | + | require.Equal(t, uint64(1), firstRecord.Committed.Generation) | |||
| 233 | + | ||||
| 234 | + | require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, false)) | |||
| 235 | + | d.mu.Lock() | |||
| 236 | + | stopped, retained := d.stopped[sess.name] | |||
| 237 | + | d.mu.Unlock() | |||
| 238 | + | require.True(t, retained, "non-purged named session must remain stopped") | |||
| 239 | + | require.False(t, stopped.purging) | |||
| 240 | + | closedRecord := snapshotLifecycleRecord(t, d, sess.name) | |||
| 241 | + | require.NotNil(t, closedRecord.Committed) | |||
| 242 | + | require.Equal(t, firstRecord.Committed.Generation+1, closedRecord.Committed.Generation, | |||
| 243 | + | "terminal checkpoint must be authoritative before same-daemon resume") | |||
| 244 | + | ||||
| 245 | + | resumed, _, err := d.route(snapshotLifecycleHello(ports.IntentAttach, sess.name), &closeTrackingTransport{}) | |||
| 246 | + | require.NoError(t, err) | |||
| 247 | + | require.NotSame(t, sess, resumed, "stopped-session attach must create a new live runtime") | |||
| 248 | + | require.Equal(t, sess.incarnation, resumed.incarnation, "resume must retain durable incarnation authority") | |||
| 249 | + | ||||
| 250 | + | resumed.activeTab().focusedPane().screen.Write([]byte("checkpoint after resume")) | |||
| 251 | + | markSnapshotDirty(resumed) | |||
| 252 | + | require.True(t, d.scheduleSnapshot(resumed)) | |||
| 253 | + | awaitSnapshotIdle(t, resumed) | |||
| 254 | + | ||||
| 255 | + | resumedRecord := snapshotLifecycleRecord(t, d, resumed.name) | |||
| 256 | + | require.NotNil(t, resumedRecord.Committed) | |||
| 257 | + | attempts, publicationErrors := repository.results() | |||
| 258 | + | require.Len(t, attempts, 3, "the resumed checkpoint must reach the repository after the initial and terminal checkpoints") | |||
| 259 | + | require.NoError(t, publicationErrors[2]) | |||
| 260 | + | require.False(t, resumed.snapDirty.Load(), "the published resumed checkpoint must clear dirty state") | |||
| 261 | + | d.snapshotNoticeMu.Lock() | |||
| 262 | + | failure := d.snapshotActiveFailureSignature | |||
| 263 | + | d.snapshotNoticeMu.Unlock() | |||
| 264 | + | require.Empty(t, failure, "successful resumed publication must not retain a failure signature") | |||
| 265 | + | require.Equal(t, closedRecord.Committed.Generation+1, resumedRecord.Committed.Generation, | |||
| 266 | + | "same-daemon resume must publish exactly committed generation+1") | |||
| 267 | + | generation, err := repository.LoadCheckpoint(context.Background(), resumed.incarnation, resumed.name, *resumedRecord.Committed) | |||
| 268 | + | require.NoError(t, err) | |||
| 269 | + | require.Equal(t, closedRecord.Committed, generation.ParentCheckpoint, | |||
| 270 | + | "same-daemon resume must use the committed terminal checkpoint as parent") | |||
| 271 | + | } | |||
| 272 | + | ||||
| 273 | + | func snapshotLifecycleHello(intent uint8, name string) ports.Hello { | |||
| 274 | + | return ports.Hello{ | |||
| 275 | + | Version: ports.ProtocolVersion, | |||
| 276 | + | Intent: intent, | |||
| 277 | + | Name: name, | |||
| 278 | + | Cwd: "/tmp", | |||
| 279 | + | Size: domain.Size{Cols: 80, Rows: 24}, | |||
| 280 | + | } | |||
| 281 | + | } | |||
| 282 | + | ||||
| 283 | + | func snapshotLifecycleRecord(t testing.TB, d *Daemon, name string) domain.CatalogueRecord { | |||
| 284 | + | t.Helper() | |||
| 285 | + | record, ok, err := d.catalogue.Record(name) | |||
| 286 | + | require.NoError(t, err) | |||
| 287 | + | require.True(t, ok, "catalogue record %q not found", name) | |||
| 288 | + | return record | |||
| 289 | + | } | |||