fix(daemon): preserve forced checkpoint state

commit: d1419bcba9c84676b4adb9a5e2b4df5197813c38
author: brice · 3mo ago · 2026-07-21 22:40 UTC
committer: brice · 3mo ago · 2026-07-21 22:41 UTC
parent: e3b7703
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +289 -15

M internal/usecase/daemon/session.go +18 -10


@@ -56,18 +56,23 @@ type session struct {
56	56		// snapshotMu serializes the dirty generation with worker completion. It is
57	57		// intentionally independent from mu: persistence never holds session state
58	58		// locks while encoding or writing.
59		-	snapshotMu sync.Mutex
60		-	snapshotGeneration uint64
61		-	snapshotCapturedGeneration uint64
62		-	snapshotNextEligibleAt time.Time
63		-	snapshotAttempted bool
64		-	snapshotAttemptKind snapshotAttemptKind
	59	+	snapshotMu sync.Mutex
	60	+	snapshotGeneration uint64
	61	+	snapshotCapturedGeneration uint64
	62	+	snapshotPublishedGeneration uint64
	63	+	snapshotNextEligibleAt time.Time
	64	+	snapshotAttempted bool
	65	+	snapshotAttemptKind snapshotAttemptKind
65	66		// The coordinator state below is guarded by snapshotMu. A capture can be
66	67		// queued globally or in flight, never more than one of each per session.
67	68		// Quarantine cancels the session publication context before destructive
68	69		// repository operations and publicationDone joins a started publication.
69		-	snapshotPending bool
70		-	snapshotPendingCaptures uint
	70	+	snapshotPending bool
	71	+	snapshotPendingCaptures uint
	72	+	// snapshotForcedGeneration is the newest dirty generation a forced
	73	+	// checkpoint must publish. It survives an older routine capture so worker
	74	+	// completion can enqueue exactly one forced successor.
	75	+	snapshotForcedGeneration uint64
71	76		snapshotQueuedCapture *snapshotCapture
72	77		snapshotInFlightCapture *snapshotCapture
73	78		snapshotPublicationDone chan struct{}

@@ -809,11 +814,14 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
809	814		}
810	815		} else {
811	816		markSnapshotDirty(sess)
812		-	if !d.scheduleFinalSnapshot(sess) {
	817	+	sess.snapshotMu.Lock()
	818	+	terminalGeneration := sess.snapshotGeneration
	819	+	sess.snapshotMu.Unlock()
	820	+	if !d.scheduleFinalSnapshot(sess) \|\| !d.waitForSnapshotGeneration(sess, terminalGeneration) {
813	821		sess.mu.Lock()
814	822		name := sess.name
815	823		sess.mu.Unlock()
816		-	terminalSnapshotErr = fmt.Errorf("retain final snapshot for session %q: snapshot worker unavailable or saturated", name)
	824	+	terminalSnapshotErr = fmt.Errorf("retain final snapshot for session %q: snapshot worker unavailable, saturated, or timed out", name)
817	825		if reason != ports.ReasonServerShutdown {
818	826		return terminalSnapshotErr
819	827		}

 	// snapshotMu serializes the dirty generation with worker completion. It is
 	// intentionally independent from mu: persistence never holds session state
 	// locks while encoding or writing.
-	snapshotMu                 sync.Mutex
+	snapshotMu                  sync.Mutex
-	snapshotGeneration         uint64
+	snapshotGeneration          uint64
-	snapshotCapturedGeneration uint64
+	snapshotCapturedGeneration  uint64
-	snapshotNextEligibleAt     time.Time
+	snapshotPublishedGeneration uint64
-	snapshotAttempted          bool
+	snapshotNextEligibleAt      time.Time
-	snapshotAttemptKind        snapshotAttemptKind
+	snapshotAttempted           bool
+	snapshotAttemptKind         snapshotAttemptKind
 	// The coordinator state below is guarded by snapshotMu. A capture can be
 	// queued globally or in flight, never more than one of each per session.
 	// Quarantine cancels the session publication context before destructive
 	// repository operations and publicationDone joins a started publication.
-	snapshotPending            bool
+	snapshotPending         bool
-	snapshotPendingCaptures    uint
+	snapshotPendingCaptures uint
+	// snapshotForcedGeneration is the newest dirty generation a forced
+	// checkpoint must publish. It survives an older routine capture so worker
+	// completion can enqueue exactly one forced successor.
+	snapshotForcedGeneration   uint64
 	snapshotQueuedCapture      *snapshotCapture
 	snapshotInFlightCapture    *snapshotCapture
 	snapshotPublicationDone    chan struct{}
 			}
 		} else {
 			markSnapshotDirty(sess)
-			if !d.scheduleFinalSnapshot(sess) {
+			sess.snapshotMu.Lock()
+			terminalGeneration := sess.snapshotGeneration
+			sess.snapshotMu.Unlock()
+			if !d.scheduleFinalSnapshot(sess) || !d.waitForSnapshotGeneration(sess, terminalGeneration) {
 				sess.mu.Lock()
 				name := sess.name
 				sess.mu.Unlock()
-				terminalSnapshotErr = fmt.Errorf("retain final snapshot for session %q: snapshot worker unavailable or saturated", name)
+				terminalSnapshotErr = fmt.Errorf("retain final snapshot for session %q: snapshot worker unavailable, saturated, or timed out", name)
 				if reason != ports.ReasonServerShutdown {
 					return terminalSnapshotErr
 				}

M internal/usecase/daemon/snapshot.go +61 -5


@@ -385,6 +385,8 @@ func resumeSnapshotCoordinatorForNewIdentity(sess *session, quarantine snapshotC
385	385		// dirty and wake the scheduler.
386	386		sess.snapshotGeneration = 1
387	387		sess.snapshotCapturedGeneration = 0
	388	+	sess.snapshotPublishedGeneration = 0
	389	+	sess.snapshotForcedGeneration = 0
388	390		sess.snapshotNextEligibleAt = time.Time{}
389	391		sess.snapshotAttempted = false
390	392		sess.snapshotAttemptKind = snapshotAttemptRoutine

@@ -499,15 +501,22 @@ func (d Daemon) scheduleSnapshotWithFinalFallback(sess session, final bool) bo
499	501		}
500	502		sess.snapshotMu.Lock()
501	503		routineRateLimited := d.snapshotRepository != nil
502		-	// Forced attempts coalesce with a routine capture too. This keeps the
503		-	// coordinator at one queued plus one in-flight capture, while the dirty
504		-	// generation ensures a later checkpoint observes mutations made meanwhile.
505		-	if sess.snapshotQuarantined \|\| !sess.snapDirty.Load() \|\| sess.snapshotPending \|\|
506		-	(!final && routineRateLimited && !sess.snapshotNextEligibleAt.IsZero() && d.clock.Now().Before(sess.snapshotNextEligibleAt)) {
	504	+	if sess.snapshotQuarantined \|\| !sess.snapDirty.Load() {
507	505		available := !sess.snapshotQuarantined
508	506		sess.snapshotMu.Unlock()
509	507		return available
510	508		}
	509	+	if final && sess.snapshotGeneration > sess.snapshotForcedGeneration {
	510	+	// Do not let an older queued or in-flight routine publication satisfy a
	511	+	// terminal checkpoint. The intent is retained until a publication at
	512	+	// this generation (or newer) succeeds.
	513	+	sess.snapshotForcedGeneration = sess.snapshotGeneration
	514	+	}
	515	+	if sess.snapshotPending \|\|
	516	+	(!final && routineRateLimited && !sess.snapshotNextEligibleAt.IsZero() && d.clock.Now().Before(sess.snapshotNextEligibleAt)) {
	517	+	sess.snapshotMu.Unlock()
	518	+	return true
	519	+	}
511	520		if sess.snapshotPublicationContext == nil {
512	521		sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
513	522		}

@@ -549,6 +558,32 @@ func (d Daemon) scheduleSnapshotWithFinalFallback(sess session, final bool) bo
549	558		return false
550	559		}
551	560
	561	+	// waitForSnapshotGeneration waits only for the bounded terminal checkpoint
	562	+	// deadline. It observes successful publication, rather than dirty state, so a
	563	+	// concurrent later mutation cannot turn a completed forced checkpoint into a
	564	+	// false timeout.
	565	+	func (d Daemon) waitForSnapshotGeneration(sess session, generation uint64) bool {
	566	+	if d == nil \|\| sess == nil \|\| generation == 0 {
	567	+	return true
	568	+	}
	569	+	timer := d.clock.NewTimer(snapshotFinalFlushTimeout)
	570	+	defer timer.Stop()
	571	+	for {
	572	+	sess.snapshotMu.Lock()
	573	+	published := sess.snapshotPublishedGeneration >= generation
	574	+	changed := sess.snapshotChangeLocked()
	575	+	sess.snapshotMu.Unlock()
	576	+	if published {
	577	+	return true
	578	+	}
	579	+	select {
	580	+	case <-changed:
	581	+	case <-timer.C():
	582	+	return false
	583	+	}
	584	+	}
	585	+	}
	586	+
552	587		// enqueueSnapshotCapture serializes worker shutdown with the non-blocking
553	588		// queue send. The queue is deliberately never closed: producers can race
554	589		// shutdown without risking a send-on-closed panic.

@@ -737,6 +772,7 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
737	772		return
738	773		}
739	774		capture.finishOnce.Do(func() {
	775	+	shouldScheduleForcedSuccessor := false
740	776		capture.session.snapshotMu.Lock()
741	777		if capture.session.snapshotPendingCaptures > 0 {
742	778		capture.session.snapshotPendingCaptures--

@@ -758,12 +794,26 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
758	794		// attempt. A forced teardown checkpoint deliberately leaves it alone.
759	795		capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval)
760	796		}
	797	+	if succeeded {
	798	+	capture.session.snapshotPublishedGeneration = max(capture.session.snapshotPublishedGeneration, capture.generation)
	799	+	}
761	800		if succeeded && capture.session.snapshotGeneration == capture.generation {
762	801		capture.session.snapDirty.Store(false)
763	802		capture.session.snapshotFailureSig = ""
764	803		} else if !succeeded {
765	804		capture.session.snapDirty.Store(true)
766	805		}
	806	+	if succeeded && capture.session.snapshotForcedGeneration != 0 &&
	807	+	capture.generation >= capture.session.snapshotForcedGeneration {
	808	+	capture.session.snapshotForcedGeneration = 0
	809	+	}
	810	+	// A failed forced publication remains dirty for ordinary retry rather
	811	+	// than recursively creating forced jobs. A later forced request records
	812	+	// a newer intent and can create its one bounded successor.
	813	+	shouldScheduleForcedSuccessor = capture.session.snapshotForcedGeneration != 0 &&
	814	+	capture.generation < capture.session.snapshotForcedGeneration &&
	815	+	capture.session.snapshotQueuedCapture == nil && capture.session.snapshotInFlightCapture == nil &&
	816	+	(succeeded \|\| capture.attemptKind != snapshotAttemptForced)
767	817		capture.session.signalSnapshotChangedLocked()
768	818		wake := capture.session.snapshotWake
769	819		capture.session.snapshotMu.Unlock()

@@ -773,6 +823,12 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
773	823		default:
774	824		}
775	825		}
	826	+	if shouldScheduleForcedSuccessor {
	827	+	// This runs after publication completion and outside snapshotMu. It
	828	+	// bypasses routine eligibility while preserving the single queued and
	829	+	// single in-flight capture bound.
	830	+	_ = d.scheduleFinalSnapshot(capture.session)
	831	+	}
776	832		if succeeded {
777	833		d.clearSnapshotFailure()
778	834		}


@@ -385,6 +385,8 @@ func resumeSnapshotCoordinatorForNewIdentity(sess *session, quarantine snapshotC
385		// dirty and wake the scheduler.	385		// dirty and wake the scheduler.
386		sess.snapshotGeneration = 1	386		sess.snapshotGeneration = 1
387		sess.snapshotCapturedGeneration = 0	387		sess.snapshotCapturedGeneration = 0
			388	+	sess.snapshotPublishedGeneration = 0
			389	+	sess.snapshotForcedGeneration = 0
388		sess.snapshotNextEligibleAt = time.Time{}	390		sess.snapshotNextEligibleAt = time.Time{}
389		sess.snapshotAttempted = false	391		sess.snapshotAttempted = false
390		sess.snapshotAttemptKind = snapshotAttemptRoutine	392		sess.snapshotAttemptKind = snapshotAttemptRoutine

@@ -499,15 +501,22 @@ func (d Daemon) scheduleSnapshotWithFinalFallback(sess session, final bool) bo
499		}	501		}
500		sess.snapshotMu.Lock()	502		sess.snapshotMu.Lock()
501		routineRateLimited := d.snapshotRepository != nil	503		routineRateLimited := d.snapshotRepository != nil
502	-	// Forced attempts coalesce with a routine capture too. This keeps the	504	+	if sess.snapshotQuarantined \|\| !sess.snapDirty.Load() {
503	-	// coordinator at one queued plus one in-flight capture, while the dirty
504	-	// generation ensures a later checkpoint observes mutations made meanwhile.
505	-	if sess.snapshotQuarantined \|\| !sess.snapDirty.Load() \|\| sess.snapshotPending \|\|
506	-	(!final && routineRateLimited && !sess.snapshotNextEligibleAt.IsZero() && d.clock.Now().Before(sess.snapshotNextEligibleAt)) {
507		available := !sess.snapshotQuarantined	505		available := !sess.snapshotQuarantined
508		sess.snapshotMu.Unlock()	506		sess.snapshotMu.Unlock()
509		return available	507		return available
510		}	508		}
			509	+	if final && sess.snapshotGeneration > sess.snapshotForcedGeneration {
			510	+	// Do not let an older queued or in-flight routine publication satisfy a
			511	+	// terminal checkpoint. The intent is retained until a publication at
			512	+	// this generation (or newer) succeeds.
			513	+	sess.snapshotForcedGeneration = sess.snapshotGeneration
			514	+	}
			515	+	if sess.snapshotPending \|\|
			516	+	(!final && routineRateLimited && !sess.snapshotNextEligibleAt.IsZero() && d.clock.Now().Before(sess.snapshotNextEligibleAt)) {
			517	+	sess.snapshotMu.Unlock()
			518	+	return true
			519	+	}
