fix(daemon): serialize final session snapshots

commit: be1bb364c4f2c113da90640bb6abed0a4bc08879
author: brice · 3mo ago · 2026-07-24 16:11 UTC
committer: brice · 3mo ago · 2026-07-24 16:14 UTC
parent: 253fc6a
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +97 -1

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


@@ -40,7 +40,11 @@ type session struct {
40	40		// and vev-cmd control) for this session, so a one-shot command cannot
41	41		// resolve focus against one state and mutate a later one. Lock order:
42	42		// dispatchMu strictly before d.mu, sess.mu, or any tab mu.
43		-	dispatchMu sync.Mutex
	43	+	dispatchMu sync.Mutex
	44	+	// teardownMu is the outermost per-session lifecycle lock. It serializes
	45	+	// competing pane-exit, control, and daemon-shutdown teardown paths before
	46	+	// they acquire daemon, session, snapshot, or pane locks.
	47	+	teardownMu sync.Mutex
44	48		mu sync.Mutex // guards tabs, active, client, clipFiles, and clipboard queue state
45	49		themeMu sync.Mutex
46	50		tabs []*tab

@@ -908,6 +912,29 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
908	912		// only immutable capture state; it never observes closed worker channels or
909	913		// session-owned cache that teardown has released.
910	914		func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, purge bool, deadline *snapshotShutdownDeadline) error {
	915	+	if sess == nil {
	916	+	return nil
	917	+	}
	918	+	// Pane EOF, explicit close, and daemon shutdown can all converge on the same
	919	+	// session. Only one owner may create the terminal mutation and checkpoint;
	920	+	// later callers observe the completed registry transition instead of
	921	+	// manufacturing a newer generation that the owner will quarantine.
	922	+	sess.teardownMu.Lock()
	923	+	defer sess.teardownMu.Unlock()
	924	+
	925	+	d.mu.Lock()
	926	+	current := d.sessions[sess.id]
	927	+	d.mu.Unlock()
	928	+	if current != sess {
	929	+	if purge {
	930	+	sess.mu.Lock()
	931	+	name := sess.name
	932	+	sess.mu.Unlock()
	933	+	return d.retryStoppedPurge(name)
	934	+	}
	935	+	return nil
	936	+	}
	937	+
911	938		ringing := sess.anyAttention()
912	939		sess.mu.Lock()
913	940		isEphemeral := sess.ephemeral


@@ -40,7 +40,11 @@ type session struct {
40		// and vev-cmd control) for this session, so a one-shot command cannot	40		// and vev-cmd control) for this session, so a one-shot command cannot
41		// resolve focus against one state and mutate a later one. Lock order:	41		// resolve focus against one state and mutate a later one. Lock order:
42		// dispatchMu strictly before d.mu, sess.mu, or any tab mu.	42		// dispatchMu strictly before d.mu, sess.mu, or any tab mu.
43	-	dispatchMu sync.Mutex	43	+	dispatchMu sync.Mutex
			44	+	// teardownMu is the outermost per-session lifecycle lock. It serializes
			45	+	// competing pane-exit, control, and daemon-shutdown teardown paths before
			46	+	// they acquire daemon, session, snapshot, or pane locks.
			47	+	teardownMu sync.Mutex
44		mu sync.Mutex // guards tabs, active, client, clipFiles, and clipboard queue state	48		mu sync.Mutex // guards tabs, active, client, clipFiles, and clipboard queue state
45		themeMu sync.Mutex	49		themeMu sync.Mutex
46		tabs []*tab	50		tabs []*tab

@@ -908,6 +912,29 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
908		// only immutable capture state; it never observes closed worker channels or	912		// only immutable capture state; it never observes closed worker channels or
909		// session-owned cache that teardown has released.	913		// session-owned cache that teardown has released.
910		func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, purge bool, deadline *snapshotShutdownDeadline) error {	914		func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, purge bool, deadline *snapshotShutdownDeadline) error {
			915	+	if sess == nil {
			916	+	return nil
			917	+	}
			918	+	// Pane EOF, explicit close, and daemon shutdown can all converge on the same
			919	+	// session. Only one owner may create the terminal mutation and checkpoint;
			920	+	// later callers observe the completed registry transition instead of
			921	+	// manufacturing a newer generation that the owner will quarantine.
			922	+	sess.teardownMu.Lock()
			923	+	defer sess.teardownMu.Unlock()
			924	+
			925	+	d.mu.Lock()
			926	+	current := d.sessions[sess.id]
			927	+	d.mu.Unlock()
			928	+	if current != sess {
			929	+	if purge {
			930	+	sess.mu.Lock()
			931	+	name := sess.name
			932	+	sess.mu.Unlock()
			933	+	return d.retryStoppedPurge(name)
			934	+	}
			935	+	return nil
			936	+	}
			937	+
911		ringing := sess.anyAttention()	938		ringing := sess.anyAttention()
912		sess.mu.Lock()	939		sess.mu.Lock()
913		isEphemeral := sess.ephemeral	940		isEphemeral := sess.ephemeral

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


@@ -4,6 +4,7 @@ import (
4	4		"context"
5	5		"errors"
6	6		"io"
	7	+	"runtime"
7	8		"sync"
8	9		"sync/atomic"
9	10		"testing"

@@ -442,6 +443,74 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
442	443		}
443	444		}
444	445
	446	+	func TestConcurrentSessionTeardownPublishesOneFinalSnapshot(t *testing.T) {
	447	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	448	+	repository := portsmocks.NewMockSnapshotRepository(t)
	449	+	WithSnapshotRepository(repository, nil)(d)
	450	+	startSnapshotEncodeWorker(t, d)
	451	+
	452	+	publicationStarted := make(chan struct{})
	453	+	releasePublication := make(chan struct{})
	454	+	var releaseOnce sync.Once
	455	+	t.Cleanup(func() { releaseOnce.Do(func() { close(releasePublication) }) })
	456	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ ports.SnapshotPublication) error {
	457	+	close(publicationStarted)
	458	+	<-releasePublication
	459	+	return nil
	460	+	}).Once()
	461	+
	462	+	sess := newSnapshotTestSession(t, "work", false, "/work")
	463	+	pty, ok := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
	464	+	require.True(t, ok)
	465	+	pty.EXPECT().Close().Return(nil).Maybe()
	466	+	d.sessions[sess.id] = sess
	467	+
	468	+	first := make(chan error, 1)
	469	+	go func() { first <- d.killSession(sess, ports.ReasonServerShutdown, false) }()
	470	+	select {
	471	+	case <-publicationStarted:
	472	+	case <-time.After(testWaitTimeout):
	473	+	t.Fatal("first terminal snapshot did not reach repository")
	474	+	}
	475	+
	476	+	const duplicateCallers = 8
	477	+	started := make(chan struct{}, duplicateCallers)
	478	+	duplicates := make(chan error, duplicateCallers)
	479	+	for range duplicateCallers {
	480	+	go func() {
	481	+	started <- struct{}{}
	482	+	duplicates <- d.killSession(sess, ports.ReasonServerShutdown, false)
	483	+	}()
	484	+	}
	485	+	for range duplicateCallers {
	486	+	<-started
	487	+	}
	488	+	for range 1000 {
	489	+	runtime.Gosched()
	490	+	sess.snapshotMu.Lock()
	491	+	generation := sess.snapshotGeneration
	492	+	sess.snapshotMu.Unlock()
	493	+	if generation > 1 {
	494	+	break
	495	+	}
	496	+	}
	497	+
	498	+	sess.snapshotMu.Lock()
	499	+	generation := sess.snapshotGeneration
	500	+	sess.snapshotMu.Unlock()
	501	+	require.Equal(t, uint64(1), generation, "duplicate teardown callers must not create synthetic terminal generations")
	502	+
	503	+	releaseOnce.Do(func() { close(releasePublication) })
	504	+	require.NoError(t, <-first)
	505	+	for range duplicateCallers {
	506	+	require.NoError(t, <-duplicates)
	507	+	}
	508	+	sess.snapshotMu.Lock()
	509	+	generation = sess.snapshotGeneration
	510	+	sess.snapshotMu.Unlock()
	511	+	require.Equal(t, uint64(1), generation, "completed duplicate teardowns must leave the terminal generation unchanged")
	512	+	}
	513	+
