1package app
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/bnema/zerowrap"
11 "github.com/spf13/viper"
12 "github.com/stretchr/testify/assert"
13 "github.com/stretchr/testify/require"
14
15 "github.com/bnema/gordon/internal/domain"
16 "github.com/bnema/gordon/internal/usecase/proxy"
17)
18
19type watchRecorder struct {
20 called bool
21 onChange func()
22}
23
24func (w *watchRecorder) Watch(ctx context.Context, onChange func()) error {
25 w.called = true
26 w.onChange = onChange
27 return nil
28}
29
30type coordinatorRecorder struct {
31 mu sync.Mutex
32 triggerCalls int
33 applyCalls int
34 err error
35}
36
37func (c *coordinatorRecorder) Trigger(context.Context) error {
38 c.mu.Lock()
39 c.triggerCalls++
40 err := c.err
41 c.mu.Unlock()
42 return err
43}
44
45func (c *coordinatorRecorder) ApplyLoadedConfig(context.Context) error {
46 c.mu.Lock()
47 c.applyCalls++
48 err := c.err
49 c.mu.Unlock()
50 return err
51}
52
53func (c *coordinatorRecorder) TriggerCalls() int {
54 c.mu.Lock()
55 defer c.mu.Unlock()
56 return c.triggerCalls
57}
58
59func (c *coordinatorRecorder) ApplyCalls() int {
60 c.mu.Lock()
61 defer c.mu.Unlock()
62 return c.applyCalls
63}
64
65type reloadRecorder struct {
66 mu sync.Mutex
67 calls int
68 err error
69 onReload func()
70}
71
72func (r *reloadRecorder) Reload(context.Context) error {
73 r.mu.Lock()
74 r.calls++
75 err := r.err
76 onReload := r.onReload
77 r.mu.Unlock()
78
79 if onReload != nil {
80 onReload()
81 }
82
83 return err
84}
85
86func (r *reloadRecorder) Calls() int {
87 r.mu.Lock()
88 defer r.mu.Unlock()
89 return r.calls
90}
91
92type proxyRecorder struct {
93 calls int
94 config proxy.Config
95}
96
97func (p *proxyRecorder) UpdateConfig(config proxy.Config) {
98 p.calls++
99 p.config = config
100}
101
102type eventBusRecorder struct {
103 mu sync.Mutex
104 calls int
105 eventType domain.EventType
106 payload any
107 err error
108}
109
110func (e *eventBusRecorder) Publish(eventType domain.EventType, payload any) error {
111 e.mu.Lock()
112 e.calls++
113 e.eventType = eventType
114 e.payload = payload
115 err := e.err
116 e.mu.Unlock()
117 return err
118}
119
120func (e *eventBusRecorder) Calls() int {
121 e.mu.Lock()
122 defer e.mu.Unlock()
123 return e.calls
124}
125
126func TestSetupConfigHotReload_WatchCallbackInvokesCoordinator(t *testing.T) {
127 ctx := context.Background()
128 configSvc := &watchRecorder{}
129 coordinator := &coordinatorRecorder{}
130
131 err := setupConfigHotReload(ctx, configSvc, coordinator)
132 require.NoError(t, err)
133 require.True(t, configSvc.called)
134 require.NotNil(t, configSvc.onChange)
135
136 configSvc.onChange()
137 require.Equal(t, 1, coordinator.ApplyCalls())
138 require.Equal(t, 0, coordinator.TriggerCalls())
139}
140
141func TestReloadCoordinator_ApplyLoadedConfig_RebuildsProxyConfigAndPublishesEvent(t *testing.T) {
142 ctx := context.Background()
143 v := viper.New()
144 v.Set("server.gordon_domain", "reload.example.com")
145 v.Set("server.registry_port", 5000)
146 v.Set("server.max_proxy_body_size", "5MB")
147 v.Set("server.max_blob_chunk_size", "6MB")
148 v.Set("server.max_proxy_response_size", "7MB")
149 v.Set("server.max_concurrent_connections", 99)
150
151 reloadSvc := &reloadRecorder{}
152 proxySvc := &proxyRecorder{}
153 events := &eventBusRecorder{}
154
155 coord := newReloadCoordinator(v, reloadSvc, proxySvc, events, zerowrap.Default())
156 require.NoError(t, coord.ApplyLoadedConfig(ctx))
157
158 require.Equal(t, 0, reloadSvc.Calls())
159 require.Equal(t, 1, proxySvc.calls)
160 require.Equal(t, 1, events.Calls())
161 require.Equal(t, domain.EventConfigReload, events.eventType)
162 require.Equal(t, proxy.Config{
163 RegistryDomain: "reload.example.com",
164 RegistryPort: 5000,
165 MaxBodySize: 5 << 20,
166 MaxResponseSize: 7 << 20,
167 MaxConcurrentConns: 99,
168 }, proxySvc.config)
169}
170
171func TestReloadCoordinator_DebouncesRepeatedWatchCallbacks(t *testing.T) {
172 ctx := context.Background()
173 v := viper.New()
174 v.Set("server.gordon_domain", "new.example.com")
175 v.Set("server.registry_port", 5000)
176 v.Set("server.max_proxy_body_size", "5MB")
177 v.Set("server.max_blob_chunk_size", "6MB")
178 v.Set("server.max_proxy_response_size", "7MB")
179 v.Set("server.max_concurrent_connections", 99)
180
181 reloadSvc := &reloadRecorder{}
182 proxySvc := &proxyRecorder{}
183 events := &eventBusRecorder{}
184
185 coord := newReloadCoordinator(v, reloadSvc, proxySvc, events, zerowrap.Default())
186 require.NoError(t, coord.Trigger(ctx))
187 require.NoError(t, coord.Trigger(ctx))
188
189 require.Equal(t, 1, reloadSvc.Calls())
190 require.Equal(t, 1, proxySvc.calls)
191 require.Equal(t, 1, events.Calls())
192 require.Equal(t, domain.EventConfigReload, events.eventType)
193 require.Equal(t, proxy.Config{
194 RegistryDomain: "new.example.com",
195 RegistryPort: 5000,
196 MaxBodySize: 5 << 20,
197 MaxResponseSize: 7 << 20,
198 MaxConcurrentConns: 99,
199 }, proxySvc.config)
200}
201
202func TestReloadCoordinator_RetriesImmediatelyAfterFailedReload(t *testing.T) {
203 ctx := context.Background()
204 v := viper.New()
205 v.Set("server.gordon_domain", "retry.example.com")
206 v.Set("server.registry_port", 5000)
207 v.Set("server.max_proxy_body_size", "2MB")
208 v.Set("server.max_blob_chunk_size", "3MB")
209 v.Set("server.max_proxy_response_size", "4MB")
210 v.Set("server.max_concurrent_connections", 11)
211
212 reloadErr := errors.New("reload failed")
213 reloadSvc := &reloadRecorder{err: reloadErr}
214 proxySvc := &proxyRecorder{}
215 coord := newReloadCoordinator(v, reloadSvc, proxySvc, nil, zerowrap.Default())
216
217 err := coord.Trigger(ctx)
218 require.ErrorIs(t, err, reloadErr)
219
220 reloadSvc.mu.Lock()
221 reloadSvc.err = nil
222 reloadSvc.mu.Unlock()
223
224 require.NoError(t, coord.Trigger(ctx))
225 require.Equal(t, 2, reloadSvc.Calls())
226 assert.Equal(t, 1, proxySvc.calls)
227}
228
229func TestReloadCoordinator_PublishErrorDoesNotAdvanceDebounceState(t *testing.T) {
230 ctx := context.Background()
231 v := viper.New()
232 v.Set("server.gordon_domain", "publish.example.com")
233 v.Set("server.registry_port", 5000)
234 v.Set("server.max_proxy_body_size", "5MB")
235 v.Set("server.max_blob_chunk_size", "6MB")
236 v.Set("server.max_proxy_response_size", "7MB")
237 v.Set("server.max_concurrent_connections", 99)
238
239 publishErr := errors.New("publish failed")
240 reloadSvc := &reloadRecorder{}
241 proxySvc := &proxyRecorder{}
242 events := &eventBusRecorder{err: publishErr}
243
244 coord := newReloadCoordinator(v, reloadSvc, proxySvc, events, zerowrap.Default())
245
246 err := coord.Trigger(ctx)
247 require.ErrorIs(t, err, publishErr)
248
249 err = coord.Trigger(ctx)
250 require.ErrorIs(t, err, publishErr)
251
252 require.Equal(t, 2, reloadSvc.Calls())
253 require.Equal(t, 2, proxySvc.calls)
254 require.Equal(t, 2, events.Calls())
255 require.Equal(t, domain.EventConfigReload, events.eventType)
256}
257
258func TestReloadCoordinator_SerializesOverlappingReloadRequests(t *testing.T) {
259 ctx := context.Background()
260 v := viper.New()
261 v.Set("server.gordon_domain", "app.example.com")
262 v.Set("server.registry_port", 5000)
263 v.Set("server.max_proxy_body_size", "2MB")
264 v.Set("server.max_blob_chunk_size", "3MB")
265 v.Set("server.max_proxy_response_size", "4MB")
266 v.Set("server.max_concurrent_connections", 11)
267
268 started := make(chan struct{})
269 release := make(chan struct{})
270 var once sync.Once
271 reloadSvc := &reloadRecorder{onReload: func() {
272 once.Do(func() { close(started) })
273 <-release
274 }}
275 proxySvc := &proxyRecorder{}
276 coord := newReloadCoordinator(v, reloadSvc, proxySvc, nil, zerowrap.Default())
277
278 firstDone := make(chan struct{})
279 go func() {
280 _ = coord.Trigger(ctx)
281 close(firstDone)
282 }()
283
284 select {
285 case <-started:
286 case <-time.After(2 * time.Second):
287 t.Fatal("first reload did not start")
288 }
289
290 secondDone := make(chan struct{})
291 go func() {
292 _ = coord.Trigger(ctx)
293 close(secondDone)
294 }()
295
296 select {
297 case <-secondDone:
298 t.Fatal("second reload finished before the first was released")
299 case <-time.After(75 * time.Millisecond):
300 }
301
302 require.Equal(t, 1, reloadSvc.Calls())
303 close(release)
304
305 select {
306 case <-firstDone:
307 case <-time.After(2 * time.Second):
308 t.Fatal("first reload did not finish")
309 }
310
311 select {
312 case <-secondDone:
313 case <-time.After(2 * time.Second):
314 t.Fatal("second reload did not finish")
315 }
316
317 require.Equal(t, 1, reloadSvc.Calls())
318 assert.Equal(t, 1, proxySvc.calls)
319}