fix(dgram): enforce single output state window

commit: 48a3b83d0702ca34840007a4e732d086e1fb24c8
author: brice · 3mo ago · 2026-07-10 11:12 UTC
parent: 1105732
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +114 -23

M internal/adapters/dgram/proxy.go +25 -1


@@ -51,6 +51,7 @@ type proxyCopyDirection struct {
51	51		recvKind proxyCopyErrKind
52	52		sendKind proxyCopyErrKind
53	53		retryRecoverable bool
	54	+	transform func(ports.Frame) ports.Frame
54	55		}
55	56
56	57		func (p ProxyRuntime) Run(ctx context.Context) error {

@@ -76,7 +77,11 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
76	77
77	78		errCh := make(chan proxyCopyErr, 2)
78	79		copier := proxyCopier{ctx: copyCtx, errCh: errCh, retryBackoff: retryBackoff, clk: clk}
79		-	clientToDaemon := proxyCopyDirection{src: p.Client, dst: p.Daemon, recvKind: proxyClientRecv, sendKind: proxyDaemonSend}
	80	+	clientToDaemon := proxyCopyDirection{
	81	+	src: p.Client, dst: p.Daemon,
	82	+	recvKind: proxyClientRecv, sendKind: proxyDaemonSend,
	83	+	transform: clampDatagramHelloOutputWindow,
	84	+	}
80	85		daemonToClient := proxyCopyDirection{src: p.Daemon, dst: p.Client, recvKind: proxyDaemonRecv, sendKind: proxyClientSend, retryRecoverable: true}
81	86		go copier.copyFrames(clientToDaemon)
82	87		go copier.copyFrames(daemonToClient)

@@ -157,6 +162,9 @@ func (c proxyCopier) copyFrames(dir proxyCopyDirection) {
157	162		c.errCh <- proxyCopyErr{kind: dir.recvKind, err: err}
158	163		return
159	164		}
	165	+	if dir.transform != nil {
	166	+	f = dir.transform(f)
	167	+	}
160	168		// Retry the same frame in place while the destination link error stays
161	169		// recoverable (congestion, transient path loss). The frame is never
162	170		// dropped — on a reliable stream a gap desynchronizes the client screen.

@@ -199,3 +207,19 @@ func recoverableClientLink(t ports.Transport, err error) bool {
199	207		return false
200	208		}
201	209		}
	210	+
	211	+	// clampDatagramHelloOutputWindow enforces the datagram-safe state window at
	212	+	// the adapter boundary. Invalid Hello payloads pass through unchanged so the
	213	+	// daemon remains the single authority for strict decoding and version checks.
	214	+	func clampDatagramHelloOutputWindow(f ports.Frame) ports.Frame {
	215	+	if f.Type != ports.MsgHello {
	216	+	return f
	217	+	}
	218	+	hello, err := ports.UnmarshalHello(f.Payload)
	219	+	if err != nil {
	220	+	return f
	221	+	}
	222	+	hello.MaxOutputInFlight = 1
	223	+	f.Payload = ports.MarshalHello(hello)
	224	+	return f
	225	+	}


