fix(daemon): gate paints on output acks
Add maxUnackedOutputStates: when the client is that many output states behind, paint defers the diff instead of composing another, marking a per-client paintDeferred flag under sendMu. The MsgOutputAck handler flushes one cumulative paint once acks catch up. Reset paints (explicit invalidations) bypass the gate and clear the flag; park clears it too. This paces the daemon's paint rate to the client's ack rate so heavy output degrades to lower fps on a slow link instead of overflowing the UDP proxy window and tripping the 1s send watchdog.
- commit
- 7e0e2785b94475e38b9745d8bd1b970265d7767c
- author
- brice · · 2026-07-08 16:52 UTC
- parent
- 5726ece
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
5 files · +133 -0
M internal/usecase/daemon/client.go +32 -0
| @@ -49,6 +49,7 @@ type attachedClient struct { | |||
| 49 | 49 | nextStateNum uint64 | |
| 50 | 50 | echoAck atomic.Uint64 | |
| 51 | 51 | outputAck atomic.Uint64 | |
| 52 | + | paintDeferred bool // guarded by sendMu; a diff paint skipped while acks lag | |
| 52 | 53 | bars barCache // only touched while sendMu is held | |
| 53 | 54 | size domain.Size | |
| 54 | 55 | keys *keys.Router | |
| @@ -225,6 +226,34 @@ func (ac *attachedClient) advanceOutputAck(state uint64) { | |||
| 225 | 226 | } | |
| 226 | 227 | } | |
| 227 | 228 | ||
| 229 | + | // outputLagAtCapLocked reports whether the client has at least | |
| 230 | + | // maxUnackedOutputStates output states in flight (sent but unacked). Caller | |
| 231 | + | // holds sendMu. | |
| 232 | + | func (ac *attachedClient) outputLagAtCapLocked() bool { | |
| 233 | + | return ac.nextStateNum >= ac.outputAck.Load()+maxUnackedOutputStates | |
| 234 | + | } | |
| 235 | + | ||
| 236 | + | // takeDeferredPaint reports whether a deferred diff paint is now flushable | |
| 237 | + | // because acks have caught up, clearing the flag when it is. The caller must | |
| 238 | + | // then paint outside sendMu (paint re-acquires it). | |
| 239 | + | func (ac *attachedClient) takeDeferredPaint() bool { | |
| 240 | + | ac.sendMu.Lock() | |
| 241 | + | defer ac.sendMu.Unlock() | |
| 242 | + | if ac.paintDeferred && !ac.outputLagAtCapLocked() { | |
| 243 | + | ac.paintDeferred = false | |
| 244 | + | return true | |
| 245 | + | } | |
| 246 | + | return false | |
| 247 | + | } | |
| 248 | + | ||
| 249 | + | // clearPaintDeferred drops any pending deferred-paint flag. Used on park so a | |
| 250 | + | // resumed client does not inherit a stale deferral. | |
| 251 | + | func (ac *attachedClient) clearPaintDeferred() { | |
| 252 | + | ac.sendMu.Lock() | |
| 253 | + | ac.paintDeferred = false | |
| 254 | + | ac.sendMu.Unlock() | |
| 255 | + | } | |
| 256 | + | ||
| 228 | 257 | // send serialises a frame onto the client's transport. | |
| 229 | 258 | func (ac *attachedClient) send(f ports.Frame) error { | |
| 230 | 259 | _, err := ac.sendTransport(f) | |
| @@ -492,6 +521,9 @@ func (d *Daemon) runConnLoop(ac *attachedClient) { | |||
| 492 | 521 | case ports.MsgAck: | |
| 493 | 522 | if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil { | |
| 494 | 523 | ac.advanceOutputAck(ack.AckedStateNum) | |
| 524 | + | if ac.takeDeferredPaint() { | |
| 525 | + | d.paint(sess, ac, false) | |
| 526 | + | } | |
| 495 | 527 | } | |
| 496 | 528 | case ports.MsgPing: | |
| 497 | 529 | _ = ac.send(framePong()) | |
| @@ -49,6 +49,7 @@ type attachedClient struct { | |||||
| 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 | bars barCache // only touched while sendMu is held | 53 | bars barCache // only touched while sendMu is held | ||
| 53 | size domain.Size | 54 | size domain.Size | ||
| 54 | keys *keys.Router | 55 | keys *keys.Router | ||
| @@ -225,6 +226,34 @@ func (ac *attachedClient) advanceOutputAck(state uint64) { | |||||
| 225 | } | 226 | } | ||
| 226 | } | 227 | } | ||
| 227 | 228 | ||||
| 229 | + | // outputLagAtCapLocked reports whether the client has at least | |||
| 230 | + | // maxUnackedOutputStates output states in flight (sent but unacked). Caller | |||
| 231 | + | // holds sendMu. | |||
| 232 | + | func (ac *attachedClient) outputLagAtCapLocked() bool { | |||
| 233 | + | return ac.nextStateNum >= ac.outputAck.Load()+maxUnackedOutputStates | |||
| 234 | + | } | |||
| 235 | + | ||||
| 236 | + | // takeDeferredPaint reports whether a deferred diff paint is now flushable | |||
| 237 | + | // because acks have caught up, clearing the flag when it is. The caller must | |||
| 238 | + | // then paint outside sendMu (paint re-acquires it). | |||
| 239 | + | func (ac *attachedClient) takeDeferredPaint() bool { | |||
| 240 | + | ac.sendMu.Lock() | |||
| 241 | + | defer ac.sendMu.Unlock() | |||
| 242 | + | if ac.paintDeferred && !ac.outputLagAtCapLocked() { | |||
| 243 | + | ac.paintDeferred = false | |||
| 244 | + | return true | |||
| 245 | + | } | |||
| 246 | + | return false | |||
| 247 | + | } | |||
| 248 | + | ||||
| 249 | + | // clearPaintDeferred drops any pending deferred-paint flag. Used on park so a | |||
| 250 | + | // resumed client does not inherit a stale deferral. | |||
| 251 | + | func (ac *attachedClient) clearPaintDeferred() { | |||
| 252 | + | ac.sendMu.Lock() | |||
| 253 | + | ac.paintDeferred = false | |||
| 254 | + | ac.sendMu.Unlock() | |||
| 255 | + | } | |||
| 256 | + | ||||
| 228 | // send serialises a frame onto the client's transport. | 257 | // send serialises a frame onto the client's transport. | ||
| 229 | func (ac *attachedClient) send(f ports.Frame) error { | 258 | func (ac *attachedClient) send(f ports.Frame) error { | ||
| 230 | _, err := ac.sendTransport(f) | 259 | _, err := ac.sendTransport(f) | ||
| @@ -492,6 +521,9 @@ func (d *Daemon) runConnLoop(ac *attachedClient) { | |||||
| 492 | case ports.MsgAck: | 521 | case ports.MsgAck: | ||
| 493 | if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil { | 522 | if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil { | ||
| 494 | ac.advanceOutputAck(ack.AckedStateNum) | 523 | ac.advanceOutputAck(ack.AckedStateNum) | ||
| 524 | + | if ac.takeDeferredPaint() { | |||
| 525 | + | d.paint(sess, ac, false) | |||
| 526 | + | } | |||
| 495 | } | 527 | } | ||
| 496 | case ports.MsgPing: | 528 | case ports.MsgPing: | ||
| 497 | _ = ac.send(framePong()) | 529 | _ = ac.send(framePong()) | ||
M internal/usecase/daemon/daemon.go +8 -0
| @@ -65,6 +65,14 @@ const defaultScrollbackRows = 10_000 | |||
| 65 | 65 | // the in-flight send. Teardown is never gated on a client draining its socket. | |
| 66 | 66 | const detachNotifyTimeout = time.Second | |
| 67 | 67 | ||
| 68 | + | // maxUnackedOutputStates caps how many output states may be in flight (sent but | |
| 69 | + | // not yet acked by the client) before paint defers rather than composing | |
| 70 | + | // another diff. It bounds the daemon's paint rate to the client's ack rate, so | |
| 71 | + | // heavy output degrades to lower fps on a slow link instead of overflowing the | |
| 72 | + | // transport. It must stay well under the UDP proxy's 32-frame client window so | |
| 73 | + | // the proxy's reliable queue never fills from painting alone. | |
| 74 | + | const maxUnackedOutputStates = 8 | |
| 75 | + | ||
| 68 | 76 | const defaultResumeParkGrace = 15 * time.Minute | |
| 69 | 77 | ||
| 70 | 78 | // defaultSize is used when a client's Hello carries no valid dimensions. | |
| @@ -65,6 +65,14 @@ const defaultScrollbackRows = 10_000 | |||||
| 65 | // the in-flight send. Teardown is never gated on a client draining its socket. | 65 | // the in-flight send. Teardown is never gated on a client draining its socket. | ||
| 66 | const detachNotifyTimeout = time.Second | 66 | const detachNotifyTimeout = time.Second | ||
| 67 | 67 | ||||
| 68 | + | // maxUnackedOutputStates caps how many output states may be in flight (sent but | |||
| 69 | + | // not yet acked by the client) before paint defers rather than composing | |||
| 70 | + | // another diff. It bounds the daemon's paint rate to the client's ack rate, so | |||
| 71 | + | // heavy output degrades to lower fps on a slow link instead of overflowing the | |||
| 72 | + | // transport. It must stay well under the UDP proxy's 32-frame client window so | |||
| 73 | + | // the proxy's reliable queue never fills from painting alone. | |||
| 74 | + | const maxUnackedOutputStates = 8 | |||
| 75 | + | ||||
| 68 | const defaultResumeParkGrace = 15 * time.Minute | 76 | const defaultResumeParkGrace = 15 * time.Minute | ||
| 69 | 77 | ||||
| 70 | // defaultSize is used when a client's Hello carries no valid dimensions. | 78 | // defaultSize is used when a client's Hello carries no valid dimensions. | ||
A internal/usecase/daemon/paint_gate_test.go +78 -0
| @@ -0,0 +1,78 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "testing" | |
| 5 | + | "time" | |
| 6 | + | ||
| 7 | + | "github.com/bnema/vev/internal/ports" | |
| 8 | + | "github.com/stretchr/testify/require" | |
| 9 | + | ) | |
| 10 | + | ||
| 11 | + | func requireNoOutputFrame(t *testing.T, sends chan ports.Frame) { | |
| 12 | + | t.Helper() | |
| 13 | + | select { | |
| 14 | + | case f := <-sends: | |
| 15 | + | if f.Type == ports.MsgOutput { | |
| 16 | + | t.Fatalf("unexpected output frame: %+v", f) | |
| 17 | + | } | |
| 18 | + | case <-time.After(50 * time.Millisecond): | |
| 19 | + | } | |
| 20 | + | } | |
| 21 | + | ||
| 22 | + | func TestPaintDefersWhenOutputAckLagsAtCap(t *testing.T) { | |
| 23 | + | p, _ := newBlockingPTY(t) | |
| 24 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |
| 25 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |
| 26 | + | ||
| 27 | + | // Client is maxUnackedOutputStates behind (outputAck stays 0). | |
| 28 | + | ac.nextStateNum = maxUnackedOutputStates | |
| 29 | + | d.paint(sess, ac, false) | |
| 30 | + | ||
| 31 | + | requireNoOutputFrame(t, sends) | |
| 32 | + | ac.sendMu.Lock() | |
| 33 | + | deferred := ac.paintDeferred | |
| 34 | + | ac.sendMu.Unlock() | |
| 35 | + | require.True(t, deferred, "diff paint at ack cap must defer") | |
| 36 | + | } | |
| 37 | + | ||
| 38 | + | func TestDeferredPaintFlushesOnceWhenAcksCatchUp(t *testing.T) { | |
| 39 | + | p, _ := newBlockingPTY(t) | |
| 40 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |
| 41 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |
| 42 | + | ||
| 43 | + | ac.nextStateNum = maxUnackedOutputStates | |
| 44 | + | d.paint(sess, ac, false) | |
| 45 | + | requireNoOutputFrame(t, sends) | |
| 46 | + | ||
| 47 | + | // Acks catch up: the deferred paint becomes flushable and emits exactly one | |
| 48 | + | // cumulative frame (the renderer shadow only advanced on send, so the skip | |
| 49 | + | // coalesces into this paint). | |
| 50 | + | ac.advanceOutputAck(maxUnackedOutputStates) | |
| 51 | + | require.True(t, ac.takeDeferredPaint(), "deferred paint must flush once acks catch up") | |
| 52 | + | d.paint(sess, ac, false) | |
| 53 | + | ||
| 54 | + | awaitFrame(t, sends, ports.MsgOutput) | |
| 55 | + | requireNoOutputFrame(t, sends) | |
| 56 | + | ac.sendMu.Lock() | |
| 57 | + | deferred := ac.paintDeferred | |
| 58 | + | ac.sendMu.Unlock() | |
| 59 | + | require.False(t, deferred, "flag must clear after the flush paint") | |
| 60 | + | } | |
| 61 | + | ||
| 62 | + | func TestResetPaintBypassesAckGate(t *testing.T) { | |
| 63 | + | p, _ := newBlockingPTY(t) | |
| 64 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |
| 65 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |
| 66 | + | ||
| 67 | + | // At the cap and already deferred: a reset paint is a full invalidation that | |
| 68 | + | // must paint anyway and clear the deferral. | |
| 69 | + | ac.nextStateNum = maxUnackedOutputStates | |
| 70 | + | ac.paintDeferred = true | |
| 71 | + | d.paint(sess, ac, true) | |
| 72 | + | ||
| 73 | + | awaitFrame(t, sends, ports.MsgOutput) | |
| 74 | + | ac.sendMu.Lock() | |
| 75 | + | deferred := ac.paintDeferred | |
| 76 | + | ac.sendMu.Unlock() | |
| 77 | + | require.False(t, deferred, "reset paint must clear the deferral") | |
| 78 | + | } | |
| @@ -0,0 +1,78 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "testing" | |||
| 5 | + | "time" | |||
| 6 | + | ||||
| 7 | + | "github.com/bnema/vev/internal/ports" | |||
| 8 | + | "github.com/stretchr/testify/require" | |||
| 9 | + | ) | |||
| 10 | + | ||||
| 11 | + | func requireNoOutputFrame(t *testing.T, sends chan ports.Frame) { | |||
| 12 | + | t.Helper() | |||
| 13 | + | select { | |||
| 14 | + | case f := <-sends: | |||
| 15 | + | if f.Type == ports.MsgOutput { | |||
| 16 | + | t.Fatalf("unexpected output frame: %+v", f) | |||
| 17 | + | } | |||
| 18 | + | case <-time.After(50 * time.Millisecond): | |||
| 19 | + | } | |||
| 20 | + | } | |||
| 21 | + | ||||
| 22 | + | func TestPaintDefersWhenOutputAckLagsAtCap(t *testing.T) { | |||
| 23 | + | p, _ := newBlockingPTY(t) | |||
| 24 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |||
| 25 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |||
| 26 | + | ||||
| 27 | + | // Client is maxUnackedOutputStates behind (outputAck stays 0). | |||
| 28 | + | ac.nextStateNum = maxUnackedOutputStates | |||
| 29 | + | d.paint(sess, ac, false) | |||
| 30 | + | ||||
| 31 | + | requireNoOutputFrame(t, sends) | |||
| 32 | + | ac.sendMu.Lock() | |||
| 33 | + | deferred := ac.paintDeferred | |||
| 34 | + | ac.sendMu.Unlock() | |||
| 35 | + | require.True(t, deferred, "diff paint at ack cap must defer") | |||
| 36 | + | } | |||
| 37 | + | ||||
| 38 | + | func TestDeferredPaintFlushesOnceWhenAcksCatchUp(t *testing.T) { | |||
| 39 | + | p, _ := newBlockingPTY(t) | |||
| 40 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |||
| 41 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |||
| 42 | + | ||||
| 43 | + | ac.nextStateNum = maxUnackedOutputStates | |||
| 44 | + | d.paint(sess, ac, false) | |||
| 45 | + | requireNoOutputFrame(t, sends) | |||
| 46 | + | ||||
| 47 | + | // Acks catch up: the deferred paint becomes flushable and emits exactly one | |||
| 48 | + | // cumulative frame (the renderer shadow only advanced on send, so the skip | |||
| 49 | + | // coalesces into this paint). | |||
| 50 | + | ac.advanceOutputAck(maxUnackedOutputStates) | |||
| 51 | + | require.True(t, ac.takeDeferredPaint(), "deferred paint must flush once acks catch up") | |||
| 52 | + | d.paint(sess, ac, false) | |||
| 53 | + | ||||
| 54 | + | awaitFrame(t, sends, ports.MsgOutput) | |||
| 55 | + | requireNoOutputFrame(t, sends) | |||
| 56 | + | ac.sendMu.Lock() | |||
| 57 | + | deferred := ac.paintDeferred | |||
| 58 | + | ac.sendMu.Unlock() | |||
| 59 | + | require.False(t, deferred, "flag must clear after the flush paint") | |||
| 60 | + | } | |||
| 61 | + | ||||
| 62 | + | func TestResetPaintBypassesAckGate(t *testing.T) { | |||
| 63 | + | p, _ := newBlockingPTY(t) | |||
| 64 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, p) | |||
| 65 | + | sess.tabs[0].focusedPane().screen.Write([]byte("A")) | |||
| 66 | + | ||||
| 67 | + | // At the cap and already deferred: a reset paint is a full invalidation that | |||
| 68 | + | // must paint anyway and clear the deferral. | |||
| 69 | + | ac.nextStateNum = maxUnackedOutputStates | |||
| 70 | + | ac.paintDeferred = true | |||
| 71 | + | d.paint(sess, ac, true) | |||
| 72 | + | ||||
| 73 | + | awaitFrame(t, sends, ports.MsgOutput) | |||
| 74 | + | ac.sendMu.Lock() | |||
| 75 | + | deferred := ac.paintDeferred | |||
| 76 | + | ac.sendMu.Unlock() | |||
| 77 | + | require.False(t, deferred, "reset paint must clear the deferral") | |||
| 78 | + | } | |||
M internal/usecase/daemon/render.go +12 -0
| @@ -280,6 +280,18 @@ func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool) { | |||
| 280 | 280 | ||
| 281 | 281 | ac.initOverlays() | |
| 282 | 282 | ac.sendMu.Lock() | |
| 283 | + | // Ack-gated coalescing: while the client is at least maxUnackedOutputStates | |
| 284 | + | // behind, skip composing another diff and mark the paint deferred. The | |
| 285 | + | // MsgOutputAck handler flushes one cumulative paint once acks catch up. | |
| 286 | + | // A reset paint is an explicit full invalidation that supersedes any queued | |
| 287 | + | // diffs, so it bypasses the gate and clears the flag. | |
| 288 | + | if reset { | |
| 289 | + | ac.paintDeferred = false | |
| 290 | + | } else if ac.outputLagAtCapLocked() { | |
| 291 | + | ac.paintDeferred = true | |
| 292 | + | ac.sendMu.Unlock() | |
| 293 | + | return | |
| 294 | + | } | |
| 283 | 295 | overlays := ac.overlays.SnapshotForRender() | |
| 284 | 296 | repaintAttachedClients := false | |
| 285 | 297 | defer func() { | |
| @@ -280,6 +280,18 @@ func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool) { | |||||
| 280 | 280 | ||||
| 281 | ac.initOverlays() | 281 | ac.initOverlays() | ||
| 282 | ac.sendMu.Lock() | 282 | ac.sendMu.Lock() | ||
| 283 | + | // Ack-gated coalescing: while the client is at least maxUnackedOutputStates | |||
| 284 | + | // behind, skip composing another diff and mark the paint deferred. The | |||
| 285 | + | // MsgOutputAck handler flushes one cumulative paint once acks catch up. | |||
| 286 | + | // A reset paint is an explicit full invalidation that supersedes any queued | |||
| 287 | + | // diffs, so it bypasses the gate and clears the flag. | |||
| 288 | + | if reset { | |||
| 289 | + | ac.paintDeferred = false | |||
| 290 | + | } else if ac.outputLagAtCapLocked() { | |||
| 291 | + | ac.paintDeferred = true | |||
| 292 | + | ac.sendMu.Unlock() | |||
| 293 | + | return | |||
| 294 | + | } | |||
| 283 | overlays := ac.overlays.SnapshotForRender() | 295 | overlays := ac.overlays.SnapshotForRender() | ||
| 284 | repaintAttachedClients := false | 296 | repaintAttachedClients := false | ||
| 285 | defer func() { | 297 | defer func() { | ||
M internal/usecase/daemon/resume.go +3 -0
| @@ -48,6 +48,9 @@ func (d *Daemon) parkAttachment(sess *session, ac *attachedClient) bool { | |||
| 48 | 48 | if !d.prepareParkAttachment(sess, ac) { | |
| 49 | 49 | return false | |
| 50 | 50 | } | |
| 51 | + | // Drop any deferred diff so a resumed client starts clean; resume forces a | |
| 52 | + | // full reset repaint anyway. | |
| 53 | + | ac.clearPaintDeferred() | |
| 51 | 54 | d.mu.Lock() | |
| 52 | 55 | if d.closing { | |
| 53 | 56 | d.mu.Unlock() | |
| @@ -48,6 +48,9 @@ func (d *Daemon) parkAttachment(sess *session, ac *attachedClient) bool { | |||||
| 48 | if !d.prepareParkAttachment(sess, ac) { | 48 | if !d.prepareParkAttachment(sess, ac) { | ||
| 49 | return false | 49 | return false | ||
| 50 | } | 50 | } | ||
| 51 | + | // Drop any deferred diff so a resumed client starts clean; resume forces a | |||
| 52 | + | // full reset repaint anyway. | |||
| 53 | + | ac.clearPaintDeferred() | |||
| 51 | d.mu.Lock() | 54 | d.mu.Lock() | ||
| 52 | if d.closing { | 55 | if d.closing { | ||
| 53 | d.mu.Unlock() | 56 | d.mu.Unlock() | ||