511		if sess.snapshotPublicationContext == nil {	520		if sess.snapshotPublicationContext == nil {
512		sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())	521		sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
513		}	522		}

@@ -549,6 +558,32 @@ func (d Daemon) scheduleSnapshotWithFinalFallback(sess session, final bool) bo
549		return false	558		return false
550		}	559		}
551			560
			561	+	// waitForSnapshotGeneration waits only for the bounded terminal checkpoint
			562	+	// deadline. It observes successful publication, rather than dirty state, so a
			563	+	// concurrent later mutation cannot turn a completed forced checkpoint into a
			564	+	// false timeout.
			565	+	func (d Daemon) waitForSnapshotGeneration(sess session, generation uint64) bool {
			566	+	if d == nil \|\| sess == nil \|\| generation == 0 {
			567	+	return true
			568	+	}
			569	+	timer := d.clock.NewTimer(snapshotFinalFlushTimeout)
			570	+	defer timer.Stop()
			571	+	for {
			572	+	sess.snapshotMu.Lock()
			573	+	published := sess.snapshotPublishedGeneration >= generation
			574	+	changed := sess.snapshotChangeLocked()
			575	+	sess.snapshotMu.Unlock()
			576	+	if published {
			577	+	return true
			578	+	}
			579	+	select {
			580	+	case <-changed:
			581	+	case <-timer.C():
			582	+	return false
			583	+	}
			584	+	}
			585	+	}
			586	+
