1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package gitcli_test
4
5import (
6 "context"
7 "crypto/sha256"
8 "errors"
9 "fmt"
10 "io"
11 "os"
12 "runtime"
13 "sync"
14 "syscall"
15 "testing"
16 "time"
17
18 "github.com/stretchr/testify/assert"
19 "github.com/stretchr/testify/require"
20
21 "bnema.dev/ermite/internal/adapters/gitcli"
22 "bnema.dev/ermite/internal/domain"
23)
24
25// alive reports whether a process still exists. The adapter collects its
26// children, so a stopped one is gone and not a zombie.
27func alive(pid int) bool { return syscall.Kill(pid, 0) == nil }
28
29func requireGone(t *testing.T, pids ...int) {
30 t.Helper()
31 require.Eventually(t, func() bool {
32 for _, p := range pids {
33 if alive(p) {
34 return false
35 }
36 }
37 return true
38 }, 5*time.Second, 10*time.Millisecond, "processes %v are still running", pids)
39}
40
41func TestCatFileProcessIsReusedAndStopsWhenIdle(t *testing.T) {
42 ctx := context.Background()
43 // The default idle timeout (60 s) never fires during the test: the hooks
44 // do what the timers would, deterministically.
45 g, data := newReader(t, nil)
46 h := newHistory(t, data)
47
48 assert.Empty(t, g.CatFilePIDs(), "nothing runs before the first read")
49 _, err := g.ResolveRevision(ctx, h.name, domain.Revision{})
50 require.NoError(t, err)
51 pids := g.CatFilePIDs()
52 require.Len(t, pids, 1)
53 require.True(t, alive(pids[0]))
54
55 for range 5 {
56 _, err = g.GetTree(ctx, h.name, domain.Revision{}, domain.TreePath{}, false)
57 require.NoError(t, err)
58 }
59 assert.Equal(t, pids, g.CatFilePIDs(), "one long-lived process serves every request")
60
61 // A timer armed before the last use is stale: use keeps the process alive.
62 g.FireStaleIdleTimers()
63 assert.Equal(t, pids, g.CatFilePIDs())
64
65 // The timer of the last use stops it once nobody used it since.
66 g.FireIdleTimers()
67 require.Empty(t, g.CatFilePIDs())
68 requireGone(t, pids...)
69
70 // The next request starts a new one.
71 _, err = g.ResolveRevision(ctx, h.name, domain.Revision{})
72 require.NoError(t, err)
73 next := g.CatFilePIDs()
74 require.Len(t, next, 1)
75 assert.NotEqual(t, pids, next)
76}
77
78func TestCatFileIdleTimerStopsAnUnusedProcess(t *testing.T) {
79 ctx := context.Background()
80 g, data := newReader(t, func(c *gitcli.Config) { c.CatFileIdle = 20 * time.Millisecond })
81 h := newHistory(t, data)
82 _, err := g.ResolveRevision(ctx, h.name, domain.Revision{})
83 require.NoError(t, err)
84 pids := g.CatFilePIDs()
85 require.Eventually(t, func() bool { return len(g.CatFilePIDs()) == 0 }, 5*time.Second, 5*time.Millisecond)
86 requireGone(t, pids...)
87}
88
89func TestCloseStopsEveryProcess(t *testing.T) {
90 ctx := context.Background()
91 g, data := newReader(t, nil) // default idle: 60 s, so only Close can stop them
92 h := newHistory(t, data)
93 f2 := newFixtureRepo(t, data, "second")
94 f2.write("x", "x").commit("x")
95 big := newFixtureRepo(t, data, "bigone")
96 big.writeBytes("big", randomBytes(600<<10, 1)).commit("big")
97
98 _, err := g.ResolveRevision(ctx, h.name, domain.Revision{})
99 require.NoError(t, err)
100 _, err = g.ResolveRevision(ctx, f2.name, domain.Revision{})
101 require.NoError(t, err)
102 _, rc, err := g.OpenBlob(ctx, big.name, domain.Revision{}, tpath(t, "big"))
103 require.NoError(t, err)
104 _ = rc // left open on purpose: Close must stop its process too
105
106 pids := append(g.CatFilePIDs(), g.BlobStreamPIDs()...)
107 require.GreaterOrEqual(t, len(pids), 4, "two cat-file processes, the big repo's one and the blob stream")
108 for _, p := range pids {
109 require.True(t, alive(p))
110 }
111
112 cctx, cancel := context.WithTimeout(ctx, 10*time.Second)
113 defer cancel()
114 require.NoError(t, g.Close(cctx))
115 requireGone(t, pids...)
116 assert.Empty(t, g.CatFilePIDs())
117
118 _, err = g.ResolveRevision(ctx, h.name, domain.Revision{})
119 assert.ErrorIs(t, err, gitcli.ErrClosed)
120 _, _, err = g.OpenBlob(ctx, big.name, domain.Revision{}, tpath(t, "big"))
121 assert.ErrorIs(t, err, gitcli.ErrClosed)
122 assert.NoError(t, g.Close(ctx), "Close is idempotent")
123 _, err = io.Copy(io.Discard, rc)
124 assert.ErrorIs(t, err, io.ErrUnexpectedEOF, "a stream cut by Close fails instead of looking complete")
125 assert.NoError(t, rc.Close())
126 assert.Equal(t, 0, g.BlobStreams())
127}
128
129func TestPoolIsBounded(t *testing.T) {
130 ctx := context.Background()
131 g, data := newReader(t, func(c *gitcli.Config) { c.MaxCatFiles = 2 })
132 var names []domain.RepoName
133 for i := range 6 {
134 f := newFixtureRepo(t, data, fmt.Sprintf("r%d", i))
135 f.write("x", "x").commit("x")
136 names = append(names, f.name)
137 }
138 var seen []int
139 for _, n := range names {
140 _, err := g.ResolveRevision(ctx, n, domain.Revision{})
141 require.NoError(t, err)
142 pids := g.CatFilePIDs()
143 assert.LessOrEqual(t, len(pids), 2)
144 seen = append(seen, pids...)
145 }
146 // The evicted processes were stopped, not leaked.
147 live := map[int]bool{}
148 for _, p := range g.CatFilePIDs() {
149 live[p] = true
150 }
151 for _, p := range seen {
152 if !live[p] {
153 requireGone(t, p)
154 }
155 }
156}
157
158func TestPoolRestartsAfterProcessDeath(t *testing.T) {
159 ctx := context.Background()
160 g, data := newReader(t, nil)
161 h := newHistory(t, data)
162
163 _, err := g.ResolveRevision(ctx, h.name, domain.Revision{})
164 require.NoError(t, err)
165 old := g.CatFilePIDs()
166 require.Len(t, old, 1)
167 require.NoError(t, syscall.Kill(old[0], syscall.SIGKILL))
168 requireGone(t, old[0])
169
170 // The request that finds the dead process succeeds on a fresh one.
171 got, err := g.ResolveRevision(ctx, h.name, domain.Revision{})
172 require.NoError(t, err)
173 assert.Equal(t, h.merge, got.CommitID.String())
174 now := g.CatFilePIDs()
175 require.Len(t, now, 1)
176 assert.NotEqual(t, old, now)
177}
178
179func TestConcurrentReadsAreSerializedPerProcess(t *testing.T) {
180 ctx := context.Background()
181 g, data := newReader(t, nil)
182 h := newHistory(t, data)
183
184 var wg sync.WaitGroup
185 errs := make(chan error, 24*10)
186 for i := range 24 {
187 wg.Go(func() {
188 for j := range 10 {
189 switch (i + j) % 4 {
190 case 0:
191 res, err := g.ResolveRevision(ctx, h.name, rev(t, "feature/x"))
192 if err == nil && res.CommitID.String() != h.side {
193 err = fmt.Errorf("resolved %s", res.CommitID)
194 }
195 errs <- err
196 case 1:
197 info, rc, err := g.OpenBlob(ctx, h.name, domain.Revision{}, tpath(t, "b.txt"))
198 if err == nil {
199 var body []byte
200 body, err = io.ReadAll(rc)
201 _ = rc.Close()
202 if err == nil && (info.Size != 19 || string(body) != "one\ntwo\nthree\nfour\n") {
203 err = fmt.Errorf("blob %d %q", info.Size, body)
204 }
205 }
206 errs <- err
207 case 2:
208 tree, err := g.GetTree(ctx, h.name, rev(t, "v2"), tpath(t, "d"), false)
209 if err == nil && len(tree.Entries) != 2 {
210 err = fmt.Errorf("tree of %d entries", len(tree.Entries))
211 }
212 errs <- err
213 default:
214 _, err := g.ResolveRefPath(ctx, h.name, "feature/x/d/f")
215 errs <- err
216 }
217 }
218 })
219 }
220 wg.Wait()
221 close(errs)
222 for err := range errs {
223 require.NoError(t, err)
224 }
225 assert.Len(t, g.CatFilePIDs(), 1, "one process for one repository")
226}
227
228func TestCancelledRequestDoesNotPoisonTheProcess(t *testing.T) {
229 g, data := newReader(t, nil)
230 h := newHistory(t, data)
231
232 // Many requests, some with a context that dies at a random point. Whatever
233 // state a killed process was in, later requests must get right answers.
234 var wg sync.WaitGroup
235 for i := range 16 {
236 wg.Go(func() {
237 for j := range 20 {
238 ctx, cancel := context.WithTimeout(context.Background(), time.Duration((i*7+j*3)%5)*time.Millisecond)
239 res, err := g.ResolveRevision(ctx, h.name, rev(t, "v2"))
240 cancel()
241 if err == nil {
242 assert.Equal(t, h.c2, res.CommitID.String())
243 } else {
244 assert.True(t, errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled), "%v", err)
245 }
246 }
247 })
248 }
249 wg.Wait()
250 res, err := g.ResolveRevision(context.Background(), h.name, rev(t, "v2"))
251 require.NoError(t, err)
252 assert.Equal(t, h.c2, res.CommitID.String())
253}
254
255func TestBlobStreamStopsWhenContextEndsOrReaderCloses(t *testing.T) {
256 g, data := newReader(t, nil)
257 f := newFixtureRepo(t, data, "bigone")
258 f.writeBytes("big", randomBytes(2<<20, 3)).commit("big")
259
260 ctx, cancel := context.WithCancel(context.Background())
261 _, rc, err := g.OpenBlob(ctx, f.name, domain.Revision{}, tpath(t, "big"))
262 require.NoError(t, err)
263 pids := g.BlobStreamPIDs()
264 require.Len(t, pids, 1)
265 _, err = io.ReadFull(rc, make([]byte, 1000))
266 require.NoError(t, err)
267
268 cancel() // the client went away mid-stream
269 requireGone(t, pids...)
270 _, err = io.Copy(io.Discard, rc)
271 assert.Error(t, err, "a cut stream does not look complete")
272 require.NoError(t, rc.Close())
273 assert.Equal(t, 0, g.BlobStreams())
274
275 // Closing early stops the process as well.
276 _, rc, err = g.OpenBlob(context.Background(), f.name, domain.Revision{}, tpath(t, "big"))
277 require.NoError(t, err)
278 pids = g.BlobStreamPIDs()
279 require.Len(t, pids, 1)
280 require.NoError(t, rc.Close())
281 requireGone(t, pids...)
282 assert.Equal(t, 0, g.BlobStreams())
283
284 // A context that is already cancelled starts nothing.
285 cctx, cancel2 := context.WithCancel(context.Background())
286 cancel2()
287 _, _, err = g.OpenBlob(cctx, f.name, domain.Revision{}, tpath(t, "big"))
288 assert.ErrorIs(t, err, context.Canceled)
289 assert.Equal(t, 0, g.BlobStreams())
290}
291
292// TestOpenBlobStreams50MiB is AC3: a 50 MiB file is streamed with memory that
293// does not depend on its size. The test reads it in 64 KiB chunks, as the
294// GetBlob RPC does, and bounds both what was allocated while reading and the
295// peak of the heap.
296func TestOpenBlobStreams50MiB(t *testing.T) {
297 if testing.Short() {
298 t.Skip("50 MiB fixture")
299 }
300 ctx := context.Background()
301 g, data := newReader(t, nil)
302 f := newFixtureRepo(t, data, "huge")
303 const size = 50 << 20
304 content := randomBytes(size, 42)
305 want := sha256.Sum256(content)
306 f.writeBytes("huge.bin", content).commit("huge")
307 content = nil //nolint:ineffassign // let the 50 MiB go before measuring
308
309 runtime.GC()
310 var before, after runtime.MemStats
311 runtime.ReadMemStats(&before)
312
313 info, rc, err := g.OpenBlob(ctx, f.name, domain.Revision{}, tpath(t, "huge.bin"))
314 require.NoError(t, err)
315 assert.Equal(t, int64(size), info.Size)
316 assert.True(t, info.Binary)
317 assert.Equal(t, "application/octet-stream", info.MediaType)
318
319 h := sha256.New()
320 buf := make([]byte, 64<<10)
321 var total int64
322 var peak uint64
323 for i := 0; ; i++ {
324 n, err := rc.Read(buf)
325 total += int64(n)
326 h.Write(buf[:n])
327 if i%64 == 0 {
328 var m runtime.MemStats
329 runtime.ReadMemStats(&m)
330 peak = max(peak, m.HeapAlloc)
331 }
332 if errors.Is(err, io.EOF) {
333 break
334 }
335 require.NoError(t, err)
336 }
337 require.NoError(t, rc.Close())
338 runtime.ReadMemStats(&after)
339
340 assert.Equal(t, int64(size), total)
341 assert.Equal(t, want[:], h.Sum(nil), "content is intact")
342
343 allocated := after.TotalAlloc - before.TotalAlloc
344 assert.Less(t, allocated, uint64(8<<20), "bytes allocated while streaming 50 MiB: %d", allocated)
345 growth := int64(peak) - int64(before.HeapAlloc)
346 assert.Less(t, growth, int64(8<<20), "heap growth while streaming 50 MiB: %d", growth)
347 t.Logf("streamed %d MiB: allocated %d KiB, peak heap growth %d KiB", size>>20, allocated>>10, growth>>10)
348}
349
350// The one-shot git commands share a bounded number of slots: a burst of
351// requests waits for a slot, and gives up with its context.
352func TestExecSemaphoreBoundsOneShotCommands(t *testing.T) {
353 ctx := context.Background()
354 gate := t.TempDir() + "/gate"
355 // A git that blocks until the gate file exists.
356 script := writeScript(t, `while [ ! -e `+gate+` ]; do sleep 0.01; done; echo "git version 2.50.0"`)
357 g := gitcli.New(gitcli.Config{Bin: script, DataDir: t.TempDir(), Path: os.Getenv("PATH"), MaxExec: 2})
358 size, taken := g.ExecSlots()
359 require.Equal(t, 2, size)
360 require.Zero(t, taken)
361
362 errs := make(chan error, 2)
363 for range 2 {
364 go func() { _, err := g.CheckVersion(ctx); errs <- err }()
365 }
366 require.Eventually(t, func() bool { _, taken := g.ExecSlots(); return taken == 2 }, 5*time.Second, time.Millisecond)
367
368 // A third waits and gives up with its deadline: it never started git.
369 short, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
370 defer cancel()
371 _, err := g.CheckVersion(short)
372 assert.ErrorIs(t, err, context.DeadlineExceeded)
373
374 require.NoError(t, os.WriteFile(gate, nil, 0o600))
375 for range 2 {
376 <-errs // their outcome is the version check's, not the subject
377 }
378 require.Eventually(t, func() bool { _, taken := g.ExecSlots(); return taken == 0 }, 5*time.Second, time.Millisecond)
379}
380
381// Open blob streams are bounded; a refused stream leaves nothing behind and a
382// closed one frees its slot.
383func TestBlobStreamsAreBounded(t *testing.T) {
384 ctx := context.Background()
385 g, data := newReader(t, func(c *gitcli.Config) { c.MaxBlobStreams = 2 })
386 f := newFixtureRepo(t, data, "bigone")
387 f.writeBytes("big", randomBytes(600<<10, 5)).commit("big")
388 open := func() (io.ReadCloser, error) {
389 _, rc, err := g.OpenBlob(ctx, f.name, domain.Revision{}, tpath(t, "big"))
390 return rc, err
391 }
392
393 a, err := open()
394 require.NoError(t, err)
395 b, err := open()
396 require.NoError(t, err)
397 _, err = open()
398 assert.ErrorIs(t, err, domain.ErrResourceExhausted)
399 assert.Equal(t, 2, g.BlobStreams())
400
401 require.NoError(t, a.Close())
402 c, err := open()
403 require.NoError(t, err, "a closed stream frees its slot")
404 require.NoError(t, b.Close())
405 require.NoError(t, c.Close())
406 assert.Zero(t, g.BlobStreams())
407}