@@ -51,6 +51,7 @@ type proxyCopyDirection struct {
51		recvKind proxyCopyErrKind	51		recvKind proxyCopyErrKind
52		sendKind proxyCopyErrKind	52		sendKind proxyCopyErrKind
53		retryRecoverable bool	53		retryRecoverable bool
			54	+	transform func(ports.Frame) ports.Frame
54		}	55		}
55			56
56		func (p ProxyRuntime) Run(ctx context.Context) error {	57		func (p ProxyRuntime) Run(ctx context.Context) error {

@@ -76,7 +77,11 @@ func (p ProxyRuntime) Run(ctx context.Context) error {
76			77
77		errCh := make(chan proxyCopyErr, 2)	78		errCh := make(chan proxyCopyErr, 2)
78		copier := proxyCopier{ctx: copyCtx, errCh: errCh, retryBackoff: retryBackoff, clk: clk}	79		copier := proxyCopier{ctx: copyCtx, errCh: errCh, retryBackoff: retryBackoff, clk: clk}
79	-	clientToDaemon := proxyCopyDirection{src: p.Client, dst: p.Daemon, recvKind: proxyClientRecv, sendKind: proxyDaemonSend}	80	+	clientToDaemon := proxyCopyDirection{
			81	+	src: p.Client, dst: p.Daemon,
			82	+	recvKind: proxyClientRecv, sendKind: proxyDaemonSend,
			83	+	transform: clampDatagramHelloOutputWindow,
			84	+	}
80		daemonToClient := proxyCopyDirection{src: p.Daemon, dst: p.Client, recvKind: proxyDaemonRecv, sendKind: proxyClientSend, retryRecoverable: true}	85		daemonToClient := proxyCopyDirection{src: p.Daemon, dst: p.Client, recvKind: proxyDaemonRecv, sendKind: proxyClientSend, retryRecoverable: true}
81		go copier.copyFrames(clientToDaemon)	86		go copier.copyFrames(clientToDaemon)
82		go copier.copyFrames(daemonToClient)	87		go copier.copyFrames(daemonToClient)

@@ -157,6 +162,9 @@ func (c proxyCopier) copyFrames(dir proxyCopyDirection) {
157		c.errCh <- proxyCopyErr{kind: dir.recvKind, err: err}	162		c.errCh <- proxyCopyErr{kind: dir.recvKind, err: err}
158		return	163		return
159		}	164		}
			165	+	if dir.transform != nil {
			166	+	f = dir.transform(f)
			167	+	}
160		// Retry the same frame in place while the destination link error stays	168		// Retry the same frame in place while the destination link error stays
161		// recoverable (congestion, transient path loss). The frame is never	169		// recoverable (congestion, transient path loss). The frame is never
162		// dropped — on a reliable stream a gap desynchronizes the client screen.	170		// dropped — on a reliable stream a gap desynchronizes the client screen.

@@ -199,3 +207,19 @@ func recoverableClientLink(t ports.Transport, err error) bool {
199		return false	207		return false
200		}	208		}
201		}	209		}
			210	+
			211	+	// clampDatagramHelloOutputWindow enforces the datagram-safe state window at
			212	+	// the adapter boundary. Invalid Hello payloads pass through unchanged so the
			213	+	// daemon remains the single authority for strict decoding and version checks.
			214	+	func clampDatagramHelloOutputWindow(f ports.Frame) ports.Frame {
			215	+	if f.Type != ports.MsgHello {
			216	+	return f
			217	+	}
			218	+	hello, err := ports.UnmarshalHello(f.Payload)
			219	+	if err != nil {
			220	+	return f
			221	+	}
			222	+	hello.MaxOutputInFlight = 1
			223	+	f.Payload = ports.MarshalHello(hello)
			224	+	return f
			225	+	}

M internal/adapters/dgram/proxy_test.go +74 -22


