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