feat(daemon): acknowledge echoed input in Output.Echo
- commit
- e73c690b26cd7b4a88c5cf54a6a480da7b729aa6
- author
- brice · · 2026-10-07 11:00 UTC
- committer
- brice · · 2026-10-07 12:15 UTC
- parent
- c297fbe
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
5 files · +248 -7
M internal/usecase/daemon/client.go +4 -1
| @@ -68,7 +68,10 @@ type attachedClient struct { | |||
| 68 | 68 | // credential instead of consuming it. | |
| 69 | 69 | resumeClaimToken uint64 | |
| 70 | 70 | parked bool | |
| 71 | - | echoAck atomic.Uint64 | |
| 71 | + | // echoAck is the newest client InputSeq applied at least echoAckDelay ago; | |
| 72 | + | // echo owns its updates. | |
| 73 | + | echoAck atomic.Uint64 | |
| 74 | + | echo echoAckTracker | |
| 72 | 75 | // focus holds the domain.TerminalFocus the client last reported. It is an | |
| 73 | 76 | // atomic so any daemon path reads it without taking a lock; see | |
| 74 | 77 | // terminalFocus. | |
| @@ -68,7 +68,10 @@ type attachedClient struct { | |||||
| 68 | // credential instead of consuming it. | 68 | // credential instead of consuming it. | ||
| 69 | resumeClaimToken uint64 | 69 | resumeClaimToken uint64 | ||
| 70 | parked bool | 70 | parked bool | ||
| 71 | - | echoAck atomic.Uint64 | 71 | + | // echoAck is the newest client InputSeq applied at least echoAckDelay ago; |
| 72 | + | // echo owns its updates. | |||
| 73 | + | echoAck atomic.Uint64 | |||
| 74 | + | echo echoAckTracker | |||
| 72 | // focus holds the domain.TerminalFocus the client last reported. It is an | 75 | // focus holds the domain.TerminalFocus the client last reported. It is an | ||
| 73 | // atomic so any daemon path reads it without taking a lock; see | 76 | // atomic so any daemon path reads it without taking a lock; see | ||
| 74 | // terminalFocus. | 77 | // terminalFocus. | ||
A internal/usecase/daemon/echo_ack.go +118 -0
| @@ -0,0 +1,118 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "sync" | |
| 5 | + | "time" | |
| 6 | + | ||
| 7 | + | "github.com/bnema/vev/internal/ports" | |
| 8 | + | ) | |
| 9 | + | ||
| 10 | + | // echoAckDelay is how long sequenced input must have been applied before the | |
| 11 | + | // daemon acknowledges it in Output.Echo. Like mosh's ECHO_TIMEOUT, it gives the | |
| 12 | + | // PTY time to echo the input, so the screen published with the acknowledgement | |
| 13 | + | // already contains whatever echo the input produced. | |
| 14 | + | const echoAckDelay = 50 * time.Millisecond | |
| 15 | + | ||
| 16 | + | // maxPendingEchoes bounds the per-attachment input history. Acknowledgements | |
| 17 | + | // are cumulative, so dropping the oldest entry only delays its confirmation to | |
| 18 | + | // a newer one. | |
| 19 | + | const maxPendingEchoes = 256 | |
| 20 | + | ||
| 21 | + | type pendingEcho struct { | |
| 22 | + | seq uint64 | |
| 23 | + | at time.Time | |
| 24 | + | } | |
| 25 | + | ||
| 26 | + | // echoAckTracker turns applied input sequence numbers into the cumulative | |
| 27 | + | // Output.Echo acknowledgement. echoAck is written only under mu. | |
| 28 | + | type echoAckTracker struct { | |
| 29 | + | mu sync.Mutex | |
| 30 | + | pending []pendingEcho | |
| 31 | + | timer pendingByteTimer | |
| 32 | + | } | |
| 33 | + | ||
| 34 | + | // noteInputApplied records that the input numbered seq reached the session and | |
| 35 | + | // schedules its acknowledgement. Sequence zero is unsequenced input. | |
| 36 | + | func (d *Daemon) noteInputApplied(ac *attachedClient, seq uint64) { | |
| 37 | + | if ac == nil || seq == 0 { | |
| 38 | + | return | |
| 39 | + | } | |
| 40 | + | t := &ac.echo | |
| 41 | + | t.mu.Lock() | |
| 42 | + | defer t.mu.Unlock() | |
| 43 | + | if seq <= ac.echoAck.Load() || len(t.pending) > 0 && seq <= t.pending[len(t.pending)-1].seq { | |
| 44 | + | return | |
| 45 | + | } | |
| 46 | + | if len(t.pending) == maxPendingEchoes { | |
| 47 | + | t.pending = append(t.pending[:0], t.pending[1:]...) | |
| 48 | + | } | |
| 49 | + | t.pending = append(t.pending, pendingEcho{seq: seq, at: d.clock.Now()}) | |
| 50 | + | if t.timer.timer == nil { | |
| 51 | + | d.armEchoAckLocked(ac, echoAckDelay) | |
| 52 | + | } | |
| 53 | + | } | |
| 54 | + | ||
| 55 | + | // armEchoAckLocked starts the single echo timer. The caller holds echo.mu. | |
| 56 | + | func (d *Daemon) armEchoAckLocked(ac *attachedClient, delay time.Duration) { | |
| 57 | + | t := &ac.echo | |
| 58 | + | t.timer.retain(d.clock, delay, func(fired ports.Timer) { | |
| 59 | + | if d.advanceEchoAck(ac, fired) { | |
| 60 | + | d.sendEchoAck(ac) | |
| 61 | + | } | |
| 62 | + | }) | |
| 63 | + | } | |
| 64 | + | ||
| 65 | + | // advanceEchoAck moves echoAck to the newest input applied at least | |
| 66 | + | // echoAckDelay ago and re-arms for the rest. It reports whether echoAck moved. | |
| 67 | + | func (d *Daemon) advanceEchoAck(ac *attachedClient, fired ports.Timer) bool { | |
| 68 | + | t := &ac.echo | |
| 69 | + | t.mu.Lock() | |
| 70 | + | defer t.mu.Unlock() | |
| 71 | + | if t.timer.timer != fired { | |
| 72 | + | return false // stopped by a reset or superseded | |
| 73 | + | } | |
| 74 | + | t.timer.timer, t.timer.done = nil, nil | |
| 75 | + | now := d.clock.Now() | |
| 76 | + | acked := uint64(0) | |
| 77 | + | n := 0 | |
| 78 | + | for n < len(t.pending) && now.Sub(t.pending[n].at) >= echoAckDelay { | |
| 79 | + | acked = t.pending[n].seq | |
| 80 | + | n++ | |
| 81 | + | } | |
| 82 | + | t.pending = append(t.pending[:0], t.pending[n:]...) | |
| 83 | + | if len(t.pending) > 0 { | |
| 84 | + | d.armEchoAckLocked(ac, max(echoAckDelay-now.Sub(t.pending[0].at), 0)) | |
| 85 | + | } | |
| 86 | + | if acked <= ac.echoAck.Load() { | |
| 87 | + | return false | |
| 88 | + | } | |
| 89 | + | ac.echoAck.Store(acked) | |
| 90 | + | return true | |
| 91 | + | } | |
| 92 | + | ||
| 93 | + | // sendEchoAck publishes a new acknowledgement in an empty side-effect Output, | |
| 94 | + | // so the client learns about it even when the screen does not change. | |
| 95 | + | func (d *Daemon) sendEchoAck(ac *attachedClient) { | |
| 96 | + | if ac.attachmentActivity() != attachmentActive { | |
| 97 | + | return | |
| 98 | + | } | |
| 99 | + | sess := ac.currentAttachmentSession() | |
| 100 | + | if sess == nil { | |
| 101 | + | return | |
| 102 | + | } | |
| 103 | + | failed, err := d.boundedSendOutputErrTransport(ac, nil) | |
| 104 | + | if err != nil && failed != nil { | |
| 105 | + | d.detachOnSendError(sess, ac, failed) | |
| 106 | + | } | |
| 107 | + | } | |
| 108 | + | ||
| 109 | + | // resetEchoAck forgets every acknowledgement. A resumed client restarts its | |
| 110 | + | // input numbering, so the replacement link must start from zero too. | |
| 111 | + | func (ac *attachedClient) resetEchoAck() { | |
| 112 | + | t := &ac.echo | |
| 113 | + | t.mu.Lock() | |
| 114 | + | defer t.mu.Unlock() | |
| 115 | + | t.timer.stop() | |
| 116 | + | t.pending = nil | |
| 117 | + | ac.echoAck.Store(0) | |
| 118 | + | } | |
| @@ -0,0 +1,118 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "sync" | |||
| 5 | + | "time" | |||
| 6 | + | ||||
| 7 | + | "github.com/bnema/vev/internal/ports" | |||
| 8 | + | ) | |||
| 9 | + | ||||
| 10 | + | // echoAckDelay is how long sequenced input must have been applied before the | |||
| 11 | + | // daemon acknowledges it in Output.Echo. Like mosh's ECHO_TIMEOUT, it gives the | |||
| 12 | + | // PTY time to echo the input, so the screen published with the acknowledgement | |||
| 13 | + | // already contains whatever echo the input produced. | |||
| 14 | + | const echoAckDelay = 50 * time.Millisecond | |||
| 15 | + | ||||
| 16 | + | // maxPendingEchoes bounds the per-attachment input history. Acknowledgements | |||
| 17 | + | // are cumulative, so dropping the oldest entry only delays its confirmation to | |||
| 18 | + | // a newer one. | |||
| 19 | + | const maxPendingEchoes = 256 | |||
| 20 | + | ||||
| 21 | + | type pendingEcho struct { | |||
| 22 | + | seq uint64 | |||
| 23 | + | at time.Time | |||
| 24 | + | } | |||
| 25 | + | ||||
| 26 | + | // echoAckTracker turns applied input sequence numbers into the cumulative | |||
| 27 | + | // Output.Echo acknowledgement. echoAck is written only under mu. | |||
| 28 | + | type echoAckTracker struct { | |||
| 29 | + | mu sync.Mutex | |||
| 30 | + | pending []pendingEcho | |||
| 31 | + | timer pendingByteTimer | |||
| 32 | + | } | |||
| 33 | + | ||||
| 34 | + | // noteInputApplied records that the input numbered seq reached the session and | |||
| 35 | + | // schedules its acknowledgement. Sequence zero is unsequenced input. | |||
| 36 | + | func (d *Daemon) noteInputApplied(ac *attachedClient, seq uint64) { | |||
| 37 | + | if ac == nil || seq == 0 { | |||
| 38 | + | return | |||
| 39 | + | } | |||
| 40 | + | t := &ac.echo | |||
| 41 | + | t.mu.Lock() | |||
| 42 | + | defer t.mu.Unlock() | |||
| 43 | + | if seq <= ac.echoAck.Load() || len(t.pending) > 0 && seq <= t.pending[len(t.pending)-1].seq { | |||
| 44 | + | return | |||
| 45 | + | } | |||
| 46 | + | if len(t.pending) == maxPendingEchoes { | |||
| 47 | + | t.pending = append(t.pending[:0], t.pending[1:]...) | |||
| 48 | + | } | |||
| 49 | + | t.pending = append(t.pending, pendingEcho{seq: seq, at: d.clock.Now()}) | |||
| 50 | + | if t.timer.timer == nil { | |||
| 51 | + | d.armEchoAckLocked(ac, echoAckDelay) | |||
| 52 | + | } | |||
| 53 | + | } | |||
| 54 | + | ||||
| 55 | + | // armEchoAckLocked starts the single echo timer. The caller holds echo.mu. | |||
| 56 | + | func (d *Daemon) armEchoAckLocked(ac *attachedClient, delay time.Duration) { | |||
| 57 | + | t := &ac.echo | |||
| 58 | + | t.timer.retain(d.clock, delay, func(fired ports.Timer) { | |||
| 59 | + | if d.advanceEchoAck(ac, fired) { | |||
| 60 | + | d.sendEchoAck(ac) | |||
| 61 | + | } | |||
| 62 | + | }) | |||
| 63 | + | } | |||
| 64 | + | ||||
| 65 | + | // advanceEchoAck moves echoAck to the newest input applied at least | |||
| 66 | + | // echoAckDelay ago and re-arms for the rest. It reports whether echoAck moved. | |||
| 67 | + | func (d *Daemon) advanceEchoAck(ac *attachedClient, fired ports.Timer) bool { | |||
| 68 | + | t := &ac.echo | |||
| 69 | + | t.mu.Lock() | |||
| 70 | + | defer t.mu.Unlock() | |||
| 71 | + | if t.timer.timer != fired { | |||
| 72 | + | return false // stopped by a reset or superseded | |||
| 73 | + | } | |||
| 74 | + | t.timer.timer, t.timer.done = nil, nil | |||
| 75 | + | now := d.clock.Now() | |||
| 76 | + | acked := uint64(0) | |||
| 77 | + | n := 0 | |||
| 78 | + | for n < len(t.pending) && now.Sub(t.pending[n].at) >= echoAckDelay { | |||
| 79 | + | acked = t.pending[n].seq | |||
| 80 | + | n++ | |||
| 81 | + | } | |||
| 82 | + | t.pending = append(t.pending[:0], t.pending[n:]...) | |||
| 83 | + | if len(t.pending) > 0 { | |||
| 84 | + | d.armEchoAckLocked(ac, max(echoAckDelay-now.Sub(t.pending[0].at), 0)) | |||
| 85 | + | } | |||
| 86 | + | if acked <= ac.echoAck.Load() { | |||
| 87 | + | return false | |||
| 88 | + | } | |||
| 89 | + | ac.echoAck.Store(acked) | |||
| 90 | + | return true | |||
| 91 | + | } | |||
| 92 | + | ||||
| 93 | + | // sendEchoAck publishes a new acknowledgement in an empty side-effect Output, | |||
| 94 | + | // so the client learns about it even when the screen does not change. | |||
| 95 | + | func (d *Daemon) sendEchoAck(ac *attachedClient) { | |||
| 96 | + | if ac.attachmentActivity() != attachmentActive { | |||
| 97 | + | return | |||
| 98 | + | } | |||
| 99 | + | sess := ac.currentAttachmentSession() | |||
| 100 | + | if sess == nil { | |||
| 101 | + | return | |||
| 102 | + | } | |||
| 103 | + | failed, err := d.boundedSendOutputErrTransport(ac, nil) | |||
| 104 | + | if err != nil && failed != nil { | |||
| 105 | + | d.detachOnSendError(sess, ac, failed) | |||
| 106 | + | } | |||
| 107 | + | } | |||
| 108 | + | ||||
| 109 | + | // resetEchoAck forgets every acknowledgement. A resumed client restarts its | |||
| 110 | + | // input numbering, so the replacement link must start from zero too. | |||
| 111 | + | func (ac *attachedClient) resetEchoAck() { | |||
| 112 | + | t := &ac.echo | |||
| 113 | + | t.mu.Lock() | |||
| 114 | + | defer t.mu.Unlock() | |||
| 115 | + | t.timer.stop() | |||
| 116 | + | t.pending = nil | |||
| 117 | + | ac.echoAck.Store(0) | |||
| 118 | + | } | |||
A internal/usecase/daemon/echo_ack_test.go +118 -0
| @@ -0,0 +1,118 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "sync" | |
| 5 | + | "testing" | |
| 6 | + | "time" | |
| 7 | + | ||
| 8 | + | "github.com/stretchr/testify/require" | |
| 9 | + | ||
| 10 | + | "github.com/bnema/vev/internal/ports" | |
| 11 | + | "github.com/bnema/vev/internal/protocol" | |
| 12 | + | ) | |
| 13 | + | ||
| 14 | + | // echoTestClock is a settable clock whose timers fire only when the test says. | |
| 15 | + | type echoTestClock struct { | |
| 16 | + | mu sync.Mutex | |
| 17 | + | now time.Time | |
| 18 | + | timers chan echoTestTimer | |
| 19 | + | } | |
| 20 | + | ||
| 21 | + | type echoTestTimer struct { | |
| 22 | + | timer *manualAttentionTimer | |
| 23 | + | delay time.Duration | |
| 24 | + | } | |
| 25 | + | ||
| 26 | + | func newEchoTestClock() *echoTestClock { | |
| 27 | + | return &echoTestClock{now: time.Unix(100, 0), timers: make(chan echoTestTimer, 16)} | |
| 28 | + | } | |
| 29 | + | ||
| 30 | + | func (c *echoTestClock) Now() time.Time { | |
| 31 | + | c.mu.Lock() | |
| 32 | + | defer c.mu.Unlock() | |
| 33 | + | return c.now | |
| 34 | + | } | |
| 35 | + | ||
| 36 | + | func (c *echoTestClock) advance(d time.Duration) { | |
| 37 | + | c.mu.Lock() | |
| 38 | + | c.now = c.now.Add(d) | |
| 39 | + | c.mu.Unlock() | |
| 40 | + | } | |
| 41 | + | ||
| 42 | + | func (c *echoTestClock) NewTimer(d time.Duration) ports.Timer { | |
| 43 | + | t := &manualAttentionTimer{ch: make(chan time.Time, 1)} | |
| 44 | + | c.timers <- echoTestTimer{timer: t, delay: d} | |
| 45 | + | return t | |
| 46 | + | } | |
| 47 | + | ||
| 48 | + | // nextTimer returns the next echo timer, skipping the send-bound timers the | |
| 49 | + | // ack-only output path creates. | |
| 50 | + | func (c *echoTestClock) nextTimer(t *testing.T) (*manualAttentionTimer, time.Duration) { | |
| 51 | + | t.Helper() | |
| 52 | + | deadline := time.After(time.Second) | |
| 53 | + | for { | |
| 54 | + | select { | |
| 55 | + | case next := <-c.timers: | |
| 56 | + | if next.delay <= echoAckDelay { | |
| 57 | + | return next.timer, next.delay | |
| 58 | + | } | |
| 59 | + | case <-deadline: | |
| 60 | + | t.Fatal("timed out waiting for echo timer") | |
| 61 | + | return nil, 0 | |
| 62 | + | } | |
| 63 | + | } | |
| 64 | + | } | |
| 65 | + | ||
| 66 | + | func TestEchoAckWaitsForEchoTimeoutAndSendsAckOnlyOutput(t *testing.T) { | |
| 67 | + | clock := newEchoTestClock() | |
| 68 | + | d, _, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY()) | |
| 69 | + | ||
| 70 | + | d.noteInputApplied(ac, 1) | |
| 71 | + | clock.advance(20 * time.Millisecond) | |
| 72 | + | d.noteInputApplied(ac, 2) | |
| 73 | + | timer, delay := clock.nextTimer(t) | |
| 74 | + | require.Equal(t, echoAckDelay, delay) | |
| 75 | + | require.Zero(t, ac.echoAck.Load(), "input is not echoed when applied") | |
| 76 | + | ||
| 77 | + | clock.advance(30 * time.Millisecond) | |
| 78 | + | timer.fire() | |
| 79 | + | output := decodeServerMessage(t, awaitFrame(t, sends, "Output")).(protocol.Output) | |
| 80 | + | require.Equal(t, uint64(1), output.Echo) | |
| 81 | + | require.Zero(t, output.New, "an ack-only output is a side effect") | |
| 82 | + | require.Empty(t, output.Data) | |
| 83 | + | ||
| 84 | + | // Input 2 is still younger than the timeout: the tracker re-arms for its | |
| 85 | + | // remaining age. | |
| 86 | + | timer, delay = clock.nextTimer(t) | |
| 87 | + | require.Equal(t, 20*time.Millisecond, delay) | |
| 88 | + | clock.advance(20 * time.Millisecond) | |
| 89 | + | timer.fire() | |
| 90 | + | output = decodeServerMessage(t, awaitFrame(t, sends, "Output")).(protocol.Output) | |
| 91 | + | require.Equal(t, uint64(2), output.Echo) | |
| 92 | + | require.Equal(t, uint64(2), ac.echoAck.Load()) | |
| 93 | + | } | |
| 94 | + | ||
| 95 | + | func TestEchoAckIgnoresUnsequencedAndStaleInput(t *testing.T) { | |
| 96 | + | clock := newEchoTestClock() | |
| 97 | + | d, _, ac, _ := newManualSessionWithPTYsClock(t, clock, newQuietPTY()) | |
| 98 | + | ac.echoAck.Store(5) | |
| 99 | + | ||
| 100 | + | d.noteInputApplied(ac, 0) | |
| 101 | + | d.noteInputApplied(ac, 5) | |
| 102 | + | require.Empty(t, ac.echo.pending) | |
| 103 | + | require.Empty(t, clock.timers) | |
| 104 | + | } | |
| 105 | + | ||
| 106 | + | func TestResetEchoAckDropsPendingInput(t *testing.T) { | |
| 107 | + | clock := newEchoTestClock() | |
| 108 | + | d, _, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY()) | |
| 109 | + | ||
| 110 | + | d.noteInputApplied(ac, 7) | |
| 111 | + | timer, _ := clock.nextTimer(t) | |
| 112 | + | ac.resetEchoAck() | |
| 113 | + | clock.advance(echoAckDelay) | |
| 114 | + | timer.fire() | |
| 115 | + | ||
| 116 | + | require.Zero(t, ac.echoAck.Load()) | |
| 117 | + | require.Empty(t, drainAllFrames(sends), "a reset tracker must not publish the old link's input") | |
| 118 | + | } | |
| @@ -0,0 +1,118 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "sync" | |||
| 5 | + | "testing" | |||
| 6 | + | "time" | |||
| 7 | + | ||||
| 8 | + | "github.com/stretchr/testify/require" | |||
| 9 | + | ||||
| 10 | + | "github.com/bnema/vev/internal/ports" | |||
| 11 | + | "github.com/bnema/vev/internal/protocol" | |||
| 12 | + | ) | |||
| 13 | + | ||||
| 14 | + | // echoTestClock is a settable clock whose timers fire only when the test says. | |||
| 15 | + | type echoTestClock struct { | |||
| 16 | + | mu sync.Mutex | |||
| 17 | + | now time.Time | |||
| 18 | + | timers chan echoTestTimer | |||
| 19 | + | } | |||
| 20 | + | ||||
| 21 | + | type echoTestTimer struct { | |||
| 22 | + | timer *manualAttentionTimer | |||
| 23 | + | delay time.Duration | |||
| 24 | + | } | |||
| 25 | + | ||||
| 26 | + | func newEchoTestClock() *echoTestClock { | |||
| 27 | + | return &echoTestClock{now: time.Unix(100, 0), timers: make(chan echoTestTimer, 16)} | |||
| 28 | + | } | |||
| 29 | + | ||||
| 30 | + | func (c *echoTestClock) Now() time.Time { | |||
| 31 | + | c.mu.Lock() | |||
| 32 | + | defer c.mu.Unlock() | |||
| 33 | + | return c.now | |||
| 34 | + | } | |||
| 35 | + | ||||
| 36 | + | func (c *echoTestClock) advance(d time.Duration) { | |||
| 37 | + | c.mu.Lock() | |||
| 38 | + | c.now = c.now.Add(d) | |||
| 39 | + | c.mu.Unlock() | |||
| 40 | + | } | |||
| 41 | + | ||||
| 42 | + | func (c *echoTestClock) NewTimer(d time.Duration) ports.Timer { | |||
| 43 | + | t := &manualAttentionTimer{ch: make(chan time.Time, 1)} | |||
| 44 | + | c.timers <- echoTestTimer{timer: t, delay: d} | |||
| 45 | + | return t | |||
| 46 | + | } | |||
| 47 | + | ||||
| 48 | + | // nextTimer returns the next echo timer, skipping the send-bound timers the | |||
| 49 | + | // ack-only output path creates. | |||
| 50 | + | func (c *echoTestClock) nextTimer(t *testing.T) (*manualAttentionTimer, time.Duration) { | |||
| 51 | + | t.Helper() | |||
| 52 | + | deadline := time.After(time.Second) | |||
| 53 | + | for { | |||
| 54 | + | select { | |||
| 55 | + | case next := <-c.timers: | |||
| 56 | + | if next.delay <= echoAckDelay { | |||
| 57 | + | return next.timer, next.delay | |||
| 58 | + | } | |||
| 59 | + | case <-deadline: | |||
| 60 | + | t.Fatal("timed out waiting for echo timer") | |||
| 61 | + | return nil, 0 | |||
| 62 | + | } | |||
| 63 | + | } | |||
| 64 | + | } | |||
| 65 | + | ||||
| 66 | + | func TestEchoAckWaitsForEchoTimeoutAndSendsAckOnlyOutput(t *testing.T) { | |||
| 67 | + | clock := newEchoTestClock() | |||
| 68 | + | d, _, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY()) | |||
| 69 | + | ||||
| 70 | + | d.noteInputApplied(ac, 1) | |||
| 71 | + | clock.advance(20 * time.Millisecond) | |||
| 72 | + | d.noteInputApplied(ac, 2) | |||
| 73 | + | timer, delay := clock.nextTimer(t) | |||
| 74 | + | require.Equal(t, echoAckDelay, delay) | |||
| 75 | + | require.Zero(t, ac.echoAck.Load(), "input is not echoed when applied") | |||
| 76 | + | ||||
| 77 | + | clock.advance(30 * time.Millisecond) | |||
| 78 | + | timer.fire() | |||
| 79 | + | output := decodeServerMessage(t, awaitFrame(t, sends, "Output")).(protocol.Output) | |||
| 80 | + | require.Equal(t, uint64(1), output.Echo) | |||
| 81 | + | require.Zero(t, output.New, "an ack-only output is a side effect") | |||
| 82 | + | require.Empty(t, output.Data) | |||
| 83 | + | ||||
| 84 | + | // Input 2 is still younger than the timeout: the tracker re-arms for its | |||
| 85 | + | // remaining age. | |||
| 86 | + | timer, delay = clock.nextTimer(t) | |||
| 87 | + | require.Equal(t, 20*time.Millisecond, delay) | |||
| 88 | + | clock.advance(20 * time.Millisecond) | |||
| 89 | + | timer.fire() | |||
| 90 | + | output = decodeServerMessage(t, awaitFrame(t, sends, "Output")).(protocol.Output) | |||
| 91 | + | require.Equal(t, uint64(2), output.Echo) | |||
| 92 | + | require.Equal(t, uint64(2), ac.echoAck.Load()) | |||
| 93 | + | } | |||
| 94 | + | ||||
| 95 | + | func TestEchoAckIgnoresUnsequencedAndStaleInput(t *testing.T) { | |||
| 96 | + | clock := newEchoTestClock() | |||
| 97 | + | d, _, ac, _ := newManualSessionWithPTYsClock(t, clock, newQuietPTY()) | |||
| 98 | + | ac.echoAck.Store(5) | |||
| 99 | + | ||||
| 100 | + | d.noteInputApplied(ac, 0) | |||
| 101 | + | d.noteInputApplied(ac, 5) | |||
| 102 | + | require.Empty(t, ac.echo.pending) | |||
| 103 | + | require.Empty(t, clock.timers) | |||
| 104 | + | } | |||
| 105 | + | ||||
| 106 | + | func TestResetEchoAckDropsPendingInput(t *testing.T) { | |||
| 107 | + | clock := newEchoTestClock() | |||
| 108 | + | d, _, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY()) | |||
| 109 | + | ||||
| 110 | + | d.noteInputApplied(ac, 7) | |||
| 111 | + | timer, _ := clock.nextTimer(t) | |||
| 112 | + | ac.resetEchoAck() | |||
| 113 | + | clock.advance(echoAckDelay) | |||
| 114 | + | timer.fire() | |||
| 115 | + | ||||
| 116 | + | require.Zero(t, ac.echoAck.Load()) | |||
| 117 | + | require.Empty(t, drainAllFrames(sends), "a reset tracker must not publish the old link's input") | |||
| 118 | + | } | |||
M internal/usecase/daemon/input.go +7 -6
| @@ -14,19 +14,20 @@ import ( | |||
| 14 | 14 | "github.com/bnema/vev/internal/usecase/mouse" | |
| 15 | 15 | ) | |
| 16 | 16 | ||
| 17 | - | func (d *Daemon) handleSequencedInput(sess *session, ac *attachedClient, _ uint64, data []byte) { | |
| 18 | - | // Do not acknowledge client-side echo prediction here: input has only been | |
| 19 | - | // accepted/routed, not necessarily echoed by the PTY and incorporated into a | |
| 20 | - | // rendered screen state. Until prediction is implemented against rendered | |
| 21 | - | // output state, Echo must remain conservative. | |
| 17 | + | // handleSequencedInput applies input and schedules its echo acknowledgement. | |
| 18 | + | // Echo is not acknowledged immediately: the PTY needs time to echo the input | |
| 19 | + | // into a rendered screen state (see echoAckDelay). | |
| 20 | + | func (d *Daemon) handleSequencedInput(sess *session, ac *attachedClient, seq uint64, data []byte) { | |
| 22 | 21 | d.handleInput(sess, ac, data) | |
| 22 | + | d.noteInputApplied(ac, seq) | |
| 23 | 23 | } | |
| 24 | 24 | ||
| 25 | - | func (d *Daemon) handleSequencedInputForAttachment(effect *attachmentEffect, _ uint64, data []byte) { | |
| 25 | + | func (d *Daemon) handleSequencedInputForAttachment(effect *attachmentEffect, seq uint64, data []byte) { | |
| 26 | 26 | if !effect.current() { | |
| 27 | 27 | return | |
| 28 | 28 | } | |
| 29 | 29 | d.handleInputForAttachment(effect, data) | |
| 30 | + | d.noteInputApplied(effect.ac, seq) | |
| 30 | 31 | } | |
| 31 | 32 | ||
| 32 | 33 | func (d *Daemon) handleInput(_ *session, ac *attachedClient, data []byte) { | |
| @@ -14,19 +14,20 @@ import ( | |||||
| 14 | "github.com/bnema/vev/internal/usecase/mouse" | 14 | "github.com/bnema/vev/internal/usecase/mouse" | ||
| 15 | ) | 15 | ) | ||
| 16 | 16 | ||||
| 17 | - | func (d *Daemon) handleSequencedInput(sess *session, ac *attachedClient, _ uint64, data []byte) { | 17 | + | // handleSequencedInput applies input and schedules its echo acknowledgement. |
| 18 | - | // Do not acknowledge client-side echo prediction here: input has only been | 18 | + | // Echo is not acknowledged immediately: the PTY needs time to echo the input |
| 19 | - | // accepted/routed, not necessarily echoed by the PTY and incorporated into a | 19 | + | // into a rendered screen state (see echoAckDelay). |
| 20 | - | // rendered screen state. Until prediction is implemented against rendered | 20 | + | func (d *Daemon) handleSequencedInput(sess *session, ac *attachedClient, seq uint64, data []byte) { |
| 21 | - | // output state, Echo must remain conservative. | |||
| 22 | d.handleInput(sess, ac, data) | 21 | d.handleInput(sess, ac, data) | ||
| 22 | + | d.noteInputApplied(ac, seq) | |||
| 23 | } | 23 | } | ||
| 24 | 24 | ||||
| 25 | - | func (d *Daemon) handleSequencedInputForAttachment(effect *attachmentEffect, _ uint64, data []byte) { | 25 | + | func (d *Daemon) handleSequencedInputForAttachment(effect *attachmentEffect, seq uint64, data []byte) { |
| 26 | if !effect.current() { | 26 | if !effect.current() { | ||
| 27 | return | 27 | return | ||
| 28 | } | 28 | } | ||
| 29 | d.handleInputForAttachment(effect, data) | 29 | d.handleInputForAttachment(effect, data) | ||
| 30 | + | d.noteInputApplied(effect.ac, seq) | |||
| 30 | } | 31 | } | ||
| 31 | 32 | ||||
| 32 | func (d *Daemon) handleInput(_ *session, ac *attachedClient, data []byte) { | 33 | func (d *Daemon) handleInput(_ *session, ac *attachedClient, data []byte) { | ||
M internal/usecase/daemon/resume.go +1 -0
| @@ -627,6 +627,7 @@ func (d *Daemon) resumeParkedLocked(h protocol.Hello, tr ports.ServerConnection, | |||
| 627 | 627 | // abandoned transport's output chain before binding the replacement so the | |
| 628 | 628 | // mandatory first paint cannot be blocked by ACKs that died with the link. | |
| 629 | 629 | ac.rebaseOutput() | |
| 630 | + | ac.resetEchoAck() | |
| 630 | 631 | ac.output.setWindow(h.MaxOutputInFlight) | |
| 631 | 632 | ac.replaceTransport(tr) | |
| 632 | 633 | // A parked attachment can resume through any route. Reapply the declared | |
| @@ -627,6 +627,7 @@ func (d *Daemon) resumeParkedLocked(h protocol.Hello, tr ports.ServerConnection, | |||||
| 627 | // abandoned transport's output chain before binding the replacement so the | 627 | // abandoned transport's output chain before binding the replacement so the | ||
| 628 | // mandatory first paint cannot be blocked by ACKs that died with the link. | 628 | // mandatory first paint cannot be blocked by ACKs that died with the link. | ||
| 629 | ac.rebaseOutput() | 629 | ac.rebaseOutput() | ||
| 630 | + | ac.resetEchoAck() | |||
| 630 | ac.output.setWindow(h.MaxOutputInFlight) | 631 | ac.output.setWindow(h.MaxOutputInFlight) | ||
| 631 | ac.replaceTransport(tr) | 632 | ac.replaceTransport(tr) | ||
| 632 | // A parked attachment can resume through any route. Reapply the declared | 633 | // A parked attachment can resume through any route. Reapply the declared | ||