@@ -3,7 +3,9 @@ package dgram
3	3		import (
4	4		"context"
5	5		"errors"
	6	+	"fmt"
6	7		"io"
	8	+	"reflect"
7	9		"sync"
8	10		"testing"
9	11		"time"

@@ -78,30 +80,49 @@ func TestProxyRuntimeCopiesFramesAndDaemonEOFIsTerminal(t *testing.T) {
78	80		}
79	81		}
80	82
81		-	func TestProxyRuntimeForwardsOutputAckCapabilityInHello(t *testing.T) {
82		-	client := newFakeTransport()
83		-	daemon := newFakeTransport()
84		-	errCh := make(chan error, 1)
85		-	go func() { errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context()) }()
86		-	hello := ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", MaxOutputInFlight: 8}
87		-	client.recv <- recvResult{frame: ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}}
	83	+	func TestProxyRuntimeClampsDatagramHelloOutputWindowAndPreservesControl(t *testing.T) {
	84	+	for _, requested := range []uint8{0, 1, 8} {
	85	+	t.Run(fmt.Sprintf("requested_%d", requested), func(t *testing.T) {
	86	+	client := newFakeTransport()
	87	+	daemon := newFakeTransport()
	88	+	errCh := make(chan error, 1)
	89	+	go func() { errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context()) }()
	90	+	hello := ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", MaxOutputInFlight: requested}
	91	+	client.recv <- recvResult{frame: ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}}
88	92
89		-	select {
90		-	case frame := <-daemon.sent:
91		-	got, err := ports.UnmarshalHello(frame.Payload)
92		-	if err != nil {
93		-	t.Fatal(err)
94		-	}
95		-	if got.MaxOutputInFlight != 8 {
96		-	t.Fatal("proxied Hello lost output window")
97		-	}
98		-	case <-time.After(time.Second):
99		-	t.Fatal("timeout waiting for proxied Hello")
100		-	}
	93	+	select {
	94	+	case frame := <-daemon.sent:
	95	+	got, err := ports.UnmarshalHello(frame.Payload)
	96	+	if err != nil {
	97	+	t.Fatal(err)
	98	+	}
	99	+	if got.MaxOutputInFlight != 1 {
	100	+	t.Fatalf("proxied output window=%d, want datagram-safe 1", got.MaxOutputInFlight)
	101	+	}
	102	+	case <-time.After(time.Second):
	103	+	t.Fatal("timeout waiting for proxied Hello")
	104	+	}
101	105
102		-	daemon.recv <- recvResult{err: io.EOF}
103		-	if err := <-errCh; err != nil {
104		-	t.Fatalf("Run err=%v, want nil", err)
	106	+	input := ports.Frame{Type: ports.MsgInput, Payload: []byte("still flowing")}
	107	+	ack := ports.Frame{Type: ports.MsgAck, Payload: ports.MarshalAck(ports.Ack{AckedStateNum: 7})}
	108	+	client.recv <- recvResult{frame: input}
	109	+	client.recv <- recvResult{frame: ack}
	110	+	for _, want := range []ports.Frame{input, ack} {
	111	+	select {
	112	+	case got := <-daemon.sent:
	113	+	if !reflect.DeepEqual(got, want) {
	114	+	t.Fatalf("proxied control frame=%+v, want %+v", got, want)
	115	+	}
	116	+	case <-time.After(time.Second):
	117	+	t.Fatalf("timeout waiting for proxied %v", want.Type)
	118	+	}
	119	+	}
	120	+
	121	+	daemon.recv <- recvResult{err: io.EOF}
	122	+	if err := <-errCh; err != nil {
	123	+	t.Fatalf("Run err=%v, want nil", err)
	124	+	}
	125	+	})
105	126		}
106	127		}
107	128

