1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package app_test
4
5import (
6 "context"
7 "errors"
8 "runtime"
9 "sync"
10 "sync/atomic"
11 "testing"
12 "time"
13
14 "github.com/stretchr/testify/assert"
15
16 "bnema.dev/ermite/internal/app"
17 "bnema.dev/ermite/internal/domain"
18 "bnema.dev/ermite/internal/ports/mocks"
19)
20
21func waitFor(t *testing.T, cond func() bool, what string) {
22 t.Helper()
23 deadline := time.Now().Add(5 * time.Second)
24 for !cond() {
25 if time.Now().After(deadline) {
26 t.Fatalf("timeout waiting for %s", what)
27 }
28 time.Sleep(2 * time.Millisecond)
29 }
30}
31
32func TestMaintainerDebounces(t *testing.T) {
33 ctx := context.Background()
34 n := name(t, "tools")
35 git := mocks.NewGitRepos(t)
36 var runs atomic.Int32
37 git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(context.Context, domain.RepoName) error {
38 runs.Add(1)
39 return nil
40 })
41 m := app.NewMaintainer(ctx, git, 200*time.Millisecond)
42 defer m.Close(ctx)
43
44 for range 10 {
45 m.Schedule(ctx, n)
46 }
47 assert.Equal(t, int32(0), runs.Load(), "waits for the delay")
48 waitFor(t, func() bool { return runs.Load() == 1 }, "one run")
49 // Idle: nothing is pending or running, so no second run is coming.
50 waitFor(t, m.Idle, "the maintainer to be idle")
51 assert.Equal(t, int32(1), runs.Load(), "a burst makes one run")
52
53 m.Schedule(ctx, n)
54 waitFor(t, func() bool { return runs.Load() == 2 }, "a second burst, a second run")
55}
56
57func TestMaintainerOneRunAtATimePerRepo(t *testing.T) {
58 ctx := context.Background()
59 n, other := name(t, "tools"), name(t, "other")
60 git := mocks.NewGitRepos(t)
61 var concurrent, maxConcurrent, runs atomic.Int32
62 release := make(chan struct{})
63 var otherRan atomic.Bool
64 git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(context.Context, domain.RepoName) error {
65 c := concurrent.Add(1)
66 for {
67 m := maxConcurrent.Load()
68 if c <= m || maxConcurrent.CompareAndSwap(m, c) {
69 break
70 }
71 }
72 runs.Add(1)
73 <-release
74 concurrent.Add(-1)
75 return nil
76 })
77 git.EXPECT().Maintain(mockAny, other).RunAndReturn(func(context.Context, domain.RepoName) error {
78 otherRan.Store(true)
79 return nil
80 })
81 m := app.NewMaintainer(ctx, git, 10*time.Millisecond)
82 defer m.Close(ctx)
83
84 m.Schedule(ctx, n)
85 waitFor(t, func() bool { return runs.Load() == 1 }, "first run to start")
86 m.Schedule(ctx, n) // during the run: queues exactly one more
87 m.Schedule(ctx, n)
88 m.Schedule(ctx, other)
89 waitFor(t, otherRan.Load, "another repository is not blocked")
90 assert.Equal(t, int32(1), runs.Load(), "the second run waits for the first")
91
92 release <- struct{}{} // finish first run
93 waitFor(t, func() bool { return runs.Load() == 2 }, "queued run")
94 release <- struct{}{}
95 waitFor(t, m.Idle, "the maintainer to be idle")
96 assert.Equal(t, int32(2), runs.Load(), "exactly one queued run")
97 assert.Equal(t, int32(1), maxConcurrent.Load())
98}
99
100func TestMaintainerTellsTheReaderAfterARun(t *testing.T) {
101 ctx := context.Background()
102 n := name(t, "tools")
103 for _, failure := range []error{nil, errors.New("boom")} {
104 git := mocks.NewGitRepos(t)
105 git.EXPECT().Maintain(mockAny, n).Return(failure)
106 fresh := &freshener{}
107 m := app.NewMaintainer(ctx, git, time.Millisecond).WithFreshener(fresh)
108 m.Schedule(ctx, n)
109 waitFor(t, func() bool { return len(fresh.seen()) == 1 }, "the reader to be told")
110 m.Close(ctx)
111 assert.Equal(t, []domain.RepoName{n}, fresh.seen())
112 }
113}
114
115func TestMaintainerFailureIsLoggedNotFatal(t *testing.T) {
116 ctx := context.Background()
117 n := name(t, "tools")
118 git := mocks.NewGitRepos(t)
119 var runs atomic.Int32
120 git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(context.Context, domain.RepoName) error {
121 runs.Add(1)
122 return errors.New("boom")
123 })
124 m := app.NewMaintainer(ctx, git, 5*time.Millisecond)
125 defer m.Close(ctx)
126 m.Schedule(ctx, n)
127 waitFor(t, func() bool { return runs.Load() == 1 }, "run")
128 m.Schedule(ctx, n)
129 waitFor(t, func() bool { return runs.Load() == 2 }, "later runs still happen")
130}
131
132func TestMaintainerClose(t *testing.T) {
133 ctx := context.Background()
134 n := name(t, "tools")
135
136 t.Run("pending runs are dropped", func(t *testing.T) {
137 git := mocks.NewGitRepos(t) // no Maintain expected
138 m := app.NewMaintainer(ctx, git, time.Hour)
139 m.Schedule(ctx, n)
140 m.Close(ctx)
141 m.Schedule(ctx, n) // after Close: ignored
142 m.Close(ctx) // twice is fine
143 })
144
145 t.Run("a run in progress is cancelled and awaited", func(t *testing.T) {
146 git := mocks.NewGitRepos(t)
147 started := make(chan struct{})
148 var cancelled atomic.Bool
149 git.EXPECT().Maintain(mockAny, n).RunAndReturn(func(ctx context.Context, _ domain.RepoName) error {
150 close(started)
151 <-ctx.Done()
152 cancelled.Store(true)
153 return ctx.Err()
154 })
155 m := app.NewMaintainer(ctx, git, time.Millisecond)
156 m.Schedule(ctx, n)
157 <-started
158 done := make(chan struct{})
159 go func() { m.Close(ctx); close(done) }()
160 select {
161 case <-done:
162 case <-time.After(5 * time.Second):
163 t.Fatal("Close did not return")
164 }
165 assert.True(t, cancelled.Load())
166 })
167}
168
169func TestMaintainerConcurrentSchedule(t *testing.T) {
170 ctx := context.Background()
171 git := mocks.NewGitRepos(t)
172 git.EXPECT().Maintain(mockAny, mockAny).Return(nil).Maybe()
173 m := app.NewMaintainer(ctx, git, time.Millisecond)
174 var wg sync.WaitGroup
175 for i := range 8 {
176 wg.Add(1)
177 go func() {
178 defer wg.Done()
179 for range 50 {
180 m.Schedule(ctx, name(t, string(rune('a'+i%3))))
181 runtime.Gosched()
182 }
183 }()
184 }
185 wg.Wait()
186 m.Close(ctx)
187}