4c3ccc7 feat(daemon): fence per-attachment output epochs · vev-vt

M internal/usecase/daemon/client.go +7 -4


@@ -289,10 +289,12 @@ func (ac *attachedClient) currentTransportIs(tr ports.Transport) bool {
289	289		return tr != nil && ac.transportIs(tr)
290	290		}
291	291
292		-	func (ac *attachedClient) ackOutputState(state uint64) {
	292	+	func (ac *attachedClient) ackOutputState(values ...uint64) {
293	293		ac.sendMu.Lock()
294	294		defer ac.sendMu.Unlock()
295		-	ac.output.ack(state)
	295	+	if ac.output != nil {
	296	+	ac.output.ack(values...)
	297	+	}
296	298		}
297	299
298	300		// ensureScreenOutput is called while sendMu owns the attachment. Production

@@ -317,8 +319,8 @@ func (ac *attachedClient) discardProxyCapture() {
317	319		}
318	320		}
319	321
320		-	// rebaseOutput resets both wire representations while retaining the shared
321		-	// state-number chain. Callers hold sendMu (or the activation barrier).
	322	+	// rebaseOutput resets only this attachment's output representations. Callers
	323	+	// hold sendMu (or the activation barrier).
322	324		func (ac *attachedClient) rebaseOutput() {
323	325		if ac == nil {
324	326		return

@@ -572,6 +574,7 @@ func (d *Daemon) prepareAttachedClientLocked(tr ports.Transport, sz domain.Size,
572	574		resumeToken: resumeToken,
573	575		proxied: opts.proxied,
574	576		}
	577	+	output.attachment = ac
575	578		if opts.proxied {
576	579		ac.screenOutput = newStructuredOutputStream(output)
577	580		}

 	return tr != nil && ac.transportIs(tr)
 }
-func (ac *attachedClient) ackOutputState(state uint64) {
+func (ac *attachedClient) ackOutputState(values ...uint64) {
 	ac.sendMu.Lock()
 	defer ac.sendMu.Unlock()
-	ac.output.ack(state)
+	if ac.output != nil {
+		ac.output.ack(values...)
+	}
 }
 // ensureScreenOutput is called while sendMu owns the attachment. Production
 	}
 }
