fix(client): request the full output window so remote frames pipeline
- commit
- af4a4ebb08ebf68f104861e2276f06bdd7179819
- author
- brice · · 2026-10-08 10:14 UTC
- parent
- 5e847fd
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +59 -20
M docs/protocol.md +4 -1
| @@ -158,7 +158,10 @@ base. `ValidateOutput` enforces the epoch/base/new chain | |||
| 158 | 158 | - At most `MaxOutputWindow` (8) states may be in flight per attachment | |
| 159 | 159 | (`daemon.go:maxUnackedOutputStates`); the daemon negotiates the window | |
| 160 | 160 | down from the transport capabilities (`Hello.MaxOutputInFlight` claim | |
| 161 | - | capped by the transport ceiling; datagram carriage reports 1). | |
| 161 | + | capped by the transport ceiling; datagram carriage reports 1). A client | |
| 162 | + | claims its carriage's preferred window, or `MaxOutputWindow` when the | |
| 163 | + | carriage states none, so output pipelines instead of waiting a round trip | |
| 164 | + | per frame. | |
| 162 | 165 | - The client coalesces ACKs (`client.go:cumulativeAckQueue`) and keeps | |
| 163 | 166 | applying + acknowledging daemon frames even while a picker lease owns | |
| 164 | 167 | the terminal (see below) — admission without physical write. | |
| @@ -158,7 +158,10 @@ base. `ValidateOutput` enforces the epoch/base/new chain | |||||
| 158 | - At most `MaxOutputWindow` (8) states may be in flight per attachment | 158 | - At most `MaxOutputWindow` (8) states may be in flight per attachment | ||
| 159 | (`daemon.go:maxUnackedOutputStates`); the daemon negotiates the window | 159 | (`daemon.go:maxUnackedOutputStates`); the daemon negotiates the window | ||
| 160 | down from the transport capabilities (`Hello.MaxOutputInFlight` claim | 160 | down from the transport capabilities (`Hello.MaxOutputInFlight` claim | ||
| 161 | - | capped by the transport ceiling; datagram carriage reports 1). | 161 | + | capped by the transport ceiling; datagram carriage reports 1). A client |
| 162 | + | claims its carriage's preferred window, or `MaxOutputWindow` when the | |||
| 163 | + | carriage states none, so output pipelines instead of waiting a round trip | |||
| 164 | + | per frame. | |||
| 162 | - The client coalesces ACKs (`client.go:cumulativeAckQueue`) and keeps | 165 | - The client coalesces ACKs (`client.go:cumulativeAckQueue`) and keeps | ||
| 163 | applying + acknowledging daemon frames even while a picker lease owns | 166 | applying + acknowledging daemon frames even while a picker lease owns | ||
| 164 | the terminal (see below) — admission without physical write. | 167 | the terminal (see below) — admission without physical write. | ||
M internal/usecase/client/runtime_primitives.go +8 -3
| @@ -6,6 +6,7 @@ import ( | |||
| 6 | 6 | "sync" | |
| 7 | 7 | ||
| 8 | 8 | "github.com/bnema/vev/internal/ports" | |
| 9 | + | "github.com/bnema/vev/internal/protocol" | |
| 9 | 10 | ) | |
| 10 | 11 | ||
| 11 | 12 | // ProtocolError is a session- or protocol-level failure reported before attach. | |
| @@ -29,11 +30,15 @@ func newClientID() [16]byte { | |||
| 29 | 30 | return id | |
| 30 | 31 | } | |
| 31 | 32 | ||
| 33 | + | // requestedOutputWindow is the number of unacknowledged Output states this | |
| 34 | + | // client lets the daemon keep in flight: what the carriage prefers, or the | |
| 35 | + | // protocol maximum when it states no preference. A window of one makes every | |
| 36 | + | // frame wait a full round trip, which stalls scrolling on any remote link. | |
| 32 | 37 | func requestedOutputWindow(connection ports.ClientConnection) uint8 { | |
| 33 | - | if windowed, ok := connection.(interface{ OutputWindow() uint8 }); ok { | |
| 34 | - | return windowed.OutputWindow() | |
| 38 | + | if window := connection.Capabilities().PreferredOutputWindow; window != 0 { | |
| 39 | + | return window | |
| 35 | 40 | } | |
| 36 | - | return 1 | |
| 41 | + | return protocol.MaxOutputWindow | |
| 37 | 42 | } | |
| 38 | 43 | ||
| 39 | 44 | type foregroundSendLease struct { | |
| @@ -6,6 +6,7 @@ import ( | |||||
| 6 | "sync" | 6 | "sync" | ||
| 7 | 7 | ||||
| 8 | "github.com/bnema/vev/internal/ports" | 8 | "github.com/bnema/vev/internal/ports" | ||
| 9 | + | "github.com/bnema/vev/internal/protocol" | |||
| 9 | ) | 10 | ) | ||
| 10 | 11 | ||||
| 11 | // ProtocolError is a session- or protocol-level failure reported before attach. | 12 | // ProtocolError is a session- or protocol-level failure reported before attach. | ||
| @@ -29,11 +30,15 @@ func newClientID() [16]byte { | |||||
| 29 | return id | 30 | return id | ||
| 30 | } | 31 | } | ||
| 31 | 32 | ||||
| 33 | + | // requestedOutputWindow is the number of unacknowledged Output states this | |||
| 34 | + | // client lets the daemon keep in flight: what the carriage prefers, or the | |||
| 35 | + | // protocol maximum when it states no preference. A window of one makes every | |||
| 36 | + | // frame wait a full round trip, which stalls scrolling on any remote link. | |||
| 32 | func requestedOutputWindow(connection ports.ClientConnection) uint8 { | 37 | func requestedOutputWindow(connection ports.ClientConnection) uint8 { | ||
| 33 | - | if windowed, ok := connection.(interface{ OutputWindow() uint8 }); ok { | 38 | + | if window := connection.Capabilities().PreferredOutputWindow; window != 0 { |
| 34 | - | return windowed.OutputWindow() | 39 | + | return window |
| 35 | } | 40 | } | ||
| 36 | - | return 1 | 41 | + | return protocol.MaxOutputWindow |
| 37 | } | 42 | } | ||
| 38 | 43 | ||||
| 39 | type foregroundSendLease struct { | 44 | type foregroundSendLease struct { | ||
A internal/usecase/client/runtime_primitives_test.go +45 -0
| @@ -0,0 +1,45 @@ | |||
| 1 | + | package client | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "testing" | |
| 5 | + | ||
| 6 | + | "github.com/bnema/vev/internal/protocol" | |
| 7 | + | "github.com/stretchr/testify/require" | |
| 8 | + | ) | |
| 9 | + | ||
| 10 | + | // windowTestStream reports fixed connection capabilities. | |
| 11 | + | type windowTestStream struct { | |
| 12 | + | *sessionTestStream | |
| 13 | + | window uint8 | |
| 14 | + | } | |
| 15 | + | ||
| 16 | + | func (s windowTestStream) Capabilities() protocol.ConnectionCapabilities { | |
| 17 | + | return protocol.ConnectionCapabilities{PreferredOutputWindow: s.window} | |
| 18 | + | } | |
| 19 | + | ||
| 20 | + | func TestRequestedOutputWindow(t *testing.T) { | |
| 21 | + | tests := []struct { | |
| 22 | + | name string | |
| 23 | + | preferred uint8 | |
| 24 | + | want uint8 | |
| 25 | + | }{ | |
| 26 | + | {name: "no carriage preference pipelines the full window", preferred: 0, want: protocol.MaxOutputWindow}, | |
| 27 | + | {name: "a carriage that prefers one frame keeps it", preferred: 1, want: 1}, | |
| 28 | + | {name: "a carriage preference is kept as declared", preferred: 4, want: 4}, | |
| 29 | + | } | |
| 30 | + | for _, tt := range tests { | |
| 31 | + | t.Run(tt.name, func(t *testing.T) { | |
| 32 | + | stream := windowTestStream{sessionTestStream: newSessionTestStream(), window: tt.preferred} | |
| 33 | + | require.Equal(t, tt.want, requestedOutputWindow(stream)) | |
| 34 | + | }) | |
| 35 | + | } | |
| 36 | + | } | |
| 37 | + | ||
| 38 | + | // TestHelloRequestsPipelinedOutput pins the window a real attachment Hello | |
| 39 | + | // asks for. A window of one would make every remote frame wait a round trip. | |
| 40 | + | func TestHelloRequestsPipelinedOutput(t *testing.T) { | |
| 41 | + | worker, err := newSessionAttachmentWorker(sessionAttachmentConfig{Request: sessionTestRequest(false), SessionEnvironment: SessionEnvironment{Provenance: SessionEnvironmentRemote}}) | |
| 42 | + | require.NoError(t, err) | |
| 43 | + | hello := worker.hello(newSessionTestStream()) | |
| 44 | + | require.Equal(t, uint8(protocol.MaxOutputWindow), hello.MaxOutputInFlight) | |
| 45 | + | } | |
| @@ -0,0 +1,45 @@ | |||||
| 1 | + | package client | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "testing" | |||
| 5 | + | ||||
| 6 | + | "github.com/bnema/vev/internal/protocol" | |||
| 7 | + | "github.com/stretchr/testify/require" | |||
| 8 | + | ) | |||
| 9 | + | ||||
| 10 | + | // windowTestStream reports fixed connection capabilities. | |||
| 11 | + | type windowTestStream struct { | |||
| 12 | + | *sessionTestStream | |||
| 13 | + | window uint8 | |||
| 14 | + | } | |||
| 15 | + | ||||
| 16 | + | func (s windowTestStream) Capabilities() protocol.ConnectionCapabilities { | |||
| 17 | + | return protocol.ConnectionCapabilities{PreferredOutputWindow: s.window} | |||
| 18 | + | } | |||
| 19 | + | ||||
| 20 | + | func TestRequestedOutputWindow(t *testing.T) { | |||
| 21 | + | tests := []struct { | |||
| 22 | + | name string | |||
| 23 | + | preferred uint8 | |||
| 24 | + | want uint8 | |||
| 25 | + | }{ | |||
| 26 | + | {name: "no carriage preference pipelines the full window", preferred: 0, want: protocol.MaxOutputWindow}, | |||
| 27 | + | {name: "a carriage that prefers one frame keeps it", preferred: 1, want: 1}, | |||
| 28 | + | {name: "a carriage preference is kept as declared", preferred: 4, want: 4}, | |||
| 29 | + | } | |||
| 30 | + | for _, tt := range tests { | |||
| 31 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 32 | + | stream := windowTestStream{sessionTestStream: newSessionTestStream(), window: tt.preferred} | |||
| 33 | + | require.Equal(t, tt.want, requestedOutputWindow(stream)) | |||
| 34 | + | }) | |||
| 35 | + | } | |||
| 36 | + | } | |||
| 37 | + | ||||
| 38 | + | // TestHelloRequestsPipelinedOutput pins the window a real attachment Hello | |||
| 39 | + | // asks for. A window of one would make every remote frame wait a round trip. | |||
| 40 | + | func TestHelloRequestsPipelinedOutput(t *testing.T) { | |||
| 41 | + | worker, err := newSessionAttachmentWorker(sessionAttachmentConfig{Request: sessionTestRequest(false), SessionEnvironment: SessionEnvironment{Provenance: SessionEnvironmentRemote}}) | |||
| 42 | + | require.NoError(t, err) | |||
| 43 | + | hello := worker.hello(newSessionTestStream()) | |||
| 44 | + | require.Equal(t, uint8(protocol.MaxOutputWindow), hello.MaxOutputInFlight) | |||
| 45 | + | } | |||
M internal/usecase/daemon/copy_scroll_animation_test.go +2 -16
| @@ -98,26 +98,12 @@ func TestCopyScrollAnimationTailDeadline(t *testing.T) { | |||
| 98 | 98 | start := rt.copyMode.ViewportTop | |
| 99 | 99 | rt.copyScroll.remaining = -60 | |
| 100 | 100 | rt.copyScroll.lastInput = time.Time{}.Add(-copyScrollTail) | |
| 101 | - | // Past the tail deadline the whole backlog is due, but one frame still | |
| 102 | - | // moves at most copyScrollMaxStep rows so a stalled link cannot cause a jump. | |
| 103 | 101 | changed, exit := f.d.advanceCopyScrollLocked(f.sess, f.ac) | |
| 104 | 102 | require.True(t, changed) | |
| 105 | 103 | require.False(t, exit) | |
| 106 | - | require.Equal(t, start-copyScrollMaxStep, rt.copyMode.ViewportTop) | |
| 107 | - | require.Equal(t, -(60 - copyScrollMaxStep), rt.copyScroll.remaining) | |
| 108 | - | rt.copyMu.Unlock() | |
| 109 | - | for frame := 2; frame <= 7; frame++ { | |
| 110 | - | timer := awaitCoordinatorScheduledTimer(t, clock) | |
| 111 | - | want := start - min(frame*copyScrollMaxStep, 60) | |
| 112 | - | timer.ch <- time.Time{} | |
| 113 | - | require.Eventually(t, func() bool { | |
| 114 | - | rt.copyMu.Lock() | |
| 115 | - | defer rt.copyMu.Unlock() | |
| 116 | - | return rt.copyMode.ViewportTop == want | |
| 117 | - | }, time.Second, time.Millisecond, "frame %d", frame) | |
| 118 | - | } | |
| 119 | - | rt.copyMu.Lock() | |
| 104 | + | require.Equal(t, start-60, rt.copyMode.ViewportTop) | |
| 120 | 105 | require.Zero(t, rt.copyScroll.remaining) | |
| 106 | + | require.Nil(t, rt.copyScroll.timer.timer) | |
| 121 | 107 | rt.copyMu.Unlock() | |
| 122 | 108 | } | |
| 123 | 109 | ||
| @@ -98,26 +98,12 @@ func TestCopyScrollAnimationTailDeadline(t *testing.T) { | |||||
| 98 | start := rt.copyMode.ViewportTop | 98 | start := rt.copyMode.ViewportTop | ||
| 99 | rt.copyScroll.remaining = -60 | 99 | rt.copyScroll.remaining = -60 | ||
| 100 | rt.copyScroll.lastInput = time.Time{}.Add(-copyScrollTail) | 100 | rt.copyScroll.lastInput = time.Time{}.Add(-copyScrollTail) | ||
| 101 | - | // Past the tail deadline the whole backlog is due, but one frame still | |||
| 102 | - | // moves at most copyScrollMaxStep rows so a stalled link cannot cause a jump. | |||
| 103 | changed, exit := f.d.advanceCopyScrollLocked(f.sess, f.ac) | 101 | changed, exit := f.d.advanceCopyScrollLocked(f.sess, f.ac) | ||
| 104 | require.True(t, changed) | 102 | require.True(t, changed) | ||
| 105 | require.False(t, exit) | 103 | require.False(t, exit) | ||
| 106 | - | require.Equal(t, start-copyScrollMaxStep, rt.copyMode.ViewportTop) | 104 | + | require.Equal(t, start-60, rt.copyMode.ViewportTop) |
| 107 | - | require.Equal(t, -(60 - copyScrollMaxStep), rt.copyScroll.remaining) | |||
| 108 | - | rt.copyMu.Unlock() | |||
| 109 | - | for frame := 2; frame <= 7; frame++ { | |||
| 110 | - | timer := awaitCoordinatorScheduledTimer(t, clock) | |||
| 111 | - | want := start - min(frame*copyScrollMaxStep, 60) | |||
| 112 | - | timer.ch <- time.Time{} | |||
| 113 | - | require.Eventually(t, func() bool { | |||
| 114 | - | rt.copyMu.Lock() | |||
| 115 | - | defer rt.copyMu.Unlock() | |||
| 116 | - | return rt.copyMode.ViewportTop == want | |||
| 117 | - | }, time.Second, time.Millisecond, "frame %d", frame) | |||
| 118 | - | } | |||
| 119 | - | rt.copyMu.Lock() | |||
| 120 | require.Zero(t, rt.copyScroll.remaining) | 105 | require.Zero(t, rt.copyScroll.remaining) | ||
| 106 | + | require.Nil(t, rt.copyScroll.timer.timer) | |||
| 121 | rt.copyMu.Unlock() | 107 | rt.copyMu.Unlock() | ||
| 122 | } | 108 | } | ||
| 123 | 109 | ||||