1fbf071 perf(daemon): publish only changed snapshot objects · vev

M internal/usecase/daemon/performance_benchmark_test.go +65 -8


@@ -5,6 +5,7 @@ import (
5	5		"fmt"
6	6		"io"
7	7		"sync"
	8	+	"sync/atomic"
8	9		"testing"
9	10		"time"
10	11

@@ -23,7 +24,7 @@ func TestCountingSnapshotRepositoryPublishCountsAndRetainsManifest(t *testing.T)
23	24		object := []byte{0x01, 0x02}
24	25
25	26		require.NoError(t, repository.Publish(context.Background(), ports.SnapshotPublication{Name: "session", Manifest: payload, Objects: []ports.SnapshotObject{{Data: object}}}))
26		-	require.Equal(t, countingSnapshotMetrics{writes: 1, manifestBytes: uint64(len(payload)), objectBytes: uint64(len(object)), headBytes: snapshotHeadBytes}, repository.metrics())
	27	+	require.Equal(t, countingSnapshotMetrics{writes: 1, manifestBytes: uint64(len(payload)), objectBytes: uint64(len(object)), suppliedObjectBytes: uint64(len(object)), headBytes: snapshotHeadBytes}, repository.metrics())
27	28		require.Equal(t, payload, repository.lastPayload())
28	29
29	30		payload[0] = 0x01

@@ -52,12 +53,61 @@ func TestIncrementalSnapshotMetricsWriteNoUnchangedHistoryBlobsAndBoundCache(t *
52	53
53	54		metrics := fixture.snapshots.metrics()
54	55		require.Zero(t, metrics.historyBlobBytes)
	56	+	require.Zero(t, metrics.suppliedHistoryBytes)
55	57		require.Zero(t, metrics.objectBytes)
56	58		require.Positive(t, metrics.manifestBytes)
57	59		require.Equal(t, uint64(snapshotHeadBytes), metrics.headBytes)
58	60		require.LessOrEqual(t, benchmarkSnapshotCacheBytes(fixture.sess), snapshotChunkCacheLimit)
59	61		}
60	62
	63	+	func TestDaemonSnapshotDoesNotResupplyUnchangedTenThousandChunkHistory(t *testing.T) {
	64	+	fixture := newPerformanceFixture(t, performanceConfig{size: domain.Size{Cols: 120, Rows: 40}, panes: 1, historyRows: 10_000})
	65	+
	66	+	markSnapshotDirty(fixture.sess)
	67	+	require.True(t, fixture.d.scheduleSnapshot(fixture.sess))
	68	+	awaitSnapshotClean(t, fixture.sess)
	69	+
	70	+	var copies atomic.Uint64
	71	+	fixture.sess.snapshotMu.Lock()
	72	+	fixture.sess.snapshotChunkCache.copyObject = func(object ports.SnapshotObject) ports.SnapshotObject {
	73	+	copies.Add(1)
	74	+	return copySnapshotObject(object)
	75	+	}
	76	+	fixture.sess.snapshotMu.Unlock()
	77	+	fixture.snapshots.reset()
	78	+	fixture.activePane.mu.Lock()
	79	+	fixture.activePane.screen.Write([]byte("\\x1b[1;1Hchanged"))
	80	+	fixture.activePane.mu.Unlock()
	81	+	markSnapshotDirty(fixture.sess)
	82	+	require.True(t, fixture.d.scheduleFinalSnapshot(fixture.sess))
	83	+	awaitSnapshotClean(t, fixture.sess)
	84	+
	85	+	metrics := fixture.snapshots.metrics()
	86	+	require.Equal(t, uint64(1), metrics.writes)
	87	+	require.Zero(t, metrics.suppliedHistoryBytes, "retained sealed history must not be resupplied")
	88	+	require.Zero(t, copies.Load(), "retained sealed history must not be deep-copied")
	89	+	require.Positive(t, metrics.suppliedObjectBytes, "changed tail and visible state must still be supplied")
	90	+	requireCompleteSnapshotManifest(t, fixture.snapshots, fixture.sess.name)
	91	+	}
	92	+
	93	+	func requireCompleteSnapshotManifest(t testing.T, repository countingSnapshotRepository, name string) {
	94	+	t.Helper()
	95	+	repository.mu.Lock()
	96	+	manifestBytes := append([]byte(nil), repository.last...)
	97	+	objects := repository.objects[name]
	98	+	repository.mu.Unlock()
	99	+	manifest, err := snapcodec.UnmarshalManifest(manifestBytes)
	100	+	require.NoError(t, err)
	101	+	for _, tab := range manifest.Tabs {
	102	+	for _, pane := range tab.Panes {
	103	+	for _, ref := range append(append([]snapcodec.ObjectRef(nil), pane.Sealed...), pane.Tail, pane.Visible) {
	104	+	_, ok := objects[ref.Digest]
	105	+	require.True(t, ok, "manifest reference %x is not retained", ref.Digest)
	106	+	}
	107	+	}
	108	+	}
	109	+	}
	110	+
61	111		func TestIncrementalSnapshotMetricScenarios(t *testing.T) {
62	112		for _, tt := range []struct {
63	113		name string

@@ -506,11 +556,13 @@ func validateSnapshotBenchmarkFixture(b testing.B, fixture performanceFixture,
506	556
507	557		func addCountingSnapshotMetrics(a, b countingSnapshotMetrics) countingSnapshotMetrics {
508	558		return countingSnapshotMetrics{
509		-	writes: a.writes + b.writes,
510		-	objectBytes: a.objectBytes + b.objectBytes,
511		-	historyBlobBytes: a.historyBlobBytes + b.historyBlobBytes,
512		-	manifestBytes: a.manifestBytes + b.manifestBytes,
513		-	headBytes: a.headBytes + b.headBytes,
	559	+	writes: a.writes + b.writes,
	560	+	objectBytes: a.objectBytes + b.objectBytes,
	561	+	historyBlobBytes: a.historyBlobBytes + b.historyBlobBytes,
	562	+	suppliedObjectBytes: a.suppliedObjectBytes + b.suppliedObjectBytes,
	563	+	suppliedHistoryBytes: a.suppliedHistoryBytes + b.suppliedHistoryBytes,
	564	+	manifestBytes: a.manifestBytes + b.manifestBytes,
	565	+	headBytes: a.headBytes + b.headBytes,
514	566		}
515	567		}
516	568

@@ -527,6 +579,7 @@ func publishBenchmarkSnapshot(b testing.TB, fixture *performanceFixture, generat
527	579		if err := fixture.snapshots.Publish(context.Background(), publication); err != nil {
528	580		b.Fatal(err)
529	581		}
	582	+	markSnapshotCaptureObjectsPublished(capture)
530	583		}
531	584
532	585		func mutateBenchmarkVisible(fixture *performanceFixture, operation int) {

@@ -1227,7 +1280,7 @@ func (t *countingOutputTransport) lastPayload() []byte {
1227	1280		const snapshotHeadBytes = 4 + 8 + 32
1228	1281
1229	1282		type countingSnapshotMetrics struct {
1230		-	writes, objectBytes, historyBlobBytes, manifestBytes, headBytes uint64
	1283	+	writes, objectBytes, historyBlobBytes, suppliedObjectBytes, suppliedHistoryBytes, manifestBytes, headBytes uint64
1231	1284		}
1232	1285		type countingSnapshotRepository struct {
1233	1286		mu sync.Mutex

@@ -1255,12 +1308,16 @@ func (s *countingSnapshotRepository) Publish(_ context.Context, publication port
1255	1308		s.manifestBytes += uint64(len(publication.Manifest))
1256	1309		s.headBytes += snapshotHeadBytes
1257	1310		for _, object := range publication.Objects {
	1311	+	s.suppliedObjectBytes += uint64(len(object.Data))
	1312	+	kind, _, err := snapcodec.PreflightObject(object.Data)
	1313	+	if err == nil && kind == snapcodec.HistoryChunk {
	1314	+	s.suppliedHistoryBytes += uint64(len(object.Data))
	1315	+	}
1258	1316		if _, exists := objects[object.Digest]; exists {
1259	1317		continue
1260	1318		}
1261	1319		objects[object.Digest] = struct{}{}
1262	1320		s.objectBytes += uint64(len(object.Data))
1263		-	kind, _, err := snapcodec.PreflightObject(object.Data)
1264	1321		if err == nil && kind == snapcodec.HistoryChunk {
1265	1322		s.historyBlobBytes += uint64(len(object.Data))
1266	1323		}

 	"fmt"
 	"io"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 	object := []byte{0x01, 0x02}
 	require.NoError(t, repository.Publish(context.Background(), ports.SnapshotPublication{Name: "session", Manifest: payload, Objects: []ports.SnapshotObject{{Data: object}}}))
-	require.Equal(t, countingSnapshotMetrics{writes: 1, manifestBytes: uint64(len(payload)), objectBytes: uint64(len(object)), headBytes: snapshotHeadBytes}, repository.metrics())
+	require.Equal(t, countingSnapshotMetrics{writes: 1, manifestBytes: uint64(len(payload)), objectBytes: uint64(len(object)), suppliedObjectBytes: uint64(len(object)), headBytes: snapshotHeadBytes}, repository.metrics())
 	require.Equal(t, payload, repository.lastPayload())
 	payload[0] = 0x01
 	metrics := fixture.snapshots.metrics()
 	require.Zero(t, metrics.historyBlobBytes)
+	require.Zero(t, metrics.suppliedHistoryBytes)
 	require.Zero(t, metrics.objectBytes)
 	require.Positive(t, metrics.manifestBytes)
 	require.Equal(t, uint64(snapshotHeadBytes), metrics.headBytes)
 	require.LessOrEqual(t, benchmarkSnapshotCacheBytes(fixture.sess), snapshotChunkCacheLimit)
 }
+func TestDaemonSnapshotDoesNotResupplyUnchangedTenThousandChunkHistory(t *testing.T) {
+	fixture := newPerformanceFixture(t, performanceConfig{size: domain.Size{Cols: 120, Rows: 40}, panes: 1, historyRows: 10_000})
+	markSnapshotDirty(fixture.sess)
+	require.True(t, fixture.d.scheduleSnapshot(fixture.sess))
+	awaitSnapshotClean(t, fixture.sess)
+	var copies atomic.Uint64
+	fixture.sess.snapshotMu.Lock()
+	fixture.sess.snapshotChunkCache.copyObject = func(object ports.SnapshotObject) ports.SnapshotObject {
+		copies.Add(1)
+		return copySnapshotObject(object)
+	}
+	fixture.sess.snapshotMu.Unlock()
+	fixture.snapshots.reset()
+	fixture.activePane.mu.Lock()
+	fixture.activePane.screen.Write([]byte("\\x1b[1;1Hchanged"))
+	fixture.activePane.mu.Unlock()
+	markSnapshotDirty(fixture.sess)
+	require.True(t, fixture.d.scheduleFinalSnapshot(fixture.sess))
+	awaitSnapshotClean(t, fixture.sess)
+	metrics := fixture.snapshots.metrics()
+	require.Equal(t, uint64(1), metrics.writes)
+	require.Zero(t, metrics.suppliedHistoryBytes, "retained sealed history must not be resupplied")
+	require.Zero(t, copies.Load(), "retained sealed history must not be deep-copied")
+	require.Positive(t, metrics.suppliedObjectBytes, "changed tail and visible state must still be supplied")
+	requireCompleteSnapshotManifest(t, fixture.snapshots, fixture.sess.name)
+}
+func requireCompleteSnapshotManifest(t *testing.T, repository *countingSnapshotRepository, name string) {
+	t.Helper()
+	repository.mu.Lock()
+	manifestBytes := append([]byte(nil), repository.last...)
+	objects := repository.objects[name]
+	repository.mu.Unlock()
+	manifest, err := snapcodec.UnmarshalManifest(manifestBytes)
+	require.NoError(t, err)
+	for _, tab := range manifest.Tabs {
+		for _, pane := range tab.Panes {
+			for _, ref := range append(append([]snapcodec.ObjectRef(nil), pane.Sealed...), pane.Tail, pane.Visible) {
+				_, ok := objects[ref.Digest]
+				require.True(t, ok, "manifest reference %x is not retained", ref.Digest)
+			}
+		}
+	}
+}
 func TestIncrementalSnapshotMetricScenarios(t *testing.T) {
 	for _, tt := range []struct {
 		name        string
 func addCountingSnapshotMetrics(a, b countingSnapshotMetrics) countingSnapshotMetrics {
 	return countingSnapshotMetrics{
-		writes:           a.writes + b.writes,
+		writes:               a.writes + b.writes,
-		objectBytes:      a.objectBytes + b.objectBytes,
+		objectBytes:          a.objectBytes + b.objectBytes,
-		historyBlobBytes: a.historyBlobBytes + b.historyBlobBytes,
+		historyBlobBytes:     a.historyBlobBytes + b.historyBlobBytes,
-		manifestBytes:    a.manifestBytes + b.manifestBytes,
+		suppliedObjectBytes:  a.suppliedObjectBytes + b.suppliedObjectBytes,
-		headBytes:        a.headBytes + b.headBytes,
+		suppliedHistoryBytes: a.suppliedHistoryBytes + b.suppliedHistoryBytes,
+		manifestBytes:        a.manifestBytes + b.manifestBytes,
+		headBytes:            a.headBytes + b.headBytes,
 	}
 }
 	if err := fixture.snapshots.Publish(context.Background(), publication); err != nil {
 		b.Fatal(err)
 	}
+	markSnapshotCaptureObjectsPublished(capture)
 }
 func mutateBenchmarkVisible(fixture *performanceFixture, operation int) {
 const snapshotHeadBytes = 4 + 8 + 32
 type countingSnapshotMetrics struct {
-	writes, objectBytes, historyBlobBytes, manifestBytes, headBytes uint64
+	writes, objectBytes, historyBlobBytes, suppliedObjectBytes, suppliedHistoryBytes, manifestBytes, headBytes uint64
 }
 type countingSnapshotRepository struct {
 	mu sync.Mutex
 	s.manifestBytes += uint64(len(publication.Manifest))
 	s.headBytes += snapshotHeadBytes
 	for _, object := range publication.Objects {
+		s.suppliedObjectBytes += uint64(len(object.Data))
+		kind, _, err := snapcodec.PreflightObject(object.Data)
+		if err == nil && kind == snapcodec.HistoryChunk {
+			s.suppliedHistoryBytes += uint64(len(object.Data))
+		}
 		if _, exists := objects[object.Digest]; exists {
 			continue
 		}
 		objects[object.Digest] = struct{}{}
 		s.objectBytes += uint64(len(object.Data))
-		kind, _, err := snapcodec.PreflightObject(object.Data)
 		if err == nil && kind == snapcodec.HistoryChunk {
 			s.historyBlobBytes += uint64(len(object.Data))
 		}

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


@@ -674,13 +674,14 @@ 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		-	deleteErr func(string) error
682		-	syncs int
683		-	closed bool
	677	+	mu sync.Mutex
	678	+	data map[string][]byte
	679	+	sets int
	680	+	dels []string
	681	+	deleteErr func(string) error
	682	+	deleteDone chan struct{}
	683	+	syncs int
	684	+	closed bool
684	685		}
685	686
686	687		func newMockStore(t testing.T) (portsmocks.MockStore, *mockStoreState) {

@@ -702,15 +703,21 @@ func newMockStore(t testing.T) (portsmocks.MockStore, *mockStoreState) {
702	703		}).Maybe()
703	704		store.EXPECT().Delete(mock.Anything).RunAndReturn(func(k []byte) error {
704	705		state.mu.Lock()
705		-	defer state.mu.Unlock()
706	706		key := string(k)
707	707		state.dels = append(state.dels, key)
708	708		if state.deleteErr != nil {
709	709		if err := state.deleteErr(key); err != nil {
	710	+	state.mu.Unlock()
710	711		return err
711	712		}
712	713		}
713	714		delete(state.data, key)
	715	+	deleteDone := state.deleteDone
	716	+	state.deleteDone = nil
	717	+	state.mu.Unlock()
	718	+	if deleteDone != nil {
	719	+	close(deleteDone)
	720	+	}
714	721		return nil
715	722		}).Maybe()
716	723		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
-	deleteErr func(string) error
+	deleteErr  func(string) error
-	syncs     int
+	deleteDone chan struct{}
-	closed    bool
+	syncs      int
+	closed     bool
 }
 func newMockStore(t *testing.T) (*portsmocks.MockStore, *mockStoreState) {
 	}).Maybe()
 	store.EXPECT().Delete(mock.Anything).RunAndReturn(func(k []byte) error {
 		state.mu.Lock()
-		defer state.mu.Unlock()
 		key := string(k)
 		state.dels = append(state.dels, key)
 		if state.deleteErr != nil {
 			if err := state.deleteErr(key); err != nil {
+				state.mu.Unlock()
 				return err
 			}
 		}
 		delete(state.data, key)
+		deleteDone := state.deleteDone
+		state.deleteDone = nil
+		state.mu.Unlock()
+		if deleteDone != nil {
+			close(deleteDone)
+		}
 		return nil
 	}).Maybe()
 	store.EXPECT().Range(mock.Anything).Run(func(fn func(k, v []byte) bool) {

M internal/usecase/daemon/snapshot.go +2 -1


@@ -40,7 +40,8 @@ type snapshotCapture struct {
40	40		active uint16
41	41		tabs []snapshotCaptureTab
42	42		publicationContext context.Context
43		-	coordinatorDiscarded bool // guarded by session.snapshotMu
	43	+	sealedRefs map[*vt.HistoryChunk]snapcodec.ObjectRef // set by the single encoder worker
	44	+	coordinatorDiscarded bool // guarded by session.snapshotMu
44	45		finishOnce sync.Once
45	46		}
46	47


@@ -40,7 +40,8 @@ type snapshotCapture struct {
40		active uint16	40		active uint16
41		tabs []snapshotCaptureTab	41		tabs []snapshotCaptureTab
42		publicationContext context.Context	42		publicationContext context.Context
43	-	coordinatorDiscarded bool // guarded by session.snapshotMu	43	+	sealedRefs map[*vt.HistoryChunk]snapcodec.ObjectRef // set by the single encoder worker
			44	+	coordinatorDiscarded bool // guarded by session.snapshotMu
44		finishOnce sync.Once	45		finishOnce sync.Once
45		}	46		}
46			47

M internal/usecase/daemon/snapshot_incremental.go +60 -20


@@ -17,44 +17,54 @@ type snapshotChunkCacheEntry struct {
17	17		}
18	18
19	19		// snapshotChunkCache is keyed by immutable HistoryChunk identity. Its owner
20		-	// must hold the named session's snapshotMu for every operation. Entries never
21		-	// expose cache-owned bytes: a publication gets a fresh slice, so a repository
22		-	// cannot mutate data used by an in-flight or later checkpoint.
	20	+	// must hold the named session's snapshotMu for every operation. Encoded bytes
	21	+	// remain cache-owned; publications receive a copy only until a successful
	22	+	// repository publication has made the immutable object available by reference.
23	23		type snapshotChunkCache struct {
24		-	limit int
25		-	used int
26		-	byPtr map[*vt.HistoryChunk]snapshotChunkCacheEntry
	24	+	limit int
	25	+	used int
	26	+	byPtr map[*vt.HistoryChunk]snapshotChunkCacheEntry
	27	+	persisted map[*vt.HistoryChunk]snapcodec.ObjectRef
27	28		// marshalChunk makes cache admission measurable without coupling the cache
28	29		// policy test to VT's wire encoding implementation.
29	30		marshalChunk func(*vt.HistoryChunk) ([]byte, error)
	31	+	copyObject func(ports.SnapshotObject) ports.SnapshotObject
30	32		}
31	33
32	34		func newSnapshotChunkCache(limit int) *snapshotChunkCache {
33	35		return &snapshotChunkCache{
34	36		limit: limit,
35	37		byPtr: make(map[*vt.HistoryChunk]snapshotChunkCacheEntry),
	38	+	persisted: make(map[*vt.HistoryChunk]snapcodec.ObjectRef),
36	39		marshalChunk: vt.MarshalHistoryChunk,
	40	+	copyObject: copySnapshotObject,
37	41		}
38	42		}
39	43
40		-	// objectLocked returns an independently owned object. Caller holds the owning
41		-	// session's snapshotMu.
42		-	func (c snapshotChunkCache) objectLocked(chunk vt.HistoryChunk) (ports.SnapshotObject, error) {
	44	+	// objectLocked returns a complete manifest reference and, when the repository
	45	+	// does not already retain it, an independently owned object to supply. Caller
	46	+	// holds the owning session's snapshotMu.
	47	+	func (c snapshotChunkCache) objectLocked(chunk vt.HistoryChunk) (snapcodec.ObjectRef, *ports.SnapshotObject, error) {
43	48		if chunk == nil {
44		-	return ports.SnapshotObject{}, fmt.Errorf("snapshot: nil history chunk")
	49	+	return snapcodec.ObjectRef{}, nil, fmt.Errorf("snapshot: nil history chunk")
	50	+	}
	51	+	if ref, ok := c.persisted[chunk]; ok {
	52	+	return ref, nil, nil
45	53		}
46	54		if entry, ok := c.byPtr[chunk]; ok {
47		-	return copySnapshotObject(entry.object), nil
	55	+	object := c.copyObject(entry.object)
	56	+	return objectRef(snapcodec.HistoryChunk, object), &object, nil
48	57		}
49	58
50	59		payload, err := c.marshalChunk(chunk)
51	60		if err != nil {
52		-	return ports.SnapshotObject{}, err
	61	+	return snapcodec.ObjectRef{}, nil, err
53	62		}
54	63		object, err := snapcodec.MarshalObject(snapcodec.HistoryChunk, payload)
55	64		if err != nil {
56		-	return ports.SnapshotObject{}, err
	65	+	return snapcodec.ObjectRef{}, nil, err
57	66		}
	67	+	ref := objectRef(snapcodec.HistoryChunk, object)
58	68
59	69		// An oldest-to-newest snapshot scan can exceed the cache. LRU would admit
60	70		// every early miss and evict the later working set, causing the next scan to

@@ -62,12 +72,12 @@ func (c snapshotChunkCache) objectLocked(chunk vt.HistoryChunk) (ports.Snapsho
62	72		// return later misses uncached. Pruning makes space when a retained chunk is
63	73		// no longer reachable by a queued or in-flight capture.
64	74		if len(object.Data) > c.limit \|\| c.used > c.limit-len(object.Data) {
65		-	return object, nil
	75	+	return ref, &object, nil
66	76		}
67		-	stored := copySnapshotObject(object)
	77	+	stored := c.copyObject(object)
68	78		c.byPtr[chunk] = snapshotChunkCacheEntry{object: stored}
69	79		c.used += len(stored.Data)
70		-	return object, nil
	80	+	return ref, &object, nil
71	81		}
72	82
73	83		// pruneLocked releases cached encodings which no currently queued or in-flight

@@ -79,6 +89,11 @@ func (c snapshotChunkCache) pruneLocked(referenced map[vt.HistoryChunk]struct{
79	89		c.used -= len(entry.object.Data)
80	90		}
81	91		}
	92	+	for chunk := range c.persisted {
	93	+	if _, keep := referenced[chunk]; !keep {
	94	+	delete(c.persisted, chunk)
	95	+	}
	96	+	}
82	97		}
83	98
84	99		func copySnapshotObject(object ports.SnapshotObject) ports.SnapshotObject {

@@ -100,6 +115,7 @@ func (d Daemon) incrementalPublication(capture snapshotCapture) (ports.Snapsho
100	115		}
101	116		manifest := snapcodec.Manifest{Generation: capture.generation, Name: capture.name, CreatedAt: capture.createdAt, Active: capture.active, Tabs: make([]snapcodec.ManifestTab, 0, len(capture.tabs))}
102	117		objects := make([]ports.SnapshotObject, 0)
	118	+	capture.sealedRefs = make(map[*vt.HistoryChunk]snapcodec.ObjectRef)
103	119		for _, tab := range capture.tabs {
104	120		outTab := snapcodec.ManifestTab{StableID: tab.stableID, Cols: tab.cols, Rows: tab.rows, NextPaneID: tab.nextPaneID, Focus: tab.focus, Tree: tab.tree, Panes: make([]snapcodec.ManifestPane, 0, len(tab.panes))}
105	121		for _, pane := range tab.panes {

@@ -111,12 +127,15 @@ func (d Daemon) incrementalPublication(capture snapshotCapture) (ports.Snapsho
111	127		}
112	128		outPane := snapcodec.ManifestPane{ID: pane.id, StableID: pane.stableID, Cwd: pane.cwd, Process: pane.process}
113	129		for i := 0; i < pane.sealed.ChunkCount(); i++ {
114		-	object, err := snapshotChunkObject(capture, pane.sealed.Chunk(i))
	130	+	ref, object, err := snapshotChunkObject(capture, pane.sealed.Chunk(i))
115	131		if err != nil {
116	132		return ports.SnapshotPublication{}, fmt.Errorf("snapshot history chunk: %w", err)
117	133		}
118		-	outPane.Sealed = append(outPane.Sealed, objectRef(snapcodec.HistoryChunk, object))
119		-	objects = append(objects, object)
	134	+	outPane.Sealed = append(outPane.Sealed, ref)
	135	+	capture.sealedRefs[pane.sealed.Chunk(i)] = ref
	136	+	if object != nil {
	137	+	objects = append(objects, *object)
	138	+	}
120	139		}
121	140		tail, err := vt.MarshalHistory(pane.tail)
122	141		if pane.tail.Len() == 0 {

@@ -161,13 +180,34 @@ func prepareSnapshotChunkCache(capture *snapshotCapture) error {
161	180		return nil
162	181		}
163	182
164		-	func snapshotChunkObject(capture snapshotCapture, chunk vt.HistoryChunk) (ports.SnapshotObject, error) {
	183	+	func snapshotChunkObject(capture snapshotCapture, chunk vt.HistoryChunk) (snapcodec.ObjectRef, *ports.SnapshotObject, error) {
165	184		sess := capture.session
166	185		sess.snapshotMu.Lock()
167	186		defer sess.snapshotMu.Unlock()
168	187		return sess.snapshotChunkCache.objectLocked(chunk)
169	188		}
170	189
	190	+	// markSnapshotCaptureObjectsPublished records that the immutable sealed
	191	+	// history referenced by capture is available in this session's repository
	192	+	// lineage. It is called only after Publish succeeds.
	193	+	func markSnapshotCaptureObjectsPublished(capture *snapshotCapture) {
	194	+	if capture == nil \|\| capture.session == nil {
	195	+	return
	196	+	}
	197	+	sess := capture.session
	198	+	sess.snapshotMu.Lock()
	199	+	defer sess.snapshotMu.Unlock()
	200	+	cache := sess.snapshotChunkCache
	201	+	if cache == nil {
	202	+	return
	203	+	}
	204	+	for chunk, ref := range capture.sealedRefs {
	205	+	if chunk != nil {
	206	+	cache.persisted[chunk] = ref
	207	+	}
	208	+	}
	209	+	}
	210	+
171	211		// pruneSnapshotChunkCacheLocked retains only chunks referenced by captures that
172	212		// can still be published. Caller holds sess.snapshotMu.
173	213		func pruneSnapshotChunkCacheLocked(sess session, capture snapshotCapture) {

 }
 // snapshotChunkCache is keyed by immutable HistoryChunk identity. Its owner
-// must hold the named session's snapshotMu for every operation. Entries never
+// must hold the named session's snapshotMu for every operation. Encoded bytes
-// expose cache-owned bytes: a publication gets a fresh slice, so a repository
+// remain cache-owned; publications receive a copy only until a successful
-// cannot mutate data used by an in-flight or later checkpoint.
+// repository publication has made the immutable object available by reference.
 type snapshotChunkCache struct {
-	limit int
+	limit     int
-	used  int
+	used      int
-	byPtr map[*vt.HistoryChunk]snapshotChunkCacheEntry
+	byPtr     map[*vt.HistoryChunk]snapshotChunkCacheEntry
+	persisted map[*vt.HistoryChunk]snapcodec.ObjectRef
 	// marshalChunk makes cache admission measurable without coupling the cache
 	// policy test to VT's wire encoding implementation.
 	marshalChunk func(*vt.HistoryChunk) ([]byte, error)
+	copyObject   func(ports.SnapshotObject) ports.SnapshotObject
 }
 func newSnapshotChunkCache(limit int) *snapshotChunkCache {
 	return &snapshotChunkCache{
 		limit:        limit,
 		byPtr:        make(map[*vt.HistoryChunk]snapshotChunkCacheEntry),
+		persisted:    make(map[*vt.HistoryChunk]snapcodec.ObjectRef),
 		marshalChunk: vt.MarshalHistoryChunk,
+		copyObject:   copySnapshotObject,
 	}
 }
-// objectLocked returns an independently owned object. Caller holds the owning
+// objectLocked returns a complete manifest reference and, when the repository
-// session's snapshotMu.
+// does not already retain it, an independently owned object to supply. Caller
-func (c *snapshotChunkCache) objectLocked(chunk *vt.HistoryChunk) (ports.SnapshotObject, error) {
+// holds the owning session's snapshotMu.
+func (c *snapshotChunkCache) objectLocked(chunk *vt.HistoryChunk) (snapcodec.ObjectRef, *ports.SnapshotObject, error) {
 	if chunk == nil {
-		return ports.SnapshotObject{}, fmt.Errorf("snapshot: nil history chunk")
+		return snapcodec.ObjectRef{}, nil, fmt.Errorf("snapshot: nil history chunk")
+	}
+	if ref, ok := c.persisted[chunk]; ok {
+		return ref, nil, nil
 	}
 	if entry, ok := c.byPtr[chunk]; ok {
-		return copySnapshotObject(entry.object), nil
+		object := c.copyObject(entry.object)
+		return objectRef(snapcodec.HistoryChunk, object), &object, nil
 	}
 	payload, err := c.marshalChunk(chunk)
 	if err != nil {
-		return ports.SnapshotObject{}, err
+		return snapcodec.ObjectRef{}, nil, err
 	}
 	object, err := snapcodec.MarshalObject(snapcodec.HistoryChunk, payload)
 	if err != nil {
-		return ports.SnapshotObject{}, err
+		return snapcodec.ObjectRef{}, nil, err
 	}
+	ref := objectRef(snapcodec.HistoryChunk, object)
 	// An oldest-to-newest snapshot scan can exceed the cache. LRU would admit
 	// every early miss and evict the later working set, causing the next scan to
 	// return later misses uncached. Pruning makes space when a retained chunk is
 	// no longer reachable by a queued or in-flight capture.
 	if len(object.Data) > c.limit || c.used > c.limit-len(object.Data) {
-		return object, nil
+		return ref, &object, nil
 	}
-	stored := copySnapshotObject(object)
+	stored := c.copyObject(object)
 	c.byPtr[chunk] = snapshotChunkCacheEntry{object: stored}
 	c.used += len(stored.Data)
-	return object, nil
+	return ref, &object, nil
 }
 // pruneLocked releases cached encodings which no currently queued or in-flight
 			c.used -= len(entry.object.Data)
 		}
 	}
+	for chunk := range c.persisted {
+		if _, keep := referenced[chunk]; !keep {
+			delete(c.persisted, chunk)
+		}
+	}
 }
 func copySnapshotObject(object ports.SnapshotObject) ports.SnapshotObject {
 	}
 	manifest := snapcodec.Manifest{Generation: capture.generation, Name: capture.name, CreatedAt: capture.createdAt, Active: capture.active, Tabs: make([]snapcodec.ManifestTab, 0, len(capture.tabs))}
 	objects := make([]ports.SnapshotObject, 0)
+	capture.sealedRefs = make(map[*vt.HistoryChunk]snapcodec.ObjectRef)
 	for _, tab := range capture.tabs {
 		outTab := snapcodec.ManifestTab{StableID: tab.stableID, Cols: tab.cols, Rows: tab.rows, NextPaneID: tab.nextPaneID, Focus: tab.focus, Tree: tab.tree, Panes: make([]snapcodec.ManifestPane, 0, len(tab.panes))}
 		for _, pane := range tab.panes {
 			}
 			outPane := snapcodec.ManifestPane{ID: pane.id, StableID: pane.stableID, Cwd: pane.cwd, Process: pane.process}
 			for i := 0; i < pane.sealed.ChunkCount(); i++ {
-				object, err := snapshotChunkObject(capture, pane.sealed.Chunk(i))
+				ref, object, err := snapshotChunkObject(capture, pane.sealed.Chunk(i))
 				if err != nil {
 					return ports.SnapshotPublication{}, fmt.Errorf("snapshot history chunk: %w", err)
 				}
-				outPane.Sealed = append(outPane.Sealed, objectRef(snapcodec.HistoryChunk, object))
+				outPane.Sealed = append(outPane.Sealed, ref)
-				objects = append(objects, object)
+				capture.sealedRefs[pane.sealed.Chunk(i)] = ref
+				if object != nil {
+					objects = append(objects, *object)
+				}
 			}
 			tail, err := vt.MarshalHistory(pane.tail)
 			if pane.tail.Len() == 0 {
 	return nil
 }
-func snapshotChunkObject(capture *snapshotCapture, chunk *vt.HistoryChunk) (ports.SnapshotObject, error) {
+func snapshotChunkObject(capture *snapshotCapture, chunk *vt.HistoryChunk) (snapcodec.ObjectRef, *ports.SnapshotObject, error) {
 	sess := capture.session
 	sess.snapshotMu.Lock()
 	defer sess.snapshotMu.Unlock()
 	return sess.snapshotChunkCache.objectLocked(chunk)
 }
+// markSnapshotCaptureObjectsPublished records that the immutable sealed
+// history referenced by capture is available in this session's repository
+// lineage. It is called only after Publish succeeds.
+func markSnapshotCaptureObjectsPublished(capture *snapshotCapture) {
+	if capture == nil || capture.session == nil {
+		return
+	}
+	sess := capture.session
+	sess.snapshotMu.Lock()
+	defer sess.snapshotMu.Unlock()
+	cache := sess.snapshotChunkCache
+	if cache == nil {
+		return
+	}
+	for chunk, ref := range capture.sealedRefs {
+		if chunk != nil {
+			cache.persisted[chunk] = ref
+		}
+	}
+}
 // pruneSnapshotChunkCacheLocked retains only chunks referenced by captures that
 // can still be published. Caller holds sess.snapshotMu.
 func pruneSnapshotChunkCacheLocked(sess *session, capture *snapshotCapture) {

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


@@ -59,7 +59,7 @@ func TestSnapshotChunkCacheResistsWarmedOverCapacityHistoryScans(t *testing.T) {
59	59
60	60		for pass := range 3 {
61	61		for i := range view.ChunkCount() {
62		-	_, err := cache.objectLocked(view.Chunk(i))
	62	+	_, _, err := cache.objectLocked(view.Chunk(i))
63	63		require.NoError(t, err)
64	64		}
65	65		require.LessOrEqual(t, cache.used, cache.limit)

@@ -82,7 +82,7 @@ func TestSnapshotChunkCacheResistsWarmedOverCapacityHistoryScans(t *testing.T) {
82	82		isolatedEncodes++
83	83		return vt.MarshalHistoryChunk(chunk)
84	84		}
85		-	_, err = isolated.objectLocked(view.Chunk(0))
	85	+	_, _, err = isolated.objectLocked(view.Chunk(0))
86	86		require.NoError(t, err)
87	87		require.Equal(t, 1, isolatedEncodes, "a session cache must not share encoded history with another session")
88	88		}


@@ -59,7 +59,7 @@ func TestSnapshotChunkCacheResistsWarmedOverCapacityHistoryScans(t *testing.T) {
59			59
60		for pass := range 3 {	60		for pass := range 3 {
61		for i := range view.ChunkCount() {	61		for i := range view.ChunkCount() {
62	-	_, err := cache.objectLocked(view.Chunk(i))	62	+	_, _, err := cache.objectLocked(view.Chunk(i))
63		require.NoError(t, err)	63		require.NoError(t, err)
64		}	64		}
65		require.LessOrEqual(t, cache.used, cache.limit)	65		require.LessOrEqual(t, cache.used, cache.limit)

@@ -82,7 +82,7 @@ func TestSnapshotChunkCacheResistsWarmedOverCapacityHistoryScans(t *testing.T) {
82		isolatedEncodes++	82		isolatedEncodes++
83		return vt.MarshalHistoryChunk(chunk)	83		return vt.MarshalHistoryChunk(chunk)
84		}	84		}
85	-	_, err = isolated.objectLocked(view.Chunk(0))	85	+	_, _, err = isolated.objectLocked(view.Chunk(0))
86		require.NoError(t, err)	86		require.NoError(t, err)
87		require.Equal(t, 1, isolatedEncodes, "a session cache must not share encoded history with another session")	87		require.Equal(t, 1, isolatedEncodes, "a session cache must not share encoded history with another session")
88		}	88		}

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


@@ -92,6 +92,8 @@ func TestRenameKeepsSnapshotCoordinatorQuarantinedUntilNewIdentityCommits(t *tes
92	92
93	93		deleteEntered := make(chan struct{})
94	94		allowDelete := make(chan struct{})
	95	+	deleteCompleted := make(chan struct{})
	96	+	state.deleteDone = deleteCompleted
95	97		published := make(chan ports.SnapshotPublication, 2)
96	98		repository.EXPECT().Tombstone(mock.Anything, "work").Return(nil).Once()
97	99		repository.EXPECT().Delete(mock.Anything, "work").RunAndReturn(func(context.Context, string) error {

@@ -103,7 +105,7 @@ func TestRenameKeepsSnapshotCoordinatorQuarantinedUntilNewIdentityCommits(t *tes
103	105		repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
104	106		published <- publication
105	107		return nil
106		-	}).Maybe()
	108	+	}).Once()
107	109
108	110		pty, release := newBlockingPTY(t)
109	111		defer release()

@@ -124,7 +126,12 @@ func TestRenameKeepsSnapshotCoordinatorQuarantinedUntilNewIdentityCommits(t *tes
124	126		// during this window must not capture or publish the old name.
125	127		d.mu.Lock()
126	128		close(allowDelete)
127		-	require.Eventually(t, func() bool { return !state.has("work") }, testWaitTimeout, time.Millisecond)
	129	+	select {
	130	+	case <-deleteCompleted:
	131	+	case <-time.After(testWaitTimeout):
	132	+	t.Fatal("old snapshot identity was not deleted")
	133	+	}
	134	+	require.False(t, state.has("work"))
128	135		sess.snapshotMu.Lock()
129	136		quarantined := sess.snapshotQuarantined
130	137		sess.snapshotMu.Unlock()

@@ -137,12 +144,7 @@ func TestRenameKeepsSnapshotCoordinatorQuarantinedUntilNewIdentityCommits(t *tes
137	144		case d.snapshotWake <- struct{}{}:
138	145		default:
139	146		}
140		-	require.True(t, d.scheduleSnapshot(sess))
141		-	select {
142		-	case publication := <-published:
143		-	t.Fatalf("published %q after old Delete before identity commit", publication.Name)
144		-	case <-time.After(50 * time.Millisecond):
145		-	}
	147	+	require.True(t, d.scheduleSnapshot(sess), "old work may be coalesced while the coordinator is quarantined")
146	148		d.mu.Unlock()
147	149
148	150		require.NoError(t, <-renameDone)

@@ -375,12 +377,11 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
375	377		// repository call is always released before cleanup asks the worker to join.
376	378		t.Cleanup(release)
377	379		repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
378		-	if publication.Name == "blocker" {
379		-	close(blockerStarted)
380		-	<-releaseBlocker
381		-	}
	380	+	require.Equal(t, "blocker", publication.Name)
	381	+	close(blockerStarted)
	382	+	<-releaseBlocker
382	383		return nil
383		-	}).Maybe()
	384	+	}).Once()
384	385		notices.EXPECT().Append(mock.MatchedBy(func(notification domain.Notification) bool {
385	386		return notification.Code == domain.NoticeSnapshotWrite && notification.Severity == domain.NoticeError
386	387		})).Return(nil).Once()

@@ -424,11 +425,20 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
424	425		// The retained capture is not published after quarantine, but releasing the
425	426		// unrelated blocked call lets the worker discard it and lets cleanup join.
426	427		release()
427		-	require.Eventually(t, func() bool {
	428	+	for {
428	429		sess.snapshotMu.Lock()
429		-	defer sess.snapshotMu.Unlock()
430		-	return !sess.snapshotPending && sess.snapshotQueuedCapture == nil
431		-	}, testWaitTimeout, time.Millisecond)
	430	+	complete := !sess.snapshotPending && sess.snapshotQueuedCapture == nil
	431	+	changed := sess.snapshotChangeLocked()
	432	+	sess.snapshotMu.Unlock()
	433	+	if complete {
	434	+	break
	435	+	}
	436	+	select {
	437	+	case <-changed:
	438	+	case <-time.After(testWaitTimeout):
	439	+	t.Fatal("quarantined capture was not discarded")
	440	+	}
	441	+	}
432	442		}
433	443
434	444		func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {

@@ -491,12 +501,14 @@ func TestServeShutdownDeadlineDoesNotWaitTwiceForUncooperativeSnapshotRepository
491	501		return n.Code == domain.NoticeSnapshotWrite && n.Severity == domain.NoticeError
492	502		})).Return(nil).Once()
493	503
	504	+	firstPublicationCompleted := make(chan struct{})
494	505		terminalPublicationStarted := make(chan struct{})
495	506		releaseTerminalPublication := make(chan struct{})
496	507		var lastValid atomic.Uint64
497	508		repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
498	509		if publication.Generation == 1 {
499	510		lastValid.Store(publication.Generation)
	511	+	close(firstPublicationCompleted)
500	512		return nil
501	513		}
502	514		close(terminalPublicationStarted)

@@ -514,7 +526,11 @@ func TestServeShutdownDeadlineDoesNotWaitTwiceForUncooperativeSnapshotRepository
514	526		awaitFrame(t, sends, ports.MsgWelcome)
515	527		markSnapshotDirty(firstSession(d))
516	528		d.snapshotWake <- struct{}{}
517		-	require.Eventually(t, func() bool { return lastValid.Load() == 1 }, testWaitTimeout, time.Millisecond)
	529	+	select {
	530	+	case <-firstPublicationCompleted:
	531	+	case <-time.After(testWaitTimeout):
	532	+	t.Fatal("initial checkpoint did not complete")
	533	+	}
518	534
519	535		cancel()
520	536		select {

@@ -626,7 +642,7 @@ func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *t
626	642
627	643		markSnapshotDirty(sess)
628	644		require.True(t, d.scheduleSnapshot(sess))
629		-	require.Eventually(t, func() bool { return publishes.Load() == 1 }, time.Second, time.Millisecond)
	645	+	require.Equal(t, int32(1), publishes.Load(), "rate-limited routine work must not publish")
630	646
631	647		require.True(t, d.scheduleFinalSnapshot(sess))
632	648		awaitSnapshotClean(t, sess)

 	deleteEntered := make(chan struct{})
 	allowDelete := make(chan struct{})
+	deleteCompleted := make(chan struct{})
+	state.deleteDone = deleteCompleted
 	published := make(chan ports.SnapshotPublication, 2)
 	repository.EXPECT().Tombstone(mock.Anything, "work").Return(nil).Once()
 	repository.EXPECT().Delete(mock.Anything, "work").RunAndReturn(func(context.Context, string) error {
 	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
 		published <- publication
 		return nil
-	}).Maybe()
+	}).Once()
 	pty, release := newBlockingPTY(t)
 	defer release()
 	// during this window must not capture or publish the old name.
 	d.mu.Lock()
 	close(allowDelete)
-	require.Eventually(t, func() bool { return !state.has("work") }, testWaitTimeout, time.Millisecond)
+	select {
+	case <-deleteCompleted:
+	case <-time.After(testWaitTimeout):
+		t.Fatal("old snapshot identity was not deleted")
+	}
+	require.False(t, state.has("work"))
 	sess.snapshotMu.Lock()
 	quarantined := sess.snapshotQuarantined
 	sess.snapshotMu.Unlock()
 	case d.snapshotWake <- struct{}{}:
 	default:
 	}
-	require.True(t, d.scheduleSnapshot(sess))
+	require.True(t, d.scheduleSnapshot(sess), "old work may be coalesced while the coordinator is quarantined")
-	select {
-	case publication := <-published:
-		t.Fatalf("published %q after old Delete before identity commit", publication.Name)
-	case <-time.After(50 * time.Millisecond):
-	}
 	d.mu.Unlock()
 	require.NoError(t, <-renameDone)
 	// 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" {
+		require.Equal(t, "blocker", publication.Name)
-			close(blockerStarted)
+		close(blockerStarted)
-			<-releaseBlocker
+		<-releaseBlocker
-		}
 		return nil
-	}).Maybe()
+	}).Once()
 	notices.EXPECT().Append(mock.MatchedBy(func(notification domain.Notification) bool {
 		return notification.Code == domain.NoticeSnapshotWrite && notification.Severity == domain.NoticeError
 	})).Return(nil).Once()
 	// 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 {
+	for {
 		sess.snapshotMu.Lock()
-		defer sess.snapshotMu.Unlock()
+		complete := !sess.snapshotPending && sess.snapshotQueuedCapture == nil
-		return !sess.snapshotPending && sess.snapshotQueuedCapture == nil
+		changed := sess.snapshotChangeLocked()
-	}, testWaitTimeout, time.Millisecond)
+		sess.snapshotMu.Unlock()
+		if complete {
+			break
+		}
+		select {
+		case <-changed:
+		case <-time.After(testWaitTimeout):
+			t.Fatal("quarantined capture was not discarded")
+		}
+	}
 }
 func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {
 		return n.Code == domain.NoticeSnapshotWrite && n.Severity == domain.NoticeError
 	})).Return(nil).Once()
