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/adapters/in/http/registry"
16 "github.com/bnema/gordon/internal/domain"
17 cfgusecase "github.com/bnema/gordon/internal/usecase/config"
18 "github.com/bnema/gordon/internal/usecase/container"
19 "github.com/bnema/gordon/internal/usecase/proxy"
20)
21
22type watchRecorder struct {
23 called bool
24 onChange func()
25}
26
27func (w *watchRecorder) Watch(ctx context.Context, onChange func()) error {
28 w.called = true
29 w.onChange = onChange
30 return nil
31}
32
33type coordinatorRecorder struct {
34 mu sync.Mutex
35 triggerCalls int
36 applyCalls int
37 err error
38}
39
40func (c *coordinatorRecorder) Trigger(context.Context) error {
41 c.mu.Lock()
42 c.triggerCalls++
43 err := c.err
44 c.mu.Unlock()
45 return err
46}
47
48func (c *coordinatorRecorder) ApplyLoadedConfig(context.Context) error {
49 c.mu.Lock()
50 c.applyCalls++
51 err := c.err
52 c.mu.Unlock()
53 return err
54}
55
56func (c *coordinatorRecorder) TriggerCalls() int {
57 c.mu.Lock()
58 defer c.mu.Unlock()
59 return c.triggerCalls
60}
61
62func (c *coordinatorRecorder) ApplyCalls() int {
63 c.mu.Lock()
64 defer c.mu.Unlock()
65 return c.applyCalls
66}
67
68type reloadRecorder struct {
69 mu sync.Mutex
70 calls int
71 err error
72 onReload func()
73}
74
75func (r *reloadRecorder) Reload(context.Context) error {
76 r.mu.Lock()
77 r.calls++
78 err := r.err
79 onReload := r.onReload
80 r.mu.Unlock()
81
82 if onReload != nil {
83 onReload()
84 }
85
86 return err
87}
88
89func (r *reloadRecorder) Calls() int {
90 r.mu.Lock()
91 defer r.mu.Unlock()
92 return r.calls
93}
94
95type proxyRecorder struct {
96 calls int
97 config proxy.Config
98}
99
100func (p *proxyRecorder) UpdateConfig(config proxy.Config) {
101 p.calls++
102 p.config = config
103}
104
105type registryLimitsRecorder struct {
106 calls int
107 maxBlobChunkSize int64
108 maxBlobSize int64
109}
110
111func (r *registryLimitsRecorder) UpdateBlobLimits(maxBlobChunkSize, maxBlobSize int64) {
112 r.calls++
113 r.maxBlobChunkSize = maxBlobChunkSize
114 r.maxBlobSize = maxBlobSize
115}
116
117type containerConfigApplyRecorder struct {
118 calls int
119 ctx context.Context
120 cfg Config
121}
122
123func (r *containerConfigApplyRecorder) Apply(ctx context.Context, cfg Config) error {
124 r.calls++
125 r.ctx = ctx
126 r.cfg = cfg
127 return nil
128}
129
130type eventBusRecorder struct {
131 mu sync.Mutex
132 calls int
133 eventType domain.EventType
134 payload any
135 err error
136}
137
138func (e *eventBusRecorder) Publish(eventType domain.EventType, payload any) error {
139 e.mu.Lock()
140 e.calls++
141 e.eventType = eventType
142 e.payload = payload
143 err := e.err
144 e.mu.Unlock()
145 return err
146}
147
148func (e *eventBusRecorder) Calls() int {
149 e.mu.Lock()
150 defer e.mu.Unlock()
151 return e.calls
152}
153
154func TestSetupConfigHotReload_WatchCallbackInvokesCoordinator(t *testing.T) {
155 ctx := context.Background()
156 configSvc := &watchRecorder{}
157 coordinator := &coordinatorRecorder{}
158
159 err := setupConfigHotReload(ctx, configSvc, coordinator)
160 require.NoError(t, err)
161 require.True(t, configSvc.called)
162 require.NotNil(t, configSvc.onChange)
163
164 configSvc.onChange()
165 require.Equal(t, 1, coordinator.ApplyCalls())
166 require.Equal(t, 0, coordinator.TriggerCalls())
167}
168
169func TestBuildProxyConfig_ParsesMaxBlobSizeWithoutChangingChunkDefault(t *testing.T) {
170 cfg := Config{}
171 cfg.Server.MaxBlobSize = "2GB"
172
173 result, err := buildProxyConfig(cfg, zerowrap.Default())
174
175 require.NoError(t, err)
176 assert.Equal(t, int64(2<<30), result.maxBlobSize)
177 assert.Equal(t, int64(registry.DefaultMaxBlobChunkSize), result.maxBlobChunkSize)
178}
179
180func TestBuildProxyConfig_DefaultMaxBlobSize(t *testing.T) {
181 result, err := buildProxyConfig(Config{}, zerowrap.Default())
182
183 require.NoError(t, err)
184 assert.Equal(t, int64(registry.DefaultMaxBlobSize), result.maxBlobSize)
185}
186
187func TestReloadCoordinator_ApplyLoadedConfig_RebuildsProxyConfigAndPublishesEvent(t *testing.T) {
188 ctx := context.Background()
189 v := viper.New()
190 v.Set("server.gordon_domain", "reload.example.com")
191 v.Set("server.registry_port", 5000)
192 v.Set("server.max_proxy_body_size", "5MB")
193 v.Set("server.max_blob_chunk_size", "6MB")
194 v.Set("server.max_blob_size", "8MB")
195 v.Set("server.max_proxy_response_size", "7MB")
196 v.Set("server.max_concurrent_connections", 99)
197 v.Set("server.legacy_registry_domains", []string{"old.example.com"})
198 v.Set("images.allowed_registries", []string{"docker.io"})
199
200 reloadSvc := &reloadRecorder{}
201 proxySvc := &proxyRecorder{}
202 events := &eventBusRecorder{}
203 containerCfg := &containerConfigApplyRecorder{}
204
205 registryLimits := ®istryLimitsRecorder{}
206 coord := newReloadCoordinator(v, reloadSvc, proxySvc, registryLimits, events, nil, zerowrap.Default())
207 coord.SetContainerConfigApplier(containerCfg.Apply)
208 require.NoError(t, coord.ApplyLoadedConfig(ctx))
209
210 require.Equal(t, 0, reloadSvc.Calls())
211 require.Equal(t, 1, proxySvc.calls)
212 require.Equal(t, 1, events.Calls())
213 require.Equal(t, 1, registryLimits.calls)
214 require.Equal(t, 1, containerCfg.calls)
215 require.Equal(t, int64(6<<20), registryLimits.maxBlobChunkSize)
216 require.Equal(t, int64(8<<20), registryLimits.maxBlobSize)
217 require.Equal(t, domain.EventConfigReload, events.eventType)
218 require.Equal(t, "reload.example.com", containerCfg.cfg.Server.GordonDomain)
219 require.Equal(t, []string{"old.example.com"}, containerCfg.cfg.Server.LegacyRegistryDomains)
220 require.Equal(t, []string{"docker.io"}, containerCfg.cfg.Images.AllowedRegistries)
221 require.Equal(t, proxy.Config{
222 RegistryDomain: "reload.example.com",
223 RegistryPort: 5000,
224 MaxBodySize: 5 << 20,
225 MaxResponseSize: 7 << 20,
226 MaxConcurrentConns: 99,
227 }, proxySvc.config)
228}
229
230func TestServiceInit_RegisterReloadCoordinatorHooks_WiresContainerConfigApplier(t *testing.T) {
231 ctx := context.Background()
232 v := viper.New()
233 v.Set("server.gordon_domain", "reload.example.com")
234 v.Set("server.registry_port", 5000)
235
236 configSvc := cfgusecase.NewService(v, nil)
237 require.NoError(t, configSvc.Load(ctx))
238
239 si := &serviceInit{
240 ctx: ctx,
241 v: v,
242 cfg: Config{},
243 log: zerowrap.Default(),
244 svc: &services{
245 configSvc: configSvc,
246 containerSvc: container.NewService(nil, nil, nil, nil, container.Config{}, configSvc),
247 internalRegUser: "gordon",
248 internalRegPass: "secret",
249 reloadCoordinator: newReloadCoordinator(v, &reloadRecorder{}, &proxyRecorder{}, nil, nil, nil, zerowrap.Default()),
250 },
251 }
252
253 si.registerReloadCoordinatorHooks()
254 require.NotNil(t, si.svc.reloadCoordinator.applyContainerConfig)
255
256 reloadCfg := Config{}
257 reloadCfg.Server.GordonDomain = "reload.example.com"
258 reloadCfg.Server.RegistryPort = 5000
259 reloadCfg.Images.AllowedRegistries = []string{"docker.io"}
260
261 require.NoError(t, si.svc.reloadCoordinator.applyContainerConfig(ctx, reloadCfg))
262}
263func TestReloadCoordinator_DebouncesRepeatedWatchCallbacks(t *testing.T) {
264 ctx := context.Background()
265 v := viper.New()
266 v.Set("server.gordon_domain", "new.example.com")
267 v.Set("server.registry_port", 5000)
268 v.Set("server.max_proxy_body_size", "5MB")
269 v.Set("server.max_blob_chunk_size", "6MB")
270 v.Set("server.max_blob_size", "8MB")
271 v.Set("server.max_proxy_response_size", "7MB")
272 v.Set("server.max_concurrent_connections", 99)
273
274 reloadSvc := &reloadRecorder{}
275 proxySvc := &proxyRecorder{}
276 events := &eventBusRecorder{}
277
278 registryLimits := ®istryLimitsRecorder{}
279 coord := newReloadCoordinator(v, reloadSvc, proxySvc, registryLimits, events, nil, zerowrap.Default())
280 require.NoError(t, coord.Trigger(ctx))
281 require.NoError(t, coord.Trigger(ctx))
282
283 require.Equal(t, 1, reloadSvc.Calls())
284 require.Equal(t, 1, proxySvc.calls)
285 require.Equal(t, 1, events.Calls())
286 require.Equal(t, 1, registryLimits.calls)
287 require.Equal(t, int64(6<<20), registryLimits.maxBlobChunkSize)
288 require.Equal(t, int64(8<<20), registryLimits.maxBlobSize)
289 require.Equal(t, domain.EventConfigReload, events.eventType)
290 require.Equal(t, proxy.Config{
291 RegistryDomain: "new.example.com",
292 RegistryPort: 5000,
293 MaxBodySize: 5 << 20,
294 MaxResponseSize: 7 << 20,
295 MaxConcurrentConns: 99,
296 }, proxySvc.config)
297}
298
299func TestReloadCoordinator_RetriesImmediatelyAfterFailedReload(t *testing.T) {
300 ctx := context.Background()
301 v := viper.New()
302 v.Set("server.gordon_domain", "retry.example.com")
303 v.Set("server.registry_port", 5000)
304 v.Set("server.max_proxy_body_size", "2MB")
305 v.Set("server.max_blob_chunk_size", "3MB")
306 v.Set("server.max_proxy_response_size", "4MB")
307 v.Set("server.max_concurrent_connections", 11)
308
309 reloadErr := errors.New("reload failed")
310 reloadSvc := &reloadRecorder{err: reloadErr}
311 proxySvc := &proxyRecorder{}
312 coord := newReloadCoordinator(v, reloadSvc, proxySvc, nil, nil, nil, zerowrap.Default())
313
314 err := coord.Trigger(ctx)
315 require.ErrorIs(t, err, reloadErr)
316
317 reloadSvc.mu.Lock()
318 reloadSvc.err = nil
319 reloadSvc.mu.Unlock()
320
321 require.NoError(t, coord.Trigger(ctx))
322 require.Equal(t, 2, reloadSvc.Calls())
323 assert.Equal(t, 1, proxySvc.calls)
324}
325
326func TestReloadCoordinator_PublishErrorDoesNotAdvanceDebounceState(t *testing.T) {
327 ctx := context.Background()
328 v := viper.New()
329 v.Set("server.gordon_domain", "publish.example.com")
330 v.Set("server.registry_port", 5000)
331 v.Set("server.max_proxy_body_size", "5MB")
332 v.Set("server.max_blob_chunk_size", "6MB")
333 v.Set("server.max_proxy_response_size", "7MB")
334 v.Set("server.max_concurrent_connections", 99)
335
336 publishErr := errors.New("publish failed")
337 reloadSvc := &reloadRecorder{}
338 proxySvc := &proxyRecorder{}
339 events := &eventBusRecorder{err: publishErr}
340
341 coord := newReloadCoordinator(v, reloadSvc, proxySvc, nil, events, nil, zerowrap.Default())
342
343 err := coord.Trigger(ctx)
344 require.ErrorIs(t, err, publishErr)
345
346 err = coord.Trigger(ctx)
347 require.ErrorIs(t, err, publishErr)
348
349 require.Equal(t, 2, reloadSvc.Calls())
350 require.Equal(t, 2, proxySvc.calls)
351 require.Equal(t, 2, events.Calls())
352 require.Equal(t, domain.EventConfigReload, events.eventType)
353}
354
355func TestReloadCoordinator_SerializesOverlappingReloadRequests(t *testing.T) {
356 ctx := context.Background()
357 v := viper.New()
358 v.Set("server.gordon_domain", "app.example.com")
359 v.Set("server.registry_port", 5000)
360 v.Set("server.max_proxy_body_size", "2MB")
361 v.Set("server.max_blob_chunk_size", "3MB")
362 v.Set("server.max_proxy_response_size", "4MB")
363 v.Set("server.max_concurrent_connections", 11)
364
365 started := make(chan struct{})
366 release := make(chan struct{})
367 var once sync.Once
368 reloadSvc := &reloadRecorder{onReload: func() {
369 once.Do(func() { close(started) })
370 <-release
371 }}
372 proxySvc := &proxyRecorder{}
373 coord := newReloadCoordinator(v, reloadSvc, proxySvc, nil, nil, nil, zerowrap.Default())
374
375 firstDone := make(chan struct{})
376 go func() {
377 _ = coord.Trigger(ctx)
378 close(firstDone)
379 }()
380
381 select {
382 case <-started:
383 case <-time.After(2 * time.Second):
384 t.Fatal("first reload did not start")
385 }
386
387 secondDone := make(chan struct{})
388 go func() {
389 _ = coord.Trigger(ctx)
390 close(secondDone)
391 }()
392
393 select {
394 case <-secondDone:
395 t.Fatal("second reload finished before the first was released")
396 case <-time.After(75 * time.Millisecond):
397 }
398
399 require.Equal(t, 1, reloadSvc.Calls())
400 close(release)
401
402 select {
403 case <-firstDone:
404 case <-time.After(2 * time.Second):
405 t.Fatal("first reload did not finish")
406 }
407
408 select {
409 case <-secondDone:
410 case <-time.After(2 * time.Second):
411 t.Fatal("second reload did not finish")
412 }
413
414 require.Equal(t, 1, reloadSvc.Calls())
415 assert.Equal(t, 1, proxySvc.calls)
416}