8f34248 fix(daemon): serialize detached output state · vev-vt

M internal/usecase/daemon/attachment_registry.go +9 -2


@@ -48,9 +48,16 @@ func (ac *attachedClient) publishView(view attachmentView) {
48	48		ac.viewMu.Lock()
49	49		// Rebase while viewMu is held, before the new revision becomes visible.
50	50		// Output fences take the same lock, so no side effect can pair the new
51		-	// ViewRevision with the previous output epoch.
	51	+	// ViewRevision with the previous output epoch. Test/headless fixtures may
	52	+	// leave output.attachment unset; use the stream-local lock for those.
52	53		if ac.output != nil && ac.view.revision != view.revision {
53		-	ac.output.rebaseLocked()
	54	+	if ac.output.attachment == ac {
	55	+	ac.output.rebaseLocked()
	56	+	} else {
	57	+	ac.output.stateMu.Lock()
	58	+	ac.output.rebaseLocked()
	59	+	ac.output.stateMu.Unlock()
	60	+	}
54	61		}
55	62		ac.view = view
56	63		ac.viewMu.Unlock()

 	ac.viewMu.Lock()
 	// Rebase while viewMu is held, before the new revision becomes visible.
 	// Output fences take the same lock, so no side effect can pair the new
-	// ViewRevision with the previous output epoch.
+	// ViewRevision with the previous output epoch. Test/headless fixtures may
+	// leave output.attachment unset; use the stream-local lock for those.
 	if ac.output != nil && ac.view.revision != view.revision {
-		ac.output.rebaseLocked()
+		if ac.output.attachment == ac {
+			ac.output.rebaseLocked()
+		} else {
+			ac.output.stateMu.Lock()
+			ac.output.rebaseLocked()
+			ac.output.stateMu.Unlock()
+		}
 	}
 	ac.view = view
 	ac.viewMu.Unlock()

M internal/usecase/daemon/client.go +6 -6


@@ -356,8 +356,8 @@ func (ac *attachedClient) sendExpectedSideEffectForAttachment(expected transport
356	356		if err := ac.beginExpectedTransportSendLocked(expected, ticket); err != nil {
357	357		return errAttachmentTransition
358	358		}
359		-	unlockView := ac.output.lockView()
360		-	defer unlockView()
	359	+	ac.output.lockView()
	360	+	defer ac.output.unlockView()
361	361		frame, err := ac.output.sideEffectLocked(data, echoAck)
362	362		if err == nil {
363	363		err = expected.transport.Send(frame)

@@ -408,8 +408,8 @@ func (d Daemon) boundedSendOutputErrTransport(ac attachedClient, b []byte) (po
408	408		if !ac.transportSnapshotCurrent(expected) {
409	409		return expected.transport, errTransportReplaced
410	410		}
411		-	unlockView := ac.output.lockView()
412		-	defer unlockView()
	411	+	ac.output.lockView()
	412	+	defer ac.output.unlockView()
413	413		frame, err := ac.output.sideEffectLocked(b, ac.echoAck.Load())
414	414		if err == nil {
415	415		err = owned.SendSynchronous(frame)

@@ -422,8 +422,8 @@ func (d Daemon) boundedSendOutputErrTransport(ac attachedClient, b []byte) (po
422	422		if !ac.transportSnapshotCurrent(expected) {
423	423		return errTransportReplaced
424	424		}
425		-	unlockView := ac.output.lockView()
426		-	defer unlockView()
	425	+	ac.output.lockView()
	426	+	defer ac.output.unlockView()
427	427		frame, err := ac.output.sideEffectLocked(b, ac.echoAck.Load())
428	428		if err != nil {
429	429		return err


@@ -356,8 +356,8 @@ func (ac *attachedClient) sendExpectedSideEffectForAttachment(expected transport
356		if err := ac.beginExpectedTransportSendLocked(expected, ticket); err != nil {	356		if err := ac.beginExpectedTransportSendLocked(expected, ticket); err != nil {
357		return errAttachmentTransition	357		return errAttachmentTransition
358		}	358		}
359	-	unlockView := ac.output.lockView()	359	+	ac.output.lockView()
360	-	defer unlockView()	360	+	defer ac.output.unlockView()
361		frame, err := ac.output.sideEffectLocked(data, echoAck)	361		frame, err := ac.output.sideEffectLocked(data, echoAck)
362		if err == nil {	362		if err == nil {
363		err = expected.transport.Send(frame)	363		err = expected.transport.Send(frame)

@@ -408,8 +408,8 @@ func (d Daemon) boundedSendOutputErrTransport(ac attachedClient, b []byte) (po
408		if !ac.transportSnapshotCurrent(expected) {	408		if !ac.transportSnapshotCurrent(expected) {
409		return expected.transport, errTransportReplaced	409		return expected.transport, errTransportReplaced
410		}	410		}
411	-	unlockView := ac.output.lockView()	411	+	ac.output.lockView()
412	-	defer unlockView()	412	+	defer ac.output.unlockView()
413		frame, err := ac.output.sideEffectLocked(b, ac.echoAck.Load())	413		frame, err := ac.output.sideEffectLocked(b, ac.echoAck.Load())
414		if err == nil {	414		if err == nil {
415		err = owned.SendSynchronous(frame)	415		err = owned.SendSynchronous(frame)

@@ -422,8 +422,8 @@ func (d Daemon) boundedSendOutputErrTransport(ac attachedClient, b []byte) (po
422		if !ac.transportSnapshotCurrent(expected) {	422		if !ac.transportSnapshotCurrent(expected) {
423		return errTransportReplaced	423		return errTransportReplaced
424		}	424		}
425	-	unlockView := ac.output.lockView()	425	+	ac.output.lockView()
426	-	defer unlockView()	426	+	defer ac.output.unlockView()
427		frame, err := ac.output.sideEffectLocked(b, ac.echoAck.Load())	427		frame, err := ac.output.sideEffectLocked(b, ac.echoAck.Load())
428		if err != nil {	428		if err != nil {
429		return err	429		return err

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


@@ -271,8 +271,8 @@ func (d Daemon) boundedSendClipboardForward(item clipboardForward, ticket atta
271	271		if !beginClipboardOwnerSend(item.owner, ticket, expected) {
272	272		return errAttachmentTransition
273	273		}
274		-	unlockView := ac.output.lockView()
275		-	defer unlockView()
	274	+	ac.output.lockView()
	275	+	defer ac.output.unlockView()
276	276		frame, err := ac.output.sideEffectLocked(item.seq, ac.echoAck.Load())
277	277		if err == nil {
278	278		if owned {


@@ -271,8 +271,8 @@ func (d Daemon) boundedSendClipboardForward(item clipboardForward, ticket atta
271		if !beginClipboardOwnerSend(item.owner, ticket, expected) {	271		if !beginClipboardOwnerSend(item.owner, ticket, expected) {
272		return errAttachmentTransition	272		return errAttachmentTransition
273		}	273		}
274	-	unlockView := ac.output.lockView()	274	+	ac.output.lockView()
275	-	defer unlockView()	275	+	defer ac.output.unlockView()
276		frame, err := ac.output.sideEffectLocked(item.seq, ac.echoAck.Load())	276		frame, err := ac.output.sideEffectLocked(item.seq, ac.echoAck.Load())
277		if err == nil {	277		if err == nil {
278		if owned {	278		if owned {

M internal/usecase/daemon/daemon_test.go +0 -1


@@ -401,7 +401,6 @@ func newManualSessionWithPTYsClockCleanup(t testing.TB, clock ports.Clock, regis
401	401		d := newTestDaemonWithCleanup(t, nil, clock, registerCleanup)
402	402		tr, sends := newCapturingTransport(t)
403	403		ac := &attachedClient{tr: tr, output: newOutputStateStream(), size: domain.Size{Cols: 80, Rows: 24}}
404		-	ac.output.attachment = ac
405	404		ac.initOverlays()
406	405		sctx, cancel := context.WithCancel(d.serveCtx)
407	406		tabs := make([]*tab, 0, len(ptys))


@@ -401,7 +401,6 @@ func newManualSessionWithPTYsClockCleanup(t testing.TB, clock ports.Clock, regis
401		d := newTestDaemonWithCleanup(t, nil, clock, registerCleanup)	401	d := newTestDaemonWithCleanup(t, nil, clock, registerCleanup)
402		tr, sends := newCapturingTransport(t)	402	tr, sends := newCapturingTransport(t)
403		ac := &attachedClient{tr: tr, output: newOutputStateStream(), size: domain.Size{Cols: 80, Rows: 24}}	403	ac := &attachedClient{tr: tr, output: newOutputStateStream(), size: domain.Size{Cols: 80, Rows: 24}}
404	-	ac.output.attachment = ac
405		ac.initOverlays()	404	ac.initOverlays()
406		sctx, cancel := context.WithCancel(d.serveCtx)	405	sctx, cancel := context.WithCancel(d.serveCtx)
407		tabs := make([]*tab, 0, len(ptys))	406	tabs := make([]*tab, 0, len(ptys))

M internal/usecase/daemon/output_state_stream.go +38 -22


@@ -2,6 +2,7 @@ package daemon
2	2
3	3		import (
4	4		"errors"
	5	+	"sync"
5	6		"sync/atomic"
6	7
7	8		"github.com/bnema/vev/internal/domain"

@@ -30,18 +31,33 @@ type outputStateStream struct {
30	31		forceSnapshot bool
31	32		initialized bool
32	33		attachment *attachedClient
	34	+	stateMu sync.Mutex
33	35		}
34	36
35	37		// lockView serializes an attachment's view publication with every output
36	38		// state transition. It deliberately does not take sendMu: callers that hold
37	39		// architecture locks can publish a view without reversing the paint path's
38	40		// sendMu -> session lock order.
39		-	func (s *outputStateStream) lockView() func() {
40		-	if s == nil \|\| s.attachment == nil {
41		-	return func() {}
	41	+	func (s *outputStateStream) lockView() {
	42	+	if s == nil {
	43	+	return
	44	+	}
	45	+	if s.attachment == nil {
	46	+	s.stateMu.Lock()
	47	+	return
42	48		}
43	49		s.attachment.viewMu.Lock()
44		-	return s.attachment.viewMu.Unlock
	50	+	}
	51	+
	52	+	func (s *outputStateStream) unlockView() {
	53	+	if s == nil {
	54	+	return
	55	+	}
	56	+	if s.attachment == nil {
	57	+	s.stateMu.Unlock()
	58	+	return
	59	+	}
	60	+	s.attachment.viewMu.Unlock()
45	61		}
46	62
47	63		func newOutputStateStream(windowSize ...uint8) *outputStateStream {

@@ -76,8 +92,8 @@ type preparedOutput struct {
76	92		}
77	93
78	94		func (s outputStateStream) prepare(frame renderer.Frame, damage []renderer.Damage, reset bool) (preparedOutput, error) {
79		-	unlockView := s.lockView()
80		-	defer unlockView()
	95	+	s.lockView()
	96	+	defer s.unlockView()
81	97		if s.epoch == 0 {
82	98		s.epoch = 1
83	99		}

@@ -113,8 +129,8 @@ func (p *preparedOutput) send(data []byte, echoAck uint64, send func(ports.Frame
113	129		if p == nil \|\| p.stream == nil \|\| p.attempted {
114	130		return nil
115	131		}
116		-	unlockView := p.stream.lockView()
117		-	defer unlockView()
	132	+	p.stream.lockView()
	133	+	defer p.stream.unlockView()
118	134		p.attempted = true
119	135		if !p.currentLocked() {
120	136		return nil

@@ -148,8 +164,8 @@ func (p *preparedOutput) commitNoSend() {
148	164		if p == nil \|\| p.stream == nil \|\| p.attempted {
149	165		return
150	166		}
151		-	unlockView := p.stream.lockView()
152		-	defer unlockView()
	167	+	p.stream.lockView()
	168	+	defer p.stream.unlockView()
153	169		p.attempted = true
154	170		if !p.currentLocked() \|\| len(p.data) != 0 {
155	171		return

@@ -164,8 +180,8 @@ func (p *preparedOutput) commit() {
164	180		}
165	181
166	182		func (s *outputStateStream) currentEpoch() uint64 {
167		-	unlockView := s.lockView()
168		-	defer unlockView()
	183	+	s.lockView()
	184	+	defer s.unlockView()
169	185		return s.currentEpochLocked()
170	186		}
171	187

@@ -218,8 +234,8 @@ func marshalOutputState(data []byte, epoch, base, next, echoAck, viewRevision ui
218	234		// an attachment is present, callers must hold its sendMu; attachment-bound
219	235		// sends use sideEffectLocked after validating their transport incarnation.
220	236		func (s *outputStateStream) sideEffect(data []byte, echoAck uint64) (ports.Frame, error) {
221		-	unlockView := s.lockView()
222		-	defer unlockView()
	237	+	s.lockView()
	238	+	defer s.unlockView()
223	239		return s.sideEffectLocked(data, echoAck)
224	240		}
225	241

@@ -239,8 +255,8 @@ func (s *outputStateStream) ack(epoch, state uint64) {
239	255		if s == nil {
240	256		return
241	257		}
242		-	unlockView := s.lockView()
243		-	defer unlockView()
	258	+	s.lockView()
	259	+	defer s.unlockView()
244	260		if epoch != s.currentEpochLocked() \|\| state > s.next \|\| state <= s.acked {
245	261		return
246	262		}

@@ -252,8 +268,8 @@ func (s *outputStateStream) rebase() {
252	268		if s == nil {
253	269		return
254	270		}
255		-	unlockView := s.lockView()
256		-	defer unlockView()
	271	+	s.lockView()
	272	+	defer s.unlockView()
257	273		s.rebaseLocked()
258	274		}
259	275

@@ -291,8 +307,8 @@ func (s *outputStateStream) syncCapacityLocked() {
291	307		if s == nil {
292	308		return
293	309		}
294		-	unlockView := s.lockView()
295		-	defer unlockView()
	310	+	s.lockView()
	311	+	defer s.unlockView()
296	312		s.maxOutstandingAtomic.Store(s.maxOutstanding)
297	313		s.publishOutstanding()
298	314		}

@@ -308,7 +324,7 @@ func (s *outputStateStream) outstanding() uint64 {
308	324		if s == nil {
309	325		return 0
310	326		}
311		-	unlockView := s.lockView()
312		-	defer unlockView()
	327	+	s.lockView()
	328	+	defer s.unlockView()
313	329		return s.next - s.acked
314	330		}

 import (
 	"errors"
+	"sync"
 	"sync/atomic"
 	"github.com/bnema/vev/internal/domain"
 	forceSnapshot        bool
 	initialized          bool
 	attachment           *attachedClient
+	stateMu              sync.Mutex
 }
 // lockView serializes an attachment's view publication with every output
 // state transition. It deliberately does not take sendMu: callers that hold
 // architecture locks can publish a view without reversing the paint path's
 // sendMu -> session lock order.
-func (s *outputStateStream) lockView() func() {
+func (s *outputStateStream) lockView() {
-	if s == nil || s.attachment == nil {
+	if s == nil {
-		return func() {}
+		return
+	}
+	if s.attachment == nil {
+		s.stateMu.Lock()
+		return
 	}
 	s.attachment.viewMu.Lock()
-	return s.attachment.viewMu.Unlock
+}
+func (s *outputStateStream) unlockView() {
+	if s == nil {
+		return
+	}
+	if s.attachment == nil {
+		s.stateMu.Unlock()
+		return
+	}
+	s.attachment.viewMu.Unlock()
 }
 func newOutputStateStream(windowSize ...uint8) *outputStateStream {
 }
 func (s *outputStateStream) prepare(frame renderer.Frame, damage []renderer.Damage, reset bool) (*preparedOutput, error) {
-	unlockView := s.lockView()
+	s.lockView()
-	defer unlockView()
+	defer s.unlockView()
 	if s.epoch == 0 {
 		s.epoch = 1
 	}
 	if p == nil || p.stream == nil || p.attempted {
 		return nil
 	}
-	unlockView := p.stream.lockView()
+	p.stream.lockView()
-	defer unlockView()
+	defer p.stream.unlockView()
 	p.attempted = true
 	if !p.currentLocked() {
 		return nil
 	if p == nil || p.stream == nil || p.attempted {
 		return
 	}
-	unlockView := p.stream.lockView()
+	p.stream.lockView()
-	defer unlockView()
+	defer p.stream.unlockView()
 	p.attempted = true
 	if !p.currentLocked() || len(p.data) != 0 {
 		return
 }
 func (s *outputStateStream) currentEpoch() uint64 {
-	unlockView := s.lockView()
+	s.lockView()
-	defer unlockView()
+	defer s.unlockView()
 	return s.currentEpochLocked()
 }
 // an attachment is present, callers must hold its sendMu; attachment-bound
 // sends use sideEffectLocked after validating their transport incarnation.
 func (s *outputStateStream) sideEffect(data []byte, echoAck uint64) (ports.Frame, error) {
-	unlockView := s.lockView()
+	s.lockView()
-	defer unlockView()
+	defer s.unlockView()
 	return s.sideEffectLocked(data, echoAck)
 }
 	if s == nil {
 		return
 	}
-	unlockView := s.lockView()
+	s.lockView()
-	defer unlockView()
+	defer s.unlockView()
 	if epoch != s.currentEpochLocked() || state > s.next || state <= s.acked {
 		return
 	}
 	if s == nil {
 		return
 	}
-	unlockView := s.lockView()
+	s.lockView()
-	defer unlockView()
+	defer s.unlockView()
 	s.rebaseLocked()
 }
 	if s == nil {
 		return
 	}
-	unlockView := s.lockView()
+	s.lockView()
-	defer unlockView()
+	defer s.unlockView()
 	s.maxOutstandingAtomic.Store(s.maxOutstanding)
 	s.publishOutstanding()
 }
 	if s == nil {
 		return 0
 	}
-	unlockView := s.lockView()
+	s.lockView()
-	defer unlockView()
+	defer s.unlockView()
 	return s.next - s.acked
 }