552		// enqueueSnapshotCapture serializes worker shutdown with the non-blocking	587		// enqueueSnapshotCapture serializes worker shutdown with the non-blocking
553		// queue send. The queue is deliberately never closed: producers can race	588		// queue send. The queue is deliberately never closed: producers can race
554		// shutdown without risking a send-on-closed panic.	589		// shutdown without risking a send-on-closed panic.

@@ -737,6 +772,7 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
737		return	772		return
738		}	773		}
739		capture.finishOnce.Do(func() {	774		capture.finishOnce.Do(func() {
			775	+	shouldScheduleForcedSuccessor := false
740		capture.session.snapshotMu.Lock()	776		capture.session.snapshotMu.Lock()
741		if capture.session.snapshotPendingCaptures > 0 {	777		if capture.session.snapshotPendingCaptures > 0 {
742		capture.session.snapshotPendingCaptures--	778		capture.session.snapshotPendingCaptures--

@@ -758,12 +794,26 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
758		// attempt. A forced teardown checkpoint deliberately leaves it alone.	794		// attempt. A forced teardown checkpoint deliberately leaves it alone.
759		capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval)	795		capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval)
760		}	796		}
			797	+	if succeeded {
			798	+	capture.session.snapshotPublishedGeneration = max(capture.session.snapshotPublishedGeneration, capture.generation)
			799	+	}
761		if succeeded && capture.session.snapshotGeneration == capture.generation {	800		if succeeded && capture.session.snapshotGeneration == capture.generation {
762		capture.session.snapDirty.Store(false)	801		capture.session.snapDirty.Store(false)
763		capture.session.snapshotFailureSig = ""	802		capture.session.snapshotFailureSig = ""
764		} else if !succeeded {	803		} else if !succeeded {
765		capture.session.snapDirty.Store(true)	804		capture.session.snapDirty.Store(true)
766		}	805		}
			806	+	if succeeded && capture.session.snapshotForcedGeneration != 0 &&
			807	+	capture.generation >= capture.session.snapshotForcedGeneration {
			808	+	capture.session.snapshotForcedGeneration = 0
			809	+	}
			810	+	// A failed forced publication remains dirty for ordinary retry rather
			811	+	// than recursively creating forced jobs. A later forced request records
			812	+	// a newer intent and can create its one bounded successor.
			813	+	shouldScheduleForcedSuccessor = capture.session.snapshotForcedGeneration != 0 &&
			814	+	capture.generation < capture.session.snapshotForcedGeneration &&
			815	+	capture.session.snapshotQueuedCapture == nil && capture.session.snapshotInFlightCapture == nil &&
			816	+	(succeeded \|\| capture.attemptKind != snapshotAttemptForced)
767		capture.session.signalSnapshotChangedLocked()	817		capture.session.signalSnapshotChangedLocked()
768		wake := capture.session.snapshotWake	818		wake := capture.session.snapshotWake
769		capture.session.snapshotMu.Unlock()	819		capture.session.snapshotMu.Unlock()

@@ -773,6 +823,12 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
773		default:	823		default:
774		}	824		}
775		}	825		}
			826	+	if shouldScheduleForcedSuccessor {
			827	+	// This runs after publication completion and outside snapshotMu. It
			828	+	// bypasses routine eligibility while preserving the single queued and
			829	+	// single in-flight capture bound.
			830	+	_ = d.scheduleFinalSnapshot(capture.session)
			831	+	}
776		if succeeded {	832		if succeeded {
777		d.clearSnapshotFailure()	833		d.clearSnapshotFailure()
778		}	834		}

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


@@ -195,6 +195,183 @@ func TestRenameRollbackAfterOldDeleteLeavesOldCoordinatorStopped(t *testing.T) {
195	195		}
196	196		}
197	197
	198	+	func TestForcedSnapshotSchedulesSuccessorAfterRoutineInFlight(t *testing.T) {
	199	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	200	+	repository := portsmocks.NewMockSnapshotRepository(t)
	201	+	WithSnapshotRepository(repository, nil)(d)
	202	+	startSnapshotEncodeWorker(t, d)
	203	+
	204	+	started := make(chan struct{})
	205	+	release := make(chan struct{})
	206	+	publications := make(chan ports.SnapshotPublication, 2)
	207	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
	208	+	publications <- publication
	209	+	if publication.Generation == 1 {
	210	+	close(started)
	211	+	<-release
	212	+	}
	213	+	return nil
	214	+	}).Twice()
	215	+
	216	+	sess := newSnapshotTestSession(t, "work", false, "/work")
	217	+	markSnapshotDirty(sess)
	218	+	require.True(t, d.scheduleSnapshot(sess))
	219	+	select {
	220	+	case <-started:
	221	+	case <-time.After(testWaitTimeout):
	222	+	t.Fatal("routine publication did not start")
	223	+	}
	224	+
	225	+	// The terminal state changed after the routine capture, so its publication
	226	+	// cannot satisfy the forced checkpoint.
	227	+	sess.tabs[0].panes["pane-1"].screen.Write([]byte(" terminal"))
	228	+	markSnapshotDirty(sess)
	229	+	require.True(t, d.scheduleFinalSnapshot(sess))
	230	+	close(release)
	231	+
	232	+	awaitSnapshotClean(t, sess)
	233	+	require.Equal(t, uint64(1), (<-publications).Generation)
	234	+	require.Equal(t, uint64(2), (<-publications).Generation)
	235	+	}
	236	+
	237	+	func TestForcedSnapshotSchedulesSuccessorForRoutineCaptureQueuedBehindWorker(t *testing.T) {
	238	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	239	+	repository := portsmocks.NewMockSnapshotRepository(t)
	240	+	WithSnapshotRepository(repository, nil)(d)
	241	+	startSnapshotEncodeWorker(t, d)
	242	+
	243	+	blockerStarted := make(chan struct{})
	244	+	releaseBlocker := make(chan struct{})
	245	+	workPublications := make(chan ports.SnapshotPublication, 2)
	246	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
	247	+	if publication.Name == "blocker" {
	248	+	close(blockerStarted)
	249	+	<-releaseBlocker
	250	+	return nil
	251	+	}
	252	+	workPublications <- publication
	253	+	return nil
	254	+	}).Times(3)
	255	+
	256	+	blocker := newSnapshotTestSession(t, "blocker", false, "/work")
	257	+	markSnapshotDirty(blocker)
	258	+	require.True(t, d.scheduleSnapshot(blocker))
	259	+	select {
	260	+	case <-blockerStarted:
	261	+	case <-time.After(testWaitTimeout):
	262	+	t.Fatal("blocking publication did not start")
	263	+	}
	264	+
	265	+	sess := newSnapshotTestSession(t, "work", false, "/work")
	266	+	markSnapshotDirty(sess)
	267	+	require.True(t, d.scheduleSnapshot(sess), "routine capture should occupy the bounded queue")
	268	+	sess.tabs[0].panes["pane-1"].screen.Write([]byte(" terminal"))
	269	+	markSnapshotDirty(sess)
	270	+	require.True(t, d.scheduleFinalSnapshot(sess))
	271	+	close(releaseBlocker)
	272	+
	273	+	awaitSnapshotClean(t, sess)
	274	+	require.Equal(t, uint64(1), (<-workPublications).Generation)
	275	+	require.Equal(t, uint64(2), (<-workPublications).Generation)
	276	+	}
	277	+
	278	+	func TestMultipleForcedSnapshotsCoalesceToOneSuccessor(t *testing.T) {
	279	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	280	+	repository := portsmocks.NewMockSnapshotRepository(t)
	281	+	WithSnapshotRepository(repository, nil)(d)
	282	+	startSnapshotEncodeWorker(t, d)
	283	+
	284	+	started := make(chan struct{})
	285	+	release := make(chan struct{})
	286	+	publications := make(chan ports.SnapshotPublication, 2)
	287	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
	288	+	publications <- publication
	289	+	if publication.Generation == 1 {
	290	+	close(started)
	291	+	<-release
	292	+	}
	293	+	return nil
	294	+	}).Twice()
	295	+
	296	+	sess := newSnapshotTestSession(t, "work", false, "/work")
	297	+	markSnapshotDirty(sess)
	298	+	require.True(t, d.scheduleSnapshot(sess))
	299	+	select {
	300	+	case <-started:
	301	+	case <-time.After(testWaitTimeout):
	302	+	t.Fatal("routine publication did not start")
	303	+	}
	304	+	markSnapshotDirty(sess)
	305	+	require.True(t, d.scheduleFinalSnapshot(sess))
	306	+	markSnapshotDirty(sess)
	307	+	require.True(t, d.scheduleFinalSnapshot(sess))
	308	+	close(release)
	309	+
	310	+	awaitSnapshotClean(t, sess)
	311	+	require.Equal(t, uint64(1), (<-publications).Generation)
	312	+	require.Equal(t, uint64(3), (<-publications).Generation)
	313	+	sess.snapshotMu.Lock()
	314	+	require.False(t, sess.snapshotPending)
	315	+	require.Nil(t, sess.snapshotQueuedCapture)
	316	+	require.Nil(t, sess.snapshotInFlightCapture)
	317	+	sess.snapshotMu.Unlock()
	318	+	}
	319	+
	320	+	func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.T) {
	321	+	clock := newSnapshotDeadlineClock()
	322	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
	323	+	repository := portsmocks.NewMockSnapshotRepository(t)
	324	+	WithSnapshotRepository(repository, nil)(d)
	325	+	notices := portsmocks.NewMockNoticeStore(t)
	326	+	WithNoticeStore(notices)(d)
	327	+	startSnapshotEncodeWorker(t, d)
	328	+
	329	+	blockerStarted := make(chan struct{})
	330	+	releaseBlocker := make(chan struct{})
	331	+	defer close(releaseBlocker)
	332	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
	333	+	if publication.Name == "blocker" {
	334	+	close(blockerStarted)
	335	+	<-releaseBlocker
	336	+	}
	337	+	return nil
	338	+	}).Maybe()
	339	+	notices.EXPECT().Append(mock.MatchedBy(func(notification domain.Notification) bool {
	340	+	return notification.Code == domain.NoticeSnapshotWrite && notification.Severity == domain.NoticeError
	341	+	})).Return(nil).Once()
	342	+
	343	+	blocker := newSnapshotTestSession(t, "blocker", false, "/work")
	344	+	markSnapshotDirty(blocker)
	345	+	require.True(t, d.scheduleSnapshot(blocker))
	346	+	select {
	347	+	case <-blockerStarted:
	348	+	case <-time.After(testWaitTimeout):
	349	+	t.Fatal("blocking publication did not start")
	350	+	}
	351	+
	352	+	sess := newSnapshotTestSession(t, "work", false, "/work")
	353	+	pty := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
	354	+	pty.EXPECT().Close().Return(nil).Maybe()
	355	+	d.sessions[sess.id] = sess
	356	+	markSnapshotDirty(sess)
	357	+	require.True(t, d.scheduleSnapshot(sess), "routine capture should occupy the bounded queue")
	358	+
	359	+	killed := make(chan error, 1)
	360	+	go func() { killed <- d.killSession(sess, ports.ReasonServerShutdown, false) }()
	361	+	timer := clock.nextTimer(t)
	362	+	timer.fire()
	363	+	require.Error(t, <-killed)
	364	+
	365	+	sess.snapshotMu.Lock()
	366	+	require.True(t, sess.snapDirty.Load())
	367	+	require.Greater(t, sess.snapshotForcedGeneration, sess.snapshotPublishedGeneration)
	368	+	require.NotNil(t, sess.snapshotQueuedCapture, "the one routine capture remains retryable")
	369	+	sess.snapshotMu.Unlock()
	370	+	d.snapshotWorkerMu.Lock()
	371	+	require.LessOrEqual(t, len(d.snapshotJobs), snapshotQueueCapacity)
	372	+	d.snapshotWorkerMu.Unlock()
	373	+	}
	374	+