@@ -464,3 +485,34 @@ func TestProxyCopierStopsRetryingWhenContextCanceled(t *testing.T) {
464	485		t.Fatal("retrying copier did not stop after context cancellation")
465	486		}
466	487		}
	488	+
	489	+	func TestClampDatagramHelloOutputWindowPreservesProtocolValidation(t *testing.T) {
	490	+	t.Run("malformed payload remains malformed", func(t *testing.T) {
	491	+	payload := append(ports.MarshalHello(ports.Hello{Version: ports.ProtocolVersion, MaxOutputInFlight: 8}), 0xff)
	492	+	frame := ports.Frame{Type: ports.MsgHello, Payload: payload}
	493	+	got := clampDatagramHelloOutputWindow(frame)
	494	+	if !reflect.DeepEqual(got, frame) {
	495	+	t.Fatalf("malformed Hello changed: got %#v, want %#v", got, frame)
	496	+	}
	497	+	if _, err := ports.UnmarshalHello(got.Payload); err == nil {
	498	+	t.Fatal("malformed Hello became decodable")
	499	+	}
	500	+	})
	501	+
	502	+	t.Run("mismatched version remains mismatched", func(t *testing.T) {
	503	+	wantVersion := ports.ProtocolVersion + 1
	504	+	frame := ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(ports.Hello{
	505	+	Version: wantVersion, MaxOutputInFlight: 8,
	506	+	})}
	507	+	got, err := ports.UnmarshalHello(clampDatagramHelloOutputWindow(frame).Payload)
	508	+	if err != nil {
	509	+	t.Fatal(err)
	510	+	}
	511	+	if got.Version != wantVersion {
	512	+	t.Fatalf("version=%d, want unchanged mismatch %d", got.Version, wantVersion)
	513	+	}
	514	+	if got.MaxOutputInFlight != 1 {
	515	+	t.Fatalf("output window=%d, want 1", got.MaxOutputInFlight)
	516	+	}
	517	+	})
	518	+	}


@@ -3,7 +3,9 @@ package dgram
3		import (	3		import (
4		"context"	4		"context"
5		"errors"	5		"errors"
			6	+	"fmt"
6		"io"	7		"io"
			8	+	"reflect"
7		"sync"	9		"sync"
8		"testing"	10		"testing"
9		"time"	11		"time"

@@ -78,30 +80,49 @@ func TestProxyRuntimeCopiesFramesAndDaemonEOFIsTerminal(t *testing.T) {
78		}	80		}
79		}	81		}
80			82
81	-	func TestProxyRuntimeForwardsOutputAckCapabilityInHello(t *testing.T) {	83	+	func TestProxyRuntimeClampsDatagramHelloOutputWindowAndPreservesControl(t *testing.T) {
82	-	client := newFakeTransport()	84	+	for _, requested := range []uint8{0, 1, 8} {
83	-	daemon := newFakeTransport()	85	+	t.Run(fmt.Sprintf("requested_%d", requested), func(t *testing.T) {
84	-	errCh := make(chan error, 1)	86	+	client := newFakeTransport()
85	-	go func() { errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context()) }()	87	+	daemon := newFakeTransport()
86	-	hello := ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", MaxOutputInFlight: 8}	88	+	errCh := make(chan error, 1)
87	-	client.recv <- recvResult{frame: ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}}	89	+	go func() { errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context()) }()
			90	+	hello := ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", MaxOutputInFlight: requested}
			91	+	client.recv <- recvResult{frame: ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}}
88			92
89	-	select {	93	+	select {
90	-	case frame := <-daemon.sent:	94	+	case frame := <-daemon.sent:
91	-	got, err := ports.UnmarshalHello(frame.Payload)	95	+	got, err := ports.UnmarshalHello(frame.Payload)
92	-	if err != nil {	96	+	if err != nil {
93	-	t.Fatal(err)	97	+	t.Fatal(err)
94	-	}	98	+	}
95	-	if got.MaxOutputInFlight != 8 {	99	+	if got.MaxOutputInFlight != 1 {
96	-	t.Fatal("proxied Hello lost output window")	100	+	t.Fatalf("proxied output window=%d, want datagram-safe 1", got.MaxOutputInFlight)
97	-	}	101	+	}
98	-	case <-time.After(time.Second):	102	+	case <-time.After(time.Second):
99	-	t.Fatal("timeout waiting for proxied Hello")	103	+	t.Fatal("timeout waiting for proxied Hello")
100	-	}	104	+	}
101			105
102	-	daemon.recv <- recvResult{err: io.EOF}	106	+	input := ports.Frame{Type: ports.MsgInput, Payload: []byte("still flowing")}
103	-	if err := <-errCh; err != nil {	107	+	ack := ports.Frame{Type: ports.MsgAck, Payload: ports.MarshalAck(ports.Ack{AckedStateNum: 7})}
104	-	t.Fatalf("Run err=%v, want nil", err)	108	+	client.recv <- recvResult{frame: input}
			109	+	client.recv <- recvResult{frame: ack}
			110	+	for _, want := range []ports.Frame{input, ack} {
			111	+	select {
			112	+	case got := <-daemon.sent:
			113	+	if !reflect.DeepEqual(got, want) {
			114	+	t.Fatalf("proxied control frame=%+v, want %+v", got, want)
			115	+	}
			116	+	case <-time.After(time.Second):
			117	+	t.Fatalf("timeout waiting for proxied %v", want.Type)
			118	+	}
			119	+	}
			120	+
			121	+	daemon.recv <- recvResult{err: io.EOF}
			122	+	if err := <-errCh; err != nil {
			123	+	t.Fatalf("Run err=%v, want nil", err)
			124	+	}
			125	+	})
