fix(navigation): fence stopped resume lifecycle
- commit
- 046cebbd3fad3e47443348593f59cba33cc00239
- author
- brice · · 2026-08-19 04:27 UTC
- parent
- 736629f
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +90 -3
M internal/usecase/daemon/coderabbit_regression_test.go +63 -0
| @@ -13,6 +13,7 @@ import ( | |||
| 13 | 13 | ||
| 14 | 14 | "github.com/bnema/vev/internal/domain" | |
| 15 | 15 | "github.com/bnema/vev/internal/ports" | |
| 16 | + | "github.com/bnema/vev/internal/usecase/picker" | |
| 16 | 17 | recoveryusecase "github.com/bnema/vev/internal/usecase/recovery" | |
| 17 | 18 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | |
| 18 | 19 | ) | |
| @@ -243,6 +244,68 @@ func TestCreateSessionRechecksShutdownAfterCatalogueRead(t *testing.T) { | |||
| 243 | 244 | require.Empty(t, d.sessions) | |
| 244 | 245 | } | |
| 245 | 246 | ||
| 247 | + | type blockingResumeCatalogue struct { | |
| 248 | + | *durableRecoveryCatalogue | |
| 249 | + | entered chan struct{} | |
| 250 | + | release chan struct{} | |
| 251 | + | once sync.Once | |
| 252 | + | } | |
| 253 | + | ||
| 254 | + | func (c *blockingResumeCatalogue) Record(name string) (domain.CatalogueRecord, bool, error) { | |
| 255 | + | c.once.Do(func() { close(c.entered) }) | |
| 256 | + | <-c.release | |
| 257 | + | return c.durableRecoveryCatalogue.Record(name) | |
| 258 | + | } | |
| 259 | + | ||
| 260 | + | func TestResumeStoppedSessionRejectsReplacementDuringCatalogueRead(t *testing.T) { | |
| 261 | + | sourcePTY, releaseSource := newBlockingPTY(t) | |
| 262 | + | defer releaseSource() | |
| 263 | + | targetPTY, releaseTarget := newBlockingPTY(t) | |
| 264 | + | defer releaseTarget() | |
| 265 | + | d, from, ac, _ := newManualSessionWithPTYs(t, sourcePTY) | |
| 266 | + | d.ptys = newFactorySeq(t, targetPTY) | |
| 267 | + | ||
| 268 | + | record := durableRecoveryRecord(1) | |
| 269 | + | record.Name = "stopped" | |
| 270 | + | catalogue := &blockingResumeCatalogue{ | |
| 271 | + | durableRecoveryCatalogue: newDurableRecoveryCatalogue([]domain.CatalogueRecord{record}), | |
| 272 | + | entered: make(chan struct{}), | |
| 273 | + | release: make(chan struct{}), | |
| 274 | + | } | |
| 275 | + | d.catalogue = catalogue | |
| 276 | + | d.persistEnabled = true | |
| 277 | + | d.recovery = recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil) | |
| 278 | + | d.stopped[record.Name] = stoppedSessionFromRecord(record, ports.SessionDown, nil) | |
| 279 | + | ||
| 280 | + | target := picker.Target{ | |
| 281 | + | Name: record.Name, | |
| 282 | + | Stopped: true, | |
| 283 | + | Incarnation: record.IncarnationID, | |
| 284 | + | ExpectedCreatedAt: &record.CreatedAt, | |
| 285 | + | } | |
| 286 | + | result := make(chan bool, 1) | |
| 287 | + | go func() { result <- d.resumeStoppedAndSwitch(from, ac, target) }() | |
| 288 | + | ||
| 289 | + | <-catalogue.entered | |
| 290 | + | replacement := record | |
| 291 | + | replacement.IncarnationID = domain.IncarnationID{9} | |
| 292 | + | replacement.CreatedAt++ | |
| 293 | + | replacement.UpdatedAt = replacement.CreatedAt | |
| 294 | + | require.NoError(t, catalogue.Replace(replacement.Name, replacement)) | |
| 295 | + | d.mu.Lock() | |
| 296 | + | d.stopped[replacement.Name] = stoppedSessionFromRecord(replacement, ports.SessionDown, nil) | |
| 297 | + | d.mu.Unlock() | |
| 298 | + | close(catalogue.release) | |
| 299 | + | ||
| 300 | + | require.False(t, <-result) | |
| 301 | + | require.Same(t, from, ac.currentSession()) | |
| 302 | + | require.Empty(t, catalogue.MetadataUpdates()) | |
| 303 | + | d.mu.Lock() | |
| 304 | + | defer d.mu.Unlock() | |
| 305 | + | require.Nil(t, d.findByNameLocked(record.Name)) | |
| 306 | + | require.Equal(t, replacement.IncarnationID, d.stopped[record.Name].incarnation) | |
| 307 | + | } | |
| 308 | + | ||
| 246 | 309 | func TestSnapshotStopContextCancelIsIdempotent(t *testing.T) { | |
| 247 | 310 | deadline := &snapshotShutdownDeadline{done: make(chan struct{})} | |
| 248 | 311 | _, cancel := snapshotStopContext(deadline) | |
| @@ -13,6 +13,7 @@ import ( | |||||
| 13 | 13 | ||||
| 14 | "github.com/bnema/vev/internal/domain" | 14 | "github.com/bnema/vev/internal/domain" | ||
| 15 | "github.com/bnema/vev/internal/ports" | 15 | "github.com/bnema/vev/internal/ports" | ||
| 16 | + | "github.com/bnema/vev/internal/usecase/picker" | |||
| 16 | recoveryusecase "github.com/bnema/vev/internal/usecase/recovery" | 17 | recoveryusecase "github.com/bnema/vev/internal/usecase/recovery" | ||
| 17 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | 18 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | ||
| 18 | ) | 19 | ) | ||
| @@ -243,6 +244,68 @@ func TestCreateSessionRechecksShutdownAfterCatalogueRead(t *testing.T) { | |||||
| 243 | require.Empty(t, d.sessions) | 244 | require.Empty(t, d.sessions) | ||
| 244 | } | 245 | } | ||
| 245 | 246 | ||||
| 247 | + | type blockingResumeCatalogue struct { | |||
| 248 | + | *durableRecoveryCatalogue | |||
| 249 | + | entered chan struct{} | |||
| 250 | + | release chan struct{} | |||
| 251 | + | once sync.Once | |||
| 252 | + | } | |||
| 253 | + | ||||
| 254 | + | func (c *blockingResumeCatalogue) Record(name string) (domain.CatalogueRecord, bool, error) { | |||
| 255 | + | c.once.Do(func() { close(c.entered) }) | |||
| 256 | + | <-c.release | |||
| 257 | + | return c.durableRecoveryCatalogue.Record(name) | |||
| 258 | + | } | |||
| 259 | + | ||||
| 260 | + | func TestResumeStoppedSessionRejectsReplacementDuringCatalogueRead(t *testing.T) { | |||
| 261 | + | sourcePTY, releaseSource := newBlockingPTY(t) | |||
| 262 | + | defer releaseSource() | |||
| 263 | + | targetPTY, releaseTarget := newBlockingPTY(t) | |||
| 264 | + | defer releaseTarget() | |||
| 265 | + | d, from, ac, _ := newManualSessionWithPTYs(t, sourcePTY) | |||
| 266 | + | d.ptys = newFactorySeq(t, targetPTY) | |||
| 267 | + | ||||
| 268 | + | record := durableRecoveryRecord(1) | |||
| 269 | + | record.Name = "stopped" | |||
| 270 | + | catalogue := &blockingResumeCatalogue{ | |||
| 271 | + | durableRecoveryCatalogue: newDurableRecoveryCatalogue([]domain.CatalogueRecord{record}), | |||
| 272 | + | entered: make(chan struct{}), | |||
| 273 | + | release: make(chan struct{}), | |||
| 274 | + | } | |||
| 275 | + | d.catalogue = catalogue | |||
| 276 | + | d.persistEnabled = true | |||
| 277 | + | d.recovery = recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil) | |||
| 278 | + | d.stopped[record.Name] = stoppedSessionFromRecord(record, ports.SessionDown, nil) | |||
| 279 | + | ||||
| 280 | + | target := picker.Target{ | |||
| 281 | + | Name: record.Name, | |||
| 282 | + | Stopped: true, | |||
| 283 | + | Incarnation: record.IncarnationID, | |||
| 284 | + | ExpectedCreatedAt: &record.CreatedAt, | |||
| 285 | + | } | |||
| 286 | + | result := make(chan bool, 1) | |||
| 287 | + | go func() { result <- d.resumeStoppedAndSwitch(from, ac, target) }() | |||
| 288 | + | ||||
| 289 | + | <-catalogue.entered | |||
| 290 | + | replacement := record | |||
| 291 | + | replacement.IncarnationID = domain.IncarnationID{9} | |||
| 292 | + | replacement.CreatedAt++ | |||
| 293 | + | replacement.UpdatedAt = replacement.CreatedAt | |||
| 294 | + | require.NoError(t, catalogue.Replace(replacement.Name, replacement)) | |||
| 295 | + | d.mu.Lock() | |||
| 296 | + | d.stopped[replacement.Name] = stoppedSessionFromRecord(replacement, ports.SessionDown, nil) | |||
| 297 | + | d.mu.Unlock() | |||
| 298 | + | close(catalogue.release) | |||
| 299 | + | ||||
| 300 | + | require.False(t, <-result) | |||
| 301 | + | require.Same(t, from, ac.currentSession()) | |||
| 302 | + | require.Empty(t, catalogue.MetadataUpdates()) | |||
| 303 | + | d.mu.Lock() | |||
| 304 | + | defer d.mu.Unlock() | |||
| 305 | + | require.Nil(t, d.findByNameLocked(record.Name)) | |||
| 306 | + | require.Equal(t, replacement.IncarnationID, d.stopped[record.Name].incarnation) | |||
| 307 | + | } | |||
| 308 | + | ||||
| 246 | func TestSnapshotStopContextCancelIsIdempotent(t *testing.T) { | 309 | func TestSnapshotStopContextCancelIsIdempotent(t *testing.T) { | ||
| 247 | deadline := &snapshotShutdownDeadline{done: make(chan struct{})} | 310 | deadline := &snapshotShutdownDeadline{done: make(chan struct{})} | ||
| 248 | _, cancel := snapshotStopContext(deadline) | 311 | _, cancel := snapshotStopContext(deadline) | ||
M internal/usecase/daemon/picker.go +7 -3
| @@ -1290,7 +1290,7 @@ func (d *Daemon) resumeStoppedAndSwitchLocked(from *session, ac *attachedClient, | |||
| 1290 | 1290 | from.mu.Lock() | |
| 1291 | 1291 | cwd, env := d.dirOrHome(current.cwd), copyEnvironment(from.env) | |
| 1292 | 1292 | from.mu.Unlock() | |
| 1293 | - | created, createErr := d.createSessionLocked(target.Name, false, cwd, ac.sizeSnapshot(), env, current.tabNames) | |
| 1293 | + | created, createErr := d.createStoppedSessionLocked(target.Name, cwd, ac.sizeSnapshot(), env, current, current.tabNames) | |
| 1294 | 1294 | targetSess = created | |
| 1295 | 1295 | return created, createErr | |
| 1296 | 1296 | }, | |
| @@ -1316,8 +1316,12 @@ func (d *Daemon) resumeStoppedAndSwitchLocked(from *session, ac *attachedClient, | |||
| 1316 | 1316 | } | |
| 1317 | 1317 | env := copyEnvironment(from.env) | |
| 1318 | 1318 | from.mu.Unlock() | |
| 1319 | - | cwd := d.dirOrHome(stopped.cwd) | |
| 1320 | - | targetSess, err := d.createSessionLocked(target.Name, false, cwd, ac.sizeSnapshot(), env, stopped.tabNames) | |
| 1319 | + | current, ok := d.stopped[target.Name] | |
| 1320 | + | if !ok || current.purging || !stoppedSessionMatchesLifecycle(current, stopped) || !targetMatchesLifecycle(target, current.name, current.createdAt, current.incarnation) { | |
| 1321 | + | return nil, attachmentTransitionResult{}, false, errAttachmentTransition | |
| 1322 | + | } | |
| 1323 | + | cwd := d.dirOrHome(current.cwd) | |
| 1324 | + | targetSess, err := d.createStoppedSessionLocked(target.Name, cwd, ac.sizeSnapshot(), env, current, current.tabNames) | |
| 1321 | 1325 | if err != nil { | |
| 1322 | 1326 | d.log.Warn("resuming stopped session failed", "err", err, "session", target.Name) | |
| 1323 | 1327 | return nil, attachmentTransitionResult{}, false, err | |
| @@ -1290,7 +1290,7 @@ func (d *Daemon) resumeStoppedAndSwitchLocked(from *session, ac *attachedClient, | |||||
| 1290 | from.mu.Lock() | 1290 | from.mu.Lock() | ||
| 1291 | cwd, env := d.dirOrHome(current.cwd), copyEnvironment(from.env) | 1291 | cwd, env := d.dirOrHome(current.cwd), copyEnvironment(from.env) | ||
| 1292 | from.mu.Unlock() | 1292 | from.mu.Unlock() | ||
| 1293 | - | created, createErr := d.createSessionLocked(target.Name, false, cwd, ac.sizeSnapshot(), env, current.tabNames) | 1293 | + | created, createErr := d.createStoppedSessionLocked(target.Name, cwd, ac.sizeSnapshot(), env, current, current.tabNames) |
| 1294 | targetSess = created | 1294 | targetSess = created | ||
| 1295 | return created, createErr | 1295 | return created, createErr | ||
| 1296 | }, | 1296 | }, | ||
| @@ -1316,8 +1316,12 @@ func (d *Daemon) resumeStoppedAndSwitchLocked(from *session, ac *attachedClient, | |||||
| 1316 | } | 1316 | } | ||
| 1317 | env := copyEnvironment(from.env) | 1317 | env := copyEnvironment(from.env) | ||
| 1318 | from.mu.Unlock() | 1318 | from.mu.Unlock() | ||
| 1319 | - | cwd := d.dirOrHome(stopped.cwd) | 1319 | + | current, ok := d.stopped[target.Name] |
| 1320 | - | targetSess, err := d.createSessionLocked(target.Name, false, cwd, ac.sizeSnapshot(), env, stopped.tabNames) | 1320 | + | if !ok || current.purging || !stoppedSessionMatchesLifecycle(current, stopped) || !targetMatchesLifecycle(target, current.name, current.createdAt, current.incarnation) { |
| 1321 | + | return nil, attachmentTransitionResult{}, false, errAttachmentTransition | |||
| 1322 | + | } | |||
| 1323 | + | cwd := d.dirOrHome(current.cwd) | |||
| 1324 | + | targetSess, err := d.createStoppedSessionLocked(target.Name, cwd, ac.sizeSnapshot(), env, current, current.tabNames) | |||
| 1321 | if err != nil { | 1325 | if err != nil { | ||
| 1322 | d.log.Warn("resuming stopped session failed", "err", err, "session", target.Name) | 1326 | d.log.Warn("resuming stopped session failed", "err", err, "session", target.Name) | ||
| 1323 | return nil, attachmentTransitionResult{}, false, err | 1327 | return nil, attachmentTransitionResult{}, false, err | ||
M internal/usecase/daemon/session.go +20 -0
| @@ -186,7 +186,21 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz | |||
| 186 | 186 | return d.createSessionLockedWithMode(name, ephemeral, cwd, sz, env, restoredTabNames...) | |
| 187 | 187 | } | |
| 188 | 188 | ||
| 189 | + | // createStoppedSessionLocked resumes exactly expectedStopped. It rechecks the | |
| 190 | + | // stopped lifecycle after catalogue I/O, which intentionally runs without d.mu. | |
| 191 | + | func (d *Daemon) createStoppedSessionLocked(name string, cwd string, sz domain.Size, env []string, expectedStopped stoppedSession, restoredTabNames ...[]string) (*session, error) { | |
| 192 | + | return d.createSessionLockedWithModeAndStoppedFence(name, false, cwd, sz, env, &expectedStopped, restoredTabNames...) | |
| 193 | + | } | |
| 194 | + | ||
| 195 | + | func stoppedSessionMatchesLifecycle(current, expected stoppedSession) bool { | |
| 196 | + | return current.name == expected.name && current.createdAt == expected.createdAt && current.incarnation == expected.incarnation | |
| 197 | + | } | |
| 198 | + | ||
| 189 | 199 | func (d *Daemon) createSessionLockedWithMode(name string, ephemeral bool, cwd string, sz domain.Size, env []string, restoredTabNames ...[]string) (*session, error) { | |
| 200 | + | return d.createSessionLockedWithModeAndStoppedFence(name, ephemeral, cwd, sz, env, nil, restoredTabNames...) | |
| 201 | + | } | |
| 202 | + | ||
| 203 | + | func (d *Daemon) createSessionLockedWithModeAndStoppedFence(name string, ephemeral bool, cwd string, sz domain.Size, env []string, expectedStopped *stoppedSession, restoredTabNames ...[]string) (*session, error) { | |
| 190 | 204 | env = copyEnvironment(env) | |
| 191 | 205 | if _, reserved := d.creating[name]; reserved { | |
| 192 | 206 | return nil, errSessionNameInUse | |
| @@ -194,6 +208,9 @@ func (d *Daemon) createSessionLockedWithMode(name string, ephemeral bool, cwd st | |||
| 194 | 208 | d.creating[name] = struct{}{} | |
| 195 | 209 | defer delete(d.creating, name) | |
| 196 | 210 | stopped, resuming := d.stopped[name] | |
| 211 | + | if expectedStopped != nil && (!resuming || !stoppedSessionMatchesLifecycle(stopped, *expectedStopped)) { | |
| 212 | + | return nil, errAttachmentTransition | |
| 213 | + | } | |
| 197 | 214 | var authoritative domain.CatalogueRecord | |
| 198 | 215 | var authoritativeExists bool | |
| 199 | 216 | if !ephemeral && resuming && d.persistEnabled { | |
| @@ -208,6 +225,9 @@ func (d *Daemon) createSessionLockedWithMode(name string, ephemeral bool, cwd st | |||
| 208 | 225 | return nil, &protoErr{ports.ErrServerShutdown, "daemon is shutting down"} | |
| 209 | 226 | } | |
| 210 | 227 | stopped, resuming = d.stopped[name] | |
| 228 | + | if expectedStopped != nil && (!resuming || !stoppedSessionMatchesLifecycle(stopped, *expectedStopped)) { | |
| 229 | + | return nil, errAttachmentTransition | |
| 230 | + | } | |
| 211 | 231 | } | |
| 212 | 232 | var createdAt int64 | |
| 213 | 233 | var incarnation domain.IncarnationID | |
| @@ -186,7 +186,21 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz | |||||
| 186 | return d.createSessionLockedWithMode(name, ephemeral, cwd, sz, env, restoredTabNames...) | 186 | return d.createSessionLockedWithMode(name, ephemeral, cwd, sz, env, restoredTabNames...) | ||
| 187 | } | 187 | } | ||
| 188 | 188 | ||||
| 189 | + | // createStoppedSessionLocked resumes exactly expectedStopped. It rechecks the | |||
| 190 | + | // stopped lifecycle after catalogue I/O, which intentionally runs without d.mu. | |||
| 191 | + | func (d *Daemon) createStoppedSessionLocked(name string, cwd string, sz domain.Size, env []string, expectedStopped stoppedSession, restoredTabNames ...[]string) (*session, error) { | |||
| 192 | + | return d.createSessionLockedWithModeAndStoppedFence(name, false, cwd, sz, env, &expectedStopped, restoredTabNames...) | |||
| 193 | + | } | |||
| 194 | + | ||||
| 195 | + | func stoppedSessionMatchesLifecycle(current, expected stoppedSession) bool { | |||
| 196 | + | return current.name == expected.name && current.createdAt == expected.createdAt && current.incarnation == expected.incarnation | |||
| 197 | + | } | |||
| 198 | + | ||||
| 189 | func (d *Daemon) createSessionLockedWithMode(name string, ephemeral bool, cwd string, sz domain.Size, env []string, restoredTabNames ...[]string) (*session, error) { | 199 | func (d *Daemon) createSessionLockedWithMode(name string, ephemeral bool, cwd string, sz domain.Size, env []string, restoredTabNames ...[]string) (*session, error) { | ||
| 200 | + | return d.createSessionLockedWithModeAndStoppedFence(name, ephemeral, cwd, sz, env, nil, restoredTabNames...) | |||
| 201 | + | } | |||
| 202 | + | ||||
| 203 | + | func (d *Daemon) createSessionLockedWithModeAndStoppedFence(name string, ephemeral bool, cwd string, sz domain.Size, env []string, expectedStopped *stoppedSession, restoredTabNames ...[]string) (*session, error) { | |||
| 190 | env = copyEnvironment(env) | 204 | env = copyEnvironment(env) | ||
| 191 | if _, reserved := d.creating[name]; reserved { | 205 | if _, reserved := d.creating[name]; reserved { | ||
| 192 | return nil, errSessionNameInUse | 206 | return nil, errSessionNameInUse | ||
| @@ -194,6 +208,9 @@ func (d *Daemon) createSessionLockedWithMode(name string, ephemeral bool, cwd st | |||||
| 194 | d.creating[name] = struct{}{} | 208 | d.creating[name] = struct{}{} | ||
| 195 | defer delete(d.creating, name) | 209 | defer delete(d.creating, name) | ||
| 196 | stopped, resuming := d.stopped[name] | 210 | stopped, resuming := d.stopped[name] | ||
| 211 | + | if expectedStopped != nil && (!resuming || !stoppedSessionMatchesLifecycle(stopped, *expectedStopped)) { | |||
| 212 | + | return nil, errAttachmentTransition | |||
| 213 | + | } | |||
| 197 | var authoritative domain.CatalogueRecord | 214 | var authoritative domain.CatalogueRecord | ||
| 198 | var authoritativeExists bool | 215 | var authoritativeExists bool | ||
| 199 | if !ephemeral && resuming && d.persistEnabled { | 216 | if !ephemeral && resuming && d.persistEnabled { | ||
| @@ -208,6 +225,9 @@ func (d *Daemon) createSessionLockedWithMode(name string, ephemeral bool, cwd st | |||||
| 208 | return nil, &protoErr{ports.ErrServerShutdown, "daemon is shutting down"} | 225 | return nil, &protoErr{ports.ErrServerShutdown, "daemon is shutting down"} | ||
| 209 | } | 226 | } | ||
| 210 | stopped, resuming = d.stopped[name] | 227 | stopped, resuming = d.stopped[name] | ||
| 228 | + | if expectedStopped != nil && (!resuming || !stoppedSessionMatchesLifecycle(stopped, *expectedStopped)) { | |||
| 229 | + | return nil, errAttachmentTransition | |||
| 230 | + | } | |||
| 211 | } | 231 | } | ||
| 212 | var createdAt int64 | 232 | var createdAt int64 | ||
| 213 | var incarnation domain.IncarnationID | 233 | var incarnation domain.IncarnationID | ||