Merge pull request #59 from bnema/fix/udp-output-state-sync-advance-on-ack
improve(daemon): advance UDP output on ack
15 files · +262 -43
M internal/adapters/dgram/proxy_test.go +27 -0
| @@ -78,6 +78,33 @@ func TestProxyRuntimeCopiesFramesAndDaemonEOFIsTerminal(t *testing.T) { | |||
| 78 | 78 | } | |
| 79 | 79 | } | |
| 80 | 80 | ||
| 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", AckOutput: true} | |
| 87 | + | client.recv <- recvResult{frame: ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}} | |
| 88 | + | ||
| 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.AckOutput { | |
| 96 | + | t.Fatal("proxied Hello lost AckOutput capability") | |
| 97 | + | } | |
| 98 | + | case <-time.After(time.Second): | |
| 99 | + | t.Fatal("timeout waiting for proxied Hello") | |
| 100 | + | } | |
| 101 | + | ||
| 102 | + | daemon.recv <- recvResult{err: io.EOF} | |
| 103 | + | if err := <-errCh; err != nil { | |
| 104 | + | t.Fatalf("Run err=%v, want nil", err) | |
| 105 | + | } | |
| 106 | + | } | |
| 107 | + | ||
| 81 | 108 | func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing.T) { | |
| 82 | 109 | tests := []struct { | |
| 83 | 110 | name string | |
| @@ -78,6 +78,33 @@ func TestProxyRuntimeCopiesFramesAndDaemonEOFIsTerminal(t *testing.T) { | |||||
| 78 | } | 78 | } | ||
| 79 | } | 79 | } | ||
| 80 | 80 | ||||
| 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", AckOutput: true} | |||
| 87 | + | client.recv <- recvResult{frame: ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}} | |||
| 88 | + | ||||
| 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.AckOutput { | |||
| 96 | + | t.Fatal("proxied Hello lost AckOutput capability") | |||
| 97 | + | } | |||
| 98 | + | case <-time.After(time.Second): | |||
| 99 | + | t.Fatal("timeout waiting for proxied Hello") | |||
| 100 | + | } | |||
| 101 | + | ||||
| 102 | + | daemon.recv <- recvResult{err: io.EOF} | |||
| 103 | + | if err := <-errCh; err != nil { | |||
| 104 | + | t.Fatalf("Run err=%v, want nil", err) | |||
| 105 | + | } | |||
| 106 | + | } | |||
| 107 | + | ||||
| 81 | func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing.T) { | 108 | func TestProxyRuntimeRetriesRecoverableClientSendWithoutDroppingFrame(t *testing.T) { | ||
| 82 | tests := []struct { | 109 | tests := []struct { | ||
| 83 | name string | 110 | name string | ||
M internal/adapters/dgram/transport.go +9 -2
| @@ -240,9 +240,16 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD | |||
| 240 | 240 | return t, nil | |
| 241 | 241 | } | |
| 242 | 242 | ||
| 243 | + | // DatagramTransport marks this transport as UDP/datagram-backed for daemon | |
| 244 | + | // output-state policy selection. | |
| 245 | + | func (t *Transport) DatagramTransport() {} | |
| 246 | + | ||
| 243 | 247 | func (t *Transport) Send(f ports.Frame) error { | |
| 244 | - | // Until full ACK-driven terminal state sync exists, every frame is reliable; | |
| 245 | - | // dropping MsgOutput can permanently desynchronize the client screen. | |
| 248 | + | // MsgOutput supersession is intentionally deferred: this reliable stream uses | |
| 249 | + | // strict sequence delivery, so deleting an older pending output without a | |
| 250 | + | // receive-side skip/ack signal would stall later frames behind the missing | |
| 251 | + | // sequence. Add an explicit droppable/skip protocol before replacing pending | |
| 252 | + | // outputs. | |
| 246 | 253 | reliable := true | |
| 247 | 254 | t.mu.Lock() | |
| 248 | 255 | var pendingWaitStarted time.Time | |
| @@ -240,9 +240,16 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD | |||||
| 240 | return t, nil | 240 | return t, nil | ||
| 241 | } | 241 | } | ||
| 242 | 242 | ||||
| 243 | + | // DatagramTransport marks this transport as UDP/datagram-backed for daemon | |||
| 244 | + | // output-state policy selection. | |||
| 245 | + | func (t *Transport) DatagramTransport() {} | |||
| 246 | + | ||||
| 243 | func (t *Transport) Send(f ports.Frame) error { | 247 | func (t *Transport) Send(f ports.Frame) error { | ||
| 244 | - | // Until full ACK-driven terminal state sync exists, every frame is reliable; | 248 | + | // MsgOutput supersession is intentionally deferred: this reliable stream uses |
| 245 | - | // dropping MsgOutput can permanently desynchronize the client screen. | 249 | + | // strict sequence delivery, so deleting an older pending output without a |
| 250 | + | // receive-side skip/ack signal would stall later frames behind the missing | |||
| 251 | + | // sequence. Add an explicit droppable/skip protocol before replacing pending | |||
| 252 | + | // outputs. | |||
| 246 | reliable := true | 253 | reliable := true | ||
| 247 | t.mu.Lock() | 254 | t.mu.Lock() | ||
| 248 | var pendingWaitStarted time.Time | 255 | var pendingWaitStarted time.Time | ||
M internal/ports/frame.go +1 -1
| @@ -1,7 +1,7 @@ | |||
| 1 | 1 | package ports | |
| 2 | 2 | ||
| 3 | 3 | // ProtocolVersion is the current vev IPC wire protocol version. | |
| 4 | - | const ProtocolVersion uint16 = 10 | |
| 4 | + | const ProtocolVersion uint16 = 11 | |
| 5 | 5 | ||
| 6 | 6 | // MsgType identifies the kind of payload carried by a Frame. | |
| 7 | 7 | type MsgType uint8 | |
| @@ -1,7 +1,7 @@ | |||||
| 1 | package ports | 1 | 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 = 10 | 4 | + | const ProtocolVersion uint16 = 11 |
| 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 uint8 | 7 | type MsgType uint8 | ||
M internal/ports/frame_test.go +2 -2
| @@ -3,8 +3,8 @@ package ports | |||
| 3 | 3 | import "testing" | |
| 4 | 4 | ||
| 5 | 5 | func TestProtocolVersion(t *testing.T) { | |
| 6 | - | if ProtocolVersion != 10 { | |
| 7 | - | t.Fatalf("ProtocolVersion = %d, want 10", ProtocolVersion) | |
| 6 | + | if ProtocolVersion != 11 { | |
| 7 | + | t.Fatalf("ProtocolVersion = %d, want 11", ProtocolVersion) | |
| 8 | 8 | } | |
| 9 | 9 | } | |
| 10 | 10 | ||
| @@ -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 != 10 { | 6 | + | if ProtocolVersion != 11 { |
| 7 | - | t.Fatalf("ProtocolVersion = %d, want 10", ProtocolVersion) | 7 | + | t.Fatalf("ProtocolVersion = %d, want 11", ProtocolVersion) |
| 8 | } | 8 | } | ||
| 9 | } | 9 | } | ||
| 10 | 10 | ||||
M internal/ports/ports.go +6 -0
| @@ -45,6 +45,12 @@ type Transport interface { | |||
| 45 | 45 | Close() error | |
| 46 | 46 | } | |
| 47 | 47 | ||
| 48 | + | // DatagramTransport marks transports with datagram/reordered/lossy delivery | |
| 49 | + | // semantics behind the framed Transport interface. | |
| 50 | + | type DatagramTransport interface { | |
| 51 | + | DatagramTransport() | |
| 52 | + | } | |
| 53 | + | ||
| 48 | 54 | // Dialer establishes outbound Transport connections. | |
| 49 | 55 | type Dialer interface { | |
| 50 | 56 | Dial(ctx context.Context) (Transport, error) | |
| @@ -45,6 +45,12 @@ type Transport interface { | |||||
| 45 | Close() error | 45 | Close() error | ||
| 46 | } | 46 | } | ||
| 47 | 47 | ||||
| 48 | + | // DatagramTransport marks transports with datagram/reordered/lossy delivery | |||
| 49 | + | // semantics behind the framed Transport interface. | |||
| 50 | + | type DatagramTransport interface { | |||
| 51 | + | DatagramTransport() | |||
| 52 | + | } | |||
| 53 | + | ||||
| 48 | // Dialer establishes outbound Transport connections. | 54 | // Dialer establishes outbound Transport connections. | ||
| 49 | type Dialer interface { | 55 | type Dialer interface { | ||
| 50 | Dial(ctx context.Context) (Transport, error) | 56 | Dial(ctx context.Context) (Transport, error) | ||
M internal/ports/wire.go +11 -0
| @@ -66,6 +66,7 @@ type Hello struct { | |||
| 66 | 66 | TermEnv string | |
| 67 | 67 | Cwd string | |
| 68 | 68 | TrueColor bool | |
| 69 | + | AckOutput bool | |
| 69 | 70 | } | |
| 70 | 71 | ||
| 71 | 72 | // Input carries raw bytes typed/pasted by the client, destined for the PTY. | |
| @@ -309,6 +310,11 @@ func MarshalHello(h Hello) []byte { | |||
| 309 | 310 | } else { | |
| 310 | 311 | w.putUint8(0) | |
| 311 | 312 | } | |
| 313 | + | if h.AckOutput { | |
| 314 | + | w.putUint8(1) | |
| 315 | + | } else { | |
| 316 | + | w.putUint8(0) | |
| 317 | + | } | |
| 312 | 318 | return w.b | |
| 313 | 319 | } | |
| 314 | 320 | ||
| @@ -355,6 +361,11 @@ func UnmarshalHello(b []byte) (Hello, error) { | |||
| 355 | 361 | return Hello{}, err | |
| 356 | 362 | } | |
| 357 | 363 | h.TrueColor = trueColor != 0 | |
| 364 | + | ackOutput, err := r.getUint8() | |
| 365 | + | if err != nil { | |
| 366 | + | return Hello{}, err | |
| 367 | + | } | |
| 368 | + | h.AckOutput = ackOutput != 0 | |
| 358 | 369 | if err := r.done(); err != nil { | |
| 359 | 370 | return Hello{}, err | |
| 360 | 371 | } | |
| @@ -66,6 +66,7 @@ type Hello struct { | |||||
| 66 | TermEnv string | 66 | TermEnv string | ||
| 67 | Cwd string | 67 | Cwd string | ||
| 68 | TrueColor bool | 68 | TrueColor bool | ||
| 69 | + | AckOutput bool | |||
| 69 | } | 70 | } | ||
| 70 | 71 | ||||
| 71 | // Input carries raw bytes typed/pasted by the client, destined for the PTY. | 72 | // Input carries raw bytes typed/pasted by the client, destined for the PTY. | ||
| @@ -309,6 +310,11 @@ func MarshalHello(h Hello) []byte { | |||||
| 309 | } else { | 310 | } else { | ||
| 310 | w.putUint8(0) | 311 | w.putUint8(0) | ||
| 311 | } | 312 | } | ||
| 313 | + | if h.AckOutput { | |||
| 314 | + | w.putUint8(1) | |||
| 315 | + | } else { | |||
| 316 | + | w.putUint8(0) | |||
| 317 | + | } | |||
| 312 | return w.b | 318 | return w.b | ||
| 313 | } | 319 | } | ||
| 314 | 320 | ||||
| @@ -355,6 +361,11 @@ func UnmarshalHello(b []byte) (Hello, error) { | |||||
| 355 | return Hello{}, err | 361 | return Hello{}, err | ||
| 356 | } | 362 | } | ||
| 357 | h.TrueColor = trueColor != 0 | 363 | h.TrueColor = trueColor != 0 | ||
| 364 | + | ackOutput, err := r.getUint8() | |||
| 365 | + | if err != nil { | |||
| 366 | + | return Hello{}, err | |||
| 367 | + | } | |||
| 368 | + | h.AckOutput = ackOutput != 0 | |||
| 358 | if err := r.done(); err != nil { | 369 | if err := r.done(); err != nil { | ||
| 359 | return Hello{}, err | 370 | return Hello{}, err | ||
| 360 | } | 371 | } | ||
M internal/ports/wire_test.go +3 -2
| @@ -69,8 +69,9 @@ func TestHelloGoldenAndRoundTrip(t *testing.T) { | |||
| 69 | 69 | TermEnv: "xterm-256color", | |
| 70 | 70 | Cwd: "/tmp/project", | |
| 71 | 71 | TrueColor: true, | |
| 72 | + | AckOutput: true, | |
| 72 | 73 | }, | |
| 73 | - | 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}, | |
| 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}, | |
| 74 | 75 | }, | |
| 75 | 76 | { | |
| 76 | 77 | name: "empty strings", | |
| @@ -83,7 +84,7 @@ func TestHelloGoldenAndRoundTrip(t *testing.T) { | |||
| 83 | 84 | Cwd: "", | |
| 84 | 85 | TrueColor: false, | |
| 85 | 86 | }, | |
| 86 | - | want: []byte{0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}, | |
| 87 | + | want: []byte{0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}, | |
| 87 | 88 | }, | |
| 88 | 89 | } | |
| 89 | 90 | ||
| @@ -69,8 +69,9 @@ func TestHelloGoldenAndRoundTrip(t *testing.T) { | |||||
| 69 | TermEnv: "xterm-256color", | 69 | TermEnv: "xterm-256color", | ||
| 70 | Cwd: "/tmp/project", | 70 | Cwd: "/tmp/project", | ||
| 71 | TrueColor: true, | 71 | TrueColor: true, | ||
| 72 | + | AckOutput: true, | |||
| 72 | }, | 73 | }, | ||
| 73 | - | 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}, | 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}, |
| 74 | }, | 75 | }, | ||
| 75 | { | 76 | { | ||
| 76 | name: "empty strings", | 77 | name: "empty strings", | ||
| @@ -83,7 +84,7 @@ func TestHelloGoldenAndRoundTrip(t *testing.T) { | |||||
| 83 | Cwd: "", | 84 | Cwd: "", | ||
| 84 | TrueColor: false, | 85 | TrueColor: false, | ||
| 85 | }, | 86 | }, | ||
| 86 | - | want: []byte{0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}, | 87 | + | want: []byte{0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}, |
| 87 | }, | 88 | }, | ||
| 88 | } | 89 | } | ||
| 89 | 90 | ||||
M internal/usecase/client/client.go +2 -0
| @@ -352,6 +352,7 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi | |||
| 352 | 352 | termEnv := os.Getenv("TERM") | |
| 353 | 353 | colorTerm := os.Getenv("COLORTERM") | |
| 354 | 354 | trueColor := DetectTrueColor(termEnv, colorTerm) | |
| 355 | + | _, ackOutput := transport.(ports.DatagramTransport) | |
| 355 | 356 | hello := ports.Hello{ | |
| 356 | 357 | Version: ports.ProtocolVersion, | |
| 357 | 358 | Intent: intent, | |
| @@ -362,6 +363,7 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi | |||
| 362 | 363 | TermEnv: termEnv, | |
| 363 | 364 | Cwd: cwd, | |
| 364 | 365 | TrueColor: trueColor, | |
| 366 | + | AckOutput: ackOutput, | |
| 365 | 367 | } | |
| 366 | 368 | if err := transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil { | |
| 367 | 369 | return attachResult{err: fmt.Errorf("vev: sending hello: %w", err)} | |
| @@ -352,6 +352,7 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi | |||||
| 352 | termEnv := os.Getenv("TERM") | 352 | termEnv := os.Getenv("TERM") | ||
| 353 | colorTerm := os.Getenv("COLORTERM") | 353 | colorTerm := os.Getenv("COLORTERM") | ||
| 354 | trueColor := DetectTrueColor(termEnv, colorTerm) | 354 | trueColor := DetectTrueColor(termEnv, colorTerm) | ||
| 355 | + | _, ackOutput := transport.(ports.DatagramTransport) | |||
| 355 | hello := ports.Hello{ | 356 | hello := ports.Hello{ | ||
| 356 | Version: ports.ProtocolVersion, | 357 | Version: ports.ProtocolVersion, | ||
| 357 | Intent: intent, | 358 | Intent: intent, | ||
| @@ -362,6 +363,7 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi | |||||
| 362 | TermEnv: termEnv, | 363 | TermEnv: termEnv, | ||
| 363 | Cwd: cwd, | 364 | Cwd: cwd, | ||
| 364 | TrueColor: trueColor, | 365 | TrueColor: trueColor, | ||
| 366 | + | AckOutput: ackOutput, | |||
| 365 | } | 367 | } | ||
| 366 | if err := transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil { | 368 | if err := transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil { | ||
| 367 | return attachResult{err: fmt.Errorf("vev: sending hello: %w", err)} | 369 | return attachResult{err: fmt.Errorf("vev: sending hello: %w", err)} | ||
M internal/usecase/client/client_test.go +6 -1
| @@ -151,6 +151,10 @@ func isType(typ ports.MsgType) any { | |||
| 151 | 151 | return mock.MatchedBy(func(f ports.Frame) bool { return f.Type == typ }) | |
| 152 | 152 | } | |
| 153 | 153 | ||
| 154 | + | type datagramMarkedTransport struct{ ports.Transport } | |
| 155 | + | ||
| 156 | + | func (datagramMarkedTransport) DatagramTransport() {} | |
| 157 | + | ||
| 154 | 158 | func TestAttachHelloIncludesTrueColor(t *testing.T) { | |
| 155 | 159 | t.Setenv("TERM", "xterm-256color") | |
| 156 | 160 | t.Setenv("COLORTERM", "truecolor") | |
| @@ -176,13 +180,14 @@ func TestAttachHelloIncludesTrueColor(t *testing.T) { | |||
| 176 | 180 | defer unblock() | |
| 177 | 181 | tr.EXPECT().Close().Return(nil).Once() | |
| 178 | 182 | ||
| 179 | - | err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "") | |
| 183 | + | err := client.Attach(context.Background(), datagramMarkedTransport{Transport: tr}, tm, realClock{}, ports.IntentEphemeral, "") | |
| 180 | 184 | require.NoError(t, err) | |
| 181 | 185 | ||
| 182 | 186 | select { | |
| 183 | 187 | case hello := <-gotHello: | |
| 184 | 188 | require.Equal(t, "xterm-256color", hello.TermEnv) | |
| 185 | 189 | require.True(t, hello.TrueColor) | |
| 190 | + | require.True(t, hello.AckOutput) | |
| 186 | 191 | case <-time.After(2 * time.Second): | |
| 187 | 192 | t.Fatal("hello frame was not sent") | |
| 188 | 193 | } | |
| @@ -151,6 +151,10 @@ func isType(typ ports.MsgType) any { | |||||
| 151 | return mock.MatchedBy(func(f ports.Frame) bool { return f.Type == typ }) | 151 | return mock.MatchedBy(func(f ports.Frame) bool { return f.Type == typ }) | ||
| 152 | } | 152 | } | ||
| 153 | 153 | ||||
| 154 | + | type datagramMarkedTransport struct{ ports.Transport } | |||
| 155 | + | ||||
| 156 | + | func (datagramMarkedTransport) DatagramTransport() {} | |||
| 157 | + | ||||
| 154 | func TestAttachHelloIncludesTrueColor(t *testing.T) { | 158 | func TestAttachHelloIncludesTrueColor(t *testing.T) { | ||
| 155 | t.Setenv("TERM", "xterm-256color") | 159 | t.Setenv("TERM", "xterm-256color") | ||
| 156 | t.Setenv("COLORTERM", "truecolor") | 160 | t.Setenv("COLORTERM", "truecolor") | ||
| @@ -176,13 +180,14 @@ func TestAttachHelloIncludesTrueColor(t *testing.T) { | |||||
| 176 | defer unblock() | 180 | defer unblock() | ||
| 177 | tr.EXPECT().Close().Return(nil).Once() | 181 | tr.EXPECT().Close().Return(nil).Once() | ||
| 178 | 182 | ||||
| 179 | - | err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "") | 183 | + | err := client.Attach(context.Background(), datagramMarkedTransport{Transport: tr}, tm, realClock{}, ports.IntentEphemeral, "") |
| 180 | require.NoError(t, err) | 184 | require.NoError(t, err) | ||
| 181 | 185 | ||||
| 182 | select { | 186 | select { | ||
| 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) | ||
| 190 | + | require.True(t, hello.AckOutput) | |||
| 186 | case <-time.After(2 * time.Second): | 191 | case <-time.After(2 * time.Second): | ||
| 187 | t.Fatal("hello frame was not sent") | 192 | t.Fatal("hello frame was not sent") | ||
| 188 | } | 193 | } | ||
M internal/usecase/daemon/client.go +68 -29
| @@ -39,28 +39,30 @@ import ( | |||
| 39 | 39 | ) | |
| 40 | 40 | ||
| 41 | 41 | type attachedClient struct { | |
| 42 | - | tr ports.Transport | |
| 43 | - | rend *renderer.Renderer | |
| 44 | - | overlays *overlayRuntime | |
| 45 | - | clientID [16]byte | |
| 46 | - | resumeCapable bool | |
| 47 | - | resumeToken uint64 | |
| 48 | - | parked bool | |
| 49 | - | nextStateNum uint64 | |
| 50 | - | echoAck atomic.Uint64 | |
| 51 | - | outputAck atomic.Uint64 | |
| 52 | - | paintDeferred bool // guarded by sendMu; a diff paint skipped while acks lag | |
| 53 | - | bars barCache // only touched while sendMu is held | |
| 54 | - | composed composedFrameCache // only touched while sendMu is held | |
| 55 | - | size domain.Size | |
| 56 | - | keys *keys.Router | |
| 57 | - | sess Guarded[*session] | |
| 58 | - | mouseScan mouse.Scanner | |
| 59 | - | themeMu sync.Mutex | |
| 60 | - | theme themeui.Theme | |
| 61 | - | clientTheme themeui.Theme | |
| 62 | - | lastCursor cursorOut | |
| 63 | - | recentNav recentSessionNavigator | |
| 42 | + | tr ports.Transport | |
| 43 | + | rend *renderer.Renderer | |
| 44 | + | overlays *overlayRuntime | |
| 45 | + | clientID [16]byte | |
| 46 | + | resumeCapable bool | |
| 47 | + | resumeToken uint64 | |
| 48 | + | parked bool | |
| 49 | + | nextStateNum uint64 | |
| 50 | + | echoAck atomic.Uint64 | |
| 51 | + | outputAck atomic.Uint64 | |
| 52 | + | paintDeferred bool // guarded by sendMu; a diff paint skipped while acks lag | |
| 53 | + | bars barCache // only touched while sendMu is held | |
| 54 | + | composed composedFrameCache // only touched while sendMu is held | |
| 55 | + | advanceOutputOnAck bool // renderer shadow advances after client MsgAck instead of send | |
| 56 | + | outputFrames map[uint64]renderer.Frame // only touched while sendMu is held; frames sent by output state | |
| 57 | + | size domain.Size | |
| 58 | + | keys *keys.Router | |
| 59 | + | sess Guarded[*session] | |
| 60 | + | mouseScan mouse.Scanner | |
| 61 | + | themeMu sync.Mutex | |
| 62 | + | theme themeui.Theme | |
| 63 | + | clientTheme themeui.Theme | |
| 64 | + | lastCursor cursorOut | |
| 65 | + | recentNav recentSessionNavigator | |
| 64 | 66 | // historyNavMu protects historyNav and historyNavTimer. When paint needs | |
| 65 | 67 | // several locks, take sendMu before historyNavMu, then Daemon.mu/session.mu. | |
| 66 | 68 | historyNavMu sync.Mutex | |
| @@ -213,6 +215,10 @@ func (ac *attachedClient) shouldResetOutputState(reset bool) bool { | |||
| 213 | 215 | func (ac *attachedClient) advanceOutputAck(state uint64) { | |
| 214 | 216 | ac.sendMu.Lock() | |
| 215 | 217 | defer ac.sendMu.Unlock() | |
| 218 | + | ac.advanceOutputAckLocked(state) | |
| 219 | + | } | |
| 220 | + | ||
| 221 | + | func (ac *attachedClient) advanceOutputAckLocked(state uint64) { | |
| 216 | 222 | if state > ac.nextStateNum { | |
| 217 | 223 | return | |
| 218 | 224 | } | |
| @@ -227,6 +233,29 @@ func (ac *attachedClient) advanceOutputAck(state uint64) { | |||
| 227 | 233 | } | |
| 228 | 234 | } | |
| 229 | 235 | ||
| 236 | + | func (ac *attachedClient) ackOutputState(state uint64) error { | |
| 237 | + | ac.sendMu.Lock() | |
| 238 | + | defer ac.sendMu.Unlock() | |
| 239 | + | if state > ac.nextStateNum || state <= ac.outputAck.Load() { | |
| 240 | + | return nil | |
| 241 | + | } | |
| 242 | + | if ac.advanceOutputOnAck { | |
| 243 | + | frame, ok := ac.outputFrames[state] | |
| 244 | + | if ok { | |
| 245 | + | if err := ac.rend.Ack(frame); err != nil { | |
| 246 | + | return err | |
| 247 | + | } | |
| 248 | + | } | |
| 249 | + | for n := range ac.outputFrames { | |
| 250 | + | if n <= state { | |
| 251 | + | delete(ac.outputFrames, n) | |
| 252 | + | } | |
| 253 | + | } | |
| 254 | + | } | |
| 255 | + | ac.advanceOutputAckLocked(state) | |
| 256 | + | return nil | |
| 257 | + | } | |
| 258 | + | ||
| 230 | 259 | // outputLagAtCapLocked reports whether the client has at least | |
| 231 | 260 | // maxUnackedOutputStates output states in flight (sent but unacked). Caller | |
| 232 | 261 | // holds sendMu. | |
| @@ -396,6 +425,7 @@ func (d *Daemon) notifiesSnapshot() []chan struct{} { | |||
| 396 | 425 | type attachClientOptions struct { | |
| 397 | 426 | clientID [16]byte | |
| 398 | 427 | resumeCapable bool | |
| 428 | + | ackOutput bool | |
| 399 | 429 | } | |
| 400 | 430 | ||
| 401 | 431 | func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, opts attachClientOptions) (*attachedClient, *attachedClient) { | |
| @@ -403,13 +433,20 @@ func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, | |||
| 403 | 433 | if opts.resumeCapable { | |
| 404 | 434 | resumeToken = d.nextResumeTokenLocked() | |
| 405 | 435 | } | |
| 436 | + | caps := renderer.Capabilities{} | |
| 437 | + | advanceOutputOnAck := opts.ackOutput | |
| 438 | + | if advanceOutputOnAck { | |
| 439 | + | caps.AdvancePolicy = renderer.AdvanceOnAck | |
| 440 | + | } | |
| 406 | 441 | ac := &attachedClient{ | |
| 407 | - | tr: tr, | |
| 408 | - | rend: renderer.New(renderer.Capabilities{}), | |
| 409 | - | size: sz, | |
| 410 | - | clientID: opts.clientID, | |
| 411 | - | resumeCapable: opts.resumeCapable, | |
| 412 | - | resumeToken: resumeToken, | |
| 442 | + | tr: tr, | |
| 443 | + | rend: renderer.New(caps), | |
| 444 | + | advanceOutputOnAck: advanceOutputOnAck, | |
| 445 | + | outputFrames: make(map[uint64]renderer.Frame), | |
| 446 | + | size: sz, | |
| 447 | + | clientID: opts.clientID, | |
| 448 | + | resumeCapable: opts.resumeCapable, | |
| 449 | + | resumeToken: resumeToken, | |
| 413 | 450 | } | |
| 414 | 451 | ac.initOverlays() | |
| 415 | 452 | ac.setSession(sess) | |
| @@ -521,7 +558,9 @@ func (d *Daemon) runConnLoop(ac *attachedClient) { | |||
| 521 | 558 | return | |
| 522 | 559 | case ports.MsgAck: | |
| 523 | 560 | if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil { | |
| 524 | - | ac.advanceOutputAck(ack.AckedStateNum) | |
| 561 | + | if err := ac.ackOutputState(ack.AckedStateNum); err != nil { | |
| 562 | + | d.log.Warn("output ack rejected", "err", err, "state", ack.AckedStateNum) | |
| 563 | + | } | |
| 525 | 564 | if ac.takeDeferredPaint() { | |
| 526 | 565 | d.paint(sess, ac, false) | |
| 527 | 566 | } | |
| @@ -39,28 +39,30 @@ import ( | |||||
| 39 | ) | 39 | ) | ||
| 40 | 40 | ||||
| 41 | type attachedClient struct { | 41 | type attachedClient struct { | ||
| 42 | - | tr ports.Transport | 42 | + | tr ports.Transport |
| 43 | - | rend *renderer.Renderer | 43 | + | rend *renderer.Renderer |
| 44 | - | overlays *overlayRuntime | 44 | + | overlays *overlayRuntime |
| 45 | - | clientID [16]byte | 45 | + | clientID [16]byte |
| 46 | - | resumeCapable bool | 46 | + | resumeCapable bool |
| 47 | - | resumeToken uint64 | 47 | + | resumeToken uint64 |
| 48 | - | parked bool | 48 | + | parked bool |
| 49 | - | nextStateNum uint64 | 49 | + | nextStateNum uint64 |
| 50 | - | echoAck atomic.Uint64 | 50 | + | echoAck atomic.Uint64 |
| 51 | - | outputAck atomic.Uint64 | 51 | + | outputAck atomic.Uint64 |
| 52 | - | paintDeferred bool // guarded by sendMu; a diff paint skipped while acks lag | 52 | + | paintDeferred bool // guarded by sendMu; a diff paint skipped while acks lag |
| 53 | - | bars barCache // only touched while sendMu is held | 53 | + | bars barCache // only touched while sendMu is held |
| 54 | - | composed composedFrameCache // only touched while sendMu is held | 54 | + | composed composedFrameCache // only touched while sendMu is held |
| 55 | - | size domain.Size | 55 | + | advanceOutputOnAck bool // renderer shadow advances after client MsgAck instead of send |
| 56 | - | keys *keys.Router | 56 | + | outputFrames map[uint64]renderer.Frame // only touched while sendMu is held; frames sent by output state |
| 57 | - | sess Guarded[*session] | 57 | + | size domain.Size |
| 58 | - | mouseScan mouse.Scanner | 58 | + | keys *keys.Router |
| 59 | - | themeMu sync.Mutex | 59 | + | sess Guarded[*session] |
| 60 | - | theme themeui.Theme | 60 | + | mouseScan mouse.Scanner |
| 61 | - | clientTheme themeui.Theme | 61 | + | themeMu sync.Mutex |
| 62 | - | lastCursor cursorOut | 62 | + | theme themeui.Theme |
| 63 | - | recentNav recentSessionNavigator | 63 | + | clientTheme themeui.Theme |
| 64 | + | lastCursor cursorOut | |||
| 65 | + | recentNav recentSessionNavigator | |||
| 64 | // historyNavMu protects historyNav and historyNavTimer. When paint needs | 66 | // historyNavMu protects historyNav and historyNavTimer. When paint needs | ||
| 65 | // several locks, take sendMu before historyNavMu, then Daemon.mu/session.mu. | 67 | // several locks, take sendMu before historyNavMu, then Daemon.mu/session.mu. | ||
| 66 | historyNavMu sync.Mutex | 68 | historyNavMu sync.Mutex | ||
| @@ -213,6 +215,10 @@ func (ac *attachedClient) shouldResetOutputState(reset bool) bool { | |||||
| 213 | func (ac *attachedClient) advanceOutputAck(state uint64) { | 215 | func (ac *attachedClient) advanceOutputAck(state uint64) { | ||
| 214 | ac.sendMu.Lock() | 216 | ac.sendMu.Lock() | ||
| 215 | defer ac.sendMu.Unlock() | 217 | defer ac.sendMu.Unlock() | ||
| 218 | + | ac.advanceOutputAckLocked(state) | |||
| 219 | + | } | |||
| 220 | + | ||||
| 221 | + | func (ac *attachedClient) advanceOutputAckLocked(state uint64) { | |||
| 216 | if state > ac.nextStateNum { | 222 | if state > ac.nextStateNum { | ||
| 217 | return | 223 | return | ||
| 218 | } | 224 | } | ||
| @@ -227,6 +233,29 @@ func (ac *attachedClient) advanceOutputAck(state uint64) { | |||||
| 227 | } | 233 | } | ||
| 228 | } | 234 | } | ||
| 229 | 235 | ||||
| 236 | + | func (ac *attachedClient) ackOutputState(state uint64) error { | |||
| 237 | + | ac.sendMu.Lock() | |||
| 238 | + | defer ac.sendMu.Unlock() | |||
| 239 | + | if state > ac.nextStateNum || state <= ac.outputAck.Load() { | |||
| 240 | + | return nil | |||
| 241 | + | } | |||
| 242 | + | if ac.advanceOutputOnAck { | |||
| 243 | + | frame, ok := ac.outputFrames[state] | |||
| 244 | + | if ok { | |||
| 245 | + | if err := ac.rend.Ack(frame); err != nil { | |||
| 246 | + | return err | |||
| 247 | + | } | |||
| 248 | + | } | |||
| 249 | + | for n := range ac.outputFrames { | |||
| 250 | + | if n <= state { | |||
| 251 | + | delete(ac.outputFrames, n) | |||
| 252 | + | } | |||
| 253 | + | } | |||
| 254 | + | } | |||
| 255 | + | ac.advanceOutputAckLocked(state) | |||
| 256 | + | return nil | |||
| 257 | + | } | |||
| 258 | + | ||||
| 230 | // outputLagAtCapLocked reports whether the client has at least | 259 | // outputLagAtCapLocked reports whether the client has at least | ||
| 231 | // maxUnackedOutputStates output states in flight (sent but unacked). Caller | 260 | // maxUnackedOutputStates output states in flight (sent but unacked). Caller | ||
| 232 | // holds sendMu. | 261 | // holds sendMu. | ||
| @@ -396,6 +425,7 @@ func (d *Daemon) notifiesSnapshot() []chan struct{} { | |||||
| 396 | type attachClientOptions struct { | 425 | type attachClientOptions struct { | ||
| 397 | clientID [16]byte | 426 | clientID [16]byte | ||
| 398 | resumeCapable bool | 427 | resumeCapable bool | ||
| 428 | + | ackOutput bool | |||
| 399 | } | 429 | } | ||
| 400 | 430 | ||||
| 401 | func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, opts attachClientOptions) (*attachedClient, *attachedClient) { | 431 | func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, opts attachClientOptions) (*attachedClient, *attachedClient) { | ||
| @@ -403,13 +433,20 @@ func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, | |||||
| 403 | if opts.resumeCapable { | 433 | if opts.resumeCapable { | ||
| 404 | resumeToken = d.nextResumeTokenLocked() | 434 | resumeToken = d.nextResumeTokenLocked() | ||
| 405 | } | 435 | } | ||
| 436 | + | caps := renderer.Capabilities{} | |||
| 437 | + | advanceOutputOnAck := opts.ackOutput | |||
| 438 | + | if advanceOutputOnAck { | |||
| 439 | + | caps.AdvancePolicy = renderer.AdvanceOnAck | |||
| 440 | + | } | |||
| 406 | ac := &attachedClient{ | 441 | ac := &attachedClient{ | ||
| 407 | - | tr: tr, | 442 | + | tr: tr, |
| 408 | - | rend: renderer.New(renderer.Capabilities{}), | 443 | + | rend: renderer.New(caps), |
| 409 | - | size: sz, | 444 | + | advanceOutputOnAck: advanceOutputOnAck, |
| 410 | - | clientID: opts.clientID, | 445 | + | outputFrames: make(map[uint64]renderer.Frame), |
| 411 | - | resumeCapable: opts.resumeCapable, | 446 | + | size: sz, |
| 412 | - | resumeToken: resumeToken, | 447 | + | clientID: opts.clientID, |
| 448 | + | resumeCapable: opts.resumeCapable, | |||
| 449 | + | resumeToken: resumeToken, | |||
| 413 | } | 450 | } | ||
| 414 | ac.initOverlays() | 451 | ac.initOverlays() | ||
| 415 | ac.setSession(sess) | 452 | ac.setSession(sess) | ||
| @@ -521,7 +558,9 @@ func (d *Daemon) runConnLoop(ac *attachedClient) { | |||||
| 521 | return | 558 | return | ||
| 522 | case ports.MsgAck: | 559 | case ports.MsgAck: | ||
| 523 | if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil { | 560 | if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil { | ||
| 524 | - | ac.advanceOutputAck(ack.AckedStateNum) | 561 | + | if err := ac.ackOutputState(ack.AckedStateNum); err != nil { |
| 562 | + | d.log.Warn("output ack rejected", "err", err, "state", ack.AckedStateNum) | |||
| 563 | + | } | |||
| 525 | if ac.takeDeferredPaint() { | 564 | if ac.takeDeferredPaint() { | ||
| 526 | d.paint(sess, ac, false) | 565 | d.paint(sess, ac, false) | ||
| 527 | } | 566 | } | ||
M internal/usecase/daemon/daemon.go +4 -4
| @@ -665,7 +665,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||
| 665 | 665 | } | |
| 666 | 666 | } | |
| 667 | 667 | d.purgeParkedForSessionLocked(sess) | |
| 668 | - | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true}) | |
| 668 | + | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, ackOutput: h.AckOutput}) | |
| 669 | 669 | sess.mu.Lock() | |
| 670 | 670 | sess.terminal = term | |
| 671 | 671 | sess.mu.Unlock() | |
| @@ -681,7 +681,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||
| 681 | 681 | return nil, nil, err | |
| 682 | 682 | } | |
| 683 | 683 | d.purgeParkedForSessionLocked(sess) | |
| 684 | - | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true}) | |
| 684 | + | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, ackOutput: h.AckOutput}) | |
| 685 | 685 | sess.mu.Lock() | |
| 686 | 686 | sess.terminal = term | |
| 687 | 687 | sess.mu.Unlock() | |
| @@ -708,7 +708,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||
| 708 | 708 | return nil, nil, err | |
| 709 | 709 | } | |
| 710 | 710 | d.purgeParkedForSessionLocked(sess) | |
| 711 | - | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true}) | |
| 711 | + | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, ackOutput: h.AckOutput}) | |
| 712 | 712 | sess.mu.Lock() | |
| 713 | 713 | sess.terminal = term | |
| 714 | 714 | sess.mu.Unlock() | |
| @@ -733,7 +733,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||
| 733 | 733 | } | |
| 734 | 734 | } | |
| 735 | 735 | d.purgeParkedForSessionLocked(sess) | |
| 736 | - | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true}) | |
| 736 | + | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, ackOutput: h.AckOutput}) | |
| 737 | 737 | sess.mu.Lock() | |
| 738 | 738 | sess.terminal = term | |
| 739 | 739 | sess.mu.Unlock() | |
| @@ -665,7 +665,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||||
| 665 | } | 665 | } | ||
| 666 | } | 666 | } | ||
| 667 | d.purgeParkedForSessionLocked(sess) | 667 | d.purgeParkedForSessionLocked(sess) | ||
| 668 | - | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true}) | 668 | + | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, ackOutput: h.AckOutput}) |
| 669 | sess.mu.Lock() | 669 | sess.mu.Lock() | ||
| 670 | sess.terminal = term | 670 | sess.terminal = term | ||
| 671 | sess.mu.Unlock() | 671 | sess.mu.Unlock() | ||
| @@ -681,7 +681,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||||
| 681 | return nil, nil, err | 681 | return nil, nil, err | ||
| 682 | } | 682 | } | ||
| 683 | d.purgeParkedForSessionLocked(sess) | 683 | d.purgeParkedForSessionLocked(sess) | ||
| 684 | - | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true}) | 684 | + | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, ackOutput: h.AckOutput}) |
| 685 | sess.mu.Lock() | 685 | sess.mu.Lock() | ||
| 686 | sess.terminal = term | 686 | sess.terminal = term | ||
| 687 | sess.mu.Unlock() | 687 | sess.mu.Unlock() | ||
| @@ -708,7 +708,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||||
| 708 | return nil, nil, err | 708 | return nil, nil, err | ||
| 709 | } | 709 | } | ||
| 710 | d.purgeParkedForSessionLocked(sess) | 710 | d.purgeParkedForSessionLocked(sess) | ||
| 711 | - | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true}) | 711 | + | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, ackOutput: h.AckOutput}) |
| 712 | sess.mu.Lock() | 712 | sess.mu.Lock() | ||
| 713 | sess.terminal = term | 713 | sess.terminal = term | ||
| 714 | sess.mu.Unlock() | 714 | sess.mu.Unlock() | ||
| @@ -733,7 +733,7 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||||
| 733 | } | 733 | } | ||
| 734 | } | 734 | } | ||
| 735 | d.purgeParkedForSessionLocked(sess) | 735 | d.purgeParkedForSessionLocked(sess) | ||
| 736 | - | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true}) | 736 | + | ac, old := d.attachClient(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, ackOutput: h.AckOutput}) |
| 737 | sess.mu.Lock() | 737 | sess.mu.Lock() | ||
| 738 | sess.terminal = term | 738 | sess.terminal = term | ||
| 739 | sess.mu.Unlock() | 739 | sess.mu.Unlock() | ||
A internal/usecase/daemon/output_state_sync_test.go +104 -0
| @@ -0,0 +1,104 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "io" | |
| 5 | + | "sync" | |
| 6 | + | "testing" | |
| 7 | + | ||
| 8 | + | "github.com/bnema/vev/internal/domain" | |
| 9 | + | "github.com/bnema/vev/internal/ports" | |
| 10 | + | "github.com/stretchr/testify/require" | |
| 11 | + | ) | |
| 12 | + | ||
| 13 | + | type datagramTestTransport struct { | |
| 14 | + | sends chan ports.Frame | |
| 15 | + | recv chan ports.Frame | |
| 16 | + | once sync.Once | |
| 17 | + | } | |
| 18 | + | ||
| 19 | + | func newDatagramTestTransport() *datagramTestTransport { | |
| 20 | + | return &datagramTestTransport{sends: make(chan ports.Frame, 64), recv: make(chan ports.Frame, 64)} | |
| 21 | + | } | |
| 22 | + | ||
| 23 | + | func (t *datagramTestTransport) Send(f ports.Frame) error { t.sends <- f; return nil } | |
| 24 | + | func (t *datagramTestTransport) Recv() (ports.Frame, error) { | |
| 25 | + | f, ok := <-t.recv | |
| 26 | + | if !ok { | |
| 27 | + | return ports.Frame{}, io.EOF | |
| 28 | + | } | |
| 29 | + | return f, nil | |
| 30 | + | } | |
| 31 | + | func (t *datagramTestTransport) Close() error { t.once.Do(func() { close(t.recv) }); return nil } | |
| 32 | + | ||
| 33 | + | func TestDatagramAttachAdvancesRendererOnlyOnAck(t *testing.T) { | |
| 34 | + | p, releasePTY := newBlockingPTY(t) | |
| 35 | + | defer releasePTY() | |
| 36 | + | d, sess, _, _ := newManualSessionWithPTYs(t, p) | |
| 37 | + | tr := newDatagramTestTransport() | |
| 38 | + | routed, ac, err := d.route(ports.Hello{ | |
| 39 | + | Version: ports.ProtocolVersion, | |
| 40 | + | Intent: ports.IntentAttach, | |
| 41 | + | Name: sess.name, | |
| 42 | + | Size: domain.Size{Cols: 80, Rows: 24}, | |
| 43 | + | AckOutput: true, | |
| 44 | + | }, tr) | |
| 45 | + | require.NoError(t, err) | |
| 46 | + | require.Same(t, sess, routed) | |
| 47 | + | ||
| 48 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |
| 49 | + | d.paint(sess, ac, false) | |
| 50 | + | first := awaitFrame(t, tr.sends, ports.MsgOutput) | |
| 51 | + | out, err := ports.UnmarshalOutput(first.Payload) | |
| 52 | + | require.NoError(t, err) | |
| 53 | + | require.Equal(t, uint64(1), out.NewStateNum) | |
| 54 | + | ||
| 55 | + | // Before the MsgAck, AdvanceOnAck must keep rediffing from the unacked | |
| 56 | + | // shadow even if the composed screen is unchanged. | |
| 57 | + | d.paint(sess, ac, false) | |
| 58 | + | awaitFrame(t, tr.sends, ports.MsgOutput) | |
| 59 | + | ||
| 60 | + | tr.recv <- ports.Frame{Type: ports.MsgAck, Payload: ports.MarshalAck(ports.Ack{AckedStateNum: out.NewStateNum})} | |
| 61 | + | require.NoError(t, tr.Close()) | |
| 62 | + | d.runConnLoop(ac) | |
| 63 | + | ||
| 64 | + | // After the production MsgAck path advances the renderer shadow to state 1, | |
| 65 | + | // an unchanged repaint is a no-op. | |
| 66 | + | d.paint(sess, ac, false) | |
| 67 | + | requireNoOutputFrame(t, tr.sends) | |
| 68 | + | } | |
| 69 | + | ||
| 70 | + | func TestLocalAttachStillAdvancesRendererOnSend(t *testing.T) { | |
| 71 | + | p, releasePTY := newBlockingPTY(t) | |
| 72 | + | defer releasePTY() | |
| 73 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |
| 74 | + | ||
| 75 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |
| 76 | + | d.paint(sess, ac, false) | |
| 77 | + | awaitFrame(t, sends, ports.MsgOutput) | |
| 78 | + | ||
| 79 | + | // Local/default transports keep AdvanceOnSend: the first paint updates the | |
| 80 | + | // renderer shadow without waiting for MsgAck. | |
| 81 | + | d.paint(sess, ac, false) | |
| 82 | + | requireNoOutputFrame(t, sends) | |
| 83 | + | } | |
| 84 | + | ||
| 85 | + | func TestLocalOutputAckDoesNotMoveRendererShadowBackward(t *testing.T) { | |
| 86 | + | p, releasePTY := newBlockingPTY(t) | |
| 87 | + | defer releasePTY() | |
| 88 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |
| 89 | + | pane := sess.tabs[0].focusedPane() | |
| 90 | + | ||
| 91 | + | pane.screen.Write([]byte("A")) | |
| 92 | + | d.paint(sess, ac, false) | |
| 93 | + | first := awaitFrame(t, sends, ports.MsgOutput) | |
| 94 | + | out, err := ports.UnmarshalOutput(first.Payload) | |
| 95 | + | require.NoError(t, err) | |
| 96 | + | ||
| 97 | + | pane.screen.Write([]byte("\rB")) | |
| 98 | + | d.paint(sess, ac, false) | |
| 99 | + | awaitFrame(t, sends, ports.MsgOutput) | |
| 100 | + | ||
| 101 | + | require.NoError(t, ac.ackOutputState(out.NewStateNum)) | |
| 102 | + | d.paint(sess, ac, false) | |
| 103 | + | requireNoOutputFrame(t, sends) | |
| 104 | + | } | |
| @@ -0,0 +1,104 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "io" | |||
| 5 | + | "sync" | |||
| 6 | + | "testing" | |||
| 7 | + | ||||
| 8 | + | "github.com/bnema/vev/internal/domain" | |||
| 9 | + | "github.com/bnema/vev/internal/ports" | |||
| 10 | + | "github.com/stretchr/testify/require" | |||
| 11 | + | ) | |||
| 12 | + | ||||
| 13 | + | type datagramTestTransport struct { | |||
| 14 | + | sends chan ports.Frame | |||
| 15 | + | recv chan ports.Frame | |||
| 16 | + | once sync.Once | |||
| 17 | + | } | |||
| 18 | + | ||||
| 19 | + | func newDatagramTestTransport() *datagramTestTransport { | |||
| 20 | + | return &datagramTestTransport{sends: make(chan ports.Frame, 64), recv: make(chan ports.Frame, 64)} | |||
| 21 | + | } | |||
| 22 | + | ||||
| 23 | + | func (t *datagramTestTransport) Send(f ports.Frame) error { t.sends <- f; return nil } | |||
| 24 | + | func (t *datagramTestTransport) Recv() (ports.Frame, error) { | |||
| 25 | + | f, ok := <-t.recv | |||
| 26 | + | if !ok { | |||
| 27 | + | return ports.Frame{}, io.EOF | |||
| 28 | + | } | |||
| 29 | + | return f, nil | |||
| 30 | + | } | |||
| 31 | + | func (t *datagramTestTransport) Close() error { t.once.Do(func() { close(t.recv) }); return nil } | |||
| 32 | + | ||||
| 33 | + | func TestDatagramAttachAdvancesRendererOnlyOnAck(t *testing.T) { | |||
| 34 | + | p, releasePTY := newBlockingPTY(t) | |||
| 35 | + | defer releasePTY() | |||
| 36 | + | d, sess, _, _ := newManualSessionWithPTYs(t, p) | |||
| 37 | + | tr := newDatagramTestTransport() | |||
| 38 | + | routed, ac, err := d.route(ports.Hello{ | |||
| 39 | + | Version: ports.ProtocolVersion, | |||
| 40 | + | Intent: ports.IntentAttach, | |||
| 41 | + | Name: sess.name, | |||
| 42 | + | Size: domain.Size{Cols: 80, Rows: 24}, | |||
| 43 | + | AckOutput: true, | |||
| 44 | + | }, tr) | |||
| 45 | + | require.NoError(t, err) | |||
| 46 | + | require.Same(t, sess, routed) | |||
| 47 | + | ||||
| 48 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |||
| 49 | + | d.paint(sess, ac, false) | |||
| 50 | + | first := awaitFrame(t, tr.sends, ports.MsgOutput) | |||
| 51 | + | out, err := ports.UnmarshalOutput(first.Payload) | |||
| 52 | + | require.NoError(t, err) | |||
| 53 | + | require.Equal(t, uint64(1), out.NewStateNum) | |||
| 54 | + | ||||
| 55 | + | // Before the MsgAck, AdvanceOnAck must keep rediffing from the unacked | |||
| 56 | + | // shadow even if the composed screen is unchanged. | |||
| 57 | + | d.paint(sess, ac, false) | |||
| 58 | + | awaitFrame(t, tr.sends, ports.MsgOutput) | |||
| 59 | + | ||||
| 60 | + | tr.recv <- ports.Frame{Type: ports.MsgAck, Payload: ports.MarshalAck(ports.Ack{AckedStateNum: out.NewStateNum})} | |||
| 61 | + | require.NoError(t, tr.Close()) | |||
| 62 | + | d.runConnLoop(ac) | |||
| 63 | + | ||||
| 64 | + | // After the production MsgAck path advances the renderer shadow to state 1, | |||
| 65 | + | // an unchanged repaint is a no-op. | |||
| 66 | + | d.paint(sess, ac, false) | |||
| 67 | + | requireNoOutputFrame(t, tr.sends) | |||
| 68 | + | } | |||
| 69 | + | ||||
| 70 | + | func TestLocalAttachStillAdvancesRendererOnSend(t *testing.T) { | |||
| 71 | + | p, releasePTY := newBlockingPTY(t) | |||
| 72 | + | defer releasePTY() | |||
| 73 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |||
| 74 | + | ||||
| 75 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |||
| 76 | + | d.paint(sess, ac, false) | |||
| 77 | + | awaitFrame(t, sends, ports.MsgOutput) | |||
| 78 | + | ||||
| 79 | + | // Local/default transports keep AdvanceOnSend: the first paint updates the | |||
| 80 | + | // renderer shadow without waiting for MsgAck. | |||
| 81 | + | d.paint(sess, ac, false) | |||
| 82 | + | requireNoOutputFrame(t, sends) | |||
| 83 | + | } | |||
| 84 | + | ||||
| 85 | + | func TestLocalOutputAckDoesNotMoveRendererShadowBackward(t *testing.T) { | |||
| 86 | + | p, releasePTY := newBlockingPTY(t) | |||
| 87 | + | defer releasePTY() | |||
| 88 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |||
| 89 | + | pane := sess.tabs[0].focusedPane() | |||
| 90 | + | ||||
| 91 | + | pane.screen.Write([]byte("A")) | |||
| 92 | + | d.paint(sess, ac, false) | |||
| 93 | + | first := awaitFrame(t, sends, ports.MsgOutput) | |||
| 94 | + | out, err := ports.UnmarshalOutput(first.Payload) | |||
| 95 | + | require.NoError(t, err) | |||
| 96 | + | ||||
| 97 | + | pane.screen.Write([]byte("\rB")) | |||
| 98 | + | d.paint(sess, ac, false) | |||
| 99 | + | awaitFrame(t, sends, ports.MsgOutput) | |||
| 100 | + | ||||
| 101 | + | require.NoError(t, ac.ackOutputState(out.NewStateNum)) | |||
| 102 | + | d.paint(sess, ac, false) | |||
| 103 | + | requireNoOutputFrame(t, sends) | |||
| 104 | + | } | |||
M internal/usecase/daemon/render.go +7 -1
| @@ -400,7 +400,13 @@ func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool) { | |||
| 400 | 400 | if sendTr == nil { | |
| 401 | 401 | serr = errors.New("client transport is nil") | |
| 402 | 402 | } else { | |
| 403 | - | serr = sendTr.Send(ac.nextOutputFrameLocked(data, reset)) | |
| 403 | + | out := ac.nextOutputFrameLocked(data, reset) | |
| 404 | + | if ac.advanceOutputOnAck && ac.outputFrames != nil { | |
| 405 | + | if msg, derr := ports.UnmarshalOutput(out.Payload); derr == nil { | |
| 406 | + | ac.outputFrames[msg.NewStateNum] = frame.Clone() | |
| 407 | + | } | |
| 408 | + | } | |
| 409 | + | serr = sendTr.Send(out) | |
| 404 | 410 | } | |
| 405 | 411 | } | |
| 406 | 412 | } | |
| @@ -400,7 +400,13 @@ func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool) { | |||||
| 400 | if sendTr == nil { | 400 | if sendTr == nil { | ||
| 401 | serr = errors.New("client transport is nil") | 401 | serr = errors.New("client transport is nil") | ||
| 402 | } else { | 402 | } else { | ||
| 403 | - | serr = sendTr.Send(ac.nextOutputFrameLocked(data, reset)) | 403 | + | out := ac.nextOutputFrameLocked(data, reset) |
| 404 | + | if ac.advanceOutputOnAck && ac.outputFrames != nil { | |||
| 405 | + | if msg, derr := ports.UnmarshalOutput(out.Payload); derr == nil { | |||
| 406 | + | ac.outputFrames[msg.NewStateNum] = frame.Clone() | |||
| 407 | + | } | |||
| 408 | + | } | |||
| 409 | + | serr = sendTr.Send(out) | |||
| 404 | } | 410 | } | ||
| 405 | } | 411 | } | ||
| 406 | } | 412 | } | ||
M pkg/renderer/frame.go +11 -0
| @@ -29,6 +29,17 @@ func NewFrame(width, height int) Frame { | |||
| 29 | 29 | return Frame{Width: width, Height: height, Cells: cells, lineOffset: lineOffset} | |
| 30 | 30 | } | |
| 31 | 31 | ||
| 32 | + | // Clone returns an independent copy preserving logical row contents. The clone | |
| 33 | + | // uses canonical row storage so later scrolls or writes to the source frame do | |
| 34 | + | // not affect it. | |
| 35 | + | func (f Frame) Clone() Frame { | |
| 36 | + | clone := NewFrame(f.Width, f.Height) | |
| 37 | + | for y := range f.Height { | |
| 38 | + | copy(clone.Row(y), f.Row(y)) | |
| 39 | + | } | |
| 40 | + | return clone | |
| 41 | + | } | |
| 42 | + | ||
| 32 | 43 | func (f Frame) Validate() error { | |
| 33 | 44 | if f.Width <= 0 || f.Height <= 0 { | |
| 34 | 45 | return fmt.Errorf("invalid frame size %dx%d", f.Width, f.Height) | |
| @@ -29,6 +29,17 @@ func NewFrame(width, height int) Frame { | |||||
| 29 | return Frame{Width: width, Height: height, Cells: cells, lineOffset: lineOffset} | 29 | return Frame{Width: width, Height: height, Cells: cells, lineOffset: lineOffset} | ||
| 30 | } | 30 | } | ||
| 31 | 31 | ||||
| 32 | + | // Clone returns an independent copy preserving logical row contents. The clone | |||
| 33 | + | // uses canonical row storage so later scrolls or writes to the source frame do | |||
| 34 | + | // not affect it. | |||
| 35 | + | func (f Frame) Clone() Frame { | |||
| 36 | + | clone := NewFrame(f.Width, f.Height) | |||
| 37 | + | for y := range f.Height { | |||
| 38 | + | copy(clone.Row(y), f.Row(y)) | |||
| 39 | + | } | |||
| 40 | + | return clone | |||
| 41 | + | } | |||
| 42 | + | ||||
| 32 | func (f Frame) Validate() error { | 43 | func (f Frame) Validate() error { | ||
| 33 | if f.Width <= 0 || f.Height <= 0 { | 44 | if f.Width <= 0 || f.Height <= 0 { | ||
| 34 | return fmt.Errorf("invalid frame size %dx%d", f.Width, f.Height) | 45 | return fmt.Errorf("invalid frame size %dx%d", f.Width, f.Height) | ||
M pkg/renderer/renderer.go +1 -1
| @@ -219,7 +219,7 @@ func (r *Renderer) lineDirty(frame Frame, y int) bool { | |||
| 219 | 219 | } | |
| 220 | 220 | ||
| 221 | 221 | // Ack advances an AdvanceOnAck renderer's shadow after the caller knows the | |
| 222 | - | // bytes for frame reached the terminal. It is harmless for AdvanceOnSend. | |
| 222 | + | // bytes for frame reached the terminal. | |
| 223 | 223 | func (r *Renderer) Ack(frame Frame) error { | |
| 224 | 224 | if err := frame.Validate(); err != nil { | |
| 225 | 225 | return err | |
| @@ -219,7 +219,7 @@ func (r *Renderer) lineDirty(frame Frame, y int) bool { | |||||
| 219 | } | 219 | } | ||
| 220 | 220 | ||||
| 221 | // Ack advances an AdvanceOnAck renderer's shadow after the caller knows the | 221 | // Ack advances an AdvanceOnAck renderer's shadow after the caller knows the | ||
| 222 | - | // bytes for frame reached the terminal. It is harmless for AdvanceOnSend. | 222 | + | // bytes for frame reached the terminal. |
| 223 | func (r *Renderer) Ack(frame Frame) error { | 223 | func (r *Renderer) Ack(frame Frame) error { | ||
| 224 | if err := frame.Validate(); err != nil { | 224 | if err := frame.Validate(); err != nil { | ||
| 225 | return err | 225 | return err | ||