fix(daemon): preserve the restore attach barrier
- commit
- c8b162c98fc34ffea8269aa51ede286883ed781a
- author
- brice · · 2026-07-26 12:04 UTC
- parent
- 84be71d
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +44 -10
M internal/usecase/daemon/daemon.go +16 -9
| @@ -944,16 +944,23 @@ func (d *Daemon) finishAttach(sess *session, tr ports.Transport, sz domain.Size, | |||
| 944 | 944 | ||
| 945 | 945 | func (d *Daemon) waitForTargetRestore(ctx context.Context, name string) error { | |
| 946 | 946 | d.mu.Lock() | |
| 947 | - | if d.findByNameLocked(name) != nil { | |
| 948 | - | d.mu.Unlock() | |
| 949 | - | return nil | |
| 950 | - | } | |
| 951 | - | stopped, ok := d.stopped[name] | |
| 952 | - | if !ok || stopped.purging { | |
| 953 | - | d.mu.Unlock() | |
| 954 | - | return nil | |
| 947 | + | var ( | |
| 948 | + | done chan struct{} | |
| 949 | + | stopped stoppedSession | |
| 950 | + | ok bool | |
| 951 | + | ) | |
| 952 | + | if sess := d.findByNameLocked(name); sess != nil { | |
| 953 | + | sess.mu.Lock() | |
| 954 | + | done = sess.restoreDone | |
| 955 | + | sess.mu.Unlock() | |
| 956 | + | } else { | |
| 957 | + | stopped, ok = d.stopped[name] | |
| 958 | + | if !ok || stopped.purging { | |
| 959 | + | d.mu.Unlock() | |
| 960 | + | return nil | |
| 961 | + | } | |
| 962 | + | done = stopped.restoreDone | |
| 955 | 963 | } | |
| 956 | - | done := stopped.restoreDone | |
| 957 | 964 | d.mu.Unlock() | |
| 958 | 965 | ||
| 959 | 966 | if done != nil { | |
| @@ -944,16 +944,23 @@ func (d *Daemon) finishAttach(sess *session, tr ports.Transport, sz domain.Size, | |||||
| 944 | 944 | ||||
| 945 | func (d *Daemon) waitForTargetRestore(ctx context.Context, name string) error { | 945 | func (d *Daemon) waitForTargetRestore(ctx context.Context, name string) error { | ||
| 946 | d.mu.Lock() | 946 | d.mu.Lock() | ||
| 947 | - | if d.findByNameLocked(name) != nil { | 947 | + | var ( |
| 948 | - | d.mu.Unlock() | 948 | + | done chan struct{} |
| 949 | - | return nil | 949 | + | stopped stoppedSession |
| 950 | - | } | 950 | + | ok bool |
| 951 | - | stopped, ok := d.stopped[name] | 951 | + | ) |
| 952 | - | if !ok || stopped.purging { | 952 | + | if sess := d.findByNameLocked(name); sess != nil { |
| 953 | - | d.mu.Unlock() | 953 | + | sess.mu.Lock() |
| 954 | - | return nil | 954 | + | done = sess.restoreDone |
| 955 | + | sess.mu.Unlock() | |||
| 956 | + | } else { | |||
| 957 | + | stopped, ok = d.stopped[name] | |||
| 958 | + | if !ok || stopped.purging { | |||
| 959 | + | d.mu.Unlock() | |||
| 960 | + | return nil | |||
| 961 | + | } | |||
| 962 | + | done = stopped.restoreDone | |||
| 955 | } | 963 | } | ||
| 956 | - | done := stopped.restoreDone | |||
| 957 | d.mu.Unlock() | 964 | d.mu.Unlock() | ||
| 958 | 965 | ||||
| 959 | if done != nil { | 966 | if done != nil { | ||
M internal/usecase/daemon/durable_recovery_test.go +20 -0
| @@ -551,6 +551,26 @@ func TestAttachWaitsForRestore(t *testing.T) { | |||
| 551 | 551 | require.Zero(t, factory.calls.Load()) | |
| 552 | 552 | } | |
| 553 | 553 | ||
| 554 | + | func TestAttachBarrierSurvivesLiveRegistryPublication(t *testing.T) { | |
| 555 | + | record := durableRecoveryRecord(0) | |
| 556 | + | d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{}) | |
| 557 | + | done := d.recordRestoreDone(record.Name) | |
| 558 | + | sess := &session{name: record.Name, incarnation: record.IncarnationID} | |
| 559 | + | ||
| 560 | + | registered, err := d.persistAndRegisterRestoredSession(t.Context(), sess) | |
| 561 | + | require.NoError(t, err) | |
| 562 | + | require.True(t, registered) | |
| 563 | + | ||
| 564 | + | cancelled, cancel := context.WithCancel(context.Background()) | |
| 565 | + | cancel() | |
| 566 | + | require.ErrorIs(t, d.waitForTargetRestore(cancelled, record.Name), context.Canceled, | |
| 567 | + | "an attach racing after live publication must still observe the restore barrier") | |
| 568 | + | ||
| 569 | + | d.finishRecordRestore(record, nil, done) | |
| 570 | + | require.NoError(t, d.waitForTargetRestore(context.Background(), record.Name), | |
| 571 | + | "attach must proceed after successful restoration completes") | |
| 572 | + | } | |
| 573 | + | ||
| 554 | 574 | func TestAttachRestoreWaitHonorsContextCancellation(t *testing.T) { | |
| 555 | 575 | record := durableRecoveryRecord(0) | |
| 556 | 576 | d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{}) | |
| @@ -551,6 +551,26 @@ func TestAttachWaitsForRestore(t *testing.T) { | |||||
| 551 | require.Zero(t, factory.calls.Load()) | 551 | require.Zero(t, factory.calls.Load()) | ||
| 552 | } | 552 | } | ||
| 553 | 553 | ||||
| 554 | + | func TestAttachBarrierSurvivesLiveRegistryPublication(t *testing.T) { | |||
| 555 | + | record := durableRecoveryRecord(0) | |||
| 556 | + | d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{}) | |||
| 557 | + | done := d.recordRestoreDone(record.Name) | |||
| 558 | + | sess := &session{name: record.Name, incarnation: record.IncarnationID} | |||
| 559 | + | ||||
| 560 | + | registered, err := d.persistAndRegisterRestoredSession(t.Context(), sess) | |||
| 561 | + | require.NoError(t, err) | |||
| 562 | + | require.True(t, registered) | |||
| 563 | + | ||||
| 564 | + | cancelled, cancel := context.WithCancel(context.Background()) | |||
| 565 | + | cancel() | |||
| 566 | + | require.ErrorIs(t, d.waitForTargetRestore(cancelled, record.Name), context.Canceled, | |||
| 567 | + | "an attach racing after live publication must still observe the restore barrier") | |||
| 568 | + | ||||
| 569 | + | d.finishRecordRestore(record, nil, done) | |||
| 570 | + | require.NoError(t, d.waitForTargetRestore(context.Background(), record.Name), | |||
| 571 | + | "attach must proceed after successful restoration completes") | |||
| 572 | + | } | |||
| 573 | + | ||||
| 554 | func TestAttachRestoreWaitHonorsContextCancellation(t *testing.T) { | 574 | func TestAttachRestoreWaitHonorsContextCancellation(t *testing.T) { | ||
| 555 | record := durableRecoveryRecord(0) | 575 | record := durableRecoveryRecord(0) | ||
| 556 | d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{}) | 576 | d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{}) | ||
M internal/usecase/daemon/session.go +4 -1
| @@ -52,7 +52,10 @@ type session struct { | |||
| 52 | 52 | teardownWaiters uint | |
| 53 | 53 | teardownChanged chan struct{} | |
| 54 | 54 | lifecycleStopOnce sync.Once | |
| 55 | - | mu sync.Mutex // guards tabs, active, client, clipFiles, and clipboard queue state | |
| 55 | + | mu sync.Mutex // guards tabs, active, client, restoreDone, clipFiles, and clipboard queue state | |
| 56 | + | // restoreDone remains attached to a restored session after it is published in | |
| 57 | + | // the live registry, so racing attaches cannot bypass restoration completion. | |
| 58 | + | restoreDone chan struct{} | |
| 56 | 59 | // metadataPersistMu serializes authority writes and post-I/O rollback. State | |
| 57 | 60 | // mutation paths must release d.mu and mu before acquiring it. After durable | |
| 58 | 61 | // I/O completes it may acquire d.mu then mu to reconcile failed revisions. | |
| @@ -52,7 +52,10 @@ type session struct { | |||||
| 52 | teardownWaiters uint | 52 | teardownWaiters uint | ||
| 53 | teardownChanged chan struct{} | 53 | teardownChanged chan struct{} | ||
| 54 | lifecycleStopOnce sync.Once | 54 | lifecycleStopOnce sync.Once | ||
| 55 | - | mu sync.Mutex // guards tabs, active, client, clipFiles, and clipboard queue state | 55 | + | mu sync.Mutex // guards tabs, active, client, restoreDone, clipFiles, and clipboard queue state |
| 56 | + | // restoreDone remains attached to a restored session after it is published in | |||
| 57 | + | // the live registry, so racing attaches cannot bypass restoration completion. | |||
| 58 | + | restoreDone chan struct{} | |||
| 56 | // metadataPersistMu serializes authority writes and post-I/O rollback. State | 59 | // metadataPersistMu serializes authority writes and post-I/O rollback. State | ||
| 57 | // mutation paths must release d.mu and mu before acquiring it. After durable | 60 | // mutation paths must release d.mu and mu before acquiring it. After durable | ||
| 58 | // I/O completes it may acquire d.mu then mu to reconcile failed revisions. | 61 | // I/O completes it may acquire d.mu then mu to reconcile failed revisions. | ||
M internal/usecase/daemon/snapshot_restore.go +4 -0
| @@ -334,6 +334,10 @@ func (d *Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess *se | |||
| 334 | 334 | } | |
| 335 | 335 | sess.id = domain.SessionID(fmt.Sprintf("sess-%d", d.nextID)) | |
| 336 | 336 | d.nextID++ | |
| 337 | + | stopped := d.stopped[sess.name] | |
| 338 | + | sess.mu.Lock() | |
| 339 | + | sess.restoreDone = stopped.restoreDone | |
| 340 | + | sess.mu.Unlock() | |
| 337 | 341 | delete(d.stopped, sess.name) | |
| 338 | 342 | d.sessions[sess.id] = sess | |
| 339 | 343 | sess.snapDirty.Store(false) | |
| @@ -334,6 +334,10 @@ func (d *Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess *se | |||||
| 334 | } | 334 | } | ||
| 335 | sess.id = domain.SessionID(fmt.Sprintf("sess-%d", d.nextID)) | 335 | sess.id = domain.SessionID(fmt.Sprintf("sess-%d", d.nextID)) | ||
| 336 | d.nextID++ | 336 | d.nextID++ | ||
| 337 | + | stopped := d.stopped[sess.name] | |||
| 338 | + | sess.mu.Lock() | |||
| 339 | + | sess.restoreDone = stopped.restoreDone | |||
| 340 | + | sess.mu.Unlock() | |||
| 337 | delete(d.stopped, sess.name) | 341 | delete(d.stopped, sess.name) | ||
| 338 | d.sessions[sess.id] = sess | 342 | d.sessions[sess.id] = sess | ||
| 339 | sess.snapDirty.Store(false) | 343 | sess.snapDirty.Store(false) | ||