improve(daemon): advance UDP output on ack
- commit
- 5d4ffc4d4841d110b233fcd171260c218783c279
- author
- brice · · 2026-07-09 16:36 UTC
- parent
- cfccd91
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
7 files · +206 -33
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/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/usecase/daemon/client.go +73 -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,12 @@ func (d *Daemon) notifiesSnapshot() []chan struct{} { | |||
| 396 | 425 | type attachClientOptions struct { | |
| 397 | 426 | clientID [16]byte | |
| 398 | 427 | resumeCapable bool | |
| 428 | + | datagram bool | |
| 429 | + | } | |
| 430 | + | ||
| 431 | + | func isDatagramTransport(tr ports.Transport) bool { | |
| 432 | + | _, ok := tr.(ports.DatagramTransport) | |
| 433 | + | return ok | |
| 399 | 434 | } | |
| 400 | 435 | ||
| 401 | 436 | func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, opts attachClientOptions) (*attachedClient, *attachedClient) { | |
| @@ -403,13 +438,20 @@ func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, | |||
| 403 | 438 | if opts.resumeCapable { | |
| 404 | 439 | resumeToken = d.nextResumeTokenLocked() | |
| 405 | 440 | } | |
| 441 | + | caps := renderer.Capabilities{} | |
| 442 | + | advanceOutputOnAck := opts.datagram || isDatagramTransport(tr) | |
| 443 | + | if advanceOutputOnAck { | |
| 444 | + | caps.AdvancePolicy = renderer.AdvanceOnAck | |
| 445 | + | } | |
| 406 | 446 | 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, | |
| 447 | + | tr: tr, | |
| 448 | + | rend: renderer.New(caps), | |
| 449 | + | advanceOutputOnAck: advanceOutputOnAck, | |
| 450 | + | outputFrames: make(map[uint64]renderer.Frame), | |
| 451 | + | size: sz, | |
| 452 | + | clientID: opts.clientID, | |
| 453 | + | resumeCapable: opts.resumeCapable, | |
| 454 | + | resumeToken: resumeToken, | |
| 413 | 455 | } | |
| 414 | 456 | ac.initOverlays() | |
| 415 | 457 | ac.setSession(sess) | |
| @@ -521,7 +563,9 @@ func (d *Daemon) runConnLoop(ac *attachedClient) { | |||
| 521 | 563 | return | |
| 522 | 564 | case ports.MsgAck: | |
| 523 | 565 | if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil { | |
| 524 | - | ac.advanceOutputAck(ack.AckedStateNum) | |
| 566 | + | if err := ac.ackOutputState(ack.AckedStateNum); err != nil { | |
| 567 | + | d.log.Warn("output ack rejected", "err", err, "state", ack.AckedStateNum) | |
| 568 | + | } | |
| 525 | 569 | if ac.takeDeferredPaint() { | |
| 526 | 570 | d.paint(sess, ac, false) | |
| 527 | 571 | } | |
| @@ -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,12 @@ 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 | + | datagram bool | |||
| 429 | + | } | |||
| 430 | + | ||||
| 431 | + | func isDatagramTransport(tr ports.Transport) bool { | |||
| 432 | + | _, ok := tr.(ports.DatagramTransport) | |||
| 433 | + | return ok | |||
| 399 | } | 434 | } | ||
| 400 | 435 | ||||
| 401 | func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, opts attachClientOptions) (*attachedClient, *attachedClient) { | 436 | func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, opts attachClientOptions) (*attachedClient, *attachedClient) { | ||
| @@ -403,13 +438,20 @@ func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size, | |||||
| 403 | if opts.resumeCapable { | 438 | if opts.resumeCapable { | ||
| 404 | resumeToken = d.nextResumeTokenLocked() | 439 | resumeToken = d.nextResumeTokenLocked() | ||
| 405 | } | 440 | } | ||
| 441 | + | caps := renderer.Capabilities{} | |||
| 442 | + | advanceOutputOnAck := opts.datagram || isDatagramTransport(tr) | |||
| 443 | + | if advanceOutputOnAck { | |||
| 444 | + | caps.AdvancePolicy = renderer.AdvanceOnAck | |||
| 445 | + | } | |||
| 406 | ac := &attachedClient{ | 446 | ac := &attachedClient{ | ||
| 407 | - | tr: tr, | 447 | + | tr: tr, |
| 408 | - | rend: renderer.New(renderer.Capabilities{}), | 448 | + | rend: renderer.New(caps), |
| 409 | - | size: sz, | 449 | + | advanceOutputOnAck: advanceOutputOnAck, |
| 410 | - | clientID: opts.clientID, | 450 | + | outputFrames: make(map[uint64]renderer.Frame), |
| 411 | - | resumeCapable: opts.resumeCapable, | 451 | + | size: sz, |
| 412 | - | resumeToken: resumeToken, | 452 | + | clientID: opts.clientID, |
| 453 | + | resumeCapable: opts.resumeCapable, | |||
| 454 | + | resumeToken: resumeToken, | |||
| 413 | } | 455 | } | ||
| 414 | ac.initOverlays() | 456 | ac.initOverlays() | ||
| 415 | ac.setSession(sess) | 457 | ac.setSession(sess) | ||
| @@ -521,7 +563,9 @@ func (d *Daemon) runConnLoop(ac *attachedClient) { | |||||
| 521 | return | 563 | return | ||
| 522 | case ports.MsgAck: | 564 | case ports.MsgAck: | ||
| 523 | if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil { | 565 | if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil { | ||
| 524 | - | ac.advanceOutputAck(ack.AckedStateNum) | 566 | + | if err := ac.ackOutputState(ack.AckedStateNum); err != nil { |
| 567 | + | d.log.Warn("output ack rejected", "err", err, "state", ack.AckedStateNum) | |||
| 568 | + | } | |||
| 525 | if ac.takeDeferredPaint() { | 569 | if ac.takeDeferredPaint() { | ||
| 526 | d.paint(sess, ac, false) | 570 | d.paint(sess, ac, false) | ||
| 527 | } | 571 | } | ||
A internal/usecase/daemon/output_state_sync_test.go +99 -0
| @@ -0,0 +1,99 @@ | |||
| 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 | + | func (t *datagramTestTransport) DatagramTransport() {} | |
| 33 | + | ||
| 34 | + | func TestDatagramAttachAdvancesRendererOnlyOnAck(t *testing.T) { | |
| 35 | + | p, releasePTY := newBlockingPTY(t) | |
| 36 | + | defer releasePTY() | |
| 37 | + | d, sess, old, _ := newManualSessionWithPTYs(t, p) | |
| 38 | + | tr := newDatagramTestTransport() | |
| 39 | + | ac, replaced := d.attachClient(sess, tr, domain.Size{Cols: 80, Rows: 24}, attachClientOptions{datagram: true}) | |
| 40 | + | require.Same(t, old, replaced) | |
| 41 | + | defer d.detachReplacedClient(replaced) | |
| 42 | + | ||
| 43 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |
| 44 | + | d.paint(sess, ac, false) | |
| 45 | + | first := awaitFrame(t, tr.sends, ports.MsgOutput) | |
| 46 | + | out, err := ports.UnmarshalOutput(first.Payload) | |
| 47 | + | require.NoError(t, err) | |
| 48 | + | require.Equal(t, uint64(1), out.NewStateNum) | |
| 49 | + | ||
| 50 | + | // Before the MsgAck, AdvanceOnAck must keep rediffing from the unacked | |
| 51 | + | // shadow even if the composed screen is unchanged. | |
| 52 | + | d.paint(sess, ac, false) | |
| 53 | + | awaitFrame(t, tr.sends, ports.MsgOutput) | |
| 54 | + | ||
| 55 | + | tr.recv <- ports.Frame{Type: ports.MsgAck, Payload: ports.MarshalAck(ports.Ack{AckedStateNum: out.NewStateNum})} | |
| 56 | + | require.NoError(t, tr.Close()) | |
| 57 | + | d.runConnLoop(ac) | |
| 58 | + | ||
| 59 | + | // After the production MsgAck path advances the renderer shadow to state 1, | |
| 60 | + | // an unchanged repaint is a no-op. | |
| 61 | + | d.paint(sess, ac, false) | |
| 62 | + | requireNoOutputFrame(t, tr.sends) | |
| 63 | + | } | |
| 64 | + | ||
| 65 | + | func TestLocalAttachStillAdvancesRendererOnSend(t *testing.T) { | |
| 66 | + | p, releasePTY := newBlockingPTY(t) | |
| 67 | + | defer releasePTY() | |
| 68 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |
| 69 | + | ||
| 70 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |
| 71 | + | d.paint(sess, ac, false) | |
| 72 | + | awaitFrame(t, sends, ports.MsgOutput) | |
| 73 | + | ||
| 74 | + | // Local/default transports keep AdvanceOnSend: the first paint updates the | |
| 75 | + | // renderer shadow without waiting for MsgAck. | |
| 76 | + | d.paint(sess, ac, false) | |
| 77 | + | requireNoOutputFrame(t, sends) | |
| 78 | + | } | |
| 79 | + | ||
| 80 | + | func TestLocalOutputAckDoesNotMoveRendererShadowBackward(t *testing.T) { | |
| 81 | + | p, releasePTY := newBlockingPTY(t) | |
| 82 | + | defer releasePTY() | |
| 83 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |
| 84 | + | pane := sess.tabs[0].focusedPane() | |
| 85 | + | ||
| 86 | + | pane.screen.Write([]byte("A")) | |
| 87 | + | d.paint(sess, ac, false) | |
| 88 | + | first := awaitFrame(t, sends, ports.MsgOutput) | |
| 89 | + | out, err := ports.UnmarshalOutput(first.Payload) | |
| 90 | + | require.NoError(t, err) | |
| 91 | + | ||
| 92 | + | pane.screen.Write([]byte("\rB")) | |
| 93 | + | d.paint(sess, ac, false) | |
| 94 | + | awaitFrame(t, sends, ports.MsgOutput) | |
| 95 | + | ||
| 96 | + | require.NoError(t, ac.ackOutputState(out.NewStateNum)) | |
| 97 | + | d.paint(sess, ac, false) | |
| 98 | + | requireNoOutputFrame(t, sends) | |
| 99 | + | } | |
| @@ -0,0 +1,99 @@ | |||||
| 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 | + | func (t *datagramTestTransport) DatagramTransport() {} | |||
| 33 | + | ||||
| 34 | + | func TestDatagramAttachAdvancesRendererOnlyOnAck(t *testing.T) { | |||
| 35 | + | p, releasePTY := newBlockingPTY(t) | |||
| 36 | + | defer releasePTY() | |||
| 37 | + | d, sess, old, _ := newManualSessionWithPTYs(t, p) | |||
| 38 | + | tr := newDatagramTestTransport() | |||
| 39 | + | ac, replaced := d.attachClient(sess, tr, domain.Size{Cols: 80, Rows: 24}, attachClientOptions{datagram: true}) | |||
| 40 | + | require.Same(t, old, replaced) | |||
| 41 | + | defer d.detachReplacedClient(replaced) | |||
| 42 | + | ||||
| 43 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |||
| 44 | + | d.paint(sess, ac, false) | |||
| 45 | + | first := awaitFrame(t, tr.sends, ports.MsgOutput) | |||
| 46 | + | out, err := ports.UnmarshalOutput(first.Payload) | |||
| 47 | + | require.NoError(t, err) | |||
| 48 | + | require.Equal(t, uint64(1), out.NewStateNum) | |||
| 49 | + | ||||
| 50 | + | // Before the MsgAck, AdvanceOnAck must keep rediffing from the unacked | |||
| 51 | + | // shadow even if the composed screen is unchanged. | |||
| 52 | + | d.paint(sess, ac, false) | |||
| 53 | + | awaitFrame(t, tr.sends, ports.MsgOutput) | |||
| 54 | + | ||||
| 55 | + | tr.recv <- ports.Frame{Type: ports.MsgAck, Payload: ports.MarshalAck(ports.Ack{AckedStateNum: out.NewStateNum})} | |||
| 56 | + | require.NoError(t, tr.Close()) | |||
| 57 | + | d.runConnLoop(ac) | |||
| 58 | + | ||||
| 59 | + | // After the production MsgAck path advances the renderer shadow to state 1, | |||
| 60 | + | // an unchanged repaint is a no-op. | |||
| 61 | + | d.paint(sess, ac, false) | |||
| 62 | + | requireNoOutputFrame(t, tr.sends) | |||
| 63 | + | } | |||
| 64 | + | ||||
| 65 | + | func TestLocalAttachStillAdvancesRendererOnSend(t *testing.T) { | |||
| 66 | + | p, releasePTY := newBlockingPTY(t) | |||
| 67 | + | defer releasePTY() | |||
| 68 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |||
| 69 | + | ||||
| 70 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |||
| 71 | + | d.paint(sess, ac, false) | |||
| 72 | + | awaitFrame(t, sends, ports.MsgOutput) | |||
| 73 | + | ||||
| 74 | + | // Local/default transports keep AdvanceOnSend: the first paint updates the | |||
| 75 | + | // renderer shadow without waiting for MsgAck. | |||
| 76 | + | d.paint(sess, ac, false) | |||
| 77 | + | requireNoOutputFrame(t, sends) | |||
| 78 | + | } | |||
| 79 | + | ||||
| 80 | + | func TestLocalOutputAckDoesNotMoveRendererShadowBackward(t *testing.T) { | |||
| 81 | + | p, releasePTY := newBlockingPTY(t) | |||
| 82 | + | defer releasePTY() | |||
| 83 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |||
| 84 | + | pane := sess.tabs[0].focusedPane() | |||
| 85 | + | ||||
| 86 | + | pane.screen.Write([]byte("A")) | |||
| 87 | + | d.paint(sess, ac, false) | |||
| 88 | + | first := awaitFrame(t, sends, ports.MsgOutput) | |||
| 89 | + | out, err := ports.UnmarshalOutput(first.Payload) | |||
| 90 | + | require.NoError(t, err) | |||
| 91 | + | ||||
| 92 | + | pane.screen.Write([]byte("\rB")) | |||
| 93 | + | d.paint(sess, ac, false) | |||
| 94 | + | awaitFrame(t, sends, ports.MsgOutput) | |||
| 95 | + | ||||
| 96 | + | require.NoError(t, ac.ackOutputState(out.NewStateNum)) | |||
| 97 | + | d.paint(sess, ac, false) | |||
| 98 | + | requireNoOutputFrame(t, sends) | |||
| 99 | + | } | |||
M internal/usecase/daemon/render.go +7 -1
| @@ -399,7 +399,13 @@ func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool) { | |||
| 399 | 399 | if sendTr == nil { | |
| 400 | 400 | serr = errors.New("client transport is nil") | |
| 401 | 401 | } else { | |
| 402 | - | serr = sendTr.Send(ac.nextOutputFrameLocked(data, reset)) | |
| 402 | + | out := ac.nextOutputFrameLocked(data, reset) | |
| 403 | + | if ac.advanceOutputOnAck && ac.outputFrames != nil { | |
| 404 | + | if msg, derr := ports.UnmarshalOutput(out.Payload); derr == nil { | |
| 405 | + | ac.outputFrames[msg.NewStateNum] = frame.Clone() | |
| 406 | + | } | |
| 407 | + | } | |
| 408 | + | serr = sendTr.Send(out) | |
| 403 | 409 | } | |
| 404 | 410 | } | |
| 405 | 411 | } | |
| @@ -399,7 +399,13 @@ func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool) { | |||||
| 399 | if sendTr == nil { | 399 | if sendTr == nil { | ||
| 400 | serr = errors.New("client transport is nil") | 400 | serr = errors.New("client transport is nil") | ||
| 401 | } else { | 401 | } else { | ||
| 402 | - | serr = sendTr.Send(ac.nextOutputFrameLocked(data, reset)) | 402 | + | out := ac.nextOutputFrameLocked(data, reset) |
| 403 | + | if ac.advanceOutputOnAck && ac.outputFrames != nil { | |||
| 404 | + | if msg, derr := ports.UnmarshalOutput(out.Payload); derr == nil { | |||
| 405 | + | ac.outputFrames[msg.NewStateNum] = frame.Clone() | |||
| 406 | + | } | |||
| 407 | + | } | |||
| 408 | + | serr = sendTr.Send(out) | |||
| 403 | } | 409 | } | ||
| 404 | } | 410 | } | ||
| 405 | } | 411 | } | ||
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 | ||