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 TestConcurrentReadsAreSerializedPerProcess(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.Len(t, g.CatFilePIDs(), 1, "one process for one repository")
227}
228
229func TestCancelledRequestDoesNotPoisonTheProcess(t *testing.T) {
230 g, data := newReader(t, nil)
231 h := readertest.NewHistory(t, data)
232
233 // Many requests, some with a context that dies at a random point. Whatever
234 // state a killed process was in, later requests must get right answers.
235 var wg sync.WaitGroup
236 for i := range 16 {
237 wg.Go(func() {
238 for j := range 20 {
239 ctx, cancel := context.WithTimeout(context.Background(), time.Duration((i*7+j*3)%5)*time.Millisecond)
240 res, err := g.ResolveRevision(ctx, h.Name, readertest.Rev(t, "v2"))
241 cancel()
242 if err == nil {
243 assert.Equal(t, h.C2, res.CommitID.String())
244 } else {
245 assert.True(t, errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled), "%v", err)
246 }
247 }
248 })
249 }
250 wg.Wait()
251 res, err := g.ResolveRevision(context.Background(), h.Name, readertest.Rev(t, "v2"))
252 require.NoError(t, err)
253 assert.Equal(t, h.C2, res.CommitID.String())
254}
255
256func TestBlobStreamStopsWhenContextEndsOrReaderCloses(t *testing.T) {
257 g, data := newReader(t, nil)
258 f := readertest.NewFixture(t, data, "bigone")
259 f.WriteBytes("big", readertest.RandomBytes(2<<20, 3)).Commit("big")
260
261 ctx, cancel := context.WithCancel(context.Background())
262 _, rc, err := g.OpenBlob(ctx, f.Name, domain.Revision{}, readertest.TreePath(t, "big"))
263 require.NoError(t, err)
264 pids := g.BlobStreamPIDs()
265 require.Len(t, pids, 1)
266 _, err = io.ReadFull(rc, make([]byte, 1000))
267 require.NoError(t, err)
268
269 cancel() // the client went away mid-stream
270 requireGone(t, pids...)
271 _, err = io.Copy(io.Discard, rc)
272 assert.Error(t, err, "a cut stream does not look complete")
273 require.NoError(t, rc.Close())
274 assert.Equal(t, 0, g.BlobStreams())
275
276 // Closing early stops the process as well.
277 _, rc, err = g.OpenBlob(context.Background(), f.Name, domain.Revision{}, readertest.TreePath(t, "big"))
278 require.NoError(t, err)
279 pids = g.BlobStreamPIDs()
280 require.Len(t, pids, 1)
281 require.NoError(t, rc.Close())
282 requireGone(t, pids...)
283 assert.Equal(t, 0, g.BlobStreams())
284
285 // A context that is already cancelled starts nothing.
286 cctx, cancel2 := context.WithCancel(context.Background())
287 cancel2()
288 _, _, err = g.OpenBlob(cctx, f.Name, domain.Revision{}, readertest.TreePath(t, "big"))
289 assert.ErrorIs(t, err, context.Canceled)
290 assert.Equal(t, 0, g.BlobStreams())
291}
292
293// TestOpenBlobStreams50MiB is AC3: a 50 MiB file is streamed with memory that
294// does not depend on its size. The test reads it in 64 KiB chunks, as the
295// GetBlob RPC does, and bounds both what was allocated while reading and the
296// peak of the heap.
297func TestOpenBlobStreams50MiB(t *testing.T) {
298 if testing.Short() {
299 t.Skip("50 MiB fixture")
300 }
301 ctx := context.Background()
302 g, data := newReader(t, nil)
303 f := readertest.NewFixture(t, data, "huge")
304 const size = 50 << 20
305 content := readertest.RandomBytes(size, 42)
306 want := sha256.Sum256(content)
307 f.WriteBytes("huge.bin", content).Commit("huge")
308 content = nil //nolint:ineffassign // let the 50 MiB go before measuring
309
310 runtime.GC()
311 var before, after runtime.MemStats
312 runtime.ReadMemStats(&before)
313
314 info, rc, err := g.OpenBlob(ctx, f.Name, domain.Revision{}, readertest.TreePath(t, "huge.bin"))
315 require.NoError(t, err)
316 assert.Equal(t, int64(size), info.Size)
317 assert.True(t, info.Binary)
318 assert.Equal(t, "application/octet-stream", info.MediaType)
319
320 h := sha256.New()
321 buf := make([]byte, 64<<10)
322 var total int64
323 var peak uint64
324 for i := 0; ; i++ {
325 n, err := rc.Read(buf)
326 total += int64(n)
327 h.Write(buf[:n])
328 if i%64 == 0 {
329 var m runtime.MemStats
330 runtime.ReadMemStats(&m)
331 peak = max(peak, m.HeapAlloc)
332 }
333 if errors.Is(err, io.EOF) {
334 break
335 }
336 require.NoError(t, err)
337 }
338 require.NoError(t, rc.Close())
339 runtime.ReadMemStats(&after)
340
341 assert.Equal(t, int64(size), total)
342 assert.Equal(t, want[:], h.Sum(nil), "content is intact")
343
344 allocated := after.TotalAlloc - before.TotalAlloc
345 assert.Less(t, allocated, uint64(8<<20), "bytes allocated while streaming 50 MiB: %d", allocated)
346 growth := int64(peak) - int64(before.HeapAlloc)
347 assert.Less(t, growth, int64(8<<20), "heap growth while streaming 50 MiB: %d", growth)
348 t.Logf("streamed %d MiB: allocated %d KiB, peak heap growth %d KiB", size>>20, allocated>>10, growth>>10)
349}
350
351// The one-shot git commands share a bounded number of slots: a burst of
352// requests waits for a slot, and gives up with its context.
353func TestExecSemaphoreBoundsOneShotCommands(t *testing.T) {
354 ctx := context.Background()
355 gate := t.TempDir() + "/gate"
356 // A git that blocks until the gate file exists.
357 script := writeScript(t, `while [ ! -e `+gate+` ]; do sleep 0.01; done; echo "git version 2.50.0"`)
358 g := gitcli.New(gitcli.Config{Bin: script, DataDir: t.TempDir(), Path: os.Getenv("PATH"), MaxExec: 2})
359 size, taken := g.ExecSlots()
360 require.Equal(t, 2, size)
361 require.Zero(t, taken)
362
363 errs := make(chan error, 2)
364 for range 2 {
365 go func() { _, err := g.CheckVersion(ctx); errs <- err }()
366 }
367 require.Eventually(t, func() bool { _, taken := g.ExecSlots(); return taken == 2 }, 5*time.Second, time.Millisecond)
368
369 // A third waits and gives up with its deadline: it never started git.
370 short, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
371 defer cancel()
372 _, err := g.CheckVersion(short)
373 assert.ErrorIs(t, err, context.DeadlineExceeded)
374
375 require.NoError(t, os.WriteFile(gate, nil, 0o600))
376 for range 2 {
377 <-errs // their outcome is the version check's, not the subject
378 }
379 require.Eventually(t, func() bool { _, taken := g.ExecSlots(); return taken == 0 }, 5*time.Second, time.Millisecond)
380}
381
382// Open blob streams are bounded; a refused stream leaves nothing behind and a
383// closed one frees its slot.
384func TestBlobStreamsAreBounded(t *testing.T) {
385 ctx := context.Background()
386 g, data := newReader(t, func(c *gitcli.Config) { c.MaxBlobStreams = 2 })
387 f := readertest.NewFixture(t, data, "bigone")
388 f.WriteBytes("big", readertest.RandomBytes(600<<10, 5)).Commit("big")
389 open := func() (io.ReadCloser, error) {
390 _, rc, err := g.OpenBlob(ctx, f.Name, domain.Revision{}, readertest.TreePath(t, "big"))
391 return rc, err
392 }
393
394 a, err := open()
395 require.NoError(t, err)
396 b, err := open()
397 require.NoError(t, err)
398 _, err = open()
399 assert.ErrorIs(t, err, domain.ErrResourceExhausted)
400 assert.Equal(t, 2, g.BlobStreams())
401
402 require.NoError(t, a.Close())
403 c, err := open()
404 require.NoError(t, err, "a closed stream frees its slot")
405 require.NoError(t, b.Close())
406 require.NoError(t, c.Close())
407 assert.Zero(t, g.BlobStreams())
408}