fix(daemon): defer echo acks while the output window is full
An ack-only side effect bypasses the output window, so it could reach the client before the withheld frame holding the echo. The ack now waits for the next frame or a retry once the window has room.
- commit
- a686aba440f17f22fdf1b8cbf97ffba519d32bc5
- author
- brice · · 2026-10-07 13:38 UTC
- parent
- 4491ca1
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +61 -0
M internal/usecase/daemon/echo_ack.go +37 -0
| @@ -62,6 +62,28 @@ func (d *Daemon) armEchoAckLocked(ac *attachedClient, delay time.Duration) { | |||
| 62 | 62 | }) | |
| 63 | 63 | } | |
| 64 | 64 | ||
| 65 | + | // armEchoAckRetryLocked re-sends the current acknowledgement after one more | |
| 66 | + | // echo delay. The caller holds echo.mu. | |
| 67 | + | func (d *Daemon) armEchoAckRetryLocked(ac *attachedClient) { | |
| 68 | + | t := &ac.echo | |
| 69 | + | t.timer.retain(d.clock, echoAckDelay, func(fired ports.Timer) { | |
| 70 | + | t.mu.Lock() | |
| 71 | + | current := t.timer.timer == fired | |
| 72 | + | if current { | |
| 73 | + | t.timer.timer, t.timer.done = nil, nil | |
| 74 | + | if len(t.pending) > 0 { | |
| 75 | + | // New input owns the next timer and its ack covers this one. | |
| 76 | + | d.armEchoAckLocked(ac, max(echoAckDelay-d.clock.Now().Sub(t.pending[0].at), 0)) | |
| 77 | + | current = false | |
| 78 | + | } | |
| 79 | + | } | |
| 80 | + | t.mu.Unlock() | |
| 81 | + | if current { | |
| 82 | + | d.sendEchoAck(ac) | |
| 83 | + | } | |
| 84 | + | }) | |
| 85 | + | } | |
| 86 | + | ||
| 65 | 87 | // advanceEchoAck moves echoAck to the newest input applied at least | |
| 66 | 88 | // echoAckDelay ago and re-arms for the rest. It reports whether echoAck moved. | |
| 67 | 89 | func (d *Daemon) advanceEchoAck(ac *attachedClient, fired ports.Timer) bool { | |
| @@ -92,6 +114,12 @@ func (d *Daemon) advanceEchoAck(ac *attachedClient, fired ports.Timer) bool { | |||
| 92 | 114 | ||
| 93 | 115 | // sendEchoAck publishes a new acknowledgement in an empty side-effect Output, | |
| 94 | 116 | // so the client learns about it even when the screen does not change. | |
| 117 | + | // | |
| 118 | + | // Side effects bypass the output window. While the window is full, the frame | |
| 119 | + | // holding the acknowledged echo may still be withheld, and an early ack would | |
| 120 | + | // make the client judge its guesses against an older screen. The ack then | |
| 121 | + | // waits: the next frame carries it, and a retry covers a screen that never | |
| 122 | + | // changes. | |
| 95 | 123 | func (d *Daemon) sendEchoAck(ac *attachedClient) { | |
| 96 | 124 | if ac.attachmentActivity() != attachmentActive { | |
| 97 | 125 | return | |
| @@ -100,6 +128,15 @@ func (d *Daemon) sendEchoAck(ac *attachedClient) { | |||
| 100 | 128 | if sess == nil { | |
| 101 | 129 | return | |
| 102 | 130 | } | |
| 131 | + | if ac.output != nil && ac.output.atCapacity() { | |
| 132 | + | t := &ac.echo | |
| 133 | + | t.mu.Lock() | |
| 134 | + | if t.timer.timer == nil { | |
| 135 | + | d.armEchoAckRetryLocked(ac) | |
| 136 | + | } | |
| 137 | + | t.mu.Unlock() | |
| 138 | + | return | |
| 139 | + | } | |
| 103 | 140 | failed, err := d.boundedSendOutputErrTransport(ac, nil) | |
| 104 | 141 | if err != nil && failed != nil { | |
| 105 | 142 | d.detachOnSendError(sess, ac, failed) | |
| @@ -62,6 +62,28 @@ func (d *Daemon) armEchoAckLocked(ac *attachedClient, delay time.Duration) { | |||||
| 62 | }) | 62 | }) | ||
| 63 | } | 63 | } | ||
| 64 | 64 | ||||
| 65 | + | // armEchoAckRetryLocked re-sends the current acknowledgement after one more | |||
| 66 | + | // echo delay. The caller holds echo.mu. | |||
| 67 | + | func (d *Daemon) armEchoAckRetryLocked(ac *attachedClient) { | |||
| 68 | + | t := &ac.echo | |||
| 69 | + | t.timer.retain(d.clock, echoAckDelay, func(fired ports.Timer) { | |||
| 70 | + | t.mu.Lock() | |||
| 71 | + | current := t.timer.timer == fired | |||
| 72 | + | if current { | |||
| 73 | + | t.timer.timer, t.timer.done = nil, nil | |||
| 74 | + | if len(t.pending) > 0 { | |||
| 75 | + | // New input owns the next timer and its ack covers this one. | |||
| 76 | + | d.armEchoAckLocked(ac, max(echoAckDelay-d.clock.Now().Sub(t.pending[0].at), 0)) | |||
| 77 | + | current = false | |||
| 78 | + | } | |||
| 79 | + | } | |||
| 80 | + | t.mu.Unlock() | |||
| 81 | + | if current { | |||
| 82 | + | d.sendEchoAck(ac) | |||
| 83 | + | } | |||
| 84 | + | }) | |||
| 85 | + | } | |||
| 86 | + | ||||
| 65 | // advanceEchoAck moves echoAck to the newest input applied at least | 87 | // advanceEchoAck moves echoAck to the newest input applied at least | ||
| 66 | // echoAckDelay ago and re-arms for the rest. It reports whether echoAck moved. | 88 | // echoAckDelay ago and re-arms for the rest. It reports whether echoAck moved. | ||
| 67 | func (d *Daemon) advanceEchoAck(ac *attachedClient, fired ports.Timer) bool { | 89 | func (d *Daemon) advanceEchoAck(ac *attachedClient, fired ports.Timer) bool { | ||
| @@ -92,6 +114,12 @@ func (d *Daemon) advanceEchoAck(ac *attachedClient, fired ports.Timer) bool { | |||||
| 92 | 114 | ||||
| 93 | // sendEchoAck publishes a new acknowledgement in an empty side-effect Output, | 115 | // 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. | 116 | // so the client learns about it even when the screen does not change. | ||
| 117 | + | // | |||
| 118 | + | // Side effects bypass the output window. While the window is full, the frame | |||
| 119 | + | // holding the acknowledged echo may still be withheld, and an early ack would | |||
| 120 | + | // make the client judge its guesses against an older screen. The ack then | |||
| 121 | + | // waits: the next frame carries it, and a retry covers a screen that never | |||
| 122 | + | // changes. | |||
| 95 | func (d *Daemon) sendEchoAck(ac *attachedClient) { | 123 | func (d *Daemon) sendEchoAck(ac *attachedClient) { | ||
| 96 | if ac.attachmentActivity() != attachmentActive { | 124 | if ac.attachmentActivity() != attachmentActive { | ||
| 97 | return | 125 | return | ||
| @@ -100,6 +128,15 @@ func (d *Daemon) sendEchoAck(ac *attachedClient) { | |||||
| 100 | if sess == nil { | 128 | if sess == nil { | ||
| 101 | return | 129 | return | ||
| 102 | } | 130 | } | ||
| 131 | + | if ac.output != nil && ac.output.atCapacity() { | |||
| 132 | + | t := &ac.echo | |||
| 133 | + | t.mu.Lock() | |||
| 134 | + | if t.timer.timer == nil { | |||
| 135 | + | d.armEchoAckRetryLocked(ac) | |||
| 136 | + | } | |||
| 137 | + | t.mu.Unlock() | |||
| 138 | + | return | |||
| 139 | + | } | |||
| 103 | failed, err := d.boundedSendOutputErrTransport(ac, nil) | 140 | failed, err := d.boundedSendOutputErrTransport(ac, nil) | ||
| 104 | if err != nil && failed != nil { | 141 | if err != nil && failed != nil { | ||
| 105 | d.detachOnSendError(sess, ac, failed) | 142 | d.detachOnSendError(sess, ac, failed) | ||
M internal/usecase/daemon/echo_ack_test.go +24 -0
| @@ -116,3 +116,27 @@ func TestResetEchoAckDropsPendingInput(t *testing.T) { | |||
| 116 | 116 | require.Zero(t, ac.echoAck.Load()) | |
| 117 | 117 | require.Empty(t, drainAllFrames(sends), "a reset tracker must not publish the old link's input") | |
| 118 | 118 | } | |
| 119 | + | ||
| 120 | + | func TestEchoAckWaitsWhileOutputWindowIsFull(t *testing.T) { | |
| 121 | + | clock := newEchoTestClock() | |
| 122 | + | d, _, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY()) | |
| 123 | + | ac.output.maxOutstandingAtomic.Store(1) | |
| 124 | + | ac.output.outstandingAtomic.Store(1) | |
| 125 | + | ||
| 126 | + | d.noteInputApplied(ac, 1) | |
| 127 | + | timer, _ := clock.nextTimer(t) | |
| 128 | + | clock.advance(echoAckDelay) | |
| 129 | + | timer.fire() | |
| 130 | + | ||
| 131 | + | retry, delay := clock.nextTimer(t) | |
| 132 | + | require.Equal(t, echoAckDelay, delay, "a full window defers the ack") | |
| 133 | + | require.Equal(t, uint64(1), ac.echoAck.Load(), "the next frame may still carry it") | |
| 134 | + | require.Empty(t, drainAllFrames(sends), "no ack-only output may overtake withheld frames") | |
| 135 | + | ||
| 136 | + | ac.output.outstandingAtomic.Store(0) | |
| 137 | + | clock.advance(echoAckDelay) | |
| 138 | + | retry.fire() | |
| 139 | + | output := decodeServerMessage(t, awaitFrame(t, sends, "Output")).(protocol.Output) | |
| 140 | + | require.Equal(t, uint64(1), output.Echo) | |
| 141 | + | require.Zero(t, output.New) | |
| 142 | + | } | |
| @@ -116,3 +116,27 @@ func TestResetEchoAckDropsPendingInput(t *testing.T) { | |||||
| 116 | require.Zero(t, ac.echoAck.Load()) | 116 | require.Zero(t, ac.echoAck.Load()) | ||
| 117 | require.Empty(t, drainAllFrames(sends), "a reset tracker must not publish the old link's input") | 117 | require.Empty(t, drainAllFrames(sends), "a reset tracker must not publish the old link's input") | ||
| 118 | } | 118 | } | ||
| 119 | + | ||||
| 120 | + | func TestEchoAckWaitsWhileOutputWindowIsFull(t *testing.T) { | |||
| 121 | + | clock := newEchoTestClock() | |||
| 122 | + | d, _, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY()) | |||
| 123 | + | ac.output.maxOutstandingAtomic.Store(1) | |||
| 124 | + | ac.output.outstandingAtomic.Store(1) | |||
| 125 | + | ||||
| 126 | + | d.noteInputApplied(ac, 1) | |||
| 127 | + | timer, _ := clock.nextTimer(t) | |||
| 128 | + | clock.advance(echoAckDelay) | |||
| 129 | + | timer.fire() | |||
| 130 | + | ||||
| 131 | + | retry, delay := clock.nextTimer(t) | |||
| 132 | + | require.Equal(t, echoAckDelay, delay, "a full window defers the ack") | |||
| 133 | + | require.Equal(t, uint64(1), ac.echoAck.Load(), "the next frame may still carry it") | |||
| 134 | + | require.Empty(t, drainAllFrames(sends), "no ack-only output may overtake withheld frames") | |||
| 135 | + | ||||
| 136 | + | ac.output.outstandingAtomic.Store(0) | |||
| 137 | + | clock.advance(echoAckDelay) | |||
| 138 | + | retry.fire() | |||
| 139 | + | output := decodeServerMessage(t, awaitFrame(t, sends, "Output")).(protocol.Output) | |||
| 140 | + | require.Equal(t, uint64(1), output.Echo) | |||
| 141 | + | require.Zero(t, output.New) | |||
| 142 | + | } | |||