fc94e20 refactor(daemon): remove snapshot test-only state · vev

M internal/adapters/snapshot/legacy.go +28 -8


@@ -23,9 +23,21 @@ type legacyDeleteMarker struct {
23	23		name string
24	24		}
25	25
	26	+	type legacyDirectory interface {
	27	+	ReadDir(int) ([]os.DirEntry, error)
	28	+	Close() error
	29	+	}
	30	+
	31	+	func (r *Repository) openLegacyDirectory(path string) (legacyDirectory, error) {
	32	+	if r.hooks.openLegacyDirectory != nil {
	33	+	return r.hooks.openLegacyDirectory(path)
	34	+	}
	35	+	return os.Open(path)
	36	+	}
	37	+
26	38		// LoadLegacy reads only pre-incremental root .snap files. It is deliberately
27	39		// isolated from repository reads so incremental directories are never traversed.
28		-	func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) {
	40	+	func (r *Repository) LoadLegacy(ctx context.Context) (out []ports.LegacySnapshot, err error) {
29	41		if err := ctx.Err(); err != nil {
30	42		return nil, err
31	43		}

@@ -34,16 +46,20 @@ func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, er
34	46		if err := r.cleanupPendingLegacyDeletes(ctx); err != nil {
35	47		return nil, err
36	48		}
37		-	f, err := os.Open(r.dir)
	49	+	f, err := r.openLegacyDirectory(r.dir)
38	50		if errors.Is(err, os.ErrNotExist) {
39	51		return nil, nil
40	52		}
41	53		if err != nil {
42	54		return nil, fmt.Errorf("read legacy snapshot directory: %w", safeFilesystemError(err))
43	55		}
44		-	defer f.Close()
	56	+	defer func() {
	57	+	if closeErr := f.Close(); closeErr != nil {
	58	+	err = errors.Join(err, fmt.Errorf("close legacy snapshot directory: %w", safeFilesystemError(closeErr)))
	59	+	}
	60	+	}()
45	61
46		-	out := make([]ports.LegacySnapshot, 0, maxLegacySnapshotFiles)
	62	+	out = make([]ports.LegacySnapshot, 0, maxLegacySnapshotFiles)
47	63		files := 0
48	64		total := 0
49	65		for {

@@ -181,13 +197,17 @@ func (r *Repository) cleanupPendingLegacyDeletes(ctx context.Context) error {
181	197		return nil
182	198		}
183	199
184		-	func (r *Repository) pendingLegacyDeleteMarkers(ctx context.Context) ([]legacyDeleteMarker, error) {
185		-	f, err := os.Open(r.dir)
	200	+	func (r *Repository) pendingLegacyDeleteMarkers(ctx context.Context) (markers []legacyDeleteMarker, err error) {
	201	+	f, err := r.openLegacyDirectory(r.dir)
186	202		if err != nil {
187	203		return nil, err
188	204		}
189		-	defer f.Close()
190		-	markers := make([]legacyDeleteMarker, 0, maxLegacyDeleteMarkers)
	205	+	defer func() {
	206	+	if closeErr := f.Close(); closeErr != nil {
	207	+	err = errors.Join(err, fmt.Errorf("close legacy deletion marker directory: %w", safeFilesystemError(closeErr)))
	208	+	}
	209	+	}()
	210	+	markers = make([]legacyDeleteMarker, 0, maxLegacyDeleteMarkers)
191	211		for {
192	212		entries, readErr := f.ReadDir(maintenanceBatch)
193	213		for _, entry := range entries {

 	name string
 }
+type legacyDirectory interface {
+	ReadDir(int) ([]os.DirEntry, error)
+	Close() error
+}
+func (r *Repository) openLegacyDirectory(path string) (legacyDirectory, error) {
+	if r.hooks.openLegacyDirectory != nil {
+		return r.hooks.openLegacyDirectory(path)
+	}
+	return os.Open(path)
+}
 // LoadLegacy reads only pre-incremental root .snap files. It is deliberately
 // isolated from repository reads so incremental directories are never traversed.
-func (r *Repository) LoadLegacy(ctx context.Context) ([]ports.LegacySnapshot, error) {
+func (r *Repository) LoadLegacy(ctx context.Context) (out []ports.LegacySnapshot, err error) {
 	if err := ctx.Err(); err != nil {
 		return nil, err
 	}
 	if err := r.cleanupPendingLegacyDeletes(ctx); err != nil {
 		return nil, err
 	}
-	f, err := os.Open(r.dir)
+	f, err := r.openLegacyDirectory(r.dir)
 	if errors.Is(err, os.ErrNotExist) {
 		return nil, nil
 	}
 	if err != nil {
 		return nil, fmt.Errorf("read legacy snapshot directory: %w", safeFilesystemError(err))
 	}
-	defer f.Close()
+	defer func() {
+		if closeErr := f.Close(); closeErr != nil {
+			err = errors.Join(err, fmt.Errorf("close legacy snapshot directory: %w", safeFilesystemError(closeErr)))
+		}
+	}()
-	out := make([]ports.LegacySnapshot, 0, maxLegacySnapshotFiles)
+	out = make([]ports.LegacySnapshot, 0, maxLegacySnapshotFiles)
 	files := 0
 	total := 0
 	for {
 	return nil
 }
-func (r *Repository) pendingLegacyDeleteMarkers(ctx context.Context) ([]legacyDeleteMarker, error) {
+func (r *Repository) pendingLegacyDeleteMarkers(ctx context.Context) (markers []legacyDeleteMarker, err error) {
-	f, err := os.Open(r.dir)
+	f, err := r.openLegacyDirectory(r.dir)
 	if err != nil {
 		return nil, err
 	}
-	defer f.Close()
+	defer func() {
-	markers := make([]legacyDeleteMarker, 0, maxLegacyDeleteMarkers)
+		if closeErr := f.Close(); closeErr != nil {
+			err = errors.Join(err, fmt.Errorf("close legacy deletion marker directory: %w", safeFilesystemError(closeErr)))
+		}
+	}()
+	markers = make([]legacyDeleteMarker, 0, maxLegacyDeleteMarkers)
 	for {
 		entries, readErr := f.ReadDir(maintenanceBatch)
 		for _, entry := range entries {

M internal/adapters/snapshot/legacy_test.go +40 -0


@@ -4,6 +4,7 @@ import (
4	4		"context"
5	5		"errors"
6	6		"fmt"
	7	+	"io"
7	8		"os"
8	9		"path/filepath"
9	10		"strings"

@@ -34,6 +35,45 @@ func TestRepositoryLegacyReadsOnlyBoundedRootSnapshots(t *testing.T) {
34	35		}
35	36		}
36	37
	38	+	func TestRepositoryLoadLegacyReportsDirectoryCloseError(t *testing.T) {
	39	+	for _, tt := range []struct {
	40	+	name string
	41	+	failOpen int
	42	+	operation string
	43	+	}{
	44	+	{name: "pending deletion markers", failOpen: 1, operation: "close legacy deletion marker directory"},
	45	+	{name: "legacy snapshots", failOpen: 2, operation: "close legacy snapshot directory"},
	46	+	} {
	47	+	t.Run(tt.name, func(t *testing.T) {
	48	+	closeErr := errors.New("injected directory close failure")
	49	+	repo := NewRepository(privateDir(t))
	50	+	openCalls := 0
	51	+	repo.hooks.openLegacyDirectory = func(string) (legacyDirectory, error) {
	52	+	openCalls++
	53	+	if openCalls == tt.failOpen {
	54	+	return fakeLegacyDirectory{closeErr: closeErr}, nil
	55	+	}
	56	+	return fakeLegacyDirectory{}, nil
	57	+	}
	58	+
	59	+	_, err := repo.LoadLegacy(context.Background())
	60	+	if !errors.Is(err, closeErr) {
	61	+	t.Fatalf("LoadLegacy error = %v, want close error", err)
	62	+	}
	63	+	if !strings.Contains(err.Error(), tt.operation) {
	64	+	t.Fatalf("LoadLegacy error = %v, want %q", err, tt.operation)
	65	+	}
	66	+	})
	67	+	}
	68	+	}
	69	+
	70	+	type fakeLegacyDirectory struct {
	71	+	closeErr error
	72	+	}
	73	+
	74	+	func (d fakeLegacyDirectory) ReadDir(int) ([]os.DirEntry, error) { return nil, io.EOF }
	75	+	func (d fakeLegacyDirectory) Close() error { return d.closeErr }
	76	+
37	77		func TestRepositoryLegacyRejectsTooManyAndTooLargeAggregateSnapshots(t *testing.T) {
38	78		t.Run("count", func(t *testing.T) {
39	79		repo := NewRepository(privateDir(t))


@@ -4,6 +4,7 @@ import (
4	"context"	4		"context"
5	"errors"	5		"errors"
6	"fmt"	6		"fmt"
		7	+	"io"
7	"os"	8		"os"
8	"path/filepath"	9		"path/filepath"
9	"strings"	10		"strings"

@@ -34,6 +35,45 @@ func TestRepositoryLegacyReadsOnlyBoundedRootSnapshots(t *testing.T) {
34	}	35		}
35	}	36		}
36		37
		38	+	func TestRepositoryLoadLegacyReportsDirectoryCloseError(t *testing.T) {
		39	+	for _, tt := range []struct {
		40	+	name string
		41	+	failOpen int
		42	+	operation string
		43	+	}{
		44	+	{name: "pending deletion markers", failOpen: 1, operation: "close legacy deletion marker directory"},
		45	+	{name: "legacy snapshots", failOpen: 2, operation: "close legacy snapshot directory"},
		46	+	} {
		47	+	t.Run(tt.name, func(t *testing.T) {
		48	+	closeErr := errors.New("injected directory close failure")
		49	+	repo := NewRepository(privateDir(t))
		50	+	openCalls := 0
		51	+	repo.hooks.openLegacyDirectory = func(string) (legacyDirectory, error) {
		52	+	openCalls++
		53	+	if openCalls == tt.failOpen {
		54	+	return fakeLegacyDirectory{closeErr: closeErr}, nil
		55	+	}
		56	+	return fakeLegacyDirectory{}, nil
		57	+	}
		58	+
		59	+	_, err := repo.LoadLegacy(context.Background())
		60	+	if !errors.Is(err, closeErr) {
		61	+	t.Fatalf("LoadLegacy error = %v, want close error", err)
		62	+	}
		63	+	if !strings.Contains(err.Error(), tt.operation) {
		64	+	t.Fatalf("LoadLegacy error = %v, want %q", err, tt.operation)
		65	+	}
		66	+	})
		67	+	}
		68	+	}
		69	+
		70	+	type fakeLegacyDirectory struct {
		71	+	closeErr error
		72	+	}
		73	+
		74	+	func (d fakeLegacyDirectory) ReadDir(int) ([]os.DirEntry, error) { return nil, io.EOF }
		75	+	func (d fakeLegacyDirectory) Close() error { return d.closeErr }
		76	+
37	func TestRepositoryLegacyRejectsTooManyAndTooLargeAggregateSnapshots(t *testing.T) {	77		func TestRepositoryLegacyRejectsTooManyAndTooLargeAggregateSnapshots(t *testing.T) {
38	t.Run("count", func(t *testing.T) {	78		t.Run("count", func(t *testing.T) {
39	repo := NewRepository(privateDir(t))	79		repo := NewRepository(privateDir(t))

M internal/adapters/snapshot/maintenance.go +6 -2


@@ -87,7 +87,7 @@ func (r *Repository) Delete(ctx context.Context, name string) error {
87	87		return nil
88	88		}
89	89
90		-	func pendingQuarantine(dir, key string) (bool, error) {
	90	+	func pendingQuarantine(dir, key string) (pending bool, err error) {
91	91		f, err := os.Open(dir)
92	92		if errors.Is(err, os.ErrNotExist) {
93	93		return false, nil

@@ -95,7 +95,11 @@ func pendingQuarantine(dir, key string) (bool, error) {
95	95		if err != nil {
96	96		return false, err
97	97		}
98		-	defer f.Close()
	98	+	defer func() {
	99	+	if closeErr := f.Close(); closeErr != nil {
	100	+	err = errors.Join(err, fmt.Errorf("close deleting snapshot directory: %w", safeFilesystemError(closeErr)))
	101	+	}
	102	+	}()
99	103		prefix := ".deleting-" + key + "-"
100	104		for {
101	105		entries, err := f.ReadDir(maintenanceBatch)

 	return nil
 }
-func pendingQuarantine(dir, key string) (bool, error) {
+func pendingQuarantine(dir, key string) (pending bool, err error) {
 	f, err := os.Open(dir)
 	if errors.Is(err, os.ErrNotExist) {
 		return false, nil
 	if err != nil {
 		return false, err
 	}
-	defer f.Close()
+	defer func() {
+		if closeErr := f.Close(); closeErr != nil {
+			err = errors.Join(err, fmt.Errorf("close deleting snapshot directory: %w", safeFilesystemError(closeErr)))
+		}
+	}()
 	prefix := ".deleting-" + key + "-"
 	for {
 		entries, err := f.ReadDir(maintenanceBatch)

M internal/adapters/snapshot/repository.go +1 -0


@@ -60,6 +60,7 @@ type repositoryHooks struct {
60	60		syncDirectory func(string) error
61	61		remove func(string) error
62	62		openMaintenanceDirectory func(string) (maintenanceDirectory, error)
	63	+	openLegacyDirectory func(string) (legacyDirectory, error)
63	64		beforeSessionLock func(string)
64	65		}
65	66


@@ -60,6 +60,7 @@ type repositoryHooks struct {
60	syncDirectory func(string) error	60		syncDirectory func(string) error
61	remove func(string) error	61		remove func(string) error
62	openMaintenanceDirectory func(string) (maintenanceDirectory, error)	62		openMaintenanceDirectory func(string) (maintenanceDirectory, error)
		63	+	openLegacyDirectory func(string) (legacyDirectory, error)
63	beforeSessionLock func(string)	64		beforeSessionLock func(string)
64	}	65		}
65		66

M internal/usecase/daemon/session.go +17 -31


@@ -58,11 +58,8 @@ type session struct {
58	58		// locks while encoding or writing.
59	59		snapshotMu sync.Mutex
60	60		snapshotGeneration uint64
61		-	snapshotCapturedGeneration uint64
62	61		snapshotPublishedGeneration uint64
63	62		snapshotNextEligibleAt time.Time
64		-	snapshotAttempted bool
65		-	snapshotAttemptKind snapshotAttemptKind
66	63		// The coordinator state below is guarded by snapshotMu. A capture can be
67	64		// queued globally or in flight, never more than one of each per session.
68	65		// Quarantine cancels the session publication context before destructive

@@ -80,9 +77,6 @@ type session struct {
80	77		snapshotPublicationCancel context.CancelFunc
81	78		snapshotQuarantined bool
82	79		snapshotQuarantineEpoch uint64
83		-	// snapshotFailureSig is the stable class of the active failed publication.
84		-	// It is guarded by snapshotMu and deliberately never contains error text.
85		-	snapshotFailureSig string
86	80		// snapshotChunkCache retains encoded sealed history only for this named
87	81		// session. snapshotMu owns it so cache touches never contend with pane
88	82		// state or mutate bytes retained by queued/in-flight publications.

@@ -899,32 +893,24 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
899	893		}
900	894		var terminalSnapshotErr error
901	895		retainSnapshotRetry := false
902		-	if d.snapsEnabled && !isEphemeral {
903		-	if purge {
904		-	if d.snapshotRepository != nil {
905		-	// Repository deletion is intentionally deferred until the render
906		-	// coordinator is cancelled and joined below. See client.go's lock
907		-	// ordering notes: no coordinator callback may race a purge.
908		-	}
909		-	} else {
910		-	markSnapshotDirty(sess)
911		-	sess.snapshotMu.Lock()
912		-	terminalGeneration := sess.snapshotGeneration
913		-	sess.snapshotMu.Unlock()
914		-	if !d.scheduleFinalSnapshot(sess) \|\| !d.waitForSnapshotGenerationWithDeadline(sess, terminalGeneration, deadline) {
915		-	sess.mu.Lock()
916		-	name := sess.name
917		-	sess.mu.Unlock()
918		-	terminalSnapshotErr = fmt.Errorf("retain final snapshot for session %q: snapshot worker unavailable, saturated, or timed out", name)
919		-	if reason != ports.ReasonServerShutdown {
920		-	return terminalSnapshotErr
921		-	}
922		-	// Preserve any already-queued routine capture until the blocked
923		-	// worker can safely discard it. This retains the dirty generation
924		-	// and forced retry intent without adding another queue entry.
925		-	retainSnapshotRetry = true
926		-	d.persistShutdownSnapshotFailure(name, terminalSnapshotErr)
	896	+	if d.snapsEnabled && !isEphemeral && !purge {
	897	+	markSnapshotDirty(sess)
	898	+	sess.snapshotMu.Lock()
	899	+	terminalGeneration := sess.snapshotGeneration
	900	+	sess.snapshotMu.Unlock()
	901	+	if !d.scheduleFinalSnapshot(sess) \|\| !d.waitForSnapshotGenerationWithDeadline(sess, terminalGeneration, deadline) {
	902	+	sess.mu.Lock()
	903	+	name := sess.name
	904	+	sess.mu.Unlock()
	905	+	terminalSnapshotErr = fmt.Errorf("retain final snapshot for session %q: snapshot worker unavailable, saturated, or timed out", name)
	906	+	if reason != ports.ReasonServerShutdown {
	907	+	return terminalSnapshotErr
927	908		}
	909	+	// Preserve any already-queued routine capture until the blocked
	910	+	// worker can safely discard it. This retains the dirty generation
	911	+	// and forced retry intent without adding another queue entry.
	912	+	retainSnapshotRetry = true
	913	+	d.persistShutdownSnapshotFailure(name, terminalSnapshotErr)
928	914		}
929	915		}
930	916		if !isEphemeral {

 	// locks while encoding or writing.
 	snapshotMu                  sync.Mutex
 	snapshotGeneration          uint64
-	snapshotCapturedGeneration  uint64
 	snapshotPublishedGeneration uint64
 	snapshotNextEligibleAt      time.Time
-	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
 	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
 	// snapshotChunkCache retains encoded sealed history only for this named
 	// session. snapshotMu owns it so cache touches never contend with pane
 	// state or mutate bytes retained by queued/in-flight publications.
 	}
 	var terminalSnapshotErr error
 	retainSnapshotRetry := false
-	if d.snapsEnabled && !isEphemeral {
+	if d.snapsEnabled && !isEphemeral && !purge {
-		if purge {
+		markSnapshotDirty(sess)
-			if d.snapshotRepository != nil {
+		sess.snapshotMu.Lock()
-				// Repository deletion is intentionally deferred until the render
+		terminalGeneration := sess.snapshotGeneration
-				// coordinator is cancelled and joined below. See client.go's lock
+		sess.snapshotMu.Unlock()
-				// ordering notes: no coordinator callback may race a purge.
+		if !d.scheduleFinalSnapshot(sess) || !d.waitForSnapshotGenerationWithDeadline(sess, terminalGeneration, deadline) {
-			}
+			sess.mu.Lock()
-		} else {
+			name := sess.name
-			markSnapshotDirty(sess)
+			sess.mu.Unlock()
-			sess.snapshotMu.Lock()
+			terminalSnapshotErr = fmt.Errorf("retain final snapshot for session %q: snapshot worker unavailable, saturated, or timed out", name)
-			terminalGeneration := sess.snapshotGeneration
+			if reason != ports.ReasonServerShutdown {
-			sess.snapshotMu.Unlock()
+				return terminalSnapshotErr
-			if !d.scheduleFinalSnapshot(sess) || !d.waitForSnapshotGenerationWithDeadline(sess, terminalGeneration, deadline) {
-				sess.mu.Lock()
-				name := sess.name
-				sess.mu.Unlock()
-				terminalSnapshotErr = fmt.Errorf("retain final snapshot for session %q: snapshot worker unavailable, saturated, or timed out", name)
-				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)
 			}
+			// 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)
 		}
 	}
 	if !isEphemeral {

M internal/usecase/daemon/snapshot_coordinator.go +0 -23


@@ -5,15 +5,6 @@ import (
5	5		"time"
6	6		)
7	7
8		-	func snapshotCoordinatorContext(sess *session) (context.Context, context.CancelFunc) {
9		-	sess.snapshotMu.Lock()
10		-	defer sess.snapshotMu.Unlock()
11		-	if sess.snapshotPublicationContext == nil {
12		-	sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
13		-	}
14		-	return sess.snapshotPublicationContext, sess.snapshotPublicationCancel
15		-	}
16		-
17	8		// snapshotCoordinatorQuarantine identifies one stop request. A later teardown
18	9		// can supersede it, in which case only that later owner may resume publication.
19	10		type snapshotCoordinatorQuarantine struct {

@@ -98,13 +89,9 @@ func resumeSnapshotCoordinatorForNewIdentity(sess *session, quarantine snapshotC
98	89		// state while eligibility remains false, then atomically make generation 1
99	90		// dirty and wake the scheduler.
100	91		sess.snapshotGeneration = 1
101		-	sess.snapshotCapturedGeneration = 0
102	92		sess.snapshotPublishedGeneration = 0
103	93		sess.snapshotForcedGeneration = 0
104	94		sess.snapshotNextEligibleAt = time.Time{}
105		-	sess.snapshotAttempted = false
106		-	sess.snapshotAttemptKind = snapshotAttemptRoutine
107		-	sess.snapshotFailureSig = ""
108	95		sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)
109	96		sess.snapDirty.Store(true)
110	97		sess.snapEligible.Store(true)

@@ -247,7 +234,6 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
247	234		}
248	235		if succeeded && capture.session.snapshotGeneration == capture.generation {
249	236		capture.session.snapDirty.Store(false)
250		-	capture.session.snapshotFailureSig = ""
251	237		} else if !succeeded {
252	238		capture.session.snapDirty.Store(true)
253	239		}

@@ -282,12 +268,3 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
282	268		}
283	269		})
284	270		}
285		-
286		-	// reportSnapshotFailure records every failed persistence attempt but only
287		-	// displays a toast when its stable phase/error class changes. This keeps a
288		-	// blocked disk from repeatedly repainting every attached client while retaining
289		-	// the count in notification history.
290		-
291		-	// snapshotRepositorySaver waits until the earliest completion-derived routine
292		-	// eligibility. State changes wake it immediately, so a newly dirty session is
293		-	// captured without waiting for an unrelated session's interval.


@@ -5,15 +5,6 @@ import (
5		"time"	5	"time"
6		)	6	)
7			7
8	-	func snapshotCoordinatorContext(sess *session) (context.Context, context.CancelFunc) {
9	-	sess.snapshotMu.Lock()
10	-	defer sess.snapshotMu.Unlock()
11	-	if sess.snapshotPublicationContext == nil {
12	-	sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
13	-	}
14	-	return sess.snapshotPublicationContext, sess.snapshotPublicationCancel
15	-	}
16	-
17		// snapshotCoordinatorQuarantine identifies one stop request. A later teardown	8	// snapshotCoordinatorQuarantine identifies one stop request. A later teardown
18		// can supersede it, in which case only that later owner may resume publication.	9	// can supersede it, in which case only that later owner may resume publication.
19		type snapshotCoordinatorQuarantine struct {	10	type snapshotCoordinatorQuarantine struct {

@@ -98,13 +89,9 @@ func resumeSnapshotCoordinatorForNewIdentity(sess *session, quarantine snapshotC
98		// state while eligibility remains false, then atomically make generation 1	89	// state while eligibility remains false, then atomically make generation 1
99		// dirty and wake the scheduler.	90	// dirty and wake the scheduler.
100		sess.snapshotGeneration = 1	91	sess.snapshotGeneration = 1
101	-	sess.snapshotCapturedGeneration = 0
102		sess.snapshotPublishedGeneration = 0	92	sess.snapshotPublishedGeneration = 0
103		sess.snapshotForcedGeneration = 0	93	sess.snapshotForcedGeneration = 0
104		sess.snapshotNextEligibleAt = time.Time{}	94	sess.snapshotNextEligibleAt = time.Time{}
105	-	sess.snapshotAttempted = false
106	-	sess.snapshotAttemptKind = snapshotAttemptRoutine
107	-	sess.snapshotFailureSig = ""
108		sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)	95	sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)
109		sess.snapDirty.Store(true)	96	sess.snapDirty.Store(true)
110		sess.snapEligible.Store(true)	97	sess.snapEligible.Store(true)

@@ -247,7 +234,6 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
247		}	234	}
248		if succeeded && capture.session.snapshotGeneration == capture.generation {	235	if succeeded && capture.session.snapshotGeneration == capture.generation {
249		capture.session.snapDirty.Store(false)	236	capture.session.snapDirty.Store(false)
250	-	capture.session.snapshotFailureSig = ""
251		} else if !succeeded {	237	} else if !succeeded {
252		capture.session.snapDirty.Store(true)	238	capture.session.snapDirty.Store(true)
253		}	239	}

@@ -282,12 +268,3 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
282		}	268	}
283		})	269	})
284		}	270	}
285	-
286	-	// reportSnapshotFailure records every failed persistence attempt but only
287	-	// displays a toast when its stable phase/error class changes. This keeps a
288	-	// blocked disk from repeatedly repainting every attached client while retaining
289	-	// the count in notification history.
290	-
291	-	// snapshotRepositorySaver waits until the earliest completion-derived routine
292	-	// eligibility. State changes wake it immediately, so a newly dirty session is
293	-	// captured without waiting for an unrelated session's interval.

M internal/usecase/daemon/snapshot_incremental_test.go +1 -1


@@ -89,7 +89,7 @@ func TestSnapshotChunkCacheResistsWarmedOverCapacityHistoryScans(t *testing.T) {
89	89
90	90		func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) {
91	91		history := vt.NewHistory(vt.HistoryConfig{MaxRows: 2, ChunkRows: 1})
92		-	history.Append([]renderer.Cell{renderer.BlankCell()})
	92	+	require.NoError(t, history.Append([]renderer.Cell{renderer.BlankCell()}))
93	93		view := history.SnapshotView()
94	94		visible := vt.NewScreen(1, 1).PrimaryVisibleSnapshot()
95	95		d := New(nil, nil, nil)


@@ -89,7 +89,7 @@ func TestSnapshotChunkCacheResistsWarmedOverCapacityHistoryScans(t *testing.T) {
89			89
90		func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) {	90		func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) {
91		history := vt.NewHistory(vt.HistoryConfig{MaxRows: 2, ChunkRows: 1})	91		history := vt.NewHistory(vt.HistoryConfig{MaxRows: 2, ChunkRows: 1})
92	-	history.Append([]renderer.Cell{renderer.BlankCell()})	92	+	require.NoError(t, history.Append([]renderer.Cell{renderer.BlankCell()}))
93		view := history.SnapshotView()	93		view := history.SnapshotView()
94		visible := vt.NewScreen(1, 1).PrimaryVisibleSnapshot()	94		visible := vt.NewScreen(1, 1).PrimaryVisibleSnapshot()
95		d := New(nil, nil, nil)	95		d := New(nil, nil, nil)

M internal/usecase/daemon/snapshot_queue.go +0 -3


@@ -43,9 +43,6 @@ func (d Daemon) scheduleSnapshotWithFinalFallback(sess session, final bool) bo
43	43		}
44	44		generation := sess.snapshotGeneration
45	45		publicationContext := sess.snapshotPublicationContext
46		-	sess.snapshotCapturedGeneration = generation
47		-	sess.snapshotAttempted = true
48		-	sess.snapshotAttemptKind = kind
49	46		sess.snapshotPendingCaptures++
50	47		sess.snapshotPending = true
51	48		sess.signalSnapshotChangedLocked()


@@ -43,9 +43,6 @@ func (d Daemon) scheduleSnapshotWithFinalFallback(sess session, final bool) bo
43		}	43	}
44		generation := sess.snapshotGeneration	44	generation := sess.snapshotGeneration
45		publicationContext := sess.snapshotPublicationContext	45	publicationContext := sess.snapshotPublicationContext
46	-	sess.snapshotCapturedGeneration = generation
47	-	sess.snapshotAttempted = true
48	-	sess.snapshotAttemptKind = kind
49		sess.snapshotPendingCaptures++	46	sess.snapshotPendingCaptures++
50		sess.snapshotPending = true	47	sess.snapshotPending = true
51		sess.signalSnapshotChangedLocked()	48	sess.signalSnapshotChangedLocked()

M internal/usecase/daemon/snapshot_scheduler_test.go +35 -13


@@ -18,21 +18,32 @@ import (
18	18		)
19	19
20	20		func TestSnapshotCoordinatorQuarantineCancelsPublicationBeforeDelete(t *testing.T) {
21		-	sess := newSnapshotTestSession(t, "work", false, "/work")
22		-	ctx, cancel := snapshotCoordinatorContext(sess)
23		-	done := quarantineSnapshotCoordinator(sess)
	21	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	22	+	repository := portsmocks.NewMockSnapshotRepository(t)
	23	+	WithSnapshotRepository(repository, nil)(d)
	24	+	startSnapshotEncodeWorker(t, d)
24	25
	26	+	started := make(chan struct{})
	27	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, _ ports.SnapshotPublication) error {
	28	+	close(started)
	29	+	<-ctx.Done()
	30	+	return ctx.Err()
	31	+	}).Once()
	32	+	sess := newSnapshotTestSession(t, "work", false, "/work")
	33	+	markSnapshotDirty(sess)
	34	+	require.True(t, d.scheduleSnapshot(sess))
25	35		select {
26		-	case <-ctx.Done():
	36	+	case <-started:
27	37		case <-time.After(testWaitTimeout):
28		-	t.Fatal("quarantine did not cancel publication")
	38	+	t.Fatal("publication did not start")
29	39		}
	40	+
	41	+	done := quarantineSnapshotCoordinator(sess)
30	42		select {
31	43		case <-done:
32	44		case <-time.After(testWaitTimeout):
33		-	t.Fatal("quarantine did not join idle coordinator")
	45	+	t.Fatal("quarantine did not join canceled publication")
34	46		}
35		-	cancel()
36	47		}
37	48
38	49		func TestSnapshotCoordinatorQuarantineJoinsInFlightPublication(t *testing.T) {

@@ -750,17 +761,28 @@ func TestSnapshotSchedulerImmediateEligibilityAndStaleCapturesRemainDirty(t *tes
750	761		WithSnapshotRepository(portsmocks.NewMockSnapshotRepository(t), nil)(d)
751	762
752	763		t.Run("no prior attempt is immediately eligible", func(t *testing.T) {
	764	+	published := make(chan ports.SnapshotPublication, 1)
	765	+	repository := portsmocks.NewMockSnapshotRepository(t)
	766	+	WithSnapshotRepository(repository, nil)(d)
	767	+	startSnapshotEncodeWorker(t, d)
	768	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
	769	+	published <- publication
	770	+	return nil
	771	+	}).Once()
	772	+
753	773		sess := newSnapshotTestSession(t, "immediate", false, "/work")
754	774		sess.snapshotWake = d.snapshotWake
755	775		sess.snapDirty.Store(true)
756	776		d.sessions[sess.id] = sess
757	777
758		-	require.Equal(t, snapshotInterval, d.scheduleEligibleRepositorySnapshots())
759		-	sess.snapshotMu.Lock()
760		-	require.True(t, sess.snapshotAttempted)
761		-	require.Equal(t, uint64(0), sess.snapshotCapturedGeneration)
762		-	require.Equal(t, now.Add(snapshotInterval), sess.snapshotNextEligibleAt)
763		-	sess.snapshotMu.Unlock()
	778	+	d.scheduleEligibleRepositorySnapshots()
	779	+	select {
	780	+	case publication := <-published:
	781	+	require.Equal(t, "immediate", publication.Name)
	782	+	case <-time.After(testWaitTimeout):
	783	+	t.Fatal("immediately eligible snapshot was not published")
	784	+	}
	785	+	awaitSnapshotClean(t, sess)
764	786		})
765	787
766	788		t.Run("queued and in-flight stale successes cannot clear newer generation", func(t *testing.T) {

 )
 func TestSnapshotCoordinatorQuarantineCancelsPublicationBeforeDelete(t *testing.T) {
-	sess := newSnapshotTestSession(t, "work", false, "/work")
+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
-	ctx, cancel := snapshotCoordinatorContext(sess)
+	repository := portsmocks.NewMockSnapshotRepository(t)
-	done := quarantineSnapshotCoordinator(sess)
+	WithSnapshotRepository(repository, nil)(d)
+	startSnapshotEncodeWorker(t, d)
+	started := make(chan struct{})
+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, _ ports.SnapshotPublication) error {
+		close(started)
+		<-ctx.Done()
+		return ctx.Err()
+	}).Once()
+	sess := newSnapshotTestSession(t, "work", false, "/work")
+	markSnapshotDirty(sess)
+	require.True(t, d.scheduleSnapshot(sess))
 	select {
-	case <-ctx.Done():
+	case <-started:
 	case <-time.After(testWaitTimeout):
-		t.Fatal("quarantine did not cancel publication")
+		t.Fatal("publication did not start")
 	}
+	done := quarantineSnapshotCoordinator(sess)
 	select {
 	case <-done:
 	case <-time.After(testWaitTimeout):
-		t.Fatal("quarantine did not join idle coordinator")
+		t.Fatal("quarantine did not join canceled publication")
 	}
-	cancel()
 }
 func TestSnapshotCoordinatorQuarantineJoinsInFlightPublication(t *testing.T) {
 	WithSnapshotRepository(portsmocks.NewMockSnapshotRepository(t), nil)(d)
 	t.Run("no prior attempt is immediately eligible", func(t *testing.T) {
+		published := make(chan ports.SnapshotPublication, 1)
+		repository := portsmocks.NewMockSnapshotRepository(t)
+		WithSnapshotRepository(repository, nil)(d)
+		startSnapshotEncodeWorker(t, d)
+		repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
+			published <- publication
+			return nil
+		}).Once()
 		sess := newSnapshotTestSession(t, "immediate", false, "/work")
 		sess.snapshotWake = d.snapshotWake
 		sess.snapDirty.Store(true)
 		d.sessions[sess.id] = sess
-		require.Equal(t, snapshotInterval, d.scheduleEligibleRepositorySnapshots())
+		d.scheduleEligibleRepositorySnapshots()
-		sess.snapshotMu.Lock()
+		select {
-		require.True(t, sess.snapshotAttempted)
+		case publication := <-published:
-		require.Equal(t, uint64(0), sess.snapshotCapturedGeneration)
+			require.Equal(t, "immediate", publication.Name)
-		require.Equal(t, now.Add(snapshotInterval), sess.snapshotNextEligibleAt)
+		case <-time.After(testWaitTimeout):
-		sess.snapshotMu.Unlock()
+			t.Fatal("immediately eligible snapshot was not published")
+		}
+		awaitSnapshotClean(t, sess)
 	})
 	t.Run("queued and in-flight stale successes cannot clear newer generation", func(t *testing.T) {

M internal/usecase/daemon/snapshot_worker.go +0 -4


@@ -11,10 +11,6 @@ func (d Daemon) reportSnapshotFailure(capture snapshotCapture, phase string, c
11	11		return
12	12		}
13	13		signature := snapshotFailureSignature(phase, cause)
14		-	capture.session.snapshotMu.Lock()
15		-	capture.session.snapshotFailureSig = signature
16		-	capture.session.snapshotMu.Unlock()
17		-
18	14		d.snapshotNoticeMu.Lock()
19	15		changed := d.snapshotActiveFailureSignature != signature
20	16		d.snapshotActiveFailureSignature = signature


@@ -11,10 +11,6 @@ func (d Daemon) reportSnapshotFailure(capture snapshotCapture, phase string, c
11		return	11	return
12		}	12	}
13		signature := snapshotFailureSignature(phase, cause)	13	signature := snapshotFailureSignature(phase, cause)
14	-	capture.session.snapshotMu.Lock()
15	-	capture.session.snapshotFailureSig = signature
16	-	capture.session.snapshotMu.Unlock()
17	-
18		d.snapshotNoticeMu.Lock()	14	d.snapshotNoticeMu.Lock()
19		changed := d.snapshotActiveFailureSignature != signature	15	changed := d.snapshotActiveFailureSignature != signature
20		d.snapshotActiveFailureSignature = signature	16	d.snapshotActiveFailureSignature = signature