198	375		func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {
199	376		clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}
200	377		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)

@@ -241,3 +418,36 @@ func (c *snapshotSchedulerClock) Now() time.Time {
241	418		}
242	419
243	420		func (*snapshotSchedulerClock) NewTimer(time.Duration) ports.Timer { return stubTimer{} }
	421	+
	422	+	type snapshotDeadlineClock struct {
	423	+	timers chan *snapshotDeadlineTimer
	424	+	}
	425	+
	426	+	func newSnapshotDeadlineClock() *snapshotDeadlineClock {
	427	+	return &snapshotDeadlineClock{timers: make(chan *snapshotDeadlineTimer, 4)}
	428	+	}
	429	+
	430	+	func (*snapshotDeadlineClock) Now() time.Time { return time.Unix(100, 0) }
	431	+	func (c *snapshotDeadlineClock) NewTimer(time.Duration) ports.Timer {
	432	+	timer := &snapshotDeadlineTimer{ch: make(chan time.Time, 1)}
	433	+	c.timers <- timer
	434	+	return timer
	435	+	}
	436	+
	437	+	func (c snapshotDeadlineClock) nextTimer(t testing.T) *snapshotDeadlineTimer {
	438	+	t.Helper()
	439	+	select {
	440	+	case timer := <-c.timers:
	441	+	return timer
	442	+	case <-time.After(testWaitTimeout):
	443	+	t.Fatal("snapshot deadline timer was not created")
	444	+	return nil
	445	+	}
	446	+	}
	447	+
	448	+	type snapshotDeadlineTimer struct{ ch chan time.Time }
	449	+
	450	+	func (t *snapshotDeadlineTimer) C() <-chan time.Time { return t.ch }
	451	+	func (*snapshotDeadlineTimer) Reset(time.Duration) bool { return true }
	452	+	func (*snapshotDeadlineTimer) Stop() bool { return true }
	453	+	func (t *snapshotDeadlineTimer) fire() { t.ch <- time.Unix(101, 0) }


@@ -195,6 +195,183 @@ func TestRenameRollbackAfterOldDeleteLeavesOldCoordinatorStopped(t *testing.T) {
195	}	195		}
196	}	196		}
197		197
		198	+	func TestForcedSnapshotSchedulesSuccessorAfterRoutineInFlight(t *testing.T) {
		199	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
		200	+	repository := portsmocks.NewMockSnapshotRepository(t)
		201	+	WithSnapshotRepository(repository, nil)(d)
		202	+	startSnapshotEncodeWorker(t, d)
		203	+
		204	+	started := make(chan struct{})
		205	+	release := make(chan struct{})
		206	+	publications := make(chan ports.SnapshotPublication, 2)
		207	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
		208	+	publications <- publication
		209	+	if publication.Generation == 1 {
		210	+	close(started)
		211	+	<-release
		212	+	}
		213	+	return nil
		214	+	}).Twice()
		215	+
		216	+	sess := newSnapshotTestSession(t, "work", false, "/work")
		217	+	markSnapshotDirty(sess)
		218	+	require.True(t, d.scheduleSnapshot(sess))
		219	+	select {
		220	+	case <-started:
		221	+	case <-time.After(testWaitTimeout):
		222	+	t.Fatal("routine publication did not start")
		223	+	}
		224	+
		225	+	// The terminal state changed after the routine capture, so its publication
		226	+	// cannot satisfy the forced checkpoint.
		227	+	sess.tabs[0].panes["pane-1"].screen.Write([]byte(" terminal"))
		228	+	markSnapshotDirty(sess)
		229	+	require.True(t, d.scheduleFinalSnapshot(sess))
		230	+	close(release)
		231	+
		232	+	awaitSnapshotClean(t, sess)
		233	+	require.Equal(t, uint64(1), (<-publications).Generation)
		234	+	require.Equal(t, uint64(2), (<-publications).Generation)
		235	+	}
		236	+
		237	+	func TestForcedSnapshotSchedulesSuccessorForRoutineCaptureQueuedBehindWorker(t *testing.T) {
		238	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
		239	+	repository := portsmocks.NewMockSnapshotRepository(t)
		240	+	WithSnapshotRepository(repository, nil)(d)
		241	+	startSnapshotEncodeWorker(t, d)
		242	+
		243	+	blockerStarted := make(chan struct{})
		244	+	releaseBlocker := make(chan struct{})
		245	+	workPublications := make(chan ports.SnapshotPublication, 2)
		246	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
		247	+	if publication.Name == "blocker" {
		248	+	close(blockerStarted)
		249	+	<-releaseBlocker
		250	+	return nil
		251	+	}
		252	+	workPublications <- publication
		253	+	return nil
		254	+	}).Times(3)
		255	+
		256	+	blocker := newSnapshotTestSession(t, "blocker", false, "/work")
		257	+	markSnapshotDirty(blocker)
		258	+	require.True(t, d.scheduleSnapshot(blocker))
		259	+	select {
		260	+	case <-blockerStarted:
		261	+	case <-time.After(testWaitTimeout):
		262	+	t.Fatal("blocking publication did not start")
		263	+	}
		264	+
		265	+	sess := newSnapshotTestSession(t, "work", false, "/work")
		266	+	markSnapshotDirty(sess)
		267	+	require.True(t, d.scheduleSnapshot(sess), "routine capture should occupy the bounded queue")
		268	+	sess.tabs[0].panes["pane-1"].screen.Write([]byte(" terminal"))
		269	+	markSnapshotDirty(sess)
		270	+	require.True(t, d.scheduleFinalSnapshot(sess))
		271	+	close(releaseBlocker)
		272	+
		273	+	awaitSnapshotClean(t, sess)
		274	+	require.Equal(t, uint64(1), (<-workPublications).Generation)
		275	+	require.Equal(t, uint64(2), (<-workPublications).Generation)
		276	+	}
		277	+
		278	+	func TestMultipleForcedSnapshotsCoalesceToOneSuccessor(t *testing.T) {
		279	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
		280	+	repository := portsmocks.NewMockSnapshotRepository(t)
		281	+	WithSnapshotRepository(repository, nil)(d)
		282	+	startSnapshotEncodeWorker(t, d)
		283	+
		284	+	started := make(chan struct{})
		285	+	release := make(chan struct{})
		286	+	publications := make(chan ports.SnapshotPublication, 2)
		287	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
		288	+	publications <- publication
		289	+	if publication.Generation == 1 {
		290	+	close(started)
		291	+	<-release
		292	+	}
		293	+	return nil
		294	+	}).Twice()
		295	+
		296	+	sess := newSnapshotTestSession(t, "work", false, "/work")
		297	+	markSnapshotDirty(sess)
		298	+	require.True(t, d.scheduleSnapshot(sess))
		299	+	select {
		300	+	case <-started:
		301	+	case <-time.After(testWaitTimeout):
		302	+	t.Fatal("routine publication did not start")
		303	+	}
		304	+	markSnapshotDirty(sess)
		305	+	require.True(t, d.scheduleFinalSnapshot(sess))
		306	+	markSnapshotDirty(sess)
		307	+	require.True(t, d.scheduleFinalSnapshot(sess))
		308	+	close(release)
		309	+
		310	+	awaitSnapshotClean(t, sess)
		311	+	require.Equal(t, uint64(1), (<-publications).Generation)
		312	+	require.Equal(t, uint64(3), (<-publications).Generation)
		313	+	sess.snapshotMu.Lock()
		314	+	require.False(t, sess.snapshotPending)
		315	+	require.Nil(t, sess.snapshotQueuedCapture)
		316	+	require.Nil(t, sess.snapshotInFlightCapture)
		317	+	sess.snapshotMu.Unlock()
		318	+	}
		319	+
		320	+	func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.T) {
		321	+	clock := newSnapshotDeadlineClock()
		322	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
		323	+	repository := portsmocks.NewMockSnapshotRepository(t)
		324	+	WithSnapshotRepository(repository, nil)(d)
		325	+	notices := portsmocks.NewMockNoticeStore(t)
		326	+	WithNoticeStore(notices)(d)
		327	+	startSnapshotEncodeWorker(t, d)
		328	+
		329	+	blockerStarted := make(chan struct{})
		330	+	releaseBlocker := make(chan struct{})
		331	+	defer close(releaseBlocker)
		332	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
		333	+	if publication.Name == "blocker" {
		334	+	close(blockerStarted)
		335	+	<-releaseBlocker
		336	+	}
		337	+	return nil
		338	+	}).Maybe()
		339	+	notices.EXPECT().Append(mock.MatchedBy(func(notification domain.Notification) bool {
		340	+	return notification.Code == domain.NoticeSnapshotWrite && notification.Severity == domain.NoticeError
		341	+	})).Return(nil).Once()
		342	+
		343	+	blocker := newSnapshotTestSession(t, "blocker", false, "/work")
		344	+	markSnapshotDirty(blocker)
		345	+	require.True(t, d.scheduleSnapshot(blocker))
		346	+	select {
		347	+	case <-blockerStarted:
		348	+	case <-time.After(testWaitTimeout):
		349	+	t.Fatal("blocking publication did not start")
		350	+	}
		351	+
		352	+	sess := newSnapshotTestSession(t, "work", false, "/work")
		353	+	pty := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
		354	+	pty.EXPECT().Close().Return(nil).Maybe()
		355	+	d.sessions[sess.id] = sess
		356	+	markSnapshotDirty(sess)
		357	+	require.True(t, d.scheduleSnapshot(sess), "routine capture should occupy the bounded queue")
		358	+
		359	+	killed := make(chan error, 1)
		360	+	go func() { killed <- d.killSession(sess, ports.ReasonServerShutdown, false) }()
		361	+	timer := clock.nextTimer(t)
		362	+	timer.fire()
		363	+	require.Error(t, <-killed)
		364	+
		365	+	sess.snapshotMu.Lock()
		366	+	require.True(t, sess.snapDirty.Load())
		367	+	require.Greater(t, sess.snapshotForcedGeneration, sess.snapshotPublishedGeneration)
		368	+	require.NotNil(t, sess.snapshotQueuedCapture, "the one routine capture remains retryable")
		369	+	sess.snapshotMu.Unlock()
		370	+	d.snapshotWorkerMu.Lock()
		371	+	require.LessOrEqual(t, len(d.snapshotJobs), snapshotQueueCapacity)
		372	+	d.snapshotWorkerMu.Unlock()
		373	+	}
		374	+