+	firstPublicationCompleted := make(chan struct{})
 	terminalPublicationStarted := make(chan struct{})
 	releaseTerminalPublication := make(chan struct{})
 	var lastValid atomic.Uint64
 	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
 		if publication.Generation == 1 {
 			lastValid.Store(publication.Generation)
+			close(firstPublicationCompleted)
 			return nil
 		}
 		close(terminalPublicationStarted)
 	awaitFrame(t, sends, ports.MsgWelcome)
 	markSnapshotDirty(firstSession(d))
 	d.snapshotWake <- struct{}{}
-	require.Eventually(t, func() bool { return lastValid.Load() == 1 }, testWaitTimeout, time.Millisecond)
+	select {
+	case <-firstPublicationCompleted:
+	case <-time.After(testWaitTimeout):
+		t.Fatal("initial checkpoint did not complete")
+	}
 	cancel()
 	select {
 	markSnapshotDirty(sess)
 	require.True(t, d.scheduleSnapshot(sess))
-	require.Eventually(t, func() bool { return publishes.Load() == 1 }, time.Second, time.Millisecond)
+	require.Equal(t, int32(1), publishes.Load(), "rate-limited routine work must not publish")
 	require.True(t, d.scheduleFinalSnapshot(sess))
 	awaitSnapshotClean(t, sess)

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


@@ -134,6 +134,9 @@ func (d *Daemon) publishSnapshotCapture(workerCtx context.Context, workerID uint
134	134		}
135	135		}
136	136		d.clearSnapshotWorkerInFlight(workerID, capture)
	137	+	if err == nil && workerCtx.Err() == nil {
	138	+	markSnapshotCaptureObjectsPublished(capture)
	139	+	}
137	140		if err != nil && workerCtx.Err() == nil && publicationContext.Err() == nil {
138	141		// Global, not session-scoped: by now the session may already be torn
139	142		// down, so a session notice would be dead-on-arrival. No lock is held


@@ -134,6 +134,9 @@ func (d *Daemon) publishSnapshotCapture(workerCtx context.Context, workerID uint
134	}	134		}
135	}	135		}
136	d.clearSnapshotWorkerInFlight(workerID, capture)	136		d.clearSnapshotWorkerInFlight(workerID, capture)
		137	+	if err == nil && workerCtx.Err() == nil {
		138	+	markSnapshotCaptureObjectsPublished(capture)
		139	+	}
137	if err != nil && workerCtx.Err() == nil && publicationContext.Err() == nil {	140		if err != nil && workerCtx.Err() == nil && publicationContext.Err() == nil {
138	// Global, not session-scoped: by now the session may already be torn	141		// Global, not session-scoped: by now the session may already be torn
139	// down, so a session notice would be dead-on-arrival. No lock is held	142		// down, so a session notice would be dead-on-arrival. No lock is held