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 "bnema.dev/ermite/internal/readertest"
24)
25
26// alive reports whether a process still exists. The adapter collects its
27// children, so a stopped one is gone and not a zombie.
28func alive(pid int) bool { return syscall.Kill(pid, 0) == nil }
29
30func requireGone(t *testing.T, pids ...int) {
31 t.Helper()
32 require.Eventually(t, func() bool {
33 for _, p := range pids {
34 if alive(p) {
35 return false
36 }
37 }
38 return true
39 }, 5*time.Second, 10*time.Millisecond, "processes %v are still running", pids)
40}
41
42func TestCatFileProcessIsReusedAndStopsWhenIdle(t *testing.T) {
43 ctx := context.Background()
44 // The default idle timeout (60 s) never fires during the test: the hooks
45 // do what the timers would, deterministically.
46 g, data := newReader(t, nil)
47 h := readertest.NewHistory(t, data)
48
49 assert.Empty(t, g.CatFilePIDs(), "nothing runs before the first read")
50 _, err := g.ResolveRevision(ctx, h.Name, domain.Revision{})
51 require.NoError(t, err)
52 pids := g.CatFilePIDs()
53 require.Len(t, pids, 1)
54 require.True(t, alive(pids[0]))
55
56 for range 5 {
57 _, err = g.GetTree(ctx, h.Name, domain.Revision{}, domain.TreePath{}, false)
58 require.NoError(t, err)
59 }
60 assert.Equal(t, pids, g.CatFilePIDs(), "one long-lived process serves every request")
61
62 // A timer armed before the last use is stale: use keeps the process alive.
63 g.FireStaleIdleTimers()
64 assert.Equal(t, pids, g.CatFilePIDs())
65
66 // The timer of the last use stops it once nobody used it since.
67 g.FireIdleTimers()
68 require.Empty(t, g.CatFilePIDs())
69 requireGone(t, pids...)
70
71 // The next request starts a new one.
72 _, err = g.ResolveRevision(ctx, h.Name, domain.Revision{})
73 require.NoError(t, err)
74 next := g.CatFilePIDs()
75 require.Len(t, next, 1)
76 assert.NotEqual(t, pids, next)
77}
78
79func TestCatFileIdleTimerStopsAnUnusedProcess(t *testing.T) {
80 ctx := context.Background()
81 g, data := newReader(t, func(c *gitcli.Config) { c.CatFileIdle = 20 * time.Millisecond })
82 h := readertest.NewHistory(t, data)
83 _, err := g.ResolveRevision(ctx, h.Name, domain.Revision{})
84 require.NoError(t, err)
85 pids := g.CatFilePIDs()
86 require.Eventually(t, func() bool { return len(g.CatFilePIDs()) == 0 }, 5*time.Second, 5*time.Millisecond)
87 requireGone(t, pids...)
88}
89
90func TestCloseStopsEveryProcess(t *testing.T) {
91 ctx := context.Background()
92 g, data := newReader(t, nil) // default idle: 60 s, so only Close can stop them
93 h := readertest.NewHistory(t, data)
94 f2 := readertest.NewFixture(t, data, "second")
95 f2.Write("x", "x").Commit("x")
96 big := readertest.NewFixture(t, data, "bigone")
97 big.WriteBytes("big", readertest.RandomBytes(600<<10, 1)).Commit("big")
98
99 _, err := g.ResolveRevision(ctx, h.Name, domain.Revision{})
100 require.NoError(t, err)
101 _, err = g.ResolveRevision(ctx, f2.Name, domain.Revision{})
102 require.NoError(t, err)
103 _, rc, err := g.OpenBlob(ctx, big.Name, domain.Revision{}, readertest.TreePath(t, "big"))
104 require.NoError(t, err)
105 _ = rc // left open on purpose: Close must stop its process too
106
107 pids := append(g.CatFilePIDs(), g.BlobStreamPIDs()...)
108 require.GreaterOrEqual(t, len(pids), 4, "two cat-file processes, the big repo's one and the blob stream")
109 for _, p := range pids {
110 require.True(t, alive(p))
111 }
112
113 cctx, cancel := context.WithTimeout(ctx, 10*time.Second)
114 defer cancel()
115 require.NoError(t, g.Close(cctx))
116 requireGone(t, pids...)
117 assert.Empty(t, g.CatFilePIDs())
118
119 _, err = g.ResolveRevision(ctx, h.Name, domain.Revision{})
120 assert.ErrorIs(t, err, gitcli.ErrClosed)
121 _, _, err = g.OpenBlob(ctx, big.Name, domain.Revision{}, readertest.TreePath(t, "big"))
122 assert.ErrorIs(t, err, gitcli.ErrClosed)
123 assert.NoError(t, g.Close(ctx), "Close is idempotent")
124 _, err = io.Copy(io.Discard, rc)
125 assert.ErrorIs(t, err, io.ErrUnexpectedEOF, "a stream cut by Close fails instead of looking complete")
126 assert.NoError(t, rc.Close())
127 assert.Equal(t, 0, g.BlobStreams())
128}
129
130func TestPoolIsBounded(t *testing.T) {
131 ctx := context.Background()
132 g, data := newReader(t, func(c *gitcli.Config) { c.MaxCatFiles = 2 })
133 var names []domain.RepoName
134 for i := range 6 {
135 f := readertest.NewFixture(t, data, fmt.Sprintf("r%d", i))
136 f.Write("x", "x").Commit("x")
137 names = append(names, f.Name)
138 }
139 var seen []int
140 for _, n := range names {
141 _, err := g.ResolveRevision(ctx, n, domain.Revision{})
142 require.NoError(t, err)
143 pids := g.CatFilePIDs()
144 assert.LessOrEqual(t, len(pids), 2)
145 seen = append(seen, pids...)
146 }
147 // The evicted processes were stopped, not leaked.
148 live := map[int]bool{}
149 for _, p := range g.CatFilePIDs() {
150 live[p] = true
151 }
152 for _, p := range seen {
153 if !live[p] {
154 requireGone(t, p)
155 }
156 }
157}
158
159func TestPoolRestartsAfterProcessDeath(t *testing.T) {
160 ctx := context.Background()
161 g, data := newReader(t, nil)
162 h := readertest.NewHistory(t, data)
163
164 _, err := g.ResolveRevision(ctx, h.Name, domain.Revision{})
165 require.NoError(t, err)
166 old := g.CatFilePIDs()
167 require.Len(t, old, 1)
168 require.NoError(t, syscall.Kill(old[0], syscall.SIGKILL))
169 requireGone(t, old[0])
170
171 // The request that finds the dead process succeeds on a fresh one.
172 got, err := g.ResolveRevision(ctx, h.Name, domain.Revision{})
173 require.NoError(t, err)
174 assert.Equal(t, h.Merge, got.CommitID.String())
175 now := g.CatFilePIDs()
176 require.Len(t, now, 1)
177 assert.NotEqual(t, old, now)
178}
179
180func TestConcurrentReadsAreCorrectAndBounded(t *testing.T) {
181 ctx := context.Background()
182 g, data := newReader(t, nil)
183 h := readertest.NewHistory(t, data)
184
185 var wg sync.WaitGroup
186 errs := make(chan error, 24*10)
187 for i := range 24 {
188 wg.Go(func() {
189 for j := range 10 {
190 switch (i + j) % 4 {
191 case 0:
192 res, err := g.ResolveRevision(ctx, h.Name, readertest.Rev(t, "feature/x"))
193 if err == nil && res.CommitID.String() != h.Side {
194 err = fmt.Errorf("resolved %s", res.CommitID)
195 }
196 errs <- err
197 case 1:
198 info, rc, err := g.OpenBlob(ctx, h.Name, domain.Revision{}, readertest.TreePath(t, "b.txt"))
199 if err == nil {
200 var body []byte
201 body, err = io.ReadAll(rc)
202 _ = rc.Close()
203 if err == nil && (info.Size != 19 || string(body) != "one\ntwo\nthree\nfour\n") {
204 err = fmt.Errorf("blob %d %q", info.Size, body)
205 }
206 }
207 errs <- err
208 case 2:
209 tree, err := g.GetTree(ctx, h.Name, readertest.Rev(t, "v2"), readertest.TreePath(t, "d"), false)
210 if err == nil && len(tree.Entries) != 2 {
211 err = fmt.Errorf("tree of %d entries", len(tree.Entries))
212 }
213 errs <- err
214 default:
215 _, err := g.ResolveRefPath(ctx, h.Name, "feature/x/d/f")
216 errs <- err
217 }
218 }
219 })
220 }
221 wg.Wait()
222 close(errs)
223 for err := range errs {
224 require.NoError(t, err)
225 }
226 assert.NotEmpty(t, g.CatFilePIDs())
227 assert.LessOrEqual(t, len(g.CatFilePIDs()), 4, "a repository holds a bounded number of processes")
228}
229
230func TestCancelledRequestDoesNotPoisonTheProcess(t *testing.T) {
231 g, data := newReader(t, nil)
232 h := readertest.NewHistory(t, data)
233
234 // Many requests, some with a context that dies at a random point. Whatever
235 // state a killed process was in, later requests must get right answers.
236 var wg sync.WaitGroup
237 for i := range 16 {
238 wg.Go(func() {
239 for j := range 20 {
240 ctx, cancel := context.WithTimeout(context.Background(), time.Duration((i*7+j*3)%5)*time.Millisecond)
241 res, err := g.ResolveRevision(ctx, h.Name, readertest.Rev(t, "v2"))
242 cancel()
243 if err == nil {
244 assert.Equal(t, h.C2, res.CommitID.String())
245 } else {
246 assert.True(t, errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled), "%v", err)
247 }
248 }
249 })
250 }
251 wg.Wait()
252 res, err := g.ResolveRevision(context.Background(), h.Name, readertest.Rev(t, "v2"))
253 require.NoError(t, err)
254 assert.Equal(t, h.C2, res.CommitID.String())
255}
256
257func TestBlobStreamStopsWhenContextEndsOrReaderCloses(t *testing.T) {
258 g, data := newReader(t, nil)
259 f := readertest.NewFixture(t, data, "bigone")
260 f.WriteBytes("big", readertest.RandomBytes(2<<20, 3)).Commit("big")
261
262 ctx, cancel := context.WithCancel(context.Background())
263 _, rc, err := g.OpenBlob(ctx, f.Name, domain.Revision{}, readertest.TreePath(t, "big"))
264 require.NoError(t, err)
265 pids := g.BlobStreamPIDs()
266 require.Len(t, pids, 1)
267 _, err = io.ReadFull(rc, make([]byte, 1000))
268 require.NoError(t, err)
269
270 cancel() // the client went away mid-stream
271 requireGone(t, pids...)
272 _, err = io.Copy(io.Discard, rc)
273 assert.Error(t, err, "a cut stream does not look complete")
274 require.NoError(t, rc.Close())
275 assert.Equal(t, 0, g.BlobStreams())
276
277 // Closing early stops the process as well.
278 _, rc, err = g.OpenBlob(context.Background(), f.Name, domain.Revision{}, readertest.TreePath(t, "big"))
279 require.NoError(t, err)
280 pids = g.BlobStreamPIDs()
281 require.Len(t, pids, 1)
282 require.NoError(t, rc.Close())
283 requireGone(t, pids...)
284 assert.Equal(t, 0, g.BlobStreams())
285
286 // A context that is already cancelled starts nothing.
287 cctx, cancel2 := context.WithCancel(context.Background())
288 cancel2()
289 _, _, err = g.OpenBlob(cctx, f.Name, domain.Revision{}, readertest.TreePath(t, "big"))
290 assert.ErrorIs(t, err, context.Canceled)
291 assert.Equal(t, 0, g.BlobStreams())
292}
293
294// TestOpenBlobStreams50MiB is AC3: a 50 MiB file is streamed with memory that
295// does not depend on its size. The test reads it in 64 KiB chunks, as the
296// GetBlob RPC does, and bounds both what was allocated while reading and the
297// peak of the heap.
298func TestOpenBlobStreams50MiB(t *testing.T) {
299 if testing.Short() {
300 t.Skip("50 MiB fixture")
301 }
302 ctx := context.Background()
303 g, data := newReader(t, nil)
304 f := readertest.NewFixture(t, data, "huge")
305 const size = 50 << 20
306 content := readertest.RandomBytes(size, 42)
307 want := sha256.Sum256(content)
308 f.WriteBytes("huge.bin", content).Commit("huge")
309 content = nil //nolint:ineffassign // let the 50 MiB go before measuring
310
311 runtime.GC()
312 var before, after runtime.MemStats
313 runtime.ReadMemStats(&before)
314
315 info, rc, err := g.OpenBlob(ctx, f.Name, domain.Revision{}, readertest.TreePath(t, "huge.bin"))
316 require.NoError(t, err)
317 assert.Equal(t, int64(size), info.Size)
318 assert.True(t, info.Binary)
319 assert.Equal(t, "application/octet-stream", info.MediaType)
320
321 h := sha256.New()
322 buf := make([]byte, 64<<10)
323 var total int64
324 var peak uint64
325 for i := 0; ; i++ {
326 n, err := rc.Read(buf)
327 total += int64(n)
328 h.Write(buf[:n])
329 if i%64 == 0 {
330 var m runtime.MemStats
331 runtime.ReadMemStats(&m)
332 peak = max(peak, m.HeapAlloc)
333 }
334 if errors.Is(err, io.EOF) {
335 break
336 }
337 require.NoError(t, err)
338 }
339 require.NoError(t, rc.Close())
340 runtime.ReadMemStats(&after)
341
342 assert.Equal(t, int64(size), total)
343 assert.Equal(t, want[:], h.Sum(nil), "content is intact")
344
345 allocated := after.TotalAlloc - before.TotalAlloc
346 assert.Less(t, allocated, uint64(8<<20), "bytes allocated while streaming 50 MiB: %d", allocated)
347 growth := int64(peak) - int64(before.HeapAlloc)
348 assert.Less(t, growth, int64(8<<20), "heap growth while streaming 50 MiB: %d", growth)
349 t.Logf("streamed %d MiB: allocated %d KiB, peak heap growth %d KiB", size>>20, allocated>>10, growth>>10)
350}
351
352// The one-shot git commands share a bounded number of slots: a burst of
353// requests waits for a slot, and gives up with its context.
354func TestExecSemaphoreBoundsOneShotCommands(t *testing.T) {
355 ctx := context.Background()
356 gate := t.TempDir() + "/gate"
357 // A git that blocks until the gate file exists.
358 script := writeScript(t, `while [ ! -e `+gate+` ]; do sleep 0.01; done; echo "git version 2.50.0"`)
359 g := gitcli.New(gitcli.Config{Bin: script, DataDir: t.TempDir(), Path: os.Getenv("PATH"), MaxExec: 2})
360 size, taken := g.ExecSlots()
361 require.Equal(t, 2, size)
362 require.Zero(t, taken)
363
364 errs := make(chan error, 2)
365 for range 2 {
366 go func() { _, err := g.CheckVersion(ctx); errs <- err }()
367 }
368 require.Eventually(t, func() bool { _, taken := g.ExecSlots(); return taken == 2 }, 5*time.Second, time.Millisecond)
369
370 // A third waits and gives up with its deadline: it never started git.
371 short, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
372 defer cancel()
373 _, err := g.CheckVersion(short)
374 assert.ErrorIs(t, err, context.DeadlineExceeded)
375
376 require.NoError(t, os.WriteFile(gate, nil, 0o600))
377 for range 2 {
378 <-errs // their outcome is the version check's, not the subject
379 }
380 require.Eventually(t, func() bool { _, taken := g.ExecSlots(); return taken == 0 }, 5*time.Second, time.Millisecond)
381}
382
383// Open blob streams are bounded; a refused stream leaves nothing behind and a
384// closed one frees its slot.
385func TestBlobStreamsAreBounded(t *testing.T) {
386 ctx := context.Background()
387 g, data := newReader(t, func(c *gitcli.Config) { c.MaxBlobStreams = 2 })
388 f := readertest.NewFixture(t, data, "bigone")
389 f.WriteBytes("big", readertest.RandomBytes(600<<10, 5)).Commit("big")
390 open := func() (io.ReadCloser, error) {
391 _, rc, err := g.OpenBlob(ctx, f.Name, domain.Revision{}, readertest.TreePath(t, "big"))
392 return rc, err
393 }
394
395 a, err := open()
396 require.NoError(t, err)
397 b, err := open()
398 require.NoError(t, err)
399 _, err = open()
400 assert.ErrorIs(t, err, domain.ErrResourceExhausted)
401 assert.Equal(t, 2, g.BlobStreams())
402
403 require.NoError(t, a.Close())
404 c, err := open()
405 require.NoError(t, err, "a closed stream frees its slot")
406 require.NoError(t, b.Close())
407 require.NoError(t, c.Close())
408 assert.Zero(t, g.BlobStreams())
409}
410
411// hold runs HoldCatFile in the background and returns once the process is
412// held. The returned func releases it and waits for the request to end.
413func hold(t *testing.T, g *gitcli.Git, repo domain.RepoName) (release func()) {
414 t.Helper()
415 held, free, done := make(chan struct{}), make(chan struct{}), make(chan error, 1)
416 go func() {
417 done <- g.HoldCatFile(context.Background(), repo, func() { close(held); <-free })
418 }()
419 select {
420 case <-held:
421 case err := <-done:
422 require.NoError(t, err)
423 t.Fatal("the request ended before it held a process")
424 }
425 var once sync.Once
426 release = func() {
427 once.Do(func() { close(free); require.NoError(t, <-done) })
428 }
429 t.Cleanup(release)
430 return release
431}
432
433// A request in flight does not make the others of the repository wait: they
434// get a process of their own, started only because of the contention.
435func TestConcurrentReadsOnOneRepoDoNotSerialize(t *testing.T) {
436 ctx := context.Background()
437 g, data := newReader(t, nil)
438 h := readertest.NewHistory(t, data)
439
440 _, err := g.ResolveRevision(ctx, h.Name, domain.Revision{})
441 require.NoError(t, err)
442 require.Len(t, g.CatFilePIDs(), 1, "no contention, one process")
443
444 release := hold(t, g, h.Name)
445 // The slow request holds the only process; this one must not wait for it
446 // (it would block forever: the holder is released after).
447 res, err := g.ResolveRevision(ctx, h.Name, domain.Revision{})
448 require.NoError(t, err)
449 assert.Equal(t, h.Merge, res.CommitID.String())
450 pids := g.CatFilePIDs()
451 assert.Len(t, pids, 2, "the contention started a second process")
452 assert.Zero(t, g.CatFileWaiting())
453
454 release()
455 // Both stay for reuse until they idle out.
456 assert.Equal(t, pids, g.CatFilePIDs())
457 g.FireIdleTimers()
458 assert.Empty(t, g.CatFilePIDs())
459 requireGone(t, pids...)
460}
461
462func TestCatFileProcessBounds(t *testing.T) {
463 tests := []struct {
464 name string
465 max int // Config.MaxCatFiles
466 // holds are the repositories, by index, whose process a request holds.
467 holds []int
468 // extra is the repository of one more request, which must wait.
469 extra int
470 wantProcs int
471 wantWaiting int
472 }{
473 {name: "per repository cap", max: 16, holds: []int{0, 0, 0, 0}, extra: 0, wantProcs: 4, wantWaiting: 1},
474 {name: "global cap, repository with a process waits", max: 3, holds: []int{0, 0, 1}, extra: 1, wantProcs: 3, wantWaiting: 1},
475 {name: "global cap lower than the per repository cap", max: 2, holds: []int{0, 0}, extra: 0, wantProcs: 2, wantWaiting: 1},
476 }
477 for _, tt := range tests {
478 t.Run(tt.name, func(t *testing.T) {
479 g, data := newReader(t, func(c *gitcli.Config) { c.MaxCatFiles = tt.max })
480 var names []domain.RepoName
481 for i := range 2 {
482 f := readertest.NewFixture(t, data, fmt.Sprintf("r%d", i))
483 f.Write("x", "x").Commit("x")
484 names = append(names, f.Name)
485 }
486 var releases []func()
487 for _, i := range tt.holds {
488 releases = append(releases, hold(t, g, names[i]))
489 }
490 assert.Len(t, g.CatFilePIDs(), tt.wantProcs)
491
492 done := make(chan error, 1)
493 go func() {
494 _, err := g.ResolveRevision(context.Background(), names[tt.extra], domain.Revision{})
495 done <- err
496 }()
497 require.Eventually(t, func() bool { return g.CatFileWaiting() == tt.wantWaiting }, 5*time.Second, time.Millisecond)
498 assert.Len(t, g.CatFilePIDs(), tt.wantProcs, "a waiting request starts nothing")
499 select {
500 case err := <-done:
501 t.Fatalf("the request did not wait: %v", err)
502 default:
503 }
504
505 // A process released to a waiter serves it, without a new process.
506 for k, i := range tt.holds {
507 if i == tt.extra {
508 releases[k]()
509 break
510 }
511 }
512 require.NoError(t, <-done)
513 assert.LessOrEqual(t, len(g.CatFilePIDs()), tt.wantProcs)
514 assert.Zero(t, g.CatFileWaiting())
515 })
516 }
517}
518
519// A waiting request gives up with its context and leaves nothing queued.
520func TestWaitingForAProcessEndsWithTheContext(t *testing.T) {
521 g, data := newReader(t, func(c *gitcli.Config) { c.MaxCatFiles = 1 })
522 h := readertest.NewHistory(t, data)
523 release := hold(t, g, h.Name)
524
525 ctx, cancel := context.WithCancel(context.Background())
526 done := make(chan error, 1)
527 go func() { _, err := g.ResolveRevision(ctx, h.Name, domain.Revision{}); done <- err }()
528 require.Eventually(t, func() bool { return g.CatFileWaiting() == 1 }, 5*time.Second, time.Millisecond)
529 cancel()
530 assert.ErrorIs(t, <-done, context.Canceled)
531 assert.Zero(t, g.CatFileWaiting())
532 assert.Len(t, g.CatFilePIDs(), 1)
533
534 release()
535 _, err := g.ResolveRevision(context.Background(), h.Name, domain.Revision{})
536 require.NoError(t, err, "the process survived the abandoned wait")
537}
538
539// The bound counts processes, not repositories: idle processes of several
540// repositories are evicted to make room.
541func TestPoolBoundCountsProcessesAcrossRepositories(t *testing.T) {
542 ctx := context.Background()
543 g, data := newReader(t, func(c *gitcli.Config) { c.MaxCatFiles = 3 })
544 var names []domain.RepoName
545 for i := range 4 {
546 f := readertest.NewFixture(t, data, fmt.Sprintf("r%d", i))
547 f.Write("x", "x").Commit("x")
548 names = append(names, f.Name)
549 }
550 // Repository 0 grows to three processes.
551 releases := []func(){hold(t, g, names[0]), hold(t, g, names[0]), hold(t, g, names[0])}
552 require.Len(t, g.CatFilePIDs(), 3)
553 for _, r := range releases {
554 r()
555 }
556 before := g.CatFilePIDs()
557
558 for _, n := range names[1:] {
559 _, err := g.ResolveRevision(ctx, n, domain.Revision{})
560 require.NoError(t, err)
561 assert.LessOrEqual(t, len(g.CatFilePIDs()), 3)
562 }
563 live := map[int]bool{}
564 for _, p := range g.CatFilePIDs() {
565 live[p] = true
566 }
567 for _, p := range before {
568 if !live[p] {
569 requireGone(t, p)
570 }
571 }
572}