1package app
2
3import (
4 "context"
5 "sync"
6 "testing"
7 "time"
8
9 "github.com/bnema/zerowrap"
10 "github.com/stretchr/testify/assert"
11 "github.com/stretchr/testify/mock"
12 "github.com/stretchr/testify/require"
13
14 outmocks "github.com/bnema/gordon/internal/boundaries/out/mocks"
15 "github.com/bnema/gordon/internal/domain"
16 "github.com/bnema/gordon/internal/usecase/apptraffic"
17)
18
19func httpAppSpec() domain.AppService {
20 return domain.AppService{
21 Name: "web",
22 HTTP: []domain.AppHTTPInterface{{Host: "blog.example.com", Port: 8080, TLS: "auto"}},
23 }
24}
25
26// newPublisherServices wires the minimum app wiring the publisher needs:
27// a state source, the ACTIVE-derived host index, and the activator.
28func newPublisherServices(state *outmocks.MockAppState) (*services, *apptraffic.HostIndex) {
29 index := apptraffic.NewHostIndex()
30 return &services{
31 appState: state,
32 appActivator: apptraffic.NewActivator(zerowrap.Default()),
33 appHostIndex: index,
34 }, index
35}
36
37func TestAppTrafficPublisher_WithdrawServiceIsFailClosedInStateAndIndex(t *testing.T) {
38 ctx := context.Background()
39 state := outmocks.NewMockAppState(t)
40 svc, index := newPublisherServices(state)
41
42 served := domain.AppActive{App: "blog", Services: map[string]domain.AppEffectiveService{
43 "web": {Container: "c-1", Spec: httpAppSpec(), BackendBinds: map[int]int{8080: 32770}},
44 }}
45 withdrawn := domain.AppActive{App: "blog", Services: map[string]domain.AppEffectiveService{
46 "web": {Container: "c-1", Spec: httpAppSpec()},
47 }}
48 state.EXPECT().LoadActive(mock.Anything, "blog").Return(served, true, nil).Once()
49 state.EXPECT().SaveActive(mock.Anything, mock.MatchedBy(func(active domain.AppActive) bool {
50 service := active.Services["web"]
51 return service.BackendBinds == nil && service.UDPBackendBinds == nil
52 })).Return(nil).Once()
53 state.EXPECT().ListApps(mock.Anything).Return([]string{"blog"}, nil).Once()
54 state.EXPECT().LoadIntent(mock.Anything, "blog").Return(domain.AppStopIntent{App: "blog"}, nil).Once()
55 state.EXPECT().LoadActive(mock.Anything, "blog").Return(withdrawn, true, nil).Once()
56
57 // Serve the host first so the withdrawal has something to remove.
58 require.NoError(t, svc.appActivator.RebuildHostIndex(ctx, index, stubAppState{active: served}, nil))
59 backend, ok := index.LookupHost("blog.example.com")
60 require.True(t, ok)
61 require.True(t, backend.Resolved(), "precondition: the host resolves to a loopback backend")
62
63 publisher := newAppTrafficPublisher(svc, Config{})
64 require.NoError(t, publisher.WithdrawService(ctx, "blog", "web"))
65
66 // The proxy fails closed on an unresolved backend (proxy/service.go
67 // resolveAppTarget), so an unresolved projection is a withdrawn one.
68 backend, ok = index.LookupHost("blog.example.com")
69 if ok {
70 assert.False(t, backend.Resolved(), "a withdrawn service must not resolve to a backend")
71 }
72}
73
74func TestAppTrafficPublisher_WithdrawServiceSurfacesStateFailure(t *testing.T) {
75 ctx := context.Background()
76 state := outmocks.NewMockAppState(t)
77 svc, index := newPublisherServices(state)
78
79 served := domain.AppActive{App: "blog", Services: map[string]domain.AppEffectiveService{
80 "web": {Container: "c-1", Spec: httpAppSpec(), BackendBinds: map[int]int{8080: 32770}},
81 }}
82 state.EXPECT().LoadActive(mock.Anything, "blog").Return(served, true, nil).Once()
83 state.EXPECT().SaveActive(mock.Anything, mock.Anything).Return(assert.AnError).Once()
84
85 require.NoError(t, svc.appActivator.RebuildHostIndex(ctx, index, stubAppState{active: served}, nil))
86
87 publisher := newAppTrafficPublisher(svc, Config{})
88 err := publisher.WithdrawService(ctx, "blog", "web")
89 require.Error(t, err)
90 assert.Contains(t, err.Error(), "persist fail-closed state")
91 // The projection could not be rebuilt, so stale forwarding may remain:
92 // the caller keeps its publication inhibition and retries.
93 _, ok := index.LookupHost("blog.example.com")
94 assert.True(t, ok)
95}
96
97func TestAppTrafficPublisher_WithdrawUnknownServiceIsANoop(t *testing.T) {
98 ctx := context.Background()
99 state := outmocks.NewMockAppState(t)
100 svc, _ := newPublisherServices(state)
101
102 state.EXPECT().LoadActive(mock.Anything, "blog").Return(domain.AppActive{App: "blog"}, true, nil).Once()
103 state.EXPECT().ListApps(mock.Anything).Return([]string{"blog"}, nil).Once()
104 state.EXPECT().LoadIntent(mock.Anything, "blog").Return(domain.AppStopIntent{App: "blog"}, nil).Once()
105 state.EXPECT().LoadActive(mock.Anything, "blog").Return(domain.AppActive{App: "blog"}, true, nil).Once()
106
107 publisher := newAppTrafficPublisher(svc, Config{})
108 require.NoError(t, publisher.WithdrawService(ctx, "blog", "web"))
109}
110
111func TestAppTrafficPublisher_RebuildTrafficReprojectsVerifiedState(t *testing.T) {
112 ctx := context.Background()
113 state := outmocks.NewMockAppState(t)
114 svc, index := newPublisherServices(state)
115
116 served := domain.AppActive{App: "blog", Services: map[string]domain.AppEffectiveService{
117 "web": {Container: "c-1", Spec: httpAppSpec(), BackendBinds: map[int]int{8080: 32770}},
118 }}
119 state.EXPECT().ListApps(mock.Anything).Return([]string{"blog"}, nil).Once()
120 state.EXPECT().LoadIntent(mock.Anything, "blog").Return(domain.AppStopIntent{App: "blog"}, nil).Once()
121 state.EXPECT().LoadActive(mock.Anything, "blog").Return(served, true, nil).Once()
122
123 publisher := newAppTrafficPublisher(svc, Config{})
124 require.NoError(t, publisher.RebuildTraffic(ctx))
125
126 backend, ok := index.LookupHost("blog.example.com")
127 require.True(t, ok)
128 assert.Equal(t, 32770, backend.Port)
129}
130
131func TestAppTrafficPublisher_SerializesConcurrentPublication(t *testing.T) {
132 ctx := context.Background()
133 state := outmocks.NewMockAppState(t)
134 svc, index := newPublisherServices(state)
135
136 served := domain.AppActive{App: "blog", Services: map[string]domain.AppEffectiveService{
137 "web": {Container: "c-1", Spec: httpAppSpec(), BackendBinds: map[int]int{8080: 32770}},
138 }}
139 state.EXPECT().LoadActive(mock.Anything, "blog").Return(served, true, nil)
140 state.EXPECT().SaveActive(mock.Anything, mock.Anything).Return(nil)
141 state.EXPECT().ListApps(mock.Anything).Return([]string{"blog"}, nil)
142 state.EXPECT().LoadIntent(mock.Anything, "blog").Return(domain.AppStopIntent{App: "blog"}, nil)
143
144 publisher := newAppTrafficPublisher(svc, Config{})
145 done := make(chan struct{})
146 for range 4 {
147 go func() {
148 defer func() { done <- struct{}{} }()
149 _ = publisher.RebuildTraffic(ctx)
150 _ = publisher.WithdrawService(ctx, "blog", "web")
151 }()
152 }
153 for range 4 {
154 <-done
155 }
156 assert.NotNil(t, index)
157}
158
159// stubAppState is a read-only app state double for index projections.
160// The zero value reports running intent.
161type stubAppState struct {
162 active domain.AppActive
163 stopped bool
164}
165
166func (s stubAppState) ListApps(context.Context) ([]string, error) {
167 return []string{s.active.App}, nil
168}
169
170func (s stubAppState) LoadActive(context.Context, string) (domain.AppActive, bool, error) {
171 return s.active, true, nil
172}
173
174func (s stubAppState) LoadIntent(context.Context, string) (domain.AppStopIntent, error) {
175 return domain.AppStopIntent{App: s.active.App, Stopped: s.stopped}, nil
176}
177
178func TestAppTrafficPublisher_WithdrawalIsIdempotent(t *testing.T) {
179 ctx := context.Background()
180 state := outmocks.NewMockAppState(t)
181 svc, _ := newPublisherServices(state)
182
183 alreadyWithdrawn := domain.AppActive{App: "blog", Services: map[string]domain.AppEffectiveService{
184 "web": {Container: "c-1", Spec: httpAppSpec()},
185 }}
186 state.EXPECT().LoadActive(mock.Anything, "blog").Return(alreadyWithdrawn, true, nil).Once()
187 state.EXPECT().ListApps(mock.Anything).Return([]string{"blog"}, nil).Once()
188 state.EXPECT().LoadIntent(mock.Anything, "blog").Return(domain.AppStopIntent{App: "blog"}, nil).Once()
189 state.EXPECT().LoadActive(mock.Anything, "blog").Return(alreadyWithdrawn, true, nil).Once()
190
191 publisher := newAppTrafficPublisher(svc, Config{})
192 require.NoError(t, publisher.WithdrawService(ctx, "blog", "web"))
193}
194
195func TestAppMonitor_RunsOnePassPerTick(t *testing.T) {
196 reconciler := &fakeReconciler{}
197 monitor := newAppMonitor(reconciler, zerowrap.Default())
198 ticks := make(chan time.Time, 8)
199 monitor.ticks = func() <-chan time.Time { return ticks }
200 monitor.Start(context.Background())
201
202 ticks <- time.Now()
203 require.Eventually(t, func() bool { return reconciler.callCount() == 1 }, time.Second, time.Millisecond)
204 ticks <- time.Now()
205 require.Eventually(t, func() bool { return reconciler.callCount() == 2 }, time.Second, time.Millisecond)
206
207 monitor.Stop()
208 assert.Equal(t, 2, reconciler.callCount())
209}
210
211func TestAppMonitor_DoesNotOverlapPasses(t *testing.T) {
212 reconciler := &fakeReconciler{enter: make(chan struct{}), release: make(chan struct{})}
213 monitor := newAppMonitor(reconciler, zerowrap.Default())
214 ticks := make(chan time.Time, 8)
215 monitor.ticks = func() <-chan time.Time { return ticks }
216 monitor.Start(context.Background())
217
218 ticks <- time.Now()
219 <-reconciler.entered()
220 // Additional ticks while a pass is in flight queue at most one
221 // further pass, never a concurrent one.
222 for range 3 {
223 ticks <- time.Now()
224 }
225 assert.Equal(t, 1, reconciler.callCount())
226 assert.Equal(t, 1, reconciler.maxConcurrent())
227
228 close(reconciler.release)
229 monitor.Stop()
230 assert.GreaterOrEqual(t, reconciler.callCount(), 1)
231}
232
233func TestAppMonitor_ReconciliationErrorDoesNotStopTheLoop(t *testing.T) {
234 reconciler := &fakeReconciler{err: assert.AnError}
235 monitor := newAppMonitor(reconciler, zerowrap.Default())
236 ticks := make(chan time.Time, 8)
237 monitor.ticks = func() <-chan time.Time { return ticks }
238 monitor.Start(context.Background())
239
240 ticks <- time.Now()
241 ticks <- time.Now()
242 require.Eventually(t, func() bool { return reconciler.callCount() == 2 }, time.Second, time.Millisecond)
243 monitor.Stop()
244}
245
246func TestAppMonitor_StartIsIdempotent(t *testing.T) {
247 reconciler := &fakeReconciler{}
248 monitor := newAppMonitor(reconciler, zerowrap.Default())
249 ticks := make(chan time.Time, 8)
250 monitor.ticks = func() <-chan time.Time { return ticks }
251
252 monitor.Start(context.Background())
253 monitor.Start(context.Background())
254 monitor.Start(context.Background())
255
256 ticks <- time.Now()
257 require.Eventually(t, func() bool { return reconciler.callCount() == 1 }, time.Second, time.Millisecond)
258 monitor.Stop()
259 assert.Equal(t, 1, reconciler.callCount(), "reload must never create a second loop")
260}
261
262func TestAppMonitor_StopCancelsAndJoins(t *testing.T) {
263 blocking := &fakeReconciler{enter: make(chan struct{}), release: make(chan struct{})}
264 monitor := newAppMonitor(blocking, zerowrap.Default())
265 ticks := make(chan time.Time, 8)
266 monitor.ticks = func() <-chan time.Time { return ticks }
267 monitor.Start(context.Background())
268
269 ticks <- time.Now()
270 <-blocking.entered()
271
272 stopped := make(chan struct{})
273 go func() {
274 monitor.Stop()
275 close(stopped)
276 }()
277 select {
278 case <-stopped:
279 t.Fatal("Stop returned while a bounded pass was still running")
280 case <-time.After(30 * time.Millisecond):
281 }
282 close(blocking.release)
283 select {
284 case <-stopped:
285 case <-time.After(time.Second):
286 t.Fatal("Stop never joined the loop")
287 }
288 monitor.Stop() // idempotent
289}
290
291func TestAppMonitor_NilReconcilerNeverStarts(t *testing.T) {
292 monitor := newAppMonitor(nil, zerowrap.Default())
293 monitor.Start(context.Background())
294 monitor.Stop()
295}
296
297// fakeReconciler records passes and can hold one open.
298type fakeReconciler struct {
299 mu sync.Mutex
300 calls int
301 inflight int
302 maxInflight int
303 enter chan struct{}
304 release chan struct{}
305 enteredOnce sync.Once
306 err error
307}
308
309func (f *fakeReconciler) ReconcileRunning(context.Context) error {
310 f.mu.Lock()
311 f.calls++
312 f.inflight++
313 if f.inflight > f.maxInflight {
314 f.maxInflight = f.inflight
315 }
316 f.mu.Unlock()
317 if f.enter != nil {
318 f.enteredOnce.Do(func() { close(f.enter) })
319 <-f.release
320 }
321 f.mu.Lock()
322 f.inflight--
323 f.mu.Unlock()
324 return f.err
325}
326
327func (f *fakeReconciler) callCount() int {
328 f.mu.Lock()
329 defer f.mu.Unlock()
330 return f.calls
331}
332
333func (f *fakeReconciler) maxConcurrent() int {
334 f.mu.Lock()
335 defer f.mu.Unlock()
336 return f.maxInflight
337}
338
339func (f *fakeReconciler) entered() <-chan struct{} { return f.enter }
340
341// TestAppTrafficPublisher_RebuildSkipsStoppedApps proves durable stopped
342// intent is honored by every publication, including the plain rebuild a
343// stopped-app convergence pass performs: a stopped service is never
344// projected back into the proxy index.
345func TestAppTrafficPublisher_RebuildSkipsStoppedApps(t *testing.T) {
346 ctx := context.Background()
347 state := outmocks.NewMockAppState(t)
348 svc, index := newPublisherServices(state)
349
350 served := domain.AppActive{App: "blog", Services: map[string]domain.AppEffectiveService{
351 "web": {Container: "c-1", Spec: httpAppSpec(), BackendBinds: map[int]int{8080: 32770}},
352 }}
353 // Precondition: the host is routable while the app runs.
354 require.NoError(t, svc.appActivator.RebuildHostIndex(ctx, index, stubAppState{active: served}, nil))
355 backend, ok := index.LookupHost("blog.example.com")
356 require.True(t, ok)
357 require.True(t, backend.Resolved())
358
359 state.EXPECT().ListApps(mock.Anything).Return([]string{"blog"}, nil).Once()
360 state.EXPECT().LoadIntent(mock.Anything, "blog").
361 Return(domain.AppStopIntent{App: "blog", Stopped: true}, nil).Once()
362
363 publisher := newAppTrafficPublisher(svc, Config{})
364 require.NoError(t, publisher.RebuildTraffic(ctx))
365
366 backend, ok = index.LookupHost("blog.example.com")
367 if ok {
368 assert.False(t, backend.Resolved(), "a stopped app must not stay routable")
369 }
370}
371
372// TestAppTrafficPublisher_AdmissionHonorsContextCancellation proves a
373// caller whose deadline expired never waits behind an in-flight
374// publication: a monitor pass must not hold its app lock for a graph
375// application someone else started, and shutdown must not block on it.
376func TestAppTrafficPublisher_AdmissionHonorsContextCancellation(t *testing.T) {
377 state := outmocks.NewMockAppState(t)
378 svc, _ := newPublisherServices(state)
379
380 release := make(chan struct{})
381 entered := make(chan struct{})
382 state.EXPECT().ListApps(mock.Anything).RunAndReturn(func(context.Context) ([]string, error) {
383 close(entered)
384 <-release
385 return nil, nil
386 }).Once()
387
388 publisher := newAppTrafficPublisher(svc, Config{})
389 holding := make(chan struct{})
390 go func() {
391 defer close(holding)
392 _ = publisher.RebuildTraffic(context.Background())
393 }()
394 <-entered
395
396 expired, cancel := context.WithCancel(context.Background())
397 cancel()
398 err := publisher.RebuildTraffic(expired)
399 require.ErrorIs(t, err, context.Canceled)
400
401 close(release)
402 <-holding
403}