198	func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {	375		func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {
199	clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}	376		clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}
200	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)	377		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)

@@ -241,3 +418,36 @@ func (c *snapshotSchedulerClock) Now() time.Time {
241	}	418		}
242		419
243	func (*snapshotSchedulerClock) NewTimer(time.Duration) ports.Timer { return stubTimer{} }	420		func (*snapshotSchedulerClock) NewTimer(time.Duration) ports.Timer { return stubTimer{} }
		421	+
		422	+	type snapshotDeadlineClock struct {
		423	+	timers chan *snapshotDeadlineTimer
		424	+	}
		425	+
		426	+	func newSnapshotDeadlineClock() *snapshotDeadlineClock {
		427	+	return &snapshotDeadlineClock{timers: make(chan *snapshotDeadlineTimer, 4)}
		428	+	}
		429	+
		430	+	func (*snapshotDeadlineClock) Now() time.Time { return time.Unix(100, 0) }
		431	+	func (c *snapshotDeadlineClock) NewTimer(time.Duration) ports.Timer {
		432	+	timer := &snapshotDeadlineTimer{ch: make(chan time.Time, 1)}
		433	+	c.timers <- timer
		434	+	return timer
		435	+	}
		436	+
		437	+	func (c snapshotDeadlineClock) nextTimer(t testing.T) *snapshotDeadlineTimer {
		438	+	t.Helper()
		439	+	select {
		440	+	case timer := <-c.timers:
		441	+	return timer
		442	+	case <-time.After(testWaitTimeout):
		443	+	t.Fatal("snapshot deadline timer was not created")
		444	+	return nil
		445	+	}
		446	+	}
		447	+
		448	+	type snapshotDeadlineTimer struct{ ch chan time.Time }
		449	+
		450	+	func (t *snapshotDeadlineTimer) C() <-chan time.Time { return t.ch }
		451	+	func (*snapshotDeadlineTimer) Reset(time.Duration) bool { return true }
		452	+	func (*snapshotDeadlineTimer) Stop() bool { return true }
		453	+	func (t *snapshotDeadlineTimer) fire() { t.ch <- time.Unix(101, 0) }