1package daemon
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/stretchr/testify/require"
11
12 "github.com/bnema/vev/internal/ports"
13 "github.com/bnema/vev/internal/protocol"
14)
15
16type commandTrackerTestTimer struct {
17 ch chan time.Time
18 once sync.Once
19}
20
21func (t *commandTrackerTestTimer) C() <-chan time.Time { return t.ch }
22func (t *commandTrackerTestTimer) Reset(time.Duration) bool { return false }
23func (t *commandTrackerTestTimer) Stop() bool { return true }
24func (t *commandTrackerTestTimer) fire() { t.once.Do(func() { t.ch <- time.Time{} }) }
25
26type commandTrackerTestClock struct {
27 timers chan *commandTrackerTestTimer
28 delay time.Duration
29}
30
31func newCommandTrackerTestClock() *commandTrackerTestClock {
32 return &commandTrackerTestClock{timers: make(chan *commandTrackerTestTimer, 8)}
33}
34
35func (c *commandTrackerTestClock) Now() time.Time { return time.Time{} }
36
37func (c *commandTrackerTestClock) NewTimer(delay time.Duration) ports.Timer {
38 c.delay = delay
39 timer := &commandTrackerTestTimer{ch: make(chan time.Time, 1)}
40 c.timers <- timer
41 return timer
42}
43
44func TestCommandRequestTrackerTimeoutAndGenerationIsolation(t *testing.T) {
45 tracker := NewCommandRequestTracker()
46 clock := newCommandTrackerTestClock()
47 requestID, outcome := tracker.Publish(4)
48 waitDone := make(chan error, 1)
49 go func() {
50 _, err := tracker.Wait(context.Background(), clock, requestID, 4, outcome)
51 waitDone <- err
52 }()
53
54 timer := <-clock.timers
55 require.Equal(t, CommandRequestTimeout, clock.delay)
56 timer.fire()
57 require.ErrorIs(t, <-waitDone, ErrCommandRequestTimeout)
58 require.Zero(t, tracker.PendingCount())
59
60 // Reuse the timed-out wire ID with a new generation. A late result from the
61 // old connection must not complete the new request.
62 secondOutcome, ok := tracker.Track(requestID, 8)
63 require.True(t, ok)
64 tracker.Complete(4, protocol.CommandResult{RequestID: requestID, Outcome: protocol.CommandSucceeded})
65 select {
66 case <-secondOutcome:
67 t.Fatal("late result from the old generation completed a newer request")
68 default:
69 }
70 tracker.Complete(8, protocol.CommandResult{RequestID: requestID, Outcome: protocol.CommandSucceeded})
71 result := <-secondOutcome
72 require.True(t, result.Result.Outcome == protocol.CommandSucceeded)
73
74 t.Run("wait outcomes", func(t *testing.T) {
75 failure := errors.New("transport failed")
76 tests := []struct {
77 name string
78 setup func(*testing.T, *CommandRequestTracker, uint64) (context.Context, ports.Clock, func())
79 wantErr error
80 wantPending int
81 }{
82 {
83 name: "canceled context",
84 setup: func(_ *testing.T, _ *CommandRequestTracker, _ uint64) (context.Context, ports.Clock, func()) {
85 ctx, cancel := context.WithCancel(context.Background())
86 cancel()
87 return ctx, newCommandTrackerTestClock(), nil
88 },
89 wantErr: context.Canceled,
90 wantPending: 0,
91 },
92 {
93 name: "nil clock",
94 setup: func(_ *testing.T, _ *CommandRequestTracker, _ uint64) (context.Context, ports.Clock, func()) {
95 return context.Background(), nil, nil
96 },
97 wantErr: ErrCommandRequestUnavailable,
98 wantPending: 1,
99 },
100 {
101 name: "fail",
102 setup: func(_ *testing.T, tracker *CommandRequestTracker, requestID uint64) (context.Context, ports.Clock, func()) {
103 tracker.Fail(requestID, 1, failure)
104 return context.Background(), newCommandTrackerTestClock(), nil
105 },
106 wantErr: failure,
107 wantPending: 0,
108 },
109 {
110 name: "remove",
111 setup: func(_ *testing.T, tracker *CommandRequestTracker, requestID uint64) (context.Context, ports.Clock, func()) {
112 tracker.Remove(requestID, 1)
113 clock := newCommandTrackerTestClock()
114 return context.Background(), clock, func() {
115 timer := <-clock.timers
116 timer.fire()
117 }
118 },
119 wantErr: ErrCommandRequestTimeout,
120 wantPending: 0,
121 },
122 {
123 name: "duplicate track",
124 setup: func(t *testing.T, tracker *CommandRequestTracker, requestID uint64) (context.Context, ports.Clock, func()) {
125 duplicate, ok := tracker.Track(requestID, 1)
126 require.False(t, ok)
127 require.Nil(t, duplicate)
128 ctx, cancel := context.WithCancel(context.Background())
129 cancel()
130 return ctx, newCommandTrackerTestClock(), nil
131 },
132 wantErr: context.Canceled,
133 wantPending: 0,
134 },
135 }
136 for _, tt := range tests {
137 t.Run(tt.name, func(t *testing.T) {
138 tracker := NewCommandRequestTracker()
139 requestID, outcome := tracker.Publish(1)
140 ctx, clock, drive := tt.setup(t, tracker, requestID)
141 waitDone := make(chan error, 1)
142 go func() {
143 _, err := tracker.Wait(ctx, clock, requestID, 1, outcome)
144 waitDone <- err
145 }()
146 if drive != nil {
147 drive()
148 }
149 require.ErrorIs(t, <-waitDone, tt.wantErr)
150 require.Equal(t, tt.wantPending, tracker.PendingCount())
151 })
152 }
153 })
154}