1package daemon
2
3import (
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.
14const 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.
19const maxPendingEchoes = 256
20
21type 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.
28type 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.
36func (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.
56func (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// armEchoAckRetryLocked re-sends the current acknowledgement after one more
66// echo delay. The caller holds echo.mu.
67func (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
87// advanceEchoAck moves echoAck to the newest input applied at least
88// echoAckDelay ago and re-arms for the rest. It reports whether echoAck moved.
89func (d *Daemon) advanceEchoAck(ac *attachedClient, fired ports.Timer) bool {
90 t := &ac.echo
91 t.mu.Lock()
92 defer t.mu.Unlock()
93 if t.timer.timer != fired {
94 return false // stopped by a reset or superseded
95 }
96 t.timer.timer, t.timer.done = nil, nil
97 now := d.clock.Now()
98 acked := uint64(0)
99 n := 0
100 for n < len(t.pending) && now.Sub(t.pending[n].at) >= echoAckDelay {
101 acked = t.pending[n].seq
102 n++
103 }
104 t.pending = append(t.pending[:0], t.pending[n:]...)
105 if len(t.pending) > 0 {
106 d.armEchoAckLocked(ac, max(echoAckDelay-now.Sub(t.pending[0].at), 0))
107 }
108 if acked <= ac.echoAck.Load() {
109 return false
110 }
111 ac.echoAck.Store(acked)
112 return true
113}
114
115// sendEchoAck publishes a new acknowledgement in an empty side-effect Output,
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.
123func (d *Daemon) sendEchoAck(ac *attachedClient) {
124 if ac.attachmentActivity() != attachmentActive {
125 return
126 }
127 sess := ac.currentAttachmentSession()
128 if sess == nil {
129 return
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 }
140 failed, err := d.boundedSendOutputErrTransport(ac, nil)
141 if err != nil && failed != nil {
142 d.detachOnSendError(sess, ac, failed)
143 }
144}
145
146// resetEchoAck forgets every acknowledgement. A resumed client restarts its
147// input numbering, so the replacement link must start from zero too.
148func (ac *attachedClient) resetEchoAck() {
149 t := &ac.echo
150 t.mu.Lock()
151 defer t.mu.Unlock()
152 t.timer.stop()
153 t.pending = nil
154 ac.echoAck.Store(0)
155}