1package daemon
2
3import (
4 "sync"
5 "time"
6
7 "github.com/bnema/vev/internal/ports"
8 "github.com/bnema/vev/internal/protocol"
9)
10
11// echoAckDelay is the protocol's echo acknowledgement delay.
12const echoAckDelay = protocol.EchoAckDelay
13
14// maxPendingEchoes bounds the per-attachment input history. Acknowledgements
15// are cumulative, so dropping the oldest entry only delays its confirmation to
16// a newer one.
17const maxPendingEchoes = 256
18
19type pendingEcho struct {
20 seq uint64
21 at time.Time
22}
23
24// echoAckTracker turns applied input sequence numbers into the cumulative
25// Output.Echo acknowledgement. echoAck is written only under mu.
26type echoAckTracker struct {
27 mu sync.Mutex
28 pending []pendingEcho
29 timer pendingByteTimer
30}
31
32// noteInputApplied records that the input numbered seq reached the session and
33// schedules its acknowledgement. Sequence zero is unsequenced input.
34func (d *Daemon) noteInputApplied(ac *attachedClient, seq uint64) {
35 if ac == nil || seq == 0 {
36 return
37 }
38 t := &ac.echo
39 t.mu.Lock()
40 defer t.mu.Unlock()
41 if seq <= ac.echoAck.Load() || len(t.pending) > 0 && seq <= t.pending[len(t.pending)-1].seq {
42 return
43 }
44 if len(t.pending) == maxPendingEchoes {
45 t.pending = append(t.pending[:0], t.pending[1:]...)
46 }
47 t.pending = append(t.pending, pendingEcho{seq: seq, at: d.clock.Now()})
48 if t.timer.timer == nil {
49 d.armEchoAckLocked(ac, echoAckDelay)
50 }
51}
52
53// armEchoAckLocked starts the single echo timer. The caller holds echo.mu.
54func (d *Daemon) armEchoAckLocked(ac *attachedClient, delay time.Duration) {
55 t := &ac.echo
56 t.timer.retain(d.clock, delay, func(fired ports.Timer) {
57 if d.advanceEchoAck(ac, fired) {
58 d.sendEchoAck(ac)
59 }
60 })
61}
62
63// armEchoAckRetryLocked re-sends the current acknowledgement after one more
64// echo delay. The caller holds echo.mu.
65func (d *Daemon) armEchoAckRetryLocked(ac *attachedClient) {
66 t := &ac.echo
67 t.timer.retain(d.clock, echoAckDelay, func(fired ports.Timer) {
68 t.mu.Lock()
69 current := t.timer.timer == fired
70 if current {
71 t.timer.timer, t.timer.done = nil, nil
72 if len(t.pending) > 0 {
73 // New input owns the next timer and its ack covers this one.
74 d.armEchoAckLocked(ac, max(echoAckDelay-d.clock.Now().Sub(t.pending[0].at), 0))
75 current = false
76 }
77 }
78 t.mu.Unlock()
79 if current {
80 d.sendEchoAck(ac)
81 }
82 })
83}
84
85// advanceEchoAck moves echoAck to the newest input applied at least
86// echoAckDelay ago and re-arms for the rest. It reports whether echoAck moved.
87func (d *Daemon) advanceEchoAck(ac *attachedClient, fired ports.Timer) bool {
88 t := &ac.echo
89 t.mu.Lock()
90 defer t.mu.Unlock()
91 if t.timer.timer != fired {
92 return false // stopped by a reset or superseded
93 }
94 t.timer.timer, t.timer.done = nil, nil
95 now := d.clock.Now()
96 acked := uint64(0)
97 n := 0
98 for n < len(t.pending) && now.Sub(t.pending[n].at) >= echoAckDelay {
99 acked = t.pending[n].seq
100 n++
101 }
102 t.pending = append(t.pending[:0], t.pending[n:]...)
103 if len(t.pending) > 0 {
104 d.armEchoAckLocked(ac, max(echoAckDelay-now.Sub(t.pending[0].at), 0))
105 }
106 if acked <= ac.echoAck.Load() {
107 return false
108 }
109 ac.echoAck.Store(acked)
110 return true
111}
112
113// sendEchoAck publishes a new acknowledgement in an empty side-effect Output,
114// so the client learns about it even when the screen does not change.
115//
116// Side effects bypass the output window. While the window is full, the frame
117// holding the acknowledged echo may still be withheld, and an early ack would
118// make the client judge its guesses against an older screen. The ack then
119// waits: the next frame carries it, and a retry covers a screen that never
120// changes.
121func (d *Daemon) sendEchoAck(ac *attachedClient) {
122 if ac.attachmentActivity() != attachmentActive {
123 return
124 }
125 sess := ac.currentAttachmentSession()
126 if sess == nil {
127 return
128 }
129 if ac.output != nil && ac.output.atCapacity() {
130 t := &ac.echo
131 t.mu.Lock()
132 if t.timer.timer == nil {
133 d.armEchoAckRetryLocked(ac)
134 }
135 t.mu.Unlock()
136 return
137 }
138 failed, err := d.boundedSendOutputErrTransport(ac, nil)
139 if err != nil && failed != nil {
140 d.detachOnSendError(sess, ac, failed)
141 }
142}
143
144// resetEchoAck forgets every acknowledgement. A resumed client restarts its
145// input numbering, so the replacement link must start from zero too.
146func (ac *attachedClient) resetEchoAck() {
147 t := &ac.echo
148 t.mu.Lock()
149 defer t.mu.Unlock()
150 t.timer.stop()
151 t.pending = nil
152 ac.echoAck.Store(0)
153}