105		}	126		}
106		}	127		}
107			128

@@ -464,3 +485,34 @@ func TestProxyCopierStopsRetryingWhenContextCanceled(t *testing.T) {
464		t.Fatal("retrying copier did not stop after context cancellation")	485		t.Fatal("retrying copier did not stop after context cancellation")
465		}	486		}
466		}	487		}
			488	+
			489	+	func TestClampDatagramHelloOutputWindowPreservesProtocolValidation(t *testing.T) {
			490	+	t.Run("malformed payload remains malformed", func(t *testing.T) {
			491	+	payload := append(ports.MarshalHello(ports.Hello{Version: ports.ProtocolVersion, MaxOutputInFlight: 8}), 0xff)
			492	+	frame := ports.Frame{Type: ports.MsgHello, Payload: payload}
			493	+	got := clampDatagramHelloOutputWindow(frame)
			494	+	if !reflect.DeepEqual(got, frame) {
			495	+	t.Fatalf("malformed Hello changed: got %#v, want %#v", got, frame)
			496	+	}
			497	+	if _, err := ports.UnmarshalHello(got.Payload); err == nil {
			498	+	t.Fatal("malformed Hello became decodable")
			499	+	}
			500	+	})
			501	+
			502	+	t.Run("mismatched version remains mismatched", func(t *testing.T) {
			503	+	wantVersion := ports.ProtocolVersion + 1
			504	+	frame := ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(ports.Hello{
			505	+	Version: wantVersion, MaxOutputInFlight: 8,
			506	+	})}
			507	+	got, err := ports.UnmarshalHello(clampDatagramHelloOutputWindow(frame).Payload)
			508	+	if err != nil {
			509	+	t.Fatal(err)
			510	+	}
			511	+	if got.Version != wantVersion {
			512	+	t.Fatalf("version=%d, want unchanged mismatch %d", got.Version, wantVersion)
			513	+	}
			514	+	if got.MaxOutputInFlight != 1 {
			515	+	t.Fatalf("output window=%d, want 1", got.MaxOutputInFlight)
			516	+	}
			517	+	})
			518	+	}

M internal/usecase/daemon/output_state_sync_test.go +15 -0


@@ -294,7 +294,20 @@ func TestDatagramWindowOneCoalescesPaintsUntilMsgAck(t *testing.T) {
294	294		pane.screen.Write([]byte("x"))
295	295		d.paint(sess, ac, false)
296	296		}
	297	+	pane.screen.Write([]byte("LATEST"))
	298	+	d.paint(sess, ac, false)
