1package app
2
3import (
4 "context"
5 "crypto/tls"
6 "errors"
7 "net"
8 "net/http"
9 "strconv"
10 "sync"
11 "testing"
12 "time"
13
14 "github.com/bnema/zerowrap"
15 "github.com/spf13/viper"
16 "github.com/stretchr/testify/assert"
17 "github.com/stretchr/testify/mock"
18 "github.com/stretchr/testify/require"
19
20 "github.com/bnema/gordon/internal/adapters/in/http/registry"
21 trafficadapter "github.com/bnema/gordon/internal/adapters/in/traffic"
22 inmocks "github.com/bnema/gordon/internal/boundaries/in/mocks"
23 "github.com/bnema/gordon/internal/domain"
24 cfgusecase "github.com/bnema/gordon/internal/usecase/config"
25 "github.com/bnema/gordon/internal/usecase/container"
26 "github.com/bnema/gordon/internal/usecase/proxy"
27 traffic "github.com/bnema/gordon/internal/usecase/traffic"
28)
29
30type watchRecorder struct {
31 called bool
32 onChange func()
33}
34
35func (w *watchRecorder) Watch(ctx context.Context, onChange func()) error {
36 w.called = true
37 w.onChange = onChange
38 return nil
39}
40
41type coordinatorRecorder struct {
42 mu sync.Mutex
43 triggerCalls int
44 applyCalls int
45 err error
46}
47
48func (c *coordinatorRecorder) Trigger(context.Context) error {
49 c.mu.Lock()
50 c.triggerCalls++
51 err := c.err
52 c.mu.Unlock()
53 return err
54}
55
56func (c *coordinatorRecorder) ApplyLoadedConfig(context.Context) error {
57 c.mu.Lock()
58 c.applyCalls++
59 err := c.err
60 c.mu.Unlock()
61 return err
62}
63
64func (c *coordinatorRecorder) TriggerCalls() int {
65 c.mu.Lock()
66 defer c.mu.Unlock()
67 return c.triggerCalls
68}
69
70func (c *coordinatorRecorder) ApplyCalls() int {
71 c.mu.Lock()
72 defer c.mu.Unlock()
73 return c.applyCalls
74}
75
76type reloadRecorder struct {
77 mu sync.Mutex
78 calls int
79 err error
80 onReload func()
81}
82
83func (r *reloadRecorder) Reload(context.Context) error {
84 r.mu.Lock()
85 r.calls++
86 err := r.err
87 onReload := r.onReload
88 r.mu.Unlock()
89
90 if onReload != nil {
91 onReload()
92 }
93
94 return err
95}
96
97func (r *reloadRecorder) Calls() int {
98 r.mu.Lock()
99 defer r.mu.Unlock()
100 return r.calls
101}
102
103type proxyRecorder struct {
104 mu sync.Mutex
105 calls int
106 config proxy.Config
107}
108
109func (p *proxyRecorder) UpdateConfig(config proxy.Config) {
110 p.mu.Lock()
111 defer p.mu.Unlock()
112 p.calls++
113 p.config = config
114}
115
116func (p *proxyRecorder) Calls() int {
117 p.mu.Lock()
118 defer p.mu.Unlock()
119 return p.calls
120}
121
122func (p *proxyRecorder) Config() proxy.Config {
123 p.mu.Lock()
124 defer p.mu.Unlock()
125 return p.config
126}
127
128type registryLimitsRecorder struct {
129 mu sync.Mutex
130 calls int
131 maxBlobChunkSize int64
132 maxBlobSize int64
133}
134
135func (r *registryLimitsRecorder) UpdateBlobLimits(maxBlobChunkSize, maxBlobSize int64) {
136 r.mu.Lock()
137 defer r.mu.Unlock()
138 r.calls++
139 r.maxBlobChunkSize = maxBlobChunkSize
140 r.maxBlobSize = maxBlobSize
141}
142
143func (r *registryLimitsRecorder) Calls() int {
144 r.mu.Lock()
145 defer r.mu.Unlock()
146 return r.calls
147}
148
149func (r *registryLimitsRecorder) Limits() (chunk, size int64) {
150 r.mu.Lock()
151 defer r.mu.Unlock()
152 return r.maxBlobChunkSize, r.maxBlobSize
153}
154
155type containerConfigApplyRecorder struct {
156 calls int
157 ctx context.Context
158 cfg Config
159}
160
161func (r *containerConfigApplyRecorder) Apply(ctx context.Context, cfg Config) error {
162 r.calls++
163 r.ctx = ctx
164 r.cfg = cfg
165 return nil
166}
167
168type publicTLSReconcileRecorder struct {
169 calls int
170}
171
172func (r *publicTLSReconcileRecorder) Reconcile(context.Context) error {
173 r.calls++
174 return nil
175}
176
177type eventBusRecorder struct {
178 mu sync.Mutex
179 calls int
180 eventType domain.EventType
181 payload any
182 err error
183}
184
185func (e *eventBusRecorder) Publish(eventType domain.EventType, payload any) error {
186 e.mu.Lock()
187 e.calls++
188 e.eventType = eventType
189 e.payload = payload
190 err := e.err
191 e.mu.Unlock()
192 return err
193}
194
195func (e *eventBusRecorder) Calls() int {
196 e.mu.Lock()
197 defer e.mu.Unlock()
198 return e.calls
199}
200
201func (e *eventBusRecorder) Type() domain.EventType {
202 e.mu.Lock()
203 defer e.mu.Unlock()
204 return e.eventType
205}
206
207func TestSetupConfigHotReload_WatchCallbackInvokesCoordinator(t *testing.T) {
208 ctx := context.Background()
209 configSvc := &watchRecorder{}
210 coordinator := &coordinatorRecorder{}
211
212 err := setupConfigHotReload(ctx, configSvc, coordinator)
213 require.NoError(t, err)
214 require.True(t, configSvc.called)
215 require.NotNil(t, configSvc.onChange)
216
217 configSvc.onChange()
218 require.Equal(t, 1, coordinator.ApplyCalls())
219 require.Equal(t, 0, coordinator.TriggerCalls())
220}
221
222func TestBuildProxyConfig_ParsesMaxBlobSizeWithoutChangingChunkDefault(t *testing.T) {
223 cfg := Config{}
224 cfg.Server.MaxBlobSize = "2GB"
225
226 result, err := buildProxyConfig(cfg, zerowrap.Default())
227
228 require.NoError(t, err)
229 assert.Equal(t, int64(2<<30), result.maxBlobSize)
230 assert.Equal(t, int64(registry.DefaultMaxBlobChunkSize), result.maxBlobChunkSize)
231}
232
233func TestBuildProxyConfig_DefaultMaxBlobSize(t *testing.T) {
234 result, err := buildProxyConfig(Config{}, zerowrap.Default())
235
236 require.NoError(t, err)
237 assert.Equal(t, int64(registry.DefaultMaxBlobSize), result.maxBlobSize)
238}
239
240func TestReloadCoordinator_ApplyLoadedConfig_RebuildsProxyConfigAndPublishesEvent(t *testing.T) {
241 ctx := context.Background()
242 v := viper.New()
243 v.Set("server.gordon_domain", "reload.example.com")
244 v.Set("server.registry_port", 5000)
245 v.Set("server.max_proxy_body_size", "5MB")
246 v.Set("server.max_blob_chunk_size", "6MB")
247 v.Set("server.max_blob_size", "8MB")
248 v.Set("server.max_proxy_response_size", "7MB")
249 v.Set("server.max_concurrent_connections", 99)
250 v.Set("server.legacy_registry_domains", []string{"old.example.com"})
251 v.Set("images.allowed_registries", []string{"docker.io"})
252
253 reloadSvc := &reloadRecorder{}
254 proxySvc := &proxyRecorder{}
255 events := &eventBusRecorder{}
256 containerCfg := &containerConfigApplyRecorder{}
257
258 registryLimits := ®istryLimitsRecorder{}
259 coord := newReloadCoordinator(v, reloadSvc, proxySvc, registryLimits, events, nil, zerowrap.Default())
260 coord.SetContainerConfigApplier(containerCfg.Apply)
261 require.NoError(t, coord.ApplyLoadedConfig(ctx))
262
263 require.Equal(t, 0, reloadSvc.Calls())
264 require.Equal(t, 1, proxySvc.Calls())
265 require.Equal(t, 1, events.Calls())
266 require.Equal(t, 1, registryLimits.Calls())
267 require.Equal(t, 1, containerCfg.calls)
268 rlChunk, rlSize := registryLimits.Limits()
269 require.Equal(t, int64(6<<20), rlChunk)
270 require.Equal(t, int64(8<<20), rlSize)
271 require.Equal(t, domain.EventConfigReload, events.Type())
272 require.Equal(t, "reload.example.com", containerCfg.cfg.Server.GordonDomain)
273 require.Equal(t, []string{"old.example.com"}, containerCfg.cfg.Server.LegacyRegistryDomains)
274 require.Equal(t, []string{"docker.io"}, containerCfg.cfg.Images.AllowedRegistries)
275 require.Equal(t, proxy.Config{
276 RegistryDomain: "reload.example.com",
277 RegistryPort: 5000,
278 MaxBodySize: 5 << 20,
279 MaxResponseSize: 7 << 20,
280 MaxConcurrentConns: 99,
281 }, proxySvc.Config())
282}
283
284// newTestServices wires the serialized traffic publisher the reload hook
285// requires. Tests that omit app state still exercise the graph apply.
286func newTestServices(svc *services) *services {
287 svc.appTrafficPublisher = newAppTrafficPublisher(svc, Config{})
288 return svc
289}
290
291func TestServiceInit_ReloadRegistersCustomTLSMuxHTTPSFallback(t *testing.T) {
292 ctx := context.Background()
293 v := viper.New()
294 v.Set("server.gordon_domain", "reload.example.com")
295 v.Set("server.registry_port", 5000)
296
297 configSvc := cfgusecase.NewService(v, nil)
298 require.NoError(t, configSvc.Load(ctx))
299 manager := trafficadapter.NewManager()
300 defer func() { require.NoError(t, manager.Shutdown(ctx)) }()
301 httpsHandler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
302 _, _ = w.Write([]byte("reload secure"))
303 })
304 pkiSvc := newTestPKIService(t)
305
306 customAddress := freeTCPAddress(t)
307 si := &serviceInit{
308 ctx: ctx,
309 v: v,
310 cfg: Config{},
311 log: zerowrap.Default(),
312 svc: newTestServices(&services{
313 configSvc: configSvc,
314 containerSvc: container.NewService(nil, nil, nil, nil, container.Config{}),
315 internalRegUser: "gordon",
316 internalRegPass: "secret",
317 reloadCoordinator: newReloadCoordinator(v, &reloadRecorder{}, &proxyRecorder{}, nil, nil, nil, zerowrap.Default()),
318 trafficManager: manager,
319 httpsProxyHandler: httpsHandler,
320 pkiSvc: pkiSvc,
321 }),
322 }
323 si.registerReloadCoordinatorHooks()
324
325 reloadCfg := Config{}
326 reloadCfg.Server.GordonDomain = "reload.example.com"
327 reloadCfg.Server.RegistryPort = 5000
328 reloadCfg.Server.TLSPort = freeTCPPort(t)
329 reloadCfg.EntryPoints = map[string]traffic.EntryPointConfig{
330 "custom-secure": {Address: customAddress, Protocol: domain.EntryPointProtocolTLSMux},
331 }
332
333 require.NoError(t, si.svc.reloadCoordinator.applyContainerConfig(ctx, reloadCfg))
334 managementCert, err := pkiSvc.GetCertificate(&tls.ClientHelloInfo{ServerName: "reload.example.com"})
335 require.NoError(t, err)
336 require.NotNil(t, managementCert)
337
338 _, portValue, err := net.SplitHostPort(customAddress)
339 require.NoError(t, err)
340 port, err := strconv.Atoi(portValue)
341 require.NoError(t, err)
342 body := httpsGetForTest(t, port, "app.example.com")
343 require.Contains(t, body, "reload secure")
344}
345
346func TestServiceInit_ReloadUpdatesManagementHosts(t *testing.T) {
347 ctx := t.Context()
348 v := viper.New()
349 v.Set("server.gordon_domain", "old.example.com")
350 v.Set("server.registry_port", 5000)
351
352 configSvc := cfgusecase.NewService(v, nil)
353 require.NoError(t, configSvc.Load(ctx))
354 publicTLS := inmocks.NewMockPublicTLSService(t)
355 publicTLS.EXPECT().SetAdditionalHosts(mock.Anything, []string{"new.example.com"}).Once()
356
357 si := &serviceInit{
358 ctx: ctx,
359 v: v,
360 cfg: Config{},
361 log: zerowrap.Default(),
362 svc: newTestServices(&services{
363 configSvc: configSvc,
364 containerSvc: container.NewService(nil, nil, nil, nil, container.Config{}),
365 internalRegUser: "gordon",
366 internalRegPass: "secret",
367 reloadCoordinator: newReloadCoordinator(v, &reloadRecorder{}, &proxyRecorder{}, nil, nil, publicTLS, zerowrap.Default()),
368 publicTLSSvc: publicTLS,
369 }),
370 }
371 si.registerReloadCoordinatorHooks()
372
373 reloadCfg := Config{}
374 reloadCfg.Server.GordonDomain = "new.example.com"
375 reloadCfg.Server.RegistryPort = 5000
376
377 require.NoError(t, si.svc.reloadCoordinator.applyContainerConfig(ctx, reloadCfg))
378}
379
380func TestServiceInit_RegisterReloadCoordinatorHooks_WiresContainerConfigApplier(t *testing.T) {
381 ctx := context.Background()
382 v := viper.New()
383 v.Set("server.gordon_domain", "reload.example.com")
384 v.Set("server.registry_port", 5000)
385
386 configSvc := cfgusecase.NewService(v, nil)
387 require.NoError(t, configSvc.Load(ctx))
388
389 si := &serviceInit{
390 ctx: ctx,
391 v: v,
392 cfg: Config{},
393 log: zerowrap.Default(),
394 svc: newTestServices(&services{
395 configSvc: configSvc,
396 containerSvc: container.NewService(nil, nil, nil, nil, container.Config{}),
397 internalRegUser: "gordon",
398 internalRegPass: "secret",
399 reloadCoordinator: newReloadCoordinator(v, &reloadRecorder{}, &proxyRecorder{}, nil, nil, nil, zerowrap.Default()),
400 }),
401 }
402
403 si.registerReloadCoordinatorHooks()
404 require.NotNil(t, si.svc.reloadCoordinator.applyContainerConfig)
405
406 reloadCfg := Config{}
407 reloadCfg.Server.GordonDomain = "reload.example.com"
408 reloadCfg.Server.RegistryPort = 5000
409 reloadCfg.Images.AllowedRegistries = []string{"docker.io"}
410
411 require.NoError(t, si.svc.reloadCoordinator.applyContainerConfig(ctx, reloadCfg))
412}
413
414func TestServiceInit_RegisterReloadCoordinatorHooks_AppliesTraffic(t *testing.T) {
415 ctx := context.Background()
416 v := viper.New()
417 v.Set("server.gordon_domain", "reload.example.com")
418 v.Set("server.registry_port", 5000)
419
420 configSvc := cfgusecase.NewService(v, nil)
421 require.NoError(t, configSvc.Load(ctx))
422 manager := trafficadapter.NewManager()
423 defer func() { require.NoError(t, manager.Shutdown(ctx)) }()
424
425 si := &serviceInit{
426 ctx: ctx,
427 v: v,
428 cfg: Config{},
429 log: zerowrap.Default(),
430 svc: newTestServices(&services{
431 configSvc: configSvc,
432 containerSvc: container.NewService(nil, nil, nil, nil, container.Config{}),
433 internalRegUser: "gordon",
434 internalRegPass: "secret",
435 reloadCoordinator: newReloadCoordinator(v, &reloadRecorder{}, &proxyRecorder{}, nil, nil, nil, zerowrap.Default()),
436 trafficManager: manager,
437 }),
438 }
439 si.registerReloadCoordinatorHooks()
440
441 reloadCfg := Config{}
442 reloadCfg.Server.GordonDomain = "reload.example.com"
443 reloadCfg.Server.RegistryPort = 5000
444 reloadCfg.EntryPoints = map[string]traffic.EntryPointConfig{"game": {Address: freeTCPAddress(t), Protocol: domain.EntryPointProtocolTCP}}
445 reloadCfg.NetworkServices = []traffic.NetworkServiceConfig{{Name: "game", Ports: []traffic.PortConfig{{Name: "game", Container: 28015, Protocol: domain.NetworkProtocolTCP}}}}
446 reloadCfg.Traffic.TCP.Routers = []traffic.RouterConfig{{Name: "game", EntryPoint: "game", Service: "network_service:game:game"}}
447
448 require.NoError(t, si.svc.reloadCoordinator.applyContainerConfig(ctx, reloadCfg))
449 assert.NotEmpty(t, manager.Status().EntryPoints)
450}
451
452func TestWaitForCoreProxyReady_WaitsWithoutPublishingEmptyTraffic(t *testing.T) {
453 ctx := context.Background()
454 manager := trafficadapter.NewManager()
455 defer func() { require.NoError(t, manager.Shutdown(ctx)) }()
456
457 ready := make(chan struct{})
458 close(ready)
459 require.NoError(t, waitForCoreProxyReady(ctx, ready, ready, nil))
460 assert.Empty(t, manager.Status().EntryPoints)
461}
462
463func TestWaitForCoreProxyReady_ObservesContextCancellation(t *testing.T) {
464 ctx, cancel := context.WithCancel(context.Background())
465 cancel()
466
467 // Neither readiness channel ever closes: the canceled startup context must
468 // unblock the wait rather than hang on a listener that will never bind.
469 err := waitForCoreProxyReady(ctx, make(chan struct{}), make(chan struct{}), make(chan error))
470 require.ErrorIs(t, err, context.Canceled)
471}
472
473func TestReloadCoordinator_StopCancelsOwnedTrailingReloadLifecycle(t *testing.T) {
474 ctx := context.Background()
475 v := viper.New()
476 v.Set("server.gordon_domain", "lifecycle.example.com")
477 v.Set("server.registry_port", 5000)
478
479 reloadSvc := &reloadRecorder{}
480 proxySvc := &proxyRecorder{}
481 coord := newReloadCoordinator(v, reloadSvc, proxySvc, nil, nil, nil, zerowrap.Default())
482 // Keep the trailing timer pending so Stop is the only thing that can tear
483 // the coordinator-owned lifecycle down.
484 coord.debounce = time.Hour
485
486 require.NoError(t, coord.Trigger(ctx)) // immediate apply records lastRun
487 require.NoError(t, coord.Trigger(ctx)) // coalesced into the trailing timer
488
489 coord.mu.Lock()
490 lifecycleCtx := coord.lifecycleCtx
491 coord.mu.Unlock()
492 require.NotNil(t, lifecycleCtx, "trailing reload must own a lifecycle context")
493 require.NoError(t, lifecycleCtx.Err())
494
495 coord.Stop()
496 require.ErrorIs(t, lifecycleCtx.Err(), context.Canceled)
497 require.Equal(t, 1, proxySvc.Calls(), "stopped coordinator must not apply the trailing reload")
498}
499
500func TestReloadCoordinator_DebouncesRepeatedWatchCallbacks(t *testing.T) {
501 ctx := context.Background()
502 v := viper.New()
503 v.Set("server.gordon_domain", "new.example.com")
504 v.Set("server.registry_port", 5000)
505 v.Set("server.max_proxy_body_size", "5MB")
506 v.Set("server.max_blob_chunk_size", "6MB")
507 v.Set("server.max_blob_size", "8MB")
508 v.Set("server.max_proxy_response_size", "7MB")
509 v.Set("server.max_concurrent_connections", 99)
510
511 reloadSvc := &reloadRecorder{}
512 proxySvc := &proxyRecorder{}
513 events := &eventBusRecorder{}
514
515 registryLimits := ®istryLimitsRecorder{}
516 coord := newReloadCoordinator(v, reloadSvc, proxySvc, registryLimits, events, nil, zerowrap.Default())
517 coord.debounce = 20 * time.Millisecond
518 require.NoError(t, coord.Trigger(ctx))
519 require.NoError(t, coord.Trigger(ctx))
520 require.NoError(t, coord.Trigger(ctx))
521
522 // Observe the trailing reload through the recorders' own locks instead of
523 // locking the coordinator: the reload event is published last, so waiting
524 // for all recorders proves the trailing reload finished applying.
525 require.Eventually(t, func() bool {
526 return reloadSvc.Calls() == 2 && proxySvc.Calls() == 2 &&
527 events.Calls() == 2 && registryLimits.Calls() == 2
528 }, time.Second, 5*time.Millisecond)
529 // No further reload may fire: the debounce window is over.
530 require.Never(t, func() bool {
531 return proxySvc.Calls() > 2 || events.Calls() > 2
532 }, 50*time.Millisecond, 5*time.Millisecond)
533 require.Equal(t, 2, proxySvc.Calls())
534 require.Equal(t, 2, events.Calls())
535 require.Equal(t, 2, registryLimits.Calls())
536 rlChunk, rlSize := registryLimits.Limits()
537 require.Equal(t, int64(6<<20), rlChunk)
538 require.Equal(t, int64(8<<20), rlSize)
539 require.Equal(t, domain.EventConfigReload, events.Type())
540 require.Equal(t, proxy.Config{
541 RegistryDomain: "new.example.com",
542 RegistryPort: 5000,
543 MaxBodySize: 5 << 20,
544 MaxResponseSize: 7 << 20,
545 MaxConcurrentConns: 99,
546 }, proxySvc.Config())
547}
548
549func TestSetupConfigHotReload_CoalescesRepeatedWatcherCallbacks(t *testing.T) {
550 ctx := context.Background()
551 v := viper.New()
552 v.Set("server.gordon_domain", "watch.example.com")
553 v.Set("server.registry_port", 5000)
554
555 configSvc := &watchRecorder{}
556 reloadSvc := &reloadRecorder{}
557 proxySvc := &proxyRecorder{}
558 coord := newReloadCoordinator(v, reloadSvc, proxySvc, nil, nil, nil, zerowrap.Default())
559 coord.debounce = 20 * time.Millisecond
560 defer coord.Stop()
561
562 require.NoError(t, setupConfigHotReload(ctx, configSvc, coord))
563 require.NotNil(t, configSvc.onChange)
564
565 // fsnotify delivers a burst of events for a single edit; each callback
566 // must not apply the config on its own.
567 configSvc.onChange()
568 configSvc.onChange()
569 configSvc.onChange()
570
571 require.Eventually(t, func() bool { return proxySvc.Calls() == 2 }, time.Second, 5*time.Millisecond)
572 require.Never(t, func() bool { return proxySvc.Calls() > 2 }, 50*time.Millisecond, 5*time.Millisecond)
573 // ApplyLoadedConfig applies what the watcher already loaded: the trailing
574 // apply must not read the config again.
575 require.Zero(t, reloadSvc.Calls())
576 require.Equal(t, "watch.example.com", proxySvc.Config().RegistryDomain)
577}
578
579func TestReloadCoordinator_TrailingReloadAppliesFinalState(t *testing.T) {
580 ctx := context.Background()
581 v := viper.New()
582 v.Set("server.gordon_domain", "initial.example.com")
583 v.Set("server.registry_port", 5000)
584
585 reloadSvc := &reloadRecorder{}
586 proxySvc := &proxyRecorder{}
587 coord := newReloadCoordinator(v, reloadSvc, proxySvc, nil, nil, nil, zerowrap.Default())
588 coord.debounce = 20 * time.Millisecond
589
590 require.NoError(t, coord.Trigger(ctx))
591 v.Set("server.gordon_domain", "intermediate.example.com")
592 require.NoError(t, coord.Trigger(ctx))
593 v.Set("server.gordon_domain", "final.example.com")
594 require.NoError(t, coord.Trigger(ctx))
595
596 // The trailing reload is the only writer of the final state: wait for
597 // the observable outcome instead of locking the coordinator.
598 require.Eventually(t, func() bool {
599 return proxySvc.Config().RegistryDomain == "final.example.com"
600 }, time.Second, 5*time.Millisecond)
601}
602
603func TestReloadCoordinator_InvalidReloadKeepsActiveConfigThenAcceptsValidChange(t *testing.T) {
604 ctx := context.Background()
605 v := viper.New()
606 v.Set("server.gordon_domain", "active.example.com")
607 v.Set("server.registry_port", 5000)
608
609 reloadSvc := &reloadRecorder{}
610 proxySvc := &proxyRecorder{}
611 coord := newReloadCoordinator(v, reloadSvc, proxySvc, nil, nil, nil, zerowrap.Default())
612 coord.debounce = 10 * time.Millisecond
613
614 require.NoError(t, coord.Trigger(ctx))
615 require.Equal(t, "active.example.com", proxySvc.Config().RegistryDomain)
616
617 time.Sleep(coord.debounce)
618 v.Set("server.max_proxy_body_size", "invalid")
619 require.Error(t, coord.Trigger(ctx))
620 require.Equal(t, "active.example.com", proxySvc.Config().RegistryDomain)
621
622 v.Set("server.max_proxy_body_size", "5MB")
623 v.Set("server.gordon_domain", "recovered.example.com")
624 require.NoError(t, coord.Trigger(ctx))
625 require.Equal(t, "recovered.example.com", proxySvc.Config().RegistryDomain)
626}
627
628func TestReloadCoordinator_RetriesImmediatelyAfterFailedReload(t *testing.T) {
629 ctx := context.Background()
630 v := viper.New()
631 v.Set("server.gordon_domain", "retry.example.com")
632 v.Set("server.registry_port", 5000)
633 v.Set("server.max_proxy_body_size", "2MB")
634 v.Set("server.max_blob_chunk_size", "3MB")
635 v.Set("server.max_proxy_response_size", "4MB")
636 v.Set("server.max_concurrent_connections", 11)
637
638 reloadErr := errors.New("reload failed")
639 reloadSvc := &reloadRecorder{err: reloadErr}
640 proxySvc := &proxyRecorder{}
641 coord := newReloadCoordinator(v, reloadSvc, proxySvc, nil, nil, nil, zerowrap.Default())
642
643 err := coord.Trigger(ctx)
644 require.ErrorIs(t, err, reloadErr)
645
646 reloadSvc.mu.Lock()
647 reloadSvc.err = nil
648 reloadSvc.mu.Unlock()
649
650 require.NoError(t, coord.Trigger(ctx))
651 require.Equal(t, 2, reloadSvc.Calls())
652 assert.Equal(t, 1, proxySvc.Calls())
653}
654
655func TestReloadCoordinator_PublishErrorDoesNotAdvanceDebounceState(t *testing.T) {
656 ctx := context.Background()
657 v := viper.New()
658 v.Set("server.gordon_domain", "publish.example.com")
659 v.Set("server.registry_port", 5000)
660 v.Set("server.max_proxy_body_size", "5MB")
661 v.Set("server.max_blob_chunk_size", "6MB")
662 v.Set("server.max_proxy_response_size", "7MB")
663 v.Set("server.max_concurrent_connections", 99)
664
665 publishErr := errors.New("publish failed")
666 reloadSvc := &reloadRecorder{}
667 proxySvc := &proxyRecorder{}
668 events := &eventBusRecorder{err: publishErr}
669 publicTLS := &publicTLSReconcileRecorder{}
670
671 coord := newReloadCoordinator(v, reloadSvc, proxySvc, nil, events, publicTLS, zerowrap.Default())
672
673 err := coord.Trigger(ctx)
674 require.ErrorIs(t, err, publishErr)
675
676 err = coord.Trigger(ctx)
677 require.ErrorIs(t, err, publishErr)
678
679 require.Equal(t, 2, reloadSvc.Calls())
680 require.Equal(t, 2, proxySvc.Calls())
681 require.Equal(t, 2, events.Calls())
682 require.Equal(t, 2, publicTLS.calls)
683 require.Equal(t, domain.EventConfigReload, events.Type())
684}
685
686func TestReloadCoordinator_SerializesOverlappingReloadRequests(t *testing.T) {
687 ctx := context.Background()
688 v := viper.New()
689 v.Set("server.gordon_domain", "app.example.com")
690 v.Set("server.registry_port", 5000)
691 v.Set("server.max_proxy_body_size", "2MB")
692 v.Set("server.max_blob_chunk_size", "3MB")
693 v.Set("server.max_proxy_response_size", "4MB")
694 v.Set("server.max_concurrent_connections", 11)
695
696 started := make(chan struct{})
697 release := make(chan struct{})
698 var once sync.Once
699 reloadSvc := &reloadRecorder{onReload: func() {
700 once.Do(func() { close(started) })
701 <-release
702 }}
703 proxySvc := &proxyRecorder{}
704 coord := newReloadCoordinator(v, reloadSvc, proxySvc, nil, nil, nil, zerowrap.Default())
705 coord.debounce = 20 * time.Millisecond
706
707 firstDone := make(chan struct{})
708 go func() {
709 _ = coord.Trigger(ctx)
710 close(firstDone)
711 }()
712
713 select {
714 case <-started:
715 case <-time.After(2 * time.Second):
716 t.Fatal("first reload did not start")
717 }
718
719 secondDone := make(chan struct{})
720 go func() {
721 _ = coord.Trigger(ctx)
722 close(secondDone)
723 }()
724
725 select {
726 case <-secondDone:
727 t.Fatal("second reload finished before the first was released")
728 case <-time.After(75 * time.Millisecond):
729 }
730
731 require.Equal(t, 1, reloadSvc.Calls())
732 close(release)
733
734 select {
735 case <-firstDone:
736 case <-time.After(2 * time.Second):
737 t.Fatal("first reload did not finish")
738 }
739
740 select {
741 case <-secondDone:
742 case <-time.After(2 * time.Second):
743 t.Fatal("second reload did not finish")
744 }
745
746 require.Eventually(t, func() bool { return proxySvc.Calls() == 2 }, time.Second, 5*time.Millisecond)
747 require.Never(t, func() bool { return proxySvc.Calls() > 2 }, 50*time.Millisecond, 5*time.Millisecond)
748 assert.Equal(t, 2, proxySvc.Calls())
749}