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