ae5758e fix(daemon): retain timed-out checkpoint state · vev

M internal/usecase/daemon/session.go +11 -1


@@ -908,6 +908,7 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
908	908		d.refreshSessionCwd(sess)
909	909		}
910	910		var terminalSnapshotErr error
	911	+	retainSnapshotRetry := false
911	912		if d.snapsEnabled && !isEphemeral {
912	913		if purge {
913	914		if d.snapshotRepository != nil {

@@ -928,6 +929,10 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
928	929		if reason != ports.ReasonServerShutdown {
929	930		return terminalSnapshotErr
930	931		}
	932	+	// Preserve any already-queued routine capture until the blocked
	933	+	// worker can safely discard it. This retains the dirty generation
	934	+	// and forced retry intent without adding another queue entry.
	935	+	retainSnapshotRetry = true
931	936		d.persistShutdownSnapshotFailure(name, terminalSnapshotErr)
932	937		}
933	938		}

@@ -938,7 +943,12 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
938	943		// budget expires, the worker is cancelled by its one final join instead;
939	944		// keep the cache reachable because an uncooperative writer may still be
940	945		// encoding immutable state.
941		-	quarantineDone := quarantineSnapshotCoordinator(sess)
	946	+	var quarantineDone <-chan struct{}
	947	+	if retainSnapshotRetry {
	948	+	quarantineDone = quarantineSnapshotCoordinatorRetainingQueuedCapture(sess)
	949	+	} else {
	950	+	quarantineDone = quarantineSnapshotCoordinator(sess)
	951	+	}
942	952		joined := true
943	953		if deadline != nil {
944	954		select {

 		d.refreshSessionCwd(sess)
 	}
 	var terminalSnapshotErr error
+	retainSnapshotRetry := false
 	if d.snapsEnabled && !isEphemeral {
 		if purge {
 			if d.snapshotRepository != nil {
 				if reason != ports.ReasonServerShutdown {
 					return terminalSnapshotErr
 				}
+				// Preserve any already-queued routine capture until the blocked
+				// worker can safely discard it. This retains the dirty generation
+				// and forced retry intent without adding another queue entry.
+				retainSnapshotRetry = true
 				d.persistShutdownSnapshotFailure(name, terminalSnapshotErr)
 			}
 		}
 		// budget expires, the worker is cancelled by its one final join instead;
 		// keep the cache reachable because an uncooperative writer may still be
 		// encoding immutable state.
-		quarantineDone := quarantineSnapshotCoordinator(sess)
+		var quarantineDone <-chan struct{}
+		if retainSnapshotRetry {
+			quarantineDone = quarantineSnapshotCoordinatorRetainingQueuedCapture(sess)
+		} else {
+			quarantineDone = quarantineSnapshotCoordinator(sess)
+		}
 		joined := true
 		if deadline != nil {
 			select {

M internal/usecase/daemon/snapshot.go +17 -4


@@ -329,7 +329,20 @@ func quarantineSnapshotCoordinator(sess *session) <-chan struct{} {
329	329		return quarantineSnapshotCoordinatorWithEpoch(sess).done
330	330		}
331	331
	332	+	// quarantineSnapshotCoordinatorRetainingQueuedCapture is used after a forced
	333	+	// shutdown checkpoint times out. Its queued capture is logical retry state, not
	334	+	// a repository operation: startSnapshotPublication rejects it while
	335	+	// quarantined, and the worker eventually discards it after its blocked call
	336	+	// returns.
	337	+	func quarantineSnapshotCoordinatorRetainingQueuedCapture(sess *session) <-chan struct{} {
	338	+	return quarantineSnapshotCoordinatorWithOptions(sess, true).done
	339	+	}
	340	+
332	341		func quarantineSnapshotCoordinatorWithEpoch(sess *session) snapshotCoordinatorQuarantine {
	342	+	return quarantineSnapshotCoordinatorWithOptions(sess, false)
	343	+	}
	344	+
	345	+	func quarantineSnapshotCoordinatorWithOptions(sess *session, retainQueuedCapture bool) snapshotCoordinatorQuarantine {
333	346		if sess == nil {
334	347		done := make(chan struct{})
335	348		close(done)

@@ -343,10 +356,10 @@ func quarantineSnapshotCoordinatorWithEpoch(sess *session) snapshotCoordinatorQu
343	356		if sess.snapshotPublicationCancel != nil {
344	357		sess.snapshotPublicationCancel()
345	358		}
346		-	// Remove a not-yet-started capture from the coordinator immediately. The
347		-	// global queue may still contain it, but startSnapshotPublication compares
348		-	// identity and will discard it even if this session is later renamed.
349		-	if sess.snapshotQueuedCapture != nil {
	359	+	if sess.snapshotQueuedCapture != nil && !retainQueuedCapture {
	360	+	// A rename or purge must discard a capture under the old identity. The
	361	+	// global queue may still contain it, but startSnapshotPublication
	362	+	// compares identity and will discard it even if this session is renamed.
350	363		sess.snapshotQueuedCapture.coordinatorDiscarded = true
351	364		sess.snapshotQueuedCapture = nil
352	365		if sess.snapshotPendingCaptures > 0 {

 	return quarantineSnapshotCoordinatorWithEpoch(sess).done
 }
+// quarantineSnapshotCoordinatorRetainingQueuedCapture is used after a forced
+// shutdown checkpoint times out. Its queued capture is logical retry state, not
+// a repository operation: startSnapshotPublication rejects it while
+// quarantined, and the worker eventually discards it after its blocked call
+// returns.
+func quarantineSnapshotCoordinatorRetainingQueuedCapture(sess *session) <-chan struct{} {
+	return quarantineSnapshotCoordinatorWithOptions(sess, true).done
+}
 func quarantineSnapshotCoordinatorWithEpoch(sess *session) snapshotCoordinatorQuarantine {
+	return quarantineSnapshotCoordinatorWithOptions(sess, false)
+}
+func quarantineSnapshotCoordinatorWithOptions(sess *session, retainQueuedCapture bool) snapshotCoordinatorQuarantine {
 	if sess == nil {
 		done := make(chan struct{})
 		close(done)
 	if sess.snapshotPublicationCancel != nil {
 		sess.snapshotPublicationCancel()
 	}
-	// Remove a not-yet-started capture from the coordinator immediately. The
+	if sess.snapshotQueuedCapture != nil && !retainQueuedCapture {
-	// global queue may still contain it, but startSnapshotPublication compares
+		// A rename or purge must discard a capture under the old identity. The
-	// identity and will discard it even if this session is later renamed.
+		// global queue may still contain it, but startSnapshotPublication
-	if sess.snapshotQueuedCapture != nil {
+		// compares identity and will discard it even if this session is renamed.
 		sess.snapshotQueuedCapture.coordinatorDiscarded = true
 		sess.snapshotQueuedCapture = nil
 		if sess.snapshotPendingCaptures > 0 {

M internal/usecase/daemon/snapshot_scheduler_test.go +24 -6


@@ -329,7 +329,11 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
329	329
330	330		blockerStarted := make(chan struct{})
331	331		releaseBlocker := make(chan struct{})
332		-	defer close(releaseBlocker)
	332	+	var releaseBlockerOnce sync.Once
	333	+	release := func() { releaseBlockerOnce.Do(func() { close(releaseBlocker) }) }
	334	+	// Register this after newTestDaemon's worker cleanup so the uncooperative
	335	+	// repository call is always released before cleanup asks the worker to join.
	336	+	t.Cleanup(release)
333	337		repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
334	338		if publication.Name == "blocker" {
335	339		close(blockerStarted)

@@ -363,14 +367,28 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
363	367		timer.fire()
364	368		require.Error(t, <-killed)
365	369
366		-	sess.snapshotMu.Lock()
367		-	require.True(t, sess.snapDirty.Load())
368		-	require.Greater(t, sess.snapshotForcedGeneration, sess.snapshotPublishedGeneration)
369		-	require.NotNil(t, sess.snapshotQueuedCapture, "the one routine capture remains retryable")
370		-	sess.snapshotMu.Unlock()
	370	+	func() {
	371	+	sess.snapshotMu.Lock()
	372	+	defer sess.snapshotMu.Unlock()
	373	+	require.True(t, sess.snapDirty.Load())
	374	+	require.Greater(t, sess.snapshotForcedGeneration, sess.snapshotPublishedGeneration)
	375	+	require.True(t, sess.snapshotPending)
	376	+	require.Equal(t, uint(1), sess.snapshotPendingCaptures)
	377	+	require.NotNil(t, sess.snapshotQueuedCapture, "the one routine capture remains retryable")
	378	+	require.Nil(t, sess.snapshotInFlightCapture)
	379	+	}()
371	380		d.snapshotWorkerMu.Lock()
372	381		require.LessOrEqual(t, len(d.snapshotJobs), snapshotQueueCapacity)
373	382		d.snapshotWorkerMu.Unlock()
	383	+
	384	+	// The retained capture is not published after quarantine, but releasing the
	385	+	// unrelated blocked call lets the worker discard it and lets cleanup join.
	386	+	release()
	387	+	require.Eventually(t, func() bool {
	388	+	sess.snapshotMu.Lock()
	389	+	defer sess.snapshotMu.Unlock()
	390	+	return !sess.snapshotPending && sess.snapshotQueuedCapture == nil
	391	+	}, testWaitTimeout, time.Millisecond)
374	392		}
375	393
376	394		func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {

 	blockerStarted := make(chan struct{})
 	releaseBlocker := make(chan struct{})
-	defer close(releaseBlocker)
+	var releaseBlockerOnce sync.Once
+	release := func() { releaseBlockerOnce.Do(func() { close(releaseBlocker) }) }
+	// Register this after newTestDaemon's worker cleanup so the uncooperative
+	// repository call is always released before cleanup asks the worker to join.
+	t.Cleanup(release)
 	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
 		if publication.Name == "blocker" {
 			close(blockerStarted)
 	timer.fire()
 	require.Error(t, <-killed)
-	sess.snapshotMu.Lock()
+	func() {
-	require.True(t, sess.snapDirty.Load())
+		sess.snapshotMu.Lock()
-	require.Greater(t, sess.snapshotForcedGeneration, sess.snapshotPublishedGeneration)
+		defer sess.snapshotMu.Unlock()
-	require.NotNil(t, sess.snapshotQueuedCapture, "the one routine capture remains retryable")
+		require.True(t, sess.snapDirty.Load())
-	sess.snapshotMu.Unlock()
+		require.Greater(t, sess.snapshotForcedGeneration, sess.snapshotPublishedGeneration)
+		require.True(t, sess.snapshotPending)
+		require.Equal(t, uint(1), sess.snapshotPendingCaptures)
+		require.NotNil(t, sess.snapshotQueuedCapture, "the one routine capture remains retryable")
+		require.Nil(t, sess.snapshotInFlightCapture)
+	}()
 	d.snapshotWorkerMu.Lock()
 	require.LessOrEqual(t, len(d.snapshotJobs), snapshotQueueCapacity)
 	d.snapshotWorkerMu.Unlock()
+	// The retained capture is not published after quarantine, but releasing the
+	// unrelated blocked call lets the worker discard it and lets cleanup join.
+	release()
+	require.Eventually(t, func() bool {
+		sess.snapshotMu.Lock()
+		defer sess.snapshotMu.Unlock()
+		return !sess.snapshotPending && sess.snapshotQueuedCapture == nil
+	}, testWaitTimeout, time.Millisecond)
 }
 func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {