detached e25b377

feat(protocol): negotiate output in-flight window

commit
e25b3778cefa7465bd14d2cf6cb93d320383ca6a
author
brice · · 2026-07-10 11:07 UTC
parent
8d15e6e
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
17 files · +432 -66
M internal/adapters/dgram/proxy_test.go +3 -3
@@ -83,7 +83,7 @@ func TestProxyRuntimeForwardsOutputAckCapabilityInHello(t *testing.T) {
8383 daemon := newFakeTransport()
8484 errCh := make(chan error, 1)
8585 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", AckOutput: true}
86+ hello := ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", MaxOutputInFlight: 8}
8787 client.recv <- recvResult{frame: ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}}
8888
8989 select {
@@ -92,8 +92,8 @@ func TestProxyRuntimeForwardsOutputAckCapabilityInHello(t *testing.T) {
9292 if err != nil {
9393 t.Fatal(err)
9494 }
95- if !got.AckOutput {
96- t.Fatal("proxied Hello lost AckOutput capability")
95+ if got.MaxOutputInFlight != 8 {
96+ t.Fatal("proxied Hello lost output window")
9797 }
9898 case <-time.After(time.Second):
9999 t.Fatal("timeout waiting for proxied Hello")
@@ -83,7 +83,7 @@ func TestProxyRuntimeForwardsOutputAckCapabilityInHello(t *testing.T) {
83 daemon := newFakeTransport()83 daemon := newFakeTransport()
84 errCh := make(chan error, 1)84 errCh := make(chan error, 1)
85 go func() { errCh <- ProxyRuntime{Client: client, Daemon: daemon, IdleTTL: time.Hour}.Run(t.Context()) }()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", AckOutput: true}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)}}87 client.recv <- recvResult{frame: ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}}
88 88
89 select {89 select {
@@ -92,8 +92,8 @@ func TestProxyRuntimeForwardsOutputAckCapabilityInHello(t *testing.T) {
92 if err != nil {92 if err != nil {
93 t.Fatal(err)93 t.Fatal(err)
94 }94 }
95- if !got.AckOutput {95+ if got.MaxOutputInFlight != 8 {
96- t.Fatal("proxied Hello lost AckOutput capability")96+ t.Fatal("proxied Hello lost output window")
97 }97 }
98 case <-time.After(time.Second):98 case <-time.After(time.Second):
99 t.Fatal("timeout waiting for proxied Hello")99 t.Fatal("timeout waiting for proxied Hello")
M internal/adapters/dgram/transport.go +5 -0
@@ -129,6 +129,8 @@ func normalizeOptions(opts Options) Options {
129129 return opts
130130 }
131131
132+var _ ports.DatagramTransport = (*Transport)(nil)
133+
132134 type Transport struct {
133135 pc net.PacketConn
134136 codec *pdgram.Codec
@@ -1514,3 +1516,6 @@ func decodeData(b []byte) (uint64, bool, ports.Frame, bool) {
15141516 }
15151517 return binary.BigEndian.Uint64(b[1:9]), b[9] == 1, ports.Frame{Type: ports.MsgType(b[10]), Payload: append([]byte(nil), b[dataRecordHeaderSize:]...)}, true
15161518 }
1519+
1520+// DatagramTransport marks Transport as a UDP-style datagram transport.
1521+func (*Transport) DatagramTransport() {}
@@ -129,6 +129,8 @@ func normalizeOptions(opts Options) Options {
129 return opts129 return opts
130 }130 }
131 131
132+var _ ports.DatagramTransport = (*Transport)(nil)
133+
132 type Transport struct {134 type Transport struct {
133 pc net.PacketConn135 pc net.PacketConn
134 codec *pdgram.Codec136 codec *pdgram.Codec
@@ -1514,3 +1516,6 @@ func decodeData(b []byte) (uint64, bool, ports.Frame, bool) {
1514 }1516 }
1515 return binary.BigEndian.Uint64(b[1:9]), b[9] == 1, ports.Frame{Type: ports.MsgType(b[10]), Payload: append([]byte(nil), b[dataRecordHeaderSize:]...)}, true1517 return binary.BigEndian.Uint64(b[1:9]), b[9] == 1, ports.Frame{Type: ports.MsgType(b[10]), Payload: append([]byte(nil), b[dataRecordHeaderSize:]...)}, true
1516 }1518 }
1519+
1520+// DatagramTransport marks Transport as a UDP-style datagram transport.
1521+func (*Transport) DatagramTransport() {}
M internal/ports/frame.go +1 -1
@@ -1,7 +1,7 @@
11 package ports
22
33 // ProtocolVersion is the current vev IPC wire protocol version.
4-const ProtocolVersion uint16 = 13
4+const ProtocolVersion uint16 = 14
55
66 // MsgType identifies the kind of payload carried by a Frame.
77 type MsgType uint8
@@ -1,7 +1,7 @@
1 package ports1 package ports
2 2
3 // ProtocolVersion is the current vev IPC wire protocol version.3 // ProtocolVersion is the current vev IPC wire protocol version.
4-const ProtocolVersion uint16 = 134+const ProtocolVersion uint16 = 14
5 5
6 // MsgType identifies the kind of payload carried by a Frame.6 // MsgType identifies the kind of payload carried by a Frame.
7 type MsgType uint87 type MsgType uint8
M internal/ports/frame_test.go +2 -2
@@ -3,8 +3,8 @@ package ports
33 import "testing"
44
55 func TestProtocolVersion(t *testing.T) {
6- if ProtocolVersion != 13 {
7- t.Fatalf("ProtocolVersion = %d, want 13", ProtocolVersion)
6+ if ProtocolVersion != 14 {
7+ t.Fatalf("ProtocolVersion = %d, want 14", ProtocolVersion)
88 }
99 }
1010
@@ -3,8 +3,8 @@ package ports
3 import "testing"3 import "testing"
4 4
5 func TestProtocolVersion(t *testing.T) {5 func TestProtocolVersion(t *testing.T) {
6- if ProtocolVersion != 13 {6+ if ProtocolVersion != 14 {
7- t.Fatalf("ProtocolVersion = %d, want 13", ProtocolVersion)7+ t.Fatalf("ProtocolVersion = %d, want 14", ProtocolVersion)
8 }8 }
9 }9 }
10 10
M internal/ports/mocks/mocks.go +208 -0
@@ -1120,6 +1120,214 @@ func (_c *MockTransport_Send_Call) RunAndReturn(run func(frame ports.Frame) erro
11201120 return _c
11211121 }
11221122
1123+// NewMockDatagramTransport creates a new instance of MockDatagramTransport. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
1124+// The first argument is typically a *testing.T value.
1125+func NewMockDatagramTransport(t interface {
1126+ mock.TestingT
1127+ Cleanup(func())
1128+}) *MockDatagramTransport {
1129+ mock := &MockDatagramTransport{}
1130+ mock.Mock.Test(t)
1131+
1132+ t.Cleanup(func() { mock.AssertExpectations(t) })
1133+
1134+ return mock
1135+}
1136+
1137+// MockDatagramTransport is an autogenerated mock type for the DatagramTransport type
1138+type MockDatagramTransport struct {
1139+ mock.Mock
1140+}
1141+
1142+type MockDatagramTransport_Expecter struct {
1143+ mock *mock.Mock
1144+}
1145+
1146+func (_m *MockDatagramTransport) EXPECT() *MockDatagramTransport_Expecter {
1147+ return &MockDatagramTransport_Expecter{mock: &_m.Mock}
1148+}
1149+
1150+// Close provides a mock function for the type MockDatagramTransport
1151+func (_mock *MockDatagramTransport) Close() error {
1152+ ret := _mock.Called()
1153+
1154+ if len(ret) == 0 {
1155+ panic("no return value specified for Close")
1156+ }
1157+
1158+ var r0 error
1159+ if returnFunc, ok := ret.Get(0).(func() error); ok {
1160+ r0 = returnFunc()
1161+ } else {
1162+ r0 = ret.Error(0)
1163+ }
1164+ return r0
1165+}
1166+
1167+// MockDatagramTransport_Close_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Close'
1168+type MockDatagramTransport_Close_Call struct {
1169+ *mock.Call
1170+}
1171+
1172+// Close is a helper method to define mock.On call
1173+func (_e *MockDatagramTransport_Expecter) Close() *MockDatagramTransport_Close_Call {
1174+ return &MockDatagramTransport_Close_Call{Call: _e.mock.On("Close")}
1175+}
1176+
1177+func (_c *MockDatagramTransport_Close_Call) Run(run func()) *MockDatagramTransport_Close_Call {
1178+ _c.Call.Run(func(args mock.Arguments) {
1179+ run()
1180+ })
1181+ return _c
1182+}
1183+
1184+func (_c *MockDatagramTransport_Close_Call) Return(err error) *MockDatagramTransport_Close_Call {
1185+ _c.Call.Return(err)
1186+ return _c
1187+}
1188+
1189+func (_c *MockDatagramTransport_Close_Call) RunAndReturn(run func() error) *MockDatagramTransport_Close_Call {
1190+ _c.Call.Return(run)
1191+ return _c
1192+}
1193+
1194+// DatagramTransport provides a mock function for the type MockDatagramTransport
1195+func (_mock *MockDatagramTransport) DatagramTransport() {
1196+ _mock.Called()
1197+ return
1198+}
1199+
1200+// MockDatagramTransport_DatagramTransport_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'DatagramTransport'
1201+type MockDatagramTransport_DatagramTransport_Call struct {
1202+ *mock.Call
1203+}
1204+
1205+// DatagramTransport is a helper method to define mock.On call
1206+func (_e *MockDatagramTransport_Expecter) DatagramTransport() *MockDatagramTransport_DatagramTransport_Call {
1207+ return &MockDatagramTransport_DatagramTransport_Call{Call: _e.mock.On("DatagramTransport")}
1208+}
1209+
1210+func (_c *MockDatagramTransport_DatagramTransport_Call) Run(run func()) *MockDatagramTransport_DatagramTransport_Call {
1211+ _c.Call.Run(func(args mock.Arguments) {
1212+ run()
1213+ })
1214+ return _c
1215+}
1216+
1217+func (_c *MockDatagramTransport_DatagramTransport_Call) Return() *MockDatagramTransport_DatagramTransport_Call {
1218+ _c.Call.Return()
1219+ return _c
1220+}
1221+
1222+func (_c *MockDatagramTransport_DatagramTransport_Call) RunAndReturn(run func()) *MockDatagramTransport_DatagramTransport_Call {
1223+ _c.Run(run)
1224+ return _c
1225+}
1226+
1227+// Recv provides a mock function for the type MockDatagramTransport
1228+func (_mock *MockDatagramTransport) Recv() (ports.Frame, error) {
1229+ ret := _mock.Called()
1230+
1231+ if len(ret) == 0 {
1232+ panic("no return value specified for Recv")
1233+ }
1234+
1235+ var r0 ports.Frame
1236+ var r1 error
1237+ if returnFunc, ok := ret.Get(0).(func() (ports.Frame, error)); ok {
1238+ return returnFunc()
1239+ }
1240+ if returnFunc, ok := ret.Get(0).(func() ports.Frame); ok {
1241+ r0 = returnFunc()
1242+ } else {
1243+ r0 = ret.Get(0).(ports.Frame)
1244+ }
1245+ if returnFunc, ok := ret.Get(1).(func() error); ok {
1246+ r1 = returnFunc()
1247+ } else {
1248+ r1 = ret.Error(1)
1249+ }
1250+ return r0, r1
1251+}
1252+
1253+// MockDatagramTransport_Recv_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Recv'
1254+type MockDatagramTransport_Recv_Call struct {
1255+ *mock.Call
1256+}
1257+
1258+// Recv is a helper method to define mock.On call
1259+func (_e *MockDatagramTransport_Expecter) Recv() *MockDatagramTransport_Recv_Call {
1260+ return &MockDatagramTransport_Recv_Call{Call: _e.mock.On("Recv")}
1261+}
1262+
1263+func (_c *MockDatagramTransport_Recv_Call) Run(run func()) *MockDatagramTransport_Recv_Call {
1264+ _c.Call.Run(func(args mock.Arguments) {
1265+ run()
1266+ })
1267+ return _c
1268+}
1269+
1270+func (_c *MockDatagramTransport_Recv_Call) Return(frame ports.Frame, err error) *MockDatagramTransport_Recv_Call {
1271+ _c.Call.Return(frame, err)
1272+ return _c
1273+}
1274+
1275+func (_c *MockDatagramTransport_Recv_Call) RunAndReturn(run func() (ports.Frame, error)) *MockDatagramTransport_Recv_Call {
1276+ _c.Call.Return(run)
1277+ return _c
1278+}
1279+
1280+// Send provides a mock function for the type MockDatagramTransport
1281+func (_mock *MockDatagramTransport) Send(frame ports.Frame) error {
1282+ ret := _mock.Called(frame)
1283+
1284+ if len(ret) == 0 {
1285+ panic("no return value specified for Send")
1286+ }
1287+
1288+ var r0 error
1289+ if returnFunc, ok := ret.Get(0).(func(ports.Frame) error); ok {
1290+ r0 = returnFunc(frame)
1291+ } else {
1292+ r0 = ret.Error(0)
1293+ }
1294+ return r0
1295+}
1296+
1297+// MockDatagramTransport_Send_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Send'
1298+type MockDatagramTransport_Send_Call struct {
1299+ *mock.Call
1300+}
1301+
1302+// Send is a helper method to define mock.On call
1303+// - frame ports.Frame
1304+func (_e *MockDatagramTransport_Expecter) Send(frame any) *MockDatagramTransport_Send_Call {
1305+ return &MockDatagramTransport_Send_Call{Call: _e.mock.On("Send", frame)}
1306+}
1307+
1308+func (_c *MockDatagramTransport_Send_Call) Run(run func(frame ports.Frame)) *MockDatagramTransport_Send_Call {
1309+ _c.Call.Run(func(args mock.Arguments) {
1310+ var arg0 ports.Frame
1311+ if args[0] != nil {
1312+ arg0 = args[0].(ports.Frame)
1313+ }
1314+ run(
1315+ arg0,
1316+ )
1317+ })
1318+ return _c
1319+}
1320+
1321+func (_c *MockDatagramTransport_Send_Call) Return(err error) *MockDatagramTransport_Send_Call {
1322+ _c.Call.Return(err)
1323+ return _c
1324+}
1325+
1326+func (_c *MockDatagramTransport_Send_Call) RunAndReturn(run func(frame ports.Frame) error) *MockDatagramTransport_Send_Call {
1327+ _c.Call.Return(run)
1328+ return _c
1329+}
1330+
11231331 // NewMockAsyncTransport creates a new instance of MockAsyncTransport. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
11241332 // The first argument is typically a *testing.T value.
11251333 func NewMockAsyncTransport(t interface {
@@ -1120,6 +1120,214 @@ func (_c *MockTransport_Send_Call) RunAndReturn(run func(frame ports.Frame) erro
1120 return _c1120 return _c
1121 }1121 }
1122 1122
1123+// NewMockDatagramTransport creates a new instance of MockDatagramTransport. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
1124+// The first argument is typically a *testing.T value.
1125+func NewMockDatagramTransport(t interface {
1126+ mock.TestingT
1127+ Cleanup(func())
1128+}) *MockDatagramTransport {
1129+ mock := &MockDatagramTransport{}
1130+ mock.Mock.Test(t)
1131+
1132+ t.Cleanup(func() { mock.AssertExpectations(t) })
1133+
1134+ return mock
1135+}
1136+
1137+// MockDatagramTransport is an autogenerated mock type for the DatagramTransport type
1138+type MockDatagramTransport struct {
1139+ mock.Mock
1140+}
1141+
1142+type MockDatagramTransport_Expecter struct {
1143+ mock *mock.Mock
1144+}
1145+
1146+func (_m *MockDatagramTransport) EXPECT() *MockDatagramTransport_Expecter {
1147+ return &MockDatagramTransport_Expecter{mock: &_m.Mock}
1148+}
1149+
1150+// Close provides a mock function for the type MockDatagramTransport
1151+func (_mock *MockDatagramTransport) Close() error {
1152+ ret := _mock.Called()
1153+
1154+ if len(ret) == 0 {
1155+ panic("no return value specified for Close")
1156+ }
1157+
1158+ var r0 error
1159+ if returnFunc, ok := ret.Get(0).(func() error); ok {
1160+ r0 = returnFunc()
1161+ } else {
1162+ r0 = ret.Error(0)
1163+ }
1164+ return r0
1165+}
1166+
1167+// MockDatagramTransport_Close_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Close'
1168+type MockDatagramTransport_Close_Call struct {
1169+ *mock.Call
1170+}
1171+
1172+// Close is a helper method to define mock.On call
1173+func (_e *MockDatagramTransport_Expecter) Close() *MockDatagramTransport_Close_Call {
1174+ return &MockDatagramTransport_Close_Call{Call: _e.mock.On("Close")}
1175+}
1176+
1177+func (_c *MockDatagramTransport_Close_Call) Run(run func()) *MockDatagramTransport_Close_Call {
1178+ _c.Call.Run(func(args mock.Arguments) {
1179+ run()
1180+ })
1181+ return _c
1182+}
1183+
1184+func (_c *MockDatagramTransport_Close_Call) Return(err error) *MockDatagramTransport_Close_Call {
1185+ _c.Call.Return(err)
1186+ return _c
1187+}
1188+
1189+func (_c *MockDatagramTransport_Close_Call) RunAndReturn(run func() error) *MockDatagramTransport_Close_Call {
1190+ _c.Call.Return(run)
1191+ return _c
1192+}
1193+
1194+// DatagramTransport provides a mock function for the type MockDatagramTransport
1195+func (_mock *MockDatagramTransport) DatagramTransport() {
1196+ _mock.Called()
1197+ return
1198+}
1199+
1200+// MockDatagramTransport_DatagramTransport_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'DatagramTransport'
1201+type MockDatagramTransport_DatagramTransport_Call struct {
1202+ *mock.Call
1203+}
1204+
1205+// DatagramTransport is a helper method to define mock.On call
1206+func (_e *MockDatagramTransport_Expecter) DatagramTransport() *MockDatagramTransport_DatagramTransport_Call {
1207+ return &MockDatagramTransport_DatagramTransport_Call{Call: _e.mock.On("DatagramTransport")}
1208+}
1209+
1210+func (_c *MockDatagramTransport_DatagramTransport_Call) Run(run func()) *MockDatagramTransport_DatagramTransport_Call {
1211+ _c.Call.Run(func(args mock.Arguments) {
1212+ run()
1213+ })
1214+ return _c
1215+}
1216+
1217+func (_c *MockDatagramTransport_DatagramTransport_Call) Return() *MockDatagramTransport_DatagramTransport_Call {
1218+ _c.Call.Return()
1219+ return _c
1220+}
1221+
1222+func (_c *MockDatagramTransport_DatagramTransport_Call) RunAndReturn(run func()) *MockDatagramTransport_DatagramTransport_Call {
1223+ _c.Run(run)
1224+ return _c
1225+}
1226+
1227+// Recv provides a mock function for the type MockDatagramTransport
1228+func (_mock *MockDatagramTransport) Recv() (ports.Frame, error) {
1229+ ret := _mock.Called()
1230+
1231+ if len(ret) == 0 {
1232+ panic("no return value specified for Recv")
1233+ }
1234+
1235+ var r0 ports.Frame
1236+ var r1 error
1237+ if returnFunc, ok := ret.Get(0).(func() (ports.Frame, error)); ok {
1238+ return returnFunc()
1239+ }
1240+ if returnFunc, ok := ret.Get(0).(func() ports.Frame); ok {
1241+ r0 = returnFunc()
1242+ } else {
1243+ r0 = ret.Get(0).(ports.Frame)
1244+ }
1245+ if returnFunc, ok := ret.Get(1).(func() error); ok {
1246+ r1 = returnFunc()
1247+ } else {
1248+ r1 = ret.Error(1)
1249+ }
1250+ return r0, r1
1251+}
1252+
1253+// MockDatagramTransport_Recv_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Recv'
1254+type MockDatagramTransport_Recv_Call struct {
1255+ *mock.Call
1256+}
1257+
1258+// Recv is a helper method to define mock.On call
1259+func (_e *MockDatagramTransport_Expecter) Recv() *MockDatagramTransport_Recv_Call {
1260+ return &MockDatagramTransport_Recv_Call{Call: _e.mock.On("Recv")}
1261+}
1262+
1263+func (_c *MockDatagramTransport_Recv_Call) Run(run func()) *MockDatagramTransport_Recv_Call {
1264+ _c.Call.Run(func(args mock.Arguments) {
1265+ run()
1266+ })
1267+ return _c
1268+}
1269+
1270+func (_c *MockDatagramTransport_Recv_Call) Return(frame ports.Frame, err error) *MockDatagramTransport_Recv_Call {
1271+ _c.Call.Return(frame, err)
1272+ return _c
1273+}
1274+
1275+func (_c *MockDatagramTransport_Recv_Call) RunAndReturn(run func() (ports.Frame, error)) *MockDatagramTransport_Recv_Call {
1276+ _c.Call.Return(run)
1277+ return _c
1278+}
1279+
1280+// Send provides a mock function for the type MockDatagramTransport
1281+func (_mock *MockDatagramTransport) Send(frame ports.Frame) error {
1282+ ret := _mock.Called(frame)
1283+
1284+ if len(ret) == 0 {
1285+ panic("no return value specified for Send")
1286+ }
1287+
1288+ var r0 error
1289+ if returnFunc, ok := ret.Get(0).(func(ports.Frame) error); ok {
1290+ r0 = returnFunc(frame)
1291+ } else {
1292+ r0 = ret.Error(0)
1293+ }
1294+ return r0
1295+}
1296+
1297+// MockDatagramTransport_Send_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Send'
1298+type MockDatagramTransport_Send_Call struct {
1299+ *mock.Call
1300+}
1301+
1302+// Send is a helper method to define mock.On call
1303+// - frame ports.Frame
1304+func (_e *MockDatagramTransport_Expecter) Send(frame any) *MockDatagramTransport_Send_Call {
1305+ return &MockDatagramTransport_Send_Call{Call: _e.mock.On("Send", frame)}
1306+}
1307+
1308+func (_c *MockDatagramTransport_Send_Call) Run(run func(frame ports.Frame)) *MockDatagramTransport_Send_Call {
1309+ _c.Call.Run(func(args mock.Arguments) {
1310+ var arg0 ports.Frame
1311+ if args[0] != nil {
1312+ arg0 = args[0].(ports.Frame)
1313+ }
1314+ run(
1315+ arg0,
1316+ )
1317+ })
1318+ return _c
1319+}
1320+
1321+func (_c *MockDatagramTransport_Send_Call) Return(err error) *MockDatagramTransport_Send_Call {
1322+ _c.Call.Return(err)
1323+ return _c
1324+}
1325+
1326+func (_c *MockDatagramTransport_Send_Call) RunAndReturn(run func(frame ports.Frame) error) *MockDatagramTransport_Send_Call {
1327+ _c.Call.Return(run)
1328+ return _c
1329+}
1330+
1123 // NewMockAsyncTransport creates a new instance of MockAsyncTransport. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.1331 // NewMockAsyncTransport creates a new instance of MockAsyncTransport. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
1124 // The first argument is typically a *testing.T value.1332 // The first argument is typically a *testing.T value.
1125 func NewMockAsyncTransport(t interface {1333 func NewMockAsyncTransport(t interface {
M internal/ports/ports.go +7 -0
@@ -45,6 +45,13 @@ type Transport interface {
4545 Close() error
4646 }
4747
48+// DatagramTransport marks transports backed by a datagram link. It lets
49+// usecases negotiate a conservative output window without importing adapters.
50+type DatagramTransport interface {
51+ Transport
52+ DatagramTransport()
53+}
54+
4855 // AsyncTransport accepts frames for ordered background transmission. SendAsync
4956 // returns once the adapter owns the frame; Send retains its synchronous wire-
5057 // attempt contract. Daemon paint output may use this capability to pipeline.
@@ -45,6 +45,13 @@ type Transport interface {
45 Close() error45 Close() error
46 }46 }
47 47
48+// DatagramTransport marks transports backed by a datagram link. It lets
49+// usecases negotiate a conservative output window without importing adapters.
50+type DatagramTransport interface {
51+ Transport
52+ DatagramTransport()
53+}
54+
48 // AsyncTransport accepts frames for ordered background transmission. SendAsync55 // AsyncTransport accepts frames for ordered background transmission. SendAsync
49 // returns once the adapter owns the frame; Send retains its synchronous wire-56 // returns once the adapter owns the frame; Send retains its synchronous wire-
50 // attempt contract. Daemon paint output may use this capability to pipeline.57 // attempt contract. Daemon paint output may use this capability to pipeline.
M internal/ports/wire.go +5 -11
@@ -66,9 +66,9 @@ type Hello struct {
6666 TermEnv string
6767 Cwd string
6868 TrueColor bool
69- // AckOutput is reserved for wire compatibility in protocol v13. The daemon
70- // ignores it because v13 clients cumulatively ACK output on every transport.
71- AckOutput bool
69+ // MaxOutputInFlight is the requested maximum number of unacknowledged
70+ // state-bearing output frames.
71+ MaxOutputInFlight uint8
7272 }
7373
7474 // Input carries raw bytes typed/pasted by the client, destined for the PTY.
@@ -312,11 +312,7 @@ func MarshalHello(h Hello) []byte {
312312 } else {
313313 w.putUint8(0)
314314 }
315- if h.AckOutput {
316- w.putUint8(1)
317- } else {
318- w.putUint8(0)
319- }
315+ w.putUint8(h.MaxOutputInFlight)
320316 return w.b
321317 }
322318
@@ -363,11 +359,9 @@ func UnmarshalHello(b []byte) (Hello, error) {
363359 return Hello{}, err
364360 }
365361 h.TrueColor = trueColor != 0
366- ackOutput, err := r.getUint8()
367- if err != nil {
362+ if h.MaxOutputInFlight, err = r.getUint8(); err != nil {
368363 return Hello{}, err
369364 }
370- h.AckOutput = ackOutput != 0
371365 if err := r.done(); err != nil {
372366 return Hello{}, err
373367 }
@@ -66,9 +66,9 @@ type Hello struct {
66 TermEnv string66 TermEnv string
67 Cwd string67 Cwd string
68 TrueColor bool68 TrueColor bool
69- // AckOutput is reserved for wire compatibility in protocol v13. The daemon69+ // MaxOutputInFlight is the requested maximum number of unacknowledged
70- // ignores it because v13 clients cumulatively ACK output on every transport.70+ // state-bearing output frames.
71- AckOutput bool71+ MaxOutputInFlight uint8
72 }72 }
73 73
74 // Input carries raw bytes typed/pasted by the client, destined for the PTY.74 // Input carries raw bytes typed/pasted by the client, destined for the PTY.
@@ -312,11 +312,7 @@ func MarshalHello(h Hello) []byte {
312 } else {312 } else {
313 w.putUint8(0)313 w.putUint8(0)
314 }314 }
315- if h.AckOutput {315+ w.putUint8(h.MaxOutputInFlight)
316- w.putUint8(1)
317- } else {
318- w.putUint8(0)
319- }
320 return w.b316 return w.b
321 }317 }
322 318
@@ -363,11 +359,9 @@ func UnmarshalHello(b []byte) (Hello, error) {
363 return Hello{}, err359 return Hello{}, err
364 }360 }
365 h.TrueColor = trueColor != 0361 h.TrueColor = trueColor != 0
366- ackOutput, err := r.getUint8()362+ if h.MaxOutputInFlight, err = r.getUint8(); err != nil {
367- if err != nil {
368 return Hello{}, err363 return Hello{}, err
369 }364 }
370- h.AckOutput = ackOutput != 0
371 if err := r.done(); err != nil {365 if err := r.done(); err != nil {
372 return Hello{}, err366 return Hello{}, err
373 }367 }
M internal/ports/wire_test.go +38 -11
@@ -8,6 +8,7 @@ import (
88
99 "github.com/bnema/vev/internal/domain"
1010 "github.com/bnema/vev/pkg/renderer"
11+ "github.com/stretchr/testify/require"
1112 )
1213
1314 // mustNotPanic runs f and fails the test (instead of crashing the test
@@ -60,18 +61,18 @@ func TestHelloGoldenAndRoundTrip(t *testing.T) {
6061 {
6162 name: "typical",
6263 msg: Hello{
63- Version: 1,
64- Intent: IntentNew,
65- ClientID: [16]byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10},
66- ResumeToken: 0x0102030405060708,
67- Name: "w0",
68- Size: domain.Size{Cols: 80, Rows: 24},
69- TermEnv: "xterm-256color",
70- Cwd: "/tmp/project",
71- TrueColor: true,
72- AckOutput: true,
64+ Version: 1,
65+ Intent: IntentNew,
66+ ClientID: [16]byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10},
67+ ResumeToken: 0x0102030405060708,
68+ Name: "w0",
69+ Size: domain.Size{Cols: 80, Rows: 24},
70+ TermEnv: "xterm-256color",
71+ Cwd: "/tmp/project",
72+ TrueColor: true,
73+ MaxOutputInFlight: 8,
7374 },
74- want: []byte{0x00, 0x01, 0x01, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x00, 0x02, 0x77, 0x30, 0x00, 0x50, 0x00, 0x18, 0x00, 0x0e, 0x78, 0x74, 0x65, 0x72, 0x6d, 0x2d, 0x32, 0x35, 0x36, 0x63, 0x6f, 0x6c, 0x6f, 0x72, 0x00, 0x0c, 0x2f, 0x74, 0x6d, 0x70, 0x2f, 0x70, 0x72, 0x6f, 0x6a, 0x65, 0x63, 0x74, 0x01, 0x01},
75+ want: []byte{0x00, 0x01, 0x01, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x00, 0x02, 0x77, 0x30, 0x00, 0x50, 0x00, 0x18, 0x00, 0x0e, 0x78, 0x74, 0x65, 0x72, 0x6d, 0x2d, 0x32, 0x35, 0x36, 0x63, 0x6f, 0x6c, 0x6f, 0x72, 0x00, 0x0c, 0x2f, 0x74, 0x6d, 0x70, 0x2f, 0x70, 0x72, 0x6f, 0x6a, 0x65, 0x63, 0x74, 0x01, 0x08},
7576 },
7677 {
7778 name: "empty strings",
@@ -568,3 +569,29 @@ func TestMsgTypeConstantsDistinct(t *testing.T) {
568569 }
569570 })
570571 }
572+
573+func TestHelloOutputWindowByteExactValues(t *testing.T) {
574+ for _, window := range []uint8{0, 1, 8} {
575+ t.Run(fmt.Sprintf("window_%d", window), func(t *testing.T) {
576+ hello := Hello{Version: 14, MaxOutputInFlight: window}
577+ // Every preceding field in this deliberately empty Hello has a fixed,
578+ // byte-exact zero encoding; the final byte is the negotiated window.
579+ want := append(make([]byte, 38), window)
580+ want[1] = 14
581+ got := MarshalHello(hello)
582+ requireBytesEqual(t, want, got)
583+ back, err := UnmarshalHello(got)
584+ require.NoError(t, err)
585+ require.Equal(t, window, back.MaxOutputInFlight)
586+ assertAllPrefixesFail(t, got, UnmarshalHello)
587+ assertTrailingGarbageFails(t, got, UnmarshalHello)
588+ })
589+ }
590+}
591+
592+func requireBytesEqual(t *testing.T, want, got []byte) {
593+ t.Helper()
594+ if !bytes.Equal(want, got) {
595+ t.Fatalf("bytes = %#v, want %#v", got, want)
596+ }
597+}
@@ -8,6 +8,7 @@ import (
8 8
9 "github.com/bnema/vev/internal/domain"9 "github.com/bnema/vev/internal/domain"
10 "github.com/bnema/vev/pkg/renderer"10 "github.com/bnema/vev/pkg/renderer"
11+ "github.com/stretchr/testify/require"
11 )12 )
12 13
13 // mustNotPanic runs f and fails the test (instead of crashing the test14 // mustNotPanic runs f and fails the test (instead of crashing the test
@@ -60,18 +61,18 @@ func TestHelloGoldenAndRoundTrip(t *testing.T) {
60 {61 {
61 name: "typical",62 name: "typical",
62 msg: Hello{63 msg: Hello{
63- Version: 1,64+ Version: 1,
64- Intent: IntentNew,65+ Intent: IntentNew,
65- ClientID: [16]byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10},66+ ClientID: [16]byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10},
66- ResumeToken: 0x0102030405060708,67+ ResumeToken: 0x0102030405060708,
67- Name: "w0",68+ Name: "w0",
68- Size: domain.Size{Cols: 80, Rows: 24},69+ Size: domain.Size{Cols: 80, Rows: 24},
69- TermEnv: "xterm-256color",70+ TermEnv: "xterm-256color",
70- Cwd: "/tmp/project",71+ Cwd: "/tmp/project",
71- TrueColor: true,72+ TrueColor: true,
72- AckOutput: true,73+ MaxOutputInFlight: 8,
73 },74 },
74- want: []byte{0x00, 0x01, 0x01, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x00, 0x02, 0x77, 0x30, 0x00, 0x50, 0x00, 0x18, 0x00, 0x0e, 0x78, 0x74, 0x65, 0x72, 0x6d, 0x2d, 0x32, 0x35, 0x36, 0x63, 0x6f, 0x6c, 0x6f, 0x72, 0x00, 0x0c, 0x2f, 0x74, 0x6d, 0x70, 0x2f, 0x70, 0x72, 0x6f, 0x6a, 0x65, 0x63, 0x74, 0x01, 0x01},75+ want: []byte{0x00, 0x01, 0x01, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x00, 0x02, 0x77, 0x30, 0x00, 0x50, 0x00, 0x18, 0x00, 0x0e, 0x78, 0x74, 0x65, 0x72, 0x6d, 0x2d, 0x32, 0x35, 0x36, 0x63, 0x6f, 0x6c, 0x6f, 0x72, 0x00, 0x0c, 0x2f, 0x74, 0x6d, 0x70, 0x2f, 0x70, 0x72, 0x6f, 0x6a, 0x65, 0x63, 0x74, 0x01, 0x08},
75 },76 },
76 {77 {
77 name: "empty strings",78 name: "empty strings",
@@ -568,3 +569,29 @@ func TestMsgTypeConstantsDistinct(t *testing.T) {
568 }569 }
569 })570 })
570 }571 }
572+
573+func TestHelloOutputWindowByteExactValues(t *testing.T) {
574+ for _, window := range []uint8{0, 1, 8} {
575+ t.Run(fmt.Sprintf("window_%d", window), func(t *testing.T) {
576+ hello := Hello{Version: 14, MaxOutputInFlight: window}
577+ // Every preceding field in this deliberately empty Hello has a fixed,
578+ // byte-exact zero encoding; the final byte is the negotiated window.
579+ want := append(make([]byte, 38), window)
580+ want[1] = 14
581+ got := MarshalHello(hello)
582+ requireBytesEqual(t, want, got)
583+ back, err := UnmarshalHello(got)
584+ require.NoError(t, err)
585+ require.Equal(t, window, back.MaxOutputInFlight)
586+ assertAllPrefixesFail(t, got, UnmarshalHello)
587+ assertTrailingGarbageFails(t, got, UnmarshalHello)
588+ })
589+ }
590+}
591+
592+func requireBytesEqual(t *testing.T, want, got []byte) {
593+ t.Helper()
594+ if !bytes.Equal(want, got) {
595+ t.Fatalf("bytes = %#v, want %#v", got, want)
596+ }
597+}
M internal/usecase/client/client.go +17 -12
@@ -339,6 +339,13 @@ func DetectTrueColor(termEnv, colorTerm string) bool {
339339 return termEnv == "xterm-direct" || strings.HasSuffix(termEnv, "-direct")
340340 }
341341
342+func requestedOutputWindow(transport ports.Transport) uint8 {
343+ if _, ok := transport.(ports.DatagramTransport); ok {
344+ return 1
345+ }
346+ return 8
347+}
348+
342349 func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms *milestones, enterRaw func() error, clearStatus func(), drawStatusStage func(reconnectStage), linkEvents <-chan ports.LinkEvent, remote bool, clipboard ports.ClipboardReader, log *slog.Logger) attachResult {
343350 // 1. Handshake: send Hello with our size and TERM.
344351 size, err := term.Size()
@@ -353,18 +360,16 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi
353360 colorTerm := os.Getenv("COLORTERM")
354361 trueColor := DetectTrueColor(termEnv, colorTerm)
355362 hello := ports.Hello{
356- Version: ports.ProtocolVersion,
357- Intent: intent,
358- ClientID: clientID,
359- ResumeToken: resumeToken,
360- Name: name,
361- Size: size,
362- TermEnv: termEnv,
363- Cwd: cwd,
364- TrueColor: trueColor,
365- // Protocol v13 requires cumulative output ACKs on every transport. Keep
366- // the field asserted while it remains in the wire layout.
367- AckOutput: true,
363+ Version: ports.ProtocolVersion,
364+ Intent: intent,
365+ ClientID: clientID,
366+ ResumeToken: resumeToken,
367+ Name: name,
368+ Size: size,
369+ TermEnv: termEnv,
370+ Cwd: cwd,
371+ TrueColor: trueColor,
372+ MaxOutputInFlight: requestedOutputWindow(transport),
368373 }
369374 if err := transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil {
370375 return attachResult{err: fmt.Errorf("vev: sending hello: %w", err)}
@@ -339,6 +339,13 @@ func DetectTrueColor(termEnv, colorTerm string) bool {
339 return termEnv == "xterm-direct" || strings.HasSuffix(termEnv, "-direct")339 return termEnv == "xterm-direct" || strings.HasSuffix(termEnv, "-direct")
340 }340 }
341 341
342+func requestedOutputWindow(transport ports.Transport) uint8 {
343+ if _, ok := transport.(ports.DatagramTransport); ok {
344+ return 1
345+ }
346+ return 8
347+}
348+
342 func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms *milestones, enterRaw func() error, clearStatus func(), drawStatusStage func(reconnectStage), linkEvents <-chan ports.LinkEvent, remote bool, clipboard ports.ClipboardReader, log *slog.Logger) attachResult {349 func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms *milestones, enterRaw func() error, clearStatus func(), drawStatusStage func(reconnectStage), linkEvents <-chan ports.LinkEvent, remote bool, clipboard ports.ClipboardReader, log *slog.Logger) attachResult {
343 // 1. Handshake: send Hello with our size and TERM.350 // 1. Handshake: send Hello with our size and TERM.
344 size, err := term.Size()351 size, err := term.Size()
@@ -353,18 +360,16 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi
353 colorTerm := os.Getenv("COLORTERM")360 colorTerm := os.Getenv("COLORTERM")
354 trueColor := DetectTrueColor(termEnv, colorTerm)361 trueColor := DetectTrueColor(termEnv, colorTerm)
355 hello := ports.Hello{362 hello := ports.Hello{
356- Version: ports.ProtocolVersion,363+ Version: ports.ProtocolVersion,
357- Intent: intent,364+ Intent: intent,
358- ClientID: clientID,365+ ClientID: clientID,
359- ResumeToken: resumeToken,366+ ResumeToken: resumeToken,
360- Name: name,367+ Name: name,
361- Size: size,368+ Size: size,
362- TermEnv: termEnv,369+ TermEnv: termEnv,
363- Cwd: cwd,370+ Cwd: cwd,
364- TrueColor: trueColor,371+ TrueColor: trueColor,
365- // Protocol v13 requires cumulative output ACKs on every transport. Keep372+ MaxOutputInFlight: requestedOutputWindow(transport),
366- // the field asserted while it remains in the wire layout.
367- AckOutput: true,
368 }373 }
369 if err := transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil {374 if err := transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil {
370 return attachResult{err: fmt.Errorf("vev: sending hello: %w", err)}375 return attachResult{err: fmt.Errorf("vev: sending hello: %w", err)}
M internal/usecase/client/client_test.go +29 -1
@@ -46,6 +46,10 @@ type recvItem struct {
4646 err error
4747 }
4848
49+type markedDatagramTransport struct{ ports.Transport }
50+
51+func (markedDatagramTransport) DatagramTransport() {}
52+
4953 // scriptRecv wires a MockTransport's Recv to yield the given items in order,
5054 // then block forever (simulating a live but idle connection). It returns a
5155 // cleanup that unblocks the parked reader.
@@ -183,12 +187,36 @@ func TestAttachHelloIncludesTrueColor(t *testing.T) {
183187 case hello := <-gotHello:
184188 require.Equal(t, "xterm-256color", hello.TermEnv)
185189 require.True(t, hello.TrueColor)
186- require.True(t, hello.AckOutput)
190+ require.Equal(t, uint8(8), hello.MaxOutputInFlight)
187191 case <-time.After(2 * time.Second):
188192 t.Fatal("hello frame was not sent")
189193 }
190194 }
191195
196+func TestAttachHelloRequestsSingleOutputForDatagramTransport(t *testing.T) {
197+ var out bytes.Buffer
198+ var restoreCount atomic.Int32
199+ resizeCh := make(chan domain.Size)
200+ tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
201+ defer in.unblock()
202+
203+ tr := portsmocks.NewMockTransport(t)
204+ tr.EXPECT().Send(isType(ports.MsgHello)).RunAndReturn(func(f ports.Frame) error {
205+ hello, err := ports.UnmarshalHello(f.Payload)
206+ require.NoError(t, err)
207+ require.Equal(t, uint8(1), hello.MaxOutputInFlight)
208+ return nil
209+ }).Once()
210+ unblock := scriptRecv(tr,
211+ recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
212+ recvItem{f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
213+ )
214+ defer unblock()
215+ tr.EXPECT().Close().Return(nil).Once()
216+
217+ require.NoError(t, client.Attach(context.Background(), markedDatagramTransport{tr}, tm, realClock{}, ports.IntentEphemeral, ""))
218+}
219+
192220 func TestAttachHappyPath(t *testing.T) {
193221 var out bytes.Buffer
194222 var restoreCount atomic.Int32
@@ -46,6 +46,10 @@ type recvItem struct {
46 err error46 err error
47 }47 }
48 48
49+type markedDatagramTransport struct{ ports.Transport }
50+
51+func (markedDatagramTransport) DatagramTransport() {}
52+
49 // scriptRecv wires a MockTransport's Recv to yield the given items in order,53 // scriptRecv wires a MockTransport's Recv to yield the given items in order,
50 // then block forever (simulating a live but idle connection). It returns a54 // then block forever (simulating a live but idle connection). It returns a
51 // cleanup that unblocks the parked reader.55 // cleanup that unblocks the parked reader.
@@ -183,12 +187,36 @@ func TestAttachHelloIncludesTrueColor(t *testing.T) {
183 case hello := <-gotHello:187 case hello := <-gotHello:
184 require.Equal(t, "xterm-256color", hello.TermEnv)188 require.Equal(t, "xterm-256color", hello.TermEnv)
185 require.True(t, hello.TrueColor)189 require.True(t, hello.TrueColor)
186- require.True(t, hello.AckOutput)190+ require.Equal(t, uint8(8), hello.MaxOutputInFlight)
187 case <-time.After(2 * time.Second):191 case <-time.After(2 * time.Second):
188 t.Fatal("hello frame was not sent")192 t.Fatal("hello frame was not sent")
189 }193 }
190 }194 }
191 195
196+func TestAttachHelloRequestsSingleOutputForDatagramTransport(t *testing.T) {
197+ var out bytes.Buffer
198+ var restoreCount atomic.Int32
199+ resizeCh := make(chan domain.Size)
200+ tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
201+ defer in.unblock()
202+
203+ tr := portsmocks.NewMockTransport(t)
204+ tr.EXPECT().Send(isType(ports.MsgHello)).RunAndReturn(func(f ports.Frame) error {
205+ hello, err := ports.UnmarshalHello(f.Payload)
206+ require.NoError(t, err)
207+ require.Equal(t, uint8(1), hello.MaxOutputInFlight)
208+ return nil
209+ }).Once()
210+ unblock := scriptRecv(tr,
211+ recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
212+ recvItem{f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
213+ )
214+ defer unblock()
215+ tr.EXPECT().Close().Return(nil).Once()
216+
217+ require.NoError(t, client.Attach(context.Background(), markedDatagramTransport{tr}, tm, realClock{}, ports.IntentEphemeral, ""))
218+}
219+
192 func TestAttachHappyPath(t *testing.T) {220 func TestAttachHappyPath(t *testing.T) {
193 var out bytes.Buffer221 var out bytes.Buffer
194 var restoreCount atomic.Int32222 var restoreCount atomic.Int32
M internal/usecase/daemon/client.go +4 -3
@@ -370,8 +370,9 @@ func (d *Daemon) notifiesSnapshot() []chan struct{} {
370370 }
371371
372372 type attachClientOptions struct {
373- clientID [16]byte
374- resumeCapable bool
373+ clientID [16]byte
374+ resumeCapable bool
375+ maxOutputInFlight uint8
375376 }
376377
377378 func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, opts attachClientOptions) (*attachedClient, *attachedClient) {
@@ -381,7 +382,7 @@ func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size,
381382 }
382383 ac := &attachedClient{
383384 tr: tr,
384- output: newOutputStateStream(),
385+ output: newOutputStateStream(opts.maxOutputInFlight),
385386 size: sz,
386387 clientID: opts.clientID,
387388 resumeCapable: opts.resumeCapable,
@@ -370,8 +370,9 @@ func (d *Daemon) notifiesSnapshot() []chan struct{} {
370 }370 }
371 371
372 type attachClientOptions struct {372 type attachClientOptions struct {
373- clientID [16]byte373+ clientID [16]byte
374- resumeCapable bool374+ resumeCapable bool
375+ maxOutputInFlight uint8
375 }376 }
376 377
377 func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, opts attachClientOptions) (*attachedClient, *attachedClient) {378 func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, opts attachClientOptions) (*attachedClient, *attachedClient) {
@@ -381,7 +382,7 @@ func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size,
381 }382 }
382 ac := &attachedClient{383 ac := &attachedClient{
383 tr: tr,384 tr: tr,
384- output: newOutputStateStream(),385+ output: newOutputStateStream(opts.maxOutputInFlight),
385 size: sz,386 size: sz,
386 clientID: opts.clientID,387 clientID: opts.clientID,
387 resumeCapable: opts.resumeCapable,388 resumeCapable: opts.resumeCapable,
M internal/usecase/daemon/daemon.go +13 -4
@@ -74,6 +74,15 @@ const detachNotifyTimeout = time.Second
7474 // the proxy's reliable queue never fills from painting alone.
7575 const maxUnackedOutputStates = 8
7676
77+// normalizeOutputWindow bounds the untrusted Hello value. Zero deliberately
78+// means the legacy/default window, so malformed or absent values remain safe.
79+func normalizeOutputWindow(window uint8) uint8 {
80+ if window == 0 || window > maxUnackedOutputStates {
81+ return maxUnackedOutputStates
82+ }
83+ return window
84+}
85+
7786 const defaultResumeParkGrace = 15 * time.Minute
7887
7988 // defaultSize is used when a client's Hello carries no valid dimensions.
@@ -667,7 +676,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl
667676 }
668677 }
669678 d.purgeParkedForSessionLocked(sess)
670- ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true})
679+ ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
671680 sess.mu.Lock()
672681 sess.terminal = term
673682 sess.mu.Unlock()
@@ -683,7 +692,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl
683692 return nil, nil, err
684693 }
685694 d.purgeParkedForSessionLocked(sess)
686- ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true})
695+ ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
687696 sess.mu.Lock()
688697 sess.terminal = term
689698 sess.mu.Unlock()
@@ -710,7 +719,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl
710719 return nil, nil, err
711720 }
712721 d.purgeParkedForSessionLocked(sess)
713- ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true})
722+ ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
714723 sess.mu.Lock()
715724 sess.terminal = term
716725 sess.mu.Unlock()
@@ -735,7 +744,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl
735744 }
736745 }
737746 d.purgeParkedForSessionLocked(sess)
738- ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true})
747+ ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
739748 sess.mu.Lock()
740749 sess.terminal = term
741750 sess.mu.Unlock()
@@ -74,6 +74,15 @@ const detachNotifyTimeout = time.Second
74 // the proxy's reliable queue never fills from painting alone.74 // the proxy's reliable queue never fills from painting alone.
75 const maxUnackedOutputStates = 875 const maxUnackedOutputStates = 8
76 76
77+// normalizeOutputWindow bounds the untrusted Hello value. Zero deliberately
78+// means the legacy/default window, so malformed or absent values remain safe.
79+func normalizeOutputWindow(window uint8) uint8 {
80+ if window == 0 || window > maxUnackedOutputStates {
81+ return maxUnackedOutputStates
82+ }
83+ return window
84+}
85+
77 const defaultResumeParkGrace = 15 * time.Minute86 const defaultResumeParkGrace = 15 * time.Minute
78 87
79 // defaultSize is used when a client's Hello carries no valid dimensions.88 // defaultSize is used when a client's Hello carries no valid dimensions.
@@ -667,7 +676,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl
667 }676 }
668 }677 }
669 d.purgeParkedForSessionLocked(sess)678 d.purgeParkedForSessionLocked(sess)
670- ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true})679+ ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
671 sess.mu.Lock()680 sess.mu.Lock()
672 sess.terminal = term681 sess.terminal = term
673 sess.mu.Unlock()682 sess.mu.Unlock()
@@ -683,7 +692,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl
683 return nil, nil, err692 return nil, nil, err
684 }693 }
685 d.purgeParkedForSessionLocked(sess)694 d.purgeParkedForSessionLocked(sess)
686- ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true})695+ ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
687 sess.mu.Lock()696 sess.mu.Lock()
688 sess.terminal = term697 sess.terminal = term
689 sess.mu.Unlock()698 sess.mu.Unlock()
@@ -710,7 +719,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl
710 return nil, nil, err719 return nil, nil, err
711 }720 }
712 d.purgeParkedForSessionLocked(sess)721 d.purgeParkedForSessionLocked(sess)
713- ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true})722+ ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
714 sess.mu.Lock()723 sess.mu.Lock()
715 sess.terminal = term724 sess.terminal = term
716 sess.mu.Unlock()725 sess.mu.Unlock()
@@ -735,7 +744,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl
735 }744 }
736 }745 }
737 d.purgeParkedForSessionLocked(sess)746 d.purgeParkedForSessionLocked(sess)
738- ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true})747+ ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
739 sess.mu.Lock()748 sess.mu.Lock()
740 sess.terminal = term749 sess.terminal = term
741 sess.mu.Unlock()750 sess.mu.Unlock()
M internal/usecase/daemon/output_state_stream.go +14 -8
@@ -14,16 +14,22 @@ import (
1414 //
1515 // Callers serialize access with attachedClient.sendMu.
1616 type outputStateStream struct {
17- renderer *renderer.Renderer
18- next uint64
19- acked uint64
20- deferred bool
21- deferredReset bool
17+ renderer *renderer.Renderer
18+ next uint64
19+ acked uint64
20+ deferred bool
21+ deferredReset bool
22+ maxOutstanding uint64
2223 }
2324
24-func newOutputStateStream() *outputStateStream {
25+func newOutputStateStream(max ...uint8) *outputStateStream {
26+ window := uint8(maxUnackedOutputStates)
27+ if len(max) > 0 {
28+ window = normalizeOutputWindow(max[0])
29+ }
2530 return &outputStateStream{
26- renderer: renderer.New(renderer.Capabilities{}),
31+ renderer: renderer.New(renderer.Capabilities{}),
32+ maxOutstanding: uint64(window),
2733 }
2834 }
2935
@@ -87,7 +93,7 @@ func (s *outputStateStream) rebase() {
8793 }
8894
8995 func (s *outputStateStream) atCapacity() bool {
90- return s.outstanding() >= maxUnackedOutputStates
96+ return s.outstanding() >= s.maxOutstanding
9197 }
9298
9399 func (s *outputStateStream) deferIfAtCapacity(reset bool) bool {
@@ -14,16 +14,22 @@ import (
14 //14 //
15 // Callers serialize access with attachedClient.sendMu.15 // Callers serialize access with attachedClient.sendMu.
16 type outputStateStream struct {16 type outputStateStream struct {
17- renderer *renderer.Renderer17+ renderer *renderer.Renderer
18- next uint6418+ next uint64
19- acked uint6419+ acked uint64
20- deferred bool20+ deferred bool
21- deferredReset bool21+ deferredReset bool
22+ maxOutstanding uint64
22 }23 }
23 24
24-func newOutputStateStream() *outputStateStream {25+func newOutputStateStream(max ...uint8) *outputStateStream {
26+ window := uint8(maxUnackedOutputStates)
27+ if len(max) > 0 {
28+ window = normalizeOutputWindow(max[0])
29+ }
25 return &outputStateStream{30 return &outputStateStream{
26- renderer: renderer.New(renderer.Capabilities{}),31+ renderer: renderer.New(renderer.Capabilities{}),
32+ maxOutstanding: uint64(window),
27 }33 }
28 }34 }
29 35
@@ -87,7 +93,7 @@ func (s *outputStateStream) rebase() {
87 }93 }
88 94
89 func (s *outputStateStream) atCapacity() bool {95 func (s *outputStateStream) atCapacity() bool {
90- return s.outstanding() >= maxUnackedOutputStates96+ return s.outstanding() >= s.maxOutstanding
91 }97 }
92 98
93 func (s *outputStateStream) deferIfAtCapacity(reset bool) bool {99 func (s *outputStateStream) deferIfAtCapacity(reset bool) bool {
M internal/usecase/daemon/output_state_stream_test.go +43 -0
@@ -123,3 +123,46 @@ func drawOutputState(t *testing.T, stream *outputStateStream, frame renderer.Fra
123123 }
124124 return stream.frame(data, reset, echoAck), true, nil
125125 }
126+
127+func TestOutputStateStreamWindowOneCoalescesUntilCumulativeAck(t *testing.T) {
128+ stream := newOutputStateStream(1)
129+ stream.frame([]byte("first"), true, 0)
130+ for range 100 {
131+ require.True(t, stream.deferIfAtCapacity(false))
132+ }
133+ require.Equal(t, uint64(1), stream.outstanding())
134+
135+ // A stale ACK cannot release the coalesced paint; the cumulative ACK for
136+ // the actually sent state does.
137+ stream.ack(0)
138+ _, ok := stream.takeDeferred()
139+ require.False(t, ok)
140+ stream.ack(1)
141+ reset, ok := stream.takeDeferred()
142+ require.True(t, ok)
143+ require.False(t, reset)
144+ stream.frame([]byte("latest"), reset, 0)
145+ require.Equal(t, uint64(1), stream.outstanding())
146+
147+ sideEffect := stream.sideEffect([]byte("reliable"), 0)
148+ output, err := ports.UnmarshalOutput(sideEffect.Payload)
149+ require.NoError(t, err)
150+ require.Zero(t, output.BaseStateNum)
151+ require.Zero(t, output.NewStateNum)
152+ require.Equal(t, uint64(2), stream.next)
153+}
154+
155+func TestOutputStateStreamDefaultsAndNormalizesWindow(t *testing.T) {
156+ require.Equal(t, uint64(8), newOutputStateStream().maxOutstanding)
157+ require.Equal(t, uint64(8), newOutputStateStream(0).maxOutstanding)
158+ require.Equal(t, uint64(8), newOutputStateStream(9).maxOutstanding)
159+ require.Equal(t, uint64(1), newOutputStateStream(1).maxOutstanding)
160+}
161+
162+func TestNormalizeOutputWindow(t *testing.T) {
163+ for _, tt := range []struct {
164+ input, want uint8
165+ }{{0, 8}, {1, 1}, {8, 8}, {9, 8}, {255, 8}} {
166+ require.Equal(t, tt.want, normalizeOutputWindow(tt.input))
167+ }
168+}
@@ -123,3 +123,46 @@ func drawOutputState(t *testing.T, stream *outputStateStream, frame renderer.Fra
123 }123 }
124 return stream.frame(data, reset, echoAck), true, nil124 return stream.frame(data, reset, echoAck), true, nil
125 }125 }
126+
127+func TestOutputStateStreamWindowOneCoalescesUntilCumulativeAck(t *testing.T) {
128+ stream := newOutputStateStream(1)
129+ stream.frame([]byte("first"), true, 0)
130+ for range 100 {
131+ require.True(t, stream.deferIfAtCapacity(false))
132+ }
133+ require.Equal(t, uint64(1), stream.outstanding())
134+
135+ // A stale ACK cannot release the coalesced paint; the cumulative ACK for
136+ // the actually sent state does.
137+ stream.ack(0)
138+ _, ok := stream.takeDeferred()
139+ require.False(t, ok)
140+ stream.ack(1)
141+ reset, ok := stream.takeDeferred()
142+ require.True(t, ok)
143+ require.False(t, reset)
144+ stream.frame([]byte("latest"), reset, 0)
145+ require.Equal(t, uint64(1), stream.outstanding())
146+
147+ sideEffect := stream.sideEffect([]byte("reliable"), 0)
148+ output, err := ports.UnmarshalOutput(sideEffect.Payload)
149+ require.NoError(t, err)
150+ require.Zero(t, output.BaseStateNum)
151+ require.Zero(t, output.NewStateNum)
152+ require.Equal(t, uint64(2), stream.next)
153+}
154+
155+func TestOutputStateStreamDefaultsAndNormalizesWindow(t *testing.T) {
156+ require.Equal(t, uint64(8), newOutputStateStream().maxOutstanding)
157+ require.Equal(t, uint64(8), newOutputStateStream(0).maxOutstanding)
158+ require.Equal(t, uint64(8), newOutputStateStream(9).maxOutstanding)
159+ require.Equal(t, uint64(1), newOutputStateStream(1).maxOutstanding)
160+}
161+
162+func TestNormalizeOutputWindow(t *testing.T) {
163+ for _, tt := range []struct {
164+ input, want uint8
165+ }{{0, 8}, {1, 1}, {8, 8}, {9, 8}, {255, 8}} {
166+ require.Equal(t, tt.want, normalizeOutputWindow(tt.input))
167+ }
168+}
M internal/usecase/daemon/output_state_sync_test.go +10 -10
@@ -71,11 +71,11 @@ func TestDatagramAttachPipelinesRendererBeforeAck(t *testing.T) {
7171 d, sess, _, _ := newManualSessionWithPTYs(t, p)
7272 tr := newDatagramTestTransport()
7373 routed, ac, err := d.route(ports.Hello{
74- Version: ports.ProtocolVersion,
75- Intent: ports.IntentAttach,
76- Name: sess.name,
77- Size: domain.Size{Cols: 80, Rows: 24},
78- AckOutput: true,
74+ Version: ports.ProtocolVersion,
75+ Intent: ports.IntentAttach,
76+ Name: sess.name,
77+ Size: domain.Size{Cols: 80, Rows: 24},
78+ MaxOutputInFlight: 8,
7979 }, tr)
8080 require.NoError(t, err)
8181 require.Same(t, sess, routed)
@@ -126,11 +126,11 @@ func TestDatagramMultipleUnackedScrollPaintsMatchLatestFrame(t *testing.T) {
126126 d, sess, _, _ := newManualSessionWithPTYs(t, p)
127127 tr := newDatagramTestTransport()
128128 _, ac, err := d.route(ports.Hello{
129- Version: ports.ProtocolVersion,
130- Intent: ports.IntentAttach,
131- Name: sess.name,
132- Size: domain.Size{Cols: 80, Rows: 24},
133- AckOutput: true,
129+ Version: ports.ProtocolVersion,
130+ Intent: ports.IntentAttach,
131+ Name: sess.name,
132+ Size: domain.Size{Cols: 80, Rows: 24},
133+ MaxOutputInFlight: 8,
134134 }, tr)
135135 require.NoError(t, err)
136136
@@ -71,11 +71,11 @@ func TestDatagramAttachPipelinesRendererBeforeAck(t *testing.T) {
71 d, sess, _, _ := newManualSessionWithPTYs(t, p)71 d, sess, _, _ := newManualSessionWithPTYs(t, p)
72 tr := newDatagramTestTransport()72 tr := newDatagramTestTransport()
73 routed, ac, err := d.route(ports.Hello{73 routed, ac, err := d.route(ports.Hello{
74- Version: ports.ProtocolVersion,74+ Version: ports.ProtocolVersion,
75- Intent: ports.IntentAttach,75+ Intent: ports.IntentAttach,
76- Name: sess.name,76+ Name: sess.name,
77- Size: domain.Size{Cols: 80, Rows: 24},77+ Size: domain.Size{Cols: 80, Rows: 24},
78- AckOutput: true,78+ MaxOutputInFlight: 8,
79 }, tr)79 }, tr)
80 require.NoError(t, err)80 require.NoError(t, err)
81 require.Same(t, sess, routed)81 require.Same(t, sess, routed)
@@ -126,11 +126,11 @@ func TestDatagramMultipleUnackedScrollPaintsMatchLatestFrame(t *testing.T) {
126 d, sess, _, _ := newManualSessionWithPTYs(t, p)126 d, sess, _, _ := newManualSessionWithPTYs(t, p)
127 tr := newDatagramTestTransport()127 tr := newDatagramTestTransport()
128 _, ac, err := d.route(ports.Hello{128 _, ac, err := d.route(ports.Hello{
129- Version: ports.ProtocolVersion,129+ Version: ports.ProtocolVersion,
130- Intent: ports.IntentAttach,130+ Intent: ports.IntentAttach,
131- Name: sess.name,131+ Name: sess.name,
132- Size: domain.Size{Cols: 80, Rows: 24},132+ Size: domain.Size{Cols: 80, Rows: 24},
133- AckOutput: true,133+ MaxOutputInFlight: 8,
134 }, tr)134 }, tr)
135 require.NoError(t, err)135 require.NoError(t, err)
136 136
M internal/usecase/daemon/resume.go +1 -0
@@ -175,6 +175,7 @@ func (d *Daemon) resumeParkedLocked(h ports.Hello, tr ports.Transport, sz domain
175175 // abandoned transport's output chain before binding the replacement so the
176176 // mandatory first paint cannot be blocked by ACKs that died with the link.
177177 ac.output.rebase()
178+ ac.output.maxOutstanding = uint64(normalizeOutputWindow(h.MaxOutputInFlight))
178179 ac.replaceTransport(tr)
179180 ac.size = sz
180181 ac.resumeToken = d.nextResumeTokenLocked()
@@ -175,6 +175,7 @@ func (d *Daemon) resumeParkedLocked(h ports.Hello, tr ports.Transport, sz domain
175 // abandoned transport's output chain before binding the replacement so the175 // abandoned transport's output chain before binding the replacement so the
176 // mandatory first paint cannot be blocked by ACKs that died with the link.176 // mandatory first paint cannot be blocked by ACKs that died with the link.
177 ac.output.rebase()177 ac.output.rebase()
178+ ac.output.maxOutstanding = uint64(normalizeOutputWindow(h.MaxOutputInFlight))
178 ac.replaceTransport(tr)179 ac.replaceTransport(tr)
179 ac.size = sz180 ac.size = sz
180 ac.resumeToken = d.nextResumeTokenLocked()181 ac.resumeToken = d.nextResumeTokenLocked()
M internal/usecase/daemon/resume_test.go +32 -0
@@ -471,3 +471,35 @@ func TestResumeParkedUpdatesTerminalEnv(t *testing.T) {
471471 defer sess.mu.Unlock()
472472 require.True(t, sess.terminal.TrueColor)
473473 }
474+
475+func TestResumeRenegotiatesOutputWindowOnReusedStream(t *testing.T) {
476+ pty, release := newBlockingPTY(t)
477+ defer release()
478+ d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
479+ firstHello := helloResumeCapable(ports.IntentNew, "work", 0)
480+ firstHello.MaxOutputInFlight = 8
481+ sess, ac, err := d.route(firstHello, &closeTrackingTransport{})
482+ require.NoError(t, err)
483+ require.Equal(t, uint64(8), ac.output.maxOutstanding)
484+ stream := ac.output
485+ token := ac.resumeToken
486+ require.True(t, sess.detachIfCurrent(ac))
487+ require.True(t, d.parkAttachment(sess, ac))
488+
489+ resumeOne := helloResumeCapable(ports.IntentResume, "work", token)
490+ resumeOne.MaxOutputInFlight = 1
491+ _, resumed, err := d.route(resumeOne, &closeTrackingTransport{})
492+ require.NoError(t, err)
493+ require.Same(t, stream, resumed.output)
494+ require.Equal(t, uint64(1), resumed.output.maxOutstanding)
495+
496+ token = resumed.resumeToken
497+ require.True(t, sess.detachIfCurrent(resumed))
498+ require.True(t, d.parkAttachment(sess, resumed))
499+ resumeEight := helloResumeCapable(ports.IntentResume, "work", token)
500+ resumeEight.MaxOutputInFlight = 8
501+ _, resumed, err = d.route(resumeEight, &closeTrackingTransport{})
502+ require.NoError(t, err)
503+ require.Same(t, stream, resumed.output)
504+ require.Equal(t, uint64(8), resumed.output.maxOutstanding)
505+}
@@ -471,3 +471,35 @@ func TestResumeParkedUpdatesTerminalEnv(t *testing.T) {
471 defer sess.mu.Unlock()471 defer sess.mu.Unlock()
472 require.True(t, sess.terminal.TrueColor)472 require.True(t, sess.terminal.TrueColor)
473 }473 }
474+
475+func TestResumeRenegotiatesOutputWindowOnReusedStream(t *testing.T) {
476+ pty, release := newBlockingPTY(t)
477+ defer release()
478+ d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
479+ firstHello := helloResumeCapable(ports.IntentNew, "work", 0)
480+ firstHello.MaxOutputInFlight = 8
481+ sess, ac, err := d.route(firstHello, &closeTrackingTransport{})
482+ require.NoError(t, err)
483+ require.Equal(t, uint64(8), ac.output.maxOutstanding)
484+ stream := ac.output
485+ token := ac.resumeToken
486+ require.True(t, sess.detachIfCurrent(ac))
487+ require.True(t, d.parkAttachment(sess, ac))
488+
489+ resumeOne := helloResumeCapable(ports.IntentResume, "work", token)
490+ resumeOne.MaxOutputInFlight = 1
491+ _, resumed, err := d.route(resumeOne, &closeTrackingTransport{})
492+ require.NoError(t, err)
493+ require.Same(t, stream, resumed.output)
494+ require.Equal(t, uint64(1), resumed.output.maxOutstanding)
495+
496+ token = resumed.resumeToken
497+ require.True(t, sess.detachIfCurrent(resumed))
498+ require.True(t, d.parkAttachment(sess, resumed))
499+ resumeEight := helloResumeCapable(ports.IntentResume, "work", token)
500+ resumeEight.MaxOutputInFlight = 8
501+ _, resumed, err = d.route(resumeEight, &closeTrackingTransport{})
502+ require.NoError(t, err)
503+ require.Same(t, stream, resumed.output)
504+ require.Equal(t, uint64(8), resumed.output.maxOutstanding)
505+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close