445	514		func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {
446	515		p, releasePTY := newBlockingPTY(t)
447	516		defer releasePTY()


@@ -4,6 +4,7 @@ import (
4	"context"	4		"context"
5	"errors"	5		"errors"
6	"io"	6		"io"
		7	+	"runtime"
7	"sync"	8		"sync"
8	"sync/atomic"	9		"sync/atomic"
9	"testing"	10		"testing"

@@ -442,6 +443,74 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
442	}	443		}
443	}	444		}
444		445
		446	+	func TestConcurrentSessionTeardownPublishesOneFinalSnapshot(t *testing.T) {
		447	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
		448	+	repository := portsmocks.NewMockSnapshotRepository(t)
		449	+	WithSnapshotRepository(repository, nil)(d)
		450	+	startSnapshotEncodeWorker(t, d)
		451	+
		452	+	publicationStarted := make(chan struct{})
		453	+	releasePublication := make(chan struct{})
		454	+	var releaseOnce sync.Once
		455	+	t.Cleanup(func() { releaseOnce.Do(func() { close(releasePublication) }) })
		456	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ ports.SnapshotPublication) error {
		457	+	close(publicationStarted)
		458	+	<-releasePublication
		459	+	return nil
		460	+	}).Once()
		461	+
		462	+	sess := newSnapshotTestSession(t, "work", false, "/work")
		463	+	pty, ok := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
		464	+	require.True(t, ok)
		465	+	pty.EXPECT().Close().Return(nil).Maybe()
		466	+	d.sessions[sess.id] = sess
		467	+
		468	+	first := make(chan error, 1)
		469	+	go func() { first <- d.killSession(sess, ports.ReasonServerShutdown, false) }()
		470	+	select {
		471	+	case <-publicationStarted:
		472	+	case <-time.After(testWaitTimeout):
		473	+	t.Fatal("first terminal snapshot did not reach repository")
		474	+	}
		475	+
		476	+	const duplicateCallers = 8
		477	+	started := make(chan struct{}, duplicateCallers)
		478	+	duplicates := make(chan error, duplicateCallers)
		479	+	for range duplicateCallers {
		480	+	go func() {
		481	+	started <- struct{}{}
		482	+	duplicates <- d.killSession(sess, ports.ReasonServerShutdown, false)
		483	+	}()
		484	+	}
		485	+	for range duplicateCallers {
		486	+	<-started
		487	+	}
		488	+	for range 1000 {
		489	+	runtime.Gosched()
		490	+	sess.snapshotMu.Lock()
		491	+	generation := sess.snapshotGeneration
		492	+	sess.snapshotMu.Unlock()
		493	+	if generation > 1 {
		494	+	break
		495	+	}
		496	+	}
		497	+
		498	+	sess.snapshotMu.Lock()
		499	+	generation := sess.snapshotGeneration
		500	+	sess.snapshotMu.Unlock()
		501	+	require.Equal(t, uint64(1), generation, "duplicate teardown callers must not create synthetic terminal generations")
		502	+
		503	+	releaseOnce.Do(func() { close(releasePublication) })
		504	+	require.NoError(t, <-first)
		505	+	for range duplicateCallers {
		506	+	require.NoError(t, <-duplicates)
		507	+	}
		508	+	sess.snapshotMu.Lock()
		509	+	generation = sess.snapshotGeneration
		510	+	sess.snapshotMu.Unlock()
		511	+	require.Equal(t, uint64(1), generation, "completed duplicate teardowns must leave the terminal generation unchanged")
		512	+	}
		513	+
445	func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {	514		func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {
446	p, releasePTY := newBlockingPTY(t)	515		p, releasePTY := newBlockingPTY(t)
447	defer releasePTY()	516		defer releasePTY()