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, _, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY())
69
70 d.noteInputApplied(ac, 1)
71 clock.advance(20 * time.Millisecond)
72 d.noteInputApplied(ac, 2)
73 timer, delay := clock.nextTimer(t)
74 require.Equal(t, echoAckDelay, delay)
75 require.Zero(t, ac.echoAck.Load(), "input is not echoed when applied")
76
77 clock.advance(30 * time.Millisecond)
78 timer.fire()
79 output := decodeServerMessage(t, awaitFrame(t, sends, "Output")).(protocol.Output)
80 require.Equal(t, uint64(1), output.Echo)
81 require.Zero(t, output.New, "an ack-only output is a side effect")
82 require.Empty(t, output.Data)
83
84 // Input 2 is still younger than the timeout: the tracker re-arms for its
85 // remaining age.
86 timer, delay = clock.nextTimer(t)
87 require.Equal(t, 20*time.Millisecond, delay)
88 clock.advance(20 * time.Millisecond)
89 timer.fire()
90 output = decodeServerMessage(t, awaitFrame(t, sends, "Output")).(protocol.Output)
91 require.Equal(t, uint64(2), output.Echo)
92 require.Equal(t, uint64(2), ac.echoAck.Load())
93}
94
95func TestEchoAckIgnoresUnsequencedAndStaleInput(t *testing.T) {
96 clock := newEchoTestClock()
97 d, _, ac, _ := newManualSessionWithPTYsClock(t, clock, newQuietPTY())
98 ac.echoAck.Store(5)
99
100 d.noteInputApplied(ac, 0)
101 d.noteInputApplied(ac, 5)
102 require.Empty(t, ac.echo.pending)
103 require.Empty(t, clock.timers)
104}
105
106func TestResetEchoAckDropsPendingInput(t *testing.T) {
107 clock := newEchoTestClock()
108 d, _, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY())
109
110 d.noteInputApplied(ac, 7)
111 timer, _ := clock.nextTimer(t)
112 ac.resetEchoAck()
113 clock.advance(echoAckDelay)
114 timer.fire()
115
116 require.Zero(t, ac.echoAck.Load())
117 require.Empty(t, drainAllFrames(sends), "a reset tracker must not publish the old link's input")
118}
119
120func 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}