61d93fc refactor(daemon): add guarded session state · vev

M internal/usecase/daemon/client.go +11 -16


@@ -46,8 +46,7 @@ type attachedClient struct {
46	46		rend *renderer.Renderer
47	47		size domain.Size
48	48		keys *keys.Router
49		-	sessMu sync.Mutex
50		-	sess *session
	49	+	sess Guarded[*session]
51	50		copyMu sync.Mutex
52	51		copyMode *scopy.Mode
53	52		copyPending []byte

@@ -108,17 +107,9 @@ func (p *pendingByteTimer) stop() {
108	107		}
109	108		}
110	109
111		-	func (ac attachedClient) currentSession() session {
112		-	ac.sessMu.Lock()
113		-	defer ac.sessMu.Unlock()
114		-	return ac.sess
115		-	}
	110	+	func (ac attachedClient) currentSession() session { return ac.sess.Get() }
116	111
117		-	func (ac attachedClient) setSession(sess session) {
118		-	ac.sessMu.Lock()
119		-	defer ac.sessMu.Unlock()
120		-	ac.sess = sess
121		-	}
	112	+	func (ac attachedClient) setSession(sess session) { ac.sess.Set(sess) }
122	113
123	114		func (ac *attachedClient) copyModeActive() bool {
124	115		ac.copyMu.Lock()

@@ -213,13 +204,17 @@ func (d Daemon) notifyDetachedAsync(ac attachedClient, reason uint8) {
213	204		// of the call has completed. Each is deadline-bounded (boundedSend), so this
214	205		// wait is bounded too.
215	206		func (d *Daemon) waitNotifies() {
	207	+	for _, c := range d.notifiesSnapshot() {
	208	+	<-c
	209	+	}
	210	+	}
	211	+
	212	+	func (d *Daemon) notifiesSnapshot() []chan struct{} {
216	213		d.mu.Lock()
	214	+	defer d.mu.Unlock()
217	215		snapshot := make([]chan struct{}, len(d.notifies))
218	216		copy(snapshot, d.notifies)
219		-	d.mu.Unlock()
220		-	for _, c := range snapshot {
221		-	<-c
222		-	}
	217	+	return snapshot
223	218		}
224	219
225	220		func (d Daemon) attachClient(sess session, tr ports.Transport, sz domain.Size) (attachedClient, attachedClient) {

 	rend           *renderer.Renderer
 	size           domain.Size
 	keys           *keys.Router
-	sessMu         sync.Mutex
+	sess           Guarded[*session]
-	sess           *session
 	copyMu         sync.Mutex
 	copyMode       *scopy.Mode
 	copyPending    []byte
 	}
 }
-func (ac *attachedClient) currentSession() *session {
+func (ac *attachedClient) currentSession() *session { return ac.sess.Get() }
-	ac.sessMu.Lock()
-	defer ac.sessMu.Unlock()
-	return ac.sess
-}
-func (ac *attachedClient) setSession(sess *session) {
+func (ac *attachedClient) setSession(sess *session) { ac.sess.Set(sess) }
-	ac.sessMu.Lock()
-	defer ac.sessMu.Unlock()
-	ac.sess = sess
-}
 func (ac *attachedClient) copyModeActive() bool {
 	ac.copyMu.Lock()
 // of the call has completed. Each is deadline-bounded (boundedSend), so this
 // wait is bounded too.
 func (d *Daemon) waitNotifies() {
+	for _, c := range d.notifiesSnapshot() {
+		<-c
+	}
+}
+func (d *Daemon) notifiesSnapshot() []chan struct{} {
 	d.mu.Lock()
+	defer d.mu.Unlock()
 	snapshot := make([]chan struct{}, len(d.notifies))
 	copy(snapshot, d.notifies)
-	d.mu.Unlock()
+	return snapshot
-	for _, c := range snapshot {
-		<-c
-	}
 }
 func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size) (*attachedClient, *attachedClient) {

M internal/usecase/daemon/daemon.go +9 -4


@@ -199,10 +199,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
199	199		func (d *Daemon) shutdownAll(reason uint8) {
200	200		d.mu.Lock()
201	201		d.closing = true
202		-	snapshot := make([]*session, 0, len(d.sessions))
203		-	for _, s := range d.sessions {
204		-	snapshot = append(snapshot, s)
205		-	}
	202	+	snapshot := d.sessionsSnapshotLocked()
206	203		empty := len(snapshot) == 0
207	204		d.mu.Unlock()
208	205		if empty {

@@ -214,6 +211,14 @@ func (d *Daemon) shutdownAll(reason uint8) {
214	211		}
215	212		}
216	213
	214	+	func (d Daemon) sessionsSnapshotLocked() []session {
	215	+	snapshot := make([]*session, 0, len(d.sessions))
	216	+	for _, s := range d.sessions {
	217	+	snapshot = append(snapshot, s)
	218	+	}
	219	+	return snapshot
	220	+	}
	221	+
217	222		// handleConn reads the first frame off a fresh connection and routes it. A
218	223		// context watcher closes the transport on serve-context cancel so a handler
219	224		// parked in Recv (mid-handshake or between input frames) unwinds on shutdown.

 func (d *Daemon) shutdownAll(reason uint8) {
 	d.mu.Lock()
 	d.closing = true
-	snapshot := make([]*session, 0, len(d.sessions))
+	snapshot := d.sessionsSnapshotLocked()
-	for _, s := range d.sessions {
-		snapshot = append(snapshot, s)
-	}
 	empty := len(snapshot) == 0
 	d.mu.Unlock()
 	if empty {
 	}
 }
+func (d *Daemon) sessionsSnapshotLocked() []*session {
+	snapshot := make([]*session, 0, len(d.sessions))
+	for _, s := range d.sessions {
+		snapshot = append(snapshot, s)
+	}
+	return snapshot
+}
 // handleConn reads the first frame off a fresh connection and routes it. A
 // context watcher closes the transport on serve-context cancel so a handler
 // parked in Recv (mid-handshake or between input frames) unwinds on shutdown.

A internal/usecase/daemon/guarded.go +27 -0

@@ -0,0 +1,27 @@
	1	+	package daemon
	2	+
	3	+	import "sync"
	4	+
	5	+	// Guarded protects one value behind a mutex.
	6	+	type Guarded[T any] struct {
	7	+	mu sync.Mutex
	8	+	v T
	9	+	}
	10	+
	11	+	func (g *Guarded[T]) Get() T {
	12	+	g.mu.Lock()
	13	+	defer g.mu.Unlock()
	14	+	return g.v
	15	+	}
	16	+
	17	+	func (g *Guarded[T]) Set(v T) {
	18	+	g.mu.Lock()
	19	+	defer g.mu.Unlock()
	20	+	g.v = v
	21	+	}
	22	+
	23	+	func (g Guarded[T]) With(fn func(T)) {
	24	+	g.mu.Lock()
	25	+	defer g.mu.Unlock()
	26	+	fn(&g.v)
	27	+	}

@@ -0,0 +1,27 @@
	1	+	package daemon
	2	+
	3	+	import "sync"
	4	+
	5	+	// Guarded protects one value behind a mutex.
	6	+	type Guarded[T any] struct {
	7	+	mu sync.Mutex
	8	+	v T
	9	+	}
	10	+
	11	+	func (g *Guarded[T]) Get() T {
	12	+	g.mu.Lock()
	13	+	defer g.mu.Unlock()
	14	+	return g.v
	15	+	}
	16	+
	17	+	func (g *Guarded[T]) Set(v T) {
	18	+	g.mu.Lock()
	19	+	defer g.mu.Unlock()
	20	+	g.v = v
	21	+	}
	22	+
	23	+	func (g Guarded[T]) With(fn func(T)) {
	24	+	g.mu.Lock()
	25	+	defer g.mu.Unlock()
	26	+	fn(&g.v)
	27	+	}

A internal/usecase/daemon/guarded_test.go +33 -0

@@ -0,0 +1,33 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"sync"
	5	+	"testing"
	6	+
	7	+	"github.com/stretchr/testify/require"
	8	+	)
	9	+
	10	+	func TestGuardedGetSetWith(t *testing.T) {
	11	+	var g Guarded[int]
	12	+
	13	+	g.Set(10)
	14	+	require.Equal(t, 10, g.Get())
	15	+
	16	+	g.With(func(v int) { v += 5 })
	17	+	require.Equal(t, 15, g.Get())
	18	+	}
	19	+
	20	+	func TestGuardedConcurrentAccess(t *testing.T) {
	21	+	var g Guarded[int]
	22	+	var wg sync.WaitGroup
	23	+	for range 20 {
	24	+	wg.Go(func() {
	25	+	for range 1000 {
	26	+	g.With(func(v int) { v++ })
	27	+	}
	28	+	})
	29	+	}
	30	+	wg.Wait()
	31	+
	32	+	require.Equal(t, 20000, g.Get())
	33	+	}

@@ -0,0 +1,33 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"sync"
	5	+	"testing"
	6	+
	7	+	"github.com/stretchr/testify/require"
	8	+	)
	9	+
	10	+	func TestGuardedGetSetWith(t *testing.T) {
	11	+	var g Guarded[int]
	12	+
	13	+	g.Set(10)
	14	+	require.Equal(t, 10, g.Get())
	15	+
	16	+	g.With(func(v int) { v += 5 })
	17	+	require.Equal(t, 15, g.Get())
	18	+	}
	19	+
	20	+	func TestGuardedConcurrentAccess(t *testing.T) {
	21	+	var g Guarded[int]
	22	+	var wg sync.WaitGroup
	23	+	for range 20 {
	24	+	wg.Go(func() {
	25	+	for range 1000 {
	26	+	g.With(func(v int) { v++ })
	27	+	}
	28	+	})
	29	+	}
	30	+	wg.Wait()
	31	+
	32	+	require.Equal(t, 20000, g.Get())
	33	+	}

M internal/usecase/daemon/picker.go +1 -4


@@ -52,10 +52,7 @@ func (d Daemon) enterPicker(sess session, ac *attachedClient) {
52	52
53	53		func (d Daemon) pickerViews(cur session) ([]picker.SessionView, int) {
54	54		d.mu.Lock()
55		-	sessions := make([]*session, 0, len(d.sessions))
56		-	for _, s := range d.sessions {
57		-	sessions = append(sessions, s)
58		-	}
	55	+	sessions := d.sessionsSnapshotLocked()
59	56		d.mu.Unlock()
60	57		sort.Slice(sessions, func(i, j int) bool { return sessions[i].name < sessions[j].name })
61	58

 func (d *Daemon) pickerViews(cur *session) ([]picker.SessionView, int) {
 	d.mu.Lock()
-	sessions := make([]*session, 0, len(d.sessions))
+	sessions := d.sessionsSnapshotLocked()
-	for _, s := range d.sessions {
-		sessions = append(sessions, s)
-	}
 	d.mu.Unlock()
 	sort.Slice(sessions, func(i, j int) bool { return sessions[i].name < sessions[j].name })