1package daemon
2
3import (
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.
15type echoTestClock struct {
16 mu sync.Mutex
17 now time.Time
18 timers chan echoTestTimer
19}
20
21type echoTestTimer struct {
22 timer *manualAttentionTimer
23 delay time.Duration
24}
25
26func newEchoTestClock() *echoTestClock {
27 return &echoTestClock{now: time.Unix(100, 0), timers: make(chan echoTestTimer, 16)}
28}
29
30func (c *echoTestClock) Now() time.Time {
31 c.mu.Lock()
32 defer c.mu.Unlock()
33 return c.now
34}
35
36func (c *echoTestClock) advance(d time.Duration) {
37 c.mu.Lock()
38 c.now = c.now.Add(d)
39 c.mu.Unlock()
40}
41
42func (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.
50func (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
66func TestEchoAckWaitsForEchoTimeoutAndSendsAckOnlyOutput(t *testing.T) {
67 clock := newEchoTestClock()
68 d, sess, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY())
69
70 // Drive the production input path so the acknowledgement wiring is
71 // covered, not just the tracker.
72 d.handleSequencedInput(sess, ac, 1, []byte("a"))
73 clock.advance(20 * time.Millisecond)
74 d.handleSequencedInput(sess, ac, 2, []byte("b"))
75 timer, delay := clock.nextTimer(t)
76 require.Equal(t, echoAckDelay, delay)
77 require.Zero(t, ac.echoAck.Load(), "input is not echoed when applied")
78
79 clock.advance(30 * time.Millisecond)
80 timer.fire()
81 output := decodeServerMessage(t, awaitFrame(t, sends, "Output")).(protocol.Output)
82 require.Equal(t, uint64(1), output.Echo)
83 require.Zero(t, output.New, "an ack-only output is a side effect")
84 require.Empty(t, output.Data)
85
86 // Input 2 is still younger than the timeout: the tracker re-arms for its
87 // remaining age.
88 timer, delay = clock.nextTimer(t)
89 require.Equal(t, 20*time.Millisecond, delay)
90 clock.advance(20 * time.Millisecond)
91 timer.fire()
92 output = decodeServerMessage(t, awaitFrame(t, sends, "Output")).(protocol.Output)
93 require.Equal(t, uint64(2), output.Echo)
94 require.Equal(t, uint64(2), ac.echoAck.Load())
95}
96
97func TestEchoAckIgnoresUnsequencedAndStaleInput(t *testing.T) {
98 clock := newEchoTestClock()
99 d, _, ac, _ := newManualSessionWithPTYsClock(t, clock, newQuietPTY())
100 ac.echoAck.Store(5)
101
102 d.noteInputApplied(ac, 0)
103 d.noteInputApplied(ac, 5)
104 require.Empty(t, ac.echo.pending)
105 require.Empty(t, clock.timers)
106}
107
108func TestResetEchoAckDropsPendingInput(t *testing.T) {
109 clock := newEchoTestClock()
110 d, _, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY())
111
112 d.noteInputApplied(ac, 7)
113 timer, _ := clock.nextTimer(t)
114 ac.resetEchoAck()
115 clock.advance(echoAckDelay)
116 timer.fire()
117
118 require.Zero(t, ac.echoAck.Load())
119 require.Empty(t, drainAllFrames(sends), "a reset tracker must not publish the old link's input")
120}
121
122func TestEchoAckWaitsWhileOutputWindowIsFull(t *testing.T) {
123 clock := newEchoTestClock()
124 d, _, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY())
125 ac.output.maxOutstandingAtomic.Store(1)
126 ac.output.outstandingAtomic.Store(1)
127
128 d.noteInputApplied(ac, 1)
129 timer, _ := clock.nextTimer(t)
130 clock.advance(echoAckDelay)
131 timer.fire()
132
133 retry, delay := clock.nextTimer(t)
134 require.Equal(t, echoAckDelay, delay, "a full window defers the ack")
135 require.Equal(t, uint64(1), ac.echoAck.Load(), "the next frame may still carry it")
136 require.Empty(t, drainAllFrames(sends), "no ack-only output may overtake withheld frames")
137
138 ac.output.outstandingAtomic.Store(0)
139 clock.advance(echoAckDelay)
140 retry.fire()
141 output := decodeServerMessage(t, awaitFrame(t, sends, "Output")).(protocol.Output)
142 require.Equal(t, uint64(1), output.Echo)
143 require.Zero(t, output.New)
144}