297	299		requireNoOutputFrame(t, tr.sends)
	300	+	require.Equal(t, uint64(1), ac.output.outstanding(), "only one state-bearing datagram output may be unacknowledged")
	301	+
	302	+	// Side-effect output is control-like: it must pass while the state window is
	303	+	// full without consuming another state number.
	304	+	require.NoError(t, d.boundedSendOutputErr(ac, []byte("side-effect")))
	305	+	sideEffect := awaitFrame(t, tr.sends, ports.MsgOutput)
	306	+	sideEffectOutput, err := ports.UnmarshalOutput(sideEffect.Payload)
	307	+	require.NoError(t, err)
	308	+	require.Zero(t, sideEffectOutput.BaseStateNum)
	309	+	require.Zero(t, sideEffectOutput.NewStateNum)
	310	+	require.Equal(t, uint64(1), ac.output.outstanding())
298	311
299	312		tr.recv <- ports.Frame{Type: ports.MsgAck, Payload: ports.MarshalAck(ports.Ack{AckedStateNum: firstOutput.NewStateNum})}
300	313		require.NoError(t, tr.Close())

@@ -304,4 +317,6 @@ func TestDatagramWindowOneCoalescesPaintsUntilMsgAck(t *testing.T) {
304	317		require.NoError(t, err)
305	318		require.Equal(t, firstOutput.NewStateNum, secondOutput.BaseStateNum)
306	319		require.Equal(t, uint64(2), secondOutput.NewStateNum)
	320	+	require.Contains(t, string(secondOutput.Data), "LATEST", "cumulative ACK must release the latest coalesced state")
	321	+	require.Equal(t, uint64(1), ac.output.outstanding())
307	322		}


@@ -294,7 +294,20 @@ func TestDatagramWindowOneCoalescesPaintsUntilMsgAck(t *testing.T) {
294	pane.screen.Write([]byte("x"))	294		pane.screen.Write([]byte("x"))
295	d.paint(sess, ac, false)	295		d.paint(sess, ac, false)
296	}	296		}
		297	+	pane.screen.Write([]byte("LATEST"))
		298	+	d.paint(sess, ac, false)
297	requireNoOutputFrame(t, tr.sends)	299		requireNoOutputFrame(t, tr.sends)
		300	+	require.Equal(t, uint64(1), ac.output.outstanding(), "only one state-bearing datagram output may be unacknowledged")
		301	+
		302	+	// Side-effect output is control-like: it must pass while the state window is
		303	+	// full without consuming another state number.
		304	+	require.NoError(t, d.boundedSendOutputErr(ac, []byte("side-effect")))
		305	+	sideEffect := awaitFrame(t, tr.sends, ports.MsgOutput)
		306	+	sideEffectOutput, err := ports.UnmarshalOutput(sideEffect.Payload)
		307	+	require.NoError(t, err)
		308	+	require.Zero(t, sideEffectOutput.BaseStateNum)
		309	+	require.Zero(t, sideEffectOutput.NewStateNum)
		310	+	require.Equal(t, uint64(1), ac.output.outstanding())
298		311
299	tr.recv <- ports.Frame{Type: ports.MsgAck, Payload: ports.MarshalAck(ports.Ack{AckedStateNum: firstOutput.NewStateNum})}	312		tr.recv <- ports.Frame{Type: ports.MsgAck, Payload: ports.MarshalAck(ports.Ack{AckedStateNum: firstOutput.NewStateNum})}
300	require.NoError(t, tr.Close())	313		require.NoError(t, tr.Close())

@@ -304,4 +317,6 @@ func TestDatagramWindowOneCoalescesPaintsUntilMsgAck(t *testing.T) {
304	require.NoError(t, err)	317		require.NoError(t, err)
305	require.Equal(t, firstOutput.NewStateNum, secondOutput.BaseStateNum)	318		require.Equal(t, firstOutput.NewStateNum, secondOutput.BaseStateNum)
306	require.Equal(t, uint64(2), secondOutput.NewStateNum)	319		require.Equal(t, uint64(2), secondOutput.NewStateNum)
		320	+	require.Contains(t, string(secondOutput.Data), "LATEST", "cumulative ACK must release the latest coalesced state")
		321	+	require.Equal(t, uint64(1), ac.output.outstanding())
307	}	322		}