run_reload_test.go Raw History Folder

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

run_reload_test.go Raw History Folder
