fix(daemon): preserve recovery metadata on updates
- commit
- 75792150b93ce4b910c5b1190fd7744afb6fcc81
- author
- brice · · 2026-07-26 00:27 UTC
- parent
- 33d8242
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
8 files · +167 -19
M internal/persist/catalogue_test.go +34 -0
| @@ -16,6 +16,7 @@ func TestCatalogue(t *testing.T) { | |||
| 16 | 16 | t.Run("duplicate-incarnations", testCatalogueDuplicateIncarnations) | |
| 17 | 17 | t.Run("open-fails-closed", testCatalogueOpenFailsClosed) | |
| 18 | 18 | t.Run("apply-rename-replace", testCatalogueApplyRenameReplace) | |
| 19 | + | t.Run("metadata-update-preserves-authority", testCatalogueMetadataUpdatePreservesAuthority) | |
| 19 | 20 | t.Run("read-only-malformed", testCatalogueLoadReadOnlyRejectsMalformedValue) | |
| 20 | 21 | } | |
| 21 | 22 | ||
| @@ -87,6 +88,39 @@ func testCatalogueApplyRenameReplace(t *testing.T) { | |||
| 87 | 88 | require.ElementsMatch(t, []domain.CatalogueRecord{renamed, two}, got) | |
| 88 | 89 | } | |
| 89 | 90 | ||
| 91 | + | func testCatalogueMetadataUpdatePreservesAuthority(t *testing.T) { | |
| 92 | + | dir := privateDir(t) | |
| 93 | + | p, _, err := openCurrentCatalogue(dir, true) | |
| 94 | + | require.NoError(t, err) | |
| 95 | + | defer func() { require.NoError(t, p.Close()) }() | |
| 96 | + | ||
| 97 | + | committed := &domain.CheckpointRef{Generation: 9, ManifestDigest: [32]byte{1}} | |
| 98 | + | fallback1 := &domain.CheckpointRef{Generation: 8, ManifestDigest: [32]byte{2}} | |
| 99 | + | fallback2 := &domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{3}} | |
| 100 | + | original := domain.CatalogueRecord{ | |
| 101 | + | Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/old", CreatedAt: 11, | |
| 102 | + | UpdatedAt: 12, LastUsedSeq: 13, TabNames: []string{"old"}, | |
| 103 | + | RecoveryState: domain.RecoveryDegraded, Committed: committed, | |
| 104 | + | Fallbacks: [2]*domain.CheckpointRef{fallback1, fallback2}, DegradedReason: "repair pending", | |
| 105 | + | } | |
| 106 | + | require.NoError(t, p.Create(original)) | |
| 107 | + | ||
| 108 | + | update := domain.CatalogueRecord{Name: "work", IncarnationID: original.IncarnationID, Cwd: "/new", UpdatedAt: 22, LastUsedSeq: 23, TabNames: []string{"editor", "logs"}} | |
| 109 | + | require.NoError(t, p.UpdateMetadata(update)) | |
| 110 | + | got, ok := p.Record("work") | |
| 111 | + | require.True(t, ok) | |
| 112 | + | expected := original | |
| 113 | + | expected.Cwd, expected.UpdatedAt, expected.LastUsedSeq, expected.TabNames = update.Cwd, update.UpdatedAt, update.LastUsedSeq, update.TabNames | |
| 114 | + | require.Equal(t, expected, got) | |
| 115 | + | ||
| 116 | + | stale := update | |
| 117 | + | stale.IncarnationID = domain.IncarnationID{2} | |
| 118 | + | require.Error(t, p.UpdateMetadata(stale)) | |
| 119 | + | unchanged, ok := p.Record("work") | |
| 120 | + | require.True(t, ok) | |
| 121 | + | require.Equal(t, expected, unchanged) | |
| 122 | + | } | |
| 123 | + | ||
| 90 | 124 | func rawCorruptCatalogueWAL() []byte { | |
| 91 | 125 | // A complete WAL record with an impossible CRC. | |
| 92 | 126 | return []byte{0, 0, 0, 3, 0, 0, 0, 0, 0xff, 0, 0} | |
| @@ -16,6 +16,7 @@ func TestCatalogue(t *testing.T) { | |||||
| 16 | t.Run("duplicate-incarnations", testCatalogueDuplicateIncarnations) | 16 | t.Run("duplicate-incarnations", testCatalogueDuplicateIncarnations) | ||
| 17 | t.Run("open-fails-closed", testCatalogueOpenFailsClosed) | 17 | t.Run("open-fails-closed", testCatalogueOpenFailsClosed) | ||
| 18 | t.Run("apply-rename-replace", testCatalogueApplyRenameReplace) | 18 | t.Run("apply-rename-replace", testCatalogueApplyRenameReplace) | ||
| 19 | + | t.Run("metadata-update-preserves-authority", testCatalogueMetadataUpdatePreservesAuthority) | |||
| 19 | t.Run("read-only-malformed", testCatalogueLoadReadOnlyRejectsMalformedValue) | 20 | t.Run("read-only-malformed", testCatalogueLoadReadOnlyRejectsMalformedValue) | ||
| 20 | } | 21 | } | ||
| 21 | 22 | ||||
| @@ -87,6 +88,39 @@ func testCatalogueApplyRenameReplace(t *testing.T) { | |||||
| 87 | require.ElementsMatch(t, []domain.CatalogueRecord{renamed, two}, got) | 88 | require.ElementsMatch(t, []domain.CatalogueRecord{renamed, two}, got) | ||
| 88 | } | 89 | } | ||
| 89 | 90 | ||||
| 91 | + | func testCatalogueMetadataUpdatePreservesAuthority(t *testing.T) { | |||
| 92 | + | dir := privateDir(t) | |||
| 93 | + | p, _, err := openCurrentCatalogue(dir, true) | |||
| 94 | + | require.NoError(t, err) | |||
| 95 | + | defer func() { require.NoError(t, p.Close()) }() | |||
| 96 | + | ||||
| 97 | + | committed := &domain.CheckpointRef{Generation: 9, ManifestDigest: [32]byte{1}} | |||
| 98 | + | fallback1 := &domain.CheckpointRef{Generation: 8, ManifestDigest: [32]byte{2}} | |||
| 99 | + | fallback2 := &domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{3}} | |||
| 100 | + | original := domain.CatalogueRecord{ | |||
| 101 | + | Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/old", CreatedAt: 11, | |||
| 102 | + | UpdatedAt: 12, LastUsedSeq: 13, TabNames: []string{"old"}, | |||
| 103 | + | RecoveryState: domain.RecoveryDegraded, Committed: committed, | |||
| 104 | + | Fallbacks: [2]*domain.CheckpointRef{fallback1, fallback2}, DegradedReason: "repair pending", | |||
| 105 | + | } | |||
| 106 | + | require.NoError(t, p.Create(original)) | |||
| 107 | + | ||||
| 108 | + | update := domain.CatalogueRecord{Name: "work", IncarnationID: original.IncarnationID, Cwd: "/new", UpdatedAt: 22, LastUsedSeq: 23, TabNames: []string{"editor", "logs"}} | |||
| 109 | + | require.NoError(t, p.UpdateMetadata(update)) | |||
| 110 | + | got, ok := p.Record("work") | |||
| 111 | + | require.True(t, ok) | |||
| 112 | + | expected := original | |||
| 113 | + | expected.Cwd, expected.UpdatedAt, expected.LastUsedSeq, expected.TabNames = update.Cwd, update.UpdatedAt, update.LastUsedSeq, update.TabNames | |||
| 114 | + | require.Equal(t, expected, got) | |||
| 115 | + | ||||
| 116 | + | stale := update | |||
| 117 | + | stale.IncarnationID = domain.IncarnationID{2} | |||
| 118 | + | require.Error(t, p.UpdateMetadata(stale)) | |||
| 119 | + | unchanged, ok := p.Record("work") | |||
| 120 | + | require.True(t, ok) | |||
| 121 | + | require.Equal(t, expected, unchanged) | |||
| 122 | + | } | |||
| 123 | + | ||||
| 90 | func rawCorruptCatalogueWAL() []byte { | 124 | func rawCorruptCatalogueWAL() []byte { | ||
| 91 | // A complete WAL record with an impossible CRC. | 125 | // A complete WAL record with an impossible CRC. | ||
| 92 | return []byte{0, 0, 0, 3, 0, 0, 0, 0, 0xff, 0, 0} | 126 | return []byte{0, 0, 0, 3, 0, 0, 0, 0, 0xff, 0, 0} | ||
M internal/persist/persist.go +35 -2
| @@ -205,8 +205,35 @@ func (p *Persister) Replace(name string, next domain.CatalogueRecord) error { | |||
| 205 | 205 | return p.Apply(map[string]*domain.CatalogueRecord{name: &next}) | |
| 206 | 206 | } | |
| 207 | 207 | ||
| 208 | + | // UpdateMetadata atomically updates ordinary session metadata on the existing | |
| 209 | + | // authoritative incarnation. Recovery and transaction fields are retained from | |
| 210 | + | // the stored record rather than accepted from a potentially stale runtime copy. | |
| 211 | + | func (p *Persister) UpdateMetadata(next domain.CatalogueRecord) error { | |
| 212 | + | if p == nil || p.store == nil { | |
| 213 | + | return nil | |
| 214 | + | } | |
| 215 | + | p.mu.Lock() | |
| 216 | + | defer p.mu.Unlock() | |
| 217 | + | value, ok := p.store.Get([]byte(next.Name)) | |
| 218 | + | if !ok { | |
| 219 | + | return errors.New("persist: session not found") | |
| 220 | + | } | |
| 221 | + | current, err := decodeRecordValue(next.Name, value) | |
| 222 | + | if err != nil { | |
| 223 | + | return err | |
| 224 | + | } | |
| 225 | + | if next.IncarnationID == (domain.IncarnationID{}) || current.IncarnationID != next.IncarnationID { | |
| 226 | + | return errors.New("persist: session incarnation changed") | |
| 227 | + | } | |
| 228 | + | current.Cwd = next.Cwd | |
| 229 | + | current.UpdatedAt = next.UpdatedAt | |
| 230 | + | current.LastUsedSeq = next.LastUsedSeq | |
| 231 | + | current.TabNames = append([]string(nil), next.TabNames...) | |
| 232 | + | return p.applyLocked(map[string]*domain.CatalogueRecord{current.Name: ¤t}) | |
| 233 | + | } | |
| 234 | + | ||
| 208 | 235 | // Touch updates metadata while retaining the durable identity and recovery state. | |
| 209 | - | func (p *Persister) Touch(name, cwd string, at int64) error { | |
| 236 | + | func (p *Persister) Touch(name string, incarnation domain.IncarnationID, cwd string, at int64) error { | |
| 210 | 237 | if p == nil || p.store == nil { | |
| 211 | 238 | return nil | |
| 212 | 239 | } | |
| @@ -220,6 +247,9 @@ func (p *Persister) Touch(name, cwd string, at int64) error { | |||
| 220 | 247 | if err != nil { | |
| 221 | 248 | return err | |
| 222 | 249 | } | |
| 250 | + | if incarnation == (domain.IncarnationID{}) || record.IncarnationID != incarnation { | |
| 251 | + | return errors.New("persist: session incarnation changed") | |
| 252 | + | } | |
| 223 | 253 | record.Cwd, record.UpdatedAt = cwd, at | |
| 224 | 254 | encoded, err := encodeRecordValue(record) | |
| 225 | 255 | if err != nil { | |
| @@ -227,7 +257,7 @@ func (p *Persister) Touch(name, cwd string, at int64) error { | |||
| 227 | 257 | } | |
| 228 | 258 | return p.store.Set([]byte(name), encoded) | |
| 229 | 259 | } | |
| 230 | - | func (p *Persister) TouchMRU(name string, sequence uint64) error { | |
| 260 | + | func (p *Persister) TouchMRU(name string, incarnation domain.IncarnationID, sequence uint64) error { | |
| 231 | 261 | if p == nil || p.store == nil { | |
| 232 | 262 | return nil | |
| 233 | 263 | } | |
| @@ -241,6 +271,9 @@ func (p *Persister) TouchMRU(name string, sequence uint64) error { | |||
| 241 | 271 | if err != nil { | |
| 242 | 272 | return err | |
| 243 | 273 | } | |
| 274 | + | if incarnation == (domain.IncarnationID{}) || record.IncarnationID != incarnation { | |
| 275 | + | return errors.New("persist: session incarnation changed") | |
| 276 | + | } | |
| 244 | 277 | record.LastUsedSeq = sequence | |
| 245 | 278 | encoded, err := encodeRecordValue(record) | |
| 246 | 279 | if err != nil { | |
| @@ -205,8 +205,35 @@ func (p *Persister) Replace(name string, next domain.CatalogueRecord) error { | |||||
| 205 | return p.Apply(map[string]*domain.CatalogueRecord{name: &next}) | 205 | return p.Apply(map[string]*domain.CatalogueRecord{name: &next}) | ||
| 206 | } | 206 | } | ||
| 207 | 207 | ||||
| 208 | + | // UpdateMetadata atomically updates ordinary session metadata on the existing | |||
| 209 | + | // authoritative incarnation. Recovery and transaction fields are retained from | |||
| 210 | + | // the stored record rather than accepted from a potentially stale runtime copy. | |||
| 211 | + | func (p *Persister) UpdateMetadata(next domain.CatalogueRecord) error { | |||
| 212 | + | if p == nil || p.store == nil { | |||
| 213 | + | return nil | |||
| 214 | + | } | |||
| 215 | + | p.mu.Lock() | |||
| 216 | + | defer p.mu.Unlock() | |||
| 217 | + | value, ok := p.store.Get([]byte(next.Name)) | |||
| 218 | + | if !ok { | |||
| 219 | + | return errors.New("persist: session not found") | |||
| 220 | + | } | |||
| 221 | + | current, err := decodeRecordValue(next.Name, value) | |||
| 222 | + | if err != nil { | |||
| 223 | + | return err | |||
| 224 | + | } | |||
| 225 | + | if next.IncarnationID == (domain.IncarnationID{}) || current.IncarnationID != next.IncarnationID { | |||
| 226 | + | return errors.New("persist: session incarnation changed") | |||
| 227 | + | } | |||
| 228 | + | current.Cwd = next.Cwd | |||
| 229 | + | current.UpdatedAt = next.UpdatedAt | |||
| 230 | + | current.LastUsedSeq = next.LastUsedSeq | |||
| 231 | + | current.TabNames = append([]string(nil), next.TabNames...) | |||
| 232 | + | return p.applyLocked(map[string]*domain.CatalogueRecord{current.Name: ¤t}) | |||
| 233 | + | } | |||
| 234 | + | ||||
| 208 | // Touch updates metadata while retaining the durable identity and recovery state. | 235 | // Touch updates metadata while retaining the durable identity and recovery state. | ||
| 209 | - | func (p *Persister) Touch(name, cwd string, at int64) error { | 236 | + | func (p *Persister) Touch(name string, incarnation domain.IncarnationID, cwd string, at int64) error { |
| 210 | if p == nil || p.store == nil { | 237 | if p == nil || p.store == nil { | ||
| 211 | return nil | 238 | return nil | ||
| 212 | } | 239 | } | ||
| @@ -220,6 +247,9 @@ func (p *Persister) Touch(name, cwd string, at int64) error { | |||||
| 220 | if err != nil { | 247 | if err != nil { | ||
| 221 | return err | 248 | return err | ||
| 222 | } | 249 | } | ||
| 250 | + | if incarnation == (domain.IncarnationID{}) || record.IncarnationID != incarnation { | |||
| 251 | + | return errors.New("persist: session incarnation changed") | |||
| 252 | + | } | |||
| 223 | record.Cwd, record.UpdatedAt = cwd, at | 253 | record.Cwd, record.UpdatedAt = cwd, at | ||
| 224 | encoded, err := encodeRecordValue(record) | 254 | encoded, err := encodeRecordValue(record) | ||
| 225 | if err != nil { | 255 | if err != nil { | ||
| @@ -227,7 +257,7 @@ func (p *Persister) Touch(name, cwd string, at int64) error { | |||||
| 227 | } | 257 | } | ||
| 228 | return p.store.Set([]byte(name), encoded) | 258 | return p.store.Set([]byte(name), encoded) | ||
| 229 | } | 259 | } | ||
| 230 | - | func (p *Persister) TouchMRU(name string, sequence uint64) error { | 260 | + | func (p *Persister) TouchMRU(name string, incarnation domain.IncarnationID, sequence uint64) error { |
| 231 | if p == nil || p.store == nil { | 261 | if p == nil || p.store == nil { | ||
| 232 | return nil | 262 | return nil | ||
| 233 | } | 263 | } | ||
| @@ -241,6 +271,9 @@ func (p *Persister) TouchMRU(name string, sequence uint64) error { | |||||
| 241 | if err != nil { | 271 | if err != nil { | ||
| 242 | return err | 272 | return err | ||
| 243 | } | 273 | } | ||
| 274 | + | if incarnation == (domain.IncarnationID{}) || record.IncarnationID != incarnation { | |||
| 275 | + | return errors.New("persist: session incarnation changed") | |||
| 276 | + | } | |||
| 244 | record.LastUsedSeq = sequence | 277 | record.LastUsedSeq = sequence | ||
| 245 | encoded, err := encodeRecordValue(record) | 278 | encoded, err := encodeRecordValue(record) | ||
| 246 | if err != nil { | 279 | if err != nil { | ||
M internal/persist/persist_test.go +2 -2
| @@ -69,8 +69,8 @@ func TestCatalogueNilStoreIsNoOp(t *testing.T) { | |||
| 69 | 69 | for _, p := range persisters { | |
| 70 | 70 | require.NoError(t, p.Save(validRecord("one", 1))) | |
| 71 | 71 | require.NoError(t, p.Apply(map[string]*domain.CatalogueRecord{"one": nil})) | |
| 72 | - | require.NoError(t, p.Touch("one", "/tmp", 1)) | |
| 73 | - | require.NoError(t, p.TouchMRU("one", 1)) | |
| 72 | + | require.NoError(t, p.Touch("one", domain.IncarnationID{1}, "/tmp", 1)) | |
| 73 | + | require.NoError(t, p.TouchMRU("one", domain.IncarnationID{1}, 1)) | |
| 74 | 74 | require.NoError(t, p.Delete("one")) | |
| 75 | 75 | records, err := p.LoadAll() | |
| 76 | 76 | require.NoError(t, err) | |
| @@ -69,8 +69,8 @@ func TestCatalogueNilStoreIsNoOp(t *testing.T) { | |||||
| 69 | for _, p := range persisters { | 69 | for _, p := range persisters { | ||
| 70 | require.NoError(t, p.Save(validRecord("one", 1))) | 70 | require.NoError(t, p.Save(validRecord("one", 1))) | ||
| 71 | require.NoError(t, p.Apply(map[string]*domain.CatalogueRecord{"one": nil})) | 71 | require.NoError(t, p.Apply(map[string]*domain.CatalogueRecord{"one": nil})) | ||
| 72 | - | require.NoError(t, p.Touch("one", "/tmp", 1)) | 72 | + | require.NoError(t, p.Touch("one", domain.IncarnationID{1}, "/tmp", 1)) |
| 73 | - | require.NoError(t, p.TouchMRU("one", 1)) | 73 | + | require.NoError(t, p.TouchMRU("one", domain.IncarnationID{1}, 1)) |
| 74 | require.NoError(t, p.Delete("one")) | 74 | require.NoError(t, p.Delete("one")) | ||
| 75 | records, err := p.LoadAll() | 75 | records, err := p.LoadAll() | ||
| 76 | require.NoError(t, err) | 76 | require.NoError(t, err) | ||
M internal/usecase/daemon/session.go +30 -11
| @@ -183,10 +183,11 @@ func (d *Daemon) touchMRU(sess *session) { | |||
| 183 | 183 | } | |
| 184 | 184 | sess.mu.Lock() | |
| 185 | 185 | name := sess.name | |
| 186 | + | incarnation := sess.incarnation | |
| 186 | 187 | ephemeral := sess.ephemeral | |
| 187 | 188 | sess.mu.Unlock() | |
| 188 | 189 | if !ephemeral && d.persist != nil { | |
| 189 | - | if err := d.persist.TouchMRU(name, seq); err != nil { | |
| 190 | + | if err := d.persist.TouchMRU(name, incarnation, seq); err != nil { | |
| 190 | 191 | d.log.Warn("touching persisted session recency failed", "err", err, "session", name) | |
| 191 | 192 | } | |
| 192 | 193 | } | |
| @@ -203,6 +204,10 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz | |||
| 203 | 204 | // the persisted identity rather than receive a fresh timestamp. | |
| 204 | 205 | createdAt = stopped.createdAt | |
| 205 | 206 | incarnation = stopped.incarnation | |
| 207 | + | if authoritative, ok := d.persist.Record(name); ok { | |
| 208 | + | incarnation = authoritative.IncarnationID | |
| 209 | + | createdAt = authoritative.CreatedAt | |
| 210 | + | } | |
| 206 | 211 | if createdAt > d.lastAllocatedCreatedAt { | |
| 207 | 212 | d.lastAllocatedCreatedAt = createdAt | |
| 208 | 213 | } | |
| @@ -281,10 +286,23 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz | |||
| 281 | 286 | } | |
| 282 | 287 | sess.snapEligible.Store(!ephemeral && name != "") | |
| 283 | 288 | if !ephemeral { | |
| 284 | - | if err := d.persist.Save(persist.Record{Name: name, IncarnationID: incarnation, Cwd: cwd, CreatedAt: createdAt, UpdatedAt: createdAt, LastUsedSeq: lastUsedSeq, TabNames: names, RecoveryState: domain.RecoveryFresh}); err != nil { | |
| 285 | - | closeTabs(tabs) | |
| 286 | - | cancel() | |
| 287 | - | return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err) | |
| 289 | + | if d.persistEnabled { | |
| 290 | + | record := persist.Record{Name: name, IncarnationID: incarnation, Cwd: cwd, CreatedAt: createdAt, UpdatedAt: createdAt, LastUsedSeq: lastUsedSeq, TabNames: names, RecoveryState: domain.RecoveryFresh} | |
| 291 | + | var err error | |
| 292 | + | if resuming { | |
| 293 | + | if _, ok := d.persist.Record(name); ok { | |
| 294 | + | err = d.persist.UpdateMetadata(record) | |
| 295 | + | } else { | |
| 296 | + | err = d.persist.Create(record) | |
| 297 | + | } | |
| 298 | + | } else { | |
| 299 | + | err = d.persist.Create(record) | |
| 300 | + | } | |
| 301 | + | if err != nil { | |
| 302 | + | closeTabs(tabs) | |
| 303 | + | cancel() | |
| 304 | + | return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err) | |
| 305 | + | } | |
| 288 | 306 | } | |
| 289 | 307 | delete(d.stopped, name) | |
| 290 | 308 | } | |
| @@ -426,7 +444,7 @@ func (d *Daemon) createTab(sess *session, sz domain.Size) error { | |||
| 426 | 444 | record := sess.persistRecordLocked(time.Now().UnixNano()) | |
| 427 | 445 | ephemeral := sess.ephemeral | |
| 428 | 446 | if !ephemeral { | |
| 429 | - | if err := d.persist.Save(record); err != nil { | |
| 447 | + | if err := d.persist.UpdateMetadata(record); err != nil { | |
| 430 | 448 | sess.tabs = sess.tabs[:len(sess.tabs)-1] | |
| 431 | 449 | sess.active = oldActive | |
| 432 | 450 | if tb.cancel != nil { | |
| @@ -695,8 +713,8 @@ func (d *Daemon) renameSession(sess *session, name string) error { | |||
| 695 | 713 | // For a named rename this is the durable commit point, deliberately after | |
| 696 | 714 | // the old tombstone transaction. Ephemeral promotion has no old durable | |
| 697 | 715 | // identity and can create its first record directly. | |
| 698 | - | if wasEphemeral || oldName != name { | |
| 699 | - | if err := d.persist.Save(record); err != nil { | |
| 716 | + | if (wasEphemeral || oldName != name) && d.persistEnabled { | |
| 717 | + | if err := d.persist.Create(record); err != nil { | |
| 700 | 718 | return rollback(err) | |
| 701 | 719 | } | |
| 702 | 720 | } | |
| @@ -756,7 +774,7 @@ func (d *Daemon) renameTab(sess *session, tb *tab, name string) error { | |||
| 756 | 774 | record := sess.persistRecordLocked(time.Now().UnixNano()) | |
| 757 | 775 | ephemeral := sess.ephemeral | |
| 758 | 776 | if !ephemeral { | |
| 759 | - | if err := d.persist.Save(record); err != nil { | |
| 777 | + | if err := d.persist.UpdateMetadata(record); err != nil { | |
| 760 | 778 | tb.name = oldName | |
| 761 | 779 | sess.mu.Unlock() | |
| 762 | 780 | d.mu.Unlock() | |
| @@ -835,7 +853,7 @@ func (d *Daemon) closeTab(sess *session, tb *tab, repaint bool) error { | |||
| 835 | 853 | record := sess.persistRecordLocked(time.Now().UnixNano()) | |
| 836 | 854 | ephemeral := sess.ephemeral | |
| 837 | 855 | if !ephemeral { | |
| 838 | - | if err := d.persist.Save(record); err != nil { | |
| 856 | + | if err := d.persist.UpdateMetadata(record); err != nil { | |
| 839 | 857 | d.log.Warn("persisting closed tab failed", "err", err, "session", name) | |
| 840 | 858 | } | |
| 841 | 859 | } | |
| @@ -1336,10 +1354,11 @@ func (d *Daemon) refreshSessionCwd(sess *session) { | |||
| 1336 | 1354 | } | |
| 1337 | 1355 | sess.cwd = cwd | |
| 1338 | 1356 | name := sess.name | |
| 1357 | + | incarnation := sess.incarnation | |
| 1339 | 1358 | ephemeral := sess.ephemeral | |
| 1340 | 1359 | sess.mu.Unlock() | |
| 1341 | 1360 | if !ephemeral { | |
| 1342 | - | if err := d.persist.Touch(name, cwd, time.Now().UnixNano()); err != nil { | |
| 1361 | + | if err := d.persist.Touch(name, incarnation, cwd, time.Now().UnixNano()); err != nil { | |
| 1343 | 1362 | d.log.Warn("touching persisted session cwd failed", "err", err, "session", name) | |
| 1344 | 1363 | } | |
| 1345 | 1364 | markSnapshotDirty(sess) | |
| @@ -183,10 +183,11 @@ func (d *Daemon) touchMRU(sess *session) { | |||||
| 183 | } | 183 | } | ||
| 184 | sess.mu.Lock() | 184 | sess.mu.Lock() | ||
| 185 | name := sess.name | 185 | name := sess.name | ||
| 186 | + | incarnation := sess.incarnation | |||
| 186 | ephemeral := sess.ephemeral | 187 | ephemeral := sess.ephemeral | ||
| 187 | sess.mu.Unlock() | 188 | sess.mu.Unlock() | ||
| 188 | if !ephemeral && d.persist != nil { | 189 | if !ephemeral && d.persist != nil { | ||
| 189 | - | if err := d.persist.TouchMRU(name, seq); err != nil { | 190 | + | if err := d.persist.TouchMRU(name, incarnation, seq); err != nil { |
| 190 | d.log.Warn("touching persisted session recency failed", "err", err, "session", name) | 191 | d.log.Warn("touching persisted session recency failed", "err", err, "session", name) | ||
| 191 | } | 192 | } | ||
| 192 | } | 193 | } | ||
| @@ -203,6 +204,10 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz | |||||
| 203 | // the persisted identity rather than receive a fresh timestamp. | 204 | // the persisted identity rather than receive a fresh timestamp. | ||
| 204 | createdAt = stopped.createdAt | 205 | createdAt = stopped.createdAt | ||
| 205 | incarnation = stopped.incarnation | 206 | incarnation = stopped.incarnation | ||
| 207 | + | if authoritative, ok := d.persist.Record(name); ok { | |||
| 208 | + | incarnation = authoritative.IncarnationID | |||
| 209 | + | createdAt = authoritative.CreatedAt | |||
| 210 | + | } | |||
| 206 | if createdAt > d.lastAllocatedCreatedAt { | 211 | if createdAt > d.lastAllocatedCreatedAt { | ||
| 207 | d.lastAllocatedCreatedAt = createdAt | 212 | d.lastAllocatedCreatedAt = createdAt | ||
| 208 | } | 213 | } | ||
| @@ -281,10 +286,23 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz | |||||
| 281 | } | 286 | } | ||
| 282 | sess.snapEligible.Store(!ephemeral && name != "") | 287 | sess.snapEligible.Store(!ephemeral && name != "") | ||
| 283 | if !ephemeral { | 288 | if !ephemeral { | ||
| 284 | - | if err := d.persist.Save(persist.Record{Name: name, IncarnationID: incarnation, Cwd: cwd, CreatedAt: createdAt, UpdatedAt: createdAt, LastUsedSeq: lastUsedSeq, TabNames: names, RecoveryState: domain.RecoveryFresh}); err != nil { | 289 | + | if d.persistEnabled { |
| 285 | - | closeTabs(tabs) | 290 | + | record := persist.Record{Name: name, IncarnationID: incarnation, Cwd: cwd, CreatedAt: createdAt, UpdatedAt: createdAt, LastUsedSeq: lastUsedSeq, TabNames: names, RecoveryState: domain.RecoveryFresh} |
| 286 | - | cancel() | 291 | + | var err error |
| 287 | - | return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err) | 292 | + | if resuming { |
| 293 | + | if _, ok := d.persist.Record(name); ok { | |||
| 294 | + | err = d.persist.UpdateMetadata(record) | |||
| 295 | + | } else { | |||
| 296 | + | err = d.persist.Create(record) | |||
| 297 | + | } | |||
| 298 | + | } else { | |||
| 299 | + | err = d.persist.Create(record) | |||
| 300 | + | } | |||
| 301 | + | if err != nil { | |||
| 302 | + | closeTabs(tabs) | |||
| 303 | + | cancel() | |||
| 304 | + | return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err) | |||
| 305 | + | } | |||
| 288 | } | 306 | } | ||
| 289 | delete(d.stopped, name) | 307 | delete(d.stopped, name) | ||
| 290 | } | 308 | } | ||
| @@ -426,7 +444,7 @@ func (d *Daemon) createTab(sess *session, sz domain.Size) error { | |||||
| 426 | record := sess.persistRecordLocked(time.Now().UnixNano()) | 444 | record := sess.persistRecordLocked(time.Now().UnixNano()) | ||
| 427 | ephemeral := sess.ephemeral | 445 | ephemeral := sess.ephemeral | ||
| 428 | if !ephemeral { | 446 | if !ephemeral { | ||
| 429 | - | if err := d.persist.Save(record); err != nil { | 447 | + | if err := d.persist.UpdateMetadata(record); err != nil { |
| 430 | sess.tabs = sess.tabs[:len(sess.tabs)-1] | 448 | sess.tabs = sess.tabs[:len(sess.tabs)-1] | ||
| 431 | sess.active = oldActive | 449 | sess.active = oldActive | ||
| 432 | if tb.cancel != nil { | 450 | if tb.cancel != nil { | ||
| @@ -695,8 +713,8 @@ func (d *Daemon) renameSession(sess *session, name string) error { | |||||
| 695 | // For a named rename this is the durable commit point, deliberately after | 713 | // For a named rename this is the durable commit point, deliberately after | ||
| 696 | // the old tombstone transaction. Ephemeral promotion has no old durable | 714 | // the old tombstone transaction. Ephemeral promotion has no old durable | ||
| 697 | // identity and can create its first record directly. | 715 | // identity and can create its first record directly. | ||
| 698 | - | if wasEphemeral || oldName != name { | 716 | + | if (wasEphemeral || oldName != name) && d.persistEnabled { |
| 699 | - | if err := d.persist.Save(record); err != nil { | 717 | + | if err := d.persist.Create(record); err != nil { |
| 700 | return rollback(err) | 718 | return rollback(err) | ||
| 701 | } | 719 | } | ||
| 702 | } | 720 | } | ||
| @@ -756,7 +774,7 @@ func (d *Daemon) renameTab(sess *session, tb *tab, name string) error { | |||||
| 756 | record := sess.persistRecordLocked(time.Now().UnixNano()) | 774 | record := sess.persistRecordLocked(time.Now().UnixNano()) | ||
| 757 | ephemeral := sess.ephemeral | 775 | ephemeral := sess.ephemeral | ||
| 758 | if !ephemeral { | 776 | if !ephemeral { | ||
| 759 | - | if err := d.persist.Save(record); err != nil { | 777 | + | if err := d.persist.UpdateMetadata(record); err != nil { |
| 760 | tb.name = oldName | 778 | tb.name = oldName | ||
| 761 | sess.mu.Unlock() | 779 | sess.mu.Unlock() | ||
| 762 | d.mu.Unlock() | 780 | d.mu.Unlock() | ||
| @@ -835,7 +853,7 @@ func (d *Daemon) closeTab(sess *session, tb *tab, repaint bool) error { | |||||
| 835 | record := sess.persistRecordLocked(time.Now().UnixNano()) | 853 | record := sess.persistRecordLocked(time.Now().UnixNano()) | ||
| 836 | ephemeral := sess.ephemeral | 854 | ephemeral := sess.ephemeral | ||
| 837 | if !ephemeral { | 855 | if !ephemeral { | ||
| 838 | - | if err := d.persist.Save(record); err != nil { | 856 | + | if err := d.persist.UpdateMetadata(record); err != nil { |
| 839 | d.log.Warn("persisting closed tab failed", "err", err, "session", name) | 857 | d.log.Warn("persisting closed tab failed", "err", err, "session", name) | ||
| 840 | } | 858 | } | ||
| 841 | } | 859 | } | ||
| @@ -1336,10 +1354,11 @@ func (d *Daemon) refreshSessionCwd(sess *session) { | |||||
| 1336 | } | 1354 | } | ||
| 1337 | sess.cwd = cwd | 1355 | sess.cwd = cwd | ||
| 1338 | name := sess.name | 1356 | name := sess.name | ||
| 1357 | + | incarnation := sess.incarnation | |||
| 1339 | ephemeral := sess.ephemeral | 1358 | ephemeral := sess.ephemeral | ||
| 1340 | sess.mu.Unlock() | 1359 | sess.mu.Unlock() | ||
| 1341 | if !ephemeral { | 1360 | if !ephemeral { | ||
| 1342 | - | if err := d.persist.Touch(name, cwd, time.Now().UnixNano()); err != nil { | 1361 | + | if err := d.persist.Touch(name, incarnation, cwd, time.Now().UnixNano()); err != nil { |
| 1343 | d.log.Warn("touching persisted session cwd failed", "err", err, "session", name) | 1362 | d.log.Warn("touching persisted session cwd failed", "err", err, "session", name) | ||
| 1344 | } | 1363 | } | ||
| 1345 | markSnapshotDirty(sess) | 1364 | markSnapshotDirty(sess) | ||
M internal/usecase/daemon/session_test.go +1 -0
| @@ -1185,6 +1185,7 @@ func TestEphemeralPromotionLifecycleFailuresLeaveStateRollbackSafe(t *testing.T) | |||
| 1185 | 1185 | store := portsmocks.NewMockStore(t) | |
| 1186 | 1186 | var attempted map[string][]byte | |
| 1187 | 1187 | store.EXPECT().Range(mock.Anything).Run(func(func([]byte, []byte) bool) {}).Twice() | |
| 1188 | + | store.EXPECT().Get([]byte("named")).Return(nil, false).Twice() | |
| 1188 | 1189 | store.EXPECT().Batch(mock.Anything).RunAndReturn(func(changes []ports.StoreChange) error { | |
| 1189 | 1190 | attempted = make(map[string][]byte, len(changes)) | |
| 1190 | 1191 | for _, change := range changes { | |
| @@ -1185,6 +1185,7 @@ func TestEphemeralPromotionLifecycleFailuresLeaveStateRollbackSafe(t *testing.T) | |||||
| 1185 | store := portsmocks.NewMockStore(t) | 1185 | store := portsmocks.NewMockStore(t) | ||
| 1186 | var attempted map[string][]byte | 1186 | var attempted map[string][]byte | ||
| 1187 | store.EXPECT().Range(mock.Anything).Run(func(func([]byte, []byte) bool) {}).Twice() | 1187 | store.EXPECT().Range(mock.Anything).Run(func(func([]byte, []byte) bool) {}).Twice() | ||
| 1188 | + | store.EXPECT().Get([]byte("named")).Return(nil, false).Twice() | |||
| 1188 | store.EXPECT().Batch(mock.Anything).RunAndReturn(func(changes []ports.StoreChange) error { | 1189 | store.EXPECT().Batch(mock.Anything).RunAndReturn(func(changes []ports.StoreChange) error { | ||
| 1189 | attempted = make(map[string][]byte, len(changes)) | 1190 | attempted = make(map[string][]byte, len(changes)) | ||
| 1190 | for _, change := range changes { | 1191 | for _, change := range changes { | ||
M internal/usecase/daemon/snapshot_restore.go +4 -3
| @@ -65,7 +65,7 @@ func validateRestoreSessionSnapshot(snap snapcodec.Session) error { | |||
| 65 | 65 | ||
| 66 | 66 | // restoreSession owns the restored tabs until registration succeeds. Every | |
| 67 | 67 | // unsuccessful path closes their PTYs and cancels their shared session context. | |
| 68 | - | func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session, repositoryGeneration uint64) error { | |
| 68 | + | func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session, repositoryGeneration uint64, checkpoint domain.CheckpointRef) error { | |
| 69 | 69 | if err := ctx.Err(); err != nil { | |
| 70 | 70 | return err | |
| 71 | 71 | } | |
| @@ -102,6 +102,7 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session, rep | |||
| 102 | 102 | // The loaded manifest is the repository head for this name. Future dirty | |
| 103 | 103 | // checkpoints must continue from it rather than reuse generation one. | |
| 104 | 104 | sess.snapshotPublishedGeneration = repositoryGeneration | |
| 105 | + | sess.snapshotPublishedCheckpoint = &checkpoint | |
| 105 | 106 | registered, err := d.persistAndRegisterRestoredSession(ctx, sess) | |
| 106 | 107 | if err != nil || !registered { | |
| 107 | 108 | return err | |
| @@ -269,8 +270,8 @@ func (d *Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess *se | |||
| 269 | 270 | } | |
| 270 | 271 | // The catalogue checkpoint already authorizes this runtime. Rewriting it | |
| 271 | 272 | // during restoration would discard committed/fallback recovery metadata. | |
| 272 | - | if d.catalogue == nil { | |
| 273 | - | if err := d.persist.Save(sess.persistRecordLocked(sess.createdAt)); err != nil { | |
| 273 | + | if d.catalogue == nil && d.persistEnabled { | |
| 274 | + | if err := d.persist.Create(sess.persistRecordLocked(sess.createdAt)); err != nil { | |
| 274 | 275 | return false, err | |
| 275 | 276 | } | |
| 276 | 277 | } | |
| @@ -65,7 +65,7 @@ func validateRestoreSessionSnapshot(snap snapcodec.Session) error { | |||||
| 65 | 65 | ||||
| 66 | // restoreSession owns the restored tabs until registration succeeds. Every | 66 | // restoreSession owns the restored tabs until registration succeeds. Every | ||
| 67 | // unsuccessful path closes their PTYs and cancels their shared session context. | 67 | // unsuccessful path closes their PTYs and cancels their shared session context. | ||
| 68 | - | func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session, repositoryGeneration uint64) error { | 68 | + | func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session, repositoryGeneration uint64, checkpoint domain.CheckpointRef) error { |
| 69 | if err := ctx.Err(); err != nil { | 69 | if err := ctx.Err(); err != nil { | ||
| 70 | return err | 70 | return err | ||
| 71 | } | 71 | } | ||
| @@ -102,6 +102,7 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session, rep | |||||
| 102 | // The loaded manifest is the repository head for this name. Future dirty | 102 | // The loaded manifest is the repository head for this name. Future dirty | ||
| 103 | // checkpoints must continue from it rather than reuse generation one. | 103 | // checkpoints must continue from it rather than reuse generation one. | ||
| 104 | sess.snapshotPublishedGeneration = repositoryGeneration | 104 | sess.snapshotPublishedGeneration = repositoryGeneration | ||
| 105 | + | sess.snapshotPublishedCheckpoint = &checkpoint | |||
| 105 | registered, err := d.persistAndRegisterRestoredSession(ctx, sess) | 106 | registered, err := d.persistAndRegisterRestoredSession(ctx, sess) | ||
| 106 | if err != nil || !registered { | 107 | if err != nil || !registered { | ||
| 107 | return err | 108 | return err | ||
| @@ -269,8 +270,8 @@ func (d *Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess *se | |||||
| 269 | } | 270 | } | ||
| 270 | // The catalogue checkpoint already authorizes this runtime. Rewriting it | 271 | // The catalogue checkpoint already authorizes this runtime. Rewriting it | ||
| 271 | // during restoration would discard committed/fallback recovery metadata. | 272 | // during restoration would discard committed/fallback recovery metadata. | ||
| 272 | - | if d.catalogue == nil { | 273 | + | if d.catalogue == nil && d.persistEnabled { |
| 273 | - | if err := d.persist.Save(sess.persistRecordLocked(sess.createdAt)); err != nil { | 274 | + | if err := d.persist.Create(sess.persistRecordLocked(sess.createdAt)); err != nil { |
| 274 | return false, err | 275 | return false, err | ||
| 275 | } | 276 | } | ||
| 276 | } | 277 | } | ||
M internal/usecase/daemon/snapshot_restore_incremental.go +1 -1
| @@ -230,7 +230,7 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor | |||
| 230 | 230 | d.log.Warn("checkpoint HEAD repair pending", "session", record.Name, "err", err) | |
| 231 | 231 | } | |
| 232 | 232 | ||
| 233 | - | if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation); err != nil { | |
| 233 | + | if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation, selected); err != nil { | |
| 234 | 234 | return err | |
| 235 | 235 | } | |
| 236 | 236 | d.setStoppedRecovery(record, runtimeHealthy) | |
| @@ -230,7 +230,7 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor | |||||
| 230 | d.log.Warn("checkpoint HEAD repair pending", "session", record.Name, "err", err) | 230 | d.log.Warn("checkpoint HEAD repair pending", "session", record.Name, "err", err) | ||
| 231 | } | 231 | } | ||
| 232 | 232 | ||||
| 233 | - | if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation); err != nil { | 233 | + | if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation, selected); err != nil { |
| 234 | return err | 234 | return err | ||
| 235 | } | 235 | } | ||
| 236 | d.setStoppedRecovery(record, runtimeHealthy) | 236 | d.setStoppedRecovery(record, runtimeHealthy) | ||
M internal/usecase/daemon/snapshot_restore_test.go +60 -0
| @@ -10,8 +10,10 @@ import ( | |||
| 10 | 10 | "github.com/stretchr/testify/require" | |
| 11 | 11 | ||
| 12 | 12 | "github.com/bnema/vev/internal/domain" | |
| 13 | + | "github.com/bnema/vev/internal/persist" | |
| 13 | 14 | "github.com/bnema/vev/internal/ports" | |
| 14 | 15 | "github.com/bnema/vev/internal/usecase/layout" | |
| 16 | + | recoveryusecase "github.com/bnema/vev/internal/usecase/recovery" | |
| 15 | 17 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | |
| 16 | 18 | "github.com/bnema/vev/pkg/renderer" | |
| 17 | 19 | "github.com/bnema/vev/pkg/vt" | |
| @@ -257,6 +259,64 @@ func TestRestoreIncrementalGenerationAcceptance(t *testing.T) { | |||
| 257 | 259 | require.Equal(t, uint64(10), repository.publishes[0].Generation, "a restored session must continue the concrete repository generation stream") | |
| 258 | 260 | } | |
| 259 | 261 | ||
| 262 | + | func TestRestoredSessionMetadataUpdatePreservesCheckpointLineage(t *testing.T) { | |
| 263 | + | snapshot := restoreAcceptanceSession(t, "restored") | |
| 264 | + | generation := acceptanceGeneration(t, snapshot, 9) | |
| 265 | + | repository := &snapshotAcceptanceRepository{names: []string{snapshot.Name}, generations: map[string]ports.SnapshotGeneration{snapshot.Name: generation}} | |
| 266 | + | store, _ := newMockStore(t) | |
| 267 | + | catalogue := persist.New(store) | |
| 268 | + | committed := domain.CheckpointRef{Generation: generation.Generation, ManifestDigest: snapcodec.ManifestDigest(generation.Manifest)} | |
| 269 | + | fallback1 := domain.CheckpointRef{Generation: 8, ManifestDigest: [32]byte{8}} | |
| 270 | + | fallback2 := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}} | |
| 271 | + | record := domain.CatalogueRecord{ | |
| 272 | + | Name: snapshot.Name, IncarnationID: generation.IncarnationID, Cwd: "/snapshot/cwd", | |
| 273 | + | CreatedAt: int64(snapshot.CreatedAt), UpdatedAt: 81, LastUsedSeq: 17, | |
| 274 | + | TabNames: []string{"before"}, RecoveryState: domain.RecoveryHealthy, | |
| 275 | + | Committed: &committed, Fallbacks: [2]*domain.CheckpointRef{&fallback1, &fallback2}, | |
| 276 | + | } | |
| 277 | + | require.NoError(t, catalogue.Create(record)) | |
| 278 | + | ||
| 279 | + | pty, release := newBlockingPTY(t) | |
| 280 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 281 | + | coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil, nil) | |
| 282 | + | WithCatalogue(catalogue, []domain.CatalogueRecord{record})(d) | |
| 283 | + | WithSnapshotRepository(repository, nil)(d) | |
| 284 | + | WithCheckpointCoordinator(coordinator)(d) | |
| 285 | + | t.Cleanup(func() { release(); d.sessWg.Wait() }) | |
| 286 | + | ||
| 287 | + | d.restoreIncrementalSnapshots(context.Background()) | |
| 288 | + | d.mu.Lock() | |
| 289 | + | restored := d.findByNameLocked(snapshot.Name) | |
| 290 | + | d.mu.Unlock() | |
| 291 | + | require.NotNil(t, restored) | |
| 292 | + | require.NoError(t, d.renameTab(restored, restored.tabs[0], "after")) | |
| 293 | + | ||
| 294 | + | updated, ok := catalogue.Record(record.Name) | |
| 295 | + | require.True(t, ok) | |
| 296 | + | require.Equal(t, record.IncarnationID, updated.IncarnationID) | |
| 297 | + | require.Equal(t, record.RecoveryState, updated.RecoveryState) | |
| 298 | + | require.Equal(t, record.Committed, updated.Committed) | |
| 299 | + | require.Equal(t, record.Fallbacks, updated.Fallbacks) | |
| 300 | + | require.Equal(t, record.DegradedReason, updated.DegradedReason) | |
| 301 | + | require.Equal(t, []string{"after"}, updated.TabNames) | |
| 302 | + | ||
| 303 | + | startSnapshotEncodeWorker(t, d) | |
| 304 | + | markSnapshotDirty(restored) | |
| 305 | + | require.True(t, d.scheduleSnapshot(restored)) | |
| 306 | + | awaitSnapshotIdle(t, restored) | |
| 307 | + | d.snapshotNoticeMu.Lock() | |
| 308 | + | failure := d.snapshotActiveFailureSignature | |
| 309 | + | d.snapshotNoticeMu.Unlock() | |
| 310 | + | require.Empty(t, failure) | |
| 311 | + | require.Len(t, repository.publishes, 1) | |
| 312 | + | require.Equal(t, record.Committed, repository.publishes[0].ParentCheckpoint) | |
| 313 | + | published, ok := catalogue.Record(record.Name) | |
| 314 | + | require.True(t, ok) | |
| 315 | + | require.Equal(t, uint64(10), published.Committed.Generation) | |
| 316 | + | require.Equal(t, committed, *published.Fallbacks[0]) | |
| 317 | + | require.Equal(t, fallback1, *published.Fallbacks[1]) | |
| 318 | + | } | |
| 319 | + | ||
| 260 | 320 | func TestRestoreIncrementalFallbackAndInvalidObjectMappings(t *testing.T) { | |
| 261 | 321 | snapshot := restoreAcceptanceSession(t, "fallback") | |
| 262 | 322 | valid := acceptanceGeneration(t, snapshot, 3) | |
| @@ -10,8 +10,10 @@ import ( | |||||
| 10 | "github.com/stretchr/testify/require" | 10 | "github.com/stretchr/testify/require" | ||
| 11 | 11 | ||||
| 12 | "github.com/bnema/vev/internal/domain" | 12 | "github.com/bnema/vev/internal/domain" | ||
| 13 | + | "github.com/bnema/vev/internal/persist" | |||
| 13 | "github.com/bnema/vev/internal/ports" | 14 | "github.com/bnema/vev/internal/ports" | ||
| 14 | "github.com/bnema/vev/internal/usecase/layout" | 15 | "github.com/bnema/vev/internal/usecase/layout" | ||
| 16 | + | recoveryusecase "github.com/bnema/vev/internal/usecase/recovery" | |||
| 15 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | 17 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | ||
| 16 | "github.com/bnema/vev/pkg/renderer" | 18 | "github.com/bnema/vev/pkg/renderer" | ||
| 17 | "github.com/bnema/vev/pkg/vt" | 19 | "github.com/bnema/vev/pkg/vt" | ||
| @@ -257,6 +259,64 @@ func TestRestoreIncrementalGenerationAcceptance(t *testing.T) { | |||||
| 257 | require.Equal(t, uint64(10), repository.publishes[0].Generation, "a restored session must continue the concrete repository generation stream") | 259 | require.Equal(t, uint64(10), repository.publishes[0].Generation, "a restored session must continue the concrete repository generation stream") | ||
| 258 | } | 260 | } | ||
| 259 | 261 | ||||
| 262 | + | func TestRestoredSessionMetadataUpdatePreservesCheckpointLineage(t *testing.T) { | |||
| 263 | + | snapshot := restoreAcceptanceSession(t, "restored") | |||
| 264 | + | generation := acceptanceGeneration(t, snapshot, 9) | |||
| 265 | + | repository := &snapshotAcceptanceRepository{names: []string{snapshot.Name}, generations: map[string]ports.SnapshotGeneration{snapshot.Name: generation}} | |||
| 266 | + | store, _ := newMockStore(t) | |||
| 267 | + | catalogue := persist.New(store) | |||
| 268 | + | committed := domain.CheckpointRef{Generation: generation.Generation, ManifestDigest: snapcodec.ManifestDigest(generation.Manifest)} | |||
| 269 | + | fallback1 := domain.CheckpointRef{Generation: 8, ManifestDigest: [32]byte{8}} | |||
| 270 | + | fallback2 := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}} | |||
| 271 | + | record := domain.CatalogueRecord{ | |||
| 272 | + | Name: snapshot.Name, IncarnationID: generation.IncarnationID, Cwd: "/snapshot/cwd", | |||
| 273 | + | CreatedAt: int64(snapshot.CreatedAt), UpdatedAt: 81, LastUsedSeq: 17, | |||
| 274 | + | TabNames: []string{"before"}, RecoveryState: domain.RecoveryHealthy, | |||
| 275 | + | Committed: &committed, Fallbacks: [2]*domain.CheckpointRef{&fallback1, &fallback2}, | |||
| 276 | + | } | |||
| 277 | + | require.NoError(t, catalogue.Create(record)) | |||
| 278 | + | ||||
| 279 | + | pty, release := newBlockingPTY(t) | |||
| 280 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 281 | + | coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil, nil) | |||
| 282 | + | WithCatalogue(catalogue, []domain.CatalogueRecord{record})(d) | |||
| 283 | + | WithSnapshotRepository(repository, nil)(d) | |||
| 284 | + | WithCheckpointCoordinator(coordinator)(d) | |||
| 285 | + | t.Cleanup(func() { release(); d.sessWg.Wait() }) | |||
| 286 | + | ||||
| 287 | + | d.restoreIncrementalSnapshots(context.Background()) | |||
| 288 | + | d.mu.Lock() | |||
| 289 | + | restored := d.findByNameLocked(snapshot.Name) | |||
| 290 | + | d.mu.Unlock() | |||
| 291 | + | require.NotNil(t, restored) | |||
| 292 | + | require.NoError(t, d.renameTab(restored, restored.tabs[0], "after")) | |||
| 293 | + | ||||
| 294 | + | updated, ok := catalogue.Record(record.Name) | |||
| 295 | + | require.True(t, ok) | |||
| 296 | + | require.Equal(t, record.IncarnationID, updated.IncarnationID) | |||
| 297 | + | require.Equal(t, record.RecoveryState, updated.RecoveryState) | |||
| 298 | + | require.Equal(t, record.Committed, updated.Committed) | |||
| 299 | + | require.Equal(t, record.Fallbacks, updated.Fallbacks) | |||
| 300 | + | require.Equal(t, record.DegradedReason, updated.DegradedReason) | |||
| 301 | + | require.Equal(t, []string{"after"}, updated.TabNames) | |||
| 302 | + | ||||
| 303 | + | startSnapshotEncodeWorker(t, d) | |||
| 304 | + | markSnapshotDirty(restored) | |||
| 305 | + | require.True(t, d.scheduleSnapshot(restored)) | |||
| 306 | + | awaitSnapshotIdle(t, restored) | |||
| 307 | + | d.snapshotNoticeMu.Lock() | |||
| 308 | + | failure := d.snapshotActiveFailureSignature | |||
| 309 | + | d.snapshotNoticeMu.Unlock() | |||
| 310 | + | require.Empty(t, failure) | |||
| 311 | + | require.Len(t, repository.publishes, 1) | |||
| 312 | + | require.Equal(t, record.Committed, repository.publishes[0].ParentCheckpoint) | |||
| 313 | + | published, ok := catalogue.Record(record.Name) | |||
| 314 | + | require.True(t, ok) | |||
| 315 | + | require.Equal(t, uint64(10), published.Committed.Generation) | |||
| 316 | + | require.Equal(t, committed, *published.Fallbacks[0]) | |||
| 317 | + | require.Equal(t, fallback1, *published.Fallbacks[1]) | |||
| 318 | + | } | |||
| 319 | + | ||||
| 260 | func TestRestoreIncrementalFallbackAndInvalidObjectMappings(t *testing.T) { | 320 | func TestRestoreIncrementalFallbackAndInvalidObjectMappings(t *testing.T) { | ||
| 261 | snapshot := restoreAcceptanceSession(t, "fallback") | 321 | snapshot := restoreAcceptanceSession(t, "fallback") | ||
| 262 | valid := acceptanceGeneration(t, snapshot, 3) | 322 | valid := acceptanceGeneration(t, snapshot, 3) | ||