-// rebaseOutput resets both wire representations while retaining the shared
+// rebaseOutput resets only this attachment's output representations. Callers
-// state-number chain. Callers hold sendMu (or the activation barrier).
+// hold sendMu (or the activation barrier).
 func (ac *attachedClient) rebaseOutput() {
 	if ac == nil {
 		return
 		resumeToken:   resumeToken,
 		proxied:       opts.proxied,
 	}
+	output.attachment = ac
 	if opts.proxied {
 		ac.screenOutput = newStructuredOutputStream(output)
 	}

M internal/usecase/daemon/client_frame_routing.go +3 -3


@@ -129,7 +129,7 @@ func (d *Daemon) handleAttachmentClientFrame(token attachmentConnectionToken, f
129	129		}
130	130		case ports.MsgAck:
131	131		if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil {
132		-	d.ackOutput(token, ack.AckedStateNum)
	132	+	d.ackOutput(token, ack.Epoch, ack.State)
133	133		}
134	134		case ports.MsgOutputResetRequest:
135	135		if _, derr := ports.UnmarshalOutputResetRequest(f.Payload); derr == nil {

@@ -263,13 +263,13 @@ func (d *Daemon) resetOutput(token attachmentConnectionToken) bool {
263	263		})
264	264		}
265	265
266		-	func (d *Daemon) ackOutput(token attachmentConnectionToken, state uint64) bool {
	266	+	func (d *Daemon) ackOutput(token attachmentConnectionToken, values ...uint64) bool {
267	267		ac := token.ac
268	268		if token.effect == nil \|\| token.effect.ended.Load() {
269	269		return false
270	270		}
271	271		ac.sendMu.Lock()
272		-	ac.output.ack(state)
	272	+	ac.output.ack(values...)
273	273		ac.sendMu.Unlock()
274	274		if rc := token.sess.core().coordinator.Load(); rc != nil {
275	275		rc.notifyAckForLease(token.lease)


@@ -129,7 +129,7 @@ func (d *Daemon) handleAttachmentClientFrame(token attachmentConnectionToken, f
129		}	129		}
130		case ports.MsgAck:	130		case ports.MsgAck:
131		if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil {	131		if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil {
132	-	d.ackOutput(token, ack.AckedStateNum)	132	+	d.ackOutput(token, ack.Epoch, ack.State)
133		}	133		}
134		case ports.MsgOutputResetRequest:	134		case ports.MsgOutputResetRequest:
135		if _, derr := ports.UnmarshalOutputResetRequest(f.Payload); derr == nil {	135		if _, derr := ports.UnmarshalOutputResetRequest(f.Payload); derr == nil {

@@ -263,13 +263,13 @@ func (d *Daemon) resetOutput(token attachmentConnectionToken) bool {
263		})	263		})
264		}	264		}
265			265
266	-	func (d *Daemon) ackOutput(token attachmentConnectionToken, state uint64) bool {	266	+	func (d *Daemon) ackOutput(token attachmentConnectionToken, values ...uint64) bool {
267		ac := token.ac	267		ac := token.ac
268		if token.effect == nil \|\| token.effect.ended.Load() {	268		if token.effect == nil \|\| token.effect.ended.Load() {
269		return false	269		return false
270		}	270		}
271		ac.sendMu.Lock()	271		ac.sendMu.Lock()
272	-	ac.output.ack(state)	272	+	ac.output.ack(values...)
273		ac.sendMu.Unlock()	273		ac.sendMu.Unlock()
274		if rc := token.sess.core().coordinator.Load(); rc != nil {	274		if rc := token.sess.core().coordinator.Load(); rc != nil {
275		rc.notifyAckForLease(token.lease)	275		rc.notifyAckForLease(token.lease)

M internal/usecase/daemon/output_state_stream.go +139 -23

@@ -1,8 +1,10 @@
1	1		package daemon
2	2
3	3		import (
	4	+	"errors"
4	5		"sync/atomic"
5	6
	7	+	"github.com/bnema/vev/internal/domain"
6	8		"github.com/bnema/vev/internal/ports"
7	9		"github.com/bnema/vev/pkg/renderer"
8	10		)

@@ -11,19 +13,23 @@ import (
11	13		// attached client. Each incremental output is rendered from the preceding
12	14		// emitted frame and therefore remains pipelineable without waiting for an
13	15		// ACK. Cumulative ACKs advance the bounded paint window.
14		-	// A reset frame is the only dependency-free output and advertises state 0 as
15		-	// its base.
	16	+	// A reset frame starts a new epoch, clears the dependency chain, and advertises
	17	+	// state 0 as its base.
16	18		//
17	19		// Callers serialize access with attachedClient.sendMu. A prepared transaction
18	20		// must be sent or committed as a no-op before the next transaction is prepared.
19	21		type outputStateStream struct {
20	22		renderer *renderer.Renderer
	23	+	epoch uint64
21	24		next uint64
22	25		acked uint64
	26	+	generation uint64
23	27		maxOutstanding uint64
24	28		outstandingAtomic atomic.Uint64
25	29		maxOutstandingAtomic atomic.Uint64
26	30		forceSnapshot bool
	31	+	initialized bool
	32	+	attachment *attachedClient
27	33		}
28	34
29	35		func newOutputStateStream(windowSize ...uint8) *outputStateStream {

@@ -33,6 +39,7 @@ func newOutputStateStream(windowSize ...uint8) *outputStateStream {
33	39		}
34	40		stream := &outputStateStream{
35	41		renderer: renderer.New(renderer.Capabilities{}),
	42	+	epoch: 1,
36	43		maxOutstanding: uint64(window),
37	44		}
38	45		stream.maxOutstandingAtomic.Store(uint64(window))

@@ -41,39 +48,71 @@ func newOutputStateStream(windowSize ...uint8) *outputStateStream {
41	48		}
42	49
43	50		type preparedOutput struct {
44		-	stream *outputStateStream
45		-	draw renderer.PreparedDraw
46		-	next uint64
47		-	data []byte
48		-	reset bool
49		-	attempted bool
	51	+	stream *outputStateStream
	52	+	draw renderer.PreparedDraw
	53	+	base uint64
	54	+	next uint64
	55	+	epoch uint64
	56	+	generation uint64
	57	+	connectionGeneration uint64
	58	+	viewRevision uint64
	59	+	size domain.Size
	60	+	data []byte
	61	+	reset bool
	62	+	attempted bool
	63	+	sent bool
50	64		}
51	65
52	66		func (s outputStateStream) prepare(frame renderer.Frame, damage []renderer.Damage, reset bool) (preparedOutput, error) {
53		-	reset = reset \|\| s.forceSnapshot
	67	+	if s.epoch == 0 {
	68	+	s.epoch = 1
	69	+	}
	70	+	if reset && s.initialized {
	71	+	s.rebase()
	72	+	}
	73	+	reset = reset \|\| s.forceSnapshot \|\| !s.initialized
	74	+	if s.next == ^uint64(0) {
	75	+	return nil, errors.New("output state number exhausted")
	76	+	}
54	77		draw, err := s.renderer.Prepare(frame, damage, reset)
55	78		if err != nil {
56	79		return nil, err
57	80		}
	81	+	s.initialized = true
	82	+	generation, epoch, viewRevision := s.fence()
58	83		return &preparedOutput{
59		-	stream: s,
60		-	draw: draw,
61		-	next: s.next + 1,
62		-	data: draw.Bytes(),
63		-	reset: reset,
	84	+	stream: s,
	85	+	draw: draw,
	86	+	base: s.next,
	87	+	next: s.next + 1,
	88	+	epoch: epoch,
	89	+	generation: s.generation,
	90	+	connectionGeneration: generation,
	91	+	viewRevision: viewRevision,
	92	+	size: domain.Size{Cols: frame.Width, Rows: frame.Height},
	93	+	data: draw.Bytes(),
	94	+	reset: reset,
64	95		}, nil
65	96		}
66	97
67	98		func (p *preparedOutput) send(data []byte, echoAck uint64, send func(ports.Frame) error) error {
68		-	if p.attempted {
	99	+	if p == nil \|\| p.attempted {
69	100		return nil
70	101		}
71	102		p.attempted = true
72		-	base := p.stream.next
	103	+	if !p.current() {
	104	+	return nil
	105	+	}
	106	+	if len(data) == 0 {
	107	+	p.commit()
	108	+	p.sent = true
	109	+	return nil
	110	+	}
	111	+	base := p.base
73	112		if p.reset {
74	113		base = 0
75	114		}
76		-	out := frameOutputState(data, base, p.next, echoAck)
	115	+	out := marshalOutputState(data, p.epoch, base, p.next, echoAck, p.viewRevision, p.size, p.reset)
77	116		if err := send(out); err != nil {
78	117		p.stream.forceSnapshot = true
79	118		return err

@@ -81,15 +120,20 @@ func (p *preparedOutput) send(data []byte, echoAck uint64, send func(ports.Frame
81	120		p.commit()
82	121		p.stream.next = p.next
83	122		p.stream.publishOutstanding()
	123	+	p.sent = true
84	124		return nil
85	125		}
86	126
87	127		func (p *preparedOutput) commitNoSend() {
88		-	if p == nil \|\| p.attempted \|\| len(p.data) != 0 {
	128	+	if p == nil \|\| p.attempted {
89	129		return
90	130		}
91	131		p.attempted = true
	132	+	if !p.current() \|\| len(p.data) != 0 {
	133	+	return
	134	+	}
92	135		p.commit()
	136	+	p.sent = true
93	137		}
94	138
95	139		func (p *preparedOutput) commit() {

@@ -97,12 +141,73 @@ func (p *preparedOutput) commit() {
97	141		p.stream.forceSnapshot = false
98	142		}
99	143
	144	+	func (s *outputStateStream) currentEpoch() uint64 {
	145	+	if s.epoch == 0 {
	146	+	s.epoch = 1
	147	+	}
	148	+	return s.epoch
	149	+	}
	150	+
	151	+	func (s *outputStateStream) fence() (uint64, uint64, uint64) {
	152	+	if s == nil {
	153	+	return 0, 1, 0
	154	+	}
	155	+	generation, revision := uint64(0), uint64(0)
	156	+	if s.attachment != nil {
	157	+	generation = s.attachment.connectionGeneration.Load()
	158	+	revision = s.attachment.viewSnapshot().revision
	159	+	}
	160	+	return generation, s.currentEpoch(), revision
	161	+	}
	162	+
	163	+	func (s *outputStateStream) preparedCurrent(generation, epoch, viewRevision, base uint64) bool {
	164	+	if s == nil \|\| s.currentEpoch() != epoch \|\| s.generation != generation \|\| s.next != base {
	165	+	return false
	166	+	}
	167	+	if s.attachment == nil {
	168	+	return true
	169	+	}
	170	+	return s.attachment.viewSnapshot().revision == viewRevision
	171	+	}
	172	+
	173	+	func (p *preparedOutput) current() bool {
	174	+	return p != nil && p.stream != nil && p.stream.preparedCurrent(p.generation, p.epoch, p.viewRevision, p.base) &&
	175	+	(p.stream.attachment == nil \|\| p.stream.attachment.connectionGeneration.Load() == p.connectionGeneration)
	176	+	}
	177	+
	178	+	func marshalOutputState(data []byte, epoch, base, next, echoAck, viewRevision uint64, size domain.Size, full bool) ports.Frame {
	179	+	if !size.Valid() {
	180	+	size = domain.Size{Cols: 1, Rows: 1}
	181	+	}
	182	+	return ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{
	183	+	Epoch: epoch, Base: base, New: next, Echo: echoAck, ViewRevision: viewRevision,
	184	+	Size: size, Full: full, Data: data,
	185	+	})}
	186	+	}
	187	+
100	188		func (s *outputStateStream) sideEffect(data []byte, echoAck uint64) ports.Frame {
101		-	return frameOutputState(data, 0, 0, echoAck)
	189	+	_, epoch, viewRevision := s.fence()
	190	+	size := domain.Size{Cols: 1, Rows: 1}
	191	+	if s != nil && s.attachment != nil && s.attachment.size.Valid() {
	192	+	size = s.attachment.size
	193	+	}
	194	+	return marshalOutputState(data, epoch, 0, 0, echoAck, viewRevision, size, false)
102	195		}
103	196
104		-	func (s *outputStateStream) ack(state uint64) {
105		-	if state > s.next \|\| state <= s.acked {
	197	+	func (s *outputStateStream) ack(values ...uint64) {
	198	+	if s == nil {
	199	+	return
	200	+	}
	201	+	epoch, state := s.currentEpoch(), uint64(0)
	202	+	switch len(values) {
	203	+	case 1:
	204	+	state = values[0]
	205	+	case 2:
	206	+	epoch, state = values[0], values[1]
	207	+	default:
	208	+	return
	209	+	}
	210	+	if epoch != s.currentEpoch() \|\| state > s.next \|\| state <= s.acked {
106	211		return
107	212		}
108	213		s.acked = state

@@ -110,11 +215,22 @@ func (s *outputStateStream) ack(state uint64) {
110	215		}
111	216
112	217		func (s *outputStateStream) rebase() {
113		-	s.acked = s.next
114		-	s.publishOutstanding()
	218	+	if s == nil {
	219	+	return
	220	+	}
	221	+	epoch := s.currentEpoch()
	222	+	if epoch == ^uint64(0) {
	223	+	return
	224	+	}
	225	+	s.epoch = epoch + 1
	226	+	s.next = 0
	227	+	s.acked = 0
	228	+	s.generation++
115	229		s.maxOutstandingAtomic.Store(s.maxOutstanding)
	230	+	s.publishOutstanding()
116	231		s.renderer.Reset()
117	232		s.forceSnapshot = true
	233	+	s.initialized = false
118	234		}
119	235
120	236		func (s *outputStateStream) publishOutstanding() {

@@ -1,8 +1,10 @@
1		package daemon	1		package daemon
2			2
3		import (	3		import (
			4	+	"errors"
4		"sync/atomic"	5		"sync/atomic"
5			6
			7	+	"github.com/bnema/vev/internal/domain"
6		"github.com/bnema/vev/internal/ports"	8		"github.com/bnema/vev/internal/ports"
7		"github.com/bnema/vev/pkg/renderer"	9		"github.com/bnema/vev/pkg/renderer"
8		)	10		)

@@ -11,19 +13,23 @@ import (
11		// attached client. Each incremental output is rendered from the preceding	13		// attached client. Each incremental output is rendered from the preceding
12		// emitted frame and therefore remains pipelineable without waiting for an	14		// emitted frame and therefore remains pipelineable without waiting for an
13		// ACK. Cumulative ACKs advance the bounded paint window.	15		// ACK. Cumulative ACKs advance the bounded paint window.
14	-	// A reset frame is the only dependency-free output and advertises state 0 as	16	+	// A reset frame starts a new epoch, clears the dependency chain, and advertises
15	-	// its base.	17	+	// state 0 as its base.
16		//	18		//
17		// Callers serialize access with attachedClient.sendMu. A prepared transaction	19		// Callers serialize access with attachedClient.sendMu. A prepared transaction
18		// must be sent or committed as a no-op before the next transaction is prepared.	20		// must be sent or committed as a no-op before the next transaction is prepared.
19		type outputStateStream struct {	21		type outputStateStream struct {
20		renderer *renderer.Renderer	22		renderer *renderer.Renderer
			23	+	epoch uint64
21		next uint64	24		next uint64
22		acked uint64	25		acked uint64
			26	+	generation uint64
23		maxOutstanding uint64	27		maxOutstanding uint64
24		outstandingAtomic atomic.Uint64	28		outstandingAtomic atomic.Uint64
25		maxOutstandingAtomic atomic.Uint64	29		maxOutstandingAtomic atomic.Uint64
26		forceSnapshot bool	30		forceSnapshot bool
			31	+	initialized bool
			32	+	attachment *attachedClient
27		}	33		}
28			34
29		func newOutputStateStream(windowSize ...uint8) *outputStateStream {	35		func newOutputStateStream(windowSize ...uint8) *outputStateStream {

@@ -33,6 +39,7 @@ func newOutputStateStream(windowSize ...uint8) *outputStateStream {
33		}	39		}
34		stream := &outputStateStream{	40		stream := &outputStateStream{
35		renderer: renderer.New(renderer.Capabilities{}),	41		renderer: renderer.New(renderer.Capabilities{}),
			42	+	epoch: 1,
36		maxOutstanding: uint64(window),	43		maxOutstanding: uint64(window),
37		}	44		}
38		stream.maxOutstandingAtomic.Store(uint64(window))	45		stream.maxOutstandingAtomic.Store(uint64(window))

@@ -41,39 +48,71 @@ func newOutputStateStream(windowSize ...uint8) *outputStateStream {
41		}	48		}
42			49
43		type preparedOutput struct {	50		type preparedOutput struct {
44	-	stream *outputStateStream	51	+	stream *outputStateStream
45	-	draw renderer.PreparedDraw	52	+	draw renderer.PreparedDraw
46	-	next uint64	53	+	base uint64
47	-	data []byte	54	+	next uint64
48	-	reset bool	55	+	epoch uint64
49	-	attempted bool	56	+	generation uint64
			57	+	connectionGeneration uint64
			58	+	viewRevision uint64
			59	+	size domain.Size
			60	+	data []byte
			61	+	reset bool
			62	+	attempted bool
			63	+	sent bool
50		}	64		}
51			65
52		func (s outputStateStream) prepare(frame renderer.Frame, damage []renderer.Damage, reset bool) (preparedOutput, error) {	66		func (s outputStateStream) prepare(frame renderer.Frame, damage []renderer.Damage, reset bool) (preparedOutput, error) {
53	-	reset = reset \|\| s.forceSnapshot	67	+	if s.epoch == 0 {
			68	+	s.epoch = 1
			69	+	}
			70	+	if reset && s.initialized {
			71	+	s.rebase()
			72	+	}
			73	+	reset = reset \|\| s.forceSnapshot \|\| !s.initialized
			74	+	if s.next == ^uint64(0) {
			75	+	return nil, errors.New("output state number exhausted")
			76	+	}
54		draw, err := s.renderer.Prepare(frame, damage, reset)	77		draw, err := s.renderer.Prepare(frame, damage, reset)
55		if err != nil {	78		if err != nil {
56		return nil, err	79		return nil, err
57		}	80		}
			81	+	s.initialized = true
			82	+	generation, epoch, viewRevision := s.fence()
58		return &preparedOutput{	83		return &preparedOutput{
59	-	stream: s,	84	+	stream: s,
60	-	draw: draw,	85	+	draw: draw,
61	-	next: s.next + 1,	86	+	base: s.next,
62	-	data: draw.Bytes(),	87	+	next: s.next + 1,
63	-	reset: reset,	88	+	epoch: epoch,
			89	+	generation: s.generation,
			90	+	connectionGeneration: generation,
			91	+	viewRevision: viewRevision,
			92	+	size: domain.Size{Cols: frame.Width, Rows: frame.Height},
			93	+	data: draw.Bytes(),
			94	+	reset: reset,
64		}, nil	95		}, nil
65		}	96		}
66			97
67		func (p *preparedOutput) send(data []byte, echoAck uint64, send func(ports.Frame) error) error {	98		func (p *preparedOutput) send(data []byte, echoAck uint64, send func(ports.Frame) error) error {
68	-	if p.attempted {	99	+	if p == nil \|\| p.attempted {
69		return nil	100		return nil
70		}	101		}
71		p.attempted = true	102		p.attempted = true
72	-	base := p.stream.next	103	+	if !p.current() {
			104	+	return nil
			105	+	}
			106	+	if len(data) == 0 {
			107	+	p.commit()
			108	+	p.sent = true
			109	+	return nil
			110	+	}
			111	+	base := p.base
73		if p.reset {	112		if p.reset {
74		base = 0	113		base = 0
75		}	114		}
76	-	out := frameOutputState(data, base, p.next, echoAck)	115	+	out := marshalOutputState(data, p.epoch, base, p.next, echoAck, p.viewRevision, p.size, p.reset)
77		if err := send(out); err != nil {	116		if err := send(out); err != nil {
78		p.stream.forceSnapshot = true	117		p.stream.forceSnapshot = true
79		return err	118		return err

@@ -81,15 +120,20 @@ func (p *preparedOutput) send(data []byte, echoAck uint64, send func(ports.Frame
81		p.commit()	120		p.commit()
82		p.stream.next = p.next	121		p.stream.next = p.next
83		p.stream.publishOutstanding()	122		p.stream.publishOutstanding()
			123	+	p.sent = true
84		return nil	124		return nil
85		}	125		}
86			126
87		func (p *preparedOutput) commitNoSend() {	127		func (p *preparedOutput) commitNoSend() {
88	-	if p == nil \|\| p.attempted \|\| len(p.data) != 0 {	128	+	if p == nil \|\| p.attempted {
89		return	129		return
90		}	130		}
91		p.attempted = true	131		p.attempted = true
			132	+	if !p.current() \|\| len(p.data) != 0 {
			133	+	return
			134	+	}
92		p.commit()	135		p.commit()
			136	+	p.sent = true
93		}	137		}
94			138
95		func (p *preparedOutput) commit() {	139		func (p *preparedOutput) commit() {

@@ -97,12 +141,73 @@ func (p *preparedOutput) commit() {
97		p.stream.forceSnapshot = false	141		p.stream.forceSnapshot = false
98		}	142		}
99			143
			144	+	func (s *outputStateStream) currentEpoch() uint64 {
			145	+	if s.epoch == 0 {
			146	+	s.epoch = 1
			147	+	}
			148	+	return s.epoch
			149	+	}
			150	+
			151	+	func (s *outputStateStream) fence() (uint64, uint64, uint64) {
			152	+	if s == nil {
			153	+	return 0, 1, 0
			154	+	}
			155	+	generation, revision := uint64(0), uint64(0)
			156	+	if s.attachment != nil {
			157	+	generation = s.attachment.connectionGeneration.Load()
			158	+	revision = s.attachment.viewSnapshot().revision
			159	+	}
			160	+	return generation, s.currentEpoch(), revision
			161	+	}
			162	+
			163	+	func (s *outputStateStream) preparedCurrent(generation, epoch, viewRevision, base uint64) bool {
			164	+	if s == nil \|\| s.currentEpoch() != epoch \|\| s.generation != generation \|\| s.next != base {
			165	+	return false
			166	+	}
			167	+	if s.attachment == nil {
			168	+	return true
			169	+	}
			170	+	return s.attachment.viewSnapshot().revision == viewRevision
			171	+	}
			172	+
			173	+	func (p *preparedOutput) current() bool {
			174	+	return p != nil && p.stream != nil && p.stream.preparedCurrent(p.generation, p.epoch, p.viewRevision, p.base) &&
			175	+	(p.stream.attachment == nil \|\| p.stream.attachment.connectionGeneration.Load() == p.connectionGeneration)
			176	+	}
			177	+
			178	+	func marshalOutputState(data []byte, epoch, base, next, echoAck, viewRevision uint64, size domain.Size, full bool) ports.Frame {
			179	+	if !size.Valid() {
			180	+	size = domain.Size{Cols: 1, Rows: 1}
			181	+	}
			182	+	return ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{
			183	+	Epoch: epoch, Base: base, New: next, Echo: echoAck, ViewRevision: viewRevision,
			184	+	Size: size, Full: full, Data: data,
			185	+	})}
			186	+	}
			187	+
100		func (s *outputStateStream) sideEffect(data []byte, echoAck uint64) ports.Frame {	188		func (s *outputStateStream) sideEffect(data []byte, echoAck uint64) ports.Frame {
101	-	return frameOutputState(data, 0, 0, echoAck)	189	+	_, epoch, viewRevision := s.fence()
			190	+	size := domain.Size{Cols: 1, Rows: 1}
			191	+	if s != nil && s.attachment != nil && s.attachment.size.Valid() {
			192	+	size = s.attachment.size
			193	+	}
			194	+	return marshalOutputState(data, epoch, 0, 0, echoAck, viewRevision, size, false)
102		}	195		}
103			196
104	-	func (s *outputStateStream) ack(state uint64) {	197	+	func (s *outputStateStream) ack(values ...uint64) {
105	-	if state > s.next \|\| state <= s.acked {	198	+	if s == nil {
			199	+	return
			200	+	}
			201	+	epoch, state := s.currentEpoch(), uint64(0)
			202	+	switch len(values) {
			203	+	case 1:
			204	+	state = values[0]
			205	+	case 2:
			206	+	epoch, state = values[0], values[1]
			207	+	default:
			208	+	return
			209	+	}
			210	+	if epoch != s.currentEpoch() \|\| state > s.next \|\| state <= s.acked {
106		return	211		return
107		}	212		}
108		s.acked = state	213		s.acked = state

@@ -110,11 +215,22 @@ func (s *outputStateStream) ack(state uint64) {
110		}	215		}
111			216
112		func (s *outputStateStream) rebase() {	217		func (s *outputStateStream) rebase() {
113	-	s.acked = s.next	218	+	if s == nil {
114	-	s.publishOutstanding()	219	+	return
			220	+	}
			221	+	epoch := s.currentEpoch()
			222	+	if epoch == ^uint64(0) {
			223	+	return
			224	+	}
			225	+	s.epoch = epoch + 1
			226	+	s.next = 0
			227	+	s.acked = 0
			228	+	s.generation++
115		s.maxOutstandingAtomic.Store(s.maxOutstanding)	229		s.maxOutstandingAtomic.Store(s.maxOutstanding)
			230	+	s.publishOutstanding()
116		s.renderer.Reset()	231		s.renderer.Reset()
117		s.forceSnapshot = true	232		s.forceSnapshot = true
			233	+	s.initialized = false
118		}	234		}
119			235
120		func (s *outputStateStream) publishOutstanding() {	236		func (s *outputStateStream) publishOutstanding() {

M internal/usecase/daemon/output_state_stream_test.go +124 -0


@@ -5,6 +5,7 @@ import (
5	5		"runtime"
6	6		"testing"
7	7
	8	+	"github.com/bnema/vev/internal/domain"
8	9		"github.com/bnema/vev/internal/ports"
9	10		"github.com/bnema/vev/pkg/renderer"
10	11		"github.com/bnema/vev/pkg/vt"

@@ -74,6 +75,129 @@ func TestOutputStateStreamCapacityProbeDoesNotRaceWithSend(t *testing.T) {
74	75		require.False(t, stream.atCapacity())
75	76		}
76	77
	78	+	func TestOutputStateStreamEpochsAreAttachmentLocalAndPreparedFramesAreFenced(t *testing.T) {
	79	+	frame := renderer.NewFrame(3, 1)
	80	+	fillOutputStateRows(frame, []string{"abc"})
	81	+	newAttachment := func() *outputStateStream {
	82	+	stream := newOutputStateStream()
	83	+	ac := &attachedClient{output: stream, size: domain.Size{Cols: 3, Rows: 1}}
	84	+	stream.attachment = ac
	85	+	return stream
	86	+	}
	87	+
	88	+	first := newAttachment()
	89	+	second := newAttachment()
	90	+	firstPrepared, err := first.prepare(frame, nil, true)
	91	+	require.NoError(t, err)
	92	+	var firstFrame ports.Frame
	93	+	require.NoError(t, firstPrepared.send(firstPrepared.data, 0, func(frame ports.Frame) error {
	94	+	firstFrame = frame
	95	+	return nil
	96	+	}))
	97	+	firstOutput, err := ports.UnmarshalOutput(firstFrame.Payload)
	98	+	require.NoError(t, err)
	99	+	require.Equal(t, uint64(1), firstOutput.Epoch)
	100	+	require.Equal(t, uint64(1), firstOutput.New)
	101	+	require.True(t, firstOutput.Full)
	102	+	first.ack(firstOutput.Epoch, firstOutput.New)
	103	+
	104	+	secondPrepared, err := second.prepare(frame, nil, true)
	105	+	require.NoError(t, err)
	106	+	require.NoError(t, secondPrepared.send(secondPrepared.data, 0, func(ports.Frame) error { return nil }))
	107	+	require.Equal(t, uint64(1), second.epoch)
	108	+	require.Equal(t, uint64(1), second.next)
	109	+	require.Equal(t, uint64(1), first.next)
	110	+
	111	+	changed := frame.Clone()
	112	+	changed.Set(0, 0, renderer.Cell{Rune: 'x', Style: renderer.DefaultStyle()})
	113	+	pending, err := first.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
	114	+	require.NoError(t, err)
	115	+	first.rebase()
	116	+	require.Equal(t, uint64(2), first.epoch)
	117	+	require.Zero(t, first.next)
	118	+	require.Zero(t, first.acked)
	119	+	var staleSent bool
	120	+	require.NoError(t, pending.send(pending.data, 0, func(ports.Frame) error {
	121	+	staleSent = true
	122	+	return nil
	123	+	}))
	124	+	require.False(t, staleSent)
	125	+	require.Zero(t, first.next)
	126	+	first.ack(1, 1)
	127	+	require.Zero(t, first.acked, "an ACK from the retired epoch must not advance the new epoch")
	128	+
	129	+	reset, err := first.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
	130	+	require.NoError(t, err)
	131	+	var resetFrame ports.Frame
	132	+	require.NoError(t, reset.send(reset.data, 0, func(frame ports.Frame) error {
	133	+	resetFrame = frame
	134	+	return nil
	135	+	}))
	136	+	resetOutput, err := ports.UnmarshalOutput(resetFrame.Payload)
	137	+	require.NoError(t, err)
	138	+	require.Equal(t, uint64(2), resetOutput.Epoch)
	139	+	require.Zero(t, resetOutput.Base)
	140	+	require.Equal(t, uint64(1), resetOutput.New)
	141	+	require.True(t, resetOutput.Full)
	142	+	}
	143	+
	144	+	func TestPreparedOutputDropsReplacedConnectionAndView(t *testing.T) {
	145	+	for _, tt := range []struct {
	146	+	name string
	147	+	change func(*attachedClient)
	148	+	}{
	149	+	{name: "connection generation", change: func(ac *attachedClient) { ac.connectionGeneration.Add(1) }},
	150	+	{name: "view revision", change: func(ac *attachedClient) {
	151	+	ac.viewMu.Lock()
	152	+	ac.view.revision++
	153	+	ac.viewMu.Unlock()
	154	+	}},
	155	+	} {
	156	+	t.Run(tt.name, func(t *testing.T) {
	157	+	stream := newOutputStateStream()
	158	+	ac := &attachedClient{output: stream, size: domain.Size{Cols: 3, Rows: 1}}
	159	+	stream.attachment = ac
	160	+	frame := renderer.NewFrame(3, 1)
	161	+	fillOutputStateRows(frame, []string{"abc"})
	162	+	initial, err := stream.prepare(frame, nil, true)
	163	+	require.NoError(t, err)
	164	+	require.NoError(t, initial.send(initial.data, 0, func(ports.Frame) error { return nil }))
	165	+	changed := frame.Clone()
	166	+	changed.Set(0, 0, renderer.Cell{Rune: 'x', Style: renderer.DefaultStyle()})
	167	+	pending, err := stream.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
	168	+	require.NoError(t, err)
	169	+	tt.change(ac)
	170	+	var sent bool
	171	+	require.NoError(t, pending.send(pending.data, 0, func(ports.Frame) error {
	172	+	sent = true
	173	+	return nil
	174	+	}))
	175	+	require.False(t, sent)
	176	+	require.Equal(t, uint64(1), stream.next)
	177	+
	178	+	retry, err := stream.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
	179	+	require.NoError(t, err)
	180	+	require.NoError(t, retry.send(retry.data, 0, func(ports.Frame) error { return nil }))
	181	+	require.Equal(t, uint64(2), stream.next)
	182	+	})
	183	+	}
	184	+	}
	185	+
	186	+	func TestOutputStateSideEffectsDoNotAdvanceEpochStateOrACK(t *testing.T) {
	187	+	stream := newOutputStateStream()
	188	+	stream.ack(1, 0)
	189	+	beforeEpoch, beforeNext, beforeAck := stream.epoch, stream.next, stream.acked
	190	+	out, err := ports.UnmarshalOutput(stream.sideEffect([]byte("pty"), 9).Payload)
	191	+	require.NoError(t, err)
	192	+	require.Equal(t, beforeEpoch, out.Epoch)
	193	+	require.Zero(t, out.Base)
	194	+	require.Zero(t, out.New)
	195	+	require.False(t, out.Full)
	196	+	require.Equal(t, uint64(9), out.Echo)
	197	+	require.Equal(t, beforeNext, stream.next)
	198	+	require.Equal(t, beforeAck, stream.acked)
	199	+	}
	200	+
77	201		func TestOutputStateStreamFailedSendRetriesSnapshotWithoutAdvancing(t *testing.T) {
78	202		stream := newOutputStateStream()
79	203		initial := renderer.NewFrame(3, 1)


@@ -5,6 +5,7 @@ import (
5	"runtime"	5		"runtime"
6	"testing"	6		"testing"
7		7
		8	+	"github.com/bnema/vev/internal/domain"
8	"github.com/bnema/vev/internal/ports"	9		"github.com/bnema/vev/internal/ports"
9	"github.com/bnema/vev/pkg/renderer"	10		"github.com/bnema/vev/pkg/renderer"
10	"github.com/bnema/vev/pkg/vt"	11		"github.com/bnema/vev/pkg/vt"

@@ -74,6 +75,129 @@ func TestOutputStateStreamCapacityProbeDoesNotRaceWithSend(t *testing.T) {
74	require.False(t, stream.atCapacity())	75		require.False(t, stream.atCapacity())
75	}	76		}
76		77
		78	+	func TestOutputStateStreamEpochsAreAttachmentLocalAndPreparedFramesAreFenced(t *testing.T) {
		79	+	frame := renderer.NewFrame(3, 1)
		80	+	fillOutputStateRows(frame, []string{"abc"})
		81	+	newAttachment := func() *outputStateStream {
		82	+	stream := newOutputStateStream()
		83	+	ac := &attachedClient{output: stream, size: domain.Size{Cols: 3, Rows: 1}}
		84	+	stream.attachment = ac
		85	+	return stream
		86	+	}
		87	+
		88	+	first := newAttachment()
		89	+	second := newAttachment()
		90	+	firstPrepared, err := first.prepare(frame, nil, true)
		91	+	require.NoError(t, err)
		92	+	var firstFrame ports.Frame
		93	+	require.NoError(t, firstPrepared.send(firstPrepared.data, 0, func(frame ports.Frame) error {
		94	+	firstFrame = frame
		95	+	return nil
		96	+	}))
		97	+	firstOutput, err := ports.UnmarshalOutput(firstFrame.Payload)
		98	+	require.NoError(t, err)
		99	+	require.Equal(t, uint64(1), firstOutput.Epoch)
		100	+	require.Equal(t, uint64(1), firstOutput.New)
		101	+	require.True(t, firstOutput.Full)
		102	+	first.ack(firstOutput.Epoch, firstOutput.New)
		103	+
		104	+	secondPrepared, err := second.prepare(frame, nil, true)
		105	+	require.NoError(t, err)
		106	+	require.NoError(t, secondPrepared.send(secondPrepared.data, 0, func(ports.Frame) error { return nil }))
		107	+	require.Equal(t, uint64(1), second.epoch)
		108	+	require.Equal(t, uint64(1), second.next)
		109	+	require.Equal(t, uint64(1), first.next)
		110	+
		111	+	changed := frame.Clone()
		112	+	changed.Set(0, 0, renderer.Cell{Rune: 'x', Style: renderer.DefaultStyle()})
		113	+	pending, err := first.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
		114	+	require.NoError(t, err)
		115	+	first.rebase()
		116	+	require.Equal(t, uint64(2), first.epoch)
		117	+	require.Zero(t, first.next)
		118	+	require.Zero(t, first.acked)
		119	+	var staleSent bool
		120	+	require.NoError(t, pending.send(pending.data, 0, func(ports.Frame) error {
		121	+	staleSent = true
		122	+	return nil
		123	+	}))
		124	+	require.False(t, staleSent)
		125	+	require.Zero(t, first.next)
		126	+	first.ack(1, 1)
		127	+	require.Zero(t, first.acked, "an ACK from the retired epoch must not advance the new epoch")
		128	+
		129	+	reset, err := first.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
		130	+	require.NoError(t, err)
		131	+	var resetFrame ports.Frame
		132	+	require.NoError(t, reset.send(reset.data, 0, func(frame ports.Frame) error {
		133	+	resetFrame = frame
		134	+	return nil
		135	+	}))
		136	+	resetOutput, err := ports.UnmarshalOutput(resetFrame.Payload)
		137	+	require.NoError(t, err)
		138	+	require.Equal(t, uint64(2), resetOutput.Epoch)
		139	+	require.Zero(t, resetOutput.Base)
		140	+	require.Equal(t, uint64(1), resetOutput.New)
		141	+	require.True(t, resetOutput.Full)
		142	+	}
		143	+
		144	+	func TestPreparedOutputDropsReplacedConnectionAndView(t *testing.T) {
		145	+	for _, tt := range []struct {
		146	+	name string
		147	+	change func(*attachedClient)
		148	+	}{
		149	+	{name: "connection generation", change: func(ac *attachedClient) { ac.connectionGeneration.Add(1) }},
		150	+	{name: "view revision", change: func(ac *attachedClient) {
		151	+	ac.viewMu.Lock()
		152	+	ac.view.revision++
		153	+	ac.viewMu.Unlock()
		154	+	}},
		155	+	} {
		156	+	t.Run(tt.name, func(t *testing.T) {
		157	+	stream := newOutputStateStream()
		158	+	ac := &attachedClient{output: stream, size: domain.Size{Cols: 3, Rows: 1}}
		159	+	stream.attachment = ac
		160	+	frame := renderer.NewFrame(3, 1)
		161	+	fillOutputStateRows(frame, []string{"abc"})
		162	+	initial, err := stream.prepare(frame, nil, true)
		163	+	require.NoError(t, err)
		164	+	require.NoError(t, initial.send(initial.data, 0, func(ports.Frame) error { return nil }))
		165	+	changed := frame.Clone()
		166	+	changed.Set(0, 0, renderer.Cell{Rune: 'x', Style: renderer.DefaultStyle()})
		167	+	pending, err := stream.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
		168	+	require.NoError(t, err)
		169	+	tt.change(ac)
		170	+	var sent bool
		171	+	require.NoError(t, pending.send(pending.data, 0, func(ports.Frame) error {
		172	+	sent = true
		173	+	return nil
		174	+	}))
		175	+	require.False(t, sent)
		176	+	require.Equal(t, uint64(1), stream.next)
		177	+
		178	+	retry, err := stream.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
		179	+	require.NoError(t, err)
		180	+	require.NoError(t, retry.send(retry.data, 0, func(ports.Frame) error { return nil }))
		181	+	require.Equal(t, uint64(2), stream.next)
		182	+	})
		183	+	}
		184	+	}
		185	+
		186	+	func TestOutputStateSideEffectsDoNotAdvanceEpochStateOrACK(t *testing.T) {
		187	+	stream := newOutputStateStream()
		188	+	stream.ack(1, 0)
		189	+	beforeEpoch, beforeNext, beforeAck := stream.epoch, stream.next, stream.acked
		190	+	out, err := ports.UnmarshalOutput(stream.sideEffect([]byte("pty"), 9).Payload)
		191	+	require.NoError(t, err)
		192	+	require.Equal(t, beforeEpoch, out.Epoch)
		193	+	require.Zero(t, out.Base)
		194	+	require.Zero(t, out.New)
		195	+	require.False(t, out.Full)
		196	+	require.Equal(t, uint64(9), out.Echo)
		197	+	require.Equal(t, beforeNext, stream.next)
		198	+	require.Equal(t, beforeAck, stream.acked)
		199	+	}
		200	+
77	func TestOutputStateStreamFailedSendRetriesSnapshotWithoutAdvancing(t *testing.T) {	201		func TestOutputStateStreamFailedSendRetriesSnapshotWithoutAdvancing(t *testing.T) {
78	stream := newOutputStateStream()	202		stream := newOutputStateStream()
79	initial := renderer.NewFrame(3, 1)	203		initial := renderer.NewFrame(3, 1)

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


@@ -283,7 +283,7 @@ func TestOutputResetRebasesFullWindowAndSchedulesBaseZeroPaint(t *testing.T) {
283	283		require.Equal(t, ports.ScreenUpdateSnapshot, out.Kind)
284	284		require.Zero(t, out.BaseStateNum)
285	285		ac.sendMu.Lock()
286		-	require.Equal(t, ac.output.maxOutstanding, ac.output.acked, "reset must retire the previously full output window")
	286	+	require.Zero(t, ac.output.acked, "reset must clear the previous output window")
287	287		require.Equal(t, uint64(1), ac.output.outstanding(), "only the authoritative reset paint remains outstanding")
288	288		require.Equal(t, ac.output.next, out.NewStateNum)
289	289		ac.sendMu.Unlock()


@@ -283,7 +283,7 @@ func TestOutputResetRebasesFullWindowAndSchedulesBaseZeroPaint(t *testing.T) {
283		require.Equal(t, ports.ScreenUpdateSnapshot, out.Kind)	283		require.Equal(t, ports.ScreenUpdateSnapshot, out.Kind)
284		require.Zero(t, out.BaseStateNum)	284		require.Zero(t, out.BaseStateNum)
285		ac.sendMu.Lock()	285		ac.sendMu.Lock()
286	-	require.Equal(t, ac.output.maxOutstanding, ac.output.acked, "reset must retire the previously full output window")	286	+	require.Zero(t, ac.output.acked, "reset must clear the previous output window")
287		require.Equal(t, uint64(1), ac.output.outstanding(), "only the authoritative reset paint remains outstanding")	287		require.Equal(t, uint64(1), ac.output.outstanding(), "only the authoritative reset paint remains outstanding")
288		require.Equal(t, ac.output.next, out.NewStateNum)	288		require.Equal(t, ac.output.next, out.NewStateNum)
289		ac.sendMu.Unlock()	289		ac.sendMu.Unlock()

M internal/usecase/daemon/render_pipeline.go +16 -1


@@ -573,7 +573,13 @@ func (d Daemon) emitFrame(entry attachmentSession, ac attachedClient, state *c
573	573		marks.attachmentEffect.endTransportSend()
574	574		}
575	575		}
576		-	endEmit(uint64(len(data)), sendErr == nil)
	576	+	emitted := sendErr == nil && ((preparedANSI != nil && preparedANSI.sent) \|\| (preparedScreen != nil && preparedScreen.sent))
	577	+	endEmit(uint64(len(data)), emitted)
	578	+	if sendErr == nil && !emitted {
	579	+	ac.discardProxyCapture()
	580	+	ac.sendMu.Unlock()
	581	+	return true
	582	+	}
577	583		if sendErr == nil && ac.proxied {
578	584		kind := ports.RuntimeScreenDelta
579	585		if preparedScreen.update.Kind == ports.ScreenUpdateSnapshot {

@@ -587,8 +593,17 @@ func (d Daemon) emitFrame(entry attachmentSession, ac attachedClient, state *c
587	593		if len(data) == 0 {
588	594		if ac.proxied {
589	595		preparedScreen.commitNoSend()
	596	+	if !preparedScreen.sent {
	597	+	ac.discardProxyCapture()
	598	+	ac.sendMu.Unlock()
	599	+	return true
	600	+	}
590	601		} else {
591	602		preparedANSI.commitNoSend()
	603	+	if !preparedANSI.sent {
	604	+	ac.sendMu.Unlock()
	605	+	return true
	606	+	}
592	607		}
593	608		}
594	609		// Publish only after output preparation and transport emission both

 					marks.attachmentEffect.endTransportSend()
 				}
 			}
-			endEmit(uint64(len(data)), sendErr == nil)
+			emitted := sendErr == nil && ((preparedANSI != nil && preparedANSI.sent) || (preparedScreen != nil && preparedScreen.sent))
+			endEmit(uint64(len(data)), emitted)
+			if sendErr == nil && !emitted {
+				ac.discardProxyCapture()
+				ac.sendMu.Unlock()
+				return true
+			}
 			if sendErr == nil && ac.proxied {
 				kind := ports.RuntimeScreenDelta
 				if preparedScreen.update.Kind == ports.ScreenUpdateSnapshot {
 		if len(data) == 0 {
 			if ac.proxied {
 				preparedScreen.commitNoSend()
+				if !preparedScreen.sent {
+					ac.discardProxyCapture()
+					ac.sendMu.Unlock()
+					return true
+				}
 			} else {
 				preparedANSI.commitNoSend()
+				if !preparedANSI.sent {
+					ac.sendMu.Unlock()
+					return true
+				}
 			}
 		}
 		// Publish only after output preparation and transport emission both

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


@@ -1082,7 +1082,7 @@ func TestResumeRebasesFullOutputWindowBeforeFirstPaint(t *testing.T) {
1082	1082		first, err := ports.UnmarshalOutput(sends[0].Payload)
1083	1083		require.NoError(t, err)
1084	1084		require.Zero(t, first.BaseStateNum)
1085		-	require.Equal(t, uint64(maxUnackedOutputStates+1), first.NewStateNum)
	1085	+	require.Equal(t, uint64(1), first.NewStateNum)
1086	1086		resumedAC.ackOutputState(first.NewStateNum)
1087	1087
1088	1088		resumedSess.tabs[0].focusedPane().screen.Write([]byte("A"))


@@ -1082,7 +1082,7 @@ func TestResumeRebasesFullOutputWindowBeforeFirstPaint(t *testing.T) {
1082		first, err := ports.UnmarshalOutput(sends[0].Payload)	1082		first, err := ports.UnmarshalOutput(sends[0].Payload)
1083		require.NoError(t, err)	1083		require.NoError(t, err)
1084		require.Zero(t, first.BaseStateNum)	1084		require.Zero(t, first.BaseStateNum)
1085	-	require.Equal(t, uint64(maxUnackedOutputStates+1), first.NewStateNum)	1085	+	require.Equal(t, uint64(1), first.NewStateNum)
1086		resumedAC.ackOutputState(first.NewStateNum)	1086		resumedAC.ackOutputState(first.NewStateNum)
1087			1087
1088		resumedSess.tabs[0].focusedPane().screen.Write([]byte("A"))	1088		resumedSess.tabs[0].focusedPane().screen.Write([]byte("A"))

M internal/usecase/daemon/screen_output_stream.go +65 -28


@@ -32,6 +32,12 @@ func (s *structuredOutputStream) prepare(frame renderer.Frame, damage []renderer
32	32		if s == nil \|\| s.state == nil {
33	33		return nil, errors.New("structured output stream is nil")
34	34		}
	35	+	if reset && s.state.initialized {
	36	+	s.state.rebase()
	37	+	s.shadow = renderer.Frame{}
	38	+	s.cursorOut = cursorOut{}
	39	+	s.forceSnapshot = true
	40	+	}
35	41		if err := frame.Validate(); err != nil {
36	42		return nil, err
37	43		}

@@ -48,14 +54,21 @@ func (s *structuredOutputStream) prepare(frame renderer.Frame, damage []renderer
48	54		}
49	55
50	56		plan := candidate.Plan
	57	+	s.state.initialized = true
51	58		snapshot := reset \|\| plan.Snapshot
52	59		cursorChanged := cursor != s.cursorOut
53	60		if !snapshot && plan.Scroll.Height == 0 && len(plan.Spans) == 0 && !cursorChanged {
	61	+	connectionGeneration, epoch, viewRevision := s.state.fence()
54	62		return &preparedStructuredOutput{
55		-	stream: s,
56		-	candidate: candidate,
57		-	cursor: cursor,
58		-	echoAck: echoAck,
	63	+	stream: s,
	64	+	candidate: candidate,
	65	+	cursor: cursor,
	66	+	base: s.state.next,
	67	+	epoch: epoch,
	68	+	generation: s.state.generation,
	69	+	connectionGeneration: connectionGeneration,
	70	+	viewRevision: viewRevision,
	71	+	echoAck: echoAck,
59	72		}, nil
60	73		}
61	74		if s.state.next == ^uint64(0) {

@@ -66,9 +79,10 @@ func (s *structuredOutputStream) prepare(frame renderer.Frame, damage []renderer
66	79		if err != nil {
67	80		return nil, err
68	81		}
	82	+	base := s.state.next
69	83		update := ports.ScreenUpdate{
70		-	BaseStateNum: s.state.next,
71		-	NewStateNum: s.state.next + 1,
	84	+	BaseStateNum: base,
	85	+	NewStateNum: base + 1,
72	86		EchoAck: echoAck,
73	87		Kind: ports.ScreenUpdateDelta,
74	88		Size: frameSize(frame),

@@ -109,15 +123,21 @@ func (s *structuredOutputStream) prepare(frame renderer.Frame, damage []renderer
109	123		if err != nil {
110	124		return nil, err
111	125		}
	126	+	connectionGeneration, epoch, viewRevision := s.state.fence()
112	127		return &preparedStructuredOutput{
113		-	stream: s,
114		-	candidate: candidate,
115		-	cursor: cursor,
116		-	update: update,
117		-	next: update.NewStateNum,
118		-	data: data,
119		-	echoAck: echoAck,
120		-	snapshot: snapshot,
	128	+	stream: s,
	129	+	candidate: candidate,
	130	+	cursor: cursor,
	131	+	base: base,
	132	+	epoch: epoch,
	133	+	generation: s.state.generation,
	134	+	connectionGeneration: connectionGeneration,
	135	+	viewRevision: viewRevision,
	136	+	update: update,
	137	+	next: update.NewStateNum,
	138	+	data: data,
	139	+	echoAck: echoAck,
	140	+	snapshot: snapshot,
121	141		}, nil
122	142		}
123	143

@@ -152,28 +172,38 @@ func wireScreenCursor(cursor cursorOut, frame renderer.Frame) (ports.ScreenCurso
152	172		}
153	173
154	174		type preparedStructuredOutput struct {
155		-	stream *structuredOutputStream
156		-	candidate renderer.DeltaCandidate
157		-	cursor cursorOut
158		-	update ports.ScreenUpdate
159		-	next uint64
160		-	echoAck uint64
161		-	data []byte
162		-	snapshot bool
163		-	attempted bool
164		-	committed bool
	175	+	stream *structuredOutputStream
	176	+	candidate renderer.DeltaCandidate
	177	+	cursor cursorOut
	178	+	base uint64
	179	+	epoch uint64
	180	+	generation uint64
	181	+	connectionGeneration uint64
	182	+	viewRevision uint64
	183	+	update ports.ScreenUpdate
	184	+	next uint64
	185	+	echoAck uint64
	186	+	data []byte
	187	+	snapshot bool
	188	+	attempted bool
	189	+	committed bool
	190	+	sent bool
165	191		}
166	192
167	193		func (p *preparedStructuredOutput) send(sender func(ports.Frame) error) error {
168	194		if p == nil \|\| p.attempted {
169	195		return nil
170	196		}
	197	+	p.attempted = true
	198	+	if !p.stream.state.preparedCurrent(p.generation, p.epoch, p.viewRevision, p.base) \|\|
	199	+	(p.stream.state.attachment != nil && p.stream.state.attachment.connectionGeneration.Load() != p.connectionGeneration) {
	200	+	return nil
	201	+	}
171	202		if len(p.data) == 0 {
172		-	p.attempted = true
173	203		p.commit()
	204	+	p.sent = true
174	205		return nil
175	206		}
176		-	p.attempted = true
177	207		if sender == nil {
178	208		p.stream.forceSnapshot = true
179	209		return errors.New("structured output: nil sender")

@@ -185,6 +215,7 @@ func (p *preparedStructuredOutput) send(sender func(ports.Frame) error) error {
185	215		p.commit()
186	216		p.stream.state.next = p.next
187	217		p.stream.state.publishOutstanding()
	218	+	p.sent = true
188	219		return nil
189	220		}
190	221

@@ -196,6 +227,10 @@ func (p *preparedStructuredOutput) commitNoSend() {
196	227		return
197	228		}
198	229		p.attempted = true
	230	+	if !p.stream.state.preparedCurrent(p.generation, p.epoch, p.viewRevision, p.base) \|\|
	231	+	(p.stream.state.attachment != nil && p.stream.state.attachment.connectionGeneration.Load() != p.connectionGeneration) {
	232	+	return
	233	+	}
199	234		if len(p.data) != 0 {
200	235		if p.stream != nil {
201	236		p.stream.forceSnapshot = true

@@ -203,6 +238,7 @@ func (p *preparedStructuredOutput) commitNoSend() {
203	238		return
204	239		}
205	240		p.commit()
	241	+	p.sent = true
206	242		}
207	243
208	244		func (p *preparedStructuredOutput) commit() {

@@ -212,12 +248,13 @@ func (p *preparedStructuredOutput) commit() {
212	248		p.candidate.Commit(&p.stream.shadow)
213	249		p.stream.cursorOut = p.cursor
214	250		p.stream.forceSnapshot = false
	251	+	p.stream.state.forceSnapshot = false
215	252		p.committed = true
216	253		}
217	254
218		-	func (s *structuredOutputStream) ack(state uint64) {
	255	+	func (s *structuredOutputStream) ack(values ...uint64) {
219	256		if s != nil && s.state != nil {
220		-	s.state.ack(state)
	257	+	s.state.ack(values...)
221	258		}
222	259		}
223	260

 	if s == nil || s.state == nil {
 		return nil, errors.New("structured output stream is nil")
 	}
+	if reset && s.state.initialized {
+		s.state.rebase()
+		s.shadow = renderer.Frame{}
+		s.cursorOut = cursorOut{}
+		s.forceSnapshot = true
+	}
 	if err := frame.Validate(); err != nil {
 		return nil, err
 	}
 	}
 	plan := candidate.Plan
+	s.state.initialized = true
 	snapshot := reset || plan.Snapshot
 	cursorChanged := cursor != s.cursorOut
 	if !snapshot && plan.Scroll.Height == 0 && len(plan.Spans) == 0 && !cursorChanged {
+		connectionGeneration, epoch, viewRevision := s.state.fence()
 		return &preparedStructuredOutput{
-			stream:    s,
+			stream:               s,
-			candidate: candidate,
+			candidate:            candidate,
-			cursor:    cursor,
+			cursor:               cursor,
-			echoAck:   echoAck,
+			base:                 s.state.next,
+			epoch:                epoch,
+			generation:           s.state.generation,
+			connectionGeneration: connectionGeneration,
+			viewRevision:         viewRevision,
+			echoAck:              echoAck,
 		}, nil
 	}
 	if s.state.next == ^uint64(0) {
 	if err != nil {
 		return nil, err
 	}
+	base := s.state.next
 	update := ports.ScreenUpdate{
-		BaseStateNum: s.state.next,
+		BaseStateNum: base,
-		NewStateNum:  s.state.next + 1,
+		NewStateNum:  base + 1,
 		EchoAck:      echoAck,
 		Kind:         ports.ScreenUpdateDelta,
 		Size:         frameSize(frame),
 	if err != nil {
 		return nil, err
 	}
+	connectionGeneration, epoch, viewRevision := s.state.fence()
 	return &preparedStructuredOutput{
-		stream:    s,
+		stream:               s,
-		candidate: candidate,
+		candidate:            candidate,
-		cursor:    cursor,
+		cursor:               cursor,
-		update:    update,
+		base:                 base,
-		next:      update.NewStateNum,
+		epoch:                epoch,
-		data:      data,
+		generation:           s.state.generation,
-		echoAck:   echoAck,
+		connectionGeneration: connectionGeneration,
-		snapshot:  snapshot,
+		viewRevision:         viewRevision,
+		update:               update,
+		next:                 update.NewStateNum,
+		data:                 data,
+		echoAck:              echoAck,
+		snapshot:             snapshot,
 	}, nil
 }
 }
 type preparedStructuredOutput struct {
-	stream    *structuredOutputStream
+	stream               *structuredOutputStream
-	candidate renderer.DeltaCandidate
+	candidate            renderer.DeltaCandidate
-	cursor    cursorOut
+	cursor               cursorOut
-	update    ports.ScreenUpdate
+	base                 uint64
-	next      uint64
+	epoch                uint64
-	echoAck   uint64
+	generation           uint64
-	data      []byte
+	connectionGeneration uint64
-	snapshot  bool
+	viewRevision         uint64
-	attempted bool
+	update               ports.ScreenUpdate
-	committed bool
+	next                 uint64
+	echoAck              uint64
+	data                 []byte
+	snapshot             bool
+	attempted            bool
+	committed            bool
+	sent                 bool
 }
 func (p *preparedStructuredOutput) send(sender func(ports.Frame) error) error {
 	if p == nil || p.attempted {
 		return nil
 	}
+	p.attempted = true
+	if !p.stream.state.preparedCurrent(p.generation, p.epoch, p.viewRevision, p.base) ||
+		(p.stream.state.attachment != nil && p.stream.state.attachment.connectionGeneration.Load() != p.connectionGeneration) {
+		return nil
+	}
 	if len(p.data) == 0 {
-		p.attempted = true
 		p.commit()
+		p.sent = true
 		return nil
 	}
-	p.attempted = true
 	if sender == nil {
 		p.stream.forceSnapshot = true
 		return errors.New("structured output: nil sender")
 	p.commit()
 	p.stream.state.next = p.next
 	p.stream.state.publishOutstanding()
+	p.sent = true
 	return nil
 }
 		return
 	}
 	p.attempted = true
+	if !p.stream.state.preparedCurrent(p.generation, p.epoch, p.viewRevision, p.base) ||
+		(p.stream.state.attachment != nil && p.stream.state.attachment.connectionGeneration.Load() != p.connectionGeneration) {
+		return
+	}
 	if len(p.data) != 0 {
 		if p.stream != nil {
 			p.stream.forceSnapshot = true
 		return
 	}
 	p.commit()
+	p.sent = true
 }
 func (p *preparedStructuredOutput) commit() {
 	p.candidate.Commit(&p.stream.shadow)
 	p.stream.cursorOut = p.cursor
 	p.stream.forceSnapshot = false
+	p.stream.state.forceSnapshot = false
 	p.committed = true
 }
-func (s *structuredOutputStream) ack(state uint64) {
+func (s *structuredOutputStream) ack(values ...uint64) {
 	if s != nil && s.state != nil {
-		s.state.ack(state)
+		s.state.ack(values...)
 	}
 }

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


@@ -137,7 +137,7 @@ func TestStructuredOutputStreamResetAndResizeUseBaseZero(t *testing.T) {
137	137		resetUpdate := decodeStructured(t, reset.data)
138	138		require.Equal(t, ports.ScreenUpdateSnapshot, resetUpdate.Kind)
139	139		require.Zero(t, resetUpdate.BaseStateNum)
140		-	require.Equal(t, uint64(3), resetUpdate.NewStateNum)
	140	+	require.Equal(t, uint64(1), resetUpdate.NewStateNum)
141	141		require.Len(t, resetUpdate.Spans, resized.Height)
142	142		for _, span := range resetUpdate.Spans {
143	143		require.Equal(t, uint16(resized.Width), uint16(len(span.Cells)))


@@ -137,7 +137,7 @@ func TestStructuredOutputStreamResetAndResizeUseBaseZero(t *testing.T) {
137		resetUpdate := decodeStructured(t, reset.data)	137		resetUpdate := decodeStructured(t, reset.data)
138		require.Equal(t, ports.ScreenUpdateSnapshot, resetUpdate.Kind)	138		require.Equal(t, ports.ScreenUpdateSnapshot, resetUpdate.Kind)
139		require.Zero(t, resetUpdate.BaseStateNum)	139		require.Zero(t, resetUpdate.BaseStateNum)
140	-	require.Equal(t, uint64(3), resetUpdate.NewStateNum)	140	+	require.Equal(t, uint64(1), resetUpdate.NewStateNum)
141		require.Len(t, resetUpdate.Spans, resized.Height)	141		require.Len(t, resetUpdate.Spans, resized.Height)
142		for _, span := range resetUpdate.Spans {	142		for _, span := range resetUpdate.Spans {
143		require.Equal(t, uint16(resized.Width), uint16(len(span.Cells)))	143		require.Equal(t, uint16(resized.Width), uint16(len(span.Cells)))

M internal/usecase/daemon/wire.go +5 -1


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

@@ -25,7 +26,10 @@ func frameError(code uint16, text string) ports.Frame {
25	26		}
26	27
27	28		func frameOutputState(b []byte, baseState uint64, state uint64, echoAck uint64) ports.Frame {
28		-	return ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: baseState, NewStateNum: state, EchoAck: echoAck, Data: b})}
	29	+	return ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{
	30	+	Epoch: 1, Base: baseState, New: state, Echo: echoAck,
	31	+	Size: domain.Size{Cols: 1, Rows: 1}, Full: state != 0 && baseState == 0, Data: b,
	32	+	})}
29	33		}
30	34
31	35		func frameDetached(reason uint8) ports.Frame {

 package daemon
 import (
+	"github.com/bnema/vev/internal/domain"
 	"github.com/bnema/vev/internal/ports"
 )
 }
 func frameOutputState(b []byte, baseState uint64, state uint64, echoAck uint64) ports.Frame {
-	return ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{BaseStateNum: baseState, NewStateNum: state, EchoAck: echoAck, Data: b})}
+	return ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{
+		Epoch: 1, Base: baseState, New: state, Echo: echoAck,
+		Size: domain.Size{Cols: 1, Rows: 1}, Full: state != 0 && baseState == 0, Data: b,
+	})}
 }
 func frameDetached(reason uint8) ports.Frame {