fix(daemon): complete snapshot repository cutover

commit: e3b770390daf33a639159a76284eb6e136822357
author: brice · 3mo ago · 2026-07-21 22:29 UTC
parent: d32ee68
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +210 -24

M internal/usecase/daemon/session.go +25 -9


@@ -74,6 +74,7 @@ type session struct {
74	74		snapshotPublicationContext context.Context
75	75		snapshotPublicationCancel context.CancelFunc
76	76		snapshotQuarantined bool
	77	+	snapshotQuarantineEpoch uint64
77	78		// snapshotFailureSig is the stable class of the active failed publication.
78	79		// It is guarded by snapshotMu and deliberately never contains error text.
79	80		snapshotFailureSig string

@@ -583,13 +584,14 @@ func (d Daemon) renameSession(sess session, name string) error {
583	584		sess.mu.Unlock()
584	585		d.mu.Unlock()
585	586
586		-	rollback := func(err error, resume bool) error {
	587	+	rollback := func(err error) error {
	588	+	// The old record remains authoritative until its deletion succeeds. If
	589	+	// repository deletion was attempted, its outcome is not safely
	590	+	// reversible, so the coordinator stays quarantined rather than risking a
	591	+	// publication that recreates the deleted old identity.
587	592		if cleanupErr := d.persist.Delete(name); cleanupErr != nil {
588	593		d.log.Warn("cleaning up renamed persisted session failed", "err", cleanupErr, "session", name)
589	594		}
590		-	if resume {
591		-	resumeSnapshotCoordinator(sess)
592		-	}
593	595		sess.mu.Lock()
594	596		sess.renameInProgress = false
595	597		sess.mu.Unlock()

@@ -603,17 +605,18 @@ func (d Daemon) renameSession(sess session, name string) error {
603	605		return err
604	606		}
605	607		}
	608	+	var quarantine snapshotCoordinatorQuarantine
606	609		if !wasEphemeral && oldName != name {
607		-	<-quarantineSnapshotCoordinator(sess)
	610	+	quarantine = quarantineSnapshotCoordinatorWithEpoch(sess)
	611	+	<-quarantine.done
608	612		if d.snapsEnabled && d.snapshotRepository != nil {
609	613		if err := d.snapshotRepository.Delete(context.Background(), oldName); err != nil {
610		-	return rollback(err, true)
	614	+	return rollback(err)
611	615		}
612	616		}
613	617		if err := d.persist.Delete(oldName); err != nil {
614		-	return rollback(err, true)
	618	+	return rollback(err)
615	619		}
616		-	resumeSnapshotCoordinator(sess)
617	620		}
618	621
619	622		d.mu.Lock()

@@ -626,9 +629,22 @@ func (d Daemon) renameSession(sess session, name string) error {
626	629		sess.createdAt = createdAt
627	630		sess.ephemeral = false
628	631		sess.renameInProgress = false
629		-	sess.snapEligible.Store(name != "")
	632	+	// Keep the coordinator in quarantine until the new in-memory identity is
	633	+	// fully committed. The fresh coordinator below makes the new repository
	634	+	// start at generation one.
	635	+	if !wasEphemeral && oldName != name {
	636	+	sess.snapEligible.Store(false)
	637	+	} else {
	638	+	sess.snapEligible.Store(name != "")
	639	+	}
630	640		sess.mu.Unlock()
631	641		d.mu.Unlock()
	642	+	if !wasEphemeral && oldName != name {
	643	+	// A concurrent teardown may have quarantined after this rename did. In
	644	+	// that case its newer epoch owns the stopped coordinator.
	645	+	resumeSnapshotCoordinatorForNewIdentity(sess, quarantine)
	646	+	return nil
	647	+	}
632	648		markSnapshotDirty(sess)
633	649		return nil
634	650		}

 	snapshotPublicationContext context.Context
 	snapshotPublicationCancel  context.CancelFunc
 	snapshotQuarantined        bool
+	snapshotQuarantineEpoch    uint64
 	// snapshotFailureSig is the stable class of the active failed publication.
 	// It is guarded by snapshotMu and deliberately never contains error text.
 	snapshotFailureSig string
 	sess.mu.Unlock()
 	d.mu.Unlock()
-	rollback := func(err error, resume bool) error {
+	rollback := func(err error) error {
+		// The old record remains authoritative until its deletion succeeds. If
+		// repository deletion was attempted, its outcome is not safely
+		// reversible, so the coordinator stays quarantined rather than risking a
+		// publication that recreates the deleted old identity.
 		if cleanupErr := d.persist.Delete(name); cleanupErr != nil {
 			d.log.Warn("cleaning up renamed persisted session failed", "err", cleanupErr, "session", name)
 		}
-		if resume {
-			resumeSnapshotCoordinator(sess)
-		}
 		sess.mu.Lock()
 		sess.renameInProgress = false
 		sess.mu.Unlock()
 			return err
 		}
 	}
+	var quarantine snapshotCoordinatorQuarantine
 	if !wasEphemeral && oldName != name {
-		<-quarantineSnapshotCoordinator(sess)
+		quarantine = quarantineSnapshotCoordinatorWithEpoch(sess)
+		<-quarantine.done
 		if d.snapsEnabled && d.snapshotRepository != nil {
 			if err := d.snapshotRepository.Delete(context.Background(), oldName); err != nil {
-				return rollback(err, true)
+				return rollback(err)
 			}
 		}
 		if err := d.persist.Delete(oldName); err != nil {
-			return rollback(err, true)
+			return rollback(err)
 		}
-		resumeSnapshotCoordinator(sess)
 	}
 	d.mu.Lock()
 	sess.createdAt = createdAt
 	sess.ephemeral = false
 	sess.renameInProgress = false
-	sess.snapEligible.Store(name != "")
+	// Keep the coordinator in quarantine until the new in-memory identity is
+	// fully committed. The fresh coordinator below makes the new repository
+	// start at generation one.
+	if !wasEphemeral && oldName != name {
+		sess.snapEligible.Store(false)
+	} else {
+		sess.snapEligible.Store(name != "")
+	}
 	sess.mu.Unlock()
 	d.mu.Unlock()
+	if !wasEphemeral && oldName != name {
+		// A concurrent teardown may have quarantined after this rename did. In
+		// that case its newer epoch owns the stopped coordinator.
+		resumeSnapshotCoordinatorForNewIdentity(sess, quarantine)
+		return nil
+	}
 	markSnapshotDirty(sess)
 	return nil
 }

M internal/usecase/daemon/session_test.go +15 -8


@@ -674,12 +674,13 @@ func TestServeReturnsDespiteWedgedClientOnShutdown(t *testing.T) {
674	674		}
675	675
676	676		type mockStoreState struct {
677		-	mu sync.Mutex
678		-	data map[string][]byte
679		-	sets int
680		-	dels []string
681		-	syncs int
682		-	closed bool
	677	+	mu sync.Mutex
	678	+	data map[string][]byte
	679	+	sets int
	680	+	dels []string
	681	+	deleteErr func(string) error
	682	+	syncs int
	683	+	closed bool
683	684		}
684	685
685	686		func newMockStore(t testing.T) (portsmocks.MockStore, *mockStoreState) {

@@ -702,8 +703,14 @@ func newMockStore(t testing.T) (portsmocks.MockStore, *mockStoreState) {
702	703		store.EXPECT().Delete(mock.Anything).RunAndReturn(func(k []byte) error {
703	704		state.mu.Lock()
704	705		defer state.mu.Unlock()
705		-	state.dels = append(state.dels, string(k))
706		-	delete(state.data, string(k))
	706	+	key := string(k)
	707	+	state.dels = append(state.dels, key)
	708	+	if state.deleteErr != nil {
	709	+	if err := state.deleteErr(key); err != nil {
	710	+	return err
	711	+	}
	712	+	}
	713	+	delete(state.data, key)
707	714		return nil
708	715		}).Maybe()
709	716		store.EXPECT().Range(mock.Anything).Run(func(fn func(k, v []byte) bool) {

 }
 type mockStoreState struct {
-	mu     sync.Mutex
+	mu        sync.Mutex
-	data   map[string][]byte
+	data      map[string][]byte
-	sets   int
+	sets      int
-	dels   []string
+	dels      []string
-	syncs  int
+	deleteErr func(string) error
-	closed bool
+	syncs     int
+	closed    bool
 }
 func newMockStore(t *testing.T) (*portsmocks.MockStore, *mockStoreState) {
 	store.EXPECT().Delete(mock.Anything).RunAndReturn(func(k []byte) error {
 		state.mu.Lock()
 		defer state.mu.Unlock()
-		state.dels = append(state.dels, string(k))
+		key := string(k)
-		delete(state.data, string(k))
+		state.dels = append(state.dels, key)
+		if state.deleteErr != nil {
+			if err := state.deleteErr(key); err != nil {
+				return err
+			}
+		}
+		delete(state.data, key)
 		return nil
 	}).Maybe()
 	store.EXPECT().Range(mock.Anything).Run(func(fn func(k, v []byte) bool) {

M internal/usecase/daemon/snapshot.go +43 -7


@@ -313,17 +313,30 @@ func snapshotCoordinatorContext(sess *session) (context.Context, context.CancelF
313	313		return sess.snapshotPublicationContext, sess.snapshotPublicationCancel
314	314		}
315	315
	316	+	// snapshotCoordinatorQuarantine identifies one stop request. A later teardown
	317	+	// can supersede it, in which case only that later owner may resume publication.
	318	+	type snapshotCoordinatorQuarantine struct {
	319	+	done <-chan struct{}
	320	+	epoch uint64
	321	+	}
	322	+
316	323		// quarantineSnapshotCoordinator prevents a deleted session from starting any
317	324		// further publication. The returned join channel is closed after an already
318	325		// started repository call returns; callers must wait without daemon or session
319	326		// locks before deleting repository state.
320	327		func quarantineSnapshotCoordinator(sess *session) <-chan struct{} {
	328	+	return quarantineSnapshotCoordinatorWithEpoch(sess).done
	329	+	}
	330	+
	331	+	func quarantineSnapshotCoordinatorWithEpoch(sess *session) snapshotCoordinatorQuarantine {
321	332		if sess == nil {
322	333		done := make(chan struct{})
323	334		close(done)
324		-	return done
	335	+	return snapshotCoordinatorQuarantine{done: done}
325	336		}
326	337		sess.snapshotMu.Lock()
	338	+	sess.snapshotQuarantineEpoch++
	339	+	epoch := sess.snapshotQuarantineEpoch
327	340		sess.snapshotQuarantined = true
328	341		sess.snapEligible.Store(false)
329	342		if sess.snapshotPublicationCancel != nil {

@@ -347,24 +360,47 @@ func quarantineSnapshotCoordinator(sess *session) <-chan struct{} {
347	360		}
348	361		sess.signalSnapshotChangedLocked()
349	362		sess.snapshotMu.Unlock()
350		-	return done
	363	+	return snapshotCoordinatorQuarantine{done: done, epoch: epoch}
351	364		}
352	365
353		-	// resumeSnapshotCoordinator is used only after a rename's old repository name
354		-	// has been quarantined and deleted. Queued captures were discarded during
355		-	// quarantine, so the new context cannot publish under the old identity.
356		-	func resumeSnapshotCoordinator(sess *session) {
	366	+	// resumeSnapshotCoordinatorForNewIdentity starts a fresh repository lineage
	367	+	// only after rename committed its new in-memory name. The epoch prevents a
	368	+	// concurrent teardown from being undone by a late rename completion.
	369	+	func resumeSnapshotCoordinatorForNewIdentity(sess *session, quarantine snapshotCoordinatorQuarantine) bool {
357	370		if sess == nil {
358		-	return
	371	+	return false
359	372		}
360	373		sess.snapshotMu.Lock()
	374	+	if !sess.snapshotQuarantined \|\| sess.snapshotQuarantineEpoch != quarantine.epoch {
	375	+	sess.snapshotMu.Unlock()
	376	+	return false
	377	+	}
361	378		if sess.snapshotPublicationCancel != nil {
362	379		sess.snapshotPublicationCancel()
363	380		}
364	381		sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
365	382		sess.snapshotQuarantined = false
	383	+	// A repository name owns its own generation stream. Reset all scheduler
	384	+	// state while eligibility remains false, then atomically make generation 1
	385	+	// dirty and wake the scheduler.
	386	+	sess.snapshotGeneration = 1
	387	+	sess.snapshotCapturedGeneration = 0
	388	+	sess.snapshotNextEligibleAt = time.Time{}
	389	+	sess.snapshotAttempted = false
	390	+	sess.snapshotAttemptKind = snapshotAttemptRoutine
	391	+	sess.snapshotFailureSig = ""
	392	+	sess.snapDirty.Store(true)
	393	+	sess.snapEligible.Store(true)
366	394		sess.signalSnapshotChangedLocked()
	395	+	wake := sess.snapshotWake
367	396		sess.snapshotMu.Unlock()
	397	+	if wake != nil {
	398	+	select {
	399	+	case wake <- struct{}{}:
	400	+	default:
	401	+	}
	402	+	}
	403	+	return true
368	404		}
369	405
370	406		func markSnapshotDirty(sess *session) {

 	return sess.snapshotPublicationContext, sess.snapshotPublicationCancel
 }
+// snapshotCoordinatorQuarantine identifies one stop request. A later teardown
+// can supersede it, in which case only that later owner may resume publication.
+type snapshotCoordinatorQuarantine struct {
+	done  <-chan struct{}
+	epoch uint64
+}
 // quarantineSnapshotCoordinator prevents a deleted session from starting any
 // further publication. The returned join channel is closed after an already
 // started repository call returns; callers must wait without daemon or session
 // locks before deleting repository state.
 func quarantineSnapshotCoordinator(sess *session) <-chan struct{} {
+	return quarantineSnapshotCoordinatorWithEpoch(sess).done
+}
+func quarantineSnapshotCoordinatorWithEpoch(sess *session) snapshotCoordinatorQuarantine {
 	if sess == nil {
 		done := make(chan struct{})
 		close(done)
-		return done
+		return snapshotCoordinatorQuarantine{done: done}
 	}
 	sess.snapshotMu.Lock()
+	sess.snapshotQuarantineEpoch++
+	epoch := sess.snapshotQuarantineEpoch
 	sess.snapshotQuarantined = true
 	sess.snapEligible.Store(false)
 	if sess.snapshotPublicationCancel != nil {
 	}
 	sess.signalSnapshotChangedLocked()
 	sess.snapshotMu.Unlock()
-	return done
+	return snapshotCoordinatorQuarantine{done: done, epoch: epoch}
 }
-// resumeSnapshotCoordinator is used only after a rename's old repository name
+// resumeSnapshotCoordinatorForNewIdentity starts a fresh repository lineage
-// has been quarantined and deleted. Queued captures were discarded during
+// only after rename committed its new in-memory name. The epoch prevents a
-// quarantine, so the new context cannot publish under the old identity.
+// concurrent teardown from being undone by a late rename completion.
-func resumeSnapshotCoordinator(sess *session) {
+func resumeSnapshotCoordinatorForNewIdentity(sess *session, quarantine snapshotCoordinatorQuarantine) bool {
 	if sess == nil {
-		return
+		return false
 	}
 	sess.snapshotMu.Lock()
+	if !sess.snapshotQuarantined || sess.snapshotQuarantineEpoch != quarantine.epoch {
+		sess.snapshotMu.Unlock()
+		return false
+	}
 	if sess.snapshotPublicationCancel != nil {
 		sess.snapshotPublicationCancel()
 	}
 	sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
 	sess.snapshotQuarantined = false
+	// A repository name owns its own generation stream. Reset all scheduler
+	// state while eligibility remains false, then atomically make generation 1
+	// dirty and wake the scheduler.
+	sess.snapshotGeneration = 1
+	sess.snapshotCapturedGeneration = 0
+	sess.snapshotNextEligibleAt = time.Time{}
+	sess.snapshotAttempted = false
+	sess.snapshotAttemptKind = snapshotAttemptRoutine
+	sess.snapshotFailureSig = ""
+	sess.snapDirty.Store(true)
+	sess.snapEligible.Store(true)
 	sess.signalSnapshotChangedLocked()
+	wake := sess.snapshotWake
 	sess.snapshotMu.Unlock()
+	if wake != nil {
+		select {
+		case wake <- struct{}{}:
+		default:
+		}
+	}
+	return true
 }
 func markSnapshotDirty(sess *session) {

M internal/usecase/daemon/snapshot_scheduler_test.go +127 -0


@@ -2,6 +2,7 @@ package daemon
2	2
3	3		import (
4	4		"context"
	5	+	"errors"
5	6		"sync"
6	7		"sync/atomic"
7	8		"testing"

@@ -10,6 +11,7 @@ import (
10	11		"github.com/stretchr/testify/mock"
11	12		"github.com/stretchr/testify/require"
12	13
	14	+	"github.com/bnema/vev/internal/domain"
13	15		"github.com/bnema/vev/internal/ports"
14	16		portsmocks "github.com/bnema/vev/internal/ports/mocks"
15	17		)

@@ -68,6 +70,131 @@ func TestSnapshotCoordinatorQuarantineJoinsInFlightPublication(t *testing.T) {
68	70		}
69	71		}
70	72
	73	+	func TestRenameKeepsSnapshotCoordinatorQuarantinedUntilNewIdentityCommits(t *testing.T) {
	74	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	75	+	store, state := newMockStore(t)
	76	+	WithStore(store)(d)
	77	+	repository := portsmocks.NewMockSnapshotRepository(t)
	78	+	WithSnapshotRepository(repository, nil)(d)
	79	+	startSnapshotEncodeWorker(t, d)
	80	+
	81	+	deleteEntered := make(chan struct{})
	82	+	allowDelete := make(chan struct{})
	83	+	published := make(chan ports.SnapshotPublication, 2)
	84	+	repository.EXPECT().Delete(mock.Anything, "work").RunAndReturn(func(context.Context, string) error {
	85	+	close(deleteEntered)
	86	+	<-allowDelete
	87	+	return nil
	88	+	}).Once()
	89	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
	90	+	published <- publication
	91	+	return nil
	92	+	}).Maybe()
	93	+
	94	+	pty, release := newBlockingPTY(t)
	95	+	defer release()
	96	+	d.ptys = newFactory(t, pty)
	97	+	sess, err := d.createSessionLocked("work", false, "/work", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, d.baseEnv)
	98	+	require.NoError(t, err)
	99	+	markSnapshotDirty(sess)
	100	+
	101	+	renameDone := make(chan error, 1)
	102	+	go func() { renameDone <- d.renameSession(sess, "renamed") }()
	103	+	select {
	104	+	case <-deleteEntered:
	105	+	case <-time.After(testWaitTimeout):
	106	+	t.Fatal("rename did not begin deleting the old snapshot identity")
	107	+	}
	108	+
	109	+	// Hold the daemon identity commit after Delete has returned. Scheduler work
	110	+	// during this window must not capture or publish the old name.
	111	+	d.mu.Lock()
	112	+	close(allowDelete)
	113	+	require.Eventually(t, func() bool { return !state.has("work") }, testWaitTimeout, time.Millisecond)
	114	+	sess.snapshotMu.Lock()
	115	+	quarantined := sess.snapshotQuarantined
	116	+	sess.snapshotMu.Unlock()
	117	+	require.True(t, quarantined, "old Delete returned before the coordinator resumed")
	118	+
	119	+	schedulerCtx, cancelScheduler := context.WithCancel(context.Background())
	120	+	defer cancelScheduler()
	121	+	go d.snapshotRepositorySaver(schedulerCtx)
	122	+	select {
	123	+	case d.snapshotWake <- struct{}{}:
	124	+	default:
	125	+	}
	126	+	require.True(t, d.scheduleSnapshot(sess))
	127	+	select {
	128	+	case publication := <-published:
	129	+	t.Fatalf("published %q after old Delete before identity commit", publication.Name)
	130	+	case <-time.After(50 * time.Millisecond):
	131	+	}
	132	+	d.mu.Unlock()
	133	+
	134	+	require.NoError(t, <-renameDone)
	135	+	select {
	136	+	case publication := <-published:
	137	+	require.Equal(t, "renamed", publication.Name)
	138	+	require.Equal(t, uint64(1), publication.Generation)
	139	+	case <-time.After(testWaitTimeout):
	140	+	t.Fatal("new identity was not published")
	141	+	}
	142	+	}
	143	+
	144	+	func TestRenameRollbackAfterOldDeleteLeavesOldCoordinatorStopped(t *testing.T) {
	145	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	146	+	store, state := newMockStore(t)
	147	+	WithStore(store)(d)
	148	+	repository := portsmocks.NewMockSnapshotRepository(t)
	149	+	WithSnapshotRepository(repository, nil)(d)
	150	+	startSnapshotEncodeWorker(t, d)
	151	+
	152	+	deleteOldRecordErr := errors.New("old record delete failed")
	153	+	state.mu.Lock()
	154	+	state.deleteErr = func(key string) error {
	155	+	if key == "work" {
	156	+	return deleteOldRecordErr
	157	+	}
	158	+	return nil
	159	+	}
	160	+	state.mu.Unlock()
	161	+	repository.EXPECT().Delete(mock.Anything, "work").Return(nil).Once()
	162	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(context.Context, ports.SnapshotPublication) {
	163	+	t.Fatal("rollback resurrected the deleted old snapshot identity")
	164	+	}).Maybe()
	165	+
	166	+	pty, release := newBlockingPTY(t)
	167	+	defer release()
	168	+	d.ptys = newFactory(t, pty)
	169	+	sess, err := d.createSessionLocked("work", false, "/work", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, d.baseEnv)
	170	+	require.NoError(t, err)
	171	+	markSnapshotDirty(sess)
	172	+
	173	+	require.ErrorIs(t, d.renameSession(sess, "renamed"), deleteOldRecordErr)
	174	+	sess.mu.Lock()
	175	+	require.Equal(t, "work", sess.name)
	176	+	require.False(t, sess.ephemeral)
	177	+	require.False(t, sess.renameInProgress)
	178	+	sess.mu.Unlock()
	179	+	require.True(t, state.has("work"))
	180	+	require.False(t, state.has("renamed"))
	181	+	sess.snapshotMu.Lock()
	182	+	quarantined := sess.snapshotQuarantined
	183	+	sess.snapshotMu.Unlock()
	184	+	require.True(t, quarantined)
	185	+
	186	+	schedulerCtx, cancelScheduler := context.WithCancel(context.Background())
	187	+	defer cancelScheduler()
	188	+	go d.snapshotRepositorySaver(schedulerCtx)
	189	+	select {
	190	+	case d.snapshotWake <- struct{}{}:
	191	+	default:
	192	+	}
	193	+	select {
	194	+	case <-time.After(50 * time.Millisecond):
	195	+	}
	196	+	}
	197	+
71	198		func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {
72	199		clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}
73	200		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)


@@ -2,6 +2,7 @@ package daemon
2		2
3	import (	3		import (
4	"context"	4		"context"
		5	+	"errors"
5	"sync"	6		"sync"
6	"sync/atomic"	7		"sync/atomic"
7	"testing"	8		"testing"

@@ -10,6 +11,7 @@ import (
10	"github.com/stretchr/testify/mock"	11		"github.com/stretchr/testify/mock"
11	"github.com/stretchr/testify/require"	12		"github.com/stretchr/testify/require"
12		13
		14	+	"github.com/bnema/vev/internal/domain"
13	"github.com/bnema/vev/internal/ports"	15		"github.com/bnema/vev/internal/ports"
14	portsmocks "github.com/bnema/vev/internal/ports/mocks"	16		portsmocks "github.com/bnema/vev/internal/ports/mocks"
15	)	17		)

@@ -68,6 +70,131 @@ func TestSnapshotCoordinatorQuarantineJoinsInFlightPublication(t *testing.T) {
68	}	70		}
69	}	71		}
70		72
		73	+	func TestRenameKeepsSnapshotCoordinatorQuarantinedUntilNewIdentityCommits(t *testing.T) {
		74	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
		75	+	store, state := newMockStore(t)
		76	+	WithStore(store)(d)
		77	+	repository := portsmocks.NewMockSnapshotRepository(t)
		78	+	WithSnapshotRepository(repository, nil)(d)
		79	+	startSnapshotEncodeWorker(t, d)
		80	+
		81	+	deleteEntered := make(chan struct{})
		82	+	allowDelete := make(chan struct{})
		83	+	published := make(chan ports.SnapshotPublication, 2)
		84	+	repository.EXPECT().Delete(mock.Anything, "work").RunAndReturn(func(context.Context, string) error {
		85	+	close(deleteEntered)
		86	+	<-allowDelete
		87	+	return nil
		88	+	}).Once()
		89	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
		90	+	published <- publication
		91	+	return nil
		92	+	}).Maybe()
		93	+
		94	+	pty, release := newBlockingPTY(t)
		95	+	defer release()
		96	+	d.ptys = newFactory(t, pty)
		97	+	sess, err := d.createSessionLocked("work", false, "/work", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, d.baseEnv)
		98	+	require.NoError(t, err)
		99	+	markSnapshotDirty(sess)
		100	+
		101	+	renameDone := make(chan error, 1)
		102	+	go func() { renameDone <- d.renameSession(sess, "renamed") }()
		103	+	select {
		104	+	case <-deleteEntered:
		105	+	case <-time.After(testWaitTimeout):
		106	+	t.Fatal("rename did not begin deleting the old snapshot identity")
		107	+	}
		108	+
		109	+	// Hold the daemon identity commit after Delete has returned. Scheduler work
		110	+	// during this window must not capture or publish the old name.
		111	+	d.mu.Lock()
		112	+	close(allowDelete)
		113	+	require.Eventually(t, func() bool { return !state.has("work") }, testWaitTimeout, time.Millisecond)
		114	+	sess.snapshotMu.Lock()
		115	+	quarantined := sess.snapshotQuarantined
		116	+	sess.snapshotMu.Unlock()
		117	+	require.True(t, quarantined, "old Delete returned before the coordinator resumed")
		118	+
		119	+	schedulerCtx, cancelScheduler := context.WithCancel(context.Background())
		120	+	defer cancelScheduler()
		121	+	go d.snapshotRepositorySaver(schedulerCtx)
		122	+	select {
		123	+	case d.snapshotWake <- struct{}{}:
		124	+	default:
		125	+	}
		126	+	require.True(t, d.scheduleSnapshot(sess))
		127	+	select {
		128	+	case publication := <-published:
		129	+	t.Fatalf("published %q after old Delete before identity commit", publication.Name)
		130	+	case <-time.After(50 * time.Millisecond):
		131	+	}
		132	+	d.mu.Unlock()
		133	+
		134	+	require.NoError(t, <-renameDone)
		135	+	select {
		136	+	case publication := <-published:
		137	+	require.Equal(t, "renamed", publication.Name)
		138	+	require.Equal(t, uint64(1), publication.Generation)
		139	+	case <-time.After(testWaitTimeout):
		140	+	t.Fatal("new identity was not published")
		141	+	}
		142	+	}
		143	+
		144	+	func TestRenameRollbackAfterOldDeleteLeavesOldCoordinatorStopped(t *testing.T) {
		145	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
		146	+	store, state := newMockStore(t)
		147	+	WithStore(store)(d)
		148	+	repository := portsmocks.NewMockSnapshotRepository(t)
		149	+	WithSnapshotRepository(repository, nil)(d)
		150	+	startSnapshotEncodeWorker(t, d)
		151	+
		152	+	deleteOldRecordErr := errors.New("old record delete failed")
		153	+	state.mu.Lock()
		154	+	state.deleteErr = func(key string) error {
		155	+	if key == "work" {
		156	+	return deleteOldRecordErr
		157	+	}
		158	+	return nil
		159	+	}
		160	+	state.mu.Unlock()
		161	+	repository.EXPECT().Delete(mock.Anything, "work").Return(nil).Once()
		162	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).Run(func(context.Context, ports.SnapshotPublication) {
		163	+	t.Fatal("rollback resurrected the deleted old snapshot identity")
		164	+	}).Maybe()
		165	+
		166	+	pty, release := newBlockingPTY(t)
		167	+	defer release()
		168	+	d.ptys = newFactory(t, pty)
		169	+	sess, err := d.createSessionLocked("work", false, "/work", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, d.baseEnv)
		170	+	require.NoError(t, err)
		171	+	markSnapshotDirty(sess)
		172	+
		173	+	require.ErrorIs(t, d.renameSession(sess, "renamed"), deleteOldRecordErr)
		174	+	sess.mu.Lock()
		175	+	require.Equal(t, "work", sess.name)
		176	+	require.False(t, sess.ephemeral)
		177	+	require.False(t, sess.renameInProgress)
		178	+	sess.mu.Unlock()
		179	+	require.True(t, state.has("work"))
		180	+	require.False(t, state.has("renamed"))
		181	+	sess.snapshotMu.Lock()
		182	+	quarantined := sess.snapshotQuarantined
		183	+	sess.snapshotMu.Unlock()
		184	+	require.True(t, quarantined)
		185	+
		186	+	schedulerCtx, cancelScheduler := context.WithCancel(context.Background())
		187	+	defer cancelScheduler()
		188	+	go d.snapshotRepositorySaver(schedulerCtx)
		189	+	select {
		190	+	case d.snapshotWake <- struct{}{}:
		191	+	default:
		192	+	}
		193	+	select {
		194	+	case <-time.After(50 * time.Millisecond):
		195	+	}
		196	+	}
		197	+
71	func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {	198		func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {
72	clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}	199		clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}
73	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)	200		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)